From 6d0214290bccf276727869340daa143ada491324 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Thu, 29 Oct 2026 19:48:24 +0000 Subject: [PATCH] feat(reporting): persist reporting-subscription engine state through a CAS store The reporting-subscription engine kept its state in a process-local bag, so a restart or a second worker lost grants, losses, capacity slots, exceptions and version snapshots, and with them what the seller owes. ReportingSubscriptionStateStore persists one JSON document per principal, written by compare-and-swap. The engine loads the document, applies a change and writes it back; on a conflict it reloads and reapplies the change, and it raises ReportingSubscriptionStateConflictError when the document keeps changing. An unchanged document is not rewritten. InMemoryReportingSubscriptionStore implements the protocol for tests and single workers. encode_principal_state and decode_principal_state define the versioned document; version configurations and destination snapshots are never pruned, and exceptions are dropped status_retention_days after they close. A PostgreSQL implementation follows in adcp-reporting-storage. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/reliable-reporting-service.md | 12 + src/adcp/reporting/subscriptions/__init__.py | 10 + src/adcp/reporting/subscriptions/engine.py | 500 ++++++++++++++---- src/adcp/server/__init__.py | 4 + src/adcp/server/principal.py | 55 ++ ...test_reporting_subscription_state_store.py | 238 +++++++++ 6 files changed, 724 insertions(+), 95 deletions(-) create mode 100644 tests/test_reporting_subscription_state_store.py diff --git a/docs/reliable-reporting-service.md b/docs/reliable-reporting-service.md index 68007ec8f..17da23fa6 100644 --- a/docs/reliable-reporting-service.md +++ b/docs/reliable-reporting-service.md @@ -400,6 +400,18 @@ the buyer runs. Under `max_accounts`, accounts already served keep their slots and the rest are filled in `account_id` byte order; the remainder get a `capacity` exception. +**Engine state.** The engine keeps each principal's state (grants and losses, +timezone changes, every version's configuration and destination snapshot, +`max_accounts` slots, exceptions and withheld counts) as one JSON document in a +`ReportingSubscriptionStateStore`, written by compare-and-swap. On a conflicting +concurrent write it reloads and reapplies the change, and it raises +`ReportingSubscriptionStateConflictError` if the document keeps changing. The +default `InMemoryReportingSubscriptionStore` is lost on restart; pass a durable +`store=` in production, and share it, with the `PrincipalService` record store, +across every worker. Version configurations are never pruned, so +`prior_versions` stay answerable while obligations cite them; exceptions are +dropped `status_retention_days` after they close. + `subscription_capabilities(...)` and `advertise_reporting_subscriptions(...)` project `configuration_tasks`, the deprecated `configuration_task`, the `subscriptions` block, the feature id and the principal section. The engine diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 4cfa39e13..b76c8b7b1 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -66,6 +66,11 @@ ReportingSubscriptionEngine, ReportingSubscriptionHost, ReportingSubscriptionPrecedenceHost, + ReportingSubscriptionStateConflictError, + ReportingSubscriptionStateStore, + StoredReportingSubscriptionState, + decode_principal_state, + encode_principal_state, ) from adcp.reporting.subscriptions.registry import ( ReportingSubscriptionError, @@ -108,6 +113,11 @@ "ReportingSubscriptionError", "ReportingSubscriptionHost", "ReportingSubscriptionPrecedenceHost", + "ReportingSubscriptionStateConflictError", + "ReportingSubscriptionStateStore", + "StoredReportingSubscriptionState", + "decode_principal_state", + "encode_principal_state", "apply_reporting_subscriptions", "refresh_destination_states", "render_path_template", diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index cf1b9232e..79d179fba 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -35,11 +35,12 @@ from __future__ import annotations import asyncio +import json import logging from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field, replace from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any, Protocol +from typing import TYPE_CHECKING, Any, Protocol, TypeVar from adcp.reporting.subscriptions._clock import PeriodCalendar, iso_z, require_zone, utc_instant from adcp.reporting.subscriptions.capabilities import ReportingSubscriptionSupport @@ -68,6 +69,11 @@ logger = logging.getLogger(__name__) +_T = TypeVar("_T") + +#: Attempts at a compare-and-swap write before giving up under contention. +_MAX_STATE_WRITE_ATTEMPTS = 8 + if TYPE_CHECKING: from adcp.server.principal import ( DeliveryConfigIdsHook, @@ -238,15 +244,266 @@ class _PrincipalState: withheld: dict[tuple[str, str, datetime], int] = field(default_factory=dict) +#: Version of the persisted engine-state document; bump on a breaking change. +STATE_DOCUMENT_VERSION = 1 + + +@dataclass(frozen=True) +class StoredReportingSubscriptionState: + """One principal's persisted engine state and its store revision.""" + + document: Mapping[str, Any] + revision: int + + +class ReportingSubscriptionStateStore(Protocol): + """Durable engine state: one JSON document per principal, written by CAS. + + The engine loads a principal's document, applies a change, and writes it + back with :meth:`compare_and_swap`; on a conflict it reloads and reapplies + the change, so every worker sharing the backing store sees one history. + The document holds grants and losses, timezone changes, version + configurations and destination snapshots, ``max_accounts`` slots, + exceptions and withheld-period counts. It is plain JSON; stores persist it + verbatim and never interpret it. + """ + + async def load(self, subject: str) -> StoredReportingSubscriptionState | None: + """Return the principal's document and revision, or ``None`` if none is stored.""" + ... + + async def compare_and_swap( + self, + subject: str, + expected_revision: int | None, + document: Mapping[str, Any], + ) -> bool: + """Store ``document`` if the current revision is ``expected_revision``. + + ``expected_revision=None`` inserts only when nothing is stored. A + successful write sets the revision to ``expected_revision + 1`` (``1`` + for an insert) and returns ``True``; any other current state returns + ``False`` and changes nothing. + """ + ... + + class InMemoryReportingSubscriptionStore: - """Process-local engine state. Durable storage is a later slice.""" + """Process-local :class:`ReportingSubscriptionStateStore` for tests and single workers. + + State is lost on restart. Production sellers use a durable store, such as + ``adcp_reporting_storage``'s PostgreSQL implementation. + """ def __init__(self) -> None: - self.principals: dict[str, _PrincipalState] = {} - self.lock = asyncio.Lock() + self._documents: dict[str, StoredReportingSubscriptionState] = {} + self._lock = asyncio.Lock() + + async def load(self, subject: str) -> StoredReportingSubscriptionState | None: + async with self._lock: + stored = self._documents.get(subject) + if stored is None: + return None + return StoredReportingSubscriptionState(_copy(stored.document), stored.revision) + + async def compare_and_swap( + self, + subject: str, + expected_revision: int | None, + document: Mapping[str, Any], + ) -> bool: + async with self._lock: + current = self._documents.get(subject) + if (current.revision if current is not None else None) != expected_revision: + return False + self._documents[subject] = StoredReportingSubscriptionState( + _copy(document), 1 if expected_revision is None else expected_revision + 1 + ) + return True + + +class ReportingSubscriptionStateConflictError(RuntimeError): + """The engine could not write its state after repeated concurrent changes.""" + + +def _copy(document: Mapping[str, Any]) -> dict[str, Any]: + copied = json.loads(json.dumps(document)) + assert isinstance(copied, dict) + return copied + + +# -- state document codec ------------------------------------------------------ - def principal(self, subject: str) -> _PrincipalState: - return self.principals.setdefault(subject, _PrincipalState()) + +def _instant(value: str | None) -> datetime | None: + return None if value is None else datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _required_instant(value: str) -> datetime: + instant = _instant(value) + assert instant is not None + return instant + + +def _iso(value: datetime | None) -> str | None: + return None if value is None else iso_z(value) + + +def _json_value(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json", exclude_none=True) + return json.loads(json.dumps(value)) + + +def encode_principal_state( + principal: _PrincipalState, *, prune_closed_before: datetime | None = None +) -> dict[str, Any]: + """Serialize engine state to a JSON document. + + Exceptions closed before ``prune_closed_before`` are dropped: the spec + lets a seller forget them ``status_retention_days`` after they close. + Version configurations and destination snapshots are never pruned, so + ``prior_versions`` stay answerable for as long as obligations cite them. + """ + exclusions: list[dict[str, Any]] = [] + for (subscription_id, account_id), entries in sorted(principal.exclusions.items()): + kept = [ + entry + for entry in entries + if prune_closed_before is None + or entry.until is None + or entry.until >= prune_closed_before + ] + exclusions.append( + { + "subscription_id": subscription_id, + "account_id": account_id, + "entries": [ + { + "reason": entry.reason, + "from": iso_z(entry.from_), + "until": _iso(entry.until), + "media_buy_id": entry.media_buy_id, + "package_ids": list(entry.package_ids), + "overridden_by": ( + None if entry.overridden_by is None else list(entry.overridden_by) + ), + "effective_at": _iso(entry.effective_at), + } + for entry in kept + ], + } + ) + return { + "document_version": STATE_DOCUMENT_VERSION, + "accounts": { + account_id: { + "timezone": account.timezone, + "operator": account.operator, + "grants": [ + { + "authorized_at": iso_z(grant.authorized_at), + "unauthorized_at": _iso(grant.unauthorized_at), + } + for grant in account.grants + ], + "timezone_changes": [ + { + "changed_at": iso_z(change.changed_at), + "timezone": change.timezone, + "offering_id": change.offering_id, + "previous_timezone": change.previous_timezone, + } + for change in account.timezone_changes + ], + } + for account_id, account in sorted(principal.accounts.items()) + }, + "versions": [ + {"subscription_id": sid, "version": version, "configuration": _json_value(config)} + for (sid, version), config in sorted(principal.versions.items()) + ], + "covered": {sid: sorted(ids) for sid, ids in sorted(principal.covered.items())}, + "served": {sid: sorted(ids) for sid, ids in sorted(principal.served.items())}, + "exclusions": exclusions, + "destinations": [ + {"subscription_id": sid, "version": version, "destination": _json_value(target)} + for (sid, version), target in sorted(principal.destinations.items()) + ], + "withheld": [ + { + "subscription_id": sid, + "account_id": account_id, + "unauthorized_at": iso_z(lost), + "count": count, + } + for (sid, account_id, lost), count in sorted(principal.withheld.items()) + ], + } + + +def decode_principal_state(document: Mapping[str, Any]) -> _PrincipalState: + """Rebuild engine state from :func:`encode_principal_state` output.""" + version = document.get("document_version") + if version != STATE_DOCUMENT_VERSION: + raise ValueError(f"unsupported reporting-subscription state document version {version!r}") + principal = _PrincipalState() + for account_id, raw in document["accounts"].items(): + principal.accounts[account_id] = _Account( + timezone=raw["timezone"], + operator=raw["operator"], + grants=[ + SubscriptionAuthorization( + _required_instant(grant["authorized_at"]), _instant(grant["unauthorized_at"]) + ) + for grant in raw["grants"] + ], + timezone_changes=[ + AccountTimezoneChange( + _required_instant(change["changed_at"]), + change["timezone"], + offering_id=change["offering_id"], + previous_timezone=change["previous_timezone"], + ) + for change in raw["timezone_changes"] + ], + ) + for item in document["versions"]: + principal.versions[(item["subscription_id"], int(item["version"]))] = dict( + item["configuration"] + ) + principal.covered = {sid: set(ids) for sid, ids in document["covered"].items()} + principal.served = {sid: set(ids) for sid, ids in document["served"].items()} + for item in document["exclusions"]: + principal.exclusions[(item["subscription_id"], item["account_id"])] = [ + _Exclusion( + reason=entry["reason"], + from_=_required_instant(entry["from"]), + until=_instant(entry["until"]), + media_buy_id=entry["media_buy_id"], + package_ids=tuple(entry["package_ids"]), + overridden_by=( + None + if entry["overridden_by"] is None + else (entry["overridden_by"][0], entry["overridden_by"][1]) + ), + effective_at=_instant(entry["effective_at"]), + ) + for entry in item["entries"] + ] + for item in document["destinations"]: + principal.destinations[(item["subscription_id"], int(item["version"]))] = item[ + "destination" + ] + for item in document["withheld"]: + principal.withheld[ + ( + item["subscription_id"], + item["account_id"], + _required_instant(item["unauthorized_at"]), + ) + ] = int(item["count"]) + return principal def _dump(value: Any) -> dict[str, Any]: @@ -272,7 +529,7 @@ def __init__( offerings: Sequence[SubscriptionOffering], support: ReportingSubscriptionSupport | None = None, status_retention_days: int = 400, - store: InMemoryReportingSubscriptionStore | None = None, + store: ReportingSubscriptionStateStore | None = None, clock: Callable[[], datetime] | None = None, ) -> None: ids = [item.offering_id for item in offerings] @@ -328,6 +585,41 @@ def principal_options( def _now(self, now: datetime | None) -> datetime: return utc_instant(now if now is not None else self._clock(), field="now") + async def _read(self, subject: str) -> _PrincipalState: + stored = await self._store.load(subject) + return _PrincipalState() if stored is None else decode_principal_state(stored.document) + + async def _mutate( + self, + subject: str, + change: Callable[[_PrincipalState], _T], + *, + now: datetime | None = None, + ) -> _T: + """Apply ``change`` to the stored state and write it back by CAS. + + ``change`` may run more than once: on a conflicting concurrent write + the state is reloaded and ``change`` reapplied. Nothing is written + when the document is unchanged. + """ + prune = None if now is None else now - self._retention + for _attempt in range(_MAX_STATE_WRITE_ATTEMPTS): + stored = await self._store.load(subject) + principal = ( + _PrincipalState() if stored is None else decode_principal_state(stored.document) + ) + result = change(principal) + document = encode_principal_state(principal, prune_closed_before=prune) + if stored is not None and document == stored.document: + return result + if await self._store.compare_and_swap( + subject, None if stored is None else stored.revision, document + ): + return result + raise ReportingSubscriptionStateConflictError( + f"reporting-subscription state for {subject!r} kept changing concurrently" + ) + # -- recording ------------------------------------------------------------- async def record_versions( @@ -341,10 +633,10 @@ async def record_versions( version's coverage. """ destinations = await self._destinations(identity) - async with self._store.lock: - principal = self._store.principal(identity.subject) - for state in states: - payload = _dump(state) + payloads = [_dump(state) for state in states] + + def change(principal: _PrincipalState) -> None: + for payload in payloads: key = (str(payload["subscription_id"]), int(payload["subscription_version"])) configuration = dict(payload["configuration"]) principal.versions.setdefault(key, configuration) @@ -352,7 +644,9 @@ async def record_versions( if destination is not None and key not in principal.destinations: target = destinations.get(destination["destination_id"]) if target is not None: - principal.destinations[key] = target + principal.destinations[key] = _json_value(target) + + await self._mutate(identity.subject, change) async def authorization_changed( self, @@ -366,14 +660,14 @@ async def authorization_changed( ) -> None: """Record a grant or a loss of authority at ``at``.""" instant = utc_instant(at, field="at") - async with self._store.lock: - principal = self._store.principal(identity.subject) + if timezone is not None: + require_zone(timezone, field="timezone") + + def change(principal: _PrincipalState) -> None: account = principal.accounts.get(account_id) if account is None: if not authorized: return - if timezone is not None: - require_zone(timezone, field="timezone") account = principal.accounts[account_id] = _Account( timezone=timezone, operator=operator ) @@ -382,6 +676,8 @@ async def authorization_changed( if operator is not None: account.operator = operator + await self._mutate(identity.subject, change, now=instant) + @staticmethod def _apply_grant(account: _Account, *, authorized: bool, at: datetime) -> None: if authorized and not account.authorized: @@ -418,8 +714,8 @@ async def reconcile(self, identity: PrincipalIdentity, *, now: datetime | None = """ instant = self._now(now) listed = {item.account_id: item for item in await self._list_accounts(identity)} - async with self._store.lock: - principal = self._store.principal(identity.subject) + + def change(principal: _PrincipalState) -> None: for account_id, item in listed.items(): account = principal.accounts.get(account_id) if account is None: @@ -443,16 +739,17 @@ async def reconcile(self, identity: PrincipalIdentity, *, now: datetime | None = for account_id, account in principal.accounts.items(): if account_id not in listed and account.authorized: self._apply_grant(account, authorized=False, at=instant) + + await self._mutate(identity.subject, change, now=instant) await self._evaluate(identity, instant) async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] destinations = await self._destinations(identity) precedence = getattr(self._host, "account_delivery_configs", None) - async with self._store.lock: - principal = self._store.principal(identity.subject) - accounts = dict(principal.accounts) - covered_ids = set().union(*principal.covered.values()) if principal.covered else set() + snapshot = await self._read(identity.subject) + accounts = dict(snapshot.accounts) + covered_ids = set().union(*snapshot.covered.values()) if snapshot.covered else set() buys: dict[str, Sequence[SubscriptionBuy]] = {} configs: dict[str, Sequence[AccountDeliveryConfig]] = {} for account_id, account in accounts.items(): @@ -461,13 +758,14 @@ async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: if account.authorized and precedence is not None: configs[account_id] = await precedence(identity, account_id) - updates: list[tuple[str, list[dict[str, Any]], list[dict[str, Any]]]] = [] - async with self._store.lock: - principal = self._store.principal(identity.subject) + def change( + principal: _PrincipalState, + ) -> list[tuple[str, list[dict[str, Any]], list[dict[str, Any]]]]: + evaluated: list[tuple[str, list[dict[str, Any]], list[dict[str, Any]]]] = [] for state in states: subscription_id = str(state["subscription_id"]) try: - updates.append( + evaluated.append( self._evaluate_subscription( principal, state, states, accounts, buys, configs, destinations, now ) @@ -481,6 +779,9 @@ async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: ) else: self._errors.pop((identity.subject, subscription_id), None) + return evaluated + + updates = await self._mutate(identity.subject, change, now=now) delivered = getattr(self._host, "is_period_delivered", None) for subscription_id, _with_exceptions, unauthorized in updates: for lost_entry in unauthorized: @@ -489,16 +790,22 @@ async def _evaluate(self, identity: PrincipalIdentity, now: datetime) -> None: continue lost = datetime.fromisoformat(lost_entry["unauthorized_at"].replace("Z", "+00:00")) cache_key = (subscription_id, lost_entry["account_id"], lost) - async with self._store.lock: - cached = self._store.principal(identity.subject).withheld.get(cache_key) + cached = (await self._read(identity.subject)).withheld.get(cache_key) if cached is None: # Delivery by a past instant never changes: ask once per loss. - cached = 0 + count = 0 for period in ended: if not await delivered(identity, lost_entry["account_id"], period, by=lost): - cached += 1 - async with self._store.lock: - self._store.principal(identity.subject).withheld[cache_key] = cached + count += 1 + + def record( + principal: _PrincipalState, + key: tuple[str, str, datetime] = cache_key, + count: int = count, + ) -> int: + return principal.withheld.setdefault(key, count) + + cached = await self._mutate(identity.subject, record) lost_entry["withheld_period_count"] = cached for subscription_id, with_exceptions, unauthorized in updates: await self._principals.update_subscription_status( @@ -861,45 +1168,44 @@ async def subscription_coverage( states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] destinations = await self._destinations(identity) buys = await self._host.account_buys(identity, account_id) - async with self._store.lock: - principal = self._store.principal(identity.subject) - account = principal.accounts.get(account_id) - if account is None or not account.authorized: - return [] - result = [] - for state in states: - configuration = state["configuration"] - if not _scope_includes(configuration, account_id, account): - continue - entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) - account_level = [e for e in entries if e.until is None and e.media_buy_id is None] - status = "active" - if any(e.reason == "action_required" for e in account_level): - status = "action_required" - elif account_level: - status = "excluded" - payload: dict[str, Any] = { - "subscription_id": state["subscription_id"], - "subscription_version": state["subscription_version"], - "status": status, - "excluded": [ - entry.wire() - for entry in entries - if entry.until is None or instant - entry.until <= self._retention - ], - } - gaps = self._gaps(state, account, buys, destinations, instant) - if gaps: - payload["gaps"] = gaps - cutovers = self._timezone_cutovers(state, account, buys, destinations, instant) - if cutovers: - payload["timezone_cutovers"] = cutovers - result.append( - ReportingSubscriptionCoverage.model_validate( - project_subscription_coverage(payload, adcp_version) - ) + principal = await self._read(identity.subject) + account = principal.accounts.get(account_id) + if account is None or not account.authorized: + return [] + result = [] + for state in states: + configuration = state["configuration"] + if not _scope_includes(configuration, account_id, account): + continue + entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) + account_level = [e for e in entries if e.until is None and e.media_buy_id is None] + status = "active" + if any(e.reason == "action_required" for e in account_level): + status = "action_required" + elif account_level: + status = "excluded" + payload: dict[str, Any] = { + "subscription_id": state["subscription_id"], + "subscription_version": state["subscription_version"], + "status": status, + "excluded": [ + entry.wire() + for entry in entries + if entry.until is None or instant - entry.until <= self._retention + ], + } + gaps = self._gaps(state, account, buys, destinations, instant) + if gaps: + payload["gaps"] = gaps + cutovers = self._timezone_cutovers(state, account, buys, destinations, instant) + if cutovers: + payload["timezone_cutovers"] = cutovers + result.append( + ReportingSubscriptionCoverage.model_validate( + project_subscription_coverage(payload, adcp_version) ) - return result + ) + return result def _gaps( self, @@ -1009,12 +1315,11 @@ async def owed_periods( raise KeyError(subscription_id) destinations = await self._destinations(identity) buys = await self._host.account_buys(identity, account_id) - async with self._store.lock: - principal = self._store.principal(identity.subject) - account = principal.accounts.get(account_id) - if account is None: - return [] - return self._owed(principal, state, account_id, account, buys, destinations, instant) + principal = await self._read(identity.subject) + account = principal.accounts.get(account_id) + if account is None: + return [] + return self._owed(principal, state, account_id, account, buys, destinations, instant) def _owed( self, @@ -1074,25 +1379,24 @@ async def receipt_roster(self, identity: PrincipalIdentity, account_id: str) -> account-level exception. """ states = [_dump(item) for item in await self._principals.reporting_subscriptions(identity)] - async with self._store.lock: - principal = self._store.principal(identity.subject) - account = principal.accounts.get(account_id) - if account is None or not account.authorized: - return [] - result = [] - for state in states: - configuration = state["configuration"] - if ( - not _active(state) - or configuration["reconciliation_mode"] != "consumer_receipt" - or not _scope_includes(configuration, account_id, account) - ): - continue - entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) - if any(e.until is None and e.media_buy_id is None for e in entries): - continue - result.append(str(state["subscription_id"])) - return sorted(result) + principal = await self._read(identity.subject) + account = principal.accounts.get(account_id) + if account is None or not account.authorized: + return [] + result = [] + for state in states: + configuration = state["configuration"] + if ( + not _active(state) + or configuration["reconciliation_mode"] != "consumer_receipt" + or not _scope_includes(configuration, account_id, account) + ): + continue + entries = principal.exclusions.get((str(state["subscription_id"]), account_id), []) + if any(e.until is None and e.media_buy_id is None for e in entries): + continue + result.append(str(state["subscription_id"])) + return sorted(result) # -- helpers ------------------------------------------------------------------- @@ -1193,4 +1497,10 @@ def _destination_for( "ReportingSubscriptionEngine", "ReportingSubscriptionHost", "ReportingSubscriptionPrecedenceHost", + "ReportingSubscriptionStateConflictError", + "ReportingSubscriptionStateStore", + "STATE_DOCUMENT_VERSION", + "StoredReportingSubscriptionState", + "decode_principal_state", + "encode_principal_state", ] diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index a545b31f1..a8fe597f5 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -155,6 +155,8 @@ async def get_products(params, context=None): PrincipalRecordStore, PrincipalService, ReportingSubscriptionOptions, + principal_record_from_document, + principal_record_to_document, ) from adcp.server.proposal import ProposalBuilder, ProposalNotSupported from adcp.server.responses import ( @@ -253,6 +255,8 @@ async def get_products(params, context=None): "PrincipalRecord", "PrincipalRecordStore", "PrincipalService", + "principal_record_from_document", + "principal_record_to_document", "ReportingSubscriptionOptions", # Proposal helpers "ProposalBuilder", diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index 03d69a70e..cc74bdf6f 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -188,6 +188,58 @@ async def compare_and_swap( return True +#: Version of the :func:`principal_record_to_document` format. +PRINCIPAL_RECORD_DOCUMENT_VERSION = 1 + + +def principal_record_to_document(record: PrincipalRecord) -> dict[str, Any]: + """Serialize a :class:`PrincipalRecord` to plain JSON for a durable store. + + ``store_revision`` is not part of the document: a durable store keeps it + alongside, as its compare-and-swap fence. + """ + return { + "document_version": PRINCIPAL_RECORD_DOCUMENT_VERSION, + "subject": record.subject, + "principal_id": record.principal_id, + "principal_kind": record.principal_kind, + "configuration_version": record.configuration_version, + "configuration": ( + None + if record.configuration is None + else record.configuration.model_dump(mode="json", exclude_none=True) + ), + "idempotency": { + key: {"digest": digest, "response": response.model_dump(mode="json", exclude_none=True)} + for key, (digest, response) in sorted(record.idempotency.items()) + }, + } + + +def principal_record_from_document( + document: Mapping[str, Any], *, store_revision: int +) -> PrincipalRecord: + """Rebuild a :class:`PrincipalRecord` from :func:`principal_record_to_document` output.""" + version = document.get("document_version") + if version != PRINCIPAL_RECORD_DOCUMENT_VERSION: + raise ValueError(f"unsupported principal record document version {version!r}") + configuration = document.get("configuration") + return PrincipalRecord( + subject=document["subject"], + principal_id=document["principal_id"], + principal_kind=document["principal_kind"], + configuration_version=document.get("configuration_version"), + configuration=( + None if configuration is None else PrincipalState.model_validate(configuration) + ), + idempotency={ + key: (entry["digest"], SyncPrincipalResponse.model_validate(entry["response"])) + for key, entry in document.get("idempotency", {}).items() + }, + store_revision=store_revision, + ) + + def _copy_record(record: PrincipalRecord) -> PrincipalRecord: return PrincipalRecord( subject=record.subject, @@ -1157,7 +1209,10 @@ def _current_response( "PrincipalIdentity", "PrincipalRecord", "PrincipalRecordStore", + "PRINCIPAL_RECORD_DOCUMENT_VERSION", "PrincipalService", "ReportingSubscriptionOptions", "SubscriptionVersionsHook", + "principal_record_from_document", + "principal_record_to_document", ] diff --git a/tests/test_reporting_subscription_state_store.py b/tests/test_reporting_subscription_state_store.py new file mode 100644 index 000000000..3f97e6c01 --- /dev/null +++ b/tests/test_reporting_subscription_state_store.py @@ -0,0 +1,238 @@ +"""Durable reporting-subscription engine state (#1695 item 6). + +The engine keeps one JSON document per principal in a +``ReportingSubscriptionStateStore`` and writes it by compare-and-swap, so a +restarted engine, or another worker on the same store, owes exactly the same +periods. +""" + +from __future__ import annotations + +import json +from datetime import timedelta + +import pytest + +from adcp.reporting.subscriptions import ( + InMemoryReportingSubscriptionStore, + ReportingSubscriptionEngine, + ReportingSubscriptionStateConflictError, + ReportingSubscriptionSupport, + decode_principal_state, + encode_principal_state, +) +from tests.test_reporting_subscription_engine import ( + ALL_IDS, + DEST_A, + IDENTITY, + MANAGED, + OFFERINGS, + World, + _sync_raw, + at, + buy, + subscription, + table, +) + +CAPACITY = ReportingSubscriptionSupport(max_accounts=1) + + +async def _scenario(world: World) -> None: + """Grants, a loss and a re-grant, a capacity exception and a destination snapshot.""" + world.engine._offerings = (*world.engine._offerings, MANAGED) + world.grant("acct-1", "2026-09-01T00:00:00Z") + world.grant("acct-2", "2026-09-01T00:00:00Z") + for account_id in ("acct-1", "acct-2"): + world.host.buys[account_id] = [ + buy( + "mb-" + account_id, + "2026-10-01T04:00:00Z", + "2026-10-12T04:00:00Z", + offering_ids=(*ALL_IDS, MANAGED.offering_id), + ) + ] + await _sync_raw( + world, + { + "reporting_destinations": [DEST_A], + "reporting_subscriptions": [ + subscription(), + subscription( + "files", + scope={"account_ids": ["acct-1", "acct-2"]}, + destination={"destination_id": "dest-a", "path_template": "x/{account_id}"}, + ), + ], + }, + ) + await world.reconcile("2026-10-01T00:00:00Z") + world.revoke("acct-1", "2026-10-04T08:00:00Z") + await world.reconcile("2026-10-04T08:00:00Z") + world.grant("acct-1", "2026-10-06T15:00:00Z") + await world.reconcile("2026-10-06T15:00:00Z") + + +async def _observed(world: World, engine: ReportingSubscriptionEngine) -> dict: + owed = {} + for subscription_id in ("analytics", "files"): + for account_id in ("acct-1", "acct-2"): + periods = await engine.owed_periods( + IDENTITY, subscription_id, account_id, now=at("2026-10-10T00:00:00Z") + ) + owed[(subscription_id, account_id)] = table(periods) + coverage = { + account_id: [ + item.model_dump(mode="json") + for item in await engine.subscription_coverage( + IDENTITY, account_id, adcp_version="3.3-beta.2" + ) + ] + for account_id in ("acct-1", "acct-2") + } + return {"owed": owed, "coverage": coverage} + + +@pytest.mark.asyncio +async def test_document_round_trips_and_is_plain_json() -> None: + world = World(support=CAPACITY) + await _scenario(world) + stored = await world.engine._store.load(IDENTITY.subject) + assert stored is not None + document = json.loads(json.dumps(stored.document)) + assert encode_principal_state(decode_principal_state(document)) == document + assert document["accounts"]["acct-1"]["grants"][0]["unauthorized_at"] == "2026-10-04T08:00:00Z" + assert document["destinations"][0]["destination"]["location"] == DEST_A["location"] + + +@pytest.mark.asyncio +async def test_a_restarted_engine_owes_the_same_periods() -> None: + store = InMemoryReportingSubscriptionStore() + world = World(support=CAPACITY) + world.engine = ReportingSubscriptionEngine( + principals=world.principals, + host=world.host, + offerings=OFFERINGS, + support=CAPACITY, + store=store, + clock=world.clock, + ) + world.principals.set_reporting_subscriptions(world.engine.principal_options()) + await _scenario(world) + before = await _observed(world, world.engine) + + restarted = ReportingSubscriptionEngine( + principals=world.principals, + host=world.host, + offerings=(*OFFERINGS, MANAGED), + support=CAPACITY, + store=store, + clock=world.clock, + ) + assert await _observed(world, restarted) == before + assert before["owed"][("files", "acct-1")] + assert any(entry for entry in before["coverage"]["acct-2"]) + + +class _ConflictingStore(InMemoryReportingSubscriptionStore): + """Loses the first ``conflicts`` writes to a simulated concurrent writer.""" + + def __init__(self, conflicts: int) -> None: + super().__init__() + self.conflicts = conflicts + self.attempts = 0 + + async def compare_and_swap(self, subject, expected_revision, document): + self.attempts += 1 + if self.conflicts: + self.conflicts -= 1 + return False + return await super().compare_and_swap(subject, expected_revision, document) + + +@pytest.mark.asyncio +async def test_a_conflicting_write_is_reloaded_and_reapplied() -> None: + world = World() + store = _ConflictingStore(conflicts=2) + world.engine._store = store + await world.engine.authorization_changed( + IDENTITY, "acct-1", authorized=True, at=at("2026-10-01T00:00:00Z"), timezone="UTC" + ) + assert store.attempts == 3 + stored = await store.load(IDENTITY.subject) + assert stored is not None and stored.revision == 1 + assert stored.document["accounts"]["acct-1"]["grants"] == [ + {"authorized_at": "2026-10-01T00:00:00Z", "unauthorized_at": None} + ] + + +@pytest.mark.asyncio +async def test_endless_conflicts_raise_instead_of_dropping_the_change() -> None: + world = World() + world.engine._store = _ConflictingStore(conflicts=1_000) + with pytest.raises(ReportingSubscriptionStateConflictError): + await world.engine.authorization_changed( + IDENTITY, "acct-1", authorized=True, at=at("2026-10-01T00:00:00Z"), timezone="UTC" + ) + + +@pytest.mark.asyncio +async def test_an_unchanged_document_is_not_rewritten() -> None: + world = World(support=CAPACITY) + await _scenario(world) + before = await world.engine._store.load(IDENTITY.subject) + await world.reconcile("2026-10-06T15:00:00Z") + after = await world.engine._store.load(IDENTITY.subject) + assert before is not None and after is not None + assert after.revision == before.revision + + +@pytest.mark.asyncio +async def test_exceptions_closed_beyond_retention_are_pruned_on_write() -> None: + world = World(support=CAPACITY, retention_days=10) + await _scenario(world) + stored = await world.engine._store.load(IDENTITY.subject) + assert stored is not None + state = decode_principal_state(stored.document) + closed = [ + entry.until + for entries in state.exclusions.values() + for entry in entries + if entry.until is not None + ] + assert closed, "the scenario must close an exception" + horizon = max(closed) + timedelta(days=11) + world.clock.now = horizon + await world.engine.authorization_changed( + IDENTITY, "acct-3", authorized=True, at=horizon, timezone="UTC" + ) + pruned = decode_principal_state((await world.engine._store.load(IDENTITY.subject)).document) + assert not [ + entry + for entries in pruned.exclusions.values() + for entry in entries + if entry.until is not None and entry.until < horizon - timedelta(days=10) + ] + + +def test_an_unknown_document_version_is_refused() -> None: + with pytest.raises(ValueError, match="document version"): + decode_principal_state({"document_version": 99}) + + +@pytest.mark.asyncio +async def test_principal_records_round_trip_through_a_json_document() -> None: + from adcp.server import principal_record_from_document, principal_record_to_document + + world = World(support=CAPACITY) + await _scenario(world) + record = await world.principals._store.get(IDENTITY.subject) + assert record is not None and record.configuration is not None + document = json.loads(json.dumps(principal_record_to_document(record))) + restored = principal_record_from_document(document, store_revision=record.store_revision) + assert principal_record_to_document(restored) == document + assert restored.store_revision == record.store_revision + assert restored.configuration == record.configuration + assert restored.idempotency.keys() == record.idempotency.keys() + with pytest.raises(ValueError, match="document version"): + principal_record_from_document({**document, "document_version": 2}, store_revision=1)