diff --git a/Cargo.toml b/Cargo.toml index 15556b2..ef84553 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ sha2 = "0.10.9" thiserror = "2.0.17" toasty = "=0.9.0" toasty-driver-turso = "=0.9.0" -tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] } +tokio = { version = "1.53.1", features = ["io-util", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } toml = "0.9.8" tower-http = { version = "0.7.0", features = ["cors", "trace"] } tracing = "0.1.44" diff --git a/components/resident-probe/src/lib.rs b/components/resident-probe/src/lib.rs index f63f891..1413b9f 100644 --- a/components/resident-probe/src/lib.rs +++ b/components/resident-probe/src/lib.rs @@ -70,9 +70,12 @@ impl ResidentGuest for ResidentProbe { command: source.kind, payload: source.payload, })], + // A TCP FIN is a read-half close. This fixture has no additional + // response to send, so it explicitly asks the Host to close the + // remaining write half and release the stream resource. + Event::StreamHalfClosed(stream_id) => vec![Effect::CloseStream(stream_id)], Event::StreamOpened(_) | Event::StreamWritable(_) - | Event::StreamHalfClosed(_) | Event::StreamClosed(_) | Event::Shutdown => Vec::new(), }) diff --git a/crates/wasmeld-console/src/api.rs b/crates/wasmeld-console/src/api.rs index ec7a83b..55ffd22 100644 --- a/crates/wasmeld-console/src/api.rs +++ b/crates/wasmeld-console/src/api.rs @@ -585,7 +585,8 @@ impl IntoResponse for GatewayError { | ConsoleError::InvalidDatabaseData(_) | ConsoleError::Runtime(_) | ConsoleError::Join(_) - | ConsoleError::LockPoisoned => ( + | ConsoleError::LockPoisoned + | ConsoleError::ResidentSupervisor(_) => ( StatusCode::INTERNAL_SERVER_ERROR, "internal_error", "internal gateway error".to_owned(), @@ -680,7 +681,10 @@ impl IntoResponse for ApiError { | ConsoleError::WitRegistry(WitRegistryError::Storage { .. }) | ConsoleError::WitRegistry(WitRegistryError::LockPoisoned) | ConsoleError::Join(_) - | ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"), + | ConsoleError::LockPoisoned + | ConsoleError::ResidentSupervisor(_) => { + (StatusCode::INTERNAL_SERVER_ERROR, "internal_error") + } }; ( diff --git a/crates/wasmeld-console/src/lib.rs b/crates/wasmeld-console/src/lib.rs index 4e8719f..80f367c 100644 --- a/crates/wasmeld-console/src/lib.rs +++ b/crates/wasmeld-console/src/lib.rs @@ -36,11 +36,12 @@ mod api; mod invocation_persistence; mod kv_backend; mod persistence; +mod resident_supervisor; mod web; mod wit_registry; use std::{ - collections::{BTreeMap, VecDeque}, + collections::{BTreeMap, HashMap, VecDeque}, fs, io, path::{Path, PathBuf}, sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard}, @@ -54,14 +55,19 @@ use wasmeld_package::{ ComponentPackage, PackageError, read_package, wit_package::WitPackageMetadata, }; use wasmeld_runtime::{ - CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey, - ServiceManifest, + ActorHandle, CapabilityDescriptor, ComponentExecution, ResourceLimits, Runtime, RuntimeConfig, + RuntimeError, ServiceKey, ServiceManifest, }; pub use api::{app, gateway_app}; use invocation_persistence::InvocationPersistence; use kv_backend::ToastyKvBackend; use persistence::{InvocationUpdate, Persistence, StoredDeployment, StoredEvent, StoredService}; +pub use resident_supervisor::{ + ResidentServiceConfig, ResidentTcpListenerConfig, ResidentTimerConfig, +}; +use resident_supervisor::{ResidentSupervisor, ResidentSupervisorError}; +pub use wasmeld_runtime::{NetworkScope, ResidentPolicy}; use wit_registry::WitRegistry; pub use wit_registry::WitRegistryError; @@ -87,6 +93,12 @@ pub struct ConsoleConfig { pub registration_limits: ResourceLimits, /// Wasmtime Engine and epoch settings. pub runtime: RuntimeConfig, + /// Host-owned resident resources keyed by stable service ID. + /// + /// Each started revision receives an independent copy of this + /// configuration and therefore never shares resource IDs, socket handles, + /// or driver tasks with another revision. + pub resident_services: BTreeMap, } impl Default for ConsoleConfig { @@ -99,6 +111,7 @@ impl Default for ConsoleConfig { max_wit_package_bytes: DEFAULT_MAX_WIT_PACKAGE_BYTES, registration_limits: ResourceLimits::default(), runtime: RuntimeConfig::default(), + resident_services: BTreeMap::new(), } } } @@ -115,6 +128,9 @@ pub struct Console { // to avoid deadlocks between HTTP control operations. runtime: RwLock>, runtime_config: RuntimeConfig, + async_runtime: tokio::runtime::Handle, + resident_services: BTreeMap, + resident_supervisors: Mutex>, artifact_dir: PathBuf, wit_registry: WitRegistry, max_artifact_bytes: usize, @@ -486,6 +502,10 @@ pub enum ConsoleError { /// An internal lifecycle, state, or Runtime lock was poisoned. #[error("console state lock was poisoned")] LockPoisoned, + + /// A resident Host driver could not start, dispatch, or drain. + #[error("resident supervisor failed: {0}")] + ResidentSupervisor(String), } impl Console { @@ -495,6 +515,12 @@ impl Console { /// configuration. Their previous Store and linear memory are not restored. pub async fn new(config: ConsoleConfig) -> Result { config.registration_limits.validate()?; + for (service_id, resident) in &config.resident_services { + validate_path_segment("resident service id", service_id)?; + resident + .validate() + .map_err(ConsoleError::ResidentSupervisor)?; + } fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage { path: config.artifact_dir.clone(), source, @@ -547,6 +573,9 @@ impl Console { let console = Self { runtime: RwLock::new(Some(runtime)), runtime_config, + async_runtime: tokio::runtime::Handle::current(), + resident_services: config.resident_services, + resident_supervisors: Mutex::new(HashMap::new()), artifact_dir, wit_registry, max_artifact_bytes: config.max_artifact_bytes, @@ -671,6 +700,16 @@ impl Console { Ok(started) => { *runtime = Some(started); drop(runtime); + if let Err(error) = self.restore_deployments() { + self.rollback_runtime_start(); + self.mark_runtime_stopped()?; + self.push_event( + EventKind::Failed, + None, + format!("Wasmeld Runtime failed to restore deployments: {error}"), + )?; + return Err(error); + } self.mark_runtime_started()?; self.push_event( EventKind::Started, @@ -704,6 +743,7 @@ impl Console { drop(runtime); return self.runtime_view(); }; + self.shutdown_all_resident_supervisors(); let stopped = running.stop_all(); drop(runtime); self.mark_runtime_stopped()?; @@ -738,23 +778,34 @@ impl Console { .write() .map_err(|_| ConsoleError::LockPoisoned)?; - if let Some(running) = runtime.take() - && let Err(error) = running.stop_all() - { - drop(runtime); - self.mark_runtime_stopped()?; - self.push_event( - EventKind::Failed, - None, - format!("Wasmeld Runtime failed to restart while stopping actors: {error}"), - )?; - return Err(error.into()); + if let Some(running) = runtime.take() { + self.shutdown_all_resident_supervisors(); + if let Err(error) = running.stop_all() { + drop(runtime); + self.mark_runtime_stopped()?; + self.push_event( + EventKind::Failed, + None, + format!("Wasmeld Runtime failed to restart while stopping actors: {error}"), + )?; + return Err(error.into()); + } } match self.build_runtime() { Ok(started) => { *runtime = Some(started); drop(runtime); + if let Err(error) = self.restore_deployments() { + self.rollback_runtime_start(); + self.mark_runtime_stopped()?; + self.push_event( + EventKind::Failed, + None, + format!("Wasmeld Runtime failed to restore deployments: {error}"), + )?; + return Err(error); + } self.mark_runtime_started()?; self.push_event( EventKind::Started, @@ -783,25 +834,120 @@ impl Console { .values() .map(|service| service.manifest.clone()) .collect::>(); - let deployments = state - .deployments - .iter() - .map(|(service_id, deployment)| { - ServiceKey::new(service_id.clone(), deployment.active_revision.clone()) - }) - .collect::, _>>()?; drop(state); let runtime = Runtime::new(self.runtime_config.clone())?; for manifest in manifests { runtime.register_from_file(manifest)?; } - for key in deployments { - runtime.start(&key, Vec::new())?; - } Ok(runtime) } + fn start_actor( + &self, + runtime: &Runtime, + key: &ServiceKey, + init_config: Vec, + ) -> Result { + let actor = runtime.start(key, init_config)?; + if let Err(error) = self.attach_resident_supervisor(actor.clone()) { + // A resident Actor without its Host driver would be running in a + // misleading half-state. Roll it back before surfacing the error. + let _ = runtime.stop(key); + return Err(error); + } + Ok(actor) + } + + fn attach_resident_supervisor(&self, actor: ActorHandle) -> Result<(), ConsoleError> { + if actor.execution() != ComponentExecution::Resident { + return Ok(()); + } + let key = actor.key().clone(); + let mut supervisors = self + .resident_supervisors + .lock() + .map_err(|_| ConsoleError::LockPoisoned)?; + if supervisors + .get(&key) + .is_some_and(|supervisor| supervisor.owns_actor(&actor)) + { + return Ok(()); + } + if let Some(mut stale) = supervisors.remove(&key) + && let Err(error) = stale.shutdown() + { + tracing::warn!( + service = %key, + %error, + "stale resident supervisor did not drain cleanly before Actor replacement" + ); + } + let config = self + .resident_services + .get(key.id()) + .cloned() + .unwrap_or_default(); + let command_capacity = actor.mailbox_capacity(); + let supervisor = + ResidentSupervisor::start(actor, config, self.async_runtime.clone(), command_capacity) + .map_err(supervisor_error)?; + supervisors.insert(key, supervisor); + Ok(()) + } + + fn shutdown_resident_supervisor(&self, key: &ServiceKey) { + let supervisor = self + .resident_supervisors + .lock() + .map(|mut supervisors| supervisors.remove(key)); + match supervisor { + Ok(Some(mut supervisor)) => { + if let Err(error) = supervisor.shutdown() { + tracing::warn!(service = %key, %error, "resident supervisor did not drain cleanly"); + } + } + Ok(None) => {} + Err(_) => tracing::warn!(service = %key, "resident supervisor lock was poisoned"), + } + } + + fn shutdown_all_resident_supervisors(&self) { + let supervisors = self + .resident_supervisors + .lock() + .map(|mut supervisors| std::mem::take(&mut *supervisors)); + match supervisors { + Ok(supervisors) => { + for (key, mut supervisor) in supervisors { + if let Err(error) = supervisor.shutdown() { + tracing::warn!(service = %key, %error, "resident supervisor did not drain cleanly"); + } + } + } + Err(_) => tracing::warn!("resident supervisor lock was poisoned"), + } + } + + fn stop_actor(&self, runtime: &Runtime, key: &ServiceKey) -> Result<(), RuntimeError> { + self.shutdown_resident_supervisor(key); + runtime.stop(key) + } + + fn rollback_runtime_start(&self) { + self.shutdown_all_resident_supervisors(); + match self.runtime.write() { + Ok(mut runtime) => { + if let Some(runtime) = runtime.take() + && let Err(error) = runtime.stop_all() + { + tracing::warn!(%error, "failed to stop Runtime during start rollback"); + } + } + Err(_) => tracing::warn!("Runtime lock was poisoned during start rollback"), + } + } + fn mark_runtime_started(&self) -> Result<(), ConsoleError> { let mut state = self.state()?; let deployed = state @@ -931,6 +1077,7 @@ impl Console { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; + self.shutdown_resident_supervisor(key); runtime.unregister(key)?; self.state()?.services.remove(key); @@ -956,7 +1103,7 @@ impl Console { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; self.ensure_managed(key)?; - runtime.start(key, init_config)?; + self.start_actor(runtime, key, init_config)?; let view = self.update_status(key, ServiceStatus::Running)?; self.push_event(EventKind::Started, Some(key), "actor started".to_owned())?; Ok(view) @@ -966,7 +1113,7 @@ impl Console { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; self.ensure_managed(key)?; - runtime.stop(key)?; + self.stop_actor(runtime, key)?; let view = self.update_status(key, ServiceStatus::Stopped)?; self.push_event(EventKind::Stopped, Some(key), "actor stopped".to_owned())?; Ok(view) @@ -976,11 +1123,11 @@ impl Console { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; self.ensure_managed(key)?; - match runtime.stop(key) { + match self.stop_actor(runtime, key) { Ok(()) | Err(RuntimeError::ActorUnavailable(_)) => {} Err(error) => return Err(error.into()), } - runtime.start(key, init_config)?; + self.start_actor(runtime, key, init_config)?; let view = self.update_status(key, ServiceStatus::Running)?; self.push_event(EventKind::Started, Some(key), "actor restarted".to_owned())?; Ok(view) @@ -1006,7 +1153,7 @@ impl Console { // Starting is idempotent. The old revision stays resident so calls // that resolved immediately before this switch can finish safely and // rollback does not require recompilation. - runtime.start(&key, init_config)?; + self.start_actor(runtime, &key, init_config)?; let updated_at_ms = unix_time_ms(); let view = { @@ -1221,7 +1368,7 @@ impl Console { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; for key in &deployments { - runtime.start(key, Vec::new())?; + self.start_actor(runtime, key, Vec::new())?; } } @@ -1362,6 +1509,22 @@ impl Console { } } +impl Drop for Console { + fn drop(&mut self) { + self.shutdown_all_resident_supervisors(); + if let Ok(runtime) = self.runtime.get_mut() + && let Some(runtime) = runtime.take() + && let Err(error) = runtime.stop_all() + { + tracing::warn!(%error, "failed to stop Runtime while dropping Console"); + } + } +} + +fn supervisor_error(error: ResidentSupervisorError) -> ConsoleError { + ConsoleError::ResidentSupervisor(error.to_string()) +} + fn runtime_capabilities( runtime: &Runtime, key: &ServiceKey, diff --git a/crates/wasmeld-console/src/resident_supervisor.rs b/crates/wasmeld-console/src/resident_supervisor.rs new file mode 100644 index 0000000..0cc34e2 --- /dev/null +++ b/crates/wasmeld-console/src/resident_supervisor.rs @@ -0,0 +1,877 @@ +//! Revision-scoped execution of Host-owned resident event sources. +//! +//! The Wasmtime Actor API is intentionally synchronous: every call waits for a +//! bounded mailbox response. Running [`ResidentSession`] directly from a Tokio +//! task would therefore block an async worker for up to the Component deadline. +//! This module keeps one session on a dedicated supervisor thread. Async +//! drivers only wait for external activity and use a bounded channel to submit +//! facts such as a timer deadline, accepted TCP connection, or stream chunk. +//! +//! ```text +//! timer / listener / stream task --> bounded queue --> supervisor thread +//! ^ | +//! |-------- validated operations --------| +//! ``` +//! +//! The supervisor is scoped to one immutable service revision. Shutdown first +//! prevents new driver events, then delivers the resident `shutdown` callback, +//! aborts every Host task, closes socket handles, releases Host resource IDs, +//! and only then allows the Console to stop the Actor. + +mod tcp; + +use std::{ + collections::{HashMap, VecDeque}, + io, + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + mpsc::{self, Receiver, SyncSender, TrySendError}, + }, + thread::{self, JoinHandle as ThreadJoinHandle}, + time::{Duration, Instant}, +}; + +use thiserror::Error; +use tokio::{net::TcpStream, runtime::Handle, task::JoinHandle as TaskJoinHandle}; +use wasmeld_runtime::{ + ActorHandle, ResidentEndpoint, ResidentHostError, ResidentOperation, ResidentPolicy, + ResidentSession, ResourceId, ServiceKey, StreamCloseReason, +}; + +pub use tcp::ResidentTcpListenerConfig; +use tcp::{TcpDriverError, TcpEndpointDriver, TcpStreamCommand, TcpStreamDriver}; + +/// Platform configuration for Host resources attached to one service ID. +/// +/// A configuration is inherited by every started revision of that service, but +/// each revision receives independent Host handles and resource IDs. +#[derive(Clone, Debug, Default)] +pub struct ResidentServiceConfig { + /// Defense-in-depth policy checked before any Host endpoint is bound. + /// + /// The default denies every network endpoint. Adding a TCP listener also + /// requires setting `tcp_listen` to `loopback` or `any`. + pub policy: ResidentPolicy, + /// Host-owned TCP listeners registered when a resident Actor starts. + pub tcp_listeners: Vec, + /// Host-owned timers registered when a resident Actor starts. + pub timers: Vec, +} + +impl ResidentServiceConfig { + pub(crate) fn validate(&self) -> Result<(), String> { + let mut names = std::collections::BTreeSet::new(); + for listener in &self.tcp_listeners { + listener.validate()?; + self.policy + .validate_endpoint(&ResidentEndpoint::Tcp { + name: listener.name.clone(), + bind: listener.bind, + }) + .map_err(|error| error.to_string())?; + if !names.insert(listener.name.as_str()) { + return Err(format!( + "resident resource name {:?} is duplicated", + listener.name + )); + } + } + for timer in &self.timers { + timer.validate()?; + if !names.insert(timer.name.as_str()) { + return Err(format!( + "resident resource name {:?} is duplicated", + timer.name + )); + } + } + Ok(()) + } +} + +/// One management-owned timer source for a resident Component. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ResidentTimerConfig { + /// Stable operator-facing name, unique within one service configuration. + pub name: String, + /// Delay before the first event after the Actor starts. + pub initial_delay_ms: u64, + /// Optional Host-managed repeat period. + /// + /// A Component may replace this schedule by returning `arm-timer` or stop + /// it with `cancel-timer`. + pub interval_ms: Option, +} + +impl ResidentTimerConfig { + fn validate(&self) -> Result<(), String> { + if self.name.trim().is_empty() { + return Err("resident timer name must not be empty".to_owned()); + } + if self.initial_delay_ms == 0 { + return Err(format!( + "resident timer {:?} initial_delay_ms must be greater than zero", + self.name + )); + } + if self.interval_ms == Some(0) { + return Err(format!( + "resident timer {:?} interval_ms must be greater than zero", + self.name + )); + } + Ok(()) + } +} + +/// Failure while creating, driving, or draining a resident session. +#[derive(Debug, Error)] +pub(crate) enum ResidentSupervisorError { + #[error(transparent)] + Host(#[from] ResidentHostError), + + #[error("resident driver queue capacity must be greater than zero")] + EmptyQueue, + + #[error("failed to start resident supervisor thread: {0}")] + ThreadStart(#[source] io::Error), + + #[error("resident supervisor for {0} stopped accepting commands")] + ChannelClosed(ServiceKey), + + #[error("resident supervisor for {0} panicked")] + WorkerPanicked(ServiceKey), + + #[error("resident timer operation referenced unknown resource {0}")] + UnknownTimer(u64), + + #[error("resident stream operation referenced unknown resource {0}")] + UnknownStream(u64), + + #[error(transparent)] + Tcp(#[from] TcpDriverError), + + #[error("resident driver cannot apply {0} before its driver is installed")] + UnsupportedOperation(&'static str), +} + +/// Clone-free owner for one revision's resident session and driver tasks. +pub(crate) struct ResidentSupervisor { + key: ServiceKey, + actor: ActorHandle, + sender: SyncSender, + accepting_events: Arc, + worker: Option>, +} + +impl ResidentSupervisor { + pub(crate) fn start( + actor: ActorHandle, + config: ResidentServiceConfig, + async_runtime: Handle, + command_capacity: usize, + ) -> Result { + if command_capacity == 0 { + return Err(ResidentSupervisorError::EmptyQueue); + } + let key = actor.key().clone(); + let session = ResidentSession::new(actor.clone(), config.policy.clone())?; + let (sender, receiver) = mpsc::sync_channel(command_capacity); + let accepting_events = Arc::new(AtomicBool::new(true)); + let core = SupervisorCore::new( + session, + config, + async_runtime, + sender.clone(), + accepting_events.clone(), + command_capacity, + Instant::now(), + )?; + let thread_name = format!("wasmeld-resident-{key}"); + let worker_key = key.clone(); + let worker = thread::Builder::new() + .name(thread_name) + .spawn(move || run_supervisor(worker_key, core, receiver)) + .map_err(ResidentSupervisorError::ThreadStart)?; + Ok(Self { + key, + actor, + sender, + accepting_events, + worker: Some(worker), + }) + } + + pub(crate) fn owns_actor(&self, actor: &ActorHandle) -> bool { + self.actor.is_same_instance(actor) + } + + /// Drains this revision and cancels all Host listener, stream, and timer tasks. + pub(crate) fn shutdown(&mut self) -> Result<(), ResidentSupervisorError> { + let Some(worker) = self.worker.take() else { + return Ok(()); + }; + // Async event producers observe this before retrying a full bounded + // queue, so the control-plane shutdown command cannot be starved by + // network traffic. + self.accepting_events.store(false, Ordering::Release); + let (response_sender, response_receiver) = mpsc::sync_channel(1); + let send_result = self + .sender + .send(SupervisorCommand::Shutdown { response_sender }); + let response = match send_result { + Ok(()) => response_receiver + .recv() + .map_err(|_| ResidentSupervisorError::ChannelClosed(self.key.clone()))?, + Err(_) => Err(ResidentSupervisorError::ChannelClosed(self.key.clone())), + }; + worker + .join() + .map_err(|_| ResidentSupervisorError::WorkerPanicked(self.key.clone()))?; + response + } +} + +impl Drop for ResidentSupervisor { + fn drop(&mut self) { + if let Err(error) = self.shutdown() { + tracing::warn!(service = %self.key, %error, "resident supervisor drop could not drain cleanly"); + } + } +} + +enum SupervisorCommand { + TcpAccepted { + endpoint_id: ResourceId, + stream: TcpStream, + peer: SocketAddr, + }, + StreamData { + stream_id: ResourceId, + bytes: Vec, + }, + StreamHalfClosed { + stream_id: ResourceId, + }, + StreamClosed { + stream_id: ResourceId, + reason: StreamCloseReason, + }, + TimerFired { + timer_id: ResourceId, + generation: u64, + scheduled_at_ns: u64, + }, + Shutdown { + response_sender: SyncSender>, + }, +} + +struct SupervisorCore { + session: ResidentSession, + async_runtime: Handle, + sender: SyncSender, + accepting_events: Arc, + stream_command_capacity: usize, + stream_read_bytes: usize, + origin: Instant, + tcp_endpoints: HashMap, + streams: HashMap, + timers: HashMap, +} + +impl SupervisorCore { + fn new( + session: ResidentSession, + config: ResidentServiceConfig, + async_runtime: Handle, + sender: SyncSender, + accepting_events: Arc, + stream_command_capacity: usize, + origin: Instant, + ) -> Result { + let stream_read_bytes = session.max_stream_chunk_bytes(); + let mut core = Self { + session, + async_runtime, + sender, + accepting_events, + stream_command_capacity, + stream_read_bytes, + origin, + tcp_endpoints: HashMap::new(), + streams: HashMap::new(), + timers: HashMap::new(), + }; + for listener in config.tcp_listeners { + core.register_tcp_listener(listener)?; + } + for timer in config.timers { + let timer_id = core.session.register_timer(timer.name)?; + core.timers.insert(timer_id, TimerDriver::default()); + core.arm_timer(timer_id, timer.initial_delay_ms, timer.interval_ms)?; + } + Ok(core) + } + + fn register_tcp_listener( + &mut self, + config: ResidentTcpListenerConfig, + ) -> Result<(), ResidentSupervisorError> { + let endpoint_id = self.session.register_endpoint(ResidentEndpoint::Tcp { + name: config.name.clone(), + bind: config.bind, + })?; + + let driver = match TcpEndpointDriver::start( + &config, + self.session.owner().clone(), + endpoint_id, + &self.async_runtime, + self.sender.clone(), + self.accepting_events.clone(), + ) { + Ok(driver) => driver, + Err(error) => { + let _ = self.session.release_resource(endpoint_id); + return Err(error.into()); + } + }; + self.tcp_endpoints.insert(endpoint_id, driver); + Ok(()) + } + + fn tcp_accepted( + &mut self, + endpoint_id: ResourceId, + stream: TcpStream, + peer: SocketAddr, + ) -> Result<(), ResidentSupervisorError> { + let (stream_id, operations) = self + .session + .accept_stream(endpoint_id, Some(peer.to_string()))?; + let driver = TcpStreamDriver::start( + self.session.owner().clone(), + stream_id, + stream, + self.stream_read_bytes, + self.stream_command_capacity, + &self.async_runtime, + self.sender.clone(), + self.accepting_events.clone(), + ); + self.streams.insert(stream_id, driver); + if let Err(error) = self.apply_operations(operations) { + self.discard_stream(stream_id); + return Err(error); + } + Ok(()) + } + + fn stream_data( + &mut self, + stream_id: ResourceId, + bytes: Vec, + ) -> Result<(), ResidentSupervisorError> { + if !self.streams.contains_key(&stream_id) { + return Ok(()); + } + let result = self + .session + .stream_data(stream_id, bytes) + .map_err(ResidentSupervisorError::from) + .and_then(|operations| self.apply_operations(operations)); + if result.is_err() { + self.discard_stream(stream_id); + } + result + } + + fn stream_half_closed(&mut self, stream_id: ResourceId) -> Result<(), ResidentSupervisorError> { + if !self.streams.contains_key(&stream_id) { + return Ok(()); + } + let result = self + .session + .stream_half_closed(stream_id) + .map_err(ResidentSupervisorError::from) + .and_then(|operations| self.apply_operations(operations)); + if result.is_err() { + self.discard_stream(stream_id); + } + result + } + + fn stream_closed( + &mut self, + stream_id: ResourceId, + reason: StreamCloseReason, + ) -> Result<(), ResidentSupervisorError> { + let Some(driver) = self.streams.remove(&stream_id) else { + // A task can race with Host-initiated termination. Resource IDs are + // never reused, so a late terminal fact is safe to ignore. + return Ok(()); + }; + drop(driver); + let operations = self.session.stream_closed(stream_id, reason)?; + self.apply_operations(operations) + } + + fn discard_stream(&mut self, stream_id: ResourceId) { + self.streams.remove(&stream_id); + let _ = self.session.release_resource(stream_id); + } + + fn timer_fired( + &mut self, + timer_id: ResourceId, + generation: u64, + scheduled_at_ns: u64, + ) -> Result<(), ResidentSupervisorError> { + let interval_ms = { + let timer = self + .timers + .get_mut(&timer_id) + .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?; + if timer.generation != generation { + return Ok(()); + } + timer.task = None; + timer.interval_ms + }; + let operations = self.session.timer_fired(timer_id, scheduled_at_ns)?; + let schedule_replaced = operations.iter().any(|operation| { + matches!( + operation, + ResidentOperation::ArmTimer { + timer_id: operation_timer, + .. + } | ResidentOperation::CancelTimer { + timer_id: operation_timer + } if *operation_timer == timer_id + ) + }); + self.apply_operations(operations)?; + if !schedule_replaced && let Some(interval_ms) = interval_ms { + self.arm_timer(timer_id, interval_ms, Some(interval_ms))?; + } + Ok(()) + } + + fn apply_operations( + &mut self, + operations: Vec, + ) -> Result<(), ResidentSupervisorError> { + let mut pending = VecDeque::from(operations); + while let Some(operation) = pending.pop_front() { + match operation { + ResidentOperation::ArmTimer { + timer_id, + delay_ms, + interval_ms, + } => self.arm_timer(timer_id, delay_ms, interval_ms)?, + ResidentOperation::CancelTimer { timer_id } => self.cancel_timer(timer_id)?, + ResidentOperation::WriteStream { stream_id, bytes } => { + if !self.queue_stream_command(stream_id, TcpStreamCommand::Write(bytes))? { + pending.extend( + self.terminate_stream(stream_id, StreamCloseReason::TransportError)?, + ); + } + } + ResidentOperation::CloseStream { stream_id } => { + if !self.queue_stream_command(stream_id, TcpStreamCommand::Close)? { + pending.extend( + self.terminate_stream(stream_id, StreamCloseReason::HostClosed)?, + ); + } + } + ResidentOperation::PauseStream { stream_id } => { + if !self.queue_stream_command(stream_id, TcpStreamCommand::Pause)? { + pending.extend( + self.terminate_stream(stream_id, StreamCloseReason::TransportError)?, + ); + } + } + ResidentOperation::ResumeStream { stream_id } => { + if !self.queue_stream_command(stream_id, TcpStreamCommand::Resume)? { + pending.extend( + self.terminate_stream(stream_id, StreamCloseReason::TransportError)?, + ); + } + } + ResidentOperation::SendDatagram { .. } => { + return Err(ResidentSupervisorError::UnsupportedOperation( + "send-datagram", + )); + } + ResidentOperation::AcknowledgeMessage { .. } => { + return Err(ResidentSupervisorError::UnsupportedOperation( + "acknowledge-message", + )); + } + ResidentOperation::SourceCommand { .. } => { + return Err(ResidentSupervisorError::UnsupportedOperation( + "source-command", + )); + } + } + } + Ok(()) + } + + /// Queues one bounded operation without ever blocking the supervisor. + /// + /// A full per-stream queue means the peer is not consuming output quickly + /// enough. The caller terminates that stream instead of stalling unrelated + /// connections or allowing an unbounded Host buffer. + fn queue_stream_command( + &self, + stream_id: ResourceId, + command: TcpStreamCommand, + ) -> Result { + let driver = self + .streams + .get(&stream_id) + .ok_or(ResidentSupervisorError::UnknownStream(stream_id.get()))?; + match driver.try_send(command) { + Ok(()) => Ok(true), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + tracing::warn!( + service = %self.session.owner(), + resource_id = stream_id.get(), + "resident TCP stream closed because its write/control queue is full" + ); + Ok(false) + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => Ok(false), + } + } + + fn terminate_stream( + &mut self, + stream_id: ResourceId, + reason: StreamCloseReason, + ) -> Result, ResidentSupervisorError> { + let driver = self + .streams + .remove(&stream_id) + .ok_or(ResidentSupervisorError::UnknownStream(stream_id.get()))?; + drop(driver); + Ok(self.session.stream_closed(stream_id, reason)?) + } + + fn arm_timer( + &mut self, + timer_id: ResourceId, + delay_ms: u64, + interval_ms: Option, + ) -> Result<(), ResidentSupervisorError> { + let timer = self + .timers + .get_mut(&timer_id) + .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?; + timer.cancel_task(); + timer.generation = next_generation(timer.generation); + timer.interval_ms = interval_ms; + let generation = timer.generation; + let delay = Duration::from_millis(delay_ms); + let scheduled_at_ns = u64::try_from(self.origin.elapsed().saturating_add(delay).as_nanos()) + .unwrap_or(u64::MAX); + let sender = self.sender.clone(); + let accepting_events = self.accepting_events.clone(); + let owner = self.session.owner().clone(); + timer.task = Some(self.async_runtime.spawn(async move { + tokio::time::sleep(delay).await; + if !accepting_events.load(Ordering::Acquire) { + return; + } + match sender.try_send(SupervisorCommand::TimerFired { + timer_id, + generation, + scheduled_at_ns, + }) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + tracing::warn!( + service = %owner, + resource_id = timer_id.get(), + "resident timer occurrence dropped because the supervisor queue is full" + ); + } + Err(TrySendError::Disconnected(_)) => {} + } + })); + Ok(()) + } + + fn cancel_timer(&mut self, timer_id: ResourceId) -> Result<(), ResidentSupervisorError> { + let timer = self + .timers + .get_mut(&timer_id) + .ok_or(ResidentSupervisorError::UnknownTimer(timer_id.get()))?; + timer.cancel_task(); + timer.generation = next_generation(timer.generation); + timer.interval_ms = None; + Ok(()) + } + + fn shutdown(&mut self) -> Result<(), ResidentSupervisorError> { + self.accepting_events.store(false, Ordering::Release); + for endpoint in self.tcp_endpoints.values_mut() { + endpoint.abort(); + } + for timer in self.timers.values_mut() { + timer.cancel_task(); + } + + let mut first_error = match self.session.shutdown() { + Ok(operations) => self.apply_operations(operations).err(), + Err(error) => Some(error.into()), + }; + + // A shutdown callback may return stream or timer operations. They have + // been validated, but no source may remain active beyond this edge. + let stream_ids = self.streams.keys().copied().collect::>(); + for stream_id in stream_ids { + self.streams.remove(&stream_id); + if let Err(error) = self.session.release_resource(stream_id) + && first_error.is_none() + { + first_error = Some(error.into()); + } + } + self.streams.clear(); + + let endpoint_ids = self.tcp_endpoints.keys().copied().collect::>(); + for endpoint_id in endpoint_ids { + self.tcp_endpoints.remove(&endpoint_id); + if let Err(error) = self.session.release_resource(endpoint_id) + && first_error.is_none() + { + first_error = Some(error.into()); + } + } + self.tcp_endpoints.clear(); + + for timer in self.timers.values_mut() { + timer.cancel_task(); + } + let timer_ids = self.timers.keys().copied().collect::>(); + for timer_id in timer_ids { + if let Err(error) = self.session.release_resource(timer_id) + && first_error.is_none() + { + first_error = Some(error.into()); + } + } + self.timers.clear(); + + match first_error { + Some(error) => Err(error), + None => Ok(()), + } + } +} + +fn run_supervisor( + key: ServiceKey, + mut core: SupervisorCore, + receiver: Receiver, +) { + while let Ok(command) = receiver.recv() { + match command { + SupervisorCommand::TcpAccepted { + endpoint_id, + stream, + peer, + } => { + if let Err(error) = core.tcp_accepted(endpoint_id, stream, peer) { + tracing::warn!( + service = %key, + resource_id = endpoint_id.get(), + %peer, + %error, + "resident TCP connection was rejected" + ); + } + } + SupervisorCommand::StreamData { stream_id, bytes } => { + if let Err(error) = core.stream_data(stream_id, bytes) { + tracing::warn!( + service = %key, + resource_id = stream_id.get(), + %error, + "resident TCP stream data failed and the connection was closed" + ); + } + } + SupervisorCommand::StreamHalfClosed { stream_id } => { + if let Err(error) = core.stream_half_closed(stream_id) { + tracing::warn!( + service = %key, + resource_id = stream_id.get(), + %error, + "resident TCP half-close event failed and the connection was closed" + ); + } + } + SupervisorCommand::StreamClosed { stream_id, reason } => { + if let Err(error) = core.stream_closed(stream_id, reason) { + tracing::warn!( + service = %key, + resource_id = stream_id.get(), + %error, + "resident TCP terminal event failed" + ); + } + } + SupervisorCommand::TimerFired { + timer_id, + generation, + scheduled_at_ns, + } => { + if let Err(error) = core.timer_fired(timer_id, generation, scheduled_at_ns) { + let _ = core.cancel_timer(timer_id); + tracing::warn!( + service = %key, + resource_id = timer_id.get(), + %error, + "resident timer event failed and was cancelled" + ); + } + } + SupervisorCommand::Shutdown { response_sender } => { + let result = core.shutdown(); + let _ = response_sender.send(result); + return; + } + } + } + + if let Err(error) = core.shutdown() { + tracing::warn!(service = %key, %error, "resident supervisor channel closed during drain"); + } +} + +#[derive(Default)] +struct TimerDriver { + generation: u64, + interval_ms: Option, + task: Option>, +} + +impl TimerDriver { + fn cancel_task(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + } +} + +impl Drop for TimerDriver { + fn drop(&mut self) { + self.cancel_task(); + } +} + +fn next_generation(current: u64) -> u64 { + let next = current.wrapping_add(1); + if next == 0 { 1 } else { next } +} + +#[cfg(test)] +mod tests { + use super::{ + ResidentServiceConfig, ResidentTcpListenerConfig, ResidentTimerConfig, next_generation, + }; + + #[test] + fn rejects_invalid_timer_configuration() { + for timer in [ + ResidentTimerConfig { + name: String::new(), + initial_delay_ms: 1, + interval_ms: None, + }, + ResidentTimerConfig { + name: "zero-delay".to_owned(), + initial_delay_ms: 0, + interval_ms: None, + }, + ResidentTimerConfig { + name: "zero-interval".to_owned(), + initial_delay_ms: 1, + interval_ms: Some(0), + }, + ] { + assert!( + ResidentServiceConfig { + timers: vec![timer], + ..ResidentServiceConfig::default() + } + .validate() + .is_err() + ); + } + } + + #[test] + fn rejects_duplicate_timer_names() { + let timer = ResidentTimerConfig { + name: "heartbeat".to_owned(), + initial_delay_ms: 1, + interval_ms: None, + }; + assert!( + ResidentServiceConfig { + timers: vec![timer.clone(), timer], + ..ResidentServiceConfig::default() + } + .validate() + .is_err() + ); + } + + #[test] + fn rejects_names_shared_by_listener_and_timer() { + assert!( + ResidentServiceConfig { + tcp_listeners: vec![ResidentTcpListenerConfig { + name: "events".to_owned(), + bind: "127.0.0.1:9000".parse().unwrap(), + }], + timers: vec![ResidentTimerConfig { + name: "events".to_owned(), + initial_delay_ms: 1, + interval_ms: None, + }], + ..ResidentServiceConfig::default() + } + .validate() + .is_err() + ); + } + + #[test] + fn rejects_tcp_listener_outside_default_policy() { + assert!( + ResidentServiceConfig { + tcp_listeners: vec![ResidentTcpListenerConfig { + name: "denied".to_owned(), + bind: "127.0.0.1:9000".parse().unwrap(), + }], + ..ResidentServiceConfig::default() + } + .validate() + .is_err() + ); + } + + #[test] + fn timer_generation_never_uses_zero() { + assert_eq!(next_generation(0), 1); + assert_eq!(next_generation(u64::MAX), 1); + } +} diff --git a/crates/wasmeld-console/src/resident_supervisor/tcp.rs b/crates/wasmeld-console/src/resident_supervisor/tcp.rs new file mode 100644 index 0000000..dc3f663 --- /dev/null +++ b/crates/wasmeld-console/src/resident_supervisor/tcp.rs @@ -0,0 +1,402 @@ +//! Tokio TCP transport owned by one resident supervisor. +//! +//! This module owns operating-system socket handles and knows nothing about +//! Wasm bindings or Component effects. It reports accepted connections and +//! bounded stream facts through [`SupervisorCommand`], while the parent +//! supervisor validates resource identity and translates Component operations +//! back into [`TcpStreamCommand`] values. + +use std::{ + io, + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + mpsc::{SyncSender, TrySendError}, + }, + time::Duration, +}; + +use thiserror::Error; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpSocket, TcpStream}, + runtime::Handle, + sync::mpsc, + task::JoinHandle, +}; +use wasmeld_runtime::{ResourceId, ServiceKey, StreamCloseReason}; + +use super::SupervisorCommand; + +/// One management-owned TCP listener for a resident Component. +/// +/// The address must be numeric; DNS and socket handles are never exposed to +/// the Component. Port zero is accepted for tests and private embedding, but a +/// fixed configured port is required when external callers must discover it. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ResidentTcpListenerConfig { + /// Stable operator-facing name, unique among all resources for the service. + pub name: String, + /// Numeric address bound by the Host after policy validation. + pub bind: SocketAddr, +} + +impl ResidentTcpListenerConfig { + pub(super) fn validate(&self) -> Result<(), String> { + if self.name.trim().is_empty() { + return Err("resident TCP listener name must not be empty".to_owned()); + } + Ok(()) + } +} + +#[derive(Debug, Error)] +pub(crate) enum TcpDriverError { + #[error("failed to {action} resident TCP listener {name:?} at {bind}: {source}")] + Listener { + name: String, + bind: SocketAddr, + action: &'static str, + #[source] + source: io::Error, + }, +} + +pub(super) struct TcpEndpointDriver { + task: JoinHandle<()>, +} + +impl TcpEndpointDriver { + pub(super) fn start( + config: &ResidentTcpListenerConfig, + owner: ServiceKey, + endpoint_id: ResourceId, + async_runtime: &Handle, + sender: SyncSender, + accepting_events: Arc, + ) -> Result { + let listener = bind_listener(config, async_runtime)?; + let name = config.name.clone(); + let task = async_runtime.spawn(async move { + run_listener(owner, endpoint_id, name, listener, sender, accepting_events).await; + }); + Ok(Self { task }) + } + + pub(super) fn abort(&mut self) { + self.task.abort(); + } +} + +impl Drop for TcpEndpointDriver { + fn drop(&mut self) { + self.abort(); + } +} + +pub(super) struct TcpStreamDriver { + commands: mpsc::Sender, + task: JoinHandle<()>, +} + +impl TcpStreamDriver { + #[allow(clippy::too_many_arguments)] + pub(super) fn start( + owner: ServiceKey, + stream_id: ResourceId, + stream: TcpStream, + read_bytes: usize, + command_capacity: usize, + async_runtime: &Handle, + sender: SyncSender, + accepting_events: Arc, + ) -> Self { + let (commands, command_receiver) = mpsc::channel(command_capacity); + let task = async_runtime.spawn(async move { + run_stream( + owner, + stream_id, + stream, + read_bytes, + command_receiver, + sender, + accepting_events, + ) + .await; + }); + Self { commands, task } + } + + pub(super) fn try_send( + &self, + command: TcpStreamCommand, + ) -> Result<(), mpsc::error::TrySendError> { + self.commands.try_send(command) + } +} + +impl Drop for TcpStreamDriver { + fn drop(&mut self) { + // Aborting the task drops both owned socket halves. The session owner + // removes the resource identity separately and in deterministic order. + self.task.abort(); + } +} + +pub(super) enum TcpStreamCommand { + Write(Vec), + Close, + Pause, + Resume, +} + +fn bind_listener( + config: &ResidentTcpListenerConfig, + async_runtime: &Handle, +) -> Result { + // TcpSocket gives the platform an explicit backlog and SO_REUSEADDR, which + // makes an immediate service restart able to reclaim its stable address + // after existing connections are closed. + let _runtime_context = async_runtime.enter(); + let socket = if config.bind.is_ipv4() { + TcpSocket::new_v4() + } else { + TcpSocket::new_v6() + } + .map_err(|source| listener_error(config, "create socket for", source))?; + socket + .set_reuseaddr(true) + .map_err(|source| listener_error(config, "configure", source))?; + socket + .bind(config.bind) + .map_err(|source| listener_error(config, "bind", source))?; + socket + .listen(1024) + .map_err(|source| listener_error(config, "listen on", source)) +} + +async fn run_listener( + owner: ServiceKey, + endpoint_id: ResourceId, + name: String, + listener: TcpListener, + sender: SyncSender, + accepting_events: Arc, +) { + while accepting_events.load(Ordering::Acquire) { + match listener.accept().await { + Ok((stream, peer)) => { + if let Err(error) = stream.set_nodelay(true) { + tracing::warn!( + service = %owner, + endpoint = %name, + %peer, + %error, + "resident TCP connection rejected because TCP_NODELAY could not be set" + ); + continue; + } + if !send_stream_event( + &sender, + &accepting_events, + SupervisorCommand::TcpAccepted { + endpoint_id, + stream, + peer, + }, + ) + .await + { + return; + } + } + Err(error) => { + if !accepting_events.load(Ordering::Acquire) { + return; + } + tracing::warn!( + service = %owner, + endpoint = %name, + %error, + "resident TCP accept failed; listener will retry" + ); + // Persistent descriptor exhaustion must not turn into a hot + // retry loop. A healthy listener resumes after the backoff. + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + } +} + +async fn run_stream( + owner: ServiceKey, + stream_id: ResourceId, + stream: TcpStream, + read_bytes: usize, + mut commands: mpsc::Receiver, + sender: SyncSender, + accepting_events: Arc, +) { + let (mut reader, mut writer) = stream.into_split(); + let mut buffer = vec![0_u8; read_bytes]; + let mut read_paused = false; + let mut peer_half_closed = false; + + loop { + if !accepting_events.load(Ordering::Acquire) { + return; + } + + if read_paused || peer_half_closed { + let Some(command) = commands.recv().await else { + return; + }; + if let Some(reason) = apply_stream_command(command, &mut writer, &mut read_paused).await + { + let _ = send_stream_terminal(&sender, &accepting_events, stream_id, reason).await; + return; + } + continue; + } + + tokio::select! { + biased; + command = commands.recv() => { + let Some(command) = command else { + return; + }; + if let Some(reason) = + apply_stream_command(command, &mut writer, &mut read_paused).await + { + let _ = + send_stream_terminal(&sender, &accepting_events, stream_id, reason).await; + return; + } + } + read = reader.read(&mut buffer) => { + match read { + Ok(0) => { + peer_half_closed = true; + if !send_stream_event( + &sender, + &accepting_events, + SupervisorCommand::StreamHalfClosed { stream_id }, + ) + .await + { + return; + } + } + Ok(read) => { + if !send_stream_event( + &sender, + &accepting_events, + SupervisorCommand::StreamData { + stream_id, + bytes: buffer[..read].to_vec(), + }, + ) + .await + { + return; + } + } + Err(error) => { + tracing::debug!( + service = %owner, + resource_id = stream_id.get(), + %error, + "resident TCP read failed" + ); + let _ = send_stream_terminal( + &sender, + &accepting_events, + stream_id, + StreamCloseReason::TransportError, + ) + .await; + return; + } + } + } + } + } +} + +async fn apply_stream_command( + command: TcpStreamCommand, + writer: &mut tokio::net::tcp::OwnedWriteHalf, + read_paused: &mut bool, +) -> Option { + match command { + TcpStreamCommand::Write(bytes) => writer + .write_all(&bytes) + .await + .err() + .map(|_| StreamCloseReason::TransportError), + TcpStreamCommand::Close => { + let _ = writer.shutdown().await; + Some(StreamCloseReason::HostClosed) + } + TcpStreamCommand::Pause => { + *read_paused = true; + None + } + TcpStreamCommand::Resume => { + *read_paused = false; + None + } + } +} + +async fn send_stream_terminal( + sender: &SyncSender, + accepting_events: &AtomicBool, + stream_id: ResourceId, + reason: StreamCloseReason, +) -> bool { + send_stream_event( + sender, + accepting_events, + SupervisorCommand::StreamClosed { stream_id, reason }, + ) + .await +} + +/// Sends a non-lossy stream fact through the bounded supervisor queue. +/// +/// Unlike timer ticks, TCP chunks cannot be dropped. Waiting here naturally +/// stops the connection task from reading more bytes and propagates +/// backpressure into the kernel receive buffer. Shutdown flips +/// `accepting_events`, which cancels the retry without blocking Actor drain. +async fn send_stream_event( + sender: &SyncSender, + accepting_events: &AtomicBool, + mut command: SupervisorCommand, +) -> bool { + while accepting_events.load(Ordering::Acquire) { + match sender.try_send(command) { + Ok(()) => return true, + Err(TrySendError::Full(returned)) => { + command = returned; + tokio::time::sleep(Duration::from_millis(1)).await; + } + Err(TrySendError::Disconnected(_)) => return false, + } + } + false +} + +fn listener_error( + config: &ResidentTcpListenerConfig, + action: &'static str, + source: io::Error, +) -> TcpDriverError { + TcpDriverError::Listener { + name: config.name.clone(), + bind: config.bind, + action, + source, + } +} diff --git a/crates/wasmeld-console/tests/api.rs b/crates/wasmeld-console/tests/api.rs index a3c2a0c..c5a2c64 100644 --- a/crates/wasmeld-console/tests/api.rs +++ b/crates/wasmeld-console/tests/api.rs @@ -1,9 +1,12 @@ use std::{ + collections::BTreeMap, fs, io::Cursor, + net::{SocketAddr, TcpListener as StdTcpListener}, path::{Path, PathBuf}, process::Command, sync::{Arc, OnceLock}, + time::Duration, }; use axum::{ @@ -14,8 +17,15 @@ use axum::{ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use serde_json::{Value, json}; use tempfile::TempDir; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, +}; use tower::ServiceExt; -use wasmeld_console::{Console, ConsoleConfig, app, gateway_app}; +use wasmeld_console::{ + Console, ConsoleConfig, NetworkScope, ResidentPolicy, ResidentServiceConfig, + ResidentTcpListenerConfig, ResidentTimerConfig, app, gateway_app, +}; use wasmeld_package::{ module::{ModuleLock, sync_dependencies}, wit_package::build_wit_package, @@ -90,6 +100,192 @@ async fn manages_a_resident_component_over_http() { assert!(events["events"].as_array().unwrap().len() >= 5); } +#[tokio::test] +async fn drives_resident_timers_across_service_and_runtime_restarts() { + let artifact_dir = TempDir::new().expect("temporary artifact directory"); + let mut resident_services = BTreeMap::new(); + resident_services.insert( + "resident-probe".to_owned(), + ResidentServiceConfig { + timers: vec![ResidentTimerConfig { + name: "heartbeat".to_owned(), + initial_delay_ms: 5, + interval_ms: None, + }], + ..ResidentServiceConfig::default() + }, + ); + let console = Arc::new( + Console::new(ConsoleConfig { + artifact_dir: artifact_dir.path().to_path_buf(), + wit_registry_dir: artifact_dir.path().join("wit-packages"), + database_path: artifact_dir.path().join("console.db"), + resident_services, + ..ConsoleConfig::default() + }) + .await + .expect("console should start"), + ); + let application = app(console, Vec::new()); + let component = fs::read(component_artifact("resident_probe_component.wasm")) + .expect("resident probe component should be readable"); + + let response = application + .clone() + .oneshot(package_request_with_world( + "resident-probe", + "0.1.0", + "component:resident-probe/resident-probe-component@0.1.0", + &component, + )) + .await + .expect("resident package registration should complete"); + assert_eq!(response.status(), StatusCode::CREATED); + + let response = application + .clone() + .oneshot(json_request( + "/api/v1/deployments/resident-probe/activate", + json!({ "revision": "0.1.0" }), + )) + .await + .expect("resident deployment activation should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert!( + wait_for_resident_count(&application).await > 0, + "configured timer should reach the resident Component" + ); + + let response = application + .clone() + .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/stop")) + .await + .expect("resident service stop should complete"); + assert_eq!(response.status(), StatusCode::OK); + + let response = application + .clone() + .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/restart")) + .await + .expect("resident service restart should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert!( + wait_for_resident_count(&application).await > 0, + "service restart should create a new timer session" + ); + + let response = application + .clone() + .oneshot(empty_post("/api/v1/runtime/restart")) + .await + .expect("Runtime restart should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert!( + wait_for_resident_count(&application).await > 0, + "deployment restore should recreate the timer supervisor" + ); +} + +#[tokio::test] +async fn drives_tcp_streams_across_service_and_runtime_restarts() { + let artifact_dir = TempDir::new().expect("temporary artifact directory"); + let occupied_listener = + StdTcpListener::bind("127.0.0.1:0").expect("occupied test listener should bind"); + let address = occupied_listener + .local_addr() + .expect("occupied test listener should have an address"); + let mut resident_services = BTreeMap::new(); + resident_services.insert( + "resident-probe".to_owned(), + ResidentServiceConfig { + policy: ResidentPolicy { + tcp_listen: NetworkScope::Loopback, + ..ResidentPolicy::default() + }, + tcp_listeners: vec![ResidentTcpListenerConfig { + name: "echo".to_owned(), + bind: address, + }], + ..ResidentServiceConfig::default() + }, + ); + let console = Arc::new( + Console::new(ConsoleConfig { + artifact_dir: artifact_dir.path().to_path_buf(), + wit_registry_dir: artifact_dir.path().join("wit-packages"), + database_path: artifact_dir.path().join("console.db"), + resident_services, + ..ConsoleConfig::default() + }) + .await + .expect("console should start"), + ); + let application = app(console, Vec::new()); + let component = fs::read(component_artifact("resident_probe_component.wasm")) + .expect("resident probe component should be readable"); + + let response = application + .clone() + .oneshot(package_request_with_world( + "resident-probe", + "0.1.0", + "component:resident-probe/resident-probe-component@0.1.0", + &component, + )) + .await + .expect("resident package registration should complete"); + assert_eq!(response.status(), StatusCode::CREATED); + + let response = application + .clone() + .oneshot(json_request( + "/api/v1/deployments/resident-probe/activate", + json!({ "revision": "0.1.0" }), + )) + .await + .expect("conflicting resident deployment activation should complete"); + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + + // Driver startup failure must stop the Actor and leave deployment state + // untouched. Once the external conflict is gone, the same revision can + // start normally without repairing Runtime state manually. + drop(occupied_listener); + let response = application + .clone() + .oneshot(json_request( + "/api/v1/deployments/resident-probe/activate", + json!({ "revision": "0.1.0" }), + )) + .await + .expect("resident deployment activation should complete"); + assert_eq!(response.status(), StatusCode::OK); + let large_payload = vec![0xA5; 2 * 64 * 1024 + 17]; + assert_tcp_echo(address, &large_payload).await; + + let response = application + .clone() + .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/stop")) + .await + .expect("resident service stop should complete"); + assert_eq!(response.status(), StatusCode::OK); + + let response = application + .clone() + .oneshot(empty_post("/api/v1/services/resident-probe/0.1.0/restart")) + .await + .expect("resident service restart should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_tcp_echo(address, b"service restart").await; + + let response = application + .clone() + .oneshot(empty_post("/api/v1/runtime/restart")) + .await + .expect("Runtime restart should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_tcp_echo(address, b"runtime restart").await; +} + #[tokio::test] async fn reports_exact_component_host_capabilities() { let artifact_dir = TempDir::new().expect("temporary artifact directory"); @@ -999,6 +1195,53 @@ async fn invoke_service(application: &Router, id: &str, revision: &str, input: & .unwrap() } +async fn wait_for_resident_count(application: &Router) -> u64 { + for _ in 0..100 { + let output = invoke_service(application, "resident-probe", "0.1.0", &[]).await; + let count = u64::from_le_bytes( + output + .try_into() + .expect("resident probe response should be a u64"), + ); + if count > 0 { + return count; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + 0 +} + +async fn assert_tcp_echo(address: SocketAddr, payload: &[u8]) { + let mut stream = connect_tcp(address).await; + stream + .write_all(payload) + .await + .expect("TCP request should be written"); + let mut echoed = vec![0_u8; payload.len()]; + tokio::time::timeout(Duration::from_secs(2), stream.read_exact(&mut echoed)) + .await + .expect("resident TCP echo should not time out") + .expect("resident TCP echo should be readable"); + assert_eq!(echoed, payload); +} + +async fn connect_tcp(address: SocketAddr) -> TcpStream { + let mut last_error = None; + for _ in 0..100 { + match TcpStream::connect(address).await { + Ok(stream) => return stream, + Err(error) => { + last_error = Some(error); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + } + panic!( + "resident TCP listener {address} did not accept connections: {}", + last_error.expect("at least one connection should have been attempted") + ); +} + fn echo_component() -> PathBuf { component_artifact("echo_component.wasm") } @@ -1010,6 +1253,7 @@ fn component_artifact(name: &str) -> PathBuf { "components/counter/Cargo.toml", "components/clock-probe/Cargo.toml", "components/kv-probe/Cargo.toml", + "components/resident-probe/Cargo.toml", ] { let status = Command::new("rustup") .args([