From 831621c01256f774822e04c41d92bdf0ae4ce545 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 9 Oct 2026 13:37:08 +0000 Subject: [PATCH 1/2] feat(reporting): sweep abandoned uploads and prune expired row sets Adds row retention (adcontextprotocol/adcp#7996, sections 4.2 and 4.3). - sweep_row_write_intents resolves abandoned object uploads. An expired open intent moves to sweeping under the account lock so no writer can commit against it, then only the objects that upload created are deleted at their recorded versions. Adopted objects and objects a committed chunk references are never deleted, and an interrupted sweep resumes. - prune_expired_row_sets retires a period's row bytes in an ordered, resumable sequence: rows stop being served first, external object versions are deleted next, and manifests and bodies go last, so a failed delete leaves every locator intact. Expired rows read like an unknown revision, never as an integrity failure. A period qualifies once max(status_retention_days, ledger_record_retention_days) has passed since its newest record and no materializer work is pending. Python keeps revision headers and their records until the shared Core layout (persistence Stage 3). Part of #1457. Co-Authored-By: Claude Sonnet 5.5 --- docs/reporting-row-storage.md | 34 ++ src/adcp/reporting/ledger/__init__.py | 3 + src/adcp/reporting/ledger/pg.py | 39 ++ src/adcp/reporting/ledger/row_retention.py | 374 ++++++++++++++++++ .../test_reporting_row_retention_pg.py | 335 ++++++++++++++++ 5 files changed, 785 insertions(+) create mode 100644 src/adcp/reporting/ledger/row_retention.py create mode 100644 tests/conformance/reporting/test_reporting_row_retention_pg.py diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index e4e6fecab..d98ccf0d8 100644 --- a/docs/reporting-row-storage.md +++ b/docs/reporting-row-storage.md @@ -237,3 +237,37 @@ azure = create_azure_blob_reporting_row_object_provider_v1( that version, then the version itself. The probe refuses containers that allow anonymous read access or that do not enforce create-only writes, and a write that returns no `version_id`. - The client you inject must belong to the binding's `account`. + +## Retention + +Two maintenance operations keep row storage bounded. Both are safe to run from any worker on a schedule, in bounded +batches; neither runs on its own, so schedule them next to your producer. + +```python +await store.sweep_row_write_intents(limit=100) # abandoned uploads +await store.prune_expired_row_sets(limit=20) # periods past their retention window +``` + +**Intent sweeps** resolve uploads whose ledger commit never happened (a crash, a lost lease, a numbering conflict). +An expired write intent moves to `sweeping` under the account lock, so no writer can commit against it. Only the +objects that upload *created* are then deleted, at their recorded native versions. Adopted objects are never recorded +as created, and any object a committed chunk references is skipped. A sweep that cannot delete (for example because +the binding is no longer configured) leaves the intent for the next pass. + +**Row pruning** retires one period's row bytes at a time, in an ordered, resumable sequence: + +1. Under the account lock, live row sets become `pruning`. Reads of those revisions now answer `REVISION_NOT_FOUND`, + exactly like an unknown revision, never as an integrity failure. +2. The period's recorded object versions are deleted outside any transaction. A failure leaves the period `pruning` + with every locator intact and the next pass resumes. +3. Under the account lock, chunk manifests and PostgreSQL bodies are deleted and the row sets become `pruned`. + +A period qualifies once the later of its period end and its newest revision or adjustment is older than +`max(status_retention_days, ledger_record_retention_days)`, and no materializer work is pending for it. The default +`ledger_record_retention_days` is the generation's `status_retention_days`; a shorter value is ignored, because a row +retention shorter than the advertised window is non-conformant. + +Python keeps revision headers and their records until the shared Core layout (persistence Stage 3). A pruned row set +stays as a small marker that makes reads answer as expired; the records are already outside the status projection, +whose horizon excludes periods past retention. Versioned buckets, soft delete and retention locks are allowed as +backstops, but a `pruned` period is not "erased" while they keep the old bytes. diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index 4209f3543..6aceac2d5 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/src/adcp/reporting/ledger/__init__.py @@ -163,6 +163,7 @@ revision_binding_algorithm, revision_content_sha256, ) +from adcp.reporting.ledger.row_retention import ReportingRowIntentSweepV1, ReportingRowPruneV1 from adcp.reporting.ledger.row_storage import ( ReportingObjectRowBindingV1, ReportingPostgresRowBindingV1, @@ -303,7 +304,9 @@ "ReportingRevisionReceiptRecord", "ReportingRevisionRecord", "ReportingRowBindingSelectionInputV1", + "ReportingRowIntentSweepV1", "ReportingRowPage", + "ReportingRowPruneV1", "ReportingRowStorageOptionsV1", "ReportingRowStoreError", "ReportingRowStoreErrorCode", diff --git a/src/adcp/reporting/ledger/pg.py b/src/adcp/reporting/ledger/pg.py index 1a25152ae..b54e167d0 100644 --- a/src/adcp/reporting/ledger/pg.py +++ b/src/adcp/reporting/ledger/pg.py @@ -150,6 +150,7 @@ from adcp.reporting._source_authorization import InlineSealPublisher from adcp.reporting.inline_source import ReportingSealStore, SealedSlice 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 try: @@ -358,6 +359,44 @@ async def migrate_inline_rows( self, limit=limit, account_id=account_id, cursor=cursor, binding_id=binding_id ) + async def sweep_row_write_intents( + self, *, limit: int = 100, account_id: str | None = None + ) -> ReportingRowIntentSweepV1: + """Resolve abandoned object uploads: delete only what an expired upload created. + + An expired write intent with no committed header moves to ``sweeping`` under + the account lock, so no writer can commit against it, then the objects it + created are deleted at their recorded versions. Adopted objects and objects a + committed chunk references are never touched. Schedule it alongside the + producer. + """ + from adcp.reporting.ledger.row_retention import sweep_row_write_intents + + return await sweep_row_write_intents(self, limit=limit, account_id=account_id) + + async def prune_expired_row_sets( + self, + *, + limit: int = 20, + account_id: str | None = None, + ledger_record_retention_days: int | None = None, + ) -> ReportingRowPruneV1: + """Retire the row bytes of periods past their retention window, resumably. + + Rows stop being served first, then external objects are deleted at their + recorded versions, then the manifests and bodies go. A failed delete leaves + the period ``pruning`` with every locator intact and the next pass resumes. + See :mod:`adcp.reporting.ledger.row_retention` for what qualifies a period. + """ + from adcp.reporting.ledger.row_retention import prune_expired_row_sets + + return await prune_expired_row_sets( + self, + limit=limit, + account_id=account_id, + ledger_record_retention_days=ledger_record_retention_days, + ) + async def retire_legacy_inline_rows(self) -> bool: """Rename ``reporting_revision_rows`` after every revision is migrated or quarantined. diff --git a/src/adcp/reporting/ledger/row_retention.py b/src/adcp/reporting/ledger/row_retention.py new file mode 100644 index 000000000..014d2615e --- /dev/null +++ b/src/adcp/reporting/ledger/row_retention.py @@ -0,0 +1,374 @@ +"""Row retention: intent sweeps and expired-period row pruning. + +Shared SDK persistence spec, adcontextprotocol/adcp#7996, sections 4.2 and 4.3. + +**Intent sweeps** resolve abandoned object uploads. An expired ``open`` write intent +whose ledger commit never happened moves to ``sweeping`` under the account lock, so no +writer can commit against it. Then only the objects that upload *created* are deleted, +at their recorded native versions. Adopted objects are never recorded as created, and +any object a committed chunk references is skipped. Interrupted sweeps resume. + +**Row pruning** retires a whole period's row bytes in an ordered, resumable sequence +(row bytes first, never the records that locate them): + +1. under the account lock, mark the period's live row sets ``pruning``. Exact reads + now answer as expired, like an unknown revision; +2. delete the period's recorded object versions outside any transaction. A failure + leaves the period ``pruning`` with every locator intact, and the next pass resumes; +3. under the account lock, delete the period's chunk manifests and PostgreSQL bodies + and mark its row sets ``pruned``. + +A period qualifies when every record in it is older than +``max(status_retention_days, ledger_record_retention_days)``, anchored at the later of +the period end and its newest revision or adjustment, and no materializer work is +pending for it. Python retains the ledger headers and their records until the shared +Core layout (persistence Stage 3), so a pruned row set stays as a small marker that +makes reads answer ``REVISION_NOT_FOUND``; the records stay outside the protocol +projection because the status horizon already excludes periods past retention. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from adcp.reporting.ledger.row_storage import ReportingRowStoreError + +if TYPE_CHECKING: + from adcp.reporting.ledger.pg import PgReportingLedgerStore + +__all__ = [ + "ReportingRowIntentSweepV1", + "ReportingRowPruneV1", + "prune_expired_row_sets", + "sweep_row_write_intents", +] + + +@dataclass(frozen=True, slots=True) +class ReportingRowIntentSweepV1: + swept: int + objects_deleted: int + failed: int + + +@dataclass(frozen=True, slots=True) +class ReportingRowPruneV1: + #: Obligations whose row bytes were fully retired in this pass. + retired: int + #: Obligations left ``pruning`` with locators intact, to resume on the next pass. + failed: tuple[str, ...] + #: Stable, secret-free cause per failed obligation. + causes: tuple[tuple[str, str], ...] + + +def _positive(value: int, name: str) -> None: + if type(value) is not int or value < 1 or value > 10_000: + raise ValueError(f"{name} must be between 1 and 10000") + + +async def sweep_row_write_intents( + store: PgReportingLedgerStore, *, limit: int = 100, account_id: str | None = None +) -> ReportingRowIntentSweepV1: + """Resolve abandoned object uploads. Safe to run on a schedule from any worker.""" + _positive(limit, "limit") + storage = store._row_storage + if storage is None: + raise ReportingRowStoreError( + "INVALID_INPUT", "sweep_row_write_intents requires row storage" + ) + await store.ready_row_storage() + async with store._pool.connection() as connection: + expired = await ( + await connection.execute( + "SELECT row_binding_id, account_id, row_set_id, content_sha256" + " FROM adcp_reporting_row_write_intents" + " WHERE (%(account)s::text IS NULL OR account_id = %(account)s)" + " AND row_binding_id = ANY(%(bindings)s::text[])" + " AND (state = 'sweeping' OR expires_at <= clock_timestamp())" + " ORDER BY expires_at LIMIT %(limit)s", + { + "account": account_id, + "bindings": list(storage._bindings), + "limit": limit, + }, + ) + ).fetchall() + swept = deleted = failed = 0 + for binding_id, owner, row_set_id, content_sha256 in expired: + key = (binding_id, row_set_id, content_sha256) + try: + async with store._pool.connection() as connection, connection.transaction(): + await store._lock_account(connection, owner) + claimed = await ( + await connection.execute( + "UPDATE adcp_reporting_row_write_intents SET state = 'sweeping'" + " WHERE row_binding_id = %s AND row_set_id = %s AND content_sha256 = %s" + " AND (state = 'sweeping' OR expires_at <= clock_timestamp())" + " RETURNING created_objects", + key, + ) + ).fetchone() + if claimed is None: + continue + # Referenced objects are kept; the check and the delete are one atomic step + # against a concurrent commit adopting the object. + orphaned = [(str(item["objectKey"]), str(item["nativeVersion"])) for item in claimed[0]] + removed = await storage.delete_unreferenced(store._pool, binding_id, orphaned) + async with store._pool.connection() as connection: + await connection.execute( + "DELETE FROM adcp_reporting_row_write_intents" + " WHERE row_binding_id = %s AND row_set_id = %s AND content_sha256 = %s" + " AND state = 'sweeping'", + key, + ) + swept += 1 + deleted += removed + except Exception: + failed += 1 + return ReportingRowIntentSweepV1(swept, deleted, failed) + + +def prune_candidate_query(materializer_hold: str, *, only: bool) -> str: + """Obligations whose row bytes are past their window. + + Shared by candidate selection and the per-period recheck under the account lock + (``only=True``), so a period that stopped qualifying between the two is skipped. + The anchor is the latest of the period end, the newest revision, the newest + adjustment and the newest row set recorded, the last on the database clock. + """ + return "".join( + ( + "SELECT o.reporting_obligation_id, o.account_id 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 EXISTS (SELECT 1 FROM adcp_reporting_row_sets s" + " WHERE s.account_id = o.account_id" + " AND s.obligation_id = o.reporting_obligation_id" + " AND s.rows_state IN ('live', 'unavailable', 'pruning'))" + " 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.reporting_obligation_id = o.reporting_obligation_id" + " AND ar.account_id = o.account_id), o.period_end)," + " COALESCE((SELECT MAX(rs.recorded_at) FROM adcp_reporting_row_sets rs" + " WHERE rs.account_id = o.account_id" + " AND rs.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()", + materializer_hold, + " ORDER BY EXISTS (SELECT 1 FROM adcp_reporting_row_sets p" + " WHERE p.obligation_id = o.reporting_obligation_id" + " AND p.rows_state = 'pruning') DESC, o.period_end," + " o.reporting_obligation_id LIMIT %(limit)s", + ) + ) + + +async def prune_expired_row_sets( + store: PgReportingLedgerStore, + *, + limit: int = 20, + account_id: str | None = None, + ledger_record_retention_days: int | None = None, +) -> ReportingRowPruneV1: + """Retire the row bytes of periods past their retention window. + + ``ledger_record_retention_days`` may only lengthen retention: a window shorter + than a generation's advertised ``status_retention_days`` is non-conformant, so + each period uses the longer of the two. + """ + _positive(limit, "limit") + 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") + storage = store._row_storage + if storage is None: + raise ReportingRowStoreError("INVALID_INPUT", "prune_expired_row_sets requires row storage") + await store.ready_row_storage() + async with store._pool.connection() as connection: + hold = await ( + await connection.execute( + "SELECT to_regclass(quote_ident(current_schema())" + " || '.reporting_materializer_work') IS NOT NULL" + ) + ).fetchone() + materializer_hold = ( + " 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" + " AND w.state = 'pending')" + if hold is not None and hold[0] + else "" + ) + candidates = await ( + await connection.execute( + prune_candidate_query(materializer_hold, only=False), + { + "account": account_id, + "only": None, + "record_days": ledger_record_retention_days, + "limit": limit, + }, + ) + ).fetchall() + retired = 0 + failed: list[str] = [] + causes: list[tuple[str, str]] = [] + for obligation_id, owner in candidates: + try: + await _retire_period( + store, obligation_id, owner, ledger_record_retention_days, materializer_hold + ) + retired += 1 + 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__ + ), + ) + ) + return ReportingRowPruneV1(retired, tuple(failed), tuple(causes)) + + +class NoLongerEligibleError(Exception): + """The period stopped qualifying between candidate selection and the account lock.""" + + +async def mark_period_pruning(connection: Any, obligation_id: str, account_id: str) -> list[str]: + """Stop serving a period's live row sets and return every set now ``pruning``. + + Call under the account lock. The returned IDs are the only sets later phases touch, so + a revision committed after this point can never lose its rows to this pass. + """ + await connection.execute( + "UPDATE adcp_reporting_row_sets SET rows_state = 'pruning'," + " rows_state_changed_at = clock_timestamp()" + " WHERE account_id = %s AND obligation_id = %s AND rows_state IN ('live', 'unavailable')", + (account_id, obligation_id), + ) + rows = await ( + await connection.execute( + "SELECT row_set_id FROM adcp_reporting_row_sets" + " WHERE account_id = %s AND obligation_id = %s AND rows_state = 'pruning'", + (account_id, obligation_id), + ) + ).fetchall() + return [row[0] for row in rows] + + +async def delete_period_objects( + store: PgReportingLedgerStore, obligation_id: str, row_set_ids: list[str] +) -> None: + """Delete the recorded object versions of ``row_set_ids``, at those versions. + + An object that a live chunk of another obligation still references is kept: the default + key template cannot collide across obligations, but a custom template can, and a + delete must never take bytes another revision still needs. See + ``ReportingRowStorageV1.delete_unreferenced`` for the locking contract. + """ + storage = store._row_storage + assert storage is not None + async with store._pool.connection() as connection: + objects = await ( + await connection.execute( + "SELECT DISTINCT s.row_binding_id, c.object_key, c.native_version" + " FROM adcp_reporting_row_chunks c JOIN adcp_reporting_row_sets s" + " ON s.row_set_id = c.row_set_id" + " WHERE c.row_set_id = ANY(%(ids)s::text[]) AND c.object_key IS NOT NULL", + {"ids": row_set_ids}, + ) + ).fetchall() + by_binding: dict[str, list[tuple[str, str]]] = {} + for binding_id, object_key, native_version in objects: + by_binding.setdefault(binding_id, []).append((object_key, native_version)) + for binding_id, pairs in by_binding.items(): + # Objects a chunk of any other row set still references are kept, checked + # atomically with the delete so a concurrent commit cannot adopt a doomed object. + await storage.delete_unreferenced( + store._pool, binding_id, pairs, obligation_id=obligation_id + ) + + +async def _retire_period( + store: PgReportingLedgerStore, + obligation_id: str, + account_id: str, + ledger_record_retention_days: int | None, + materializer_hold: str, +) -> None: + # Phase 1: stop serving the rows. Exact reads answer as expired from here on. The + # period is re-checked under the account lock: a late revision or restatement 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_row_sets WHERE account_id = %s AND obligation_id = %s" + " AND rows_state = 'pruning' LIMIT 1", + (account_id, obligation_id), + ) + ).fetchone() + if resuming is None: + still = await ( + await connection.execute( + prune_candidate_query(materializer_hold, only=True), + { + "account": account_id, + "only": obligation_id, + "record_days": ledger_record_retention_days, + "limit": 1, + }, + ) + ).fetchone() + if still is None: + raise NoLongerEligibleError + row_set_ids = await mark_period_pruning(connection, obligation_id, account_id) + # Phase 2: external bytes, at their recorded versions. + await delete_period_objects(store, obligation_id, row_set_ids) + # Phase 3: the manifests and bodies, together, then the marker state. + async with store._pool.connection() as connection, connection.transaction(): + await store._lock_account(connection, account_id) + await _finish_period(connection, obligation_id, account_id, row_set_ids) + + +async def _finish_period( + connection: Any, obligation_id: str, account_id: str, row_set_ids: list[str] +) -> None: + await connection.execute( + "DELETE FROM adcp_reporting_row_chunks c USING adcp_reporting_row_sets s" + " WHERE s.row_set_id = c.row_set_id AND s.rows_state = 'pruning'" + " AND s.row_set_id = ANY(%s::text[])", + (row_set_ids,), + ) + # Bodies are content-addressed within one obligation, so only bodies no remaining + # chunk of this obligation still names may go. + await connection.execute( + "DELETE FROM adcp_reporting_chunk_bodies b WHERE b.account_id = %s AND b.obligation_id = %s" + " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_row_chunks c" + " WHERE c.account_id = b.account_id AND c.obligation_id = b.obligation_id" + " AND c.sha256 = b.sha256)", + (account_id, obligation_id), + ) + await connection.execute( + "UPDATE adcp_reporting_row_sets SET rows_state = 'pruned'," + " rows_state_changed_at = clock_timestamp()" + " WHERE row_set_id = ANY(%s::text[]) AND rows_state = 'pruning'", + (row_set_ids,), + ) diff --git a/tests/conformance/reporting/test_reporting_row_retention_pg.py b/tests/conformance/reporting/test_reporting_row_retention_pg.py new file mode 100644 index 000000000..3ee11e7fa --- /dev/null +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -0,0 +1,335 @@ +"""Intent sweeps and expired-period row pruning.""" + +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.row_storage_object import ( + ReportingRowObjectContextV1, + ReportingRowObjectPutInputV1, + create_filesystem_reporting_row_object_provider_v1, +) +from adcp.reporting.ledger.row_write import prepare_revision_row_set +from adcp.reporting.ledger.store import LedgerConflictError + +from ._generation_support import isolated_reporting_pool +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, walk + + +async def staged_revision( + store: PgReportingLedgerStore, + obligation: Any, + suffix: str, + rows: list[dict[str, Any]], + **kwargs: Any, +) -> Any: + revision = typed_revision(obligation, rows, suffix=suffix, **kwargs) + assert await store.stage_revision_rows(revision, rows) is True + return revision + + +async def expire_intents(pool: Any) -> None: + async with pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_row_write_intents" + " SET expires_at = clock_timestamp() - INTERVAL '1 second'" + ) + + +async def test_a_sweep_deletes_only_the_objects_an_abandoned_upload_created(tmp_path: Path) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path)) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + rows = many_rows(10) + await staged_revision(store, obligation, "abandoned", rows) + assert len(list_objects(tmp_path / "rows")) == 1 + + # A live intent is nobody's to sweep yet. + idle = await store.sweep_row_write_intents() + assert (idle.swept, idle.objects_deleted, idle.failed) == (0, 0, 0) + assert len(list_objects(tmp_path / "rows")) == 1 + + await expire_intents(pool) + result = await store.sweep_row_write_intents() + assert (result.swept, result.objects_deleted, result.failed) == (1, 1, 0) + assert list_objects(tmp_path / "rows") == [] + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_row_write_intents") == [ + (0,) + ] + + +async def test_a_sweep_never_deletes_adopted_or_committed_objects(tmp_path: Path) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path)) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + + # An adopted object is not recorded as created, so the sweep leaves it alone. + rows = many_rows(10) + revision = typed_revision(obligation, rows, suffix="adopted") + prepared = await prepare_revision_row_set( + store, revision, rows, period_start="2026-09-01T00:00:00Z" + ) + provider = create_filesystem_reporting_row_object_provider_v1( + roots={"rows": tmp_path / "rows"} + ) + await provider.put_if_absent( + ReportingRowObjectPutInputV1( + {"root": "rows"}, + prepared.objects[0].key, + prepared.objects[0].physical_bytes, + "application/gzip", + {}, + ), + ReportingRowObjectContextV1(), + ) + await store.stage_revision_rows(revision, rows) + await expire_intents(pool) + result = await store.sweep_row_write_intents() + assert (result.swept, result.objects_deleted) == (1, 0) + assert len(list_objects(tmp_path / "rows")) == 1 + + # An intent whose chunk was committed (a crash after commit, before cleanup) is + # swept without touching the object a header now references. + other = many_rows(11) + committed = await staged_revision(store, obligation, "committed", other) + await store.commit_revision(committed, other) + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_row_write_intents (row_binding_id, account_id," + " row_set_id, content_sha256, state, owner_token, created_objects, expires_at)" + " SELECT 'reports', account_id, 'rpr_ghost', %s, 'open', 'ghost'," + " jsonb_build_array(jsonb_build_object('objectKey', object_key," + " 'nativeVersion', native_version))," + " clock_timestamp() - INTERVAL '1 second'" + " FROM adcp_reporting_row_chunks WHERE row_set_id = %s", + ("f" * 64, committed.reporting_revision_id), + ) + before = list_objects(tmp_path / "rows") + result = await store.sweep_row_write_intents() + assert (result.swept, result.objects_deleted) == (1, 0) + assert list_objects(tmp_path / "rows") == before + walked, _ = await walk(store, obligation.account_id, committed.reporting_revision_id, 500) + assert walked == other + + +async def test_an_interrupted_sweep_resumes_and_a_swept_intent_blocks_its_commit( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path)) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + rows = many_rows(10) + revision = await staged_revision(store, obligation, "resume", rows) + # A sweep that crashed after claiming the intent but before deleting anything. + async with pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_row_write_intents SET state = 'sweeping'" + ) + with pytest.raises(ReportingRowStoreError): + await store.commit_revision(revision, rows) + result = await store.sweep_row_write_intents() + assert (result.swept, result.objects_deleted, result.failed) == (1, 1, 0) + assert list_objects(tmp_path / "rows") == [] + + +async def test_a_sweep_that_cannot_delete_keeps_the_intent_for_the_next_pass( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + configured = options(tmp_path) + store = PgReportingLedgerStore(pool=pool, row_storage=configured) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + await staged_revision(store, obligation, "stuck", many_rows(10)) + await expire_intents(pool) + # A host that no longer configures the binding cannot delete: the intent stays. + without = PgReportingLedgerStore( + pool=pool, + row_storage=type(configured)(deployment_namespace="test"), + ) + result = await without.sweep_row_write_intents() + # Intents of bindings this host does not configure are skipped, so they can never + # starve the intents it can delete. + assert (result.swept, result.failed) == (0, 0) + assert await fetch_all(pool, "SELECT state FROM adcp_reporting_row_write_intents") == [ + ("open",) + ] + again = await store.sweep_row_write_intents() + assert (again.swept, again.objects_deleted) == (1, 1) + assert list_objects(tmp_path / "rows") == [] + + +async def age(pool: Any, *, days: int) -> None: + """Move every recorded instant ``days`` into the past, as time passing would.""" + async with pool.connection() as connection: + for table in ("reporting_revisions", "reporting_adjustments"): + await connection.execute(f"ALTER TABLE {table} DISABLE TRIGGER USER") # noqa: S608 + await connection.execute( + f"UPDATE {table} SET created_at = created_at - make_interval(days => %s)", # noqa: S608 + (days,), + ) + 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,), + ) + 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") + + +async def test_expired_periods_lose_their_row_bytes_in_every_binding(tmp_path: Path) -> None: + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path)) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + inline_rows = many_rows(5) + # canonical_byte_count is above zero, so the selector routes both to object + # storage; force one revision into PostgreSQL bodies through the migration path. + object_rows = many_rows(40) + first = typed_revision(obligation, inline_rows, suffix="a") + second = typed_revision( + obligation, object_rows, suffix="b", supersedes=first.reporting_revision_id + ) + for revision, rows in ((first, inline_rows), (second, object_rows)): + await store.stage_revision_rows(revision, rows) + await store.commit_revision(revision, rows) + assert len(list_objects(tmp_path / "rows")) == 2 + + # Inside the retention window nothing is pruned. + young = await store.prune_expired_row_sets() + assert young.retired == 0 and len(list_objects(tmp_path / "rows")) == 2 + + await age(pool, days=500) + with pytest.raises(ValueError): + await store.prune_expired_row_sets(ledger_record_retention_days=0) + pruned = await store.prune_expired_row_sets() + assert (pruned.retired, pruned.failed) == (1, ()) + assert list_objects(tmp_path / "rows") == [] + assert await fetch_all( + pool, "SELECT rows_state FROM adcp_reporting_row_sets ORDER BY row_set_id" + ) == [("pruned",), ("pruned",)] + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_row_chunks") == [(0,)] + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_chunk_bodies") == [(0,)] + + # Expired rows read like an unknown revision, never as an integrity failure. + for revision in (first, second): + with pytest.raises(LedgerConflictError) as expired: + await store.read_revision_rows( + account_id=obligation.account_id, + reporting_revision_id=revision.reporting_revision_id, + ) + assert expired.value.code == "REVISION_NOT_FOUND" + # The headers remain: records leave the ledger with the shared Core layout. + assert await fetch_all(pool, "SELECT count(*) FROM reporting_revisions") == [(2,)] + assert (await store.prune_expired_row_sets()).retired == 0 + + +async def test_postgres_bound_periods_lose_their_bodies(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(30) + revision = typed_revision(obligation, rows, suffix="pg") + await store.commit_revision(revision, rows) + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_chunk_bodies") == [(1,)] + await age(pool, days=500) + assert (await store.prune_expired_row_sets()).retired == 1 + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_chunk_bodies") == [(0,)] + + +async def test_a_longer_record_retention_keeps_the_period_a_pending_hold_keeps_it_too( + 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) + revision = typed_revision(obligation, rows, suffix="held") + await store.commit_revision(revision, rows) + await age(pool, days=500) + assert (await store.prune_expired_row_sets(ledger_record_retention_days=3650)).retired == 0 + assert await fetch_all(pool, "SELECT rows_state FROM adcp_reporting_row_sets") == [ + ("live",) + ] + assert (await store.prune_expired_row_sets(ledger_record_retention_days=30)).retired == 1 + + +async def test_a_failed_delete_leaves_the_period_pruning_with_its_locators_and_resumes( + tmp_path: Path, +) -> None: + async with isolated_reporting_pool() as pool: + configured = options(tmp_path) + store = PgReportingLedgerStore(pool=pool, row_storage=configured) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + rows = many_rows(10) + revision = typed_revision(obligation, rows, suffix="resume") + await store.stage_revision_rows(revision, rows) + await store.commit_revision(revision, rows) + await age(pool, days=500) + + # The host stopped configuring the binding: the delete cannot run. + without = PgReportingLedgerStore( + pool=pool, row_storage=type(configured)(deployment_namespace="test") + ) + stuck = await without.prune_expired_row_sets() + assert stuck.retired == 0 and stuck.failed and stuck.causes[0][1] == "STATE_UNAVAILABLE" + assert await fetch_all(pool, "SELECT rows_state FROM adcp_reporting_row_sets") == [ + ("pruning",) + ] + assert await fetch_all(pool, "SELECT count(*) FROM adcp_reporting_row_chunks") == [(1,)] + assert len(list_objects(tmp_path / "rows")) == 1 + # While pruning, reads already answer as expired. + with pytest.raises(LedgerConflictError) as expired: + await store.read_revision_rows( + account_id=obligation.account_id, + reporting_revision_id=revision.reporting_revision_id, + ) + assert expired.value.code == "REVISION_NOT_FOUND" + + resumed = await store.prune_expired_row_sets() + assert resumed.retired == 1 + assert list_objects(tmp_path / "rows") == [] + assert await fetch_all(pool, "SELECT rows_state FROM adcp_reporting_row_sets") == [ + ("pruned",) + ] + + +async def test_retention_requires_row_storage_and_sane_limits() -> None: + async with isolated_reporting_pool() as pool: + plain = PgReportingLedgerStore(pool=pool) + await plain.create_schema() + with pytest.raises(ReportingRowStoreError): + await plain.prune_expired_row_sets() + with pytest.raises(ReportingRowStoreError): + await plain.sweep_row_write_intents() + store = PgReportingLedgerStore(pool=pool, row_storage=True) + await store.create_schema() + for bad in (0, 10_001): + with pytest.raises(ValueError): + await store.sweep_row_write_intents(limit=bad) + with pytest.raises(ValueError): + await store.prune_expired_row_sets(limit=bad) From 34d05a88a70400741ba9a6e966cf91294db6eedb Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 11 Oct 2026 15:21:31 +0000 Subject: [PATCH 2/2] fix(reporting): report why an intent sweep failed Adds per-intent failure causes and a skipped-unconfigured count to the sweep result. Co-Authored-By: Claude Sonnet 5.5 --- docs/reporting-row-storage.md | 6 ++-- src/adcp/reporting/ledger/row_retention.py | 30 +++++++++++++++++-- .../test_reporting_row_retention_pg.py | 25 ++++++++++++++++ 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index d98ccf0d8..b843fc214 100644 --- a/docs/reporting-row-storage.md +++ b/docs/reporting-row-storage.md @@ -251,8 +251,10 @@ await store.prune_expired_row_sets(limit=20) # periods past their retention win **Intent sweeps** resolve uploads whose ledger commit never happened (a crash, a lost lease, a numbering conflict). An expired write intent moves to `sweeping` under the account lock, so no writer can commit against it. Only the objects that upload *created* are then deleted, at their recorded native versions. Adopted objects are never recorded -as created, and any object a committed chunk references is skipped. A sweep that cannot delete (for example because -the binding is no longer configured) leaves the intent for the next pass. +as created, and any object a committed chunk references is skipped. A sweep that cannot delete leaves the intent for +the next pass and reports a stable, secret-free code per failed intent in `causes` (`(row_set_id, code)`), so a +scheduled sweep that keeps failing is diagnosable. Intents of bindings this host does not configure are not this +host's to sweep; they are counted in `skipped_unconfigured` and left for the host that does. **Row pruning** retires one period's row bytes at a time, in an ordered, resumable sequence: diff --git a/src/adcp/reporting/ledger/row_retention.py b/src/adcp/reporting/ledger/row_retention.py index 014d2615e..01560c744 100644 --- a/src/adcp/reporting/ledger/row_retention.py +++ b/src/adcp/reporting/ledger/row_retention.py @@ -50,6 +50,10 @@ class ReportingRowIntentSweepV1: swept: int objects_deleted: int failed: int + #: Stable, secret-free cause per failed intent: ``(row_set_id, code)``. + causes: tuple[tuple[str, str], ...] = () + #: Expired intents of bindings this host does not configure; another host owns them. + skipped_unconfigured: int = 0 @dataclass(frozen=True, slots=True) @@ -94,6 +98,18 @@ async def sweep_row_write_intents( }, ) ).fetchall() + async with store._pool.connection() as connection: + skipped_row = await ( + await connection.execute( + "SELECT count(*) FROM adcp_reporting_row_write_intents" + " WHERE (%(account)s::text IS NULL OR account_id = %(account)s)" + " AND NOT (row_binding_id = ANY(%(bindings)s::text[]))" + " AND (state = 'sweeping' OR expires_at <= clock_timestamp())", + {"account": account_id, "bindings": list(storage._bindings)}, + ) + ).fetchone() + skipped = int(skipped_row[0]) if skipped_row else 0 + causes: list[tuple[str, str]] = [] swept = deleted = failed = 0 for binding_id, owner, row_set_id, content_sha256 in expired: key = (binding_id, row_set_id, content_sha256) @@ -124,9 +140,19 @@ async def sweep_row_write_intents( ) swept += 1 deleted += removed - except Exception: + except Exception as error: failed += 1 - return ReportingRowIntentSweepV1(swept, deleted, failed) + causes.append( + ( + row_set_id, + ( + error.code + if isinstance(error, ReportingRowStoreError) + else type(error).__name__ + ), + ) + ) + return ReportingRowIntentSweepV1(swept, deleted, failed, tuple(causes), skipped) def prune_candidate_query(materializer_hold: str, *, only: bool) -> str: diff --git a/tests/conformance/reporting/test_reporting_row_retention_pg.py b/tests/conformance/reporting/test_reporting_row_retention_pg.py index 3ee11e7fa..551adf74e 100644 --- a/tests/conformance/reporting/test_reporting_row_retention_pg.py +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -163,6 +163,7 @@ async def test_a_sweep_that_cannot_delete_keeps_the_intent_for_the_next_pass( # Intents of bindings this host does not configure are skipped, so they can never # starve the intents it can delete. assert (result.swept, result.failed) == (0, 0) + assert result.skipped_unconfigured == 1 assert await fetch_all(pool, "SELECT state FROM adcp_reporting_row_write_intents") == [ ("open",) ] @@ -333,3 +334,27 @@ async def test_retention_requires_row_storage_and_sane_limits() -> None: await store.sweep_row_write_intents(limit=bad) with pytest.raises(ValueError): await store.prune_expired_row_sets(limit=bad) + + +async def test_a_failed_sweep_reports_its_stable_cause(tmp_path: Path) -> None: + import json + + async with isolated_reporting_pool() as pool: + store = PgReportingLedgerStore(pool=pool, row_storage=options(tmp_path)) + await store.create_schema() + obligation = await seeded(store) + await lease(pool) + revision = await staged_revision(store, obligation, "corrupt", many_rows(10)) + # A corrupted locator outside the installation's owned prefix is never deleted. + async with pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_row_write_intents SET created_objects = %s::jsonb", + (json.dumps([{"objectKey": "other/tenant/object", "nativeVersion": "1:2:3:4"}]),), + ) + await expire_intents(pool) + result = await store.sweep_row_write_intents() + assert (result.swept, result.failed) == (0, 1) + assert result.causes == ((revision.reporting_revision_id, "UNSAFE_BINDING"),) + assert await fetch_all(pool, "SELECT state FROM adcp_reporting_row_write_intents") == [ + ("sweeping",) + ]