diff --git a/contracts/embedder-api.md b/contracts/embedder-api.md index 69de056..f68f85e 100644 --- a/contracts/embedder-api.md +++ b/contracts/embedder-api.md @@ -347,6 +347,14 @@ instance↔rep mapping; when the guest drops its last own handle the runtime calls `instance[Symbol.dispose]?.()`. Method `self` is the instance. +Overlapping host-originated borrows retain the mapping until the last +borrowing call ends. A guest drop during that interval defers disposal +until the final borrow ends; the pending-drop instance cannot be passed +as own again. A deferred disposal error is reported by the last borrowing +call, after all its borrow mappings are released. An existing call failure +remains primary; results that cannot be delivered because cleanup failed +are released rather than abandoned. + **Constructors are synchronous** (a JS constructor cannot await). A guest constructor that does not complete synchronously raises a named error rather than half-constructing; its plain entry is one instance of diff --git a/crates/translator-shim/src/fact_string_limits.rs b/crates/translator-shim/src/fact_string_limits.rs new file mode 100644 index 0000000..e41e784 --- /dev/null +++ b/crates/translator-shim/src/fact_string_limits.rs @@ -0,0 +1,317 @@ +//! Correct the pinned FACT generator's pre-realloc string limit, not guest code. +//! Its old checks use destination widths (and retry expansion factors). The +//! reference limits SOURCE bytes instead. That bound also makes the old allocation +//! arithmetic checks redundant: even 3 * source units is below 2^31. +use anyhow::{bail, ensure, Context, Result}; +use std::collections::HashMap; +use wasmtime_environ::wasmparser::{self, BlockType, Operator as Op, Parser, Payload, TypeRef}; + +const MAX: i64 = (1 << 28) - 1; +const OLD: i64 = (1 << 31) - 1; +const PIN: &str = "rev = \"4675ee16b703b33948073a5ff6b961367371e7a1\""; + +pub(super) fn correct(wasm: &[u8]) -> Result> { + ensure!( + include_str!("../../../Cargo.toml").contains(PIN), + "FACT string-limit correction needs review for new environ pin" + ); + let mut output = wasm.to_vec(); + let mut transcodes = HashMap::new(); + let mut trap = None; + let mut function_index = 0; + for payload in Parser::new(0).parse_all(wasm) { + match payload? { + Payload::ImportSection(imports) => { + for import in imports.into_imports() { + let import = import?; + if !matches!(import.ty, TypeRef::Func(_)) { + continue; + } + if import.module == "transcode" { + let op = import + .name + .split_once(" (mem") + .context("FACT transcode name drift")? + .0; + let (width, args) = match op { + "utf8-to-utf8" | "latin1-to-latin1" | "latin1-to-utf16" + | "utf8-to-utf16" | "utf8-to-latin1" => (1, 3), + "utf16-to-utf16" + | "utf16-to-latin1" + | "utf16-to-compact-probably-utf16" => (2, 3), + "latin1-to-utf8" | "utf8-to-compact-utf16" => (1, 5), + "utf16-to-utf8" | "utf16-to-compact-utf16" => (2, 5), + _ => bail!("FACT transcode operation drift: {op}"), + }; + transcodes.insert(function_index, (width, args)); + } + if import.module == "runtime" + && import.name + == format!("trap{}", wasmtime_environ::Trap::StringOutOfBounds as u8) + { + trap = Some(function_index); + } + function_index += 1; + } + } + Payload::CodeSectionEntry(body) if !transcodes.is_empty() => { + let ops = body + .get_operators_reader()? + .into_iter_with_offsets() + .map(|op| op.map(|(op, offset)| (op, offset as usize))) + .collect::, _>>()?; + rewrite_body( + &ops, + &transcodes, + trap.context("FACT string trap missing")?, + &mut output, + )?; + } + _ => {} + } + } + ensure!( + output.len() == wasm.len(), + "FACT correction changed module length" + ); + wasmparser::Validator::new_with_features(super::features()) + .validate_all(&output) + .context("invalid FACT adapter after string-limit correction")?; + Ok(output) +} + +// Parse only the straight-line argument expressions FACT emits. In particular, +// calls and control flow cannot be mistaken for a source-length expression. +fn expression_start(ops: &[(Op<'_>, usize)], end: usize) -> Result { + ensure!(end > 0, "FACT transcode argument underflow"); + let i = end - 1; + match ops[i].0 { + Op::LocalGet { .. } | Op::I32Const { .. } | Op::I64Const { .. } => Ok(i), + Op::I32WrapI64 | Op::I64ExtendI32U => expression_start(ops, i), + Op::I32Add | Op::I64Add | Op::I32Sub | Op::I64Sub | Op::I32Shl | Op::I64Shl => { + expression_start(ops, expression_start(ops, i)?) + } + _ => bail!("FACT transcode argument shape drift"), + } +} + +fn rewrite_body( + ops: &[(Op<'_>, usize)], + transcodes: &HashMap, + trap: u32, + output: &mut [u8], +) -> Result<()> { + // A path identifies branches, not just nesting depth. A guard in one arm + // cannot authorize a transcode in its sibling, or after that arm has ended. + let mut path = Vec::new(); + let mut guards: Vec<(u32, Vec, i64)> = Vec::new(); + let mut pending: Option<(u32, Vec, usize)> = None; + let mut i = 0; + while i < ops.len() { + let value = match ops[i].0 { + Op::I32Const { value } => Some(i64::from(value)), + Op::I64Const { value } => Some(value), + _ => None, + }; + if value.is_some_and(|v| [OLD, OLD / 2, OLD / 3].contains(&v)) + && i > 0 + && matches!(ops[i - 1].0, Op::LocalGet { .. }) + && matches!(ops.get(i + 3).map(|o| &o.0), Some(Op::Call { function_index }) if *function_index == trap) + { + let local = match ops[i - 1].0 { + Op::LocalGet { local_index } => local_index, + _ => unreachable!(), + }; + let wide = matches!(ops[i].0, Op::I64Const { .. }); + ensure!( + matches!(ops.get(i + 1).map(|o| &o.0), Some(Op::I64GtU)) && wide + || matches!(ops.get(i + 1).map(|o| &o.0), Some(Op::I32GtU)) && !wide, + "FACT string comparison drift" + ); + ensure!( + matches!( + ops.get(i + 2).map(|o| &o.0), + Some(Op::If { + blockty: BlockType::Empty + }) + ) && matches!(ops.get(i + 3).map(|o| &o.0), Some(Op::Call { function_index }) if *function_index == trap) + && matches!(ops.get(i + 4).map(|o| &o.0), Some(Op::Unreachable)) + && matches!(ops.get(i + 5).map(|o| &o.0), Some(Op::End)), + "FACT string guard shape drift" + ); + ensure!(pending.is_none(), "FACT unassociated string guard"); + pending = Some((local, path.clone(), i)); + i += 6; + continue; + } + match &ops[i].0 { + Op::Block { .. } | Op::Loop { .. } | Op::If { .. } | Op::TryTable { .. } => { + path.push(i) + } + Op::Else | Op::End => { + if let Some((_, scope, _)) = &pending { + ensure!( + path.len() > scope.len(), + "FACT string guard crosses control-flow scope" + ); + } + path.pop(); + if matches!(ops[i].0, Op::Else) { + path.push(i); + } + guards.retain(|(_, scope, _)| path.starts_with(scope)); + } + Op::LocalSet { local_index } | Op::LocalTee { local_index } => { + ensure!( + !pending + .as_ref() + .is_some_and(|(local, _, _)| local == local_index), + "FACT guarded source length overwritten" + ); + guards.retain(|(local, _, _)| local != local_index); + } + Op::Br { .. } | Op::BrTable { .. } | Op::Return => { + ensure!(pending.is_none(), "FACT string guard crosses branch"); + } + Op::BrIf { relative_depth } if pending.is_some() => { + let scope = &pending.as_ref().unwrap().1; + ensure!( + path.len() - *relative_depth as usize > scope.len(), + "FACT string guard crosses conditional branch" + ); + } + Op::Call { function_index } if transcodes.contains_key(function_index) => { + let (width, nargs) = transcodes[function_index]; + let mut end = i; + // Skip destination arguments; the second argument is source units. + for _ in 2..nargs { + end = expression_start(ops, end)?; + } + let start = expression_start(ops, end)?; + let (local, retry) = match &ops[start..end] { + [(Op::LocalGet { local_index }, _)] => (*local_index, false), + [(Op::LocalGet { local_index }, _), (Op::LocalGet { .. }, _), (Op::I32Sub | Op::I64Sub, _)] => { + (*local_index, true) + } + _ => bail!("FACT source length expression drift"), + }; + if retry { + ensure!( + guards.iter().any(|(l, scope, w)| *l == local + && *w == width + && path.starts_with(scope)), + "FACT retry missing original source bound" + ); + } + if let Some((guard_local, scope, constant)) = pending.take() { + ensure!( + scope == path && guard_local == local, + "FACT string guard/transcode association drift" + ); + let mut value = MAX / width; + // Keep the original signed-LEB width, including legal padding. + let bytes = &mut output[ops[constant].1 + 1..ops[constant + 1].1]; + let len = bytes.len(); + for (j, byte) in bytes.iter_mut().enumerate() { + *byte = (value as u8 & 0x7f) | if j + 1 < len { 0x80 } else { 0 }; + value >>= 7; + } + ensure!(value == 0, "FACT string immediate width drift"); + guards.push((local, scope, width)); + } + ensure!( + guards + .iter() + .any(|(l, scope, w)| *l == local && *w == width && path.starts_with(scope)), + "FACT transcode missing dominating source guard" + ); + } + _ => {} + } + i += 1; + } + ensure!(pending.is_none(), "FACT string guard has no transcode"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn module(body: &str) -> Vec { + wat::parse_str(format!( + r#"(module + (import "runtime" "trap{}" (func $trap)) + (import "transcode" "utf8-to-utf8 (mem0 => mem1)" (func $copy (param i32 i32 i32))) + (func (param i32 i32) {body}))"#, + wasmtime_environ::Trap::StringOutOfBounds as u8 + )) + .unwrap() + } + + const GUARD: &str = "local.get 0 i32.const 2147483647 i32.gt_u if call $trap unreachable end"; + const CALL: &str = "i32.const 0 local.get 0 i32.const 0 call $copy"; + + #[test] + fn padded_immediate_and_unrelated_constants() { + let input = module(&format!( + "{GUARD} {CALL} local.get 1 i32.const 2147483647 i32.gt_u drop" + )); + let output = correct(&input).unwrap(); + assert_eq!(input.len(), output.len()); + let differences = input.iter().zip(&output).filter(|(a, b)| a != b).count(); + assert_eq!(differences, 1); // only the top group in the five-byte LEB + assert_eq!( + input.iter().filter(|b| **b == 7).count(), + output.iter().filter(|b| **b == 7).count() + 1 + ); + } + + #[test] + fn no_strings_is_byte_identical() { + let input = wat::parse_str( + "(module (func (param i32) local.get 0 i32.const 2147483647 i32.gt_u drop))", + ) + .unwrap(); + assert_eq!(correct(&input).unwrap(), input); + } + + #[test] + fn drift_fails_closed() { + let cases = [ + CALL.to_string(), + format!("{} {CALL}", GUARD.replace("i32.gt_u", "i32.ge_u")), + format!("{} {CALL}", GUARD.replace("2147483647", "2147483646")), + format!("{} {CALL}", GUARD.replace("unreachable", "nop")), + format!("{GUARD} i32.const 0 local.set 0 {CALL}"), + format!("{GUARD} i32.const 0 local.tee 0 drop {CALL}"), + format!("i32.const 1 if {GUARD} else {CALL} end"), + format!("block {GUARD} end {CALL}"), + format!("block {GUARD} br 0 {CALL} end"), + format!("block {GUARD} i32.const 1 br_if 0 {CALL} end"), + format!("{GUARD} i32.const 0 local.get 1 i32.const 0 call $copy"), + format!("{GUARD} i32.const 0 local.get 0 local.get 1 i32.sub i32.const 0 call $copy"), + format!("{GUARD} return"), + ]; + for body in cases { + assert!(correct(&module(&body)).is_err(), "accepted drift: {body}"); + } + } + + #[test] + fn memory64_guard_uses_same_source_bound() { + let input = wat::parse_str(format!( + r#"(module + (import "runtime" "trap{}" (func $trap)) + (import "transcode" "utf16-to-utf16 (mem0 => mem1)" (func $copy (param i64 i64 i64))) + (func (param i64) + local.get 0 i64.const 1073741823 i64.gt_u if call $trap unreachable end + i64.const 0 local.get 0 i64.const 0 call $copy))"#, + wasmtime_environ::Trap::StringOutOfBounds as u8 + )) + .unwrap(); + let output = correct(&input).unwrap(); + assert_eq!(input.len(), output.len()); + assert_ne!(input, output); + } +} diff --git a/crates/translator-shim/src/lib.rs b/crates/translator-shim/src/lib.rs index 5c8e645..f989b3c 100644 --- a/crates/translator-shim/src/lib.rs +++ b/crates/translator-shim/src/lib.rs @@ -42,6 +42,7 @@ use wasmtime_environ::{ScopeVec, Tunables, wasmparser}; pub const WASMTIME_ENVIRON_VERSION: &str = "49.0.0-dev+4675ee1"; pub mod error; +mod fact_string_limits; pub mod plan; pub use error::{Phase, TranslateError}; @@ -244,7 +245,7 @@ fn map_translation( ); adapters.push(AdapterArtifact { file, - wasm: mt.wasm.to_vec(), + wasm: fact_string_limits::correct(mt.wasm)?, }); } } diff --git a/crates/translator-shim/tests/fact_string_source_limits.rs b/crates/translator-shim/tests/fact_string_source_limits.rs new file mode 100644 index 0000000..a171bb3 --- /dev/null +++ b/crates/translator-shim/tests/fact_string_source_limits.rs @@ -0,0 +1,62 @@ +use translator_shim::{plan::ModuleEntry, translate}; +use wasmtime_environ::wasmparser::{Operator, Parser, Payload}; + +#[test] +fn fact_string_encoding_matrix() { + let bytes = wat::parse_str(include_str!( + "../../../runtime/tests/fixtures/fact-string-source-limits.wat" + )) + .unwrap(); + let translated = translate(&bytes).unwrap(); + assert_eq!( + translated.plan.producer.wasmtime_environ, + "49.0.0-dev+4675ee1" + ); + let mut limits = [0, 0]; + for adapter in &translated.adapters { + for payload in Parser::new(0).parse_all(&adapter.wasm) { + if let Payload::CodeSectionEntry(body) = payload.unwrap() { + for op in body.get_operators_reader().unwrap() { + if let Operator::I32Const { value } = op.unwrap() { + match value { + 268435455 => limits[0] += 1, + 134217727 => limits[1] += 1, + 2147483647 | 1073741823 | 715827882 => panic!("old FACT limit remains"), + _ => {} + } + } + } + } + } + assert!(translated.plan.modules.iter().any(|m| matches!(m, ModuleEntry::Adapter { file, len, .. } if file == &adapter.file && *len == adapter.wasm.len()))); + } + // Nine encoding pairs, compact's two branches, plus four retry checks. + assert_eq!(limits, [8, 8]); +} + +#[test] +fn embedded_guest_string_guard_is_not_rewritten() { + let core = format!( + r#"(module + (import "runtime" "trap{}" (func $trap)) + (import "transcode" "utf8-to-utf8 (mem0 => mem1)" (func $copy (param i32 i32 i32))) + (func (param i32) + local.get 0 i32.const 2147483647 i32.gt_u if call $trap unreachable end + i32.const 0 local.get 0 i32.const 0 call $copy))"#, + wasmtime_environ::Trap::StringOutOfBounds as u8 + ); + let expected = wat::parse_str(&core).unwrap(); + let component = format!( + "(component {})", + core.replacen("(module", "(core module", 1) + ); + let bytes = wat::parse_str(component).unwrap(); + let original = bytes.clone(); + let translated = translate(&bytes).unwrap(); + assert!(translated.adapters.is_empty()); + assert_eq!(bytes, original); + let ModuleEntry::Embedded { offset, len } = &translated.plan.modules[0] else { + panic!("guest module became adapter") + }; + assert_eq!(&bytes[*offset as usize..*offset as usize + len], expected); +} diff --git a/harness/src/runtime-executor.ts b/harness/src/runtime-executor.ts index 38c0b96..4cdbe3e 100644 --- a/harness/src/runtime-executor.ts +++ b/harness/src/runtime-executor.ts @@ -13,7 +13,7 @@ // understood but the capability plainly doesn't exist yet — precise // hand-off to whichever track builds it. -import { loadPlan, PlanError } from "@polyengine/runtime/plan"; +import { loadPlan, PlanError, TranslateError } from "@polyengine/runtime/plan"; import type { LoadedPlan } from "@polyengine/runtime/plan"; import type { WireExport } from "@polyengine/runtime/plan"; import { Translator } from "@polyengine/runtime/shim"; @@ -21,11 +21,15 @@ import { type ComponentHandle, instantiateComponent, } from "../../runtime/src/exec/mod.ts"; -import { AssertionError, NotImplemented, Trap } from "../../runtime/src/cabi/mod.ts"; +import { + AssertionError, + NotImplemented, + Trap, +} from "../../runtime/src/cabi/mod.ts"; import { type Artifact, - CoreOnlyExecutor, type CommandExecutor, + CoreOnlyExecutor, type InstanceRef, type InstantiateExpectation, type InvokeOutcome, @@ -132,8 +136,9 @@ export class RuntimeExecutor implements CommandExecutor { this.#translator.translate(artifact.bytes); return Promise.resolve({ valid: true }); } catch (e) { + if (!(e instanceof TranslateError) || !e.isValidationVerdict) throw e; return Promise.resolve( - { valid: false, error: e instanceof Error ? e.message : String(e) }, + { valid: false, error: e.message }, ); } } @@ -142,7 +147,9 @@ export class RuntimeExecutor implements CommandExecutor { artifact: Artifact, expect: InstantiateExpectation, ): Promise { - if (artifact.kind === "module") return this.#core.instantiate(artifact, expect); + if (artifact.kind === "module") { + return this.#core.instantiate(artifact, expect); + } return await this.#instantiateComponent(artifact.bytes, expect); } diff --git a/harness/tests/fixtures/validation-imported-module.wasm b/harness/tests/fixtures/validation-imported-module.wasm new file mode 100644 index 0000000..1a8ea7c Binary files /dev/null and b/harness/tests/fixtures/validation-imported-module.wasm differ diff --git a/harness/tests/fixtures/validation-imported-module.wat b/harness/tests/fixtures/validation-imported-module.wat new file mode 100644 index 0000000..89cfbdd --- /dev/null +++ b/harness/tests/fixtures/validation-imported-module.wat @@ -0,0 +1,6 @@ +;; Valid component; instantiating an imported core module is unsupported. +;; Generate: wasm-tools parse validation-imported-module.wat -o validation-imported-module.wasm +(component + (import "m" (core module $m)) + (core instance (instantiate $m)) +) diff --git a/harness/tests/runtime_validation_test.ts b/harness/tests/runtime_validation_test.ts new file mode 100644 index 0000000..2cf5e05 --- /dev/null +++ b/harness/tests/runtime_validation_test.ts @@ -0,0 +1,152 @@ +import { PlanError, TranslateError } from "@polyengine/runtime/plan"; +import { Translator } from "@polyengine/runtime/shim"; +import type { Artifact } from "../src/executor.ts"; +import { runWastJson } from "../src/runner.ts"; +import { RuntimeExecutor } from "../src/runtime-executor.ts"; + +const shim = await Deno.readFile( + new URL( + "../../target/wasm32-unknown-unknown/release/translator_shim.wasm", + import.meta.url, + ), +); +const emptyComponent = new Uint8Array([0, 0x61, 0x73, 0x6d, 0x0d, 0, 1, 0]); + +function artifact(bytes: Uint8Array): Artifact { + return { + filename: "validation.wasm", + kind: "component", + moduleType: "binary", + bytes, + }; +} + +async function thrown(fn: () => unknown): Promise { + try { + await fn(); + } catch (error) { + return error; + } + throw new Error("expected an exception, but returned normally"); +} + +async function negativeAssertions( + executor: RuntimeExecutor, + bytes: Uint8Array, +) { + return (await runWastJson( + { + source_filename: "validation.wast", + commands: (["assert_invalid", "assert_malformed"] as const).map(( + type, + line, + ) => ({ + type, + line, + filename: "validation.wasm", + module_type: "binary", + text: "invalid component", + })), + }, + () => Promise.resolve(bytes), + executor, + )).results; +} + +Deno.test("runtime validation: valid unsupported component fails negative assertions", async () => { + const bytes = await Deno.readFile( + new URL( + "fixtures/validation-imported-module.wasm", + import.meta.url, + ), + ); + const translator = await Translator.create(shim); + const error = await thrown(() => translator.translate(bytes)); + if (!(error instanceof TranslateError) || error.phase !== "unsupported") { + throw new Error(`expected structured unsupported error, got ${error}`); + } + const executor = await RuntimeExecutor.create(shim); + const propagated = await thrown(() => executor.validate(artifact(bytes))); + if ( + !(propagated instanceof TranslateError) || + propagated.phase !== "unsupported" + ) { + throw new Error( + `expected unsupported error to propagate, got ${propagated}`, + ); + } + for (const result of await negativeAssertions(executor, bytes)) { + if (result.status !== "failed" || result.detail !== String(error)) { + throw new Error( + `unsupported translation satisfied ${result.type}: ${ + JSON.stringify(result) + }`, + ); + } + } +}); + +Deno.test("runtime validation: genuine malformed component satisfies negative assertions", async () => { + // Valid component preamble followed by an invalid section ID. + const bytes = new Uint8Array([...emptyComponent, 0xff]); + const translator = await Translator.create(shim); + const error = await thrown(() => translator.translate(bytes)); + if (!(error instanceof TranslateError) || !error.isValidationVerdict) { + throw new Error(`expected structured validation error, got ${error}`); + } + const executor = await RuntimeExecutor.create(shim); + const verdict = await executor.validate(artifact(bytes)); + if (verdict.valid || verdict.error !== error.message) { + throw new Error( + `expected invalid verdict with translator message: ${ + JSON.stringify(verdict) + }`, + ); + } + for (const result of await negativeAssertions(executor, bytes)) { + if (result.status !== "passed") throw new Error(JSON.stringify(result)); + } + if (!(await executor.validate(artifact(emptyComponent))).valid) { + throw new Error("empty component should validate"); + } +}); + +Deno.test("runtime validation: pipeline failures propagate unchanged and fail negative assertions", async () => { + const executor = await RuntimeExecutor.create(shim); + const original = Translator.prototype.translate; + try { + for ( + const error of [ + new TranslateError({ + phase: "internal", + message: "adapter validation failed", + }), + new PlanError("invalid plan"), + new Error("unexpected translator failure"), + { phase: "validation", isValidationVerdict: true }, + "unexpected non-Error failure", + ] + ) { + Translator.prototype.translate = () => { + throw error; + }; + if ( + await thrown(() => executor.validate(artifact(emptyComponent))) !== + error + ) { + throw new Error("pipeline failure was replaced"); + } + for (const result of await negativeAssertions(executor, emptyComponent)) { + if (result.status !== "failed" || result.detail !== String(error)) { + throw new Error( + `pipeline failure satisfied ${result.type}: ${ + JSON.stringify(result) + }`, + ); + } + } + } + } finally { + Translator.prototype.translate = original; + } +}); diff --git a/runtime/src/cabi/async_values.ts b/runtime/src/cabi/async_values.ts index 84313df..1d4d9df 100644 --- a/runtime/src/cabi/async_values.ts +++ b/runtime/src/cabi/async_values.ts @@ -37,6 +37,7 @@ import { SharedFutureImpl, SharedStreamImpl, } from "../task/streams.ts"; +import { removeHandleWithUnwind } from "../task/scheduler.ts"; /** * Diagnostic for a handle-table entry that carries the error-context brand @@ -86,57 +87,60 @@ function liftAsyncValue( assert_(!containsBorrow(t), `${what} may not contain a borrow`); const inst = cx.inst; assert_(inst !== null, `${what} lift requires a component instance`); - const e = inst!.handles.remove(i); - trapIf(!(e instanceof EndT), `${what} lift: handle is not a ${what} end`); - const end = e as { - shared: SharedBase; - state: CopyState; - inWaitableSet(): boolean; - }; - trapIf( - !sameElemType(end.shared.t, elem), - `${what} lift: element type mismatch`, - ); - trapIf( - end.state === CopyState.DONE, - what === "future" - ? "cannot lift future after previous read succeeded" - : "cannot lift stream after being notified that the writable end dropped", - ); - trapIf(end.state !== CopyState.IDLE, `cannot remove busy ${what}`); - trapIf( - end.inWaitableSet(), - `cannot lift ${what} while it's in a waitable set`, - ); - // Remember the driving store so a host wrapper can pump the guest later. - // Single-store only: a shared object crossing into a SECOND store is - // unsupported misuse — fail loudly rather than silently pumping the first - // (review advisory, host-streams round). Class field initializes to null; - // != null covers both sentinels. - const holder = end.shared as { boundStore?: unknown }; - const store = (inst as unknown as { store?: unknown }).store; - if (holder.boundStore != null && store != null) { - // module identity: when several runtime copies are loaded, "a second store" is very - // often "a second COPY" — the shared object was minted by one runtime and - // is being driven by another. The two stores are indistinguishable from - // here (stores carry no copy identity), so the census is appended as the - // hypothesis it is, rather than asserted (issue #83). - const census = copyCensus(); - assert_( - holder.boundStore === store, - `${what} crossed into a second store; multi-store is unsupported` + - (census === "" - ? "" - : ` (${census} — a value from one copy cannot be lowered through ` + - `another)`), + return removeHandleWithUnwind(inst!, i, (e) => { + trapIf(!(e instanceof EndT), `${what} lift: handle is not a ${what} end`); + const end = e as { + shared: SharedBase; + state: CopyState; + inWaitableSet(): boolean; + }; + trapIf( + !sameElemType(end.shared.t, elem), + `${what} lift: element type mismatch`, ); - } - holder.boundStore ??= store; - // Host-wrapper re-arm hook (#162, contracts/embedder-api.md §"Streams and futures"): the readable - // end just left a guest table, so whoever receives it can act on it again. - // See `bindOnLower` in exec/host_streams.ts for the retention rule. - (end.shared as { onLifted?: ((i: unknown) => void) | null }).onLifted?.(inst); - return end.shared; + trapIf( + end.state === CopyState.DONE, + what === "future" + ? "cannot lift future after previous read succeeded" + : "cannot lift stream after being notified that the writable end dropped", + ); + trapIf(end.state !== CopyState.IDLE, `cannot remove busy ${what}`); + trapIf( + end.inWaitableSet(), + `cannot lift ${what} while it's in a waitable set`, + ); + // Remember the driving store so a host wrapper can pump the guest later. + // Single-store only: a shared object crossing into a SECOND store is + // unsupported misuse — fail loudly rather than silently pumping the first + // (review advisory, host-streams round). Class field initializes to null; + // != null covers both sentinels. + const holder = end.shared as { boundStore?: unknown }; + const store = (inst as unknown as { store?: unknown }).store; + if (holder.boundStore != null && store != null) { + // module identity: when several runtime copies are loaded, "a second store" is very + // often "a second COPY" — the shared object was minted by one runtime and + // is being driven by another. The two stores are indistinguishable from + // here (stores carry no copy identity), so the census is appended as the + // hypothesis it is, rather than asserted (issue #83). + const census = copyCensus(); + assert_( + holder.boundStore === store, + `${what} crossed into a second store; multi-store is unsupported` + + (census === "" + ? "" + : ` (${census} — a value from one copy cannot be lowered through ` + + `another)`), + ); + } + holder.boundStore ??= store; + // Host-wrapper re-arm hook (#162, contracts/embedder-api.md §"Streams and futures"): the readable + // end just left a guest table, so whoever receives it can act on it again. + // See `bindOnLower` in exec/host_streams.ts for the retention rule. + (end.shared as { onLifted?: ((i: unknown) => void) | null }).onLifted?.( + inst, + ); + return end.shared; + }); } export function liftStream( diff --git a/runtime/src/cabi/handles.ts b/runtime/src/cabi/handles.ts index ba6d677..19409d6 100644 --- a/runtime/src/cabi/handles.ts +++ b/runtime/src/cabi/handles.ts @@ -7,18 +7,14 @@ // - canon_resource_* take the instance explicitly instead of reading // current_instance() from the running thread; // - canon_resource_drop routes the dtor through `callDtorGated` below, -// which reconstructs the reference's store.lift/store.lower bracket -// (entry refusal + trap poisoning) around the destructor call (#85). +// which uses the reference's fresh synchronous lift task/thread. // Host-initiated drops do NOT come here: they run the dtor through the // real lift harness (`hostDtorCall`, exec/boundary.ts) — see #160. -import { assert_, Trap, trap, trapIf } from "./trap.ts"; -import { - entryRefusal, - NeedsJspi, - notifyInstancePoisoned, - PendingCapability, -} from "../task/scheduler.ts"; +import { assert_, trapIf } from "./trap.ts"; +import { removeHandleWithUnwind } from "../task/scheduler.ts"; +import { createDtorEntry } from "../exec/boundary.ts"; +import type { ComponentInstanceState } from "../task/mod.ts"; import { COMPONENT_INSTANCE } from "./context.ts"; import type { ComponentInstanceLike, @@ -95,13 +91,14 @@ export function liftOwn( i: number, t: OwnType, ): number { - const h = requireInst(cx).handles.remove(i); - trapIf(!(h instanceof ResourceHandle), "not a resource handle"); - const rh = h as ResourceHandle; - trapIf(rh.rt !== t.rt, "resource type mismatch"); - trapIf(rh.numLends !== 0, "handle still lent out"); - trapIf(!rh.own, "expected own handle"); - return rh.rep; + return removeHandleWithUnwind(requireInst(cx), i, (h) => { + trapIf(!(h instanceof ResourceHandle), "not a resource handle"); + const rh = h as ResourceHandle; + trapIf(rh.rt !== t.rt, "resource type mismatch"); + trapIf(rh.numLends !== 0, "handle still lent out"); + trapIf(!rh.own, "expected own handle"); + return rh.rep; + }); } export function liftBorrow( @@ -227,8 +224,8 @@ function isThenable(v: unknown): v is PromiseLike { * SCOPE (#160): this is the **guest-initiated** path only. A guest-initiated * drop must complete synchronously (the reference lifts the dtor with * `async_ = False`), so a thenable here is a trap. The host-initiated path - * goes through the full lift harness instead (`hostDtorCall` in - * exec/boundary.ts), which is what definitions.py actually does. + * uses the same lift harness with host completion policy (`hostDtorCall` in + * exec/boundary.ts). Guest entry uses only the reference synchronous drive. */ export function callDtorGated( rt: ResourceTypeInfo, @@ -236,8 +233,6 @@ export function callDtorGated( caller: unknown, ): void { const impl = isComponentInstance(rt.impl); - // Always the raw synchronous dtor: `dtorHost` is the host path's lifted - // entry, which is not callable from inside a guest activation. const dtorFn = rt.dtor; // No component instance behind the resource: an imported (host-implemented) // resource has `impl === null` by construction (executor.ts @@ -257,46 +252,11 @@ export function callDtorGated( // (Store.invoke). It feeds `entryRefusal`'s `caller !== callee` guard below. const callerInst = isComponentInstance(caller) === null ? null : caller; - // A poisoned target's refusal names the original trap (polyengine#145). - // `callerInst` can legitimately BE `impl` here (a guest dropping its own - // resource): `entryRefusal`'s self-call guard keeps that entry allowed - // even against a marked instance. - { - const refusal = entryRefusal( - impl, - callerInst, - "cannot enter component instance", - ); - if (refusal !== null) trap(refusal); - } - - const poison = (e: unknown): void => { - // Capability signals are not traps: the operation they stand in for - // completes normally in the reference, so the instance stays healthy. - if (e instanceof NeedsJspi || e instanceof PendingCapability) return; - // A real trap buries the implementing instance, and its live - // stream/future ends are retired (#66) through the same seam - // fact_calls.ts uses for its poisoning sites. - notifyInstancePoisoned(impl, e); - }; - - let out: unknown; - try { - out = dtorFn?.(rep) as unknown; - } catch (e) { - poison(e); - throw e; - } - if (isThenable(out)) { - // A guest-initiated drop is lifted with `async_ = False`: the dtor must - // resolve before `canon_resource_drop` returns. Reaching here means the - // dtor's activation escaped, which is a trap that poisons the impl. - const e = new Trap( - "resource destructor did not complete synchronously", - ); - poison(e); - throw e; - } + createDtorEntry({ + dtor: dtorFn, + instance: impl as ComponentInstanceState, + guestCaller: callerInst as ComponentInstanceState | null, + })(rep); } export function canonResourceDrop( @@ -305,25 +265,26 @@ export function canonResourceDrop( i: number, ): void { trapIf(!inst.mayLeave, "may_leave violation"); - const h = inst.handles.remove(i); - trapIf(!(h instanceof ResourceHandle), "not a resource handle"); - const rh = h as ResourceHandle; - trapIf(rh.rt !== rt, "resource type mismatch"); - trapIf(rh.numLends !== 0, "handle still lent out"); - if (rh.own) { - assert_(rh.borrowScope === null); - // definitions.py line 2326-2333: the dtor runs through the store's - // lift/lower bracket. SCOPE NOTE (#85): the call below is a JS frame - // inside the drop trampoline, so a *guest*-initiated drop whose dtor - // suspends traps under the JSPI frame rule. That is deterministic and - // loud, and routing guest-initiated dtor calls through generated wasm is - // explicitly out of scope for #85 (docs/architecture.md §5/§7 carry the - // known-limitation note). - callDtorGated(rt, rh.rep, inst); - } else { - assert_(rh.borrowScope !== null); - rh.borrowScope!.numBorrows -= 1; - } + removeHandleWithUnwind(inst, i, (h) => { + trapIf(!(h instanceof ResourceHandle), "not a resource handle"); + const rh = h as ResourceHandle; + trapIf(rh.rt !== rt, "resource type mismatch"); + trapIf(rh.numLends !== 0, "handle still lent out"); + if (rh.own) { + assert_(rh.borrowScope === null); + // definitions.py line 2326-2333: the dtor runs through the store's + // lift/lower bracket. SCOPE NOTE (#85): the call below is a JS frame + // inside the drop trampoline, so a *guest*-initiated drop whose dtor + // suspends traps under the JSPI frame rule. That is deterministic and + // loud, and routing guest-initiated dtor calls through generated wasm is + // explicitly out of scope for #85 (docs/architecture.md §5/§7 carry the + // known-limitation note). + callDtorGated(rt, rh.rep, inst); + } else { + assert_(rh.borrowScope !== null); + rh.borrowScope!.numBorrows -= 1; + } + }); } export function canonResourceRep( diff --git a/runtime/src/cabi/types.ts b/runtime/src/cabi/types.ts index 7506ac9..4e04322 100644 --- a/runtime/src/cabi/types.ts +++ b/runtime/src/cabi/types.ts @@ -66,9 +66,9 @@ export interface InstanceLike { * in lazily for tokens built directly. It returns `undefined` or a Promise, * so it is NOT callable from inside a guest activation. * - * Guest-initiated drops (`callDtorGated`) always use `dtor` directly: they - * must complete synchronously (reference lifts the dtor with - * `async_ = False`), and any thenable there is a trap. + * Guest-initiated drops (`callDtorGated`) lift `dtor` through the same + * machinery with a fresh synchronous task/thread, the guest caller identity, + * and no host-wide drive. Any thenable there is a trap. */ export class ResourceTypeInfo { constructor( diff --git a/runtime/src/embedder/instantiate.ts b/runtime/src/embedder/instantiate.ts index cca5716..497e463 100644 --- a/runtime/src/embedder/instantiate.ts +++ b/runtime/src/embedder/instantiate.ts @@ -15,6 +15,8 @@ import type { LoadedPlan } from "../plan/loader.ts"; import { loadEnvelope, loadPlan, PlanError } from "../plan/loader.ts"; import type { FuncType, ResourceTypeInfo, ValType } from "../cabi/types.ts"; import type { ComponentValue, VariantValue } from "../cabi/types.ts"; +import { despecialize } from "../cabi/types.ts"; +import { hostFutureFor, hostStreamFor } from "../exec/host_streams.ts"; import { Trap } from "../cabi/trap.ts"; import { type ComponentHandle, @@ -392,7 +394,7 @@ class Facade { // The rt is supplied per wrapper, so an anonymous class needs none here. { impl: null, dtor: null } as unknown as ResourceTypeInfo, () => () => Promise.reject(new TypeError("no methods")), - () => [], + () => ({ lowered: [], release: () => {} }), ); return b.cls; } @@ -448,21 +450,16 @@ class Facade { lowerOwn(v, t) { const b = self.#binding(t.rt); if (b.kind === "host") return b.registry.repFor(v); - return takeRep(v, true, `own<${b.name}>`); + return takeRep(v, t.rt, true, `own<${b.name}>`); }, lowerBorrow(v, t) { const b = self.#binding(t.rt); if (b.kind === "host") { - // Contract 2x4 table, bottom-right: "a never-registered instance - // gets a rep allocated **for the call's duration**". A rep minted - // here is call-scoped, so it is released when the call returns — - // otherwise it would sit in the registry's STRONG rep->instance map - // forever, since a guest dropping a borrow handle runs no dtor. - const known = b.registry.hasInstance(v); - const rep = b.registry.repFor(v); - if (!known) { - self.#lowerScope?.push(() => b.registry.releaseIfPresent(rep)); - } + // Each overlapping call retains the rep; the final borrow release + // removes only temporary mappings, never a guest-owned registration. + const { rep, release } = b.registry.borrowFor(v); + if (self.#lowerScope === null) release(); + else self.#lowerScope.push(release); return rep; } // Host `own` wrapper lowered as `borrow` (#86): record the lend @@ -471,7 +468,7 @@ class Facade { // definitions.py `lift_borrow` -> `Subtask.add_lender` (line 890); // `#lowerScope` is released where that subtask delivers its // resolution, i.e. when the call ends. - const rep = takeRep(v, false, `borrow<${b.name}>`); + const rep = takeRep(v, t.rt, false, `borrow<${b.name}>`); const release = lendWrapper(v as object); if (self.#lowerScope === null) { // No enclosing lowering scope (a raw/one-off lowering): the lend @@ -1053,7 +1050,7 @@ class Facade { (raw, params, results, async, where) => this.#wrapExportFn(raw, { params, results, async }, where), (args, params, where) => - args.map((a, i) => fromHost(a, params[i], this.#opts(where))), + this.#lowerParams(params, args, this.#opts(where)), ); obj[claim(pascalCase(name), name)] = cls; const index = this.#tokenIndex.get(rt); @@ -1078,26 +1075,93 @@ class Facade { o: AdapterOptions, ): { lowered: ComponentValue[]; release: () => void } { const scope: (() => void)[] = []; + let released = false; + const release = () => { + if (released) return; + released = true; + let failed = false; + let error: unknown; + for (const r of scope) { + try { + r(); + } catch (e) { + if (!failed) error = e; + failed = true; + } + } + if (failed) throw error; + }; const outer = this.#lowerScope; this.#lowerScope = scope; let lowered: ComponentValue[]; try { lowered = params.map((p, i) => fromHost(args[i], p, o)); } catch (e) { - for (const r of scope) r(); - throw e; + try { + release(); + } finally { + throw e; + } } finally { this.#lowerScope = outer; } - let released = false; - return { - lowered, - release: () => { - if (released) return; - released = true; - for (const r of scope) r(); - }, - }; + return { lowered, release }; + } + + /** Cleanup cannot abandon a result already transferred out of the guest. */ + #finishCall( + release: () => void, + succeeded: boolean, + raw: unknown, + type: ValType | null, + ): void { + try { + release(); + } catch (e) { + if (!succeeded) return; // Preserve the original call failure. + if (type !== null) this.#dropResult(raw as ComponentValue, type); + throw e; + } + } + + #dropResult(raw: ComponentValue, type: ValType): void { + // Already failing cleanup: retire every owned leaf, preserving that error + // even when a result destructor also throws. + try { + const t = despecialize(type); + switch (t.kind) { + case "own": + this.#bridge.dropOwn(raw as number, t); + break; + case "future": + hostFutureFor(raw).drop(); + break; + case "stream": + hostStreamFor(raw).readable.drop(); + break; + case "list": + for (const v of raw as ComponentValue[]) { + this.#dropResult(v, t.element); + } + break; + case "record": + for (const f of t.fields) { + this.#dropResult( + (raw as Record)[f.label], + f.type, + ); + } + break; + case "variant": { + const v = raw as VariantValue; + const payload = t.cases.find((c) => c.label === v.kind)?.type; + if (payload != null) this.#dropResult(v.value, payload); + break; + } + } + } catch { + // The argument cleanup error remains primary. + } } /** @@ -1136,13 +1200,14 @@ class Facade { try { pending = Promise.resolve(fn(...lowered)) as Promise; } catch (e) { - release(); + this.#finishCall(release, false, undefined, resultType); throw e; } - void pending.then(release, release); return Future.deferred( pending, elementCodec(element, o), + (succeeded, raw) => + this.#finishCall(release, succeeded, raw, resultType), ) as unknown as Promise; }; } else { @@ -1156,11 +1221,11 @@ class Facade { let raw: unknown; try { raw = await fn(...lowered); - } finally { - // Call-scoped reps minted for `borrow` arguments of a - // host-implemented resource live exactly as long as the call. - release(); + } catch (e) { + this.#finishCall(release, false, undefined, resultType); + throw e; } + this.#finishCall(release, true, raw, resultType); if (resultType === null) return undefined; if (resultType.kind === "result") { // Internal result: `{kind: "ok"|"error", value}` (cabi/types.ts @@ -1243,9 +1308,11 @@ class Facade { let raw: unknown; try { raw = entry(...lowered); - } finally { - release(); + } catch (e) { + this.#finishCall(release, false, undefined, resultType); + throw e; } + this.#finishCall(release, true, raw, resultType); if (isThenable(raw)) unreachableThenable(raw); return Future.fromLifted( raw as ComponentValue, @@ -1263,9 +1330,11 @@ class Facade { let raw: unknown; try { raw = entry(...lowered); - } finally { - release(); + } catch (e) { + this.#finishCall(release, false, undefined, resultType); + throw e; } + this.#finishCall(release, true, raw, resultType); if (isThenable(raw)) unreachableThenable(raw); if (resultType === null) return undefined; if (resultType.kind === "result") { diff --git a/runtime/src/embedder/resources.ts b/runtime/src/embedder/resources.ts index e53d398..e748ca4 100644 --- a/runtime/src/embedder/resources.ts +++ b/runtime/src/embedder/resources.ts @@ -299,14 +299,27 @@ export function invalidateWrapper(w: object): void { } /** Read a wrapper's rep for a lowering site, applying the ownership rule. */ -export function takeRep(w: unknown, own: boolean, what: string): number { +export function takeRep( + w: unknown, + rt: ResourceTypeInfo, + own: boolean, + what: string, +): number { if (typeof w !== "object" || w === null) { throw new InvalidHandleError( `${what}: expected a resource class instance, got ${typeof w}`, ); } const s = requireLive(w, what); + if (s.rt !== rt) { + throw new InvalidHandleError(`${what}: resource type mismatch`); + } if (own) { + if (!s.owns) { + throw new InvalidHandleError( + `${what}: a borrowed ${s.className} handle cannot be transferred as own`, + ); + } // definitions.py `lift_own` (line 1508): `trap_if(h.num_lends != 0)`. A // handle currently lent to an in-flight call cannot be transferred away. if (s.lends > 0) { @@ -377,7 +390,11 @@ export function buildGuestResourceClass( spec: GuestResourceSpec, rt: ResourceTypeInfo, wrapExport: ExportWrapper, - lowerArgs: (args: unknown[], params: ValType[], where: string) => unknown[], + lowerArgs: ( + args: unknown[], + params: ValType[], + where: string, + ) => { lowered: unknown[]; release: () => void }, // deno-lint-ignore no-explicit-any ): any { const className = pascalCase(spec.name); @@ -390,8 +407,30 @@ export function buildGuestResourceClass( ); } const where = `${className} constructor`; - const lowered = lowerArgs(args, spec.ctorParams ?? [], where); - const rep = spec.ctor(...lowered); + const { lowered, release } = lowerArgs( + args, + spec.ctorParams ?? [], + where, + ); + let rep: unknown; + try { + rep = spec.ctor(...lowered); + } catch (e) { + try { + release(); + } finally { + throw e; + } + } + try { + release(); + } catch (e) { + try { + if (typeof rep === "number") hostDtorCall(rt, rep); + } finally { + throw e; + } + } if (rep !== null && typeof rep === "object" && "then" in rep) { throw new TypeError( `${where}: the guest constructor did not complete synchronously. ` + @@ -501,14 +540,16 @@ export function makeWrapper( * a registry. */ export class HostResourceRegistry { - readonly #byRep = new Map(); + readonly #byRep = new Map< + number, + { instance: object; owns: boolean; borrows: number; pendingDrop: boolean } + >(); readonly #byInstance = new WeakMap(); #next = 1; constructor(readonly className: string) {} - /** The host is passing an instance to the guest: allocate (or reuse) a rep. */ - repFor(instance: unknown): number { + #repFor(instance: unknown): number { if (instance === null || typeof instance !== "object") { throw new TypeError( `${this.className}: expected a class instance, got ${typeof instance}`, @@ -517,11 +558,53 @@ export class HostResourceRegistry { const held = this.#byInstance.get(instance); if (held !== undefined && this.#byRep.has(held)) return held; const rep = this.#next++; - this.#byRep.set(rep, instance); + this.#byRep.set(rep, { + instance, + owns: false, + borrows: 0, + pendingDrop: false, + }); this.#byInstance.set(instance, rep); return rep; } + /** The host is passing an own to the guest: retain until release or drop. */ + repFor(instance: unknown): number { + const rep = this.#repFor(instance); + const entry = this.#byRep.get(rep)!; + if (entry.pendingDrop) { + throw new InvalidHandleError( + `${this.className}: cannot transfer an instance pending drop as own`, + ); + } + entry.owns = true; + return rep; + } + + /** Retain a mapping for every overlapping call, independently of ownership. */ + borrowFor(instance: unknown): { rep: number; release: () => void } { + const rep = this.#repFor(instance); + const entry = this.#byRep.get(rep)!; + entry.borrows += 1; + let released = false; + return { + rep, + release: () => { + if (released) return; + released = true; + entry.borrows -= 1; + if (entry.borrows === 0 && !entry.owns) { + this.#byRep.delete(rep); + if (entry.pendingDrop) { + entry.pendingDrop = false; + (entry.instance as { [Symbol.dispose]?: () => void }) + [Symbol.dispose]?.(); + } + } + }, + }; + } + /** Is this instance already registered with a live rep? */ hasInstance(instance: unknown): boolean { if (instance === null || typeof instance !== "object") return false; @@ -534,11 +617,6 @@ export class HostResourceRegistry { return this.#byRep.has(rep); } - /** Release a rep if it is still live; no dtor, no error when already gone. */ - releaseIfPresent(rep: number): void { - this.#byRep.delete(rep); - } - /** A `borrow` arrived from the guest: the host's own instance, mapping kept. */ lookup(rep: number): object { const inst = this.#byRep.get(rep); @@ -547,7 +625,7 @@ export class HostResourceRegistry { `${this.className}: no live instance for rep ${rep}`, ); } - return inst; + return inst.instance; } /** @@ -556,7 +634,9 @@ export class HostResourceRegistry { */ release(rep: number): object { const inst = this.lookup(rep); - this.#byRep.delete(rep); + const entry = this.#byRep.get(rep)!; + entry.owns = false; + if (entry.borrows === 0) this.#byRep.delete(rep); return inst; } @@ -565,9 +645,14 @@ export class HostResourceRegistry { * `HostResourceType` dtor the executor calls from `canon_resource_drop`. */ dtor(rep: number): void { - const inst = this.#byRep.get(rep); - if (inst === undefined) return; - this.#byRep.delete(rep); + const entry = this.#byRep.get(rep); + if (entry === undefined || !entry.owns) return; + if (entry.borrows > 0) { + entry.owns = false; + entry.pendingDrop = true; + return; + } + const inst = this.release(rep); (inst as { [Symbol.dispose]?: () => void })[Symbol.dispose]?.(); } diff --git a/runtime/src/embedder/streams.ts b/runtime/src/embedder/streams.ts index e904bc3..91682bf 100644 --- a/runtime/src/embedder/streams.ts +++ b/runtime/src/embedder/streams.ts @@ -479,19 +479,35 @@ export class StreamWriter implements ProtocolStreamWriter { * chunk a BORROW until the returned promise settles; mutating it in that * window is misuse. Plain-array chunks are lowered (copied) up front. */ - async write(values: Chunk): Promise { + write(values: Chunk): Promise { + return this.#write(values, false); + } + + async #write(values: Chunk, all: boolean): Promise { await this.#stream.whenBound(); const host = hostOf(this.#stream); 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 (§"Streams and futures"). A full take - // genuinely completed before the trap and stays a success. - if (n < values.length) throwIfPeerTrapped(host.value, where, n); + const codec = this.#stream.codec!; + const lowered = packChunk(values, codec); + const info = codec.release === undefined ? undefined : { progress: 0 }; + let n: number; + try { + n = await host.writable[all ? "writeAll" : "write"]( + lowered as unknown as T[], + info, + ); + // Full takes completed before a later peer fault and keep their result. + if (n < values.length) throwIfPeerTrapped(host.value, where, n); + } catch (e) { + try { + releaseUntaken(lowered, info?.progress ?? 0, codec); + } catch { + // Cleanup attempted every tail element; preserve the write failure. + } + throw e; + } + releaseUntaken(lowered, n, codec); return n; } @@ -536,16 +552,8 @@ export class StreamWriter implements ProtocolStreamWriter { } /** 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); - 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; + writeAll(values: Chunk): Promise { + return this.#write(values, true); } cancelWrite(): void { @@ -636,11 +644,16 @@ export class Future implements ProtocolFuture { static deferred( pending: Promise, codec: ElemCodec, + finish?: (succeeded: boolean, raw: unknown) => void, ): Future { const hostP = pending.then((v) => { + finish?.(true, v); const h = hostFutureFor(v); (f as unknown as { adopt(h: HostFuture): void }).adopt(h); return h; + }, (e) => { + finish?.(false, e); + throw e; }); // Backstop (issue #182): a deferred handle that is never awaited, // dropped, or cancelled still has `#hostP` sitting there uninspected — if @@ -857,13 +870,18 @@ function packChunk( codec: ElemCodec, ): ComponentValue[] | Uint8Array { const u8 = isU8Element(codec.element); - if (values instanceof Uint8Array) { - if (u8) return values; - return Array.from(values as ArrayLike).map((v) => - codec.fromHost(v as T) - ) as ComponentValue[]; + if (values instanceof Uint8Array && u8) return values; + const lowered: ComponentValue[] = []; + try { + for (const v of values) lowered.push(codec.fromHost(v as T)); + } catch (e) { + try { + releaseUntaken(lowered, 0, codec); + } catch { + // Preserve the invalid element's error after releasing the prefix. + } + throw e; } - const lowered = (values as readonly T[]).map((v) => codec.fromHost(v)); return u8 ? Uint8Array.from(lowered as number[]) : (lowered as ComponentValue[]); @@ -879,6 +897,7 @@ async function pump( ): Promise { const where = codec.where ?? "stream producer"; let failure: unknown; + let failed = false; let produced = 0; // resource stream cancellation companion: the pump learns of the reader dropping // through short writes, but a producer PARKED on an external event (an @@ -894,33 +913,36 @@ async function pump( // Lowering is the likeliest failure (a value of the wrong shape) and it // must be attributed to the site, not swallowed into a short stream. const lowered = packChunk(batch, codec) as unknown as T[]; + const info = codec.release === undefined ? undefined : { progress: 0 }; let n: number; try { - n = await host.writable.writeAll(lowered); + n = await host.writable.writeAll(lowered, info); + if (n < lowered.length) throwIfPeerTrapped(host.value, where, n); } catch (e) { - // resource stream: elements past the fault's progress point were lowered but - // will never be taken — destroy them (an `own` element may hold a - // live platform resource). `PeerTrappedError.progress` reports - // delivered-before-the-fault; anything else delivered nothing. - releaseUntaken( - lowered as unknown as ComponentValue[], - e instanceof PeerTrappedError ? e.progress ?? 0 : 0, - codec, - ); + try { + releaseUntaken( + lowered as unknown as ComponentValue[], + info?.progress ?? 0, + codec, + ); + } catch { + // Preserve the producer/peer failure after attempting every release. + } throw e; } + releaseUntaken(lowered as unknown as ComponentValue[], n, codec); produced += n; if (n < lowered.length) { // The reader went away: a clean end — but the un-taken tail of this // chunk was already lowered and must be destroyed, not leaked. - releaseUntaken(lowered as unknown as ComponentValue[], n, codec); break; } } } catch (e) { failure = e; + failed = true; } - if (failure !== undefined) { + if (failed) { void produced; // Report BEFORE dropping: the drop is what lets the guest see // end-of-stream and resolve, and the driving loop checks `hostFailure` @@ -946,7 +968,17 @@ function releaseUntaken( ): void { const release = codec.release; if (release === undefined || lowered instanceof Uint8Array) return; - for (let i = taken; i < lowered.length; i++) release(lowered[i]); + let failure: unknown; + let failed = false; + for (let i = taken; i < lowered.length; i++) { + try { + release(lowered[i]); + } catch (e) { + if (!failed) failure = e; + failed = true; + } + } + if (failed) throw failure; } /** diff --git a/runtime/src/exec/boundary.ts b/runtime/src/exec/boundary.ts index 1b4f3e7..e0b6a94 100644 --- a/runtime/src/exec/boundary.ts +++ b/runtime/src/exec/boundary.ts @@ -1621,6 +1621,9 @@ export function createLiftedFunction(input: { * Promise, which it can never advance, and declares a bogus deadlock. */ allowAsyncCompletion?: boolean; + /** Nested guest destructor only: preserve the caller and use the reference + * sync lift drive, not the host's store-wide completion policy. */ + guestDtorCaller?: ComponentInstanceState | null; /** * Refuse — synchronously, before entering — a call made while the instance * has HOP-parked activations, instead of deferring it (§"Functions and async", @@ -1662,6 +1665,7 @@ export function createLiftedFunction(input: { const inst = opts.instance; const store = inst.store; const mode: SuspensionMode = input.suspensionMode ?? "plain"; + const guestDtor = input.guestDtorCaller !== undefined; // Entry wrapping, half of jspi/bridge.ts's invariant: a lifted export's core // function is one of the three activations that can reach a blocking // built-in, so it is `promising`-wrapped exactly when the imports are @@ -1701,7 +1705,7 @@ export function createLiftedFunction(input: { stats.liftedCalls++; // A trap remembered during an earlier call must never be attributed to // this one (see intrinsics `HostTrapState`). - if (trapState !== undefined) trapState.pending = undefined; + if (!guestDtor && trapState !== undefined) trapState.pending = undefined; // Depth of the sync-call scope stack on entry; see the `finally` below. const syncCallDepth = syncCallStack?.length ?? 0; @@ -1713,7 +1717,7 @@ export function createLiftedFunction(input: { { const refusal = entryRefusal( inst, - null, + input.guestDtorCaller ?? null, `cannot enter component instance ${inst.index}`, ); if (refusal !== null) trap(refusal); @@ -1815,7 +1819,7 @@ export function createLiftedFunction(input: { // `poison` below) and must stay exactly as the trap left it. Restoring // its `may_leave` would be tidying the state of an instance that is no // longer allowed to run at all. - for (const i of allInstances?.() ?? []) { + for (const i of guestDtor ? [] : allInstances?.() ?? []) { if (i as unknown as ComponentInstanceState !== inst) { i.mayLeave = true; } @@ -1911,6 +1915,9 @@ export function createLiftedFunction(input: { if (!ft.async && mode !== "jspi" && !input.allowAsyncCompletion) { driveSyncLift(task); } + // The dropping activation is still on the stack. In particular, its + // own JSPI hop must not turn this completed sync call into a Promise. + if (guestDtor) return finishHostEntry(); } catch (e) { unwind(); if (!isCapabilitySignal(e)) poison(e); @@ -2282,8 +2289,11 @@ export function createDtorEntry(input: { trapState?: { pending: unknown }; syncCallStack?: LenderScope[]; allInstances?: () => Iterable<{ mayLeave: boolean }>; + /** Present only for guest drops; null is a guest call without a real caller. */ + guestCaller?: ComponentInstanceState | null; }): (rep: number) => unknown { - const mode = input.suspensionMode ?? "plain"; + const guest = input.guestCaller !== undefined; + const mode = guest ? "plain" : input.suspensionMode ?? "plain"; const raw: CoreFn = input.dtor ?? (() => undefined); // A dtor's core type is `(i32) -> ()`, but the *host*-supplied dtors this // helper also serves (embedder test doubles, `ResourceTypeInfo` built @@ -2295,6 +2305,10 @@ export function createDtorEntry(input: { // real wasm dtor returns nothing by construction). const core: CoreFn = mode === "jspi" ? raw : ((rep: number) => { const r = raw(rep); + trapIf( + guest && isPromiseLike(r), + "resource destructor did not complete synchronously", + ); return isPromiseLike(r) ? r : undefined; }); const lifted = createLiftedFunction({ @@ -2309,7 +2323,8 @@ export function createDtorEntry(input: { allInstances: input.allInstances, // The host does not wait for a destructor: `drop(): void` is // non-blocking, and an unfinished dtor's tail is driven by the store. - allowAsyncCompletion: true, + allowAsyncCompletion: !guest, + guestDtorCaller: input.guestCaller, }); return (rep: number) => lifted(rep); } diff --git a/runtime/src/exec/executor.ts b/runtime/src/exec/executor.ts index 51c32ba..d44535c 100644 --- a/runtime/src/exec/executor.ts +++ b/runtime/src/exec/executor.ts @@ -788,9 +788,7 @@ class Executor { if (table.kind === "concrete" && table.resource === resourceIndex) { const token = this.loaded.resourceTokens[tableIndex]; token.impl = inst; - token.dtor = dtor === null ? null : (rep: number) => { - dtor(rep); - }; + token.dtor = dtor; // #85/#160: the host-initiated-drop entry. A host-initiated // drop is a full canonical LIFT of the dtor (definitions.py // `canon_resource_drop`, line 2319), so it is built here with diff --git a/runtime/src/exec/host_streams.ts b/runtime/src/exec/host_streams.ts index c296063..95d11bc 100644 --- a/runtime/src/exec/host_streams.ts +++ b/runtime/src/exec/host_streams.ts @@ -845,7 +845,8 @@ export interface HostWritableEnd { * stay parked across several partial reads); mutating it in that window is * misuse. Readers always receive their own copy. */ - write(values: T[]): Promise; + // Internal out-parameter: transferred prefix, including on rejection. + write(values: T[], info?: { progress: number }): Promise; /** * Offer `values` repeatedly until all of them have been taken or the reader * goes away. Convenience over `write`, and the shape most embedders want. @@ -860,7 +861,7 @@ export interface HostWritableEnd { * Resolves with the total accepted, which is less than `values.length` only * if the reader dropped. */ - writeAll(values: T[]): Promise; + writeAll(values: T[], info?: { progress: number }): Promise; /** * Park a **direct session** on this end (`stream` only — embedder-api * §"Streams and futures" ("Direct-access byte edges") (polyengine#128)). @@ -1040,6 +1041,7 @@ function mkStreamEnds( // `SharedBase.cancel` retires whatever is parked, so cancelling is only // legal (and only meaningful) while the parked side is ours. const parked = { read: false, write: false }; + let writeAll: "active" | "cancelled" | null = null; /** * Settle bookkeeping for a completed copy. `DROPPED` means the peer end is * gone: no further host activity on this end is possible, so the activity @@ -1167,9 +1169,45 @@ function mkStreamEnds( activity.notify(); activity.pump(); }; + const write = (values: T[], info?: { progress: number }): Promise => { + const start = info?.progress ?? 0; + const buf = new HostBuffer( + shared.t, + values as unknown as ComponentValue[], + values.length, + ); + return new Promise((resolve, reject) => { + parked.write = true; + const done = (result: CopyResult): void => { + parked.write = false; + if (info !== undefined) info.progress = start + buf.progress; + settle(result); + resolve(buf.progress); + }; + try { + shared.write( + writeInst, + buf as never, + (reclaim) => { + // A host offer stays parked across partial peer reads. + if (buf.remain() > 0) return; + reclaim(); + done(CopyResult.COMPLETED); + }, + done, + ); + activity.notify(); + activity.pump(); + } catch (e) { + if (info !== undefined) info.progress = start + buf.progress; + reject(e); + withdraw("write", buf); + } + }); + }; return { writable: { - write(values: T[]): Promise { + write(values: T[], info?: { progress: number }): 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 @@ -1178,70 +1216,47 @@ function mkStreamEnds( // 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) { + if (parked.write || writeAll !== null) { 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[], - values.length, - ); - return new Promise((resolve) => { - parked.write = true; - shared.write( - writeInst, - buf as never, - // `on_copy`: a partial rendezvous happened. A guest end would be - // handed a COMPLETED event here and decide for itself whether to - // re-offer; a host end has no event loop, so we make the useful - // choice and **stay parked** until the offer is exhausted. That is - // exactly the shape wit-bindgen's `wit_stream::new()` produces — - // a background write that the reader drains a few elements at a - // time — and it is why `reclaim` is deliberately not called while - // values remain: reclaiming retires the pending buffer and the - // next guest read would find nothing. - (reclaim) => { - if (buf.remain() > 0) return; // still parked; more to give - reclaim(); - parked.write = false; - activity.notify(); - resolve(buf.progress); - }, - (result: CopyResult) => { - parked.write = false; - settle(result); - resolve(buf.progress); - }, - ); - activity.notify(); - try { - activity.pump(); - } catch (e) { - withdraw("write", buf); - throw e; - } - }); + return write(values, info); }, - async writeAll(values: T[]): Promise { + async writeAll( + values: T[], + info?: { progress: number }, + ): Promise { + if (parked.write || writeAll !== null) { + throw new TypeError( + "a write is already in flight on this stream's writable end; " + + "await it or cancelWrite() first", + ); + } + writeAll = "active"; let sent = 0; - while (sent < values.length && !shared.dropped) { - // Re-offers keep `write`'s borrow semantics: the first round is the - // chunk itself and later rounds a `subarray` VIEW for typed chunks - // (review F1: a `slice` here cost a second full copy on the very - // path the one-copy contract names), a `slice` for plain arrays. - const rest = sent === 0 - ? values - : values instanceof Uint8Array - ? values.subarray(sent) as unknown as T[] - : values.slice(sent); - const n = await this.write(rest); - if (n === 0) break; // reader gone; nothing more will be taken - sent += n; + try { + while ( + sent < values.length && !shared.dropped && writeAll === "active" + ) { + // Re-offers keep `write`'s borrow semantics: the first round is the + // chunk itself and later rounds a `subarray` VIEW for typed chunks + // (review F1: a `slice` here cost a second full copy on the very + // path the one-copy contract names), a `slice` for plain arrays. + const rest = sent === 0 + ? values + : values instanceof Uint8Array + ? values.subarray(sent) as unknown as T[] + : values.slice(sent); + const n = await write(rest, info); + if (n === 0) break; // reader gone; nothing more will be taken + sent += n; + } + return sent; + } finally { + writeAll = null; } - return sent; }, writeDirect( produce: (dest: DirectDestination) => DirectVerdict, @@ -1249,7 +1264,7 @@ function mkStreamEnds( ): Promise { // Same one-in-flight-per-end rule, same wording shape as `write`: // `writeDirect` participates in it exactly as `write` does. - if (parked.write) { + if (parked.write || writeAll !== null) { throw new TypeError( "a write is already in flight on this stream's writable end; " + "await it or cancelWrite() first", @@ -1265,6 +1280,8 @@ function mkStreamEnds( }); }, cancelWrite() { + // Cancellation owns the whole helper, including gaps between offers. + if (writeAll !== null) writeAll = "cancelled"; if (!parked.write) return; const session = direct.write; if (session !== null) return cancelDirect(session); @@ -1497,27 +1514,26 @@ function mkFuture( // Distinct rendezvous identities per end — see `hostEndInstance`. const writeInst = hostEndInstance("write"); const readInst = hostEndInstance("read"); - const parked = { any: false }; + const parked = { read: false, write: false }; /** Set once the future's one value has actually crossed (#90). */ let delivered = false; - const settle = (result: CopyResult): void => { - parked.any = false; + const settle = (side: "read" | "write", result: CopyResult): void => { + parked[side] = false; if (result === CopyResult.COMPLETED) delivered = true; if (result === CopyResult.DROPPED) activity.close(); else activity.notify(); }; /** See `mkStreamEnds`' `withdraw`: the pump-trap unwind path (F1). */ - const withdraw = (buf: unknown): void => { - if (!parked.any) return; - parked.any = false; + const withdraw = (side: "read" | "write", buf: unknown): void => { + if (!parked[side]) return; + parked[side] = false; if (shared.pendingBuffer === buf as never) shared.cancel(); activity.notify(); }; 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) { + // Opposite ends may rendezvous after a guest round trip. + if (parked.write) { throw new TypeError( "an operation is already in flight on this future; " + "await it or cancel() first", @@ -1526,24 +1542,24 @@ function mkFuture( // 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); - return new Promise((resolve) => { - parked.any = true; - shared.write(writeInst, buf as never, (result: CopyResult) => { - settle(result); - resolve(); - }); - activity.notify(); + return new Promise((resolve, reject) => { + parked.write = true; try { + shared.write(writeInst, buf as never, (result: CopyResult) => { + settle("write", result); + resolve(); + }); + activity.notify(); activity.pump(); } catch (e) { - withdraw(buf); - throw e; + reject(e); + withdraw("write", buf); } }); }, readResult(): Promise<{ value: T | undefined; result: CopyResult }> { - // One in-flight operation per wrapper — see write(). - if (parked.any) { + // One in-flight operation per readable end — see write(). + if (parked.read) { throw new TypeError( "an operation is already in flight on this future; " + "await it or cancel() first", @@ -1559,21 +1575,21 @@ function mkFuture( }); } const buf = new HostBuffer(shared.t, null, 1); - return new Promise((resolve) => { - parked.any = true; - shared.read(readInst, buf as never, (result: CopyResult) => { - settle(result); - resolve({ - value: buf.taken()[0] as unknown as T | undefined, - result, - }); - }); - activity.notify(); + return new Promise((resolve, reject) => { + parked.read = true; try { + shared.read(readInst, buf as never, (result: CopyResult) => { + settle("read", result); + resolve({ + value: buf.taken()[0] as unknown as T | undefined, + result, + }); + }); + activity.notify(); activity.pump(); } catch (e) { - withdraw(buf); - throw e; + reject(e); + withdraw("read", buf); } }); }, @@ -1581,8 +1597,7 @@ function mkFuture( return (await self.readResult()).value; }, cancel(): void { - if (!parked.any) return; - parked.any = false; + if (!parked.read && !parked.write) return; shared.cancel(); activity.notify(); activity.pump(); diff --git a/runtime/src/intrinsics/async_builtins.ts b/runtime/src/intrinsics/async_builtins.ts index 89c7dea..bd77597 100644 --- a/runtime/src/intrinsics/async_builtins.ts +++ b/runtime/src/intrinsics/async_builtins.ts @@ -73,6 +73,7 @@ import type { ComponentInstanceState } from "../task/mod.ts"; import type { CoreFn, ResolvedOptions } from "../exec/boundary.ts"; import { cabiOptions, normalizeCoreValues } from "../exec/boundary.ts"; import { traceCopy } from "./stream_builtins.ts"; +import { removeHandleWithUnwind } from "../task/scheduler.ts"; /** Services these built-ins need from the executor. */ export interface AsyncTrampolineContext { @@ -385,12 +386,13 @@ export function createWaitableSetDrop(inst: ComponentInstanceState): CoreFn { !inst.mayLeave, "waitable-set.drop: cannot leave component instance", ); - const wset = inst.handles.remove(i); - trapIf( - !(wset instanceof WaitableSet), - "waitable-set.drop: handle is not a waitable set", - ); - (wset as WaitableSet).drop(); + removeHandleWithUnwind(inst, i, (wset) => { + trapIf( + !(wset instanceof WaitableSet), + "waitable-set.drop: handle is not a waitable set", + ); + (wset as WaitableSet).drop(); + }); }; } @@ -429,16 +431,17 @@ export function createSubtaskDrop(inst: ComponentInstanceState): CoreFn { return (i?: number) => { i = (i ?? 0) >>> 0; trapIf(!inst.mayLeave, "subtask.drop: cannot leave component instance"); - const s = inst.handles.remove(i); - trapIf(!(s instanceof Subtask), "subtask.drop: handle is not a subtask"); - (s as Subtask).drop(); + removeHandleWithUnwind(inst, i, (s) => { + trapIf(!(s instanceof Subtask), "subtask.drop: handle is not a subtask"); + (s as Subtask).drop(); + }); }; } /** * definitions.py `canon_subtask_cancel` (line 2469). * - * The synchronous form blocks (`subtask.wait_for_pending_event()`) when the + * The synchronous form blocks (`thread.wait_until(subtask.resolved)`) when the * callee does not resolve promptly; from a stackless guest that is JSPI * territory. The async form returns `BLOCKED` instead of blocking, and is * fully supported. @@ -591,8 +594,7 @@ export function createSubtaskCancel( // (lit), mirroring SITE 4 (stream_builtins.ts) and // `Waitable.waitForPendingEvent`. The ASYNC form answers BLOCKED as // soon as the callee is determinate and still unresolved. - const ready = (): boolean => - determinate() && (async_ || st.hasPendingEvent()); + const ready = (): boolean => determinate() && (async_ || st.resolved()); if (mode !== "jspi") { if (st.resolved()) return finish(); diff --git a/runtime/src/intrinsics/mod.ts b/runtime/src/intrinsics/mod.ts index e696986..0cb78f0 100644 --- a/runtime/src/intrinsics/mod.ts +++ b/runtime/src/intrinsics/mod.ts @@ -21,6 +21,7 @@ import { trap, } from "../cabi/mod.ts"; import { ResourceHandle } from "../cabi/handles.ts"; +import { removeHandleWithUnwind } from "../task/scheduler.ts"; import { trapIf } from "../cabi/trap.ts"; import { assert_ } from "../cabi/trap.ts"; import type { ResourceTypeInfo } from "../cabi/types.ts"; @@ -832,17 +833,21 @@ function transferOwn( const srcRt = ctx.resourceToken(srcTable); const dstRt = ctx.resourceToken(dstTable); - const h = src.handles.remove(handle); - trapIf(!(h instanceof ResourceHandle), "transfer-own: not a resource handle"); - const rh = h as ResourceHandle; - trapIf(rh.rt !== srcRt, "transfer-own: resource type mismatch"); - // definitions.py `lift_own`: `trap_if(h.num_lends != 0)`. - trapIf( - rh.numLends !== 0, - "cannot remove owned resource while borrowed (handle still lent out)", - ); - trapIf(!rh.own, "transfer-own: expected an owning handle"); - return dst.handles.add(new ResourceHandle(dstRt, rh.rep, true)); + return removeHandleWithUnwind(src, handle, (h) => { + trapIf( + !(h instanceof ResourceHandle), + "transfer-own: not a resource handle", + ); + const rh = h as ResourceHandle; + trapIf(rh.rt !== srcRt, "transfer-own: resource type mismatch"); + // definitions.py `lift_own`: `trap_if(h.num_lends != 0)`. + trapIf( + rh.numLends !== 0, + "cannot remove owned resource while borrowed (handle still lent out)", + ); + trapIf(!rh.own, "transfer-own: expected an owning handle"); + return dst.handles.add(new ResourceHandle(dstRt, rh.rep, true)); + }); } /** diff --git a/runtime/src/intrinsics/stream_builtins.ts b/runtime/src/intrinsics/stream_builtins.ts index e89d8ca..a62960c 100644 --- a/runtime/src/intrinsics/stream_builtins.ts +++ b/runtime/src/intrinsics/stream_builtins.ts @@ -55,6 +55,7 @@ import { type ResolvedOptions, } from "../exec/boundary.ts"; import { BLOCKED } from "./async_builtins.ts"; +import { removeHandleWithUnwind } from "../task/scheduler.ts"; /** * Standing probe (CE_COPY_TRACE=1): per-call return codes of the copy / @@ -500,11 +501,12 @@ function dropEnd( // Guest-supplied index is u32; core wasm delivers i32 args signed (F3, R2). hi = hi >>> 0; trapIf(!inst.mayLeave, `${what}: cannot leave component instance`); - const e = inst.handles.remove(hi); - trapIf(!(e instanceof EndT), `${what}: wrong end type for this handle`); - const end = e as CopyEnd; - trapIf(!sameElem(end.shared.t, elem), `${what}: element type mismatch`); - end.drop(); + removeHandleWithUnwind(inst, hi, (e) => { + trapIf(!(e instanceof EndT), `${what}: wrong end type for this handle`); + const end = e as CopyEnd; + trapIf(!sameElem(end.shared.t, elem), `${what}: element type mismatch`); + end.drop(); + }); } // --------------------------------------------------------------------------- @@ -573,11 +575,12 @@ export function createErrorContextDrop( !inst.mayLeave, "error-context.drop: cannot leave component instance", ); - const e = inst.handles.remove(i); - trapIf( - !(e instanceof ErrorContext), - errorContextTrapMessage("error-context.drop", e), - ); + removeHandleWithUnwind(inst, i, (e) => { + trapIf( + !(e instanceof ErrorContext), + errorContextTrapMessage("error-context.drop", e), + ); + }); }; } @@ -846,33 +849,40 @@ function transferAsyncEnd(input: { what: string; }): number { const { EndT, srcInst, dstInst, srcElem, dstElem, srcIdx, what } = input; - const e = srcInst.handles.remove(srcIdx); - trapIf(!(e instanceof EndT), `${what}: handle is not a readable ${what} end`); - const end = e as CopyEnd; - trapIf(!sameElem(end.shared.t, srcElem), `${what}: source element mismatch`); - trapIf( - !sameElem(end.shared.t, dstElem), - `${what}: destination element mismatch`, - ); - // definitions.py `lift_async_value`: an end that is mid-copy or parked in a - // waitable set cannot be handed on. The messages match the suite's - // `assert_trap` text. - trapIf( - end.state === CopyState.DONE, - what === "future" - ? "cannot lift future after previous read succeeded" - : "cannot lift stream after being notified that the writable end dropped", - ); - trapIf( - end.state !== CopyState.IDLE, - `cannot remove busy ${what}`, - ); - trapIf( - end.inWaitableSet(), - `cannot lift ${what} while it's in a waitable set`, - ); - const Ctor = EndT as unknown as new (shared: unknown) => CopyEnd; - return dstInst.handles.add(new Ctor(end.shared)); + return removeHandleWithUnwind(srcInst, srcIdx, (e) => { + trapIf( + !(e instanceof EndT), + `${what}: handle is not a readable ${what} end`, + ); + const end = e as CopyEnd; + trapIf( + !sameElem(end.shared.t, srcElem), + `${what}: source element mismatch`, + ); + trapIf( + !sameElem(end.shared.t, dstElem), + `${what}: destination element mismatch`, + ); + // definitions.py `lift_async_value`: an end that is mid-copy or parked in a + // waitable set cannot be handed on. The messages match the suite's + // `assert_trap` text. + trapIf( + end.state === CopyState.DONE, + what === "future" + ? "cannot lift future after previous read succeeded" + : "cannot lift stream after being notified that the writable end dropped", + ); + trapIf( + end.state !== CopyState.IDLE, + `cannot remove busy ${what}`, + ); + trapIf( + end.inWaitableSet(), + `cannot lift ${what} while it's in a waitable set`, + ); + const Ctor = EndT as unknown as new (shared: unknown) => CopyEnd; + return dstInst.handles.add(new Ctor(end.shared)); + }); } export function createStreamTransfer(ctx: AsyncTransferContext): CoreFn { diff --git a/runtime/src/task/scheduler.ts b/runtime/src/task/scheduler.ts index a3df2b0..aa55e2a 100644 --- a/runtime/src/task/scheduler.ts +++ b/runtime/src/task/scheduler.ts @@ -47,6 +47,7 @@ // rather than pretending (see `needsJspi`). import { assert_, trapIf } from "../cabi/trap.ts"; +import type { ComponentInstanceLike } from "../cabi/context.ts"; /** definitions.py `Cancelled` (line 248). */ export const CANCELLED_FALSE = false; @@ -156,11 +157,41 @@ let onInstancePoisoned: | ((inst: { handles: Iterable }, cause: unknown) => void) | null = null; +let onHandleRemovalFailed: + | ((inst: ComponentInstanceLike, entry: unknown, cause: unknown) => void) + | null = null; + +/** Preserve destructive table removal, retiring an async end only if its + * validation/drop fails. The hook avoids a handles -> streams import cycle. */ +export function removeHandleWithUnwind( + inst: ComponentInstanceLike, + i: number, + use: (entry: unknown) => T, +): T { + const entry = inst.handles.remove(i); + try { + return use(entry); + } catch (cause) { + try { + onHandleRemovalFailed?.(inst, entry, cause); + } catch { + // A peer notification must not replace the original failure. + } + throw cause; + } +} + /** @internal — see `onInstancePoisoned`; registered once by task/streams.ts. */ export function setOnInstancePoisoned( f: (inst: { handles: Iterable }, cause: unknown) => void, + removalFailed?: ( + inst: ComponentInstanceLike, + entry: unknown, + cause: unknown, + ) => void, ): void { onInstancePoisoned = f; + if (removalFailed !== undefined) onHandleRemovalFailed = removalFailed; } /** diff --git a/runtime/src/task/streams.ts b/runtime/src/task/streams.ts index f03a67f..a133aba 100644 --- a/runtime/src/task/streams.ts +++ b/runtime/src/task/streams.ts @@ -1037,19 +1037,18 @@ export function dropSharedForTeardown( ): void { if (shared.dropped) return; shared.dropped = true; - if (shared.pendingBuffer) { - const pi = shared.pendingInst; - const parkedInDeadGuest = typeof pi === "object" && pi !== null && - (isInstancePoisoned(pi) || retiredInstances.has(pi)); - if (parkedInDeadGuest) shared.resetPending(); - else shared.resetAndNotifyPending(CopyResult.DROPPED); + try { + if (shared.pendingBuffer) { + const pi = shared.pendingInst; + const parkedInDeadGuest = typeof pi === "object" && pi !== null && + (isInstancePoisoned(pi) || retiredInstances.has(pi)); + if (parkedInDeadGuest) shared.resetPending(); + else shared.resetAndNotifyPending(CopyResult.DROPPED); + } + } finally { + // Release producer/host retention even if the peer's notification throws. + shared.notifyDropped(); } - // The drop observers also fire on the teardown path: a stream producer - // parked behind a trap-poisoned reader must be cancelled the same as behind - // a cleanly-dropped one, and a host wrapper's activity arm must be - // released the same way (#162, §"Streams and futures"). Both classes carry the - // observer machinery, so this is unconditional. - shared.notifyDropped(); } /** @@ -1097,12 +1096,20 @@ export function retireInstanceAsyncEnds( ): void { if (retiredInstances.has(inst)) return; retiredInstances.add(inst); - const where = inst.index !== undefined - ? `component instance ${inst.index}` - : "a component instance"; // Snapshot: the notifications below can run peer code that mutates tables. const ends: CopyEnd[] = []; for (const e of inst.handles) if (e instanceof CopyEnd) ends.push(e); + retireAsyncEnds(inst, ends, cause); +} + +function retireAsyncEnds( + inst: PoisonedInstanceLike, + ends: CopyEnd[], + cause: unknown, +): void { + const where = inst.index !== undefined + ? `component instance ${inst.index}` + : "a component instance"; // Pass 1: record the failure, and mark abandoned every future this table // owes a value on. Done before ANY notification, so the reader-side trap @@ -1147,7 +1154,14 @@ export function retireInstanceAsyncEnds( // `Store.tick`'s poisoning 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); +setOnInstancePoisoned(retireInstanceAsyncEnds, (inst, entry, cause) => { + if (!(entry instanceof CopyEnd)) return; + const shared = entry.shared as SharedStreamImpl | SharedFutureImpl; + // The removed end is unreachable even before boundary poisoning. Retract + // its buffer silently without marking the whole instance dead or retired. + if (shared.pendingInst === inst) shared.resetPending(); + retireAsyncEnds(inst, [entry], cause); +}); // --------------------------------------------------------------------------- // error-context (definitions.py `class ErrorContext`, line 2775) diff --git a/runtime/tests/async_builtins_test.ts b/runtime/tests/async_builtins_test.ts index 9f42047..57df1b6 100644 --- a/runtime/tests/async_builtins_test.ts +++ b/runtime/tests/async_builtins_test.ts @@ -567,14 +567,9 @@ Deno.test( assertEq(f.subtask.resolved(), false); assertEq(f.subtask.hasSyncWaiter, true); - // The callee resolves later; the park's `readyFunc` (`hasPendingEvent`) - // only fires once the event is actually armed — not merely once - // `resolved()` is true — so drive both steps to pin the ordering SITE 5 - // now depends on. - f.subtask.resolve(SubtaskState.CANCELLED_BEFORE_RETURNED, []); - // Not yet armed: the park must not be ready on `resolved()` alone. - assertEq(f.store.tick(), false); - f.subtask.setSubtaskPendingEvent(f.subtaski); + // Production resolution and its progress notification are one synchronous + // transition; there is no scheduler turn between them. + resolveSubtask(f.subtask, f.subtaski); assertEq(f.store.tick(), true); const rc = await pending; diff --git a/runtime/tests/dtor_guest_context.wasm b/runtime/tests/dtor_guest_context.wasm new file mode 100644 index 0000000..f830c2d Binary files /dev/null and b/runtime/tests/dtor_guest_context.wasm differ diff --git a/runtime/tests/dtor_guest_context.wat b/runtime/tests/dtor_guest_context.wat new file mode 100644 index 0000000..65abf80 --- /dev/null +++ b/runtime/tests/dtor_guest_context.wat @@ -0,0 +1,40 @@ +(component + (canon context.get i32 0 (core func $get)) + (canon context.set i32 0 (core func $set)) + (canon task.return (core func $return)) + (core module $D + (import "" "get" (func $get (result i32))) + (import "" "set" (func $set (param i32))) + (global $seen (mut i32) (i32.const -1)) + (func (export "dtor") (param i32) + (global.set $seen (call $get)) + (call $set (i32.const 99))) + (func (export "seen") (result i32) (global.get $seen))) + (core instance $d (instantiate $D (with "" (instance + (export "get" (func $get)) (export "set" (func $set)))))) + (type $R (resource (rep i32) (dtor (core func $d "dtor")))) + (canon resource.new $R (core func $new)) + (canon resource.drop $R (core func $drop)) + (core module $M + (import "" "new" (func $new (param i32) (result i32))) + (import "" "drop" (func $drop (param i32))) + (import "" "get" (func $get (result i32))) + (import "" "set" (func $set (param i32))) + (import "" "return" (func $return)) + (func $probe (export "probe") (result i32) + (call $set (i32.const 42)) + (call $drop (call $new (i32.const 7))) + (call $get)) + (func (export "async-probe") (result i32) + (if (i32.ne (call $probe) (i32.const 42)) (then unreachable)) + (call $return) + (i32.const 0)) + (func (export "callback") (param i32 i32 i32) (result i32) unreachable)) + (core instance $m (instantiate $M (with "" (instance + (export "new" (func $new)) (export "drop" (func $drop)) + (export "get" (func $get)) (export "set" (func $set)) + (export "return" (func $return)))))) + (func (export "probe") (result u32) (canon lift (core func $m "probe"))) + (func (export "async-probe") async + (canon lift (core func $m "async-probe") async (callback (core func $m "callback")))) + (func (export "seen") (result u32) (canon lift (core func $d "seen")))) diff --git a/runtime/tests/dtor_guest_context_test.ts b/runtime/tests/dtor_guest_context_test.ts new file mode 100644 index 0000000..d0cdc0c --- /dev/null +++ b/runtime/tests/dtor_guest_context_test.ts @@ -0,0 +1,224 @@ +// definitions.py canon_resource_drop: a fresh synchronous lift/task/thread, +// even when the dropper is async. Host completion policy does not apply. +import { + canonResourceDrop, + canonResourceNew, + ResourceTypeInfo, + Trap, +} from "../src/cabi/mod.ts"; +import { + ComponentInstanceState, + currentTask, + currentThread, + Store, + type Thread, +} from "../src/task/mod.ts"; +import { + isInstancePoisoned, + maybeCurrentThread, + NeedsJspi, +} from "../src/task/scheduler.ts"; +import { + createDtorEntry, + createLiftedFunction, + newStats, + type ResolvedOptions, +} from "../src/exec/boundary.ts"; +import { instantiateComponent } from "../src/exec/mod.ts"; +import { Translator } from "../src/shim/mod.ts"; +import { assertEq } from "./support/asserts.ts"; + +function options(instance: ComponentInstanceState): ResolvedOptions { + return { + instance, + stringEncoding: "utf8", + memory: null, + realloc: null, + postReturn: null, + callback: null, + async: false, + cancellable: false, + coreType: { params: [], results: [] }, + }; +} + +Deno.test("guest dtor isolates task, both context slots, and post-return attribution", () => { + const store = new Store(); + const caller = new ComponentInstanceState(0, store); + const impl = new ComponentInstanceState(1, store); + let outer: Thread; + const rt = new ResourceTypeInfo(impl, (rep) => { + assertEq(rep, 7); + assertEq(currentTask().inst === impl, true); + assertEq(currentTask().ft.async, false); + assertEq(currentTask().opts.async_, false); + const thread = currentThread(); + assertEq(thread === outer, false); + assertEq(thread.storage, [0, 0]); + thread.storage[0] = 99; + thread.storage[1] = 100; + }); + const failure = new Trap("post-return trap"); + const opts = options(caller); + opts.postReturn = () => () => { + assertEq(currentThread() === outer, true); + assertEq(currentTask().inst === caller, true); + assertEq(outer.storage, [42, 43]); + throw failure; + }; + const call = createLiftedFunction({ + name: "drop", + ft: { params: [], results: [] }, + opts, + stats: newStats(), + core: () => { + outer = currentThread(); + outer.storage[0] = 42; + outer.storage[1] = 43; + canonResourceDrop(caller, rt, canonResourceNew(caller, rt, 7)); + assertEq(currentThread() === outer, true); + assertEq([...impl.threads].length, 0); + }, + }); + let caught: unknown; + try { + call(); + } catch (e) { + caught = e; + } + assertEq(caught === failure, true); + assertEq(isInstancePoisoned(caller), true); + assertEq(isInstancePoisoned(impl), false); + assertEq(maybeCurrentThread(), undefined); +}); + +for (const capability of [false, true]) { + Deno.test(`guest dtor unwind preserves outer state (capability=${capability})`, () => { + const store = new Store(); + const caller = new ComponentInstanceState(0, store); + const impl = new ComponentInstanceState(1, store); + const failure = capability + ? new NeedsJspi("dtor probe") + : new Trap("dtor trap"); + const trapState = { pending: failure as unknown }; + const entry = createDtorEntry({ + instance: impl, + guestCaller: caller, + trapState, + allInstances: () => [caller, impl], + dtor: () => { + assertEq(trapState.pending === failure, true); + assertEq(currentTask().inst === impl, true); + throw failure; + }, + }); + const call = createLiftedFunction({ + name: "outer", + ft: { params: [], results: [] }, + opts: options(caller), + stats: newStats(), + core: () => { + const outer = currentThread(); + caller.mayLeave = false; + let caught: unknown; + try { + entry(0); + } catch (e) { + caught = e; + } + assertEq(caught === failure, true); + assertEq(currentThread() === outer, true); + assertEq(caller.mayLeave, false); + assertEq(trapState.pending === failure, true); + caller.mayLeave = true; + }, + }); + call(); + assertEq(isInstancePoisoned(impl), !capability); + assertEq(isInstancePoisoned(caller), false); + }); +} + +for (const jspi of [false, true]) { + Deno.test(`translated guest dtor context, sync and async caller (jspi=${jspi})`, async () => { + const translator = await Translator.create( + await Deno.readFile( + new URL( + "../../target/wasm32-unknown-unknown/release/translator_shim.wasm", + import.meta.url, + ), + ), + ); + const componentBytes = await Deno.readFile( + new URL("dtor_guest_context.wasm", import.meta.url), + ); + const { plan, adapters } = translator.translate(componentBytes); + const component = await instantiateComponent({ + plan, + adapters, + componentBytes, + jspi, + }); + const exports = component.exports as Record unknown>; + assertEq(await exports.probe(), 42); + assertEq(await exports.seen(), 0); + await exports["async-probe"](); + assertEq(await exports.seen(), 0); + }); +} + +Deno.test("async caller cannot give guest dtor async completion or a host-wide drive", () => { + const store = new Store(); + const caller = new ComponentInstanceState(0, store); + const impl = new ComponentInstanceState(1, store); + const opts = options(caller); + opts.async = true; + opts.callback = () => () => { + throw new Error("unexpected callback"); + }; + opts.coreType.results = ["i32"]; + let ranSibling = false; + // Ready store work is the host driver's responsibility, not resource.drop's. + const sibling = { + ready: () => true, + resume: () => { + ranSibling = true; + }, + }; + const call = createLiftedFunction({ + name: "async dropper", + ft: { params: [], results: [], async: true }, + opts, + stats: newStats(), + core: () => { + const outer = currentThread(); + store.waiting.push(sibling as never); + const quick = new ResourceTypeInfo(impl, () => { + assertEq(currentTask().ft.async, false); + assertEq(currentTask().opts.async_, false); + }); + canonResourceDrop(caller, quick, canonResourceNew(caller, quick, 0)); + assertEq(ranSibling, false); + store.waiting.pop(); + const slow = new ResourceTypeInfo(impl, () => Promise.resolve()); + let caught: unknown; + try { + canonResourceDrop(caller, slow, canonResourceNew(caller, slow, 0)); + } catch (e) { + caught = e; + } + assertEq(caught instanceof Trap, true); + assertEq( + (caught as Error).message.includes("did not complete synchronously"), + true, + ); + assertEq(currentThread() === outer, true); + assertEq(currentTask().state, "started"); + currentTask().return_([]); + return 0; + }, + }); + call(); + assertEq(isInstancePoisoned(impl), true); + assertEq(isInstancePoisoned(caller), false); +}); diff --git a/runtime/tests/embedder/cross_copy_test.ts b/runtime/tests/embedder/cross_copy_test.ts index 0b8c000..0dd51e7 100644 --- a/runtime/tests/embedder/cross_copy_test.ts +++ b/runtime/tests/embedder/cross_copy_test.ts @@ -147,7 +147,7 @@ Deno.test("module identity: a foreign resource wrapper is named cross-copy, not owns: true, }; assertEq(wrapperState(w), undefined, "a foreign wrapper has no state HERE"); - const e = caught(() => takeRep(w, false, "export 'f'")); + const e = caught(() => takeRep(w, {} as never, false, "export 'f'")); const m = String((e as Error).message); assertEq((e as Error).name, "InvalidHandleError"); assertTrue(m.includes("DIFFERENT polyengine runtime copy"), m); @@ -168,11 +168,11 @@ Deno.test("module identity: this copy's own wrappers are unaffected", () => { }); assertEq(wrapperState(w)?.rep, 3); assertEq(wrapperState(w)?.copyUrl, COPY_URL); - assertEq(takeRep(w, false, "export 'f'"), 3); + assertEq(takeRep(w, wrapperState(w)!.rt, false, "export 'f'"), 3); }); Deno.test("module identity: a non-handle object still gets the plain diagnosis", () => { - const e = caught(() => takeRep({}, false, "export 'f'")); + const e = caught(() => takeRep({}, {} as never, false, "export 'f'")); assertTrue(String((e as Error).message).includes("not a resource handle")); }); diff --git a/runtime/tests/embedder/future_result_test.ts b/runtime/tests/embedder/future_result_test.ts index 989b5a5..3d4fa83 100644 --- a/runtime/tests/embedder/future_result_test.ts +++ b/runtime/tests/embedder/future_result_test.ts @@ -13,11 +13,79 @@ import { assertEq } from "../support/asserts.ts"; import { guest, haveFixture, instantiateFixture } from "./support.ts"; -import { Stream } from "../../src/embedder/streams.ts"; +import { Future, Stream } from "../../src/embedder/streams.ts"; +import { hostFuture } from "../../src/exec/host_streams.ts"; +import { HostResourceRegistry } from "../../src/embedder/resources.ts"; +import { INTERNAL_HOST_REGISTRIES } from "../../src/embedder/instantiate.ts"; +import { sync } from "../../src/embedder/sync.ts"; +import { Trap } from "@polyengine/protocol"; const FIXTURE = guest("future-import"); const have = await haveFixture(FIXTURE); +const cleanupFixture = "runtime/tests/embedder/resource-overlap-future.wasm"; +const cleanupReady = await haveFixture(cleanupFixture); +for (const synchronous of [false, true]) { + for (const callFails of [false, true]) { + Deno.test({ + name: + `future result cleanup: sync=${synchronous}, callFails=${callFails}`, + ignore: !cleanupReady, + async fn() { + const cleanupError = new Error("borrow disposal failed"); + const primary = new Trap("primary future call failure"); + let disposals = 0; + class R { + [Symbol.dispose]() { + disposals++; + throw cleanupError; + } + } + const source = hostFuture({ kind: "u32" }); + let resultDrops = 0; + const drop = source.drop.bind(source); + source.drop = () => { + resultDrops++; + drop(); + }; + const future = Future.fromHostFuture(source, { + element: { kind: "u32" }, + where: "cleanup test", + toHost: (v) => v as number, + fromHost: (v) => v, + }); + let registry: HostResourceRegistry; + let rep: number; + const c = await instantiateFixture(cleanupFixture, { + r: R, + next: () => future, + finish: () => { + registry.dtor(rep); + if (callFails) throw primary; + }, + }); + registry = + (c as unknown as Record>)[ + INTERNAL_HOST_REGISTRIES + ].get(0)!; + const cell = new R(); + rep = registry.repFor(cell); + let observed: unknown; + try { + await (synchronous ? sync(c.exports.run)(cell) : c.exports.run(cell)); + } catch (e) { + observed = e; + } + assertEq(observed, callFails ? primary : cleanupError); + assertEq(disposals, 1); + assertEq(registry.liveCount, 0); + assertEq(resultDrops, callFails ? 0 : 1); + if (callFails) source.drop(); + }, + }); + } +} + Deno.test({ name: "futures: a sync import returning future accepts a plain Promise", ignore: !have, diff --git a/runtime/tests/embedder/host_imports_test.ts b/runtime/tests/embedder/host_imports_test.ts index 2225b05..b482ae6 100644 --- a/runtime/tests/embedder/host_imports_test.ts +++ b/runtime/tests/embedder/host_imports_test.ts @@ -14,8 +14,10 @@ import { instantiateFixture, testdata, } from "./support.ts"; -import { ComponentException, Trap } from "@polyengine/protocol"; +import { ComponentException, suspending, Trap } from "@polyengine/protocol"; import { INTERNAL_HOST_REGISTRIES } from "../../src/embedder/instantiate.ts"; +import { HostResourceRegistry } from "../../src/embedder/resources.ts"; +import { sync } from "../../src/embedder/sync.ts"; const ready = await haveFixture(testdata("imports")); @@ -339,6 +341,288 @@ const borrowReady = await haveFixture( "runtime/tests/embedder/host-borrow.wasm", ); +const overlapFixture = "runtime/tests/embedder/resource-overlap.wasm"; +const overlapReady = await haveFixture(overlapFixture); + +for (const mode of ["constructor", "promise", "sync"] as const) { + Deno.test({ + name: + `host resources: ${mode} successful own result is dropped when cleanup throws`, + ignore: !overlapReady, + fn: async () => { + const boom = new Error("borrow cleanup failed"); + let drops = 0; + class R { + [Symbol.dispose]() { + drops++; + throw boom; + } + } + let registry: HostResourceRegistry; + let rep: number; + const c = await instantiateFixture(overlapFixture, { + "host:api/res": { + R, + value: () => { + registry.dtor(rep); + return 7; + }, + }, + }); + registry = + (c as unknown as Record>)[ + INTERNAL_HOST_REGISTRIES + ].get(0)!; + const cell = new R(); + rep = registry.repFor(cell); + const e = await caught(() => + mode === "constructor" + ? new c.exports.Ticket(cell) + : mode === "sync" + ? sync(c.exports.makeTicket)(cell) + : c.exports.makeTicket(cell) + ); + assertEq(e, boom); + assertEq(drops, 1); + assertEq(registry.liveCount, 0); + assertEq( + await c.exports.ticketDrops(), + 1, + "undeliverable result dropped once", + ); + }, + }); +} + +for (const mode of ["constructor", "promise", "sync"] as const) { + Deno.test({ + name: `host resources: ${mode} original failure survives throwing cleanup`, + ignore: !overlapReady, + fn: async () => { + const primary = new Trap("guest call failed"); + class R { + [Symbol.dispose]() { + throw new Error("secondary disposal"); + } + } + let registry: HostResourceRegistry; + let rep: number; + const c = await instantiateFixture(overlapFixture, { + "host:api/res": { + R, + value: () => { + registry.dtor(rep); + throw primary; + }, + }, + }); + registry = + (c as unknown as Record>)[ + INTERNAL_HOST_REGISTRIES + ].get(0)!; + const cell = new R(); + rep = registry.repFor(cell); + const e = await caught(() => + mode === "constructor" + ? new c.exports.Ticket(cell) + : mode === "sync" + ? sync(c.exports.makeTicket)(cell) + : c.exports.makeTicket(cell) + ); + assertEq(e, primary); + assertEq(registry.liveCount, 0); + }, + }); +} + +Deno.test("host resources: borrow releases retain overlapping and owned mappings", () => { + for (const ownAt of ["never", "before", "during"] as const) { + const registry = new HostResourceRegistry("Cell"); + const cell = new Cell(7); + if (ownAt === "before") registry.repFor(cell); + const first = registry.borrowFor(cell); + const second = registry.borrowFor(cell); + assertEq(first.rep, second.rep); + if (ownAt === "during") assertEq(registry.repFor(cell), first.rep); + first.release(); + first.release(); + assertEq(registry.lookup(second.rep) === cell, true); + second.release(); + assertEq(registry.liveCount, ownAt === "never" ? 0 : 1); + if (ownAt !== "never") { + assertEq(registry.release(first.rep) === cell, true); + assertEq(registry.liveCount, 0); + } + } +}); + +Deno.test("host resources: returning an own does not remove an overlapping borrow", () => { + const registry = new HostResourceRegistry("Cell"); + const cell = new Cell(7); + const rep = registry.repFor(cell); + const borrow = registry.borrowFor(cell); + assertEq(registry.release(rep) === cell, true); + assertEq(registry.lookup(rep) === cell, true); + borrow.release(); + assertEq(registry.liveCount, 0); +}); + +for (const reverse of [false, true]) { + for (const throwing of [false, true]) { + Deno.test({ + name: + `host resources: deferred drop, reverse=${reverse}, throwing=${throwing}`, + ignore: !overlapReady, + fn: async () => { + const boom = new Error("host destructor failed"); + class Droppable { + drops = 0; + [Symbol.dispose]() { + this.drops++; + if (throwing) throw boom; + } + } + const resolvers: (() => void)[] = []; + const bothEntered = Promise.withResolvers(); + const c = await instantiateFixture(overlapFixture, { + "host:api/res": { + R: Droppable, + value: suspending((r: Droppable) => { + assertEq(r.drops, 0, "no lookup reaches a disposed object"); + if (resolvers.length >= 2) return 7; + return new Promise((resolve) => { + resolvers.push(() => resolve(7)); + if (resolvers.length === 2) bothEntered.resolve(); + }); + }), + }, + }, { jspi: true }); + const registry = (c as unknown as Record< + symbol, + Map + >)[INTERNAL_HOST_REGISTRIES].get(0)!; + const cell = new Droppable(); + const other = new Droppable(); + const persistent = new Droppable(); + const held = await c.exports.hold(cell); + const persistentHeld = await c.exports.hold(persistent); + // The second argument needs cleanup even if the first's final release + // runs a throwing destructor. It is temporary in both overlapping calls. + const calls = [ + c.exports.peekTwo(cell, other), + c.exports.peekTwo(cell, other), + ]; + const outcomes = calls.map((p) => caught(() => p)); + await bothEntered.promise; + await c.exports.dropHeld(held); + assertEq(cell.drops, 0); + const reacquire = await caught(() => c.exports.hold(cell)); + assertEq(String(reacquire).includes("pending drop"), true); + const first = reverse ? 1 : 0; + resolvers[first](); + assertEq(await outcomes[first], undefined); + assertEq(cell.drops, 0, "one remaining borrow still protects disposal"); + assertEq(registry.hasInstance(cell), true); + resolvers[1 - first](); + assertEq(await outcomes[1 - first], throwing ? boom : undefined); + assertEq(cell.drops, 1); + assertEq(registry.hasInstance(cell), false); + assertEq( + registry.hasInstance(other), + false, + "later releases still run", + ); + assertEq(registry.hasInstance(persistent), true); + assertEq(registry.liveCount, 1); + // With no borrow, disposal still happens in the guest call itself. + const drop = c.exports.dropHeld(persistentHeld); + assertEq(persistent.drops, 1); + const dropError = await caught(() => drop); + assertEq(dropError === undefined, !throwing); + assertEq(registry.liveCount, 0); + }, + }); + } +} + +for (const ownAt of ["never", "before", "during"] as const) { + Deno.test({ + name: + `host resources: overlapping JSPI borrows preserve ${ownAt}-owned mapping`, + ignore: !overlapReady, + fn: async () => { + const resolvers: (() => void)[] = []; + const bothEntered = Promise.withResolvers(); + const c = await instantiateFixture(overlapFixture, { + "host:api/res": { + R: Cell, + value: suspending((r: Cell) => { + if (resolvers.length >= 2) return r.v; + return new Promise((resolve) => { + resolvers.push(() => resolve(r.v)); + if (resolvers.length === 2) bothEntered.resolve(); + }); + }), + }, + }, { jspi: true }); + const registry = (c as unknown as Record< + symbol, + Map + >)[INTERNAL_HOST_REGISTRIES].get(0)!; + Cell.disposed = []; + const cell = new Cell(7); + let held: number | undefined; + if (ownAt === "before") held = await c.exports.hold(cell); + const first = c.exports.peek(cell); + const second = c.exports.peek(cell); + const secondOutcome = caught(() => second); + await bothEntered.promise; + if (ownAt === "during") held = await c.exports.hold(cell); + resolvers[0](); + assertEq(await first, 7); + const liveDuringSecond = registry.hasInstance(cell); + resolvers[1](); + const error = await secondOutcome; + assertEq(liveDuringSecond, true, "the second borrow keeps its mapping"); + assertEq(error, undefined); + assertEq(await second, 7, "the second guest lookup succeeds"); + assertEq(registry.liveCount, ownAt === "never" ? 0 : 1); + assertEq(Cell.disposed, [], "ending a borrow never disposes"); + if (held !== undefined) { + await c.exports.dropHeld(held); + assertEq(registry.liveCount, 0); + assertEq(Cell.disposed, [7]); + } + }, + }); +} + +Deno.test({ + name: + "host resources: constructor borrow mappings last through the call only", + ignore: !overlapReady, + fn: async () => { + const cell = new Cell(7); + let seen: Cell | undefined; + const c = await instantiateFixture(overlapFixture, { + "host:api/res": { + R: Cell, + value: (r: Cell) => { + seen = r; + return r.v; + }, + }, + }); + const registry = (c as unknown as Record< + symbol, + Map + >)[INTERNAL_HOST_REGISTRIES].get(0)!; + using ticket = new c.exports.Ticket(cell); + assertEq(seen === cell, true); + assertEq(registry.liveCount, 0); + }, +}); + Deno.test({ name: "host resources: a borrow-allocated rep is released when the call returns", diff --git a/runtime/tests/embedder/passthrough_test.ts b/runtime/tests/embedder/passthrough_test.ts index 9e5287d..513a177 100644 --- a/runtime/tests/embedder/passthrough_test.ts +++ b/runtime/tests/embedder/passthrough_test.ts @@ -12,6 +12,7 @@ import { assertEq } from "../support/asserts.ts"; import { artifactsOf, + caught, guest, haveFixture, instantiateFixture, @@ -353,8 +354,7 @@ Deno.test({ // The read genuinely happened while the host owned the end; only reads // STARTED after the transfer are refused. Modelled on a LIFTED future // (the host holds the readable end) with a guest-shaped write completing - // the rendezvous — a host-created future cannot read and write through - // one wrapper (one in-flight operation per wrapper). + // the rendezvous. const codec = { element: { kind: "u32" } as ValType, toHost: (v: ComponentValue) => v as number, @@ -390,3 +390,71 @@ Deno.test({ assertEq(await pending, 7, "the pre-transfer read resolves"); }, }); + +const streamHostFixture = "runtime/tests/embedder/stream-host.wasm"; +const streamHostReady = await haveFixture(streamHostFixture); + +for (const delayed of [false, true]) { + Deno.test({ + name: `future pass-through: ${ + delayed ? "reader" : "producer" + } arrives first`, + ignore: !streamHostReady, + async fn() { + const c = await instantiateFixture(streamHostFixture, { + "host:streams/api": { ticket: class {} }, + }); + let resolve!: (n: number) => void; + const source = delayed + ? new Promise((r) => resolve = r) + : Promise.resolve(7); + const f = c.exports.passFuture(source) as Future; + const result = Promise.resolve(f); + if (delayed) { + await new Promise((r) => setTimeout(r, 0)); + resolve(7); + } + assertEq(await result, 7); + f.drop(); + }, + }); +} + +Deno.test("future ends: same-direction exclusion, cancellation and retry", async () => { + const h = hostFuture({ kind: "u32" }); + const write = h.write(1); + assertEq(await caught(() => h.write(2)) instanceof TypeError, true); + h.cancel(); + await write; + const read = h.readResult(); + assertEq(await caught(() => h.readResult()) instanceof TypeError, true); + h.cancel(); + assertEq((await read).value, undefined); + const retried = h.read(); + await h.write(3); + assertEq(await retried, 3); + h.drop(); +}); + +for (const side of ["read", "write"] as const) { + Deno.test(`future ${side}: a throwing shared operation clears only its end`, async () => { + const h = hostFuture({ kind: "u32" }); + const shared = h.value as SharedFutureImpl; + const original = shared[side]; + const error = new Error("shared operation failed"); + shared[side] = (...args: Parameters) => { + original.apply(shared, args); + throw error; + }; + assertEq( + await caught(() => side === "read" ? h.read() : h.write(1)), + error, + ); + assertEq(shared.pendingBuffer, null, "the failed operation was withdrawn"); + shared[side] = original as typeof shared[typeof side]; + const read = h.read(); + await h.write(2); + assertEq(await read, 2); + h.drop(); + }); +} diff --git a/runtime/tests/embedder/resource-overlap-future.wasm b/runtime/tests/embedder/resource-overlap-future.wasm new file mode 100644 index 0000000..4006a1d Binary files /dev/null and b/runtime/tests/embedder/resource-overlap-future.wasm differ diff --git a/runtime/tests/embedder/resource-overlap-future.wat b/runtime/tests/embedder/resource-overlap-future.wat new file mode 100644 index 0000000..cac321b --- /dev/null +++ b/runtime/tests/embedder/resource-overlap-future.wat @@ -0,0 +1,23 @@ +(component + (import "r" (type $R (sub resource))) + (import "next" (func $next (result (future u32)))) + (import "finish" (func $finish (param "r" (borrow $R)))) + (canon lower (func $next) (core func $next')) + (canon lower (func $finish) (core func $finish')) + (canon resource.drop $R (core func $drop)) + (core module $M + (import "" "next" (func $next (result i32))) + (import "" "finish" (func $finish (param i32))) + (import "" "drop" (func $drop (param i32))) + (func (export "run") (param $r i32) (result i32) + (local $f i32) + (call $finish (local.get $r)) + (local.set $f (call $next)) + (call $drop (local.get $r)) + (local.get $f))) + (core instance $m (instantiate $M (with "" (instance + (export "next" (func $next')) + (export "finish" (func $finish')) + (export "drop" (func $drop)))))) + (func (export "run") (param "r" (borrow $R)) (result (future u32)) + (canon lift (core func $m "run")))) diff --git a/runtime/tests/embedder/resource-overlap.wasm b/runtime/tests/embedder/resource-overlap.wasm new file mode 100644 index 0000000..c972c44 Binary files /dev/null and b/runtime/tests/embedder/resource-overlap.wasm differ diff --git a/runtime/tests/embedder/resource-overlap.wat b/runtime/tests/embedder/resource-overlap.wat new file mode 100644 index 0000000..1b0bfc0 --- /dev/null +++ b/runtime/tests/embedder/resource-overlap.wat @@ -0,0 +1,55 @@ +;; wasm-tools parse resource-overlap.wat -o resource-overlap.wasm +(component + (import "host:api/res" (instance $api + (export "R" (type $R (sub resource))) + (export "value" (func (param "r" (borrow $R)) (result u32))))) + (alias export $api "R" (type $R)) + (alias export $api "value" (func $value)) + (canon lower (func $value) (core func $value')) + (canon resource.drop $R (core func $drop)) + (core module $D + (global $drops (mut i32) (i32.const 0)) + (func (export "drop") (param i32) + (global.set $drops (i32.add (global.get $drops) (i32.const 1)))) + (func (export "drops") (result i32) (global.get $drops))) + (core instance $d (instantiate $D)) + (type $ticket (resource (rep i32) (dtor (func $d "drop")))) + (canon resource.new $ticket (core func $new)) + (core module $M + (import "" "value" (func $value (param i32) (result i32))) + (import "" "drop" (func $drop (param i32))) + (import "" "new" (func $new (param i32) (result i32))) + (func $peek (export "peek") (param $h i32) (result i32) + (local $out i32) + (drop (call $value (local.get $h))) + (local.set $out (call $value (local.get $h))) + (call $drop (local.get $h)) + (local.get $out)) + (func (export "peek-two") (param $a i32) (param $b i32) (result i32) + (call $peek (local.get $a)) + (call $drop (local.get $b))) + (func (export "hold") (param i32) (result i32) (local.get 0)) + (export "drop-held" (func $drop)) + (func (export "ticket") (param $h i32) (result i32) + (local $out i32) + (local.set $out (call $value (local.get $h))) + (call $drop (local.get $h)) + (call $new (local.get $out)))) + (core instance $i (instantiate $M (with "" (instance + (export "value" (func $value')) (export "drop" (func $drop)) + (export "new" (func $new)))))) + (func (export "peek") (param "r" (borrow $R)) (result u32) + (canon lift (core func $i "peek"))) + (func (export "hold") (param "r" (own $R)) (result u32) + (canon lift (core func $i "hold"))) + (func (export "peek-two") (param "a" (borrow $R)) (param "b" (borrow $R)) (result u32) + (canon lift (core func $i "peek-two"))) + (func (export "drop-held") (param "h" u32) + (canon lift (core func $i "drop-held"))) + (export $Ticket "ticket" (type $ticket)) + (func (export "[constructor]ticket") (param "r" (borrow $R)) (result (own $Ticket)) + (canon lift (core func $i "ticket"))) + (func (export "make-ticket") (param "r" (borrow $R)) (result (own $Ticket)) + (canon lift (core func $i "ticket"))) + (func (export "ticket-drops") (result u32) + (canon lift (core func $d "drops")))) diff --git a/runtime/tests/embedder/resource_stream_test.ts b/runtime/tests/embedder/resource_stream_test.ts index 618c082..73a0598 100644 --- a/runtime/tests/embedder/resource_stream_test.ts +++ b/runtime/tests/embedder/resource_stream_test.ts @@ -14,10 +14,15 @@ // element is a live accepted connection that must be closed. import { assertEq } from "../support/asserts.ts"; -import { guest, haveFixture, instantiateFixture } from "./support.ts"; +import { caught, guest, haveFixture, instantiateFixture } from "./support.ts"; +import { createStream } from "../../src/embedder/mod.ts"; +import type { Stream } from "@polyengine/protocol"; +import { PeerTrappedError, StreamProducerError } from "@polyengine/protocol"; const FIXTURE = guest("resource-stream"); const have = await haveFixture(FIXTURE); +const hostFixture = "runtime/tests/embedder/stream-host.wasm"; +const haveHost = await haveFixture(hostFixture); class Ticket { static disposed: number[] = []; @@ -40,6 +45,266 @@ function reset(): void { Ticket.created = 0; } +for (const mode of ["pump", "write", "writeAll"] as const) { + Deno.test({ + name: + `resource streams: ${mode} packing failure releases lowered prefix exactly once`, + ignore: !haveHost, + async fn() { + reset(); + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: Ticket }, + }); + const first = new Ticket(1); + if (mode === "pump") { + let out: Stream | undefined; + const error = await caught(async () => { + out = await c.exports.passTickets([[first, null]]); + await out!.read(1); + }); + assertEq(error instanceof StreamProducerError, true, String(error)); + out?.drop(); + } else { + const { stream, writer } = createStream(); + const out = await c.exports.passTickets(stream) as Stream; + assertEq( + await caught(() => writer[mode]([first, null as never])) instanceof + TypeError, + true, + ); + assertEq(Ticket.disposed, [1]); + const retry = writer[mode]([new Ticket(2)]); + const [ticket] = await out.read(1); + assertEq(ticket.v, 2); + assertEq(await retry, 1); + ticket[Symbol.dispose](); + await writer.close(); + out.drop(); + } + assertEq(Ticket.disposed, mode === "pump" ? [1] : [1, 2]); + }, + }); +} + +for (const mode of ["pump", "write", "writeAll"] as const) { + for (const trap of [false, true]) { + Deno.test({ + name: + `resource streams: ${mode} releases only the untaken tail on guest ${ + trap ? "fault" : "drop" + }`, + ignore: !haveHost, + async fn() { + reset(); + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: Ticket }, + }); + const { stream, writer } = createStream(); + const out = await c.exports.passTickets( + mode === "pump" ? [[new Ticket(1), new Ticket(2)]] : stream, + ) as Stream; + const pending = mode === "pump" + ? null + : writer[mode]([new Ticket(1), new Ticket(2)]); + // Writer methods bind asynchronously; the fixture takes one element + // synchronously, so let this offer park before entering the guest. + await Promise.resolve(); + const error = await caught(() => c.exports.takeTicket(out, trap)); + assertEq(error instanceof Error, trap, String(error)); + if (pending !== null) { + if (trap) { + const fault = await caught(() => pending); + assertEq(fault instanceof PeerTrappedError, true, String(fault)); + assertEq((fault as PeerTrappedError).progress, 1); + } else { + assertEq(await pending, 1); + } + } + // Pump cleanup has a generator-finally continuation after the write. + await new Promise((r) => setTimeout(r, 0)); + assertEq( + Ticket.disposed.sort(), + [1, 2], + "delivered guest drop and untaken cleanup each run once", + ); + }, + }); + } +} + +for (const mode of ["write", "writeAll"] as const) { + Deno.test({ + name: + `resource streams: ${mode} cancellation releases only untaken owns and permits retry`, + ignore: !haveHost, + async fn() { + reset(); + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: Ticket }, + }); + const { stream, writer } = createStream(); + const out = await c.exports.passTickets(stream) as Stream; + const pending = writer[mode]([new Ticket(1), new Ticket(2)]); + await Promise.resolve(); + const [delivered] = await out.read(1); + assertEq(Ticket.disposed, []); + writer.cancelWrite(); + assertEq(await pending, 1); + assertEq(Ticket.disposed, [2]); + const retry = writer[mode]([new Ticket(3)]); + const [next] = await out.read(1); + assertEq(next.v, 3); + assertEq(await retry, 1); + delivered[Symbol.dispose](); + next[Symbol.dispose](); + await writer.close(); + out.drop(); + assertEq(Ticket.disposed.sort(), [1, 2, 3]); + }, + }); +} + +Deno.test({ + name: + "resource streams: packing rollback continues after a destructor throws", + ignore: !haveHost, + async fn() { + reset(); + class ThrowingTicket extends Ticket { + override [Symbol.dispose](): void { + super[Symbol.dispose](); + if (this.v === 1) throw new Error("ticket disposal failed"); + } + } + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: ThrowingTicket }, + }); + const { stream, writer } = createStream(); + const out = await c.exports.passTickets(stream) as Stream; + const error = await caught(() => + writer.write([ + new ThrowingTicket(1), + new ThrowingTicket(2), + null as never, + ]) + ); + assertEq( + error instanceof TypeError, + true, + "packing error keeps attribution", + ); + assertEq(Ticket.disposed, [1, 2]); + await writer.close(); + out.drop(); + }, +}); + +Deno.test({ + name: "resource streams: pump reports tail cleanup throwing undefined", + ignore: !haveHost, + async fn() { + reset(); + class ThrowingTicket extends Ticket { + override [Symbol.dispose](): void { + super[Symbol.dispose](); + if (this.v === 2) throw undefined; + } + } + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: ThrowingTicket }, + }); + const out = await c.exports.passTickets([[ + new ThrowingTicket(1), + new ThrowingTicket(2), + new ThrowingTicket(3), + ]]) as Stream; + const [delivered] = await out.read(1); + out.drop(); + await new Promise((r) => setTimeout(r, 0)); + const error = await caught(() => out.read(1)); + assertEq( + error instanceof StreamProducerError, + true, + "cleanup fault is not clean EOS", + ); + assertEq((error as StreamProducerError).cause, undefined); + assertEq( + Ticket.disposed, + [2, 3], + "cleanup continues past the throwing tail", + ); + delivered[Symbol.dispose](); + assertEq( + Ticket.disposed, + [2, 3, 1], + "delivered own belongs only to its receiver", + ); + }, +}); + +for (const trap of [false, true]) { + Deno.test({ + name: `resource streams: throwing tail cleanup ${ + trap ? "preserves peer fault" : "surfaces its own failure" + }`, + ignore: !haveHost, + async fn() { + reset(); + class ThrowingTicket extends Ticket { + override [Symbol.dispose](): void { + super[Symbol.dispose](); + if (this.v === 2) throw new Error("ticket disposal failed"); + } + } + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: ThrowingTicket }, + }); + const { stream, writer } = createStream(); + const out = await c.exports.passTickets(stream) as Stream; + const pending = writer.writeAll([ + new ThrowingTicket(1), + new ThrowingTicket(2), + new ThrowingTicket(3), + ]); + await Promise.resolve(); + await caught(() => c.exports.takeTicket(out, trap)); + const error = await caught(() => pending); + if (trap) { + assertEq(error instanceof PeerTrappedError, true, String(error)); + assertEq((error as PeerTrappedError).progress, 1); + } else { + assertEq(String(error).includes("ticket disposal failed"), true); + } + assertEq( + Ticket.disposed, + [1, 2, 3], + "all tails released, delivered own dropped only by guest", + ); + }, + }); +} + +Deno.test({ + name: + "resource streams: writer arriving second releases its short-write tail", + ignore: !haveHost, + async fn() { + reset(); + const c = await instantiateFixture(hostFixture, { + "host:streams/api": { ticket: Ticket }, + }); + const { stream, writer } = createStream(); + const out = await c.exports.passTickets(stream) as Stream; + const read = out.read(1); + assertEq(await writer.write([new Ticket(1), new Ticket(2)]), 1); + assertEq(Ticket.disposed, [2]); + (await read)[0][Symbol.dispose](); + await writer.close(); + out.drop(); + assertEq(Ticket.disposed.sort(), [1, 2]); + }, +}); + /** One ticket per chunk: the pump lowers (and parks on) one element at a time. */ function ticketSource(count: number): AsyncIterable { return (async function* () { diff --git a/runtime/tests/embedder/resources_test.ts b/runtime/tests/embedder/resources_test.ts index e4c0a36..0292b46 100644 --- a/runtime/tests/embedder/resources_test.ts +++ b/runtime/tests/embedder/resources_test.ts @@ -9,10 +9,81 @@ import { assertEq } from "../support/asserts.ts"; import { caught, guest, haveFixture, instantiateFixture } from "./support.ts"; import { InvalidHandleError } from "@polyengine/protocol"; +import { ResourceTypeInfo } from "../../src/cabi/types.ts"; +import { + GuestResource, + makeWrapper, + takeRep, + wrapperState, +} from "../../src/embedder/resources.ts"; const ready = await haveFixture(guest("resources")); const IFACE = "polyengine:resources/counters"; +Deno.test("resources: identity and ownership rejection leave a wrapper intact", async () => { + const rt = new ResourceTypeInfo(null, null); + const other = new ResourceTypeInfo(null, null); + const owned = makeWrapper(GuestResource, 7, rt, true); + for (const own of [false, true]) { + const e = await caught(() => takeRep(owned, other, own, "resource")); + assertEq(e instanceof InvalidHandleError, true); + assertEq(String(e).includes("resource type mismatch"), true); + assertEq(wrapperState(owned)?.valid, true); + } + assertEq(takeRep(owned, rt, true, "own"), 7); + assertEq(wrapperState(owned)?.valid, false); + + const borrowed = makeWrapper(GuestResource, 7, rt, false); + const e = await caught(() => takeRep(borrowed, rt, true, "own")); + assertEq(e instanceof InvalidHandleError, true); + assertEq(String(e).includes("borrowed"), true); + assertEq(wrapperState(borrowed)?.valid, true); + assertEq(takeRep(borrowed, rt, false, "borrow"), 7); + borrowed.drop(); +}); + +Deno.test({ + name: + "resources: foreign instantiation owns and borrows are rejected before entry", + ignore: !ready, + fn: async () => { + const a = await counters(); + const b = await counters(); + using x = new a.Counter(11n); + using y = new b.Counter(22n); + for (const call of [() => b.bump(x, 1n), () => b.consume(x)]) { + const e = await caught(call); + assertEq(e instanceof InvalidHandleError, true); + assertEq(String(e).includes("resource type mismatch"), true); + assertEq(await x.get(), 11n, "source wrapper remains live"); + assertEq(await y.get(), 22n, "target guest was not entered"); + assertEq(await a.liveCounters(), 1); + assertEq(await b.liveCounters(), 1); + } + }, +}); + +Deno.test({ + name: + "resources: own-lowering refuses a live borrowed wrapper before consumption", + ignore: !ready, + fn: async () => { + const c = await counters(); + using owned = new c.Counter(11n); + const state = wrapperState(owned)!; + // Same wrapper producer as the incoming-borrow bridge, followed by the + // production export own-lowering path. Not a guest-to-host borrow fixture. + using borrowed = makeWrapper(c.Counter, state.rep, state.rt, false); + const e = await caught(() => c.consume(borrowed)); + assertEq(e instanceof InvalidHandleError, true); + assertEq(String(e).includes("borrowed"), true); + assertEq(wrapperState(borrowed)?.valid, true); + assertEq(await c.bump(borrowed, 1n), 12n, "reborrowing remains legal"); + assertEq(await owned.get(), 12n, "the actual owner was not consumed"); + assertEq(await c.liveCounters(), 1); + }, +}); + // deno-lint-ignore no-explicit-any async function counters(): Promise { const inst = await instantiateFixture(guest("resources")); diff --git a/runtime/tests/embedder/stream-host.wasm b/runtime/tests/embedder/stream-host.wasm new file mode 100644 index 0000000..12b4639 Binary files /dev/null and b/runtime/tests/embedder/stream-host.wasm differ diff --git a/runtime/tests/embedder/stream-host.wat b/runtime/tests/embedder/stream-host.wat new file mode 100644 index 0000000..31adc89 --- /dev/null +++ b/runtime/tests/embedder/stream-host.wat @@ -0,0 +1,37 @@ +;; Host stream ownership and future round trips through real canonical edges. +;; Regenerate: wasm-tools parse stream-host.wat -o stream-host.wasm +(component + (import "host:streams/api" (instance $api + (export "ticket" (type $ticket (sub resource))))) + (alias export $api "ticket" (type $ticket)) + (type $tickets (stream (own $ticket))) + (type $future (future u32)) + + (core module $memory (memory (export "memory") 1)) + (core instance $mem (instantiate $memory)) + (canon stream.read $tickets (memory $mem "memory") async (core func $read)) + (canon stream.drop-readable $tickets (core func $drop-stream)) + (canon resource.drop $ticket (core func $drop-ticket)) + (core module $M + (import "" "memory" (memory 1)) + (import "" "read" (func $read (param i32 i32 i32) (result i32))) + (import "" "drop-stream" (func $drop-stream (param i32))) + (import "" "drop-ticket" (func $drop-ticket (param i32))) + (func (export "pass") (param i32) (result i32) local.get 0) + (func (export "take") (param $s i32) (param $trap i32) + (if (i32.ne (call $read (local.get $s) (i32.const 0) (i32.const 1)) (i32.const 16)) + (then unreachable)) + (call $drop-ticket (i32.load (i32.const 0))) + (if (local.get $trap) (then unreachable)) + (call $drop-stream (local.get $s)))) + (core instance $i (instantiate $M (with "" (instance + (export "memory" (memory $mem "memory")) + (export "read" (func $read)) + (export "drop-stream" (func $drop-stream)) + (export "drop-ticket" (func $drop-ticket)))))) + (func (export "pass-future") (param "f" $future) (result $future) + (canon lift (core func $i "pass"))) + (func (export "pass-tickets") (param "s" $tickets) (result $tickets) + (canon lift (core func $i "pass"))) + (func (export "take-ticket") (param "s" $tickets) (param "trap" bool) + (canon lift (core func $i "take")))) diff --git a/runtime/tests/embedder/streams_test.ts b/runtime/tests/embedder/streams_test.ts index 171f1a1..93b5cd2 100644 --- a/runtime/tests/embedder/streams_test.ts +++ b/runtime/tests/embedder/streams_test.ts @@ -18,6 +18,54 @@ const ready = await haveFixture(guest("async-probe")) && await haveFixture(guest("stream-echo")) && await haveFixture(guest("future-user")); +for (const readerFirst of [false, true]) { + Deno.test({ + name: `writeAll cancellation: ${ + readerFirst ? "arriving" : "parked" + } writer retracts the whole tail`, + ignore: !(await haveFixture(guest("stream-pass"))), + async fn() { + const c = await instantiateFixture(guest("stream-pass"), { + sink: () => 0n, + }); + const { stream, writer } = Stream.create(); + const out = await c.exports.passThrough(stream) as Stream; + const read = readerFirst ? out.read(2) : null; + const pending = writer.writeAll(new Uint8Array([1, 2, 3, 4])); + assertEq([...(await (read ?? out.read(2)))], [1, 2]); + writer.cancelWrite(); + assertEq(await pending, 2); + const retry = writer.writeAll(new Uint8Array([5, 6])); + assertEq( + [...(await out.read(4))], + [5, 6], + "cancelled tail is never reoffered", + ); + assertEq(await retry, 2); + await writer.close(); + out.drop(); + }, + }); +} + +Deno.test("writeAll: cancellation in a reoffer gap preserves per-end exclusion", async () => { + const h = hostStream({ kind: "u8" }); + const read = h.readable.read(1); + const pending = h.writable.writeAll([1, 2]); + // The first copy completed synchronously; the helper has not resumed yet. + let busy: unknown; + try { + h.writable.write([3]); + } catch (e) { + busy = e; + } + assertEq(busy instanceof TypeError, true); + h.writable.cancelWrite(); + assertEq(await pending, 1); + assertEq([...(await read)], [1]); + h.writable.drop(); +}); + Deno.test({ name: "async: an async export is Promise-shaped and suspends transparently", ignore: !ready, diff --git a/runtime/tests/fact_string_source_limits_test.ts b/runtime/tests/fact_string_source_limits_test.ts new file mode 100644 index 0000000..4929129 --- /dev/null +++ b/runtime/tests/fact_string_source_limits_test.ts @@ -0,0 +1,101 @@ +import { Translator } from "../src/shim/translator.ts"; +import { instantiateComponent } from "../src/exec/mod.ts"; +import { Trap } from "../src/cabi/mod.ts"; +import { assertEq } from "./support/asserts.ts"; + +const translator = await Translator.create( + await Deno.readFile( + new URL("../../translator/translator_shim.wasm", import.meta.url), + ), +); +const componentBytes = await Deno.readFile( + new URL("./fixtures/fact-string-source-limits.wasm", import.meta.url), +); +const translated = translator.translate(componentBytes); +const MAX = 2 ** 28 - 1; +const TAG = 2 ** 31; + +async function probe(name: string, ptr: number, len: number, trapped: boolean) { + const instance = await instantiateComponent({ + ...translated, + componentBytes, + }); + const sink = instance.coreInstances.find((i) => + i.exports["allocation-count"] + ); + if (!sink) throw new Error("missing production realloc counter"); + let result: unknown; + let didTrap = false; + try { + result = await (instance.exports[name] as (...args: number[]) => unknown)( + ptr, + len, + ); + } catch (e) { + if (!(e instanceof Trap)) throw e; + didTrap = true; + } + assertEq(didTrap, trapped, `${name}(${ptr},${len}) trap`); + return { + result, + calls: (sink.exports["allocation-count"] as WebAssembly.Global).value, + bytes: (sink.exports["allocation-size"] as WebAssembly.Global).value, + }; +} + +// Every destination for UTF-8, UTF-16, and both compact source tag paths. +for ( + const [names, width, tag] of [ + ["abc", 1, 0], + ["def", 2, 0], + ["ghi", 1, 0], + ["ghi", 2, TAG], + ] as const +) { + for (const name of names) { + Deno.test(`FACT source limit: ${name}, width=${width}, tag=${tag}`, async () => { + const maxUnits = Math.floor(MAX / width); + assertEq( + (await probe(name, 0, tag + maxUnits + 1, true)).calls, + 0, + "oversized source rejected BEFORE guest realloc", + ); + const atLimit = await probe(name, 0, tag + maxUnits, true); + assertEq( + atLimit.calls, + 1, + "valid source reaches realloc, then destination bounds trap", + ); + if (name === "b" || name === "h") { + assertEq( + atLimit.bytes, + 2 * maxUnits, + "destination expansion is not source byte length", + ); + } + assertEq((await probe(name, 64, tag, false)).result, 0, "empty string"); + assertEq( + (await probe(name, 64, tag + 1, false)).result, + 1, + "one zero code unit", + ); + }); + } +} + +Deno.test("FACT source limit: transcode retry paths remain valid", async () => { + for ( + const [name, ptr, len, result, calls] of [ + ["b", 0, 2, 1, 2], // UTF-8 -> UTF-16, shrink pessimistic allocation + ["c", 0, 2, TAG + 1, 3], // UTF-8 -> compact, inflate then shrink + ["d", 16, 1, 2, 3], // UTF-16 -> UTF-8, grow then shrink + ["f", 16, 1, TAG + 1, 2], // UTF-16 -> compact, inflate + ["g", 32, 1, 2, 2], // Latin-1 -> UTF-8, grow + ["g", 16, TAG + 1, 2, 3], // compact UTF-16 -> UTF-8 + ["i", 16, TAG + 1, TAG + 1, 1], // compact UTF-16 stays wide + ] as const + ) { + const got = await probe(name, ptr, len, false); + assertEq([got.result, got.calls], [result, calls]); + } +}); diff --git a/runtime/tests/fixtures/fact-string-source-limits.wasm b/runtime/tests/fixtures/fact-string-source-limits.wasm new file mode 100644 index 0000000..596458f Binary files /dev/null and b/runtime/tests/fixtures/fact-string-source-limits.wasm differ diff --git a/runtime/tests/fixtures/fact-string-source-limits.wat b/runtime/tests/fixtures/fact-string-source-limits.wat new file mode 100644 index 0000000..1a946c2 --- /dev/null +++ b/runtime/tests/fixtures/fact-string-source-limits.wat @@ -0,0 +1,77 @@ +;; The source memory is reserved lazily; oversized tests never read or copy it. +;; The destination has only one page: boundary-valid large sources must call +;; realloc, then trap on destination bounds, without allocating a second 256MB. +(component + (component $sink + (core module $M + (memory (export "memory") 1) + (global $calls (export "allocation-count") (mut i32) (i32.const 0)) + (global $size (export "allocation-size") (mut i32) (i32.const 0)) + (func (export "realloc") (param i32 i32 i32 i32) (result i32) + global.get $calls i32.const 1 i32.add global.set $calls + local.get 3 global.set $size i32.const 0) + (func (export "consume") (param i32 i32) (result i32) local.get 1)) + (core instance $m (instantiate $M)) + (func (export "consume8") (param "s" string) (result u32) + (canon lift (core func $m "consume") (memory $m "memory") + (realloc (func $m "realloc")) string-encoding=utf8)) + (func (export "consume16") (param "s" string) (result u32) + (canon lift (core func $m "consume") (memory $m "memory") + (realloc (func $m "realloc")) string-encoding=utf16)) + (func (export "consumec") (param "s" string) (result u32) + (canon lift (core func $m "consume") (memory $m "memory") + (realloc (func $m "realloc")) string-encoding=latin1+utf16))) + (instance $sink-i (instantiate $sink)) + (component $source + (import "consume8" (func $consume8 (param "s" string) (result u32))) + (import "consume16" (func $consume16 (param "s" string) (result u32))) + (import "consumec" (func $consumec (param "s" string) (result u32))) + (core module $Mem + (memory (export "memory") 4096) + ;; U+0100 in UTF-8 / UTF-16, and U+00FF in Latin-1 for retry paths. + (data (i32.const 0) "\c4\80") + (data (i32.const 16) "\00\01") + (data (i32.const 32) "\ff")) + (core instance $mem (instantiate $Mem)) + (core func $a (canon lower (func $consume8) (memory $mem "memory") string-encoding=utf8)) + (core func $b (canon lower (func $consume16) (memory $mem "memory") string-encoding=utf8)) + (core func $c (canon lower (func $consumec) (memory $mem "memory") string-encoding=utf8)) + (core func $d (canon lower (func $consume8) (memory $mem "memory") string-encoding=utf16)) + (core func $e (canon lower (func $consume16) (memory $mem "memory") string-encoding=utf16)) + (core func $f (canon lower (func $consumec) (memory $mem "memory") string-encoding=utf16)) + (core func $g (canon lower (func $consume8) (memory $mem "memory") string-encoding=latin1+utf16)) + (core func $h (canon lower (func $consume16) (memory $mem "memory") string-encoding=latin1+utf16)) + (core func $i (canon lower (func $consumec) (memory $mem "memory") string-encoding=latin1+utf16)) + (core module $Run + (import "host" "a" (func $a (param i32 i32) (result i32))) + (import "host" "b" (func $b (param i32 i32) (result i32))) + (import "host" "c" (func $c (param i32 i32) (result i32))) + (import "host" "d" (func $d (param i32 i32) (result i32))) + (import "host" "e" (func $e (param i32 i32) (result i32))) + (import "host" "f" (func $f (param i32 i32) (result i32))) + (import "host" "g" (func $g (param i32 i32) (result i32))) + (import "host" "h" (func $h (param i32 i32) (result i32))) + (import "host" "i" (func $i (param i32 i32) (result i32))) + (export "a" (func $a)) (export "b" (func $b)) (export "c" (func $c)) + (export "d" (func $d)) (export "e" (func $e)) (export "f" (func $f)) + (export "g" (func $g)) (export "h" (func $h)) (export "i" (func $i))) + (core instance $run (instantiate $Run (with "host" (instance + (export "a" (func $a)) (export "b" (func $b)) (export "c" (func $c)) + (export "d" (func $d)) (export "e" (func $e)) (export "f" (func $f)) + (export "g" (func $g)) (export "h" (func $h)) (export "i" (func $i)))))) + (func (export "a") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "a"))) + (func (export "b") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "b"))) + (func (export "c") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "c"))) + (func (export "d") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "d"))) + (func (export "e") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "e"))) + (func (export "f") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "f"))) + (func (export "g") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "g"))) + (func (export "h") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "h"))) + (func (export "i") (param "ptr" u32) (param "len" u32) (result u32) (canon lift (core func $run "i")))) + (instance $source-i (instantiate $source + (with "consume8" (func $sink-i "consume8")) + (with "consume16" (func $sink-i "consume16")) + (with "consumec" (func $sink-i "consumec")))) + (export "a" (func $source-i "a")) (export "b" (func $source-i "b")) (export "c" (func $source-i "c")) + (export "d" (func $source-i "d")) (export "e" (func $source-i "e")) (export "f" (func $source-i "f")) + (export "g" (func $source-i "g")) (export "h" (func $source-i "h")) (export "i" (func $source-i "i"))) diff --git a/runtime/tests/resource_lifetime_test.ts b/runtime/tests/resource_lifetime_test.ts index e811965..cc893bc 100644 --- a/runtime/tests/resource_lifetime_test.ts +++ b/runtime/tests/resource_lifetime_test.ts @@ -330,11 +330,11 @@ Deno.test("#86: transferring a lent handle as own is refused (lift_own)", () const release = lendWrapper(w); let msg = ""; try { - takeRep(w, true, "own"); + takeRep(w, rt, true, "own"); } catch (e) { msg = (e as Error).message; } assert(msg.includes("still lent out"), `expected a lend refusal, got ${msg}`); release(); - assertEq(takeRep(w, true, "own"), 26); + assertEq(takeRep(w, rt, true, "own"), 26); }); diff --git a/runtime/tests/streams_teardown_test.ts b/runtime/tests/streams_teardown_test.ts index efde958..0cff7fb 100644 --- a/runtime/tests/streams_teardown_test.ts +++ b/runtime/tests/streams_teardown_test.ts @@ -30,11 +30,22 @@ import { assertEq } from "./support/asserts.ts"; import { AssertionError, Trap } from "../src/cabi/mod.ts"; import { + createErrorContextDrop, + createFutureDropReadable, + createFutureDropWritable, createFutureRead, + createFutureTransfer, + createFutureWrite, + createStreamDropReadable, + createStreamDropWritable, createStreamRead, + createStreamTransfer, + createStreamWrite, } from "../src/intrinsics/stream_builtins.ts"; import { BLOCKED, + createSubtaskDrop, + createWaitableSetDrop, createWaitableSetWait, } from "../src/intrinsics/async_builtins.ts"; import type { ResolvedOptions } from "../src/exec/boundary.ts"; @@ -62,6 +73,12 @@ import { WritableStreamEnd, } from "../src/task/mod.ts"; import type { FuncType } from "../src/cabi/types.ts"; +import { liftFuture, liftStream } from "../src/cabi/async_values.ts"; +import { LiftLowerContext, mkCanonicalOptions } from "../src/cabi/context.ts"; +import { createLiftedFunction, newStats } from "../src/exec/boundary.ts"; +import { poisonFailureOf } from "../src/task/streams.ts"; +import { Stream } from "../src/embedder/streams.ts"; +import { PeerTrappedError } from "@polyengine/protocol"; function assert(cond: boolean, msg: string): asserts cond { if (!cond) throw new Error(`assertion failed: ${msg}`); @@ -716,3 +733,320 @@ Deno.test("#100: a host peer parked on a poisoned guest's end is still notified" notifyInstancePoisoned(guest, new Trap("unreachable")); assertEq(result, CopyResult.DROPPED); }); + +for (const future of [false, true]) { + for (const writable of [false, true]) { + Deno.test(`removed busy ${future ? "future" : "stream"} ${writable ? "writer" : "reader"} cannot copy stale guest bytes`, async () => { + const inst = new ComponentInstanceState(0, new Store()); + const { memory, view } = mkMemory(); + const u8 = { kind: "u8" } as const; + const shared = future + ? new SharedFutureImpl(u8) + : new SharedStreamImpl(u8); + const end = future + ? writable + ? new WritableFutureEnd(shared as SharedFutureImpl) + : new ReadableFutureEnd(shared as SharedFutureImpl) + : writable + ? new WritableStreamEnd(shared as SharedStreamImpl) + : new ReadableStreamEnd(shared as SharedStreamImpl); + const i = inst.handles.add(end); + const ctx = { + ...mkCtx(inst, view), + streamElem: () => u8, + futureElem: () => u8, + }; + const copy = future + ? (writable ? createFutureWrite : createFutureRead)( + { futureTable: 0, options: 0 }, + ctx, + inst, + ) + : (writable ? createStreamWrite : createStreamRead)( + { streamTable: 0, options: 0 }, + ctx, + inst, + ); + const bytes = new Uint8Array(memory.buffer, 64, 4); + bytes.set([1, 2, 3, 4]); + assertEq(copy(i, 64, 4), BLOCKED); + const buffer = shared.pendingBuffer!; + const drop = future + ? (writable ? createFutureDropWritable : createFutureDropReadable)( + { futureTable: 0 }, + ctx, + inst, + ) + : (writable ? createStreamDropWritable : createStreamDropReadable)( + { streamTable: 0 }, + ctx, + inst, + ); + const call = createLiftedFunction({ + name: "busy-drop", + ft: { params: [], results: [] }, + opts: { + ...ctx.options(), + async: false, + coreType: { params: [], results: [] }, + }, + stats: newStats(), + core: () => drop(i), + }); + const trap = caughtSync(call); + assert( + trap instanceof Trap, + "busy drop traps through the export boundary", + ); + assertEq(isInstancePoisoned(inst), true); + assertEq(inst.handles.array[i], null, "reference removal is destructive"); + assertEq(shared.pendingBuffer, null); + assertEq( + end.hasPendingEvent(), + false, + "removed end gets no phantom event", + ); + assertEq(poisonFailureOf(shared)?.cause, trap); + const peer = new HostBuffer(u8, writable ? null : new Uint8Array([9]), 1); + let result: CopyResult | undefined; + if (future) { + if (writable) { + assertAbandonTrap( + caughtSync(() => + (shared as SharedFutureImpl).read({}, peer as never, () => {}) + ), + "trapped while it held an end", + ); + } else {(shared as SharedFutureImpl).write({}, peer as never, (r) => + result = r);} + } else if (writable) { + (shared as SharedStreamImpl).read( + {}, + peer as never, + () => {}, + (r) => result = r, + ); + } else { + (shared as SharedStreamImpl).write( + {}, + peer as never, + () => {}, + (r) => result = r, + ); + } + if (!future || !writable) assertEq(result, CopyResult.DROPPED); + assertEq(buffer.progress, 0); + assertEq(peer.progress, 0); + assertEq( + bytes, + new Uint8Array([1, 2, 3, 4]), + "no later write reaches guest memory", + ); + }); + } +} + +for ( + const path of [ + "drop-type", + "drop-element", + "lift-busy", + "lift-set", + "lift-type", + "transfer-busy", + "transfer-element", + "error-context", + "waitable-set", + "subtask", + ] as const +) { + Deno.test(`invalid removed endpoint retires on ${path}, without prematurely retiring its instance`, () => { + const inst = new ComponentInstanceState(0, new Store()); + const dst = new ComponentInstanceState(1, inst.store); + const shared = new SharedStreamImpl(null); + const end = new ReadableStreamEnd(shared); + const i = inst.handles.add(end); + const untouched = new SharedStreamImpl(null); + inst.handles.add(new WritableStreamEnd(untouched)); + const ctx = { + ...mkCtx(inst, null), + streamTableInstance: (table: number) => table === 0 ? inst : dst, + futureTableInstance: (table: number) => table === 0 ? inst : dst, + streamElem: (table: number) => + path.endsWith("element") && table === 1 + ? { kind: "u8" } as const + : null, + }; + let notified = false; + let dropped = 0; + shared.whenDropped(() => dropped++); + shared.write( + {}, + new HostBuffer(null, [null], 1) as never, + () => {}, + (r) => { + assertEq(r, CopyResult.DROPPED); + assertEq(poisonFailureOf(shared) !== undefined, true); + notified = true; + throw new Error("peer notification must not hide validation trap"); + }, + ); + const cx = new LiftLowerContext(mkCanonicalOptions(), inst); + let run: () => unknown; + switch (path) { + case "drop-type": + run = () => createFutureDropReadable({ futureTable: 0 }, ctx, inst)(i); + break; + case "drop-element": + run = () => createStreamDropReadable({ streamTable: 1 }, ctx, inst)(i); + break; + case "lift-busy": + end.state = CopyState.COPYING; + run = () => liftStream(cx, i, { kind: "stream", element: null }); + break; + case "lift-set": + end.join(new WaitableSet()); + run = () => liftStream(cx, i, { kind: "stream", element: null }); + break; + case "lift-type": + run = () => liftFuture(cx, i, { kind: "future", element: null }); + break; + case "transfer-busy": + end.state = CopyState.COPYING; + run = () => createStreamTransfer(ctx)(i, 0, 1); + break; + case "transfer-element": + run = () => createStreamTransfer(ctx)(i, 0, 1); + break; + case "error-context": + run = () => createErrorContextDrop(inst)(i); + break; + case "waitable-set": + run = () => createWaitableSetDrop(inst)(i); + break; + case "subtask": + run = () => createSubtaskDrop(inst)(i); + break; + } + const trap = caughtSync(run); + assert( + trap instanceof Trap, + "original validation failure survives peer throw", + ); + assertEq(notified, true); + assertEq(inst.handles.array[i], null); + assertEq(shared.dropped, true); + assertEq(shared.pendingBuffer, null); + assertEq(dropped, 1, "peer failure cannot skip drop observers"); + assertEq(poisonFailureOf(shared)?.cause, trap); + assertEq(isInstancePoisoned(inst), false); + assertEq( + untouched.dropped, + false, + "local unwind does not retire the whole instance", + ); + notifyInstancePoisoned(inst, trap); + assertEq(untouched.dropped, true, "later poison walk still runs"); + shared.notifyDropped(); + assertEq(dropped, 1, "drop observers fire exactly once"); + }); +} + +Deno.test("an invalid unwritten future drop notifies a healthy guest reader with the fault", () => { + const f = mkFutureSplit(); + assertEq(f.run(() => f.read(f.ri, 0)), BLOCKED); + const trap = caughtSync(() => + createFutureDropWritable({ futureTable: 0 }, f.ctx, f.writer)(f.wi) + ); + assert(trap instanceof Trap, "unwritten future drop traps"); + assertEq(f.writer.handles.array[f.wi], null); + assertEq(f.readEnd.hasPendingEvent(), true); + assertAbandonTrap( + caughtSync(() => f.readEnd.getPendingEvent()), + "trapped while it held an end", + ); +}); + +Deno.test("a removed invalid end reports PeerTrappedError to parked and later host readers", async () => { + const inst = new ComponentInstanceState(0, new Store()); + const u8 = { kind: "u8" } as const; + const shared = new SharedStreamImpl(u8); + const i = inst.handles.add(new WritableStreamEnd(shared)); + const stream = Stream.fromLifted(shared as never, { + element: u8, + toHost: (v) => v as number, + fromHost: (v) => v, + }); + const pending = caughtAsync(stream.read(1)); + const trap = caughtSync(() => createErrorContextDrop(inst)(i)); + const error = await pending; + assert(error instanceof PeerTrappedError, "parked peer sees branded fault"); + assertEq((error.cause as Error).cause, trap); + assert( + await caughtAsync(stream.read(1)) instanceof PeerTrappedError, + "later peer sees branded fault", + ); +}); + +for (const future of [false, true]) { + for (const transfer of [false, true]) { + Deno.test(`successful ${future ? "future" : "stream"} ${transfer ? "FACT transfer" : "lift"} keeps its parked peer live`, () => { + const src = new ComponentInstanceState(0, new Store()); + const dst = new ComponentInstanceState(1, src.store); + const shared = future + ? new SharedFutureImpl(null) + : new SharedStreamImpl(null); + const end = future + ? new ReadableFutureEnd(shared as SharedFutureImpl) + : new ReadableStreamEnd(shared as SharedStreamImpl); + const i = src.handles.add(end); + let notified = false; + const buffer = new HostBuffer(null, [null], 1); + if (future) { + (shared as SharedFutureImpl).write( + {}, + buffer as never, + () => notified = true, + ); + } else {(shared as SharedStreamImpl).write({}, buffer as never, () => + notified = true, () => + notified = true);} + if (transfer) { + const ctx = { + ...mkCtx(src, null), + streamTableInstance: (t: number) => t ? dst : src, + futureTableInstance: (t: number) => t ? dst : src, + }; + const index = (future ? createFutureTransfer : createStreamTransfer)( + ctx, + )(i, 0, 1) as number; + assertEq( + (dst.handles.get(index) as ReadableStreamEnd).shared === shared, + true, + ); + } else { + const cx = new LiftLowerContext(mkCanonicalOptions(), src); + const value = future + ? liftFuture(cx, i, { kind: "future", element: null }) + : liftStream(cx, i, { kind: "stream", element: null }); + assertEq(value === shared, true); + } + assertEq(src.handles.array[i], null); + notifyInstancePoisoned(src, new Trap("after transfer")); + assertEq(notified, false); + assertEq(shared.dropped, false); + assertEq(shared.pendingBuffer === (buffer as unknown), true); + assertEq(poisonFailureOf(shared), undefined); + const out = new HostBuffer(null, null, 1); + if (future) (shared as SharedFutureImpl).read({}, out as never, () => {}); + else {(shared as SharedStreamImpl).read( + {}, + out as never, + () => {}, + () => {}, + );} + assertEq(out.progress, 1); + assertEq(notified, true); + }); + } +} diff --git a/runtime/tests/subtask_cancel_sync_waiter_window_test.ts b/runtime/tests/subtask_cancel_sync_waiter_window_test.ts index 33e91c9..a639f05 100644 --- a/runtime/tests/subtask_cancel_sync_waiter_window_test.ts +++ b/runtime/tests/subtask_cancel_sync_waiter_window_test.ts @@ -12,7 +12,7 @@ // `store.waiting`, so `determinate()` is false), a sibling thread's // `waitable.join` on the same subtask handle succeeds during the park. -import { assertEq } from "./support/asserts.ts"; +import { assertEq, assertTrap } from "./support/asserts.ts"; import { BLOCKED, createSubtaskCancel, @@ -20,16 +20,28 @@ import { } from "../src/intrinsics/async_builtins.ts"; import { ComponentInstanceState, + currentTask, + EventCode, popCurrentThread, pushCurrentThread, Store, Subtask, + SubtaskState, Task, type TaskOptions, Thread, + unpackSubtaskResult, WaitableSet, + withActivation, } from "../src/task/mod.ts"; import type { FuncType } from "../src/cabi/types.ts"; +import { + createAsyncStartCall, + createPrepareCall, + type FactCallContext, + START_FLAG_ASYNC_CALLEE, +} from "../src/intrinsics/fact_calls.ts"; +import { newStats } from "../src/exec/boundary.ts"; const FT: FuncType = { params: [], results: [], async: true }; const OPTS: TaskOptions = { @@ -101,3 +113,93 @@ Deno.test( assertEq(subtask.hasSyncWaiter, false); }, ); + +for (const progressBeforeCancel of [true, false]) { + Deno.test(`sync subtask.cancel waits past STARTED progress ${progressBeforeCancel ? "before" : "during"} its park`, async () => { + const store = new Store(); + const caller = new ComponentInstanceState(0, store); + const callee = new ComponentInstanceState(1, store); + const set = new WaitableSet(); + const seti = callee.handles.add(set); + const wait = (seti << 4) | 2; + let cancelled = false; + let resolve = false; + const ctx: FactCallContext = { + componentInstance: (i) => [caller, callee][i], + resultTypes: () => [], + resultTypesForTuple: () => [], + callback: () => (code: number) => { + if (code === EventCode.TASK_CANCELLED) cancelled = true; + if (resolve) { + (currentTask() as Task).cancel(); + return 0; + } + return wait; + }, + memoryToken: () => null, + stats: newStats(), + suspensionMode: "jspi", + prepared: { current: null }, + factStartScopes: [], + }; + const callerTask = new Task(FT, OPTS, caller, () => [], () => {}); + const callerThread = new Thread(callerTask, (function* () {})()); + const asGuest = (f: () => T) => withActivation(callerThread, f); + callee.backpressure = 1; + const packed = asGuest(() => { + createPrepareCall({ memory: null }, ctx)( + () => undefined, + () => undefined, + 0, + 1, + 0, + 1, + 0, + 0xffff_ffff, + ); + return createAsyncStartCall({ callback: 0, postReturn: null }, ctx)( + () => wait, + 0, + 0, + START_FLAG_ASYNC_CALLEE, + ); + }); + const [state, i] = unpackSubtaskResult(packed as number); + assertEq(state, SubtaskState.STARTING); + callee.backpressure = 0; + assertEq(store.tick(), true); + const st = caller.handles.get(i) as Subtask; + assertEq(st.state, SubtaskState.STARTED); + assertEq(st.hasPendingEvent(), true); + if (!progressBeforeCancel) st.getPendingEvent(); + const lender = { numLends: 0 }; + st.addLender(lender); + const pending = asGuest(() => + createSubtaskCancel({ async: false }, caller, "jspi")(i) + ) as unknown as Promise; + assertEq(pending instanceof Promise, true); + assertEq(cancelled, true); + assertEq(st.hasSyncWaiter, true); + if (!progressBeforeCancel) st.setSubtaskPendingEvent(i); + assertEq(store.tick(), false, "STARTED is not resolution"); + const joinSet = caller.handles.add(new WaitableSet()); + assertTrap(() => asGuest(() => createWaitableJoin(caller)(i, joinSet))); + assertEq(lender.numLends, 1); + + // A normal callback event lets the cooperatively cancelled callee resolve. + resolve = true; + const signal = new Subtask(); + signal.join(set); + signal.setSubtaskPendingEvent(1); + assertEq(store.tick(), true); + assertEq(st.resolved(), true); + assertEq(st.hasSyncWaiter, true); + assertEq(store.tick(), true); + assertEq(await pending, SubtaskState.CANCELLED_BEFORE_RETURNED); + assertEq(st.resolveDelivered(), true); + assertEq(st.hasSyncWaiter, false); + assertEq(st.hasPendingEvent(), false); + assertEq(lender.numLends, 0); + asGuest(() => createWaitableJoin(caller)(i, joinSet)); + }); +}