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
12 changes: 12 additions & 0 deletions docs/reliable-reporting-service.md
Original file line number Diff line number Diff line change
Expand Up @@ -338,3 +338,15 @@ exact-read assertions after service recreation or durable recovery.

See [`examples/reliable_reporting_adapters.py`](../examples/reliable_reporting_adapters.py)
for compact GAM-like and FreeWheel-like adapter definitions.

Installed Core and managed production reporting use one error policy at service
admission. A created, stopping, closed, or failed service returns
`SERVICE_UNAVAILABLE` with transient recovery. A missing authenticated caller
returns `AUTH_REQUIRED`; explicit buyer configuration validation can raise
`ReliableReportingConfigurationError(message, kind="invalid_request")`.
Composition errors retain `INTERNAL_ERROR` and terminal recovery, with internal
configuration details omitted from buyer responses. Ledger codes are preserved
and receive the SDK's recovery classification for that code. Reporting cursor/
input extensions are correctable, storage-unavailability extensions are transient,
and unknown extension or integrity failures default to terminal. Disabled optional methods and aggregate delivery
continue to delegate to the application.
130 changes: 130 additions & 0 deletions src/adcp/reporting/_errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
"""One buyer error policy for installed Core and managed reporting routes."""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from typing import Literal, TypeVar

from adcp.exceptions import ADCPTaskError
from adcp.reporting.ledger.store import LedgerConflictError
from adcp.reporting.service_lifecycle import ReliableReportingUnavailableError
from adcp.server.helpers import STANDARD_ERROR_CODES, adcp_error
from adcp.types import Error

_T = TypeVar("_T")

# Reporting extensions have local meanings. Normative SDK codes always take
# precedence; unlisted extension/invariant failures retain terminal recovery.
_REPORTING_CORRECTABLE = frozenset(
{
"CURSOR_REVISION_MISMATCH",
"CURSOR_SNAPSHOT_MISMATCH",
"INVALID_CHECKPOINT",
"INVALID_CURSOR",
"INVALID_PAGE_SIZE",
"INVALID_PERIOD",
"INVALID_STATUS_PERIOD",
"INVALID_STATUS_TIME",
"INVALID_VIEW",
"MISSING_REVISION_ID",
"EMPTY_BATCH",
"UNKNOWN_CONFIGURATION_GENERATION",
"CONFIGURATION_GENERATION_IMMUTABLE",
"CONFIGURATION_GENERATION_MISMATCH",
"REPORT_DEFINITION_MISMATCH",
"OBLIGATION_IDENTITY_MISMATCH",
"SELLER_SNAPSHOT_EVIDENCE_INCOMPLETE",
"STATUS_NOT_DUE",
"STATUS_SUPERSEDES_REQUIRED",
"STATUS_SUPERSEDES_STALE",
"REVISION_NOT_CURRENTLY_REQUIRED",
"REPORTING_FEED_VERSION_MISMATCH",
}
)
_REPORTING_TRANSIENT = frozenset(
{
"REPORTING_CONFIGURATION_UNAVAILABLE",
"REPORTING_STATUS_UNAVAILABLE",
"REPORTING_CONTENT_UNAVAILABLE",
"REPORTING_FEED_STORAGE_UNAVAILABLE",
"RECEIPT_STORAGE_UNAVAILABLE",
}
)


class ReliableReportingConfigurationError(ValueError):
"""A composition failure, missing caller, or invalid buyer configuration.

Composition failures default to terminal INTERNAL_ERROR. Only explicit
request-time failures may ask the buyer to correct authentication/input.
"""

def __init__(
self,
message: str,
*,
kind: Literal["misconfiguration", "auth_required", "invalid_request"] = "misconfiguration",
) -> None:
super().__init__(message)
self.kind = kind


def reporting_error(code: str, message: str) -> Error:
"""Preserve the domain code and use the SDK's normative recovery defaults."""
recovery = None
if code not in STANDARD_ERROR_CODES:
if code in _REPORTING_CORRECTABLE:
recovery = "correctable"
elif code in _REPORTING_TRANSIENT:
recovery = "transient"
return Error.model_validate(adcp_error(code, message, recovery=recovery)["errors"][0])


async def installed_reporting_call(
task: str, call: Callable[[], Awaitable[_T]], *, decisioning: bool = False
) -> _T:
"""Translate at admission, including lifecycle refusals before handler entry."""
try:
return await call()
except ReliableReportingUnavailableError as error:
wire = reporting_error(
"SERVICE_UNAVAILABLE", "Reporting service is temporarily unavailable"
)
cause: Exception = error
except LedgerConflictError as error:
wire = reporting_error(error.code, str(error))
cause = error
except ReliableReportingConfigurationError as error:
code, message = {
"misconfiguration": ("INTERNAL_ERROR", "Reporting service configuration failed"),
"auth_required": ("AUTH_REQUIRED", "Authenticated reporting caller is required"),
"invalid_request": ("INVALID_REQUEST", str(error)),
}[error.kind]
wire = reporting_error(code, message)
cause = error
except ADCPTaskError as error:
# Existing production routes already redact ACL/storage errors. Apply
# the same code policy without dropping their structured metadata.
errors = [
(
Error.model_validate(
{
**item.model_dump(exclude_none=True),
"recovery": reporting_error(item.code, item.message).recovery,
}
)
if isinstance(item, Error)
else item
)
for item in error.errors
]
if not decisioning:
raise ADCPTaskError(operation=task, errors=errors) from error
wire = errors[0]
cause = error
if decisioning:
from adcp.decisioning import AdcpError

assert wire.recovery is not None
raise AdcpError(wire.code, message=wire.message, recovery=wire.recovery.value) from cause
raise ADCPTaskError(operation=task, errors=[wire]) from cause
76 changes: 72 additions & 4 deletions src/adcp/reporting/production/handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,11 @@

from adcp.decisioning.context import RequestContext
from adcp.exceptions import ADCPTaskError
from adcp.reporting._errors import (
ReliableReportingConfigurationError,
installed_reporting_call,
reporting_error,
)
from adcp.reporting.canonical_json import canonical_json_utf8_v1
from adcp.reporting.ledger.consumer_status import ConsumerStatusIngest
from adcp.reporting.ledger.delivery_models import ReportingDeliveryPrincipal
Expand All @@ -21,11 +26,14 @@
ReportingReceiptHandler,
_consumer,
)
from adcp.reporting.service_lifecycle import (
ReliableReportingState,
ReliableReportingUnavailableError,
)
from adcp.server.base import ADCPHandler, NotImplementedResponse, ToolContext
from adcp.server.mcp_tools import ADCP_TOOL_DEFINITIONS, get_tools_for_handler
from adcp.server.responses import capabilities_response
from adcp.types import (
Error,
GetAdcpCapabilitiesRequest,
GetMediaBuyDeliveryRequest,
GetReportingStatusRequest,
Expand All @@ -48,7 +56,7 @@ def _request(value: Any) -> dict[str, Any]:


def _task_error(task: str, code: str, message: str) -> ADCPTaskError:
return ADCPTaskError(operation=task, errors=[Error(code=code, message=message)])
return ADCPTaskError(operation=task, errors=[reporting_error(code, message)])


_Method = TypeVar("_Method", bound=Callable[..., Any])
Expand Down Expand Up @@ -76,9 +84,61 @@ async def call(self: ReportingProductionHandler, *args: Any, **kwargs: Any) -> A
method.__name__ == "get_media_buy_delivery"
and "reporting_revision_id" not in _request(args[0] if args else kwargs["params"])
)
if lifecycle is None or aggregate:
if aggregate:
return await method(self, *args, **kwargs)

async def invoke() -> Any:
support = self.production
if support._closed or support._failed or support._task is None or support._task.done():
state = (
ReliableReportingState.CLOSED
if support._closed
else (
ReliableReportingState.FAILED
if support._failed
else ReliableReportingState.NEW
)
)
raise ReliableReportingUnavailableError(state)
try:
support._assert_components()
except ReportingNotificationError as error:
raise ReliableReportingConfigurationError(
"production composition failed"
) from error
context = args[1] if len(args) > 1 else kwargs.get("context")
# A RequestContext's caller_identity is an account cache key. Only
# transport auth/registry evidence establishes the reporting caller.
present = context is not None and (
any(
getattr(context, name, None) is not None
for name in (
"auth_info",
"auth_principal",
"buyer_agent",
)
)
or (
not isinstance(context, RequestContext)
and getattr(context, "caller_identity", None) is not None
)
or any(
context.metadata.get(key) is not None
for key in (
"adcp.auth_info",
"auth_info",
"adcp.buyer_agent",
)
)
)
if not present:
raise ReliableReportingConfigurationError("missing caller", kind="auth_required")
return await method(self, *args, **kwargs)
return await lifecycle.call(lambda: method(self, *args, **kwargs))

return await installed_reporting_call(
method.__name__,
lambda: invoke() if lifecycle is None else lifecycle.call(invoke),
)

return cast(_Method, call)

Expand Down Expand Up @@ -184,6 +244,8 @@ async def _authorize(
if isinstance(context, RequestContext) and context.account.id != account:
raise ReportingReceiptError("UNAUTHORIZED")
return ReportingDeliveryPrincipal(account, consumer)
except (ReliableReportingConfigurationError, ReliableReportingUnavailableError):
raise
except Exception:
raise ReportingReceiptError("UNAUTHORIZED") from None

Expand Down Expand Up @@ -257,6 +319,8 @@ async def sync_accounts(
code, message = error.code, str(error)
except ReportingReceiptError as error:
code, message = error.code, str(error)
except (ReliableReportingConfigurationError, ReliableReportingUnavailableError):
raise
except Exception:
code, message = (
"REPORTING_CONFIGURATION_UNAVAILABLE",
Expand Down Expand Up @@ -285,6 +349,8 @@ async def sync_reporting_status(
return result
except (ReportingReceiptError, LedgerConflictError) as error:
code, message = error.code, str(error)
except (ReliableReportingConfigurationError, ReliableReportingUnavailableError):
raise
except Exception:
code, message = "REPORTING_STATUS_UNAVAILABLE", "reporting status is unavailable"
raise _task_error("sync_reporting_status", code, message)
Expand Down Expand Up @@ -412,6 +478,8 @@ async def get_media_buy_delivery(
raise
except (ReportingReceiptError, LedgerConflictError) as error:
code, message = error.code, str(error)
except (ReliableReportingConfigurationError, ReliableReportingUnavailableError):
raise
except Exception:
code, message = "REPORTING_CONTENT_UNAVAILABLE", "reporting content is unavailable"
raise _task_error("get_media_buy_delivery", code, message)
Expand Down
41 changes: 31 additions & 10 deletions src/adcp/reporting/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,10 @@

from pydantic import BaseModel

from adcp.reporting._errors import (
ReliableReportingConfigurationError,
installed_reporting_call,
)
from adcp.reporting._source_authorization import source_turn
from adcp.reporting.inline_source import (
InlineFetchResult,
Expand Down Expand Up @@ -92,10 +96,6 @@
logger = logging.getLogger(__name__)


class ReliableReportingConfigurationError(ValueError):
"""The requested service composition cannot make truthful guarantees."""


@runtime_checkable
class ReportingAdapter(Protocol):
"""The minimum source adapter: declaration plus one normalized slice read.
Expand Down Expand Up @@ -1043,7 +1043,8 @@ def _default_caller(request: Any, context: Any | None) -> ReportingStatusCaller:
"default caller resolution requires AccountAwareToolContext.account_id and "
"authenticated caller_identity, or RequestContext.account.id and an "
"authenticated current_principal or verified auth_principal; "
"provide caller_resolver for another trust model"
"provide caller_resolver for another trust model",
kind="auth_required" if not consumer_id else "misconfiguration",
)
return ReportingStatusCaller(account_id=account_id, consumer_id=consumer_id)

Expand Down Expand Up @@ -1268,26 +1269,42 @@ def install(self, platform: Any) -> Any:

class ReportingInstallMixin:
async def get_reporting_status(self, params: Any, context: Any | None = None) -> Any:
return await service.get_reporting_status(params, context)
return await installed_reporting_call(
"get_reporting_status",
lambda: service.get_reporting_status(params, context),
decisioning=decisioning,
)

async def sync_reporting_status(self, params: Any, context: Any | None = None) -> Any:
if not service._consumer_status_enabled:
return await _resolve(
getattr(super(), "sync_reporting_status")(params, context)
)
return await service.sync_reporting_status(params, context)
return await installed_reporting_call(
"sync_reporting_status",
lambda: service.sync_reporting_status(params, context),
decisioning=decisioning,
)

async def sync_reporting_receipts(self, params: Any, context: Any | None = None) -> Any:
if service._receipt_handler is None:
return await _resolve(
getattr(super(), "sync_reporting_receipts")(params, context)
)
return await service.sync_reporting_receipts(params, context)
return await installed_reporting_call(
"sync_reporting_receipts",
lambda: service.sync_reporting_receipts(params, context),
decisioning=decisioning,
)

async def _get_reporting_revision_content(
self, params: Any, context: Any | None = None
) -> Any:
return await service.get_revision_content(params, context)
return await installed_reporting_call(
"get_media_buy_delivery",
lambda: service.get_revision_content(params, context),
decisioning=decisioning,
)

attrs: dict[str, Any] = {"__module__": original.__module__}
if not decisioning:
Expand All @@ -1305,7 +1322,11 @@ async def get_media_buy_delivery(
return await _resolve(
original.get_media_buy_delivery(instance, params, context)
)
return await service.get_revision_content(params, context)
return await installed_reporting_call(
"get_media_buy_delivery",
lambda: service.get_revision_content(params, context),
decisioning=decisioning,
)

attrs.update(
advertised_tools=tools,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -411,7 +411,6 @@ async def broken(*args, **kwargs):
async def test_service_failure_stops_admission_but_preserves_ordinary_delegate(tmp_path):
from adcp.reporting.service import (
ReliableReportingServiceError,
ReliableReportingUnavailableError,
)

async with production_harness(
Expand All @@ -427,8 +426,12 @@ async def test_service_failure_stops_admission_but_preserves_ordinary_delegate(t
with pytest.raises(ReliableReportingServiceError):
await asyncio.wait_for(h.service.wait(), 2)
assert not h.service.ready
with pytest.raises(ReliableReportingUnavailableError):
from adcp.exceptions import ADCPTaskError

with pytest.raises(ADCPTaskError) as unavailable:
await h.production.handler.sync_accounts({})
assert unavailable.value.error_codes == ["SERVICE_UNAVAILABLE"]
assert unavailable.value.error_info[0].recovery == "transient"
assert await h.production.handler.get_products({}) == {"products": []}
assert (await h.production.handler.get_media_buy_delivery({}))["aggregated_totals"][
"impressions"
Expand Down
Loading
Loading