From 572fed47b411a3f76a51632addaefa89082b8a6b Mon Sep 17 00:00:00 2001 From: Maofeng Date: Thu, 30 Jul 2026 07:40:08 +0800 Subject: [PATCH] feat(runtime): supervise resident component resources Detect the resident actor export, route Host events through the bounded Actor mailbox, enforce fuel, deadlines, payload limits, and translate WIT effects into typed runtime values. Add a revision-scoped ResidentSession with deny-by-default TCP, UDP, and Unix endpoint policies, non-reused resource IDs, stream lifecycle and backpressure state, atomic effect validation, timers, subscriptions, extension sources, and shutdown handling. Cover passive rejection, real Component ABI round trips, resource ownership, endpoint policy, and driver operation validation with integration tests. --- crates/wasmeld-runtime/src/bindings.rs | 8 + crates/wasmeld-runtime/src/error.rs | 12 + crates/wasmeld-runtime/src/lib.rs | 11 + crates/wasmeld-runtime/src/manifest.rs | 8 +- crates/wasmeld-runtime/src/resident.rs | 317 +++++ crates/wasmeld-runtime/src/resident_host.rs | 1057 +++++++++++++++++ crates/wasmeld-runtime/src/runtime.rs | 309 ++++- .../tests/runtime_components.rs | 268 ++++- 8 files changed, 1983 insertions(+), 7 deletions(-) create mode 100644 crates/wasmeld-runtime/src/resident.rs create mode 100644 crates/wasmeld-runtime/src/resident_host.rs diff --git a/crates/wasmeld-runtime/src/bindings.rs b/crates/wasmeld-runtime/src/bindings.rs index 82d2a93..a12faca 100644 --- a/crates/wasmeld-runtime/src/bindings.rs +++ b/crates/wasmeld-runtime/src/bindings.rs @@ -5,3 +5,11 @@ wasmtime::component::bindgen!({ path: "../../wit/service", world: "service-component", }); + +/// Optional exports implemented only by active resident Components. +pub(crate) mod resident { + wasmtime::component::bindgen!({ + path: "../../wit/resident", + world: "resident-component", + }); +} diff --git a/crates/wasmeld-runtime/src/error.rs b/crates/wasmeld-runtime/src/error.rs index 1a56e12..832d83b 100644 --- a/crates/wasmeld-runtime/src/error.rs +++ b/crates/wasmeld-runtime/src/error.rs @@ -68,6 +68,18 @@ pub enum RuntimeError { #[error("actor for {0} stopped before returning a result")] ActorStopped(ServiceKey), + #[error("service revision {0} does not export the resident actor interface")] + NotResident(ServiceKey), + + #[error("resident event for {service} exceeds the {limit}-byte payload limit")] + ResidentEventTooLarge { service: ServiceKey, limit: usize }, + + #[error("resident event for {service} returned more than {limit} effects")] + TooManyResidentEffects { service: ServiceKey, limit: usize }, + + #[error("invalid resident effect: {0}")] + InvalidResidentEffect(String), + #[error("input for {service} exceeds the {limit}-byte limit")] InputTooLarge { service: ServiceKey, limit: usize }, diff --git a/crates/wasmeld-runtime/src/lib.rs b/crates/wasmeld-runtime/src/lib.rs index 448494a..94d796d 100644 --- a/crates/wasmeld-runtime/src/lib.rs +++ b/crates/wasmeld-runtime/src/lib.rs @@ -9,6 +9,8 @@ mod bindings; mod capability; mod error; mod manifest; +mod resident; +mod resident_host; mod runtime; pub use capability::{ @@ -16,5 +18,14 @@ pub use capability::{ }; pub use error::RuntimeError; pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest}; +pub use resident::{ + ComponentExecution, ResidentEffect, ResidentEvent, ResidentLimits, ResourceId, + StreamCloseReason, +}; +pub use resident_host::{ + NetworkScope, ResidentEndpoint, ResidentHostError, ResidentOperation, ResidentPolicy, + ResidentResourceInfo, ResidentResourceKind, ResidentResourceMetadata, ResidentResourceState, + ResidentSession, +}; pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats}; pub use wasmeld_package::SERVICE_WORLD; diff --git a/crates/wasmeld-runtime/src/manifest.rs b/crates/wasmeld-runtime/src/manifest.rs index 0719a59..04119e8 100644 --- a/crates/wasmeld-runtime/src/manifest.rs +++ b/crates/wasmeld-runtime/src/manifest.rs @@ -4,7 +4,7 @@ use std::{fmt, time::Duration}; use serde::{Deserialize, Serialize}; -use crate::error::RuntimeError; +use crate::{error::RuntimeError, resident::ResidentLimits}; /// Immutable identity of one service revision. #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] @@ -103,6 +103,9 @@ pub struct ResourceLimits { pub max_input_bytes: usize, /// Maximum response payload size. pub max_output_bytes: usize, + /// Limits for active events and Host-owned resources. + #[serde(default)] + pub resident: ResidentLimits, } impl Default for ResourceLimits { @@ -114,6 +117,7 @@ impl Default for ResourceLimits { mailbox_capacity: 16, max_input_bytes: 16 * 1024, max_output_bytes: 16 * 1024, + resident: ResidentLimits::default(), } } } @@ -151,7 +155,7 @@ impl ResourceLimits { )); } - Ok(()) + self.resident.validate() } /// Returns the configured call deadline. diff --git a/crates/wasmeld-runtime/src/resident.rs b/crates/wasmeld-runtime/src/resident.rs new file mode 100644 index 0000000..7143ff7 --- /dev/null +++ b/crates/wasmeld-runtime/src/resident.rs @@ -0,0 +1,317 @@ +//! Host-owned event and effect protocol for active resident Components. +//! +//! A [`ResidentEvent`] is always produced by a Host driver and delivered +//! through the same bounded Actor mailbox as ordinary service invocations. +//! The Component never receives an operating-system descriptor. It can only +//! reference opaque [`ResourceId`] values and return [`ResidentEffect`] values +//! for the owning driver to validate and apply. + +use serde::{Deserialize, Serialize}; + +use crate::{RuntimeError, bindings::resident::exports::wasmeld::resident::actor as wit}; + +pub(crate) const RESIDENT_INTERFACE: &str = "wasmeld:resident/actor@0.1.0"; + +/// Component execution model discovered from its exported WIT interfaces. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ComponentExecution { + /// Retains memory but runs only for `invoke` mailbox commands. + Service, + /// Also accepts Host events and returns effects for Host-owned resources. + Resident, +} + +/// Opaque Host resource identity scoped to one resident Actor revision. +/// +/// IDs are never reused while the Actor is alive. `0` is reserved so an +/// uninitialized ID cannot accidentally address a real resource. +#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)] +#[serde(transparent)] +pub struct ResourceId(u64); + +impl ResourceId { + /// Creates a valid opaque resource identity. + pub fn new(value: u64) -> Option { + (value != 0).then_some(Self(value)) + } + + /// Returns the WIT-compatible integer representation. + pub const fn get(self) -> u64 { + self.0 + } +} + +/// Why a Host-owned stream stopped accepting events and effects. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum StreamCloseReason { + PeerClosed, + HostClosed, + IdleTimeout, + ProtocolError, + TransportError, +} + +/// One bounded event delivered to a resident Component. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResidentEvent { + Timer { + timer_id: ResourceId, + scheduled_at_ns: u64, + }, + StreamOpened { + endpoint_id: ResourceId, + stream_id: ResourceId, + peer: Option, + }, + StreamData { + stream_id: ResourceId, + bytes: Vec, + }, + StreamWritable { + stream_id: ResourceId, + }, + StreamHalfClosed { + stream_id: ResourceId, + }, + StreamClosed { + stream_id: ResourceId, + reason: StreamCloseReason, + }, + Datagram { + endpoint_id: ResourceId, + peer: String, + bytes: Vec, + }, + Message { + subscription_id: ResourceId, + message_id: u64, + bytes: Vec, + }, + Source { + source_id: ResourceId, + event: String, + payload: Vec, + }, + Shutdown, +} + +/// A requested operation returned by a resident Component. +/// +/// Effects are inert until the Host supervisor checks resource ownership, +/// payload limits, current stream state, and driver policy. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResidentEffect { + WriteStream { + stream_id: ResourceId, + bytes: Vec, + }, + CloseStream { + stream_id: ResourceId, + }, + PauseStream { + stream_id: ResourceId, + }, + ResumeStream { + stream_id: ResourceId, + }, + SendDatagram { + endpoint_id: ResourceId, + peer: String, + bytes: Vec, + }, + ArmTimer { + timer_id: ResourceId, + delay_ms: u64, + interval_ms: Option, + }, + CancelTimer { + timer_id: ResourceId, + }, + AcknowledgeMessage { + subscription_id: ResourceId, + message_id: u64, + }, + SourceCommand { + source_id: ResourceId, + command: String, + payload: Vec, + }, +} + +/// Execution limits specific to active resident Components. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ResidentLimits { + /// Maximum number of Host resources owned by one Actor revision. + pub max_resources: usize, + /// Maximum effects accepted from one event handler call. + pub max_effects_per_event: usize, + /// Maximum bytes in one stream read or write. + pub max_stream_chunk_bytes: usize, + /// Maximum bytes in one received or sent datagram. + pub max_datagram_bytes: usize, + /// Minimum delay accepted for Component-created timers. + pub min_timer_delay_ms: u64, +} + +impl Default for ResidentLimits { + fn default() -> Self { + Self { + max_resources: 1024, + max_effects_per_event: 64, + max_stream_chunk_bytes: 64 * 1024, + max_datagram_bytes: 64 * 1024, + min_timer_delay_ms: 1, + } + } +} + +impl ResidentLimits { + pub(crate) fn validate(&self) -> Result<(), RuntimeError> { + if self.max_resources == 0 { + return Err(invalid_limit("max_resources")); + } + if self.max_effects_per_event == 0 { + return Err(invalid_limit("max_effects_per_event")); + } + if self.max_stream_chunk_bytes == 0 { + return Err(invalid_limit("max_stream_chunk_bytes")); + } + if self.max_datagram_bytes == 0 { + return Err(invalid_limit("max_datagram_bytes")); + } + if self.min_timer_delay_ms == 0 { + return Err(invalid_limit("min_timer_delay_ms")); + } + Ok(()) + } +} + +pub(crate) fn into_wit_event(event: ResidentEvent) -> wit::Event { + match event { + ResidentEvent::Timer { + timer_id, + scheduled_at_ns, + } => wit::Event::Timer(wit::TimerFired { + timer_id: timer_id.get(), + scheduled_at_ns, + }), + ResidentEvent::StreamOpened { + endpoint_id, + stream_id, + peer, + } => wit::Event::StreamOpened(wit::StreamOpened { + endpoint_id: endpoint_id.get(), + stream_id: stream_id.get(), + peer, + }), + ResidentEvent::StreamData { stream_id, bytes } => { + wit::Event::StreamData(wit::StreamChunk { + stream_id: stream_id.get(), + bytes, + }) + } + ResidentEvent::StreamWritable { stream_id } => wit::Event::StreamWritable(stream_id.get()), + ResidentEvent::StreamHalfClosed { stream_id } => { + wit::Event::StreamHalfClosed(stream_id.get()) + } + ResidentEvent::StreamClosed { stream_id, reason } => { + wit::Event::StreamClosed(wit::StreamClosed { + stream_id: stream_id.get(), + reason: match reason { + StreamCloseReason::PeerClosed => wit::CloseReason::PeerClosed, + StreamCloseReason::HostClosed => wit::CloseReason::HostClosed, + StreamCloseReason::IdleTimeout => wit::CloseReason::IdleTimeout, + StreamCloseReason::ProtocolError => wit::CloseReason::ProtocolError, + StreamCloseReason::TransportError => wit::CloseReason::TransportError, + }, + }) + } + ResidentEvent::Datagram { + endpoint_id, + peer, + bytes, + } => wit::Event::Datagram(wit::Datagram { + endpoint_id: endpoint_id.get(), + peer, + bytes, + }), + ResidentEvent::Message { + subscription_id, + message_id, + bytes, + } => wit::Event::Message(wit::Message { + subscription_id: subscription_id.get(), + message_id, + bytes, + }), + ResidentEvent::Source { + source_id, + event, + payload, + } => wit::Event::Source(wit::SourceEvent { + source_id: source_id.get(), + kind: event, + payload, + }), + ResidentEvent::Shutdown => wit::Event::Shutdown, + } +} + +pub(crate) fn from_wit_effects( + effects: Vec, +) -> Result, RuntimeError> { + effects.into_iter().map(from_wit_effect).collect() +} + +fn from_wit_effect(effect: wit::Effect) -> Result { + Ok(match effect { + wit::Effect::WriteStream(effect) => ResidentEffect::WriteStream { + stream_id: resource_id(effect.stream_id)?, + bytes: effect.bytes, + }, + wit::Effect::CloseStream(stream_id) => ResidentEffect::CloseStream { + stream_id: resource_id(stream_id)?, + }, + wit::Effect::PauseStream(stream_id) => ResidentEffect::PauseStream { + stream_id: resource_id(stream_id)?, + }, + wit::Effect::ResumeStream(stream_id) => ResidentEffect::ResumeStream { + stream_id: resource_id(stream_id)?, + }, + wit::Effect::SendDatagram(effect) => ResidentEffect::SendDatagram { + endpoint_id: resource_id(effect.endpoint_id)?, + peer: effect.peer, + bytes: effect.bytes, + }, + wit::Effect::ArmTimer(effect) => ResidentEffect::ArmTimer { + timer_id: resource_id(effect.timer_id)?, + delay_ms: effect.delay_ms, + interval_ms: effect.interval_ms, + }, + wit::Effect::CancelTimer(timer_id) => ResidentEffect::CancelTimer { + timer_id: resource_id(timer_id)?, + }, + wit::Effect::AcknowledgeMessage(effect) => ResidentEffect::AcknowledgeMessage { + subscription_id: resource_id(effect.subscription_id)?, + message_id: effect.message_id, + }, + wit::Effect::SourceCommand(effect) => ResidentEffect::SourceCommand { + source_id: resource_id(effect.source_id)?, + command: effect.command, + payload: effect.payload, + }, + }) +} + +fn resource_id(value: u64) -> Result { + ResourceId::new(value) + .ok_or_else(|| RuntimeError::InvalidResidentEffect("resource ID 0 is reserved".to_owned())) +} + +fn invalid_limit(field: &str) -> RuntimeError { + RuntimeError::InvalidManifest(format!("limits.resident.{field} must be greater than zero")) +} diff --git a/crates/wasmeld-runtime/src/resident_host.rs b/crates/wasmeld-runtime/src/resident_host.rs new file mode 100644 index 0000000..41923f2 --- /dev/null +++ b/crates/wasmeld-runtime/src/resident_host.rs @@ -0,0 +1,1057 @@ +//! Host-side ownership and state machine for resident Component resources. +//! +//! # Architecture +//! +//! A resident Component is not a native daemon and never owns a socket, file +//! descriptor, timer task, broker consumer, or async runtime. Those objects +//! belong to a Host driver. The driver translates external activity into +//! [`ResidentEvent`](crate::ResidentEvent) values and applies the validated +//! [`ResidentOperation`] values returned by this module. +//! +//! ```text +//! TCP / UDP / Unix / timer / broker / custom driver +//! | +//! v +//! ResidentSession +//! resource + state checks +//! | +//! v +//! bounded Actor mailbox +//! | +//! v +//! Wasm Component +//! | +//! raw effects +//! | +//! v +//! ResidentSession +//! ownership + policy checks +//! | +//! v +//! driver operations +//! ``` +//! +//! One session is pinned to one [`ServiceKey`]. It must not be reused when a +//! deployment switches revisions. The deployment manager should start a new +//! Actor and session, attach new drivers, drain the old session, deliver +//! `shutdown`, close its Host resources, and finally stop the old Actor. +//! +//! Driver event loops may run concurrently, but calls into a session must be +//! serialized by their owner (normally one supervisor task). The Actor mailbox +//! is the final serialization boundary for Component memory. + +use std::{ + collections::HashMap, + fs, + net::SocketAddr, + path::{Component as PathComponent, Path, PathBuf}, +}; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::{ + ActorHandle, ComponentExecution, ResidentEffect, ResidentEvent, ResourceId, RuntimeError, + ServiceKey, StreamCloseReason, +}; + +/// Network exposure allowed for one transport direction. +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum NetworkScope { + /// No endpoint of this class may be registered. + #[default] + Disabled, + /// Only numeric loopback addresses are accepted. + Loopback, + /// Any numeric IP address is accepted. + Any, +} + +impl NetworkScope { + fn allows(self, address: SocketAddr) -> bool { + match self { + Self::Disabled => false, + Self::Loopback => address.ip().is_loopback(), + Self::Any => true, + } + } +} + +/// Platform policy applied before a Host driver binds an endpoint. +/// +/// The default is deny-all. Unix roots must already exist so their canonical +/// paths can be checked before a socket is created. A Unix driver must perform +/// the same check again immediately before bind to prevent path replacement +/// between validation and use. +#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +pub struct ResidentPolicy { + pub tcp_listen: NetworkScope, + pub udp_bind: NetworkScope, + pub unix_listen_roots: Vec, +} + +/// A management-defined endpoint. Components cannot create listeners. +/// +/// Using `SocketAddr` instead of a hostname avoids DNS changes bypassing the +/// configured network scope. Protocols such as HTTP, WebSocket, gRPC, TLS, and +/// framed streams are implemented by Host drivers above these transports. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "transport", rename_all = "snake_case")] +pub enum ResidentEndpoint { + Tcp { name: String, bind: SocketAddr }, + Udp { name: String, bind: SocketAddr }, + Unix { name: String, path: PathBuf }, +} + +impl ResidentEndpoint { + fn name(&self) -> &str { + match self { + Self::Tcp { name, .. } | Self::Udp { name, .. } | Self::Unix { name, .. } => name, + } + } +} + +/// Stable category exposed in management resource snapshots. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ResidentResourceKind { + TcpEndpoint, + UdpEndpoint, + UnixEndpoint, + Stream, + Timer, + MessageSubscription, + Source, +} + +/// Observable lifecycle state without exposing driver handles. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ResidentResourceState { + Ready, + Open, + ReadPaused, + PeerHalfClosed, + Closing, +} + +/// Management-safe snapshot of one Host-owned resource. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ResidentResourceInfo { + pub id: ResourceId, + pub name: String, + pub kind: ResidentResourceKind, + pub state: ResidentResourceState, + pub metadata: ResidentResourceMetadata, +} + +/// Kind-specific metadata required to correlate a resource with its driver. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResidentResourceMetadata { + Endpoint { endpoint: ResidentEndpoint }, + Stream { endpoint_id: ResourceId }, + Timer { armed: bool }, + MessageSubscription, + Source { source_kind: String }, +} + +/// An operation that passed payload, ownership, kind, and state validation. +/// +/// This is the only output a network or source driver should execute. Raw +/// [`ResidentEffect`] values are intentionally kept inside [`ResidentSession`]. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ResidentOperation { + WriteStream { + stream_id: ResourceId, + bytes: Vec, + }, + CloseStream { + stream_id: ResourceId, + }, + PauseStream { + stream_id: ResourceId, + }, + ResumeStream { + stream_id: ResourceId, + }, + SendDatagram { + endpoint_id: ResourceId, + peer: String, + bytes: Vec, + }, + ArmTimer { + timer_id: ResourceId, + delay_ms: u64, + interval_ms: Option, + }, + CancelTimer { + timer_id: ResourceId, + }, + AcknowledgeMessage { + subscription_id: ResourceId, + message_id: u64, + }, + SourceCommand { + source_id: ResourceId, + command: String, + payload: Vec, + }, +} + +/// Host resource, policy, or Actor failure while processing resident work. +#[derive(Debug, Error)] +pub enum ResidentHostError { + #[error(transparent)] + Runtime(#[from] RuntimeError), + + #[error("resident session for {0} is shutting down")] + ShuttingDown(ServiceKey), + + #[error("resident resource limit for {service} is {limit}")] + ResourceLimit { service: ServiceKey, limit: usize }, + + #[error("resident resource ids exhausted for {0}")] + ResourceIdsExhausted(ServiceKey), + + #[error("resident resource name must not be empty")] + EmptyResourceName, + + #[error("resident resource name {0} is already registered")] + DuplicateResourceName(String), + + #[error("resident resource {resource_id} for {service} still owns active streams")] + ResourceInUse { + service: ServiceKey, + resource_id: u64, + }, + + #[error("resident resource {resource_id} does not belong to {service}")] + UnknownResource { + service: ServiceKey, + resource_id: u64, + }, + + #[error("resident resource {resource_id} for {service} is {actual:?}, expected {expected}")] + WrongResourceKind { + service: ServiceKey, + resource_id: u64, + actual: ResidentResourceKind, + expected: &'static str, + }, + + #[error("resident resource {resource_id} for {service} is in invalid state {state:?}")] + InvalidResourceState { + service: ServiceKey, + resource_id: u64, + state: ResidentResourceState, + }, + + #[error("resident endpoint denied by policy: {0}")] + EndpointDenied(String), +} + +#[derive(Clone, Debug)] +enum ResourceEntry { + Endpoint(ResidentEndpoint), + Stream { + name: String, + endpoint_id: ResourceId, + state: StreamState, + }, + Timer { + name: String, + armed: bool, + }, + MessageSubscription { + name: String, + }, + Source { + name: String, + source_kind: String, + }, +} + +impl ResourceEntry { + fn name(&self) -> &str { + match self { + Self::Endpoint(endpoint) => endpoint.name(), + Self::Stream { name, .. } + | Self::Timer { name, .. } + | Self::MessageSubscription { name } + | Self::Source { name, .. } => name, + } + } + + fn kind(&self) -> ResidentResourceKind { + match self { + Self::Endpoint(ResidentEndpoint::Tcp { .. }) => ResidentResourceKind::TcpEndpoint, + Self::Endpoint(ResidentEndpoint::Udp { .. }) => ResidentResourceKind::UdpEndpoint, + Self::Endpoint(ResidentEndpoint::Unix { .. }) => ResidentResourceKind::UnixEndpoint, + Self::Stream { .. } => ResidentResourceKind::Stream, + Self::Timer { .. } => ResidentResourceKind::Timer, + Self::MessageSubscription { .. } => ResidentResourceKind::MessageSubscription, + Self::Source { .. } => ResidentResourceKind::Source, + } + } + + fn state(&self) -> ResidentResourceState { + match self { + Self::Endpoint(_) + | Self::Timer { .. } + | Self::MessageSubscription { .. } + | Self::Source { .. } => ResidentResourceState::Ready, + Self::Stream { state, .. } => (*state).into(), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum StreamState { + Open, + ReadPaused, + PeerHalfClosed, + PeerHalfClosedReadPaused, + Closing, +} + +impl From for ResidentResourceState { + fn from(state: StreamState) -> Self { + match state { + StreamState::Open => Self::Open, + StreamState::ReadPaused | StreamState::PeerHalfClosedReadPaused => Self::ReadPaused, + StreamState::PeerHalfClosed => Self::PeerHalfClosed, + StreamState::Closing => Self::Closing, + } + } +} + +/// Revision-scoped supervisor for Host-owned resident resources. +/// +/// This type intentionally performs no operating-system I/O. It makes driver +/// implementations replaceable and keeps Tokio or another async executor out +/// of the Wasmtime Actor thread. +pub struct ResidentSession { + actor: ActorHandle, + policy: ResidentPolicy, + resources: HashMap, + next_resource_id: u64, + shutting_down: bool, +} + +impl ResidentSession { + /// Creates a resource session for an Actor exporting resident WIT 0.1.0. + pub fn new(actor: ActorHandle, policy: ResidentPolicy) -> Result { + if actor.execution() != ComponentExecution::Resident { + return Err(RuntimeError::NotResident(actor.key().clone()).into()); + } + Ok(Self { + actor, + policy, + resources: HashMap::new(), + next_resource_id: 1, + shutting_down: false, + }) + } + + /// Returns the immutable service revision that owns every resource ID. + pub fn owner(&self) -> &ServiceKey { + self.actor.key() + } + + /// Registers a configured listener or datagram endpoint after policy checks. + pub fn register_endpoint( + &mut self, + endpoint: ResidentEndpoint, + ) -> Result { + self.ensure_active()?; + self.policy.validate_endpoint(&endpoint)?; + self.insert_resource(ResourceEntry::Endpoint(endpoint), true) + } + + /// Registers a Host timer source. The driver decides its initial schedule. + pub fn register_timer( + &mut self, + name: impl Into, + ) -> Result { + self.ensure_active()?; + self.insert_resource( + ResourceEntry::Timer { + name: name.into(), + armed: false, + }, + true, + ) + } + + /// Registers a queue, broker, or internal mailbox subscription. + pub fn register_message_subscription( + &mut self, + name: impl Into, + ) -> Result { + self.ensure_active()?; + self.insert_resource( + ResourceEntry::MessageSubscription { name: name.into() }, + true, + ) + } + + /// Registers an extension driver such as a file watcher or serial device. + /// + /// `source_kind` identifies the separately versioned capability contract + /// understood by both the driver and Component. + pub fn register_source( + &mut self, + name: impl Into, + source_kind: impl Into, + ) -> Result { + self.ensure_active()?; + let source_kind = source_kind.into(); + if source_kind.trim().is_empty() { + return Err(ResidentHostError::EmptyResourceName); + } + self.insert_resource( + ResourceEntry::Source { + name: name.into(), + source_kind, + }, + true, + ) + } + + /// Allocates a stream for a connection accepted by a TCP or Unix driver. + pub fn accept_stream( + &mut self, + endpoint_id: ResourceId, + peer: Option, + ) -> Result<(ResourceId, Vec), ResidentHostError> { + self.ensure_active()?; + let endpoint = self.resource(endpoint_id)?; + match endpoint { + ResourceEntry::Endpoint(ResidentEndpoint::Tcp { .. }) + | ResourceEntry::Endpoint(ResidentEndpoint::Unix { .. }) => {} + other => { + return Err(self.wrong_kind(endpoint_id, other.kind(), "TCP or Unix endpoint")); + } + } + + let stream_id = self.insert_resource( + ResourceEntry::Stream { + name: format!("stream-{}", self.next_resource_id), + endpoint_id, + state: StreamState::Open, + }, + false, + )?; + let result = self.dispatch(ResidentEvent::StreamOpened { + endpoint_id, + stream_id, + peer, + }); + if result.is_err() { + self.resources.remove(&stream_id); + } + result.map(|operations| (stream_id, operations)) + } + + /// Delivers one bounded stream read. Paused and closing streams reject data. + pub fn stream_data( + &mut self, + stream_id: ResourceId, + bytes: Vec, + ) -> Result, ResidentHostError> { + self.ensure_stream_state(stream_id, &[StreamState::Open])?; + self.dispatch(ResidentEvent::StreamData { stream_id, bytes }) + } + + /// Signals that a driver can accept writes again. + pub fn stream_writable( + &mut self, + stream_id: ResourceId, + ) -> Result, ResidentHostError> { + self.ensure_stream_not_closing(stream_id)?; + self.dispatch(ResidentEvent::StreamWritable { stream_id }) + } + + /// Records a peer read-half close while preserving the local write half. + pub fn stream_half_closed( + &mut self, + stream_id: ResourceId, + ) -> Result, ResidentHostError> { + let state = self.stream_state_mut(stream_id)?; + *state = match *state { + StreamState::Open => StreamState::PeerHalfClosed, + StreamState::ReadPaused => StreamState::PeerHalfClosedReadPaused, + other => return Err(self.invalid_state(stream_id, other.into())), + }; + self.dispatch(ResidentEvent::StreamHalfClosed { stream_id }) + } + + /// Delivers terminal stream state and releases the resource identity. + pub fn stream_closed( + &mut self, + stream_id: ResourceId, + reason: StreamCloseReason, + ) -> Result, ResidentHostError> { + self.ensure_kind(stream_id, ResidentResourceKind::Stream, "stream")?; + let result = self.dispatch(ResidentEvent::StreamClosed { stream_id, reason }); + self.resources.remove(&stream_id); + result + } + + /// Delivers one UDP datagram to its configured endpoint. + pub fn datagram( + &mut self, + endpoint_id: ResourceId, + peer: String, + bytes: Vec, + ) -> Result, ResidentHostError> { + self.ensure_kind( + endpoint_id, + ResidentResourceKind::UdpEndpoint, + "UDP endpoint", + )?; + self.dispatch(ResidentEvent::Datagram { + endpoint_id, + peer, + bytes, + }) + } + + /// Delivers a timer occurrence. + pub fn timer_fired( + &mut self, + timer_id: ResourceId, + scheduled_at_ns: u64, + ) -> Result, ResidentHostError> { + self.ensure_kind(timer_id, ResidentResourceKind::Timer, "timer")?; + if let ResourceEntry::Timer { armed, .. } = self.resource_mut(timer_id)? { + *armed = false; + } + self.dispatch(ResidentEvent::Timer { + timer_id, + scheduled_at_ns, + }) + } + + /// Delivers a broker or internal queue message. + pub fn message( + &mut self, + subscription_id: ResourceId, + message_id: u64, + bytes: Vec, + ) -> Result, ResidentHostError> { + self.ensure_kind( + subscription_id, + ResidentResourceKind::MessageSubscription, + "message subscription", + )?; + self.dispatch(ResidentEvent::Message { + subscription_id, + message_id, + bytes, + }) + } + + /// Delivers an event from a separately versioned extension driver. + pub fn source_event( + &mut self, + source_id: ResourceId, + event: String, + payload: Vec, + ) -> Result, ResidentHostError> { + self.ensure_kind(source_id, ResidentResourceKind::Source, "source")?; + self.dispatch(ResidentEvent::Source { + source_id, + event, + payload, + }) + } + + /// Gives the Component one final bounded callback and rejects later events. + /// + /// After applying the returned operations, the owner closes every driver + /// resource and stops the Actor. Shutdown is not a drain barrier by itself. + pub fn shutdown(&mut self) -> Result, ResidentHostError> { + self.ensure_active()?; + let result = self.dispatch(ResidentEvent::Shutdown); + self.shutting_down = true; + result + } + + /// Removes a Host resource after its driver has released the actual handle. + pub fn release_resource( + &mut self, + resource_id: ResourceId, + ) -> Result { + if self.resources.values().any(|entry| { + matches!( + entry, + ResourceEntry::Stream { endpoint_id, .. } if *endpoint_id == resource_id + ) + }) { + return Err(ResidentHostError::ResourceInUse { + service: self.owner().clone(), + resource_id: resource_id.get(), + }); + } + let entry = self + .resources + .remove(&resource_id) + .ok_or_else(|| self.unknown_resource(resource_id))?; + Ok(resource_info(resource_id, &entry)) + } + + /// Returns one management-safe resource snapshot. + pub fn resource_info( + &self, + resource_id: ResourceId, + ) -> Result { + self.resource(resource_id) + .map(|entry| resource_info(resource_id, entry)) + } + + /// Returns all resource snapshots ordered by their non-reused IDs. + pub fn resources(&self) -> Vec { + let mut resources = self + .resources + .iter() + .map(|(id, entry)| resource_info(*id, entry)) + .collect::>(); + resources.sort_unstable_by_key(|resource| resource.id); + resources + } + + fn dispatch( + &mut self, + event: ResidentEvent, + ) -> Result, ResidentHostError> { + let effects = self.actor.dispatch_event(event)?; + self.apply_effects(effects) + } + + fn apply_effects( + &mut self, + effects: Vec, + ) -> Result, ResidentHostError> { + // Validate the whole batch against a copy. A bad later effect must not + // leave pause, close, or timer state from an earlier effect committed. + let mut next_resources = self.resources.clone(); + let mut operations = Vec::with_capacity(effects.len()); + for effect in effects { + operations.push(self.validate_effect(&mut next_resources, effect)?); + } + self.resources = next_resources; + Ok(operations) + } + + fn validate_effect( + &self, + resources: &mut HashMap, + effect: ResidentEffect, + ) -> Result { + match effect { + ResidentEffect::WriteStream { stream_id, bytes } => { + ensure_stream_not_closing(resources, self.owner(), stream_id)?; + Ok(ResidentOperation::WriteStream { stream_id, bytes }) + } + ResidentEffect::CloseStream { stream_id } => { + *stream_state_mut(resources, self.owner(), stream_id)? = StreamState::Closing; + Ok(ResidentOperation::CloseStream { stream_id }) + } + ResidentEffect::PauseStream { stream_id } => { + let state = stream_state_mut(resources, self.owner(), stream_id)?; + *state = match *state { + StreamState::Open => StreamState::ReadPaused, + StreamState::PeerHalfClosed => StreamState::PeerHalfClosedReadPaused, + other => return Err(invalid_state(self.owner(), stream_id, other.into())), + }; + Ok(ResidentOperation::PauseStream { stream_id }) + } + ResidentEffect::ResumeStream { stream_id } => { + let state = stream_state_mut(resources, self.owner(), stream_id)?; + *state = match *state { + StreamState::ReadPaused => StreamState::Open, + StreamState::PeerHalfClosedReadPaused => StreamState::PeerHalfClosed, + other => return Err(invalid_state(self.owner(), stream_id, other.into())), + }; + Ok(ResidentOperation::ResumeStream { stream_id }) + } + ResidentEffect::SendDatagram { + endpoint_id, + peer, + bytes, + } => { + ensure_kind( + resources, + self.owner(), + endpoint_id, + ResidentResourceKind::UdpEndpoint, + "UDP endpoint", + )?; + Ok(ResidentOperation::SendDatagram { + endpoint_id, + peer, + bytes, + }) + } + ResidentEffect::ArmTimer { + timer_id, + delay_ms, + interval_ms, + } => { + let entry = resource_mut(resources, self.owner(), timer_id)?; + let ResourceEntry::Timer { armed, .. } = entry else { + return Err(wrong_kind(self.owner(), timer_id, entry.kind(), "timer")); + }; + *armed = true; + Ok(ResidentOperation::ArmTimer { + timer_id, + delay_ms, + interval_ms, + }) + } + ResidentEffect::CancelTimer { timer_id } => { + let entry = resource_mut(resources, self.owner(), timer_id)?; + let ResourceEntry::Timer { armed, .. } = entry else { + return Err(wrong_kind(self.owner(), timer_id, entry.kind(), "timer")); + }; + *armed = false; + Ok(ResidentOperation::CancelTimer { timer_id }) + } + ResidentEffect::AcknowledgeMessage { + subscription_id, + message_id, + } => { + ensure_kind( + resources, + self.owner(), + subscription_id, + ResidentResourceKind::MessageSubscription, + "message subscription", + )?; + Ok(ResidentOperation::AcknowledgeMessage { + subscription_id, + message_id, + }) + } + ResidentEffect::SourceCommand { + source_id, + command, + payload, + } => { + ensure_kind( + resources, + self.owner(), + source_id, + ResidentResourceKind::Source, + "source", + )?; + Ok(ResidentOperation::SourceCommand { + source_id, + command, + payload, + }) + } + } + } + + fn insert_resource( + &mut self, + entry: ResourceEntry, + unique_name: bool, + ) -> Result { + if entry.name().trim().is_empty() { + return Err(ResidentHostError::EmptyResourceName); + } + if unique_name + && self + .resources + .values() + .any(|existing| existing.name() == entry.name()) + { + return Err(ResidentHostError::DuplicateResourceName( + entry.name().to_owned(), + )); + } + let limit = self.actor.resident_resource_limit(); + if self.resources.len() >= limit { + return Err(ResidentHostError::ResourceLimit { + service: self.owner().clone(), + limit, + }); + } + + let id = ResourceId::new(self.next_resource_id) + .ok_or_else(|| ResidentHostError::ResourceIdsExhausted(self.owner().clone()))?; + self.next_resource_id = self + .next_resource_id + .checked_add(1) + .ok_or_else(|| ResidentHostError::ResourceIdsExhausted(self.owner().clone()))?; + self.resources.insert(id, entry); + Ok(id) + } + + fn ensure_active(&self) -> Result<(), ResidentHostError> { + if self.shutting_down { + Err(ResidentHostError::ShuttingDown(self.owner().clone())) + } else { + Ok(()) + } + } + + fn resource(&self, id: ResourceId) -> Result<&ResourceEntry, ResidentHostError> { + self.resources + .get(&id) + .ok_or_else(|| self.unknown_resource(id)) + } + + fn resource_mut(&mut self, id: ResourceId) -> Result<&mut ResourceEntry, ResidentHostError> { + let owner = self.owner().clone(); + self.resources + .get_mut(&id) + .ok_or(ResidentHostError::UnknownResource { + service: owner, + resource_id: id.get(), + }) + } + + fn ensure_kind( + &self, + id: ResourceId, + expected_kind: ResidentResourceKind, + expected: &'static str, + ) -> Result<(), ResidentHostError> { + let entry = self.resource(id)?; + if entry.kind() == expected_kind { + Ok(()) + } else { + Err(self.wrong_kind(id, entry.kind(), expected)) + } + } + + fn stream_state_mut(&mut self, id: ResourceId) -> Result<&mut StreamState, ResidentHostError> { + let owner = self.owner().clone(); + stream_state_mut(&mut self.resources, &owner, id) + } + + fn ensure_stream_state( + &self, + id: ResourceId, + allowed: &[StreamState], + ) -> Result<(), ResidentHostError> { + let entry = self.resource(id)?; + let ResourceEntry::Stream { state, .. } = entry else { + return Err(self.wrong_kind(id, entry.kind(), "stream")); + }; + if allowed.contains(state) { + Ok(()) + } else { + Err(self.invalid_state(id, (*state).into())) + } + } + + fn ensure_stream_not_closing(&self, id: ResourceId) -> Result<(), ResidentHostError> { + ensure_stream_not_closing(&self.resources, self.owner(), id) + } + + fn unknown_resource(&self, id: ResourceId) -> ResidentHostError { + ResidentHostError::UnknownResource { + service: self.owner().clone(), + resource_id: id.get(), + } + } + + fn wrong_kind( + &self, + id: ResourceId, + actual: ResidentResourceKind, + expected: &'static str, + ) -> ResidentHostError { + wrong_kind(self.owner(), id, actual, expected) + } + + fn invalid_state(&self, id: ResourceId, state: ResidentResourceState) -> ResidentHostError { + invalid_state(self.owner(), id, state) + } +} + +impl ResidentPolicy { + fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> { + match endpoint { + ResidentEndpoint::Tcp { bind, .. } if !self.tcp_listen.allows(*bind) => { + Err(ResidentHostError::EndpointDenied(format!( + "TCP listen address {bind} is outside {:?}", + self.tcp_listen + ))) + } + ResidentEndpoint::Udp { bind, .. } if !self.udp_bind.allows(*bind) => { + Err(ResidentHostError::EndpointDenied(format!( + "UDP bind address {bind} is outside {:?}", + self.udp_bind + ))) + } + ResidentEndpoint::Unix { path, .. } => self.validate_unix_path(path), + ResidentEndpoint::Tcp { .. } | ResidentEndpoint::Udp { .. } => Ok(()), + } + } + + fn validate_unix_path(&self, path: &Path) -> Result<(), ResidentHostError> { + if !path.is_absolute() + || path + .components() + .any(|component| matches!(component, PathComponent::ParentDir)) + { + return Err(ResidentHostError::EndpointDenied(format!( + "Unix socket path {} must be absolute and contain no parent traversal", + path.display() + ))); + } + let parent = path.parent().ok_or_else(|| { + ResidentHostError::EndpointDenied(format!( + "Unix socket path {} has no parent", + path.display() + )) + })?; + let resolved_parent = fs::canonicalize(parent).map_err(|error| { + ResidentHostError::EndpointDenied(format!( + "Unix socket parent {} cannot be resolved: {error}", + parent.display() + )) + })?; + + for root in &self.unix_listen_roots { + let resolved_root = fs::canonicalize(root).map_err(|error| { + ResidentHostError::EndpointDenied(format!( + "configured Unix root {} cannot be resolved: {error}", + root.display() + )) + })?; + if resolved_parent.starts_with(resolved_root) { + return Ok(()); + } + } + Err(ResidentHostError::EndpointDenied(format!( + "Unix socket path {} is outside configured roots", + path.display() + ))) + } +} + +fn resource_info(id: ResourceId, entry: &ResourceEntry) -> ResidentResourceInfo { + ResidentResourceInfo { + id, + name: entry.name().to_owned(), + kind: entry.kind(), + state: entry.state(), + metadata: match entry { + ResourceEntry::Endpoint(endpoint) => ResidentResourceMetadata::Endpoint { + endpoint: endpoint.clone(), + }, + ResourceEntry::Stream { endpoint_id, .. } => ResidentResourceMetadata::Stream { + endpoint_id: *endpoint_id, + }, + ResourceEntry::Timer { armed, .. } => ResidentResourceMetadata::Timer { armed: *armed }, + ResourceEntry::MessageSubscription { .. } => { + ResidentResourceMetadata::MessageSubscription + } + ResourceEntry::Source { source_kind, .. } => ResidentResourceMetadata::Source { + source_kind: source_kind.clone(), + }, + }, + } +} + +fn resource_mut<'a>( + resources: &'a mut HashMap, + owner: &ServiceKey, + id: ResourceId, +) -> Result<&'a mut ResourceEntry, ResidentHostError> { + resources + .get_mut(&id) + .ok_or_else(|| ResidentHostError::UnknownResource { + service: owner.clone(), + resource_id: id.get(), + }) +} + +fn stream_state_mut<'a>( + resources: &'a mut HashMap, + owner: &ServiceKey, + id: ResourceId, +) -> Result<&'a mut StreamState, ResidentHostError> { + let entry = resource_mut(resources, owner, id)?; + let ResourceEntry::Stream { state, .. } = entry else { + return Err(wrong_kind(owner, id, entry.kind(), "stream")); + }; + Ok(state) +} + +fn ensure_stream_not_closing( + resources: &HashMap, + owner: &ServiceKey, + id: ResourceId, +) -> Result<(), ResidentHostError> { + let entry = resources + .get(&id) + .ok_or_else(|| ResidentHostError::UnknownResource { + service: owner.clone(), + resource_id: id.get(), + })?; + let ResourceEntry::Stream { state, .. } = entry else { + return Err(wrong_kind(owner, id, entry.kind(), "stream")); + }; + if *state == StreamState::Closing { + Err(invalid_state(owner, id, (*state).into())) + } else { + Ok(()) + } +} + +fn ensure_kind( + resources: &HashMap, + owner: &ServiceKey, + id: ResourceId, + expected_kind: ResidentResourceKind, + expected: &'static str, +) -> Result<(), ResidentHostError> { + let entry = resources + .get(&id) + .ok_or_else(|| ResidentHostError::UnknownResource { + service: owner.clone(), + resource_id: id.get(), + })?; + if entry.kind() == expected_kind { + Ok(()) + } else { + Err(wrong_kind(owner, id, entry.kind(), expected)) + } +} + +fn wrong_kind( + owner: &ServiceKey, + id: ResourceId, + actual: ResidentResourceKind, + expected: &'static str, +) -> ResidentHostError { + ResidentHostError::WrongResourceKind { + service: owner.clone(), + resource_id: id.get(), + actual, + expected, + } +} + +fn invalid_state( + owner: &ServiceKey, + id: ResourceId, + state: ResidentResourceState, +) -> ResidentHostError { + ResidentHostError::InvalidResourceState { + service: owner.clone(), + resource_id: id.get(), + state, + } +} diff --git a/crates/wasmeld-runtime/src/runtime.rs b/crates/wasmeld-runtime/src/runtime.rs index 3560485..2b8730e 100644 --- a/crates/wasmeld-runtime/src/runtime.rs +++ b/crates/wasmeld-runtime/src/runtime.rs @@ -31,10 +31,15 @@ use wasmtime_wasi::{ }; use crate::{ - CapabilityDescriptor, KvBackend, RuntimeError, ServiceKey, ServiceManifest, - bindings::{ServiceComponent, ServiceError}, + CapabilityDescriptor, ComponentExecution, KvBackend, ResidentEffect, ResidentEvent, + RuntimeError, ServiceKey, ServiceManifest, + bindings::{ + ServiceComponent, ServiceError, + resident::{ResidentComponent, exports::wasmeld::resident::actor::ResidentError}, + }, capability::{Capability, CapabilityRegistry, HostCapabilities}, manifest::ResourceLimits, + resident::{RESIDENT_INTERFACE, from_wit_effects, into_wit_event}, }; const DEFAULT_EPOCH_TICK: Duration = Duration::from_millis(5); @@ -171,6 +176,7 @@ struct RegisteredService { manifest: ServiceManifest, component: Arc, capabilities: Vec, + execution: ComponentExecution, } impl Runtime { @@ -231,6 +237,7 @@ impl Runtime { let component = Component::new(&self.inner.engine, component_bytes.as_ref()) .map_err(RuntimeError::ComponentCompilation)?; let capabilities = validate_component_imports(&component, &self.inner.engine)?; + let execution = component_execution(&component, &self.inner.engine); if capabilities.contains(&Capability::KvStore) && self.inner.kv_backend.is_none() { return Err(RuntimeError::CapabilityUnavailable( Capability::KvStore.descriptor().interface().to_owned(), @@ -253,6 +260,7 @@ impl Runtime { manifest, component: Arc::new(component), capabilities, + execution, }, ); @@ -372,6 +380,34 @@ impl Runtime { .collect()) } + /// Returns whether a registered Component is passive or event-driven. + pub fn execution(&self, key: &ServiceKey) -> Result { + self.inner + .services + .lock() + .map_err(|_| RuntimeError::LockPoisoned)? + .get(key) + .map(|service| service.execution) + .ok_or_else(|| RuntimeError::ServiceNotRegistered(key.clone())) + } + + /// Delivers one Host event through the Actor's bounded serial mailbox. + pub fn dispatch_event( + &self, + key: &ServiceKey, + event: ResidentEvent, + ) -> Result, RuntimeError> { + let actor = self + .inner + .actors + .lock() + .map_err(|_| RuntimeError::LockPoisoned)? + .get(key) + .cloned() + .ok_or_else(|| RuntimeError::ActorUnavailable(key.clone()))?; + actor.dispatch_event(event) + } + /// Stops one Actor and releases its Store and Component Instance. pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> { let _lifecycle = self @@ -528,6 +564,7 @@ impl Runtime { #[derive(Clone)] pub struct ActorHandle { key: ServiceKey, + execution: ComponentExecution, limits: Arc, sender: SyncSender, status: Arc, @@ -540,6 +577,15 @@ impl ActorHandle { &self.key } + /// Returns the execution model detected from the Component exports. + pub fn execution(&self) -> ComponentExecution { + self.execution + } + + pub(crate) fn resident_resource_limit(&self) -> usize { + self.limits.resident.max_resources + } + /// Sends an invocation and waits up to the remaining service deadline. pub fn invoke(&self, input: Vec) -> Result, RuntimeError> { if !self.is_available() { @@ -583,6 +629,45 @@ impl ActorHandle { } } + /// Sends a Host-owned resource event and waits for validated effects. + pub fn dispatch_event( + &self, + event: ResidentEvent, + ) -> Result, RuntimeError> { + if !self.is_available() { + return Err(RuntimeError::ActorUnavailable(self.key.clone())); + } + validate_resident_event(&self.key, &self.limits, &event)?; + + let (response_sender, response_receiver) = mpsc::sync_channel(1); + let deadline = Instant::now() + self.limits.deadline(); + let command = ActorCommand::ResidentEvent { + event, + deadline, + response_sender, + }; + match self.sender.try_send(command) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + return Err(RuntimeError::ActorOverloaded(self.key.clone())); + } + Err(TrySendError::Disconnected(_)) => { + return Err(RuntimeError::ActorUnavailable(self.key.clone())); + } + } + + match response_receiver.recv_timeout(deadline.saturating_duration_since(Instant::now())) { + Ok(result) => result, + Err(mpsc::RecvTimeoutError::Timeout) => Err(RuntimeError::DeadlineExceeded { + service: self.key.clone(), + deadline: self.limits.deadline(), + }), + Err(mpsc::RecvTimeoutError::Disconnected) => { + Err(RuntimeError::ActorStopped(self.key.clone())) + } + } + } + /// Prevents new calls, asks the worker to stop, and waits for acknowledgement. pub fn stop(&self) -> Result<(), RuntimeError> { if !self.status.accepting.swap(false, Ordering::AcqRel) { @@ -622,6 +707,11 @@ enum ActorCommand { deadline: Instant, response_sender: SyncSender, RuntimeError>>, }, + ResidentEvent { + event: ResidentEvent, + deadline: Instant, + response_sender: SyncSender, RuntimeError>>, + }, Stop { response_sender: SyncSender<()>, }, @@ -719,6 +809,7 @@ struct ActorWorker { epoch_tick: Duration, store: Store, bindings: ServiceComponent, + resident: Option, } impl ActorWorker { @@ -750,8 +841,18 @@ impl ActorWorker { let epoch_ticks = ticks_for(limits.deadline(), epoch_tick); configure_call_budget(&mut store, &limits, epoch_ticks)?; - let bindings = ServiceComponent::instantiate(&mut store, &service.component, &linker) + let instance = linker + .instantiate(&mut store, &service.component) .map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?; + let bindings = ServiceComponent::new(&mut store, &instance) + .map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?; + let resident = match service.execution { + ComponentExecution::Service => None, + ComponentExecution::Resident => Some( + ResidentComponent::new(&mut store, &instance) + .map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?, + ), + }; configure_call_budget(&mut store, &limits, epoch_ticks)?; @@ -771,6 +872,7 @@ impl ActorWorker { epoch_tick, store, bindings, + resident, }) } @@ -807,6 +909,54 @@ impl ActorWorker { }), } } + + fn dispatch_event( + &mut self, + event: ResidentEvent, + deadline: Instant, + ) -> Result, RuntimeError> { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(RuntimeError::DeadlineExceeded { + service: self.key.clone(), + deadline: self.limits.deadline(), + }); + } + let resident = self + .resident + .as_ref() + .ok_or_else(|| RuntimeError::NotResident(self.key.clone()))?; + configure_call_budget( + &mut self.store, + &self.limits, + ticks_for(remaining, self.epoch_tick), + )?; + + match resident + .wasmeld_resident_actor() + .call_handle_event(&mut self.store, &into_wit_event(event)) + { + Ok(Ok(effects)) => { + if effects.len() > self.limits.resident.max_effects_per_event { + return Err(RuntimeError::TooManyResidentEffects { + service: self.key.clone(), + limit: self.limits.resident.max_effects_per_event, + }); + } + let effects = from_wit_effects(effects)?; + validate_resident_effects(&self.key, &self.limits, &effects)?; + Ok(effects) + } + Ok(Err(ResidentError::EventFailed(message))) => Err(RuntimeError::ComponentError { + kind: "resident-event", + message, + }), + Err(error) => Err(RuntimeError::ActorFault { + service: self.key.clone(), + message: error.to_string(), + }), + } + } } fn add_restricted_wasi(linker: &mut Linker) -> Result<(), RuntimeError> { @@ -883,6 +1033,18 @@ fn validate_component_imports( Ok(capabilities) } +fn component_execution(component: &Component, engine: &Engine) -> ComponentExecution { + if component + .component_type() + .exports(engine) + .any(|(name, _)| name == RESIDENT_INTERFACE) + { + ComponentExecution::Resident + } else { + ComponentExecution::Service + } +} + fn spawn_actor( engine: Arc, ticker: Arc, @@ -892,6 +1054,7 @@ fn spawn_actor( init_config: Vec, kv_backend: Option>, ) -> Result { + let execution = service.execution; let limits = Arc::new(service.manifest.limits.clone()); let (sender, receiver) = mpsc::sync_channel(limits.mailbox_capacity); let (ready_sender, ready_receiver) = mpsc::sync_channel(1); @@ -925,6 +1088,7 @@ fn spawn_actor( Ok(ActorHandle { key: thread_key, + execution, limits, sender, status, @@ -961,6 +1125,29 @@ fn run_actor(worker: &mut ActorWorker, receiver: Receiver, status: return; } } + ActorCommand::ResidentEvent { + event, + deadline, + response_sender, + } => { + let result = if Instant::now() >= deadline { + Err(RuntimeError::DeadlineExceeded { + service: worker.key.clone(), + deadline: worker.limits.deadline(), + }) + } else { + worker.dispatch_event(event, deadline) + }; + let fatal = is_fatal_resident(&result); + if fatal { + status.mark_stopped(); + } + let _ = response_sender.send(result); + if fatal { + drain_after_failure(receiver, &worker.key); + return; + } + } ActorCommand::Stop { response_sender } => { status.mark_stopped(); let _ = response_sender.send(()); @@ -980,6 +1167,11 @@ fn drain_after_failure(receiver: Receiver, key: &ServiceKey) { } => { let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone()))); } + ActorCommand::ResidentEvent { + response_sender, .. + } => { + let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone()))); + } ActorCommand::Stop { response_sender } => { let _ = response_sender.send(()); } @@ -987,6 +1179,113 @@ fn drain_after_failure(receiver: Receiver, key: &ServiceKey) { } } +fn validate_resident_event( + key: &ServiceKey, + limits: &ResourceLimits, + event: &ResidentEvent, +) -> Result<(), RuntimeError> { + let (payload_len, limit) = match event { + ResidentEvent::StreamData { bytes, .. } => { + (bytes.len(), limits.resident.max_stream_chunk_bytes) + } + ResidentEvent::Datagram { bytes, .. } => (bytes.len(), limits.resident.max_datagram_bytes), + ResidentEvent::Message { bytes, .. } => (bytes.len(), limits.max_input_bytes), + ResidentEvent::Source { event, payload, .. } => ( + event.len().saturating_add(payload.len()), + limits.max_input_bytes, + ), + ResidentEvent::Timer { .. } + | ResidentEvent::StreamOpened { .. } + | ResidentEvent::StreamWritable { .. } + | ResidentEvent::StreamHalfClosed { .. } + | ResidentEvent::StreamClosed { .. } + | ResidentEvent::Shutdown => return Ok(()), + }; + if payload_len > limit { + Err(RuntimeError::ResidentEventTooLarge { + service: key.clone(), + limit, + }) + } else { + Ok(()) + } +} + +fn validate_resident_effects( + key: &ServiceKey, + limits: &ResourceLimits, + effects: &[ResidentEffect], +) -> Result<(), RuntimeError> { + for effect in effects { + match effect { + ResidentEffect::WriteStream { bytes, .. } + if bytes.len() > limits.resident.max_stream_chunk_bytes => + { + return Err(invalid_resident_effect( + key, + "stream write", + limits.resident.max_stream_chunk_bytes, + )); + } + ResidentEffect::SendDatagram { peer, bytes, .. } => { + if peer.is_empty() { + return Err(RuntimeError::InvalidResidentEffect(format!( + "{key} returned a datagram without a peer" + ))); + } + if bytes.len() > limits.resident.max_datagram_bytes { + return Err(invalid_resident_effect( + key, + "datagram", + limits.resident.max_datagram_bytes, + )); + } + } + ResidentEffect::ArmTimer { + delay_ms, + interval_ms, + .. + } => { + let minimum = limits.resident.min_timer_delay_ms; + if *delay_ms < minimum || interval_ms.is_some_and(|value| value < minimum) { + return Err(RuntimeError::InvalidResidentEffect(format!( + "{key} returned a timer below the {minimum} ms minimum" + ))); + } + } + ResidentEffect::SourceCommand { + command, payload, .. + } => { + if command.is_empty() { + return Err(RuntimeError::InvalidResidentEffect(format!( + "{key} returned an empty source command" + ))); + } + if command.len().saturating_add(payload.len()) > limits.max_output_bytes { + return Err(invalid_resident_effect( + key, + "source command", + limits.max_output_bytes, + )); + } + } + ResidentEffect::WriteStream { .. } + | ResidentEffect::CloseStream { .. } + | ResidentEffect::PauseStream { .. } + | ResidentEffect::ResumeStream { .. } + | ResidentEffect::CancelTimer { .. } + | ResidentEffect::AcknowledgeMessage { .. } => {} + } + } + Ok(()) +} + +fn invalid_resident_effect(key: &ServiceKey, kind: &str, limit: usize) -> RuntimeError { + RuntimeError::InvalidResidentEffect(format!( + "{key} returned a {kind} payload above the {limit}-byte limit" + )) +} + fn configure_call_budget( store: &mut Store, limits: &ResourceLimits, @@ -1017,3 +1316,7 @@ fn component_error(kind: &'static str, error: ServiceError) -> RuntimeError { fn is_fatal(result: &Result, RuntimeError>) -> bool { matches!(result, Err(RuntimeError::ActorFault { .. })) } + +fn is_fatal_resident(result: &Result, RuntimeError>) -> bool { + matches!(result, Err(RuntimeError::ActorFault { .. })) +} diff --git a/crates/wasmeld-runtime/tests/runtime_components.rs b/crates/wasmeld-runtime/tests/runtime_components.rs index c63a567..1ae8802 100644 --- a/crates/wasmeld-runtime/tests/runtime_components.rs +++ b/crates/wasmeld-runtime/tests/runtime_components.rs @@ -1,6 +1,7 @@ use std::{ collections::HashMap, fs, + net::SocketAddr, path::{Path, PathBuf}, process::Command, sync::{Arc, Mutex, OnceLock}, @@ -9,8 +10,11 @@ use std::{ }; use wasmeld_runtime::{ - KV_MAX_VALUE_BYTES, KvBackend, KvBackendError, ResourceLimits, Runtime, RuntimeConfig, - RuntimeError, ServiceManifest, + ComponentExecution, KV_MAX_VALUE_BYTES, KvBackend, KvBackendError, NetworkScope, + ResidentEffect, ResidentEndpoint, ResidentEvent, ResidentHostError, ResidentLimits, + ResidentOperation, ResidentPolicy, ResidentResourceKind, ResidentResourceMetadata, + ResidentSession, ResourceId, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, + ServiceManifest, StreamCloseReason, }; static COMPONENTS_BUILT: OnceLock<()> = OnceLock::new(); @@ -192,6 +196,253 @@ fn explicit_clock_capability_is_linked() { runtime.stop(&key).unwrap(); } +#[test] +fn resident_component_handles_host_events_in_its_actor_mailbox() { + let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); + let (key, actor) = start_component(&runtime, "resident-probe", "resident_probe_component.wasm"); + let stream_id = resource_id(1); + let endpoint_id = resource_id(2); + let timer_id = resource_id(3); + let subscription_id = resource_id(4); + let source_id = resource_id(5); + + assert_eq!( + runtime.execution(&key).unwrap(), + ComponentExecution::Resident + ); + assert_eq!( + actor + .dispatch_event(ResidentEvent::StreamData { + stream_id, + bytes: b"stream".to_vec(), + }) + .unwrap(), + vec![ResidentEffect::WriteStream { + stream_id, + bytes: b"stream".to_vec(), + }] + ); + assert_eq!( + runtime + .dispatch_event( + &key, + ResidentEvent::Datagram { + endpoint_id, + peer: "127.0.0.1:9000".to_owned(), + bytes: b"datagram".to_vec(), + }, + ) + .unwrap(), + vec![ResidentEffect::SendDatagram { + endpoint_id, + peer: "127.0.0.1:9000".to_owned(), + bytes: b"datagram".to_vec(), + }] + ); + assert_eq!( + actor + .dispatch_event(ResidentEvent::Timer { + timer_id, + scheduled_at_ns: 42, + }) + .unwrap(), + vec![ResidentEffect::ArmTimer { + timer_id, + delay_ms: 10, + interval_ms: None, + }] + ); + assert_eq!( + actor + .dispatch_event(ResidentEvent::Message { + subscription_id, + message_id: 99, + bytes: b"message".to_vec(), + }) + .unwrap(), + vec![ResidentEffect::AcknowledgeMessage { + subscription_id, + message_id: 99, + }] + ); + assert_eq!( + actor + .dispatch_event(ResidentEvent::Source { + source_id, + event: "flush".to_owned(), + payload: b"source".to_vec(), + }) + .unwrap(), + vec![ResidentEffect::SourceCommand { + source_id, + command: "flush".to_owned(), + payload: b"source".to_vec(), + }] + ); + assert_eq!(decode_counter(actor.invoke(Vec::new()).unwrap()), 5); + + runtime.stop(&key).unwrap(); +} + +#[test] +fn passive_component_rejects_resident_events_without_stopping() { + let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); + let (key, actor) = start_component(&runtime, "passive-echo", "echo_component.wasm"); + + assert_eq!( + runtime.execution(&key).unwrap(), + ComponentExecution::Service + ); + assert!(matches!( + actor.dispatch_event(ResidentEvent::Shutdown), + Err(RuntimeError::NotResident(service)) if service == key + )); + assert_eq!( + actor.invoke(b"still-running".to_vec()).unwrap(), + b"still-running" + ); + + runtime.stop(&key).unwrap(); +} + +#[test] +fn resident_session_owns_resources_and_validates_driver_operations() { + let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); + let (key, actor) = start_component( + &runtime, + "resident-session", + "resident_probe_component.wasm", + ); + let policy = ResidentPolicy { + tcp_listen: NetworkScope::Loopback, + udp_bind: NetworkScope::Loopback, + unix_listen_roots: vec![std::env::temp_dir()], + }; + let mut session = ResidentSession::new(actor, policy).unwrap(); + let tcp_id = session + .register_endpoint(ResidentEndpoint::Tcp { + name: "ingress".to_owned(), + bind: socket_addr("127.0.0.1:7000"), + }) + .unwrap(); + let udp_id = session + .register_endpoint(ResidentEndpoint::Udp { + name: "discovery".to_owned(), + bind: socket_addr("127.0.0.1:7001"), + }) + .unwrap(); + let unix_id = session + .register_endpoint(ResidentEndpoint::Unix { + name: "local-ingress".to_owned(), + path: std::env::temp_dir().join("wasmeld-resident-test.sock"), + }) + .unwrap(); + let timer_id = session.register_timer("heartbeat").unwrap(); + let subscription_id = session.register_message_subscription("jobs").unwrap(); + let source_id = session + .register_source("watcher", "fs-watch@0.1.0") + .unwrap(); + + let (stream_id, opened) = session + .accept_stream(tcp_id, Some("127.0.0.1:51000".to_owned())) + .unwrap(); + assert!(opened.is_empty()); + assert_eq!( + session.stream_data(stream_id, b"stream".to_vec()).unwrap(), + vec![ResidentOperation::WriteStream { + stream_id, + bytes: b"stream".to_vec(), + }] + ); + assert_eq!( + session + .datagram(udp_id, "127.0.0.1:51001".to_owned(), b"datagram".to_vec(),) + .unwrap(), + vec![ResidentOperation::SendDatagram { + endpoint_id: udp_id, + peer: "127.0.0.1:51001".to_owned(), + bytes: b"datagram".to_vec(), + }] + ); + assert_eq!( + session.timer_fired(timer_id, 1).unwrap(), + vec![ResidentOperation::ArmTimer { + timer_id, + delay_ms: 10, + interval_ms: None, + }] + ); + assert!(matches!( + session.resource_info(timer_id).unwrap().metadata, + ResidentResourceMetadata::Timer { armed: true } + )); + assert_eq!( + session + .message(subscription_id, 7, b"work".to_vec()) + .unwrap(), + vec![ResidentOperation::AcknowledgeMessage { + subscription_id, + message_id: 7, + }] + ); + assert_eq!( + session + .source_event(source_id, "flush".to_owned(), b"state".to_vec()) + .unwrap(), + vec![ResidentOperation::SourceCommand { + source_id, + command: "flush".to_owned(), + payload: b"state".to_vec(), + }] + ); + assert!(matches!( + session.datagram(tcp_id, "127.0.0.1:1".to_owned(), Vec::new()), + Err(ResidentHostError::WrongResourceKind { .. }) + )); + + session + .stream_closed(stream_id, StreamCloseReason::PeerClosed) + .unwrap(); + assert!(matches!( + session.resource_info(stream_id), + Err(ResidentHostError::UnknownResource { .. }) + )); + assert_eq!(session.owner(), &key); + assert_eq!( + session.resource_info(tcp_id).unwrap().kind, + ResidentResourceKind::TcpEndpoint + ); + assert_eq!( + session.resource_info(unix_id).unwrap().kind, + ResidentResourceKind::UnixEndpoint + ); + + session.shutdown().unwrap(); + assert!(matches!( + session.register_timer("late"), + Err(ResidentHostError::ShuttingDown(_)) + )); + runtime.stop(&key).unwrap(); +} + +#[test] +fn resident_endpoint_policy_is_deny_by_default() { + let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); + let (key, actor) = + start_component(&runtime, "resident-policy", "resident_probe_component.wasm"); + let mut session = ResidentSession::new(actor, ResidentPolicy::default()).unwrap(); + + assert!(matches!( + session.register_endpoint(ResidentEndpoint::Tcp { + name: "public".to_owned(), + bind: socket_addr("0.0.0.0:8080"), + }), + Err(ResidentHostError::EndpointDenied(_)) + )); + + runtime.stop(&key).unwrap(); +} + #[test] fn kv_capability_is_service_scoped_and_shared_across_revisions() { let backend = Arc::new(MemoryKvBackend::default()); @@ -379,6 +630,7 @@ fn build_components_once() { "components/spin/Cargo.toml", "components/clock-probe/Cargo.toml", "components/kv-probe/Cargo.toml", + "components/resident-probe/Cargo.toml", "components/wasi-clock-probe/Cargo.toml", ] { let status = Command::new("rustup") @@ -421,9 +673,18 @@ fn test_limits() -> ResourceLimits { mailbox_capacity: 16, max_input_bytes: 16 * 1024, max_output_bytes: 16 * 1024, + resident: ResidentLimits::default(), } } +fn resource_id(value: u64) -> ResourceId { + ResourceId::new(value).expect("test resource id must be non-zero") +} + +fn socket_addr(value: &str) -> SocketAddr { + value.parse().expect("test socket address must be valid") +} + fn decode_counter(bytes: Vec) -> u64 { let array: [u8; 8] = bytes .try_into() @@ -447,6 +708,9 @@ fn component_world(artifact_name: &str) -> &'static str { "spin_component.wasm" => "component:spin/spin-component@0.1.0", "clock_probe_component.wasm" => "component:clock-probe/clock-probe-component@0.1.0", "kv_probe_component.wasm" => "component:kv-probe/kv-probe-component@0.1.0", + "resident_probe_component.wasm" => { + "component:resident-probe/resident-probe-component@0.1.0" + } "wasi_clock_probe_component.wasm" => { "component:wasi-clock-probe/wasi-clock-probe-component@0.1.0" }