feat(reporting): schedule retention in the production worker

docs/reporting-row-storage.md

@@ -381,3 +381,23 @@ reads answer as expired, while its records wait for the shared Core layout (pers
 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.
+
+## Scheduling
+
+`ReportingProductionSupport` runs the maintenance for the features your store opted into, every
+`maintenance_seconds` (default 300; `None` disables it), on the same worker loop as the producer and materializer:
+
+| Store option | Operation | Bound |
+| --- | --- | --- |
+| `row_storage` | `sweep_row_write_intents` | 100 intents |
+| `row_storage` or `change_feed` | `retire_expired_periods` | 20 periods |
+| `change_feed` | `prune_changes` | 10,000 rows |
+
+Each operation is bounded and resumable. A failing operation never stops the worker: the next interval retries it, and
+only the closed operation name (`REPORTING_MAINTENANCE_FAILED`, `intent_sweep`, `period_retirement` or
+`change_pruning`) is logged, never an exception, provider message or identifier. An operation that finishes but reports
+failed items (a binding that cannot delete, for example) is logged the same way on every pass it does, so a stuck
+binding is visible; read `causes` from the result for the stable per-item code. `period_retirement` also removes the
+row bytes of every period it retires, so `prune_expired_row_sets` is not scheduled: call it yourself only if you want
+row bytes retired earlier than whole periods. Outside the production service, call
+the same methods from your own scheduler.

src/adcp/reporting/production/_diagnostics.py

@@ -41,3 +41,31 @@ def _worker_stopped(*, boundary: _WorkerBoundary) -> None:
         # The failed/stop latch has already committed. A broken operator sink
         # cannot prevent sibling shutdown or replace the closed public error.
         return
+
+
+_MAINTENANCE_OPERATIONS = frozenset({"intent_sweep", "period_retirement", "change_pruning"})
+
+
+def _maintenance_failed(*, operation: str) -> None:
+    """A maintenance pass failed. The worker keeps running; the next interval retries.
+
+    Only the closed operation name is recorded, never an exception, a provider
+    message or an identifier.
+    """
+    if not _LOGGER.isEnabledFor(logging.WARNING):
+        return
+    record = logging.LogRecord(
+        _LOGGER.name, logging.WARNING, "", 0, "Reporting maintenance pass failed", (), None
+    )
+    record.threadName = None
+    record.processName = None
+    if hasattr(record, "taskName"):
+        record.taskName = None
+    record.__dict__.update(
+        code="REPORTING_MAINTENANCE_FAILED",
+        operation=operation if operation in _MAINTENANCE_OPERATIONS else "maintenance",
+    )
+    try:
+        _LOGGER.handle(record)
+    except Exception:
+        return

src/adcp/reporting/production/service.py

@@ -173,6 +173,7 @@ def __init__(
         poll_seconds: float = 0.25,
         adcp_version: str | None = None,
         source_registry: ReportingProductionSourceRegistry | None = None,
+        maintenance_seconds: float | None = 300.0,
     ) -> None:
         from adcp.reporting.outbox.worker import ReportingNotificationWorker
         from adcp.reporting.production.handler import ReportingProductionHandler
@@ -195,6 +196,7 @@ def __init__(
             or not 1 <= status_retention_days <= 36500
             or automated_recovery_window <= timedelta(0)
             or not 0.01 <= poll_seconds <= 60
+            or (maintenance_seconds is not None and not 0.01 <= maintenance_seconds <= 86_400)
             or type(notification_workers) is not tuple
             or any(type(w) is not ReportingNotificationWorker for w in notification_workers)
         ):
@@ -221,6 +223,8 @@ def __init__(
             status_retention_days,
         )
         self.notification_workers, self.poll_seconds = notification_workers, poll_seconds
+        self.maintenance_seconds = maintenance_seconds
+        self._next_maintenance = 0.0
         from adcp.reporting.production.notifications import worker_identity
 
         self._notification_identity = tuple(worker_identity(w) for w in notification_workers)
@@ -688,6 +692,7 @@ async def _run(self) -> None:
                 await self.projection.rebuild_one()
                 boundary = "sweeper"
                 await self.projection.sweep_one()
+                await self._maintain()
                 self._started.set()
                 try:
                     await asyncio.wait_for(self._stop.wait(), self.poll_seconds)
@@ -705,6 +710,44 @@ async def _run(self) -> None:
         finally:
             self._started.set()
 
+    async def _maintain(self) -> None:
+        """Scheduled retention: abandoned uploads, expired periods, old change rows.
+
+        Each operation is bounded and resumable, runs only for the features the store
+        opted into (row storage, the change feed), and a failure never stops the worker:
+        the next interval retries and only the operation's name is logged.
+        """
+        if self.maintenance_seconds is None:
+            return
+        loop = asyncio.get_running_loop()
+        if loop.time() < self._next_maintenance:
+            return
+        self._next_maintenance = loop.time() + self.maintenance_seconds
+        from adcp.reporting.production._diagnostics import _maintenance_failed
+
+        store: Any = self.store
+        operations: list[tuple[str, Any]] = []
+        if getattr(store, "_row_storage", None) is not None:
+            operations.append(("intent_sweep", lambda: store.sweep_row_write_intents(limit=100)))
+        if getattr(store, "_row_storage", None) is not None or getattr(
+            store, "_change_feed", False
+        ):
+            operations.append(("period_retirement", lambda: store.retire_expired_periods(limit=20)))
+        if getattr(store, "_change_feed", False):
+            operations.append(("change_pruning", lambda: store.prune_changes()))
+        for name, run in operations:
+            try:
+                result = await run()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                _maintenance_failed(operation=name)
+                continue
+            # Sweeps and retirement report per-item failures in their result instead of
+            # raising; a binding that fails every pass must not be silent.
+            if getattr(result, "failed", None):
+                _maintenance_failed(operation=name)
+
     async def _run_notifications(self) -> None:
         from adcp.reporting.production.notifications import notification_turn
 

tests/test_reporting_production_maintenance.py

@@ -0,0 +1,105 @@
+"""The production worker schedules retention: sweeps, period retirement and change pruning."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from adcp.reporting.production.service import ReportingProductionSupport
+
+
+class Store:
+    def __init__(self, *, rows: bool, feed: bool, fail: str | None = None) -> None:
+        self._row_storage = object() if rows else None
+        self._change_feed = feed
+        self.calls: list[str] = []
+        self.fail = fail
+        self.report: Any = None
+
+    async def _record(self, name: str) -> None:
+        self.calls.append(name)
+        if self.fail == name:
+            raise RuntimeError("secret bucket gs://x token=SECRET")
+
+    async def sweep_row_write_intents(self, **_: Any) -> Any:
+        await self._record("sweep")
+        return self.report
+
+    async def retire_expired_periods(self, **_: Any) -> Any:
+        await self._record("retire")
+        return self.report
+
+    async def prune_changes(self, **_: Any) -> None:
+        await self._record("prune")
+
+
+def service(store: Any, seconds: float | None = 300.0) -> ReportingProductionSupport:
+    instance = object.__new__(ReportingProductionSupport)
+    instance.store = store
+    instance.maintenance_seconds = seconds
+    instance._next_maintenance = 0.0
+    return instance
+
+
+async def test_maintenance_runs_only_what_the_store_opted_into() -> None:
+    cases = [
+        (dict(rows=True, feed=True), ["sweep", "retire", "prune"]),
+        (dict(rows=True, feed=False), ["sweep", "retire"]),
+        (dict(rows=False, feed=True), ["retire", "prune"]),
+        (dict(rows=False, feed=False), []),
+    ]
+    for options, expected in cases:
+        store = Store(**options)
+        await service(store)._maintain()
+        assert store.calls == expected, options
+
+
+async def test_maintenance_waits_for_its_interval_and_can_be_disabled() -> None:
+    store = Store(rows=True, feed=True)
+    worker = service(store, seconds=3600)
+    await worker._maintain()
+    await worker._maintain()
+    assert store.calls == ["sweep", "retire", "prune"]
+    disabled = Store(rows=True, feed=True)
+    await service(disabled, seconds=None)._maintain()
+    assert disabled.calls == []
+
+
+async def test_a_failing_operation_never_stops_the_worker_or_leaks(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    store = Store(rows=True, feed=True, fail="sweep")
+    with caplog.at_level(logging.WARNING, logger="adcp.reporting.production"):
+        await service(store)._maintain()
+    # The other operations still ran, and only the closed operation name is recorded.
+    assert store.calls == ["sweep", "retire", "prune"]
+    (record,) = [
+        r for r in caplog.records if getattr(r, "code", "") == "REPORTING_MAINTENANCE_FAILED"
+    ]
+    assert record.operation == "intent_sweep"
+    assert "SECRET" not in record.getMessage() and record.exc_info is None
+
+
+async def test_per_item_failures_in_a_result_are_logged_by_operation_name(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    from types import SimpleNamespace
+
+    store = Store(rows=True, feed=False)
+    store.report = SimpleNamespace(failed=("rpo_secret_account",))
+    with caplog.at_level(logging.WARNING, logger="adcp.reporting.production"):
+        await service(store)._maintain()
+    records = [
+        r for r in caplog.records if getattr(r, "code", "") == "REPORTING_MAINTENANCE_FAILED"
+    ]
+    assert [r.operation for r in records] == ["intent_sweep", "period_retirement"]
+    assert all("secret" not in r.getMessage() and r.exc_info is None for r in records)
+    store.report = SimpleNamespace(failed=())
+    caplog.clear()
+    worker = service(store)
+    await worker._maintain()
+    assert not [
+        r for r in caplog.records if getattr(r, "code", "") == "REPORTING_MAINTENANCE_FAILED"
+    ]