From 2f6fa98c0235c4ac605a7e110992c0455edf6063 Mon Sep 17 00:00:00 2001 From: fffonion Date: Wed, 26 Aug 2026 08:53:17 +0800 Subject: [PATCH] feat(vm): expose cancellable invocation item streams --- build.rs | 17 +- crates/rustscript/tests/alias_smoke.rs | 28 + docs/callable-runtime.md | 12 + src/builtins/runtime/context.rs | 74 ++ src/builtins/runtime/context_host.rs | 12 + src/builtins/runtime/error.rs | 138 ++++ src/builtins/runtime/event.rs | 195 +++++ src/builtins/runtime/mod.rs | 7 + src/compiler/typing/context.rs | 86 ++ src/compiler/typing/helpers.rs | 2 + src/compiler/typing/state.rs | 8 + src/lib.rs | 18 +- src/vm/host.rs | 17 + src/vm/instance.rs | 26 + src/vm/invocation.rs | 510 ++++++++++++ src/vm/mod.rs | 3 + src/vm/run_context.rs | 14 +- src/vm/tests.rs | 104 +++ tests/compiler/compiler_rustscript_tests.rs | 22 + tests/invocation_stream_tests.rs | 870 ++++++++++++++++++++ 20 files changed, 2144 insertions(+), 19 deletions(-) create mode 100644 src/builtins/runtime/context.rs create mode 100644 src/builtins/runtime/context_host.rs create mode 100644 src/builtins/runtime/error.rs create mode 100644 src/builtins/runtime/event.rs create mode 100644 src/vm/invocation.rs create mode 100644 tests/invocation_stream_tests.rs diff --git a/build.rs b/build.rs index 33dae8b8..8c4f49ac 100644 --- a/build.rs +++ b/build.rs @@ -166,11 +166,18 @@ fn main() { catalog.retain(|entry| !entry.source_name.starts_with("sqlite::")); } - let host_sources = [SourceSpec { - path: "src/builtins/runtime/host.rs".to_string(), - module: "host".to_string(), - category: SourceCategory::DefaultHost, - }]; + let host_sources = vec![ + SourceSpec { + path: "src/builtins/runtime/host.rs".to_string(), + module: "host".to_string(), + category: SourceCategory::DefaultHost, + }, + SourceSpec { + path: "src/builtins/runtime/context_host.rs".to_string(), + module: "context_host".to_string(), + category: SourceCategory::DefaultHost, + }, + ]; let builtin_sources = builtin_source_specs(&namespaces); let core_sources = [SourceSpec { path: "src/builtins/runtime/core.rs".to_string(), diff --git a/crates/rustscript/tests/alias_smoke.rs b/crates/rustscript/tests/alias_smoke.rs index c58bb302..5b709560 100644 --- a/crates/rustscript/tests/alias_smoke.rs +++ b/crates/rustscript/tests/alias_smoke.rs @@ -21,3 +21,31 @@ fn alias_exports_op_code() { let _ = rustscript::OpCode::Nop; let _ = rustscript::OpCode::Add; } + +#[cfg(feature = "runtime")] +#[test] +fn alias_exports_public_invocation_stream_contract() { + fn accept_item(_item: rustscript::InvocationItem) {} + + accept_item(rustscript::InvocationItem::Complete( + rustscript::Value::Null, + )); + accept_item(rustscript::InvocationItem::Event(rustscript::Value::Bool( + true, + ))); + + fn accept_poll(_poll: rustscript::InvocationPoll) {} + accept_poll(rustscript::InvocationPoll::Pending); + accept_poll(rustscript::InvocationPoll::Ready(None)); + accept_poll(rustscript::InvocationPoll::Ready(Some(Ok( + rustscript::InvocationItem::Complete(rustscript::Value::Null), + )))); + + fn accept_error(_error: rustscript::InvocationError) {} + accept_error(rustscript::InvocationError::Cancelled( + rustscript::operation::OperationCancelReason::Requested, + )); + accept_error(rustscript::InvocationError::Host { + message: "boom".to_string(), + }); +} diff --git a/docs/callable-runtime.md b/docs/callable-runtime.md index ed0cfd4b..a012c8b9 100644 --- a/docs/callable-runtime.md +++ b/docs/callable-runtime.md @@ -43,6 +43,18 @@ Reset clears Program runtime values and rebinds root function items from Program PDRC recordings preserve full execution-frame metadata. Callable environments use identity-table encoding, so aliases still share one environment after decode. +## Invocation item stream + +`Vm::start_invocation` starts one exported callable with ordinary `Value` arguments and returns an `Invocation` handle that behaves like a fused `Stream>`: + +- `InvocationItem::Event(value)` items arrive in order for each `stream::emit(value)` call; `stream::emit` still evaluates to `()` inside RSS. +- exactly one `InvocationItem::Complete(value)` carries the callable return value; events never replace it; +- cancellation, fuel exhaustion, epoch deadline expiry, runtime capability failures (including event payload bound violations), and host failures each produce exactly one typed `InvocationError` item; +- every poll after `Complete` or the error item returns `Ready(None)` (fused end of stream); +- `InvocationPoll::Pending` means the VM is paused on an outstanding host operation; drive it through the embedding-owned async bridge and poll again. + +Polling drives execution and provides backpressure: at most one event item is buffered between polls, and the VM does not produce items while the consumer is not polling. `stream::emit` validates only the configured per-item value bound (payload bytes and nesting depth); sequence assignment, receipts, persistence, and delivery policy belong to the embedding. At most one invocation is active per VM, `Invocation::cancel(reason)` cancels with a typed `OperationCancelReason`, dropping the handle retires the invocation synchronously for immediate VM reuse, and the low-level `Vm::run` pump is unchanged for custom drivers. + ## Optimized backends Whole-program AOT and Trace JIT use the same builtin call path (static catalog IDs) for environment binding and native frame dispatch for `callvalue`. Script-frame entry and return preserve frame-relative locals and typed continuations. diff --git a/src/builtins/runtime/context.rs b/src/builtins/runtime/context.rs new file mode 100644 index 00000000..8a3068c3 --- /dev/null +++ b/src/builtins/runtime/context.rs @@ -0,0 +1,74 @@ +//! Run-scoped invocation stream configuration. +//! +//! The [`RuntimeContext`] carries only the per-item event bound applied by +//! `stream::emit`. Event values are owned by the active invocation's single +//! pending-event slot; there is no ambient input, no embedding event sink, and +//! no sequence or persistence policy here. + +use super::error::RuntimeResult; +use super::event::EventLimits; + +/// The authoritative `stream::emit` builtin identity. +#[allow(dead_code)] +pub const STREAM_EMIT_NAME: &str = "stream::emit"; + +/// Configuration for one VM/run-scoped invocation stream. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct RuntimeContextConfig { + event_limits: EventLimits, +} + +#[allow(dead_code)] +impl RuntimeContextConfig { + pub fn new(event_limits: EventLimits) -> Self { + Self { event_limits } + } + + #[allow(dead_code)] + pub const fn event_limits(self) -> EventLimits { + self.event_limits + } +} + +/// Run-scoped invocation stream configuration. +#[derive(Debug, Default)] +pub struct RuntimeContext { + event_limits: EventLimits, +} + +#[allow(dead_code)] +impl RuntimeContext { + pub fn with_config(config: RuntimeContextConfig) -> RuntimeResult { + Ok(Self { + event_limits: config.event_limits, + }) + } + + pub fn config(&self) -> RuntimeContextConfig { + RuntimeContextConfig::new(self.event_limits) + } + + pub fn event_limits(&self) -> EventLimits { + self.event_limits + } +} + +#[cfg(test)] +mod tests { + use super::{EventLimits, RuntimeContext, RuntimeContextConfig, STREAM_EMIT_NAME}; + + #[test] + fn host_name_is_generic_and_stable() { + assert_eq!(STREAM_EMIT_NAME, "stream::emit"); + assert!(std::mem::size_of::() > 0); + } + + #[test] + fn per_item_event_limits_are_configurable() { + let limits = EventLimits::new(128, 4).expect("limits should be valid"); + let context = RuntimeContext::with_config(RuntimeContextConfig::new(limits)) + .expect("context should be constructible"); + assert_eq!(context.event_limits(), limits); + assert_eq!(context.config().event_limits(), limits); + } +} diff --git a/src/builtins/runtime/context_host.rs b/src/builtins/runtime/context_host.rs new file mode 100644 index 00000000..0cc3ba32 --- /dev/null +++ b/src/builtins/runtime/context_host.rs @@ -0,0 +1,12 @@ +use pd_host_function::pd_host_function; + +use super::AnyValue; +use crate::vm::{CallOutcome, Vm, VmResult}; + +/// Places one bounded event item on the active invocation stream and yields +/// control to the invocation poller. `stream::emit` still evaluates to `()` +/// inside RSS. +#[pd_host_function(name = "stream::emit")] +fn stream_emit_impl(vm: &mut Vm, value: AnyValue) -> VmResult { + vm.emit_stream_item(value) +} diff --git a/src/builtins/runtime/error.rs b/src/builtins/runtime/error.rs new file mode 100644 index 00000000..e8062a7e --- /dev/null +++ b/src/builtins/runtime/error.rs @@ -0,0 +1,138 @@ +//! Structured runtime error types shared by the invocation stream. +//! +//! A [`RuntimeError`] carries a stable machine-readable [`RuntimeErrorCode`], +//! the offending builtin operation name, and optional numeric limit/value +//! fields. The invocation stream preserves these instead of flattening them +//! to a string, so an embedding can branch on the code and inspect the +//! numeric state (payload bytes, depth) without string matching. + +use std::fmt; + +/// Result alias used by runtime builtin surfaces. +pub type RuntimeResult = Result; + +/// Stable machine-readable runtime error codes. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RuntimeErrorCode { + InvalidConfiguration, + EventPayloadTooLarge, + EventDepthExceeded, + ResourceLimitExceeded, + InvalidResourceHandle, + ResourceHandleWrongTable, + OperationFailed, + OperationAlreadyTerminal, + OperationCancelled, + SyncResourceUnavailable, + CloseFailed, +} + +impl RuntimeErrorCode { + /// Stable snake_case string form, used for transport and tests. + pub const fn as_str(self) -> &'static str { + match self { + Self::InvalidConfiguration => "invalid_configuration", + Self::EventPayloadTooLarge => "event_payload_too_large", + Self::EventDepthExceeded => "event_depth_exceeded", + Self::ResourceLimitExceeded => "resource_limit_exceeded", + Self::InvalidResourceHandle => "invalid_resource_handle", + Self::ResourceHandleWrongTable => "resource_handle_wrong_table", + Self::OperationFailed => "operation_failed", + Self::OperationAlreadyTerminal => "operation_already_terminal", + Self::OperationCancelled => "operation_cancelled", + Self::SyncResourceUnavailable => "sync_resource_unavailable", + Self::CloseFailed => "close_failed", + } + } +} + +/// A structured runtime error with a stable code and optional numeric state. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RuntimeError { + code: RuntimeErrorCode, + operation: String, + message: String, + limit: Option, + value: Option, +} + +impl RuntimeError { + pub fn new(code: RuntimeErrorCode, operation: &str, message: impl Into) -> Self { + Self { + code, + operation: operation.to_string(), + message: message.into(), + limit: None, + value: None, + } + } + + /// Attaches the configured bound that was violated. + pub fn with_limit(mut self, limit: usize) -> Self { + self.limit = Some(limit as u64); + self + } + + /// Attaches the offending value (for example the measured payload size). + pub fn with_value(mut self, value: usize) -> Self { + self.value = Some(value as u64); + self + } + + pub fn code(&self) -> RuntimeErrorCode { + self.code + } + + pub fn operation(&self) -> &str { + &self.operation + } + + pub fn limit(&self) -> Option { + self.limit + } + + pub fn value(&self) -> Option { + self.value + } + + pub fn message(&self) -> &str { + &self.message + } +} + +impl fmt::Display for RuntimeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}: {}", self.code.as_str(), self.message)?; + if let Some(limit) = self.limit { + write!(f, " (limit {limit})")?; + } + if let Some(value) = self.value { + write!(f, " (value {value})")?; + } + Ok(()) + } +} + +impl std::error::Error for RuntimeError {} + +#[cfg(test)] +mod tests { + use super::{RuntimeError, RuntimeErrorCode}; + + #[test] + fn structured_error_preserves_code_and_fields() { + let error = RuntimeError::new( + RuntimeErrorCode::EventPayloadTooLarge, + "stream::emit", + "event payload exceeds the configured bound", + ) + .with_limit(32) + .with_value(64); + + assert_eq!(error.code(), RuntimeErrorCode::EventPayloadTooLarge); + assert_eq!(error.operation(), "stream::emit"); + assert_eq!(error.limit(), Some(32)); + assert_eq!(error.value(), Some(64)); + assert!(error.to_string().contains("event_payload_too_large")); + } +} diff --git a/src/builtins/runtime/event.rs b/src/builtins/runtime/event.rs new file mode 100644 index 00000000..f5778290 --- /dev/null +++ b/src/builtins/runtime/event.rs @@ -0,0 +1,195 @@ +//! Per-item event bounds for the invocation item stream. +//! +//! [`EventLimits`] configures the per-item bound applied to one +//! `stream::emit(value)` call: a maximum payload byte estimate and a maximum +//! nesting depth. The core validates only this per-item value bound before +//! placing the value in the active invocation's single pending-event slot. +//! Sequence assignment, cumulative byte accounting, event receipts, and +//! embedding-owned sinks are not part of the core contract; delivery policy +//! belongs to the embedding. + +use crate::vm::Value; + +use super::error::{RuntimeError, RuntimeErrorCode, RuntimeResult}; + +pub const DEFAULT_MAX_EVENT_PAYLOAD_BYTES: usize = 64 * 1024; +pub const DEFAULT_MAX_EVENT_DEPTH: usize = 64; + +/// Per-item bounds applied to one `stream::emit(value)` call. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct EventLimits { + max_payload_bytes: usize, + max_depth: usize, +} + +#[allow(dead_code)] +impl EventLimits { + pub fn new(max_payload_bytes: usize, max_depth: usize) -> RuntimeResult { + if max_payload_bytes == 0 || max_depth == 0 { + return Err(RuntimeError::new( + RuntimeErrorCode::InvalidConfiguration, + "stream::emit", + "event payload and depth limits must be positive", + )); + } + Ok(Self { + max_payload_bytes, + max_depth, + }) + } + + pub const fn max_payload_bytes(self) -> usize { + self.max_payload_bytes + } + + pub const fn max_depth(self) -> usize { + self.max_depth + } +} + +impl Default for EventLimits { + fn default() -> Self { + Self { + max_payload_bytes: DEFAULT_MAX_EVENT_PAYLOAD_BYTES, + max_depth: DEFAULT_MAX_EVENT_DEPTH, + } + } +} + +/// An event value whose per-item bound has been validated. +#[derive(Clone, Debug, PartialEq)] +pub struct EventPayload { + value: Value, + size_bytes: usize, +} + +impl EventPayload { + /// Validates a value against the per-item bound and preserves the + /// validated value plus its bounded size estimate. + pub fn try_new(value: Value, limits: EventLimits) -> RuntimeResult { + let size_bytes = measure_value(&value, 0, limits)?; + Ok(Self { value, size_bytes }) + } + + pub fn into_value(self) -> Value { + self.value + } + + #[allow(dead_code)] + pub fn size_bytes(&self) -> usize { + self.size_bytes + } +} + +/// Estimates the bounded representation size of a value. +/// +/// The estimate is deliberately independent of serialization formats. It counts scalar tags, +/// container headers, string/byte contents, and recursively contained values. The host transport +/// may produce larger or smaller blobs; this bound is a conservative per-item budget used to +/// reject oversized event payloads before they enter the single pending-event slot. +fn measure_value(value: &Value, depth: usize, limits: EventLimits) -> RuntimeResult { + if depth > limits.max_depth { + return Err(RuntimeError::new( + RuntimeErrorCode::EventDepthExceeded, + "stream::emit", + "event payload nesting exceeds the configured bound", + ) + .with_limit(limits.max_depth) + .with_value(depth)); + } + + let size = match value { + Value::Null => 1, + Value::Bool(_) => 1, + Value::Int(_) => 8, + Value::Float(_) => 8, + Value::String(text) => 2 * text.len() + 1, + Value::Bytes(bytes) => bytes.len() + 2, + Value::Array(items) => { + let mut size = 2usize; + for item in items.iter() { + size = checked_payload_add(size, measure_value(item, depth + 1, limits)?, limits)?; + } + size + } + Value::Map(entries) => { + let mut size = 2usize; + for (key, value) in entries.iter() { + size = checked_payload_add(size, measure_value(key, depth + 1, limits)?, limits)?; + size = checked_payload_add(size, measure_value(value, depth + 1, limits)?, limits)?; + } + size + } + Value::Callable(_) => 8, + }; + if size > limits.max_payload_bytes { + return Err(RuntimeError::new( + RuntimeErrorCode::EventPayloadTooLarge, + "stream::emit", + "event payload exceeds the configured byte bound", + ) + .with_limit(limits.max_payload_bytes) + .with_value(size)); + } + Ok(size) +} + +fn checked_payload_add( + current: usize, + additional: usize, + limits: EventLimits, +) -> RuntimeResult { + let total = current.checked_add(additional).ok_or_else(|| { + RuntimeError::new( + RuntimeErrorCode::EventPayloadTooLarge, + "stream::emit", + "event payload size overflowed", + ) + .with_limit(limits.max_payload_bytes) + })?; + if total > limits.max_payload_bytes { + return Err(RuntimeError::new( + RuntimeErrorCode::EventPayloadTooLarge, + "stream::emit", + "event payload exceeds the configured byte bound", + ) + .with_limit(limits.max_payload_bytes) + .with_value(total)); + } + Ok(total) +} + +#[cfg(test)] +mod tests { + use super::{EventLimits, EventPayload}; + use crate::vm::Value; + + #[test] + fn per_item_limits_validate_payload_and_depth() { + let limits = EventLimits::new(32, 4).expect("limits should be valid"); + let payload = + EventPayload::try_new(Value::string("event"), limits).expect("payload should fit"); + assert!(payload.size_bytes() >= 5); + assert_eq!(payload.into_value(), Value::string("event")); + } + + #[test] + fn oversized_or_too_deep_values_are_rejected_before_placement() { + let limits = EventLimits::new(8, 2).expect("limits should be valid"); + let too_large = EventPayload::try_new(Value::string("payload-too-large"), limits) + .expect_err("oversized event should be rejected"); + assert_eq!( + too_large.code(), + super::super::error::RuntimeErrorCode::EventPayloadTooLarge + ); + let too_deep = EventPayload::try_new( + Value::array(vec![Value::array(vec![Value::array(vec![Value::Int(1)])])]), + limits, + ) + .expect_err("too-deep event should be rejected"); + assert_eq!( + too_deep.code(), + super::super::error::RuntimeErrorCode::EventDepthExceeded + ); + } +} diff --git a/src/builtins/runtime/mod.rs b/src/builtins/runtime/mod.rs index 29c27e21..b5bf31f3 100644 --- a/src/builtins/runtime/mod.rs +++ b/src/builtins/runtime/mod.rs @@ -10,7 +10,11 @@ use crate::vm::{CallOutcome, CallReturn, HostOpId, Value, Vm, VmError, VmResult} mod aot; mod bytes; +pub(crate) mod context; +pub(crate) mod context_host; pub(crate) mod core; +pub(crate) mod error; +pub(crate) mod event; mod host; #[cfg(not(target_arch = "wasm32"))] mod io; @@ -30,6 +34,9 @@ mod typed; #[cfg(target_arch = "wasm32")] use io_wasm as io; +pub(crate) use context::{RuntimeContext, RuntimeContextConfig, STREAM_EMIT_NAME}; +pub use error::{RuntimeError, RuntimeErrorCode, RuntimeResult}; +pub(crate) use event::{EventLimits, EventPayload}; pub(crate) use io::IoState; #[cfg(not(target_arch = "wasm32"))] pub use io::{IoHostExt, IoPolicy}; diff --git a/src/compiler/typing/context.rs b/src/compiler/typing/context.rs index efdfb045..10de4df3 100644 --- a/src/compiler/typing/context.rs +++ b/src/compiler/typing/context.rs @@ -1987,6 +1987,26 @@ impl<'a> TypeContext<'a> { } return Ok(()); } + // `stream::emit(value)` accepts any single value; the per-item event + // bound is validated at runtime by the invocation stream. The + // exemption is tied to the authoritative runtime builtin identity; a + // same-name function registered through another catalog does not + // inherit it. The identity constant lives in the `runtime`-featured + // builtins module, so in non-runtime builds the comparison is + // compiled out and the exemption does not apply. + #[cfg(feature = "runtime")] + if signature.runtime_builtin && signature.name == crate::builtins::runtime::STREAM_EMIT_NAME + { + return validate_host_signature( + &signature.name, + &signature.params, + args, + state, + self, + line_context, + source_name, + ); + } if self.is_strict() && signature .params @@ -2609,3 +2629,69 @@ fn literal_int_index(key: &Expr) -> Option { }; usize::try_from(*index).ok() } + +#[cfg(test)] +mod tests { + use super::*; + use crate::builtins::{CallableParam, CallableParamType}; + + /// The authoritative `stream::emit` signature: one `any` payload. + #[cfg(feature = "runtime")] + fn emit_signature(runtime_builtin: bool) -> HostCallableSignature { + HostCallableSignature { + name: crate::builtins::runtime::STREAM_EMIT_NAME.to_string(), + params: vec![CallableParam { + name: "value", + ty: CallableParamType::Any, + optional: false, + }], + runtime_builtin, + } + } + + #[test] + #[cfg(feature = "runtime")] + fn stream_emit_any_payload_exemption_requires_authoritative_builtin_identity() { + let empty_impls: HashMap = HashMap::new(); + let empty_decls: HashMap = HashMap::new(); + let empty_structs: HashMap = HashMap::new(); + let empty_names: HashMap = HashMap::new(); + let empty_returns: HashMap = HashMap::new(); + let empty_signatures: HashMap = HashMap::new(); + let mut context = TypeContext::new( + &empty_impls, + &empty_decls, + &empty_structs, + &empty_names, + &empty_returns, + &empty_signatures, + TypingMode::StrictRustScript, + ); + let state = LocalTypeState::default(); + let args = [Expr::Int(1)]; + + assert!( + context + .validate_host_argument_types(&emit_signature(true), &args, &state, None, None) + .is_ok(), + "the authoritative stream::emit builtin must accept any payload in strict mode" + ); + + // A same-name signature that is not the authoritative runtime builtin + // (for example one registered through another host catalog) must not + // inherit the strict-typing exemption. + assert!( + matches!( + context.validate_host_argument_types( + &emit_signature(false), + &args, + &state, + None, + None, + ), + Err(CompileError::StrictTypingRequired { .. }) + ), + "a same-name non-builtin signature must not inherit the stream::emit exemption" + ); + } +} diff --git a/src/compiler/typing/helpers.rs b/src/compiler/typing/helpers.rs index 28e76a8e..4c6d8a31 100644 --- a/src/compiler/typing/helpers.rs +++ b/src/compiler/typing/helpers.rs @@ -1234,6 +1234,7 @@ pub(super) fn known_host_signature(name: &str) -> Option return Some(HostCallableSignature { name: callable.name.to_string(), params: callable.signature.params.to_vec(), + runtime_builtin: true, }); } @@ -1253,6 +1254,7 @@ pub(super) fn known_host_signature(name: &str) -> Option optional: false, }) .collect(), + runtime_builtin: false, }) } diff --git a/src/compiler/typing/state.rs b/src/compiler/typing/state.rs index 5020bb27..ef273e49 100644 --- a/src/compiler/typing/state.rs +++ b/src/compiler/typing/state.rs @@ -435,4 +435,12 @@ pub(crate) struct TypeInferenceResult { pub(crate) struct HostCallableSignature { pub(crate) name: String, pub(crate) params: Vec, + /// True when this signature came from the authoritative runtime builtin + /// catalog (`default_host_callable`), false when it came from another + /// catalog such as edge ABI host functions. Strict-typing exemptions that + /// are tied to a builtin identity must check this marker so a same-name + /// function from another catalog cannot inherit them. Only read under the + /// `runtime` feature; in non-runtime builds the exemption is compiled out. + #[cfg_attr(not(feature = "runtime"), allow(dead_code))] + pub(crate) runtime_builtin: bool, } diff --git a/src/lib.rs b/src/lib.rs index 5abe5d2c..4f1c3c34 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,14 +27,14 @@ pub use builtins::runtime::print::{PrintHostFunction, PrintlnHostFunction, forma #[cfg(all(feature = "runtime", feature = "sqlite", not(target_arch = "wasm32")))] pub use builtins::runtime::sqlite::{SqliteLimits, SqlitePolicy}; #[cfg(feature = "runtime")] -pub use builtins::runtime::standard_composition; -#[cfg(feature = "runtime")] pub use builtins::runtime::{ BorrowVmValue, FromVmValue, HostCallResult, IntoHostCallOutcome, TakeVmValue, arg, borrow_arg, return_one, take_arg, }; #[cfg(all(feature = "runtime", not(target_arch = "wasm32")))] pub use builtins::runtime::{IoHostExt, IoPolicy}; +#[cfg(feature = "runtime")] +pub use builtins::runtime::{RuntimeError, RuntimeErrorCode, RuntimeResult, standard_composition}; pub use builtins::{ BUILTIN_CATALOG, BuiltinFunction, BuiltinNamespaceMemberSpec, BuiltinNamespaceSpec, CallableDef, CallableParam, CallableParamType, CallableSignature, HostExecution, @@ -102,13 +102,13 @@ pub use vm::{ FuelCheckpoint, HostArgsFunction, HostAsyncBridge, HostBindingPlan, HostContext, HostContextError, HostContextErrorKind, HostContextResult, HostExtension, HostFunction, HostFunctionRegistry, HostFuture, HostFutureOutput, HostImportParam, HostImportSchema, - HostModule, HostModuleState, HostOpId, HostStackFunction, IntoScriptValue, - QueuedScriptInvocation, ResourceCloseReason, ScriptArgs, ScriptCallback, ScriptResult, - StandardSurfaceComposition, StaticHostArgsFunction, StaticHostFunction, - StaticHostStackFunction, Store, Vm, VmError, VmResult, VmStatus, VmYieldReason, async_host, - catalog_import_schemas, execution_scope, host_context, host_extension, operation, - register_catalog_function, resource, validate_catalog_import_schemas, - validate_catalog_import_schemas_with_fingerprints, + HostModule, HostModuleState, HostOpId, HostStackFunction, IntoScriptValue, Invocation, + InvocationError, InvocationItem, InvocationPoll, QueuedScriptInvocation, ResourceCloseReason, + ScriptArgs, ScriptCallback, ScriptResult, StandardSurfaceComposition, StaticHostArgsFunction, + StaticHostFunction, StaticHostStackFunction, Store, Vm, VmError, VmResult, VmStatus, + VmYieldReason, async_host, catalog_import_schemas, execution_scope, host_context, + host_extension, operation, register_catalog_function, resource, + validate_catalog_import_schemas, validate_catalog_import_schemas_with_fingerprints, }; #[cfg(feature = "runtime")] pub use vmbc::{ diff --git a/src/vm/host.rs b/src/vm/host.rs index bc160fe2..692aa37f 100644 --- a/src/vm/host.rs +++ b/src/vm/host.rs @@ -1124,6 +1124,23 @@ impl Vm { self.host.runtime_print_sink = None; } + /// Configures the per-item event bound applied by `stream::emit` on the + /// invocation stream. + /// + /// `max_payload_bytes` bounds the estimated payload size of one emitted + /// event value; `max_depth` bounds its nesting depth. Both must be + /// positive. The bound is run-scoped configuration and survives a VM + /// reset, matching the other run-scoped configuration on the facade. + pub fn set_event_limits(&mut self, max_payload_bytes: usize, max_depth: usize) -> VmResult<()> { + let limits = crate::builtins::runtime::EventLimits::new(max_payload_bytes, max_depth) + .map_err(|error| VmError::HostError(error.to_string()))?; + self.run_ctx.runtime_context = crate::builtins::runtime::RuntimeContext::with_config( + crate::builtins::runtime::RuntimeContextConfig::new(limits), + ) + .map_err(|error| VmError::HostError(error.to_string()))?; + Ok(()) + } + pub(crate) fn write_runtime_print(&mut self, rendered: String) -> VmResult<()> { let Some(sink) = self.host.runtime_print_sink.as_mut() else { return Err(VmError::HostError( diff --git a/src/vm/instance.rs b/src/vm/instance.rs index baecdfb0..86175bac 100644 --- a/src/vm/instance.rs +++ b/src/vm/instance.rs @@ -19,6 +19,7 @@ use std::sync::{Arc, Weak}; use crate::bytecode::{CallableValue, Program, SharedCaptureCell, Value}; use crate::vm::host::WaitingHostOp; +use crate::vm::invocation::{InvocationPhase, InvocationState}; use crate::vm::map_iter::MapIteratorState; use crate::vm::{DEFAULT_MAX_SCRIPT_CALL_DEPTH, VmYieldReason}; @@ -84,6 +85,7 @@ pub(crate) struct Instance { pub(crate) shutdown: bool, pub(super) waiting_host_op: Option, pub(crate) last_yield_reason: Option, + pub(crate) invocation: Option, pub(crate) map_iterators: Vec>>, pub(crate) drop_contract_events_enabled: bool, pub(crate) drop_contract_events: u64, @@ -120,6 +122,7 @@ impl Instance { shutdown: false, waiting_host_op: None, last_yield_reason: None, + invocation: None, map_iterators: Vec::new(), drop_contract_events_enabled: false, drop_contract_events: 0, @@ -161,6 +164,8 @@ impl Instance { self.draining_queued_callables = false; self.shutdown = false; self.waiting_host_op = None; + self.drop_invocation_state(); + self.invocation = None; self.map_iterators.clear(); self.clear_interpreter_metrics(); } @@ -168,12 +173,33 @@ impl Instance { /// Releases interpreter-owned values with drop-contract accounting. Used by /// the facade's `Drop` (and by `shutdown`). pub(crate) fn drop_cleanup(&mut self) { + self.drop_invocation_state(); self.clear_stack_with_drop_contract(); self.capture_cells.clear(); self.shared_capture_slots.clear(); self.clear_locals_with_drop_contract(); } + /// Drops pending invocation stream values with drop-contract accounting and + /// rewinds the invocation state to a fresh, fused position. + pub(crate) fn drop_invocation_state(&mut self) { + let Some(state) = self.invocation.as_mut() else { + return; + }; + let value = match std::mem::replace(&mut state.phase, InvocationPhase::Fused) { + InvocationPhase::EventPending(value) | InvocationPhase::CompletePending(value) => { + Some(value) + } + _ => None, + }; + state.emit_yield_pending = false; + state.pending_error = None; + state.cancel_reason = None; + if let Some(value) = value { + self.drop_value_with_contract(value); + } + } + pub(crate) fn invalidate_callback_registries(&mut self) { for active in self .callback_registry_flags diff --git a/src/vm/invocation.rs b/src/vm/invocation.rs new file mode 100644 index 00000000..62bd49b4 --- /dev/null +++ b/src/vm/invocation.rs @@ -0,0 +1,510 @@ +//! Invocation item stream. +//! +//! One exported callable started with ordinary arguments behaves like +//! `Stream>`: zero or more +//! `Event` items produced by `stream::emit`, then exactly one `Complete` item +//! or one typed error, then a fused end of stream. Polling drives execution; +//! the VM does not produce items while the consumer is not polling, and at most +//! one event item is buffered between polls (natural backpressure). +//! +//! The invocation reuses the existing callable execution state +//! ([`Vm::start_callable`], [`Vm::run`], [`Vm::take_callable_result`]) and the +//! existing async host bridge; it does not duplicate interpreter or host loops, +//! and it does not add an executor, generator syntax, an event queue, or event +//! persistence policy. Cancellation is a per-invocation typed reason carried +//! on the invocation state and forwarded to outstanding waiting host +//! operations; there is no standalone cancellation-token graph or parallel +//! event subsystem. + +use std::fmt; +use std::task::{Context, Poll, Waker}; + +use crate::builtins::runtime::EventPayload; +use crate::builtins::runtime::error::RuntimeError; +use crate::vm::operation::reason::OperationCancelReason; +use crate::vm::{CallOutcome, CallReturn, Value, Vm, VmError, VmResult, VmStatus, VmYieldReason}; + +/// One item yielded by an invocation stream. +#[derive(Clone, Debug, PartialEq)] +pub enum InvocationItem { + /// One bounded event produced by `stream::emit(value)`. + Event(Value), + /// The callable's return value; exactly one per invocation. + Complete(Value), +} + +/// Typed terminal failure of an invocation stream. +/// +/// The failure is machine-readable: cancellation keeps its reason, fuel and +/// deadline failures keep their numeric state, and stream::emit validation +/// keeps its structured [`RuntimeError`] instead of being flattened to a +/// string. +#[derive(Debug)] +pub enum InvocationError { + /// The invocation was cancelled with this reason. + Cancelled(OperationCancelReason), + /// The configured fuel budget was exhausted. + OutOfFuel { needed: u64, remaining: u64 }, + /// The configured epoch deadline expired. + DeadlineReached { current: u64, deadline: u64 }, + /// A structured runtime error (for example event payload validation). + Capability(RuntimeError), + /// An embedding host failure without a structured runtime code. + Host { message: String }, + /// A low-level VM failure (script error or invalid frame state). + Vm(VmError), +} + +/// Poll outcome of an invocation stream. +#[derive(Debug)] +pub enum InvocationPoll { + /// The VM is paused (waiting on a host operation or a host-driven yield); + /// drive the outstanding work and poll again. + Pending, + /// One stream item, or `None` after the fused end of stream. + Ready(Option>), +} + +/// Run-scoped state of the single active invocation on a VM. +#[derive(Debug)] +pub(crate) struct InvocationState { + pub(crate) phase: InvocationPhase, + /// True while the VM is yielded at a `stream::emit` call site whose event + /// has already been delivered. The resumed call site re-enters + /// `stream::emit` and consumes this marker instead of emitting a second + /// event for the same call. + pub(crate) emit_yield_pending: bool, + /// A structured runtime error produced by `stream::emit` validation, + /// preserved for the terminal error item without string flattening. + pub(crate) pending_error: Option, + /// A typed cancellation request made through [`Invocation::cancel`], + /// consumed by the poller to produce exactly one `Cancelled` item. + /// Per-invocation: cleared on fusion so it cannot leak into a later + /// invocation started on the same VM. + pub(crate) cancel_reason: Option, + /// Stack and frame position recorded when the invocation started, used to + /// release interpreter state on terminal failure. + pub(crate) stack_base: usize, + pub(crate) frame_count: usize, +} + +#[derive(Debug)] +pub(crate) enum InvocationPhase { + Running, + EventPending(Value), + CompletePending(Value), + ErrorPending(InvocationError), + Fused, +} + +/// One active invocation handle borrowing the VM. +/// +/// Polling drives execution. Dropping a handle that has not fused retires its +/// invocation synchronously, including any waiting host operation, so the VM +/// can be reused immediately. +pub struct Invocation<'vm> { + vm: &'vm mut Vm, +} + +impl fmt::Debug for Invocation<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("Invocation").finish_non_exhaustive() + } +} + +impl Invocation<'_> { + /// Polls the invocation stream. + /// + /// Returns `Ready(Some(Ok(Event(value))))` for each emitted event, + /// `Ready(Some(Ok(Complete(value))))` exactly once for the callable return + /// value, `Ready(Some(Err(error)))` exactly once for a typed terminal + /// failure, and `Ready(None)` on every poll after the stream has fused. + /// `Pending` means the VM is paused on an outstanding host operation or + /// host-driven yield; drive it and poll again. + pub fn poll_next(&mut self) -> VmResult { + self.vm.poll_invocation() + } + + /// Cancels the active invocation with a typed reason. + /// + /// Outstanding waiting host operations are cancelled. The next poll + /// produces exactly one `Cancelled(reason)` error item, after which the + /// stream is fused. + pub fn cancel(&mut self, reason: OperationCancelReason) -> VmResult<()> { + let state = self + .vm + .instance + .invocation + .as_mut() + .ok_or(VmError::InvalidFrameState( + "no invocation is active on this vm", + ))?; + if matches!(state.phase, InvocationPhase::Fused) { + return Err(VmError::InvalidFrameState( + "the active invocation has already fused", + )); + } + state.cancel_reason = Some(reason); + self.vm.cancel_waiting_host_op(); + Ok(()) + } +} + +impl Drop for Invocation<'_> { + fn drop(&mut self) { + let active = self + .vm + .instance + .invocation + .as_ref() + .is_some_and(|state| !matches!(state.phase, InvocationPhase::Fused)); + if active { + self.vm.release_invocation(); + } + } +} + +/// One poll step selected from the current invocation phase. +enum InvocationAction { + Cancelled, + Event, + Complete, + Error, + Fused, + Drive, +} + +impl Vm { + /// Starts one invocation of an exported callable with ordinary arguments. + /// + /// The VM must be halted (complete the root frame with [`Vm::run`] first), + /// and must not already have an active invocation. A second invocation on + /// the same VM is rejected while one is active. + pub fn start_invocation( + &mut self, + callable: Value, + args: Vec, + ) -> VmResult> { + if !matches!(callable, Value::Callable(_)) { + return Err(VmError::InvalidCallable); + } + if self + .instance + .invocation + .as_ref() + .is_some_and(|state| !matches!(state.phase, InvocationPhase::Fused)) + { + return Err(VmError::InvalidFrameState( + "an invocation is already active on this vm", + )); + } + let stack_base = self.instance.stack.len(); + let frame_count = self.instance.execution_frames.len(); + self.instance.invocation = Some(InvocationState { + phase: InvocationPhase::Running, + emit_yield_pending: false, + pending_error: None, + cancel_reason: None, + stack_base, + frame_count, + }); + + match self.start_callable(callable, &args) { + Ok(VmStatus::Halted) => { + let result = self + .take_callable_result() + .ok_or(VmError::InvalidFrameState( + "invocation halted without a callable result", + ))?; + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::CompletePending(result); + } + Ok(VmStatus::Yielded) => { + // Either `stream::emit` placed one pending event, or the + // embedding must drive a host-owned yield; both are serviced by + // the next poll. + } + Ok(VmStatus::Waiting(_)) => {} + Err(error) => { + let error = self.map_invocation_error(error); + self.release_invocation(); + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::ErrorPending(error); + } + } + Ok(Invocation { vm: self }) + } + + fn poll_invocation(&mut self) -> VmResult { + loop { + let action = match self.instance.invocation.as_ref() { + Some(state) => { + // Authoritative cancellation supersedes a pending Event or + // Complete: the pending value is discarded (through the + // drop-contract path) and the stream transitions to one + // Cancelled item, then a fused end. + if state.cancel_reason.is_some() + && matches!( + state.phase, + InvocationPhase::EventPending(_) | InvocationPhase::CompletePending(_) + ) + { + InvocationAction::Cancelled + } else { + match state.phase { + InvocationPhase::EventPending(_) => InvocationAction::Event, + InvocationPhase::CompletePending(_) => InvocationAction::Complete, + InvocationPhase::ErrorPending(_) => InvocationAction::Error, + InvocationPhase::Fused => InvocationAction::Fused, + InvocationPhase::Running => InvocationAction::Drive, + } + } + } + None => return Ok(InvocationPoll::Ready(None)), + }; + match action { + InvocationAction::Cancelled => { + let reason = self + .instance + .invocation + .as_ref() + .and_then(|state| state.cancel_reason) + .expect("a cancelled action requires a cancellation reason"); + let discarded = self.replace_invocation_phase(InvocationPhase::ErrorPending( + InvocationError::Cancelled(reason), + )); + match discarded { + InvocationPhase::EventPending(value) + | InvocationPhase::CompletePending(value) => { + self.drop_value_with_contract(value); + } + _ => unreachable!("the cancelled action matched a pending phase above"), + } + } + InvocationAction::Event => { + let value = match self.replace_invocation_phase(InvocationPhase::Running) { + InvocationPhase::EventPending(value) => value, + _ => unreachable!("phase matched above"), + }; + // `emit_yield_pending` stays set until the resumed call + // site re-enters `stream::emit`. + return Ok(InvocationPoll::Ready(Some(Ok(InvocationItem::Event( + value, + ))))); + } + InvocationAction::Complete => { + let value = match self.replace_invocation_phase(InvocationPhase::Fused) { + InvocationPhase::CompletePending(value) => value, + _ => unreachable!("phase matched above"), + }; + self.release_invocation(); + return Ok(InvocationPoll::Ready(Some(Ok(InvocationItem::Complete( + value, + ))))); + } + InvocationAction::Error => { + let error = match self.replace_invocation_phase(InvocationPhase::Fused) { + InvocationPhase::ErrorPending(error) => error, + _ => unreachable!("phase matched above"), + }; + self.release_invocation(); + return Ok(InvocationPoll::Ready(Some(Err(error)))); + } + InvocationAction::Fused => return Ok(InvocationPoll::Ready(None)), + InvocationAction::Drive => { + let result = self.drive_invocation(); + match result { + DriveOutcome::Continue => {} + DriveOutcome::Pending => return Ok(InvocationPoll::Pending), + DriveOutcome::Error(error) => { + self.release_invocation(); + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::ErrorPending(error); + } + } + } + } + } + } + + /// Runs the low-level pump once and folds the outcome into the invocation + /// phase. `Vm::run` itself is unchanged. + fn drive_invocation(&mut self) -> DriveOutcome { + if let Some(reason) = self + .instance + .invocation + .as_ref() + .and_then(|state| state.cancel_reason) + { + return DriveOutcome::Error(InvocationError::Cancelled(reason)); + } + match self.run() { + Ok(VmStatus::Halted) => { + let result = match self.take_callable_result() { + Some(result) => result, + None => { + return DriveOutcome::Error(InvocationError::Vm( + VmError::InvalidFrameState( + "invocation halted without a callable result", + ), + )); + } + }; + self.instance + .invocation + .as_mut() + .expect("invocation state") + .phase = InvocationPhase::CompletePending(result); + DriveOutcome::Continue + } + Ok(VmStatus::Yielded) => match self.last_yield_reason() { + Some(VmYieldReason::Fuel) => DriveOutcome::Error(InvocationError::OutOfFuel { + needed: u64::from(self.run_ctx.fuel_check_interval), + remaining: self.run_ctx.fuel_remaining, + }), + Some(VmYieldReason::Epoch) => { + DriveOutcome::Error(InvocationError::DeadlineReached { + current: self.run_ctx.epoch_handle.current(), + deadline: self.run_ctx.epoch_deadline, + }) + } + _ => { + // A `stream::emit` yield leaves one pending event; any other + // host-driven yield is paused for the embedding. + let event_pending = matches!( + self.instance.invocation.as_ref().map(|state| &state.phase), + Some(InvocationPhase::EventPending(_)) + ); + if event_pending { + DriveOutcome::Continue + } else { + DriveOutcome::Pending + } + } + }, + Ok(VmStatus::Waiting(_)) => { + // Poll the outstanding host operation once with a noop waker. + // The embedding-owned driver completes it; re-polling observes + // readiness. + let waker = Waker::noop(); + let mut cx = Context::from_waker(waker); + match self.poll_waiting_host_op(&mut cx) { + Poll::Ready(Ok(())) => DriveOutcome::Continue, + Poll::Ready(Err(error)) => { + DriveOutcome::Error(self.map_invocation_error(error)) + } + Poll::Pending => DriveOutcome::Pending, + } + } + Err(error) => DriveOutcome::Error(self.map_invocation_error(error)), + } + } + + /// Maps a low-level VM failure to the typed invocation error, preserving + /// structured runtime errors from `stream::emit` validation. + fn map_invocation_error(&mut self, error: VmError) -> InvocationError { + if let Some(state) = self.instance.invocation.as_mut() + && let Some(runtime_error) = state.pending_error.take() + { + return InvocationError::Capability(runtime_error); + } + match error { + VmError::OutOfFuel { needed, remaining } => { + InvocationError::OutOfFuel { needed, remaining } + } + VmError::EpochDeadlineReached { current, deadline } => { + InvocationError::DeadlineReached { current, deadline } + } + VmError::ExecutionScope(scope_error) => InvocationError::Capability(RuntimeError::new( + crate::builtins::runtime::error::RuntimeErrorCode::OperationFailed, + "execution_scope", + scope_error.to_string(), + )), + VmError::HostError(message) => InvocationError::Host { message }, + other => InvocationError::Vm(other), + } + } + + /// Replaces the active invocation phase, returning the previous one so the + /// caller can consume it or drop it (the pending-event drop contract stays + /// with the caller). + fn replace_invocation_phase(&mut self, phase: InvocationPhase) -> InvocationPhase { + std::mem::replace( + &mut self + .instance + .invocation + .as_mut() + .expect("invocation state") + .phase, + phase, + ) + } + + /// Releases the active invocation: cancels outstanding waiting host + /// operations, drops interpreter frames and stack entries introduced by + /// the invocation, and fuses the stream. The per-invocation cancellation + /// reason is cleared by the drop, so it cannot leak into a later + /// invocation started on the same VM. + fn release_invocation(&mut self) { + let (stack_base, frame_count) = self + .instance + .invocation + .as_ref() + .map(|state| (state.stack_base, state.frame_count)) + .unwrap_or((0, 0)); + self.cancel_waiting_host_op(); + self.abort_host_invocation(stack_base, frame_count); + self.instance.drop_invocation_state(); + } + + /// Implements the script-visible `stream::emit(value)` builtin: validates + /// the per-item bound, places one pending event, and yields control to the + /// invocation poller. `stream::emit` still evaluates to `()` inside RSS. + /// + /// When the poller has delivered the event and the VM resumes, the call + /// site re-executes; the second entry consumes the `emit_yield_pending` + /// marker and returns normally instead of emitting a second event. + pub(crate) fn emit_stream_item(&mut self, value: Value) -> VmResult { + let state = self.instance.invocation.as_mut().ok_or_else(|| { + VmError::HostError("stream::emit requires an active invocation".to_string()) + })?; + if !matches!(state.phase, InvocationPhase::Running) { + return Err(VmError::HostError( + "stream::emit is only valid while the invocation is running".to_string(), + )); + } + if state.emit_yield_pending { + state.emit_yield_pending = false; + return Ok(CallOutcome::Return(CallReturn::none())); + } + let limits = self.run_ctx.runtime_context.event_limits(); + match EventPayload::try_new(value, limits) { + Ok(payload) => { + state.phase = InvocationPhase::EventPending(payload.into_value()); + state.emit_yield_pending = true; + Ok(CallOutcome::Yield) + } + Err(runtime_error) => { + let message = runtime_error.to_string(); + state.pending_error = Some(runtime_error); + Err(VmError::HostError(message)) + } + } + } +} + +/// Outcome of one low-level drive step. +enum DriveOutcome { + Continue, + Pending, + Error(InvocationError), +} diff --git a/src/vm/mod.rs b/src/vm/mod.rs index 5f5ea3b3..ff07a305 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -16,6 +16,7 @@ pub mod host_context; pub mod host_extension; mod host_runtime; mod instance; +pub mod invocation; pub(crate) mod jit; mod map_iter; pub(crate) mod native; @@ -52,6 +53,7 @@ pub use self::host_extension::{ }; use self::host_runtime::HostRuntime; use self::instance::{ExecutionFrame, FrameContinuation, Instance, QueuedCallable}; +pub use self::invocation::{Invocation, InvocationError, InvocationItem, InvocationPoll}; pub use self::resource::ResourceCloseReason; use self::run_context::{InterruptMode, RunContext}; pub use self::standard_composition::StandardSurfaceComposition; @@ -2883,6 +2885,7 @@ impl Vm { pub fn shutdown(&mut self) { self.invalidate_callback_registries(); self.cancel_waiting_host_op(); + self.instance.drop_invocation_state(); // Begin execution-scope shutdown (first-reason-wins; sealing the // operation registry) before tearing down interpreter state. let _ = self diff --git a/src/vm/run_context.rs b/src/vm/run_context.rs index b8dcde73..a0e4fdf9 100644 --- a/src/vm/run_context.rs +++ b/src/vm/run_context.rs @@ -7,12 +7,14 @@ //! //! The embedder-facing fuel/epoch APIs live on the VM facade (see //! `crate::vm::fuel` and `crate::vm::epoch`) and delegate here. Cancellation of -//! pending host operations lives in the facade because it crosses into -//! [`HostRuntime`](super::host_runtime::HostRuntime) state. There is no per-run -//! input/event state here by design: this mechanical decomposition only moves -//! budgets and interruption state, and new runtime semantics (input/event -//! scopes, cancellation tokens) are intentionally left out of this commit. +//! pending host operations is per-invocation and lives in the invocation layer +//! (see `crate::vm::invocation`), because it crosses into +//! [`HostRuntime`](super::host_runtime::HostRuntime) waiting-host-op state. The +//! run-scoped invocation stream configuration (per-item event limits) also +//! lives here. There is no per-run input/event buffer state here: the event +//! value is owned by the active invocation's single pending-event slot. +use crate::builtins::runtime::RuntimeContext; use crate::vm::VmError; use crate::vm::VmResult; use crate::vm::epoch::EpochHandle; @@ -42,6 +44,7 @@ impl InterruptMode { /// shared; one facade owns one context. Clone semantics: not `Clone` — a clone /// would duplicate budget state across runs. pub(crate) struct RunContext { + pub(crate) runtime_context: RuntimeContext, pub(crate) interrupt_mode: InterruptMode, pub(crate) fuel_remaining: u64, pub(crate) fuel_check_interval: u32, @@ -62,6 +65,7 @@ impl RunContext { let epoch_handle = EpochHandle::default(); let epoch_counter_ptr = epoch_handle.as_ptr() as usize; Self { + runtime_context: RuntimeContext::default(), interrupt_mode: InterruptMode::None, fuel_remaining: 0, fuel_check_interval: 1, diff --git a/src/vm/tests.rs b/src/vm/tests.rs index e3446e55..0c6ab4e6 100644 --- a/src/vm/tests.rs +++ b/src/vm/tests.rs @@ -2054,3 +2054,107 @@ fn capability_profile_allow_all_and_deny_all_differ() { assert!(!deny_all.allows_host_import("anything::at::all")); assert_ne!(allow_all.fingerprint(), deny_all.fingerprint()); } + +#[test] +fn dropping_cancelled_invocation_consumes_cancellation_at_the_boundary() { + // Dropping an invocation with a pending typed cancellation retires that + // invocation without manufacturing an unobservable terminal item. The + // per-invocation cancellation reason is cleared on fusion, so the VM can + // be reused immediately without inheriting the old reason. + let compiled = crate::compile_source( + r#" + pub fn run() -> int { + 42; + } + "#, + ) + .expect("invocation source should compile"); + let mut vm = Vm::new(compiled.program); + assert_eq!(vm.run().expect("root frame should halt"), VmStatus::Halted); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + let mut invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("invocation should start"); + invocation + .cancel(crate::vm::operation::OperationCancelReason::Requested) + .expect("cancellation should be accepted"); + // Dropping the handle retires the invocation and clears the reason. + } + assert!( + vm.instance.invocation.as_ref().is_none_or(|state| matches!( + state.phase, + crate::vm::invocation::InvocationPhase::Fused + )) && vm + .instance + .invocation + .as_ref() + .is_none_or(|state| state.cancel_reason.is_none()), + "dropping the invocation must fuse it and clear its cancellation reason" + ); + + let mut replacement = vm + .start_invocation(callable, vec![]) + .expect("the vm should be reusable after the dropped invocation"); + assert!(matches!( + replacement.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); +} + +#[test] +fn cancelled_invocation_delivers_one_typed_error_then_fused_end() { + // Functional contract of the cancellation path: exactly one typed + // Cancelled item, a fused end, and the cancellation consumed at the + // invocation boundary (a later invocation runs normally). + let compiled = crate::compile_source( + r#" + pub fn run() -> int { + 42; + } + "#, + ) + .expect("invocation source should compile"); + let mut vm = Vm::new(compiled.program); + assert_eq!(vm.run().expect("root frame should halt"), VmStatus::Halted); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + let mut invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("invocation should start"); + invocation + .cancel(crate::vm::operation::OperationCancelReason::Deadline) + .expect("cancellation should be accepted"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + crate::vm::operation::OperationCancelReason::Deadline, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + } + + // A new invocation on the same VM must run to completion instead of + // being cancelled on arrival. + let mut second = vm + .start_invocation(callable, vec![]) + .expect("a new invocation may start after fusion"); + match second.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) => {} + other => panic!("the second invocation must complete normally, got {other:?}"), + } + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} diff --git a/tests/compiler/compiler_rustscript_tests.rs b/tests/compiler/compiler_rustscript_tests.rs index b7fa4afc..8641db26 100644 --- a/tests/compiler/compiler_rustscript_tests.rs +++ b/tests/compiler/compiler_rustscript_tests.rs @@ -3809,3 +3809,25 @@ fn rustscript_generic_schema_errors_are_reported() { run_source_error_cases(&cases); } + +#[test] +fn rustscript_strict_stream_emit_accepts_any_payload() { + // In strict RustScript, `stream::emit` is the one host function whose + // `any` payload is accepted at compile time; the per-item event bound is + // validated at runtime by the invocation stream. The exemption is tied to + // the authoritative runtime builtin identity (see the compiler unit test + // `stream_emit_any_payload_exemption_requires_authoritative_builtin_identity`), + // so a same-name function registered through another catalog cannot + // inherit it. + compile_source( + r#" + use stream; + pub fn run() -> int { + stream::emit({"a": 1, "b": 2}); + stream::emit("text"); + 42; + } + "#, + ) + .expect("strict stream::emit with any payloads must compile"); +} diff --git a/tests/invocation_stream_tests.rs b/tests/invocation_stream_tests.rs new file mode 100644 index 00000000..cc6df966 --- /dev/null +++ b/tests/invocation_stream_tests.rs @@ -0,0 +1,870 @@ +#![cfg(feature = "runtime")] + +//! Invocation item stream contract tests. +//! +//! An invocation behaves like `Stream>`: +//! zero or more `Event` items, then exactly one `Complete` item or one typed error, +//! then a fused end of stream. Input enters through ordinary callable arguments and +//! polling drives execution (backpressure). + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use vm::{ + HostFunctionRegistry, InvocationError, InvocationItem, InvocationPoll, Value, Vm, VmError, + compile_source, operation::OperationCancelReason, +}; + +/// Compiles a source, binds the default runtime host registry, and completes the +/// root frame so exported callables can be started. +fn compiled_vm(source: &str) -> Vm { + let program = compile_source(source) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + HostFunctionRegistry::new() + .bind_vm_cached(&mut vm) + .expect("default runtime host registry should bind"); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + vm +} + +/// Drives one exported `run` callable to the end of its invocation stream. +fn collect_items(vm: &mut Vm, args: Vec) -> Vec> { + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, args) + .expect("invocation should start"); + let mut items = Vec::new(); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + assert!( + Instant::now() < deadline, + "invocation drive loop must terminate" + ); + match invocation + .poll_next() + .expect("invocation poll should not fail") + { + InvocationPoll::Ready(Some(item)) => items.push(item), + InvocationPoll::Ready(None) => break, + InvocationPoll::Pending => std::thread::sleep(Duration::from_millis(1)), + } + } + items +} + +#[test] +fn invocation_input_arrives_as_ordinary_callable_arguments() { + let mut vm = compiled_vm( + r#" + pub fn run(input: map) -> map { + input; + } + "#, + ); + let input = Value::map(vec![(Value::string("kind"), Value::string("message"))]); + let items = collect_items(&mut vm, vec![input.clone()]); + assert_eq!(items.len(), 1, "expected exactly one stream item"); + assert!( + matches!(&items[0], Ok(InvocationItem::Complete(value)) if *value == input), + "the exact structured argument must be the callable input, got {:?}", + items + ); +} + +#[test] +fn invocation_without_events_yields_complete_then_fused_end() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + assert!( + matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + ), + "the stream must stay fused after Complete" + ); + drop(invocation); + + // Once the first invocation has fused, a new invocation may start on the + // same VM. + let mut second = vm + .start_invocation(callable, vec![]) + .expect("a new invocation may start after fusion"); + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn dropping_an_unpolled_invocation_allows_a_second_invocation() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + { + // Dropping the handle retires even a CompletePending invocation whose + // terminal item was never observed. + let _invocation = vm + .start_invocation(callable.clone(), vec![]) + .expect("first invocation should start"); + } + let mut second = vm + .start_invocation(callable, vec![]) + .expect("dropping the first handle must release the vm immediately"); + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) + )); +} + +#[test] +fn invocation_failures_are_typed_items_without_stack_or_string_inspection() { + let mut vm = compiled_vm( + r#" + pub fn run(input: int) -> int { + 100 / input; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::Int(0)]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Vm(VmError::DivisionByZero)))) => {} + other => panic!("expected a typed division-by-zero item, got {other:?}"), + } + assert!( + matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + ), + "the stream must fuse after the error item" + ); +} + +/// Records one script-visible progress note per call. +struct ProgressNote(Arc>>); + +impl vm::HostArgsFunction for ProgressNote { + fn call(&mut self, args: &[Value]) -> vm::VmResult { + if let Some(value) = args.first() { + self.0 + .lock() + .expect("progress note lock should not be poisoned") + .push(value.clone()); + } + Ok(vm::CallOutcome::Return(vm::CallReturn::one( + args.first().cloned().unwrap_or(Value::Null), + ))) + } +} + +#[test] +fn invocation_emits_events_then_complete_in_order() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("first"); + stream::emit("second"); + "done"; + } + "#, + ); + let items = collect_items(&mut vm, vec![]); + assert_eq!( + items.len(), + 3, + "expected event, event, complete; got {items:?}" + ); + assert!( + matches!(&items[0], Ok(InvocationItem::Event(value)) if *value == Value::string("first")) + ); + assert!( + matches!(&items[1], Ok(InvocationItem::Event(value)) if *value == Value::string("second")) + ); + assert!( + matches!(&items[2], Ok(InvocationItem::Complete(value)) if *value == Value::string("done")) + ); +} + +#[test] +fn invocation_event_values_never_replace_the_callable_return_value() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> int { + stream::emit("payload"); + 42; + } + "#, + ); + let items = collect_items(&mut vm, vec![]); + assert_eq!( + items.len(), + 2, + "expected event then complete; got {items:?}" + ); + assert!( + matches!(&items[0], Ok(InvocationItem::Event(value)) if *value == Value::string("payload")) + ); + assert!(matches!( + &items[1], + Ok(InvocationItem::Complete(Value::Int(42))) + )); +} + +#[test] +fn invocation_polling_pauses_execution_and_exposes_one_event_at_a_time() { + let program = compile_source( + r#" + use stream; + fn note_progress(value: string) -> string; + pub fn run() -> string { + stream::emit("a"); + note_progress("after-a"); + stream::emit("b"); + note_progress("after-b"); + "done"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + let notes = Arc::new(Mutex::new(Vec::::new())); + vm.bind_args_function("note_progress", Box::new(ProgressNote(Arc::clone(¬es)))); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + // First poll: the script paused at the first emit; nothing after it ran. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + assert!( + notes.lock().expect("notes lock").is_empty(), + "execution must not advance while polling is paused" + ); + + // Second poll: resume past emit(a), run note_progress("after-a"), pause at + // emit(b). Exactly one progress note may exist. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("b") + )); + assert_eq!( + notes.lock().expect("notes lock").len(), + 1, + "exactly one progress note between polls" + ); + + // Third poll: resume past emit(b), run note_progress("after-b"), complete. + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) if value == Value::string("done") + )); + assert_eq!(notes.lock().expect("notes lock").len(), 2); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancellation_produces_one_typed_error_item_then_fused_end() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("before"); + while true { + 1; + } + "unreachable"; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("before") + )); + + invocation + .cancel(OperationCancelReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + OperationCancelReason::Requested, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_fuel_exhaustion_produces_one_typed_error_item() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + while true { + 1; + } + 42; + } + "#, + ); + vm.set_fuel(8); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::OutOfFuel { + needed: _, + remaining: 0, + }))) => {} + other => panic!("expected a typed out-of-fuel item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_deadline_expiry_produces_one_typed_error_item() { + let mut vm = compiled_vm( + r#" + pub fn run() -> int { + 42; + } + "#, + ); + vm.set_epoch_deadline(0) + .expect("epoch deadline should be configured"); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::DeadlineReached { + current: 0, + deadline: 0, + }))) => {} + other => panic!("expected a typed deadline item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_host_failure_produces_one_typed_error_item() { + let program = compile_source( + r#" + fn fail_host() -> int; + pub fn run() -> int { + fail_host(); + 42; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("fail_host", Box::new(FailingHost)); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Host { message }))) => { + assert_eq!(message, "boom"); + } + other => panic!("expected a typed host failure item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_event_bound_violations_are_typed_capability_errors() { + let mut vm = compiled_vm( + r#" + use stream; + pub fn run(input: string) -> int { + stream::emit(input); + 42; + } + "#, + ); + let oversized = "x".repeat(70 * 1024); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![Value::string(oversized)]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Capability(error)))) => { + assert_eq!(error.code(), vm::RuntimeErrorCode::EventPayloadTooLarge); + } + other => panic!("expected a typed capability error item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +/// Fails every host call with a plain embedding error. +struct FailingHost; + +impl vm::HostStackFunction for FailingHost { + fn call(&mut self, _vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + Err(vm::VmError::HostError("boom".to_string())) + } +} + +#[cfg(feature = "async")] +#[path = "support/async_test_bridge.rs"] +mod async_test_bridge; + +/// Waits asynchronously through the embedding-owned host bridge. +#[cfg(feature = "async")] +struct AsyncWaitHost; + +#[cfg(feature = "async")] +impl vm::HostStackFunction for AsyncWaitHost { + fn call(&mut self, vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + vm.submit_host_future(Box::pin(async move { + tokio::time::sleep(Duration::from_millis(20)).await; + Ok(vm::HostFutureOutput::returning(vm::CallReturn::one( + Value::Int(7), + ))) + })) + } +} + +#[cfg(feature = "async")] +#[test] +fn invocation_waiting_host_operation_returns_pending_and_preserves_item_order() { + let program = compile_source( + r#" + use stream; + fn wait_host() -> int; + pub fn run() -> string { + stream::emit("a"); + wait_host(); + stream::emit("b"); + "done"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("wait_host", Box::new(AsyncWaitHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + + // The outstanding host operation maps to Pending; drive it and poll again. + let deadline = Instant::now() + Duration::from_secs(10); + let mut polled_pending = false; + let next = loop { + assert!( + Instant::now() < deadline, + "waiting invocation must resume through the host driver" + ); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Pending => { + polled_pending = true; + std::thread::sleep(Duration::from_millis(1)); + } + ready => break ready, + } + }; + assert!( + polled_pending, + "the waiting host op must surface as Pending" + ); + assert!(matches!( + next, + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("b") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(value)))) if value == Value::string("done") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancellation_is_consumed_at_the_invocation_boundary() { + // Regression: after a cancelled invocation emits its typed error and + // fuses, the VM-level cancellation reason must not leak into a later + // invocation started on the same VM. + let mut vm = compiled_vm( + r#" + use stream; + pub fn run() -> string { + stream::emit("before"); + while true { + 1; + } + "unreachable"; + } + pub fn plain() -> int { + 42; + } + "#, + ); + let cancellable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(cancellable, vec![]) + .expect("invocation should start"); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("before") + )); + + invocation + .cancel(OperationCancelReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + OperationCancelReason::Requested, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + drop(invocation); + + // A fresh invocation on the same VM must not inherit the old reason: it + // runs to completion instead of being cancelled on arrival. + let plain = vm + .resolve_exported_callable("plain") + .expect("exported plain callable should resolve"); + let mut second = vm + .start_invocation(plain, vec![]) + .expect("a new invocation may start after fusion"); + match second.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Ok(InvocationItem::Complete(Value::Int(42))))) => {} + other => panic!("the second invocation must complete normally, got {other:?}"), + } + assert!(matches!( + second.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[test] +fn invocation_cancel_during_event_pending_discards_the_pending_event() { + // Cancellation is authoritative: a pending event that was placed but not + // yet delivered must be discarded (through the drop-contract path) and + // the stream must produce exactly one Cancelled item, then a fused end. + let program = compile_source( + r#" + use stream; + pub fn run() -> string { + stream::emit({"a": 1, "b": 2}); + while true { + 1; + } + "unreachable"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.set_drop_contract_events_enabled(true); + HostFunctionRegistry::new() + .bind_vm_cached(&mut vm) + .expect("default runtime host registry should bind"); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let drops_before_cancel = vm.drop_contract_event_count(); + // `start_callable` runs to the first `stream::emit` yield, so the + // invocation is already in EventPending with the map payload. + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + invocation + .cancel(OperationCancelReason::Requested) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + OperationCancelReason::Requested, + )))) => {} + other => panic!("cancellation must supersede the pending event, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); + drop(invocation); + + // The discarded event payload (map plus its two key/value pairs) must be + // dropped through the VM drop-contract path, not leaked. + assert!( + vm.drop_contract_event_count() >= drops_before_cancel + 5, + "the discarded pending event payload must be dropped through the drop contract path" + ); +} + +#[test] +fn invocation_cancel_during_complete_pending_discards_the_pending_complete() { + // Cancellation is authoritative over a not-yet-delivered Complete item: + // the callable result is discarded and the stream produces exactly one + // Cancelled item, then a fused end. + let mut vm = compiled_vm( + r#" + pub fn run() -> map { + {"a": 1, "b": 2}; + } + "#, + ); + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + // The callable completes during `start_callable`, so the invocation is + // already in CompletePending with the return map. + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + invocation + .cancel(OperationCancelReason::Deadline) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + OperationCancelReason::Deadline, + )))) => {} + other => panic!("cancellation must supersede the pending complete, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +/// Fails asynchronously on the first poll of its submitted host operation. +#[cfg(feature = "async")] +struct AsyncFailHost; + +#[cfg(feature = "async")] +impl vm::HostStackFunction for AsyncFailHost { + fn call(&mut self, vm: &mut Vm, _args: &[Value]) -> vm::VmResult { + vm.submit_host_future(Box::pin(async move { + Err(vm::VmError::HostError("bridge future failed".to_string())) + })) + } +} + +#[cfg(feature = "async")] +#[test] +fn invocation_host_op_first_poll_failure_keeps_typed_host_error() { + // Regression: the waiting host op is polled once with a noop waker; if the + // first poll fails and clears the waiting state, the typed mapping must + // still surface (here a `Host` error) on the invocation stream. + let program = compile_source( + r#" + fn fail_host() -> int; + pub fn run() -> int { + fail_host(); + 42; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("fail_host", Box::new(AsyncFailHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Host { message }))) => { + assert_eq!(message, "bridge future failed"); + } + other => panic!("expected a typed host error item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +} + +#[cfg(feature = "async")] +#[test] +fn invocation_cancellation_while_waiting_produces_one_typed_error_item() { + let program = compile_source( + r#" + use stream; + fn wait_host() -> int; + pub fn run() -> string { + stream::emit("a"); + wait_host(); + "unreachable"; + } + "#, + ) + .expect("invocation source should compile") + .program; + let mut vm = Vm::new(program); + vm.bind_stack_function("wait_host", Box::new(AsyncWaitHost)); + async_test_bridge::install(&mut vm); + assert_eq!( + vm.run().expect("root frame should halt"), + vm::VmStatus::Halted + ); + + let callable = vm + .resolve_exported_callable("run") + .expect("exported run callable should resolve"); + let mut invocation = vm + .start_invocation(callable, vec![]) + .expect("invocation should start"); + + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(Some(Ok(InvocationItem::Event(value)))) if value == Value::string("a") + )); + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Pending + )); + + invocation + .cancel(OperationCancelReason::Deadline) + .expect("cancellation should be accepted"); + match invocation.poll_next().expect("poll should succeed") { + InvocationPoll::Ready(Some(Err(InvocationError::Cancelled( + OperationCancelReason::Deadline, + )))) => {} + other => panic!("expected a typed cancellation item, got {other:?}"), + } + assert!(matches!( + invocation.poll_next().expect("poll should succeed"), + InvocationPoll::Ready(None) + )); +}