fix: a refused detach no longer starts the drop grace

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`

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())

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())

pg_partsmith/utils.py

@@ -229,20 +229,22 @@ 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.
 
     Line one is the ownership marker :func:`orphan_table_comment` has always
     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()}")

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")