driver: stale settled entry, poisoned-probe spin, corpse settlements, exception-exit hand-off (review)

runtime/src/exec/boundary.ts

@@ -34,6 +34,7 @@
   EventCode,
   withActivation,
   hasRealHostCall,
+  isInstancePoisoned,
   type EventTuple,
   NeedsJspi,
   needsJspi,
@@ -217,7 +218,7 @@
     stringEncoding: opts.stringEncoding,
     memory: opts.memory,
     realloc: opts.realloc === null ? null : (o, os, a, n) => {
       const realloc = require(opts.realloc, "realloc")!;
       const p = callCore(realloc, [o, os, a, n]);
       trapIf(p.length !== 1 || typeof p[0] !== "number", "realloc result");
       return (p[0] as number) >>> 0;
@@ -442,6 +443,24 @@
   store: Store,
   done: () => boolean,
   what: string,
+): void | Promise<void> {
+  try {
+    return driveLoop(store, done, what);
+  } catch (e) {
+    // EXIT BY EXCEPTION IS STILL AN EXIT. A trap unwinds this call, but the
+    // sibling work this loop already started — a registered host call, a
+    // queued tail (which gates `Store.tick`) — does not unwind with it. The
+    // `done()` path hands both to the settlement pump; so must this one.
+    ensureSettlementPump(store);
+    throw e;
+  }
+}
+
+/** `drive`'s loop proper; see `drive` for the exception-exit hand-off. */
+function driveLoop(
+  store: Store,
+  done: () => boolean,
+  what: string,
 ): void | Promise<void> {
   for (;;) {
     traceDrive("drive", store, done, "top");
@@ -832,6 +851,18 @@
   }
 }
 
+/**
+ * What an exiting driver must hand over: real host calls it was watching, a
+ * queued activation tail (which gates `Store.tick` for every later driver),
+ * or a hop-parked thread — the #280 rule ("a driver is not done while ANY
+ * thread of ANY task is hop-parked") applied to the exits that cannot
+ * evaluate their `done` predicate, i.e. the exception exits.
+ */
+function pumpWork(store: Store): boolean {
+  return hasRealHostCall(store) || store.settled.length > 0 ||
+    entryHopThreads(store).length > 0;
+}
+
 /**
  * Ensure a settlement pump is watching `store`'s real outstanding host calls.
  * Idempotent and cheap; called at every driver exit. Never throws.
@@ -844,7 +875,7 @@
     return;
   }
   if (store.hostFailure !== undefined) return;
-  if (!hasRealHostCall(store)) return;
+  if (!pumpWork(store)) return;
   settlementPumps.add(store);
   void settlementPumpLoop(store);
 }
@@ -863,20 +894,32 @@
       // it in a loop.
       if (store.hostFailure !== undefined) return;
       const real = realHostCalls(store);
-      if (real.length === 0) return;
-      const nudge = armSettlementNudge(store);
-      // The host-call arrival one-shot does NOT ride here, deliberately: a
-      // registration this snapshot misses always reaches `ensureSettlementPump`
-      // (which fires the nudge) — `driveAsync`'s finally, `drive`'s synchronous
-      // completion, and `HostActivity.pump()`'s async half is itself a
-      // `driveStoreAsync`. Any driver live meanwhile is what we stand down for.
-      // Rejections are not this pump's to report: the registration site's
-      // own continuation parks them on `store.hostFailure`.
-      await Promise.race([
-        ...real.map((p) => p.then(() => {}, () => {})),
-        nudge,
-      ]);
-      if (storeDriverDepth(store) > 0) continue;
+      // A queued tail is serviceable RIGHT NOW: drive without parking. A
+      // hop-parked thread lands on the engine's own schedule, so its promise
+      // is raced alongside the host calls — that is how an exception exit's
+      // orphaned hop (F4/#280) gets an owner.
+      const hops = store.settled.length > 0
+        ? []
+        : entryHopThreads(store).map((t) => t.awaiting).filter((
+          p,
+        ): p is Promise<unknown> => p !== null);
+      if (store.settled.length === 0) {
+        if (real.length === 0 && hops.length === 0) return;
+        const nudge = armSettlementNudge(store);
+        // The host-call arrival one-shot does NOT ride here, deliberately: a
+        // registration this snapshot misses always reaches `ensureSettlementPump`
+        // (which fires the nudge) — `driveAsync`'s finally, `drive`'s synchronous
+        // completion, and `HostActivity.pump()`'s async half is itself a
+        // `driveStoreAsync`. Any driver live meanwhile is what we stand down for.
+        // Rejections are not this pump's to report: the registration site's
+        // own continuation parks them on `store.hostFailure`.
+        await Promise.race([
+          ...real.map((p) => p.then(() => {}, () => {})),
+          ...hops.map((p) => p.then(() => {}, () => {})),
+          nudge,
+        ]);
+        if (storeDriverDepth(store) > 0) continue;
+      }
       // Drive unconditionally after a wake: `storeQuiescent` cannot see a
       // READY waiting thread (the usual product of a settlement — the
       // continuation readied the guest and deleted its own host call), so
@@ -906,7 +949,7 @@
     // fired the nudge after our last snapshot check must not be lost.
     if (
       !failed && store.hostFailure === undefined &&
-      storeDriverDepth(store) === 0 && hasRealHostCall(store)
+      storeDriverDepth(store) === 0 && pumpWork(store)
     ) {
       ensureSettlementPump(store);
     }
@@ -1237,8 +1280,13 @@
       // resumption site here re-checks membership and promise identity
       // synchronously — mechanisms (a) and (b), which is where that note
       // already puts the weight.
+      // ONLY IF WE ADDED IT (issue #158, same rule as the `finally` below):
+      // `pendingResumptions` is a Set by identity, so a genuine entry for
+      // `chosen` minted meanwhile — or already held — collapses with ours,
+      // and removing "ours" would drop the genuine one.
       const sole = storeDriverDepth(store) === 1;
-      if (sole) store.addPendingResumption(chosen);
+      const added = sole && !store.pendingResumptions.has(chosen);
+      if (added) store.addPendingResumption(chosen);
       let winner: AwaitWinner | null;
       try {
         // `armDriverArrival` rides the race for every driver, not just the one
@@ -1257,7 +1305,7 @@
           armHostCallArrival(store),
         ]);
       } finally {
-        if (sole) store.removePendingResumption(chosen);
+        if (added) store.removePendingResumption(chosen);
       }
       // Resume whichever thread actually settled -- not necessarily the one we
       // claimed. Resuming only the claimed thread would spin: its promise may
@@ -1271,10 +1319,18 @@
       // promise, after which its OLD promise settles late — membership is
       // true again but the tag's value belongs to a settlement this thread
       // has already consumed. Compare promise identity too.
+      // ONE SETTLEMENT, ONE DELIVERY (definitions.py `Thread.resume` is
+      // atomic). `noteAwaiting` records settlements EAGERLY, so this
+      // promise's `store.settled` entry is already queued; left there, a
+      // body that re-parks SYNCHRONOUSLY inside `resumeWith` gets the OLD
+      // value delivered against its NEW park by the next `serviceSettled`.
       if (
         winner !== null && store.awaiting.has(winner.t) &&
         winner.t.awaiting === winner.p
       ) {
+        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);
       }
       continue;
@@ -2132,7 +2188,7 @@
     task.return_(results);
     // Post-return runs after the results were read out of guest memory,
     // with may_leave cleared (reference canon_lift).
     const postReturn = require(opts.postReturn, `${name} post-return`);
     if (postReturn !== null) {
       assert_(inst.mayLeave, "post-return with may_leave already false");
       inst.mayLeave = false;
@@ -2178,7 +2234,7 @@
   // *mixed* activation, which pin (c) punishes: the first Suspending import
   // it reached would trap.
   const callback = enterWasm(
     require(opts.callback, `${name} callback`)!,
     input.mode,
   );
   const [packed] = normalizeCoreValues(
@@ -2505,7 +2561,9 @@
           // subtask that never started") and park that AssertionError on
           // `store.hostFailure`, poisoning whatever unrelated embedder call
           // came next.
-          if (subtask.resolved()) return;
+          // POISONED is the same discard (arch §6 #173): no addressee, and
+          // lowering would write into the corpse's memory via its `realloc`.
+          if (subtask.resolved() || isInstancePoisoned(opts.instance)) return;
           try {
             onResolve(toResults(v));
           } catch (e) {
@@ -2517,8 +2575,9 @@
           // Same guard, different reason: a rejection of a RENOUNCED call is
           // not a host failure. The guest cancelled and was told so; surfacing
           // the rejection would fail an unrelated later call with the error of
-          // an operation nobody is waiting for.
-          if (subtask.resolved()) return;
+          // an operation nobody is waiting for. POISONED is the same discard
+          // (arch §6 #173): it would fail a HEALTHY sibling's export call.
+          if (subtask.resolved() || isInstancePoisoned(opts.instance)) return;
           store.hostFailure = e;
         },
       );

runtime/src/task/scheduler.ts

@@ -937,9 +937,21 @@ export class Store {
     this.waiting.splice(i, 1);
   }
 
-  /** Ready waiting threads, in wait order (the FIFO of the default policy). */
+  /**
+   * Ready waiting threads, in wait order (the FIFO of the default policy).
+   *
+   * A POISONED instance's threads are not candidates: they are a corpse's and
+   * must never resume (polyengine's per-instance poisoning divergence). The
+   * filter lives here, not in `tick` alone, because the answer is also a
+   * VERDICT elsewhere — the drivers' deadlock probe asks "did anything become
+   * ready?" and must get an answer that agrees with what `tick` will actually
+   * run, or it re-arms forever on a thread `tick` refuses (exec/boundary.ts's
+   * probe, against `canon_lift`'s `trap_if(not candidates)`).
+   */
   readyCandidates(): SchedulableThread[] {
-    return this.waiting.filter((t) => t.ready());
+    return this.waiting.filter((t) =>
+      t.ready() && !isInstancePoisoned(t.task?.inst)
+    );
   }
 
   /**
@@ -1166,11 +1178,10 @@ export class Store {
     //
     // What is added is polyengine's per-instance poisoning divergence: a
     // poisoned instance is a corpse, its threads must never resume, and the
-    // MARKER is the whole test. `Thread.resumeWith` makes the same call on
-    // the tail path.
-    const candidates = this.readyCandidates().filter((t) =>
-      !isInstancePoisoned(t.task.inst)
-    );
+    // MARKER is the whole test. That filter lives in `readyCandidates` (so
+    // the drivers' deadlock probe reads the same candidate set this does).
+    // `Thread.resumeWith` makes the same call on the tail path.
+    const candidates = this.readyCandidates();
     if (candidates.length === 0) return false;
     const thread = chooseCandidate(candidates);
     const inst = thread.task.inst;

runtime/tests/driver_poisoned_probe_test.ts

@@ -0,0 +1,88 @@
+// F2: the deadlock probe consults `readyCandidates()`, which `tick` does not.
+//
+// THE SHAPE (store-level: the end-to-end participants are two live activations
+// of ONE instance under stackful async lifts, one JSPI-suspended while its
+// sibling traps — no checked-in example guest has them).
+//
+//   * instance I is poisoned; a waiting entry of I is `ready()`;
+//   * `Store.tick` filters poisoned instances out of its candidate set
+//     (task/scheduler.ts:1171) and returns false — nothing can run;
+//   * the driver's deadlock probe reads the UNFILTERED `readyCandidates()`
+//     (exec/boundary.ts:1111), concludes "a thread became READY", and
+//     `continue`s. Next turn: identical. Forever, one macrotask per turn.
+//
+// definitions.py `canon_lift`'s sync loop traps on an empty candidate set. The
+// verdict must agree with what `tick` can actually run, so the abandoned
+// export call must REJECT with the deadlock trap, not idle-spin.
+
+import { assertEq } from "./support/asserts.ts";
+import { driveStoreAsync } from "../src/exec/mod.ts";
+import { notifyInstancePoisoned, Store } from "../src/task/mod.ts";
+
+function assert(cond: boolean, msg: string): asserts cond {
+  if (!cond) throw new Error(`assertion failed: ${msg}`);
+}
+
+Deno.test({
+  name:
+    "F2: a driver whose only ready thread belongs to a poisoned instance traps instead of spinning",
+  fn: async () => {
+    const store = new Store();
+    const inst = { handles: [] as unknown[] };
+    notifyInstancePoisoned(inst, new Error("the sibling activation trapped"));
+
+    // The abandoned export's own activation: parked on a promise nobody will
+    // ever settle (its wasm frame died with the instance).
+    const parked = {
+      task: { inst },
+      awaiting: new Promise<unknown>(() => {}) as Promise<unknown> | null,
+      resumeWith(): void {
+        throw new Error("a poisoned instance's thread must never resume");
+      },
+    };
+    store.noteAwaiting(parked, parked.awaiting!);
+
+    // The sibling suspension point that stays `ready()` forever: `tick` will
+    // never pick it (poisoned), but `readyCandidates()` still reports it.
+    const sp = {
+      owner: parked,
+      task: { inst },
+      ready: () => true,
+      waiting: () => true,
+      resume(): void {
+        throw new Error("a poisoned instance's thread must never resume");
+      },
+    };
+    // deno-lint-ignore no-explicit-any
+    store.startWaiting(sp as any);
+
+    let finished = false;
+    const driving = driveStoreAsync(store, () => finished, "export 'abandoned'");
+    let outcome: { ok: true } | { err: unknown } | undefined;
+    driving.then(() => (outcome = { ok: true }), (e) => (outcome = { err: e }));
+
+    // The probe costs one macrotask per turn; a correct verdict needs a couple
+    // of them. Bounded, per the brief.
+    for (let i = 0; i < 20 && outcome === undefined; i++) {
+      await new Promise((r) => setTimeout(r, 0));
+    }
+
+    // Whatever happened, do not leave a spinning loop behind.
+    finished = true;
+    await driving.catch(() => {});
+    for (let i = 0; i < 5 && outcome === undefined; i++) {
+      await Promise.resolve();
+    }
+
+    assert(
+      outcome !== undefined && "err" in outcome,
+      "the driver never settled: the deadlock probe kept re-arming because " +
+        "`readyCandidates()` reports a thread `tick` refuses to run",
+    );
+    const msg = String(
+      (outcome.err as { message?: string })?.message ?? outcome.err,
+    );
+    assertEq(msg.includes("deadlock detected"), true);
+    assertEq(msg.includes("export 'abandoned'"), true);
+  },
+});