Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions src/adcp/reporting/subscriptions/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,8 @@
)
from adcp.reporting.subscriptions.consumer import (
REPORTING_SUBSCRIPTION_CHANGED,
ReportingSubscriptionClientError,
REPORTING_SUBSCRIPTION_ERROR_CODES,
ReportingSubscriptionError,
ReportingSubscriptionReadback,
ReportingSubscriptionTracker,
is_subscription_change,
Expand Down Expand Up @@ -79,7 +80,7 @@
subscription_obligation_id,
)
from adcp.reporting.subscriptions.registry import (
ReportingSubscriptionError,
ReportingSubscriptionValidationError,
apply_reporting_subscriptions,
refresh_destination_states,
render_path_template,
Expand Down Expand Up @@ -107,7 +108,6 @@
"SubscriptionPlanResult",
"subscription_obligation_id",
"REPORTING_SUBSCRIPTION_CHANGED",
"ReportingSubscriptionClientError",
"ReportingSubscriptionReadback",
"ReportingSubscriptionTracker",
"is_subscription_change",
Expand All @@ -120,7 +120,9 @@
"InMemoryReportingSubscriptionStore",
"ReportingSubscriptionDeliveryHost",
"ReportingSubscriptionEngine",
"REPORTING_SUBSCRIPTION_ERROR_CODES",
"ReportingSubscriptionError",
"ReportingSubscriptionValidationError",
"ReportingSubscriptionHost",
"ReportingSubscriptionPrecedenceHost",
"ReportingSubscriptionStateConflictError",
Expand Down Expand Up @@ -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}")
93 changes: 85 additions & 8 deletions src/adcp/reporting/subscriptions/consumer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Medium: dispatched = False keys solely on the code string, so it cannot tell an SDK-raised local refusal from a seller refusal that happens to carry the same §2.12 code. _readback (L190-194) feeds the seller's wire code straight into this constructor. A conformant counterpart that refuses a dispatched write with SUBSCRIPTION_ID_CLASH/DESTINATION_WOULD_REVOKE/SUBSCRIPTION_WOULD_DEACTIVATE/SUBSCRIPTION_OPERATOR_SCOPE_REQUIRED/SUBSCRIPTION_SELECTOR_UNMATCHED lands here, gets stamped dispatched=False, and is_pre_dispatch_error returns True — contradicting the PR's own invariant "a seller refusal is not pre-dispatch" (asserted in test_rejection_surfaces_the_seller_code, which only exercises INVALID_REQUEST). The known-branch also discards the wire recovery for these codes. This SDK's seller emits INVALID_REQUEST/UNSUPPORTED_FEATURE today, so it won't bite in-tree, but the §2.12 vocabulary is a wire vocabulary and the readback path has no test with a §2.12 code on the seller side.

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):
Expand Down Expand Up @@ -105,17 +179,19 @@ 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
kind = str(getattr(outcome, "kind", ""))
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 ())
Expand Down Expand Up @@ -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 ())
Expand Down Expand Up @@ -493,7 +569,8 @@ def expected_periods(
__all__ = [
"REPORTING_SUBSCRIPTION_CHANGED",
"PrincipalClient",
"ReportingSubscriptionClientError",
"REPORTING_SUBSCRIPTION_ERROR_CODES",
"ReportingSubscriptionError",
"ReportingSubscriptionReadback",
"ReportingSubscriptionTracker",
"SubscriptionStatusClient",
Expand Down
52 changes: 28 additions & 24 deletions src/adcp/reporting/subscriptions/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -111,79 +111,83 @@ 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"
)
scope = configuration.get("scope") or {}
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",
)
selector = configuration["selector"]
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"
)
template = destination.get("path_template")
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"
)
Expand All @@ -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"
)
Expand All @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"
)
Expand All @@ -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"
)
Expand Down Expand Up @@ -481,7 +485,7 @@ def refresh_destination_states(


__all__ = [
"ReportingSubscriptionError",
"ReportingSubscriptionValidationError",
"apply_reporting_subscriptions",
"refresh_destination_states",
"render_path_template",
Expand Down
Loading
Loading