diff --git a/src/adcp/decisioning/dispatch.py b/src/adcp/decisioning/dispatch.py index 5ccc9331..a37d3f4f 100644 --- a/src/adcp/decisioning/dispatch.py +++ b/src/adcp/decisioning/dispatch.py @@ -59,6 +59,7 @@ TaskHandoffContext, TaskRegistry, ) +from adcp.decisioning.time_budget import SyncExecutorAdmission, _bind_routed_sync_execution from adcp.decisioning.types import ( AdcpError, TaskHandoff, @@ -85,6 +86,11 @@ logger = logging.getLogger(__name__) +# Strong references for synchronous adopter lifecycles that outlive a +# cancelled request. A Python thread cannot be cancelled; its completion hooks +# must still settle durable proposal/idempotency state. +_SUPERVISED_SYNC_LIFECYCLES: set[asyncio.Task[Any]] = set() + # --------------------------------------------------------------------------- # Specialism enum — spec slugs known to the framework # --------------------------------------------------------------------------- @@ -1313,6 +1319,7 @@ async def _invoke_platform_method( webhook_target: WebhookDeliveryTarget | None = None, webhook_auto_emit: bool = True, pre_handoff_reject: Callable[[], None] | None = None, + sync_admission: SyncExecutorAdmission | None = None, ) -> Any: """Invoke a platform method, projecting hybrid returns. @@ -1383,12 +1390,16 @@ async def _invoke_platform_method( off on a ``wholesale`` request is rejected cleanly instead of leaking a task the buyer was told was rejected. Runs only on the ``TaskHandoff`` arm; sync / workflow-handoff returns ignore it. + :param sync_admission: Optional bounded admission controller for a sync + method. Its permit remains held until the underlying thread future + actually completes, including after caller cancellation. """ # pydantic is a required dep; import here (not at module level) to mirror # the lazy-import discipline used throughout this module. from pydantic import ValidationError as _ValidationError # noqa: PLC0415 method = getattr(platform, method_name) + sync_lifecycle_continues = False # Re-validate through the platform method's own annotation when it's a # stricter subclass of the shim's already-deserialized type. Skipped # when arg_projector is set — that path replaces positional args entirely. @@ -1407,31 +1418,74 @@ async def _invoke_platform_method( try: if asyncio.iscoroutinefunction(method): - if arg_projector is not None: - result = await method(**arg_projector, ctx=ctx) - elif extra_kwargs: - result = await method(params, ctx, **extra_kwargs) - else: - result = await method(params, ctx) + # Async router delegates may resolve to synchronous tenant + # children only after account routing. Propagate the same bounded + # admission controller and configured executor through ContextVars + # so that path cannot bypass the timed-sync limit. + with _bind_routed_sync_execution(sync_admission, executor): + if arg_projector is not None: + result = await method(**arg_projector, ctx=ctx) + elif extra_kwargs: + result = await method(params, ctx, **extra_kwargs) + else: + result = await method(params, ctx) else: ctx_snapshot = contextvars.copy_context() loop = asyncio.get_running_loop() if arg_projector is not None: projected_kwargs = {**arg_projector, "ctx": ctx} - result = await loop.run_in_executor( - executor, - functools.partial(ctx_snapshot.run, method, **projected_kwargs), - ) + worker_call = functools.partial(ctx_snapshot.run, method, **projected_kwargs) elif extra_kwargs: - result = await loop.run_in_executor( - executor, - functools.partial(ctx_snapshot.run, method, params, ctx, **extra_kwargs), + worker_call = functools.partial( + ctx_snapshot.run, method, params, ctx, **extra_kwargs ) else: - result = await loop.run_in_executor( - executor, - functools.partial(ctx_snapshot.run, method, params, ctx), - ) + worker_call = functools.partial(ctx_snapshot.run, method, params, ctx) + + if sync_admission is not None: + await sync_admission.acquire() + try: + worker_future = executor.submit(worker_call) + except BaseException: + if sync_admission is not None: + sync_admission.release() + raise + + if sync_admission is not None: + + def _release_admission(_future: object) -> None: + try: + loop.call_soon_threadsafe(sync_admission.release) + except RuntimeError: + # Event loop already closed during process teardown. + pass + + worker_future.add_done_callback(_release_admission) + + worker_async_future = asyncio.wrap_future(worker_future, loop=loop) + try: + result = await asyncio.shield(worker_async_future) + except asyncio.CancelledError: + if on_failure is not None: + sync_lifecycle_continues = True + lifecycle = asyncio.create_task( + _settle_cancelled_sync_lifecycle( + worker_async_future, + ctx=ctx, + method_name=method_name, + registry=registry, + executor=executor, + on_complete=on_complete, + on_failure=on_failure, + pre_handoff_reject=pre_handoff_reject, + request_params=params, + webhook_target=webhook_target, + webhook_auto_emit=webhook_auto_emit, + ) + ) + _SUPERVISED_SYNC_LIFECYCLES.add(lifecycle) + lifecycle.add_done_callback(_SUPERVISED_SYNC_LIFECYCLES.discard) + raise except AdcpError as exc: # Adopter raised structured error — propagate verbatim. The # outer middleware projects to the wire envelope. Fire @@ -1543,7 +1597,64 @@ async def _invoke_platform_method( if on_failure is not None: await _safe_on_failure_call(on_failure, wrapped, method_name) raise wrapped from exc + except BaseException as exc: + # ``asyncio.CancelledError`` (and shutdown BaseExceptions) bypass the + # wire-error wrapping above, but must still release framework state + # reserved before adapter dispatch. Preserve the exact exception. + nested_sync_future = getattr(exc, "_adcp_sync_worker_future", None) + if isinstance(nested_sync_future, asyncio.Future) and on_failure is not None: + sync_lifecycle_continues = True + lifecycle = asyncio.create_task( + _settle_cancelled_sync_lifecycle( + nested_sync_future, + ctx=ctx, + method_name=method_name, + registry=registry, + executor=executor, + on_complete=on_complete, + on_failure=on_failure, + pre_handoff_reject=pre_handoff_reject, + request_params=params, + webhook_target=webhook_target, + webhook_auto_emit=webhook_auto_emit, + ) + ) + _SUPERVISED_SYNC_LIFECYCLES.add(lifecycle) + lifecycle.add_done_callback(_SUPERVISED_SYNC_LIFECYCLES.discard) + if on_failure is not None and not sync_lifecycle_continues: + await _safe_on_failure_call(on_failure, exc, method_name) + raise + + return await _project_invocation_result( + result, + ctx=ctx, + method_name=method_name, + registry=registry, + executor=executor, + on_complete=on_complete, + on_failure=on_failure, + pre_handoff_reject=pre_handoff_reject, + request_params=params, + webhook_target=webhook_target, + webhook_auto_emit=webhook_auto_emit, + ) + +async def _project_invocation_result( + result: Any, + *, + ctx: RequestContext[Any], + method_name: str, + registry: TaskRegistry, + executor: ThreadPoolExecutor, + on_complete: Callable[[Any], Awaitable[None]] | None, + on_failure: Callable[[BaseException], Awaitable[None]] | None, + pre_handoff_reject: Callable[[], None] | None, + request_params: BaseModel, + webhook_target: WebhookDeliveryTarget | None, + webhook_auto_emit: bool, +) -> Any: + """Project a raw adopter result and settle its framework lifecycle hooks.""" if is_task_handoff(result): # Reject before any side effect (registry row, background task, # completion webhook) is created. The wholesale discovery guard @@ -1551,7 +1662,12 @@ async def _invoke_platform_method( # wholesale request never leaks a task the buyer is told is # rejected. if pre_handoff_reject is not None: - pre_handoff_reject() + try: + pre_handoff_reject() + except BaseException as exc: + if on_failure is not None: + await _safe_on_failure_call(on_failure, exc, method_name) + raise return await _project_handoff( result, ctx, @@ -1560,7 +1676,7 @@ async def _invoke_platform_method( executor=executor, on_complete=on_complete, on_failure=on_failure, - request_params=params, + request_params=request_params, webhook_target=webhook_target, webhook_auto_emit=webhook_auto_emit, ) @@ -1571,7 +1687,7 @@ async def _invoke_platform_method( method_name=method_name, registry=registry, executor=executor, - request_params=params, + request_params=request_params, ) # Sync return path. Fire on_complete with the typed result before @@ -1596,6 +1712,51 @@ async def _invoke_platform_method( return strip_credentials_from_wire_result(method_name, result) +async def _settle_cancelled_sync_lifecycle( + worker_future: asyncio.Future[Any], + *, + ctx: RequestContext[Any], + method_name: str, + registry: TaskRegistry, + executor: ThreadPoolExecutor, + on_complete: Callable[[Any], Awaitable[None]] | None, + on_failure: Callable[[BaseException], Awaitable[None]] | None, + pre_handoff_reject: Callable[[], None] | None, + request_params: BaseModel, + webhook_target: WebhookDeliveryTarget | None, + webhook_auto_emit: bool, +) -> None: + """Settle a sync worker after its request task has been cancelled.""" + try: + result = await worker_future + except BaseException as exc: + if on_failure is not None: + await _safe_on_failure_call(on_failure, exc, method_name) + return + try: + await _project_invocation_result( + result, + ctx=ctx, + method_name=method_name, + registry=registry, + executor=executor, + on_complete=on_complete, + on_failure=on_failure, + pre_handoff_reject=pre_handoff_reject, + request_params=request_params, + webhook_target=webhook_target, + webhook_auto_emit=webhook_auto_emit, + ) + except BaseException: + # Lifecycle hooks already apply their own rollback semantics. There is + # no request waiter left to receive this exception, so retain it in + # server logs rather than producing an unhandled-task warning. + logger.exception( + "Cancelled request's synchronous %s lifecycle failed while settling", + method_name, + ) + + async def _safe_on_failure_call( on_failure: Callable[[BaseException], Awaitable[None]], exc: BaseException, @@ -1610,10 +1771,9 @@ async def _safe_on_failure_call( """ try: await on_failure(exc) - except Exception: + except BaseException: logger.exception( - "on_failure hook raised while handling %s for %s — original " - "exception still propagates", + "on_failure hook raised while handling %s for %s — original exception still propagates", type(exc).__name__, method_name, ) @@ -1809,6 +1969,12 @@ async def _run() -> None: ) await _fail(wrapped) return + except BaseException as exc: + # Background task cancellation must release any proposal + # reservation, then remain cancellation to the task scheduler. + if on_failure is not None: + await _safe_on_failure_call(on_failure, exc, method_name) + raise # Framework completion hook (e.g., proposal_store.commit for # finalize, mark_proposal_consumed for create_media_buy). Runs diff --git a/src/adcp/decisioning/handler.py b/src/adcp/decisioning/handler.py index e47b3102..00edebba 100644 --- a/src/adcp/decisioning/handler.py +++ b/src/adcp/decisioning/handler.py @@ -55,6 +55,8 @@ ) from adcp.decisioning.dispatch import ( _build_request_context, + _internal_error_details, + _internal_error_message, _invoke_platform_method, ) from adcp.decisioning.implementation_config import ProductConfigStore @@ -79,7 +81,11 @@ has_refine_support, project_refine_response, ) -from adcp.decisioning.time_budget import project_incomplete_response, resolve_time_budget +from adcp.decisioning.time_budget import ( + SyncExecutorAdmission, + project_incomplete_response, + resolve_time_budget, +) from adcp.decisioning.types import ( Account as _DecisioningAccount, ) @@ -1282,6 +1288,7 @@ def __init__( property_list_fetcher: PropertyListFetcher | None = None, media_buy_store: MediaBuyStore | None = None, advertise_all: bool = False, + timed_sync_get_products_limit: int | None = None, ) -> None: super().__init__() self._platform = platform @@ -1298,6 +1305,16 @@ def __init__( self._property_list_fetcher = property_list_fetcher self._media_buy_store = media_buy_store self._advertise_all = advertise_all + # ThreadPoolExecutor exposes no public worker-count property. Standard + # instances carry ``_max_workers``; executor-compatible wrappers fall + # back to the conservative one-call limit. + worker_count = int(getattr(executor, "_max_workers", 1)) + admission_limit = ( + timed_sync_get_products_limit + if timed_sync_get_products_limit is not None + else max(1, worker_count // 2) + ) + self._timed_sync_get_products_admission = SyncExecutorAdmission(admission_limit) # Cache whether the platform's create_media_buy accepts 'configs' # so we only pay the inspect.signature cost at construction time. @@ -1630,17 +1647,9 @@ async def get_adcp_capabilities( ) raise AdcpError( "INTERNAL_ERROR", - message=( - "Unhandled exception in platform.get_adcp_capabilities_for_request: " - f"{type(exc).__name__}: {exc}" - ), + message=_internal_error_message("get_adcp_capabilities_for_request", exc), recovery="terminal", - details={ - "caused_by": { - "type": type(exc).__name__, - "message": str(exc), - } - }, + details=_internal_error_details(exc), ) from exc has_scoped_caps = scoped_caps is not None if scoped_caps is not None: @@ -1965,6 +1974,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: registry=self._registry, on_complete=_persist_draft_hook, pre_handoff_reject=pre_handoff_reject, + sync_admission=( + self._timed_sync_get_products_admission if deadline is not None else None + ), **self._handoff_webhook_kwargs(), ) try: @@ -1973,9 +1985,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: ) except asyncio.TimeoutError: # Deadline expired. The platform coroutine is cancelled; for - # sync adopters the underlying thread runs to completion but the - # asyncio side has moved on (thread-pool slot leak documented in - # adcp.decisioning.time_budget module header). + # sync adopters an admitted underlying thread runs to completion, + # retaining its bounded admission permit. Saturated calls that + # never acquired a permit were not submitted to the executor. tb = params.time_budget interval = tb.interval if tb is not None else 0 unit_raw = tb.unit if tb is not None else None @@ -1988,7 +2000,9 @@ async def _persist_draft_hook(get_products_result: Any) -> None: "[adcp.decisioning] get_products timed out after %ds " "(time_budget=%d %s); returning incomplete response. " "To avoid timeout cancellations, optimise get_products " - "latency or reduce the platform's search scope.", + "latency or reduce the platform's search scope. Saturated " + "sync calls wait behind timed_sync_get_products_limit and " + "are not submitted after their budget expires.", deadline, interval, unit, diff --git a/src/adcp/decisioning/platform_router.py b/src/adcp/decisioning/platform_router.py index 5619e68b..9f23005e 100644 --- a/src/adcp/decisioning/platform_router.py +++ b/src/adcp/decisioning/platform_router.py @@ -96,6 +96,8 @@ from __future__ import annotations import asyncio +import contextvars +import functools import inspect import time from collections import OrderedDict @@ -119,6 +121,7 @@ SalesPlatform, SignalsPlatform, ) +from adcp.decisioning.time_budget import _routed_sync_execution from adcp.decisioning.types import AdcpError if TYPE_CHECKING: @@ -128,6 +131,42 @@ from adcp.decisioning.proposal_store import ProposalStore +async def _run_sync_delegate(method: Any, *args: Any, **kwargs: Any) -> Any: + """Run a sync child and expose its live future across request cancellation.""" + admission, executor = _routed_sync_execution() + if admission is None or executor is None: + worker: asyncio.Future[Any] = asyncio.create_task( + asyncio.to_thread(method, *args, **kwargs) + ) + else: + await admission.acquire() + loop = asyncio.get_running_loop() + snapshot = contextvars.copy_context() + call = functools.partial(snapshot.run, method, *args, **kwargs) + try: + concurrent_worker = executor.submit(call) + except BaseException: + admission.release() + raise + + def _release_admission(_future: object) -> None: + try: + loop.call_soon_threadsafe(admission.release) + except RuntimeError: + # The event loop closed during process teardown. + pass + + concurrent_worker.add_done_callback(_release_admission) + worker = asyncio.wrap_future(concurrent_worker, loop=loop) + try: + return await asyncio.shield(worker) + except asyncio.CancelledError as exc: + # dispatch._invoke_platform_method consumes this private marker and + # settles lifecycle hooks from the worker's actual terminal outcome. + setattr(exc, "_adcp_sync_worker_future", worker) + raise + + # Every specialism Protocol the framework knows about. New Protocol # classes added to ``adcp.decisioning.specialisms`` get picked up by # adding them here. Walking ``__protocol_attrs__`` (set by the runtime @@ -503,7 +542,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(manager, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) # No proposal_manager for this tenant — fall through to the # platform. Reuses the same lookup helper as the synthesized @@ -512,7 +551,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def _make_delegate(self, method_name: str) -> Any: """Create a delegating callable for ``method_name``. @@ -549,7 +588,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: # contextvars snapshot; ``asyncio.to_thread`` does the same # using the running loop's default executor with copied # context. - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"PlatformRouter.{method_name}" @@ -1023,7 +1062,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: method = getattr(platform, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"LazyPlatformRouter.{method_name}" @@ -1057,13 +1096,13 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(manager, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) platform = await self._platform_for_method(ctx, "get_products") method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def proposal_manager_for_tenant(self, tenant_id: str) -> ProposalManager | None: """Return the :class:`ProposalManager` for ``tenant_id``, or ``None``.""" @@ -1245,7 +1284,7 @@ async def _delegate(*args: Any, **kwargs: Any) -> Any: method = getattr(platform, method_name) if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) _delegate.__name__ = method_name _delegate.__qualname__ = f"_RegistryPlatformAdapter.{method_name}" @@ -1272,7 +1311,7 @@ async def get_products(self, *args: Any, **kwargs: Any) -> Any: method = getattr(platform, "get_products") if inspect.iscoroutinefunction(method): return await method(*args, **kwargs) - return await asyncio.to_thread(method, *args, **kwargs) + return await _run_sync_delegate(method, *args, **kwargs) def _make_registry_platform_adapter( diff --git a/src/adcp/decisioning/proposal_dispatch.py b/src/adcp/decisioning/proposal_dispatch.py index e989cc7d..f3526d4d 100644 --- a/src/adcp/decisioning/proposal_dispatch.py +++ b/src/adcp/decisioning/proposal_dispatch.py @@ -744,18 +744,15 @@ async def maybe_hydrate_recipes_for_create_media_buy( recipes=record.recipes, field_path_prefix="packages", ) - except Exception: - # Narrow to Exception (not BaseException): CancelledError / - # SystemExit / KeyboardInterrupt skip the release path — under - # cancellation the next worker will read the stale reservation - # and eviction handles it; under shutdown we want fast exit. - # release_consumption is idempotent on already-COMMITTED so a - # release that races with a concurrent worker is harmless. + except BaseException: + # Cancellation and shutdown must not strand a durable reservation. + # Once the exception is caught, this cleanup await is allowed to run; + # the original BaseException is re-raised below. try: await _await_maybe( store.release_consumption(proposal_id, expected_account_id=ctx.account.id) ) - except Exception: + except BaseException: logger.exception( "Failed to release proposal reservation %s after " "post-reserve validation error; record may be stuck in " @@ -888,7 +885,7 @@ async def release_proposal_reservation( expected_account_id=proposal_record.account_id, ) ) - except Exception: + except BaseException: logger.exception( "Failed to release consumption reservation for proposal %s; " "the record may be stuck in CONSUMING until eviction. " diff --git a/src/adcp/decisioning/serve.py b/src/adcp/decisioning/serve.py index 05e4ac2c..c3fd8044 100644 --- a/src/adcp/decisioning/serve.py +++ b/src/adcp/decisioning/serve.py @@ -81,6 +81,7 @@ def create_adcp_server_from_platform( *, executor: ThreadPoolExecutor | None = None, thread_pool_size: int | None = None, + timed_sync_get_products_limit: int | None = None, registry: TaskRegistry | None = None, state_reader: StateReader | None = None, resource_resolver: ResourceResolver | None = None, @@ -125,6 +126,12 @@ def create_adcp_server_from_platform( :param thread_pool_size: Size the default framework-allocated executor. Mutually exclusive with ``executor``. Default is :func:`_default_thread_pool_size`. + :param timed_sync_get_products_limit: Maximum synchronous + ``get_products`` calls with SDK-managed deadlines admitted to the + executor at once. Saturated calls wait within their own time budget + and return ``incomplete`` without being submitted if it expires. + Defaults to half the executor worker count (minimum one), reserving + capacity for other tools. :param registry: Bring-your-own :class:`TaskRegistry` — typically a v6.1 durable backing store. Default is :class:`InMemoryTaskRegistry`, which the production-mode @@ -371,6 +378,7 @@ def create_adcp_server_from_platform( property_list_fetcher=property_list_fetcher, media_buy_store=media_buy_store, advertise_all=advertise_all, + timed_sync_get_products_limit=timed_sync_get_products_limit, ) # Boot-time fail-fast: property_list_filtering declared but no fetcher wired. @@ -455,6 +463,7 @@ def serve( name: str | None = None, executor: ThreadPoolExecutor | None = None, thread_pool_size: int | None = None, + timed_sync_get_products_limit: int | None = None, registry: TaskRegistry | None = None, state_reader: StateReader | None = None, resource_resolver: ResourceResolver | None = None, @@ -486,6 +495,9 @@ def serve( :param executor: BYO :class:`ThreadPoolExecutor` per :func:`create_adcp_server_from_platform` D5 contract. :param thread_pool_size: Default-executor size override. + :param timed_sync_get_products_limit: Bounded admission limit for + deadline-managed synchronous ``get_products`` calls. See + :func:`create_adcp_server_from_platform`. :param registry: BYO :class:`TaskRegistry`. Default is :class:`InMemoryTaskRegistry` (gated for production). :param state_reader: Custom :class:`StateReader` impl (D15). @@ -573,6 +585,7 @@ def serve( platform, executor=executor, thread_pool_size=thread_pool_size, + timed_sync_get_products_limit=timed_sync_get_products_limit, registry=registry, state_reader=state_reader, resource_resolver=resource_resolver, diff --git a/src/adcp/decisioning/time_budget.py b/src/adcp/decisioning/time_budget.py index 65fd40d4..261564ed 100644 --- a/src/adcp/decisioning/time_budget.py +++ b/src/adcp/decisioning/time_budget.py @@ -17,15 +17,17 @@ past the ``except Exception`` in ``_invoke_platform_method`` cleanly. This invariant MUST be preserved if ``get_products`` ever gains registry work. -* **Thread-pool warning for sync adopters.** When a sync adopter runs via +* **Bounded sync-adopter admission.** When a sync adopter runs via ``loop.run_in_executor`` and ``asyncio.wait_for`` fires, the asyncio side moves on but the underlying thread continues until its blocking call - returns. No Python mechanism can interrupt a running thread. The pool - slot is occupied for the full duration; on a short-budget burst against a - slow sync adopter this can exhaust the pool. Async adopters are - unaffected. Adopters who need to co-operate with deadline cancellation - should implement the ``IncrementalGetProducts`` protocol or migrate to an - async ``get_products``. + returns. No Python mechanism can interrupt a running thread. The framework + therefore admits only a bounded number of deadline-managed synchronous + calls and holds each permit until the worker really exits, even after the + response timed out. Saturated calls spend their budget waiting for a permit + and return ``incomplete[]`` without entering the executor. By default the + limit is half the executor workers (minimum one), preserving capacity for + other tools; operators can tune it at server construction. Async adopters + are unaffected. * **``campaign`` unit → no SDK-managed deadline.** ``unit='campaign'`` means "the seller has the full campaign flight to respond" — this is a @@ -61,15 +63,74 @@ from __future__ import annotations import logging +from asyncio import Semaphore from collections.abc import AsyncIterator +from contextlib import contextmanager +from contextvars import ContextVar from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable if TYPE_CHECKING: + from collections.abc import Iterator + from concurrent.futures import ThreadPoolExecutor + from adcp.decisioning.context import RequestContext from adcp.types import GetProductsRequest logger = logging.getLogger(__name__) + +class SyncExecutorAdmission: + """Bound outstanding deadline-managed synchronous executor work. + + A permit represents a worker submission, not a waiting HTTP request. It + is released only by the underlying ``concurrent.futures.Future`` done + callback, because cancelling its asyncio wrapper cannot stop a running + Python thread. + """ + + def __init__(self, limit: int) -> None: + if not isinstance(limit, int) or isinstance(limit, bool) or limit < 1: + raise ValueError("sync executor admission limit must be a positive integer") + self.limit = limit + self._semaphore = Semaphore(limit) + + async def acquire(self) -> None: + """Wait until a bounded worker slot is available.""" + await self._semaphore.acquire() + + def release(self) -> None: + """Return a worker slot after its real thread future completes.""" + self._semaphore.release() + + +_ROUTED_SYNC_ADMISSION: ContextVar[SyncExecutorAdmission | None] = ContextVar( + "adcp_routed_sync_admission", default=None +) +_ROUTED_SYNC_EXECUTOR: ContextVar[ThreadPoolExecutor | None] = ContextVar( + "adcp_routed_sync_executor", default=None +) + + +@contextmanager +def _bind_routed_sync_execution( + admission: SyncExecutorAdmission | None, + executor: ThreadPoolExecutor, +) -> Iterator[None]: + """Expose deadline admission to an async router's eventual sync child.""" + admission_token = _ROUTED_SYNC_ADMISSION.set(admission) + executor_token = _ROUTED_SYNC_EXECUTOR.set(executor) + try: + yield + finally: + _ROUTED_SYNC_EXECUTOR.reset(executor_token) + _ROUTED_SYNC_ADMISSION.reset(admission_token) + + +def _routed_sync_execution() -> tuple[SyncExecutorAdmission | None, ThreadPoolExecutor | None]: + """Return the admission/executor inherited by a router delegate.""" + return _ROUTED_SYNC_ADMISSION.get(), _ROUTED_SYNC_EXECUTOR.get() + + # ---- Unit conversion ---- _UNIT_TO_SECONDS: dict[str, float] = { @@ -258,6 +319,7 @@ async def get_products_incremental( __all__ = [ "IncrementalGetProducts", "ProductsCheckpoint", + "SyncExecutorAdmission", "project_incomplete_response", "resolve_time_budget", ] diff --git a/tests/test_decisioning_capabilities_projection.py b/tests/test_decisioning_capabilities_projection.py index 0c21b2fc..68529534 100644 --- a/tests/test_decisioning_capabilities_projection.py +++ b/tests/test_decisioning_capabilities_projection.py @@ -639,6 +639,9 @@ def get_adcp_capabilities_for_request(self, params=None, context=None): assert exc_info.value.code == "INTERNAL_ERROR" assert exc_info.value.details["caused_by"]["type"] == "RuntimeError" + assert exc_info.value.details["caused_by"] == {"type": "RuntimeError"} + assert "tenant lookup failed" not in str(exc_info.value) + assert "tenant lookup failed" not in str(exc_info.value.details) def test_request_scoped_capabilities_hook_may_be_async( diff --git a/tests/test_decisioning_dispatch.py b/tests/test_decisioning_dispatch.py index 55841987..ff0587a9 100644 --- a/tests/test_decisioning_dispatch.py +++ b/tests/test_decisioning_dispatch.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import threading import warnings from concurrent.futures import ThreadPoolExecutor from contextvars import ContextVar @@ -1642,6 +1643,143 @@ async def get_products(self, req: _StrictSubRequest, ctx): assert on_failure_calls[0] is exc_info.value +@pytest.mark.asyncio +async def test_cancellation_fires_on_failure_and_propagates_unchanged( + executor: ThreadPoolExecutor, +) -> None: + entered = asyncio.Event() + on_failure_calls: list[BaseException] = [] + + class _WaitingPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + async def get_products(self, req: _BaseRequest, ctx): + entered.set() + await asyncio.Event().wait() + + async def _on_failure(exc: BaseException) -> None: + on_failure_calls.append(exc) + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _WaitingPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_failure=_on_failure, + ) + ) + await entered.wait() + task.cancel("client disconnected") + + with pytest.raises(asyncio.CancelledError, match="client disconnected") as exc_info: + await task + assert on_failure_calls == [exc_info.value] + + +@pytest.mark.asyncio +async def test_sync_cancellation_settles_success_before_on_complete( + executor: ThreadPoolExecutor, +) -> None: + entered = threading.Event() + release = threading.Event() + settled = asyncio.Event() + completed: list[Any] = [] + failed: list[BaseException] = [] + + class _SyncPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + def get_products(self, req: _BaseRequest, ctx): + entered.set() + release.wait(timeout=2) + return {"products": []} + + async def _on_complete(result: Any) -> None: + completed.append(result) + settled.set() + + async def _on_failure(exc: BaseException) -> None: + failed.append(exc) + settled.set() + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _SyncPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_complete=_on_complete, + on_failure=_on_failure, + ) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel("client disconnected") + with pytest.raises(asyncio.CancelledError, match="client disconnected"): + await task + assert completed == [] and failed == [] + + release.set() + await asyncio.wait_for(settled.wait(), 1) + assert completed == [{"products": []}] + assert failed == [] + + +@pytest.mark.asyncio +async def test_sync_cancellation_settles_real_failure_before_on_failure( + executor: ThreadPoolExecutor, +) -> None: + entered = threading.Event() + release = threading.Event() + settled = asyncio.Event() + failures: list[BaseException] = [] + + class _FailingSyncPlatform(DecisioningPlatform): + capabilities = DecisioningCapabilities() + accounts = SingletonAccounts(account_id="x") + + def get_products(self, req: _BaseRequest, ctx): + entered.set() + release.wait(timeout=2) + raise RuntimeError("worker failed") + + async def _on_failure(exc: BaseException) -> None: + failures.append(exc) + settled.set() + + ctx = _build_request_context(ToolContext(), Account(id="x"), None) + task = asyncio.create_task( + _invoke_platform_method( + _FailingSyncPlatform(), + "get_products", + _BaseRequest(known_field="wait"), + ctx, + executor=executor, + registry=InMemoryTaskRegistry(), + on_failure=_on_failure, + ) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert failures == [] + + release.set() + await asyncio.wait_for(settled.wait(), 1) + assert len(failures) == 1 + assert isinstance(failures[0], RuntimeError) + assert str(failures[0]) == "worker failed" + + def test_coerce_varargs_annotation_is_noop() -> None: """Annotated *args should not trigger coercion — VAR_POSITIONAL guard fires.""" diff --git a/tests/test_proposal_lifecycle_e2e.py b/tests/test_proposal_lifecycle_e2e.py index e41a1677..791e330e 100644 --- a/tests/test_proposal_lifecycle_e2e.py +++ b/tests/test_proposal_lifecycle_e2e.py @@ -31,6 +31,7 @@ import asyncio import sys +import threading from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from pathlib import Path @@ -1398,6 +1399,96 @@ async def _seed_committed_proposal(handler: PlatformHandler) -> None: ) +@pytest.mark.asyncio +async def test_create_media_buy_cancellation_releases_reservation( + executor: ThreadPoolExecutor, + registry: InMemoryTaskRegistry, +) -> None: + from examples.sales_proposal_mode_seller.src.app import build_router + + router = build_router() + store = router.proposal_store_for_tenant("default") + handler = _build_handler(router, executor, registry) + await _seed_committed_proposal(handler) + + entered = asyncio.Event() + target_platform = router._platforms["default"] # noqa: SLF001 + original = target_platform.create_media_buy + + async def _waiting_create(req: Any, ctx: Any) -> Any: + del req, ctx + entered.set() + await asyncio.Event().wait() + + target_platform.create_media_buy = _waiting_create # type: ignore[method-assign] + try: + task = asyncio.create_task( + handler.create_media_buy(_build_create_media_buy_request("cancel"), ToolContext()) + ) + await entered.wait() + + reserved = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert reserved is not None and reserved.state == ProposalState.CONSUMING + + task.cancel("buyer disconnected") + with pytest.raises(asyncio.CancelledError, match="buyer disconnected"): + await task + + released = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert released is not None and released.state == ProposalState.COMMITTED + finally: + target_platform.create_media_buy = original # type: ignore[method-assign] + + +@pytest.mark.asyncio +async def test_sync_create_media_buy_cancellation_waits_for_worker_success( + executor: ThreadPoolExecutor, + registry: InMemoryTaskRegistry, +) -> None: + """A cancelled request cannot release a reservation while its thread runs.""" + router = build_router() + store = router.proposal_store_for_tenant("default") + handler = _build_handler(router, executor, registry) + await _seed_committed_proposal(handler) + + entered = threading.Event() + release = threading.Event() + target_platform = router._platforms["default"] # noqa: SLF001 + original = target_platform.create_media_buy + + def _blocking_create(req: Any, ctx: Any) -> Any: + del req, ctx + entered.set() + release.wait(timeout=2) + return {"media_buy_id": "mb_sync_cancel", "status": "active"} + + target_platform.create_media_buy = _blocking_create # type: ignore[method-assign] + try: + task = asyncio.create_task( + handler.create_media_buy(_build_create_media_buy_request("sync-cancel"), ToolContext()) + ) + assert await asyncio.to_thread(entered.wait, 1) + task.cancel("buyer disconnected") + with pytest.raises(asyncio.CancelledError, match="buyer disconnected"): + await task + + reserved = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + assert reserved is not None and reserved.state == ProposalState.CONSUMING + + release.set() + for _ in range(100): + consumed = await store.get(PROPOSAL_ID, expected_account_id="acct_demo") + if consumed is not None and consumed.state == ProposalState.CONSUMED: + break + await asyncio.sleep(0.01) + else: + pytest.fail("sync worker success did not finalize proposal reservation") + assert consumed.media_buy_id == "mb_sync_cancel" + finally: + release.set() + target_platform.create_media_buy = original # type: ignore[method-assign] + + @pytest.mark.asyncio async def test_create_media_buy_handoff_finalizes_consumption_on_completion( executor: ThreadPoolExecutor, diff --git a/tests/test_time_budget.py b/tests/test_time_budget.py index d8b08ebd..974b111e 100644 --- a/tests/test_time_budget.py +++ b/tests/test_time_budget.py @@ -14,8 +14,8 @@ from __future__ import annotations import asyncio +import threading from concurrent.futures import ThreadPoolExecutor -from unittest.mock import AsyncMock import pytest @@ -24,6 +24,8 @@ DecisioningPlatform, IncrementalGetProducts, InMemoryTaskRegistry, + LazyPlatformRouter, + PlatformRouter, ProductsCheckpoint, SingletonAccounts, ) @@ -35,7 +37,6 @@ from adcp.server.base import ToolContext from adcp.types import GetProductsRequest - # --------------------------------------------------------------------------- # resolve_time_budget # --------------------------------------------------------------------------- @@ -204,12 +205,20 @@ async def get_products(self, req, ctx): ) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 pid = products[0].get("product_id") if isinstance(products[0], dict) else products[0].product_id # type: ignore[union-attr] assert pid == "p1" # No incomplete key / field when fully resolved - incomplete = result.get("incomplete") if isinstance(result, dict) else getattr(result, "incomplete", None) + incomplete = ( + result.get("incomplete") + if isinstance(result, dict) + else getattr(result, "incomplete", None) + ) assert not incomplete @@ -232,7 +241,11 @@ async def get_products(self, req, ctx): ) req = GetProductsRequest.model_construct(account=None, time_budget=None) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 @@ -258,10 +271,204 @@ async def get_products(self, req, ctx): time_budget=_make_time_budget(interval=1, unit="campaign"), ) result = await handler.get_products(req, context=ToolContext()) - products = result.get("products") if isinstance(result, dict) else list(getattr(result, "products", [])) # type: ignore[union-attr] + products = ( + result.get("products") + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) # type: ignore[union-attr] assert len(products) == 1 +@pytest.mark.asyncio +async def test_sync_timeout_admission_saturates_without_executor_queue_growth( + executor: ThreadPoolExecutor, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Timed-out threads retain permits; later short-budget calls are not submitted.""" + release = threading.Event() + two_started = threading.Event() + calls = 0 + calls_lock = threading.Lock() + deadline = [0.05] + + class _BlockingSyncSeller(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="test") + + def get_products(self, req, ctx): + nonlocal calls + with calls_lock: + calls += 1 + if calls == 2: + two_started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": f"p{calls}", "name": "Recovered"}]} + + monkeypatch.setattr("adcp.decisioning.handler.resolve_time_budget", lambda _value: deadline[0]) + handler = PlatformHandler( + _BlockingSyncSeller(), + executor=executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=2, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="seconds"), + ) + + first_two = [ + asyncio.create_task(handler.get_products(req, context=ToolContext())) for _ in range(2) + ] + assert await asyncio.to_thread(two_started.wait, 1.0) + timed_out = await asyncio.gather(*first_two) + assert all(getattr(result, "incomplete", None) for result in timed_out) + + # Both permits remain attached to the still-running worker threads. This + # request exhausts its budget waiting and never reaches executor.submit. + saturated = await handler.get_products(req, context=ToolContext()) + assert getattr(saturated, "incomplete", None) + assert calls == 2 + + # Once real worker completion callbacks return the permits, admission + # recovers and a later call executes normally. + release.set() + deadline[0] = 0.5 + recovered = await handler.get_products(req, context=ToolContext()) + products = ( + recovered.get("products", []) + if isinstance(recovered, dict) + else list(getattr(recovered, "products", [])) + ) + assert len(products) == 1 + assert calls == 3 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("router_kind", ["eager", "lazy"]) +async def test_router_sync_timeout_uses_bounded_admission( + router_kind: str, + executor: ThreadPoolExecutor, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Eager and lazy async routers must not bypass sync-child admission.""" + release = threading.Event() + started = threading.Event() + calls = 0 + deadline = [0.05] + + class _BlockingChild(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="child") + + def get_products(self, req, ctx): + nonlocal calls + calls += 1 + started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": f"p{calls}", "name": "Recovered"}]} + + accounts = SingletonAccounts( + account_id="router", + metadata_factory=lambda: {"tenant_id": "tenant-a"}, + ) + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + if router_kind == "eager": + platform: DecisioningPlatform = PlatformRouter( + accounts=accounts, + platforms={"tenant-a": _BlockingChild()}, + capabilities=capabilities, + ) + else: + platform = LazyPlatformRouter( + accounts=accounts, + factory=lambda _tenant_id: _BlockingChild(), + capabilities=capabilities, + ) + + monkeypatch.setattr("adcp.decisioning.handler.resolve_time_budget", lambda _value: deadline[0]) + handler = PlatformHandler( + platform, + executor=executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=1, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="seconds"), + ) + + first = asyncio.create_task(handler.get_products(req, context=ToolContext())) + assert await asyncio.to_thread(started.wait, 1.0) + assert getattr(await first, "incomplete", None) + + # The first timed-out child still owns the sole permit, so this request + # times out waiting for admission and is never submitted. + assert getattr(await handler.get_products(req, context=ToolContext()), "incomplete", None) + assert calls == 1 + + release.set() + deadline[0] = 0.5 + recovered = await handler.get_products(req, context=ToolContext()) + products = ( + recovered.get("products", []) + if isinstance(recovered, dict) + else list(getattr(recovered, "products", [])) + ) + assert len(products) == 1 + assert calls == 2 + + +@pytest.mark.asyncio +async def test_sync_campaign_requests_bypass_deadline_admission() -> None: + """Campaign-unit semantics remain unlimited by the deadline-only gate.""" + release = threading.Event() + two_started = threading.Event() + calls = 0 + calls_lock = threading.Lock() + + class _CampaignSyncSeller(DecisioningPlatform): + capabilities = DecisioningCapabilities(specialisms=["sales-non-guaranteed"]) + accounts = SingletonAccounts(account_id="test") + + def get_products(self, req, ctx): + nonlocal calls + with calls_lock: + calls += 1 + if calls == 2: + two_started.set() + release.wait(timeout=2.0) + return {"products": [{"product_id": "campaign", "name": "Campaign"}]} + + with ThreadPoolExecutor(max_workers=2) as campaign_executor: + handler = PlatformHandler( + _CampaignSyncSeller(), + executor=campaign_executor, + registry=InMemoryTaskRegistry(), + timed_sync_get_products_limit=1, + ) + req = GetProductsRequest.model_construct( + account=None, + time_budget=_make_time_budget(interval=1, unit="campaign"), + ) + tasks = [ + asyncio.create_task(handler.get_products(req, context=ToolContext())) for _ in range(2) + ] + assert await asyncio.to_thread(two_started.wait, 1.0) + release.set() + results = await asyncio.gather(*tasks) + + assert calls == 2 + assert all( + len( + result.get("products", []) + if isinstance(result, dict) + else list(getattr(result, "products", [])) + ) + == 1 + for result in results + ) + + @pytest.mark.asyncio async def test_get_products_timeout_logs_warning(executor, caplog): """A timeout emits a WARNING with budget info.""" @@ -287,15 +494,15 @@ async def test_get_products_timeout_logs_warning(executor, caplog): def test_incremental_get_products_importable_from_decisioning(): - from adcp.decisioning import IncrementalGetProducts as IGP # noqa: F401 + from adcp.decisioning import IncrementalGetProducts as ImportedIncrementalGetProducts - assert IGP is IncrementalGetProducts + assert ImportedIncrementalGetProducts is IncrementalGetProducts def test_products_checkpoint_importable_from_decisioning(): - from adcp.decisioning import ProductsCheckpoint as PC # noqa: F401 + from adcp.decisioning import ProductsCheckpoint as ImportedProductsCheckpoint - assert PC is ProductsCheckpoint + assert ImportedProductsCheckpoint is ProductsCheckpoint def test_products_checkpoint_accumulates_batches():