From b393dce18ce5433f39cf87019233c71877930fc0 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 18:03:23 +0000 Subject: [PATCH] feat(reporting)!: rename the reporting-subscription errors to the cross-SDK names The buyer and seller errors swap to the names TypeScript uses (DX plan section 2.12), before beta.3 freezes them. - Seller: ReportingSubscriptionError(ValueError), the sync_principal validation error, is renamed ReportingSubscriptionValidationError. - Buyer: ReportingSubscriptionClientError(RuntimeError) becomes ReportingSubscriptionError(ADCPError). It carries code, recovery, dispatched, would_deactivate and readback. The SDK codes and their classification are published as REPORTING_SUBSCRIPTION_ERROR_CODES: SUBSCRIPTION_WOULD_DEACTIVATE, DESTINATION_WOULD_REVOKE, SUBSCRIPTION_ID_CLASH, SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED and SUBSCRIPTION_SELECTOR_UNMATCHED are correctable pre-dispatch refusals; SUBSCRIPTION_CONFLICT_EXHAUSTED is transient and SUBSCRIPTION_READBACK_MISMATCH terminal. A seller refusal keeps its wire code and takes the wire recovery, else the code table. - ReportingSubscriptionClientError remains a deprecated alias for one beta; reading it warns. BREAKING CHANGE: code that caught the seller's ReportingSubscriptionError must catch ReportingSubscriptionValidationError. The buyer error is no longer a RuntimeError; catch ReportingSubscriptionError or ADCPError. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/__init__.py | 16 +++- src/adcp/reporting/subscriptions/consumer.py | 93 +++++++++++++++++-- src/adcp/reporting/subscriptions/registry.py | 52 ++++++----- src/adcp/server/principal.py | 10 +- tests/test_reporting_subscription_consumer.py | 35 ++++++- 5 files changed, 164 insertions(+), 42 deletions(-) diff --git a/src/adcp/reporting/subscriptions/__init__.py b/src/adcp/reporting/subscriptions/__init__.py index 2ace83dcc..6e0488b1b 100644 --- a/src/adcp/reporting/subscriptions/__init__.py +++ b/src/adcp/reporting/subscriptions/__init__.py @@ -35,7 +35,8 @@ ) from adcp.reporting.subscriptions.consumer import ( REPORTING_SUBSCRIPTION_CHANGED, - ReportingSubscriptionClientError, + REPORTING_SUBSCRIPTION_ERROR_CODES, + ReportingSubscriptionError, ReportingSubscriptionReadback, ReportingSubscriptionTracker, is_subscription_change, @@ -79,7 +80,7 @@ subscription_obligation_id, ) from adcp.reporting.subscriptions.registry import ( - ReportingSubscriptionError, + ReportingSubscriptionValidationError, apply_reporting_subscriptions, refresh_destination_states, render_path_template, @@ -107,7 +108,6 @@ "SubscriptionPlanResult", "subscription_obligation_id", "REPORTING_SUBSCRIPTION_CHANGED", - "ReportingSubscriptionClientError", "ReportingSubscriptionReadback", "ReportingSubscriptionTracker", "is_subscription_change", @@ -120,7 +120,9 @@ "InMemoryReportingSubscriptionStore", "ReportingSubscriptionDeliveryHost", "ReportingSubscriptionEngine", + "REPORTING_SUBSCRIPTION_ERROR_CODES", "ReportingSubscriptionError", + "ReportingSubscriptionValidationError", "ReportingSubscriptionHost", "ReportingSubscriptionPrecedenceHost", "ReportingSubscriptionStateConflictError", @@ -163,3 +165,11 @@ "project_subscription_state", "serves_settled_subscription_fields", ] + + +def __getattr__(name: str) -> object: + if name == "ReportingSubscriptionClientError": + from adcp.reporting.subscriptions import consumer + + return getattr(consumer, name) # warns: renamed ReportingSubscriptionError + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/adcp/reporting/subscriptions/consumer.py b/src/adcp/reporting/subscriptions/consumer.py index a1d46bfb0..8adc40edc 100644 --- a/src/adcp/reporting/subscriptions/consumer.py +++ b/src/adcp/reporting/subscriptions/consumer.py @@ -28,6 +28,7 @@ from typing import Any, Protocol from uuid import uuid4 +from adcp.exceptions import ADCPError, RecoveryLiteral, _normalize_recovery from adcp.reporting.subscriptions._clock import require_zone, utc_instant from adcp.reporting.subscriptions.coverage import ( AccountTimezoneChange, @@ -56,12 +57,85 @@ REPORTING_SUBSCRIPTION_CHANGED = "reporting_subscription_changed" -class ReportingSubscriptionClientError(RuntimeError): - """The seller refused or failed a subscription read or write.""" +#: The SDK's own reporting-subscription error codes, with their recovery and +#: whether the SDK refuses them before sending anything (DX plan section 2.12). +REPORTING_SUBSCRIPTION_ERROR_CODES: Mapping[str, tuple[RecoveryLiteral, bool]] = { + "SUBSCRIPTION_WOULD_DEACTIVATE": ("correctable", True), + "DESTINATION_WOULD_REVOKE": ("correctable", True), + "SUBSCRIPTION_ID_CLASH": ("correctable", True), + "SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED": ("correctable", True), + "SUBSCRIPTION_SELECTOR_UNMATCHED": ("correctable", True), + "SUBSCRIPTION_CONFLICT_EXHAUSTED": ("transient", False), + "SUBSCRIPTION_READBACK_MISMATCH": ("terminal", False), +} + + +class ReportingSubscriptionError(ADCPError): + """A reporting-subscription read or write was refused or failed (buyer side). + + Attributes: + code: One of :data:`REPORTING_SUBSCRIPTION_ERROR_CODES`, or the + seller's wire code (``INVALID_REQUEST``, ``CONFLICT``, ...). + ``None`` when the seller gave none. + recovery: ``"transient"``, ``"correctable"`` or ``"terminal"``: the + SDK code's classification, else the seller's wire ``recovery``, + else the code table. + dispatched: ``False`` when the SDK refused the write before sending + it. See :func:`adcp.exceptions.is_pre_dispatch_error`. + would_deactivate: The subscription ids a refused write would have + deactivated (``SUBSCRIPTION_WOULD_DEACTIVATE``). + readback: The seller's readback (``SUBSCRIPTION_READBACK_MISMATCH``). + + The message ends with the fix. + """ - def __init__(self, message: str, *, code: str | None = None) -> None: + def __init__( + self, + message: str, + *, + code: str | None = None, + recovery: RecoveryLiteral | None = None, + would_deactivate: Sequence[str] = (), + readback: ReportingSubscriptionReadback | None = None, + ) -> None: super().__init__(message) self.code = code + known = REPORTING_SUBSCRIPTION_ERROR_CODES.get(code or "") + if known is not None: + self.recovery = known[0] + if known[1]: + self.dispatched = False + else: + self.recovery = recovery or _code_recovery(code) + self.would_deactivate = tuple(would_deactivate) + self.readback = readback + + +def _code_recovery(code: str | None) -> RecoveryLiteral | None: + """A wire code's registered classification; an unknown code is transient.""" + if not code: + return None + from adcp.server.helpers import CORRECTABLE_CODES, TERMINAL_CODES + + if code in TERMINAL_CODES: + return "terminal" + if code in CORRECTABLE_CODES: + return "correctable" + return "transient" + + +def __getattr__(name: str) -> Any: + if name == "ReportingSubscriptionClientError": + import warnings + + warnings.warn( + "ReportingSubscriptionClientError is renamed ReportingSubscriptionError, " + "now an ADCPError; the old name is removed in the next beta", + DeprecationWarning, + stacklevel=2, + ) + return ReportingSubscriptionError + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") class PrincipalClient(Protocol): @@ -105,7 +179,7 @@ def _dump(value: Any) -> dict[str, Any]: def _readback(result: Any, *, task: str) -> ReportingSubscriptionReadback: response = getattr(result, "data", None) if not getattr(result, "success", False) or response is None: - raise ReportingSubscriptionClientError( + raise ReportingSubscriptionError( f"{task} failed: {getattr(result, 'error', None) or 'no response'}" ) outcome = response.result @@ -113,9 +187,11 @@ def _readback(result: Any, *, task: str) -> ReportingSubscriptionReadback: if kind == "failed": errors = list(getattr(outcome, "errors", None) or []) first = errors[0] if errors else None - raise ReportingSubscriptionClientError( + recovery = getattr(first, "recovery", None) + raise ReportingSubscriptionError( f"{task} was rejected: {getattr(first, 'message', 'unknown error')}", - code=None if first is None else str(getattr(first, "code", "")), + code=None if first is None else (str(getattr(first, "code", "")) or None), + recovery=_normalize_recovery(getattr(recovery, "value", recovery)), ) configuration = getattr(outcome, "configuration", None) states = tuple(getattr(configuration, "reporting_subscriptions", None) or ()) @@ -186,7 +262,7 @@ async def load_subscription_coverage( result = await client.get_reporting_status(request) response = getattr(result, "data", None) if not getattr(result, "success", False) or response is None: - raise ReportingSubscriptionClientError( + raise ReportingSubscriptionError( f"get_reporting_status failed: {getattr(result, 'error', None) or 'no response'}" ) return tuple(getattr(response, "subscription_coverage", None) or ()) @@ -493,7 +569,8 @@ def expected_periods( __all__ = [ "REPORTING_SUBSCRIPTION_CHANGED", "PrincipalClient", - "ReportingSubscriptionClientError", + "REPORTING_SUBSCRIPTION_ERROR_CODES", + "ReportingSubscriptionError", "ReportingSubscriptionReadback", "ReportingSubscriptionTracker", "SubscriptionStatusClient", diff --git a/src/adcp/reporting/subscriptions/registry.py b/src/adcp/reporting/subscriptions/registry.py index 7da09cdd8..9eb800951 100644 --- a/src/adcp/reporting/subscriptions/registry.py +++ b/src/adcp/reporting/subscriptions/registry.py @@ -3,7 +3,7 @@ The section is complete desired state, like the principal's other sections. :func:`apply_reporting_subscriptions` turns the caller's desired list and the previously stored states into the new ``core/reporting-subscription-state.json`` -states, or raises :class:`ReportingSubscriptionError` so the whole request +states, or raises :class:`ReportingSubscriptionValidationError` so the whole request fails. It enforces what the generated model cannot express (the schema's ``allOf`` conditionals and ``x-adcp-validation`` rules): @@ -49,7 +49,7 @@ _IN_PLACE_FIELDS = frozenset({"active", "replace_account_configs", "subscription_version"}) -class ReportingSubscriptionError(ValueError): +class ReportingSubscriptionValidationError(ValueError): """The section is invalid; the whole ``sync_principal`` request fails.""" def __init__(self, message: str, *, code: str = "INVALID_REQUEST") -> None: @@ -111,7 +111,7 @@ def _validate_one( where = f"reporting_subscriptions[{index}]" subscription_id = configuration["subscription_id"] if subscription_id in delivery_config_ids: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.subscription_id equals a delivery_config_id this caller uses; " "subscription_id and delivery_config_id share one namespace" ) @@ -119,19 +119,21 @@ def _validate_one( has_all = "all_authorized_accounts" in scope has_ids = "account_ids" in scope if has_all == has_ids: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.scope must be exactly one of all_authorized_accounts or account_ids" ) if has_all and scope["all_authorized_accounts"] is not True: - raise ReportingSubscriptionError(f"{where}.scope.all_authorized_accounts must be true") + raise ReportingSubscriptionValidationError( + f"{where}.scope.all_authorized_accounts must be true" + ) if configuration.get("exclude_account_ids") and not has_all: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.exclude_account_ids is valid only with all_authorized_accounts scope" ) if configuration.get("replace_account_configs") is True: # Not applied yet: accepting it would leave the caller's account-level # configurations in force while it believes they were cut over. - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.replace_account_configs is not supported by this seller", code="UNSUPPORTED_FEATURE", ) @@ -139,28 +141,30 @@ def _validate_one( period_timezone = selector.get("period_timezone") if period_timezone is not None: if not has_ids or len(scope["account_ids"]) != 1: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.selector.period_timezone requires account_ids naming exactly one account" ) try: require_zone(period_timezone, field="period_timezone") except ValueError as error: - raise ReportingSubscriptionError(f"{where}.selector.{error}") from None + raise ReportingSubscriptionValidationError(f"{where}.selector.{error}") from None mode = configuration["reconciliation_mode"] if selector["feed_purpose"] == "billing" and ( selector["required_finality"] != "official" or mode != "consumer_receipt" ): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}: billing subscriptions require official finality and consumer_receipt" ) destination = configuration.get("destination") if mode == "consumer_receipt" and destination is None: - raise ReportingSubscriptionError(f"{where}: consumer_receipt requires a destination") + raise ReportingSubscriptionValidationError( + f"{where}: consumer_receipt requires a destination" + ) if destination is None: return target = destinations.get(destination["destination_id"]) if target is None: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.destination.destination_id {destination['destination_id']!r} is not in " "reporting_destinations after this request" ) @@ -168,22 +172,22 @@ def _validate_one( if template is not None: unknown = {name for name in _PLACEHOLDER.findall(template)} - _ALLOWED_PLACEHOLDERS if unknown: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.destination.path_template uses unknown placeholders {sorted(unknown)}" ) if "{" in _PLACEHOLDER.sub("", template) or "}" in _PLACEHOLDER.sub("", template): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.destination.path_template has braces outside a placeholder" ) if _multi_account(configuration): pattern = (target.get("configuration") or target).get("pattern") if template is None and pattern == "file_transfer": - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.destination.path_template is required for a multi-account " "file_transfer subscription" ) if template is not None and "{account_id}" not in template: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"{where}.destination.path_template must contain {{account_id}} when the " "scope can cover more than one account" ) @@ -200,12 +204,12 @@ def _validate_set(active: Sequence[dict[str, Any]], destinations: Mapping[str, A continue templates = [(item.get("destination") or {}).get("path_template") for item in sharing] if any(template is None or "{account_id}" not in template for template in templates): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"active subscriptions sharing destination {destination_id!r} must each have a " "path_template containing {account_id}" ) if len(set(templates)) != len(templates): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"active subscriptions sharing destination {destination_id!r} must use " "different path_template values" ) @@ -214,7 +218,7 @@ def _validate_set(active: Sequence[dict[str, Any]], destinations: Mapping[str, A if first["selector"]["feed_purpose"] != second["selector"]["feed_purpose"]: continue if _same_level_overlap(first, second): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"active subscriptions {first['subscription_id']!r} and " f"{second['subscription_id']!r} would both apply to one account and " "feed_purpose at the same scope level" @@ -268,7 +272,7 @@ def apply_reporting_subscriptions( for index, configuration in enumerate(configurations): subscription_id = configuration["subscription_id"] if subscription_id in seen: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"reporting_subscriptions[{index}].subscription_id duplicates an earlier entry" ) seen.add(subscription_id) @@ -314,7 +318,7 @@ def _deactivated(old: dict[str, Any], *, stamp: str) -> ReportingSubscriptionSta def _intervals(intervals: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: """Every interval, or an error: dropping one would make its periods owed again.""" if len(intervals) > _MAX_INACTIVE_INTERVALS: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"the subscription has reached {_MAX_INACTIVE_INTERVALS} inactive intervals, the " "readback's limit; register a new subscription_id instead of deactivating again" ) @@ -340,13 +344,13 @@ def _next_state( else: old_version = int(old["subscription_version"]) if version < old_version: - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"subscription {configuration['subscription_id']!r} version {version} is older " f"than the accepted version {old_version}" ) if version == old_version: if _semantic(configuration) != _semantic(old["configuration"]): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"subscription {configuration['subscription_id']!r} reuses version " f"{version} with different semantic content; increment subscription_version" ) @@ -481,7 +485,7 @@ def refresh_destination_states( __all__ = [ - "ReportingSubscriptionError", + "ReportingSubscriptionValidationError", "apply_reporting_subscriptions", "refresh_destination_states", "render_path_template", diff --git a/src/adcp/server/principal.py b/src/adcp/server/principal.py index 9ee679afa..bd162b7cc 100644 --- a/src/adcp/server/principal.py +++ b/src/adcp/server/principal.py @@ -677,7 +677,9 @@ async def _sync_principal( action = ( "would_clear" if changed and cleared - else "would_update" if changed else "would_be_unchanged" + else "would_update" + if changed + else "would_be_unchanged" ) return SyncPrincipalResponse.model_validate( { @@ -996,7 +998,7 @@ async def _replace_subscriptions( ) -> None: """Apply ``reporting_subscriptions`` against the destinations after this request.""" from adcp.reporting.subscriptions.registry import ( - ReportingSubscriptionError, + ReportingSubscriptionValidationError, apply_reporting_subscriptions, refresh_destination_states, ) @@ -1023,13 +1025,13 @@ async def _replace_subscriptions( and configuration.destination is not None and configuration.destination.destination_id not in names ): - raise ReportingSubscriptionError( + raise ReportingSubscriptionValidationError( f"subscription {item.subscription_id!r} names destination " f"{configuration.destination.destination_id!r}, which this " "request removes" ) payload["reporting_subscriptions"] = refresh_destination_states(prior, destinations) - except ReportingSubscriptionError as error: + except ReportingSubscriptionValidationError as error: raise _PrincipalValidationError(error.code, str(error)) from None def _now(self) -> datetime: diff --git a/tests/test_reporting_subscription_consumer.py b/tests/test_reporting_subscription_consumer.py index c25a2e591..b5d3211f2 100644 --- a/tests/test_reporting_subscription_consumer.py +++ b/tests/test_reporting_subscription_consumer.py @@ -11,12 +11,14 @@ import pytest +from adcp.exceptions import ADCPError, is_pre_dispatch_error from adcp.reporting import plan_consumer_statuses from adcp.reporting.subscriptions import ( + REPORTING_SUBSCRIPTION_ERROR_CODES, AuthorizedAccount, AuthorizedAccountPage, - ReportingSubscriptionClientError, ReportingSubscriptionEngine, + ReportingSubscriptionError, ReportingSubscriptionTracker, SubscriptionBuy, SubscriptionOffering, @@ -148,9 +150,36 @@ async def test_sync_and_read_back() -> None: @pytest.mark.asyncio async def test_rejection_surfaces_the_seller_code() -> None: client = Client(Seller()) - with pytest.raises(ReportingSubscriptionClientError) as raised: + with pytest.raises(ReportingSubscriptionError) as raised: await sync_reporting_subscriptions(client, [{**SUBSCRIPTION, "scope": {}}]) - assert raised.value.code == "INVALID_REQUEST" + error = raised.value + assert isinstance(error, ADCPError) + assert error.code == "INVALID_REQUEST" and error.recovery == "correctable" + # The seller refused it: the request was sent. + assert not is_pre_dispatch_error(error) + + +def test_sdk_codes_carry_their_recovery_and_dispatch() -> None: + for code, (recovery, local) in REPORTING_SUBSCRIPTION_ERROR_CODES.items(): + error = ReportingSubscriptionError("refused. Fix it.", code=code) + assert error.recovery == recovery + assert is_pre_dispatch_error(error) is local + assert ReportingSubscriptionError("x", code="X_VENDOR_NEW").recovery == "transient" + assert ReportingSubscriptionError("x", code="CONFLICT", recovery="terminal").recovery == ( + "terminal" + ) + guarded = ReportingSubscriptionError( + "x", code="SUBSCRIPTION_WOULD_DEACTIVATE", would_deactivate=["pacing"] + ) + assert guarded.would_deactivate == ("pacing",) + + +def test_the_old_client_error_name_is_a_deprecated_alias() -> None: + import adcp.reporting.subscriptions as subscriptions + + with pytest.warns(DeprecationWarning, match="renamed ReportingSubscriptionError"): + alias = subscriptions.ReportingSubscriptionClientError + assert alias is ReportingSubscriptionError def test_notification_filter() -> None: