feat(reporting): retire expired periods with tombstones and holds

docs/reporting-row-storage.md

@@ -352,3 +352,32 @@ that receives `CURSOR_EXPIRED` resynchronizes from this read.
 `reporting_obligations_account_period_idx` to back that order; a host without it can create the same index itself on
 `reporting_obligations (account_id, period_start, reporting_obligation_id)`. A cursor whose position is not a
 timestamp with a zone raises `ValueError`, never a driver error.
+
+## Period retirement
+
+`store.retire_expired_periods()` retires a whole period once its retention window has elapsed, as one ordered,
+resumable sequence that never deletes the records that locate row bytes while the bytes remain stored:
+
+1. **Tombstone.** Under the account lock, a `retiring` tombstone records the final revision, its digests, and the
+   revision and adjustment counts, and the period's row sets stop being served.
+2. **External bytes.** The period's recorded object versions are deleted. A failure leaves the period `retiring` with
+   every locator intact; the next pass resumes `retiring` periods before new candidates.
+3. **Records.** Under the account lock the manifests, bodies and the period's ledger records are deleted, the
+   tombstone becomes `retired`, and a `retirement` change is appended when the change feed is on.
+
+A period qualifies once the later of its period end, its newest revision's `created_at` and its newest adjustment's
+`created_at`, plus `max(status_retention_days, ledger_record_retention_days)`, has passed. It is held, and simply not a
+candidate this pass, while a materialization, a receipt or pending materializer work references it, or while a
+registered feed consumer updated within `max_feed_hold_days` has not passed its change rows. A silent consumer stops
+holding after `max_feed_hold_days`, so retention is never blocked forever. Eligibility is re-checked under the account
+lock before the tombstone is written, so a late revision cannot be retired with a stale candidate. A tombstoned period
+is never recreated or extended: `commit_obligation`, `commit_revision` and `commit_adjustment` refuse it with
+`PERIOD_RETIRED`.
+
+**Current limit.** Several Python extension tables (provisional observations, reconciliation records, notification
+events) are append-only by trigger and foreign key, and their guards are frozen catalog objects. A period that still has
+such records finishes phases 1 and 2, then stays `retiring` and is reported in `protected`: its row bytes are gone and
+reads answer as expired, while its records wait for the shared Core layout (persistence Stage 3), which carries
+retirement exemptions. Phase 3 deletes every record of a period or none.
+
+Tombstones live in `adcp_reporting_period_tombstones`, created with row storage or the change feed.

src/adcp/reporting/ledger/__init__.py

@@ -160,6 +160,7 @@
     derive_period,
     iso_duration_to_timedelta,
 )
+from adcp.reporting.ledger.period_retirement import ReportingPeriodRetirementV1
 from adcp.reporting.ledger.producer import (
     CurrencyResolver,
     FixedCurrencyResolver,
@@ -294,6 +295,7 @@
     "ReportingObligationRecord",
     "ReportingPeriodBoundary",
     "ReportingObjectRowBindingV1",
+    "ReportingPeriodRetirementV1",
     "ReportingPhysicalChecksum",
     "ReportingPostgresRowBindingV1",
     "ReportingProducer",

src/adcp/reporting/ledger/pg.py

@@ -152,6 +152,7 @@
     from adcp.reporting._source_authorization import InlineSealPublisher
     from adcp.reporting.inline_source import ReportingSealStore, SealedSlice
     from adcp.reporting.ledger.change_feed import ReportingChangeKindV1, ReportingChangesPageV1
+    from adcp.reporting.ledger.period_retirement import ReportingPeriodRetirementV1
     from adcp.reporting.ledger.row_migration import ReportingRowMigrationBatchV1
     from adcp.reporting.ledger.row_retention import ReportingRowIntentSweepV1, ReportingRowPruneV1
     from adcp.reporting.ledger.status_projection import ReportingStatusSnapshot
@@ -176,6 +177,7 @@
 _ACTIVITY_DDL_PATH = Path(__file__).parent / "reporting_webhook_activity.sql"
 _ROW_MIGRATION_DDL_PATH = Path(__file__).parent / "reporting_row_migration.sql"
 _CHANGES_DDL_PATH = Path(__file__).parent / "reporting_changes.sql"
+_RETIREMENT_DDL_PATH = Path(__file__).parent / "reporting_period_retirement.sql"
 
 __all__ = ["PG_AVAILABLE", "PgReportingLedgerStore"]
 
@@ -243,6 +245,7 @@ def __init__(
             )
         )
         self._row_sets_known = False
+        self._tombstones_known = False
         # Opt-in shared host change feed (adcp_reporting_changes), PostgreSQL 13+.
         self._change_feed = change_feed is True
         self._staged_rows: dict[tuple[str, str, str], Any] = {}
@@ -447,6 +450,32 @@ async def prune_changes(
             limit=limit,
         )
 
+    async def retire_expired_periods(
+        self,
+        *,
+        limit: int = 20,
+        account_id: str | None = None,
+        ledger_record_retention_days: int | None = None,
+        max_feed_hold_days: int = 7,
+    ) -> ReportingPeriodRetirementV1:
+        """Retire whole periods past ``max(status_retention_days, ledger_record_retention_days)``.
+
+        Tombstone and stop serving rows, delete external bytes, then delete the period's
+        records: ordered and resumable, so the records that locate row bytes are never
+        deleted while the bytes remain. Periods referenced by a materialization, a receipt
+        or an unread registered feed consumer are held. See
+        :mod:`adcp.reporting.ledger.period_retirement` for what Python can delete today.
+        """
+        from adcp.reporting.ledger.period_retirement import retire_expired_periods
+
+        return await retire_expired_periods(
+            self,
+            limit=limit,
+            account_id=account_id,
+            ledger_record_retention_days=ledger_record_retention_days,
+            max_feed_hold_days=max_feed_hold_days,
+        )
+
     async def retire_legacy_inline_rows(self) -> bool:
         """Rename ``reporting_revision_rows`` after every revision is migrated or quarantined.
 
@@ -474,6 +503,8 @@ async def _create_schema_on(self, connection: Any) -> None:
             await connection.execute(path.read_text())
         if self._change_feed:
             await connection.execute(_CHANGES_DDL_PATH.read_text())
+        if self._change_feed or self._row_storage is not None:
+            await connection.execute(_RETIREMENT_DDL_PATH.read_text())
         if self._row_storage is not None:
             await connection.execute(REPORTING_ROW_STORAGE_DDL_PATH.read_text())
             await connection.execute(_ROW_MIGRATION_DDL_PATH.read_text())
@@ -923,6 +954,58 @@ async def _require_writable_generation(
 
     # -- obligations ------------------------------------------------------
 
+    async def _period_is_tombstoned(
+        self, connection: Any, obligation: ReportingObligationRecord
+    ) -> bool:
+        """Whether retirement already took this period, so a replay cannot resurrect it."""
+        if not self._tombstones_known:
+            present = await (
+                await connection.execute(
+                    "SELECT to_regclass(quote_ident(current_schema())"
+                    " || '.adcp_reporting_period_tombstones')"
+                )
+            ).fetchone()
+            if present is None or present[0] is None:
+                return False
+            self._tombstones_known = True
+        key = obligation.generation_key
+        row = await (
+            await connection.execute(
+                "SELECT 1 FROM adcp_reporting_period_tombstones"
+                " WHERE reporting_obligation_id = %s OR (account_id = %s AND consumer_id = %s"
+                " AND delivery_config_id = %s AND delivery_config_version = %s"
+                " AND period_start = %s AND period_end = %s) LIMIT 1",
+                (
+                    obligation.reporting_obligation_id,
+                    key.account_id,
+                    key.consumer_id,
+                    key.delivery_config_id,
+                    key.delivery_config_version,
+                    obligation.period.start,
+                    obligation.period.end,
+                ),
+            )
+        ).fetchone()
+        return row is not None
+
+    async def _obligation_is_tombstoned(self, connection: Any, obligation_id: str) -> bool:
+        present = await (
+            await connection.execute(
+                "SELECT to_regclass(quote_ident(current_schema())"
+                " || '.adcp_reporting_period_tombstones')"
+            )
+        ).fetchone()
+        if present is None or present[0] is None:
+            return False
+        row = await (
+            await connection.execute(
+                "SELECT 1 FROM adcp_reporting_period_tombstones"
+                " WHERE reporting_obligation_id = %s LIMIT 1",
+                (obligation_id,),
+            )
+        ).fetchone()
+        return row is not None
+
     async def commit_obligation(
         self, obligation: ReportingObligationRecord
     ) -> ReportingObligationRecord:
@@ -947,6 +1030,11 @@ async def commit_obligation(
             ).fetchone()
             if existing_row is not None:
                 return _obligation_from_row(existing_row)
+            if await self._period_is_tombstoned(connection, obligation):
+                raise LedgerConflictError(
+                    "PERIOD_RETIRED",
+                    "this period was retired; a retired obligation is never recreated",
+                )
             require_frozen_currency(obligation.currency)
             try:
                 inserted = await (
@@ -1123,6 +1211,11 @@ async def commit_revision(
             await self._require_writable_generation(
                 connection, _obligation_from_row(obligation).generation_key
             )
+            if await self._period_is_tombstoned(connection, _obligation_from_row(obligation)):
+                raise LedgerConflictError(
+                    "PERIOD_RETIRED",
+                    "this period is being retired; no revision may be added to it",
+                )
             validate_revision_currency(_obligation_from_row(obligation), revision, rows)
             if revision.supersedes_reporting_revision_id:
                 await self._require_current_leaf(connection, revision)
@@ -2145,6 +2238,11 @@ async def commit_adjustment(
                 raise LedgerConflictError(
                     "REVISION_NOT_FOUND", "an adjustment must name a committed revision"
                 )
+            if await self._obligation_is_tombstoned(connection, revision[1]):
+                raise LedgerConflictError(
+                    "PERIOD_RETIRED",
+                    "this period is being retired; no adjustment may be added to it",
+                )
             if revision[0] != "official":
                 raise LedgerConflictError(
                     "ADJUSTMENT_REQUIRES_OFFICIAL",

src/adcp/reporting/ledger/reporting_period_retirement.sql

@@ -0,0 +1,31 @@
+-- Period tombstones for reporting period retirement.
+--
+-- A retired period keeps one compact record: it refuses replays that would resurrect
+-- the period, and it preserves the audit conclusion after the period's ledger records
+-- are deleted (adcontextprotocol/adcp#7996, section 4.2). Applied when row storage or the
+-- change feed is enabled. Python only: the JavaScript tombstone table is keyed by its own
+-- configuration identity, so this one keeps the Python generation key until the shared
+-- Core layout (persistence Stage 3).
+
+SELECT pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema()));
+
+CREATE TABLE IF NOT EXISTS adcp_reporting_period_tombstones (
+    reporting_obligation_id TEXT COLLATE "C" NOT NULL PRIMARY KEY,
+    account_id              TEXT COLLATE "C" NOT NULL,
+    consumer_id             TEXT COLLATE "C" NOT NULL,
+    delivery_config_id      TEXT COLLATE "C" NOT NULL,
+    delivery_config_version INTEGER NOT NULL,
+    period_start            TIMESTAMPTZ NOT NULL,
+    period_end              TIMESTAMPTZ NOT NULL,
+    state                   TEXT NOT NULL CHECK (state IN ('retiring', 'retired')),
+    final_revision_id       TEXT COLLATE "C",
+    final_content_sha256    TEXT COLLATE "C",
+    final_row_manifest_sha256 TEXT COLLATE "C",
+    revision_count          INTEGER NOT NULL CHECK (revision_count >= 0),
+    adjustment_count        INTEGER NOT NULL CHECK (adjustment_count >= 0),
+    retiring_at             TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
+    retired_at              TIMESTAMPTZ,
+    CHECK ((state = 'retired') = (retired_at IS NOT NULL))
+);
+CREATE INDEX IF NOT EXISTS adcp_reporting_period_tombstones_account
+    ON adcp_reporting_period_tombstones (account_id, delivery_config_id, period_start);