From 902b4d850820cc73ed572e7f911852846d23dea9 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 9 Oct 2026 15:56:12 +0000 Subject: [PATCH 1/2] feat(reporting): retire expired periods with tombstones and holds Adds retire_expired_periods (adcontextprotocol/adcp#7996 sections 4.2 and 4.3, period anchor per adcontextprotocol/adcp#8100): an ordered, resumable, period-aligned retirement in three phases. 1. Under the account lock, write a retiring tombstone (final revision, digests, revision and adjustment counts) and stop serving the period's row sets. 2. Delete the period's recorded object versions; a failure leaves the period retiring with every locator intact and the next pass resumes it first. 3. Under the account lock, delete the manifests, bodies and the period's ledger records, mark the tombstone retired and append a retirement change. A period qualifies once max(status_retention_days, ledger_record_retention_days) has passed since the later of its end, newest revision and newest adjustment. It is held by pending materializer work, reconciliation records, and registered feed consumers that have not passed its changes, until max_feed_hold_days. A retired period is never recreated: commit_obligation refuses it with PERIOD_RETIRED. Python extension tables guarded by append-only triggers cannot be deleted without the shared Core layout's retirement exemptions. A period that still has such records finishes phases 1 and 2, stays retiring and is reported as protected; phase 3 removes all of a period's records or none. Part of #1457. Co-Authored-By: Claude Sonnet 5.5 --- docs/reporting-row-storage.md | 29 ++ src/adcp/reporting/ledger/__init__.py | 2 + .../reporting/ledger/period_retirement.py | 369 ++++++++++++++++++ src/adcp/reporting/ledger/pg.py | 98 +++++ .../ledger/reporting_period_retirement.sql | 31 ++ .../test_reporting_period_retirement_pg.py | 250 ++++++++++++ .../test_reporting_row_retention_pg.py | 9 +- 7 files changed, 786 insertions(+), 2 deletions(-) create mode 100644 src/adcp/reporting/ledger/period_retirement.py create mode 100644 src/adcp/reporting/ledger/reporting_period_retirement.sql create mode 100644 tests/conformance/reporting/test_reporting_period_retirement_pg.py 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..5f15e1646 --- /dev/null +++ b/src/adcp/reporting/ledger/period_retirement.py @@ -0,0 +1,369 @@ +"""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", + ) + ) + + +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( + "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," + " (SELECT s.row_manifest_sha256 FROM adcp_reporting_row_sets s" + " WHERE s.row_set_id = latest.reporting_revision_id) AS 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", + (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..7175f7374 --- /dev/null +++ b/tests/conformance/reporting/test_reporting_period_retirement_pg.py @@ -0,0 +1,250 @@ +"""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 diff --git a/tests/conformance/reporting/test_reporting_row_retention_pg.py b/tests/conformance/reporting/test_reporting_row_retention_pg.py index 551adf74e..083e645f7 100644 --- a/tests/conformance/reporting/test_reporting_row_retention_pg.py +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -184,8 +184,13 @@ 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") From 5b2cba9e76ade6f2440d9d528fc35232ee73f775 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 11 Oct 2026 20:51:30 +0000 Subject: [PATCH 2/2] fix(reporting): retire periods on change-feed hosts that have no row storage The tombstone insert read adcp_reporting_row_sets unconditionally, so every candidate failed on a host with the change feed but no row storage. Co-Authored-By: Claude Sonnet 5.5 --- .../reporting/ledger/period_retirement.py | 71 +++++++++++-------- .../test_reporting_period_retirement_pg.py | 22 ++++++ .../test_reporting_row_retention_pg.py | 20 ++++-- 3 files changed, 78 insertions(+), 35 deletions(-) diff --git a/src/adcp/reporting/ledger/period_retirement.py b/src/adcp/reporting/ledger/period_retirement.py index 5f15e1646..7601f83bd 100644 --- a/src/adcp/reporting/ledger/period_retirement.py +++ b/src/adcp/reporting/ledger/period_retirement.py @@ -209,6 +209,48 @@ def _candidate_query(holds: str, *, only: bool) -> str: ) +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, @@ -247,34 +289,7 @@ async def _retire_period( ).fetchone() if still is None: raise NoLongerEligibleError - await connection.execute( - "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," - " (SELECT s.row_manifest_sha256 FROM adcp_reporting_row_sets s" - " WHERE s.row_set_id = latest.reporting_revision_id) AS 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", - (obligation_id, account_id), - ) + await connection.execute(_tombstone_insert(has_rows), (obligation_id, account_id)) if has_rows: from adcp.reporting.ledger.row_retention import mark_period_pruning diff --git a/tests/conformance/reporting/test_reporting_period_retirement_pg.py b/tests/conformance/reporting/test_reporting_period_retirement_pg.py index 7175f7374..18762b13c 100644 --- a/tests/conformance/reporting/test_reporting_period_retirement_pg.py +++ b/tests/conformance/reporting/test_reporting_period_retirement_pg.py @@ -248,3 +248,25 @@ async def test_a_period_that_stopped_qualifying_is_not_retired(tmp_path: Path) - ) 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 083e645f7..b693dae97 100644 --- a/tests/conformance/reporting/test_reporting_row_retention_pg.py +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -193,13 +193,19 @@ async def age(pool: Any, *, days: int) -> None: (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: