From ee749edc928e5869284b108bf6d31dac082985be Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 13 Sep 2026 11:08:01 -0400 Subject: [PATCH] runtime: separate call completion from store driver outcomes --- docs/architecture.md | 10 + runtime/src/exec/boundary.ts | 434 +++++++++++------- runtime/src/exec/host_streams.ts | 24 +- runtime/src/intrinsics/fact_calls.ts | 4 + runtime/src/jspi/bridge.ts | 11 + runtime/src/task/mod.ts | 32 ++ runtime/src/task/scheduler.ts | 103 ++++- runtime/src/task/thread.ts | 19 + runtime/tests/call_owned_completion_test.ts | 100 ++++ .../tests/driver_trap_exit_liveness_test.ts | 40 ++ runtime/tests/embedder/long_poll_test.ts | 10 +- .../fixtures/two-instance-scheduler.wasm | Bin 0 -> 2704 bytes .../tests/fixtures/two-instance-scheduler.wat | 143 ++++++ .../tests/host_boundary_preparation_test.ts | 43 +- runtime/tests/lift_done_verdict_test.ts | 63 +++ runtime/tests/poison_cause_test.ts | 33 ++ 16 files changed, 883 insertions(+), 186 deletions(-) create mode 100644 runtime/tests/call_owned_completion_test.ts create mode 100644 runtime/tests/fixtures/two-instance-scheduler.wasm create mode 100644 runtime/tests/fixtures/two-instance-scheduler.wat diff --git a/docs/architecture.md b/docs/architecture.md index 013b42b..3db54d5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -255,6 +255,16 @@ order, not the time readiness became true; pending events use join order. Tests can use seeded shuffling through `POLYENGINE_SCHED_SEED` to exercise spec-permitted scheduling variation. See `runtime/src/task/scheduler.ts`. +**Call completion.** Canonical resolution is captured at `Task.return_`, in +the reference-defined order. Public delivery becomes eligible when that call's +current activation returns or is confirmed parked at a real scheduler +`SuspensionPoint`; a promising-entry implementation hop is not such a boundary. +The call then publishes independently of unrelated store work. A later sibling +fault cannot revoke a published value, while an unpublished originating call is +failed through its task/instance ownership channel. Cleanup and terminal +notification are both attempted, and the first poison cause is retained even +when cleanup fails. + **Overlapping drivers.** Concurrent exports may run overlapping `driveAsync` loops on one store. The invariant is that an activation consumes a settlement at most once and never resumes from an obsolete diff --git a/runtime/src/exec/boundary.ts b/runtime/src/exec/boundary.ts index c7aa719..bf1b9f3 100644 --- a/runtime/src/exec/boundary.ts +++ b/runtime/src/exec/boundary.ts @@ -31,10 +31,10 @@ import { } from "../cabi/values.ts"; import { assert_, AssertionError } from "../cabi/trap.ts"; import { - addInstancePoisonedListener, type BlockRequest, type Cancelled, type ComponentInstanceState, + consumeSchedulerFailure, driveSyncLift, entryRefusal, EventCode, @@ -45,6 +45,7 @@ import { NeedsJspi, needsJspi, notifyInstancePoisoned, + OriginatedSchedulerFailure, packSubtaskResult, PendingCapability, realHostCalls, @@ -396,6 +397,11 @@ type IdlePolicy = "trap" | "exit"; */ type DriveExit = "done" | "idle"; +/** A pre-existing store-global report, distinct from a guest scheduler trap. */ +class HostFailureReport { + constructor(readonly cause: unknown) {} +} + /** * Pump `store` until `done()` holds. Returns the exit verdict directly if * that was settled synchronously, or a Promise of it otherwise. @@ -431,9 +437,13 @@ function driveLoop( // starve them. Hand those stores to the interleaved async drain. while (store.awaiting.size === 0 && store.tick()) { traceDrive("drive", store, done, "ticked"); - if (store.hostFailure !== undefined) throw takeHostFailure(store); + if (store.hostFailure !== undefined) { + throw takeHostFailure(store); + } + } + if (store.hostFailure !== undefined) { + throw takeHostFailure(store); } - if (store.hostFailure !== undefined) throw takeHostFailure(store); if (done()) { traceDrive("drive", store, done, "EXIT-done"); // No async finally will run: hand off any background work here. @@ -510,7 +520,16 @@ export async function driveStoreAsync( ): Promise { // The exit verdict is for `drive`'s lift caller (see `DriveExit`); the // pumps drive to quiescence and have nothing to decide on it. - await driveAsync(store, done, what); + for (;;) { + try { + await driveAsync(store, done, what); + return; + } catch (e) { + if (consumeSchedulerFailure(store, e)) continue; + if (e instanceof HostFailureReport) throw e.cause; + throw e; + } + } } /** @@ -764,7 +783,9 @@ async function driveAsync( traceDrive("driveAsync", store, done, "top"); // Complete settled activation bookkeeping before any scheduling decision. store.serviceSettled(); - if (store.hostFailure !== undefined) throw takeHostFailure(store); + if (store.hostFailure !== undefined) { + throw takeHostFailure(store); + } // Yield for this store's engine resumptions. Only their execution/park // or settlement may release them; never clear other owners' entries. if (store.hasPendingResumptions()) { @@ -786,7 +807,9 @@ async function driveAsync( } claimHops = 0; while (store.tick()) { - if (store.hostFailure !== undefined) throw takeHostFailure(store); + if (store.hostFailure !== undefined) { + throw takeHostFailure(store); + } // A READY/YIELD loop must not starve promise settlements. Give engine // continuations a microtask per tick and service any landed tails first. if (store.awaiting.size > 0) { @@ -794,7 +817,9 @@ async function driveAsync( if (store.hasServiceableSettled()) break; } } - if (store.hostFailure !== undefined) throw takeHostFailure(store); + if (store.hostFailure !== undefined) { + throw takeHostFailure(store); + } if (done()) { traceDrive("driveAsync", store, done, "EXIT-done"); return "done"; @@ -933,7 +958,15 @@ async function driveAsync( for (let i = store.settled.length - 1; i >= 0; i--) { if (store.settled[i].t === winner.t) store.settled.splice(i, 1); } - winner.t.resumeWith(winner.value, winner.failure); + const task = (winner.t as { + task?: { failureOwner?: unknown }; + }).task; + const origin = task?.failureOwner ?? task; + try { + winner.t.resumeWith(winner.value, winner.failure); + } catch (e) { + throw new OriginatedSchedulerFailure(origin, e); + } } continue; } @@ -975,7 +1008,7 @@ async function driveAsync( function takeHostFailure(store: Store): unknown { const e = store.hostFailure; store.hostFailure = undefined; - return e; + return new HostFailureReport(e); } // --------------------------------------------------------------------------- @@ -992,39 +1025,6 @@ function takeHostFailure(store: Store): unknown { */ export const SYNC_ENTRY: unique symbol = Symbol("polyengine.syncEntry"); -// --------------------------------------------------------------------------- -// Pending async-typed lift results -// --------------------------------------------------------------------------- -// -// An idle exit transfers result settlement to the task's onResolve callback. -// If a later driver poisons the instance first and async-end retirement returns, -// this listener rejects pending results. A throwing retirement hook prevents -// this notification; recording the poison cause alone does not settle them. -const pendingLifts = new WeakMap void>>(); - -function registerPendingLift(inst: object, reject: (c: unknown) => void): void { - let s = pendingLifts.get(inst); - if (s === undefined) pendingLifts.set(inst, s = new Set()); - s.add(reject); -} - -function unregisterPendingLift( - inst: object, - reject: (c: unknown) => void, -): void { - pendingLifts.get(inst)?.delete(reject); -} - -addInstancePoisonedListener((inst, cause) => { - const s = pendingLifts.get(inst as object); - if (s === undefined || s.size === 0) return; - // Drained before dispatch: a rejection handler running synchronously must - // not see, or re-enter, this set. - const waiters = [...s]; - s.clear(); - for (const r of waiters) r(cause); -}); - /** Build a `Store.lift` / `canon_lift` entry with a Task and implicit Thread. * Canonical options select sync result lifting, callback dispatch, or stackful * execution; the function type separately selects the sync/async idle policy. */ @@ -1161,14 +1161,83 @@ export function createLiftedFunction(input: { prepared?.cleanup(e); throw e; } - let completed = false; - let resolved: ComponentValue[] | null = null; let resolvedSeen = false; - /** - * Idle-path result waiter. onResolve, not thread drain, supplies the answer. - */ - let onResolvedHook: (() => void) | null = null; + let eligible = false; + let terminal = false; + let terminalFailed = false; + let terminalValue: unknown; + let terminalCause: unknown; + let invocationReturned = false; + let futurePublicationQueued = false; + let waiter: { + promise: Promise; + resolve: (value: unknown) => void; + reject: (cause: unknown) => void; + } | null = null; + + const terminalPromise = (): Promise => { + if (waiter === null) { + let resolve!: (value: unknown) => void; + let reject!: (cause: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + // The origin can fail while a sibling driver is still inside invokeNow, + // before the facade receives this Promise. Observe internally from + // creation; the returned promise still rejects for its caller. + void promise.catch(() => {}); + waiter = { promise, resolve, reject }; + if (terminal) { + if (terminalFailed) reject(terminalCause); + else resolve(terminalValue); + } + } + return waiter.promise; + }; + + const publishIfEligible = (): void => { + if (terminal || !eligible || !resolvedSeen) return; + if (!invocationReturned && ft.async && ft.results[0]?.kind === "future") { + return; + } + if ( + ft.async && ft.results[0]?.kind === "future" && + !futurePublicationQueued + ) { + futurePublicationQueued = true; + Promise.resolve().then(publishIfEligible); + return; + } + // Existing host producer/conversion failures are store-global reports, + // not task-origin traps. They must remain loud if recorded before this + // call publishes, but do not poison the instance (#118). + if (store.hostFailure !== undefined) { + const report = takeHostFailure(store); + failCall( + report instanceof HostFailureReport ? report.cause : report, + ); + return; + } + if (resolved === null) { + terminalFailed = true; + terminalCause = new AssertionError( + `${name}: task resolved as cancelled, but the host never requested cancellation`, + ); + terminal = true; + task.detachCall(); + waiter?.reject(terminalCause); + return; + } + terminalValue = resultsToHost(resolved); + terminal = true; + task.detachCall(); + // This runs only at a control-return boundary, outside canonical state + // mutation, so Promise thenable inspection is safe here. + waiter?.resolve(terminalValue); + }; + const task = new Task( ft, taskOpts, @@ -1178,11 +1247,6 @@ export function createLiftedFunction(input: { resolved = result; resolvedSeen = true; stats.tasksResolved++; - if (onResolvedHook !== null) { - const f = onResolvedHook; - onResolvedHook = null; - f(); - } }, ); @@ -1203,36 +1267,36 @@ export function createLiftedFunction(input: { ); const finishHostEntry = (): unknown => { - completed = true; trapIf( - !resolvedSeen, + !terminal, `${name}: task finished without resolving (deadlock)`, ); - if (resolved === null) { - // definitions.py `Task.cancel`: `on_resolve(None)`. A host-initiated - // call has no way to express "cancelled" in its return value, and the - // host never requests cancellation, so reaching this is a bug. - throw new AssertionError( - `${name}: task resolved as cancelled, but the host never ` + - `requested cancellation`, - ); - } - return resultsToHost(resolved); + if (terminalFailed) throw terminalCause; + return terminalValue; }; const unwind = (...primary: [] | [unknown]): void => { // Failed adapters may skip exit-sync-call; release this task's lenders // so unaffected instances do not retain abandoned borrows. - if (completed) return; + let cleanupFailure: unknown; + let cleanupFailed = false; try { if (primary.length === 0) prepared?.cleanup(); else prepared?.cleanup(primary[0]); - } catch { - // An active boundary failure remains primary. + } catch (e) { + cleanupFailed = true; + cleanupFailure = e; } for (const t of task.threads as { syncCallStack: unknown[] }[]) { while (t.syncCallStack.length > 0) { - (t.syncCallStack.pop() as LenderScope).releaseLenders(); + try { + (t.syncCallStack.pop() as LenderScope).releaseLenders(); + } catch (e) { + if (!cleanupFailed) { + cleanupFailed = true; + cleanupFailure = e; + } + } } } void syncCallStack; @@ -1245,6 +1309,7 @@ export function createLiftedFunction(input: { i.mayLeave = true; } } + if (primary.length === 0 && cleanupFailed) throw cleanupFailure; }; /** @@ -1266,6 +1331,36 @@ export function createLiftedFunction(input: { const isCapabilitySignal = (e: unknown): boolean => e instanceof NeedsJspi || e instanceof PendingCapability; + const failCall = (e: unknown): boolean => { + if (terminal) return false; + terminal = true; + task.detachCall(); + terminalFailed = true; + terminalCause = e; + if ( + task.state === "initial" && thread.waiting() && thread.cancellable + ) { + thread.abandonWaiting(); + } + try { + unwind(e); + } finally { + waiter?.reject(e); + } + return true; + }; + task.onFailure = (e): boolean => { + if (terminal) return false; + if (!isCapabilitySignal(e)) poison(e); + return failCall(e); + }; + task.onControlReturn = (owner) => { + if (owner.task !== task || terminal) return; + eligible = true; + publishIfEligible(); + }; + task.attachCall(); + try { thread.resume(); // The reference sync loop drives callee-instance threads. JSPI and @@ -1278,116 +1373,100 @@ export function createLiftedFunction(input: { // own JSPI hop must not turn this completed sync call into a Promise. if (guestDtor) return finishHostEntry(); } catch (e) { - unwind(e); - if (!isCapabilitySignal(e)) poison(e); + if (consumeSchedulerFailure(store, e)) return terminalPromise(); + if (e instanceof HostFailureReport) { + failCall(e.cause); + if (ft.async) return terminalPromise(); + throw e.cause; + } + task.onFailure(e); + if (ft.async && task.state === "initial") return terminalPromise(); throw e; } + const crossedPromiseHop = thread.awaiting !== null; + invocationReturned = true; + // A plain core invocation has actually returned to JS here. Only a JSPI + // promising-entry Promise is an implementation hop whose tail may still + // contain result lifting/post-return work. + if (resolvedSeen && mode !== "jspi") eligible = true; + publishIfEligible(); - /** - * After an idle exit, settle from onResolve, not the last thread's exit - * (#315 result-settlement rule; definitions.py `Task.return_`). Background - * producers may retain threads indefinitely. Already-captured results - * settle immediately; otherwise register resolution and poison waiters. - * - * This callback only shapes lifted values. It does not stop whichever - * driver now owns the task, nor unwind another driver's active FACT scopes. - */ - const backgroundCompletion = (): Promise => - new Promise((resolve, reject) => { - // An idle verdict can follow resolution while a driver-liveness - // clause remains false. Do not wait for an event that already fired. - if (resolvedSeen) { - try { - resolve(finishHostEntry()); - } catch (e) { - unwind(e); - reject(e); - } - return; + let outcome: DriveExit | Promise; + for (;;) { + try { + // Call-owned terminal state is published by explicit activation return + // or genuine park. No unrelated store quiescence is a memory barrier. + const driveDone = () => terminal; + outcome = drive( + store, + driveDone, + `export '${name}'`, + idlePolicy, + ); + break; + } catch (e) { + if (consumeSchedulerFailure(store, e)) { + // A synchronous driver can encounter a foreign ready task. Routing + // that task's failure is one scheduler step, not a reason to change + // this call's return shape or abandon its remaining work. + continue; } - const onPoison = (cause: unknown): void => { - onResolvedHook = null; - // A task blocked before admission has no guest activation to retire. - // Resume its cancellable gate so enterImplicitThread's finally - // removes the waiting slot, without synthesizing task cancellation. - if ( - task.state === "initial" && thread.waiting() && thread.cancellable - ) { - thread.abandonWaiting(); - } - unwind(cause); - reject(cause); - }; - // Poison notification may have run after the driver returned idle but - // before this continuation installed its listener. Preserve an - // already-captured task result above; otherwise poisoning first must - // reject the pending async export with the recorded cause. Defer this - // already-recorded case one microtask so the caller receives and can - // observe the returned Promise before it rejects. - // CONTRACT: contracts/embedder-api.md:180-185; per-instance poisoning - // is the runtime policy in docs/architecture.md:300-307. - if (isInstancePoisoned(inst)) { - const cause = instancePoisonCause(inst); - Promise.resolve().then(() => onPoison(cause)); - return; + if (e instanceof HostFailureReport) { + failCall(e.cause); + if (ft.async) return terminalPromise(); + throw e.cause; } - registerPendingLift(inst, onPoison); - onResolvedHook = () => { - unregisterPendingLift(inst, onPoison); - try { - // Safe synchronously inside the resolve callback: pure over - // already-lifted `ComponentValue`s (`canon_task_return` lifted - // them into `resolved`; `resultsToHost` reshapes). Host `.then` - // handlers run on a microtask regardless. - resolve(finishHostEntry()); - } catch (e) { - unwind(e); - reject(e); - } - }; - }); - - let outcome: DriveExit | Promise; - try { - const midWasmCall = () => task.threads.some((t) => store.awaiting.has(t)); - const hopParked = () => entryHopThreads(store).length > 0; - // Driver completion is not thread exhaustion. CONTRACT: an async result - // is independent of producer lifetime - // (embedder-api.md:180-185; definitions.py:521-526,2360-2369). A - // genuine SuspensionPoint may therefore be handed to the settlement - // pump once the result exists. Entry hops remain part of result-memory - // ordering and must complete first. Sync lifts retain full activation - // liveness. Callback tasks may retain waiting producer threads after - // task.return; awaiting their final exit would prevent result delivery. - const driveDone = () => - resolvedSeen && !hopParked() && (ft.async || !midWasmCall()); - outcome = drive( - store, - driveDone, - `export '${name}'`, - idlePolicy, - ); - } catch (e) { - unwind(e); - throw e; + // A store-global host report is checked before the driver's done + // predicate. If this call already published, the report belongs to the + // subsequent entry/diagnostic channel, not to this healthy result. + if (terminal) throw e; + task.onFailure(e); + if (ft.async) return terminalPromise(); + throw e; + } } /** * Use the captured exit verdict, not a fresh shared-store predicate. */ const finish = (verdict: DriveExit): unknown => - verdict === "idle" ? backgroundCompletion() : finishHostEntry(); + verdict === "idle" ? terminalPromise() : finishHostEntry(); if (!isPromiseLike(outcome)) { try { - return finish(outcome); + const value = finish(outcome); + return crossedPromiseHop ? terminalPromise() : value; } catch (e) { - unwind(e); + if (consumeSchedulerFailure(store, e)) return terminalPromise(); + if (e instanceof HostFailureReport) { + failCall(e.cause); + if (ft.async) return terminalPromise(); + throw e.cause; + } + task.onFailure(e); + if (ft.async && task.state === "initial") return terminalPromise(); + if (ft.async && terminal) return terminalPromise(); throw e; } } - return outcome.then(finish, (e) => { - unwind(e); - throw e; - }); + // The driver may remain parked on unrelated host work after this call's + // result becomes eligible. Observe it for this call's own deadlock/fault, + // but return the call-owned completion channel instead (#350/#357). + void outcome.then( + () => publishIfEligible(), + (e) => { + if (consumeSchedulerFailure(store, e)) return; + if (e instanceof HostFailureReport) { + if (!terminal) failCall(e.cause); + else store.hostFailure ??= e.cause; + return; + } + if (terminal) { + store.hostFailure ??= e; + return; + } + task.onFailure?.(e); + }, + ); + return terminalPromise(); }; return (...rawHostArgs: ComponentValue[]): unknown => { @@ -1396,8 +1475,29 @@ export function createLiftedFunction(input: { ? rawHostArgs[0] as unknown as PreparedTransfer : null; const hostArgs = prepared ?? rawHostArgs; + // A completed JSPI hop may have queued its canonical tail without an + // active driver. Finish that tail before classifying the window as busy; + // never bypass an unsettled hop, which would expose result memory. + if (input.refuseOnEntryHops && store.hasServiceableSettled()) { + try { + store.serviceSettled(); + } catch (e) { + if (!consumeSchedulerFailure(store, e)) { + prepared?.cleanup(e); + throw e; + } + } + } // Synchronous callers refuse before entry rather than waiting on JSPI hops. - if (input.refuseOnEntryHops && entryHopThreads(store, inst).length > 0) { + if ( + input.refuseOnEntryHops && entryHopThreads(store, inst).some((t) => { + const task = (t as { + task?: { onFailure?: unknown; inst?: { activeCalls?: Set } }; + }).task; + return task?.onFailure === null || task?.onFailure === undefined || + task.inst?.activeCalls?.has(task) !== false; + }) + ) { const refusal = new SyncEntryBusy(name); // Prepared arguments may already own resources. This is a pre-entry // refusal, so retire that custody without allowing cleanup failure to @@ -1471,7 +1571,11 @@ async function awaitHopQuiescence(store: Store, inst: unknown): Promise { ) ), ); - store.serviceSettled(); + try { + store.serviceSettled(); + } catch (e) { + if (!consumeSchedulerFailure(store, e)) throw e; + } } } diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index 4269131..5563fa9 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -26,6 +26,7 @@ import { BUFFER_MAX_LENGTH, type ByteWindow, type ComponentInstanceState, + consumeSchedulerFailure, CopyResult, type DirectBuffer, type DirectOutcome, @@ -36,6 +37,7 @@ import { SharedStreamImpl, type Store, storeQuiescent as quiescent, + unwrapSchedulerFailure, } from "../task/mod.ts"; /** @@ -249,18 +251,21 @@ class HostActivity { pump(): void { const store = this.#store; if (store === null) return; - try { - // Settled activation tails gate `tick` (Store.settled); a driver that - // never services them wedges the store — this loop runs BETWEEN export - // calls, when no driveAsync exists to do it. - for (;;) { + // Settled activation tails gate `tick` (Store.settled); a driver that + // never services them wedges the store — this loop runs BETWEEN export + // calls, when no driveAsync exists to do it. An attributed failure is one + // completed step; keep draining healthy ready siblings. + for (;;) { + try { const serviced = store.serviceSettled(); const ticked = store.tick(); if (!serviced && !ticked) break; + } catch (e) { + if (consumeSchedulerFailure(store, e)) continue; + const cause = unwrapSchedulerFailure(e); + store.hostFailure ??= cause; + throw cause; } - } catch (e) { - store.hostFailure ??= e; - throw e; } if (this.#pumping) return; // Retention alone is not work to drive; leave the host Promise pending. @@ -293,10 +298,11 @@ class HostActivity { ); } } catch (e) { + if (consumeSchedulerFailure(store, e)) return; // Nothing is awaiting this pump, so park the failure where the next // driving loop will surface it (same channel as a host-import // rejection). - store.hostFailure ??= e; + store.hostFailure ??= unwrapSchedulerFailure(e); } finally { this.#pumping = false; } diff --git a/runtime/src/intrinsics/fact_calls.ts b/runtime/src/intrinsics/fact_calls.ts index 75abc9c..a7426df 100644 --- a/runtime/src/intrinsics/fact_calls.ts +++ b/runtime/src/intrinsics/fact_calls.ts @@ -403,6 +403,10 @@ function mkCalleeTask(input: { ); task.factPassthrough = true; task.factResultTypesKnown = declaredResults !== null; + // Autonomous continuation failure belongs to the host call at the root of + // this nested FACT call. Synchronous propagation is unchanged because this + // hook is consulted only after a scheduler entry catches the escape. + task.failureOwner = maybeCurrentTask()?.failureOwner ?? task; const body = function* ( thread: Thread, diff --git a/runtime/src/jspi/bridge.ts b/runtime/src/jspi/bridge.ts index c04f225..979e5f4 100644 --- a/runtime/src/jspi/bridge.ts +++ b/runtime/src/jspi/bridge.ts @@ -379,6 +379,8 @@ export class SuspensionPoint implements SchedulableThread { #done = false; #finished = false; #store: Store; + /** The suspending import hook has returned to the platform. */ + boundaryReturned = false; /** * WHO the engine will resume when this point's promise settles. @@ -671,5 +673,14 @@ export function blockCurrentActivation(input: { owner, input.onSettled, ); + Promise.resolve().then(() => { + point.boundaryReturned = true; + if ( + point.waiting() && owner?.awaiting !== null && + owner?.awaiting !== undefined + ) { + input.task?.controlReturned?.(owner); + } + }); return point.promise; } diff --git a/runtime/src/task/mod.ts b/runtime/src/task/mod.ts index 552da6a..47db6f9 100644 --- a/runtime/src/task/mod.ts +++ b/runtime/src/task/mod.ts @@ -52,6 +52,8 @@ export class ComponentInstanceState implements ComponentInstanceLike { /** definitions.py `exclusive_thread`. */ exclusiveThread: Thread | null = null; readonly store: Store; + /** Host-visible calls not yet terminally published. */ + readonly activeCalls: Set<{ fail(cause: unknown): boolean }> = new Set(); constructor(index: number, store?: Store) { this.index = index; @@ -134,6 +136,36 @@ export class Task { */ factResultTypesKnown = false; + /** + * Host-call lifecycle hooks. `onControlReturn` is deliberately separate + * from `onResolve`: definitions.py delivers canonical resolution inside + * `Task.return_`, while public delivery is eligible only when the activation + * that performed it has returned or genuinely blocked (CanonicalABI.md + * 961-983). FACT tasks leave these unset. + */ + onControlReturn: ((thread: Thread) => void) | null = null; + onFailure: ((cause: unknown) => boolean) | null = null; + /** Root call that receives autonomous failures from this task. */ + failureOwner: Task = this; + /** Route an autonomous scheduler failure to this task's owning call. */ + fail(cause: unknown): boolean { + if (this.onFailure === null) return false; + return this.onFailure(cause); + } + + /** A generator activation finished or reached a real scheduler park. */ + controlReturned(thread: Thread): void { + this.onControlReturn?.(thread); + } + + attachCall(): void { + this.inst.activeCalls.add(this); + } + + detachCall(): void { + this.inst.activeCalls.delete(this); + } + constructor( public ft: FuncType, public opts: TaskOptions, diff --git a/runtime/src/task/scheduler.ts b/runtime/src/task/scheduler.ts index 7ba210d..e05aa39 100644 --- a/runtime/src/task/scheduler.ts +++ b/runtime/src/task/scheduler.ts @@ -155,9 +155,37 @@ export function notifyInstancePoisoned( cause: unknown, ): void { // Preserve the original cause across follow-on failures. - if (!poisonedInstances.has(inst)) poisonedInstances.set(inst, cause); - onInstancePoisoned?.(inst, cause); - for (const f of instancePoisonedListeners) f(inst, cause); + const first = !poisonedInstances.has(inst); + if (first) poisonedInstances.set(inst, cause); + const original = poisonedInstances.get(inst); + if (!first) return; + // Cleanup and terminal observers are independent obligations. A failing + // retirement hook must neither replace the first poison cause (including + // `undefined`) nor prevent pending calls from receiving terminal notice. + try { + onInstancePoisoned?.(inst, original); + } catch { + // The originating scheduler entry still routes `original` below. + } + for (const f of instancePoisonedListeners) { + try { + f(inst, original); + } catch { + // One observer cannot suppress the rest. + } + } + const calls = (inst as { + activeCalls?: Set<{ fail(cause: unknown): boolean }>; + }).activeCalls; + if (calls !== undefined) { + for (const call of [...calls]) { + try { + call.fail(original); + } catch { + // Cleanup from one call cannot suppress terminal notification to peers. + } + } + } } /** Poison causes shared by late-settle retirement and entry diagnostics. */ @@ -551,6 +579,50 @@ export interface SchedulableThread { task: any; } +/** A scheduler entry observed a failure from a specific task. The envelope is + * internal; host/public boundaries consume it and route `cause` to `origin`. */ +export class OriginatedSchedulerFailure { + constructor( + // deno-lint-ignore no-explicit-any + readonly origin: any, + readonly cause: unknown, + ) {} +} + +/** Consume an originated scheduler fault without attributing it to the + * scheduler sibling that happened to drive it. */ +export function consumeSchedulerFailure(_store: Store, e: unknown): boolean { + if (!(e instanceof OriginatedSchedulerFailure)) return false; + if (e.origin?.onFailure === null || e.origin?.onFailure === undefined) { + throw e.cause; + } + // Always address the specific root. Instance poison observers may already + // have failed a different call (or this one); aggregate membership changes + // cannot establish that this origin was observed. + e.origin.fail(e.cause); + return true; +} + +/** Raw cause for direct low-level callers; runtime entry points use + * `consumeSchedulerFailure` before crossing a public boundary. */ +export function unwrapSchedulerFailure(e: unknown): unknown { + return e instanceof OriginatedSchedulerFailure ? e.cause : e; +} + +function originatedFailure( + origin: { onFailure?: unknown } | null | undefined, + cause: unknown, +): OriginatedSchedulerFailure { + // Low-level scheduler clients have no call boundary that can consume an + // attribution envelope. Preserve their historical/raw throw contract. + if ( + origin?.onFailure === null || origin?.onFailure === undefined + ) { + throw cause; + } + return new OriginatedSchedulerFailure(origin, cause); +} + /** * Scheduler state shared by the component instances of an Executor. * `waiting` preserves insertion order for the default candidate policy. @@ -708,9 +780,24 @@ export class Store { continue scan; } this.settled.splice(i, 1); - (s.t as { + const t = s.t as { resumeWith(v: unknown, f?: { error: unknown }): void; - }).resumeWith(s.value, s.failure); + task?: { + inst?: object; + failureOwner?: unknown; + fail?(cause: unknown): boolean; + }; + }; + const origin = (t.task?.failureOwner ?? t.task) as + | { onFailure?: unknown } + | undefined; + try { + t.resumeWith(s.value, s.failure); + } catch (e) { + // Autonomous tails belong to their originating task, not whichever + // sibling happened to service the store queue (#357). + throw originatedFailure(origin, e); + } did = true; continue scan; } @@ -775,6 +862,7 @@ export class Store { if (candidates.length === 0) return false; const thread = chooseCandidate(candidates); const inst = thread.task.inst; + const origin = thread.task?.failureOwner ?? thread.task; // Capability failures do not poison; other escaping failures do. try { thread.resume(); @@ -785,7 +873,10 @@ export class Store { e, ); } - throw e; + // A store-wide tick may run a sibling. Consume the fault only when that + // task has an owning completion channel; nested FACT calls deliberately + // lack one and retain ordinary propagation to their caller. + throw originatedFailure(origin, e); } return true; } diff --git a/runtime/src/task/thread.ts b/runtime/src/task/thread.ts index 542bc5a..a93b5d2 100644 --- a/runtime/src/task/thread.ts +++ b/runtime/src/task/thread.ts @@ -193,6 +193,11 @@ export class Thread implements SchedulableThread { } if (step.done) { this.#state = "done"; + // CONTRACT: canonical resolution is captured at Task.return_, but the + // host result becomes eligible only after this execution quantum has + // returned (CanonicalABI.md:961-983). This also keeps sync post-return + // inside the quantum that may still fail. + this.task.controlReturned?.(this); return; } const req = step.value; @@ -203,6 +208,19 @@ export class Thread implements SchedulableThread { this.#state = "suspended"; this.awaiting = req.awaitValue; this.#store.noteAwaiting(this, req.awaitValue); + // If the blocking import's boundary notification already ran, this is + // now known to be a genuine park. Otherwise that notification will find + // this awaiting owner. Plain-mode test doubles have no SuspensionPoint + // and therefore remain an implementation hop. + if ( + this.#store.waiting.some((w) => + (w as { owner?: unknown; boundaryReturned?: boolean }).owner === + this && + (w as { boundaryReturned?: boolean }).boundaryReturned === true + ) + ) { + this.task.controlReturned?.(this); + } return; } if (req.readyFunc === null) { @@ -212,6 +230,7 @@ export class Thread implements SchedulableThread { this.#state = "suspended"; this.#startWaiting(req.readyFunc); } + this.task.controlReturned?.(this); } /** diff --git a/runtime/tests/call_owned_completion_test.ts b/runtime/tests/call_owned_completion_test.ts new file mode 100644 index 0000000..326bb92 --- /dev/null +++ b/runtime/tests/call_owned_completion_test.ts @@ -0,0 +1,100 @@ +import { assert, assertEquals } from "./jspi/asserts.ts"; +import { isTrap } from "@polyengine/protocol"; +import { Translator } from "../src/shim/mod.ts"; +import { instantiate } from "../src/embedder/mod.ts"; + +const root = new URL("../../", import.meta.url); +const shim = await Deno.readFile( + new URL("target/wasm32-unknown-unknown/release/translator_shim.wasm", root), +); +const fixture = await Deno.readFile( + new URL("./fixtures/two-instance-scheduler.wasm", import.meta.url), +); +const translator = await Translator.create(shim); + +function deferred() { + return Promise.withResolvers(); +} + +async function component(jspi: boolean, hostX: () => unknown) { + const translated = translator.translate(fixture); + return await instantiate({ componentBytes: fixture, ...translated }, { + hostX, + hostY: () => 1, + }, { jspi }) as { + exports: Record Promise>; + }; +} + +for (const jspi of [false, true]) { + Deno.test(`#350 call completion ignores unrelated pending host work (jspi=${jspi})`, async () => { + const gate = deferred(); + const c = await component(jspi, () => gate.promise); + const unrelated = c.exports.xCallHost(); + const next = c.exports.yNext(); + await Promise.resolve(); + assertEquals(await c.exports.yPush(9), undefined); + assertEquals(await next, 9); + gate.resolve(4); + assertEquals(await unrelated, 4); + }); + + Deno.test(`#357 sibling trap is routed to its origin (jspi=${jspi})`, async () => { + const gate = deferred(); + const c = await component(jspi, () => gate.promise); + await c.exports.xArmBad(); + const origin = c.exports.xCallHost(); + const healthy = c.exports.yNext(); + await Promise.resolve(); + gate.resolve(5); + let failure: unknown; + try { + await origin; + } catch (e) { + failure = e; + } + assert( + isTrap(failure), + `originating call must reject with Trap: ${failure}`, + ); + const beforePush = await Promise.race([ + healthy.then(() => "settled", () => "settled"), + Promise.resolve("pending"), + ]); + assertEquals(beforePush, "pending", "healthy sibling remains pending"); + assertEquals(await c.exports.yPush(9), undefined); + assertEquals(await healthy, 9); + assertEquals(await c.exports.yPing(), 42); + }); + + Deno.test(`#357 FACT root survives another poisoned callee call (jspi=${jspi})`, async () => { + const gate = deferred(); + const c = await component(jspi, () => gate.promise); + await c.exports.xArmBad(); + const unrelated = c.exports.xNext(); + const origin = c.exports.aCallX(); + await Promise.resolve(); + gate.resolve(5); + + let originFailure: unknown; + let unrelatedFailure: unknown; + try { + await origin; + } catch (e) { + originFailure = e; + } + try { + await unrelated; + } catch (e) { + unrelatedFailure = e; + } + assert( + isTrap(originFailure), + `FACT root must reject with Trap: ${originFailure}`, + ); + assert( + isTrap(unrelatedFailure), + `callee sibling must reject: ${unrelatedFailure}`, + ); + }); +} diff --git a/runtime/tests/driver_trap_exit_liveness_test.ts b/runtime/tests/driver_trap_exit_liveness_test.ts index d5efd93..163f65f 100644 --- a/runtime/tests/driver_trap_exit_liveness_test.ts +++ b/runtime/tests/driver_trap_exit_liveness_test.ts @@ -122,6 +122,46 @@ Deno.test({ }, }); +Deno.test("origin fault routing continues to a healthy ready sibling", async () => { + const store = new Store(); + const failed = { value: false }; + const origin = { + onFailure: true, + fail(cause: unknown): boolean { + assert(cause instanceof Trap, `expected Trap, got ${cause}`); + failed.value = true; + return true; + }, + }; + let healthyRan = false; + let originReady = true; + const originThread = { + task: { inst: { handles: [] }, failureOwner: origin }, + ready: () => originReady, + waiting: () => true, + resume: () => { + originReady = false; + store.stopWaiting(originThread); + throw new Trap("origin boom"); + }, + }; + store.startWaiting(originThread); + const healthyThread = { + task: { inst: { handles: [] } }, + ready: () => !healthyRan, + waiting: () => true, + resume: () => { + healthyRan = true; + store.stopWaiting(healthyThread); + }, + }; + store.startWaiting(healthyThread); + + await driveStoreAsync(store, () => healthyRan, "origin-routing probe"); + assertEq(failed.value, true); + assertEq(healthyRan, true); +}); + Deno.test({ name: "F4b: driveAsync's trap exit leaves no unserviced sibling tail in store.settled", diff --git a/runtime/tests/embedder/long_poll_test.ts b/runtime/tests/embedder/long_poll_test.ts index d1eb503..10781c6 100644 --- a/runtime/tests/embedder/long_poll_test.ts +++ b/runtime/tests/embedder/long_poll_test.ts @@ -81,14 +81,14 @@ for (const jspi of [false, true]) { // settles it synchronously, inside `push-bad`'s own driver. const nextOutcome = caught(() => next); await macrotask(); - // `push-bad` writes the future, which readies `next`'s callback, which - // traps. `push-bad`'s own call rejects... + // `push-bad` completes its own task before the independent `next` + // callback traps. Origin routing (#357) therefore preserves its result. const pushErr = await caught(() => (c.exports.pushBad as (v: number) => Promise)(7) ); - assertEq(pushErr !== undefined, true, "push-bad must reject"); - // ...and so must the Promise nobody was driving. Without the poisoning - // seam this hangs forever (the #66 failure, for lifts). + assertEq(pushErr, undefined, "completed push-bad must keep its result"); + // The callback's originating pending call rejects and poisons subsequent + // entry; the scheduler sibling that happened to drive it does not. const err = await nextOutcome; assertEq(err !== undefined, true, "the pending next() must reject"); assertEq( diff --git a/runtime/tests/fixtures/two-instance-scheduler.wasm b/runtime/tests/fixtures/two-instance-scheduler.wasm new file mode 100644 index 0000000000000000000000000000000000000000..8d546521d3bae0a86dba018c64bd3adfdd95b0ed GIT binary patch literal 2704 zcmcIm&2A$_5U#53=^jt}XR>Hnv_iy@BV~o;91vP0P8lL)Pb)4Ut~-eb;@$WM+ethJ z))ypr0!~~w?HlkqNW1}WKvjEYJmXyu7Y-ixPgU3N_jNw30a4X@Ih8w4DP00KaucxElCNj4z>r>E&81FfF(REvPsnBfm$pazEz(X5$r^U*ixyjuqoTbvwyRWqEkw4*+8X}~O84+2d_kKHya3$r@3l;Lgt7ws z$78jTCHe^??-y|bI``>#m0sRtr!QyYsRXgiR;SbKp7)|}Z&$aA%+Cm1FHT87%4EL0yoGSsAto~WO&KW#(F-Ueg_B9pV)dz z_!>7c9YG)79P{Uk{?BiYxyVTvt7P*`pMi_9E#1P_kBVa`Yh=`A+O)_=H&FdhQ$*R zMt@eH5~`VIv7l0SiaXJ;Oq5QL=CzTHUw4NuuXw) zVY`bQLSP_1z=Ar=M_de~ZPOOmmxZ!@LyjC;$|z$avLT)$-sK3Tns4p8TG7c~&9R!b zZB?3j;!V|XP*ZFm!fSU}JG5QR+Gh-G(_u&0uL2;U;b{*7zYDpE{H_O$--^Mwrvdo5 zgm^p%ktuLz9@(ApIMgD-h{FKjckEuxCiB@en=%~;VG_qZXWc#O`5#&|s%ZQK9UKq{ zRU^Z~!2_NMjpGD69MZ0&$drt9@s7DdZ%O^b}u9wu2sx*`a0QZy71h?`j&`kh;=uzd}fYk)*6a$Je{a zar409?G{h)Gk!3D-S*4LuDP2|sfJHJ5C77s5gRMwtoNkv@^!OM{aTI~95PI?*52GF z^(mUJM6d&Lz|??P_Bk1OY54wd{4NSp|0o)+8SY#LR`i_Un9R-EU-k_7Vnf)h*pMMw z*f-@45ZrouqWH^iAJdjYOCoVSM&^20uoNezJes+sJX&a`v)$l(kDm#CBi}mHD>o`< za(_|b!PXhv#qFKRYtFp4bp{*G { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + inst.backpressure = 1; + let cleaned = 0; + const prepared = new PreparedValues([], []); + prepared.custody.acquire(() => cleaned++); + const failure = new TypeError("host conversion failed"); + store.hostFailure = failure; + const call = createLiftedFunction({ + name: "raw-host-failure", + ft: { params: [], results: [], async: true }, + opts: { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: () => () => [0], + async: true, + cancellable: false, + coreType: { params: [], results: ["i32"] }, + instance: inst, + }, + core: () => [0], + stats: newStats(), + }); + + const pending = call(prepared as unknown as ComponentValue) as Promise< + unknown + >; + assertEq(await rejected(pending), failure); + assertEq(isInstancePoisoned(inst), false); + assertEq(cleaned, 1); + assertEq(inst.numWaitingToEnter, 0); +}); + function caught(fn: () => unknown): unknown { try { fn(); @@ -297,7 +334,11 @@ for (const timing of ["before-listener", "after-listener"] as const) { >; assertEq(acquired, 1); assertEq(entered, 0); - assertEq(inst.numWaitingToEnter, 1); + assertEq( + inst.numWaitingToEnter, + timing === "before-listener" ? 0 : 1, + "an origin failure consumed by the starting driver retires admission immediately", + ); if (timing === "after-listener") notifyInstancePoisoned(inst, poison); assertEq(await rejected(pending), poison); assertEq(cleaned, 1); diff --git a/runtime/tests/lift_done_verdict_test.ts b/runtime/tests/lift_done_verdict_test.ts index f7a07f4..cbf4b73 100644 --- a/runtime/tests/lift_done_verdict_test.ts +++ b/runtime/tests/lift_done_verdict_test.ts @@ -160,3 +160,66 @@ Deno.test({ assertEq(store.pendingHostCalls.size, 0); }, }); + +Deno.test("sync entry keeps a synchronous result after a foreign routed fault", () => { + const store = new Store(); + const healthyInst = new ComponentInstanceState(0, store); + const failedInst = new ComponentInstanceState(1, store); + let originFailed = false; + let ready = true; + let healthyReadyRan = false; + const origin = { + onFailure: true, + fail() { + originFailed = true; + return true; + }, + }; + const foreign = { + task: { inst: failedInst, failureOwner: origin }, + ready: () => ready, + waiting: () => ready, + resume: () => { + ready = false; + store.stopWaiting(foreign); + throw new Error("foreign sync fault"); + }, + }; + store.startWaiting(foreign); + const healthyReady = { + task: { inst: new ComponentInstanceState(2, store) }, + ready: () => !healthyReadyRan, + waiting: () => !healthyReadyRan, + resume: () => { + healthyReadyRan = true; + store.stopWaiting(healthyReady); + }, + }; + store.startWaiting(healthyReady); + const fn = createLiftedFunction({ + name: "sync-result", + ft: { params: [], results: [{ kind: "u32" }], async: false }, + opts: { + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: null, + async: false, + cancellable: false, + coreType: { params: [], results: ["i32"] }, + instance: healthyInst, + }, + core: () => [42], + stats: newStats(), + }); + const result = fn(); + assertEq(result, 42); + assertEq(result instanceof Promise, false); + assertEq(originFailed, true); + assertEq( + healthyReadyRan, + true, + "the export driver must continue to the healthy ready thread", + ); +}); diff --git a/runtime/tests/poison_cause_test.ts b/runtime/tests/poison_cause_test.ts index 4bb23b8..c9bd27a 100644 --- a/runtime/tests/poison_cause_test.ts +++ b/runtime/tests/poison_cause_test.ts @@ -15,6 +15,7 @@ import { assertEq } from "./support/asserts.ts"; import { + addInstancePoisonedListener, instancePoisonCause, isInstancePoisoned, notifyInstancePoisoned, @@ -85,3 +86,35 @@ Deno.test("poison cause: non-Error and unprintable causes degrade safely", () => "base — instance poisoned by: (unprintable poison cause)", ); }); + +Deno.test("poison cause: thrown undefined remains first and all waiters run", () => { + const inst = fakeInst() as { + handles: Iterable; + activeCalls: Set<{ fail(cause: unknown): boolean }>; + }; + inst.activeCalls = new Set(); + const seen: unknown[] = []; + inst.activeCalls.add({ + fail(cause) { + seen.push(cause); + throw new Error("cleanup failed"); + }, + }); + inst.activeCalls.add({ + fail(cause) { + seen.push(cause); + return true; + }, + }); + addInstancePoisonedListener((candidate, cause) => { + if (candidate === inst) seen.push(cause); + }); + + notifyInstancePoisoned(inst, undefined); + notifyInstancePoisoned(inst, new Error("secondary")); + + assertEq(isInstancePoisoned(inst), true); + assertEq(instancePoisonCause(inst), undefined); + assertEq(seen.length, 3, "a throwing cleanup must not skip later waiters"); + assertEq(seen.every((cause) => cause === undefined), true); +});