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)