From ebb794b5a5f83c9775a01b2722a9a4c5bc4d7982 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 22:44:18 +0000 Subject: [PATCH 1/2] fix(reporting): bound each seller read in reconcile and fleet reconcile reconcile(), explain(), poll()/events() and multi.reporting.reconcile() take read_timeout (timedelta, default 30 seconds, None for no bound; a non-positive or non-timedelta value is refused as INVALID_REQUEST before anything is read). It bounds each read the run makes (get_principal, get_adcp_capabilities, list_accounts, get_media_buys, get_reporting_status) and the receipts policy's sync_reporting_receipts write, mirroring the TS SDK's withReadTimeout and readTimeoutMilliseconds. A read that does not answer in time fails as TIMEOUT, naming read_timeout, and degrades the same way any other failed read does: a hung get_principal is SUBSCRIPTION_PRINCIPAL_UNAVAILABLE (a fleet issue for that seller), a hung account read fails that account. The read is cancelled and abandoned at the deadline, so a hung seller frees its fleet concurrency slot. Co-Authored-By: Claude Opus 5.5 --- docs/reporting-subscriptions-buyer.md | 4 +- docs/reporting-subscriptions-errors.md | 2 +- src/adcp/reporting/subscriptions/events.py | 8 +- src/adcp/reporting/subscriptions/facade.py | 33 ++++- src/adcp/reporting/subscriptions/reconcile.py | 102 ++++++++++++- tests/reporting_inprocess.py | 7 + .../test_reporting_subscription_reconcile.py | 137 ++++++++++++++++++ 7 files changed, 282 insertions(+), 11 deletions(-) diff --git a/docs/reporting-subscriptions-buyer.md b/docs/reporting-subscriptions-buyer.md index 1ff05c5f41..7543829f2a 100644 --- a/docs/reporting-subscriptions-buyer.md +++ b/docs/reporting-subscriptions-buyer.md @@ -57,7 +57,7 @@ for line in report.periods: print(line.account_id, line.period["start"], line.status, line.reason, line.message) ``` -`reconcile()` reads everything from the wire on each run. Never compute periods yourself. +`reconcile()` reads everything from the wire on each run. Never compute periods yourself. Each read is bounded by `read_timeout` (default 30 seconds, `None` for no bound), so one unresponsive seller cannot stall the report: a read that does not answer in time fails as `TIMEOUT` like any other failed read. - **Roster:** `list_accounts` ∩ your subscriptions' scopes, plus each subscription's `accounts_unauthorized[]`. There is no account list to keep. - **Grant history:** `authorization.authorized_at` and `regranted_at` from `list_accounts`, and `unauthorized_at` from `accounts_unauthorized[]`. You never record grants. - **Facts:** buys from `get_media_buys`, offerings from capabilities, and `subscription_coverage[]` and the obligations from a complete `get_reporting_status` periods walk. @@ -95,7 +95,7 @@ for issue in fleet.issues: # sellers that could not be reconciled at all print(issue.agent_id, issue.code, issue.message) ``` -`multi.reporting.reconcile()` runs `reconcile()` on every seller, `concurrency` at a time, with the same options. Each seller is read through its own `multi.agent(agent_id).reporting`. `fleet.summary` sums the sellers' summaries. A seller whose principal could not be read is in `fleet.issues` (`SUBSCRIPTION_PRINCIPAL_UNAVAILABLE`, with `agent_id`), not in `by_seller`. `fleet.definitive` is true only when every seller's report is definitive and no seller failed. `agent_ids=` narrows the run. +`multi.reporting.reconcile()` runs `reconcile()` on every seller, `concurrency` at a time, with the same options. Each seller is read through its own `multi.agent(agent_id).reporting`. `fleet.summary` sums the sellers' summaries. A seller whose principal could not be read is in `fleet.issues` (`SUBSCRIPTION_PRINCIPAL_UNAVAILABLE`, with `agent_id`), not in `by_seller`. `fleet.definitive` is true only when every seller's report is definitive and no seller failed. `agent_ids=` narrows the run. `read_timeout` bounds each read of each seller, so a hung seller lands in `fleet.issues` (or fails its accounts) and frees its `concurrency` slot instead of holding the run. From a shell: `python -m adcp reporting reconcile --agent [--since 2026-11-01] [--json]`. diff --git a/docs/reporting-subscriptions-errors.md b/docs/reporting-subscriptions-errors.md index 597ba42ff3..6ff20c3e2b 100644 --- a/docs/reporting-subscriptions-errors.md +++ b/docs/reporting-subscriptions-errors.md @@ -48,7 +48,7 @@ A seller rejection is a value: `AccountRejected`. Its `requires_new_account` is | `SUBSCRIPTION_ACCOUNT_NOT_COVERED` | an explicit `accounts=` id is not readable, or no subscription covers it | check `list_accounts` and the subscription scopes | | `COVERAGE_INPUT_INCOMPLETE`, `COVERAGE_INPUT_INVALID` | an owed set could not be computed, for example a missing account zone | the `detail` names the input | -An account an issue names is also `status: "failed"` in `report.accounts`, with `error.task` set to the read that failed (`get_media_buys`, `get_reporting_status`, `coverage`, ...). A principal or roster failure is the entry `"*"`. +An account an issue names is also `status: "failed"` in `report.accounts`, with `error.task` set to the read that failed (`get_media_buys`, `get_reporting_status`, `coverage`, ...). A principal or roster failure is the entry `"*"`. A read that did not answer within `read_timeout` has `error.code` `TIMEOUT` and a message naming `read_timeout`. `multi.reporting.reconcile()` lifts a seller whose principal could not be read, or whose run raised, into `fleet.issues` as `SUBSCRIPTION_PRINCIPAL_UNAVAILABLE` with its `agent_id`. Every other issue stays on that seller's report in `fleet.by_seller`. diff --git a/src/adcp/reporting/subscriptions/events.py b/src/adcp/reporting/subscriptions/events.py index 2dadd4e7ad..62ad02db5f 100644 --- a/src/adcp/reporting/subscriptions/events.py +++ b/src/adcp/reporting/subscriptions/events.py @@ -45,6 +45,7 @@ from adcp.reporting.subscriptions._clock import iso_z, utc_instant from adcp.reporting.subscriptions.reconcile import ( DEFAULT_GRACE, + DEFAULT_READ_TIMEOUT, BuyerReportingState, ReconcileReport, _reconcile, @@ -496,11 +497,12 @@ async def poll( grace: timedelta = DEFAULT_GRACE, baseline: bool = False, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> EventBatch: """One batch of events since the last acknowledgement; call ``batch.ack()`` after handling. ``baseline=True`` on a consumer with no snapshot yet records the - current state without emitting it. + current state without emitting it. ``read_timeout`` bounds each read. """ instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) report, context = await _reconcile( @@ -513,6 +515,7 @@ async def poll( include_not_owed=False, now=instant, receipts=None, + read_timeout=read_timeout, ) current = _snapshot(report, context) loaded = await self.store.load(self.agent_id) @@ -539,6 +542,7 @@ async def events( poll_every: timedelta = timedelta(minutes=5), stop: asyncio.Event | None = None, grace: timedelta = DEFAULT_GRACE, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> AsyncIterator[ReportingEvent]: """Every event, forever (or until ``stop`` is set). @@ -553,7 +557,7 @@ async def events( # Consume the ring before polling: one that lands during the # poll, the loop body or the ack survives and polls again at once. self._doorbell.clear() - batch = await self.poll(grace=grace) + batch = await self.poll(grace=grace, read_timeout=read_timeout) for event in batch.events: yield event try: diff --git a/src/adcp/reporting/subscriptions/facade.py b/src/adcp/reporting/subscriptions/facade.py index 832edf6b51..a2a4c4de4a 100644 --- a/src/adcp/reporting/subscriptions/facade.py +++ b/src/adcp/reporting/subscriptions/facade.py @@ -49,6 +49,9 @@ #: The default ``grace`` of :meth:`ReportingClient.reconcile`: one hour, never zero. DEFAULT_GRACE = timedelta(hours=1) +#: The default ``read_timeout`` of :meth:`ReportingClient.reconcile`: each +#: read of one seller is bounded to 30 seconds, as in the TS SDK. +DEFAULT_READ_TIMEOUT = timedelta(seconds=30) SubscribeAction = Literal["created", "updated", "unchanged", "reactivated", "deactivated"] @@ -443,6 +446,7 @@ async def reconcile( include_not_owed: bool = False, receipts: ReceiptPolicy | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> ReconcileReport: """What the seller owes every covered account, and whose move each period is. @@ -468,6 +472,11 @@ async def reconcile( The default is off: nothing is receipted on read. ``mode="verified"`` receipts a billing revision only after the SDK verified its materialization and your ``commit`` hook returned a reference. + :param read_timeout: bounds each read (and the receipts policy's write), + so one unresponsive seller cannot stall the report. A read that + does not answer in time fails as ``TIMEOUT``, naming + ``read_timeout``, the same way any other failed read does. Defaults + to 30 seconds; ``None`` for no bound. ``report.definitive`` is true only when every read completed; otherwise ``report.issues`` say what to retry. @@ -484,6 +493,7 @@ async def reconcile( include_not_owed=include_not_owed, receipts=receipts, now=now, + read_timeout=read_timeout, ) async def submit_receipts( @@ -532,6 +542,7 @@ async def explain( media_buy_id: str | None = None, offering_id: str | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> list[CoverageTrace]: """Why each selected period is, or is not, owed: the buyer's ``CoverageTrace`` list. @@ -540,6 +551,9 @@ async def explain( ``YYYY-MM-DD`` local date read in each offering's zone, an instant, or ``{"start", "end"}``. Compare with a seller trace through :func:`~adcp.reporting.subscriptions.diff_coverage_traces`. + ``read_timeout`` bounds each read, as in :meth:`reconcile`; a read that + does not answer in time raises a ``TIMEOUT`` + :class:`ReportingSubscriptionError`. """ from adcp.reporting.subscriptions.reconcile import explain_buyer_coverage @@ -552,6 +566,7 @@ async def explain( media_buy_id=media_buy_id, offering_id=offering_id, now=now, + read_timeout=read_timeout, ) def _event_source(self) -> ReportingEvents: @@ -567,6 +582,7 @@ async def poll( grace: timedelta = DEFAULT_GRACE, baseline: bool = False, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> EventBatch: """One batch of reporting events since the last ``ack``; serverless-safe. @@ -575,8 +591,11 @@ async def poll( ``reconcile`` and coverage with the snapshot saved at the last ack, so a crash before ``ack`` redelivers the same events with the same ids. Pass ``event_store=`` to :class:`ReportingClient` for a durable store. + ``read_timeout`` bounds each read, as in :meth:`reconcile`. """ - return await self._event_source().poll(grace=grace, baseline=baseline, now=now) + return await self._event_source().poll( + grace=grace, baseline=baseline, now=now, read_timeout=read_timeout + ) def events( self, @@ -584,13 +603,16 @@ def events( poll_every: timedelta = timedelta(minutes=5), stop: Any = None, grace: timedelta = DEFAULT_GRACE, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> AsyncIterator[ReportingEvent]: """``async for event in client.reporting.events(): ...``, acked per batch. Polls every ``poll_every``, or at once when :meth:`handle_webhook` rings. ``stop`` is an ``asyncio.Event`` that ends the loop. """ - return self._event_source().events(poll_every=poll_every, stop=stop, grace=grace) + return self._event_source().events( + poll_every=poll_every, stop=stop, grace=grace, read_timeout=read_timeout + ) async def handle_webhook(self, verified: Any) -> EventBatch | None: """Treat a verified seller webhook as a doorbell. @@ -1013,6 +1035,7 @@ async def reconcile( include_not_owed: bool = False, receipts: ReceiptPolicy | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> FleetReconcileReport: """:meth:`ReportingClient.reconcile` on every seller, ``concurrency`` at a time. @@ -1021,7 +1044,10 @@ async def reconcile( be reconciled at all, each with ``agent_id``), ``definitive`` and the summed ``summary``. ``agent_ids`` narrows the run; an unknown id is refused before anything is sent. The other options are - :meth:`ReportingClient.reconcile`'s, applied to every seller. + :meth:`ReportingClient.reconcile`'s, applied to every seller: + ``read_timeout`` bounds each read of each seller, so a seller that + stops answering is reported in ``issues`` and frees its slot instead + of holding the run. """ from adcp.reporting.subscriptions.reconcile import reconcile_fleet @@ -1036,6 +1062,7 @@ async def reconcile( include_not_owed=include_not_owed, receipts=receipts, now=now, + read_timeout=read_timeout, ) diff --git a/src/adcp/reporting/subscriptions/reconcile.py b/src/adcp/reporting/subscriptions/reconcile.py index 0825ab2923..fa41cc89a6 100644 --- a/src/adcp/reporting/subscriptions/reconcile.py +++ b/src/adcp/reporting/subscriptions/reconcile.py @@ -40,6 +40,7 @@ from typing import Any, Literal from adcp.reporting.subscriptions._clock import CoverageInputError, iso_z, utc_instant +from adcp.reporting.subscriptions.consumer import ReportingSubscriptionError from adcp.reporting.subscriptions.coverage import ( OwedSubscriptionPeriod, SubscriptionAuthorization, @@ -53,6 +54,7 @@ _Candidate, sort_traces, ) +from adcp.reporting.subscriptions.facade import DEFAULT_READ_TIMEOUT, _invalid from adcp.reporting.subscriptions.receipts import ( ReceiptPolicy, ReceiptReport, @@ -285,8 +287,93 @@ def __init__(self, task: str, detail: str, code: str = "TASK_FAILED") -> None: self.code = code +class _ReadTimeoutError(ReportingSubscriptionError): + """A read that did not answer within ``read_timeout``: code ``TIMEOUT``, transient.""" + + def __init__(self, task: str, seconds: float) -> None: + super().__init__( + f"{task} did not answer within read_timeout ({seconds:g}s). Retry once the " + "seller answers, or pass a longer read_timeout.", + code="TIMEOUT", + recovery="transient", + ) + self.task = task + + +#: The client methods a run calls, each bounded by ``read_timeout``, and their +#: wire task names. The receipts policy's write is bounded too, so a hung +#: seller never holds the report (the TS SDK's ``withReadTimeout`` set). +_BOUNDED_TASKS = { + "get_principal": "get_principal", + "fetch_capabilities": "get_adcp_capabilities", + "list_accounts": "list_accounts", + "get_media_buys": "get_media_buys", + "get_reporting_status": "get_reporting_status", + "sync_reporting_receipts": "sync_reporting_receipts", +} + + +def _read_timeout_seconds(read_timeout: timedelta | None) -> float | None: + """``read_timeout`` in seconds; ``None`` is no bound, a non-positive value is refused.""" + if read_timeout is None: + return None + if not isinstance(read_timeout, timedelta) or read_timeout <= timedelta(0): + raise _invalid( + "read_timeout must be a positive timedelta, or None for no bound. " + "Pass read_timeout=timedelta(seconds=30)." + ) + return read_timeout.total_seconds() + + +def _consume(task: asyncio.Future[Any]) -> None: + if not task.cancelled(): + task.exception() # retrieved, so an abandoned read never logs "never retrieved" + + +class _BoundedClient: + """The client with every read bounded: one that does not answer in time raises ``TIMEOUT``. + + The read is cancelled and abandoned at the deadline, without waiting for it + to unwind, so a seller that ignores cancellation still releases the run (and + its fleet concurrency slot). Every other attribute is the client's own. + """ + + def __init__(self, client: Any, seconds: float) -> None: + self._client = client + self._seconds = seconds + + def __getattr__(self, name: str) -> Any: + attribute = getattr(self._client, name) + task_name = _BOUNDED_TASKS.get(name) + if task_name is None or not callable(attribute): + return attribute + seconds = self._seconds + + async def bounded(*args: Any, **kwargs: Any) -> Any: + call = asyncio.ensure_future(attribute(*args, **kwargs)) + try: + done, _pending = await asyncio.wait({call}, timeout=seconds) + except BaseException: + call.cancel() + raise + if call in done: + return call.result() + call.cancel() + call.add_done_callback(_consume) + raise _ReadTimeoutError(task_name, seconds) + + return bounded + + +def _bounded(client: Any, read_timeout: timedelta | None) -> Any: + seconds = _read_timeout_seconds(read_timeout) + return client if seconds is None else _BoundedClient(client, seconds) + + def _failure(error: BaseException, task: str) -> ReconcileAccountError: """The ``error`` of a ``failed`` account: the task that failed, its code and detail.""" + if isinstance(error, _ReadTimeoutError): + return ReconcileAccountError(error.task, "TIMEOUT", str(error)) if isinstance(error, _UnreadableError): return ReconcileAccountError(error.task, error.code, str(error)) if isinstance(error, CoverageInputError): @@ -897,6 +984,7 @@ async def reconcile_subscriptions( include_not_owed: bool = False, receipts: ReceiptPolicy | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> ReconcileReport: """What every covered account is owed, and its status. See :meth:`ReportingClient.reconcile`.""" report, _context = await _reconcile( @@ -909,6 +997,7 @@ async def reconcile_subscriptions( grace=grace, include_not_owed=include_not_owed, now=now, + read_timeout=read_timeout, ) return report @@ -924,10 +1013,12 @@ async def _reconcile( include_not_owed: bool, now: datetime | None, receipts: ReceiptPolicy | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> tuple[ReconcileReport, _Context]: instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) if not isinstance(grace, timedelta) or grace < timedelta(0): raise ValueError("grace must be a non-negative timedelta") + client = _bounded(client, read_timeout) lower = _since(since) policy = receipts receipt_report = ReceiptReport() @@ -1148,17 +1239,19 @@ async def reconcile_fleet( include_not_owed: bool = False, receipts: ReceiptPolicy | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> FleetReconcileReport: """Reconcile each seller with bounded concurrency. See :meth:`FleetReportingClient.reconcile`. ``clients`` maps an agent id to an object with a ``reconcile`` coroutine taking :func:`reconcile_subscriptions`' keyword options (a :class:`~adcp.reporting.subscriptions.facade.ReportingClient`). + ``read_timeout`` bounds each read of each seller, so a hung seller fails + as a fleet issue and frees its concurrency slot. """ - from adcp.reporting.subscriptions.facade import _invalid - if type(concurrency) is not int or concurrency < 1: raise _invalid("concurrency must be a positive integer. Pass concurrency=4 or more.") + _read_timeout_seconds(read_timeout) # refused once here, not once per seller if agent_ids is not None: unknown = sorted(set(agent_ids) - set(clients)) if unknown: @@ -1172,6 +1265,7 @@ async def reconcile_fleet( "include_not_owed": include_not_owed, "receipts": receipts, "now": now, + "read_timeout": read_timeout, } reports: dict[str, ReconcileReport] = {} issues: list[FleetReconcileIssue] = [] @@ -1226,10 +1320,11 @@ async def explain_buyer_coverage( media_buy_id: str | None = None, offering_id: str | None = None, now: datetime | None = None, + read_timeout: timedelta | None = DEFAULT_READ_TIMEOUT, ) -> list[CoverageTrace]: """The buyer's :class:`CoverageTrace` list. See :meth:`ReportingClient.explain`.""" instant = utc_instant(now, field="now") if now is not None else datetime.now(timezone.utc) - context = _Context(client, state or BuyerReportingState()) + context = _Context(_bounded(client, read_timeout), state or BuyerReportingState()) records = await context.load(accounts=[account_id], strict=True) record = next((item for item in records if item.account_id == account_id), None) if record is None: @@ -1260,6 +1355,7 @@ async def explain_buyer_coverage( __all__ = [ "DEFAULT_GRACE", + "DEFAULT_READ_TIMEOUT", "BuyerReportingState", "FleetReconcileIssue", "FleetReconcileReport", diff --git a/tests/reporting_inprocess.py b/tests/reporting_inprocess.py index 0f214e68cf..c2213c8717 100644 --- a/tests/reporting_inprocess.py +++ b/tests/reporting_inprocess.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio from typing import Any from adcp import types as wire @@ -30,8 +31,12 @@ def __init__(self, platform: Any, context: Any) -> None: self._platform = platform self._context = context self.fail: set[str] = set() + #: Tasks that never answer, as a hung seller. + self.hang: set[str] = set() async def _call(self, task: str, request: Any, model: Any) -> TaskResult[Any]: + if task in self.hang: + await asyncio.Event().wait() if task in self.fail: return TaskResult.model_construct( status=TaskStatus.FAILED, data=None, success=False, error=f"{task} unavailable" @@ -55,4 +60,6 @@ async def get_reporting_status(self, request: Any) -> TaskResult[Any]: return await self._call("get_reporting_status", request, wire.GetReportingStatusResponse) async def fetch_capabilities(self) -> Any: + if "get_adcp_capabilities" in self.hang: + await asyncio.Event().wait() return _plain(await self._platform.get_adcp_capabilities({}, self._context())) diff --git a/tests/test_reporting_subscription_reconcile.py b/tests/test_reporting_subscription_reconcile.py index 88e367c51e..8e6fa508ff 100644 --- a/tests/test_reporting_subscription_reconcile.py +++ b/tests/test_reporting_subscription_reconcile.py @@ -277,3 +277,140 @@ async def test_the_multi_agent_client_exposes_the_fleet() -> None: fleet = multi.reporting assert isinstance(fleet, FleetReportingClient) assert fleet._reporting("s1") is multi.agent("s1").reporting + + +# -- read_timeout ------------------------------------------------------------------------- + +_FAST = timedelta(milliseconds=50) + + +async def test_a_hung_seller_times_out_in_the_fleet_and_frees_its_slot() -> None: + import asyncio + + from adcp.reporting.subscriptions.facade import FleetReportingClient + + world = await _delivered_world() + hung = InProcessBuyer(world.platform, _ctx) + hung.hang.add("get_principal") + fleet = FleetReportingClient( + { + "hung": hung, + "a": InProcessBuyer(world.platform, _ctx), + "b": InProcessBuyer(world.platform, _ctx), + } + ) + # One slot and the hung seller first: the others still run once it times out. + report = await asyncio.wait_for( + fleet.reconcile(now=world.clock(), concurrency=1, read_timeout=_FAST), timeout=10 + ) + assert sorted(report.by_seller) == ["a", "b"] + assert all(item.definitive for item in report.by_seller.values()) + ((agent_id, code, message),) = [ + (issue.agent_id, issue.code, issue.message) for issue in report.issues + ] + assert (agent_id, code) == ("hung", "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE") + assert "get_principal did not answer within read_timeout" in message + assert not report.definitive + assert report.summary.owed == 4 + + +async def test_a_read_that_does_not_answer_fails_its_account_as_timeout() -> None: + world = await _delivered_world() + reporting, client = _buyer(world) + client.hang.add("get_media_buys") + report = await reporting.reconcile(now=world.clock(), read_timeout=_FAST) + assert not report.definitive + (entry,) = report.accounts + assert (entry.account_id, entry.status) == (ACCOUNT, "failed") + assert entry.error is not None + assert (entry.error.task, entry.error.code) == ("get_media_buys", "TIMEOUT") + assert "read_timeout" in entry.error.message + assert [issue.code for issue in report.issues] == ["SUBSCRIPTION_ACCOUNT_UNREADABLE"] + + +async def test_a_principal_read_that_does_not_answer_is_reported_and_raises_for_explain() -> None: + from adcp.reporting.subscriptions import ReportingSubscriptionError + + world = await _delivered_world() + reporting, client = _buyer(world) + client.hang.add("get_principal") + report = await reporting.reconcile(now=world.clock(), read_timeout=_FAST) + assert not report.definitive and report.periods == () + (issue,) = report.issues + assert issue.code == "SUBSCRIPTION_PRINCIPAL_UNAVAILABLE" + assert "read_timeout" in issue.message + (roster,) = report.accounts + assert roster.error is not None + assert (roster.account_id, roster.error.task, roster.error.code) == ( + "*", + "get_principal", + "TIMEOUT", + ) + with pytest.raises(ReportingSubscriptionError, match="read_timeout") as raised: + await reporting.explain( + account_id=ACCOUNT, period="2026-11-01", now=world.clock(), read_timeout=_FAST + ) + assert raised.value.code == "TIMEOUT" and raised.value.recovery == "transient" + + +async def test_read_timeout_none_disables_the_bound_and_non_positive_is_refused() -> None: + import asyncio + + from adcp.reporting.subscriptions import ReportingSubscriptionError + from adcp.reporting.subscriptions.facade import FleetReportingClient + + world = await _delivered_world() + reporting, client = _buyer(world) + unbounded = await reporting.reconcile(now=world.clock(), read_timeout=None) + assert unbounded.definitive + client.hang.add("get_principal") + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(reporting.reconcile(now=world.clock(), read_timeout=None), 0.2) + fleet = FleetReportingClient({"a": InProcessBuyer(world.platform, _ctx)}) + for bad in (timedelta(0), timedelta(seconds=-1), 30): + with pytest.raises(ReportingSubscriptionError, match="read_timeout") as raised: + await reporting.reconcile(read_timeout=bad) # type: ignore[arg-type] + assert raised.value.code == "INVALID_REQUEST" + # Refused before any seller is read, never one issue per seller. + with pytest.raises(ReportingSubscriptionError, match="read_timeout"): + await fleet.reconcile(read_timeout=bad) # type: ignore[arg-type] + with pytest.raises(ReportingSubscriptionError, match="read_timeout"): + await reporting.explain( + account_id=ACCOUNT, + period="2026-11-01", + read_timeout=bad, # type: ignore[arg-type] + ) + + +async def test_a_read_that_ignores_cancellation_still_releases_the_run() -> None: + import asyncio + + from adcp.reporting.subscriptions.reconcile import _BoundedClient, _ReadTimeoutError + + finished = asyncio.Event() + + class Stubborn: + agent_config = "kept" + + async def get_principal(self, request: object) -> str: + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + await asyncio.sleep(0.2) # unwinds slowly instead of stopping + finally: + finished.set() + return "late" + + async def list_accounts(self, request: object) -> object: + raise TimeoutError("the transport's own timeout") + + bounded = _BoundedClient(Stubborn(), 0.05) + assert bounded.agent_config == "kept" + started = asyncio.get_running_loop().time() + with pytest.raises(_ReadTimeoutError, match="get_principal did not answer"): + await bounded.get_principal({}) + assert asyncio.get_running_loop().time() - started < 0.2 + # A client's own TimeoutError is its own failure, not read_timeout's. + with pytest.raises(TimeoutError, match="transport"): + await bounded.list_accounts({}) + await asyncio.wait_for(finished.wait(), 1) From 62385830b2bd04b0b82a041731629245ef7cd834 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Mon, 2 Nov 2026 22:44:18 +0000 Subject: [PATCH 2/2] fix(accounts): run account_requests[] and request_account through the credential scrubber list_accounts scrubbed its accounts[] rows and then attached account_requests[] unscrubbed, and request_account's response skipped the scrubber entirely. A resolved AccountRequest carries a full Account, so its billing_entity.bank, governance_agents[].authentication and notification_configs[].authentication.credentials now get the same defense-in-depth strip as the rows beside them. request_account joins CREDENTIAL_BEARING_METHODS so the task registry and webhook boundaries scrub it too. Co-Authored-By: Claude Opus 5.5 --- src/adcp/decisioning/account_projection.py | 1 + src/adcp/decisioning/handler.py | 12 ++++- tests/test_account_requests_seller.py | 59 ++++++++++++++++++++++ 3 files changed, 70 insertions(+), 2 deletions(-) diff --git a/src/adcp/decisioning/account_projection.py b/src/adcp/decisioning/account_projection.py index e414ee116f..e8c6ee187a 100644 --- a/src/adcp/decisioning/account_projection.py +++ b/src/adcp/decisioning/account_projection.py @@ -364,6 +364,7 @@ def to_wire_sync_governance_row(row: SyncGovernanceResultRow) -> dict[str, Any]: CREDENTIAL_BEARING_METHODS: frozenset[str] = frozenset( { "list_accounts", + "request_account", "sync_accounts", "sync_governance", "create_media_buy", diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index 8c1f2c670c..707b687043 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -3594,7 +3594,13 @@ async def list_accounts( # type: ignore[override] page = _call_with_optional_ctx(list_requests, params, ctx=resolve_ctx) if inspect.isawaitable(page): page = await page - served.update(_project_account_requests_page(page)) + # The rows above were scrubbed before the requests were attached; + # a resolved request carries a full Account, so scrub them too. + served.update( + strip_credentials_from_wire_result( + "list_accounts", _project_account_requests_page(page) + ) + ) reporting = getattr(self._platform, "_reliable_reporting_service", None) if reporting is not None: # Managed reporting subscriptions serve the grant journal's @@ -3623,7 +3629,9 @@ async def request_account( # type: ignore[override] result = _call_with_optional_ctx(request_account, params, ctx=resolve_ctx) if inspect.isawaitable(result): result = await result - return _project_request_account(result) + return strip_credentials_from_wire_result( + "request_account", _project_request_account(result) + ) # ----- Optional-method gate ----- diff --git a/tests/test_account_requests_seller.py b/tests/test_account_requests_seller.py index 8f2202d5d8..ab9e1eecea 100644 --- a/tests/test_account_requests_seller.py +++ b/tests/test_account_requests_seller.py @@ -288,3 +288,62 @@ async def list_accounts(params: dict[str, Any]) -> TaskResult: again = await client.accounts.ensure(KEY) assert isinstance(again, AccountReady) and again.account_id == "acc-1" assert len(store.requests) == 1 + + +_BANK = {"account_holder": "Acme Inc", "iban": "DE89370400440532013000"} +_GOVERNANCE_SECRET = "gov-secret-token-0123456789abcdef" +_WEBHOOK_SECRET = "webhook-secret-0123456789abcdefghij" + + +def _credentialed_account() -> dict[str, Any]: + """A wire Account carrying every credential-shaped field the scrubber strips.""" + return { + "account_id": "acc-1", + "name": "Acme", + "status": "active", + "billing_entity": {"legal_name": "Acme Inc", "bank": dict(_BANK)}, + "governance_agents": [ + { + "url": "https://governance.example", + "authentication": {"schemes": ["Bearer"], "credentials": _GOVERNANCE_SECRET}, + } + ], + "notification_configs": [ + { + "url": "https://buyer.example/hooks", + "authentication": {"schemes": ["Bearer"], "credentials": _WEBHOOK_SECRET}, + } + ], + } + + +def _assert_scrubbed(account: dict[str, Any]) -> None: + assert account["billing_entity"] == {"legal_name": "Acme Inc"} + assert account["governance_agents"] == [{"url": "https://governance.example"}] + assert "credentials" not in account["notification_configs"][0]["authentication"] + for secret in (_BANK["iban"], _GOVERNANCE_SECRET, _WEBHOOK_SECRET): + assert secret not in repr(account) + + +@pytest.mark.asyncio +async def test_account_requests_are_scrubbed_like_the_accounts_beside_them() -> None: + store = _Store() + store.requests["req-1"] = _record( + "req-1", + "resolved", + resolution="created", + account_id="acc-1", + account=_credentialed_account(), + ) + handler = _handler(store) + listed = await handler.list_accounts(ListAccountsRequest(include_requests=True), _ctx()) + (item,) = listed["account_requests"] + _assert_scrubbed(item["account"]) + # The hook's record is not mutated: the scrub copies. + assert store.requests["req-1"]["account"]["billing_entity"]["bank"] == _BANK + + answered = await handler.request_account( + RequestAccountRequest.model_validate({"idempotency_key": "k" * 16, **KEY}), _ctx() + ) + assert answered["request"]["state"] == "resolved" + _assert_scrubbed(answered["request"]["account"])