From d59b23302238582e8e746af96973d350cebd5770 Mon Sep 17 00:00:00 2001 From: Lann Martin Date: Sun, 6 Sep 2026 22:03:26 -0400 Subject: [PATCH] host streams: a trap out of pump()'s sync half is recorded and the op withdrawn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `HostActivity.pump()`'s synchronous `serviceSettled()/tick()` loop had no try/catch, and every host stream/future op invokes it inside its `new Promise(executor)` after setting `parked.*`. A trap from `tick()` therefore went one of two bad ways: - the trapping instance held an end of THIS stream: the poisoning retirement walk had already settled the op's promise, so the executor's throw was discarded — the component fault was mute (unlike `#pumpAsync`, which records `store.hostFailure`); - otherwise: the promise rejected, but `parked.*` stayed true and the host buffer stayed in the shared pending slot forever — every later op on that end threw "a write is already in flight", and a direct session's dead `pending` state re-ran the embedder's callback on the next guest op. The sync half now records `store.hostFailure ??= e` and rethrows (the one channel `#pumpAsync` already uses), and each op's executor withdraws on a throw — resets its `parked` flag and, while the pending slot is still its own buffer/session, `shared.cancel()`s it, the same withdrawal `cancelWrite`/`cancelRead`/`cancelDirect` perform (`retractDirect` is `cancelDirect`'s pending-slot half, factored out and shared). Regression: host_pump_trap_test.ts (write, read, and the silent-fault variant). Neighbouring suites (host_pump, host_arm, direct_streams, streams_teardown, e2e_streams, embedder/) unchanged. --- runtime/src/exec/host_streams.ts | 105 +++++++++++-- runtime/tests/host_pump_trap_test.ts | 215 +++++++++++++++++++++++++++ 2 files changed, 304 insertions(+), 16 deletions(-) create mode 100644 runtime/tests/host_pump_trap_test.ts diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index 4b7907f..7878107 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -368,18 +368,29 @@ class HostActivity { * Promise-returning host import stalled the reader). * * Traps from the synchronous half propagate to the caller of the host - * operation, which is the only place that can report them. + * operation AND are recorded on `store.hostFailure`, the same channel + * `#pumpAsync` uses: propagation alone is not enough, because the caller is + * a host op's promise executor whose promise may already have been settled + * by the poisoning retirement walk (the trapping instance held an end of + * this very stream), in which case the throw is discarded and the fault + * would be mute. A component fault is always loud + * (contracts/embedder-api.md §"Streams and futures"). */ pump(): void { const store = this.#store; if (store === null) return; - // 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 (;;) { - const serviced = store.serviceSettled(); - const ticked = store.tick(); - if (!serviced && !ticked) break; + 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 (;;) { + const serviced = store.serviceSettled(); + const ticked = store.tick(); + if (!serviced && !ticked) break; + } + } catch (e) { + store.hostFailure ??= e; + throw e; } if (this.#pumping) return; // Nothing is outstanding that only an event-loop turn could advance ⇒ no @@ -1034,6 +1045,22 @@ function mkStreamEnds( if (result === CopyResult.DROPPED) activity.close(); else activity.notify(); }; + /** + * Withdraw an operation that never got to finish: a trap out of + * `activity.pump()`'s synchronous half unwinds through the op's promise + * executor with our bookkeeping half-done — the `parked` flag set and our + * buffer still in the shared object's pending slot, which wedges the end + * ("a write is already in flight") for good. Same withdrawal + * `cancelWrite`/`cancelRead` perform; `shared.cancel()` only while the + * pending side is still literally ours, since `SharedBase.cancel` asserts + * that something is pending and the poisoning walk may have retired it. + */ + const withdraw = (side: "read" | "write", buf: unknown): void => { + if (!parked[side]) return; + parked[side] = false; + if (shared.pendingBuffer === buf as never) shared.cancel(); + activity.notify(); + }; /** The live direct session on each end, if any (direct-access byte edge, polyengine#128). */ const direct: { read: DirectSession | null; write: DirectSession | null } = { read: null, @@ -1092,7 +1119,16 @@ function mkStreamEnds( shared.read(readInst, session as never, onCopy, onCopyDone); } activity.notify(); - activity.pump(); + try { + activity.pump(); + } catch (e) { + // The `finally` below clears `parked`/`direct`, but the session + // itself would stay in the pending slot with `pending` true and + // its promise rejected — the next guest op would re-run the + // embedder's callback on a dead session. Retract it first. + retractDirect(session); + throw e; + } }); if (step === "done") break; } @@ -1102,6 +1138,18 @@ function mkStreamEnds( } return session.total; }; + /** + * Retract a direct session from the rendezvous: the pending-slot half of + * `cancelDirect`, shared with the pump-trap unwind above. + */ + const retractDirect = (session: DirectSession): void => { + session.cancelled = true; + if (session.pending && shared.pendingBuffer === session as never) { + shared.cancel(); + } else { + session.finish(); + } + }; /** Shared tail of `cancelWrite`/`cancelRead` for a parked direct session. */ const cancelDirect = (session: DirectSession): void => { // direct-access byte edge: cancelling RETRACTS the session — it resolves with its running @@ -1109,9 +1157,7 @@ function mkStreamEnds( // only when the session actually holds the pending slot: a session caught // between two issuances holds nothing, and `SharedBase.cancel` asserts // that something is pending. - session.cancelled = true; - if (session.pending) shared.cancel(); - else session.finish(); + retractDirect(session); activity.notify(); activity.pump(); }; @@ -1165,7 +1211,12 @@ function mkStreamEnds( }, ); activity.notify(); - activity.pump(); + try { + activity.pump(); + } catch (e) { + withdraw("write", buf); + throw e; + } }); }, async writeAll(values: T[]): Promise { @@ -1255,7 +1306,12 @@ function mkStreamEnds( }, ); activity.notify(); - activity.pump(); + try { + activity.pump(); + } catch (e) { + withdraw("read", buf); + throw e; + } }); }, readDirect( @@ -1444,6 +1500,13 @@ function mkFuture( if (result === CopyResult.DROPPED) activity.close(); else activity.notify(); }; + /** See `mkStreamEnds`' `withdraw`: the pump-trap unwind path (F1). */ + const withdraw = (buf: unknown): void => { + if (!parked.any) return; + parked.any = false; + if (shared.pendingBuffer === buf as never) shared.cancel(); + activity.notify(); + }; const self: HostFuture = { write(v: T): Promise { // One in-flight operation per wrapper — see mkStreamEnds' guards: a @@ -1464,7 +1527,12 @@ function mkFuture( resolve(); }); activity.notify(); - activity.pump(); + try { + activity.pump(); + } catch (e) { + withdraw(buf); + throw e; + } }); }, readResult(): Promise<{ value: T | undefined; result: CopyResult }> { @@ -1495,7 +1563,12 @@ function mkFuture( }); }); activity.notify(); - activity.pump(); + try { + activity.pump(); + } catch (e) { + withdraw(buf); + throw e; + } }); }, async read(): Promise { diff --git a/runtime/tests/host_pump_trap_test.ts b/runtime/tests/host_pump_trap_test.ts new file mode 100644 index 0000000..6edfc12 --- /dev/null +++ b/runtime/tests/host_pump_trap_test.ts @@ -0,0 +1,215 @@ +// A trap raised by `HostActivity.pump()`'s SYNCHRONOUS half is mishandled. +// +// Every host stream/future op sets `parked.write`/`parked.read` and then calls +// `activity.pump()` from INSIDE its `new Promise(executor)`. `pump()`'s +// `serviceSettled()/tick()` loop has no try/catch, so a guest thread that +// traps under that `tick` throws out through the executor. Two outcomes, both +// wrong (exec/host_streams.ts:373-394 vs the `#pumpAsync` catch at :447-451): +// (a) the trapping instance held an end of this stream — the poison +// retirement walk already settled the op's promise, so the executor's +// throw lands on a settled promise and is DISCARDED; nothing records +// `store.hostFailure`. A component fault is "always loud" +// (contracts/embedder-api.md §"Streams and futures"); this one is mute. +// (b) the trapping instance held no end — the promise rejects, but +// `parked.*` stays true and the host buffer stays in the shared pending +// slot, so the end is wedged: every later op throws "already in flight". + +import { assertEq } from "./support/asserts.ts"; +import { hostStreamFor } from "../src/exec/mod.ts"; +import { + ReadableStreamEnd, + SharedStreamImpl, + Store, +} from "../src/task/mod.ts"; +import { Trap } from "../src/cabi/trap.ts"; +import type { ComponentValue, ValType } from "../src/cabi/types.ts"; + +const U8: ValType = { kind: "u8" }; + +function assert(cond: boolean, msg: string): asserts cond { + if (!cond) throw new Error(`assertion failed: ${msg}`); +} + +/** + * The slice of `ComponentInstance` the poisoning path touches: `Store.tick` + * routes a resume trap through `notifyInstancePoisoned`, whose walk + * (task/streams.ts `retireInstanceAsyncEnds`) iterates `inst.handles` looking + * for `CopyEnd`s. `handles: []` = an instance holding no stream end. + */ +function fakeInst(handles: unknown[] = []) { + return { handles }; +} + +/** A guest thread stand-in; `Store.tick` resumes it whenever `ready()`. */ +class FakeThread { + #ready = true; + readonly task: { inst: ReturnType }; + constructor( + private readonly body: () => void, + inst: ReturnType, + ) { + this.task = { inst }; + } + ready(): boolean { + return this.#ready; + } + waiting(): boolean { + return !this.#ready; + } + wake(): void { + this.#ready = true; + } + park(): void { + this.#ready = false; + } + resume(): void { + this.#ready = false; + this.body(); + } +} + +/** A host stream end wired to `store`, as a lower into a guest would wire it. */ +function hostEndOn(store: Store, element: ValType | null) { + const shared = new SharedStreamImpl(element); + (shared as unknown as { boundStore: unknown }).boundStore = store; + return { + shared, + host: hostStreamFor(shared as unknown as ComponentValue), + }; +} + +/** Bounded settle probe — N macrotask turns, no long sleeps. */ +async function settleTurns(n = 20): Promise { + for (let i = 0; i < n; i++) await new Promise((r) => setTimeout(r, 0)); +} + +// --------------------------------------------------------------------------- +// (b) the trapping instance holds no end of this stream: the end must not be +// left permanently "in flight". +// --------------------------------------------------------------------------- + +Deno.test({ + name: + "host write: a trap from the sync pump does not wedge the writable end in flight", + fn: async () => { + const store = new Store(); + const { host } = hostEndOn(store, U8); + + // A ready thread of an unrelated instance (no stream ends) that traps the + // moment the host op's `pump()` ticks the store. + const trap = new Trap("boom"); + store.startWaiting(new FakeThread(() => { + throw trap; + }, fakeInst())); + + const first = host.writable.write([1]); + let firstErr: unknown = undefined; + let firstOk = false; + first.then(() => (firstOk = true), (e) => (firstErr = e)); + await settleTurns(3); + assert( + firstOk || firstErr !== undefined, + "the first write neither resolved nor rejected", + ); + + // The fault has been delivered (or recorded). The END, however, belongs + // to the embedder and must still be usable: the trap was raised by an + // instance that holds no end of this stream, so nothing about this + // stream's state legitimately changed. + let second: Promise | undefined; + let secondThrow: unknown = undefined; + try { + second = host.writable.write([2]); + } catch (e) { + secondThrow = e; + } + assert( + secondThrow === undefined, + `the writable end is wedged after the pump trap: ${ + (secondThrow as Error)?.message + }`, + ); + second?.catch(() => {}); + host.writable.drop(); + await settleTurns(2); + }, +}); + +Deno.test({ + name: + "host read: a trap from the sync pump does not wedge the readable end in flight", + fn: async () => { + const store = new Store(); + const { host } = hostEndOn(store, U8); + + const trap = new Trap("boom"); + store.startWaiting(new FakeThread(() => { + throw trap; + }, fakeInst())); + + const first = host.readable.read(8); + let settled = false; + first.then(() => (settled = true), () => (settled = true)); + await settleTurns(3); + assert(settled, "the first read neither resolved nor rejected"); + + let second: Promise | undefined; + let secondThrow: unknown = undefined; + try { + second = host.readable.read(8); + } catch (e) { + secondThrow = e; + } + assert( + secondThrow === undefined, + `the readable end is wedged after the pump trap: ${ + (secondThrow as Error)?.message + }`, + ); + second?.catch(() => {}); + host.readable.drop(); + await settleTurns(2); + }, +}); + +// --------------------------------------------------------------------------- +// (a) the trapping instance holds an end of THIS stream: the fault must stay +// loud (embedder-api.md §"Streams and futures"). +// --------------------------------------------------------------------------- + +Deno.test({ + name: "host write: a trap from the sync pump is not swallowed silently", + fn: async () => { + const store = new Store(); + const shared = new SharedStreamImpl(U8); + (shared as unknown as { boundStore: unknown }).boundStore = store; + const host = hostStreamFor(shared as unknown as ComponentValue); + + // This time the trapping instance holds the guest end of the very stream + // the host is writing to: `Store.tick`'s poisoning path runs the + // retirement walk over `handles`, which drops `shared` and settles our + // parked write (DROPPED-shaped) BEFORE the trap propagates out of + // `pump()` — so the executor's throw hits an already-settled promise. + const guestEnd = new ReadableStreamEnd(shared); + const trap = new Trap("boom"); + store.startWaiting(new FakeThread(() => { + throw trap; + }, fakeInst([guestEnd]))); + + const p = host.writable.write([1]); + let rejected: unknown = undefined; + let resolved: number | undefined = undefined; + await p.then((n) => (resolved = n), (e) => (rejected = e)); + await settleTurns(3); + + // Loud, one way or the other: either the op rejects with the fault, or the + // driver channel carries it (what `#pumpAsync` does with the same trap). + assert( + rejected !== undefined || store.hostFailure !== undefined, + `the component fault was swallowed: write resolved ${resolved}, ` + + `store.hostFailure=${store.hostFailure}`, + ); + if (rejected !== undefined) assertEq(rejected, trap); + await settleTurns(2); + }, +});