runtime: settle async results before suspended producer exit

runtime/src/exec/boundary.ts

@@ -206,7 +206,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;
@@ -1283,15 +1283,18 @@
 
     let outcome: DriveExit | Promise<DriveExit>;
     try {
-      // Driver completion is not thread exhaustion. Require a captured result,
-      // no awaiting wasm call of this task, and no entry hop anywhere in the
-      // store. Other tasks' genuine SuspensionPoint parks may remain background
-      // work, but any engine hop this driver started still needs servicing.
-      // Callback tasks can retain waiting threads after task.return; awaiting
-      // those threads' final exit would prevent long-lived producers returning.
       const midWasmCall = () => task.threads.some((t) => store.awaiting.has(t));
       const hopParked = () => entryHopThreads(store).length > 0;
-      const driveDone = () => resolvedSeen && !midWasmCall() && !hopParked();
+      // 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,
@@ -1610,7 +1613,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;
@@ -1639,7 +1642,7 @@
   // Callback ABI waits between invocations without JSPI, but callbacks in
   // JSPI mode need the same promising wrapper as the initial core entry.
   const callback = enterWasm(
     require(opts.callback, `${name} callback`)!,
     input.mode,
   );
   const [packed] = normalizeCoreValues(

runtime/tests/jspi/fixtures/task-return-settlement.wat

@@ -0,0 +1,105 @@
+;; An async export publishes its result, then remains in the same activation
+;; while waitable-set.wait JSPI-suspends on a pending async host import.
+(component
+  (import "before-gate" (func $before-gate async))
+  (import "gate" (func $gate async))
+  (import "hop" (func $hop))
+  (import "continued" (func $continued))
+  (import "probe-entered" (func $probe-entered))
+
+  (core module $Mem (memory (export "mem") 1))
+  (core instance $mem (instantiate $Mem))
+
+  (canon task.return (result u32) (core func $task.return))
+  (canon lower (func $before-gate) async (core func $before-gate.async))
+  (canon lower (func $gate) async (core func $gate.async))
+  (canon lower (func $hop) (core func $hop.sync))
+  (canon lower (func $continued) (core func $continued.sync))
+  (canon lower (func $probe-entered) (core func $probe-entered.sync))
+  (canon waitable-set.new (core func $ws.new))
+  (canon waitable.join (core func $w.join))
+  (canon waitable-set.wait (memory (core memory $mem "mem"))
+    (core func $ws.wait))
+  (canon subtask.drop (core func $subtask.drop))
+
+  (core module $Core
+    (import "" "mem" (memory 1))
+    (import "" "task.return" (func $task.return (param i32)))
+    (import "" "before-gate" (func $before-gate (result i32)))
+    (import "" "gate" (func $gate (result i32)))
+    (import "" "hop" (func $hop))
+    (import "" "continued" (func $continued))
+    (import "" "probe-entered" (func $probe-entered))
+    (import "" "waitable-set.new" (func $ws.new (result i32)))
+    (import "" "waitable.join" (func $w.join (param i32 i32)))
+    (import "" "waitable-set.wait"
+      (func $ws.wait (param i32 i32) (result i32)))
+    (import "" "subtask.drop" (func $subtask.drop (param i32)))
+
+    (func (export "run") (result i32)
+      (local $started i32) (local $sub i32) (local $ws i32)
+      ;; First suspend before the result exists. Reaching task.return after
+      ;; this resumption exercises settlement beyond the initial activation.
+      (local.set $started (call $before-gate))
+      (if (i32.ne
+            (i32.and (local.get $started) (i32.const 15))
+            (i32.const 1))
+        (then unreachable))
+      (local.set $sub (i32.shr_u (local.get $started) (i32.const 4)))
+      (local.set $ws (call $ws.new))
+      (call $w.join (local.get $sub) (local.get $ws))
+      (drop (call $ws.wait (local.get $ws) (i32.const 0)))
+      (call $subtask.drop (local.get $sub))
+
+      (call $task.return (i32.const 42))
+      ;; A nonblocking Suspending import adds an engine entry hop between
+      ;; task.return and the genuine second suspension.
+      (call $hop)
+      (local.set $started (call $gate))
+      (if (i32.ne
+            (i32.and (local.get $started) (i32.const 15))
+            (i32.const 1))
+        (then unreachable))
+      (local.set $sub (i32.shr_u (local.get $started) (i32.const 4)))
+      (local.set $ws (call $ws.new))
+      (call $w.join (local.get $sub) (local.get $ws))
+      (if (i32.ne
+            (call $ws.wait (local.get $ws) (i32.const 0))
+            (i32.const 1))
+        (then unreachable))
+      (if (i32.ne (i32.load (i32.const 0)) (local.get $sub))
+        (then unreachable))
+      (if (i32.ne (i32.load (i32.const 4)) (i32.const 2))
+        (then unreachable))
+      (call $subtask.drop (local.get $sub))
+      (call $continued)
+      ;; The result is already delivered. This later producer failure must be
+      ;; recorded for subsequent entry, not turn into an unhandled rejection.
+      unreachable)
+
+    (func (export "run-cb") (param i32 i32 i32) (result i32)
+      unreachable)
+
+    (func (export "probe") (result i32)
+      (call $probe-entered)
+      (i32.const 7)))
+
+  (core instance $core (instantiate $Core (with "" (instance
+    (export "mem" (memory $mem "mem"))
+    (export "task.return" (func $task.return))
+    (export "before-gate" (func $before-gate.async))
+    (export "gate" (func $gate.async))
+    (export "hop" (func $hop.sync))
+    (export "continued" (func $continued.sync))
+    (export "probe-entered" (func $probe-entered.sync))
+    (export "waitable-set.new" (func $ws.new))
+    (export "waitable.join" (func $w.join))
+    (export "waitable-set.wait" (func $ws.wait))
+    (export "subtask.drop" (func $subtask.drop))))))
+
+  (func (export "run") async (result u32)
+    (canon lift (core func $core "run") async
+      (callback (core func $core "run-cb"))))
+  (func (export "probe") (result u32)
+    (canon lift (core func $core "probe")))
+)

runtime/tests/jspi/task_return_settlement_test.ts

@@ -0,0 +1,202 @@
+import { instantiate } from "../../src/embedder/mod.ts";
+import { Translator } from "../../src/shim/mod.ts";
+import { isTrap, suspending } from "@polyengine/protocol";
+import { assert, assertEquals, assertRejects } from "./asserts.ts";
+
+const root = new URL("../../../", import.meta.url);
+
+async function readIfPresent(rel: string): Promise<Uint8Array | null> {
+  try {
+    return await Deno.readFile(new URL(rel, root));
+  } catch {
+    return null;
+  }
+}
+
+const shimWasm = await readIfPresent(
+  "target/wasm32-unknown-unknown/release/translator_shim.wasm",
+);
+if (shimWasm === null) {
+  console.warn(
+    "SKIP task.return settlement: missing translator_shim.wasm " +
+      "(cargo build -p translator-shim --release --target wasm32-unknown-unknown)",
+  );
+}
+const componentBytes = await Deno.readFile(
+  new URL("./fixtures/task-return-settlement.wasm", import.meta.url),
+);
+
+Deno.test({
+  name:
+    "#323: task.return settles before a producer's genuine JSPI suspension ends",
+  ignore: shimWasm === null,
+  fn: async () => {
+    const gate = Promise.withResolvers<void>();
+    const entered = Promise.withResolvers<void>();
+    const continued = Promise.withResolvers<void>();
+    const translator = await Translator.create(shimWasm!);
+    const translated = translator.translate(componentBytes);
+    let probeEntered = false;
+    const component = await instantiate({
+      componentBytes,
+      ...translated,
+    }, {
+      "before-gate": () => Promise.resolve(),
+      hop: suspending(() => undefined),
+      gate: () => {
+        entered.resolve();
+        return gate.promise;
+      },
+      continued: () => continued.resolve(),
+      "probe-entered": () => void (probeEntered = true),
+    });
+
+    const result = component.exports.run() as Promise<number>;
+    try {
+      await entered.promise;
+      const early = await Promise.race([
+        result.then((value) => ({ state: "resolved", value })),
+        new Promise<{ state: "pending" }>((resolve) =>
+          setTimeout(() => resolve({ state: "pending" }), 50)
+        ),
+      ]);
+      assertEquals(
+        JSON.stringify(early),
+        JSON.stringify({ state: "resolved", value: 42 }),
+        "task.return result was withheld behind the producer gate",
+      );
+    } finally {
+      // Never strand the real JSPI activation when an assertion fails.
+      gate.resolve();
+    }
+
+    // Observe work after the suspension independently of the settled export.
+    await continued.promise;
+    assertEquals(await result, 42);
+
+    // The producer traps after continuation. The export result stays delivered,
+    // while the failure is retained and surfaced by subsequent entry.
+    const later = await assertRejects(
+      () => component.exports.probe() as Promise<unknown>,
+      "post-delivery producer failure must reach a later entry",
+    );
+    assertEquals(
+      probeEntered,
+      false,
+      "retained poison must precede probe entry",
+    );
+    assert(
+      isTrap(later) && String(later).includes("unreachable"),
+      `expected the original branded producer trap, got ${later}`,
+    );
+  },
+});
+
+Deno.test({
+  name: "#323: a late host rejection remains observable after result delivery",
+  ignore: shimWasm === null,
+  fn: async () => {
+    const gate = Promise.withResolvers<void>();
+    const entered = Promise.withResolvers<void>();
+    const boom = new Error("late gate rejection");
+    let probeEntered = false;
+    const translator = await Translator.create(shimWasm!);
+    const component = await instantiate({
+      componentBytes,
+      ...translator.translate(componentBytes),
+    }, {
+      "before-gate": () => Promise.resolve(),
+      hop: suspending(() => undefined),
+      gate: () => {
+        entered.resolve();
+        return gate.promise;
+      },
+      continued: () => {
+        throw new Error("continued after rejected gate");
+      },
+      "probe-entered": () => void (probeEntered = true),
+    });
+
+    const result = component.exports.run() as Promise<number>;
+    try {
+      await entered.promise;
+      const early = await Promise.race([
+        result.then((value) => ({ state: "resolved", value })),
+        new Promise<{ state: "pending" }>((resolve) =>
+          setTimeout(() => resolve({ state: "pending" }), 50)
+        ),
+      ]);
+      assertEquals(
+        JSON.stringify(early),
+        JSON.stringify({ state: "resolved", value: 42 }),
+      );
+    } finally {
+      gate.reject(boom);
+    }
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    const later = await assertRejects(
+      () => component.exports.probe() as Promise<unknown>,
+      "late host rejection must reach a later entry",
+    );
+    assert(
+      isTrap(later) && String(later).includes(boom.message),
+      `expected the branded host rejection with its original cause, got ${later}`,
+    );
+    assertEquals(
+      probeEntered,
+      true,
+      "the probe entered before the resumed producer's rejection surfaced",
+    );
+  },
+});
+
+Deno.test({
+  name: "#323: task.return after one suspension settles at the next suspension",
+  ignore: shimWasm === null,
+  fn: async () => {
+    const before = Promise.withResolvers<void>();
+    const after = Promise.withResolvers<void>();
+    const enteredBefore = Promise.withResolvers<void>();
+    const enteredAfter = Promise.withResolvers<void>();
+    const continued = Promise.withResolvers<void>();
+    const translator = await Translator.create(shimWasm!);
+    const component = await instantiate({
+      componentBytes,
+      ...translator.translate(componentBytes),
+    }, {
+      "before-gate": () => {
+        enteredBefore.resolve();
+        return before.promise;
+      },
+      hop: suspending(() => undefined),
+      gate: () => {
+        enteredAfter.resolve();
+        return after.promise;
+      },
+      continued: () => continued.resolve(),
+      "probe-entered": () => {},
+    });
+
+    const result = component.exports.run() as Promise<number>;
+    try {
+      await enteredBefore.promise;
+      before.resolve();
+      await enteredAfter.promise;
+      const early = await Promise.race([
+        result.then((value) => ({ state: "resolved", value })),
+        new Promise<{ state: "pending" }>((resolve) =>
+          setTimeout(() => resolve({ state: "pending" }), 50)
+        ),
+      ]);
+      assertEquals(
+        JSON.stringify(early),
+        JSON.stringify({ state: "resolved", value: 42 }),
+      );
+    } finally {
+      before.resolve();
+      after.resolve();
+    }
+    await continued.promise;
+  },
+});