diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index d4801e9887..0e5fdd7251 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -649,6 +649,104 @@ Supply `bind_managed_generation(configuration, version)`. Keep the engine's `status_retention_days` equal to the service's. +## Observe reporting issues + +A `CONSUMER_STATUS_MISMATCH` is caller-scoped: only the buyer whose +`sync_reporting_status` statement caused it sees it in `get_reporting_status`. +To hear about one without waiting for the buyer to call, pass an +`issue_observer` ([#1258](https://github.com/adcontextprotocol/adcp-client-python/issues/1258)): + +```python +from adcp.reporting import ReliableReportingService, ReportingIssueEvent + +async def on_issue(event: ReportingIssueEvent) -> None: + if event.kind == "resolved": + await alerts.close(event.issue_id, dedupe_key=event.event_id) + elif event.issue is not None and event.issue.responsible_party != "buyer": + await alerts.open( + event.issue_id, + dedupe_key=event.event_id, # delivery is at least once + escalated=event.kind == "escalated", + action=event.issue.recommended_action, + ) + +reporting = ReliableReportingService( + storage=postgres_storage(pool=pool), + account_context=resolve_account_context, + consumer_status_enabled=True, + escalation=ReportingDeliveryEscalation( + consumer_mismatch_escalation=timedelta(hours=4), + operations_contact_email="reporting-ops@seller.example", + ), + issue_observer=on_issue, + worker_error_handler=report_to_sentry, +) +``` + +**When it is called.** Once per lifecycle transition of an issue occurrence, +after the transaction that made the transition commits: + +| `event.kind` | The transition | +|---|---| +| `opened` | An occurrence opened: a `sync_reporting_status` statement landed, or a status read first observed it. | +| `escalated` | An open or acknowledged occurrence passed `opened_at` plus `consumer_mismatch_escalation`. | +| `acknowledged` | An operator called `store.set_issue_state(..., state="acknowledged")`. | +| `waived` | An operator recorded a bilateral waiver (`state="waived"`). | +| `resolved` | The condition cleared: the buyer filed an agreeing statement, or the seller's record no longer conflicts. | + +`opened`, `acknowledged`, `waived` and `resolved` are the spec's +`issue_state` values. `escalated` is not a state: it is the +`consumer_mismatch_escalation` boundary, where the issue must turn +`action_required` with a `contact_*` action. + +The ledger records each event in the same transaction as its transition, and +the worker turn delivers it. Schedule a worker turn somewhere: set +`worker_interval`, or call `run_worker()` yourself. Every process that writes +the ledger records events, whether or not it runs a worker. The worker turn +also records `escalated` once the boundary passes. It settles the account +first, so a condition that cleared unobserved resolves instead of escalating, +and an unattended mismatch escalates on time without anyone reading its +status. Each occurrence's events arrive in the order they were recorded. + +**At least once: deduplicate on `event_id`.** Each transition is recorded +exactly once, even when several replicas make it at the same moment. +Delivery is at least once: a worker that stops after your observer returns, +but before the event is marked delivered, delivers it again. `event_id` is +derived from the occurrence and the kind, so every delivery of one transition +carries the same value. An exact retry of `sync_reporting_status` (`unchanged`) +and repeated status reads record nothing. + +**Failures never reach the buyer.** The observer runs on the worker turn, not +in the buyer's call. When it raises, or an async observer runs past 10 +seconds, the error goes to `worker_error_handler` as +`issue_observer:{event_id}`. The event is retried on a later turn, after a +backoff that starts at 30 seconds and doubles to one hour, with no limit. +`event.attempt` counts deliveries. A synchronous observer runs on the event +loop and cannot be timed out, so keep it short. + +**The event.** `ReportingIssueEvent` carries `event_id`, `kind`, +`occurred_at`, `account_id`, `consumer_id`, `issue_id`, `attempt`, and `issue`, +a `ReportingIssue` rendered as `get_reporting_status` showed it to +`consumer_id` at `occurred_at`. That gives you `responsible_party`, +`recommended_action`, `severity`, `opened_at` and the causing +`reporting_status_id` (also `event.reporting_status_id`). `issue` is `None` for +`resolved`, since the condition no longer projects, and when the evidence is +no longer retained. + +**Storage.** The in-memory store records issue events. On PostgreSQL you need +adcp-reporting-storage 0.2.0b4 or later, with migration `0008` applied +(`python -m adcp migrate`, which is safe while earlier-release pods still +run; see the +[ledger migration notes](reporting-ledger-migration.md#issue-lifecycle-events-migration-0008)). +The service refuses `issue_observer` at construction against a store that +does not record events, and at startup before the table exists. Delivered +events are pruned after seven days. `production=` does not support an issue +observer yet ([#1873](https://github.com/adcontextprotocol/adcp-client-python/issues/1873)). + +Composing the ledger without the service? Schedule +`ReportingIssueEventDispatcher(store, observer, escalation=...).run_once()` +yourself. + ## Advanced composition `postgres_reliable_reporting_service()` (from `adcp_reporting_storage`) without @@ -672,7 +770,10 @@ Other configurations and extensions still run, and the background scheduler retries on its next turn. These failures do not change service readiness. `worker_error_handler(component, error)` receives the original exception and the component name: `configuration:{account_id}:{delivery_config_id}@{version}`, -`materialization`, or `notification`. SDK logs contain only the component kind; +`materialization`, `notification`, `issue_observer:{event_id}` (the issue +observer raised), `issue_events` (claiming or sweeping issue events failed) or +`issue_events:snapshot` (an event could not be rendered yet and is retried). +SDK logs contain only the component kind; adopter callbacks control any further diagnostics. A callback failure is logged without interrupting the remaining work. diff --git a/docs/reporting-ledger-migration.md b/docs/reporting-ledger-migration.md index 601ebc10ca..640acde18a 100644 --- a/docs/reporting-ledger-migration.md +++ b/docs/reporting-ledger-migration.md @@ -152,6 +152,36 @@ nothing on a schema without status notifications. A foreign key outside the SDK that references `reporting_status_scope_checkpoints` refuses the upgrade instead of being dropped. +### Issue-lifecycle events (migration `0008`) + +`python -m adcp migrate` (migration `0008 reporting_issue_events`, header +`-- adcp-upgrade: additive`) creates `reporting_issue_events`, the outbox the +issue observer reads ([#1258](https://github.com/adcontextprotocol/adcp-client-python/issues/1258)), +and its indexes. Nothing else changes: no existing table, key, function or +trigger is touched, and the storage contract stays `2`. + +- **Rolling-deploy safe, including before contract.** `migrate` applies an + additive migration even while the contract step (`0007`) waits, so `0008` + lands on an expanded ledger beside processes of + adcp-reporting-storage 0.1.x/0.2.0b0. Those processes neither read nor write + the table and keep starting, reading, writing and running worker turns; the + published-wheel rolling test runs them beside it. They record no events for + the issue transitions they make. +- **Applied by `migrate` only.** `PgReportingLedgerStore.create_schema()` does + not install it. Until it exists, issue-lifecycle writes record nothing and + `PgReportingLedgerStore.issue_events_ready()` is `False`; nothing else + changes. `postgres_storage(...)` refuses to start while `0008` is pending, + like any other non-contract migration. +- **What it records.** Every write that moves a `reporting_issue_lifecycle` + row (an occurrence opening, being acknowledged or waived, or resolving) + records one row in the same transaction. The escalation sweep + (`sweep_issue_escalations`) records `escalated` once an open or acknowledged + occurrence passes `opened_at + consumer_mismatch_escalation_seconds`. + `UNIQUE (issue_id, kind)` keeps each record exactly once across replicas. + Delivered rows can be pruned with `prune_issue_events`. Nothing reads the + table until an issue observer is configured, and rows accumulate only for + consumer-mismatch transitions. + ## Historical account-generation upgrade The following notes describe the pre-ownership SDK release and are retained for @@ -385,7 +415,8 @@ To deploy: 1. Back up the database. 2. If you use `postgres_storage(...)` or status notifications, run `python -m adcp migrate` (for example as a pre-deploy job). It is safe while earlier - pods still run: it expands and stops before the contract step, printing + pods still run: it expands, holds the contract step and applies only the + additive migrations after it (`0008`, the issue-event outbox), printing `not applied 0007 … (contract)`. Otherwise the first new process expands the ledger when it starts. 3. Roll out the new release normally. diff --git a/packages/adcp-reporting-storage/README.md b/packages/adcp-reporting-storage/README.md index af4d3f6152..7636e55e6b 100644 --- a/packages/adcp-reporting-storage/README.md +++ b/packages/adcp-reporting-storage/README.md @@ -110,6 +110,11 @@ expand/contract pair: subscription's generation raises `ReportingUpgradeNotContractedError` (code `REPORTING_UPGRADE_NOT_CONTRACTED`). After it, the earlier release cannot start against the ledger, so rolling back means restoring the backup. +- **Issue events** (migration `0008`, additive): `python -m adcp migrate` adds + `reporting_issue_events`, the outbox the issue observer reads (#1258), even + while the contract step waits. It is one new table and its indexes, so + earlier-release processes keep running beside it; they record no issue + events for their own transitions. `postgres_storage(...)` starts on a ledger whose only pending migration is the contract step. A new ledger is created contracted. A ledger that 0.2.0b1 or diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py index 5e7a7f8660..e9fd572b33 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/__init__.py @@ -61,6 +61,9 @@ from adcp_reporting_storage.ledger.delivery_pg import ( PgReportingReconciliationStore as PgReportingReconciliationStore, ) + from adcp_reporting_storage.ledger.issue_events import ( + ReportingIssueEventLease as ReportingIssueEventLease, + ) from adcp_reporting_storage.ledger.orphan_collector import ( ReportingOrphanBindingResultV1 as ReportingOrphanBindingResultV1, ) @@ -283,6 +286,8 @@ "replace_empty_owned_reporting": ("migration", "replace_empty_owned_reporting"), "replace_empty_owned_reporting_sync": ("migration", "replace_empty_owned_reporting_sync"), "require_owned_schema_or_empty": ("migration", "require_owned_schema_or_empty"), + # The issue-lifecycle event outbox the issue observer reads (#1258) + "ReportingIssueEventLease": ("ledger.issue_events", "ReportingIssueEventLease"), # Host change feed "ReportingChangeCursorExpiredError": ( "ledger.change_feed", diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/issue_events.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/issue_events.py new file mode 100644 index 0000000000..5671cc86cc --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/issue_events.py @@ -0,0 +1,362 @@ +"""The issue-lifecycle event outbox over PostgreSQL (adcp-client-python#1258). + +Every write that moves a ``reporting_issue_lifecycle`` row records the +transition in ``reporting_issue_events`` inside the same transaction, so an +event exists exactly when its transition committed. The escalation sweep adds +``escalated`` when an open or acknowledged occurrence passes ``opened_at`` plus +the seller's ``consumer_mismatch_escalation_seconds``. A worker claims pending +events under a lease, hands each to the adopter's issue observer, and stamps it +delivered. + +Delivery is at least once. ``UNIQUE (issue_id, kind)`` makes the *record* +exactly once across replicas, and ``event_id`` is derived from that pair, so a +redelivery (a worker that crashed after the observer returned, or a lease that +expired mid-call) carries the same ``event_id`` and an observer deduplicates on +it. + +The table is installed by migration ``0008`` (``python -m adcp migrate``). +Until it exists, lifecycle writes record nothing and +:meth:`~adcp_reporting_storage.ledger.pg.PgReportingLedgerStore.issue_events_ready` +is ``False``; nothing else changes, so a store of this release runs on a +ledger that has not been migrated yet. + +Only existing ``adcp`` APIs are used here, so this module works with every +``adcp`` this distribution supports. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any, Literal +from uuid import uuid4 + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 +from adcp.reporting.ledger._record_rows import _ISSUE_COLUMNS, _issue_from_row +from adcp.reporting.ledger.models import ReportingIssueLifecycle + +if TYPE_CHECKING: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + +__all__ = [ + "ISSUE_EVENT_KINDS", + "ReportingIssueEventKind", + "ReportingIssueEventLease", + "issue_event_id", +] + +ReportingIssueEventKind = Literal["opened", "escalated", "acknowledged", "waived", "resolved"] + +#: Every kind, in lifecycle order. +ISSUE_EVENT_KINDS: tuple[ReportingIssueEventKind, ...] = ( + "opened", + "escalated", + "acknowledged", + "waived", + "resolved", +) + +#: The event that entering each ``issue_state`` records. +_KIND_FOR_STATE: dict[str, ReportingIssueEventKind] = { + "open": "opened", + "acknowledged": "acknowledged", + "waived": "waived", + "resolved": "resolved", +} + +#: Claimed columns in :class:`ReportingIssueEventLease` order, then the recording order. +_CLAIMED_COLUMNS = ( + "e.event_id, e.account_id, e.consumer_id, e.issue_key, e.issue_id, e.generation, e.kind," + " e.occurred_at, e.recorded_at, e.attempts, e.lease_token, e.lease_expires_at, e.seq" +) + + +def issue_event_id(issue_id: str, kind: str) -> str: + """The stable id of one transition of one occurrence. + + The lifecycle is forward-only, so ``(issue_id, kind)`` happens at most + once; every redelivery of that transition carries this id. + """ + payload = canonical_json_utf8_v1(["reporting-issue-event-v1", issue_id, kind]) + return "rpie_" + hashlib.sha256(payload).hexdigest()[:40] + + +@dataclass(frozen=True) +class ReportingIssueEventLease: + """One claimed issue event: what happened, and the lease to settle it with.""" + + event_id: str + account_id: str + consumer_id: str | None + issue_key: str + issue_id: str + generation: int + kind: ReportingIssueEventKind + #: When the transition took effect. + occurred_at: datetime + #: When the transition was recorded. + recorded_at: datetime + #: Delivery attempts, this one included. + attempts: int + lease_token: str = field(repr=False) + lease_expires_at: datetime + + +def _lease_from_row(row: Any) -> ReportingIssueEventLease: + return ReportingIssueEventLease( + event_id=row[0], + account_id=row[1], + consumer_id=row[2], + issue_key=row[3], + issue_id=row[4], + generation=int(row[5]), + kind=row[6], + occurred_at=row[7], + recorded_at=row[8], + attempts=int(row[9]), + lease_token=row[10], + lease_expires_at=row[11], + ) + + +def _utc(value: datetime) -> datetime: + return value.astimezone(timezone.utc) + + +async def issue_event_storage_on(connection: Any) -> bool: + """Whether migration ``0008`` installed the outbox in this schema.""" + row = await ( + await connection.execute( + "SELECT to_regclass(quote_ident(current_schema()) || '.reporting_issue_events')" + " IS NOT NULL" + ) + ).fetchone() + return bool(row and row[0]) + + +async def record_issue_transition_on( + connection: Any, + *, + before_state: str | None, + after: ReportingIssueLifecycle, + clock: Callable[[], datetime] | None, + occurred_at: datetime | None = None, +) -> None: + """Record the events one lifecycle write produced, in the writer's transaction. + + Runs inside the caller's account transaction, so the events commit or roll + back with the write. ``before_state`` is the occurrence's state before the + write, or ``None`` when the write created it. A write that leaves the + state unchanged (an idempotent re-acknowledge, an ``external_ref`` + attachment) records nothing. ``occurred_at`` defaults to the occurrence's + ``retired_at`` for retiring states and to the recording instant otherwise. + """ + from adcp_reporting_storage.outbox.pg import database_now + + if not await issue_event_storage_on(connection): + return + now = await database_now(connection, clock) + kinds: list[tuple[ReportingIssueEventKind, datetime]] = [] + if before_state is None: + kinds.append(("opened", after.opened_at)) + before_state = "open" + if after.issue_state != before_state: + at = occurred_at or after.retired_at or now + kinds.append((_KIND_FOR_STATE[after.issue_state], at)) + for kind, at in kinds: + await _insert_on(connection, after, kind, occurred_at=at, now=now) + + +async def _insert_on( + connection: Any, + issue: ReportingIssueLifecycle, + kind: ReportingIssueEventKind, + *, + occurred_at: datetime, + now: datetime, +) -> bool: + row = await ( + await connection.execute( + "INSERT INTO reporting_issue_events" + " (event_id, account_id, issue_key, generation, issue_id, consumer_id, kind," + " occurred_at, recorded_at, due_at)" + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" + " ON CONFLICT (issue_id, kind) DO NOTHING RETURNING 1", + ( + issue_event_id(issue.issue_id, kind), + issue.account_id, + issue.issue_key, + issue.generation, + issue.issue_id, + issue.consumer_id, + kind, + _utc(occurred_at), + _utc(now), + _utc(now), + ), + ) + ).fetchone() + return row is not None + + +async def sweep_escalations(store: PgReportingLedgerStore, *, after: timedelta, limit: int) -> int: + """Record ``escalated`` for occurrences unresolved past ``opened_at + after``. + + Each candidate account is swept in its own transaction under its advisory + lock: the account's lifecycle is settled first, as a status read settles + it, so a condition that cleared is retired (recording ``resolved``) + instead of escalating. ``ON CONFLICT DO NOTHING`` keeps the record exactly + once when replicas sweep together. Returns the events recorded. + """ + from adcp_reporting_storage.ledger.status_snapshot import settle_snapshot_on + from adcp_reporting_storage.outbox.pg import database_now + + if after < timedelta(0): + raise ValueError("after must be nonnegative") + if limit < 1: + raise ValueError("limit must be positive") + async with store._connection() as connection, connection.transaction(): + if not await issue_event_storage_on(connection): + return 0 + now = await database_now(connection, store._clock) + accounts = await ( + await connection.execute( + "SELECT DISTINCT l.account_id FROM reporting_issue_lifecycle l" + " WHERE l.issue_state IN ('open', 'acknowledged') AND l.opened_at <= %s" + " AND NOT EXISTS (SELECT 1 FROM reporting_issue_events e" + " WHERE e.issue_id = l.issue_id AND e.kind = 'escalated')" + " ORDER BY l.account_id LIMIT %s", + (now - after, limit), + ) + ).fetchall() + recorded = 0 + for (account_id,) in accounts: + async with store._connection() as connection, connection.transaction(): + await store._lock_account(connection, account_id) + now = await database_now(connection, store._clock) + await settle_snapshot_on(store, connection, account_id=account_id, as_of=now) + rows = await ( + await connection.execute( + f"SELECT {_ISSUE_COLUMNS} FROM reporting_issue_lifecycle l" # noqa: S608 # nosec B608 + " WHERE l.account_id = %s AND l.issue_state IN ('open', 'acknowledged')" + " AND l.opened_at <= %s" + " ORDER BY l.opened_at, l.issue_id", + (account_id, now - after), + ) + ).fetchall() + for row in rows: + issue = _issue_from_row(row) + if await _insert_on( + connection, + issue, + "escalated", + occurred_at=_utc(issue.opened_at) + after, + now=now, + ): + recorded += 1 + return recorded + + +async def claim( + store: PgReportingLedgerStore, *, lease_seconds: float, limit: int +) -> tuple[ReportingIssueEventLease, ...]: + """Lease due events, at most one per occurrence, oldest first. + + An occurrence's events are delivered in the order they were recorded: an + event is claimable only once every earlier event of its occurrence has + been delivered. ``SKIP LOCKED`` lets workers on several replicas claim + disjoint events; an expired lease is claimable again. + """ + from adcp_reporting_storage.outbox.pg import database_now + + if lease_seconds <= 0: + raise ValueError("lease_seconds must be positive") + if limit < 1: + raise ValueError("limit must be positive") + async with store._connection() as connection, connection.transaction(): + if not await issue_event_storage_on(connection): + return () + now = await database_now(connection, store._clock) + token = uuid4().hex + rows = await ( + await connection.execute( + "WITH due AS (" + " SELECT e.event_id FROM reporting_issue_events e" + " WHERE e.delivered_at IS NULL AND e.due_at <= %s" + " AND (e.lease_expires_at IS NULL OR e.lease_expires_at <= %s)" + " AND NOT EXISTS (SELECT 1 FROM reporting_issue_events p" + " WHERE p.issue_id = e.issue_id AND p.delivered_at IS NULL" + " AND p.seq < e.seq)" + " ORDER BY e.due_at, e.seq LIMIT %s" + " FOR UPDATE SKIP LOCKED)" + " UPDATE reporting_issue_events e" + " SET lease_token = %s, lease_expires_at = %s, attempts = e.attempts + 1" + " FROM due WHERE e.event_id = due.event_id" + f" RETURNING {_CLAIMED_COLUMNS}", # noqa: S608 # nosec B608 + (now, now, limit, token, now + timedelta(seconds=lease_seconds)), + ) + ).fetchall() + return tuple(_lease_from_row(row) for row in sorted(rows, key=lambda row: row[-1])) + + +async def complete(store: PgReportingLedgerStore, lease: ReportingIssueEventLease) -> bool: + """Stamp a delivered event. ``False`` when the lease was lost to another worker.""" + from adcp_reporting_storage.outbox.pg import database_now + + async with store._connection() as connection, connection.transaction(): + now = await database_now(connection, store._clock) + row = await ( + await connection.execute( + "UPDATE reporting_issue_events" + " SET delivered_at = %s, lease_token = NULL, lease_expires_at = NULL" + " WHERE event_id = %s AND lease_token = %s AND delivered_at IS NULL" + " RETURNING 1", + (now, lease.event_id, lease.lease_token), + ) + ).fetchone() + return row is not None + + +async def release( + store: PgReportingLedgerStore, lease: ReportingIssueEventLease, *, retry_after: timedelta +) -> bool: + """Give a failed delivery back, due again ``retry_after`` from now (the store's clock).""" + from adcp_reporting_storage.outbox.pg import database_now + + async with store._connection() as connection, connection.transaction(): + now = await database_now(connection, store._clock) + row = await ( + await connection.execute( + "UPDATE reporting_issue_events" + " SET due_at = %s, lease_token = NULL, lease_expires_at = NULL" + " WHERE event_id = %s AND lease_token = %s AND delivered_at IS NULL" + " RETURNING 1", + (now + max(retry_after, timedelta(0)), lease.event_id, lease.lease_token), + ) + ).fetchone() + return row is not None + + +async def prune(store: PgReportingLedgerStore, *, older_than: timedelta, limit: int) -> int: + """Delete up to ``limit`` events delivered more than ``older_than`` ago.""" + from adcp_reporting_storage.outbox.pg import database_now + + if limit < 1: + raise ValueError("limit must be positive") + async with store._connection() as connection, connection.transaction(): + if not await issue_event_storage_on(connection): + return 0 + delivered_before = await database_now(connection, store._clock) - older_than + rows = await ( + await connection.execute( + "DELETE FROM reporting_issue_events WHERE event_id IN (" + " SELECT event_id FROM reporting_issue_events" + " WHERE delivered_at IS NOT NULL AND delivered_at < %s" + " ORDER BY delivered_at LIMIT %s)" + " RETURNING 1", + (delivered_before, limit), + ) + ).fetchall() + return len(rows) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 9a58e82cb2..2db4beda56 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -173,6 +173,7 @@ ReportingChangeKindV1, ReportingChangesPageV1, ) + from adcp_reporting_storage.ledger.issue_events import ReportingIssueEventLease from adcp_reporting_storage.ledger.orphan_collector import ( ReportingOrphanCollectionOptionsV1, ReportingOrphanCollectionResultV1, @@ -253,6 +254,10 @@ class PgReportingLedgerStore: #: The reporting storage contract this store's DDL implements; adcp refuses a #: store whose contract differs from its own (``adcp_reporting_storage.contract``). storage_contract: ClassVar[int] = STORAGE_CONTRACT + #: This store records issue-lifecycle events and serves them to the issue + #: observer (adcp-client-python#1258); ``issue_events_ready()`` says whether + #: migration 0008 installed their table. + issue_events_supported: ClassVar[bool] = True def __init__( self, @@ -3069,6 +3074,7 @@ async def _ensure_issue_opened_on( generation=generation, ) await self._dirty_issue(connection, record, status_scope, enqueue=enqueue) + await self._record_issue_transition(connection, before_state=None, after=record) return record async def set_issue_state( @@ -3140,6 +3146,12 @@ async def set_issue_state( # nothing, while anything that does move retained evidence stays # reconstructable for the projector. await self._dirty_issue(connection, refreshed, status_scope, live) + await self._record_issue_transition( + connection, + before_state=live.issue_state, + after=refreshed, + occurred_at=_utc(at) if state == "acknowledged" else None, + ) return refreshed async def retire_issue( @@ -3183,12 +3195,95 @@ async def _retire_issue_on( retired = await self._issue_row(connection, issue_key, account_id, live.generation) assert retired is not None await self._dirty_issue(connection, retired, status_scope, live, enqueue=enqueue) + await self._record_issue_transition( + connection, before_state=live.issue_state, after=retired + ) return retired async def get_issue(self, *, issue_key: str, account_id: str) -> ReportingIssueLifecycle | None: async with self._connection() as connection: return await self._live_issue(connection, issue_key, account_id) + async def _record_issue_transition( + self, + connection: Any, + *, + before_state: str | None, + after: ReportingIssueLifecycle, + occurred_at: datetime | None = None, + ) -> None: + from adcp_reporting_storage.ledger.issue_events import record_issue_transition_on + + await record_issue_transition_on( + connection, + before_state=before_state, + after=after, + clock=self._clock, + occurred_at=occurred_at, + ) + + # -- issue events (adcp-client-python#1258) ---------------------------- + # + # A transactional outbox of issue-lifecycle transitions for the adopter's + # issue observer: see :mod:`adcp_reporting_storage.ledger.issue_events`. + + async def issue_events_ready(self) -> bool: + """Whether migration ``0008`` installed the issue-event outbox. + + Until it has, lifecycle writes record no events. Install it with + ``python -m adcp migrate``. + """ + from adcp_reporting_storage.ledger.issue_events import issue_event_storage_on + + async with self._connection() as connection, connection.transaction(): + return await issue_event_storage_on(connection) + + async def sweep_issue_escalations(self, *, after: timedelta, limit: int = 100) -> int: + """Record ``escalated`` for occurrences still open or acknowledged past + ``opened_at + after`` (the seller's ``consumer_mismatch_escalation``). + + Settles each candidate account first, so a cleared condition resolves + instead of escalating. Exactly once per occurrence across replicas. + Sweeps at most ``limit`` accounts; returns the events recorded. + """ + from adcp_reporting_storage.ledger.issue_events import sweep_escalations + + return await sweep_escalations(self, after=after, limit=limit) + + async def claim_issue_events( + self, *, lease_seconds: float = 60.0, limit: int = 20 + ) -> tuple[ReportingIssueEventLease, ...]: + """Lease up to ``limit`` due events, at most one per occurrence. + + An occurrence's events are claimable in recording order, each once + the one before it is delivered. Settle each lease with + :meth:`complete_issue_event` or :meth:`release_issue_event`; an + unsettled lease expires after ``lease_seconds`` and is claimed again. + """ + from adcp_reporting_storage.ledger.issue_events import claim + + return await claim(self, lease_seconds=lease_seconds, limit=limit) + + async def complete_issue_event(self, lease: ReportingIssueEventLease) -> bool: + """Stamp a delivered event; ``False`` when its lease was lost.""" + from adcp_reporting_storage.ledger.issue_events import complete + + return await complete(self, lease) + + async def release_issue_event( + self, lease: ReportingIssueEventLease, *, retry_after: timedelta + ) -> bool: + """Give back a failed delivery, due again ``retry_after`` from now.""" + from adcp_reporting_storage.ledger.issue_events import release + + return await release(self, lease, retry_after=retry_after) + + async def prune_issue_events(self, *, older_than: timedelta, limit: int = 1000) -> int: + """Delete up to ``limit`` events delivered more than ``older_than`` ago.""" + from adcp_reporting_storage.ledger.issue_events import prune + + return await prune(self, older_than=older_than, limit=limit) + @staticmethod async def _live_issue( connection: Any, issue_key: str, account_id: str diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_issue_events.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_issue_events.sql new file mode 100644 index 0000000000..a98cdfd0e3 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_issue_events.sql @@ -0,0 +1,57 @@ +-- Issue-lifecycle events for the adopter's issue observer (adcp-client-python#1258). +-- +-- A transactional outbox: every write that moves a reporting_issue_lifecycle +-- row (an occurrence opening, being acknowledged or waived, or being retired as +-- resolved) records one row here in the same transaction, and the escalation +-- sweep records 'escalated' when an open or acknowledged occurrence passes +-- opened_at + consumer_mismatch_escalation_seconds. A worker claims pending +-- rows under a lease, hands each to the observer and stamps delivered_at. +-- +-- UNIQUE (issue_id, kind) is the exactly-once guarantee: the issue lifecycle is +-- forward-only, so an occurrence makes each transition at most once, and two +-- replicas recording the same transition collide here instead of producing two +-- events. event_id is derived from (issue_id, kind), so it is stable across +-- retries and redeliveries and is what an observer deduplicates on. +-- +-- Additive only: one new table and its indexes. No existing table, key, +-- function or trigger changes, so processes of earlier releases, which neither +-- read nor write this table, keep running beside it. Those processes record no +-- events for the transitions they make. +CREATE TABLE IF NOT EXISTS reporting_issue_events ( + event_id TEXT COLLATE "C" NOT NULL PRIMARY KEY, + -- Recording order. Per account it is commit order (every write here holds + -- the account's advisory lock), and the worker delivers one occurrence's + -- events in this order. + seq BIGSERIAL NOT NULL, + account_id TEXT COLLATE "C" NOT NULL, + issue_key TEXT COLLATE "C" NOT NULL, + generation INTEGER NOT NULL, + issue_id TEXT COLLATE "C" NOT NULL, + consumer_id TEXT COLLATE "C", + kind TEXT COLLATE "C" NOT NULL CHECK + (kind IN ('opened', 'escalated', 'acknowledged', 'waived', 'resolved')), + -- When the transition took effect: opened_at, the escalation boundary, the + -- operator's instant, or retired_at. + occurred_at TIMESTAMPTZ NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + -- The next delivery attempt; a failed attempt moves it forward. + due_at TIMESTAMPTZ NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + lease_token TEXT COLLATE "C", + lease_expires_at TIMESTAMPTZ, + delivered_at TIMESTAMPTZ, + UNIQUE (issue_id, kind), + CHECK ((lease_token IS NULL) = (lease_expires_at IS NULL)) +); + +CREATE INDEX IF NOT EXISTS reporting_issue_events_pending + ON reporting_issue_events (due_at, seq) + WHERE delivered_at IS NULL; + +CREATE INDEX IF NOT EXISTS reporting_issue_events_issue_pending + ON reporting_issue_events (issue_id, seq) + WHERE delivered_at IS NULL; + +CREATE INDEX IF NOT EXISTS reporting_issue_events_delivered + ON reporting_issue_events (delivered_at) + WHERE delivered_at IS NOT NULL; diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/status_snapshot.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/status_snapshot.py index ea6648f483..a323c2f808 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/status_snapshot.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/status_snapshot.py @@ -123,43 +123,65 @@ async def persist_replay_lifecycles_on( rows used by polling, including repairs for an old writer's interrupted status/issue pair. The enclosing account lock fences concurrent ingestion. """ + from adcp_reporting_storage.ledger.issue_events import record_issue_transition_on from adcp_reporting_storage.ledger.pg import _save_waiver_binding scopes = dict(snapshot.issue_scopes) for issue in sorted(snapshot.lifecycles, key=lambda i: (i.issue_key, i.generation)): - await connection.execute( - "INSERT INTO reporting_issue_lifecycle" - " (account_id, issue_key, generation, issue_id, consumer_id, opened_at," - " issue_state, external_ref, retired_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)" - " ON CONFLICT (account_id, issue_key, generation) DO NOTHING", - ( - issue.account_id, - issue.issue_key, - issue.generation, - issue.issue_id, - issue.consumer_id, - issue.opened_at, - issue.issue_state, - issue.external_ref, - issue.retired_at, - ), - ) - await _save_waiver_binding(connection, issue) - if issue.issue_state != "open": + inserted = await ( await connection.execute( - "UPDATE reporting_issue_lifecycle SET issue_state=%s, external_ref=%s," - " retired_at=%s" - " WHERE account_id=%s AND issue_key=%s AND generation=%s" - " AND issue_state IN ('open','acknowledged')", + "INSERT INTO reporting_issue_lifecycle" + " (account_id, issue_key, generation, issue_id, consumer_id, opened_at," + " issue_state, external_ref, retired_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)" + " ON CONFLICT (account_id, issue_key, generation) DO NOTHING RETURNING 1", ( - issue.issue_state, - issue.external_ref, - issue.retired_at, issue.account_id, issue.issue_key, issue.generation, + issue.issue_id, + issue.consumer_id, + issue.opened_at, + issue.issue_state, + issue.external_ref, + issue.retired_at, ), ) + ).fetchone() + if inserted is not None: + await record_issue_transition_on( + connection, before_state=None, after=issue, clock=store._clock + ) + await _save_waiver_binding(connection, issue) + if issue.issue_state != "open": + # The account lock is held, so the state read here is the one the + # update below replaces. + before = await ( + await connection.execute( + "SELECT issue_state FROM reporting_issue_lifecycle" + " WHERE account_id=%s AND issue_key=%s AND generation=%s", + (issue.account_id, issue.issue_key, issue.generation), + ) + ).fetchone() + moved = await ( + await connection.execute( + "UPDATE reporting_issue_lifecycle SET issue_state=%s, external_ref=%s," + " retired_at=%s" + " WHERE account_id=%s AND issue_key=%s AND generation=%s" + " AND issue_state IN ('open','acknowledged') RETURNING 1", + ( + issue.issue_state, + issue.external_ref, + issue.retired_at, + issue.account_id, + issue.issue_key, + issue.generation, + ), + ) + ).fetchone() + if moved is not None and before is not None: + await record_issue_transition_on( + connection, before_state=before[0], after=issue, clock=store._clock + ) scope = scopes.get(issue.issue_id) if scope is not None: # Existing scope may already be more precise because a later source diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0008_reporting_issue_events.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0008_reporting_issue_events.sql new file mode 100644 index 0000000000..483ec80434 --- /dev/null +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0008_reporting_issue_events.sql @@ -0,0 +1,10 @@ +-- adcp-migration: 0008 reporting_issue_events +-- adcp-upgrade: additive +-- The issue-lifecycle event outbox the issue observer reads +-- (adcp-client-python#1258): one new table and its indexes, nothing else. +-- +-- Rolling-deploy safe, and applied even while the contract step (0007) waits: +-- it creates only new objects and depends on nothing 0007 changes, so +-- processes of adcp-reporting-storage 0.1.x/0.2.0b0 keep booting, reading, +-- writing and running worker turns beside it. They record no issue events. +-- include: ledger/reporting_issue_events.sql diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py index 97fcdef5e6..9db1fc9f4a 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/__init__.py @@ -9,6 +9,9 @@ releases cannot run beside: stop every reporting worker first; * ``-- adcp-upgrade: contract`` marks the contract half of an expand/contract upgrade (below); +* ``-- adcp-upgrade: additive`` marks a migration that only creates new + objects and depends on nothing a contract step changes, so it is applied + even while a contract step waits (below); * ``-- include: `` applies one of the package's idempotent DDL files, in order, so each statement lives in exactly one place. Any other SQL in the file runs after its includes. @@ -40,6 +43,11 @@ (a new one, or one that 0.2.0b1/b2 widened in place) the contract step changes nothing and is applied and recorded without the request. ``postgres_storage`` starts on a ledger whose only pending migration is a contract step. + +A migration after a waiting contract step is applied only when it is +``additive`` (``0008``, the issue-event outbox); the walk stops at the first +one that is not, since it may depend on the contract. An additive migration +is therefore applied, and recorded, before the contract step it follows. """ from __future__ import annotations @@ -154,6 +162,9 @@ class Migration: #: The contract half of an expand/contract upgrade: applied to an expanded #: ledger only on request (:data:`CONTRACT_COMMAND`). contract: bool = False + #: Creates only new objects and depends on no contract step: applied even + #: while an earlier contract step waits for :data:`CONTRACT_COMMAND`. + additive: bool = False def statements(self) -> list[str]: """The SQL this migration runs, in order.""" @@ -200,7 +211,7 @@ def _parse(filename: str, text: str) -> Migration: body.append(line) if declared != f"{version} {name}": raise ValueError(f"migration file {filename!r} must declare '{version} {name}'") - if upgrade not in ("", "stopped-worker", "contract"): + if upgrade not in ("", "stopped-worker", "contract", "additive"): raise ValueError(f"migration file {filename!r} declares an unknown upgrade {upgrade!r}") return Migration( version=version, @@ -210,6 +221,7 @@ def _parse(filename: str, text: str) -> Migration: body="\n".join(body), description=" ".join(item for item in description if item), contract=upgrade == "contract", + additive=upgrade == "additive", ) @@ -291,9 +303,11 @@ async def migrate(pool: Any, *, dry_run: bool = False, contract: bool = False) - ``search_path`` selects the schema. Each migration commits on its own, so a failure leaves the earlier ones applied and the failed one pending. - The walk stops before a contract step whose ledger is expanded (or is an - earlier release's, which this run expands) unless ``contract`` is set: pass - it only once every earlier-release reporting process has stopped. + The walk holds a contract step whose ledger is expanded (or is an earlier + release's, which this run expands) unless ``contract`` is set: pass it only + once every earlier-release reporting process has stopped. Past a held + contract step it applies only ``additive`` migrations, and stops at the + first one that is not. """ if hasattr(pool, "getconn"): # a pool async with pool.connection() as connection: @@ -308,20 +322,44 @@ async def _migrate(connection: Any, *, dry_run: bool, contract: bool) -> list[Mi async with connection.transaction(): expanded = await needs_contract(connection) if expanded: - first = next(index for index, item in enumerate(pending) if item.contract) - return pending[:first] + return _past_held_contract(pending) return pending applied: list[Migration] = [] + held = False for migration in pending: async with connection.transaction(): await connection.execute(_LOCK) + if held and not migration.additive: + break # it may depend on the contract step still waiting if migration.contract and not contract and await needs_contract(connection): - break # stopped-worker: the operator asks for it explicitly + # The operator asks for the contract step explicitly; what + # follows it is applied only when it is additive. + held = True + continue if await _apply(connection, migration): applied.append(migration) return applied +def _past_held_contract(pending: Sequence[Migration]) -> list[Migration]: + """The plan when every pending contract step is held: up to the first, then additive only. + + Mirrors :func:`_migrate` step for step: past the held contract step the + walk stops at the first migration that is not additive, another contract + step included. + """ + plan: list[Migration] = [] + held = False + for migration in pending: + if held and not migration.additive: + break + if migration.contract: + held = True + continue + plan.append(migration) + return plan + + async def apply_pending( connection: Any, *, within: frozenset[str] | None = None ) -> list[Migration]: diff --git a/src/adcp/reporting/__init__.py b/src/adcp/reporting/__init__.py index 46beb2df4d..e6285fcb32 100644 --- a/src/adcp/reporting/__init__.py +++ b/src/adcp/reporting/__init__.py @@ -76,6 +76,7 @@ from adcp.reporting import currency as currency from adcp.reporting import fixtures as fixtures from adcp.reporting import inline_source as inline_source + from adcp.reporting import issue_events as issue_events from adcp.reporting import ledger as ledger from adcp.reporting import materializer as materializer from adcp.reporting import revision_selection as revision_selection @@ -83,6 +84,18 @@ from adcp.reporting import service_production as service_production from adcp.reporting import source as source from adcp.reporting import testing as testing + from adcp.reporting.issue_events import ( + ReportingIssueEvent as ReportingIssueEvent, + ) + from adcp.reporting.issue_events import ( + ReportingIssueEventDispatcher as ReportingIssueEventDispatcher, + ) + from adcp.reporting.issue_events import ( + ReportingIssueEventKind as ReportingIssueEventKind, + ) + from adcp.reporting.issue_events import ( + ReportingIssueObserver as ReportingIssueObserver, + ) from adcp.reporting.service import ( ReliableReportingConfigurationError as ReliableReportingConfigurationError, ) @@ -161,6 +174,7 @@ "currency", "fixtures", "inline_source", + "issue_events", "ledger", "materializer", "revision_selection", @@ -186,6 +200,10 @@ "ReportingServiceResource": ("service_lifecycle", "ReportingServiceResource"), "ReportingAccountContext": ("service", "ReportingAccountContext"), "ReportingAdapter": ("service", "ReportingAdapter"), + "ReportingIssueEvent": ("issue_events", "ReportingIssueEvent"), + "ReportingIssueEventDispatcher": ("issue_events", "ReportingIssueEventDispatcher"), + "ReportingIssueEventKind": ("issue_events", "ReportingIssueEventKind"), + "ReportingIssueObserver": ("issue_events", "ReportingIssueObserver"), "ReportingProductionOptions": ("service_production", "ReportingProductionOptions"), "ReportingServiceOffering": ("service_production", "ReportingServiceOffering"), "SubscriptionOptions": ("service_subscriptions", "SubscriptionOptions"), @@ -242,6 +260,10 @@ def __dir__() -> list[str]: "ReportingServiceResource", "ReportingAccountContext", "ReportingAdapter", + "ReportingIssueEvent", + "ReportingIssueEventDispatcher", + "ReportingIssueEventKind", + "ReportingIssueObserver", "ReportingProductionOptions", "ReportingServiceOffering", "PrincipalRef", diff --git a/src/adcp/reporting/issue_events.py b/src/adcp/reporting/issue_events.py new file mode 100644 index 0000000000..758867e2b9 --- /dev/null +++ b/src/adcp/reporting/issue_events.py @@ -0,0 +1,408 @@ +"""Tell the seller when a reporting issue opens, escalates or resolves (#1258). + +A ``CONSUMER_STATUS_MISMATCH`` is caller-scoped: only the buyer whose +statement caused it sees it in ``get_reporting_status``. A mismatch the seller +must act on is invisible to the seller's operators unless the buyer calls +them. The issue observer closes that gap:: + + from adcp.reporting import ReliableReportingService, ReportingIssueEvent + + async def on_issue(event: ReportingIssueEvent) -> None: + if event.issue is not None and event.issue.responsible_party != "buyer": + await page_operations(event) # dedupe on event.event_id + + service = ReliableReportingService(..., issue_observer=on_issue) + +When it is called +----------------- + +Once per lifecycle transition of an issue occurrence, after the transaction +that made the transition commits: + +========================= =========================================================== +``opened`` An occurrence opened: a ``sync_reporting_status`` + statement landed (or a status read first observed it). +``escalated`` An open or acknowledged occurrence passed ``opened_at`` + plus the seller's ``consumer_mismatch_escalation``. +``acknowledged`` An operator acknowledged it (``store.set_issue_state``). +``waived`` Buyer and seller agreed off-protocol to disregard it. +``resolved`` The condition cleared: the buyer filed an agreeing + statement, or the seller's record no longer conflicts. +========================= =========================================================== + +The transition and its event commit in one transaction, and the worker turn +(``run_worker``, or the ``worker_interval`` loop) delivers committed events to +the observer. ``escalated`` is recorded by that turn too, when the boundary +passes, so an unattended mismatch escalates whether or not anyone reads its +status. Each occurrence's events arrive in the order they were recorded. + +Delivery is at least once +------------------------- + +Each transition is recorded exactly once, even when several replicas make it +together. Delivery is **at least once**: a worker that stops after the +observer returned, but before the event was marked delivered, delivers it +again. Every delivery of one transition carries the same ``event_id``, derived +from the occurrence and the kind, so **deduplicate on** ``event_id``. + +The observer cannot fail the protocol operation: it runs on the worker turn, +not in the buyer's call. When it raises or times out, the error goes to the +service's ``worker_error_handler`` (or the log), and the event is retried on a +later turn with exponential backoff, without limit. A replayed or retried +``sync_reporting_status`` that changes nothing records nothing. +""" + +from __future__ import annotations + +import asyncio +import inspect +import logging +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass, replace +from datetime import datetime, timedelta +from typing import TYPE_CHECKING + +from adcp.reporting.ledger.issue_events import ( + ReportingIssueEventKind, + ReportingIssueEventLease, + ReportingIssueEventStore, + issue_event_id, + supports_issue_events, +) +from adcp.reporting.ledger.models import ( + ConsumerStatusRecord, + ReportingDeliveryEscalation, + ReportingIssue, + ReportingIssueLifecycle, + ReportingIssueStateValue, +) + +if TYPE_CHECKING: + from adcp.reporting.ledger.status_projection import ReportingStatusSnapshot + +__all__ = [ + "ReportingIssueDelivery", + "ReportingIssueEvent", + "ReportingIssueEventDispatcher", + "ReportingIssueEventKind", + "ReportingIssueObserver", + "issue_event_id", + "render_issue_event", +] + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ReportingIssueEvent: + """One lifecycle transition of one reporting-issue occurrence. + + Delivered at least once: deduplicate on ``event_id``, which is the same on + every delivery of this transition. + """ + + #: Stable across redeliveries of this transition; deduplicate on it. + event_id: str + kind: ReportingIssueEventKind + #: When the transition took effect: the issue's ``opened_at``, the + #: escalation boundary, the operator's instant, or when it was retired. + occurred_at: datetime + account_id: str + #: The buyer whose statement caused a caller-scoped issue. + consumer_id: str | None + issue_id: str + #: The issue as ``get_reporting_status`` rendered it for ``consumer_id`` at + #: ``occurred_at``: ``responsible_party``, ``recommended_action``, + #: ``severity``, ``opened_at``, the causing ``reporting_status_id``. Its + #: ``issue_state`` is this event's. ``None`` for ``resolved`` (the condition + #: no longer projects) or when the evidence is no longer retained. + issue: ReportingIssue | None + #: This delivery's attempt number, from 1. + attempt: int = 1 + + @property + def reporting_status_id(self) -> str | None: + """The consumer statement that caused the issue, when it is rendered.""" + return self.issue.reporting_status_id if self.issue is not None else None + + +#: The adopter's callback: sync or async (an awaitable result is awaited), its +#: return value otherwise ignored. +ReportingIssueObserver = Callable[[ReportingIssueEvent], object] + +#: The issue state each kind renders the occurrence in. +_RENDER_STATE: dict[str, ReportingIssueStateValue] = { + "opened": "open", + "acknowledged": "acknowledged", + "waived": "waived", +} + + +def render_issue_event( + lease: ReportingIssueEventLease, + snapshot: ReportingStatusSnapshot | None, + *, + escalation: ReportingDeliveryEscalation | None = None, +) -> ReportingIssueEvent: + """Build the observer's event, rendering the issue as of ``occurred_at``. + + The issue is projected exactly as ``get_reporting_status`` projects it for + ``consumer_id``, over the evidence that existed at ``occurred_at``: + statements recorded by then (with the chain leaf as of then), obligations + and revisions created by then, and the occurrence in this event's state. + Rendering is best effort: anything it cannot reconstruct yields + ``issue=None`` rather than a wrong issue. + """ + issue: ReportingIssue | None = None + if lease.kind != "resolved" and snapshot is not None and lease.consumer_id is not None: + try: + issue = _render(lease, snapshot, escalation) + except Exception: # noqa: BLE001 - rendering must never block delivery + logger.warning("Reliable Reporting could not render issue event %s", lease.event_id) + return ReportingIssueEvent( + event_id=lease.event_id, + kind=lease.kind, + occurred_at=lease.occurred_at, + account_id=lease.account_id, + consumer_id=lease.consumer_id, + issue_id=lease.issue_id, + issue=issue, + attempt=lease.attempts, + ) + + +def _render( + lease: ReportingIssueEventLease, + snapshot: ReportingStatusSnapshot, + escalation: ReportingDeliveryEscalation | None, +) -> ReportingIssue | None: + from adcp.reporting.ledger.notification_models import ReportingStatusScope + from adcp.reporting.ledger.status_projection import StatusProjectionInput, _project + + at = lease.occurred_at + occurrence = next( + ( + item + for item in snapshot.lifecycles + if item.issue_key == lease.issue_key and item.generation == lease.generation + ), + None, + ) + if occurrence is None: + return None + state = _RENDER_STATE.get(lease.kind) + if state is None: # escalated: as it stands, unless it has since been retired + state = occurrence.issue_state if occurrence.issue_state == "acknowledged" else "open" + # A waived occurrence leaves the projection by design, so render the + # disagreement it waived and label it waived. + rendered: ReportingIssueLifecycle = replace( + occurrence, + issue_state="acknowledged" if state == "waived" else state, + retired_at=None, + external_ref=None if lease.kind == "opened" else occurrence.external_ref, + waived_reporting_status_id=None, + waived_conflict_sha256=None, + ) + as_of = replace( + snapshot, + as_of=at, + statuses=_statuses_as_of(snapshot.statuses, at), + obligations=tuple(o for o in snapshot.obligations if o.created_at <= at), + revisions=tuple(r for r in snapshot.revisions if r.created_at <= at), + lifecycles=tuple(rendered if item is occurrence else item for item in snapshot.lifecycles), + ) + result, _ = _project( + StatusProjectionInput( + as_of, + ReportingStatusScope(lease.account_id, consumer_id=lease.consumer_id), + escalation, + ) + ) + issue = next((item for item in result.issues if item.issue_id == lease.issue_id), None) + if issue is None: + return None + return replace(issue, issue_state=state) + + +def _statuses_as_of( + statuses: Sequence[ConsumerStatusRecord], at: datetime +) -> tuple[ConsumerStatusRecord, ...]: + """The statements recorded by ``at``, superseded only by statements recorded by then.""" + recorded = [status for status in statuses if status.recorded_at <= at] + superseded = { + (status.consumer_id, status.supersedes_reporting_status_id) + for status in recorded + if status.supersedes_reporting_status_id is not None + } + return tuple( + replace( + status, + superseded=(status.consumer_id, status.reporting_status_id) in superseded, + ) + for status in recorded + ) + + +@dataclass(frozen=True) +class ReportingIssueDelivery: + """What one dispatcher turn did.""" + + #: Escalations recorded by this turn's sweep. + escalations_recorded: int = 0 + #: Events the observer accepted. + delivered: int = 0 + #: Events the observer raised on or timed out on; each is retried later. + failed: int = 0 + + @property + def did_work(self) -> bool: + return bool(self.escalations_recorded or self.delivered or self.failed) + + +ReportingIssueErrorReporter = Callable[[str, BaseException], Awaitable[None]] + + +class ReportingIssueEventDispatcher: + """Deliver a store's recorded issue events to an observer, at least once. + + One :meth:`run_once` sweeps escalations, then claims due events in small + leased batches (at most one per occurrence at a time, so an occurrence's + events arrive in order), renders each, calls the observer, and settles the + lease: delivered, or released for a retry after an exponential backoff + (``retry_base`` doubling to ``retry_cap``). Delivered events are pruned + after ``retention`` (``None`` keeps them). An event whose status snapshot + cannot be read is released for a retry rather than delivered unrendered. + + :class:`~adcp.reporting.service.ReliableReportingService` runs one on each + worker turn when given ``issue_observer``. Schedule it yourself only when + you compose the ledger without the service. + """ + + def __init__( + self, + store: ReportingIssueEventStore, + observer: ReportingIssueObserver, + *, + escalation: ReportingDeliveryEscalation | None = None, + on_error: ReportingIssueErrorReporter | None = None, + observer_timeout: timedelta = timedelta(seconds=10), + batch_size: int = 10, + max_batches: int = 10, + retry_base: timedelta = timedelta(seconds=30), + retry_cap: timedelta = timedelta(hours=1), + retention: timedelta | None = timedelta(days=7), + ) -> None: + if not supports_issue_events(store): + raise TypeError( + "the ledger store does not record issue events; use " + "InMemoryReportingLedgerStore, or PgReportingLedgerStore from " + "adcp-reporting-storage 0.2.0b4 or later" + ) + if not callable(observer): + raise TypeError("observer must be callable") + if observer_timeout <= timedelta(0) or batch_size < 1 or max_batches < 1: + raise ValueError("observer_timeout, batch_size and max_batches must be positive") + if ( + retry_base <= timedelta(0) + or retry_cap < retry_base + or (retention is not None and retention < timedelta(0)) + ): + raise ValueError("invalid retry or retention policy") + self.store = store + self.observer = observer + self.escalation = escalation + self._on_error = on_error + self._timeout = observer_timeout + self._batch_size = batch_size + self._max_batches = max_batches + self._retry_base = retry_base + self._retry_cap = retry_cap + self._retention = retention + + @property + def lease_seconds(self) -> float: + """Long enough for a whole batch to time out, so a lease never lapses mid-batch.""" + return self._batch_size * self._timeout.total_seconds() + 30 + + async def run_once(self) -> ReportingIssueDelivery: + escalations = 0 + window = self.escalation.consumer_mismatch_escalation if self.escalation else None + if window is not None: + escalations = await self.store.sweep_issue_escalations(after=window) + delivered = failed = 0 + for _ in range(self._max_batches): + leases = await self.store.claim_issue_events( + lease_seconds=self.lease_seconds, limit=self._batch_size + ) + if not leases: + break + snapshots, unreadable = await self._snapshots(leases) + for lease in leases: + if (lease.account_id, lease.consumer_id) in unreadable: + failed += 1 + await self.store.release_issue_event( + lease, retry_after=self._backoff(lease.attempts) + ) + continue + event = render_issue_event( + lease, + snapshots.get((lease.account_id, lease.consumer_id)), + escalation=self.escalation, + ) + try: + await self._call(event) + except Exception as error: + failed += 1 + await self.store.release_issue_event( + lease, retry_after=self._backoff(lease.attempts) + ) + await self._report(f"issue_observer:{lease.event_id}", error) + continue + if await self.store.complete_issue_event(lease): + delivered += 1 + if self._retention is not None: + await self.store.prune_issue_events(older_than=self._retention) + return ReportingIssueDelivery(escalations, delivered, failed) + + async def _snapshots(self, leases: Sequence[ReportingIssueEventLease]) -> tuple[ + dict[tuple[str, str | None], ReportingStatusSnapshot], + set[tuple[str, str | None]], + ]: + """One status snapshot per caller for rendering, and the callers that failed.""" + from adcp.reporting.ledger.status import ReportingStatusCaller + from adcp.reporting.ledger.status_snapshot import ReportingStatusParticipant + + snapshots: dict[tuple[str, str | None], ReportingStatusSnapshot] = {} + unreadable: set[tuple[str, str | None]] = set() + if not isinstance(self.store, ReportingStatusParticipant): + return snapshots, unreadable + for account_id, consumer_id in sorted( + {(lease.account_id, lease.consumer_id) for lease in leases if lease.kind != "resolved"}, + key=lambda item: (item[0], item[1] or ""), + ): + if consumer_id is None: + continue + try: + snapshots[(account_id, consumer_id)] = await self.store.read_status_snapshot( + caller=ReportingStatusCaller(account_id=account_id, consumer_id=consumer_id) + ) + except Exception as error: # noqa: BLE001 - retried with the event + unreadable.add((account_id, consumer_id)) + await self._report("issue_events:snapshot", error) + return snapshots, unreadable + + async def _call(self, event: ReportingIssueEvent) -> None: + result = self.observer(event) + if inspect.isawaitable(result): + await asyncio.wait_for(result, self._timeout.total_seconds()) + + def _backoff(self, attempts: int) -> timedelta: + exponent = min(max(attempts - 1, 0), 20) + delay: timedelta = self._retry_base * (1 << exponent) + return min(delay, self._retry_cap) + + async def _report(self, component: str, error: BaseException) -> None: + if self._on_error is not None: + await self._on_error(component, error) + return + logger.error("Reliable Reporting %s failed", component.split(":", 1)[0]) diff --git a/src/adcp/reporting/ledger/__init__.py b/src/adcp/reporting/ledger/__init__.py index 7b5b3b774e..85578d7522 100644 --- a/src/adcp/reporting/ledger/__init__.py +++ b/src/adcp/reporting/ledger/__init__.py @@ -132,6 +132,12 @@ project_obligation_health, ) from adcp.reporting.ledger.host_reads import ReportingCurrentRevisionsPageV1 +from adcp.reporting.ledger.issue_events import ( + ReportingIssueEventLease, + ReportingIssueEventStore, + issue_event_id, + supports_issue_events, +) from adcp.reporting.ledger.models import ( ConsumerStatusRecord, ConsumerStatusValue, @@ -221,6 +227,10 @@ __all__ = [ "ReportingCaller", + "ReportingIssueEventLease", + "ReportingIssueEventStore", + "issue_event_id", + "supports_issue_events", "REPORTING_SELECTOR_VERSION", "ReportingRevisionCorrupt", "ReportingRevisionNotReady", diff --git a/src/adcp/reporting/ledger/issue_events.py b/src/adcp/reporting/ledger/issue_events.py new file mode 100644 index 0000000000..02210f92c3 --- /dev/null +++ b/src/adcp/reporting/ledger/issue_events.py @@ -0,0 +1,193 @@ +"""The issue-lifecycle event outbox a ledger store keeps for the issue observer. + +Every write that moves an issue occurrence's lifecycle -- opening it, +acknowledging or waiving it, retiring it as resolved -- records one event in +the same transaction, and an escalation sweep records ``escalated`` when an +open or acknowledged occurrence passes ``opened_at`` plus the seller's +``consumer_mismatch_escalation``. A worker claims pending events under a lease +and settles each one after handing it to the observer (see +:class:`adcp.reporting.issue_events.ReportingIssueEventDispatcher`). + +A store opts in by implementing :class:`ReportingIssueEventStore` and setting +``issue_events_supported = True``. :class:`InMemoryReportingLedgerStore +` does; +``PgReportingLedgerStore`` does from adcp-reporting-storage 0.2.0b4, once +migration ``0008`` is applied (``python -m adcp migrate``). + +Each transition is recorded exactly once. The lifecycle is forward-only, so an +occurrence makes each transition at most once, and ``(issue_id, kind)`` is +unique; :func:`issue_event_id` derives the event's id from that pair, so every +redelivery of one transition carries the same id. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Literal, Protocol, runtime_checkable + +from adcp.reporting.canonical_json import canonical_json_utf8_v1 + +__all__ = [ + "ISSUE_EVENT_KINDS", + "ReportingIssueEventKind", + "ReportingIssueEventLease", + "ReportingIssueEventStore", + "issue_event_id", + "supports_issue_events", +] + +ReportingIssueEventKind = Literal["opened", "escalated", "acknowledged", "waived", "resolved"] + +#: Every kind, in lifecycle order. +ISSUE_EVENT_KINDS: tuple[ReportingIssueEventKind, ...] = ( + "opened", + "escalated", + "acknowledged", + "waived", + "resolved", +) + +#: The event that entering each ``issue_state`` records. +KIND_FOR_STATE: dict[str, ReportingIssueEventKind] = { + "open": "opened", + "acknowledged": "acknowledged", + "waived": "waived", + "resolved": "resolved", +} + + +def issue_event_id(issue_id: str, kind: str) -> str: + """The stable id of one transition of one issue occurrence. + + Every store derives it the same way (adcp-reporting-storage included), so + an observer can deduplicate on it whichever store recorded the event. + """ + payload = canonical_json_utf8_v1(["reporting-issue-event-v1", issue_id, kind]) + return "rpie_" + hashlib.sha256(payload).hexdigest()[:40] + + +class ReportingIssueEventLease(Protocol): + """One claimed issue event, as a store hands it to a worker.""" + + @property + def event_id(self) -> str: ... + + @property + def account_id(self) -> str: ... + + @property + def consumer_id(self) -> str | None: ... + + @property + def issue_key(self) -> str: ... + + @property + def issue_id(self) -> str: ... + + @property + def generation(self) -> int: ... + + @property + def kind(self) -> ReportingIssueEventKind: ... + + @property + def occurred_at(self) -> datetime: + """When the transition took effect.""" + ... + + @property + def attempts(self) -> int: + """Delivery attempts, this one included.""" + ... + + +@runtime_checkable +class ReportingIssueEventStore(Protocol): + """A ledger store that records issue events and serves them under a lease. + + ``issue_events_supported`` is the opt-in: a store that merely has methods + of these names is not taken to record events beside its lifecycle writes. + """ + + issue_events_supported: bool + + async def issue_events_ready(self) -> bool: + """Whether the store can record events now (its schema is installed).""" + ... + + async def sweep_issue_escalations(self, *, after: timedelta, limit: int = 100) -> int: + """Record ``escalated`` for occurrences open or acknowledged past ``opened_at + after``. + + Settles each swept account first, as a status read would, so a + condition that cleared resolves instead of escalating. Returns the + events recorded. + """ + ... + + async def claim_issue_events( + self, *, lease_seconds: float = 60.0, limit: int = 20 + ) -> tuple[ReportingIssueEventLease, ...]: + """Lease due events, at most one per occurrence, each in recording order.""" + ... + + async def complete_issue_event(self, lease: ReportingIssueEventLease) -> bool: + """Mark a leased event delivered; ``False`` when the lease was lost.""" + ... + + async def release_issue_event( + self, lease: ReportingIssueEventLease, *, retry_after: timedelta + ) -> bool: + """Give back a failed delivery, due again ``retry_after`` from now.""" + ... + + async def prune_issue_events(self, *, older_than: timedelta, limit: int = 1000) -> int: + """Delete up to ``limit`` events delivered more than ``older_than`` ago.""" + ... + + +def supports_issue_events(store: object) -> bool: + """Whether ``store`` opted in to recording issue events.""" + return getattr(store, "issue_events_supported", False) is True and isinstance( + store, ReportingIssueEventStore + ) + + +@dataclass(frozen=True) +class MemoryIssueEventLease: + """The reference store's claimed event.""" + + event_id: str + account_id: str + consumer_id: str | None + issue_key: str + issue_id: str + generation: int + kind: ReportingIssueEventKind + occurred_at: datetime + recorded_at: datetime + attempts: int + lease_token: str = field(repr=False) + lease_expires_at: datetime + + +@dataclass +class MemoryIssueEvent: + """One recorded event in the reference store.""" + + seq: int + event_id: str + account_id: str + consumer_id: str | None + issue_key: str + issue_id: str + generation: int + kind: ReportingIssueEventKind + occurred_at: datetime + recorded_at: datetime + due_at: datetime + attempts: int = 0 + lease_token: str | None = None + lease_expires_at: datetime | None = None + delivered_at: datetime | None = None diff --git a/src/adcp/reporting/ledger/status_snapshot.py b/src/adcp/reporting/ledger/status_snapshot.py index 71b0b330ac..44040c1258 100644 --- a/src/adcp/reporting/ledger/status_snapshot.py +++ b/src/adcp/reporting/ledger/status_snapshot.py @@ -153,9 +153,18 @@ def apply_memory_intents( for intent in intents: issue = intent.lifecycle key = (issue.account_id, issue.issue_key) + previous = store._issues.get(key) store._issues[key] = issue store._issue_generations[key] = max(store._issue_generations.get(key, 0), issue.generation) store._dirty_issue(issue, intent.scope, enqueue=False) + store._record_issue_transition( + ( + previous.issue_state + if previous is not None and previous.issue_id == issue.issue_id + else None + ), + issue, + ) def settle_memory_snapshot( diff --git a/src/adcp/reporting/ledger/store.py b/src/adcp/reporting/ledger/store.py index e336d24870..34bbd323a1 100644 --- a/src/adcp/reporting/ledger/store.py +++ b/src/adcp/reporting/ledger/store.py @@ -40,8 +40,8 @@ from contextvars import ContextVar from copy import deepcopy from dataclasses import dataclass, field, replace -from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, Literal, Protocol, runtime_checkable +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, runtime_checkable from uuid import uuid4 from adcp.reporting.canonical_json import canonical_json_utf8_v1 @@ -53,6 +53,14 @@ ) from adcp.reporting.evidence import ReportingControlTotalRecord from adcp.reporting.ledger.health import issue_id_for_occurrence +from adcp.reporting.ledger.issue_events import ( + KIND_FOR_STATE, + MemoryIssueEvent, + MemoryIssueEventLease, + ReportingIssueEventKind, + ReportingIssueEventLease, + issue_event_id, +) from adcp.reporting.ledger.models import ( ConsumerStatusRecord, LedgerRecordKind, @@ -808,6 +816,10 @@ class InMemoryReportingLedgerStore: place a test's ledger boundary at a deliberate instant. """ + #: Records issue-lifecycle events for the issue observer (see + #: :mod:`adcp.reporting.ledger.issue_events`). + issue_events_supported: ClassVar[bool] = True + def __init__( self, *, clock: Callable[[], datetime] | None = None, notifications: bool = False ) -> None: @@ -845,6 +857,9 @@ def __init__( self._issues: dict[tuple[str, str], ReportingIssueLifecycle] = {} self._issue_generations: dict[tuple[str, str], int] = {} self._issue_status_scopes: dict[tuple[str, str], ReportingStatusScope] = {} + # The issue-event outbox, by event id in recording order. + self._issue_events: dict[str, MemoryIssueEvent] = {} + self._issue_event_seq = 0 self._notification_state: NotificationState | None = None self._status_notification_state: Any = None if notifications: @@ -1790,6 +1805,7 @@ async def ensure_issue_opened( self._issue_generations[key] = generation self._issues[key] = record self._dirty_issue(record, status_scope) + self._record_issue_transition(None, record) return record async def set_issue_state( @@ -1854,6 +1870,11 @@ async def set_issue_state( # nothing, while anything that does move retained evidence stays # reconstructable for the projector. self._dirty_issue(updated, status_scope, live) + self._record_issue_transition( + live.issue_state, + updated, + occurred_at=_utc(at) if state == "acknowledged" else None, + ) return updated async def retire_issue( @@ -1884,6 +1905,7 @@ async def retire_issue( ) self._issues[key] = retired self._dirty_issue(retired, status_scope, live) + self._record_issue_transition(live.issue_state, retired) return retired async def get_issue(self, *, issue_key: str, account_id: str) -> ReportingIssueLifecycle | None: @@ -1894,6 +1916,192 @@ async def get_issue(self, *, issue_key: str, account_id: str) -> ReportingIssueL live = self._issues.get((account_id, issue_key)) return live if live is not None and live.live else None + # -- issue events (adcp-client-python#1258) --------------------------- + + def _record_issue_transition( + self, + before_state: str | None, + after: ReportingIssueLifecycle, + *, + occurred_at: datetime | None = None, + ) -> None: + """Record the events one lifecycle write produced. Caller holds the mutation. + + ``before_state`` is ``None`` when the write created the occurrence. A + write that leaves the state unchanged records nothing. + """ + kinds: list[tuple[ReportingIssueEventKind, datetime]] = [] + if before_state is None: + kinds.append(("opened", _utc(after.opened_at))) + before_state = "open" + if after.issue_state != before_state: + at = occurred_at or after.retired_at or self._clock() + kinds.append((KIND_FOR_STATE[after.issue_state], _utc(at))) + for kind, at in kinds: + self._insert_issue_event(after, kind, occurred_at=at) + + def _insert_issue_event( + self, + issue: ReportingIssueLifecycle, + kind: ReportingIssueEventKind, + *, + occurred_at: datetime, + ) -> bool: + event_id = issue_event_id(issue.issue_id, kind) + if event_id in self._issue_events: + return False # exactly once per (issue_id, kind) + now = _utc(self._clock()) + self._issue_event_seq += 1 + self._issue_events[event_id] = MemoryIssueEvent( + seq=self._issue_event_seq, + event_id=event_id, + account_id=issue.account_id, + consumer_id=issue.consumer_id, + issue_key=issue.issue_key, + issue_id=issue.issue_id, + generation=issue.generation, + kind=kind, + occurred_at=occurred_at, + recorded_at=now, + due_at=now, + ) + return True + + async def issue_events_ready(self) -> bool: + return True + + async def sweep_issue_escalations(self, *, after: timedelta, limit: int = 100) -> int: + from adcp.reporting.ledger.status_snapshot import settle_memory_snapshot + + if after < timedelta(0): + raise ValueError("after must be nonnegative") + if limit < 1: + raise ValueError("limit must be positive") + async with self._mutation(): + now = _utc(self._clock()) + escalated = {e.issue_id for e in self._issue_events.values() if e.kind == "escalated"} + accounts = sorted( + { + issue.account_id + for issue in self._issues.values() + if issue.issue_state in {"open", "acknowledged"} + and _utc(issue.opened_at) + after <= now + and issue.issue_id not in escalated + } + )[:limit] + recorded = 0 + for account_id in accounts: + # Settle first, as a status read would: a condition that + # cleared resolves instead of escalating. + settle_memory_snapshot(self, account_id) + for issue in sorted( + (i for i in self._issues.values() if i.account_id == account_id), + key=lambda i: (i.opened_at, i.issue_id), + ): + if ( + issue.issue_state in {"open", "acknowledged"} + and _utc(issue.opened_at) + after <= now + and self._insert_issue_event( + issue, "escalated", occurred_at=_utc(issue.opened_at) + after + ) + ): + recorded += 1 + return recorded + + async def claim_issue_events( + self, *, lease_seconds: float = 60.0, limit: int = 20 + ) -> tuple[MemoryIssueEventLease, ...]: + if lease_seconds <= 0: + raise ValueError("lease_seconds must be positive") + if limit < 1: + raise ValueError("limit must be positive") + async with self._mutation(): + now = _utc(self._clock()) + pending = sorted( + (e for e in self._issue_events.values() if e.delivered_at is None), + key=lambda e: e.seq, + ) + first_pending: dict[str, int] = {} + for event in pending: + first_pending.setdefault(event.issue_id, event.seq) + due = sorted( + ( + e + for e in pending + if first_pending[e.issue_id] == e.seq + and e.due_at <= now + and (e.lease_expires_at is None or e.lease_expires_at <= now) + ), + key=lambda e: (e.due_at, e.seq), + )[:limit] + token = uuid4().hex + expires = now + timedelta(seconds=lease_seconds) + claimed: list[MemoryIssueEventLease] = [] + for event in due: + event.lease_token, event.lease_expires_at = token, expires + event.attempts += 1 + claimed.append( + MemoryIssueEventLease( + event_id=event.event_id, + account_id=event.account_id, + consumer_id=event.consumer_id, + issue_key=event.issue_key, + issue_id=event.issue_id, + generation=event.generation, + kind=event.kind, + occurred_at=event.occurred_at, + recorded_at=event.recorded_at, + attempts=event.attempts, + lease_token=token, + lease_expires_at=expires, + ) + ) + return tuple(claimed) + + def _leased(self, lease: ReportingIssueEventLease) -> MemoryIssueEvent | None: + event = self._issue_events.get(lease.event_id) + token = getattr(lease, "lease_token", None) + if event is None or event.delivered_at is not None or event.lease_token != token: + return None + return event + + async def complete_issue_event(self, lease: ReportingIssueEventLease) -> bool: + async with self._mutation(): + event = self._leased(lease) + if event is None: + return False + event.delivered_at = _utc(self._clock()) + event.lease_token = event.lease_expires_at = None + return True + + async def release_issue_event( + self, lease: ReportingIssueEventLease, *, retry_after: timedelta + ) -> bool: + async with self._mutation(): + event = self._leased(lease) + if event is None: + return False + event.due_at = _utc(self._clock()) + max(retry_after, timedelta(0)) + event.lease_token = event.lease_expires_at = None + return True + + async def prune_issue_events(self, *, older_than: timedelta, limit: int = 1000) -> int: + if limit < 1: + raise ValueError("limit must be positive") + async with self._mutation(): + horizon = _utc(self._clock()) - older_than + expired = sorted( + ( + e + for e in self._issue_events.values() + if e.delivered_at is not None and e.delivered_at < horizon + ), + key=lambda e: e.delivered_at or horizon, + )[:limit] + for event in expired: + del self._issue_events[event.event_id] + return len(expired) + # -- snapshots ------------------------------------------------------- async def open_snapshot( diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index 302ae45326..18ef026a92 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -17,7 +17,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable from uuid import uuid4 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -69,6 +69,11 @@ from adcp.types import ReportingDeliveryOffering if TYPE_CHECKING: + from adcp.reporting.issue_events import ( + ReportingIssueDelivery, + ReportingIssueEventDispatcher, + ReportingIssueObserver, + ) from adcp.reporting.production.contracts import ReportingProductionSourceBinding from adcp.reporting.production.service import ReportingProductionSupport from adcp.reporting.service_production import ReportingProductionOptions @@ -469,6 +474,8 @@ class ReliableReportingTurn: subscription_errors: dict[tuple[str, str | None, str | None], BaseException] = field( default_factory=dict ) + #: What the issue observer's delivery did this turn, when one is configured. + issue_events: ReportingIssueDelivery | None = None @property def did_work(self) -> bool: @@ -479,6 +486,7 @@ def did_work(self) -> bool: or bool(self.extension_errors) or bool(self.subscription_obligations_written) or bool(self.subscription_errors) + or (self.issue_events is not None and self.issue_events.did_work) ) @@ -509,6 +517,7 @@ def __init__( receipt_handler: ReportingReceiptHandler | None = None, reconciled_billing: bool = False, worker_error_handler: ReportingWorkerErrorHandler | None = None, + issue_observer: ReportingIssueObserver | None = None, owned_resources: Sequence[ReportingServiceResource] = (), production: ReportingProductionOptions | None = None, automated_recovery_window: timedelta = timedelta(hours=6), @@ -537,7 +546,18 @@ def __init__( ledger ``store``, the default ``staging``/``seals``, and the principal and subscription-state stores of the managed subscription path. Durable storage checks its schema version at startup instead of changing it. - Pass ``storage`` or ``store``, not both.""" + Pass ``storage`` or ``store``, not both. + + ``issue_observer`` is called once per lifecycle transition of a + reporting issue (``opened``, ``escalated``, ``acknowledged``, + ``waived``, ``resolved``) with a + :class:`~adcp.reporting.issue_events.ReportingIssueEvent`, from the + worker turn, after the transition commits. Delivery is at least once: + deduplicate on ``event.event_id``. An observer that raises is reported + to ``worker_error_handler`` and retried; it never fails a buyer's call. + It needs a store that records issue events (the in-memory store, or + adcp-reporting-storage 0.2.0b4+ with migration ``0008``) and a + scheduled worker turn somewhere. See :mod:`adcp.reporting.issue_events`.""" if storage is not None: from adcp.storage import ReportingStorage @@ -572,6 +592,12 @@ def __init__( raise ReliableReportingConfigurationError( "production options compose their own workers, admission and receipt handlers" ) + if production is not None and issue_observer is not None: + raise ReliableReportingConfigurationError( + "issue_observer is not supported with production= yet " + "(adcp-client-python#1873); compose the service without production= to " + "observe reporting issues" + ) if subscriptions is not None and reporting_subscriptions is not None: raise ReliableReportingConfigurationError( "pass subscriptions (managed) or reporting_subscriptions (advanced), not both" @@ -629,6 +655,9 @@ def __init__( self._receipt_handler = receipt_handler self._reconciled_billing = reconciled_billing self._worker_error_handler = worker_error_handler + self._issue_events: ReportingIssueEventDispatcher | None = None + if issue_observer is not None: + self._issue_events = self._issue_event_dispatcher(store, issue_observer) self._bindings: dict[ReportingConfigurationGenerationKey, _Binding] = {} self._pending_configurations: dict[ ReportingConfigurationGenerationKey, ReportingConfiguration @@ -673,6 +702,31 @@ def __init__( configuration_error=ReliableReportingConfigurationError, ) + def _issue_event_dispatcher( + self, store: ReportingLedgerStore, observer: ReportingIssueObserver + ) -> ReportingIssueEventDispatcher: + from adcp.reporting.issue_events import ReportingIssueEventDispatcher + from adcp.reporting.ledger.issue_events import ( + ReportingIssueEventStore, + supports_issue_events, + ) + + if not callable(observer): + raise ReliableReportingConfigurationError("issue_observer must be callable") + if not supports_issue_events(store): + raise ReliableReportingConfigurationError( + f"issue_observer needs a ledger store that records issue events, and " + f"{type(store).__name__} does not: use memory_storage(), or " + "postgres_storage(...) with adcp-reporting-storage 0.2.0b4 or later " + "(pip install -U adcp-reporting-storage), then run python -m adcp migrate" + ) + return ReportingIssueEventDispatcher( + cast(ReportingIssueEventStore, store), + observer, + escalation=self._escalation, + on_error=self._report_worker_error, + ) + @classmethod def from_production( cls, @@ -1440,6 +1494,12 @@ async def _initialize(self) -> None: await self.store.create_schema() if self._initialize_source_storage is not None: await self._initialize_source_storage() + issue_events = self._issue_events + if issue_events is not None and not await issue_events.store.issue_events_ready(): + raise ReliableReportingConfigurationError( + "issue_observer needs the issue-event table (adcp-reporting-storage " + "migration 0008): run python -m adcp migrate" + ) for configuration in self._pending_configurations.values(): await self.store.put_configuration(configuration) self._pending_configurations.clear() @@ -1611,6 +1671,14 @@ async def _run_worker(self, *, now: datetime | None) -> ReliableReportingTurn: continue if result is not None: turn.extension_results.append(result) + if self._issue_events is not None and not self._lifecycle.stopping: + try: + turn.issue_events = await self._issue_events.run_once() + except Exception as error: + # The observer's own failures are reported per event and + # retried; this is the store (claim, sweep) failing. + turn.extension_errors["issue_events"] = error + await self._report_worker_error("issue_events", error) return turn async def caller_for(self, request: Any, context: Any | None) -> ReportingStatusCaller: diff --git a/tests/conformance/reporting/test_reporting_issue_events_pg.py b/tests/conformance/reporting/test_reporting_issue_events_pg.py new file mode 100644 index 0000000000..645661591a --- /dev/null +++ b/tests/conformance/reporting/test_reporting_issue_events_pg.py @@ -0,0 +1,356 @@ +"""The issue-lifecycle event outbox against real PostgreSQL (adcp-client-python#1258). + +Migration ``0008`` installs ``reporting_issue_events``; every write that moves +a ``reporting_issue_lifecycle`` row records its transition there in the same +transaction, the escalation sweep records ``escalated`` exactly once, and a +worker claims, completes or releases events under a lease. The issue observer +in ``adcp`` reads this outbox; these tests pin the storage half. + +Requires ``ADCP_PG_TEST_URL``; skips otherwise. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import timedelta +from typing import Any + +import pytest + +from adcp.reporting.ledger.status_projection import mismatch_key + +from ._generation_support import ( + configuration, + isolated_reporting_pool, + obligation_for, + revision_for, +) +from ._reliable_support import ManualClock +from .test_reporting_notification_outbox import statement + +ACCOUNT = "acct_a" +HOUR = timedelta(hours=1) + + +async def _store(pool: Any, clock: ManualClock, *, migrated: bool = True) -> Any: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + from adcp_reporting_storage.migrations import migrate + + store = PgReportingLedgerStore(pool=pool, clock=clock) + await store.create_schema() + if migrated: + await migrate(pool) + return store + + +async def _events(pool: Any) -> list[dict[str, Any]]: + async with pool.connection() as connection: + cursor = await connection.execute( + "SELECT event_id, issue_id, kind, occurred_at, attempts, delivered_at, generation" + " FROM reporting_issue_events ORDER BY seq" + ) + names = [column.name for column in cursor.description] + return [dict(zip(names, row)) for row in await cursor.fetchall()] + + +async def _open(store: Any, clock: ManualClock, key: str = "rpik_condition") -> Any: + return await store.ensure_issue_opened( + issue_key=key, account_id=ACCOUNT, consumer_id="buyer", observed_at=clock() + ) + + +async def _catalog(pool: Any) -> set[tuple[str, str]]: + async with pool.connection() as connection: + rows = await ( + await connection.execute( + "SELECT c.relkind::text, c.relname FROM pg_class c" + " JOIN pg_namespace n ON n.oid = c.relnamespace" + " WHERE n.nspname = current_schema()" + " UNION ALL" + " SELECT 'column', table_name || '.' || column_name || ':' || data_type" + " FROM information_schema.columns WHERE table_schema = current_schema()" + " UNION ALL" + " SELECT 'constraint', conrelid::regclass::text || '.' || conname" + " FROM pg_constraint WHERE connamespace = current_schema()::regnamespace" + ) + ).fetchall() + return {(kind, name) for kind, name in rows} + + +async def test_migration_0008_adds_only_the_event_table() -> None: + """Purely additive: dropping its objects leaves exactly the 0001-0007 catalog.""" + from adcp_reporting_storage.migrations import migrate + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + with_events = await _catalog(pool) + async with pool.connection() as connection: + await connection.execute("DROP TABLE reporting_issue_events") + await connection.execute("DELETE FROM adcp_schema_migrations WHERE version = '0008'") + without = await _catalog(pool) + added = with_events - without + assert without <= with_events # nothing existing was changed or removed + # Only the table, its columns, constraints, indexes and sequence. + assert all(name.startswith("reporting_issue_events") for _, name in added), added + assert ("r", "reporting_issue_events") in added + # Re-applying records it again and restores the same catalog. + assert [item.version for item in await migrate(pool)] == ["0008"] + assert await _catalog(pool) == with_events + + +async def test_an_unmigrated_ledger_records_nothing_and_says_so() -> None: + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock, migrated=False) + assert store.issue_events_supported is True + assert not await store.issue_events_ready() + issue = await _open(store, clock) # the lifecycle write itself is unchanged + await store.retire_issue(issue_key=issue.issue_key, account_id=ACCOUNT, at=clock()) + assert await store.claim_issue_events() == () + assert await store.sweep_issue_escalations(after=timedelta(0)) == 0 + assert await store.prune_issue_events(older_than=timedelta(0)) == 0 + from adcp_reporting_storage.migrations import migrate + + await migrate(pool) + assert await store.issue_events_ready() + assert await _events(pool) == [] # history is not backfilled + + +async def test_each_lifecycle_write_records_its_transition_once() -> None: + from adcp_reporting_storage.ledger.issue_events import issue_event_id + + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + opened_at = clock() + first = await _open(store, clock) + clock.advance(timedelta(minutes=1)) + assert (await _open(store, clock)).issue_id == first.issue_id # idempotent: no event + acknowledged_at = clock.advance(timedelta(minutes=1)) + for _ in range(2): # a re-acknowledge changes nothing and records nothing + await store.set_issue_state( + issue_key=first.issue_key, + account_id=ACCOUNT, + state="acknowledged", + at=acknowledged_at, + ) + await store.set_issue_state( # external_ref alone is not a transition + issue_key=first.issue_key, + account_id=ACCOUNT, + state="acknowledged", + at=clock.advance(), + external_ref="TICKET-1", + ) + resolved_at = clock.advance(timedelta(minutes=1)) + await store.retire_issue(issue_key=first.issue_key, account_id=ACCOUNT, at=resolved_at) + assert ( + await store.retire_issue( # convergent: nothing live, nothing recorded + issue_key=first.issue_key, account_id=ACCOUNT, at=clock.advance() + ) + is None + ) + reopened_at = clock.advance(timedelta(minutes=1)) + second = await _open(store, clock) + assert second.issue_id != first.issue_id and second.generation == 2 + waived_at = clock.advance(timedelta(minutes=1)) + await store.set_issue_state( + issue_key=second.issue_key, account_id=ACCOUNT, state="waived", at=waived_at + ) + events = await _events(pool) + assert [(e["issue_id"], e["kind"], e["occurred_at"]) for e in events] == [ + (first.issue_id, "opened", opened_at), + (first.issue_id, "acknowledged", acknowledged_at), + (first.issue_id, "resolved", resolved_at), + (second.issue_id, "opened", reopened_at), + (second.issue_id, "waived", waived_at), + ] + assert [e["event_id"] for e in events] == [ + issue_event_id(e["issue_id"], e["kind"]) for e in events + ] + assert all(e["attempts"] == 0 and e["delivered_at"] is None for e in events) + + +async def test_a_rolled_back_lifecycle_write_records_no_event() -> None: + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + with pytest.raises(RuntimeError, match="abort"): + async with store.transaction(): + await _open(store, clock) + raise RuntimeError("abort") + assert await store.get_issue(issue_key="rpik_condition", account_id=ACCOUNT) is None + assert await _events(pool) == [] + + +async def _seed_mismatch(store: Any, clock: ManualClock) -> tuple[Any, Any, Any]: + """A readable revision the buyer reports unreadable: a seller-side mismatch.""" + config = configuration() + await store.put_configuration(config) + obligation = await store.commit_obligation(obligation_for(config)) + revision, rows = revision_for(obligation) + await store.commit_revision(revision, rows) + unreadable = replace( + statement(obligation), + consumer_status="unreadable", + failure_code="transport_failed", + reporting_revision_id=revision.reporting_revision_id, + status_as_of=clock(), + recorded_at=clock(), + ) + await store.record_consumer_status_with_lifecycle(unreadable) + return obligation, revision, unreadable + + +async def test_ingest_and_agreement_record_opened_and_resolved_in_the_status_transaction() -> None: + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + _, revision, unreadable = await _seed_mismatch(store, clock) + issue = await store.get_issue(account_id=ACCOUNT, issue_key=mismatch_key(unreadable)) + assert issue is not None and issue.opened_at == unreadable.recorded_at + clock.advance(timedelta(minutes=5)) + agreeing = replace( + unreadable, + reporting_status_id="consumer-status-buyer-agrees", + consumer_status="received", + failure_code=None, + observed_revision_content_sha256=revision.revision_content_sha256, + supersedes_reporting_status_id=unreadable.reporting_status_id, + status_as_of=clock(), + recorded_at=clock(), + ) + await store.record_consumer_status_with_lifecycle(agreeing) + events = await _events(pool) + assert [(e["issue_id"], e["kind"]) for e in events] == [ + (issue.issue_id, "opened"), + (issue.issue_id, "resolved"), + ] + assert events[0]["occurred_at"] == unreadable.recorded_at + assert events[1]["occurred_at"] == clock() + + +async def test_escalation_is_recorded_once_at_the_boundary_across_sweepers() -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + opened = await _open(store, clock, "rpik_open") + acknowledged = await _open(store, clock, "rpik_acknowledged") + await store.set_issue_state( + issue_key=acknowledged.issue_key, account_id=ACCOUNT, state="acknowledged", at=clock() + ) + resolved = await _open(store, clock, "rpik_resolved") + await store.retire_issue(issue_key=resolved.issue_key, account_id=ACCOUNT, at=clock()) + waived = await _open(store, clock, "rpik_waived") + await store.set_issue_state( + issue_key=waived.issue_key, account_id=ACCOUNT, state="waived", at=clock() + ) + boundary = opened.opened_at + HOUR + clock.now = boundary - timedelta(seconds=1) + assert await store.sweep_issue_escalations(after=HOUR) == 0 + clock.now = boundary + replicas = [PgReportingLedgerStore(pool=pool, clock=clock) for _ in range(4)] + recorded = await asyncio.gather( + *(replica.sweep_issue_escalations(after=HOUR) for replica in replicas) + ) + assert sum(recorded) == 2 # open and acknowledged; never resolved or waived + assert await store.sweep_issue_escalations(after=HOUR) == 0 + escalated = [e for e in await _events(pool) if e["kind"] == "escalated"] + assert {e["issue_id"] for e in escalated} == {opened.issue_id, acknowledged.issue_id} + assert {e["occurred_at"] for e in escalated} == {boundary} + + +async def test_a_condition_that_cleared_unobserved_resolves_instead_of_escalating() -> None: + """The sweep settles the account first, as a status read would.""" + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + _, revision, unreadable = await _seed_mismatch(store, clock) + issue = await store.get_issue(account_id=ACCOUNT, issue_key=mismatch_key(unreadable)) + # The seller now agrees the revision is unreadable; nothing has re-read since. + await store.set_revision_readable( + account_id=ACCOUNT, reporting_revision_id=revision.reporting_revision_id, readable=False + ) + assert await store.get_issue(account_id=ACCOUNT, issue_key=issue.issue_key) is not None + clock.now = issue.opened_at + HOUR + assert await store.sweep_issue_escalations(after=HOUR) == 0 + assert await store.get_issue(account_id=ACCOUNT, issue_key=issue.issue_key) is None + assert [e["kind"] for e in await _events(pool)] == ["opened", "resolved"] + + +async def test_claims_deliver_each_occurrence_in_order_under_exclusive_leases() -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + a = await _open(store, clock, "rpik_a") + await store.set_issue_state( + issue_key=a.issue_key, account_id=ACCOUNT, state="acknowledged", at=clock.advance() + ) + b = await _open(store, clock, "rpik_b") + # One event per occurrence, oldest first; the second worker gets nothing. + first = await store.claim_issue_events(lease_seconds=30, limit=10) + assert [(lease.issue_id, lease.kind, lease.attempts) for lease in first] == [ + (a.issue_id, "opened", 1), + (b.issue_id, "opened", 1), + ] + other = PgReportingLedgerStore(pool=pool, clock=clock) + assert await other.claim_issue_events(lease_seconds=30, limit=10) == () + # A's next event becomes claimable once its first is delivered. + assert await store.complete_issue_event(first[0]) + assert not await store.complete_issue_event(first[0]) # already settled + (acknowledged,) = await other.claim_issue_events(lease_seconds=30, limit=10) + assert (acknowledged.issue_id, acknowledged.kind) == (a.issue_id, "acknowledged") + # A failed delivery is due again retry_after from now, and not before. + retry_at = clock() + timedelta(seconds=10) # before B's lease expires + assert await other.release_issue_event(acknowledged, retry_after=timedelta(seconds=10)) + assert await store.claim_issue_events(lease_seconds=30) == () + clock.now = retry_at + (again,) = await store.claim_issue_events(lease_seconds=30) + assert (again.event_id, again.attempts) == (acknowledged.event_id, 2) + # An expired lease is claimed again; the stale holder can no longer settle it. + clock.now = retry_at + timedelta(seconds=31) + reclaimed = await other.claim_issue_events(lease_seconds=30) + assert {lease.event_id for lease in reclaimed} == {first[1].event_id, again.event_id} + assert not await store.complete_issue_event(first[1]) + assert not await store.release_issue_event(again, retry_after=timedelta(0)) + for lease in reclaimed: + assert await other.complete_issue_event(lease) + assert await store.claim_issue_events() == () + events = await _events(pool) + assert all(e["delivered_at"] is not None for e in events) + + +async def test_concurrent_claims_never_lease_one_event_twice() -> None: + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + for index in range(12): + await _open(store, clock, f"rpik_{index:02d}") + workers = [PgReportingLedgerStore(pool=pool, clock=clock) for _ in range(4)] + claimed = await asyncio.gather( + *(worker.claim_issue_events(lease_seconds=30, limit=5) for worker in workers) + ) + ids = [lease.event_id for batch in claimed for lease in batch] + assert len(ids) == len(set(ids)) == 12 + + +async def test_prune_deletes_only_events_delivered_before_the_horizon() -> None: + async with isolated_reporting_pool(autocommit=True) as pool: + clock = ManualClock() + store = await _store(pool, clock) + for key in ("rpik_old", "rpik_new", "rpik_pending"): + await _open(store, clock, key) + old, new, pending = await store.claim_issue_events(limit=3) + assert await store.complete_issue_event(old) + clock.advance(timedelta(days=1)) + assert await store.complete_issue_event(new) + clock.advance(timedelta(hours=1)) + assert await store.prune_issue_events(older_than=timedelta(days=1)) == 1 + remaining = {e["issue_id"] for e in await _events(pool)} + assert remaining == {new.issue_id, pending.issue_id} diff --git a/tests/conformance/reporting/test_reporting_issue_observer_pg.py b/tests/conformance/reporting/test_reporting_issue_observer_pg.py new file mode 100644 index 0000000000..32ea5da1ad --- /dev/null +++ b/tests/conformance/reporting/test_reporting_issue_observer_pg.py @@ -0,0 +1,106 @@ +"""The issue observer through ``ReliableReportingService`` on PostgreSQL (#1258). + +The same flows as the memory reference (``tests/test_reporting_issue_observer.py``) +over ``postgres_storage``: the issue opens on ``sync_reporting_status``, escalates +on the worker turn when the boundary passes, resolves on agreement, and a +failing observer is reported and retried with the same ``event_id``. Two +services on one ledger (two replicas) deliver each event once. + +Requires ``ADCP_PG_TEST_URL``; skips otherwise. +""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta + +from adcp.reporting.service import ReliableReportingService +from adcp.reporting.testing import DeterministicReportingClock +from tests.test_reliable_reporting_service import NOW +from tests.test_reporting_issue_observer import ( + ESCALATION, + Observed, + published_period, + run_failing_observer_flow, + run_lifecycle_flow, + service_options, + unreadable, +) + +from ._generation_support import isolated_reporting_pool + + +async def _service(pool, clock, observed) -> ReliableReportingService: + from adcp.storage.postgres import postgres_storage + + return ReliableReportingService( + storage=postgres_storage(pool=pool, clock=clock), **service_options(clock, observed) + ) + + +async def test_the_lifecycle_reaches_the_observer_once_per_transition_on_postgres() -> None: + from adcp_reporting_storage.migrations import migrate + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = await _service(pool, clock, observed) + try: + await run_lifecycle_flow(service, clock, observed) + finally: + await service.close() + + +async def test_a_failing_observer_is_retried_with_the_same_event_id_on_postgres() -> None: + from adcp_reporting_storage.migrations import migrate + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = await _service(pool, clock, observed) + try: + await run_failing_observer_flow(service, clock, observed) + finally: + await service.close() + + +async def test_two_replicas_deliver_each_event_once() -> None: + from adcp_reporting_storage.migrations import migrate + + async with isolated_reporting_pool(autocommit=True) as pool: + await migrate(pool) + clock = DeterministicReportingClock(NOW) + observed = Observed() + first = await _service(pool, clock, observed) + second = await _service(pool, clock, observed) + try: + period, revision = await published_period(first) + clock.advance(timedelta(minutes=1)) + await first.sync_reporting_status({"statuses": [unreadable(period, revision, clock())]}) + clock.advance(ESCALATION.consumer_mismatch_escalation) + await asyncio.gather(first.run_worker(), second.run_worker()) + await asyncio.gather(first.run_worker(), second.run_worker()) + assert observed.kinds() == ["opened", "escalated"] + assert observed.errors == [] + finally: + await first.close() + await second.close() + + +async def test_the_service_refuses_to_start_before_migration_0008() -> None: + import pytest + from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore + + from adcp.reporting.service import ReliableReportingConfigurationError + + async with isolated_reporting_pool(autocommit=True) as pool: + clock = DeterministicReportingClock(NOW) + observed = Observed() + options = service_options(clock, observed) + service = ReliableReportingService( + store=PgReportingLedgerStore(pool=pool, clock=clock), **options + ) + with pytest.raises(ReliableReportingConfigurationError, match="python -m adcp migrate"): + await service.initialize() diff --git a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py index c3cfdf542d..0ba8b2777d 100644 --- a/tests/conformance/reporting/test_reporting_storage_migrations_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_migrations_pg.py @@ -34,8 +34,9 @@ async def test_migrate_applies_every_version_once_and_status_reads_it() -> None: lines = describe(await migration_status(pool)) assert all(": applied " in line for line in lines) # A new ledger was never expanded: its contract step applied without --contract. - assert lines[-1].startswith("0007 reporting_contract_generation_keys: applied ") - assert lines[-1].endswith("[contract]") + assert lines[-2].startswith("0007 reporting_contract_generation_keys: applied ") + assert lines[-2].endswith("[contract]") + assert lines[-1].startswith("0008 reporting_issue_events: applied ") assert not any("stopped-worker" in line for line in lines) @@ -54,7 +55,7 @@ async def test_storage_refuses_a_schema_that_is_behind_and_starts_once_migrated( error = refused.value assert error.code == "REPORTING_SCHEMA_BEHIND" and error.recovery == "correctable" # The contract step alone never makes a schema "behind" (0007 is not listed). - assert "pending: 0001, 0002, 0003, 0004, 0005, 0006)" in str(error) + assert "pending: 0001, 0002, 0003, 0004, 0005, 0006, 0008)" in str(error) assert "python -m adcp migrate" in str(error) assert "stopped-worker" not in str(error) # every upgrade is rolling-deploy safe await migrate(pool) @@ -97,7 +98,7 @@ async def test_the_cli_migrates_and_reports_status(capsys) -> None: "0007 reporting_contract_generation_keys: pending [contract: run once every" " earlier-release reporting process has stopped: python -m adcp migrate --contract]" ) in status - assert "7 pending" in status + assert "8 pending" in status assert await _run(dsn, status=False, dry_run=False) == 0 applied = capsys.readouterr().out assert "applied 0002 reporting_subscription_generations\n" in applied @@ -271,6 +272,7 @@ async def test_create_schema_builds_its_own_schema_beside_a_migrated_one() -> No "0005", "0006", "0007", + "0008", ] assert await _catalog(other, other_schema) == before @@ -320,6 +322,7 @@ async def test_a_new_ledger_is_created_contracted(entry: str, capsys) -> None: "0005", "0006", "0007", + "0008", ] assert await _expand_objects(pool) == [] assert not await needs_contract(pool) diff --git a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py index fa9bd3f7b0..a74d558e8b 100644 --- a/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py +++ b/tests/conformance/reporting/test_reporting_storage_published_upgrade_pg.py @@ -244,6 +244,8 @@ def _rest(*, contract: bool = True) -> list[str]: #: The contract steps: applied to an expanded ledger only with ``contract=True``. CONTRACT_VERSIONS = ["0007"] +#: Additive migrations after the contract step: applied even while it is held. +ADDITIVE_AFTER_CONTRACT = ["0008"] #: The objects that exist only while a ledger is expanded (catalog lines). EXPANDED_LEDGER = { "constraint reporting_configurations reporting_configurations_generation_key", @@ -695,6 +697,9 @@ async def test_a_rolling_deploy_runs_published_and_new_pods_side_by_side_until_c expanded = EXPANDED_LEDGER | (EXPANDED_STATUS if notifications else set()) assert _expand_only(await _catalog(pool)) == expanded assert_evidence_unchanged(seeded["image"], await _image_here(pool)) + # The additive issue-event outbox (0008, #1258) is installed before the + # contract step, so every published-pod step below runs beside it. + assert await PgReportingLedgerStore(pool=pool).issue_events_ready() # 3. A published pod restarts, reads, records a generation, closes a # period and runs a full worker turn. old = await _old_pod( @@ -993,7 +998,7 @@ async def test_a_ledger_the_published_0_2_0b1_widened_counts_as_contracted( assert await _catalog(pool) == before # The contract step is recorded without --contract: it changes nothing here. applied = [item.version for item in await migrate(pool)] - assert applied[-1:] == CONTRACT_VERSIONS + assert applied[-2:] == [*CONTRACT_VERSIONS, *ADDITIVE_AFTER_CONTRACT] await postgres_storage(pool=pool).check_schema() after = await _catalog(pool) # 0004 and 0005 add their own tables assert set(before) <= set(after) and _expand_only(after) == set() @@ -1025,6 +1030,7 @@ async def test_migrate_holds_the_contract_step_and_status_says_when_to_run_it( planned = capsys.readouterr().out assert "would apply 0006 reporting_status_checkpoint_offerings\n" in planned assert "would apply 0007" not in planned # held + assert "would apply 0008 reporting_issue_events\n" in planned # additive: not held assert ( "not applied 0007 reporting_contract_generation_keys (contract): run" f' "python -m adcp migrate --contract" {when}' @@ -1034,19 +1040,21 @@ async def test_migrate_holds_the_contract_step_and_status_says_when_to_run_it( applied = capsys.readouterr().out assert "applied 0002 reporting_subscription_generations\n" in applied assert "\napplied 0007" not in applied + assert "applied 0008 reporting_issue_events\n" in applied assert "not applied 0007 reporting_contract_generation_keys (contract)" in applied await postgres_storage(pool=pool).check_schema() # starts: only the contract pends assert await _run(dsn, status=True, dry_run=False) == 0 status = capsys.readouterr().out.strip().splitlines() - assert status[-2] == ( + assert status[-3] == ( "0007 reporting_contract_generation_keys: pending [contract: run" f" {when}: python -m adcp migrate --contract]" ) + assert status[-2].startswith("0008 reporting_issue_events: applied ") assert status[-1] == ( f'up to date except the contract step: run "python -m adcp migrate --contract" {when}' ) - assert describe(await migration_status(pool))[-1] == status[-2] + assert describe(await migration_status(pool))[-2] == status[-3] _all_ok(await _old_pod(python, pool, workspace, notifications=False, ordinal=2)) assert await _run(dsn, status=False, dry_run=False, contract=True) == 0 diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 90565a87ba..c86c38b366 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -801,6 +801,10 @@ "ReportingFailureCode": "union[value:'access_denied',value:'integrity_mismatch',value:'reader_incompatible',value:'resource_not_found',value:'transport_failed']", "ReportingFailurePlan": "adcp.reporting._testing_failures", "ReportingInspectionContext": "adcp.reporting._reconcile", + "ReportingIssueEvent": "adcp.reporting.issue_events", + "ReportingIssueEventDispatcher": "adcp.reporting.issue_events", + "ReportingIssueEventKind": "union[value:'acknowledged',value:'escalated',value:'opened',value:'resolved',value:'waived']", + "ReportingIssueObserver": "union[builtins:object,instance:builtins:list]", "ReportingLedger": "adcp.reporting._reconcile", "ReportingLifecycleSnapshot": "adcp.reporting._testing_lifecycle", "ReportingObservation": "adcp.reporting._reconcile", @@ -873,6 +877,8 @@ "ReportingFinality": "union[value:'official',value:'snapshot']", "ReportingHealth": "union[value:'action_required',value:'complete',value:'delayed',value:'healthy',value:'waiting']", "ReportingIssue": "adcp.reporting.ledger.models", + "ReportingIssueEventLease": "adcp.reporting.ledger.issue_events", + "ReportingIssueEventStore": "adcp.reporting.ledger.issue_events", "ReportingIssueLifecycle": "adcp.reporting.ledger.models", "ReportingIssueStateValue": "union[value:'acknowledged',value:'open',value:'resolved',value:'waived']", "ReportingLedgerStore": "adcp.reporting.ledger.store", @@ -944,6 +950,7 @@ "current_required_revision": "adcp.reporting.ledger.health", "derive_period": "adcp.reporting.ledger.models", "iso_duration_to_timedelta": "adcp.reporting.ledger.models", + "issue_event_id": "adcp.reporting.ledger.issue_events", "issue_id_for": "adcp.reporting.ledger.health", "issue_id_for_occurrence": "adcp.reporting.ledger.health", "issue_is_retirable": "adcp.reporting.ledger.store", @@ -960,6 +967,7 @@ "revision_to_wire": "adcp.reporting.ledger.delivery", "select_reporting_revision": "adcp.reporting.revision_selection", "stale_received_grace_deadline": "adcp.reporting.ledger.consumer_status", + "supports_issue_events": "adcp.reporting.ledger.issue_events", "validate_currency": "adcp.reporting.currency" } }, @@ -9034,6 +9042,7 @@ "ReportingChangeKindV1": "union[value:'adjustment',value:'obligation',value:'retirement',value:'revision']", "ReportingChangeRecordV1": "adcp_reporting_storage.ledger.change_feed", "ReportingChangesPageV1": "adcp_reporting_storage.ledger.change_feed", + "ReportingIssueEventLease": "adcp_reporting_storage.ledger.issue_events", "ReportingObjectRowBindingV1": "adcp_reporting_storage.ledger.row_storage", "ReportingOrphanBindingResultV1": "adcp_reporting_storage.ledger.orphan_collector", "ReportingOrphanCandidateV1": "adcp_reporting_storage.ledger.orphan_collector", diff --git a/tests/test_reporting_issue_observer.py b/tests/test_reporting_issue_observer.py new file mode 100644 index 0000000000..22ff98e42c --- /dev/null +++ b/tests/test_reporting_issue_observer.py @@ -0,0 +1,477 @@ +"""The reporting-issue observer: lifecycle events recorded with each transition (#1258). + +The memory reference store records an event in the same mutation as every +issue-lifecycle write, the dispatcher delivers them at least once, and +``ReliableReportingService(issue_observer=...)`` runs it on the worker turn. +``tests/conformance/reporting/test_reporting_issue_observer_pg.py`` runs the +service flow below against PostgreSQL. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from datetime import timedelta +from types import SimpleNamespace +from typing import Any + +import pytest + +from adcp.reporting import ReportingIssueEvent, ReportingIssueEventDispatcher +from adcp.reporting.fixtures import redacted_capabilities +from adcp.reporting.ledger import ( + InMemoryReportingLedgerStore, + ReportingDeliveryEscalation, + ReportingStatusCaller, + issue_event_id, + supports_issue_events, +) +from adcp.reporting.service import ( + ReliableReportingConfigurationError, + ReliableReportingService, +) +from adcp.reporting.testing import DeterministicReportingClock, ScriptedReportingAdapter +from tests.test_reliable_reporting_service import ( + NOW, + _account_context, + _configuration, + _rows, +) + +ACCOUNT = "account-redacted" +CONSUMER = "buyer-1" +ESCALATION = ReportingDeliveryEscalation( + consumer_mismatch_escalation=timedelta(hours=1), + operations_contact_email="reporting-ops@example.test", +) + + +# -- the service flow, shared with the PostgreSQL conformance test ----------- + + +@dataclass +class Observed: + """What the observer and the worker error handler saw.""" + + events: list[ReportingIssueEvent] = field(default_factory=list) + errors: list[tuple[str, BaseException]] = field(default_factory=list) + fail: set[str] = field(default_factory=set) + + async def observe(self, event: ReportingIssueEvent) -> None: + if event.kind in self.fail: + self.fail.discard(event.kind) + raise RuntimeError(f"observer rejected {event.kind}") + self.events.append(event) + + def report(self, component: str, error: BaseException) -> None: + self.errors.append((component, error)) + + def kinds(self) -> list[str]: + return [event.kind for event in self.events] + + +def service_options(clock: DeterministicReportingClock, observed: Observed) -> dict[str, Any]: + caller = ReportingStatusCaller(account_id=ACCOUNT, consumer_id=CONSUMER) + return { + "account_context": _account_context, + "caller_resolver": lambda _request, _context: caller, + "clock": clock, + "consumer_status_enabled": True, + "escalation": ESCALATION, + "issue_observer": observed.observe, + "worker_error_handler": observed.report, + } + + +async def published_period(service: ReliableReportingService) -> tuple[dict[str, Any], dict]: + service.sources.register("gam", ScriptedReportingAdapter(redacted_capabilities(), [_rows(17)])) + await service.configure(_configuration()) + await service.run_worker(now=NOW) + status = await service.get_reporting_status({"view": "periods"}) + return status["periods"][0], status["revisions"][0] + + +def unreadable(period: dict[str, Any], revision: dict[str, Any], at: Any) -> dict[str, Any]: + """The buyer cannot read a revision the seller serves: the provider must act.""" + return { + "reporting_status_id": "status-unreadable-0001", + "delivery_config_id": period["delivery_config_id"], + "delivery_config_version": period["delivery_config_version"], + "report_definition_id": period["report_definition_id"], + "period": period["period"], + "consumer_status": "unreadable", + "failure_code": "transport_failed", + "reporting_obligation_id": period["reporting_obligation_id"], + "reporting_revision_id": revision["reporting_revision_id"], + "status_as_of": at.isoformat(), + } + + +def agreeing(statement: dict[str, Any], revision: dict[str, Any], at: Any) -> dict[str, Any]: + agreed = { + **statement, + "reporting_status_id": "status-received-0001", + "supersedes_reporting_status_id": statement["reporting_status_id"], + "consumer_status": "received", + "observed_revision_content_sha256": revision["revision_content_sha256"], + "status_as_of": at.isoformat(), + } + del agreed["failure_code"] + return agreed + + +async def run_lifecycle_flow( + service: ReliableReportingService, clock: DeterministicReportingClock, observed: Observed +) -> None: + """Open on ingest, escalate on time, resolve on agreement: one event each.""" + period, revision = await published_period(service) + clock.advance(timedelta(minutes=1)) + opened_at = clock() + statement = unreadable(period, revision, opened_at) + recorded = await service.sync_reporting_status({"statuses": [statement]}) + assert recorded["results"][0]["result"] == "recorded" + assert observed.events == [] # delivered by the worker turn, after commit + + turn = await service.run_worker() + assert turn.issue_events is not None and turn.issue_events.delivered == 1 + (opened,) = observed.events + assert opened.kind == "opened" and opened.occurred_at == opened_at + assert opened.event_id == issue_event_id(opened.issue_id, "opened") + assert (opened.account_id, opened.consumer_id, opened.attempt) == (ACCOUNT, CONSUMER, 1) + assert opened.reporting_status_id == statement["reporting_status_id"] + issue = opened.issue + assert issue is not None and issue.issue_id == opened.issue_id + assert issue.code == "CONSUMER_STATUS_MISMATCH" + assert (issue.responsible_party, issue.recommended_action) == ("provider", "contact_provider") + assert (issue.severity, issue.issue_state, issue.opened_at) == ( + "action_required", + "open", + opened_at, + ) + assert issue.reporting_obligation_id == period["reporting_obligation_id"] + + # An exact retry, a status read and an idle turn change nothing. + replay = await service.sync_reporting_status({"statuses": [statement]}) + assert replay["results"][0]["result"] == "unchanged" + status = await service.get_reporting_status({"view": "summary"}) + assert [item["issue_id"] for item in status["issues"]] == [opened.issue_id] + assert not (await service.run_worker()).issue_events.did_work + assert observed.kinds() == ["opened"] + + # Escalation fires when the boundary passes, with nobody reading status. + clock.set(opened_at + ESCALATION.consumer_mismatch_escalation - timedelta(seconds=1)) + assert not (await service.run_worker()).issue_events.did_work + clock.set(opened_at + ESCALATION.consumer_mismatch_escalation) + turn = await service.run_worker() + assert turn.issue_events.escalations_recorded == 1 and turn.issue_events.delivered == 1 + escalated = observed.events[-1] + assert escalated.kind == "escalated" and escalated.issue_id == opened.issue_id + assert escalated.occurred_at == opened_at + ESCALATION.consumer_mismatch_escalation + assert escalated.issue is not None + assert escalated.issue.recommended_action == "contact_provider" + assert escalated.issue.severity == "action_required" + assert not (await service.run_worker()).issue_events.did_work # once + + # The buyer files an agreeing statement: resolved. + clock.advance(timedelta(minutes=1)) + agreed = await service.sync_reporting_status( + {"statuses": [agreeing(statement, revision, clock())]} + ) + assert agreed["results"][0]["result"] == "recorded" + await service.run_worker() + resolved = observed.events[-1] + assert (resolved.kind, resolved.issue_id, resolved.issue) == ("resolved", opened.issue_id, None) + assert resolved.occurred_at == clock() + assert observed.kinds() == ["opened", "escalated", "resolved"] + assert len({event.event_id for event in observed.events}) == 3 + assert observed.errors == [] + + +async def run_failing_observer_flow( + service: ReliableReportingService, clock: DeterministicReportingClock, observed: Observed +) -> None: + """A raising observer fails no call, is reported, and is retried with the same id.""" + period, revision = await published_period(service) + clock.advance(timedelta(minutes=1)) + observed.fail.add("opened") + statement = unreadable(period, revision, clock()) + recorded = await service.sync_reporting_status({"statuses": [statement]}) + assert recorded["results"][0]["result"] == "recorded" + turn = await service.run_worker() + assert turn.issue_events is not None + assert (turn.issue_events.delivered, turn.issue_events.failed) == (0, 1) + assert turn.extension_errors == {} # the turn itself succeeded + ((component, error),) = observed.errors + assert component.startswith("issue_observer:rpie_") + assert str(error) == "observer rejected opened" + # Not due again until the backoff passes; then delivered once. + assert not (await service.run_worker()).issue_events.did_work + clock.advance(timedelta(seconds=30)) + await service.run_worker() + (opened,) = observed.events + assert opened.kind == "opened" and opened.attempt == 2 + assert component == f"issue_observer:{opened.event_id}" + assert not (await service.run_worker()).issue_events.did_work + + +# -- the memory reference ----------------------------------------------------- + + +async def test_the_lifecycle_reaches_the_observer_once_per_transition() -> None: + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = ReliableReportingService.memory(**service_options(clock, observed)) + await run_lifecycle_flow(service, clock, observed) + + +async def test_a_failing_observer_is_reported_and_retried_with_the_same_event_id() -> None: + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = ReliableReportingService.memory(**service_options(clock, observed)) + await run_failing_observer_flow(service, clock, observed) + + +def test_issue_event_ids_are_stable_and_match_the_storage_distribution() -> None: + assert issue_event_id("rpti_x", "opened") == issue_event_id("rpti_x", "opened") + assert issue_event_id("rpti_x", "opened") != issue_event_id("rpti_x", "resolved") + assert issue_event_id("rpti_x", "opened").startswith("rpie_") + storage = pytest.importorskip("adcp_reporting_storage.ledger.issue_events") + for kind in ("opened", "escalated", "acknowledged", "waived", "resolved"): + assert storage.issue_event_id("rpti_x", kind) == issue_event_id("rpti_x", kind) + + +async def _events(store: InMemoryReportingLedgerStore) -> list[tuple[str, str, Any]]: + return [ + (event.issue_id, event.kind, event.occurred_at) + for event in sorted(store._issue_events.values(), key=lambda e: e.seq) + ] + + +async def _open(store: InMemoryReportingLedgerStore, clock: Any, key: str = "rpik_condition"): + return await store.ensure_issue_opened( + issue_key=key, account_id=ACCOUNT, consumer_id=CONSUMER, observed_at=clock() + ) + + +async def test_the_memory_store_records_each_lifecycle_write_once() -> None: + clock = DeterministicReportingClock(NOW) + store = InMemoryReportingLedgerStore(clock=clock) + assert supports_issue_events(store) and await store.issue_events_ready() + first = await _open(store, clock) + await _open(store, clock) # idempotent + acknowledged_at = clock.advance(timedelta(minutes=1)) + for ref in (None, None, "TICKET-1"): # re-acknowledging and attaching a ref record nothing + await store.set_issue_state( + issue_key=first.issue_key, + account_id=ACCOUNT, + state="acknowledged", + at=acknowledged_at, + external_ref=ref, + ) + resolved_at = clock.advance(timedelta(minutes=1)) + await store.retire_issue(issue_key=first.issue_key, account_id=ACCOUNT, at=resolved_at) + second = await _open(store, clock) + waived_at = clock.advance(timedelta(minutes=1)) + await store.set_issue_state( + issue_key=second.issue_key, account_id=ACCOUNT, state="waived", at=waived_at + ) + assert await _events(store) == [ + (first.issue_id, "opened", NOW), + (first.issue_id, "acknowledged", acknowledged_at), + (first.issue_id, "resolved", resolved_at), + (second.issue_id, "opened", resolved_at), + (second.issue_id, "waived", waived_at), + ] + + +async def test_a_rolled_back_lifecycle_write_records_no_event() -> None: + clock = DeterministicReportingClock(NOW) + store = InMemoryReportingLedgerStore(clock=clock) + with pytest.raises(RuntimeError, match="abort"): + async with store.transaction(): + await _open(store, clock) + raise RuntimeError("abort") + assert await _events(store) == [] + + +async def test_memory_claims_lease_one_event_per_occurrence_in_order() -> None: + clock = DeterministicReportingClock(NOW) + store = InMemoryReportingLedgerStore(clock=clock) + a = await _open(store, clock, "rpik_a") + await store.set_issue_state( + issue_key=a.issue_key, account_id=ACCOUNT, state="acknowledged", at=clock() + ) + b = await _open(store, clock, "rpik_b") + first = await store.claim_issue_events(lease_seconds=30, limit=10) + assert [(lease.issue_id, lease.kind) for lease in first] == [ + (a.issue_id, "opened"), + (b.issue_id, "opened"), + ] + assert await store.claim_issue_events(lease_seconds=30) == () # leased + assert await store.complete_issue_event(first[0]) + assert not await store.complete_issue_event(first[0]) + (acknowledged,) = await store.claim_issue_events(lease_seconds=30) + assert (acknowledged.issue_id, acknowledged.kind) == (a.issue_id, "acknowledged") + assert await store.release_issue_event(acknowledged, retry_after=timedelta(seconds=10)) + assert await store.claim_issue_events(lease_seconds=30) == () + clock.advance(timedelta(seconds=10)) + (again,) = await store.claim_issue_events(lease_seconds=30) + assert (again.event_id, again.attempts) == (acknowledged.event_id, 2) + clock.advance(timedelta(seconds=31)) # both leases lapse + reclaimed = await store.claim_issue_events(lease_seconds=30) + assert {lease.event_id for lease in reclaimed} == {first[1].event_id, again.event_id} + assert not await store.complete_issue_event(first[1]) # the stale lease lost it + for lease in reclaimed: + assert await store.complete_issue_event(lease) + clock.advance(timedelta(days=8)) + assert await store.prune_issue_events(older_than=timedelta(days=7)) == 3 + assert await _events(store) == [] + + +async def test_escalation_is_recorded_once_and_never_for_a_retired_occurrence() -> None: + clock = DeterministicReportingClock(NOW) + store = InMemoryReportingLedgerStore(clock=clock) + opened = await _open(store, clock, "rpik_open") + resolved = await _open(store, clock, "rpik_resolved") + await store.retire_issue(issue_key=resolved.issue_key, account_id=ACCOUNT, at=clock()) + waived = await _open(store, clock, "rpik_waived") + await store.set_issue_state( + issue_key=waived.issue_key, account_id=ACCOUNT, state="waived", at=clock() + ) + clock.set(NOW + timedelta(minutes=59, seconds=59)) + assert await store.sweep_issue_escalations(after=timedelta(hours=1)) == 0 + clock.set(NOW + timedelta(hours=1)) + results = await asyncio.gather( + *(store.sweep_issue_escalations(after=timedelta(hours=1)) for _ in range(3)) + ) + assert sum(results) == 1 + escalated = [event for event in await _events(store) if event[1] == "escalated"] + assert escalated == [(opened.issue_id, "escalated", NOW + timedelta(hours=1))] + + +async def test_an_occurrence_waits_for_its_earlier_event_to_be_delivered() -> None: + clock = DeterministicReportingClock(NOW) + observed = Observed(fail={"opened"}) + store = InMemoryReportingLedgerStore(clock=clock) + dispatcher = ReportingIssueEventDispatcher( + store, observed.observe, escalation=ESCALATION, on_error=_collect(observed) + ) + await _open(store, clock) + clock.set(NOW + timedelta(hours=1)) + first = await dispatcher.run_once() + assert (first.escalations_recorded, first.delivered, first.failed) == (1, 0, 1) + assert observed.events == [] # 'escalated' waits behind the failed 'opened' + clock.advance(timedelta(seconds=30)) + second = await dispatcher.run_once() + assert second.delivered == 2 + assert observed.kinds() == ["opened", "escalated"] + + +def _collect(observed: Observed): + async def report(component: str, error: BaseException) -> None: + observed.report(component, error) + + return report + + +async def test_a_hung_observer_times_out_and_a_sync_observer_works() -> None: + clock = DeterministicReportingClock(NOW) + store = InMemoryReportingLedgerStore(clock=clock) + await _open(store, clock) + observed = Observed() + + async def hang(event: ReportingIssueEvent) -> None: + await asyncio.sleep(60) + + hung = ReportingIssueEventDispatcher( + store, hang, observer_timeout=timedelta(milliseconds=20), on_error=_collect(observed) + ) + assert (await hung.run_once()).failed == 1 + ((component, error),) = observed.errors + assert component.startswith("issue_observer:") and isinstance(error, asyncio.TimeoutError) + seen: list[ReportingIssueEvent] = [] + clock.advance(timedelta(seconds=30)) + assert (await ReportingIssueEventDispatcher(store, seen.append).run_once()).delivered == 1 + assert [event.kind for event in seen] == ["opened"] + assert seen[0].attempt == 2 + + +async def test_operator_transitions_render_in_their_state() -> None: + """Acknowledged and waived come from store.set_issue_state, outside any request.""" + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = ReliableReportingService.memory(**service_options(clock, observed)) + period, revision = await published_period(service) + clock.advance(timedelta(minutes=1)) + await service.sync_reporting_status({"statuses": [unreadable(period, revision, clock())]}) + await service.run_worker() + (opened,) = observed.events + key = next(iter(service.store._issues.values())).issue_key + clock.advance(timedelta(minutes=5)) + await service.store.set_issue_state( + issue_key=key, account_id=ACCOUNT, state="acknowledged", at=clock() + ) + clock.advance(timedelta(minutes=5)) + await service.store.set_issue_state( + issue_key=key, account_id=ACCOUNT, state="waived", at=clock() + ) + await service.run_worker() + assert observed.kinds() == ["opened", "acknowledged", "waived"] + acknowledged, waived = observed.events[1:] + assert acknowledged.issue is not None and acknowledged.issue.issue_state == "acknowledged" + assert waived.issue is not None and waived.issue.issue_state == "waived" + assert waived.issue.responsible_party == "provider" + assert {event.issue_id for event in observed.events} == {opened.issue_id} + + +async def test_a_condition_that_cleared_unobserved_resolves_instead_of_escalating() -> None: + clock = DeterministicReportingClock(NOW) + observed = Observed() + service = ReliableReportingService.memory(**service_options(clock, observed)) + period, revision = await published_period(service) + clock.advance(timedelta(minutes=1)) + await service.sync_reporting_status({"statuses": [unreadable(period, revision, clock())]}) + # The seller now agrees the revision is unreadable; nobody reads status. + await service.store.set_revision_readable( + account_id=ACCOUNT, reporting_revision_id=revision["reporting_revision_id"], readable=False + ) + clock.advance(ESCALATION.consumer_mismatch_escalation) + turn = await service.run_worker() + assert turn.issue_events is not None and turn.issue_events.escalations_recorded == 0 + assert observed.kinds() == ["opened", "resolved"] + + +def test_the_service_refuses_production_and_stores_without_issue_events() -> None: + with pytest.raises(ReliableReportingConfigurationError, match="#1873"): + ReliableReportingService( + store=InMemoryReportingLedgerStore(), + account_context=_account_context, + production=SimpleNamespace(), + issue_observer=lambda event: None, + ) + + class LegacyStore(InMemoryReportingLedgerStore): + issue_events_supported = False + + with pytest.raises(ReliableReportingConfigurationError, match="0.2.0b4"): + ReliableReportingService( + store=LegacyStore(), + account_context=_account_context, + issue_observer=lambda event: None, + ) + with pytest.raises(ReliableReportingConfigurationError, match="callable"): + ReliableReportingService.memory( + account_context=_account_context, issue_observer="not callable" + ) + + +async def test_the_service_refuses_to_start_before_the_event_table_exists() -> None: + class Unmigrated(InMemoryReportingLedgerStore): + async def issue_events_ready(self) -> bool: + return False + + service = ReliableReportingService( + store=Unmigrated(), account_context=_account_context, issue_observer=lambda event: None + ) + with pytest.raises(ReliableReportingConfigurationError, match="python -m adcp migrate"): + await service.initialize() diff --git a/tests/test_reporting_storage_facade.py b/tests/test_reporting_storage_facade.py index 052d14cc91..486943ea40 100644 --- a/tests/test_reporting_storage_facade.py +++ b/tests/test_reporting_storage_facade.py @@ -124,8 +124,9 @@ def test_migrations_are_numbered_files_with_their_upgrade_notes() -> None: "0005", "0006", "0007", + "0008", ] - widened, offerings, contract = shipped[1], shipped[5], shipped[6] + widened, offerings, contract, events = shipped[1], shipped[5], shipped[6], shipped[7] assert widened.name == "reporting_subscription_generations" assert widened.includes == ("ledger/reporting_ledger_account_generations.sql",) assert offerings.name == "reporting_status_checkpoint_offerings" @@ -145,6 +146,10 @@ def test_migrations_are_numbered_files_with_their_upgrade_notes() -> None: ) assert "python -m adcp migrate --contract" in contract.description assert [item.version for item in shipped if item.contract] == ["0007"] + # The issue-event outbox (#1258) is additive: applied even while 0007 waits. + assert events.name == "reporting_issue_events" and events.additive and not events.contract + assert events.includes == ("ledger/reporting_issue_events.sql",) + assert [item.version for item in shipped if item.additive] == ["0008"] for item in shipped: assert item.statements(), item.version # every include resolves lines = describe([MigrationStatus(item, None) for item in shipped]) @@ -162,6 +167,30 @@ def test_an_unknown_upgrade_header_is_refused() -> None: with pytest.raises(ValueError, match="unknown upgrade"): _parse("0009_x.sql", "-- adcp-migration: 0009 x\n-- adcp-upgrade: contracted\n") assert _parse("0009_x.sql", "-- adcp-migration: 0009 x\n-- adcp-upgrade: contract\n").contract + assert _parse("0009_x.sql", "-- adcp-migration: 0009 x\n-- adcp-upgrade: additive\n").additive + + +def test_a_held_contract_step_lets_only_the_additive_migrations_after_it_through() -> None: + pytest.importorskip("adcp_reporting_storage") + from adcp_reporting_storage.migrations import Migration, _past_held_contract + + def step(version: str, *, contract: bool = False, additive: bool = False) -> Migration: + return Migration(version, f"m{version}", False, (), "", "", contract, additive) + + pending = [ + step("0006"), + step("0007", contract=True), + step("0008", additive=True), + step("0009"), + step("0010", additive=True), + ] + # 0009 may depend on the contract, so the walk stops there; 0010 waits behind it. + assert [item.version for item in _past_held_contract(pending)] == ["0006", "0008"] + assert [item.version for item in _past_held_contract(pending[:2])] == ["0006"] + # A second contract step is not additive either: nothing after it is planned, + # exactly as migrate() stops there. + second = [*pending[:3], step("0009", contract=True), step("0010", additive=True)] + assert [item.version for item in _past_held_contract(second)] == ["0006", "0008"] def test_only_the_uncontracted_check_is_the_contract_refusal() -> None: diff --git a/tests/type_checks/reporting_issue_observer.py b/tests/type_checks/reporting_issue_observer.py new file mode 100644 index 0000000000..b8f7cb29bf --- /dev/null +++ b/tests/type_checks/reporting_issue_observer.py @@ -0,0 +1,42 @@ +"""The reporting-issue observer as a strictly typed adopter writes it (#1258).""" + +from datetime import datetime + +from adcp.reporting import ( + ReliableReportingService, + ReportingIssueEvent, + ReportingIssueEventKind, + ReportingIssueObserver, +) +from adcp.reporting.ledger import ReportingIssue, ReportingLedgerStore +from adcp.reporting.service import ReportingContextResolver + +seen: set[str] = set() + + +async def on_issue(event: ReportingIssueEvent) -> None: + # Delivery is at least once: deduplicate on the stable event id. + if event.event_id in seen: + return + seen.add(event.event_id) + kind: ReportingIssueEventKind = event.kind + occurred_at: datetime = event.occurred_at + attempt: int = event.attempt + issue: ReportingIssue | None = event.issue + status_id: str | None = event.reporting_status_id + assert occurred_at.tzinfo is not None and attempt >= 1 and kind + if issue is not None and issue.responsible_party != "buyer": + assert issue.recommended_action and issue.issue_id == event.issue_id + assert status_id is None or status_id + + +def log_issue(event: ReportingIssueEvent) -> None: + """A synchronous observer is accepted too.""" + print(event.kind, event.issue_id, event.account_id, event.consumer_id) + + +def compose( + store: ReportingLedgerStore, context: ReportingContextResolver, sync: bool +) -> ReliableReportingService: + observer: ReportingIssueObserver = log_issue if sync else on_issue + return ReliableReportingService(store=store, account_context=context, issue_observer=observer)