From 84f0e5645b33bb7e68f446edf4971dcb6c74ba9a Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 8 Nov 2026 14:13:24 +0000 Subject: [PATCH] feat(reporting-storage): record issue-lifecycle transitions in a transactional outbox Migration 0008 (header "-- adcp-upgrade: additive") adds reporting_issue_events and its indexes: one new table, nothing else, and the storage contract stays 2. Every write that moves a reporting_issue_lifecycle row (ensure opens, set_issue_state acknowledges or waives, the projection retires as resolved, replay persistence) records its transition there in the same transaction, so an event exists exactly when its transition committed. UNIQUE (issue_id, kind) keeps the record exactly once across replicas, and event_id is derived from (issue_id, kind), so every redelivery of a transition carries the same id. PgReportingLedgerStore gains the outbox the issue observer reads (adcp-client-python#1258): issue_events_supported and issue_events_ready(), sweep_issue_escalations(after=...) (settles each candidate account under its lock, then records 'escalated' for occurrences still open or acknowledged past opened_at + after, ON CONFLICT DO NOTHING), claim_issue_events (SKIP LOCKED leases, at most one event per occurrence, each occurrence in recording order), complete_issue_event, release_issue_event(retry_after=...) and prune_issue_events(older_than=...), both measured on the store's clock. ReportingIssueEventLease is exported. Until 0008 is applied, lifecycle writes record nothing and issue_events_ready() is False; nothing else changes. python -m adcp migrate now applies an additive migration even while a contract step waits, so 0008 lands on an expanded ledger beside adcp-reporting-storage 0.1.x/0.2.0b0 processes, which neither read nor write the table. The published-wheel rolling test runs them beside it. Co-Authored-By: Claude Opus 5.5 --- docs/reporting-ledger-migration.md | 33 +- packages/adcp-reporting-storage/README.md | 5 + .../src/adcp_reporting_storage/__init__.py | 5 + .../ledger/issue_events.py | 362 ++++++++++++++++++ .../src/adcp_reporting_storage/ledger/pg.py | 95 +++++ .../ledger/reporting_issue_events.sql | 57 +++ .../ledger/status_snapshot.py | 74 ++-- .../0008_reporting_issue_events.sql | 10 + .../migrations/__init__.py | 52 ++- .../test_reporting_issue_events_pg.py | 356 +++++++++++++++++ .../test_reporting_storage_migrations_pg.py | 11 +- ..._reporting_storage_published_upgrade_pg.py | 14 +- tests/fixtures/public_api_snapshot.json | 1 + tests/test_reporting_storage_facade.py | 31 +- 14 files changed, 1064 insertions(+), 42 deletions(-) create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/issue_events.py create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_issue_events.sql create mode 100644 packages/adcp-reporting-storage/src/adcp_reporting_storage/migrations/0008_reporting_issue_events.sql create mode 100644 tests/conformance/reporting/test_reporting_issue_events_pg.py 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/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_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..4e7451aa21 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -9034,6 +9034,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_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: