diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index 73bee17a4..8e9bdb507 100644 --- a/docs/reporting-row-storage.md +++ b/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. diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index 152be4cd9..9ec33ee75 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/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", diff --git a/src/adcp/reporting/ledger/period_retirement.py b/src/adcp/reporting/ledger/period_retirement.py new file mode 100644 index 000000000..7601f83bd --- /dev/null +++ b/src/adcp/reporting/ledger/period_retirement.py @@ -0,0 +1,384 @@ +"""Period retirement: retire a whole period once its retention window has elapsed. + +Shared SDK persistence spec, adcontextprotocol/adcp#7996, sections 4.2 and 4.3, and the +period-aligned anchor of adcontextprotocol/adcp#8100. A period is retired as a unit by one +ordered, resumable sequence, so the records that locate row bytes are never deleted while +the bytes remain stored: + +1. **Tombstone.** Under the account lock, write a ``retiring`` tombstone (final revision, + its digests, revision and adjustment counts) and mark the period's row sets + ``pruning``. Exact reads now answer as expired. +2. **External bytes.** Delete the period's recorded object versions. 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, delete the manifests and bodies, then the period's + ledger records, mark the tombstone ``retired`` and append a ``retirement`` change. + +**Eligibility.** A period qualifies once the later of its period end, its newest +revision's ``created_at`` and its newest adjustment's ``created_at`` is older than +``max(status_retention_days, ledger_record_retention_days)``. It is held, and simply not a +candidate this pass, while any materializer work or reconciliation record (a live +materialization or a receipt) references it, or while a live registered feed consumer +(updated within ``max_feed_hold_days``) has not passed its change rows in the consumer's +own feed order. A consumer silent past ``max_feed_hold_days`` stops holding, so retention +is never blocked forever. + +**What Python can delete today.** 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`` with cause ``RECORDS_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. Nothing is +half-deleted: phase 3 either removes every record of the period or none. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from adcp.reporting.ledger.row_retention import NoLongerEligibleError +from adcp.reporting.ledger.row_storage import ReportingRowStoreError + +if TYPE_CHECKING: + from adcp.reporting.ledger.pg import PgReportingLedgerStore + +__all__ = ["ReportingPeriodRetirementV1", "retire_expired_periods"] + + +@dataclass(frozen=True, slots=True) +class ReportingPeriodRetirementV1: + #: Periods whose records are now deleted and whose tombstone is ``retired``. + retired: int + #: Periods whose row bytes are gone but whose records remain, cause ``RECORDS_PROTECTED``. + protected: tuple[str, ...] + #: Periods that failed, with a secret-free cause, and resume on the next pass. + failed: tuple[str, ...] + causes: tuple[tuple[str, str], ...] + + +async def _table_exists(connection: Any, name: str) -> bool: + row = await ( + await connection.execute( + "SELECT to_regclass(quote_ident(current_schema()) || '.' || %s)", (name,) + ) + ).fetchone() + return row is not None and row[0] is not None + + +async def retire_expired_periods( + store: PgReportingLedgerStore, + *, + limit: int = 20, + account_id: str | None = None, + ledger_record_retention_days: int | None = None, + max_feed_hold_days: int = 7, +) -> ReportingPeriodRetirementV1: + if type(limit) is not int or not 1 <= limit <= 10_000: + raise ValueError("limit must be between 1 and 10000") + if type(max_feed_hold_days) is not int or not 1 <= max_feed_hold_days <= 36_500: + raise ValueError("max_feed_hold_days must be between 1 and 36500") + if ledger_record_retention_days is not None and ( + type(ledger_record_retention_days) is not int or ledger_record_retention_days < 1 + ): + raise ValueError("ledger_record_retention_days must be a positive integer") + async with store._pool.connection() as connection: + if not await _table_exists(connection, "adcp_reporting_period_tombstones"): + raise ReportingRowStoreError( + "STATE_UNAVAILABLE", + "period retirement requires row_storage or change_feed on the store", + ) + holds: list[str] = [] + if await _table_exists(connection, "reporting_materializer_work"): + holds.append( + " AND NOT EXISTS (SELECT 1 FROM reporting_materializer_work w" + " WHERE w.account_id = o.account_id" + " AND w.reporting_obligation_id = o.reporting_obligation_id)" + ) + if await _table_exists(connection, "reporting_reconciliation_records"): + holds.append( + " AND NOT EXISTS (SELECT 1 FROM reporting_reconciliation_records x" + " WHERE x.account_id = o.account_id" + " AND x.reporting_obligation_id = o.reporting_obligation_id)" + ) + if await _table_exists(connection, "adcp_reporting_changes") and await _table_exists( + connection, "adcp_reporting_feed_consumers" + ): + holds.append( + " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_changes change" + " JOIN adcp_reporting_feed_consumers consumer" + " ON consumer.updated_at > clock_timestamp()" + " - (%(hold_days)s::integer * INTERVAL '1 day')" + " WHERE change.obligation_id = o.reporting_obligation_id" + " AND ((consumer.account_id IS NULL AND (change.xid, change.seq) >" + " (consumer.cursor_xid, consumer.cursor_seq))" + " OR (consumer.account_id = change.account_id" + " AND change.seq > consumer.cursor_seq)))" + ) + query = _candidate_query("".join(holds), only=False) + candidates = await ( + await connection.execute( + query, + { + "account": account_id, + "only": None, + "record_days": ledger_record_retention_days, + "hold_days": max_feed_hold_days, + "limit": limit, + }, + ) + ).fetchall() + has_rows = await _table_exists(connection, "adcp_reporting_row_sets") + retired = 0 + protected: list[str] = [] + failed: list[str] = [] + causes: list[tuple[str, str]] = [] + options = (ledger_record_retention_days, max_feed_hold_days, "".join(holds)) + for obligation_id, owner in candidates: + try: + outcome = await _retire_period(store, obligation_id, owner, has_rows, options) + except NoLongerEligibleError: + continue + except Exception as error: + failed.append(obligation_id) + causes.append( + ( + obligation_id, + ( + error.code + if isinstance(error, ReportingRowStoreError) + else type(error).__name__ + ), + ) + ) + continue + if outcome == "retired": + retired += 1 + elif outcome == "protected": + protected.append(obligation_id) + return ReportingPeriodRetirementV1(retired, tuple(protected), tuple(failed), tuple(causes)) + + +def _candidate_query(holds: str, *, only: bool) -> str: + """Resumable ``retiring`` periods first, then expired periods that nothing holds. + + ``only=True`` is the recheck under the account lock: it asks whether this one + obligation still qualifies, so a late revision cannot be retired with its period. + """ + resumable = ( + "" + if only + else "".join( + ( + " SELECT t.reporting_obligation_id, t.account_id, 0 AS resume, t.period_end", + " FROM adcp_reporting_period_tombstones t WHERE t.state = 'retiring'", + " AND (%(account)s::text IS NULL OR t.account_id = %(account)s)", + " UNION ALL", + ) + ) + ) + return "".join( + ( + "SELECT reporting_obligation_id, account_id FROM (", + resumable, + " SELECT o.reporting_obligation_id, o.account_id, 1 AS resume, o.period_end", + " FROM reporting_obligations o JOIN reporting_configurations c", + " ON c.account_id = o.account_id AND c.consumer_id = o.consumer_id", + " AND c.delivery_config_id = o.delivery_config_id", + " AND c.delivery_config_version = o.delivery_config_version", + " WHERE (%(account)s::text IS NULL OR o.account_id = %(account)s)", + " AND o.reporting_obligation_id = %(only)s" if only else "", + " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_period_tombstones t", + " WHERE t.reporting_obligation_id = o.reporting_obligation_id)", + " AND GREATEST(o.period_end,", + " COALESCE((SELECT MAX(r.created_at) FROM reporting_revisions r", + " WHERE r.account_id = o.account_id", + " AND r.reporting_obligation_id = o.reporting_obligation_id),", + " o.period_end),", + " COALESCE((SELECT MAX(a.created_at) FROM reporting_adjustments a", + " JOIN reporting_revisions ar ON ar.reporting_revision_id =", + " a.adjusts_reporting_revision_id", + " WHERE ar.account_id = o.account_id", + " AND ar.reporting_obligation_id = o.reporting_obligation_id),", + " o.period_end))", + " + (GREATEST(c.status_retention_days, COALESCE(%(record_days)s::integer,", + " c.status_retention_days)) * INTERVAL '1 day') <= clock_timestamp()", + holds, + ") candidate ORDER BY resume, period_end, reporting_obligation_id LIMIT %(limit)s", + ) + ) + + +def _tombstone_insert(has_rows: bool) -> str: + """The phase 1 tombstone insert. ``adcp_reporting_row_sets`` exists only with row storage, + so a change-feed-only host records no row manifest instead of failing on the missing table.""" + manifest = ( + "(SELECT s.row_manifest_sha256 FROM adcp_reporting_row_sets s" + " WHERE s.row_set_id = latest.reporting_revision_id)" + if has_rows + else "NULL::text" + ) + return "".join( + ( + "INSERT INTO adcp_reporting_period_tombstones", + " (reporting_obligation_id, account_id, consumer_id, delivery_config_id,", + " delivery_config_version, period_start, period_end, state, final_revision_id,", + " final_content_sha256, final_row_manifest_sha256, revision_count,", + " adjustment_count)", + " SELECT o.reporting_obligation_id, o.account_id, o.consumer_id,", + " o.delivery_config_id, o.delivery_config_version, o.period_start,", + " o.period_end, 'retiring', latest.reporting_revision_id,", + " latest.revision_content_sha256, ", + manifest, + ", (SELECT count(*) FROM reporting_revisions r", + " WHERE r.account_id = o.account_id", + " AND r.reporting_obligation_id = o.reporting_obligation_id),", + " (SELECT count(*) FROM reporting_adjustments a", + " JOIN reporting_revisions ar", + " ON ar.reporting_revision_id = a.adjusts_reporting_revision_id", + " WHERE ar.account_id = o.account_id", + " AND ar.reporting_obligation_id = o.reporting_obligation_id)", + " FROM reporting_obligations o LEFT JOIN LATERAL (", + " SELECT r.reporting_revision_id, r.revision_content_sha256", + " FROM reporting_revisions r", + " WHERE r.account_id = o.account_id", + " AND r.reporting_obligation_id = o.reporting_obligation_id", + " ORDER BY (r.finality = 'official') DESC, r.created_at DESC,", + " r.reporting_revision_id DESC LIMIT 1) latest ON TRUE", + " WHERE o.reporting_obligation_id = %s AND o.account_id = %s", + " ON CONFLICT (reporting_obligation_id) DO NOTHING", + ) + ) + + +async def _retire_period( + store: PgReportingLedgerStore, + obligation_id: str, + account_id: str, + has_rows: bool, + options: tuple[int | None, int, str], +) -> str: + from psycopg import errors + + record_days, hold_days, holds = options + row_set_ids: list[str] = [] + # Phase 1: tombstone first, so expired rows read as expired from here on. A period + # that is not yet tombstoned is re-checked under the account lock: a late revision or + # adjustment may have committed since the candidate list was read. + async with store._pool.connection() as connection, connection.transaction(): + await store._lock_account(connection, account_id) + resuming = await ( + await connection.execute( + "SELECT 1 FROM adcp_reporting_period_tombstones" + " WHERE reporting_obligation_id = %s AND state = 'retiring'", + (obligation_id,), + ) + ).fetchone() + if resuming is None: + still = await ( + await connection.execute( + _candidate_query(holds, only=True), + { + "account": account_id, + "only": obligation_id, + "record_days": record_days, + "hold_days": hold_days, + "limit": 1, + }, + ) + ).fetchone() + if still is None: + raise NoLongerEligibleError + await connection.execute(_tombstone_insert(has_rows), (obligation_id, account_id)) + if has_rows: + from adcp.reporting.ledger.row_retention import mark_period_pruning + + row_set_ids = await mark_period_pruning(connection, obligation_id, account_id) + # Phase 2: external bytes, at their recorded versions. + if has_rows and store._row_storage is not None: + from adcp.reporting.ledger.row_retention import delete_period_objects + + await delete_period_objects(store, obligation_id, row_set_ids) + # Phase 3: the records, together. + async with store._pool.connection() as connection, connection.transaction(): + await store._lock_account(connection, account_id) + if has_rows: + from adcp.reporting.ledger.row_retention import _finish_period + + await _finish_period(connection, obligation_id, account_id, row_set_ids) + try: + async with connection.transaction(): # a savepoint: all of the records or none + await _delete_records(connection, obligation_id, account_id) + except (errors.RaiseException, errors.ForeignKeyViolation, errors.CheckViolation): + return "protected" + await connection.execute( + "UPDATE adcp_reporting_period_tombstones SET state = 'retired'," + " retired_at = clock_timestamp()" + " WHERE reporting_obligation_id = %s AND state = 'retiring'", + (obligation_id,), + ) + if store._change_feed: + await connection.execute( + "INSERT INTO adcp_reporting_changes" + " (account_id, record_kind, record_id, obligation_id)" + " VALUES (%s, 'retirement', %s, %s)", + (account_id, obligation_id, obligation_id), + ) + return "retired" + + +async def _delete_records(connection: Any, obligation_id: str, account_id: str) -> None: + revisions = ( + "SELECT reporting_revision_id FROM reporting_revisions" + " WHERE account_id = %(a)s AND reporting_obligation_id = %(o)s" + ) + params = {"a": account_id, "o": obligation_id} + if await _table_exists(connection, "adcp_reporting_row_sets"): + await connection.execute( + "DELETE FROM adcp_reporting_row_sets WHERE account_id = %(a)s" + " AND obligation_id = %(o)s AND rows_state IN ('pruned', 'pruning')", + params, + ) + await connection.execute( + "DELETE FROM reporting_consumer_statuses WHERE account_id = %(a)s" + f" AND (reporting_obligation_id = %(o)s OR reporting_revision_id IN ({revisions}))", # noqa: S608 # nosec B608 + params, + ) + await connection.execute( + "DELETE FROM reporting_adjustments WHERE account_id = %(a)s" + f" AND adjusts_reporting_revision_id IN ({revisions})", # noqa: S608 # nosec B608 + params, + ) + if await _table_exists(connection, "reporting_revision_rows"): + await connection.execute( + f"DELETE FROM reporting_revision_rows WHERE reporting_revision_id IN ({revisions})", # noqa: S608 # nosec B608 + params, + ) + if await _table_exists(connection, "adcp_reporting_legacy_revision_rows_v0"): + await connection.execute( + "DELETE FROM adcp_reporting_legacy_revision_rows_v0" + f" WHERE reporting_revision_id IN ({revisions})", # noqa: S608 # nosec B608 + params, + ) + await connection.execute( + "DELETE FROM reporting_restatement_checkpoints WHERE account_id = %(a)s" + " AND reporting_obligation_id = %(o)s", + params, + ) + for table in ("reporting_provisional_observations", "reporting_provisional_acquisitions"): + if await _table_exists(connection, table): + await connection.execute( + f"DELETE FROM {table} WHERE account_id = %(a)s" # noqa: S608 # nosec B608 + " AND reporting_obligation_id = %(o)s", + params, + ) + await connection.execute( + "DELETE FROM reporting_revisions WHERE account_id = %(a)s" + " AND reporting_obligation_id = %(o)s", + params, + ) + await connection.execute( + "DELETE FROM reporting_obligations WHERE account_id = %(a)s" + " AND reporting_obligation_id = %(o)s", + params, + ) diff --git a/src/adcp/reporting/ledger/pg.py b/src/adcp/reporting/ledger/pg.py index 952a9d319..0eba1baa5 100644 --- a/src/adcp/reporting/ledger/pg.py +++ b/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", diff --git a/src/adcp/reporting/ledger/reporting_period_retirement.sql b/src/adcp/reporting/ledger/reporting_period_retirement.sql new file mode 100644 index 000000000..46046029e --- /dev/null +++ b/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); diff --git a/tests/conformance/reporting/test_reporting_period_retirement_pg.py b/tests/conformance/reporting/test_reporting_period_retirement_pg.py new file mode 100644 index 000000000..18762b13c --- /dev/null +++ b/tests/conformance/reporting/test_reporting_period_retirement_pg.py @@ -0,0 +1,272 @@ +"""Whole-period retirement: ordered, resumable, held by unread consumers, tombstoned.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from adcp.reporting.ledger import PgReportingLedgerStore, ReportingRowStoreError +from adcp.reporting.ledger.change_feed import encode_reporting_change_cursor_v1 +from adcp.reporting.ledger.store import LedgerConflictError + +from ._generation_support import configuration, isolated_reporting_pool, obligation_for +from .test_reporting_row_retention_pg import age +from .test_reporting_row_storage_object_pg import lease, list_objects, options +from .test_reporting_row_storage_pg import fetch_all, many_rows, seeded, typed_revision + + +async def count(pool: Any, table: str) -> int: + return (await fetch_all(pool, f"SELECT count(*) FROM {table}"))[0][0] # noqa: S608 + + +async def committed_period(store: PgReportingLedgerStore, pool: Any, revisions: int = 2) -> Any: + obligation = await seeded(store) + await lease(pool) + previous = None + for index in range(revisions): + rows = many_rows(10 + index) + revision = typed_revision(obligation, rows, suffix=f"r{index}", supersedes=previous) + await store.stage_revision_rows(revision, rows) + await store.commit_revision(revision, rows) + previous = revision.reporting_revision_id + return obligation + + +async def test_an_expired_period_is_retired_in_three_ordered_phases(tmp_path: Path) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path), change_feed=True) + await store.create_schema() + await committed_period(store, pool) + assert len(list_objects(tmp_path / "rows")) == 2 + + # Inside the window nothing happens. + young = await store.retire_expired_periods() + assert (young.retired, young.protected, young.failed) == (0, (), ()) + + await age(pool, days=500) + result = await store.retire_expired_periods() + assert (result.retired, result.protected, result.failed) == (1, (), ()) + + assert list_objects(tmp_path / "rows") == [] + for table in ( + "reporting_obligations", + "reporting_revisions", + "adcp_reporting_row_sets", + "adcp_reporting_row_chunks", + "adcp_reporting_chunk_bodies", + ): + assert await count(pool, table) == 0, table + (tombstone,) = await fetch_all( + pool, + "SELECT state, final_revision_id, revision_count, adjustment_count," + " final_row_manifest_sha256 IS NOT NULL, retired_at IS NOT NULL" + " FROM adcp_reporting_period_tombstones", + ) + assert tombstone == ("retired", "rpr_acct_a_r1", 2, 0, True, True) + kinds = [r.kind for r in (await store.changes_after(account_id="acct_a")).records] + assert kinds == ["obligation", "revision", "revision", "retirement"] + + # A retired period is gone from reads and is never resurrected by a replay. + with pytest.raises(LedgerConflictError) as unknown: + await store.read_revision_rows( + account_id="acct_a", reporting_revision_id="rpr_acct_a_r1" + ) + assert unknown.value.code == "REVISION_NOT_FOUND" + with pytest.raises(LedgerConflictError) as replay: + await store.commit_obligation(obligation_for(configuration())) + assert replay.value.code == "PERIOD_RETIRED" + assert (await store.retire_expired_periods()).retired == 0 + + +async def test_the_window_is_anchored_at_the_newest_record_and_may_be_lengthened( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=True) + await store.create_schema() + obligation = await seeded(store) + rows = many_rows(3) + await store.commit_revision(typed_revision(obligation, rows, suffix="a"), rows) + await age(pool, days=500) + assert (await store.retire_expired_periods(ledger_record_retention_days=3650)).retired == 0 + # A newer revision moves the anchor: the period is not retired until it ages too. + newest = typed_revision(obligation, rows, suffix="b", supersedes="rpr_acct_a_a") + await store.commit_revision(newest, rows) + assert (await store.retire_expired_periods()).retired == 0 + await age(pool, days=500) + assert (await store.retire_expired_periods()).retired == 1 + for bad in (0, 10_001): + with pytest.raises(ValueError): + await store.retire_expired_periods(limit=bad) + with pytest.raises(ValueError): + await store.retire_expired_periods(ledger_record_retention_days=0) + + +async def test_an_unread_feed_consumer_holds_the_period_until_it_passes_or_goes_silent( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=True, change_feed=True) + await store.create_schema() + obligation = await seeded(store) + rows = many_rows(3) + await store.commit_revision(typed_revision(obligation, rows, suffix="a"), rows) + await age(pool, days=500) + await store.save_feed_consumer_cursor( + "sink", encode_reporting_change_cursor_v1(account="acct_a", seq="0") + ) + held = await store.retire_expired_periods() + assert (held.retired, held.failed) == (0, ()) + assert await count(pool, "reporting_obligations") == 1 + + # Silent past max_feed_hold_days: the hold releases and retention proceeds. + async with pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_feed_consumers SET updated_at = updated_at - INTERVAL '10 days'" + ) + assert (await store.retire_expired_periods(max_feed_hold_days=7)).retired == 1 + + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=True, change_feed=True) + await store.create_schema() + obligation = await seeded(store) + rows = many_rows(3) + await store.commit_revision(typed_revision(obligation, rows, suffix="a"), rows) + await age(pool, days=500) + everything = await store.changes_after(account_id="acct_a") + await store.save_feed_consumer_cursor("sink", everything.cursor) + assert (await store.retire_expired_periods()).retired == 1 + + +async def test_a_failed_external_delete_leaves_the_period_retiring_and_resumes( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + configured = options(tmp_path) + store = PgReportingLedgerStore(pool=pool, row_storage=configured, change_feed=True) + await store.create_schema() + await committed_period(store, pool, revisions=1) + await age(pool, days=500) + + without = PgReportingLedgerStore( + pool=pool, row_storage=type(configured)(deployment_namespace="test"), change_feed=True + ) + stuck = await without.retire_expired_periods() + assert stuck.retired == 0 and stuck.causes[0][1] == "STATE_UNAVAILABLE" + assert await fetch_all(pool, "SELECT state FROM adcp_reporting_period_tombstones") == [ + ("retiring",) + ] + assert await fetch_all(pool, "SELECT rows_state FROM adcp_reporting_row_sets") == [ + ("pruning",) + ] + assert await count(pool, "reporting_obligations") == 1 + assert len(list_objects(tmp_path / "rows")) == 1 + + resumed = await store.retire_expired_periods() + assert resumed.retired == 1 + assert list_objects(tmp_path / "rows") == [] + assert await count(pool, "reporting_obligations") == 0 + + +async def test_a_period_with_protected_records_loses_its_row_bytes_but_keeps_its_records() -> None: + from tests.test_reporting_settling import _capabilities, _harness, _revisions + + async with isolated_reporting_pool() as pool: + producer, store, _, _ = await _harness( + _capabilities(restatement_window="P3D"), + store_factory=lambda clock: PgReportingLedgerStore( + pool=pool, clock=clock, row_storage=True, change_feed=True + ), + ) + await producer.run_worker() + (revision,) = await _revisions(store) + await age(pool, days=500) + result = await store.retire_expired_periods() + # Provisional observations are append-only by trigger: all records or none. + assert result.retired == 0 and result.protected == (revision.reporting_obligation_id,) + assert await count(pool, "reporting_revisions") == 1 + assert await count(pool, "adcp_reporting_chunk_bodies") == 0 + assert await fetch_all(pool, "SELECT state FROM adcp_reporting_period_tombstones") == [ + ("retiring",) + ] + assert await fetch_all(pool, "SELECT rows_state FROM adcp_reporting_row_sets") == [ + ("pruned",) + ] + with pytest.raises(LedgerConflictError) as expired: + await store.read_revision_rows( + account_id=revision.account_id, + reporting_revision_id=revision.reporting_revision_id, + ) + assert expired.value.code == "REVISION_NOT_FOUND" + + +async def test_retirement_needs_the_tombstone_table() -> None: + async with isolated_reporting_pool() as pool: + plain = PgReportingLedgerStore(pool=pool) + await plain.create_schema() + with pytest.raises(ReportingRowStoreError) as failure: + await plain.retire_expired_periods() + assert failure.value.code == "STATE_UNAVAILABLE" + + +async def test_a_retiring_period_refuses_new_revisions(tmp_path: Path) -> None: + async with isolated_reporting_pool() as pool: + configured = options(tmp_path) + store = PgReportingLedgerStore(pool=pool, row_storage=configured, change_feed=True) + await store.create_schema() + obligation = await committed_period(store, pool, revisions=1) + await age(pool, days=500) + without = PgReportingLedgerStore( + pool=pool, row_storage=type(configured)(deployment_namespace="test"), change_feed=True + ) + await without.retire_expired_periods() # stuck in phase 2: the tombstone is 'retiring' + late_rows = many_rows(10) + late = typed_revision(obligation, late_rows, suffix="late") + with pytest.raises(LedgerConflictError) as refused: + await store.commit_revision(late, late_rows) + assert refused.value.code == "PERIOD_RETIRED" + + +async def test_a_period_that_stopped_qualifying_is_not_retired(tmp_path: Path) -> None: + from adcp.reporting.ledger.period_retirement import _retire_period + from adcp.reporting.ledger.row_retention import NoLongerEligibleError + + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path), change_feed=True) + await store.create_schema() + obligation = await committed_period(store, pool, revisions=1) + # Young records: a candidate list read earlier is stale by the time the lock is held. + with pytest.raises(NoLongerEligibleError): + await _retire_period( + store, + obligation.reporting_obligation_id, + obligation.account_id, + True, + (None, 7, ""), + ) + assert await count(pool, "adcp_reporting_period_tombstones") == 0 + assert await count(pool, "reporting_obligations") == 1 + + +async def test_a_change_feed_host_without_row_storage_retires_periods() -> None: + """``adcp_reporting_row_sets`` exists only with row storage; retirement must not need it.""" + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, change_feed=True) + await store.create_schema() + obligation = await seeded(store) + rows = many_rows(10) + revision = typed_revision(obligation, rows, suffix="inline") + await store.commit_revision(revision, rows) + await age(pool, days=500) + + result = await store.retire_expired_periods() + assert (result.retired, result.protected, result.failed) == (1, (), ()) + for table in ("reporting_obligations", "reporting_revisions", "reporting_revision_rows"): + assert await count(pool, table) == 0, table + assert await fetch_all( + pool, + "SELECT state, final_revision_id, final_row_manifest_sha256, revision_count" + " FROM adcp_reporting_period_tombstones", + ) == [("retired", revision.reporting_revision_id, None, 1)] diff --git a/tests/conformance/reporting/test_reporting_row_retention_pg.py b/tests/conformance/reporting/test_reporting_row_retention_pg.py index 551adf74e..b693dae97 100644 --- a/tests/conformance/reporting/test_reporting_row_retention_pg.py +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -184,17 +184,28 @@ async def age(pool: Any, *, days: int) -> None: await connection.execute(f"ALTER TABLE {table} ENABLE TRIGGER USER") # noqa: S608 await connection.execute("ALTER TABLE reporting_obligations DISABLE TRIGGER USER") await connection.execute( - "UPDATE reporting_obligations SET period_end = period_end - make_interval(days => %s)", - (days,), + "UPDATE reporting_obligations SET period_end = period_end - make_interval(days => %s)," + " period_start = period_start - make_interval(days => %s)," + " scope_resolved_at = scope_resolved_at - make_interval(days => %s)," + " expected_at = expected_at - make_interval(days => %s)," + " automated_recovery_deadline_at =" + " automated_recovery_deadline_at - make_interval(days => %s)", + (days, days, days, days, days), ) await connection.execute("ALTER TABLE reporting_obligations ENABLE TRIGGER USER") - await connection.execute("ALTER TABLE adcp_reporting_row_sets DISABLE TRIGGER USER") - await connection.execute( - "UPDATE adcp_reporting_row_sets" - " SET recorded_at = recorded_at - make_interval(days => %s)", - (days,), - ) - await connection.execute("ALTER TABLE adcp_reporting_row_sets ENABLE TRIGGER USER") + present = await ( + await connection.execute( + "SELECT to_regclass(quote_ident(current_schema()) || '.adcp_reporting_row_sets')" + ) + ).fetchone() + if present is not None and present[0] is not None: # row storage is optional + await connection.execute("ALTER TABLE adcp_reporting_row_sets DISABLE TRIGGER USER") + await connection.execute( + "UPDATE adcp_reporting_row_sets" + " SET recorded_at = recorded_at - make_interval(days => %s)", + (days,), + ) + await connection.execute("ALTER TABLE adcp_reporting_row_sets ENABLE TRIGGER USER") async def test_expired_periods_lose_their_row_bytes_in_every_binding(tmp_path: Path) -> None: