diff --git a/crates/wasmeld-runtime/src/capability/kv.rs b/crates/wasmeld-runtime/src/capability/kv.rs index 3dd216c..94813fa 100644 --- a/crates/wasmeld-runtime/src/capability/kv.rs +++ b/crates/wasmeld-runtime/src/capability/kv.rs @@ -49,10 +49,13 @@ pub trait KvBackend: fmt::Debug + Send + Sync { /// Storage failures returned by a [`KvBackend`]. #[derive(Debug, Error)] pub enum KvBackendError { + /// Backend queue or connection pool is temporarily saturated. #[error("KV backend is busy")] Busy, + /// Backend did not finish within the Host-call timeout. #[error("KV backend timed out")] Timeout, + /// Backend rejected or failed the operation for another reason. #[error("KV backend operation failed: {0}")] Operation(String), } diff --git a/crates/wasmeld-runtime/src/error.rs b/crates/wasmeld-runtime/src/error.rs index 832d83b..3249567 100644 --- a/crates/wasmeld-runtime/src/error.rs +++ b/crates/wasmeld-runtime/src/error.rs @@ -9,98 +9,158 @@ use crate::manifest::ServiceKey; /// Failures from Runtime configuration, sandbox validation, or Actor execution. #[derive(Debug, Error)] pub enum RuntimeError { + /// Service ID or revision is empty. #[error("service id and revision must not be empty")] InvalidServiceKey, + /// Manifest identity, world, path, or resource limits are invalid. #[error("invalid manifest: {0}")] InvalidManifest(String), + /// Runtime manifest TOML could not be decoded. #[error("manifest parse failed: {0}")] ManifestParse(#[from] toml::de::Error), + /// A lifecycle operation addressed an unknown service revision. #[error("service revision {0} is not registered")] ServiceNotRegistered(ServiceKey), + /// Registration attempted to overwrite an immutable service revision. #[error("service revision {0} is already registered")] ServiceAlreadyRegistered(ServiceKey), + /// Wasmtime rejected the Component before it entered the registry. #[error("component compilation failed: {0}")] ComponentCompilation(#[source] wasmtime::Error), + /// The Component imports a WASI or Wasmeld interface outside the allowlist. #[error("component imports unsupported capability {0}")] UnsupportedImport(String), + /// A supported capability has no configured Host backend. #[error("component requires unavailable capability {0}")] CapabilityUnavailable(String), + /// The shared Wasmtime Engine could not be created. #[error("runtime creation failed: {0}")] RuntimeCreation(#[source] wasmtime::Error), + /// The configured compiled-Component cache directory could not be prepared. #[error("failed to prepare Component compilation cache {path}: {source}")] CacheDirectory { + /// Requested cache directory. path: PathBuf, + /// Underlying filesystem failure. #[source] source: std::io::Error, }, + /// Wasmtime rejected the persistent cache configuration. #[error("invalid Component compilation cache configuration: {0}")] CacheConfiguration(#[source] wasmtime::Error), + /// A Component path from a manifest could not be read. #[error("failed to read component artifact {path}: {source}")] ArtifactRead { + /// Manifest-provided artifact path. path: String, + /// Underlying filesystem failure. #[source] source: std::io::Error, }, + /// The operating system refused to create an Actor worker thread. #[error("failed to spawn actor thread: {0}")] ActorThread(#[source] std::io::Error), + /// Linking, instantiation, or the Component `init` export failed. #[error("actor initialization failed: {0}")] ActorInitialization(String), + /// The Actor is stopped, faulted, starting, or no longer accepting work. #[error("actor for {0} is unavailable")] ActorUnavailable(ServiceKey), + /// The bounded Actor mailbox has no capacity for another command. #[error("actor mailbox for {0} is full")] ActorOverloaded(ServiceKey), + /// The worker exited before returning a queued response. #[error("actor for {0} stopped before returning a result")] ActorStopped(ServiceKey), + /// Resident dispatch was attempted on a passive service Component. #[error("service revision {0} does not export the resident actor interface")] NotResident(ServiceKey), + /// A Host event payload exceeds its configured resident or input limit. #[error("resident event for {service} exceeds the {limit}-byte payload limit")] - ResidentEventTooLarge { service: ServiceKey, limit: usize }, + ResidentEventTooLarge { + /// Actor revision receiving the event. + service: ServiceKey, + /// Maximum accepted bytes. + limit: usize, + }, + /// A Component returned more effects than one event is allowed to produce. #[error("resident event for {service} returned more than {limit} effects")] - TooManyResidentEffects { service: ServiceKey, limit: usize }, + TooManyResidentEffects { + /// Actor revision that returned the effect list. + service: ServiceKey, + /// Maximum effects accepted per event. + limit: usize, + }, + /// A raw Component effect has invalid IDs, payloads, names, or timer values. #[error("invalid resident effect: {0}")] InvalidResidentEffect(String), + /// Invocation input exceeds the service manifest limit. #[error("input for {service} exceeds the {limit}-byte limit")] - InputTooLarge { service: ServiceKey, limit: usize }, + InputTooLarge { + /// Actor revision receiving the input. + service: ServiceKey, + /// Maximum accepted bytes. + limit: usize, + }, + /// Invocation output exceeds the service manifest limit. #[error("output for {service} exceeds the {limit}-byte limit")] - OutputTooLarge { service: ServiceKey, limit: usize }, + OutputTooLarge { + /// Actor revision returning the output. + service: ServiceKey, + /// Maximum accepted bytes. + limit: usize, + }, + /// Queue waiting and Component execution exceeded the end-to-end deadline. #[error("actor for {service} exceeded its {deadline:?} execution deadline")] DeadlineExceeded { + /// Actor revision whose command expired. service: ServiceKey, + /// Configured end-to-end command deadline. deadline: Duration, }, + /// A Wasm trap made the current Actor instance unusable. #[error("actor for {service} faulted: {message}")] ActorFault { + /// Faulted Actor revision. service: ServiceKey, + /// Wasmtime trap or call failure detail. message: String, }, + /// Guest code returned a typed WIT error without trapping the Actor. #[error("component returned {kind}: {message}")] - ComponentError { kind: &'static str, message: String }, + ComponentError { + /// Export phase such as `init`, `invoke`, or `resident-event`. + kind: &'static str, + /// Guest-provided diagnostic text. + message: String, + }, + /// An internal registry or lifecycle mutex was poisoned by a panic. #[error("runtime lock was poisoned")] LockPoisoned, } diff --git a/crates/wasmeld-runtime/src/lib.rs b/crates/wasmeld-runtime/src/lib.rs index 94d796d..dd17e42 100644 --- a/crates/wasmeld-runtime/src/lib.rs +++ b/crates/wasmeld-runtime/src/lib.rs @@ -4,6 +4,34 @@ //! It provides a restricted WASI compatibility context, links explicitly //! imported host capabilities, loads a Component that implements the stable //! service exports, and invokes it through one serial Actor. +//! +//! # Request/response example +//! +//! ```no_run +//! use wasmeld_runtime::{ResourceLimits, Runtime, RuntimeConfig, ServiceManifest}; +//! +//! let runtime = Runtime::new(RuntimeConfig::default())?; +//! let manifest = ServiceManifest { +//! id: "echo".into(), +//! revision: "1.0.0".into(), +//! component: "echo.wasm".into(), +//! world: "component:echo/echo-component@1.0.0".into(), +//! limits: ResourceLimits::default(), +//! }; +//! let bytes = std::fs::read(&manifest.component)?; +//! let key = runtime.register(manifest, bytes)?; +//! let actor = runtime.start(&key, Vec::new())?; +//! assert_eq!(actor.invoke(b"hello".to_vec())?, b"hello"); +//! runtime.stop(&key)?; +//! # Ok::<(), Box>(()) +//! ``` +//! +//! A started Actor retains Component memory between calls, but Component code +//! is never allowed to monopolize the thread between calls. Active timers, +//! sockets, subscriptions, and other long-lived resources are Host-owned and +//! enter resident Components as bounded events through [`ResidentSession`]. + +#![warn(missing_docs)] mod bindings; mod capability; diff --git a/crates/wasmeld-runtime/src/resident.rs b/crates/wasmeld-runtime/src/resident.rs index 7143ff7..987c504 100644 --- a/crates/wasmeld-runtime/src/resident.rs +++ b/crates/wasmeld-runtime/src/resident.rs @@ -5,6 +5,11 @@ //! 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. +//! +//! Driver authors normally use [`ResidentSession`](crate::ResidentSession) +//! instead of constructing events directly. The session proves that every ID +//! belongs to the current service revision and converts raw effects into +//! validated [`ResidentOperation`](crate::ResidentOperation) values. use serde::{Deserialize, Serialize}; @@ -46,10 +51,15 @@ impl ResourceId { #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum StreamCloseReason { + /// The remote peer completed or reset the connection. PeerClosed, + /// The platform or Component requested an orderly close. HostClosed, + /// The Host driver enforced its configured inactivity timeout. IdleTimeout, + /// A protocol driver rejected malformed or out-of-order bytes. ProtocolError, + /// The operating system or transport driver reported an I/O failure. TransportError, } @@ -57,44 +67,77 @@ pub enum StreamCloseReason { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum ResidentEvent { + /// A registered Host timer reached its deadline. Timer { + /// Timer resource previously registered with the session. timer_id: ResourceId, + /// Intended firing instant on the Host monotonic clock, in nanoseconds. + /// + /// This is not wall-clock time and may be earlier than delivery when + /// the Actor mailbox is busy. scheduled_at_ns: u64, }, + /// A TCP or Unix listener accepted a new byte stream. StreamOpened { + /// Listener resource that accepted the connection. endpoint_id: ResourceId, + /// Newly allocated stream resource scoped to this Actor revision. stream_id: ResourceId, + /// Driver-formatted peer address, when the transport provides one. peer: Option, }, + /// A bounded chunk was read from an open stream. StreamData { + /// Stream that produced the bytes. stream_id: ResourceId, + /// Raw transport bytes; framing remains the Component or protocol driver's job. bytes: Vec, }, + /// Backpressure cleared and the stream can accept another write. StreamWritable { + /// Stream whose driver write buffer has capacity. stream_id: ResourceId, }, + /// The peer sent EOF but the local write half remains available. StreamHalfClosed { + /// Stream entering the peer-half-closed state. stream_id: ResourceId, }, + /// A stream reached terminal state and will not produce more events. StreamClosed { + /// Stream being released after this callback. stream_id: ResourceId, + /// Terminal reason reported by the Host driver. reason: StreamCloseReason, }, + /// One UDP packet arrived on a registered datagram endpoint. Datagram { + /// UDP endpoint that received the packet. endpoint_id: ResourceId, + /// Driver-formatted source address used for a possible reply. peer: String, + /// Complete datagram payload; datagram boundaries are preserved. bytes: Vec, }, + /// A broker or internal queue delivered one message. Message { + /// Host-owned subscription that delivered the message. subscription_id: ResourceId, + /// Driver-scoped delivery identity used when acknowledging. message_id: u64, + /// Opaque message body. bytes: Vec, }, + /// An event arrived from a separately versioned extension driver. Source { + /// Registered extension source, such as a file watcher or serial port. source_id: ResourceId, + /// Capability-defined event name. event: String, + /// Capability-defined binary payload. payload: Vec, }, + /// The deployment is draining and no later events should be accepted. Shutdown, } @@ -105,39 +148,65 @@ pub enum ResidentEvent { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum ResidentEffect { + /// Queue bytes for an open stream. WriteStream { + /// Stream to write; it must belong to the current session. stream_id: ResourceId, + /// One bounded write chunk. bytes: Vec, }, + /// Ask the driver to close both halves of a stream. CloseStream { + /// Stream to transition into `closing`. stream_id: ResourceId, }, + /// Stop delivering read events until a matching resume effect. PauseStream { + /// Stream whose read side should be paused. stream_id: ResourceId, }, + /// Resume read delivery after Component backpressure clears. ResumeStream { + /// Previously paused stream. stream_id: ResourceId, }, + /// Send one UDP packet while preserving datagram boundaries. SendDatagram { + /// Registered UDP endpoint used to send the packet. endpoint_id: ResourceId, + /// Driver-formatted destination address. peer: String, + /// Complete bounded datagram payload. bytes: Vec, }, + /// Schedule or reschedule a registered Host timer. ArmTimer { + /// Timer resource to schedule. timer_id: ResourceId, + /// Delay from the time the Host applies this effect. delay_ms: u64, + /// Optional repeat period; `None` creates a one-shot timer. interval_ms: Option, }, + /// Cancel a registered timer. CancelTimer { + /// Timer whose pending occurrence should be removed. timer_id: ResourceId, }, + /// Confirm successful processing of a delivered message. AcknowledgeMessage { + /// Subscription that produced the delivery. subscription_id: ResourceId, + /// Exact message identity received in [`ResidentEvent::Message`]. message_id: u64, }, + /// Send a capability-defined command to an extension driver. SourceCommand { + /// Extension source that should execute the command. source_id: ResourceId, + /// Capability-defined command name. command: String, + /// Capability-defined bounded binary payload. payload: Vec, }, } diff --git a/crates/wasmeld-runtime/src/resident_host.rs b/crates/wasmeld-runtime/src/resident_host.rs index 41923f2..c394fbb 100644 --- a/crates/wasmeld-runtime/src/resident_host.rs +++ b/crates/wasmeld-runtime/src/resident_host.rs @@ -86,8 +86,13 @@ impl NetworkScope { /// between validation and use. #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] pub struct ResidentPolicy { + /// Addresses on which TCP listener drivers may bind. pub tcp_listen: NetworkScope, + /// Addresses on which UDP drivers may bind. pub udp_bind: NetworkScope, + /// Canonical parent directories allowed to contain Unix listener sockets. + /// + /// An empty list disables Unix endpoints. pub unix_listen_roots: Vec, } @@ -99,9 +104,27 @@ pub struct ResidentPolicy { #[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 }, + /// Host-owned TCP listener that produces byte-stream resources. + Tcp { + /// Unique operator-facing name within the session. + name: String, + /// Numeric address checked against [`ResidentPolicy::tcp_listen`]. + bind: SocketAddr, + }, + /// Host-owned UDP socket that preserves datagram boundaries. + Udp { + /// Unique operator-facing name within the session. + name: String, + /// Numeric address checked against [`ResidentPolicy::udp_bind`]. + bind: SocketAddr, + }, + /// Host-owned Unix stream listener. + Unix { + /// Unique operator-facing name within the session. + name: String, + /// Absolute socket path beneath an allowed canonical root. + path: PathBuf, + }, } impl ResidentEndpoint { @@ -116,12 +139,19 @@ impl ResidentEndpoint { #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ResidentResourceKind { + /// TCP listener endpoint. TcpEndpoint, + /// UDP datagram endpoint. UdpEndpoint, + /// Unix stream listener endpoint. UnixEndpoint, + /// Accepted TCP or Unix byte stream. Stream, + /// Host scheduler timer. Timer, + /// Queue, broker, or internal mailbox subscription. MessageSubscription, + /// Separately versioned extension driver. Source, } @@ -129,20 +159,30 @@ pub enum ResidentResourceKind { #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ResidentResourceState { + /// Non-stream resource is registered and available to its driver. Ready, + /// Stream accepts read and write activity. Open, + /// Component backpressure has temporarily disabled read delivery. ReadPaused, + /// Peer sent EOF; local writes are still permitted. PeerHalfClosed, + /// Close was requested and only the terminal close event remains valid. Closing, } /// Management-safe snapshot of one Host-owned resource. #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct ResidentResourceInfo { + /// Opaque identity scoped to one Actor revision. pub id: ResourceId, + /// Operator-facing name used in status and diagnostics. pub name: String, + /// Stable resource category. pub kind: ResidentResourceKind, + /// Current state after all validated effects have been applied. pub state: ResidentResourceState, + /// Driver correlation details that do not expose operating-system handles. pub metadata: ResidentResourceMetadata, } @@ -150,11 +190,28 @@ pub struct ResidentResourceInfo { #[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 }, + /// Complete management configuration for a listener or datagram endpoint. + Endpoint { + /// Validated endpoint definition. + endpoint: ResidentEndpoint, + }, + /// Parent listener of an accepted stream. + Stream { + /// TCP or Unix endpoint that accepted this stream. + endpoint_id: ResourceId, + }, + /// Current scheduler status for a timer. + Timer { + /// Whether an arm operation is waiting to fire. + armed: bool, + }, + /// Queue or broker details remain private to its driver. MessageSubscription, - Source { source_kind: String }, + /// Extension source identity. + Source { + /// Versioned capability name understood by the Component and driver. + source_kind: String, + }, } /// An operation that passed payload, ownership, kind, and state validation. @@ -164,39 +221,65 @@ pub enum ResidentResourceMetadata { #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum ResidentOperation { + /// Queue a bounded chunk on an open stream. WriteStream { + /// Target stream resource. stream_id: ResourceId, + /// Bytes to enqueue in order. bytes: Vec, }, + /// Close a stream and later report its terminal close event. CloseStream { + /// Target stream resource. stream_id: ResourceId, }, + /// Disable driver reads without closing the stream. PauseStream { + /// Target stream resource. stream_id: ResourceId, }, + /// Re-enable reads on a previously paused stream. ResumeStream { + /// Target stream resource. stream_id: ResourceId, }, + /// Send one complete UDP datagram. SendDatagram { + /// UDP endpoint used for the send. endpoint_id: ResourceId, + /// Driver-formatted destination address. peer: String, + /// Complete datagram payload. bytes: Vec, }, + /// Create or replace a timer schedule. ArmTimer { + /// Registered timer resource. timer_id: ResourceId, + /// Delay measured from operation application. delay_ms: u64, + /// Optional repeating interval. interval_ms: Option, }, + /// Remove a pending timer schedule. CancelTimer { + /// Registered timer resource. timer_id: ResourceId, }, + /// Acknowledge a delivered message after Component processing succeeds. AcknowledgeMessage { + /// Subscription that produced the message. subscription_id: ResourceId, + /// Driver-scoped delivery identity. message_id: u64, }, + /// Invoke a separately versioned extension driver command. SourceCommand { + /// Target extension source. source_id: ResourceId, + /// Capability-defined command name. command: String, + /// Capability-defined binary payload. payload: Vec, }, } @@ -204,51 +287,78 @@ pub enum ResidentOperation { /// Host resource, policy, or Actor failure while processing resident work. #[derive(Debug, Error)] pub enum ResidentHostError { + /// Actor dispatch, deadline, trap, or Component contract failure. #[error(transparent)] Runtime(#[from] RuntimeError), + /// An event or resource registration was attempted after shutdown began. #[error("resident session for {0} is shutting down")] ShuttingDown(ServiceKey), + /// The Actor revision reached its configured Host-resource count. #[error("resident resource limit for {service} is {limit}")] - ResourceLimit { service: ServiceKey, limit: usize }, + ResourceLimit { + /// Actor revision that owns the full resource table. + service: ServiceKey, + /// Maximum simultaneous Host resources. + limit: usize, + }, + /// The monotonic `u64` ID space was exhausted. #[error("resident resource ids exhausted for {0}")] ResourceIdsExhausted(ServiceKey), + /// An operator-facing endpoint or source name was empty. #[error("resident resource name must not be empty")] EmptyResourceName, + /// A named top-level resource conflicts with an existing name. #[error("resident resource name {0} is already registered")] DuplicateResourceName(String), + /// An endpoint cannot be released until all accepted streams are released. #[error("resident resource {resource_id} for {service} still owns active streams")] ResourceInUse { + /// Owning Actor revision. service: ServiceKey, + /// Endpoint that still has child streams. resource_id: u64, }, + /// An ID is unknown, released, or belongs to another session. #[error("resident resource {resource_id} does not belong to {service}")] UnknownResource { + /// Session in which the lookup was attempted. service: ServiceKey, + /// Unknown opaque ID. resource_id: u64, }, + /// An operation addressed the wrong category of resource. #[error("resident resource {resource_id} for {service} is {actual:?}, expected {expected}")] WrongResourceKind { + /// Owning Actor revision. service: ServiceKey, + /// Addressed opaque ID. resource_id: u64, + /// Actual category stored for the ID. actual: ResidentResourceKind, + /// Human-readable category required by the operation. expected: &'static str, }, + /// The resource exists but the requested transition is not legal. #[error("resident resource {resource_id} for {service} is in invalid state {state:?}")] InvalidResourceState { + /// Owning Actor revision. service: ServiceKey, + /// Addressed opaque ID. resource_id: u64, + /// State that rejected the transition. state: ResidentResourceState, }, + /// A bind address or Unix path falls outside platform policy. #[error("resident endpoint denied by policy: {0}")] EndpointDenied(String), } @@ -333,6 +443,47 @@ impl From for ResidentResourceState { /// 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. +/// +/// # Driver example +/// +/// ```no_run +/// use std::net::SocketAddr; +/// use wasmeld_runtime::{ +/// ActorHandle, NetworkScope, ResidentEndpoint, ResidentOperation, +/// ResidentPolicy, ResidentSession, +/// }; +/// +/// # let actor: ActorHandle = todo!("start a resident Actor with Runtime::start"); +/// let policy = ResidentPolicy { +/// tcp_listen: NetworkScope::Loopback, +/// ..ResidentPolicy::default() +/// }; +/// let mut session = ResidentSession::new(actor, policy)?; +/// let endpoint_id = session.register_endpoint(ResidentEndpoint::Tcp { +/// name: "local-api".into(), +/// bind: "127.0.0.1:9000".parse::()?, +/// })?; +/// +/// // The Tokio driver binds only after policy validation. For every accepted +/// // socket it allocates a stream, then applies returned operations in order. +/// let (stream_id, operations) = +/// session.accept_stream(endpoint_id, Some("127.0.0.1:51000".into()))?; +/// for operation in operations { +/// match operation { +/// ResidentOperation::WriteStream { stream_id, bytes } => { +/// // Queue `bytes` on the driver handle mapped to `stream_id`. +/// } +/// _ => { /* apply the other validated operation variants */ } +/// } +/// } +/// # let _ = stream_id; +/// # Ok::<(), Box>(()) +/// ``` +/// +/// The session is deliberately `&mut self`: one supervisor task must serialize +/// all driver events. Do not put separate per-socket tasks around independent +/// copies of its state, because resource transitions and Component memory must +/// observe the same total order. pub struct ResidentSession { actor: ActorHandle, policy: ResidentPolicy, @@ -343,6 +494,9 @@ pub struct ResidentSession { impl ResidentSession { /// Creates a resource session for an Actor exporting resident WIT 0.1.0. + /// + /// Resource IDs start at one and are never reused during this session. + /// Creating a new deployment revision requires a new session. pub fn new(actor: ActorHandle, policy: ResidentPolicy) -> Result { if actor.execution() != ComponentExecution::Resident { return Err(RuntimeError::NotResident(actor.key().clone()).into()); @@ -362,6 +516,9 @@ impl ResidentSession { } /// Registers a configured listener or datagram endpoint after policy checks. + /// + /// Registration does not bind an operating-system socket. The driver binds + /// after this succeeds and calls [`Self::release_resource`] if bind fails. pub fn register_endpoint( &mut self, endpoint: ResidentEndpoint, @@ -422,6 +579,9 @@ impl ResidentSession { } /// Allocates a stream for a connection accepted by a TCP or Unix driver. + /// + /// If Component dispatch fails, the allocation is rolled back and the + /// caller must close the accepted operating-system socket. pub fn accept_stream( &mut self, endpoint_id: ResourceId, @@ -457,6 +617,10 @@ impl ResidentSession { } /// Delivers one bounded stream read. Paused and closing streams reject data. + /// + /// Drivers must stop reading immediately after applying + /// [`ResidentOperation::PauseStream`]; queued data should remain in the + /// driver rather than bypassing Component backpressure. pub fn stream_data( &mut self, stream_id: ResourceId, @@ -490,6 +654,10 @@ impl ResidentSession { } /// Delivers terminal stream state and releases the resource identity. + /// + /// The identity is released even when the Component callback fails because + /// terminal transport state is already an external fact. IDs remain + /// non-reusable, so a late driver event is rejected as unknown. pub fn stream_closed( &mut self, stream_id: ResourceId, @@ -521,6 +689,10 @@ impl ResidentSession { } /// Delivers a timer occurrence. + /// + /// `scheduled_at_ns` is the driver's monotonic deadline, not wall time. The + /// timer is marked unarmed before dispatch; a returned arm operation creates + /// the next schedule. pub fn timer_fired( &mut self, timer_id: ResourceId, @@ -537,6 +709,9 @@ impl ResidentSession { } /// Delivers a broker or internal queue message. + /// + /// A driver acknowledges externally only after it receives and successfully + /// applies [`ResidentOperation::AcknowledgeMessage`]. pub fn message( &mut self, subscription_id: ResourceId, @@ -638,6 +813,8 @@ impl ResidentSession { ) -> 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. + // Drivers also receive operations only after this loop completes, so + // no external side effect can precede discovery of an invalid effect. let mut next_resources = self.resources.clone(); let mut operations = Vec::with_capacity(effects.len()); for effect in effects { diff --git a/crates/wasmeld-runtime/src/runtime.rs b/crates/wasmeld-runtime/src/runtime.rs index 2b8730e..2224006 100644 --- a/crates/wasmeld-runtime/src/runtime.rs +++ b/crates/wasmeld-runtime/src/runtime.rs @@ -64,12 +64,22 @@ const ALLOWED_WASI_IMPORTS: &[&str] = &[ #[derive(Clone, Debug)] pub struct RuntimeConfig { /// Frequency used to advance Wasmtime epoch interruption. + /// + /// Lower values make wall-time traps more precise but wake the ticker more + /// often. Service deadlines must be at least one tick. pub epoch_tick: Duration, /// Maximum native stack reservation for WebAssembly execution. + /// + /// This is separate from Component linear-memory limits. pub max_wasm_stack: usize, /// Optional persistent cache for compiled Component machine code. + /// + /// The directory is process-shared and contains derived artifacts only; it + /// may be deleted without losing registered packages. pub component_cache_dir: Option, /// Service-scoped storage used by Components importing the KV capability. + /// + /// Registration rejects a KV-importing Component when this is `None`. pub kv_backend: Option>, } @@ -227,6 +237,9 @@ impl Runtime { /// /// Imports are checked against the restricted WASI and explicit Wasmeld /// capability allowlist before the service becomes visible. + /// + /// Registration is immutable: another Component cannot replace the same + /// [`ServiceKey`]. Register a new revision and switch the deployment instead. pub fn register( &self, manifest: ServiceManifest, @@ -281,7 +294,11 @@ impl Runtime { self.register(manifest, bytes) } - /// Starts or returns the resident Actor for one registered service revision. + /// Starts or returns the Actor for one registered service revision. + /// + /// This operation is idempotent while the Actor is healthy. When an Actor + /// already exists, the supplied `init_config` is ignored because `init` ran + /// exactly once for that instance. pub fn start( &self, key: &ServiceKey, @@ -348,6 +365,10 @@ impl Runtime { } /// Enqueues one call on the service Actor's bounded serial mailbox. + /// + /// The deadline covers both time waiting behind earlier commands and Wasm + /// execution. A full mailbox returns [`RuntimeError::ActorOverloaded`] + /// immediately instead of blocking an unbounded number of callers. pub fn invoke(&self, key: &ServiceKey, input: Vec) -> Result, RuntimeError> { let actor = self .inner @@ -392,6 +413,10 @@ impl Runtime { } /// Delivers one Host event through the Actor's bounded serial mailbox. + /// + /// This low-level method returns raw Component effects. Network and source + /// drivers should use [`crate::ResidentSession`] so resource ownership and + /// state are validated before any effect is executed. pub fn dispatch_event( &self, key: &ServiceKey, @@ -409,6 +434,10 @@ impl Runtime { } /// Stops one Actor and releases its Store and Component Instance. + /// + /// Registered Component code remains available for a later [`Self::start`]. + /// A resident deployment should first deliver shutdown and close its Host + /// resources through [`crate::ResidentSession`]. pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> { let _lifecycle = self .inner @@ -435,6 +464,10 @@ impl Runtime { } /// Registers a new revision and immediately starts its Actor. + /// + /// This does not switch a deployment pointer or stop an older revision. If + /// Actor initialization fails, the new revision remains registered so the + /// management layer can inspect or explicitly unregister it. pub fn reload( &self, manifest: ServiceManifest, @@ -587,6 +620,10 @@ impl ActorHandle { } /// Sends an invocation and waits up to the remaining service deadline. + /// + /// Cloned handles compete for the same bounded mailbox. A Wasm trap is + /// fatal for the Actor, while a typed WIT `call-failed` result is returned + /// as [`RuntimeError::ComponentError`] and leaves the Actor available. pub fn invoke(&self, input: Vec) -> Result, RuntimeError> { if !self.is_available() { return Err(RuntimeError::ActorUnavailable(self.key.clone())); @@ -629,7 +666,10 @@ impl ActorHandle { } } - /// Sends a Host-owned resource event and waits for validated effects. + /// Sends a Host-owned resource event and waits for size-validated effects. + /// + /// Resource ownership and stream-state checks happen later in + /// [`crate::ResidentSession`]; drivers should not execute this raw result. pub fn dispatch_event( &self, event: ResidentEvent, @@ -669,6 +709,9 @@ impl ActorHandle { } /// Prevents new calls, asks the worker to stop, and waits for acknowledgement. + /// + /// Stop is not a graceful resident drain. The owner must first stop external + /// event production, deliver shutdown, and close Host resources. pub fn stop(&self) -> Result<(), RuntimeError> { if !self.status.accepting.swap(false, Ordering::AcqRel) { return Err(RuntimeError::ActorUnavailable(self.key.clone())); @@ -765,6 +808,10 @@ impl HostState { .build(); let mut wasi = WasiCtxBuilder::new(); + // Network access is denied twice: socket interfaces are absent from the + // linker allowlist, and the WASI context rejects TCP, UDP, DNS, and + // address checks. Resident network drivers live outside the Store and + // expose only opaque resource events. wasi.allow_tcp(false) .allow_udp(false) .allow_ip_name_lookup(false) @@ -839,6 +886,8 @@ impl ActorWorker { add_restricted_wasi(&mut linker)?; CapabilityRegistry::add_to_linker(&mut linker, &service.capabilities)?; + // Fuel limits deterministic instruction work; epoch interruption limits + // elapsed wall time. Both are reset before init and every later call. let epoch_ticks = ticks_for(limits.deadline(), epoch_tick); configure_call_budget(&mut store, &limits, epoch_ticks)?; let instance = linker @@ -1082,6 +1131,8 @@ fn spawn_actor( }) .map_err(RuntimeError::ActorThread)?; + // Do not publish a handle until instantiation and guest init completed. + // This prevents callers from observing an Actor that cannot serve work. ready_receiver.recv().map_err(|_| { RuntimeError::ActorInitialization("actor thread stopped during initialization".to_owned()) })??; @@ -1160,6 +1211,8 @@ fn run_actor(worker: &mut ActorWorker, receiver: Receiver, status: } fn drain_after_failure(receiver: Receiver, key: &ServiceKey) { + // A trap invalidates the Store. Explicitly answer queued callers so they do + // not wait for their individual deadlines after the worker exits. while let Ok(command) = receiver.try_recv() { match command { ActorCommand::Invoke {