diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index e4e6fecab..b843fc214 100644 --- a/docs/reporting-row-storage.md +++ b/docs/reporting-row-storage.md @@ -237,3 +237,39 @@ 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 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: + +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..01560c744 --- /dev/null +++ b/src/adcp/reporting/ledger/row_retention.py @@ -0,0 +1,400 @@ +"""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 + #: 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) +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() + 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) + 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 as error: + failed += 1 + 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: + """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..551adf74e --- /dev/null +++ b/tests/conformance/reporting/test_reporting_row_retention_pg.py @@ -0,0 +1,360 @@ +"""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 result.skipped_unconfigured == 1 + 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) + + +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",) + ]