From 2f1438c129d3acc9a713d8cd936bfb81c463c5e2 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 21 Jul 2026 19:20:47 +0400 Subject: [PATCH 1/2] feat(runtime): add process-wide stream capacity --- CHANGELOG.md | 16 ++ Cargo.lock | 1 + crates/astrid-capsule-types/Cargo.toml | 1 + crates/astrid-capsule-types/src/lib.rs | 2 +- crates/astrid-capsule-types/src/limits.rs | 135 +++++++++++++++ .../src/engine/wasm/host/ipc.rs | 19 ++- .../src/engine/wasm/host/net/mod.rs | 53 +++--- .../src/engine/wasm/host/net/stream.rs | 94 +++++++--- .../src/engine/wasm/host/net/tcp_stream.rs | 26 +-- .../src/engine/wasm/host/net/unix_listener.rs | 160 +++++++++++++++--- .../src/engine/wasm/host_state.rs | 24 ++- .../src/engine/wasm/host_state_hook.rs | 3 + crates/astrid-capsule/src/engine/wasm/mod.rs | 19 +++ crates/astrid-capsule/src/engine/wasm/pool.rs | 8 + .../src/engine/wasm/test_fixtures.rs | 3 + crates/astrid-capsule/src/lib.rs | 5 +- crates/astrid-capsule/src/loader.rs | 34 +++- crates/astrid-config/src/defaults.toml | 5 + crates/astrid-config/src/env.rs | 5 + crates/astrid-config/src/merge/restrict.rs | 11 ++ crates/astrid-config/src/merge/tests.rs | 29 ++++ crates/astrid-config/src/types.rs | 5 + crates/astrid-config/src/validate.rs | 21 +++ crates/astrid-daemon/src/lib.rs | 46 ++++- crates/astrid-events/src/bus_tests.rs | 25 +++ crates/astrid-events/src/route/receiver.rs | 13 ++ crates/astrid-kernel/src/lib.rs | 31 +++- 27 files changed, 698 insertions(+), 96 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cd1fbcf4..396e5eb96 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,24 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ## [Unreleased] +### Added + +- **Host-derived, operator-tunable persistent network-stream capacity.** A + process-wide admission budget is shared by every capsule and pooled Store, + exposed through `capsule.host_net_streams`, + `ASTRID_CAPSULE_HOST_NET_STREAMS`, and `--host-net-streams`. The derived + default reserves file-descriptor headroom, live usage/limit gauges are + exported, and lowering the ceiling does not disrupt existing streams. +- **Real Component Model readiness for IPC, TCP/Unix streams, and the kernel + Unix listener.** Capsules can block on heterogeneous pollables without + timeout-scanning idle resources. + ### Fixed +- **Fragmented framed socket reads preserve partial decoder state.** A slow + peer can no longer split the four-byte length prefix across non-blocking read + windows and desynchronize the connection. + - **Stable crates publication installs its authenticated-hash prerequisite.** The protected publisher installs the pinned `b3sum` binary before validating the exact dev candidate, so BLAKE3 release metadata checks run before any diff --git a/Cargo.lock b/Cargo.lock index 33792f8d8..c2ff37375 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -528,6 +528,7 @@ version = "0.10.4" dependencies = [ "astrid-core", "dashmap", + "event-listener", "nix 0.31.3", "parking_lot", "semver", diff --git a/crates/astrid-capsule-types/Cargo.toml b/crates/astrid-capsule-types/Cargo.toml index 79c86a1eb..c14298ae9 100644 --- a/crates/astrid-capsule-types/Cargo.toml +++ b/crates/astrid-capsule-types/Cargo.toml @@ -12,6 +12,7 @@ description = "Engine-agnostic capsule types shared by all Astrid capsule engine web-time = "1.1.0" astrid-core = { workspace = true } dashmap = { workspace = true } +event-listener = "5.4.1" parking_lot = { workspace = true } semver = { workspace = true, features = ["serde"] } serde = { workspace = true } diff --git a/crates/astrid-capsule-types/src/lib.rs b/crates/astrid-capsule-types/src/lib.rs index c1c36d9fc..2200fccd6 100644 --- a/crates/astrid-capsule-types/src/lib.rs +++ b/crates/astrid-capsule-types/src/lib.rs @@ -19,5 +19,5 @@ pub mod memory_ledger; pub use capsule::CapsuleId; pub use error::{CapsuleError, CapsuleResult}; pub use fuel_ledger::{FuelLedger, FuelRateLimiter}; -pub use limits::{CapsuleRuntimeLimits, HttpLimits}; +pub use limits::{CapsuleRuntimeLimits, HttpLimits, NetStreamBudget, NetStreamLease}; pub use memory_ledger::MemoryLedger; diff --git a/crates/astrid-capsule-types/src/limits.rs b/crates/astrid-capsule-types/src/limits.rs index 3ae914a6c..d0f3a09e4 100644 --- a/crates/astrid-capsule-types/src/limits.rs +++ b/crates/astrid-capsule-types/src/limits.rs @@ -20,6 +20,10 @@ //! resolved by the caller (CLI > config file > env > host-derived default); a //! `None` override here means "use the host-derived default". +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; use std::thread::available_parallelism; use std::time::Duration; @@ -109,6 +113,113 @@ pub fn host_io_concurrency_default() -> usize { } } +/// Host-derived process-wide ceiling for persistent network streams. +/// +/// A stream occupies a file descriptor for its whole lifetime. Taking half of +/// the already fd-clamped async-I/O budget reserves at most one quarter of +/// `RLIMIT_NOFILE` for persistent capsule streams and leaves the rest for +/// in-flight I/O, listeners, storage, logs, and unrelated descriptors. +#[must_use] +pub fn host_net_stream_limit_default() -> usize { + (host_io_concurrency_default() / 2).max(1) +} + +/// Process-wide admission budget shared by every capsule engine and Store. +/// Lowering the limit is non-destructive: existing streams remain valid and +/// new admissions pause until usage falls below the new ceiling. +#[derive(Debug)] +pub struct NetStreamBudget { + limit: AtomicUsize, + active: AtomicUsize, + available: event_listener::Event, +} + +impl NetStreamBudget { + /// Construct a stream budget. Programmatic zero is clamped to one; config + /// and CLI validation reject explicit zero earlier with a clearer error. + #[must_use] + pub fn new(limit: usize) -> Self { + Self { + limit: AtomicUsize::new(limit.max(1)), + active: AtomicUsize::new(0), + available: event_listener::Event::new(), + } + } + + /// Atomically acquire one stream slot and return its RAII lease. + #[must_use] + pub fn try_acquire(self: &Arc) -> Option { + self.active + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| { + (active < self.limit.load(Ordering::Acquire)).then_some(active + 1) + }) + .ok()?; + Some(NetStreamLease { + budget: Arc::clone(self), + }) + } + + /// Change the admission ceiling without disrupting existing streams. + pub fn set_limit(&self, limit: usize) { + let limit = limit.max(1); + let previous = self.limit.swap(limit, Ordering::AcqRel); + if limit > previous { + self.available.notify(usize::MAX); + } + } + + /// Current admission ceiling. + #[must_use] + pub fn limit(&self) -> usize { + self.limit.load(Ordering::Acquire) + } + + /// Current number of admitted live streams. + #[must_use] + pub fn active(&self) -> usize { + self.active.load(Ordering::Acquire) + } + + /// Whether a new stream can be admitted at this instant. + #[must_use] + pub fn has_capacity(&self) -> bool { + self.active() < self.limit() + } + + /// Wait without polling until at least one stream slot may be available. + /// Callers must still use [`try_acquire`](Self::try_acquire): another task + /// may win the slot between this wake and admission. + pub async fn wait_available(&self) { + loop { + let listener = self.available.listen(); + if self.has_capacity() { + return; + } + listener.await; + } + } +} + +impl Default for NetStreamBudget { + fn default() -> Self { + Self::new(host_net_stream_limit_default()) + } +} + +/// RAII ownership of one slot in a [`NetStreamBudget`]. +#[derive(Debug)] +pub struct NetStreamLease { + budget: Arc, +} + +impl Drop for NetStreamLease { + fn drop(&mut self) { + let previous = self.budget.active.fetch_sub(1, Ordering::AcqRel); + debug_assert!(previous > 0, "net stream budget underflow"); + self.budget.available.notify(1); + } +} + /// Host-derived default for the dynamic instance pool's **max** size — the /// ceiling on a capsule's concurrent interceptor invocations. /// @@ -384,6 +495,30 @@ mod tests { } } + #[test] + fn net_stream_default_is_derived_from_fd_clamped_io_budget() { + assert_eq!( + host_net_stream_limit_default(), + (host_io_concurrency_default() / 2).max(1) + ); + } + + #[test] + fn net_stream_budget_is_atomic_raii_and_live_tunable() { + let budget = Arc::new(NetStreamBudget::new(2)); + let first = budget.try_acquire().expect("first stream"); + let second = budget.try_acquire().expect("second stream"); + assert!(budget.try_acquire().is_none()); + assert_eq!(budget.active(), 2); + + budget.set_limit(1); + drop(first); + assert!(budget.try_acquire().is_none()); + drop(second); + assert_eq!(budget.active(), 0); + assert!(budget.try_acquire().is_some()); + } + #[test] fn resolve_prefers_overrides_and_clamps_zero() { let r = CapsuleRuntimeLimits::resolve(Some(7), Some(900), Some(40)); diff --git a/crates/astrid-capsule/src/engine/wasm/host/ipc.rs b/crates/astrid-capsule/src/engine/wasm/host/ipc.rs index 8faa96fb8..577f13ff8 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/ipc.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/ipc.rs @@ -107,6 +107,13 @@ pub(super) struct SubscriptionEntry { pub(super) topic_pattern: String, } +#[async_trait::async_trait] +impl wasmtime_wasi::p2::Pollable for SubscriptionEntry { + async fn ready(&mut self) { + self.receiver.lock().await.ready().await; + } +} + /// Convert an `AstridEvent::Ipc` arc into the internal message clone the /// WIT translation layer expects. Returns `None` for non-IPC events; /// the routed demux already filters non-IPC, so this is just defensive. @@ -665,12 +672,12 @@ impl HostSubscription for HostState { result } - fn subscribe_readiness(&mut self, _self_: Resource) -> Resource { - // Real pollable wiring (sourced from the receiver's notify - // channel) lands with the dedicated pollable commit. Until - // then, hand out an always-ready sentinel so guests get a - // clean poll-then-recv loop rather than a host panic. - super::stubs::always_ready_pollable(&mut self.resource_table) + fn subscribe_readiness(&mut self, self_: Resource) -> Resource { + wasmtime_wasi::p2::subscribe( + &mut self.resource_table, + Resource::::new_borrow(self_.rep()), + ) + .unwrap_or_else(|_| Resource::new_own(0)) } fn drop(&mut self, rep: Resource) -> wasmtime::Result<()> { diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs index 3e9491725..d369af581 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -31,6 +31,7 @@ use std::sync::Arc; use wasmtime::component::Resource; +use wasmtime_wasi::p2::Pollable; use crate::audit_sink::{HostAuditEvent, HostAuditOutcome}; use crate::engine::wasm::bindings::astrid::net::host::{ @@ -50,12 +51,21 @@ mod unix_listener; use stream::CONNECT_TIMEOUT; -/// Maximum concurrent socket connections per capsule. Defense-in-depth -/// cap on top of the per-principal profile quota. Tracked via -/// [`HostState::net_stream_count`], bumped on every successful -/// `accept` / `connect-tcp` push and decremented in the resource -/// drop path. -pub(super) const MAX_ACTIVE_STREAMS: usize = 8; +#[async_trait::async_trait] +impl Pollable for NetStream { + async fn ready(&mut self) { + match self { + Self::Unix(stream) => { + let stream = stream.lock().await; + let _ = stream.readable().await; + }, + Self::Tcp(slot) => { + let stream = slot.stream.lock().await; + let _ = stream.readable().await; + }, + } + } +} /// Stamp marking a resource slot in the table as a `UnixListener` handle. /// The kernel pre-binds the listener; the resource handle is just a @@ -131,6 +141,15 @@ pub(super) fn audit_net( } } +pub(super) fn record_net_stream_metrics(state: &HostState) { + metrics::gauge!("astrid_capsule_net_streams_active").set(f64::from( + u32::try_from(state.net_stream_budget.active()).unwrap_or(u32::MAX), + )); + metrics::gauge!("astrid_capsule_net_streams_limit").set(f64::from( + u32::try_from(state.net_stream_budget.limit()).unwrap_or(u32::MAX), + )); +} + /// Audit an outbound TCP connect, carrying the destination host + port. /// /// Wraps the generic [`audit_net`] tracing line and additionally reports a @@ -324,11 +343,11 @@ impl net::Host for HostState { } } - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + let Some(stream_lease) = self.net_stream_budget.try_acquire() else { let result: Result, ErrorCode> = Err(ErrorCode::Quota); audit_net_connect(self, &host, port, &result); return result; - } + }; let rt_handle = self.runtime_handle.clone(); let blocking_semaphore = self.blocking_semaphore.clone(); @@ -377,13 +396,6 @@ impl net::Host for HostState { }, }; - if self.net_stream_count >= MAX_ACTIVE_STREAMS { - drop(stream); - let result: Result, ErrorCode> = Err(ErrorCode::Quota); - audit_net_connect(self, &host, port, &result); - return result; - } - let net_stream = NetStream::Tcp(TcpStreamSlot { stream: Arc::new(tokio::sync::Mutex::new(stream)), read_timeout: None, @@ -402,6 +414,12 @@ impl net::Host for HostState { return result; }, }; + let previous = self.net_stream_leases.insert(res.rep(), stream_lease); + debug_assert!( + previous.is_none(), + "net stream resource rep reused while live" + ); + record_net_stream_metrics(self); self.net_stream_count += 1; let result: Result, ErrorCode> = Ok(Resource::new_own(res.rep())); audit_net_connect(self, &host, port, &result); @@ -467,11 +485,6 @@ impl net::Host for HostState { mod tests { use super::*; - #[test] - fn max_active_streams_pinned() { - assert_eq!(MAX_ACTIVE_STREAMS, 8); - } - #[test] fn validate_host_accepts_normal_names() { assert!(validate_host("example.com").is_ok()); diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs index fd28b01c6..203e44ce9 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/stream.rs @@ -11,6 +11,8 @@ pub(super) const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::fro /// Host-side cap on a single byte-stream read/peek buffer. pub(super) const MAX_BYTES_PER_CALL: usize = 10 * 1024 * 1024; +use crate::engine::wasm::host_state::FrameReadState; + /// Returns true for IO errors that represent a normal peer disconnect. pub(super) fn is_peer_disconnect(e: &std::io::Error) -> bool { matches!( @@ -23,44 +25,66 @@ pub(super) fn is_peer_disconnect(e: &std::io::Error) -> bool { } /// Read one length-prefixed frame from `stream`. -pub(super) async fn read_frame(stream: &mut S) -> Result +pub(super) async fn read_frame( + stream: &mut S, + state: &mut FrameReadState, +) -> Result where S: tokio::io::AsyncRead + Unpin, { use tokio::io::AsyncReadExt; - let mut len_buf = [0u8; 4]; - match tokio::time::timeout( - std::time::Duration::from_millis(50), - stream.read_exact(&mut len_buf), - ) - .await - { + let header_result = tokio::time::timeout(std::time::Duration::from_millis(50), async { + while state.header_read < state.header.len() { + let read = stream.read(&mut state.header[state.header_read..]).await?; + if read == 0 { + return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof)); + } + state.header_read += read; + } + Ok::<(), std::io::Error>(()) + }) + .await; + match header_result { Err(_) => return Ok(NetReadStatus::Pending), Ok(Err(e)) if is_peer_disconnect(&e) => return Ok(NetReadStatus::Closed), Ok(Err(e)) => return Err(format!("socket read error: {e}")), - Ok(Ok(_)) => {}, + Ok(Ok(())) => {}, } - let len = u32::from_be_bytes(len_buf) as usize; - if len > MAX_BYTES_PER_CALL { - return Err("Payload too large (max 10MB)".to_string()); + if state.payload.is_empty() { + let len = u32::from_be_bytes(state.header) as usize; + if len > MAX_BYTES_PER_CALL { + *state = FrameReadState::default(); + return Err("Payload too large (max 10MB)".to_string()); + } + state.payload.resize(len, 0); } - let mut payload = vec![0u8; len]; - let timeout_ms = 5000 + (len as u64 / 1024); - match tokio::time::timeout( - std::time::Duration::from_millis(timeout_ms), - stream.read_exact(&mut payload), - ) - .await - { + let timeout_ms = 5000 + (state.payload.len() as u64 / 1024); + let payload_result = + tokio::time::timeout(std::time::Duration::from_millis(timeout_ms), async { + while state.payload_read < state.payload.len() { + let read = stream + .read(&mut state.payload[state.payload_read..]) + .await?; + if read == 0 { + return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof)); + } + state.payload_read += read; + } + Ok::<(), std::io::Error>(()) + }) + .await; + match payload_result { Err(_) => return Err("Payload read timed out".to_string()), Ok(Err(e)) if is_peer_disconnect(&e) => return Ok(NetReadStatus::Closed), Ok(Err(e)) => return Err(format!("socket payload read error: {e}")), - Ok(Ok(_)) => {}, + Ok(Ok(())) => {}, } + let payload = std::mem::take(&mut state.payload); + *state = FrameReadState::default(); Ok(NetReadStatus::Data(payload)) } @@ -195,7 +219,33 @@ mod tests { // and `read_frame` converts it to `NetReadStatus::Closed`. let (tx, mut rx) = tokio::io::duplex(64); drop(tx); - let status = read_frame(&mut rx).await.expect("classified, not error"); + let status = read_frame(&mut rx, &mut FrameReadState::default()) + .await + .expect("classified, not error"); assert!(matches!(status, NetReadStatus::Closed)); } + + #[tokio::test] + async fn read_frame_preserves_fragmented_header_across_pending() { + use tokio::io::AsyncWriteExt; + + let (mut tx, mut rx) = tokio::io::duplex(64); + let mut state = FrameReadState::default(); + tx.write_all(&[0]).await.expect("first header byte"); + + let pending = read_frame(&mut rx, &mut state) + .await + .expect("partial header is not an error"); + assert!(matches!(pending, NetReadStatus::Pending)); + assert_eq!(state.header_read, 1); + + tx.write_all(&[0, 0, 3, b'a', b'b', b'c']) + .await + .expect("remaining frame"); + let complete = read_frame(&mut rx, &mut state) + .await + .expect("fragmented frame completes"); + assert!(matches!(complete, NetReadStatus::Data(bytes) if bytes == b"abc")); + assert_eq!(state.header_read, 0); + } } diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs index 0821789f1..b8ae4ee2e 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_stream.rs @@ -46,7 +46,8 @@ fn write_deadline(stream: &NetStream, data_len: usize) -> Duration { } } use super::{ - HostState, NetStream, audit_net, map_io_err, net_stream, with_tcp_slot_mut, with_tcp_stream, + HostState, NetStream, audit_net, map_io_err, net_stream, record_net_stream_metrics, + with_tcp_slot_mut, with_tcp_stream, }; use crate::engine::wasm::bindings::astrid::io::streams::{InputStream, OutputStream}; use crate::engine::wasm::bindings::astrid::net::host::{ @@ -56,19 +57,21 @@ use crate::engine::wasm::host::util; impl HostTcpStream for HostState { fn read(&mut self, self_: Resource) -> Result { - let stream = net_stream(&self.resource_table, self_.rep())?; + let rep = self_.rep(); + let stream = net_stream(&self.resource_table, rep)?; let rt = self.runtime_handle.clone(); let sem = self.blocking_semaphore.clone(); let tok = self.effective_cancel_token(); + let frame_state = self.net_frame_states.entry(rep).or_default(); let status = util::bounded_block_on_cancellable(&rt, &sem, &tok, async { match stream { NetStream::Unix(arc) => { let mut s = arc.lock().await; - read_frame(&mut *s).await + read_frame(&mut *s, frame_state).await }, NetStream::Tcp(slot) => { let mut s = slot.stream.lock().await; - read_frame(&mut *s).await + read_frame(&mut *s, frame_state).await }, } }); @@ -490,12 +493,12 @@ impl HostTcpStream for HostState { }) } - fn subscribe_readable(&mut self, _self_: Resource) -> Resource { - // Real pollable wiring (tokio AsyncRead readiness over the - // NetStream) lands with the stream-half adapter commit. - // Always-ready sentinel until then; guests poll then call - // read-bytes which handles real readability internally. - super::super::stubs::always_ready_pollable(&mut self.resource_table) + fn subscribe_readable(&mut self, self_: Resource) -> Resource { + wasmtime_wasi::p2::subscribe( + &mut self.resource_table, + Resource::::new_borrow(self_.rep()), + ) + .unwrap_or_else(|_| Resource::new_own(0)) } fn read_stream(&mut self, _self_: Resource) -> Resource { @@ -519,6 +522,9 @@ impl HostTcpStream for HostState { .is_ok() { self.net_stream_count = self.net_stream_count.saturating_sub(1); + self.net_stream_leases.remove(&table_rep); + self.net_frame_states.remove(&table_rep); + record_net_stream_metrics(self); } // Drop any verified per-connection principal binding (issue #45/#852) // so the registry does not leak entries for closed connections. A diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs b/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs index 5130a5969..338a53574 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/unix_listener.rs @@ -7,26 +7,58 @@ use std::sync::Arc; use std::time::Duration; +#[cfg(unix)] +use std::os::fd::{AsFd, OwnedFd}; + use wasmtime::component::Resource; -use wasmtime_wasi::p2::DynPollable; +use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe}; use super::client_lifecycle; use super::handshake::validate_handshake; #[cfg(unix)] use super::handshake::verify_peer_credentials; -use super::{HostState, MAX_ACTIVE_STREAMS, NetStream, UnixListenerSlot, audit_net, map_io_err}; +use super::{ + HostState, NetStream, UnixListenerSlot, audit_net, map_io_err, record_net_stream_metrics, +}; use crate::engine::wasm::bindings::astrid::net::host::{ ErrorCode, HostUnixListener, TcpStream, UnixListener, }; use crate::engine::wasm::host::util; +#[cfg(unix)] +struct UnixListenerReadiness { + descriptor: tokio::io::unix::AsyncFd, + budget: Arc, +} + +#[cfg(unix)] +#[async_trait::async_trait] +impl Pollable for UnixListenerReadiness { + async fn ready(&mut self) { + loop { + self.budget.wait_available().await; + if let Ok(mut readiness) = self.descriptor.readable().await { + // The listener itself remains level-triggered. Clear only this + // duplicate descriptor's cached Tokio readiness so the next + // poll waits for the post-accept state instead of firing + // forever. + readiness.clear_ready(); + if self.budget.has_capacity() { + return; + } + } else { + return; + } + } + } +} + impl HostUnixListener for HostState { fn accept(&mut self, _self_: Resource) -> Result, ErrorCode> { - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + let listener_arc = self.cli_socket_listener.clone().ok_or(ErrorCode::Closed)?; + if !self.net_stream_budget.has_capacity() { return Err(ErrorCode::Quota); } - - let listener_arc = self.cli_socket_listener.clone().ok_or(ErrorCode::Closed)?; let rt_handle = self.runtime_handle.clone(); let cancel_token = self.effective_cancel_token(); let session_token = self.session_token.clone(); @@ -58,11 +90,11 @@ impl HostUnixListener for HostState { &blocking_semaphore, &cancel_token, async { - let l = listener_arc.lock().await; - l.accept().await + let listener = listener_arc.lock().await; + listener.accept().await.map(|(stream, _)| stream) }, ); - let (stream, _addr) = match accept_result { + let stream = match accept_result { Some(result) => result.map_err(map_io_err)?, None => return Err(ErrorCode::Closed), }; @@ -117,11 +149,13 @@ impl HostUnixListener for HostState { } }; - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + // Waiting for an inbound connection does not consume another file + // descriptor, so acquire only after accept and authentication. The + // atomic acquisition closes the race between concurrent acceptors. + let Some(stream_lease) = self.net_stream_budget.try_acquire() else { drop(stream); return Err(ErrorCode::Quota); - } - + }; let net_stream = NetStream::Unix(Arc::new(tokio::sync::Mutex::new(stream))); let res = self .resource_table @@ -129,6 +163,12 @@ impl HostUnixListener for HostState { .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?; self.net_stream_count += 1; let rep = res.rep(); + let previous = self.net_stream_leases.insert(rep, stream_lease); + debug_assert!( + previous.is_none(), + "net stream resource rep reused while live" + ); + record_net_stream_metrics(self); // Record the verified principal AND its authenticating device key_id // (issue #45/#852) keyed by the stream resource rep, now that the rep // is known. Storage only; enforcement reads this registry separately — @@ -166,7 +206,7 @@ impl HostUnixListener for HostState { let session_token = self.session_token.clone(); let blocking_semaphore = self.blocking_semaphore.clone(); - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + if !self.net_stream_budget.has_capacity() { return Ok(None); } @@ -176,16 +216,16 @@ impl HostUnixListener for HostState { &blocking_semaphore, &cancel_token, async { - let l = listener_arc.lock().await; - tokio::time::timeout(Duration::from_millis(timeout_ms), l.accept()).await + let listener = listener_arc.lock().await; + tokio::time::timeout(Duration::from_millis(timeout_ms), listener.accept()).await }, ); - let (stream, _addr) = match accept_result { + let stream = match accept_result { None => return Ok(None), Some(Err(_)) => return Ok(None), Some(Ok(Err(e))) => return Err(map_io_err(e)), - Some(Ok(Ok(pair))) => pair, + Some(Ok(Ok((stream, _)))) => stream, }; #[cfg(unix)] @@ -236,11 +276,10 @@ impl HostUnixListener for HostState { } } - if self.net_stream_count >= MAX_ACTIVE_STREAMS { + let Some(stream_lease) = self.net_stream_budget.try_acquire() else { drop(stream); return Ok(None); - } - + }; let net_stream = NetStream::Unix(Arc::new(tokio::sync::Mutex::new(stream))); let res = self .resource_table @@ -248,6 +287,12 @@ impl HostUnixListener for HostState { .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?; self.net_stream_count += 1; let rep = res.rep(); + let previous = self.net_stream_leases.insert(rep, stream_lease); + debug_assert!( + previous.is_none(), + "net stream resource rep reused while live" + ); + record_net_stream_metrics(self); // Same per-connection principal + device-key binding as `accept` // (issue #45/#852). let verified_principal = verified_identity.as_ref().map(|(p, _)| p.clone()); @@ -265,11 +310,27 @@ impl HostUnixListener for HostState { } fn subscribe_readiness(&mut self, _self_: Resource) -> Resource { - // Real wiring (tokio UnixListener::poll_accept-backed) lands - // with the dedicated pollable commit. Always-ready sentinel - // until then — guests poll, call accept, get a connection or - // wait inside accept's own blocking path. - super::super::stubs::always_ready_pollable(&mut self.resource_table) + let Some(listener) = self.cli_socket_listener.clone() else { + return Resource::new_own(0); + }; + let runtime = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let cancel = self.effective_cancel_token(); + let async_fd = util::bounded_block_on_cancellable(&runtime, &semaphore, &cancel, async { + let listener = listener.lock().await; + let descriptor = listener.as_fd().try_clone_to_owned()?; + tokio::io::unix::AsyncFd::new(descriptor) + }); + let Some(Ok(async_fd)) = async_fd else { + return Resource::new_own(0); + }; + let Ok(readiness) = self.resource_table.push(UnixListenerReadiness { + descriptor: async_fd, + budget: Arc::clone(&self.net_stream_budget), + }) else { + return Resource::new_own(0); + }; + subscribe(&mut self.resource_table, readiness).unwrap_or_else(|_| Resource::new_own(0)) } fn drop(&mut self, rep: Resource) -> wasmtime::Result<()> { @@ -279,3 +340,54 @@ impl HostUnixListener for HostState { Ok(()) } } + +#[cfg(all(test, unix))] +mod readiness_tests { + use std::os::fd::AsFd; + + use wasmtime_wasi::p2::Pollable; + + use super::UnixListenerReadiness; + + #[tokio::test] + async fn listener_readiness_waits_for_connection_and_capacity_without_accepting() { + let directory = tempfile::tempdir().expect("temporary socket directory"); + let socket_path = directory.path().join("readiness.sock"); + let listener = tokio::net::UnixListener::bind(&socket_path).expect("bind test listener"); + let descriptor = listener + .as_fd() + .try_clone_to_owned() + .expect("duplicate listener descriptor"); + let mut readiness = UnixListenerReadiness { + descriptor: tokio::io::unix::AsyncFd::new(descriptor) + .expect("register duplicate descriptor"), + budget: std::sync::Arc::new(crate::NetStreamBudget::new(1)), + }; + let lease = readiness + .budget + .try_acquire() + .expect("occupy the only stream slot"); + + let connector = tokio::spawn(async move { + tokio::net::UnixStream::connect(socket_path) + .await + .expect("connect test client") + }); + + assert!( + tokio::time::timeout(std::time::Duration::from_millis(30), readiness.ready()) + .await + .is_err(), + "a readable listener must not wake while stream capacity is full" + ); + drop(lease); + tokio::time::timeout(std::time::Duration::from_secs(2), readiness.ready()) + .await + .expect("readiness should fire"); + tokio::time::timeout(std::time::Duration::from_secs(2), listener.accept()) + .await + .expect("readiness must not consume the connection") + .expect("accept test connection"); + connector.await.expect("connector task"); + } +} diff --git a/crates/astrid-capsule/src/engine/wasm/host_state.rs b/crates/astrid-capsule/src/engine/wasm/host_state.rs index f6dac198c..29f560c6c 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -56,6 +56,19 @@ pub struct TcpStreamSlot { pub write_timeout: Option, } +/// Incremental decoder state for one length-prefixed stream. +/// +/// A readiness wake promises only one readable byte. Preserving partial header +/// and payload bytes prevents fragmented clients from desynchronizing framing +/// when a non-blocking read window expires between bytes. +#[derive(Debug, Default)] +pub(crate) struct FrameReadState { + pub(crate) header: [u8; 4], + pub(crate) header_read: usize, + pub(crate) payload: Vec, + pub(crate) payload_read: usize, +} + /// The lifecycle phase a capsule is currently executing in. /// /// Set on [`HostState`] during `#[install]` or `#[upgrade]` dispatch. @@ -605,9 +618,18 @@ pub struct HostState { /// here — off the wasmtime resource table — which is what lets them /// survive instance churn. pub persistent_processes: Arc, + /// Kernel-owned process-wide network-stream admission budget shared by + /// every capsule engine and pooled Store. + pub net_stream_budget: Arc, + /// RAII leases keyed by resource-table rep. Clearing this map releases + /// global capacity even when a pool reset replaces the table without + /// invoking the guest-visible stream `drop` method. + pub net_stream_leases: HashMap, + /// Incremental length-prefixed decoder state keyed by network-stream rep. + pub(crate) net_frame_states: HashMap, /// Live count of `NetStream` entries currently in the resource table. /// Maintained alongside `ResourceTable` insertions / drops so the - /// `MAX_ACTIVE_STREAMS` gate is O(1) instead of iterating every + /// local accounting and diagnostics stay O(1) instead of iterating every /// resource (the table may hold hundreds of pollables / errors / /// http handles unrelated to net). Single-threaded: wasmtime /// stores are owned by exactly one OS thread. diff --git a/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs b/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs index c2f6a2662..8c936e509 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs @@ -122,6 +122,9 @@ impl HostState { identity_store: None, process_tracker, persistent_processes, + net_stream_budget: Arc::new(crate::NetStreamBudget::default()), + net_stream_leases: HashMap::new(), + net_frame_states: HashMap::new(), net_stream_count: 0, subscription_count: 0, process_count_total: 0, diff --git a/crates/astrid-capsule/src/engine/wasm/mod.rs b/crates/astrid-capsule/src/engine/wasm/mod.rs index af345f609..00fdd4bb9 100644 --- a/crates/astrid-capsule/src/engine/wasm/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/mod.rs @@ -264,6 +264,10 @@ pub struct WasmEngine { /// `blocking_semaphore` / `io_semaphore` at load time. `Default` (all /// host-derived) in tests. runtime_limits: limits::CapsuleRuntimeLimits, + /// Process-wide persistent network-stream budget shared across every + /// capsule and pooled Store. Standalone engines receive an isolated + /// host-derived budget; the kernel loader replaces it with one global Arc. + net_stream_budget: Arc, /// Resolved operator ceilings for the `astrid:http` host. A GLOBAL value /// (same for every capsule), resolved once by the daemon from the `[http]` /// config section and handed down the loader chain like `runtime_limits`; @@ -343,6 +347,7 @@ impl WasmEngine { fuel_rate, group_config: None, runtime_limits, + net_stream_budget: Arc::new(crate::NetStreamBudget::default()), http_limits, workspace_cow: None, process_tracker: None, @@ -350,6 +355,13 @@ impl WasmEngine { } } + /// Use the kernel-owned process-wide persistent network-stream budget. + #[must_use] + pub fn with_net_stream_budget(mut self, budget: Arc) -> Self { + self.net_stream_budget = budget; + self + } + /// Promote/rollback the OS-level copy-on-write workspace behind a QUIESCENCE /// INTERLOCK, so the merged tree is never swapped/deleted under a running /// invocation or spawned child (which would corrupt or destroy its work — @@ -1674,6 +1686,7 @@ impl ExecutionEngine for WasmEngine { // One IPC rate limiter shared by every pooled instance, so the // per-capsule throughput budget is not multiplied by pool size. let ipc_limiter = Arc::new(astrid_events::ipc::IpcRateLimiter::new()); + let net_stream_budget = Arc::clone(&self.net_stream_budget); // ── Run-loop resource bound (CPU epoch interrupt + linear memory) ─ // @@ -1896,6 +1909,9 @@ impl ExecutionEngine for WasmEngine { identity_store: st_identity_store.clone(), process_tracker: process_tracker.clone(), persistent_processes: persistent_registry.clone(), + net_stream_budget: Arc::clone(&net_stream_budget), + net_stream_leases: std::collections::HashMap::new(), + net_frame_states: std::collections::HashMap::new(), net_stream_count: 0, subscription_count: 0, process_count_total: 0, @@ -3017,6 +3033,9 @@ pub async fn run_lifecycle( persistent_processes: Arc::new(host::process::PersistentProcessRegistry::new( tokio::runtime::Handle::current(), )), + net_stream_budget: Arc::new(crate::NetStreamBudget::default()), + net_stream_leases: std::collections::HashMap::new(), + net_frame_states: std::collections::HashMap::new(), net_stream_count: 0, subscription_count: 0, process_count_total: 0, diff --git a/crates/astrid-capsule/src/engine/wasm/pool.rs b/crates/astrid-capsule/src/engine/wasm/pool.rs index 8c9f869c5..e136bd705 100644 --- a/crates/astrid-capsule/src/engine/wasm/pool.rs +++ b/crates/astrid-capsule/src/engine/wasm/pool.rs @@ -460,6 +460,14 @@ fn clear_on_return(state: &mut HostState, reset_resources: bool) { // them to the empty-table baseline so the per-(principal) gates start // from zero for the next lease. state.active_http_streams.clear(); + state.net_stream_leases.clear(); + state.net_frame_states.clear(); + metrics::gauge!("astrid_capsule_net_streams_active").set(f64::from( + u32::try_from(state.net_stream_budget.active()).unwrap_or(u32::MAX), + )); + metrics::gauge!("astrid_capsule_net_streams_limit").set(f64::from( + u32::try_from(state.net_stream_budget.limit()).unwrap_or(u32::MAX), + )); state.net_stream_count = 0; state.subscription_count = 0; state.process_count_total = 0; diff --git a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs index 2d591011e..ffb87f1d5 100644 --- a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs +++ b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs @@ -144,6 +144,9 @@ pub(crate) fn minimal_host_state(rt: tokio::runtime::Handle) -> HostState { persistent_processes: Arc::new( crate::engine::wasm::host::process::PersistentProcessRegistry::new(rt), ), + net_stream_budget: Arc::new(crate::NetStreamBudget::default()), + net_stream_leases: HashMap::new(), + net_frame_states: HashMap::new(), net_stream_count: 0, subscription_count: 0, process_count_total: 0, diff --git a/crates/astrid-capsule/src/lib.rs b/crates/astrid-capsule/src/lib.rs index d4cd79651..d2bd8b8b3 100644 --- a/crates/astrid-capsule/src/lib.rs +++ b/crates/astrid-capsule/src/lib.rs @@ -44,7 +44,10 @@ pub mod toposort; pub(crate) mod watcher; pub use access::CapsuleAccessResolver; -pub use astrid_capsule_types::limits::{CapsuleRuntimeLimits, HttpLimits}; +pub use astrid_capsule_types::limits::{ + CapsuleRuntimeLimits, HttpLimits, NetStreamBudget, NetStreamLease, + host_net_stream_limit_default, +}; pub use audit_sink::{HostAuditEvent, HostAuditOutcome, HostAuditSink}; pub use fuel_ledger::{FuelLedger, FuelRateLimiter}; pub use memory_ledger::MemoryLedger; diff --git a/crates/astrid-capsule/src/loader.rs b/crates/astrid-capsule/src/loader.rs index 491f4b47f..42e6dd5fd 100644 --- a/crates/astrid-capsule/src/loader.rs +++ b/crates/astrid-capsule/src/loader.rs @@ -1,6 +1,7 @@ //! Factory and routing logic for instantiating Composite Capsules. use std::path::PathBuf; +use std::sync::Arc; use crate::capsule::{Capsule, CompositeCapsule}; use crate::engine::wasm::limits::{CapsuleRuntimeLimits, HttpLimits}; @@ -32,6 +33,10 @@ pub struct CapsuleLoader { /// semaphores. A plain `Copy` value, not a shared handle. See /// [`CapsuleRuntimeLimits`]. runtime_limits: CapsuleRuntimeLimits, + /// Process-wide persistent network-stream budget. Kernel construction + /// replaces the default with its one shared handle before loading any + /// capsule; the default preserves standalone loader compatibility. + net_stream_budget: Arc, /// Resolved `astrid:http` host ceilings (timeouts, redirect/stream caps, /// buffered-body limit), resolved once by the daemon from the `[http]` /// config section and handed to every `WasmEngine`. A global `Copy` value. @@ -66,10 +71,18 @@ impl CapsuleLoader { fuel_rate, memory_ledger, runtime_limits, + net_stream_budget: Arc::new(crate::NetStreamBudget::default()), http_limits, } } + /// Use the kernel-owned process-wide persistent network-stream budget. + #[must_use] + pub fn with_net_stream_budget(mut self, budget: Arc) -> Self { + self.net_stream_budget = budget; + self + } + /// Parse a `CapsuleManifest` and build a unified `CompositeCapsule`. /// /// This method is the "router" of the Manifest-First architecture. It inspects @@ -88,15 +101,18 @@ impl CapsuleLoader { // 1. WASM Component Engine if !manifest.components.is_empty() { - composite.add_engine(Box::new(crate::engine::WasmEngine::new( - manifest.clone(), - capsule_dir.clone(), - self.fuel_ledger.clone(), - self.fuel_rate.clone(), - self.memory_ledger.clone(), - self.runtime_limits, - self.http_limits, - ))); + composite.add_engine(Box::new( + crate::engine::WasmEngine::new( + manifest.clone(), + capsule_dir.clone(), + self.fuel_ledger.clone(), + self.fuel_rate.clone(), + self.memory_ledger.clone(), + self.runtime_limits, + self.http_limits, + ) + .with_net_stream_budget(Arc::clone(&self.net_stream_budget)), + )); } // 2. Legacy Host MCP Engine (The Airlock Override) diff --git a/crates/astrid-config/src/defaults.toml b/crates/astrid-config/src/defaults.toml index 32ba5c8c9..a2ca50539 100644 --- a/crates/astrid-config/src/defaults.toml +++ b/crates/astrid-config/src/defaults.toml @@ -445,6 +445,11 @@ timeout_secs = 300 # rides on. Host default is cores-scaled and clamped by half the fd limit. # host_io_concurrency = 512 +# Process-wide persistent capsule network-stream budget. Host default derives +# from the file-descriptor envelope (half of the fd-clamped async-I/O budget), +# and is shared across every capsule and pooled instance. +# host_net_streams = 256 + # Max size of a capsule's dynamic instance pool (concurrent interceptor # invocations). Replaces the old fixed 16; host default is cores-scaled. The # pool warm-starts small and grows lazily toward this, so it bounds the peak, diff --git a/crates/astrid-config/src/env.rs b/crates/astrid-config/src/env.rs index 99f6e49eb..26f565b00 100644 --- a/crates/astrid-config/src/env.rs +++ b/crates/astrid-config/src/env.rs @@ -72,6 +72,10 @@ const ENV_MAPPINGS: &[EnvMapping] = &[ var_name: "ASTRID_CAPSULE_HOST_IO_CONCURRENCY", field_path: "capsule.host_io_concurrency", }, + EnvMapping { + var_name: "ASTRID_CAPSULE_HOST_NET_STREAMS", + field_path: "capsule.host_net_streams", + }, EnvMapping { var_name: "ASTRID_CAPSULE_INSTANCE_POOL_SIZE", field_path: "capsule.instance_pool_size", @@ -295,6 +299,7 @@ fn coerce_to_toml_value(path: &str, val: &str) -> toml::Value { | "subagents.timeout_secs" | "capsule.host_blocking_concurrency" | "capsule.host_io_concurrency" + | "capsule.host_net_streams" | "capsule.instance_pool_size" | "retry.llm_max_attempts" | "retry.mcp_max_attempts" diff --git a/crates/astrid-config/src/merge/restrict.rs b/crates/astrid-config/src/merge/restrict.rs index b2693fe25..7a3a0e9f7 100644 --- a/crates/astrid-config/src/merge/restrict.rs +++ b/crates/astrid-config/src/merge/restrict.rs @@ -101,6 +101,17 @@ pub fn enforce_restrictions( // global config can set `[http]`. block_workspace_override(merged, baseline, workspace_layer, &["http"], "http"); + // Process-wide persistent stream capacity is operator-owned. A workspace + // may neither widen it toward descriptor exhaustion nor shrink it into a + // daemon-wide denial of service for unrelated principals. + block_workspace_override( + merged, + baseline, + workspace_layer, + &["capsule", "host_net_streams"], + "capsule.host_net_streams", + ); + // workspace.auto_allow_read: cannot expand beyond baseline. block_workspace_expansion( merged, diff --git a/crates/astrid-config/src/merge/tests.rs b/crates/astrid-config/src/merge/tests.rs index dcf5d9eb3..de0bd6d97 100644 --- a/crates/astrid-config/src/merge/tests.rs +++ b/crates/astrid-config/src/merge/tests.rs @@ -454,6 +454,35 @@ fn test_capsule_local_egress_workspace_cannot_widen_operator_value() { assert_eq!(openai[0].as_str().unwrap(), "127.0.0.1:1234"); } +#[test] +fn test_host_net_streams_is_operator_only() { + let operator: toml::Value = toml::from_str( + r#" + [capsule] + host_net_streams = 64 + "#, + ) + .unwrap(); + let workspace: toml::Value = toml::from_str( + r#" + [capsule] + host_net_streams = 4096 + "#, + ) + .unwrap(); + + let mut merged = operator.clone(); + deep_merge(&mut merged, &workspace); + enforce_restrictions(&mut merged, &operator, &workspace); + assert_eq!(merged["capsule"]["host_net_streams"].as_integer(), Some(64)); + + let empty_operator: toml::Value = toml::from_str("").unwrap(); + let mut no_operator = empty_operator.clone(); + deep_merge(&mut no_operator, &workspace); + enforce_restrictions(&mut no_operator, &empty_operator, &workspace); + assert!(no_operator["capsule"].get("host_net_streams").is_none()); +} + #[test] fn test_http_section_cannot_be_set_by_workspace() { // The [http] host limits are widening controls (raising a timeout, redirect diff --git a/crates/astrid-config/src/types.rs b/crates/astrid-config/src/types.rs index a28a860b7..6a878a0a6 100644 --- a/crates/astrid-config/src/types.rs +++ b/crates/astrid-config/src/types.rs @@ -818,6 +818,11 @@ pub struct CapsuleSection { /// the outbound-throughput gate the LLM path rides on; sizing it well above /// the blocking ceiling is the point of the split. pub host_io_concurrency: Option, + /// Process-wide ceiling on persistent capsule network streams. `None` uses + /// a host-derived share of the file-descriptor budget. Unlike the old + /// per-Store constant, this limit is shared across every capsule and pooled + /// instance, so capacity cannot multiply past the process fd envelope. + pub host_net_streams: Option, /// **Max** size of a capsule's dynamic instance pool — the ceiling on its /// concurrent interceptor invocations. `None` → cores-scaled (replacing the /// old fixed 16). The pool warm-starts well below this and grows lazily, so diff --git a/crates/astrid-config/src/validate.rs b/crates/astrid-config/src/validate.rs index a0c786d36..2b3693e7c 100644 --- a/crates/astrid-config/src/validate.rs +++ b/crates/astrid-config/src/validate.rs @@ -300,6 +300,13 @@ fn validate_capsule(config: &Config) -> ConfigResult<()> { }); } + if c.host_net_streams == Some(0) { + return Err(ConfigError::ValidationError { + field: "capsule.host_net_streams".to_owned(), + message: "host_net_streams must be greater than 0".to_owned(), + }); + } + if c.instance_pool_size == Some(0) { return Err(ConfigError::ValidationError { field: "capsule.instance_pool_size".to_owned(), @@ -396,6 +403,7 @@ mod tests { let config = Config::default(); assert!(config.capsule.host_blocking_concurrency.is_none()); assert!(config.capsule.host_io_concurrency.is_none()); + assert!(config.capsule.host_net_streams.is_none()); assert!(validate(&config).is_ok()); } @@ -404,6 +412,7 @@ mod tests { let mut config = Config::default(); config.capsule.host_blocking_concurrency = Some(4); config.capsule.host_io_concurrency = Some(256); + config.capsule.host_net_streams = Some(128); assert!(validate(&config).is_ok()); } @@ -443,6 +452,18 @@ mod tests { )); } + #[test] + fn test_capsule_zero_net_streams_rejected() { + let mut config = Config::default(); + config.capsule.host_net_streams = Some(0); + let err = validate(&config).unwrap_err(); + assert!(matches!( + err, + ConfigError::ValidationError { field, .. } + if field == "capsule.host_net_streams" + )); + } + #[test] fn test_yolo_workspace_mode() { let mut config = Config::default(); diff --git a/crates/astrid-daemon/src/lib.rs b/crates/astrid-daemon/src/lib.rs index 06e679cf3..d3890e5ef 100644 --- a/crates/astrid-daemon/src/lib.rs +++ b/crates/astrid-daemon/src/lib.rs @@ -48,6 +48,11 @@ pub struct Args { #[arg(long, value_parser = parse_nonzero_concurrency)] pub host_blocking_concurrency: Option, + /// Override the process-wide persistent capsule network-stream budget. + /// Defaults to a host-derived share of the file-descriptor envelope. + #[arg(long, value_parser = parse_nonzero_concurrency)] + pub host_net_streams: Option, + /// Override the max size of each capsule's dynamic instance pool (concurrent /// interceptor invocations). Highest-precedence override; defaults to a /// host-derived value (cores-scaled, replacing the old fixed 16). @@ -111,6 +116,15 @@ fn resolve_capsule_limits( ) } +/// Resolve the process-wide persistent network-stream budget from the same +/// precedence chain as the other capsule runtime limits. +fn resolve_net_stream_limit(args: &Args, cfg: Option<&astrid_config::Config>) -> usize { + args.host_net_streams + .or_else(|| cfg.and_then(|c| c.capsule.host_net_streams)) + .unwrap_or_else(astrid_capsule::host_net_stream_limit_default) + .max(1) +} + /// Resolve the `astrid:http` operator host policy from the `[http]` config /// section into the typed [`HttpLimits`](astrid_capsule::HttpLimits) the kernel /// forwards to every capsule. The timeout fields are per-request DEFAULTS (a @@ -198,6 +212,7 @@ pub async fn run() -> Result<()> { // host-derived default); the kernel forwards them to every `WasmEngine`. // Done before `args.workspace` is consumed below. let runtime_limits = resolve_capsule_limits(&args, unified_cfg.as_ref()); + let net_stream_limit = resolve_net_stream_limit(&args, unified_cfg.as_ref()); // Operator-approved per-capsule local-egress allowlist (SSRF-airlock // exemptions). Operator config only — the kernel hands each capsule its @@ -220,6 +235,7 @@ pub async fn run() -> Result<()> { ) .await .map_err(|e| anyhow::anyhow!("Failed to boot Kernel: {e}"))?; + kernel.set_net_stream_limit(net_stream_limit); // In ephemeral mode, shut down immediately when the last client disconnects. if args.ephemeral { @@ -427,8 +443,9 @@ fn spawn_gateway( #[cfg(test)] mod tests { - use super::provides_cli_socket_uplink; + use super::{Args, provides_cli_socket_uplink, resolve_net_stream_limit}; use astrid_capsule::manifest::CapsuleManifest; + use clap::Parser; fn manifest(name: &str, uplink: bool, net_bind: &[&str]) -> CapsuleManifest { let mut manifest = CapsuleManifest::default(); @@ -461,4 +478,31 @@ mod tests { ); } } + + #[test] + fn net_stream_limit_uses_cli_then_config_then_host_default() { + let mut config = astrid_config::Config::default(); + config.capsule.host_net_streams = Some(123); + + let from_config = + Args::try_parse_from(["astrid-daemon"]).expect("default daemon arguments should parse"); + assert_eq!(resolve_net_stream_limit(&from_config, Some(&config)), 123); + + let from_cli = Args::try_parse_from(["astrid-daemon", "--host-net-streams", "456"]) + .expect("CLI stream override should parse"); + assert_eq!(resolve_net_stream_limit(&from_cli, Some(&config)), 456); + + assert_eq!( + resolve_net_stream_limit(&from_config, None), + astrid_capsule::host_net_stream_limit_default() + ); + } + + #[test] + fn net_stream_cli_rejects_zero() { + let Err(error) = Args::try_parse_from(["astrid-daemon", "--host-net-streams", "0"]) else { + panic!("zero must fail at the CLI boundary"); + }; + assert!(error.to_string().contains("must be >= 1")); + } } diff --git a/crates/astrid-events/src/bus_tests.rs b/crates/astrid-events/src/bus_tests.rs index c67219b29..a5b2cf1d3 100644 --- a/crates/astrid-events/src/bus_tests.rs +++ b/crates/astrid-events/src/bus_tests.rs @@ -702,6 +702,31 @@ async fn routed_recv_wakes_on_publish() { publisher.await.expect("publisher task"); } +#[tokio::test] +async fn routed_ready_wakes_without_consuming() { + let bus = EventBus::new(); + let mut sub = + bus.subscribe_topic_routed(uuid::Uuid::new_v4(), "t.*", "capsule-ready", "test_sub"); + + let publisher_bus = bus.clone(); + let publisher = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + publisher_bus.publish(ipc_evt("t.ready", Some("alice"))); + }); + + tokio::time::timeout(std::time::Duration::from_secs(2), sub.ready()) + .await + .expect("readiness should wake"); + assert!(sub.total_bytes() > 0, "ready must not consume the event"); + let event = sub.try_recv_one().expect("event remains queued"); + if let AstridEvent::Ipc { message, .. } = &*event { + assert_eq!(message.topic, "t.ready"); + } else { + panic!("expected IPC event"); + } + publisher.await.expect("publisher task"); +} + #[tokio::test] async fn routed_recv_timeout_returns_none_when_idle() { let bus = EventBus::new(); diff --git a/crates/astrid-events/src/route/receiver.rs b/crates/astrid-events/src/route/receiver.rs index f3a04c357..6c8a22abc 100644 --- a/crates/astrid-events/src/route/receiver.rs +++ b/crates/astrid-events/src/route/receiver.rs @@ -39,6 +39,19 @@ impl std::fmt::Debug for RoutedEventReceiver { } impl RoutedEventReceiver { + /// Wait until at least one routed event is available without consuming it. + /// Used by the Component Model pollable adapter so a guest can multiplex + /// IPC and socket readiness without timeout polling. + pub async fn ready(&mut self) { + loop { + let notified = self.notify.notified(); + if self.route_entry.lock().total_bytes > 0 { + return; + } + notified.await; + } + } + /// Non-blocking receive of one event from the next DRR round. /// /// This is the host-boundary counterpart to [`recv`](Self::recv): WASM diff --git a/crates/astrid-kernel/src/lib.rs b/crates/astrid-kernel/src/lib.rs index 96d2a18ae..aea000937 100644 --- a/crates/astrid-kernel/src/lib.rs +++ b/crates/astrid-kernel/src/lib.rs @@ -188,6 +188,10 @@ pub struct Kernel { /// `Copy` value — no resolution logic lives here. See /// [`CapsuleRuntimeLimits`](astrid_capsule_types::CapsuleRuntimeLimits). runtime_limits: astrid_capsule_types::CapsuleRuntimeLimits, + /// Process-wide persistent network-stream admission budget. One shared + /// handle is forwarded through every loader and engine so capsule count and + /// instance-pool size cannot multiply the file-descriptor envelope. + net_stream_budget: Arc, /// Operator-approved per-capsule local-egress allowlist /// (`[security.capsule_local_egress]`), keyed by capsule id. Resolved /// once from config by the daemon; the kernel only stores it and hands @@ -341,6 +345,28 @@ impl KernelResources { } impl Kernel { + /// Set the process-wide persistent network-stream admission ceiling. + /// Existing streams remain valid when the ceiling is lowered; new + /// admissions resume after usage falls below the configured limit. + pub fn set_net_stream_limit(&self, limit: usize) { + self.net_stream_budget.set_limit(limit); + metrics::gauge!("astrid_capsule_net_streams_limit").set(f64::from( + u32::try_from(self.net_stream_budget.limit()).unwrap_or(u32::MAX), + )); + metrics::gauge!("astrid_capsule_net_streams_active").set(f64::from( + u32::try_from(self.net_stream_budget.active()).unwrap_or(u32::MAX), + )); + } + + /// Return `(active, limit)` for capacity reporting and telemetry. + #[must_use] + pub fn net_stream_usage(&self) -> (usize, usize) { + ( + self.net_stream_budget.active(), + self.net_stream_budget.limit(), + ) + } + /// Per-project runtime layout selected at boot. #[must_use] pub fn workspace_layout(&self) -> &WorkspaceLayout { @@ -732,6 +758,7 @@ impl Kernel { fuel_rate: astrid_capsule_types::FuelRateLimiter::default(), memory_ledger: astrid_capsule_types::MemoryLedger::default(), runtime_limits, + net_stream_budget: Arc::new(astrid_capsule_types::NetStreamBudget::default()), local_egress, http_limits, full_reload_in_flight: AtomicBool::new(false), @@ -985,7 +1012,8 @@ impl Kernel { self.memory_ledger.clone(), self.runtime_limits, self.http_limits, - ); + ) + .with_net_stream_budget(Arc::clone(&self.net_stream_budget)); let mut capsule = loader.create_capsule(manifest, dir.to_path_buf())?; let kv = astrid_storage::ScopedKvStore::new( @@ -2456,6 +2484,7 @@ pub(crate) async fn test_kernel_with_home(home: astrid_core::dirs::AstridHome) - fuel_rate: astrid_capsule_types::FuelRateLimiter::default(), memory_ledger: astrid_capsule_types::MemoryLedger::default(), runtime_limits: astrid_capsule_types::CapsuleRuntimeLimits::default(), + net_stream_budget: Arc::new(astrid_capsule_types::NetStreamBudget::default()), local_egress: std::collections::HashMap::new(), http_limits: astrid_capsule_types::HttpLimits::default(), full_reload_in_flight: AtomicBool::new(false), From 2c39f9d7ccb924d5fad29090da45d11983a3fd56 Mon Sep 17 00:00:00 2001 From: "Joshua J. Bouw" Date: Tue, 21 Jul 2026 19:36:39 +0400 Subject: [PATCH 2/2] fix(config): satisfy test-target clippy --- crates/astrid-config/src/merge/tests.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/astrid-config/src/merge/tests.rs b/crates/astrid-config/src/merge/tests.rs index de0bd6d97..83a54cb2b 100644 --- a/crates/astrid-config/src/merge/tests.rs +++ b/crates/astrid-config/src/merge/tests.rs @@ -457,17 +457,17 @@ fn test_capsule_local_egress_workspace_cannot_widen_operator_value() { #[test] fn test_host_net_streams_is_operator_only() { let operator: toml::Value = toml::from_str( - r#" + r" [capsule] host_net_streams = 64 - "#, + ", ) .unwrap(); let workspace: toml::Value = toml::from_str( - r#" + r" [capsule] host_net_streams = 4096 - "#, + ", ) .unwrap();