From f5c119da9e67ecf85fe071bdef43e26cd4b63742 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 22:38:10 +0000 Subject: [PATCH 1/2] feat(reporting): mirror the TypeScript SDK's advanced path, engine errors and readback name Parity with adcp-client#3542, #3544 and #3546: - adcp.reporting.advanced is the documented home of the low-level building blocks, as @adcp/sdk/reporting/advanced is: the engine and its stores, the coverage function, calendar and selection, the capability builder, principal-section validation and the service planner hooks. adcp.reporting.subscriptions and service_subscriptions keep defining and exporting them. ReportingSubscriptionServiceOptions leaves the adcp.reporting facade (it was added there earlier in this stack, never released); the facade keeps SubscriptionOptions and PrincipalRef. - ReportingSubscriptionEngineError carries code and recovery for the engine's runtime failures, as TypeScript names them: - SUBSCRIPTION_STATE_CONFLICT: ReportingSubscriptionStateConflictError is now a subclass. - SUBSCRIPTION_STATE_CORRUPT: an unreadable state document; still a ValueError. - SUBSCRIPTION_ROSTER_INCOMPLETE: the roster walk is capped at 10,000 pages, and an unfinished one is refused, never used partially. - SUBSCRIPTION_HOST_FACT_INVALID: a host hook returned the wrong shape. - ReliableReportingConfigurationError gains config_field (TypeScript's configField), set on the seller configuration codes the SDK raises. - PrincipalService.update_subscription_status is renamed record_subscription_readback (TypeScript's recordSubscriptionReadback); the old name warns and delegates. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/adcp_reporting_storage/storage.py | 1 + src/adcp/reporting/__init__.py | 10 +- src/adcp/reporting/_errors.py | 8 +- src/adcp/reporting/advanced.py | 168 ++++++++++++++++++ src/adcp/reporting/subscriptions/engine.py | 102 ++++++++++- src/adcp/reporting/subscriptions/managed.py | 5 + src/adcp/server/principal.py | 18 +- src/adcp/storage/__init__.py | 1 + tests/fixtures/public_api_snapshot.json | 1 - .../test_principal_reporting_subscriptions.py | 4 +- .../test_reporting_subscription_ts_parity.py | 105 +++++++++++ 11 files changed, 399 insertions(+), 24 deletions(-) create mode 100644 src/adcp/reporting/advanced.py create mode 100644 tests/test_reporting_subscription_ts_parity.py diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py index 7498112f7..7fd83087c 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/storage.py @@ -88,6 +88,7 @@ async def check_schema() -> None: f"the reporting schema is behind this SDK (pending: {versions}). " f'Run "{MIGRATE_COMMAND}" against this database{note}.', code="REPORTING_SCHEMA_BEHIND", + config_field="python -m adcp migrate", ) await staging.check_ready() diff --git a/src/adcp/reporting/__init__.py b/src/adcp/reporting/__init__.py index de4411c8e..46beb2df4 100644 --- a/src/adcp/reporting/__init__.py +++ b/src/adcp/reporting/__init__.py @@ -70,6 +70,7 @@ ) if TYPE_CHECKING: + from adcp.reporting import advanced as advanced from adcp.reporting import canonical_json as canonical_json from adcp.reporting import conformance as conformance from adcp.reporting import currency as currency @@ -118,9 +119,6 @@ from adcp.reporting.service_production import ( ReportingServiceOffering as ReportingServiceOffering, ) - from adcp.reporting.service_subscriptions import ( - ReportingSubscriptionServiceOptions as ReportingSubscriptionServiceOptions, - ) from adcp.reporting.service_subscriptions import ( SubscriptionOptions as SubscriptionOptions, ) @@ -157,6 +155,7 @@ _LAZY_SUBMODULES = frozenset( { + "advanced", "canonical_json", "conformance", "currency", @@ -189,10 +188,6 @@ "ReportingAdapter": ("service", "ReportingAdapter"), "ReportingProductionOptions": ("service_production", "ReportingProductionOptions"), "ReportingServiceOffering": ("service_production", "ReportingServiceOffering"), - "ReportingSubscriptionServiceOptions": ( - "service_subscriptions", - "ReportingSubscriptionServiceOptions", - ), "SubscriptionOptions": ("service_subscriptions", "SubscriptionOptions"), "DeterministicReportingClock": ("testing", "DeterministicReportingClock"), "FaultInjectingReportingSealStore": ("testing", "FaultInjectingReportingSealStore"), @@ -250,7 +245,6 @@ def __dir__() -> list[str]: "ReportingProductionOptions", "ReportingServiceOffering", "PrincipalRef", - "ReportingSubscriptionServiceOptions", "SubscriptionOptions", "DeterministicReportingClock", "FaultInjectingReportingSealStore", diff --git a/src/adcp/reporting/_errors.py b/src/adcp/reporting/_errors.py index 848beaf70..a42a44788 100644 --- a/src/adcp/reporting/_errors.py +++ b/src/adcp/reporting/_errors.py @@ -81,8 +81,9 @@ class ReliableReportingConfigurationError(ValueError): vocabulary (``REPORTING_STORAGE_NOT_DURABLE``, ``REPORTING_SCHEMA_BEHIND``, ``SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS``, ``SUBSCRIPTION_ROSTER_UNAVAILABLE``); those carry ``recovery="correctable"`` - and a message whose last sentence is the fix. ``dispatched`` is always - ``None``: a configuration check sends nothing. + and a message whose last sentence is the fix; ``config_field`` names the + configuration key to change. ``dispatched`` is always ``None``: a + configuration check sends nothing. """ dispatched = None @@ -93,12 +94,15 @@ def __init__( *, kind: Literal["misconfiguration", "auth_required", "invalid_request"] = "misconfiguration", code: str | None = None, + config_field: str | None = None, ) -> None: super().__init__(message) if code is not None and code not in SELLER_CONFIGURATION_CODES: raise ValueError(f"unknown seller configuration code {code!r}") self.kind = kind self.code = code + #: The configuration key to change (the TypeScript SDK's ``configField``). + self.config_field = config_field self.recovery: Literal["correctable", "terminal"] | None = ( "correctable" if code is not None else None ) diff --git a/src/adcp/reporting/advanced.py b/src/adcp/reporting/advanced.py new file mode 100644 index 000000000..3f44b8b3d --- /dev/null +++ b/src/adcp/reporting/advanced.py @@ -0,0 +1,168 @@ +"""Reporting-subscription building blocks for hand-wired compositions (experimental). + +Rung 3 of the reporting-subscriptions design: most sellers enable subscriptions +with ``ReliableReportingService(subscriptions=SubscriptionOptions(...))`` and +never touch these. Use them to bring your own ledger, build the engine and link +the principal runtime yourself +(``ReliableReportingService(reporting_subscriptions=ReportingSubscriptionServiceOptions(...))``), +or compute coverage directly. The TypeScript SDK publishes the same building +blocks from ``@adcp/sdk/reporting/advanced``. + +Everything here is experimental and may change between betas. The names are +defined in :mod:`adcp.reporting.subscriptions` and +:mod:`adcp.reporting.service_subscriptions`, which keep exporting them. +""" + +from __future__ import annotations + +from adcp.reporting.service_subscriptions import ( + ManagedGenerationBinder, + PlanningErrorHandler, + PrincipalLister, + PrincipalResolver, + ProductionSubscriptionHooks, + ReportingSubscriptionServiceOptions, + SubscriptionPlanningFailure, +) +from adcp.reporting.subscriptions._clock import COVERAGE_INPUT_DETAILS, CoverageInputError +from adcp.reporting.subscriptions.capabilities import ( + ReportingSubscriptionSupport, + advertise_reporting_subscriptions, + reporting_configuration_tasks, + subscription_capabilities, + with_subscription_dimensions, +) +from adcp.reporting.subscriptions.coverage import ( + DEFAULT_MAX_ITERATIONS, + DEFAULT_MAX_PERIODS, + AccountTimezoneChange, + ExpectedSubscriptionPeriod, + OwedSubscriptionPeriod, + SubscriptionAuthorization, + SubscriptionBuy, + SubscriptionBuyFlight, + SubscriptionBuyPackage, + SubscriptionException, + SubscriptionVersionWindow, + authorization_chain_start, + expected_subscription_periods, + owed_subscription_periods, + subscription_covers_account, + subscription_exceptions, + subscription_version_windows, +) +from adcp.reporting.subscriptions.engine import ( + ENGINE_ERROR_CODES, + AccountConfigsEnder, + AccountDeliveryConfig, + AuthorizedAccount, + AuthorizedAccountPage, + GrantObservation, + InMemoryReportingSubscriptionStore, + ReportingSubscriptionDeliveryHost, + ReportingSubscriptionEngine, + ReportingSubscriptionEngineError, + ReportingSubscriptionHost, + ReportingSubscriptionPrecedenceHost, + ReportingSubscriptionStateConflictError, + ReportingSubscriptionStateCorruptError, + ReportingSubscriptionStateStore, + StoredReportingSubscriptionState, + decode_principal_state, + encode_principal_state, +) +from adcp.reporting.subscriptions.ledger import ( + FrozenOwedSet, + SubscriptionObligationPlanner, + SubscriptionPlanResult, + subscription_obligation_id, +) +from adcp.reporting.subscriptions.registry import ( + ReportingSubscriptionValidationError, + apply_reporting_subscriptions, + refresh_destination_states, + render_path_template, +) +from adcp.reporting.subscriptions.resolution import ( + STANDARD_REPORTING_DIMENSIONS, + SubscriptionDestination, + SubscriptionMatch, + SubscriptionOffering, + SubscriptionResolution, + SubscriptionSelector, + resolve_subscription_offerings, +) + +__all__ = [ + # Engine + "ENGINE_ERROR_CODES", + "AccountConfigsEnder", + "AccountDeliveryConfig", + "AuthorizedAccount", + "AuthorizedAccountPage", + "FrozenOwedSet", + "GrantObservation", + "ReportingSubscriptionDeliveryHost", + "ReportingSubscriptionEngine", + "ReportingSubscriptionEngineError", + "ReportingSubscriptionHost", + "ReportingSubscriptionPrecedenceHost", + "ReportingSubscriptionStateConflictError", + "ReportingSubscriptionStateCorruptError", + # Engine stores + "InMemoryReportingSubscriptionStore", + "ReportingSubscriptionStateStore", + "StoredReportingSubscriptionState", + "decode_principal_state", + "encode_principal_state", + # Coverage function + "DEFAULT_MAX_ITERATIONS", + "DEFAULT_MAX_PERIODS", + "AccountTimezoneChange", + "ExpectedSubscriptionPeriod", + "OwedSubscriptionPeriod", + "SubscriptionAuthorization", + "SubscriptionBuy", + "SubscriptionBuyFlight", + "SubscriptionBuyPackage", + "SubscriptionException", + "SubscriptionVersionWindow", + "authorization_chain_start", + "expected_subscription_periods", + "owed_subscription_periods", + "subscription_covers_account", + "subscription_exceptions", + "subscription_version_windows", + # Calendar and selection + "COVERAGE_INPUT_DETAILS", + "CoverageInputError", + "STANDARD_REPORTING_DIMENSIONS", + "SubscriptionDestination", + "SubscriptionMatch", + "SubscriptionOffering", + "SubscriptionResolution", + "SubscriptionSelector", + "resolve_subscription_offerings", + # Capability builder + "ReportingSubscriptionSupport", + "advertise_reporting_subscriptions", + "reporting_configuration_tasks", + "subscription_capabilities", + "with_subscription_dimensions", + # Principal section validation + "ReportingSubscriptionValidationError", + "apply_reporting_subscriptions", + "refresh_destination_states", + "render_path_template", + # Service planner hooks + "ManagedGenerationBinder", + "PlanningErrorHandler", + "PrincipalLister", + "PrincipalResolver", + "ProductionSubscriptionHooks", + "ReportingSubscriptionServiceOptions", + "SubscriptionObligationPlanner", + "SubscriptionPlanResult", + "SubscriptionPlanningFailure", + "subscription_obligation_id", +] diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 806b68bba..890d9e00a 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -73,6 +73,9 @@ _T = TypeVar("_T") +#: Pages one roster walk may take before it is refused as incomplete. +_MAX_ROSTER_PAGES = 10_000 + #: Attempts at a compare-and-swap write before giving up under contention. _MAX_STATE_WRITE_ATTEMPTS = 8 @@ -404,9 +407,58 @@ async def list_principals(self) -> list[str]: return sorted(self._documents) -class ReportingSubscriptionStateConflictError(RuntimeError): +#: Runtime codes of :class:`ReportingSubscriptionEngineError`, as the TypeScript +#: SDK names them. The spec's ``error-codes.json`` has no seller runtime codes. +ENGINE_ERROR_CODES = frozenset( + { + "SUBSCRIPTION_STATE_CONFLICT", + "SUBSCRIPTION_STATE_CORRUPT", + "SUBSCRIPTION_ROSTER_INCOMPLETE", + "SUBSCRIPTION_HOST_FACT_INVALID", + } +) + + +class ReportingSubscriptionEngineError(RuntimeError): + """A reporting-subscription engine runtime failure, with ``code`` and ``recovery``. + + ``SUBSCRIPTION_STATE_CONFLICT`` (transient): the state kept changing + concurrently. ``SUBSCRIPTION_STATE_CORRUPT`` (terminal): a stored document + cannot be read. ``SUBSCRIPTION_ROSTER_INCOMPLETE`` (transient): the host's + account roster did not finish paging. ``SUBSCRIPTION_HOST_FACT_INVALID`` + (terminal): a host hook returned the wrong shape. ``dispatched`` is + always ``None``: the engine sends nothing. + """ + + dispatched = None + _RECOVERY = { + "SUBSCRIPTION_STATE_CONFLICT": "transient", + "SUBSCRIPTION_STATE_CORRUPT": "terminal", + "SUBSCRIPTION_ROSTER_INCOMPLETE": "transient", + "SUBSCRIPTION_HOST_FACT_INVALID": "terminal", + } + + def __init__(self, code: str, message: str) -> None: + if code not in ENGINE_ERROR_CODES: + raise ValueError(f"unknown reporting-subscription engine code {code!r}") + super().__init__(message) + self.code = code + self.recovery = self._RECOVERY[code] + + +class ReportingSubscriptionStateConflictError(ReportingSubscriptionEngineError): """The engine could not write its state after repeated concurrent changes.""" + def __init__(self, message: str) -> None: + super().__init__("SUBSCRIPTION_STATE_CONFLICT", message) + + +class ReportingSubscriptionStateCorruptError(ReportingSubscriptionEngineError, ValueError): + """A stored engine-state document cannot be read (still a ``ValueError``).""" + + def __init__(self, message: str) -> None: + super().__init__("SUBSCRIPTION_STATE_CORRUPT", message) + def _copy(document: Mapping[str, Any]) -> dict[str, Any]: copied = json.loads(json.dumps(document)) @@ -556,7 +608,9 @@ 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}") + raise ReportingSubscriptionStateCorruptError( + f"unsupported reporting-subscription state document version {version!r}" + ) principal = _PrincipalState() for account_id, raw in document["accounts"].items(): principal.accounts[account_id] = _Account( @@ -1096,14 +1150,41 @@ def _apply_grant( ) async def _list_accounts(self, identity: PrincipalLike) -> list[AuthorizedAccount]: + """Walk every page of the roster; a walk that does not finish is refused, never partial.""" accounts: list[AuthorizedAccount] = [] cursor: str | None = None - while True: + for _page in range(_MAX_ROSTER_PAGES): page = await self._host.list_authorized_accounts(identity, cursor=cursor) + if not isinstance(page, AuthorizedAccountPage) or not all( + isinstance(item, AuthorizedAccount) for item in page.accounts + ): + raise ReportingSubscriptionEngineError( + "SUBSCRIPTION_HOST_FACT_INVALID", + "list_authorized_accounts must return an AuthorizedAccountPage of " + "AuthorizedAccount values", + ) accounts.extend(page.accounts) if not page.next_cursor: return accounts cursor = page.next_cursor + raise ReportingSubscriptionEngineError( + "SUBSCRIPTION_ROSTER_INCOMPLETE", + f"the account roster did not finish within {_MAX_ROSTER_PAGES} pages; " + "the partial roster is not used", + ) + + async def _account_buys( + self, identity: PrincipalLike, account_id: str + ) -> Sequence[SubscriptionBuy]: + buys: Any = await self._host.account_buys(identity, account_id) + if isinstance(buys, (str, bytes)) or not all( + isinstance(item, SubscriptionBuy) for item in buys + ): + raise ReportingSubscriptionEngineError( + "SUBSCRIPTION_HOST_FACT_INVALID", + "account_buys must return SubscriptionBuy values", + ) + return tuple(buys) # -- reconcile ------------------------------------------------------------- @@ -1196,7 +1277,7 @@ async def _apply_replacements(self, identity: PrincipalLike, now: datetime) -> N configuration, account_id, account ): continue - buys = await self._host.account_buys(identity, account_id) + buys = await self._account_buys(identity, account_id) product_ids = _buy_offering_ids(buys) or [ item.offering_id for item in self._offerings ] @@ -1235,7 +1316,7 @@ async def _evaluate(self, identity: PrincipalLike, now: datetime) -> None: configs: dict[str, Sequence[AccountDeliveryConfig]] = {} for account_id, account in accounts.items(): if account.authorized or account_id in covered_ids: - buys[account_id] = await self._host.account_buys(identity, account_id) + buys[account_id] = await self._account_buys(identity, account_id) if account.authorized and precedence is not None: configs[account_id] = await precedence(identity, account_id) @@ -1289,7 +1370,7 @@ def record( 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( + await self._principals.record_subscription_readback( identity, subscription_id, accounts_with_exceptions=with_exceptions, @@ -1648,7 +1729,7 @@ async def subscription_coverage( instant = self._now(now) 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) + buys = await self._account_buys(identity, account_id) principal = await self._read(identity.subject) account = principal.accounts.get(account_id) if account is None or not account.authorized: @@ -1832,7 +1913,7 @@ async def owed_periods( if state is None: raise KeyError(subscription_id) destinations = await self._destinations(identity) - buys = await self._host.account_buys(identity, account_id) + buys = await self._account_buys(identity, account_id) principal = await self._read(identity.subject) account = principal.accounts.get(account_id) if account is None: @@ -1914,7 +1995,7 @@ async def frozen_owed_sets( for key, value in principal.accounts.items() if account_id is None or key == account_id } - buys = {key: await self._host.account_buys(identity, key) for key in accounts} + buys = {key: await self._account_buys(identity, key) for key in accounts} result: list[FrozenOwedSet] = [] for state in states: subscription_id = str(state["subscription_id"]) @@ -2084,6 +2165,9 @@ def _destination_for( __all__ = [ + "ENGINE_ERROR_CODES", + "ReportingSubscriptionEngineError", + "ReportingSubscriptionStateCorruptError", "GrantObservation", "AccountDeliveryConfig", "AuthorizedAccount", diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index 8883d561c..ec3ad6d4a 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -351,6 +351,7 @@ def _roster_unavailable(principal: PrincipalRef, cause: str) -> ReliableReportin f"under its identity ({cause}); reporting subscriptions never read that as an empty " "roster. Supply SubscriptionOptions.accounts to list them another way.", code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + config_field="SubscriptionOptions.accounts", ) @@ -601,6 +602,7 @@ def _offerings( "none is registered. Register each source with capability_offerings, publishing " "standard_dimensions on every offering.", code="SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS", + config_field="capability_offerings", ) mapped: list[SubscriptionOffering] = [] for offering in offerings: @@ -614,6 +616,7 @@ def _offerings( "standard_dimensions (empty for totals only) in its capability_offerings " "entry.", code="SUBSCRIPTIONS_REQUIRE_STANDARD_DIMENSIONS", + config_field="capability_offerings", ) try: mapped.append(SubscriptionOffering.from_offering(offering)) @@ -644,6 +647,7 @@ def check_roster(self) -> None: "and this platform does not serve list_accounts. Implement list_accounts (a " "decisioning platform's accounts.list), or supply SubscriptionOptions.accounts.", code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + config_field="SubscriptionOptions.accounts", ) if self.options.buys is None and not self._platform.decisioning: from adcp.server import ADCPHandler @@ -912,6 +916,7 @@ async def _platform_accounts( f"account {account_id!r} has no IANA timezone, and coverage never " "substitutes one. Serve timezone on every account list_accounts returns.", code="SUBSCRIPTION_ROSTER_UNAVAILABLE", + config_field="SubscriptionOptions.accounts", ) authorization = account.get("authorization") tasks = None diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index fb6a70feb..c63ab9e28 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -1044,7 +1044,18 @@ def _now(self) -> datetime: clock = self._subscriptions.clock if self._subscriptions else None return clock() if clock else datetime.now(timezone.utc) - async def update_subscription_status( + async def update_subscription_status(self, *args: Any, **kwargs: Any) -> bool: + """Deprecated: renamed :meth:`record_subscription_readback`; kept for one beta.""" + import warnings + + warnings.warn( + "update_subscription_status is renamed record_subscription_readback", + DeprecationWarning, + stacklevel=2, + ) + return await self.record_subscription_readback(*args, **kwargs) + + async def record_subscription_readback( self, identity: PrincipalIdentity | PrincipalRef, subscription_id: str, @@ -1053,7 +1064,10 @@ async def update_subscription_status( accounts_unauthorized: Sequence[Mapping[str, Any]] | None = None, changed_at: datetime | None = None, ) -> bool: - """Record seller-derived subscription status without version churn. + """Record a subscription's seller-derived readback lists without version churn. + + Writes ``accounts_with_exceptions[]`` and ``accounts_unauthorized[]`` + (the TypeScript SDK's ``recordSubscriptionReadback``). Like destination proof transitions, this does not advance the caller-owned ``configuration_version``. A change emits diff --git a/src/adcp/storage/__init__.py b/src/adcp/storage/__init__.py index 245f3b82f..5701e803e 100644 --- a/src/adcp/storage/__init__.py +++ b/src/adcp/storage/__init__.py @@ -139,4 +139,5 @@ def require_durable(durable: bool, *, stores: str) -> None: "which do not survive a restart. Pass storage=postgres_storage(dsn=...) to " "ReliableReportingService, or set ADCP_ENV=development to run in memory.", code="REPORTING_STORAGE_NOT_DURABLE", + config_field="storage", ) diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index cfe5d7cf1..93b5332c6 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -802,7 +802,6 @@ "ReportingServiceOffering": "adcp.reporting.service_production", "ReportingServiceResource": "adcp.reporting.service_lifecycle", "ReportingStatusClient": "adcp.reporting._reconcile", - "ReportingSubscriptionServiceOptions": "adcp.reporting.service_subscriptions", "ReportingTestBarrier": "adcp.reporting._testing_failures", "ReportingTier": "adcp.reporting._reconcile", "ScriptedReportingAdapter": "adcp.reporting.testing", diff --git a/tests/test_principal_reporting_subscriptions.py b/tests/test_principal_reporting_subscriptions.py index e8375dfaf..e7cf5278f 100644 --- a/tests/test_principal_reporting_subscriptions.py +++ b/tests/test_principal_reporting_subscriptions.py @@ -581,14 +581,14 @@ async def test_seller_status_updates_do_not_bump_the_configuration_version() -> emitted: list[PrincipalChange] = [] principals, _clock = service(emitted=emitted) first = await sync(principals, {"reporting_subscriptions": [subscription()]}) - changed = await principals.update_subscription_status( + changed = await principals.record_subscription_readback( IDENTITY, "analytics-daily", accounts_with_exceptions=[{"account_id": "acct-1", "reasons": ["capacity"]}], accounts_unauthorized=[], ) assert changed is True - unchanged = await principals.update_subscription_status( + unchanged = await principals.record_subscription_readback( IDENTITY, "analytics-daily", accounts_with_exceptions=[{"account_id": "acct-1", "reasons": ["capacity"]}], diff --git a/tests/test_reporting_subscription_ts_parity.py b/tests/test_reporting_subscription_ts_parity.py new file mode 100644 index 000000000..d06cb88eb --- /dev/null +++ b/tests/test_reporting_subscription_ts_parity.py @@ -0,0 +1,105 @@ +"""Names and error shapes that mirror the TypeScript SDK (adcp-client#3542-#3546).""" + +from __future__ import annotations + +import pytest + +from adcp.reporting.subscriptions import AuthorizedAccountPage +from tests.test_reporting_subscription_engine import IDENTITY, World + + +def test_the_advanced_module_is_the_home_of_the_building_blocks() -> None: + import adcp.reporting as reporting + import adcp.reporting.advanced as advanced + + for name in ( + "ReportingSubscriptionEngine", + "InMemoryReportingSubscriptionStore", + "owed_subscription_periods", + "resolve_subscription_offerings", + "subscription_capabilities", + "ReportingSubscriptionServiceOptions", + "SubscriptionObligationPlanner", + "ReportingSubscriptionEngineError", + ): + assert name in advanced.__all__ and getattr(advanced, name) is not None, name + # The facade surface keeps only the managed path's names. + assert "ReportingSubscriptionServiceOptions" not in reporting.__all__ + assert {"SubscriptionOptions", "PrincipalRef"} <= set(reporting.__all__) + + +def test_engine_errors_carry_code_and_recovery() -> None: + from adcp.reporting.advanced import ( + ReportingSubscriptionEngineError, + ReportingSubscriptionStateConflictError, + decode_principal_state, + ) + + conflict = ReportingSubscriptionStateConflictError("kept changing") + assert isinstance(conflict, ReportingSubscriptionEngineError) + assert (conflict.code, conflict.recovery, conflict.dispatched) == ( + "SUBSCRIPTION_STATE_CONFLICT", + "transient", + None, + ) + with pytest.raises(ValueError) as corrupt: + decode_principal_state({"document_version": 99}) + assert isinstance(corrupt.value, ReportingSubscriptionEngineError) + assert (corrupt.value.code, corrupt.value.recovery) == ( + "SUBSCRIPTION_STATE_CORRUPT", + "terminal", + ) + + +@pytest.mark.asyncio +async def test_a_roster_that_never_finishes_is_refused(monkeypatch) -> None: + from adcp.reporting.advanced import ReportingSubscriptionEngineError + + world = World() + + async def endless(principal, *, cursor): + return AuthorizedAccountPage((), next_cursor=str(int(cursor or 0) + 1)) + + world.host.list_authorized_accounts = endless # type: ignore[method-assign] + monkeypatch.setattr("adcp.reporting.subscriptions.engine._MAX_ROSTER_PAGES", 3) + with pytest.raises(ReportingSubscriptionEngineError) as refused: + await world.engine.reconcile(IDENTITY) + assert (refused.value.code, refused.value.recovery) == ( + "SUBSCRIPTION_ROSTER_INCOMPLETE", + "transient", + ) + + +@pytest.mark.asyncio +async def test_a_host_fact_of_the_wrong_shape_is_refused() -> None: + from adcp.reporting.advanced import ReportingSubscriptionEngineError + + world = World() + + async def wrong(principal, *, cursor): + return {"accounts": []} + + world.host.list_authorized_accounts = wrong # type: ignore[method-assign] + with pytest.raises(ReportingSubscriptionEngineError) as refused: + await world.engine.reconcile(IDENTITY) + assert refused.value.code == "SUBSCRIPTION_HOST_FACT_INVALID" + + +@pytest.mark.asyncio +async def test_update_subscription_status_is_renamed_record_subscription_readback() -> None: + from adcp.server import PrincipalService + + principals = PrincipalService() + with pytest.warns(DeprecationWarning, match="record_subscription_readback"): + with pytest.raises(KeyError): # delegated: no such subscription + await principals.update_subscription_status(IDENTITY, "missing") + + +def test_configuration_errors_name_the_key_to_change() -> None: + from adcp.reporting import ReliableReportingConfigurationError + + error = ReliableReportingConfigurationError( + "x", code="SUBSCRIPTION_ROSTER_UNAVAILABLE", config_field="SubscriptionOptions.accounts" + ) + assert error.config_field == "SubscriptionOptions.accounts" + assert ReliableReportingConfigurationError("y").config_field is None From 258b435fa74274ce17c3e09966b3ae42e53736f2 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 23:57:07 +0000 Subject: [PATCH 2/2] fix(reporting): count a rejected authorized_at once and refuse buys of the wrong shape Review follow-ups: - An authorized_at regression is counted and logged after the engine-state write, not inside the change, which reruns on a compare-and-swap conflict. Under contention one regression used to count and log once per attempt. - account_buys returning None, a non-sequence or a string now fails with SUBSCRIPTION_HOST_FACT_INVALID instead of a raw TypeError. - The managed path's page-cap comment covers buys only; the engine caps the roster walk. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/engine.py | 45 ++++++++++++----- src/adcp/reporting/subscriptions/managed.py | 4 +- .../test_reporting_subscription_ts_parity.py | 50 +++++++++++++++++++ 3 files changed, 84 insertions(+), 15 deletions(-) diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 890d9e00a..72152c22c 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -909,7 +909,10 @@ async def observe_grants( reported = None if item.authorized_at is None else utc_instant(item.authorized_at) prepared.append((item, tasks, _covers(tasks), reported)) + regressions: list[tuple[str, datetime]] = [] + def change(state: _PrincipalState) -> dict[str, datetime | None]: + regressions.clear() # the change may rerun on a write conflict served: dict[str, datetime | None] = {} for item, tasks, covered, reported in prepared: account = state.accounts.get(item.account_id) @@ -919,9 +922,11 @@ def change(state: _PrincipalState) -> dict[str, datetime | None]: continue account = _Account(timezone=item.timezone, operator=item.operator) state.accounts[item.account_id] = account - self._observe( + if self._observe( account, covered=covered, reported=reported, at=instant, source="listed" - ) + ): + assert reported is not None + regressions.append((item.account_id, reported)) account.allowed_tasks = tasks if covered else None account.set_timezone(item.timezone, instant) if item.operator is not None: @@ -931,7 +936,9 @@ def change(state: _PrincipalState) -> dict[str, datetime | None]: if not prepared: return {} - return await self._mutate(principal.subject, change, now=instant) + served = await self._mutate(principal.subject, change, now=instant) + self._report_regressions(regressions) + return served async def refresh(self, principal: PrincipalLike, *, now: datetime | None = None) -> None: """Re-evaluate exceptions from the journal without listing the host's accounts. @@ -1118,17 +1125,22 @@ def _observe( self._apply_grant(account, authorized=True, at=granted, source=source) return False start = self._chain_start(account) - if reported is not None and start is not None and reported < start: + return reported is not None and start is not None and reported < start + + def _report_regressions(self, regressions: Sequence[tuple[str, datetime]]) -> None: + """Count and log rejected ``authorized_at`` values once, after the state write. + + Called outside ``_mutate``, whose change may run more than once. + """ + for account_id, reported in regressions: self._regressions += 1 logger.warning( - "authorized_at_regressed: the host reported authorized_at %s, before the " - "chain start %s the grant journal holds; the journal keeps its value", + "authorized_at_regressed: the host reported authorized_at %s for account %s, " + "before the chain start the grant journal holds; the journal keeps its value", iso_z(reported), - iso_z(start), + account_id, extra={"event": "reporting.subscription.authorized_at_regressed"}, ) - return True - return False def _chain_start(self, account: _Account) -> datetime | None: return authorization_chain_start(account.grants, status_retention_days=self._retention_days) @@ -1177,8 +1189,10 @@ async def _account_buys( self, identity: PrincipalLike, account_id: str ) -> Sequence[SubscriptionBuy]: buys: Any = await self._host.account_buys(identity, account_id) - if isinstance(buys, (str, bytes)) or not all( - isinstance(item, SubscriptionBuy) for item in buys + if ( + not isinstance(buys, Sequence) + or isinstance(buys, (str, bytes)) + or not all(isinstance(item, SubscriptionBuy) for item in buys) ): raise ReportingSubscriptionEngineError( "SUBSCRIPTION_HOST_FACT_INVALID", @@ -1207,20 +1221,24 @@ async def reconcile(self, identity: PrincipalLike, *, now: datetime | None = Non if _covers(item.allowed_tasks) } + regressions: list[tuple[str, datetime]] = [] + def change(principal: _PrincipalState) -> None: + regressions.clear() # the change may rerun on a write conflict for account_id, item in listed.items(): account = principal.accounts.get(account_id) if account is None: account = principal.accounts[account_id] = _Account( timezone=item.timezone, operator=item.operator ) - self._observe( + if self._observe( account, covered=True, reported=item.authorized_at, at=instant, source="listed", - ) + ): + regressions.append((account_id, item.authorized_at)) account.set_timezone(item.timezone, instant) account.operator = item.operator account.allowed_tasks = item.allowed_tasks @@ -1232,6 +1250,7 @@ def change(principal: _PrincipalState) -> None: principal.reconciled_at = max(instant, principal.reconciled_at or instant) await self._mutate(identity.subject, change, now=instant) + self._report_regressions(regressions) await self._apply_replacements(identity, instant) await self._evaluate(identity, instant) diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index ec3ad6d4a..e72633e8c 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -99,8 +99,8 @@ "rejected", "canceled", ) -#: Hard cap on pages walked for one principal's roster or one account's buys; -#: a walk that does not end is refused, never truncated. +#: Hard cap on pages walked for one account's buys; a walk that does not end is +#: refused, never truncated. The engine caps the roster walk the same way. _MAX_PAGES = 10_000 diff --git a/tests/test_reporting_subscription_ts_parity.py b/tests/test_reporting_subscription_ts_parity.py index d06cb88eb..3970a320d 100644 --- a/tests/test_reporting_subscription_ts_parity.py +++ b/tests/test_reporting_subscription_ts_parity.py @@ -103,3 +103,53 @@ def test_configuration_errors_name_the_key_to_change() -> None: ) assert error.config_field == "SubscriptionOptions.accounts" assert ReliableReportingConfigurationError("y").config_field is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("returned", [None, 7, "mb-1", ({"media_buy_id": "x"},)]) +async def test_account_buys_of_the_wrong_shape_are_refused(returned) -> None: + from datetime import datetime, timezone + + from adcp.reporting.advanced import ReportingSubscriptionEngineError + from adcp.reporting.subscriptions import AuthorizedAccount + + world = World() + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone="UTC", authorized_at=datetime(2026, 9, 1, tzinfo=timezone.utc) + ) + + async def wrong(principal, account_id): + return returned + + world.host.account_buys = wrong # type: ignore[method-assign] + with pytest.raises(ReportingSubscriptionEngineError) as refused: + await world.engine._account_buys(IDENTITY, "acct-1") + assert refused.value.code == "SUBSCRIPTION_HOST_FACT_INVALID" + + +@pytest.mark.asyncio +async def test_a_regression_is_counted_once_under_write_conflicts() -> None: + from datetime import datetime, timezone + + from adcp.reporting.subscriptions import AuthorizedAccount + from tests.test_reporting_subscription_state_store import _ConflictingStore + + world = World() + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone="UTC", authorized_at=datetime(2026, 9, 1, tzinfo=timezone.utc) + ) + await world.engine.reconcile(IDENTITY) + store = _ConflictingStore(conflicts=2) + store._documents = world.engine._store._documents # same state, conflicting writes + world.engine._store = store + world.host.accounts["acct-1"] = AuthorizedAccount( + "acct-1", timezone="UTC", authorized_at=datetime(2026, 8, 1, tzinfo=timezone.utc) + ) + await world.engine.observe_grant( + IDENTITY, + "acct-1", + allowed_tasks=["get_reporting_status"], # a change, so the write is attempted + authorized_at=datetime(2026, 8, 1, tzinfo=timezone.utc), + ) + assert store.attempts == 3 # the change ran three times + assert world.engine.authorized_at_regressions == 1