fix(decisioning): supervise timed and canceled work - #1000
Conversation
There was a problem hiding this comment.
Approving. Right shape: a cancelled or timed-out request can no longer release a durable reservation while the worker thread is still mutating state, and the fix bounds the pre-existing thread-pool slot leak instead of papering over it.
The load-bearing move is asyncio.shield over the wrapped worker future plus a supervised settle-task (_settle_cancelled_sync_lifecycle) that fires on_complete/on_failure from the worker's real terminal outcome — so the CONSUMING→CONSUMED transition happens even after the buyer disconnects. Verified in test_sync_create_media_buy_cancellation_waits_for_worker_success.
Things I checked
- Permit accounting is leak-free.
SyncExecutorAdmissionacquires beforeexecutor.submit, releases exactly once via the future's done-callback (fires on success and worker exception), and releases in theexcepton submit failure without adding the callback. Release is marshaled back withloop.call_soon_threadsafe— correct, sinceasyncio.Semaphoreis not thread-safe (time_budget.py:95,dispatch.py:1445-1470). - No double settlement. Direct-sync cancel sets
sync_lifecycle_continues=Truein the innerexcept asyncio.CancelledError; the re-raisedCancelledErrorthen re-enters the outerexcept BaseException, which sees the flag and skips_safe_on_failure_call. Router-path cancel carries the_adcp_sync_worker_futuremarker and is settled once in the outer handler.security-reviewerandcode-reviewerboth traced this — single settlement each path. _project_invocation_resultextraction is faithful.params→request_paramsrename threads through; the sync/handoff/workflow arms are unchanged; happy path is unchanged.get_adcp_capabilitiesINTERNAL_ERROR alignment closes a wire leak. It was the last outlier hand-rollingcaused_by={type, message}; it now routes through_internal_error_message/_internal_error_details, which emit{type}only.str(exc)is deliberately dropped so a platform that raises on secret material can't leak an OAuth secret on the wire. Confirmed by the new assertions attest_decisioning_capabilities_projection.py:640-642.- Not a wire-contract break.
ad-tech-protocol-expert: sound — perschemas/cache/3.1/core/error.json,detailsisadditionalProperties: trueandcaused_byis not a defined property; the normative recovery signal iserror.recovery(stillterminal).incomplete[]-when-saturated is invisible to the buyer and matches the existing timeout contract. - No tenant/auth regression.
_run_sync_delegatecarries only admission + executor through two ContextVars; the tenant-scopedctxis still a positional arg, and_bind_routed_sync_executionresets both tokens infinallyso nothing bleeds across sibling delegates. proposal_dispatch.pyexcept Exception→except BaseException. Awaitingrelease_consumptioninside aCancelledErrorhandler is safe — the exception is already caught, the bareraisere-raises it, and the innerexcept BaseExceptionabsorbs a second cancel during release. This is the point: cancellation must not strand a CONSUMING reservation.
Follow-ups (non-blocking — file as issues)
- Routed, no-deadline
get_productscancel drops itson_complete. In_run_sync_delegatethesync_admission is Nonebranch (platform_router.py:135-139) sets the marker and re-raises, but dispatch's outer handler only builds a settle-task whenon_failure is not None.get_productswireson_complete=_persist_draft_hookwith noon_failure, so on client-disconnect the worker is orphaned (stray "Task exception was never retrieved" on failure; draft-persist silently skipped on success). No reservation is held on this path, so it's not corruption — but it's the one gap in "supervise cancelled work." Supervise wheneveron_complete is not None or on_failure is not None. asyncio.BoundedSemaphoreoverSemaphore. Accounting is correct today, but a future strayrelease()would silently raise the ceiling abovetimed_sync_get_products_limitrather than fail-fast (time_budget.py:95).- Admission is per-handler, not per-tenant. One tenant bursting slow/cancelled timed
get_productscan hold allworkers//2permits and stall other tenants' timed calls. Strictly better than the pre-PR shared-pool exhaustion; key onctx.accountif per-tenant fairness matters.
Minor nits (non-blocking)
- Stale comment now contradicts the invariant.
dispatch.py:1529-1532still reads "We expose only the exception class name + str (not the traceback)."_internal_error_detailsno longer exposesstr— that omission is the whole point of the sanitization this PR leans on. Pre-existing, but adjacent enough that a maintainer could "restore" thestrto match the comment and reintroduce the leak. Drop "+ str."
Independent concurrency review claimed in the PR body; the test matrix (direct-sync, eager/lazy router, campaign-unit bypass, saturation-without-queue-growth) covers the new paths well. Safe to merge once CI is green.
KonstantinMirin
left a comment
There was a problem hiding this comment.
Review — PR #1000
Overview — The direct-sync path now does what the title says. asyncio.shield over the wrapped worker plus _settle_cancelled_sync_lifecycle fires the lifecycle hooks from the worker's real terminal outcome, and test_sync_create_media_buy_cancellation_waits_for_worker_success proves the CONSUMING→CONSUMED transition survives a buyer disconnect. Permit accounting is single-release on both submit sites. Two things need work before this lands: the new except BaseException arm in _run() releases a proposal reservation when a background handoff task is cancelled, which lets a retry double-book a media buy; and both supervision gates key on on_failure, which get_products never wires, so the timed path the PR is named for is the one path that stays unsupervised.
Should fix
Findings 2, 4, 5 and 10 are one root, already open as #1004: a policy re-detected by inspecting objects (on_failure is not None, executor._max_workers) or hand-copied to a subset of its sites, instead of owned at one boundary. Evidence from this PR is added there; the sites below stand on their own.
1. Cancelling a background handoff task releases the reservation while the adopter's work is still outstanding
src/adcp/decisioning/dispatch.py:1972-1977:
except BaseException as exc:
if on_failure is not None:
await _safe_on_failure_call(on_failure, exc, method_name)
raiseFor create_media_buy that hook is _release_reservation_hook → release_proposal_reservation → CONSUMING → COMMITTED. Background handoff tasks are detached create_tasks (dispatch.py:2069); ordinary loop/SIGTERM shutdown cancels them. Cancel one after the handoff fn has issued its upstream create:
HEAD state after bg cancel: ProposalState.COMMITTED | registry row: submitted
RETRY ACCEPTED -> {'media_buy_id': 'mb_duplicate_2'}
BASE state after bg cancel: ProposalState.CONSUMING
RETRY REJECTED -> AdcpError[PROPOSAL_NOT_COMMITTED / correctable]
One committed proposal, two media buys upstream. The arm also skips _fail(), so the registry row stays submitted with no terminal webhook — the buyer is left holding a task that never terminates, which is exactly what provokes the retry.
Root cause: this arm infers "our asyncio task was cancelled" ⇒ "the adopter's work did not happen". That is the inference the rest of this PR exists to refute on the request path. The handoff fn is not shielded here, and for a sync fn it is a run_in_executor thread that cannot be stopped at all.
Either do not fire on_failure from this arm (leave CONSUMING, which fails closed and is what eviction exists for), or mirror _settle_cancelled_sync_lifecycle — shield the fn and settle from its real terminal outcome. Either way a release must be paired with a terminal registry state so the buyer is not invited to double-book. Lines 1972-1977 never execute under the suite; add a test that cancels a background handoff task with a reserved proposal.
Separately, "then remain cancellation to the task scheduler" in that comment does not parse.
2. Supervision keys on on_failure, so no timed get_products is supervised
Raised in the aao-ipr-bot review of 2026-07-29 ("Routed, no-deadline get_products cancel drops its on_complete"), still open on this push, and wider than the routed no-deadline case — it covers the whole timed path.
Both settle sites gate on the same field:
src/adcp/decisioning/dispatch.py:1469—except asyncio.CancelledError: if on_failure is not None:src/adcp/decisioning/dispatch.py:1605—if isinstance(nested_sync_future, asyncio.Future) and on_failure is not None:
get_products is the only method that carries a deadline and the only caller that passes sync_admission, and it wires on_complete=_persist_draft_hook with no on_failure (src/adcp/decisioning/handler.py:1974-1979). Cancel a sync get_products mid-worker and let the worker finish:
on_complete only -> ON_COMPLETE CALLS: []
add on_failure=noop -> ON_COMPLETE CALLS: [{'products': []}]
The gate is the whole difference. Three consequences follow from it:
- The persist-draft hook is silently dropped on every timed-out sync
get_products. - The shielded future has no owner in the else-branch, so a worker that raises after the deadline lands in asyncio's default handler. New at HEAD, absent at
b7ef1dfc:That is the adopter's raw exception text at ERROR, outside the framework's error funnel — while the same PR stripsERROR:asyncio:Future exception was never retrieved future: <Future finished exception=RuntimeError('upstream token sk-SECRET-LEAK rejected')> Traceback (most recent call last): ... (full adopter traceback)str(exc)fromdetails.caused_byso a platform raising on secret material cannot leak it. The routed branch has the same shape viaasyncio.create_task(platform_router.py:137-139) and produces "Task exception was never retrieved". _adcp_sync_worker_future, set atplatform_router.py:166, has one consumer (dispatch.py:1604) behind the same gate, so on theget_productspath the marker is written and never read.
The same root shows up one function over: the new pre_handoff_reject → on_failure branch at dispatch.py:1668-1669 is dead by construction. pre_handoff_reject= is passed by exactly two callers (handler.py:1976 get_products, handler.py:2676 get_signals); on_failure= by exactly one (handler.py:2192 create_media_buy). No caller passes both, and line 1669 never executes.
Root cause: the lifecycle contract is the pair (on_complete, on_failure) and the supervision predicate reads one half of it. It should ask whether a lifecycle exists at all. The spawn block — create_task(_settle_cancelled_sync_lifecycle(...)) + _SUPERVISED_SYNC_LIFECYCLES.add + discard — is also written out twice verbatim at 1471-1487 and 1607-1623, which is what let both copies share the wrong predicate.
Gate both sites on on_complete is not None or on_failure is not None, extract the spawn into one _supervise_sync_lifecycle(...), and attach a consumer to the shielded future in every branch so a post-cancellation worker failure is logged by the framework rather than by asyncio's GC hook. Add a cancellation test for an on_complete-only timed sync get_products, covering both the late-success and late-raise legs.
3. except BaseException around awaited cleanup swallows the cancellation
Three handlers widened from except Exception to except BaseException around an await, then log and continue:
src/adcp/decisioning/dispatch.py:1774(_safe_on_failure_call)src/adcp/decisioning/proposal_dispatch.py:755src/adcp/decisioning/proposal_dispatch.py:888
on_failure hooks do durable-store I/O, so they contain await points where a cancellation lands. When it does, the hook's CancelledError is logged and discarded and the caller re-raises the original exception. Adopter raises AdcpError, request task is cancelled while on_failure is mid store round-trip:
HEAD RESULT: cancellation SWALLOWED, task raised AdcpError[INTERNAL_ERROR / terminal]
cancelled() -> False | reservation released -> []
BASE RESULT: task honoured cancellation
cancelled() -> True | reservation released -> []
Worst of both: a graceful-shutdown drain doing task.cancel(); await task gets a completed task back, and the reservation release is left half-done anyway.
Root cause: log-and-continue is the right policy for Exception and the wrong one for a BaseException that is not an Exception. Catch BaseException so the cleanup runs, then re-raise when the caught exception is not an Exception — the hook's cancellation is not the framework's to absorb.
One more at proposal_dispatch.py:747: the widened handler guards the try: at lines 698-746, and that block contains no await — _derive_packages, the params.packages assignment and validate_capability_overlap are all synchronous. CancelledError cannot be delivered there, so the new comment ("Cancellation and shutdown must not strand a durable reservation") describes a state the code cannot reach, and no test can be written for it. Narrow it back, or state the reachable trigger (a synchronous SystemExit/KeyboardInterrupt inside derivation) and test that. Lines 755, 756 and 888 never execute under the suite.
4. The admission limit is read off executor._max_workers
src/adcp/decisioning/handler.py:1311-1317:
worker_count = int(getattr(executor, "_max_workers", 1))
admission_limit = ... else max(1, worker_count // 2)serve.py:120-124 documents BYO executors as "for operators with audit-instrumented thread pools or wrappers around stdlib's executor". A wrapper is exactly the shape that has no _max_workers:
wrapper around a 64-worker pool -> admission limit 1
plain 64-worker pool -> admission limit 32
Every deadline-managed sync get_products in that deployment then serialises behind one permit held for the full worker duration, with no warning and no public way to observe the value. The parameter is annotated executor: ThreadPoolExecutor (handler.py:1275-1278), so by the declared type the getattr default is dead code; for the shape the docs advertise, it drops the limit to 1 with no warning. int() on an executor whose _max_workers is not int-able raises at handler construction.
Root cause: the composition root owns pool sizing — serve.py already resolved thread_pool_size / _default_thread_pool_size() at serve.py:70,268 — and the handler re-derives it downstream from a CPython-internal field that is not part of any stability contract. Thread the resolved worker count (or the resolved admission limit) from create_adcp_server_from_platform into PlatformHandler and keep timed_sync_get_products_limit as the explicit override. If a BYO executor cannot supply a count, require the override rather than silently choosing 1.
5. Supervised submit has no owned home: written twice, and a third site was left behind
src/adcp/decisioning/dispatch.py:1445-1467 and src/adcp/decisioning/platform_router.py:141-162 run the same sequence in the same order — await admission.acquire() → executor.submit(call) → except BaseException: admission.release(); raise → a _release_admission done-callback wrapping loop.call_soon_threadsafe(admission.release) in try/except RuntimeError → asyncio.wrap_future(..., loop=loop) → await asyncio.shield(worker). They differ only in identifiers and indentation, and already diverge on the fallback branch and on what happens at cancellation. Every step is load-bearing for permit accounting.
The third framework sync-submit site was not migrated. src/adcp/decisioning/proposal_dispatch.py:244 is still the pre-PR shape:
result = await loop.run_in_executor(
executor, functools.partial(ctx_snapshot.run, method, finalize_req, ctx)
)No shield, no admission, no settlement. Cancel a sync ProposalManager.finalize_proposal and the asyncio side unwinds immediately while the thread runs to completion and applies the adopter's finalize side effects; the framework's store.commit below never runs:
worker completed: True
proposal state after cancelled finalize: ProposalState.DRAFT
That is the same "worker mutated state, ledger disagrees" failure the PR body claims to close, on the file this PR opened to close it.
Root cause: "submit sync adopter work under supervision" is inline in dispatch and re-implemented in platform_router, so the third call site has nothing to call. SyncExecutorAdmission (time_budget.py:82) exposes only acquire/release and leaves the ordering contract — release exactly once, from the concurrent future's callback, marshalled back to the loop — to be re-derived at each site. Extract one submit_supervised(executor, admission, call) -> asyncio.Future next to the semaphore and route all three callers through it, with an e2e test for the cancelled finalize alongside test_sync_create_media_buy_cancellation_waits_for_worker_success.
6. The settle task reruns the full projection, so a cancelled request can mint a task the buyer never learns about
_settle_cancelled_sync_lifecycle (src/adcp/decisioning/dispatch.py:1715) calls _project_invocation_result at dispatch.py:1737, which reruns the full result projection rather than only the lifecycle hooks. When the cancelled sync adopter returns a TaskHandoff, the settle path runs _project_handoff: it issues a registry task_id, launches the background handoff, persists the terminal artifact, and emits the completion webhook if a push config was supplied. The Submitted envelope it builds is then discarded — there is no waiter. Cancel a sync create_media_buy that returns a handoff, then release the worker:
REGISTRY RECORDS: {'task_dc9d25f8c16e4add': TaskRecord(state='completed',
task_type='create_media_buy', result={'media_buy_id': 'mb_orphan', ...})}
Absent a push config the buyer holds no task_id, so the task is unreachable via tasks/get and their only recourse is a retry that re-executes the work. This arm could not run before the PR (await run_in_executor raised CancelledError before the handoff branch), so the diff introduces it, and neither of the two new dispatch tests exercises it — they cover a plain dict return and a RuntimeError.
Decide and state the contract: either restrict the settle path to the lifecycle hooks it exists for and refuse to promote a handoff once the caller is gone, or keep the promotion and make the task recoverable — log the issued task_id with the request correlation id at WARNING, and name the push-notification surface (schemas/cache/3.1/core/protocol-envelope.json @ AdCP 3.1.8) as the delivery channel that makes it legal. Either way pin the chosen behavior with a test for the TaskHandoff and WorkflowHandoff arms.
7. The bound executor is ignored without an admission controller, and the worker future travels on the exception
Two defects, both in _run_sync_delegate.
src/adcp/decisioning/platform_router.py:137 reads if admission is None or executor is None: → asyncio.to_thread(...). _bind_routed_sync_execution(sync_admission, executor) always binds a non-None executor (dispatch.py:1425), but the delegate only uses it when an admission controller is also present, i.e. only for deadline-managed get_products. Every other routed sync child — create_media_buy, update_media_buy, refine_get_products, all synthesized delegates — runs on the loop's default executor, outside the adopter's BYO pool, with the right executor sitting in the ContextVar one line above. With a BYO pool named FRAMEWORK, a routed sync child on a no-deadline get_products ran on thread asyncio_0. That undercuts the D5 BYO-executor contract documented in serve.py.
src/adcp/decisioning/platform_router.py:166 signals dispatch by setattr(exc, "_adcp_sync_worker_future", worker), read back at dispatch.py:1604 as Any and narrowed only by isinstance(..., asyncio.Future). platform_router is a DecisioningPlatform implementation below dispatch, and it now encodes dispatch's settlement protocol on an exception instance. Nothing fails at type-check time if the attribute name, producer or consumer drifts, and the channel is lossy: any frame between the router and dispatch that catches CancelledError and re-raises a fresh instance (adopter middleware, a wrapping platform) drops the marker with no log at either end.
Root cause for both: the executor choice is execution plumbing, not time-budget policy, and putting the ThreadPoolExecutor ContextVar in time_budget.py next to the deadline machinery is what made "no deadline" read as "no configured executor" and left the live worker future with nowhere typed to live. Split the condition (executor is not None → submit to it; admission is not None → additionally gate on a permit), carry the worker future on the same scope object the two modules already share (or on a SyncWorkerCancelled(CancelledError) subclass with a typed worker: asyncio.Future[Any]), and log when dispatch unwinds a cancelled routed sync call with no marker. Add a test asserting a routed sync child runs on the handler's executor (thread_name_prefix) with and without a time budget.
8. New surfaces shipped without a test that fails without them
Four groups, each verified by mutation or by coverage of the added lines:
- The public knob.
timed_sync_get_products_limitwas added tocreate_adcp_server_from_platform(serve.py:84,381) andserve(serve.py:466,588), and appears in tests only as a directPlatformHandler(...)kwarg (tests/test_time_budget.py:312,393,447). Deleting both plumbing lines leaves the suite green — the parameter becomes a documented no-op and nothing notices. TheValueErrorguard attime_budget.py:93never executes, sotimed_sync_get_products_limit=0producing a permanently saturated server at boot is unasserted, as is the documentedmax(1, worker_count // 2)default. - Permit accounting on return and on submit failure. Making
SyncExecutorAdmission.release()release twice — an unbounded ceiling after every completed worker — leavestests/test_time_budget.pyandtests/test_decisioning_dispatch.pygreen (84 passed). The submit-failure release paths atdispatch.py:1449-1452andplatform_router.py:148-150never execute. Add: an executor stub whosesubmitraisesRuntimeError(post-shutdownbehavior) → the next timed call is still admitted; and after N timed calls fully complete, a burst of N+1 blocks exactly one. - The lazy proposal-manager arm.
platform_router.py:1099is the only one of the eight migratedasyncio.to_thread→_run_sync_delegatesites with no execution.test_router_sync_timeout_uses_bounded_admission[lazy]covers the platform arm, notproposal_manager_for_tenant, so that path silently changed thread-dispatch semantics with no test. - The settle task's failure arms.
dispatch.py:1750,1754(except BaseException+logger.exception) anddispatch.py:1774never execute; reverting_safe_on_failure_calltoexcept Exceptionleaves the suite green.
9. Admission saturation emits the timeout incomplete[] verbatim
The bounded-admission path introduces a genuinely new server-side condition — the seller never searched, because no permit came free before the budget expired — and reuses the pre-existing timeout payload word for word (src/adcp/decisioning/time_budget.py:219-233): "time_budget exhausted (N unit); return the best results achievable within the budget. Retry with a larger time_budget…". Per schemas/cache/3.1/media-buy/get-products-response.json @ AdCP 3.1.8, incomplete[] "Declares what the seller could not finish within the buyer's time_budget or due to internal limits", and the per-entry description is the "Human-readable explanation of what is missing and why". The spec separates exactly these two causes; today a buyer cannot tell "searched and ran out of time" from "declined admission and searched nothing". scope: "products" is correct on both.
Grading is thin on the same path: both new saturation tests assert only getattr(result, "incomplete", None) truthiness, so a regression emitting an off-enum scope or an empty incomplete[] (minItems: 1) would still pass. Give the saturation path its own description naming the internal admission limit, and assert the full projected incomplete[0] (scope, description, products == []).
10. The sanitized caused_by shape was made uniform, then stopped one site short
The diff imports _internal_error_details into handler.py and routes the get_adcp_capabilities INTERNAL_ERROR through it (handler.py:1650-1653), dropping str(exc) from details.caused_by. That is the right call and the new assertions at tests/test_decisioning_capabilities_projection.py:640-642 pin it. Three follow-throughs are missing:
handler.py:2107still hand-rollsdetails={"caused_by": {"type": type(exc).__name__}}for theProductConfigStore.lookup_implementation_configsSERVICE_UNAVAILABLEwrap — the same wire shape under a different name, so the next change to the sanitizeddetailscontract has to be made twice. Route it through_internal_error_details(exc), deciding explicitly whether the helper'sdetails.validation_errorsaddition is wanted there, or factor thecaused_by-only core into a helper both call.dispatch.py:1583-1586still reads "We expose only the exception class name + str (not the traceback)" directly above the_internal_error_details(exc)call, which emits{"type": ...}only. This PR makes that sanitization the single shape, so the comment is now the one place telling a future maintainer to putstr(exc)back on the wire. Drop "+ str" and state the invariant positively: class name only, message in the server log vialogger.exception. Raised in the 2026-07-29 review, still open.- The change is buyer-visible on two values — the
messagestring and the loss ofdetails.caused_by.message— and reaches MCP and A2A buyers. The PR's Compatibility section says only "Existing direct asynchronous platform behavior is unchanged". Note thedetails.caused_by.messageremoval there and in the changelog, and citeschemas/cache/3.1/core/error.json@ AdCP 3.1.8 (detailsopen withadditionalProperties: true,caused_bynon-normative,recoveryunchanged atterminal) as the grounding.
Notes
- Permit accounting under a cancelled
acquire()was the shape most likely to bite — shorttime_budgetwhile permits are held, eachwait_forcancelling a pendingacquire, capacity ratcheting to zero. 200 cancelled acquires per trial, 5 trials, on 3.12 and on 3.11'sasyncio.Semaphore: effective capacity returned to exactly the limit every time. Not exercised on 3.10, which ships the olderSemaphore.acquire— worth one CI run. - Per-tenant fairness of
SyncExecutorAdmissionis out of scope: one tenant can hold every permit, but the diff replaces no bound with a bound, and keying the semaphore onctx.accountis a capacity-policy decision rather than a defect this PR introduces. _SUPERVISED_SYNC_LIFECYCLESis a module-global task set with no shutdown drain. Out of scope: it follows the module's existing convention (_BACKGROUND_HANDOFF_TASKSatdispatch.py:2091,_BACKGROUND_WEBHOOK_TASKSatwebhook_emit.py:107), so changing it is a repo-wide decision.tests/test_pg_idempotency_backend.py::test_delete_expired_defaults_to_wall_clockfailed once across 16 full-suite runs, while two suites ran concurrently, and passed in the other 15. Pre-existing wall-clock sensitivity, unrelated to this diff — recorded so the single red line in the logs is not read as a finding.
fcc68f5 to
d714761
Compare
There was a problem hiding this comment.
Right fix, right shape. A Python thread can't be cancelled, so tying the reservation/idempotency lifecycle to the real concurrent.futures.Future — not to the cancellable asyncio wrapper — is the only correct model, and this holds the admission permit until the worker actually exits. fail-closed beats fail-open: a timed-out sync request keeps its slot rather than releasing a reservation while the thread still mutates state.
Things I checked
- Permit lifecycle (
dispatch.pysync branch,platform_router.py:_run_sync_delegate):await acquire()→executor.submitwith noawaitin between, released exactly once via either the submit-failure branch or the concurrent future'sadd_done_callback→loop.call_soon_threadsafe(release). No acquire-without-release, no double-release.call_soon_threadsafecorrectly marshals the non-thread-safeSemaphore.releaseback to the loop; loop-closedRuntimeErrorswallowed.code-reviewer: sound. - No spurious reservation release on the sync path. On sync cancellation the worker runs under
asyncio.shield,sync_lifecycle_continuesis set, the live future is handed to_settle_cancelled_sync_lifecycle, andon_failuredoes not fire. The outerexcept BaseExceptionconsumes the_adcp_sync_worker_futuremarker guarded bynot sync_lifecycle_continues, so the supervisor is created exactly once and never double-fires.security-reviewer: reservation staysCONSUMINGthrough cancellation, reachesCONSUMED/COMMITTEDonly after the thread returns —test_sync_create_media_buy_cancellation_waits_for_worker_successproves it. - Async cancellation still releases (
test_cancellation_fires_on_failure_and_propagates_unchanged): an async adopter's cancellation truly stops the coroutine, so firingon_failurewith theCancelledErrorthere is correct — the distinction from the sync path is the load-bearing part. _project_invocation_resultrefactor is behavior-preserving vs the old inline arms; the only delta is wrappingpre_handoff_reject()to fireon_failureon rejection — a strict improvement.- Contextvar bind/reset (
_bind_routed_sync_execution,time_budget.py): symmetricset/resetintry/finally, per-task context copies, and the objects bound are the process-wide executor + one admission semaphore — no per-tenant data to cross-contaminate. No leak across requests. - Bounding is real:
acquire()precedessubmit, so saturated timed calls exhaust their budget waiting and returnincomplete[]without ever entering the executor queue —test_sync_timeout_admission_saturates_without_executor_queue_growth+ the eager/lazy router parametrization confirm no queue growth. Supervisor tasks only await an existing future; they submit no new work and self-discard. get_adcp_capabilitieserror routing: now goes through_internal_error_message/_internal_error_details, droppingcaused_by.messagefrom the wire (class name only). Strictly less exposed — a credential-leak hardening, not a new leak.caused_by.typeis a documented debug breadcrumb, not a wire contract, so the shape change is safe underfix:.- Public surface:
timed_sync_get_products_limitadded toserve/create_adcp_server_from_platform/PlatformHandleris additive and optional — non-breaking.
Follow-ups (non-blocking — file as issues)
- Sync adopter that raises
asyncio.CancelledErroritself.code-revieweredge case: the supervisor'sexcept asyncio.CancelledError: raisecan't distinguish "worker produced CancelledError" from "supervisor was cancelled," stranding the reservation. Reachable only if a sync adopter raisesCancelledError, which is never legitimate — but aworker_future.cancelled()check or a documented constraint would close it. - Process-global admission = cross-tenant noisy neighbor. One
SyncExecutorAdmissionper handler, shared across all tenants of aPlatformRouter. A single tenant whose syncget_productsthreads block indefinitely can pin every permit and force other tenants toincomplete[]. Strictly better than the pre-PR unbounded behavior; keying admission byctx.account/tenant would contain the blast radius. - Only timed
get_productsis bounded.sync_admissionis passed forget_productswith a deadline only; synccreate_media_buyetc. still hit the executor's unbounded queue (pre-existing, bounded by connection count). Worth extending if upstream connection count isn't capped. - Post-timeout draft persistence is now observable: a timed-out
get_productswhose sync worker later completes runs_persist_draft_hook. Intended and documented in the module header — confirm downstream expects it.
Minor nits (non-blocking)
Semaphore.acquire()cancellation on early 3.10/3.11. The known FIFO-waiter permit-leak was fixed in later patch releases; CI runs 3.10–3.13 so patched runtimes are fine. A one-line note nearSyncExecutorAdmissionwould save a future reader the archaeology.
Both mandatory experts came back clean — code-reviewer: sound; security-reviewer: net security improvement, no High/Medium. Careful piece of concurrency work with tests covering success, real-failure, supervisor-cancellation isolation, saturation, eager+lazy routers, and the campaign bypass.
Approving on the strength of the permit-tied-to-real-future invariant plus the exactly-once supervisor guard. Ship it once CI validates the matrix.
fd017b0 to
94cf835
Compare
There was a problem hiding this comment.
Approving. This closes the real bug — a timed-out or cancelled request releasing a proposal/idempotency reservation while a non-cancellable worker thread is still mutating state — and it does it fail-closed: the permit and the reservation stay held until the underlying concurrent.futures.Future actually settles, never on the asyncio side moving on.
Things I checked
- Hooks fire exactly once across all four cancel shapes. Direct-sync cancel sets
sync_lifecycle_continues=Truein the innerexcept CancelledErrorand the re-raised error reaches the outerexcept BaseExceptionwith no_adcp_sync_worker_futuremarker, so it suppresses its ownon_failure— one supervisor, one settle. Router→sync-child cancel carries the marker from_run_sync_delegate, hits only the outer handler, same single settle. Pure-async cancel fireson_failure(CancelledError)once and releases the reservation — correct, because async work truly stopped (test_cancellation_fires_on_failure_and_propagates_unchanged). - No permit leak / double-release.
SyncExecutorAdmission.acquire()runs once per submission; release is exclusively via theconcurrent.futuresdone-callback →loop.call_soon_threadsafe(release).executor.submitraising releases in itsexcept BaseException. No await betweensubmitandadd_done_callback, and an already-done future invokes the callback synchronously — no lost-wakeup window. Cancel duringawait sync_admission.acquire()submits nothing and holds nothing (test_sync_timeout_admission_saturates_without_executor_queue_growth). - Supervisor strong-ref pattern is right.
_SUPERVISED_SYNC_LIFECYCLES(dispatch.py:88-90) +add_done_callback(...discard)is the standard keep-alive; cancelling the supervisor re-raises without cancelling the shielded worker (test_cancelling_sync_supervisor_does_not_cancel_worker_or_release), and a handoff returned after cancellation is demoted toon_failurerather than minting a task id the disconnected buyer never received (dispatch.py_settle_cancelled_sync_lifecycle). - Tenant isolation holds. Supervisor settles with the same
ctxas the originating request; reservation-release paths keepexpected_account_id=ctx.account.id/proposal_record.account_id(proposal_dispatch.py). The new_ROUTED_SYNC_ADMISSION/_ROUTED_SYNC_EXECUTORContextVars carry only concurrency primitives, never tenant data, and each request task gets its own context copy — no cross-tenant carry.security-reviewer: no-high-findings. - Error-detail redaction is a leak-removal, not a regression. handler.py now routes
get_adcp_capabilitiesINTERNAL_ERROR through_internal_error_message/_internal_error_details(dispatch.py:574-611), strippingstr(exc)from the wire and keeping only the class name — the same shape every other dispatch path already emits.error.jsonhasdetails: additionalProperties: trueandcaused_by.typeis unchanged, so no buyer deserialization breaks. Dropping the unspecifiedcaused_by.messagedebug field does not needfix!:. - Happy path unchanged. Async adopters get only a cheap ContextVar bind around the coroutine await.
Follow-ups (non-blocking — file as issues)
- Routed sync children changed pools. platform_router.py swaps
asyncio.to_thread(...)for_run_sync_delegate(...)at all six delegate sites, so bound routed sync children now run on the framework's configured executor instead of the loop default pool. Intended (admission needs a known pool), but adopters with a small BYO executor should be told routed sync children now compete for it. Worth a line in the release notes. - New public surface — doc drift.
timed_sync_get_products_limitonserve/create_adcp_server_from_platformand theSyncExecutorAdmissionexport are additive; confirm the executor/thread-pool section of the adopter docs mentions the new knob and its half-the-pool default. asyncio.Semaphorecancelled-acquire edge on 3.10/3.11. Pre-3.12 CPython can under-count a permit under a precise release/cancel race. Upstream, not introduced here, low probability — noting only because the CI matrix includes 3.10-3.11.
Minor nits (non-blocking)
- Documentation-only
except BaseException: raise. The arm added around dispatch.py:1989 in_project_handoff._runis a control-flow no-op — aCancelledErrorpropagates identically without it. Keep it for the comment if you like, but it reads like it's doing work it isn't.
Notable that the two experts traced every settle path independently and landed on the same four-shape enumeration the tests already assert. Ship it once CI is green.
There was a problem hiding this comment.
One open question on the async-cancel path holds this back from approve; the sync-supervision core and the error-message narrowing are both correct. The fix respects the right principle for threads — a Python thread can't be cancelled, so hold the permit and the reservation until it really exits — but applies the opposite principle to async cancellation in the same reservation seam.
Things I checked
- Error-message narrowing is a real leak reduction.
_internal_error_details(dispatch.py:592) now emitscaused_by = {"type": <class>}only;str(exc)is gone from every INTERNAL_ERROR site (dispatch.py:1553/1593/1986/2022, handler.py:1673/2141).security-reviewer: confirmed safe, no path re-introduces the message.ad-tech-protocol-expert: sound —schemas/cache/3.1.0-rc.13/core/error.jsontypesdetailsasadditionalProperties: truewith nocaused_byproperty, so droppingcaused_by.messageis not a wire-contract break; buyers branch onrecovery, which is untouched. Test asserts"tenant lookup failed"no longer appears on the wire. - Permit accounting is single-release. Submit-failure branch releases manually; success branch registers exactly one concurrent-future done-callback via
call_soon_threadsafe(release); cancel-during-acquire never reachessubmit, so no permit is held.SyncExecutorAdmissionfails fast on a non-positive limit. - Sync cancellation is correctly fail-closed. Direct-sync and async-router→sync-child (via the
_adcp_sync_worker_futuremarker) both spawn_settle_cancelled_sync_lifecycle, which awaits the real thread outcome underasyncio.shieldand only then fireson_complete/on_failure. Reservation stays CONSUMING while the thread mutates.expected_account_idtenant filters intact on all release paths (proposal_dispatch.py:753/854/887). Good coverage:test_sync_cancellation_settles_success_before_on_complete,_settles_real_failure_before_on_failure,_does_not_cancel_worker_or_release. - Admission is a DoS mitigation, not a new DoS. Gates only deadline-managed
get_products;campaign-unit and every other tool passsync_admission=None, so a saturatedget_productscan't lock outcreate_media_buy. - Semver signal.
timed_sync_get_products_limitonserve()/create_adcp_server_from_platform()is additive, keyword-only, defaults toNone.SyncExecutorAdmissionis a new additive export. No public signature break —fix:is the right prefix.
The open question (would flip me to approve)
dispatch.py:1600-1626 — the new except BaseException fires on_failure on a bare async CancelledError, releasing the create_media_buy reservation. For an async adopter the CancelledError carries no _adcp_sync_worker_future marker, sync_lifecycle_continues stays False, so line 1626 runs _safe_on_failure_call(on_failure, exc, ...) → _release_reservation_hook (handler.py:2187) → release_proposal_reservation → CONSUMING→COMMITTED.
Failure scenario: an async create_media_buy commits its seller-side write, then is cancelled at its next await (client disconnect — the handler task is not wrapped in wait_for, but ASGI cancels it on disconnect) and does not roll back on CancelledError. The framework releases the proposal to COMMITTED; the buyer retries; the proposal is re-consumed and a second media buy is booked — the exact inventory double-spend the two-phase CONSUMING reservation exists to prevent (handler.py:2148-2150).
This is a behavior change from main, which had no BaseException handler here — the CancelledError propagated uncaught, on_failure never fired, and the reservation stayed fail-closed for eviction/reconciliation. Both code-reviewer (Issue) and security-reviewer (Medium) landed on this line independently. It is gated behind an adopter that commits-before-await and does not roll back, and partly mitigated by idempotency-key replay, which is why it is not a hard block.
Notable that the same diff argues both sides of fail-closed within 300 lines: the background _run() path you added at dispatch.py:~1990 carries the comment "Cancellation does not prove adopter work stopped. Leave any reservation fail-closed ... rather than release it while side effects may still be outstanding" and re-raises without releasing — while the foreground async path here releases.
Two ways to close this, either flips me to approve:
- Fail-closed on bare async
CancelledErrorat the consumption seam — keep the reservation CONSUMING and let eviction/reconciliation settle it, matching_run(). Then the async-cancelon_failurefires only for real (non-cancellation) failures. - Justify in the PR body why async cancellation is guaranteed to give the adopter its rollback window and is therefore safe to release on, where the sync/background paths are not — and cover the committed-then-cancelled window with a test (
test_create_media_buy_cancellation_releases_reservationonly exercises cancel-before-any-work).
Minor nits (non-blocking)
- Stale comment. dispatch.py:1583-1586 still says the sync return exposes "exception class name + str" — the code now emits class name only. Same at dispatch.py:~656 ("
caused_by.messagealready carries the truncated repr"). Scrub both. BoundedSemaphorewould fail loud.SyncExecutorAdmissionusesasyncio.Semaphore; a future double-release would silently inflate the limit past its bound.BoundedSemaphoreraises instead. Single-release is correct today.- Misleading log on ordinary cancel.
_safe_on_failure_callnow catchesBaseException; if the hook is itself cancelled during cleanup it logs atexceptionlevel "on_failure hook raised" before re-raising — noise on a normal cancellation, not a real hook failure. - asyncio.Semaphore cancellation on CPython <3.12.
acquire()is routinely cancelled by the deadline while siblingsrelease(). Pre-3.12 stdlib has a known lost-wakeup race in this interleaving; both reviewers traced it as self-healing (worst case a spuriousincomplete[], no permanent lockout) since the project supports 3.10+. Worth a note, not a fix.
Answer the async-cancel question — fix it or justify it — and I approve.
There was a problem hiding this comment.
Correct fix for a real concurrency bug. A cancelled or timed-out request must not release a durable proposal/idempotency reservation while a non-cancellable worker thread is still mutating state — fail-closed beats fail-open, and the permit-tied-to-thread-completion design is the right shape.
Traced the acquire/release accounting on every branch with code-reviewer, security-reviewer, and python-expert. All three came back clean: no double-release, no leak, no cross-loop hazard on the happy path, and the wire-error change is a hardening, not a regression.
Things I checked
- Permit accounting is balanced on every path.
dispatch.py:95-113— acquire →executor.submitraises → synchronousrelease()→ re-raise, no done-callback registered, so no double-release. Success/cancel → released once viaadd_done_callbackmarshaled to the loop withloop.call_soon_threadsafe(sync_admission.release)(dispatch.py:106-113). CallingSemaphore.releasedirectly from the worker thread would have been the bug; it doesn't. There is noawaitbetweenacquire()and the synchronoussubmit, so no cancellation window strands a granted permit. - Cancellation supervision fires hooks exactly once.
_settle_cancelled_sync_lifecycle(dispatch.py:1717-1773) re-shields the same worker future and only settleson_complete/on_failureafter the thread actually exits. Double-await of anasyncio.Futurefrom two coroutines is legal and both awaiters get the same result. Thesync_lifecycle_continuesguard (dispatch.py:1600-1623) suppresses the directon_failureso a cancelled request never releases a reservation while the thread runs._SUPERVISED_SYNC_LIFECYCLESstrong-ref set + discard callback mirrors the existing_BACKGROUND_HANDOFF_TASKSGC-safety pattern. - Router path doesn't double-admit. For an async router → sync child,
_invoke_platform_methodtakes the coroutine branch and binds admission+executor into ContextVars (_bind_routed_sync_execution); the single acquire happens inplatform_router.py:_run_sync_delegate. Exactly one acquire per call. The_adcp_sync_worker_futuremarker smuggled onto theCancelledError(platform_router.py:499-503) is robust — it only needs to survive up the same coroutine stack with noTaskboundary, andwait_forreads it before the boundary. - Wire INTERNAL_ERROR sanitization is load-bearing.
_internal_error_details(dispatch.py:586-653) now emits{"type": <classname>}only; the exceptionstr()is gone from bothmessageandcaused_by.validation_errorsusesinclude_input=False, so buyer-supplied secret-bearing input isn't serialized.tests/test_decisioning_capabilities_projection.py:410-414asserts the secret-shaped string appears in neitherstr(exc)norstr(details). A 200-char truncation ofstr(exc)on an adopter who raised on an OAuth secret would have round-tripped into the idempotency replay cache — this closes that. - Tenant isolation intact on the release path.
release_proposal_reservationandmark_proposal_consumedstill scope onexpected_account_id=proposal_record.account_id; hydrate-time release usesctx.account.id. Every store mutation is account-scoped. - Semaphore construction is loop-safe.
asyncio.Semaphorebinds lazily on firstacquire(), not atPlatformHandler.__init__— fine on 3.10-3.13. - Public surface is additive.
timed_sync_get_products_limitis a keyword-only param withNonedefault onserve,create_adcp_server_from_platform, andPlatformHandler.__init__;SyncExecutorAdmissionis a new export. Non-breaking —fix(decisioning):is the right prefix.
Follow-ups (non-blocking — file as issues)
- Per-handler admission is process-global across tenants. In a
PlatformRouterdeployment oneSyncExecutorAdmissionfronts every tenant, so one slow (or hostile) tenant holding all permits for each thread's full post-timeout runtime forces every other tenant's timed syncget_productstoincomplete[]. Bounded and graceful — strictly better than the pre-PR whole-pool exhaustion — but if per-tenant availability matters, key admission onctx.account/tenant. (security-reviewer: Low-Medium.) - Async-adapter cancellation now releases the reservation (
CONSUMING → COMMITTED), where before aCancelledErrorleft it held for eviction. An async adapter cancelled after its external buy-create but before finalize could be retried into a duplicate — narrow, and mitigated bycreate_media_buyidempotency-key dedup. Worth confirming idempotency is mandatory on that path. (security-reviewer: Low, and it's asserted bytest_create_media_buy_cancellation_releases_reservation, so it's deliberate.) get_productspersists a draft for a response the buyer was told was incomplete. Withon_complete=_persist_draft_hook, a timed-outget_productsstill runs the draft-persist hook on the late worker result. Consistent with the "supervise timed work" intent, but confirm that's the intended product behavior. (python-expert.)- Consider
BoundedSemaphorefor the admission limiter. Accounting is balanced today, but a plainasyncio.Semaphoresilently inflates capacity abovelimitif a future refactor introduces a stray double-release, rather than failing loudly. Cheap self-verification for a safety-critical bound. (python-expert.)
Minor nits (non-blocking)
- Stale comment references a field that no longer exists.
dispatch.py:650-651still says "The caused_by.message already carries the truncated repr" — but the PR removedcaused_by.message. Correct it so a future editor doesn't "restore" the message field and reopen the leak. _max_workersfallback silently caps BYO wrappers at 1.handler.py:1330reads the privateexecutor._max_workers; an executor-compatible wrapper without it gets an admission limit of 1. Guarded and documented, so acceptable — but a wrapper author will wonder why their pool serializes.- Hookless sync cancel can log "Future exception was never retrieved." When a directly-sync method with both hooks
Noneis cancelled and the late worker raises, the orphanedworker_async_futureproduces asyncio log noise (not a leak — the permit still releases). Not reachable forget_products(always setson_complete); only hookless sync reads.
Approving on the strength of the balanced permit accounting plus the wire-error hardening. Follow-ups noted above.
94cf835 to
08fd877
Compare
There was a problem hiding this comment.
Approving. Ties framework permit and lifecycle ownership to the real thread's completion instead of the cancellable asyncio wrapper — the only correct lifetime, since a Python thread can't be cancelled.
Things I checked
- Permit accounting balances.
submit_supervised(time_budget.py:903-935) acquires beforeexecutor.submit, releases on the submit-failure path, and otherwise releases exactly once from theconcurrent.futures.Futuredone callback vialoop.call_soon_threadsafe(admission.release). Noawaitsits betweenacquire()andsubmit, so a cancelled acquire strips no permit;BoundedSemaphore's over-releaseValueErroris unreachable. - Hooks fire exactly once under cancellation. Direct sync path: inner
except CancelledError(dispatch.py:152-170) setssync_lifecycle_continues=True, supervises the worker, re-raises into the outerexcept BaseException(dispatch.py:189-215) whererouted_sync_execution is Nonesuppresses a second supervise and the flag suppresses the outeron_failure. Async-routed path supervisesrouted_sync_execution.workeror fireson_failureonce with theCancelledError.code-reviewer: no double-fire, no missed reservation release. - DoS bound is real. Saturated timed
get_productsblocks onawait admission.acquire()inside the deadline'swait_forand is cancelled beforeexecutor.submit— returnsincomplete[]without entering the executor queue. Defaultmax(1, size//2)reserves capacity for other tools; onlyget_productspasses a non-Nonesync_admission, so no cross-tool starvation.security-reviewer: no permit-leak regression. - Error sanitization stops the leak.
_exception_cause_details(dispatch.py:46-48) is type-only; the previously-leakyhandler.py get_adcp_capabilitieswrap that builtcaused_by.message = str(exc)now routes through the sanitized helpers.security-reviewerswept the decisioning tree — no remainingstr(exc)/repr(exc)on a wiremessage/details. - Wire change is non-breaking.
ad-tech-protocol-expert:grep -c caused_by schemas/cache/3.1/core/error.json→ 0.caused_byis a framework breadcrumb inside the openadditionalProperties: truedetailsobject, never a normative field.recovery=\"terminal\"preserved and enum-valid.fix(decisioning):is the correct semver signal — no!needed. - Tenant isolation intact on the cancellation path.
_settle_cancelled_finalizecommits withexpected_account_id=account_id(proposal_dispatch.py:657-663), the samectx.account.idthe non-cancelled paths use.
Follow-ups (non-blocking — file as issues)
- Orphaned task in the
execution is Nonerouter branch._run_sync_delegate(platform_router.py) doesworker = asyncio.create_task(asyncio.to_thread(...))thenawait asyncio.shield(worker); on caller cancellation the shield re-raises butworkerruns unsupervised, and a later raise surfaces asTask exception was never retrieved. The pre-PR inlineawait asyncio.to_thread(...)didn't create this dangling Task. Low reachability — the bound-ContextVar (routed) branch is supervised by dispatch — but worth a background settler or dropping the redundantcreate_taskwrapper. - CHANGELOG bullet hand-inserted into a release-please-owned section. The new
[7.0.0-rc]bullet has no commit/PR link, unlike its siblings; release-please regenerates that section from thefix(decisioning):message and will overwrite or conflict on the next release PR. Let the commit carry the entry.
Minor nits (non-blocking)
- Stale comment.
dispatch.py:650-651still reads "The caused_by.message already carries the truncated repr" — there is no longer amessage. Misleading now. - No-op guard.
platform_router.pyexcept asyncio.CancelledError: raiseadds nothing over letting cancellation propagate. - Bundled wire change. The
caused_by.messageremoval is a security-positive change but orthogonal to "supervise timed/canceled work" — it inflates the diff of a concurrency fix. Justified in the body and covered bytest_decisioning_capabilities_projection.py; consider splitting next time.
Test coverage is strong — cancellation-before/after-worker for success and real failure, handoff discard, supervisor-cancel-does-not-cancel-worker, BYO-executor-requires-limit, and the sync finalize/create_media_buy reservation e2e all exercise behavior, not implementation.
Safe to merge.
9921d72 to
08a8490
Compare
There was a problem hiding this comment.
The supervision machinery is sound — three independent reviews confirm it. The block is narrow: a breaking change to a public construction contract shipped under fix: without the semver signal release-please needs.
Behavior change: on prior versions create_adcp_server_from_platform(platform, executor=my_pool) (and serve(platform, executor=my_pool)) constructs a working server. After this PR it raises ValueError at src/adcp/decisioning/serve.py (the if timed_sync_get_products_limit is None: guard on the executor= branch) unless the adopter also passes timed_sync_get_products_limit=. Both functions are public exports reachable from adcp.decisioning. The existing tests had to add timed_sync_get_products_limit=1 next to every executor= call — that diff is the proof the prior call shape is now rejected.
This is a breaking public-API change. It ships as fix(decisioning): supervise timed and canceled work — no !, no BREAKING CHANGE: footer. release-please derives semver from the commit, so this cuts a non-breaking bump over a diff that breaks BYO-executor adopters. ad-tech-protocol-expert independently flagged the same construction-contract break and asked to confirm it lands as minor+, not patch.
To unblock: retitle to fix(decisioning)!: or add a BREAKING CHANGE: footer naming the migration (BYO executor= adopters must now pass timed_sync_get_products_limit=). The PR body's Compatibility section already documents it — the commit metadata just needs to match. No code change required; the fail-closed ValueError with a clear message is the right shape for the break itself.
Things I checked
- Permit accounting is exactly-once.
submit_supervisedintime_budget.py: cancel-during-acquire()holds no permit (re-raises without decrement);executor.submitraising releases once and registers no callback; worker completion releases once viacall_soon_threadsafe; loop-closed-at-teardown swallowsRuntimeError.BoundedSemaphoreturns any accidental over-release into aValueErrorinstead of silent capacity inflation.code-reviewer: sound. - Cancellation hooks fire exactly once, and no reservation releases while a thread still mutates state. Direct sync cancel defers
on_complete/on_failureto_settle_cancelled_sync_lifecycleafter the thread actually exits (test_sync_create_media_buy_cancellation_waits_for_worker_successconfirms CONSUMING→CONSUMED only post-release.set()); async cancel fireson_failure(CancelledError)once and releases. The innerexcept CancelledErrorsetssync_lifecycle_continuesso the outerexcept BaseExceptiondoes not double-fire. - INTERNAL_ERROR sanitization has no residual leak.
_internal_error_message/_internal_error_details/_exception_cause_detailsemit onlytype(exc).__name__; the create_media_buy transient path and both background-task wraps carry nostr(exc).security-reviewer: net improvement, no High — this also keeps secret-bearing exception strings out of the idempotency replay cache. - Wire shape holds.
caused_byis a framework extension, not normative —schemas/cache/3.1/core/error.json:93-96isadditionalProperties: trueand never definescaused_by. Removingdetails.caused_by.messageis a reduction of a non-normative breadcrumb; requiredcode/messageand normativerecoveryare unchanged.ad-tech-protocol-expert: sound. - No cross-tenant settle.
_settle_cancelled_finalizecommits withexpected_account_id=ctx.account.idfrom the same per-request closure;proposal_store.commitfail-closes on account mismatch. - Admission resolution. Framework pools default to
max(1, size // 2);size=4→2 matches the test. Non-positive explicit limits are rejected bySyncExecutorAdmission.__init__.
Follow-ups (non-blocking — file as issues)
- Cross-tenant admission fairness (
security-reviewerLow). The admission semaphore lives on the singlePlatformHandler, so it's global across tenants — one tenant's slowget_productscan pin all permits and push other tenants toincomplete[]. Degraded service, not exposure. A per-tenant sub-bucket if fairness matters for the multi-tenant deployment. - CHANGELOG hand-edit. The added Bug Fixes bullet has no commit link, unlike its siblings. release-please regenerates this from the conventional commit post-merge — let it own the entry rather than duplicating.
Minor nits (non-blocking)
execution.workeroverwrite._run_sync_delegatepublishes only the last sync child if a routed delegate submits more than one. Not reachable for single-tenantget_productsrouting — worth a one-line comment noting the single-child assumption.- Lazy semaphore loop-binding.
SyncExecutorAdmissionis built inPlatformHandler.__init__, possibly outside a loop;asyncio.BoundedSemaphorebinds lazily on firstacquire(). Fine as long as construction and serving share one loop — a one-line comment would save a future reader the trace.
Fix the semver signal and this ships. The implementation is right.
08a8490 to
825f2bf
Compare
There was a problem hiding this comment.
The supervision core is right — sync work now holds its admission permit and settles its proposal/idempotency lifecycle past a cancellation instead of leaking a pool slot. But one arm moves the async create_media_buy cancellation path in the opposite direction from everywhere else in this diff, and I want that asymmetry confirmed before I approve.
The concern (async cancel now releases the reservation)
The new except BaseException arm in _invoke_platform_method fires on_failure on CancelledError when there's no routed sync worker. For create_media_buy, on_failure is _release_reservation_hook (handler.py:2173-2181) → release_proposal_reservation, which flips the reservation CONSUMING → COMMITTED.
On main, _invoke_platform_method had no BaseException handler — CancelledError propagated without firing on_failure, so an async create_media_buy cancellation left the reservation CONSUMING (fail-closed, reconciled/expired later). This PR flips that arm to fail-open for async.
That contradicts the fail-closed stance this same PR takes two other places:
_project_handoff._runnew arm:except BaseException: raise— "Cancellation does not prove adopter work stopped ... rather than release it while side effects may still be outstanding."maybe_hydrate_recipes_for_create_media_buy— deliberately still narrowed toexcept ExceptionsoCancelledErrorskips the release.
Failure scenario: an async create_media_buy adapter sends the buy to the ad server, the request is cancelled (client disconnect) with the side effect already committed server-side, the arm releases the reservation, and the buyer's retry re-consumes the proposal and creates a second buy — the double-spend the two-phase commit exists to prevent. Idempotency doesn't cover it: the cancelled request cached no response, so a retry executes fresh against a now-COMMITTED proposal.
test_create_media_buy_cancellation_releases_reservation asserts the release is intended — but its adapter does no side effect (await asyncio.Event().wait()), so it only proves the safe case, not the side-effect-then-cancel edge.
What flips me to approve: either confirm the async/sync asymmetry is deliberate and say why fail-open is acceptable for async when the sibling paths chose fail-closed (e.g. async cancel is treated as a clean interrupt with no committed side effect), or fail-closed the reservation-bearing async path the way _run does. A test covering async create_media_buy cancellation after a completed side effect would settle it either way. This is the only thing holding approval.
security-reviewer: no High — the cancellation/admission path fails closed for DoS and holds no cross-tenant isolation break; the expected_account_id guard on the deferred commits is correct. ad-tech-protocol-expert: sound-with-caveats — dropping details.caused_by.message is wire-compatible (error.json leaves details open, caused_by is a non-normative breadcrumb, recovery unchanged), and incomplete[] on a saturated-never-submitted call is the strongest instance of scope: "products", not a stretch.
Things I checked
SyncExecutorAdmissionpermit accounting: acquired once beforeexecutor.submit, released once in the submit-failureexcept, otherwise released once by theconcurrent.futuresdone-callback vialoop.call_soon_threadsafe. Noawaitbetween acquire and callback registration, so no cancellation window leaks a permit. Saturation cancels a pendingacquire()with no decrement.BoundedSemaphoreguards over-release.- INTERNAL_ERROR sanitization: the real fix is
handler.get_adcp_capabilities(handler.py:1654-1667on base), the last path emittingstr(exc)on the wire in bothmessageanddetails.caused_by.message. Now routed through_internal_error_message/_internal_error_details, which emit class name only.logger.exceptionstill records the full trace server-side. Strictly reduced leak surface. _project_invocation_resultextraction: handoff-reject ordering, on_complete-then-strip, and theparams→request_paramsrename all match the base inline body.- Multi-tenant isolation:
_ROUTED_SYNC_EXECUTIONis a per-requestContextVarunder a copied context; deferred commits capture the tenant-scopedctxand passexpected_account_id. - Wire compat of the error-detail change against
schemas/cache/3.1/core/error.json(detailsisadditionalProperties: true, rootrequiredis["code","message"]).
Follow-ups (non-blocking — file as issues)
- BYO-executor break needs a
BREAKING CHANGE:footer.create_adcp_server_from_platform/serve(public,adcp.decisioning.__all__) now raiseValueErrorwhenexecutor=is passed withouttimed_sync_get_products_limit=. It lands inside the 7.0.0 major already cutting and the PR body documents the migration, so the semver boundary is covered — but the commit isfix(decisioning):with an empty body. Add aBREAKING CHANGE:footer so it surfaces in the 7.0.0 changelog rather than under Bug Fixes. - Handler-lifetime
asyncio.BoundedSemaphoreis single-loop-bound. Firstacquire()binds it to that loop; reusing aPlatformHandleracross loops (repeatedasyncio.run, thread-per-loop ASGI) raises "bound to a different event loop." Single-loop serving is unaffected — worth a one-line note in thetimed_sync_get_products_limitdocs that the handler is loop-scoped once used. - Shared per-server admission couples tenant availability. One admission bucket across all tenants, each permit held for the full uninterruptible thread lifetime.
limitpermanent hangs degrade every tenant's deadline-managedget_productstoincomplete[]for the process life. Better-contained than base (which leaked to all tools), but no watchdog — consider a per-tenant bucket or a saturation alarm, and document that adopters must bound syncget_productsI/O.
Minor nits (non-blocking)
- CHANGELOG hand-edit under a tagged section.
CHANGELOG.md:17inserts the bullet into the released[7.0.0-rc.1]Bug Fixes list. That section is release-please-managed; the entry should come from the conventional-commit message and can be clobbered/reordered on the next run.
Not approving yet — answer the async/sync asymmetry question and I'll flip. Everything else is clean.
Summary
Why
Timed-out or cancelled requests could release reservations while worker threads were still mutating state. Under load this allowed duplicate work and unbounded executor queues.
Validation
origin/mainCompatibility
executor=wiring now requires an explicittimed_sync_get_products_limit=because executor wrappers expose no public worker-count contract; framework-allocated pools retain an automatic half-capacity default.details.caused_by.message; they retain only the exception type and the normativerecoveryvalue. This is wire-compatible with AdCP 3.1.8 becauseschemas/cache/3.1/core/error.jsonleavesdetailsopen (additionalProperties: true) and does not definecaused_by.