diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index 8e9bdb507..75cab075d 100644 --- a/docs/reporting-row-storage.md +++ b/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. diff --git a/src/adcp/reporting/production/_diagnostics.py b/src/adcp/reporting/production/_diagnostics.py index f2c852cf1..10b9c0942 100644 --- a/src/adcp/reporting/production/_diagnostics.py +++ b/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 diff --git a/src/adcp/reporting/production/service.py b/src/adcp/reporting/production/service.py index ce719c5cc..badcfe9fa 100644 --- a/src/adcp/reporting/production/service.py +++ b/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 diff --git a/tests/test_reporting_production_maintenance.py b/tests/test_reporting_production_maintenance.py new file mode 100644 index 000000000..db6bf8ae6 --- /dev/null +++ b/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" + ]