feat(reporting): sweep abandoned uploads and prune expired row sets

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.

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",

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.