From d505fbbc197842bfdd1a486ec28c0908ffeda1a0 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev Date: Fri, 16 Oct 2026 18:45:13 +0400 Subject: [PATCH] fix: a refused detach no longer starts the drop grace A refused DETACH ... CONCURRENTLY left its orphan marker, detached-at included, on a partition that stayed attached, and later detaches kept that instant, so DropAfter's grace counted from the first refusal. A refusal (55006, 23503) now takes the marker back off, keeping the table's own comment, and a detach of a still-attached partition always records its own instant. Closes #105 --- CHANGELOG.md | 13 ++++ pg_partsmith/aio/repositories/remover.py | 78 ++++++++++++++++++--- pg_partsmith/sync/repositories/remover.py | 77 ++++++++++++++++++-- pg_partsmith/utils.py | 6 +- tests/integration/aio/test_foreign_keys.py | 39 ++++++++++- tests/integration/sync/test_foreign_keys.py | 39 ++++++++++- 6 files changed, 231 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a59bba8..50c81b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * a backfilled window always attaches with writers held at the parent ([#89](https://github.com/bedrock-python/pg-partsmith/issues/89)) ([b1260c1](https://github.com/bedrock-python/pg-partsmith/commit/b1260c1a8bcdd6388c3836d57eb548f8fe5959bb)) +## 1.8.3 — a refused detach no longer starts the drop grace + +A `DETACH … CONCURRENTLY` writes the orphan marker, with its `detached-at` instant, in a +transaction of its own before the statement runs. When PostgreSQL refused the statement — a +partition still referenced through a foreign key, say — the marker stayed on a partition that +was still attached, and every later detach kept its instant. The day the references went, the +detach succeeded and the next run dropped the table at once, its grace counted from the first +refusal ([#105](https://github.com/bedrock-python/pg-partsmith/issues/105)). + +A refusal now takes the marker back off, keeping whatever comment the table had before, and a +detach of a partition that is still attached always records its own instant, so a marker that +was left behind anyway cannot shorten the grace either. + ## 1.8.2 — cancellation stops a run_maintenance_safe `run_maintenance_safe()` caught `asyncio.CancelledError`, `KeyboardInterrupt` and `SystemExit` diff --git a/pg_partsmith/aio/repositories/remover.py b/pg_partsmith/aio/repositories/remover.py index 5d008fe..b53a648 100644 --- a/pg_partsmith/aio/repositories/remover.py +++ b/pg_partsmith/aio/repositories/remover.py @@ -24,6 +24,7 @@ from pg_partsmith.utils import ( build_ddl_statement, coerce_str, + comment_without_markers, orphan_comment, parse_orphan_comment, pg_sqlstate, @@ -298,6 +299,7 @@ async def _detach_concurrently( try: return await self._run_concurrent_detach(table_name, partition_name) except (SQLAlchemyError, OSError, TimeoutError) as exc: + await self._unmark_if_refused(exc, partition_name, None) return self._classify_concurrent_failure(exc, table_name, partition_name, fallback=fallback) async def _detach_concurrently_pinned( @@ -327,6 +329,7 @@ async def statement() -> bool: statement_pid=lambda: pid_box[0], ) except (SQLAlchemyError, OSError, TimeoutError) as exc: + await self._unmark_if_refused(exc, partition_name, expected_oid) return self._classify_concurrent_failure(exc, table_name, partition_name, fallback=fallback) async def _run_concurrent_detach( @@ -343,17 +346,61 @@ async def _run_concurrent_detach( if pid_box is not None: backend = (await conn.execute(text("SELECT pg_backend_pid()"))).scalar() pid_box[0] = None if backend is None else int(backend) - already_marked = ( - await conn.execute( - text("SELECT obj_description(to_regclass(:partition_name), 'pg_class')"), - {"partition_name": to_regclass_argument(partition_name)}, - ) - ).scalar() - if parse_orphan_comment(coerce_str(already_marked), marker_prefix=self._marker_prefix) is None: - await self._mark_orphaned(conn, table_name, partition_name) + await self._mark_orphaned(conn, table_name, partition_name) await conn.execute(stmt) return True + async def _unmark_if_refused(self, exc: Exception, partition_name: str, expected_oid: int | None) -> None: + """Take the marker back off a partition PostgreSQL refused to detach. + + The marker went on in its own committed transaction before the + statement, so a refusal leaves an attached partition saying it was + detached. Only the marker lines go; the comment the table had before + stays. Best effort: the refusal is what the caller reports, and a + marker left behind is harmless because the next detach stamps its own + instant (see :meth:`_mark_orphaned`). + """ + if pg_sqlstate(exc) not in {"55006", "23503"}: + return + try: + async with asyncio.timeout(self._ddl_timeout), self._engine.begin() as conn: + await conn.execute(text(SET_LOCK_TIMEOUT_SQL), {"timeout": str(self._lock_timeout_ms)}) + foreign = await _relkind(conn, partition_name) == "f" + if not foreign: + # What COMMENT takes anyway, taken first, so the checks below see one relation. + await conn.execute( + build_ddl_statement( + "LOCK TABLE ONLY {partition} IN SHARE UPDATE EXCLUSIVE MODE", partition=partition_name + ) + ) + await self._ensure_expected_oid(conn, partition_name, expected_oid) + if not await _plainly_attached(conn, partition_name): + return + existing = coerce_str( + ( + await conn.execute( + text("SELECT obj_description(to_regclass(:partition_name), 'pg_class')"), + {"partition_name": to_regclass_argument(partition_name)}, + ) + ).scalar() + ) + if parse_orphan_comment(existing, marker_prefix=self._marker_prefix) is None: + return + rest = comment_without_markers(existing, marker_prefix=self._marker_prefix) + relation = "FOREIGN TABLE" if foreign else "TABLE" + if rest: + stmt = build_ddl_statement( + f"COMMENT ON {relation} {{partition}} IS [comment]", partition=partition_name, comment=rest + ) + else: + stmt = build_ddl_statement(f"COMMENT ON {relation} {{partition}} IS NULL", partition=partition_name) + await conn.execute(stmt) + except (SQLAlchemyError, OSError, TimeoutError, PlanStaleError) as cleanup_exc: + logger.warning( + "Could not take the orphan marker back off a partition whose detach was refused", + extra={"partition_name": partition_name, "error": str(cleanup_exc)}, + ) + def _classify_concurrent_failure( self, exc: Exception, table_name: str, partition_name: str, *, fallback: bool ) -> bool: @@ -393,6 +440,11 @@ async def _mark_orphaned( Written *before* the DETACH so an interrupted run leaves a marked table rather than an unmarked one that orphan discovery would never see. + An instant the table already carries is kept while it is detached or its + detach is pending. On a partition that is still plainly attached it is + what a refused ``DETACH … CONCURRENTLY`` left behind, not a detach, so + this detach records its own. + Args: stamp_now: Record the current instant as the detach time. Off for adoption, where the table was detached at an unknown time. @@ -411,6 +463,7 @@ async def _mark_orphaned( detached_at=instant, existing_comment=existing_comment, marker_prefix=self._marker_prefix, + keep_instant=not (stamp_now and await _plainly_attached(conn, partition_name)), ) if new_comment == existing_comment: return @@ -613,3 +666,12 @@ async def _handle_retry_delay(self, attempt: int, last_exc: BaseException | None async def _relkind(conn: AsyncConnection, name: str) -> str | None: """``pg_class.relkind`` of the relation holding ``name``, or None when there is none.""" return await relation_kind(conn, name) + + +async def _plainly_attached(conn: AsyncConnection, name: str) -> bool: + """Whether the relation is attached to a parent with no detach pending.""" + result = await conn.execute( + text("SELECT NOT inhdetachpending FROM pg_inherits WHERE inhrelid = to_regclass(:name)"), + {"name": to_regclass_argument(name)}, + ) + return bool(result.scalar()) diff --git a/pg_partsmith/sync/repositories/remover.py b/pg_partsmith/sync/repositories/remover.py index e6e2ca7..b7d9b05 100644 --- a/pg_partsmith/sync/repositories/remover.py +++ b/pg_partsmith/sync/repositories/remover.py @@ -24,6 +24,7 @@ from pg_partsmith.utils import ( build_ddl_statement, coerce_str, + comment_without_markers, orphan_comment, parse_orphan_comment, pg_sqlstate, @@ -301,6 +302,7 @@ def _detach_concurrently( try: return self._run_concurrent_detach(table_name, partition_name) except (SQLAlchemyError, OSError, TimeoutError) as exc: + self._unmark_if_refused(exc, partition_name, None) return self._classify_concurrent_failure(exc, table_name, partition_name, fallback=fallback) def _detach_concurrently_pinned( @@ -331,6 +333,7 @@ def statement() -> bool: statement_pid=lambda: pid_box[0], ) except (SQLAlchemyError, OSError, TimeoutError) as exc: + self._unmark_if_refused(exc, partition_name, expected_oid) return self._classify_concurrent_failure(exc, table_name, partition_name, fallback=fallback) def _run_concurrent_detach( @@ -348,16 +351,61 @@ def _run_concurrent_detach( if pid_box is not None: backend = (conn.execute(text("SELECT pg_backend_pid()"))).scalar() pid_box[0] = None if backend is None else int(backend) - already_marked = ( + self._mark_orphaned(conn, table_name, partition_name) + conn.execute(stmt) + return True + + def _unmark_if_refused(self, exc: Exception, partition_name: str, expected_oid: int | None) -> None: + """Take the marker back off a partition PostgreSQL refused to detach. + + The marker went on in its own committed transaction before the + statement, so a refusal leaves an attached partition saying it was + detached. Only the marker lines go; the comment the table had before + stays. Best effort: the refusal is what the caller reports, and a + marker left behind is harmless because the next detach stamps its own + instant (see :meth:`_mark_orphaned`). + """ + if pg_sqlstate(exc) not in {"55006", "23503"}: + return + try: + with self._engine.begin() as conn: + apply_local_statement_timeout(conn, self._ddl_timeout) + conn.execute(text(SET_LOCK_TIMEOUT_SQL), {"timeout": str(self._lock_timeout_ms)}) + foreign = _relkind(conn, partition_name) == "f" + if not foreign: + # What COMMENT takes anyway, taken first, so the checks below see one relation. conn.execute( - text("SELECT obj_description(to_regclass(:partition_name), 'pg_class')"), - {"partition_name": to_regclass_argument(partition_name)}, + build_ddl_statement( + "LOCK TABLE ONLY {partition} IN SHARE UPDATE EXCLUSIVE MODE", partition=partition_name + ) ) - ).scalar() - if parse_orphan_comment(coerce_str(already_marked), marker_prefix=self._marker_prefix) is None: - self._mark_orphaned(conn, table_name, partition_name) + self._ensure_expected_oid(conn, partition_name, expected_oid) + if not _plainly_attached(conn, partition_name): + return + existing = coerce_str( + ( + conn.execute( + text("SELECT obj_description(to_regclass(:partition_name), 'pg_class')"), + {"partition_name": to_regclass_argument(partition_name)}, + ) + ).scalar() + ) + if parse_orphan_comment(existing, marker_prefix=self._marker_prefix) is None: + return + rest = comment_without_markers(existing, marker_prefix=self._marker_prefix) + relation = "FOREIGN TABLE" if foreign else "TABLE" + if rest: + stmt = build_ddl_statement( + f"COMMENT ON {relation} {{partition}} IS [comment]", partition=partition_name, comment=rest + ) + else: + stmt = build_ddl_statement(f"COMMENT ON {relation} {{partition}} IS NULL", partition=partition_name) conn.execute(stmt) - return True + except (SQLAlchemyError, OSError, TimeoutError, PlanStaleError) as cleanup_exc: + logger.warning( + "Could not take the orphan marker back off a partition whose detach was refused", + extra={"partition_name": partition_name, "error": str(cleanup_exc)}, + ) def _classify_concurrent_failure( self, exc: Exception, table_name: str, partition_name: str, *, fallback: bool @@ -399,6 +447,11 @@ def _mark_orphaned( Written *before* the DETACH so an interrupted run leaves a marked table rather than an unmarked one that orphan discovery would never see. + An instant the table already carries is kept while it is detached or its + detach is pending. On a partition that is still plainly attached it is + what a refused ``DETACH … CONCURRENTLY`` left behind, not a detach, so + this detach records its own. + Args: stamp_now: Record the current instant as the detach time. Off for adoption, where the table was detached at an unknown time. @@ -417,6 +470,7 @@ def _mark_orphaned( detached_at=instant, existing_comment=existing_comment, marker_prefix=self._marker_prefix, + keep_instant=not (stamp_now and _plainly_attached(conn, partition_name)), ) if new_comment == existing_comment: return @@ -619,3 +673,12 @@ def _handle_retry_delay(self, attempt: int, last_exc: BaseException | None, part def _relkind(conn: Connection, name: str) -> str | None: """``pg_class.relkind`` of the relation holding ``name``, or None when there is none.""" return relation_kind(conn, name) + + +def _plainly_attached(conn: Connection, name: str) -> bool: + """Whether the relation is attached to a parent with no detach pending.""" + result = conn.execute( + text("SELECT NOT inhdetachpending FROM pg_inherits WHERE inhrelid = to_regclass(:name)"), + {"name": to_regclass_argument(name)}, + ) + return bool(result.scalar()) diff --git a/pg_partsmith/utils.py b/pg_partsmith/utils.py index c5e5ac3..9b53afb 100644 --- a/pg_partsmith/utils.py +++ b/pg_partsmith/utils.py @@ -229,6 +229,7 @@ def orphan_comment( detached_at: datetime | None, existing_comment: str | None, marker_prefix: str | None = None, + keep_instant: bool = True, ) -> str: """Compose the whole COMMENT for a table this library detached. @@ -236,13 +237,14 @@ def orphan_comment( written; line two records the detach instant; whatever comment the table carried before is kept below them. Re-marking an already marked table keeps the instant it already carries, so a repeated detach does not restart a - grace period. + grace period -- unless ``keep_instant`` is off, for a marker whose instant + records no detach at all. """ marker = orphan_table_comment(parent_table, marker_prefix=marker_prefix) lines = [marker] parsed = parse_orphan_comment(existing_comment, marker_prefix=marker_prefix) - if parsed is not None and parsed[1] is not None: + if keep_instant and parsed is not None and parsed[1] is not None: detached_at = parsed[1] if detached_at is not None: lines.append(f"{DETACHED_AT_MARKER}{detached_at.astimezone(UTC).isoformat()}") diff --git a/tests/integration/aio/test_foreign_keys.py b/tests/integration/aio/test_foreign_keys.py index 484f45a..35256ed 100644 --- a/tests/integration/aio/test_foreign_keys.py +++ b/tests/integration/aio/test_foreign_keys.py @@ -8,15 +8,16 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING import pytest import pytest_asyncio from pg_partsmith.entities import MaintenanceIssueStep -from pg_partsmith.lifecycle import AllOf, CreateAhead, ExpireIf, KeepNewest, LifecyclePolicy, Unreferenced +from pg_partsmith.lifecycle import AllOf, CreateAhead, DropAfter, ExpireIf, KeepNewest, LifecyclePolicy, Unreferenced from pg_partsmith.topology import FactKind +from pg_partsmith.utils import parse_orphan_comment from tests.integration.aio.support import ( exec_sql, is_attached, @@ -25,6 +26,7 @@ range_children_of, run_maintenance, scalar, + table_comment, ) from tests.integration.nested_support import MONTHLY_TABLE_DDL, monthly_config @@ -158,3 +160,36 @@ async def test__without_unreferenced__refused_detach_is_an_issue_and_the_run_goe assert not await is_attached(db_engine, f"{events}__2026_07") assert set(await range_children_of(db_engine, events)) == {f"{events}__2026_06", f"{events}__2026_08"} assert await scalar(db_engine, f'SELECT count(*) FROM "{refs}"') == 1 # noqa: S608 + + +async def test__refused_detach__leaves_no_marker_and_the_later_detach_stamps_its_own_instant( + db_engine: AsyncEngine, events: str, refs: str +) -> None: + """A refused detach records no detach, so the drop grace counts from the detach that happens.""" + # Arrange -- June is referenced and carries a comment of its own; detached tables wait a month + await _seed(db_engine, events, refs) + june = f"{events}__2026_06" + await exec_sql(db_engine, f"COMMENT ON TABLE \"{june}\" IS 'june archive'") + config = monthly_config( + events, + lifecycle=LifecyclePolicy( + creation=CreateAhead(count=1), retention=KeepNewest(count=1), drop=DropAfter(grace=timedelta(days=30)) + ), + ) + refused = await run_maintenance(db_engine, config, at_time="2026-08-26") + comment_after_refusal = await table_comment(db_engine, june) + + # Act -- the reference goes, and a run two days later detaches June + await exec_sql(db_engine, f'DELETE FROM "{refs}"') # noqa: S608 + detached = await run_maintenance(db_engine, config, at_time="2026-08-28") + + # Assert + assert [issue.step for issue in refused.issues] == [MaintenanceIssueStep.DETACH] + assert comment_after_refusal == "june archive" + assert detached.detached_count == 1 + assert not await is_attached(db_engine, june) + comment = await table_comment(db_engine, june) + marker = parse_orphan_comment(comment) + assert marker is not None + assert marker[1] == datetime(2026, 8, 28, tzinfo=UTC) + assert (comment or "").endswith("june archive") diff --git a/tests/integration/sync/test_foreign_keys.py b/tests/integration/sync/test_foreign_keys.py index 54820c7..66b5a6e 100644 --- a/tests/integration/sync/test_foreign_keys.py +++ b/tests/integration/sync/test_foreign_keys.py @@ -8,14 +8,15 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING import pytest from pg_partsmith.entities import MaintenanceIssueStep -from pg_partsmith.lifecycle import AllOf, CreateAhead, ExpireIf, KeepNewest, LifecyclePolicy, Unreferenced +from pg_partsmith.lifecycle import AllOf, CreateAhead, DropAfter, ExpireIf, KeepNewest, LifecyclePolicy, Unreferenced from pg_partsmith.topology import FactKind +from pg_partsmith.utils import parse_orphan_comment from tests.integration.nested_support import MONTHLY_TABLE_DDL, monthly_config from tests.integration.sync.support import ( exec_sql, @@ -25,6 +26,7 @@ range_children_of, run_maintenance, scalar, + table_comment, ) if TYPE_CHECKING: @@ -156,3 +158,36 @@ def test__without_unreferenced__refused_detach_is_an_issue_and_the_run_goes_on( assert not is_attached(sync_db_engine, f"{events}__2026_07") assert set(range_children_of(sync_db_engine, events)) == {f"{events}__2026_06", f"{events}__2026_08"} assert scalar(sync_db_engine, f'SELECT count(*) FROM "{refs}"') == 1 # noqa: S608 + + +def test__refused_detach__leaves_no_marker_and_the_later_detach_stamps_its_own_instant( + sync_db_engine: Engine, events: str, refs: str +) -> None: + """A refused detach records no detach, so the drop grace counts from the detach that happens.""" + # Arrange -- June is referenced and carries a comment of its own; detached tables wait a month + _seed(sync_db_engine, events, refs) + june = f"{events}__2026_06" + exec_sql(sync_db_engine, f"COMMENT ON TABLE \"{june}\" IS 'june archive'") + config = monthly_config( + events, + lifecycle=LifecyclePolicy( + creation=CreateAhead(count=1), retention=KeepNewest(count=1), drop=DropAfter(grace=timedelta(days=30)) + ), + ) + refused = run_maintenance(sync_db_engine, config, at_time="2026-08-26") + comment_after_refusal = table_comment(sync_db_engine, june) + + # Act -- the reference goes, and a run two days later detaches June + exec_sql(sync_db_engine, f'DELETE FROM "{refs}"') # noqa: S608 + detached = run_maintenance(sync_db_engine, config, at_time="2026-08-28") + + # Assert + assert [issue.step for issue in refused.issues] == [MaintenanceIssueStep.DETACH] + assert comment_after_refusal == "june archive" + assert detached.detached_count == 1 + assert not is_attached(sync_db_engine, june) + comment = table_comment(sync_db_engine, june) + marker = parse_orphan_comment(comment) + assert marker is not None + assert marker[1] == datetime(2026, 8, 28, tzinfo=UTC) + assert (comment or "").endswith("june archive")