host streams: a trap out of pump()'s sync half is recorded and the op withdrawn (review)

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<T>(
     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<T>(
             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,16 +1138,26 @@ function mkStreamEnds<T>(
     }
     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
     // total (future abandonment's indistinguishability caveats unchanged). `shared.cancel()`
     // 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<T>(
             },
           );
           activity.notify();
-          activity.pump();
+          try {
+            activity.pump();
+          } catch (e) {
+            withdraw("write", buf);
+            throw e;
+          }
         });
       },
       async writeAll(values: T[]): Promise<number> {
@@ -1255,7 +1306,12 @@ function mkStreamEnds<T>(
             },
           );
           activity.notify();
-          activity.pump();
+          try {
+            activity.pump();
+          } catch (e) {
+            withdraw("read", buf);
+            throw e;
+          }
         });
       },
       readDirect(
@@ -1444,6 +1500,13 @@ function mkFuture<T>(
     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<T> = {
     write(v: T): Promise<void> {
       // One in-flight operation per wrapper — see mkStreamEnds' guards: a
@@ -1464,7 +1527,12 @@ function mkFuture<T>(
           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<T>(
           });
         });
         activity.notify();
-        activity.pump();
+        try {
+          activity.pump();
+        } catch (e) {
+          withdraw(buf);
+          throw e;
+        }
       });
     },
     async read(): Promise<T | undefined> {

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<typeof fakeInst> };
+  constructor(
+    private readonly body: () => void,
+    inst: ReturnType<typeof fakeInst>,
+  ) {
+    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<T>(store: Store, element: ValType | null) {
+  const shared = new SharedStreamImpl(element);
+  (shared as unknown as { boundStore: unknown }).boundStore = store;
+  return {
+    shared,
+    host: hostStreamFor<T>(shared as unknown as ComponentValue),
+  };
+}
+
+/** Bounded settle probe — N macrotask turns, no long sleeps. */
+async function settleTurns(n = 20): Promise<void> {
+  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<number>(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<number> | 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<number>(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<number[]> | 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<number>(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);
+  },
+});