diff --git a/contracts/embedder-api.md b/contracts/embedder-api.md index d1a4ce8..3352581 100644 --- a/contracts/embedder-api.md +++ b/contracts/embedder-api.md @@ -14,7 +14,12 @@ translation envelope as the build-time artifact of one stream/future idempotent (pass-through round trips — host→guest→host — hand back the same handle machinery instead of asserting), legalizes host↔host rendezvous for every element type, and -pins u8 stream chunks as `Uint8Array` in both directions.** This document supersedes `descriptor-ir.md`'s interim +pins u8 stream chunks as `Uint8Array` in both directions; amendment A6 +(2026-08-11) ships the wasi-shims parking kernel always-on (§"WASI +examination", renumbered from a colliding second "A5"); amendment A7 +(2026-08-11) makes component faults loud on host stream/future +operations (`PeerTrappedError`, never a hang or a fake end-of-stream) +and limits host ends to one in-flight operation per direction.** This document supersedes `descriptor-ir.md`'s interim "host value mapping" table as the destination for host-facing value shapes. The runtime's *raw* boundary (`instance.exports`, `HostImports`) keeps the `definitions.py` interpreter shapes as an **internal** surface; the @@ -197,6 +202,10 @@ class WitError extends Error { constructor(payload: E, message?: string); } class Trap extends Error { … } // existing; component-fatal, never a value +class PeerTrappedError extends Error { // A7: a stream/future op whose peer instance trapped + readonly cause: unknown; // chains to the Trap + readonly progress?: number; // write ops: elements delivered before the fault +} ``` - **Guest export with `result`**: the call resolves to `T` on ok and @@ -380,6 +389,29 @@ class DroppedError extends Error { … } // awaiting a dropped future rejects underneath; the conventions layer exposes them as `Stream.create(): { stream: Stream, writer: StreamWriter }` with `write`/`writeAll`/`cancelWrite`/`close`. +- **Component faults are loud on stream/future operations** (amendment + A7). When the component instance holding the peer end traps, its live + ends are retired: a parked host `read`/`write`/`writeAll`/future-await + **rejects with `PeerTrappedError`** (`cause` chains to the trap; a + write's `progress` reports elements delivered before the fault), and so + does any operation started afterwards. A fault is never presented as a + clean end-of-stream or a bare `DroppedError` — the same + no-wrong-data-as-success rule the producer direction has + (`StreamProducerError`) — with one precision: an operation that + genuinely COMPLETED before the trap keeps its result (a full write, a + read that copied data), and the fault surfaces on the export call and + on the handle's next operation. A trapping host **import** drops the + lifted stream/future arguments it abandoned, so their peers settle with + the truthful short count / end-of-stream. Only embedder negligence — + lowering a host end and never acting on it — still hangs, as documented + since v0.2. +- **One in-flight operation per host end, per direction** (amendment A7): + a second `write` while one is parked (or a second `read`, or a second + future operation) throws a `TypeError` synchronously — the host-side + spelling of the `CopyEnd` busy trap. Reading while a write is parked on + the same stream stays legal (they are different ends). Previously the + second operation could "rendezvous" against the first one's parked + buffer and report data as taken by a peer that never existed. ## Module wiring and instantiation @@ -463,7 +495,7 @@ signatures for the representative slice: **wasi:io@0.2.x pollable + streams** (p2): `pollable.block()`, `poll()` and `blocking-read`/`blocking-write-and-flush` are **sync** WIT functions that must park — the one p2 idiom that fights a JS host. The shim package -ships the PARKING KERNEL, always on (amendment A5, 2026-08-11; +ships the PARKING KERNEL, always on (amendment A6, 2026-08-11; supersedes the original three-tier ruling and its "never (c) in this package" mission line — the polymorph-iroh upstream-iroh consumer class genuinely parks, which the always-ready stubs turned into a livelock): diff --git a/examples/guests/stream-pass/Cargo.lock b/examples/guests/stream-pass/Cargo.lock index 75d4281..42fa214 100644 --- a/examples/guests/stream-pass/Cargo.lock +++ b/examples/guests/stream-pass/Cargo.lock @@ -26,6 +26,67 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" +[[package]] +name = "futures" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a88cf1f829d945f548cf8fec32c61b1f202b6d93b45848602fc02af4b12ad218" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" + +[[package]] +name = "futures-io" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" + +[[package]] +name = "futures-sink" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" + +[[package]] +name = "futures-task" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b231ed28831efb4a61a08580c4bc233ec56bc009f4cd8f52da2c3cb97df0c109" + +[[package]] +name = "futures-util" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a77a90a256fce34da66415271e30f94ee91c57b04b8a2c042d9cf3220179deaa" +dependencies = [ + "futures-core", + "futures-sink", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "guest-stream-pass" version = "0.1.0" @@ -101,6 +162,12 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "prettyplease" version = "0.2.37" @@ -177,6 +244,12 @@ dependencies = [ "zmij", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "syn" version = "2.0.119" @@ -246,6 +319,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a301904d6657d6364c758d869e5389d05d393b16d5b65db60b4f03cbe71bb80d" dependencies = [ "bitflags", + "futures", "wit-bindgen-rust-macro", ] diff --git a/examples/guests/stream-pass/Cargo.toml b/examples/guests/stream-pass/Cargo.toml index 7d74dbb..871432a 100644 --- a/examples/guests/stream-pass/Cargo.toml +++ b/examples/guests/stream-pass/Cargo.toml @@ -9,8 +9,9 @@ crate-type = ["cdylib"] [dependencies] # `async` is a default feature; it provides the guest-side task executor and -# stream/future support. No `async-spawn`: these exports never pump anything. -wit-bindgen = "=0.60.0" +# stream/future support. `async-spawn` backs `open-then-trap`'s background +# writer; the pass-through exports themselves never pump anything. +wit-bindgen = { version = "=0.60.0", features = ["async-spawn"] } [profile.release] opt-level = "s" diff --git a/examples/guests/stream-pass/src/lib.rs b/examples/guests/stream-pass/src/lib.rs index 798dc3b..275eaf4 100644 --- a/examples/guests/stream-pass/src/lib.rs +++ b/examples/guests/stream-pass/src/lib.rs @@ -10,7 +10,7 @@ wit_bindgen::generate!({ async: true, }); -use wit_bindgen::rt::async_support::StreamReader; +use wit_bindgen::rt::async_support::{FutureReader, StreamReader}; struct Component; @@ -26,6 +26,34 @@ impl Guest for Component { async fn pass_through_text(input: StreamReader) -> StreamReader { input } + + async fn consume_then_trap(mut input: StreamReader, count: u32) { + for _ in 0..count { + let _ = input.next().await; + } + core::arch::wasm32::unreachable() + } + + async fn open_then_trap(n: u32) -> StreamReader { + let (mut writer, reader) = wit_stream::new(); + wit_bindgen::rt::async_support::spawn_local(async move { + let _ = writer.write(vec![7u8; n as usize]).await; + core::arch::wasm32::unreachable() + }); + reader + } + + async fn future_then_trap(mut gate: StreamReader) -> FutureReader { + let (writer, reader) = wit_future::new(|| 0u32); + wit_bindgen::rt::async_support::spawn_local(async move { + // Park until the host releases the gate — giving it time to park + // a read on the future — then trap without ever writing. + let _ = gate.next().await; + let _hold = writer; + core::arch::wasm32::unreachable() + }); + reader + } } export!(Component); diff --git a/examples/guests/stream-pass/wit/world.wit b/examples/guests/stream-pass/wit/world.wit index 4adfc80..9646ef2 100644 --- a/examples/guests/stream-pass/wit/world.wit +++ b/examples/guests/stream-pass/wit/world.wit @@ -15,4 +15,16 @@ world stream-pass { export forward: async func(input: stream) -> u64; /// Result-position pass-through for a non-numeric element type. export pass-through-text: async func(input: stream) -> stream; + + /// Reads `count` elements off `input`, then traps (unreachable) — the #66 + /// shape: the stream's readable end dies inside this instance's poisoned + /// handle table while host writes are still parked. + export consume-then-trap: async func(input: stream, count: u32); + /// Returns a fresh stream, writes `n` bytes from a background task, then + /// traps — the write end dies in the poisoned table while the host reads. + export open-then-trap: async func(n: u32) -> stream; + /// Returns a fresh future whose write end dies unwritten in the poisoned + /// table: the background task reads one gate byte (so the host can first + /// park a read on the future), then traps before delivering the value. + export future-then-trap: async func(gate: stream) -> future; } diff --git a/ports/webrtc/deno.lock b/ports/webrtc/deno.lock index bb143de..6abb716 100644 --- a/ports/webrtc/deno.lock +++ b/ports/webrtc/deno.lock @@ -592,9 +592,8 @@ }, "links": { "jsr:@deltic/ct-runner@0.1.0": {}, - "jsr:@deltic/examples@0.0.0": {}, - "jsr:@deltic/harness": {}, - "jsr:@deltic/runtime": {}, + "jsr:@deltic/runtime@0.1.0": {}, + "jsr:@deltic/translator@0.1.0": {}, "jsr:@deltic/wasi-shims@0.1.0": {} } } diff --git a/ports/websocket/deno.lock b/ports/websocket/deno.lock index 94cd957..15efa4a 100644 --- a/ports/websocket/deno.lock +++ b/ports/websocket/deno.lock @@ -18,9 +18,8 @@ "workspace": { "links": { "jsr:@deltic/ct-runner@0.1.0": {}, - "jsr:@deltic/examples@0.0.0": {}, - "jsr:@deltic/harness": {}, - "jsr:@deltic/runtime": {}, + "jsr:@deltic/runtime@0.1.0": {}, + "jsr:@deltic/translator@0.1.0": {}, "jsr:@deltic/wasi-shims@0.1.0": {} } } diff --git a/runtime/src/embedder/errors.ts b/runtime/src/embedder/errors.ts index 133a99a..73ac980 100644 --- a/runtime/src/embedder/errors.ts +++ b/runtime/src/embedder/errors.ts @@ -10,7 +10,9 @@ // trap, so the defensive wrapper is unnecessary by construction. // * `Trap` — component-fatal, never a value (re-exported from cabi). // * `DroppedError` — awaiting a future whose write end dropped without a -// value (R-fix review note 4). +// value (R-fix review note 4). Its uncomely sibling `PeerTrappedError` +// (below) is a drop that happened because the peer's instance trapped — +// branded separately so a fault is never mistaken for a clean end. export { Trap } from "../cabi/trap.ts"; @@ -61,6 +63,34 @@ export class InvalidHandleError extends Error { } } +/** + * A stream/future operation whose peer end died in a trap-poisoned component + * instance (#66; contracts/embedder-api.md amendment A7). + * + * Discriminated from `DroppedError` on purpose: a clean drop is a normal + * outcome (end-of-stream, "no value"), while a poisoned peer means the + * component faulted — resolving the operation as if the stream simply ended + * would be wrong data reported as success, the same shape + * `StreamProducerError` exists to prevent in the other direction. `cause` is + * the recorded poisoning failure (its own `cause` is the underlying `Trap`); + * `progress` is how many elements a write had delivered before the peer died. + */ +export class PeerTrappedError extends Error { + override readonly cause: unknown; + readonly progress?: number; + + constructor(where: string, cause: unknown, progress?: number) { + super( + `${where}: the peer component instance trapped, so this ` + + `stream/future operation can never complete — ` + + `${cause instanceof Error ? cause.message : String(cause)}`, + ); + this.name = "PeerTrappedError"; + this.cause = cause; + if (progress !== undefined) this.progress = progress; + } +} + function describePayload(p: unknown): string { if (p === null || p === undefined) return String(p); if (typeof p === "object" && "tag" in (p as Record)) { diff --git a/runtime/src/embedder/instantiate.ts b/runtime/src/embedder/instantiate.ts index 38add8e..19ec410 100644 --- a/runtime/src/embedder/instantiate.ts +++ b/runtime/src/embedder/instantiate.ts @@ -10,7 +10,7 @@ // Governing contract: contracts/embedder-api.md (all sections). Secondary: // contracts/plan-format.md for the wire shapes read here. -import type { WirePlan, WireExport } from "../plan/format.ts"; +import type { WireExport, WirePlan } from "../plan/format.ts"; import type { LoadedPlan } from "../plan/loader.ts"; import { loadEnvelope, loadPlan, PlanError } from "../plan/loader.ts"; import type { FuncType, ResourceTypeInfo, ValType } from "../cabi/types.ts"; @@ -19,8 +19,8 @@ import { Trap } from "../cabi/trap.ts"; import { type ComponentHandle, CONSTRUCTOR_SYNC_ENTRY, - hostResourceType, type HostImports, + hostResourceType, instantiateComponent, } from "../exec/mod.ts"; import { camelCase, parseLeafName, pascalCase } from "./casing.ts"; @@ -45,7 +45,7 @@ import { type ValueBridge, } from "./values.ts"; import { ImportResolver } from "./version.ts"; -import { type ElemCodec, Future } from "./streams.ts"; +import { type ElemCodec, Future, Stream } from "./streams.ts"; /** Per-element codec for a `future` returned in function-result position. */ function elementCodec( @@ -146,7 +146,12 @@ type RawFn = (...a: unknown[]) => unknown; /** How a resource type is implemented, keyed by `ResourceIndex`. */ type Binding = | { kind: "guest"; name: string; cls?: unknown } - | { kind: "host"; name: string; registry: HostResourceRegistry; cls: unknown }; + | { + kind: "host"; + name: string; + registry: HostResourceRegistry; + cls: unknown; + }; /** * Instantiate a component behind the embedder conventions. @@ -361,7 +366,9 @@ class Facade { #bindHostResources(): void { const importedResources = this.artifacts.plan.importedResources ?? []; for (const p of this.#pendingHostResources) { - const at = importedResources.findIndex((ir) => ir.import === p.importIndex); + const at = importedResources.findIndex((ir) => + ir.import === p.importIndex + ); if (at < 0) continue; this.#bindings.set(at, { kind: "host", @@ -549,7 +556,10 @@ class Facade { }[] = []; /** The JS call a lifted import leaf dispatches to. */ - #dispatcher(leaf: ImportLeaf, provider: unknown): (args: unknown[]) => unknown { + #dispatcher( + leaf: ImportLeaf, + provider: unknown, + ): (args: unknown[]) => unknown { const m = leaf.member; if (m.form === "plain") { const fn = leaf.path.length === 0 @@ -663,14 +673,24 @@ class Facade { } return fromHost(v, resultType, o); }; - const fail = (e: unknown): ComponentValue => { - if (e instanceof Trap) throw e; + const fail = (e: unknown, args: unknown[]): ComponentValue => { if (e instanceof WitError && isResult) { const rt = resultType as ValType & { kind: "result" }; return { error: rt.error === null ? null : fromHost(e.payload, rt.error, o), }; } + // Every remaining branch traps the component. The import's lifted + // stream/future arguments were transferred to the host when the params + // were converted (the guest's ends are gone), and a trapping import is + // a declared host bug — nothing owns them anymore, so drop them here: + // a peer parked on one (a host writer feeding the stream this import + // just received, the #66 E2 shape) settles with the truthful "reader + // went away" instead of hanging forever. The err-VALUE branch above + // deliberately does NOT do this: a fallible import returning err is a + // normal outcome whose implementation may retain the handles. + releaseAsyncArgs(args); + if (e instanceof Trap) throw e; if (e instanceof WitError) { throw new Trap( `${where} threw a WitError, but its WIT type has no err side; ` + @@ -694,7 +714,7 @@ class Facade { out = dispatch(args); } catch (e) { scope.end(); - return fail(e); + return fail(e, args); } if (isThenable(out)) { return (out as PromiseLike).then( @@ -704,7 +724,7 @@ class Facade { }, (e) => { scope.end(); - return fail(e); + return fail(e, args); }, ); } @@ -872,7 +892,9 @@ class Facade { s, rt, (fn, params, results, where, args) => - this.#wrapExportFn(fn, { params, results }, where)(...args) as Promise< + this.#wrapExportFn(fn, { params, results }, where)( + ...args, + ) as Promise< unknown >, (args, params, where) => @@ -1057,8 +1079,29 @@ function isThenable(v: unknown): boolean { typeof (v as { then: unknown }).then === "function"; } +/** + * Drop the lifted stream/future arguments a trapping import abandoned (#66). + * Top-level parameters only: those are the shapes whose peers park host + * operations; a stream nested inside a record is exotic enough to leave to + * the negligence rules. Uses the teardown drop, not the plain one — the + * calling instance is about to be poisoned by this very trap, and a DROPPED + * notification must not queue a phantom event into its waitables (review + * B2; see task/streams.ts `dropSharedForTeardown`). + */ +function releaseAsyncArgs(args: unknown[]): void { + for (const a of args) { + if (a instanceof Stream || a instanceof Future) { + try { + a.dropForTeardown(); + } catch { + // Best-effort teardown: the component is already trapping, and that + // trap — not a secondary drop failure — is the error to surface. + } + } + } +} + function describeThrow(e: unknown): string { if (e instanceof Error) return `${e.name}: ${e.message}`; return describe(e); } - diff --git a/runtime/src/embedder/mod.ts b/runtime/src/embedder/mod.ts index db9a64c..51aa493 100644 --- a/runtime/src/embedder/mod.ts +++ b/runtime/src/embedder/mod.ts @@ -23,6 +23,7 @@ export { DroppedError, InvalidHandleError, NameCollisionError, + PeerTrappedError, Trap, WitError, } from "./errors.ts"; diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts index 8eef313..fd82a5b 100644 --- a/runtime/src/embedder/streams.ts +++ b/runtime/src/embedder/streams.ts @@ -19,8 +19,13 @@ import { hostStream, hostStreamFor, } from "../exec/host_streams.ts"; -import { CopyResult, ErrorContext as InternalErrorContext } from "../task/mod.ts"; -import { DroppedError } from "./errors.ts"; +import { + CopyResult, + dropSharedForTeardown, + ErrorContext as InternalErrorContext, + poisonFailureOf, +} from "../task/mod.ts"; +import { DroppedError, PeerTrappedError } from "./errors.ts"; /** `Chunk` is a `Uint8Array`; every other element type chunks as `T[]`. */ export type Chunk = T extends number ? Uint8Array | T[] : T[]; @@ -111,9 +116,28 @@ function reportProducerFailure( } /** @internal — raise a recorded producer failure, if any. */ -function throwIfFailed(value: unknown): void { +function throwIfFailed(value: unknown, where = "stream"): void { const e = producerFailures.get(value as object); if (e !== undefined) throw e; + throwIfPeerTrapped(value, where); +} + +/** + * @internal — raise the recorded poisoning failure, if any (#66, amendment + * A7). Pre-op: an operation started after the peer's instance trapped must + * reject rather than park forever. Post-await (with the op's outcome in + * hand): an operation the retirement walk settled DROPPED-shaped must reject + * rather than fake a clean end — but an op that genuinely COMPLETED before + * the trap keeps its result (the fault still surfaces on the export call, + * and on this handle's next operation). + */ +function throwIfPeerTrapped( + value: unknown, + where: string, + progress?: number, +): void { + const p = poisonFailureOf(value); + if (p !== undefined) throw new PeerTrappedError(where, p, progress); } /** True for `stream` / `future`, whose chunks are `Uint8Array`. */ @@ -147,7 +171,10 @@ export class Stream { } /** Wrap a freshly created host-owned stream of a known element type. */ - static fromHostStream(host: HostStream, codec: ElemCodec): Stream { + static fromHostStream( + host: HostStream, + codec: ElemCodec, + ): Stream { return new Stream(host, codec); } @@ -221,10 +248,16 @@ export class Stream { /** Low-level read: up to `max` elements; an empty chunk means end-of-stream. */ async read(max: number): Promise> { const host = this.#require(); - throwIfFailed(host.value); + const where = this.#codec?.where ?? "stream read"; + throwIfFailed(host.value, where); const raw = await host.readable.read(max) as unknown as | ComponentValue[] | Uint8Array; + // An empty chunk normally means clean end-of-stream; when the peer's + // instance trapped it means the retirement walk settled us — reject + // instead of faking EOS (amendment A7). A non-empty chunk was really + // copied before the trap and is delivered; the next read rejects. + if (raw.length === 0) throwIfPeerTrapped(host.value, where); return this.#chunk(raw); } @@ -256,6 +289,28 @@ export class Stream { this.#host?.readable.drop(); } + /** + * @internal — teardown after a trapping import abandoned this handle + * (#66, instantiate.ts `releaseAsyncArgs`). Unlike `drop()`, this goes + * through `dropSharedForTeardown`, whose parked-side discipline never + * wakes the about-to-be-poisoned caller (review B2: a plain drop queued a + * DROPPED event into the trapping instance's waitables, and a later + * driving loop asserted on the corpse). + * + * Known asymmetry (review advisory, non-blocking): unlike + * `readable.drop()`, this path does not close the wrapper's HostActivity + * arm when nothing was parked, so the arm can outlive the stream on an + * already-faulted store — at worst misreporting a later genuine deadlock + * as the documented hang, on a store that has already trapped. + */ + dropForTeardown(): void { + if (this.#dropped) return; + this.#dropped = true; + if (this.#host !== null) { + dropSharedForTeardown(this.#host.value as never); + } + } + [Symbol.dispose](): void { this.drop(); } @@ -311,20 +366,30 @@ export class StreamWriter { async write(values: Chunk): Promise { await this.#stream.whenBound(); const host = hostOf(this.#stream); - throwIfFailed(host.value); - return await host.writable.write( + const where = this.#stream.codec?.where ?? "stream write"; + throwIfFailed(host.value, where); + const n = await host.writable.write( packChunk(values, this.#stream.codec!) as unknown as T[], ); + // A short take normally means "re-offer later" / "reader done"; when the + // reader's instance trapped it means the retirement walk settled us — + // reject, carrying the delivered count (amendment A7). A full take + // genuinely completed before the trap and stays a success. + if (n < values.length) throwIfPeerTrapped(host.value, where, n); + return n; } /** Offer values until all are taken or the reader goes away. */ async writeAll(values: Chunk): Promise { await this.#stream.whenBound(); const host = hostOf(this.#stream); - throwIfFailed(host.value); - return await host.writable.writeAll( + const where = this.#stream.codec?.where ?? "stream write"; + throwIfFailed(host.value, where); + const n = await host.writable.writeAll( packChunk(values, this.#stream.codec!) as unknown as T[], ); + if (n < values.length) throwIfPeerTrapped(host.value, where, n); + return n; } cancelWrite(): void { @@ -387,7 +452,10 @@ export class Future implements PromiseLike { return new Future(h, Promise.resolve(h), codec); } - static fromHostFuture(host: HostFuture, codec: ElemCodec): Future { + static fromHostFuture( + host: HostFuture, + codec: ElemCodec, + ): Future { return new Future(host, Promise.resolve(host), codec); } @@ -446,8 +514,12 @@ export class Future implements PromiseLike { #read(): Promise { this.#settled ??= (async () => { - const { value, result } = await (await this.#hostP).readResult(); + const host = await this.#hostP; + const { value, result } = await host.readResult(); if (result !== CopyResult.COMPLETED) { + // A drop caused by the writer's instance trapping is a fault, not a + // "no value" outcome — brand it (#66, amendment A7). + throwIfPeerTrapped(host.value, this.#codec.where ?? "future read"); throw new DroppedError( result === CopyResult.CANCELLED ? "the future read was cancelled" @@ -476,6 +548,17 @@ export class Future implements PromiseLike { else void this.#hostP.then((h) => h.drop()); } + /** @internal — see `Stream.dropForTeardown` (#66). */ + dropForTeardown(): void { + if (this.#host !== null) { + dropSharedForTeardown(this.#host.value as never); + } else { + // A deferred future (still in flight) cannot be an import argument; + // fall back to the plain drop for completeness. + this.drop(); + } + } + [Symbol.dispose](): void { this.drop(); } diff --git a/runtime/src/exec/boundary.ts b/runtime/src/exec/boundary.ts index 03f4069..1fedd59 100644 --- a/runtime/src/exec/boundary.ts +++ b/runtime/src/exec/boundary.ts @@ -40,6 +40,7 @@ import { setResumingThread, packSubtaskResult, PendingCapability, + retireInstanceAsyncEnds, Store, Subtask, WaitableSet, @@ -1114,9 +1115,15 @@ export function createLiftedFunction(input: { * Only the entered set is affected; sibling instances stay usable, which * is why the lock is released per-instance rather than by poisoning a * whole store the way wasmtime does. + * + * Poisoned instances can never rendezvous again, so their handle tables' + * live stream/future ends are retired here (#66): parked host operations + * settle (DROPPED) instead of hanging forever, and the recorded failure + * lets the embedder layer reject them loudly. */ - const poison = (): void => { + const poison = (e: unknown): void => { entered = false; // consumed: the lock is now permanent + for (const i of enteredSet) retireInstanceAsyncEnds(i, e); }; /** @@ -1181,7 +1188,7 @@ export function createLiftedFunction(input: { } catch (e) { unwind(); if (isCapabilitySignal(e)) leave(); - else poison(); + else poison(e); throw e; } // The reentrance gate is released here, before the store is pumped: diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index 8010d2c..4a87407 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -45,6 +45,13 @@ // trap. That is the honest outcome — the component is not deadlocked, the // embedder simply has not done its half — and it matches how any other // unresolved Promise behaves in JS. +// +// The inverse case is NOT a hang (#66, embedder-api amendment A7): when the +// GUEST side dies — a trap poisons the instance holding the peer end — the +// poisoned table's ends are retired (task/streams.ts +// `retireInstanceAsyncEnds`), so a parked host operation settles DROPPED- +// shaped here and the conventions layer rejects it with `PeerTrappedError`. +// Only embedder negligence hangs; a component fault is always loud. import { assert_ } from "../cabi/trap.ts"; import { despecialize } from "../cabi/types.ts"; @@ -232,7 +239,9 @@ class HostActivity { const p = this.#promise, r = this.#resolve; this.#promise = null; this.#resolve = null; - if (p !== null && this.#store !== null) this.#store.pendingHostCalls.delete(p); + if (p !== null && this.#store !== null) { + this.#store.pendingHostCalls.delete(p); + } r?.(); this.#arm(); } @@ -358,7 +367,9 @@ class HostActivity { this.#closed = true; this.#promise = null; this.#resolve = null; - if (p !== null && this.#store !== null) this.#store.pendingHostCalls.delete(p); + if (p !== null && this.#store !== null) { + this.#store.pendingHostCalls.delete(p); + } r?.(); } } @@ -481,6 +492,20 @@ function mkStreamEnds( return { writable: { write(values: T[]): Promise { + // One in-flight operation per end — the host-side spelling of the + // `CopyEnd` busy trap guests get from the table. Without it a second + // write would find the FIRST write's buffer in the shared object's + // pending slot and "rendezvous" write-against-write, silently + // copying into the parked buffer's accumulation (observed as a + // write resolving `1` against a peer that no longer exists — the + // #66 repro). Reading while a write is parked stays legal: that is + // the pass-through data plane (two different ends). + if (parked.write) { + throw new TypeError( + "a write is already in flight on this stream's writable end; " + + "await it or cancelWrite() first", + ); + } const buf = new HostBuffer( shared.t, values as unknown as ComponentValue[], @@ -550,6 +575,15 @@ function mkStreamEnds( }, readable: { read(max: number): Promise { + // One in-flight operation per end — see the write() guard: a second + // read would rendezvous read-against-read with our own parked + // buffer. + if (parked.read) { + throw new TypeError( + "a read is already in flight on this stream's readable end; " + + "await it or cancelRead() first", + ); + } const buf = new HostBuffer(shared.t, null, max); return new Promise((resolve) => { parked.read = true; @@ -700,6 +734,14 @@ function mkFuture( }; const self: HostFuture = { write(v: T): Promise { + // One in-flight operation per wrapper — see mkStreamEnds' guards: a + // second op would rendezvous against our own parked buffer. + if (parked.any) { + throw new TypeError( + "an operation is already in flight on this future; " + + "await it or cancel() first", + ); + } // definitions.py `SharedFutureImpl.write` asserts `remain() == 1`: a // future carries exactly one element. const buf = new HostBuffer(shared.t, [v as unknown as ComponentValue], 1); @@ -714,6 +756,13 @@ function mkFuture( }); }, readResult(): Promise<{ value: T | undefined; result: CopyResult }> { + // One in-flight operation per wrapper — see write(). + if (parked.any) { + throw new TypeError( + "an operation is already in flight on this future; " + + "await it or cancel() first", + ); + } // definitions.py `SharedFutureImpl.read` asserts `not self.dropped`, so // a read after the write end went away must be answered here rather // than by tripping an internal assertion. diff --git a/runtime/src/intrinsics/fact_calls.ts b/runtime/src/intrinsics/fact_calls.ts index f05b109..e151fbb 100644 --- a/runtime/src/intrinsics/fact_calls.ts +++ b/runtime/src/intrinsics/fact_calls.ts @@ -77,6 +77,7 @@ import { currentTask, maybeCurrentTask, needsJspi, + notifyInstancePoisoned, packSubtaskResult, PendingCapability, Subtask, @@ -641,6 +642,12 @@ export function createSyncStartCall( // not — see the `isCapabilitySignal` note in exec/boundary.ts. if (e instanceof NeedsJspi || e instanceof PendingCapability) { prepared.calleeInst.leaveTo(prepared.callerInst); + } else { + // Retire the poisoned CALLEE's stream/future ends (#66): this is a + // bracket-break site like `Store.tick`'s, and the trap unwinds to a + // hooked site that walks only the CALLER's chain — a composed + // component's callee would otherwise strand its host peers. + notifyInstancePoisoned(prepared.calleeInst, e); } throw e; } @@ -813,6 +820,10 @@ export function createAsyncStartCall( // See the sync form above and `isCapabilitySignal` in exec/boundary.ts. if (e instanceof NeedsJspi || e instanceof PendingCapability) { prepared.calleeInst.leaveTo(prepared.callerInst); + } else { + // Bracket-break site — retire the poisoned callee's ends (#66), + // as in the sync form above. + notifyInstancePoisoned(prepared.calleeInst, e); } throw e; } diff --git a/runtime/src/task/scheduler.ts b/runtime/src/task/scheduler.ts index 2aff197..b0cc163 100644 --- a/runtime/src/task/scheduler.ts +++ b/runtime/src/task/scheduler.ts @@ -117,6 +117,37 @@ export class PendingCapability extends Error { } } +/** + * Hook invoked when a trap breaks an instance's enter/leave bracket in + * `Store.tick` (instance poisoning — see the comment at the call site). + * task/streams.ts registers the stream/future-end retirement walk here + * (#66). An injection seam rather than an import: streams.ts (via + * waitable.ts) already imports this module, and a scheduler → streams import + * would make `CopyEnd extends Waitable` evaluation-order-sensitive. + */ +let onInstancePoisoned: + | ((inst: { handles: Iterable }, cause: unknown) => void) + | null = null; + +/** @internal — see `onInstancePoisoned`; registered once by task/streams.ts. */ +export function setOnInstancePoisoned( + f: (inst: { handles: Iterable }, cause: unknown) => void, +): void { + onInstancePoisoned = f; +} + +/** + * @internal — invoke the poisoning hook. For the bracket-break sites that + * live outside this module (`Thread.resumeWith`, exec/boundary.ts `poison`): + * one seam, all sites. + */ +export function notifyInstancePoisoned( + inst: { handles: Iterable }, + cause: unknown, +): void { + onInstancePoisoned?.(inst, cause); +} + // --------------------------------------------------------------------------- // Deterministic choice // --------------------------------------------------------------------------- @@ -876,6 +907,11 @@ export class Store { } catch (e) { if (e instanceof NeedsJspi || e instanceof PendingCapability) { inst.leaveTo(null); + } else { + // The bracket stays broken (instance poisoned, comment above), so + // its live stream/future ends can never rendezvous again — retire + // them so parked host peers settle instead of hanging (#66). + onInstancePoisoned?.(inst, e); } throw e; } @@ -899,7 +935,10 @@ export class Store { * (the spec's deadlock trap), not a hang. */ export function driveSyncLift( - task: { state: string; inst: { threads: Iterable; exclusiveThread: unknown } }, + task: { + state: string; + inst: { threads: Iterable; exclusiveThread: unknown }; + }, ): void { while (task.state !== "resolved") { const candidates = [...task.inst.threads].filter( diff --git a/runtime/src/task/streams.ts b/runtime/src/task/streams.ts index e2c04e7..5db35de 100644 --- a/runtime/src/task/streams.ts +++ b/runtime/src/task/streams.ts @@ -48,10 +48,11 @@ import { assert_, trapIf } from "../cabi/trap.ts"; import { LiftLowerContext } from "../cabi/context.ts"; import { loadListFromValidRange } from "../cabi/load.ts"; import { storeListIntoValidRange } from "../cabi/store.ts"; -import { alignTo, alignment, elemSize } from "../cabi/layout.ts"; +import { alignment, alignTo, elemSize } from "../cabi/layout.ts"; import { despecialize, valTypeEqual } from "../cabi/types.ts"; import type { ComponentValue, ValType } from "../cabi/types.ts"; import { Waitable } from "./waitable.ts"; +import { setOnInstancePoisoned } from "./scheduler.ts"; /** Structural element-type equality (`null` = the zero-width payload). * Delegates to `valTypeEqual`: naive `JSON.stringify` comparison throws on @@ -312,7 +313,9 @@ export class SharedStreamImpl implements SharedBase { this.pendingOnCopy!(() => this.resetPending()); } onCopyDone(CopyResult.COMPLETED); - } else if (srcBuffer.isZeroLength() && this.pendingBuffer.isZeroLength()) { + } else if ( + srcBuffer.isZeroLength() && this.pendingBuffer.isZeroLength() + ) { // Zero-length rendezvous: both sides are empty, which is a *completed* // handshake rather than a parked write (definitions.py line 1064 — // the case `test/async/zero-length.wast` exists to pin). @@ -516,6 +519,115 @@ export class WritableFutureEnd extends CopyEnd { } } +// --------------------------------------------------------------------------- +// Poisoned-instance retirement (#66) +// --------------------------------------------------------------------------- + +/** + * Failures recorded against shared stream/future objects whose peer end died + * inside a trap-poisoned instance's handle table. The embedder layer consults + * this to reject host operations loudly (contracts/embedder-api.md amendment + * A7) instead of letting them hang forever or fake a clean end-of-stream. + */ +const poisonFailures = new WeakMap(); + +/** The recorded poisoning failure for a shared stream/future value, if any. */ +export function poisonFailureOf(shared: unknown): Error | undefined { + return typeof shared === "object" && shared !== null + ? poisonFailures.get(shared) + : undefined; +} + +/** Instances whose async ends have already been retired (idempotence). */ +const retiredInstances = new WeakSet(); + +/** The structural slice of `ComponentInstanceState` the walk needs. */ +interface PoisonedInstanceLike { + readonly index?: number; + handles: Iterable; +} + +/** + * Drop a shared stream/future as *teardown*, without waking a doomed guest. + * + * Same outcome as `drop()` for host ends and healthy guest peers (a DROPPED + * notification), with one difference: a parked side belonging to an entered + * — and on every teardown path, about-to-be- or already-poisoned — guest + * instance (`mayEnter === false`) is retired silently via `resetPending`. + * Notifying it would queue a phantom event into the corpse's waitables, and + * a later driving loop servicing it would resume machinery whose instance + * can no longer be entered (`tick` asserts enterability). Host sentinels + * carry no `mayEnter` key and healthy guest peers park only outside the + * bracket (`mayEnter === true`), so both are always notified. + * + * Used by the poisoning walk below and by the trapping-import abandonment + * path (embedder/instantiate.ts `releaseAsyncArgs`). Idempotent. + */ +export function dropSharedForTeardown( + shared: SharedStreamImpl | SharedFutureImpl, +): void { + if (shared.dropped) return; + shared.dropped = true; + if (shared.pendingBuffer) { + const pi = shared.pendingInst as { mayEnter?: boolean } | null; + const parkedInDeadGuest = pi !== null && typeof pi === "object" && + typeof pi.mayEnter === "boolean" && !pi.mayEnter; + if (parkedInDeadGuest) shared.resetPending(); + else shared.resetAndNotifyPending(CopyResult.DROPPED); + } +} + +/** + * Retire every live stream/future end in a trap-poisoned instance's handle + * table (#66). + * + * Rationale: after a trap breaks the enter/leave bracket, `mayEnter` stays + * false forever, so no task of this instance can ever rendezvous again. Its + * table's `CopyEnd`s are therefore unreachable-forever — leaving their shared + * objects live strands the peers: a parked HOST operation never settles (its + * promise hangs), and a LATER host operation would "succeed" against the + * corpse (a copy into memory nothing will ever read — silent data loss). + * Dropping the shared object now converts both into the spec-shaped DROPPED + * outcome, and the recorded failure lets the embedder layer brand it. + * + * Called from every bracket-break site — exec/boundary.ts `poison()` (the + * sync-lift path), scheduler.ts `Store.tick` and thread.ts + * `Thread.resumeWith` (traps during a resumed thread), and the FACT + * cross-component catches (intrinsics/fact_calls.ts, callee side) — with the + * trap as `cause`. Idempotent per instance. The parked-side notification + * discipline lives in `dropSharedForTeardown` above. + */ +export function retireInstanceAsyncEnds( + inst: PoisonedInstanceLike, + cause: unknown, +): void { + if (retiredInstances.has(inst)) return; + retiredInstances.add(inst); + for (const e of inst.handles) { + if (!(e instanceof CopyEnd)) continue; + const shared = e.shared as SharedStreamImpl | SharedFutureImpl; + if (poisonFailures.get(shared) === undefined) { + const where = inst.index !== undefined + ? `component instance ${inst.index}` + : "a component instance"; + poisonFailures.set( + shared, + new Error( + `${where} trapped while it held an end of this stream/future; ` + + `the peer can never rendezvous again`, + { cause }, + ), + ); + } + dropSharedForTeardown(shared); + } +} + +// `Store.tick`'s bracket-break site reaches the walk through this seam (its +// module cannot import ours — see `setOnInstancePoisoned`); the sync-lift +// site (exec/boundary.ts `poison`) imports it directly. +setOnInstancePoisoned(retireInstanceAsyncEnds); + // --------------------------------------------------------------------------- // error-context (definitions.py `class ErrorContext`, line 2782) // --------------------------------------------------------------------------- diff --git a/runtime/src/task/thread.ts b/runtime/src/task/thread.ts index 338538d..dfb0836 100644 --- a/runtime/src/task/thread.ts +++ b/runtime/src/task/thread.ts @@ -23,6 +23,7 @@ import { CANCELLED_FALSE, CANCELLED_TRUE, NeedsJspi, + notifyInstancePoisoned, PendingCapability, popCurrentThread, pushCurrentThread, @@ -164,6 +165,14 @@ export class Thread implements SchedulableThread { } catch (e) { if (e instanceof NeedsJspi || e instanceof PendingCapability) { inst.leaveTo(null); + } else { + // The bracket stays broken (instance poisoned, comment above) — same + // as `Store.tick`: retire the poisoned table's stream/future ends so + // parked host peers settle instead of hanging (#66). + notifyInstancePoisoned( + inst as unknown as { handles: Iterable }, + e, + ); } throw e; } diff --git a/runtime/tests/embedder/trap_retire_test.ts b/runtime/tests/embedder/trap_retire_test.ts new file mode 100644 index 0000000..0d11d20 --- /dev/null +++ b/runtime/tests/embedder/trap_retire_test.ts @@ -0,0 +1,211 @@ +// #66: a component fault must never strand or silently satisfy a host +// stream/future operation (contracts/embedder-api.md amendment A7). +// +// Mechanism under test: a trap breaks the enter/leave bracket, the instance +// is poisoned (mayEnter stays false forever), and the retirement walk +// (task/streams.ts `retireInstanceAsyncEnds`, hooked at both bracket-break +// sites) drops every live stream/future end in the poisoned table and +// records the failure — so parked host peers settle and the conventions +// layer rejects them with `PeerTrappedError` instead of hanging or faking a +// clean end-of-stream. Import-position: a trapping host import drops the +// lifted stream/future args it abandoned (instantiate.ts releaseAsyncArgs). +// +// Fixture: examples/guests/stream-pass — `consume-then-trap` (reads, then +// unreachable), `open-then-trap` (writes from a background task, then +// unreachable), plus `forward`/`sink` for the import-position shape. + +import { assertEq } from "../support/asserts.ts"; +import { caught, guest, haveFixture, instantiateFixture } from "./support.ts"; +import { PeerTrappedError, Stream, Trap } from "../../src/embedder/mod.ts"; +import { hostStream } from "../../src/exec/mod.ts"; + +const FIXTURE = guest("stream-pass"); +const ready = await haveFixture(FIXTURE); + +Deno.test({ + name: "trap retire: a write parked on a trapped consumer rejects with PeerTrappedError", + ignore: !ready, + fn: async () => { + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + // More bytes than the guest will consume: whatever wit-bindgen's read + // granularity is, the offer cannot be fully taken before the trap. + const big = new Uint8Array(256 * 1024); + const w = writer.writeAll(big).then( + (n) => ({ outcome: "resolved" as const, n }), + (e) => ({ outcome: "rejected" as const, e }), + ); + const callErr = await caught(() => c.exports.consumeThenTrap(stream, 2)); + assertEq(callErr instanceof Trap, true, `export call traps: ${callErr}`); + + const settled = await w; + assertEq( + settled.outcome, + "rejected", + `parked writeAll must reject, got ${Deno.inspect(settled)}`, + ); + const err = (settled as { e: unknown }).e; + assertEq(err instanceof PeerTrappedError, true, `branded: ${err}`); + assertEq( + typeof (err as PeerTrappedError).progress === "number" && + (err as PeerTrappedError).progress! < big.length, + true, + "progress rides the error and is short of the offer", + ); + assertEq( + String((err as Error).message).includes("trapped"), + true, + `names the fault: ${err}`, + ); + }, +}); + +Deno.test({ + name: "trap retire: reads from a stream whose writer trapped reject, data first", + ignore: !ready, + fn: async () => { + // `open-then-trap` writes n bytes from a background task, then traps: + // the write end dies in the poisoned table. Data copied BEFORE the trap + // is delivered; after that a read rejects instead of resolving [] (EOS + // would be wrong data reported as success). + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const out = await c.exports.openThenTrap(3) as Stream; + let got = 0; + let err: unknown; + try { + for (;;) { + const chunk = await out.read(4096); + if (chunk.length === 0) break; + got += chunk.length; + } + } catch (e) { + err = e; + } + assertEq(err instanceof PeerTrappedError, true, `branded: ${err}`); + assertEq(got, 3, "bytes written before the trap were delivered"); + }, +}); + +Deno.test({ + name: "trap retire: operations started after the trap reject immediately", + ignore: !ready, + fn: async () => { + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const w = writer.writeAll(new Uint8Array(64 * 1024)).catch(() => {}); + await caught(() => c.exports.consumeThenTrap(stream, 1)); + await w; + const e = await caught(() => writer.write(Uint8Array.from([9]))); + assertEq(e instanceof PeerTrappedError, true, `pre-op branding: ${e}`); + }, +}); + +Deno.test({ + name: "trap retire: a future whose writer trapped rejects PeerTrappedError, not DroppedError", + ignore: !ready, + fn: async () => { + // The guest parks on the gate stream, so the call resolves and the host + // can park a read on the future FIRST; releasing the gate then makes the + // guest trap holding the unwritten write end. The parked await must + // brand the fault rather than report the "write end dropped without a + // value" clean-drop shape. (A trap that fires while the export call is + // still driving arrives as the call's own Trap rejection instead — that + // path is loud already and not this test's subject.) + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream: gate, writer: gateWriter } = Stream.create(); + const f = c.exports.futureThenTrap(gate); + const fRead = caught(() => Promise.resolve(f)); + // Let the future read park, then release the gate. + await new Promise((r) => setTimeout(r, 0)); + await gateWriter.write(Uint8Array.from([1])); + const e = await fRead; + assertEq(e instanceof PeerTrappedError, true, `branded: ${e}`); + }, +}); + +Deno.test({ + name: "trap retire: a trapping import drops its lifted stream args (E2 shape)", + ignore: !ready, + fn: async () => { + // The guest hands the stream to `sink`, which throws unbranded (a host + // bug -> trap). The lift already transferred the readable end to the + // host, so the poison walk cannot see it; the import's fail path drops + // the abandoned args instead. In this pure shape — the trapping guest + // holds NO other end of the stream — the parked writer settles with the + // truthful "reader went away" short count, not a PeerTrappedError (the + // call itself carries the trap). A guest that DOES die holding another + // end of the same stream gets the poison branding from the walk instead; + // the two rules overlap there and branding wins. + const c = await instantiateFixture(FIXTURE, { + sink: () => { + throw new Error("sink exploded"); + }, + }); + const { stream, writer } = Stream.create(); + const w = writer.writeAll(Uint8Array.from([1, 2, 3])); + const callErr = await caught(() => c.exports.forward(stream)); + assertEq(callErr instanceof Trap, true, `the call traps: ${callErr}`); + assertEq(await w, 0, "parked writeAll settles short instead of hanging"); + }, +}); + +Deno.test({ + name: "clean paths stay unbranded: writer close is end-of-stream, not an error", + ignore: !ready, + fn: async () => { + const c = await instantiateFixture(FIXTURE, { sink: () => 0n }); + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + const fed = (async () => { + await writer.writeAll(Uint8Array.from([5])); + await writer.close(); + })(); + assertEq([...await out.read(4)], [5]); + await fed; + assertEq((await out.read(4)).length, 0, "clean EOS resolves, no throw"); + out.drop(); + }, +}); + +// --------------------------------------------------------------------------- +// One in-flight operation per host end (the guard the #66 repro exposed: +// a second same-direction op used to "rendezvous" against our own parked +// buffer and resolve as if a peer took the data). +// --------------------------------------------------------------------------- + +Deno.test({ + name: "host ends: a second same-direction op throws instead of self-rendezvousing", + ignore: false, + fn: async () => { + const hs = hostStream({ kind: "u8" }); + const w1 = hs.writable.write([1, 2, 3]); // parks (no reader) + let err: unknown; + try { + await hs.writable.write([4]); + } catch (e) { + err = e; + } + assertEq(err instanceof TypeError, true, `write guard: ${err}`); + assertEq( + String(err).includes("already in flight"), + true, + `message names the misuse: ${err}`, + ); + + // Reading while a write is parked stays legal (different ends): it is + // the rendezvous itself. + assertEq([...(await hs.readable.read(8)) as unknown as Uint8Array], [1, 2, 3]); + assertEq(await w1, 3); + + const r1 = hs.readable.read(8); // parks (no writer) + let err2: unknown; + try { + await hs.readable.read(8); + } catch (e) { + err2 = e; + } + assertEq(err2 instanceof TypeError, true, `read guard: ${err2}`); + hs.readable.cancelRead(); + await r1; + }, +});