feat(runtime): add host-wide stream capacity and readiness

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

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 }

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;

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<Self>) -> Option<NetStreamLease> {
+        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<NetStreamBudget>,
+}
+
+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));

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<Subscription>) -> Resource<DynPollable> {
-        // 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<Subscription>) -> Resource<DynPollable> {
+        wasmtime_wasi::p2::subscribe(
+            &mut self.resource_table,
+            Resource::<SubscriptionEntry>::new_borrow(self_.rep()),
+        )
+        .unwrap_or_else(|_| Resource::new_own(0))
     }
 
     fn drop(&mut self, rep: Resource<Subscription>) -> wasmtime::Result<()> {

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<T, E: std::fmt::Debug>(
     }
 }
 
+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<Resource<TcpStream>, 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<Resource<TcpStream>, 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<Resource<TcpStream>, 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());