diff --git a/docs/architecture.md b/docs/architecture.md index ce4fee9..12e1cde 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,6 +47,17 @@ only to CM-3. A schedule-dependent assertion cannot invoke the exception: if two conforming schedulers can answer differently, the assertion pins a policy rather than semantics. Wasmtime behavior alone is insufficient. +**Pending upstream stream/future assumption.** Until +[component-model PR #719](https://github.com/WebAssembly/component-model/pull/719) +is adjudicated, this checkout adopts only its drop-delivery rule at head +`35e9769957627c2bee5cd445b998b08b3c652c86`; the submodule remains pinned at +`7c676115e93cd7d54c1732d95c54c6a3de7c5ae0`. A stream event tests for peer +drop when the event is consumed, upgrades the result to `DROPPED`, preserves +the accumulated element count, and retires the end. A future `COMPLETED` stays +`COMPLETED` because its payload transferred; a pending future `CANCELLED` +upgrades to `DROPPED` with zero progress if the peer is gone. Already-delivered +events are not rewritten. The PR's broader paired-end refactor is not adopted. + ## 2. Non-goals - **WASI in the runtime core.** Providers live in the separate @@ -321,6 +332,8 @@ Named differences from the reference or other hosts: This is the sole §1 corpus/reference exception, [CM-3](../upstream-component-model-repo-findings.md#cm-3-cancel_copy-returns-a-stale-completed-where-wasmtime-reports-cancelled). See `takeCancelEvent` in `runtime/src/intrinsics/stream_builtins.ts`. + This cancellation-time `COMPLETED`→`CANCELLED` remap is distinct from the + PR #719 consumption-time peer-drop upgrade above; peer drop takes precedence. ## 7. Canonical ABI decisions diff --git a/docs/references.md b/docs/references.md index 0710295..577b545 100644 --- a/docs/references.md +++ b/docs/references.md @@ -24,6 +24,11 @@ reference, and WAST corpus used by this checkout. Its principal sources are: Current upstream sources are at [WebAssembly/component-model](https://github.com/WebAssembly/component-model). +The narrow provisional semantics adopted from +[PR #719](https://github.com/WebAssembly/component-model/pull/719) are fixed to +head `35e9769957627c2bee5cd445b998b08b3c652c86` and recorded in +[architecture §1](architecture.md#1-goals); they do not move the submodule pin +or adopt that PR's structural refactor. The [Component Model book](https://component-model.bytecodealliance.org/) is introductory documentation, not the semantic tie-breaker. Local discrepancies belong in diff --git a/harness/browser/expectations/chromium.ts b/harness/browser/expectations/chromium.ts index aadc40f..ace3ad9 100644 --- a/harness/browser/expectations/chromium.ts +++ b/harness/browser/expectations/chromium.ts @@ -17,9 +17,9 @@ export const chromium: LaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/browser/expectations/firefox.ts b/harness/browser/expectations/firefox.ts index 0f4fc6c..d94052d 100644 --- a/harness/browser/expectations/firefox.ts +++ b/harness/browser/expectations/firefox.ts @@ -21,9 +21,9 @@ export const firefox: LaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/bun-pinned.ts b/harness/shell/expectations/bun-pinned.ts index a5737e7..8560a1e 100644 --- a/harness/shell/expectations/bun-pinned.ts +++ b/harness/shell/expectations/bun-pinned.ts @@ -45,9 +45,9 @@ export const bunPinned: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/jsc-pinned.ts b/harness/shell/expectations/jsc-pinned.ts index 1debbc7..dd1dd3d 100644 --- a/harness/shell/expectations/jsc-pinned.ts +++ b/harness/shell/expectations/jsc-pinned.ts @@ -21,9 +21,9 @@ export const jscPinned: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/jsc-trunk.ts b/harness/shell/expectations/jsc-trunk.ts index 5f14d66..17e0d22 100644 --- a/harness/shell/expectations/jsc-trunk.ts +++ b/harness/shell/expectations/jsc-trunk.ts @@ -55,9 +55,9 @@ export const jscTrunk: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/node-pinned.ts b/harness/shell/expectations/node-pinned.ts index 97cdf88..48b788d 100644 --- a/harness/shell/expectations/node-pinned.ts +++ b/harness/shell/expectations/node-pinned.ts @@ -35,9 +35,9 @@ export const nodePinned: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/sm-nightly.ts b/harness/shell/expectations/sm-nightly.ts index bb93295..35e4bd9 100644 --- a/harness/shell/expectations/sm-nightly.ts +++ b/harness/shell/expectations/sm-nightly.ts @@ -44,9 +44,9 @@ export const smNightly: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/shell/expectations/sm-pinned.ts b/harness/shell/expectations/sm-pinned.ts index 0e89624..fbe16b1 100644 --- a/harness/shell/expectations/sm-pinned.ts +++ b/harness/shell/expectations/sm-pinned.ts @@ -24,9 +24,9 @@ export const smPinned: ShellLaneExpectation = { totals: { commands: 1511, executed: 1411, - passed: 1285, + passed: 1286, failed: 0, - xfail: 126, + xfail: 125, pendingRuntime: 95, pendingCapability: 0, unsupportedDirective: 5, diff --git a/harness/src/wasmtime-expectations.ts b/harness/src/wasmtime-expectations.ts index 44e03e6..eb29855 100644 --- a/harness/src/wasmtime-expectations.ts +++ b/harness/src/wasmtime-expectations.ts @@ -78,15 +78,6 @@ export const WASMTIME_EXPECTATION_GROUPS: readonly WasmtimeExpectationGroup[] = status: "failed", }], }, - { - file: "async/context-in-compositions.json", - rows: [{ - lines: [435, 436, 437], - cause: - "Error: expected return, got trap: cannot enter component instance 2 — instance poisoned by: Trap: guest trapped: unreachable", - status: "failed", - }], - }, { file: "async/context-in-resource-drop.json", rows: [{ @@ -103,15 +94,6 @@ export const WASMTIME_EXPECTATION_GROUPS: readonly WasmtimeExpectationGroup[] = status: "failed", }], }, - { - file: "async/stream-big-read-and-writes.json", - rows: [{ - lines: [43], - cause: - 'Error: expected trap "stream read/write count too large", got "cannot enter component instance 0 — instance poisoned by: RangeError: Invalid array length"', - status: "failed", - }], - }, { file: "async/task-deletion.json", rows: [{ @@ -437,90 +419,6 @@ export const WASMTIME_EXPECTATION_GROUPS: readonly WasmtimeExpectationGroup[] = { class: "runtime-semantics", files: [ - { - file: "async/context-in-compositions.json", - rows: [{ - lines: [155, 434], - cause: - "Error: expected return, got trap: guest trapped: unreachable", - status: "failed", - }], - }, - { - file: "async/error-context.json", - rows: [{ - lines: [86], - cause: "AssertionError: store out of bounds", - status: "failed", - }], - }, - { - file: "async/futures.json", - rows: [{ - lines: [54, 64], - cause: - "AssertionError: suspension mode jspi wrapped imports without wrapping any entry (entries=false, imports=true) — a Suspending import reached from a non-promising activation traps unconditionally (jspi pin (c))", - status: "failed", - }], - }, - { - file: "async/stackful.json", - rows: [{ - lines: [110, 132], - cause: - "AssertionError: suspension mode jspi wrapped imports without wrapping any entry (entries=false, imports=true) — a Suspending import reached from a non-promising activation traps unconditionally (jspi pin (c))", - status: "failed", - }], - }, - { - file: "async/stream-big-read-and-writes.json", - rows: [{ - lines: [42], - cause: "RangeError: Invalid array length", - status: "failed", - }], - }, - { - file: "async/streams.json", - rows: [{ - lines: [73, 83], - cause: - "AssertionError: suspension mode jspi wrapped imports without wrapping any entry (entries=false, imports=true) — a Suspending import reached from a non-promising activation traps unconditionally (jspi pin (c))", - status: "failed", - }], - }, - { - file: "async/sync-call-context-slots.json", - rows: [{ - lines: [76, 152], - cause: - "Error: expected return, got trap: guest trapped: unreachable", - status: "failed", - }], - }, - { - file: "async/sync-call-context.json", - rows: [{ - lines: [61, 126, 207, 298, 389, 447, 544], - cause: - "Error: expected return, got trap: guest trapped: unreachable", - status: "failed", - }], - }, - { - file: "async/task-builtins.json", - rows: [{ - lines: [34, 56, 74], - cause: - "AssertionError: suspension mode jspi wrapped imports without wrapping any entry (entries=false, imports=true) — a Suspending import reached from a non-promising activation traps unconditionally (jspi pin (c))", - status: "failed", - }, { - lines: [201, 466, 723], - cause: - "Error: expected return, got trap: guest trapped: unreachable", - status: "failed", - }], - }, { file: "import.json", rows: [{ diff --git a/harness/src/xfail.ts b/harness/src/xfail.ts index 7cff9f6..c0c3d86 100644 --- a/harness/src/xfail.ts +++ b/harness/src/xfail.ts @@ -661,32 +661,6 @@ export const XFAIL: XfailEntry[] = [ line: 369, reason: "same cascade as line 355, see that entry", }, - // --- async/sync-streams.json: test/async/sync-streams.wast expects three - // values polyengine does not produce (STARTING vs STARTED at the - // sync-lowered `set` call, and a COMPLETED<->DROPPED completion-code swap - // on the paired stream.read/write) under CM#705's blocking semantics, so - // the file's single all-in-one assert_return hits a guest `unreachable`. - // Classed `cm705-sync-sched` - // (https://github.com/polymorph-components/polyengine/issues/249). - // - // For the rest of the file polyengine implements wasmtime's model (#43): - // the async-lowered call's initial status is decided only after the callee - // instance's runnable work has been drained to quiescence — by which time - // the producer has exited and the next task reports STARTED. Adjudicated - // 2026-08-10 (issue #43): the test's hard STARTED assertion is - // schedule-dependent — an upstream test defect overfitting wasmtime's - // deferred-entry policy (pristine definitions.py answers STARTING) — and - // polyengine's drain policy satisfies it as written under any seed. --- - { - file: "async/sync-streams.json", - line: 208, - reason: - "expected return, got trap: guest trapped: unreachable — this file's " + - "expected STARTING/STARTED and COMPLETED/DROPPED codes track CM#705's " + - "blocking semantics, which polyengine's sync scheduling does not yet " + - "produce, so the guest's own assertion traps; cm705-sync-sched, " + - "https://github.com/polymorph-components/polyengine/issues/249", - }, // --- async/trap-if-block-and-sync.json: (history: at the prior pin the // whole file was blocked by the now-exited wasmparser/wast pin-drift // class — see the EXIT note at the top of this file, $Tester's canonical diff --git a/runtime/src/exec/boundary.ts b/runtime/src/exec/boundary.ts index 76aa85e..bf5ad7d 100644 --- a/runtime/src/exec/boundary.ts +++ b/runtime/src/exec/boundary.ts @@ -58,6 +58,7 @@ import { Thread, WaitableSet, withActivation, + withSynchronousActivation, } from "../task/mod.ts"; import { currentTask } from "../task/scheduler.ts"; import type { @@ -219,7 +220,13 @@ export function cabiOptions(opts: ResolvedOptions): CanonicalOptions { 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]); + // definitions.py LiftLowerContext.reallocate creates and invokes a fresh + // sync canon_lift (lines 642-658), hence fresh task/thread context. Keep + // post-return on its originating task; only realloc takes this boundary. + const p = withSynchronousActivation( + opts.instance, + () => callCore(realloc, [o, os, a, n]), + ); trapIf(p.length !== 1 || typeof p[0] !== "number", "realloc result"); return (p[0] as number) >>> 0; }, diff --git a/runtime/src/exec/executor.ts b/runtime/src/exec/executor.ts index c37d01b..87856ce 100644 --- a/runtime/src/exec/executor.ts +++ b/runtime/src/exec/executor.ts @@ -291,13 +291,15 @@ class Executor { * identity, so task.return and its lifted export must share this wrapper. */ readonly liveMemories = new Map(); - /** Set by the entry/import wrapping sites; checked in `finish`. */ - wrappedEntries = false; + /** Counted by the entry wrapping site; checked in `finish`. */ + entriesConstructed = 0; + wrappedEntries = 0; wrappedImports = false; /** Record that an entry / import wrapping site ran under the current mode. */ noteEntry(): SuspensionMode { - if (this.suspensionMode === "jspi") this.wrappedEntries = true; + this.entriesConstructed++; + if (this.suspensionMode === "jspi") this.wrappedEntries++; return this.suspensionMode; } @@ -530,6 +532,23 @@ class Executor { ); } const importObject: WebAssembly.Imports = {}; + const adapter = this.wire.modules[init.module]; + const adapterImports = + init.instance === null && adapter?.kind === "adapter" + ? adapter.intrinsics + : null; + if ( + adapterImports !== null && adapterImports.length !== declared.length + ) { + throw new PlanError( + `adapter module ${init.module}: ${declared.length} compiled imports but ` + + `${adapterImports.length} manifest entries`, + ); + } + const postReturnWrappers = new Map< + CoreFn, + { owner: ComponentInstanceState; wrapper: CoreFn } + >(); // Collect suspendability from this core module's imports. FACT // needs callee-specific evidence, not the whole plan's mode choice. this.sawBlockingImport = false; @@ -547,7 +566,64 @@ class Executor { const seenAt = new Map(); declared.forEach((imp, i) => { const before = this.sawBlockingImport; - const value = this.importValue(init.args[i]); + const def = init.args[i]; + const manifest = adapterImports?.[i]; + if ( + adapterImports !== null && + (manifest === undefined || + manifest.module !== imp.module || manifest.name !== imp.name || + !sameCoreDef(manifest.def, def)) + ) { + throw new PlanError( + `adapter module ${init.module}: import ${i} does not match its manifest`, + ); + } + let value = this.importValue(def); + if (adapterImports !== null && imp.module === "post_return") { + // Pinned FACT assigns this exact namespace only to the lifted + // function's post-return import (wasmtime-environ fact.rs:322-327). + // Wrap this adapter edge, never the extracted function globally: + // the same core function may also be called as an ordinary export. + if ( + manifest?.module !== "post_return" || + typeof value !== "function" + ) { + throw new PlanError( + `adapter module ${init.module}: invalid post_return import ${i}`, + ); + } + const owner = this.coreDefOwner(def); + if (owner === null) { + throw new PlanError( + `adapter module ${init.module}: post_return import ${i} has no ` + + `declaring component instance`, + ); + } + const original = value as CoreFn; + const prior = postReturnWrappers.get(original); + if (prior !== undefined && prior.owner !== owner) { + throw new PlanError( + `adapter module ${init.module}: one post_return function has ` + + `multiple declaring component instances`, + ); + } + let wrapped = prior?.wrapper; + if (wrapped === undefined) { + wrapped = (...args: unknown[]) => { + if (!owner.mayLeave) { + throw new Trap("cannot leave component instance"); + } + owner.mayLeave = false; + // Deliberately no finally: definitions.py canon_lift leaves + // may_leave false when post-return traps (lines 2144-2149). + const result = original(...args); + owner.mayLeave = true; + return result; + }; + postReturnWrappers.set(original, { owner, wrapper: wrapped }); + } + value = wrapped; + } if (!before && this.sawBlockingImport && SUSPENDABLE_TRACE) { console.error( `[suspendable] module ${init.module}: import ` + @@ -731,9 +807,10 @@ class Executor { // Structural check of jspi/bridge.ts's invariant, run once both wrapping // sites have had their chance: entries are wrapped while building exports // (just above) and imports while running `instantiate-module`. Neither - // flag can be set by accident — only the wrapping helpers set them. + // values can be set by accident — only the wrapping helpers set them. assertModeConsistent( this.suspensionMode, + this.entriesConstructed, this.wrappedEntries, this.wrappedImports, ); @@ -982,6 +1059,32 @@ class Executor { } } + /** Component instance statically declaring a callable CoreDef, if known. */ + coreDefOwner(def: WireCoreDef): ComponentInstanceState | null { + if (def.kind === "trampoline") { + const decl = this.wire.trampolines[def.index] as + | { instance?: unknown } + | undefined; + return typeof decl?.instance === "number" + ? this.componentInstance(decl.instance) + : null; + } + if (def.kind !== "export") return null; + + // RuntimeInstanceIndex is the ordinal of instantiate-module initializers. + // Derive its owner from that existing source of truth rather than keeping a + // second ownership map (contracts/plan-format.md:70-80). + let runtimeInstance = 0; + for (const init of this.wire.initializers) { + if (init.op !== "instantiate-module") continue; + if (runtimeInstance++ !== def.instance) continue; + return init.instance === null + ? null + : this.componentInstance(init.instance); + } + return null; + } + resolveCoreExport(ref: WireCoreExport): Importable { const instance = this.instances[ref.instance]; if (instance === undefined) { @@ -1051,6 +1154,15 @@ class Executor { } return fn; }, + postReturn: (i) => { + const fn = this.postReturns[i]; + if (fn === undefined) { + throw new PlanError( + `post-return ${i} accessed before its extract-post-return initializer ran`, + ); + } + return fn; + }, memoryToken: (i) => this.liveMemory(i), streamElem: (i) => { if (i >= this.loaded.streamElems.length) { @@ -1257,3 +1369,19 @@ function describe(v: unknown): string { if (typeof v === "object") return `a ${v.constructor?.name ?? "object"}`; return `a ${typeof v}`; } + +/** Exact equality for the small, closed CoreDef wire union. */ +function sameCoreDef(a: WireCoreDef, b: WireCoreDef): boolean { + if (a.kind !== b.kind) return false; + switch (a.kind) { + case "export": + return b.kind === "export" && a.instance === b.instance && + a.item.name === b.item.name && a.item.space === b.item.space; + case "instance-flags": + return b.kind === "instance-flags" && a.instance === b.instance; + case "trampoline": + return b.kind === "trampoline" && a.index === b.index; + case "unsafe-intrinsic": + return b.kind === "unsafe-intrinsic" && a.intrinsic === b.intrinsic; + } +} diff --git a/runtime/src/intrinsics/async_builtins.ts b/runtime/src/intrinsics/async_builtins.ts index 8349ee6..f55ef0f 100644 --- a/runtime/src/intrinsics/async_builtins.ts +++ b/runtime/src/intrinsics/async_builtins.ts @@ -70,6 +70,7 @@ export const BLOCKED = 0xffff_ffff; export function createTaskReturn( decl: { results: number; resultType: number | null; options: number }, ctx: AsyncTrampolineContext, + declaredInst?: ComponentInstanceState, ): CoreFn { const opts = ctx.options(decl.options); // `resultType` is the interned `plan.types` entry; `results` is the @@ -80,6 +81,10 @@ export function createTaskReturn( ? [] : ctx.resultTypes(decl.resultType); return (...flatArgs: CoreValue[]) => { + trapIf( + declaredInst !== undefined && !declaredInst.mayLeave, + "task.return: cannot leave component instance (may_leave violation)", + ); const task = currentTask() as Task; trapIf( !task.inst.mayLeave, @@ -134,8 +139,14 @@ export function createTaskReturn( } /** definitions.py `canon_task_cancel`. */ -export function createTaskCancel(): CoreFn { +export function createTaskCancel( + declaredInst?: ComponentInstanceState, +): CoreFn { return () => { + trapIf( + declaredInst !== undefined && !declaredInst.mayLeave, + "task.cancel: cannot leave component instance (may_leave violation)", + ); const task = currentTask() as Task; trapIf( !task.inst.mayLeave, @@ -504,9 +515,14 @@ export function createSubtaskCancel( export function createThreadYield( decl: { cancellable?: boolean }, mode: SuspensionMode = "plain", + declaredInst?: ComponentInstanceState, ): CoreFn { const cancellable = decl.cancellable === true; return () => { + trapIf( + declaredInst !== undefined && !declaredInst.mayLeave, + "thread.yield: cannot leave component instance", + ); const thread = currentThread(); trapIf( !thread.task.inst.mayLeave, diff --git a/runtime/src/intrinsics/fact_calls.ts b/runtime/src/intrinsics/fact_calls.ts index a7426df..5fada06 100644 --- a/runtime/src/intrinsics/fact_calls.ts +++ b/runtime/src/intrinsics/fact_calls.ts @@ -147,6 +147,8 @@ export interface FactCallContext { resultTypesForTuple(tupleIndex: number): ValType[] | null; /** `RuntimeCallbackIndex` -> the callee's callback core function. */ callback(index: number): CoreFn; + /** `RuntimePostReturnIndex` -> the callee's post-return core function. */ + postReturn(index: number): CoreFn; /** `RuntimeMemoryIndex` -> the memory `task.return` must match, if any. */ memoryToken(index: number): unknown; stats: ExecutionStats; @@ -661,7 +663,7 @@ export function createAsyncStartCall( callback, postReturn: decl.postReturn === null ? null - : ctx.callback(decl.postReturn), + : ctx.postReturn(decl.postReturn), ctx, // `compile_async_to_async_adapter` sets START_FLAG_ASYNC_CALLEE; // `compile_async_to_sync_adapter` passes 0. @@ -769,11 +771,6 @@ export function createAsyncStartCall( // so any held gate belongs to the currently executing activation. if (ctx.suspensionMode === "jspi") { const store = prepared.callerInst.store; - const calleeInst = prepared.calleeInst; - // The caller's task: excluded from the drain scan (it is the asker). - // `maybeCurrentTask` rather than `currentTask` because a host-driven - // entry can reach here with no ambient task at all. - const callerTask = maybeCurrentTask(); // STARTING + parked == parked at the entry gate: `[async-start]` runs // immediately after `enter_implicit_thread` succeeds, so any callee // that got past the gate has already left STARTING. @@ -784,7 +781,11 @@ export function createAsyncStartCall( thread.waiting(); const determinate = (): boolean => gatedAtEntry() - ? !store.hasRunnableWork(calleeInst, callerTask) + // canon_lower reports STARTING as soon as the new callee blocks on + // admission. Running an unrelated exclusive holder here can make + // the new call complete before its caller observes it, contrary to + // definitions.py:2257-2282 and task-builtins.wast:530-532. + ? true : subtask.resolved() || thread.done() || store.waiting.some((w) => w.task === task); diff --git a/runtime/src/intrinsics/mod.ts b/runtime/src/intrinsics/mod.ts index 26957bb..0521254 100644 --- a/runtime/src/intrinsics/mod.ts +++ b/runtime/src/intrinsics/mod.ts @@ -14,7 +14,17 @@ import { trapIf } from "../cabi/trap.ts"; import { assert_ } from "../cabi/trap.ts"; import type { ResourceTableInfo } from "../cabi/types.ts"; import type { ComponentInstanceState } from "../task/mod.ts"; -import { dbgId, entryRefusal, maybeCurrentThread } from "../task/mod.ts"; +import { + claimActivationAmbient, + currentTask, + dbgId, + entryRefusal, + maybeCurrentThread, + popLogicalActivation, + SynchronousActivation, +} from "../task/mod.ts"; +import { blockCurrentActivation, setContinuationOwner } from "../jspi/mod.ts"; +import { needsJspi } from "../task/scheduler.ts"; import type { WireTrampoline } from "../plan/format.ts"; import type { CoreFn, ExecutionStats } from "../exec/boundary.ts"; import { UnsupportedFeatureError } from "./errors.ts"; @@ -164,6 +174,8 @@ export class SyncCallScope { numBorrows = 0; readonly lenders: ResourceHandle[] = []; + constructor(readonly activation?: SynchronousActivation) {} + /** * definitions.py `Subtask.add_lender`: borrowed handles can be lent onward * too. `canon_resource_drop` checks `num_lends` for both own and borrow @@ -243,6 +255,8 @@ export interface TrampolineContext { resultTypes(index: number): import("../cabi/types.ts").ValType[]; /** `RuntimeCallbackIndex` -> the extracted callback core function. */ callback(index: number): CoreFn; + /** `RuntimePostReturnIndex` -> the extracted post-return core function. */ + postReturn(index: number): CoreFn; /** `RuntimeMemoryIndex` -> an identity token for `task.return` checks. */ memoryToken(index: number): unknown; /** The single in-flight FACT preparation (see `PreparedCall`). */ @@ -422,10 +436,71 @@ function createTrampolineBody( // the prepare/start protocol creates any separate callee task. void async_; ctx.stats.enterSyncCalls++; + // Capture before a possible admission park. The store later calls + // `produce` without the caller's synchronous bracket, but the nested + // activation remains a logical child of this canonical caller. + const caller = maybeCurrentThread(); // Normal return must match enter/exit on the same activation's stack; // trap unwind releases any scopes whose exit was skipped. - const scopes = syncScopes(ctx, "enter"); - scopes.push(new SyncCallScope()); + const enter = () => { + const activation = typeof calleeInstance === "number" + ? new SynchronousActivation( + ctx.componentInstance(calleeInstance >>> 0), + async_ !== 0, + caller, + ) + : undefined; + // Resolve again after publishing the callee: its activation owns the + // bracket, including resource transfers and nested calls. + syncScopes(ctx, "enter-callee").push(new SyncCallScope(activation)); + return activation; + }; + if (typeof calleeInstance !== "number" || async_ === 0) return enter(); + if (caller === undefined) { + // Core start functions can legally make an unblocked synchronous + // guest call without an enclosing canonical task. There is no stack + // to suspend if admission is blocked, so retain the existing + // synchronous-only limitation rather than inventing a root task. + const callee = ctx.componentInstance(calleeInstance >>> 0); + if (callee.backpressure === 0 && callee.exclusiveThread === null) { + return enter(); + } + needsJspi("enter-sync-call from core start cannot block"); + } + const callee = ctx.componentInstance(calleeInstance >>> 0); + const blocked = () => + callee.backpressure > 0 || callee.exclusiveThread !== null; + if (!blocked() && callee.numWaitingToEnter === 0) return enter(); + if (ctx.suspensionMode !== "jspi") { + needsJspi("enter-sync-call blocked by backpressure or exclusivity"); + } + callee.numWaitingToEnter += 1; + const admission = blockCurrentActivation({ + store: callee.store, + task: currentTask(), + readyFunc: () => !blocked(), + // CONTRACT: this is the already-running caller's synchronous wait, + // not a new callee task's `enter_implicit_thread` wait + // (definitions.py:458-465). It cannot consume caller cancellation. + cancellable: false, + produce: () => { + const activation = enter(); + if (activation !== undefined) { + // Deferred admission runs from the store drain, outside the + // captured caller bracket. Convert the child's temporary stack + // publication into the claim consumed by the resumed wasm + // continuation; leaving it on threadStack would look live and + // prevent autonomous service of a later callee yield. + popLogicalActivation(activation.thread); + claimActivationAmbient(activation.thread); + setContinuationOwner(admission, activation.thread); + } + }, + onSettled: () => { + callee.numWaitingToEnter -= 1; + }, + }); + return admission; }; case "exit-sync-call": return (..._args: unknown[]) => { @@ -443,11 +518,21 @@ function createTrampolineBody( ); // definitions.py `Task.return_`: the callee may not return while it // still holds borrow handles. - trapIf( - scope!.numBorrows > 0, - "borrow handles still remain at the end of the call", - ); - scope!.releaseLenders(); + try { + trapIf( + scope!.numBorrows > 0, + "borrow handles still remain at the end of the call", + ); + scope!.releaseLenders(); + scope!.activation?.finish(); + } catch (e) { + scope!.releaseLenders(); + scope!.activation?.abort(); + if (scope!.activation !== undefined) { + popLogicalActivation(scope!.activation.thread); + } + throw e; + } }; // Guest-side resource built-ins; reps and handle indices are i32. @@ -514,9 +599,10 @@ function createTrampolineBody( options: number; }, ctx as AsyncTrampolineContext, + declaredInstance(decl, ctx), ); case "task-cancel": - return createTaskCancel(); + return createTaskCancel(declaredInstance(decl, ctx)); case "backpressure-inc": return createBackpressureInc(declaredInstance(decl, ctx)); case "backpressure-dec": @@ -552,6 +638,7 @@ function createTrampolineBody( return createThreadYield( decl as unknown as { cancellable?: boolean }, ctx.suspensionMode, + declaredInstance(decl, ctx), ); // --- FACT cross-component calls (see ./fact_calls.ts) ----------------- diff --git a/runtime/src/intrinsics/stream_builtins.ts b/runtime/src/intrinsics/stream_builtins.ts index ec16b08..c4ca407 100644 --- a/runtime/src/intrinsics/stream_builtins.ts +++ b/runtime/src/intrinsics/stream_builtins.ts @@ -183,17 +183,25 @@ function streamCopy(input: { ): EventTuple => { reclaim(); assert_(end.copying(), "stream event on a non-copying end"); - end.state = result === CopyResult.DROPPED ? CopyState.DONE : CopyState.IDLE; + // CONTRACT: WebAssembly/component-model#719 at head 35e9769957627c2: + // stream drop is observed when the pending event is consumed, not when it + // is armed. Preserve all progress accumulated before delivery. This runs + // before takeCancelEvent's CM-3 COMPLETED -> CANCELLED remap, so peer drop + // has precedence and remains DROPPED. + const delivered = end.shared.dropped ? CopyResult.DROPPED : result; + end.state = delivered === CopyResult.DROPPED + ? CopyState.DONE + : CopyState.IDLE; assert_( buffer.progress <= BUFFER_MAX_LENGTH, "stream progress out of packing range", ); // Low four bits hold the result; the remaining bits count elements. assert_( - result >= 0 && result < 2 ** 4, + delivered >= 0 && delivered < 2 ** 4, "stream event: packed result out of 4-bit range", ); - return [eventCode, i, (result | (buffer.progress << 4)) >>> 0]; + return [eventCode, i, (delivered | (buffer.progress << 4)) >>> 0]; }; end.state = CopyState.COPYING; @@ -258,11 +266,18 @@ function futureCopy(input: { "future event/progress disagreement", ); assert_(end.copying(), "future event on a non-copying end"); + // CONTRACT: WebAssembly/component-model#719 at head 35e9769957627c2: + // transferred payload makes COMPLETED final even if the peer later drops; + // only an undelivered CANCELLED is upgraded when its peer is gone. + const delivered = result === CopyResult.CANCELLED && end.shared.dropped + ? CopyResult.DROPPED + : result; // A future is single-shot: both COMPLETED and DROPPED retire the end. - end.state = result === CopyResult.DROPPED || result === CopyResult.COMPLETED + end.state = delivered === CopyResult.DROPPED || + delivered === CopyResult.COMPLETED ? CopyState.DONE : CopyState.IDLE; - return [eventCode, i, result]; + return [eventCode, i, delivered]; }; end.state = CopyState.COPYING; @@ -522,6 +537,13 @@ export function createErrorContextDebugMessage( errorContextTrapMessage("error-context.debug-message", e), ); const cx = new LiftLowerContext(cabiOptions(opts), inst, null); + const mem = cx.opts.memory; + assert_(mem !== null, "error-context.debug-message requires a memory"); + const resultSize = 2 * mem.ptrSize(); + trapIf( + ptr % mem.ptrSize() !== 0 || ptr + resultSize > mem.length, + "invalid debug message pointer", + ); storeString(cx, (e as ErrorContext).debugMessage, ptr); }; } diff --git a/runtime/src/jspi/bridge.ts b/runtime/src/jspi/bridge.ts index 52205cb..3abaabf 100644 --- a/runtime/src/jspi/bridge.ts +++ b/runtime/src/jspi/bridge.ts @@ -47,10 +47,16 @@ import { claimActivationAmbient, dbgId, maybeCurrentThread, + physicalOwnerOf, releaseActivationAmbient, withActivation, } from "../task/mod.ts"; -import type { Cancelled, SchedulableThread, Store } from "../task/mod.ts"; +import type { + Cancelled, + CurrentThreadLike, + SchedulableThread, + Store, +} from "../task/mod.ts"; /** Which suspension discipline an instantiation runs under. */ export type SuspensionMode = "plain" | "jspi"; @@ -152,7 +158,8 @@ export function trampolineCanBlock( t: { kind: string; async?: unknown; options?: unknown }, optionsAsync: (index: number) => boolean, ): boolean { - return t.kind === "async-start-call" || t.kind === "subtask-cancel" || + return t.kind === "async-start-call" || t.kind === "enter-sync-call" || + t.kind === "subtask-cancel" || trampolineNeedsSuspension(t, optionsAsync); } @@ -247,6 +254,16 @@ function sentinelFor(owner: unknown): void { SENTINEL_TICK.then(() => claimActivationAmbient(owner)); } const SENTINEL_TICK = Promise.resolve(); +const continuationOwners = new WeakMap(); + +/** Override the logical owner when a blocking import creates its continuation + * only at resume time (currently deferred `enter-sync-call`). */ +export function setContinuationOwner( + promise: Promise, + owner: unknown, +): void { + continuationOwners.set(promise, owner); +} /** Queue attribution before the wrapped Promise settles. Engine continuation * timing need not make the sentinel and wasm chunk adjacent; instance-scoped @@ -257,11 +274,14 @@ function attributeContinuation( ): Promise { return Promise.resolve(r).then( (v) => { - sentinelFor(owner); + // A deferred enter-sync-call can replace the logical continuation while + // this Promise is pending. Preserve that leaf rather than reasserting + // the import-time caller over it. + sentinelFor(continuationOwners.get(r as object) ?? owner); return v; }, (e) => { - sentinelFor(owner); + sentinelFor(continuationOwners.get(r as object) ?? owner); throw e; }, ); @@ -277,9 +297,13 @@ export function suspendingImport unknown>( // The activation calling us — read while its bracket (or its hop claim) // is still the ambient. const owner = maybeCurrentThread() ?? null; - const invoke = () => - (fn as unknown as (...a: unknown[]) => unknown)(...args); let r: unknown; + let continuation = owner; + const invoke = () => { + const result = (fn as unknown as (...a: unknown[]) => unknown)(...args); + continuation = maybeCurrentThread() ?? owner; + return result; + }; try { // Bracket our own JS frame with the caller. Without this, a built-in // that synchronously enters ANOTHER activation's wasm (`async-start-call` @@ -289,6 +313,9 @@ export function suspendingImport unknown>( // `fact_calls.ts:820`'s determinacy wait). The nesting is a stack, and // this is the frame that owns it. r = owner === null ? invoke() : withActivation(owner, invoke); + // A FACT enter-sync-call can publish a nested canonical activation while + // this Suspending import runs. The wasm continuation after the mandatory + // hop belongs to that exact nested frame, not to the import's parent. } catch (e) { // A synchronous trap out of a built-in also unwinds the guest through // the hop, and the guest's trap-path built-ins run there. @@ -299,8 +326,8 @@ export function suspendingImport unknown>( // Plain values still return through an engine hop. Claim synchronously // for pre-hop reads and queue a sentinel before returning; other claims // may interleave before wasm resumes, as described above. - claimActivationAmbient(owner); - sentinelFor(owner); + claimActivationAmbient(continuation); + sentinelFor(continuation); return r; } if (SP_TRACE) { @@ -318,9 +345,11 @@ export function suspendingImport unknown>( * * The dangerous direction, per jspi pin (c), is a `Suspending` import * reachable from a non-`promising` activation — that traps unconditionally, - * even on the plain-value path. Entry wrapping in jspi mode is unconditional - * (every lifted export, callback, and block-capable FACT callee), so the - * structural invariant is `importsWrapped ⇒ entriesWrapped`, per mode. + * even on the plain-value path. Entry construction records both the total and + * wrapped counts, so this check proves every constructed lifted entry used the + * mode selected for the instantiation. A component with no lifted entries is + * valid: declaration-only components can instantiate core modules importing + * blocking built-ins without providing any path that calls those imports. * * Entries-without-imports is legitimate: per-declaration classification * (`trampolineNeedsSuspension`) wraps no imports in a component whose @@ -332,23 +361,25 @@ export function suspendingImport unknown>( */ export function assertModeConsistent( mode: SuspensionMode, - entriesWrapped: boolean, + entriesConstructed: number, + entriesWrapped: number, importsWrapped: boolean, ): void { - if (mode === "plain") { - assert_( - !entriesWrapped && !importsWrapped, - `plain mode with wrapped entries=${entriesWrapped} / ` + - `imports=${importsWrapped} — wrapping ran under the wrong mode`, - ); - return; - } assert_( - entriesWrapped || !importsWrapped, - `suspension mode jspi wrapped imports without wrapping any entry ` + - `(entries=${entriesWrapped}, imports=${importsWrapped}) — a ` + - `Suspending import reached from a non-promising activation traps ` + - `unconditionally (jspi pin (c))`, + entriesConstructed >= 0 && entriesWrapped >= 0 && + entriesWrapped <= entriesConstructed, + `invalid entry wrapping counts: constructed=${entriesConstructed}, ` + + `wrapped=${entriesWrapped}`, + ); + const expectedWrapped = mode === "jspi" ? entriesConstructed : 0; + assert_( + entriesWrapped === expectedWrapped, + `${mode} mode constructed ${entriesConstructed} entries but wrapped ` + + `${entriesWrapped}; every constructed entry must use the selected mode`, + ); + assert_( + mode === "jspi" || !importsWrapped, + `plain mode with wrapped imports — wrapping ran under the wrong mode`, ); } @@ -396,6 +427,8 @@ export class SuspensionPoint implements SchedulableThread { */ // deno-lint-ignore no-explicit-any readonly owner: any; + /** Canonical task/context identity active inside the physical continuation. */ + readonly logicalOwner: CurrentThreadLike | null; constructor( store: Store, @@ -415,6 +448,7 @@ export class SuspensionPoint implements SchedulableThread { private readonly produce: (cancelled: Cancelled) => T, // deno-lint-ignore no-explicit-any owner?: any, + logicalOwner?: CurrentThreadLike | null, /** * `finally`-style hook: runs EXACTLY ONCE, on whichever terminal * transition this point takes — produce-success, produce-throw, or @@ -439,6 +473,7 @@ export class SuspensionPoint implements SchedulableThread { ) { this.#store = store; this.owner = owner ?? maybeCurrentThread() ?? task?.implicitThread ?? null; + this.logicalOwner = logicalOwner ?? this.owner; if (SP_TRACE) { console.error( `[sp] mint ${dbgId(this)} owner=${dbgId(this.owner)} task=${ @@ -554,9 +589,11 @@ export class SuspensionPoint implements SchedulableThread { // still be retired here or the store stays gated on a finished window. this.#store.consumePendingIfRunning(); if (maybeCurrentThread() === undefined) { - claimActivationAmbient(this.owner); + claimActivationAmbient(this.logicalOwner); } - this.#store.addPendingResumption(this.task?.implicitThread ?? null); + // Scheduling gates track the physical JSPI continuation. `logicalOwner` + // is only the canonical context published while that continuation runs. + this.#store.addPendingResumption(this.owner); this.#fail(e); return; } @@ -580,8 +617,10 @@ export class SuspensionPoint implements SchedulableThread { // first `Suspending` call (site (ii)); the retired tier-3 slot used to // cover the window before that, and the measurement behind its retirement // (#158, see `resolveAmbient`) says nothing ever read it there. - if (maybeCurrentThread() === undefined) claimActivationAmbient(this.owner); - this.#store.addPendingResumption(this.task?.implicitThread ?? null); + if (maybeCurrentThread() === undefined) { + claimActivationAmbient(this.logicalOwner); + } + this.#store.addPendingResumption(this.owner); this.#settle(value); } @@ -658,13 +697,15 @@ export function blockCurrentActivation(input: { // ambient claim ends here. Reading it after the release yields `undefined` // and strands the point with no owner (measured: `cancellable.wast:322` // then reported `pending-capability: instantiation-time task context`). - const owner = maybeCurrentThread() ?? input.task?.implicitThread ?? null; + const logicalOwner = maybeCurrentThread() ?? input.task?.implicitThread ?? + null; + const owner = logicalOwner === null ? null : physicalOwnerOf(logicalOwner); // The activation is parking: if it still carried the pending-resumption // entry from the settle that resumed it, that entry's window closes here // (the other closing edge — the activation FINISHING — is handled by // `Store.noteAwaiting`'s settle continuation). input.store.consumePendingIfRunning(); - releaseActivationAmbient(owner); + releaseActivationAmbient(logicalOwner); const point = new SuspensionPoint( input.store, input.task, @@ -672,6 +713,7 @@ export function blockCurrentActivation(input: { input.cancellable, input.produce, owner, + logicalOwner, input.onSettled, ); Promise.resolve().then(() => { diff --git a/runtime/src/task/mod.ts b/runtime/src/task/mod.ts index d28fc12..b4c6757 100644 --- a/runtime/src/task/mod.ts +++ b/runtime/src/task/mod.ts @@ -10,11 +10,18 @@ import { type Cancelled, CANCELLED_TRUE, chooseCandidate, + claimActivationAmbient, + type CurrentThreadLike, dbgId, isInstancePoisoned, + isSynchronousAmbient, + maybeCurrentThread, NeedsJspi, notifyInstancePoisoned, PendingCapability, + physicalOwnerOf, + popLogicalActivation, + pushLogicalActivation, Store, } from "./scheduler.ts"; import { Thread } from "./thread.ts"; @@ -404,5 +411,169 @@ export class Task { } } +/** + * A synchronous nested canonical call. The reference always creates a fresh + * Task and Thread for `canon_lift`, including realloc and sync-to-sync FACT + * calls (definitions.py:2128-2194, 642-658). The native implementation may + * defer allocation, but still saves/zeros/restores both context slots + * (wasmtime component_sync_call.rs:40-43,98-123,183-203). + * + * This is an actual Task/Thread identity, not a slots-only shim, so task-scoped + * builtins reached by the callee resolve against the callee. The body itself is + * driven by the already-running wasm stack; no second generator is needed. + */ +export class SynchronousActivation { + readonly task: Task; + readonly thread: Thread; + readonly logicalActivation: CurrentThreadLike["logicalActivation"]; + readonly parent: CurrentThreadLike | undefined; + + constructor( + readonly inst: ComponentInstanceState, + asyncTyped: boolean, + parent: CurrentThreadLike | undefined, + ) { + this.parent = parent; + this.task = new Task( + { params: [], results: [], async: asyncTyped }, + { + async_: false, + callback: false, + stringEncoding: "utf8", + memory: null, + }, + inst, + () => [], + () => {}, + ); + this.thread = new Thread(this.task, (function* () {})()); + const physical = this.parent === undefined + ? this.thread + : physicalOwnerOf(this.parent) as Thread; + this.thread.physicalOwner = physical; + if (physical !== this.thread) { + if (physical.logicalDescendants === undefined) { + (physical as CurrentThreadLike & { + logicalDescendants: Set; + }).logicalDescendants = new Set(); + } + physical.logicalDescendants.add(this.thread); + } + this.task.implicitThread = this.thread; + if (asyncTyped) { + // A sync-ABI implementation of an async-typed function needs the + // instance's exclusive slot (definitions.py Task.needs_exclusive and + // enter_implicit_thread, lines 447-469). enter-sync-call is already the + // admitted synchronous path; an occupied slot would have blocked before + // this adapter invocation. + assert_( + inst.exclusiveThread === null, + "synchronous activation entered with exclusive thread occupied", + ); + inst.exclusiveThread = this.thread; + } + this.task.registerThread(this.thread); + this.task.start(); + this.logicalActivation = { + active: true, + finish: () => this.finish(), + abort: () => this.abort(), + }; + this.thread.logicalActivation = this.logicalActivation; + pushLogicalActivation(this.thread); + } + + finish(): void { + if (!this.logicalActivation!.active) return; + trapIf( + this.task.numBorrows > 0, + "borrow handles still remain at the end of the call", + ); + // FACT exits this task before translating results into the caller and + // temporarily restores the saved callee context for post-return + // (wasmtime fact/trampoline.rs:853-904). This bookkeeping task intentionally + // has an empty host-visible result tuple: core result locals remain in the + // adapter. Resolve through the canonical state transition rather than + // mutating state as an unregister workaround. + this.task.return_([]); + this.task.unregisterThread(this.thread); + if (this.task.ft.async) { + assert_( + this.inst.exclusiveThread === this.thread, + "synchronous activation lost exclusive thread", + ); + this.inst.exclusiveThread = null; + this.inst.store.requestService(); + } + this.logicalActivation!.active = false; + this.thread.physicalOwner.logicalDescendants.delete(this.thread); + popLogicalActivation(this.thread); + if (this.parent && !isSynchronousAmbient(this.parent)) { + // The nested wasm continuation has returned through exit-sync-call. Its + // caller's continuation is now the executing canonical activation. + claimActivationAmbient(this.parent); + } + } + + abort(): void { + if (!this.logicalActivation!.active) return; + // Trap/capability unwind does not complete the task, but the synthetic + // thread must leave the instance table and ambient chain. + // FACT traps can skip exit-sync-call. Release all lenders owned by the + // abandoned nested activation before removing its task identity. + try { + // Cleanup is best-effort here: abort always runs while preserving an + // already escaping trap/capability signal. Attempt every lender scope. + while (this.thread.syncCallStack.length > 0) { + const scope = this.thread.syncCallStack.pop() as { + releaseLenders(): void; + }; + try { + scope.releaseLenders(); + } catch { + // The original activation failure remains authoritative. + } + } + } finally { + const i = this.task.threads.indexOf(this.thread); + if (i !== -1) this.task.threads.splice(i, 1); + if (this.thread.index !== null) { + this.inst.threads.remove(this.thread.index); + this.thread.index = null; + } + if (this.inst.exclusiveThread === this.thread) { + this.inst.exclusiveThread = null; + this.inst.store.requestService(); + } + this.inst.store.removePendingResumption(this.thread); + this.logicalActivation!.active = false; + this.thread.physicalOwner.logicalDescendants.delete(this.thread); + } + } +} + +/** Run a host-invoked synchronous canonical helper (notably realloc). */ +export function withSynchronousActivation( + inst: ComponentInstanceState, + fn: () => T, +): T { + // This helper enters immediately, so sampling its caller here cannot cross + // an asynchronous admission boundary. + const activation = new SynchronousActivation( + inst, + false, + maybeCurrentThread(), + ); + try { + const result = fn(); + activation.finish(); + return result; + } catch (e) { + activation.abort(); + popLogicalActivation(activation.thread); + throw e; + } +} + /** Convenience re-exports so `../task/mod.ts` remains the single entry point. */ export { Store, Subtask, Thread, Waitable, WaitableSet }; diff --git a/runtime/src/task/scheduler.ts b/runtime/src/task/scheduler.ts index 7cf4a5a..7eb02bf 100644 --- a/runtime/src/task/scheduler.ts +++ b/runtime/src/task/scheduler.ts @@ -318,6 +318,19 @@ export interface CurrentThreadLike { storage: number[]; // deno-lint-ignore no-explicit-any task: any; + /** Generator-backed activation whose JSPI Promise/awaiting entry drives us. */ + physicalOwner?: CurrentThreadLike; + /** Promise park owned by generator-backed threads. */ + awaiting?: Promise | null; + /** Logical FACT descendants which survive synchronous-stack unpublication. */ + logicalDescendants?: Set; + /** A canonical call frame represented without a separately-driven body. */ + logicalActivation?: { + active: boolean; + finish(): void; + abort(): void; + parent?: CurrentThreadLike; + }; } export function pushCurrentThread(t: CurrentThreadLike): void { @@ -325,10 +338,50 @@ export function pushCurrentThread(t: CurrentThreadLike): void { } export function popCurrentThread(t: CurrentThreadLike): void { + // A nested logical activation can outlive this JS bracket across a JSPI hop. + // Remove only its synchronous publication; completion/abort are separate. + while ( + threadStack.length > 0 && threadStack[threadStack.length - 1] !== t + ) { + const nested = threadStack.pop() as CurrentThreadLike; + assert_( + nested.logicalActivation !== undefined, + "current-thread stack contains an unrelated activation", + ); + } const top = threadStack.pop(); assert_(top === t, "current-thread stack imbalance"); } +/** Abnormal wasm-stack unwind: retire logical descendants whose exits skipped. */ +export function abortLogicalChildren(t: CurrentThreadLike): void { + const physical = physicalOwnerOf(t); + for (const nested of [...(physical.logicalDescendants ?? [])]) { + nested.logicalActivation?.abort(); + releaseActivationAmbient(nested); + } +} + +export function physicalOwnerOf(t: CurrentThreadLike): CurrentThreadLike { + return t.physicalOwner ?? t; +} + +/** Publish a synchronous nested canonical activation as `current_thread`. */ +export function pushLogicalActivation(t: CurrentThreadLike): void { + threadStack.push(t); +} + +/** Finish a nested activation whether it is still synchronous or JSPI-resumed. */ +export function popLogicalActivation(t: CurrentThreadLike): "stack" | "claim" { + if (threadStack[threadStack.length - 1] === t) { + threadStack.pop(); + return "stack"; + } else { + releaseActivationAmbient(t); + return "claim"; + } +} + /** * Run `fn` with `t` as ambient for its synchronous extent only. Wasm entries * and built-in bodies use this even during engine-driven resumptions, when @@ -340,8 +393,20 @@ export function withActivation(t: any, fn: () => T): T { entryStack.push(t); try { return fn(); + } catch (e) { + abortLogicalChildren(t); + throw e; } finally { entryStack.pop(); + while ( + threadStack.length > 0 && threadStack[threadStack.length - 1] !== t + ) { + const nested = threadStack.pop() as CurrentThreadLike; + assert_( + nested.logicalActivation !== undefined, + "withActivation crossed an unrelated activation", + ); + } const top = threadStack.pop(); assert_(top === t, "withActivation: current-thread stack imbalance"); } @@ -426,6 +491,9 @@ export function dbgId(t: unknown): string { export function releaseActivationAmbient(t: any): void { if (t === null || t === undefined) return; if (AMBIENT_TRACE) traceAmbient("release", t); + // This releases only an engine-hop claim. Synchronous publication belongs + // to push/popCurrentThread and withActivation; touching it here double-pops + // a live bracket when a logical child parks in a Suspending import. let i = activationClaims.indexOf(t); if (i === -1) { const implicit = (t as { task?: { implicitThread?: unknown } })?.task @@ -507,6 +575,11 @@ export function maybeCurrentThread(): CurrentThreadLike | undefined { return resolveAmbient(); } +/** Whether `t` is supplied by a live synchronous JS bracket, not a claim. */ +export function isSynchronousAmbient(t: CurrentThreadLike): boolean { + return threadStack[threadStack.length - 1] === t; +} + /** Whether guest code is executing on the current JavaScript stack. Engine * continuation claims deliberately do not count: they describe attribution * across a hop, not live execution. Ordinary store service must never enter a @@ -723,7 +796,8 @@ export class Store { consumePendingIfRunning(): void { const a = activationOf(); if ( - a !== null && a !== undefined && this.pendingResumptions.delete(a) + a !== null && a !== undefined && + this.pendingResumptions.delete(physicalOwnerOf(a)) ) { // This request is queued while the activation is still live. The // coordinator's live-execution guard delays service until the canonical @@ -739,11 +813,15 @@ export class Store { // deno-lint-ignore no-explicit-any releasePendingOf(t: any): void { releaseActivationAmbient(t); - let released = this.pendingResumptions.delete(t); + const physical = physicalOwnerOf(t); + let released = this.pendingResumptions.delete(physical); const implicit = (t as { task?: { implicitThread?: unknown } })?.task ?.implicitThread; if (implicit !== undefined && implicit !== null) { - released = this.pendingResumptions.delete(implicit) || released; + released = this.pendingResumptions.delete( + physicalOwnerOf(implicit as CurrentThreadLike), + ) || + released; } if (released) this.requestService(); } diff --git a/runtime/src/task/streams.ts b/runtime/src/task/streams.ts index b6bdc80..1e8c8a0 100644 --- a/runtime/src/task/streams.ts +++ b/runtime/src/task/streams.ts @@ -72,7 +72,10 @@ export class GuestBuffer { public ptr: number, readonly length: number, ) { - trapIf(length > BUFFER_MAX_LENGTH, "buffer length exceeds MAX_LENGTH"); + trapIf( + length > BUFFER_MAX_LENGTH, + "stream read/write count too large", + ); if (t !== null && length > 0) { const mem = cx.opts.memory; assert_(mem !== null, "buffer requires a memory"); @@ -110,6 +113,17 @@ export class GuestBuffer { return vs; } + /** + * Advance a zero-width guest buffer without materializing `n` placeholder + * values. This is the allocation-free form of the pinned reference's + * `n * [()]`, used only when both rendezvous buffers are guest buffers. + */ + advanceZeroWidth(n: number): void { + assert_(this.t === null, "count advance on a payload-bearing buffer"); + assert_(n <= this.remain(), "count advance beyond remaining"); + this.progress += n; + } + /** definitions.py `WritableBufferGuestImpl.write`. */ write(vs: PayloadChunk): void { assert_(vs.length <= this.remain(), "buffer write beyond remaining"); @@ -261,6 +275,19 @@ function rendezvousCopy( const srcDirect = isDirectBuffer(src); const dstDirect = isDirectBuffer(dst); if (!srcDirect && !dstDirect) { + // A payloadless guest-to-guest stream transfers only a count. Constructing + // the reference's conceptual unit-value list makes the legal maximum + // (2**28 - 1 elements) exceed JS Array's practical allocation limits. + // Keep host buffers on the ordinary path so their chunk semantics remain + // unchanged. + if ( + src instanceof GuestBuffer && dst instanceof GuestBuffer && + src.t === null && dst.t === null + ) { + src.advanceZeroWidth(n); + dst.advanceZeroWidth(n); + return "chunk"; + } dst.write(src.read(n)); return "chunk"; } diff --git a/runtime/src/task/thread.ts b/runtime/src/task/thread.ts index ed56b14..0bc554b 100644 --- a/runtime/src/task/thread.ts +++ b/runtime/src/task/thread.ts @@ -15,6 +15,7 @@ import { assert_ } from "../cabi/trap.ts"; import { + abortLogicalChildren, type BlockRequest, type Cancelled, CANCELLED_FALSE, @@ -25,6 +26,7 @@ import { PendingCapability, popCurrentThread, pushCurrentThread, + releaseActivationAmbient, type SchedulableThread, type Store, type ThreadBody, @@ -33,6 +35,16 @@ import { type ThreadState = "running" | "suspended" | "waiting" | "done"; export class Thread implements SchedulableThread { + /** Physical generator activation used for JSPI awaiting/resumption. */ + physicalOwner: Thread = this; + /** Persistent logical descendants, including while their stack is unpublished. */ + readonly logicalDescendants: Set = new Set(); + /** Present when this Thread is driven by an enclosing wasm sync call. */ + logicalActivation?: { + active: boolean; + finish(): void; + abort(): void; + }; /** * Per-thread slots for `canon_context_get` / `canon_context_set`, not * task-shared state. Number storage is for the supported i32 context; @@ -188,6 +200,9 @@ export class Thread implements SchedulableThread { // is finished either way; the exception propagates to whoever was // driving the scheduler. this.#state = "done"; + abortLogicalChildren(this); + releaseActivationAmbient(this); + this.#store.removePendingResumption(this); throw e; } finally { popCurrentThread(this); diff --git a/runtime/tests/builtin_index_normalization_test.ts b/runtime/tests/builtin_index_normalization_test.ts index e313b21..90e931f 100644 --- a/runtime/tests/builtin_index_normalization_test.ts +++ b/runtime/tests/builtin_index_normalization_test.ts @@ -249,7 +249,10 @@ Deno.test("stream.write(n=0xFFFFFFFF) traps instead of parking or completing wit // and this must trap. Ours: `length > BUFFER_MAX_LENGTH` sees the signed // JS number -1, the comparison is false, and the write silently parks // (returns BLOCKED) instead. - assertTraps(() => write(wi, 0, NEG_ONE), "MAX_LENGTH"); + assertTraps( + () => write(wi, 0, NEG_ONE), + "stream read/write count too large", + ); } finally { popCurrentThread(thread); } diff --git a/runtime/tests/embedder/host_settlement_test.ts b/runtime/tests/embedder/host_settlement_test.ts index 5cb0cb8..0e0a248 100644 --- a/runtime/tests/embedder/host_settlement_test.ts +++ b/runtime/tests/embedder/host_settlement_test.ts @@ -13,6 +13,7 @@ import type { HostResourceRegistry } from "../../src/embedder/resources.ts"; import { type ComponentInstanceState, currentTask, + currentThread, isInstancePoisoned, NeedsJspi, } from "../../src/task/mod.ts"; @@ -41,10 +42,21 @@ for (const rejection of [false, true]) { const p = deferred(); const started = Promise.withResolvers(); let caller!: ComponentInstanceState; + let callee!: ComponentInstanceState; const c = await instantiateFixture(factFixture, { r: R, makeSync: suspending(() => { - caller = currentTask().inst; + callee = currentTask().inst; + const physical = currentThread().physicalOwner; + if (physical === undefined) { + throw new Error("FACT callee has no physical caller"); + } + caller = physical.task.inst; + assertEq( + caller === callee, + false, + "FACT caller and callee are distinct", + ); started.resolve(); return p.promise; }), @@ -55,7 +67,7 @@ for (const rejection of [false, true]) { const cause = await caught(() => sync(c.exports.trap)()); assertEq(cause instanceof Trap, true); assertEq( - isInstancePoisoned(c.handle.componentInstances[1]), + isInstancePoisoned(callee), true, "FACT callee poisoned", ); diff --git a/runtime/tests/enter_sync_call_reentrance_test.ts b/runtime/tests/enter_sync_call_reentrance_test.ts index 2473c63..8a03a66 100644 --- a/runtime/tests/enter_sync_call_reentrance_test.ts +++ b/runtime/tests/enter_sync_call_reentrance_test.ts @@ -26,8 +26,25 @@ import { type TrampolineContext, } from "../src/intrinsics/mod.ts"; import { newStats } from "../src/exec/boundary.ts"; -import { ComponentInstanceState, Store } from "../src/task/mod.ts"; -import { notifyInstancePoisoned } from "../src/task/scheduler.ts"; +import { + ComponentInstanceState, + Store, + SynchronousActivation, + Task, + Thread, +} from "../src/task/mod.ts"; +import { + ambientResidue, + currentTask, + currentThread, + maybeCurrentThread, + notifyInstancePoisoned, + withActivation, +} from "../src/task/scheduler.ts"; +import { + blockCurrentActivation, + type SuspensionPoint, +} from "../src/jspi/bridge.ts"; function fixture() { const store = new Store(); @@ -64,23 +81,29 @@ const A = 0; const C = 1; Deno.test("enter-sync-call: an idle sibling callee is enterable", () => { - const { enter, exit, syncCallStack } = fixture(); + const { enter, exit, inst } = fixture(); enter(A, 0, C); - assertEq(syncCallStack.length, 1, "bracket opened"); + assertEq(currentTask().inst === inst(C), true, "callee task is current"); + assertEq(currentThread().storage, [0, 0], "callee slots start fresh"); exit(); - assertEq(syncCallStack.length, 0, "bracket closed"); + assertEq(maybeCurrentThread(), undefined, "bracket closed"); }); Deno.test("enter-sync-call: a sibling cycle A -> C -> A no longer traps (CM#705)", () => { // Host entered A; A is mid-call into C; C calls back into A. That is // simply a valid call (CM#705). - const { enter, exit, syncCallStack } = fixture(); + const { enter, exit, inst } = fixture(); enter(A, 0, C); + const c = currentThread(); + c.storage[0] = 41; enter(C, 0, A); - assertEq(syncCallStack.length, 2, "both brackets opened, nothing refused"); + assertEq(currentTask().inst === inst(A), true); + assertEq(currentThread().storage, [0, 0]); exit(); + assertEq(currentThread() === c, true, "parent activation restored"); + assertEq(c.storage[0], 41); exit(); - assertEq(syncCallStack.length, 0); + assertEq(maybeCurrentThread(), undefined); }); Deno.test("enter-sync-call: a POISONED callee is refused, naming the trap", () => { @@ -105,18 +128,133 @@ Deno.test("enter-sync-call: a POISONED callee is refused, naming the trap", () = Deno.test("enter-sync-call: a poisoned instance calling ITSELF passes vacuously", () => { // `entryRefusal`'s `caller !== callee` guard passes a self-call // vacuously, even against a marked instance. - const { enter, inst } = fixture(); + const { enter, exit, inst } = fixture(); notifyInstancePoisoned(inst(A), new Error("earlier boom")); enter(A, 0, A); + exit(); }); Deno.test("enter-sync-call: an acyclic sibling chain A -> B -> C never traps", () => { - const { enter, exit, syncCallStack } = fixture(); + const { enter, exit, inst } = fixture(); const B = 2; enter(A, 0, B); enter(B, 0, C); - assertEq(syncCallStack.length, 2); + assertEq(currentTask().inst === inst(C), true); exit(); + assertEq(currentTask().inst === inst(B), true); exit(); - assertEq(syncCallStack.length, 0); + assertEq(maybeCurrentThread(), undefined); +}); + +Deno.test("enter-sync-call: trap unwind retires nested task identity", () => { + const { enter, inst } = fixture(); + const parent = { storage: [7, 8], task: { inst: inst(A) } }; + const boom = new Error("nested trap"); + let caught: unknown; + try { + withActivation(parent, () => { + enter(A, 0, C); + assertEq(currentTask().inst === inst(C), true); + throw boom; + }); + } catch (e) { + caught = e; + } + assertEq(caught === boom, true); + assertEq([...inst(C).threads].length, 0, "callee task retired on unwind"); + assertEq(maybeCurrentThread(), undefined, "ambient parent also unwound"); +}); + +Deno.test("nested sync suspension keeps physical owner and logical task separate", async () => { + const store = new Store(); + const outerInst = new ComponentInstanceState(0, store); + const innerInst = new ComponentInstanceState(1, store); + const outerTask = new Task( + { params: [], results: [], async: true }, + { async_: true, callback: false, stringEncoding: "utf8", memory: null }, + outerInst, + () => [], + () => {}, + ); + let logical!: SynchronousActivation; + const physical = new Thread( + outerTask, + (function* () { + logical = new SynchronousActivation(innerInst, true, currentThread()); + const promise = blockCurrentActivation({ + store, + task: logical.task, + readyFunc: () => true, + cancellable: false, + produce: () => undefined, + }); + yield { readyFunc: null, cancellable: false, awaitValue: promise }; + logical.finish(); + })(), + ); + physical.resume(); + const point = store.waiting[0] as SuspensionPoint; + assertEq(point.owner === physical, true, "scheduler owner is physical"); + assertEq( + point.logicalOwner === logical.thread, + true, + "context owner is logical", + ); + assertEq(point.task === logical.task, true, "built-in task remains logical"); + point.resume(); + await Promise.resolve(); + store.serviceSettled(); + assertEq([...innerInst.threads].length, 0); + assertEq(physical.logicalDescendants.size, 0); + assertEq(ambientResidue(), { stack: 0, claim: false }); +}); + +Deno.test("nested sync post-hop trap retires persistent logical descendants", async () => { + const store = new Store(); + const outerInst = new ComponentInstanceState(0, store); + const innerInst = new ComponentInstanceState(1, store); + const outerTask = new Task( + { params: [], results: [], async: true }, + { async_: true, callback: false, stringEncoding: "utf8", memory: null }, + outerInst, + () => [], + () => {}, + ); + const boom = new Error("after-hop trap"); + const physical = new Thread( + outerTask, + (function* () { + const logical = new SynchronousActivation( + innerInst, + true, + currentThread(), + ); + const promise = blockCurrentActivation({ + store, + task: logical.task, + readyFunc: () => true, + cancellable: false, + produce: () => undefined, + }); + yield { readyFunc: null, cancellable: false, awaitValue: promise }; + throw boom; + })(), + ); + physical.resume(); + (store.waiting[0] as SuspensionPoint).resume(); + await Promise.resolve(); + let caught: unknown; + try { + store.serviceSettled(); + } catch (e) { + caught = e; + } + assertEq( + (caught as { cause?: unknown })?.cause === boom || caught === boom, + true, + ); + assertEq([...innerInst.threads].length, 0); + assertEq(physical.logicalDescendants.size, 0); + assertEq(store.pendingResumptions.size, 0); + assertEq(ambientResidue(), { stack: 0, claim: false }); }); diff --git a/runtime/tests/fact_call_test.ts b/runtime/tests/fact_call_test.ts index 8433fc1..4b5b356 100644 --- a/runtime/tests/fact_call_test.ts +++ b/runtime/tests/fact_call_test.ts @@ -77,6 +77,9 @@ function runPrepared(input: { callback: () => { throw new Error("no callback expected"); }, + postReturn: () => { + throw new Error("unexpected post-return"); + }, memoryToken: () => null, stats: newStats(), prepared, @@ -256,3 +259,54 @@ Deno.test("FACT: a subtask reports STARTING until [async-start] actually runs", // Sanity: the eager path resolved, so no handle was allocated. assertEq(SubtaskState.RETURNED, 2); }); + +Deno.test("FACT: post-return and callback index spaces do not alias", () => { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + let callbackRan = false; + let postReturnRan = false; + const ctx = { + componentInstance: () => inst, + resultTypes: () => [] as ValType[], + resultTypesForTuple: () => null, + callback: (i: number) => { + assertEq(i, 0); + return () => { + callbackRan = true; + throw new Error("callback must not be used as post-return"); + }; + }, + postReturn: (i: number) => { + assertEq(i, 0); + return () => { + postReturnRan = true; + }; + }, + memoryToken: () => null, + stats: newStats(), + prepared: { current: null }, + factStartScopes: [], + suspensionMode: "plain" as const, + }; + const prep = createPrepareCall( + { memory: null }, + ctx as unknown as FactCallContext, + ); + const startCall = createAsyncStartCall( + { callback: null, postReturn: 0 }, + ctx as unknown as FactCallContext, + ); + prep( + () => undefined, + () => undefined, + 0, + 0, + 0, + 0, + 0, + PREPARE_ASYNC_NO_RESULT, + ); + assertEq(startCall(() => undefined, 0, 0, 0), 2); + assertEq(postReturnRan, true); + assertEq(callbackRan, false); +}); diff --git a/runtime/tests/fact_post_return_context.wasm b/runtime/tests/fact_post_return_context.wasm new file mode 100644 index 0000000..3318789 Binary files /dev/null and b/runtime/tests/fact_post_return_context.wasm differ diff --git a/runtime/tests/fact_post_return_context.wat b/runtime/tests/fact_post_return_context.wat new file mode 100644 index 0000000..933b6d5 --- /dev/null +++ b/runtime/tests/fact_post_return_context.wat @@ -0,0 +1,79 @@ +(component + (component $callee + (canon task.return (core func $task-return)) + (canon context.get i32 0 (core func $context-get)) + (canon context.set i32 0 (core func $context-set)) + (core module $m + (import "" "task-return" (func $task-return)) + (import "" "context-get" (func $context-get (result i32))) + (import "" "context-set" (func $context-set (param i32))) + (global $seen (mut i32) (i32.const -1)) + (func (export "f") (result i32) + (i32.const 0)) + (func (export "g") (result i32) + (call $context-set (i32.const 42)) + (i32.const 1)) + ;; The zero result passed by post-return takes the invalid path. The same + ;; core function remains callable normally with a nonzero argument. + (func (export "post") (param i32) + (if (i32.eqz (local.get 0)) + (then (call $task-return)) + (else + (global.set $seen (call $context-get)) + (call $context-set (i32.const 99))))) + (func (export "seen") (result i32) (global.get $seen))) + (core instance $m (instantiate $m (with "" (instance + (export "task-return" (func $task-return)) + (export "context-get" (func $context-get)) + (export "context-set" (func $context-set)))))) + (func (export "f") (result u32) (canon lift + (core func $m "f") + (post-return (core func $m "post")))) + (func (export "g") (result u32) (canon lift + (core func $m "g") + (post-return (core func $m "post")))) + (func (export "post") (param "value" u32) (canon lift + (core func $m "post"))) + (func (export "seen") (result u32) (canon lift (core func $m "seen")))) + + (component $caller + (import "f" (func $f (result u32))) + (import "g" (func $g (result u32))) + (canon context.get i32 0 (core func $context-get)) + (canon context.set i32 0 (core func $context-set)) + (canon lower (func $f) (core func $f-core)) + (canon lower (func $g) (core func $g-core)) + (core module $m + (import "" "f" (func $f (result i32))) + (import "" "g" (func $g (result i32))) + (import "" "context-get" (func $context-get (result i32))) + (import "" "context-set" (func $context-set (param i32))) + (func (export "run") (result i32) + (drop (call $f)) + (i32.const 0)) + (func (export "context") (result i32) + (call $context-set (i32.const 7)) + (drop (call $g)) + (call $context-get)) + (func (export "callback") (param i32 i32 i32) (result i32) + (i32.const 0))) + (core instance $m (instantiate $m (with "" (instance + (export "f" (func $f-core)) + (export "g" (func $g-core)) + (export "context-get" (func $context-get)) + (export "context-set" (func $context-set)))))) + ;; An async caller makes the erroneous task.return target a valid live task + ;; after FACT's exit-sync-call has retired the callee task. Callback ABI + ;; keeps the fixture executable in both plain and JSPI modes. + (func (export "run") async (canon lift (core func $m "run") async + (callback (core func $m "callback")))) + (func (export "context") (result u32) (canon lift (core func $m "context")))) + + (instance $callee (instantiate $callee)) + (instance $caller (instantiate $caller + (with "f" (func $callee "f")) + (with "g" (func $callee "g")))) + (func (export "run") (alias export $caller "run")) + (func (export "context") (alias export $caller "context")) + (func (export "post") (alias export $callee "post")) + (func (export "seen") (alias export $callee "seen"))) diff --git a/runtime/tests/fact_post_return_context_test.ts b/runtime/tests/fact_post_return_context_test.ts new file mode 100644 index 0000000..a3de8ed --- /dev/null +++ b/runtime/tests/fact_post_return_context_test.ts @@ -0,0 +1,66 @@ +import { assertEq, assertTrap } from "./support/asserts.ts"; +import { Trap } from "../src/cabi/trap.ts"; +import { instantiateComponent } from "../src/exec/mod.ts"; +import { Translator } from "../src/shim/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("fact_post_return_context.wasm", import.meta.url), +); + +for (const jspi of [false, true]) { + Deno.test(`FACT sync post-return retains callee restrictions (jspi=${jspi})`, async () => { + const translator = await Translator.create(shim); + const { plan, adapters } = translator.translate(fixture); + const postReturnImports = plan.initializers.flatMap((init) => { + if (init.op !== "instantiate-module" || init.instance !== null) return []; + const module = plan.modules[init.module]; + if (module.kind !== "adapter") return []; + return module.intrinsics.filter((entry) => + entry.module === "post_return" + ); + }); + assertEq(postReturnImports.length, 1, "pinned FACT post_return imports"); + for (const entry of postReturnImports) assertEq(entry.def.kind, "export"); + const component = await instantiateComponent({ + plan, + adapters, + componentBytes: fixture, + jspi, + }); + const exports = component.exports as Record< + string, + (...a: unknown[]) => unknown + >; + + // FACT restores the callee's saved context around post-return and then the + // caller's context afterward; the wrapper changes neither task identity. + assertEq(await exports.context(), 7); + assertEq(await exports.seen(), 42); + // The same core function is legal through its ordinary lifted export. The + // post-return restriction belongs only to FACT's post_return import edge. + assertEq(await exports.post(1), undefined); + + const run = () => exports.run(); + if (jspi) { + let caught: unknown; + try { + await run(); + } catch (e) { + caught = e; + } + assertEq(caught instanceof Trap, true, "expected rejected Trap"); + assertEq( + String((caught as Error).message).includes( + "cannot leave component instance", + ), + true, + ); + } else { + assertTrap(run, "cannot leave component instance"); + } + }); +} diff --git a/runtime/tests/fixtures/nested-sync-yield.wasm b/runtime/tests/fixtures/nested-sync-yield.wasm new file mode 100644 index 0000000..2a7dd84 Binary files /dev/null and b/runtime/tests/fixtures/nested-sync-yield.wasm differ diff --git a/runtime/tests/fixtures/nested-sync-yield.wat b/runtime/tests/fixtures/nested-sync-yield.wat new file mode 100644 index 0000000..dfddd53 --- /dev/null +++ b/runtime/tests/fixtures/nested-sync-yield.wat @@ -0,0 +1,33 @@ +;; A real translated cross-component call where an async-typed function has a +;; synchronous core implementation. Its thread.yield parks the physical caller +;; while the canonical callee task remains the logical context owner. +(component + (component $A + (core func $yield (canon thread.yield)) + (core module $M + (import "" "yield" (func $yield (result i32))) + (func (export "run") (param $fail i32) (result i32) + (drop (call $yield)) + (if (local.get $fail) (then unreachable)) + (i32.const 42))) + (core instance $m (instantiate $M (with "" (instance + (export "yield" (func $yield)))))) + (func (export "run") async (param "fail" u32) (result u32) + (canon lift (core func $m "run")))) + + (component $B + (import "run" (func $run async (param "fail" u32) (result u32))) + ;; A synchronous lower is the FACT async-typed/sync-ABI path under test. + (core func $run' (canon lower (func $run))) + (core module $M + (import "" "run" (func $run (param i32) (result i32))) + (func (export "run") (param i32) (result i32) + (call $run (local.get 0)))) + (core instance $m (instantiate $M (with "" (instance + (export "run" (func $run')))))) + (func (export "run") async (param "fail" u32) (result u32) + (canon lift (core func $m "run")))) + + (instance $a (instantiate $A)) + (instance $b (instantiate $B (with "run" (func $a "run")))) + (export "run" (func $b "run"))) diff --git a/runtime/tests/fixtures/sync-admission-cancel.wasm b/runtime/tests/fixtures/sync-admission-cancel.wasm new file mode 100644 index 0000000..a7e066f Binary files /dev/null and b/runtime/tests/fixtures/sync-admission-cancel.wasm differ diff --git a/runtime/tests/fixtures/sync-admission-cancel.wat b/runtime/tests/fixtures/sync-admission-cancel.wat new file mode 100644 index 0000000..aad4106 --- /dev/null +++ b/runtime/tests/fixtures/sync-admission-cancel.wat @@ -0,0 +1,89 @@ +;; A pending caller cancellation must not cancel the synchronous wait used to +;; admit an async-typed/sync-ABI callee under backpressure. +(component + (component $A + (core func $inc (canon backpressure.inc)) + (core func $dec (canon backpressure.dec)) + (core func $yield (canon thread.yield)) + (core module $M + (import "" "inc" (func $inc)) + (import "" "dec" (func $dec)) + (import "" "yield" (func $yield (result i32))) + (func (export "inc") (call $inc)) + (func (export "dec") (call $dec)) + (func (export "run") (param $fail i32) (result i32) + (drop (call $yield)) + (if (local.get $fail) (then unreachable)) + (i32.const 7))) + (core instance $m (instantiate $M (with "" (instance + (export "inc" (func $inc)) + (export "dec" (func $dec)) + (export "yield" (func $yield)))))) + (func (export "inc") (canon lift (core func $m "inc"))) + (func (export "dec") (canon lift (core func $m "dec"))) + (func (export "run") async (param "fail" u32) (result u32) + (canon lift (core func $m "run")))) + + (component $B + (import "run" (func $run async (param "fail" u32) (result u32))) + (core func $run' (canon lower (func $run))) + (core module $Mem (memory (export "mem") 1)) + (core instance $mem (instantiate $Mem)) + (core func $return (canon task.return (result u32) + (memory (core memory $mem "mem")))) + (core module $M + (import "" "run" (func $run (param i32) (result i32))) + (import "" "return" (func $return (param i32))) + (func (export "run") (param $fail i32) (result i32) + (call $return (call $run (local.get $fail))) + (i32.const 0)) + (func (export "callback") (param i32 i32 i32) (result i32) + unreachable)) + (core instance $m (instantiate $M (with "" (instance + (export "run" (func $run')) + (export "return" (func $return)))))) + (func (export "run") async (param "fail" u32) (result u32) + (canon lift (core func $m "run") async + (memory (core memory $mem "mem")) + (callback (core func $m "callback"))))) + + (component $C + (import "run" (func $run async (param "fail" u32) (result u32))) + (core module $Mem (memory (export "mem") 1)) + (core instance $mem (instantiate $Mem)) + (core func $run' (canon lower (func $run) async + (memory (core memory $mem "mem")))) + (core func $cancel (canon subtask.cancel)) + (core func $return (canon task.return (result u32) + (memory (core memory $mem "mem")))) + (core module $M + (import "" "mem" (memory 1)) + (import "" "run" (func $run (param i32 i32) (result i32))) + (import "" "cancel" (func $cancel (param i32) (result i32))) + (import "" "return" (func $return (param i32))) + (func (export "run") (result i32) + (local $packed i32) + (i32.store (i32.const 0) (i32.const 0)) + (local.set $packed (call $run (i32.const 0) (i32.const 0))) + (drop (call $cancel (i32.shr_u (local.get $packed) (i32.const 4)))) + (call $return (i32.load (i32.const 0))) + (i32.const 0)) + (func (export "callback") (param i32 i32 i32) (result i32) + unreachable)) + (core instance $m (instantiate $M (with "" (instance + (export "mem" (memory $mem "mem")) + (export "run" (func $run')) + (export "cancel" (func $cancel)) + (export "return" (func $return)))))) + (func (export "run") async (result u32) + (canon lift (core func $m "run") async + (memory (core memory $mem "mem")) + (callback (core func $m "callback"))))) + + (instance $a (instantiate $A)) + (instance $b (instantiate $B (with "run" (func $a "run")))) + (instance $c (instantiate $C (with "run" (func $b "run")))) + (export "set-pressure" (func $a "inc")) + (export "clear-pressure" (func $a "dec")) + (export "run" (func $b "run")) + (export "run-cancel" (func $c "run"))) diff --git a/runtime/tests/jspi/bridge_test.ts b/runtime/tests/jspi/bridge_test.ts index 7874635..a3aeaa6 100644 --- a/runtime/tests/jspi/bridge_test.ts +++ b/runtime/tests/jspi/bridge_test.ts @@ -175,33 +175,37 @@ Deno.test("bridge: JSPI-less degradation — plain mode wraps nothing", () => { Deno.test("bridge: the invariant is checked, not hoped for", () => { // plain mode: neither site wrapped. - assertModeConsistent("plain", false, false); + assertModeConsistent("plain", 1, 0, false); // jspi mode: both sites wrapped. - assertModeConsistent("jspi", true, true); + assertModeConsistent("jspi", 1, 1, true); // jspi mode with NO wrapped imports is legal: per-declaration // classification (trampolineNeedsSuspension) wraps nothing in a component // whose built-ins are all async forms, while the mode can still be jspi // via planNeedsSuspension's lift-shape over-approximation. - assertModeConsistent("jspi", true, false); - // The dangerous mixtures are rejected — fact (c) makes these fatal: - // a Suspending import with no promising entry, and any wrapping at all - // in plain mode. + assertModeConsistent("jspi", 1, 1, false); + // A declaration-only component has no entry path from which its wrapped + // imports can be reached, so it needs no synthetic promising entry. + assertModeConsistent("jspi", 0, 0, true); + // Every entry that does exist must be wrapped according to the selected + // mode, and imports must never be wrapped in plain mode. for ( - const [mode, e, i] of [ - ["jspi", false, true], - ["plain", true, false], - ["plain", false, true], + const [mode, constructed, wrapped, imports] of [ + ["jspi", 1, 0, true], + ["jspi", 2, 1, false], + ["plain", 1, 1, false], + ["plain", 0, 0, true], ] as const ) { let threw = false; try { - assertModeConsistent(mode, e, i); + assertModeConsistent(mode, constructed, wrapped, imports); } catch { threw = true; } assert( threw, - `mixture (${mode}, entries=${e}, imports=${i}) must be rejected`, + `mixture (${mode}, constructed=${constructed}, wrapped=${wrapped}, ` + + `imports=${imports}) must be rejected`, ); } }); @@ -305,6 +309,17 @@ Deno.test("bridge: planNeedsSuspension recognises both sources of blocking", () false, "async-start-call", ); + // Like async-start-call, enter-sync-call is wrapped when the surrounding + // plan already selected JSPI, but it does not itself promote eager FACT + // pass-through callees. Admission can only park under runtime contention. + assertEq( + planNeedsSuspension({ + canonicalOptions: [], + trampolines: [{ kind: "enter-sync-call" }], + }), + false, + "enter-sync-call", + ); }); Deno.test("plain mode: lifted exports still return values, not Promises", async () => { diff --git a/runtime/tests/jspi/nested_sync_yield_test.ts b/runtime/tests/jspi/nested_sync_yield_test.ts new file mode 100644 index 0000000..49160e8 --- /dev/null +++ b/runtime/tests/jspi/nested_sync_yield_test.ts @@ -0,0 +1,56 @@ +import { instantiate } from "../../src/embedder/mod.ts"; +import { Translator } from "../../src/shim/mod.ts"; +import { isTrap } from "@polyengine/protocol"; +import { assert, assertEquals, assertRejects } from "./asserts.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/nested-sync-yield.wasm", import.meta.url), +); +const translator = await Translator.create(shim); + +async function fresh() { + return await instantiate( + { + componentBytes: fixture, + ...translator.translate(fixture), + }, + {}, + { jspi: true }, + ); +} + +Deno.test("physical caller owns a nested sync callee's successful yield resumption", async () => { + const component = await fresh(); + assertEquals(await component.exports.run(0), 42); + for (const inst of component.handle.componentInstances) { + if (inst === undefined) continue; + assertEquals( + [...inst.threads].length, + 0, + "successful call leaked instance threads", + ); + } +}); + +Deno.test("physical caller owns a nested sync callee's trapping yield resumption", async () => { + const component = await fresh(); + const failure = await assertRejects( + () => component.exports.run(1) as Promise, + ); + assert( + isTrap(failure) && String(failure).includes("unreachable"), + `expected unreachable Trap, got ${failure}`, + ); + assertEquals( + component.handle.componentInstances.some((inst) => + inst !== undefined && + [...inst.threads].some((thread) => thread.physicalOwner !== thread) + ), + false, + "trapping call leaked synthetic instance threads", + ); +}); diff --git a/runtime/tests/jspi/sync_admission_cancel_test.ts b/runtime/tests/jspi/sync_admission_cancel_test.ts new file mode 100644 index 0000000..7d1dd8f --- /dev/null +++ b/runtime/tests/jspi/sync_admission_cancel_test.ts @@ -0,0 +1,88 @@ +import { instantiate } from "../../src/embedder/mod.ts"; +import { Translator } from "../../src/shim/mod.ts"; +import { isTrap } from "@polyengine/protocol"; +import { assert, assertEquals, assertRejects } from "./asserts.ts"; +import { ambientResidue } from "../../src/task/scheduler.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/sync-admission-cancel.wasm", import.meta.url), +); +const translator = await Translator.create(shim); + +async function fresh() { + return await instantiate( + { componentBytes: fixture, ...translator.translate(fixture) }, + {}, + { jspi: true }, + ); +} + +function clearPressureLater( + component: Awaited>, +): void { + setTimeout(() => component.exports.clearPressure(), 0); +} + +function assertNoSyntheticThreads( + component: Awaited>, +): void { + assertEquals( + component.handle.componentInstances.some((inst) => + inst !== undefined && + [...inst.threads].some((thread) => thread.physicalOwner !== thread) + ), + false, + "call leaked synthetic instance threads", + ); + const stores = new Set( + component.handle.componentInstances.filter((inst) => inst !== undefined) + .map((inst) => inst.store), + ); + for (const store of stores) { + assertEquals( + store.pendingResumptions.size, + 0, + "call leaked a pending claim", + ); + } + assertEquals( + JSON.stringify(ambientResidue()), + JSON.stringify({ stack: 0, claim: false }), + "call leaked ambient attribution", + ); +} + +Deno.test("deferred sync admission preserves caller through callee yield", async () => { + const component = await fresh(); + component.exports.setPressure(); + const result = component.exports.run(0) as Promise; + clearPressureLater(component); + assertEquals(await result, 7); + assertNoSyntheticThreads(component); +}); + +Deno.test("deferred sync admission cleans a callee trapping after yield", async () => { + const component = await fresh(); + component.exports.setPressure(); + const result = component.exports.run(1) as Promise; + clearPressureLater(component); + const failure = await assertRejects(() => result); + assert( + isTrap(failure) && String(failure).includes("unreachable"), + `expected unreachable Trap, got ${failure}`, + ); + assertNoSyntheticThreads(component); +}); + +Deno.test("sync admission leaves outer cancellation pending", async () => { + const component = await fresh(); + component.exports.setPressure(); + const result = component.exports.runCancel() as Promise; + clearPressureLater(component); + assertEquals(await result, 7); + assertNoSyntheticThreads(component); +}); diff --git a/runtime/tests/realloc_may_leave_test.ts b/runtime/tests/realloc_may_leave_test.ts index ca496c8..99a0a93 100644 --- a/runtime/tests/realloc_may_leave_test.ts +++ b/runtime/tests/realloc_may_leave_test.ts @@ -15,7 +15,13 @@ import { newStats, type ResolvedOptions, } from "../src/exec/boundary.ts"; -import { ComponentInstanceState, Store } from "../src/task/mod.ts"; +import { + ComponentInstanceState, + currentTask, + currentThread, + Store, + type Thread, +} from "../src/task/mod.ts"; import type { FuncType } from "../src/cabi/types.ts"; import { LiftLowerContext, mkCanonicalOptions } from "../src/cabi/mod.ts"; import { adaptHostFunction } from "../src/exec/host_settlement.ts"; @@ -141,6 +147,35 @@ Deno.test("#147: host-entry param lowering runs realloc inside the may_leave win ); }); +Deno.test("realloc has a fresh synchronous task and both context slots", () => { + const h = mkHarness(); + let outer: Thread | undefined; + h.duringRealloc = () => { + const nested = currentThread(); + assertEq(nested === outer, false); + assertEq(currentTask().inst === h.inst, true); + assertEq(currentTask().ft.async, false); + assertEq(nested.storage, [0, 0]); + nested.storage[0] = 99; + nested.storage[1] = 100; + }; + const lifted = createLiftedFunction({ + name: "takes-string", + ft: TAKES_STRING, + opts: h.mkOpts({ coreType: { params: ["i32", "i32"], results: [] } }), + core: () => { + outer = currentThread(); + assertEq(outer.storage, [0, 0]); + return []; + }, + stats: newStats(), + }); + + lifted("fresh"); + assertEq(outer?.storage, [0, 0], "realloc slots do not leak into callee"); + assertEq([...h.inst.threads].length, 0, "nested realloc task retired"); +}); + Deno.test("#147: a host-entry realloc that lowers an import traps", () => { const h = mkHarness(); const importCall = createLoweredImport({ diff --git a/runtime/tests/resource_lender_park_settle_test.ts b/runtime/tests/resource_lender_park_settle_test.ts index a4ed2ad..36c6274 100644 --- a/runtime/tests/resource_lender_park_settle_test.ts +++ b/runtime/tests/resource_lender_park_settle_test.ts @@ -89,6 +89,7 @@ function mkHarness(): Harness { resultTypes: () => [] as ValType[], resultTypesForTuple: () => null, callback: (_i: number) => null, + postReturn: (_i: number) => null, memoryToken: () => null, stats: newStats(), prepared, diff --git a/runtime/tests/resource_lender_unwind_test.ts b/runtime/tests/resource_lender_unwind_test.ts index 45528d7..26fa41c 100644 --- a/runtime/tests/resource_lender_unwind_test.ts +++ b/runtime/tests/resource_lender_unwind_test.ts @@ -64,6 +64,7 @@ function mkHarness(postReturn: (() => void) | null = null): Harness { resultTypes: () => [] as ValType[], resultTypesForTuple: () => null, callback: (_i: number) => postReturn, + postReturn: (_i: number) => postReturn, memoryToken: () => null, stats: newStats(), prepared, diff --git a/runtime/tests/stream_delivery_delta_test.ts b/runtime/tests/stream_delivery_delta_test.ts new file mode 100644 index 0000000..48d5eb6 --- /dev/null +++ b/runtime/tests/stream_delivery_delta_test.ts @@ -0,0 +1,380 @@ +// Focused regression coverage for the stream/future delivery delta authorized +// from WebAssembly/component-model#719 head 35e9769957627c2, plus the two +// independently identified Wasmtime-row bugs (payloadless maximum counts and +// error-context result-area bounds). + +import { assertEq } from "./support/asserts.ts"; +import { Trap } from "../src/cabi/mod.ts"; +import { + BLOCKED, + createWaitableSetPoll, + createWaitableSetWait, +} from "../src/intrinsics/async_builtins.ts"; +import { + createErrorContextDebugMessage, + createFutureDropReadable, + createFutureNew, + createFutureRead, + createFutureWrite, + createStreamCancelRead, + createStreamCancelWrite, + createStreamDropReadable, + createStreamDropWritable, + createStreamNew, + createStreamRead, + createStreamWrite, + type StreamTrampolineContext, +} from "../src/intrinsics/stream_builtins.ts"; +import type { ResolvedOptions } from "../src/exec/boundary.ts"; +import { + BUFFER_MAX_LENGTH, + ComponentInstanceState, + CopyResult, + CopyState, + ErrorContext, + EventCode, + popCurrentThread, + pushCurrentThread, + Store, + Task, + Thread, + WaitableSet, + type WritableFutureEnd, + type WritableStreamEnd, +} from "../src/task/mod.ts"; + +function assert(cond: boolean, message: string): asserts cond { + if (!cond) throw new Error(`assertion failed: ${message}`); +} + +function assertTrap(fn: () => unknown, text: string): void { + try { + fn(); + } catch (e) { + assert(e instanceof Trap, `expected Trap, got ${Deno.inspect(e)}`); + assert(String(e.message).includes(text), `unexpected trap: ${e.message}`); + return; + } + throw new Error("expected trap"); +} + +function memoryView(memory: WebAssembly.Memory) { + return { + addrType: "i32" as const, + get bytes() { + return new Uint8Array(memory.buffer); + }, + get view() { + return new DataView(memory.buffer); + }, + get length() { + return memory.buffer.byteLength; + }, + ptrType: () => "i32" as const, + ptrSize: () => 4 as const, + }; +} + +function fixture(kind: "stream" | "future", elem: { kind: "u8" } | null) { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + const memory = new WebAssembly.Memory({ initial: 1 }); + const opts: ResolvedOptions = { + stringEncoding: "utf8", + // deno-lint-ignore no-explicit-any + memory: memoryView(memory) as any, + realloc: null, + postReturn: null, + callback: null, + async: true, + cancellable: false, + coreType: { params: [], results: [] }, + instance: inst, + }; + const ctx = { + componentInstance: () => inst, + options: () => opts, + streamElem: () => elem, + futureElem: () => elem, + resultTypes: () => [], + suspensionMode: "plain" as const, + } as unknown as StreamTrampolineContext; + const task = new Task( + { params: [], results: [], async: true }, + { async_: true, callback: true, stringEncoding: "utf8", memory: null }, + inst, + () => [], + () => {}, + ); + const thread = new Thread(task, (function* () {})()); + const newPair = kind === "stream" + ? createStreamNew({ streamTable: 0 }, ctx, inst) + : createFutureNew({ futureTable: 0 }, ctx, inst); + const packed = newPair() as bigint; + const ri = Number(packed & 0xffff_ffffn); + const wi = Number(packed >> 32n); + return { + inst, + memory, + ctx, + ri, + wi, + run(fn: () => T): T { + pushCurrentThread(thread); + try { + return fn(); + } finally { + popCurrentThread(thread); + } + }, + }; +} + +for (const pending of ["read", "write"] as const) { + for ( + const scenario of [ + { name: "zero", originalLength: 0, transfers: [] }, + { name: "exact-full", originalLength: 4, transfers: [4] }, + { name: "multiple-partial", originalLength: 8, transfers: [3, 2, 3] }, + ] as const + ) { + Deno.test(`PR719 stream ${pending} event observes peer drop after detaching ${scenario.name} buffer`, () => { + const f = fixture("stream", { kind: "u8" }); + const read = createStreamRead( + { streamTable: 0, options: 0 }, + f.ctx, + f.inst, + ); + const write = createStreamWrite( + { streamTable: 0, options: 0 }, + f.ctx, + f.inst, + ); + const originalEnd = f.inst.handles.get( + pending === "read" ? f.ri : f.wi, + ) as WritableStreamEnd; + f.run(() => { + const first = pending === "read" + ? read(f.ri, 64, scenario.originalLength) + : write(f.wi, 0, scenario.originalLength); + assertEq(first, BLOCKED); + for (const n of scenario.transfers) { + const arriving = pending === "read" + ? write(f.wi, 0, n) + : read(f.ri, 64, n); + assertEq((arriving as number) & 0xf, CopyResult.COMPLETED); + } + // The opposite end starts one more operation. Because the original + // buffer is now full (including the zero-length case), this detaches + // that original buffer and leaves the new peer operation parked. + const peerPark = pending === "read" + ? write(f.wi, 0, 1) + : read(f.ri, 64, 1); + assertEq(peerPark, BLOCKED); + }); + assertEq(originalEnd.hasPendingEvent(), true); + + // Cancel the newly parked peer operation, consume its cancellation, and + // then drop that peer through the real canonical intrinsics. The + // original event remains pending throughout. + const cancelPeer = pending === "read" + ? createStreamCancelWrite( + { streamTable: 0, async: true }, + f.ctx, + f.inst, + ) + : createStreamCancelRead( + { streamTable: 0, async: true }, + f.ctx, + f.inst, + ); + const peerIndex = pending === "read" ? f.wi : f.ri; + assertEq( + (f.run(() => cancelPeer(peerIndex)) as number) & 0xf, + CopyResult.CANCELLED, + ); + const dropPeer = pending === "read" + ? createStreamDropWritable({ streamTable: 0 }, f.ctx, f.inst) + : createStreamDropReadable({ streamTable: 0 }, f.ctx, f.inst); + f.run(() => dropPeer(peerIndex)); + + // Wait and poll each exercise a real event-consumption path. + const set = new WaitableSet(); + const seti = f.inst.handles.add(set); + originalEnd.join(set); + const consume = pending === "read" + // The fixture context includes both stream and async methods. + // deno-lint-ignore no-explicit-any + ? createWaitableSetWait({ options: 0 }, f.ctx as any, f.inst) + // deno-lint-ignore no-explicit-any + : createWaitableSetPoll({ options: 0 }, f.ctx as any, f.inst); + const code = f.run(() => consume(seti, 128)) as number; + const payload = new DataView(f.memory.buffer).getUint32(132, true); + assertEq( + code, + pending === "read" ? EventCode.STREAM_READ : EventCode.STREAM_WRITE, + ); + assertEq(payload & 0xf, CopyResult.DROPPED); + assertEq(payload >>> 4, scenario.originalLength); + assertEq(originalEnd.state, CopyState.DONE); + }); + } +} + +Deno.test("PR719 stream peer drop takes precedence over CM-3 cancel remap", () => { + const f = fixture("stream", { kind: "u8" }); + const read = createStreamRead({ streamTable: 0, options: 0 }, f.ctx, f.inst); + const write = createStreamWrite( + { streamTable: 0, options: 0 }, + f.ctx, + f.inst, + ); + const cancel = createStreamCancelWrite( + { streamTable: 0, async: true }, + f.ctx, + f.inst, + ); + f.run(() => { + assertEq(write(f.wi, 0, 4), BLOCKED); + assertEq((read(f.ri, 64, 4) as number) >>> 4, 4); + }); + const writer = f.inst.handles.get(f.wi) as WritableStreamEnd; + // Detach the completed original write by parking a fresh peer read, cancel + // that read, then drop the peer through its intrinsic. + assertEq(f.run(() => read(f.ri, 64, 1)), BLOCKED); + const cancelRead = createStreamCancelRead( + { streamTable: 0, async: true }, + f.ctx, + f.inst, + ); + assertEq( + (f.run(() => cancelRead(f.ri)) as number) & 0xf, + CopyResult.CANCELLED, + ); + const dropRead = createStreamDropReadable( + { streamTable: 0 }, + f.ctx, + f.inst, + ); + f.run(() => dropRead(f.ri)); + const result = f.run(() => cancel(f.wi)) as number; + assertEq(result & 0xf, CopyResult.DROPPED); + assertEq(result >>> 4, 4); + assertEq(writer.state, CopyState.DONE); +}); + +Deno.test("PR719 does not retroactively rewrite a delivered stream event", () => { + const f = fixture("stream", { kind: "u8" }); + const read = createStreamRead({ streamTable: 0, options: 0 }, f.ctx, f.inst); + const write = createStreamWrite( + { streamTable: 0, options: 0 }, + f.ctx, + f.inst, + ); + const result = f.run(() => { + assertEq(write(f.wi, 0, 1), BLOCKED); + return read(f.ri, 64, 1) as number; + }); + assertEq(result & 0xf, CopyResult.COMPLETED); + const dropRead = createStreamDropReadable( + { streamTable: 0 }, + f.ctx, + f.inst, + ); + f.run(() => dropRead(f.ri)); + assertEq(result & 0xf, CopyResult.COMPLETED); +}); + +Deno.test("PR719 future CANCELLED upgrades to DROPPED, while COMPLETED stays stable", () => { + const cancelled = fixture("future", null); + const write = createFutureWrite( + { futureTable: 0, options: 0 }, + cancelled.ctx, + cancelled.inst, + ); + assertEq(cancelled.run(() => write(cancelled.wi, 0)), BLOCKED); + const cancelledEnd = cancelled.inst.handles.get( + cancelled.wi, + ) as WritableFutureEnd; + // The public cancel intrinsic consumes an immediately-produced event in the + // same call, so this delivery-window state is necessarily driven at the + // CopyEnd seam: arm cancellation without consuming it, then lose the peer. + cancelledEnd.state = CopyState.CANCELLING_COPY; + cancelledEnd.shared.cancel(); + cancelledEnd.shared.drop(); + const [, , dropped] = cancelledEnd.getPendingEvent(); + assertEq(dropped, CopyResult.DROPPED); + assertEq(cancelledEnd.state, CopyState.DONE); + + const completed = fixture("future", null); + const read2 = createFutureRead( + { futureTable: 0, options: 0 }, + completed.ctx, + completed.inst, + ); + const write2 = createFutureWrite( + { futureTable: 0, options: 0 }, + completed.ctx, + completed.inst, + ); + assertEq(completed.run(() => write2(completed.wi, 0)), BLOCKED); + assertEq(completed.run(() => read2(completed.ri, 0)), CopyResult.COMPLETED); + const completedEnd = completed.inst.handles.get( + completed.wi, + ) as WritableFutureEnd; + const dropRead = createFutureDropReadable( + { futureTable: 0 }, + completed.ctx, + completed.inst, + ); + completed.run(() => dropRead(completed.ri)); + const [, , kept] = completedEnd.getPendingEvent(); + assertEq(kept, CopyResult.COMPLETED); + assertEq(completedEnd.state, CopyState.DONE); +}); + +Deno.test("payloadless builtins transfer Buffer.MAX_LENGTH without allocation", () => { + const f = fixture("stream", null); + const read = createStreamRead({ streamTable: 0, options: 0 }, f.ctx, f.inst); + const write = createStreamWrite( + { streamTable: 0, options: 0 }, + f.ctx, + f.inst, + ); + f.run(() => { + assertEq(read(f.ri, 0, BUFFER_MAX_LENGTH), BLOCKED); + assertEq(write(f.wi, 0, BUFFER_MAX_LENGTH), 0xfffffff0); + }); +}); + +Deno.test("error-context debug-message validates its result area before realloc", () => { + const store = new Store(); + const inst = new ComponentInstanceState(0, store); + const memory = new WebAssembly.Memory({ initial: 1 }); + let reallocs = 0; + const opts: ResolvedOptions = { + stringEncoding: "utf8", + // deno-lint-ignore no-explicit-any + memory: memoryView(memory) as any, + realloc: () => (_old, _oldSize, _align, _newSize) => { + reallocs++; + return 100; + }, + postReturn: null, + callback: null, + async: false, + cancellable: false, + coreType: { params: [], results: [] }, + instance: inst, + }; + const ctx = { + options: () => opts, + } as unknown as StreamTrampolineContext; + const debug = createErrorContextDebugMessage({ options: 0 }, ctx, inst); + const i = inst.handles.add(new ErrorContext("a")); + debug(i, 65528); + assertEq(reallocs, 1); + assertEq(new DataView(memory.buffer).getUint32(65528, true), 100); + assertTrap(() => debug(i, 65532), "invalid debug message pointer"); + assertEq(reallocs, 1, "invalid result area traps before realloc"); +}); diff --git a/runtime/tests/subtask_cancel_sync_waiter_window_test.ts b/runtime/tests/subtask_cancel_sync_waiter_window_test.ts index 68dd8ab..fd664e1 100644 --- a/runtime/tests/subtask_cancel_sync_waiter_window_test.ts +++ b/runtime/tests/subtask_cancel_sync_waiter_window_test.ts @@ -154,6 +154,9 @@ function startFactSubtask( } return wait; }, + postReturn: () => { + throw new Error("unexpected post-return"); + }, memoryToken: () => null, stats: newStats(), suspensionMode: "jspi", diff --git a/upstream-component-model-repo-findings.md b/upstream-component-model-repo-findings.md index 3da3830..bcf18b9 100644 --- a/upstream-component-model-repo-findings.md +++ b/upstream-component-model-repo-findings.md @@ -79,6 +79,13 @@ adjudicates otherwise. The current pin still returns the pending payload unchanged in `cancel_copy`. **Found:** 2026-08-08, implementing the stream copy protocol. +This finding remains separate from the provisional drop-delivery rule adopted +from component-model PR #719 at head +`35e9769957627c2bee5cd445b998b08b3c652c86`. That rule upgrades an undelivered +event when its peer is found dropped at consumption; CM-3 instead remaps an +unobserved stream `COMPLETED|count` to `CANCELLED|count` when this end explicitly +cancels. A peer-drop `DROPPED|count` therefore wins and is not remapped. + ### Evidence `definitions.py` `cancel_copy`: