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" }