From 08e0fc43ebad0260b32e11b2b9ffbd0004420c9c Mon Sep 17 00:00:00 2001 From: Jamie Steiner Date: Mon, 13 Jul 2026 17:12:46 -0700 Subject: [PATCH 1/3] feat(net): implement capsule inbound TCP bind (bind_tcp) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fills in the daemon's stubbed astrid:net bind_tcp host fn so a capsule can bind a loopback TCP listener and accept inbound connections — the missing substrate for capsule-hosted HTTP servers (e.g. an Anthropic-Messages shim Claude Code points ANTHROPIC_BASE_URL at, routed by srouter). Design: - Authorization reuses the existing `net_bind` manifest field, whose own doc already reads "Unix/TCP socket bind addresses". TCP entries are `host:port` / `host:*` patterns matched with the SAME semantics as net_connect; a `unix:*` entry (the CLI proxy) never matches a TCP host:port, so the two socket families share the field without cross-authorizing. New gate method `check_net_tcp_bind(capsule, host, port)`, fail-closed default in the trait, allowlist match in ManifestSecurityGate. - Host fn `bind_tcp`: capability-gate → loopback-confinement rail → tokio bind → resource-table slot. Loopback-only is enforced host-side (is_loopback_bind_host) regardless of the allowlist, mirroring how connect_tcp runs its is_safe_ip airlock AFTER the capability gate. Non-loopback bind is refused (AirlockRejected), not downgraded. - TcpListenerSlot now holds the live Arc. accept / poll_accept register the accepted stream as a NetStream::Tcp — the SAME representation outbound connect_tcp uses — so every existing read/write/peek/timeout host fn works on accepted connections with no extra wiring. Per-capsule MAX_ACTIVE_STREAMS cap applies; accept sets recv_yielded so a bound accept-loop is not epoch-trapped as a spinner; cancellable so capsule unload wins over a blocked accept. Proven: a probe capsule bound 127.0.0.1:8799, accepted a curl connection, and served an HTTP 200 through the patched daemon (LISTENING → ACCEPTED → wrote 110 bytes). 6 new unit tests (gate host:port matching incl. the unix-entry-doesn't-authorize-TCP case; loopback host classification). POC for the Astrid router work — informs the upstream feature request (gotchas + security posture documented separately). --- .../src/engine/wasm/host/net/mod.rs | 121 ++++++++++++++++-- .../src/engine/wasm/host/net/tcp_listener.rs | 113 ++++++++++++++-- .../src/security/manifest_gate.rs | 27 ++++ .../src/security/manifest_gate_tests.rs | 40 ++++++ crates/astrid-capsule/src/security/mod.rs | 26 ++++ 5 files changed, 304 insertions(+), 23 deletions(-) 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..f72fb3ff7 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -62,11 +62,13 @@ pub(super) const MAX_ACTIVE_STREAMS: usize = 8; /// capability token that the capsule must hold to call `accept`. pub(super) struct UnixListenerSlot; -/// Stamp marking a resource slot as a `TcpListener` for future inbound -/// TCP server support. Pre-allocated so the type is in scope even though -/// `bind-tcp` is still a stub. -#[allow(dead_code)] -pub(super) struct TcpListenerSlot; +/// Resource slot holding a bound inbound TCP listener. The +/// `Resource` handed to the guest is a token over this slot; +/// `accept` / `poll-accept` / `local-addr` reach the `tokio` listener +/// through it, and `Drop` closes the socket. +pub(super) struct TcpListenerSlot { + pub(super) listener: Arc, +} /// Stamp marking a resource slot as a `UdpSocket`. Same reason as above. #[allow(dead_code)] @@ -86,6 +88,20 @@ pub(super) fn validate_host(host: &str) -> Result<(), ErrorCode> { Ok(()) } +/// Whether a TCP-bind host names a loopback interface. Capsule-hosted +/// servers are confined to loopback (see `bind_tcp`): `127.0.0.0/8`, `::1`, +/// or the literal `localhost`. A hostname other than `localhost` is refused +/// rather than resolved — binding must name a concrete local interface, and +/// resolving arbitrary names for a bind target is an SSRF-shaped footgun. +pub(super) fn is_loopback_bind_host(host: &str) -> bool { + if host.eq_ignore_ascii_case("localhost") { + return true; + } + host.parse::() + .map(|ip| ip.is_loopback()) + .unwrap_or(false) +} + /// Classify a tokio io::Error into the typed `net::ErrorCode`. pub(super) fn map_io_err(err: std::io::Error) -> ErrorCode { use std::io::ErrorKind; @@ -291,12 +307,75 @@ impl net::Host for HostState { Ok(Resource::new_own(res.rep())) } - fn bind_tcp(&mut self, _host: String, _port: u16) -> Result, ErrorCode> { - // Inbound TCP server hosting — needs a fresh tokio listener + - // capability gate (net_tcp_bind allowlist) + per-capsule accept - // loop. Lands in a follow-up commit; capsules importing - // `bind-tcp` today see CapabilityDenied so they fail closed. - Err(ErrorCode::CapabilityDenied) + fn bind_tcp(&mut self, host: String, port: u16) -> Result, ErrorCode> { + validate_host(&host)?; + let bind_addr = format!("tcp:{host}:{port}"); + + // Capability gate: host:port must match the capsule's `net_bind` + // allowlist (TCP entries share that field with unix binds). + if let Some(ref gate) = self.security { + let capsule_id = self.capsule_id.as_str().to_owned(); + let host_for_check = host.clone(); + let gate = gate.clone(); + let rt = self.runtime_handle.clone(); + let semaphore = self.blocking_semaphore.clone(); + let check = util::bounded_block_on(&rt, &semaphore, async move { + gate.check_net_tcp_bind(&capsule_id, &host_for_check, port) + .await + }); + if let Err(reason) = check { + // Deny path records before the early return (exactly-once). + record_net_denied(self, HostAuditEvent::NetBind { addr: &bind_addr }, &reason); + return Err(ErrorCode::CapabilityDenied); + } + } + + // Security rail: capsule-hosted servers are loopback-only. Exposing a + // capsule listener beyond loopback is a deliberate future opt-in; this + // mirrors `connect-tcp`, which runs its `is_safe_ip` airlock AFTER the + // capability gate. A non-loopback bind is refused here, not silently + // downgraded. + if !is_loopback_bind_host(&host) { + let reason = "non-loopback TCP bind refused (capsule servers are loopback-only)"; + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(reason)); + return Err(ErrorCode::AirlockRejected); + } + + // Bind a fresh tokio listener on the daemon runtime. Quick op — the + // non-cancellable bounded_block_on is fine (accept, which blocks + // indefinitely, uses the cancellable variant instead). + let rt = self.runtime_handle.clone(); + let sem = self.blocking_semaphore.clone(); + let host_owned = host.clone(); + let bind_result: Result = + util::bounded_block_on(&rt, &sem, async move { + tokio::net::TcpListener::bind((host_owned.as_str(), port)).await + }); + let listener = match bind_result { + Ok(l) => l, + Err(e) => { + let mapped = map_io_err(e); + let reason = format!("{mapped:?}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(mapped); + }, + }; + + let slot = TcpListenerSlot { + listener: Arc::new(listener), + }; + let res = match self.resource_table.push(slot) { + Ok(res) => res, + Err(e) => { + // The socket is already bound; the push consumes and drops the + // listener here, releasing it. Record the failure. + let reason = format!("resource table: {e}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(ErrorCode::Unknown(reason)); + }, + }; + audit_net_bind(self, &bind_addr, HostAuditOutcome::Allowed); + Ok(Resource::new_own(res.rep())) } fn connect_tcp(&mut self, host: String, port: u16) -> Result, ErrorCode> { @@ -501,4 +580,24 @@ mod tests { let max = "a".repeat(255); assert!(validate_host(&max).is_ok()); } + + #[test] + fn loopback_bind_host_accepts_loopback() { + assert!(is_loopback_bind_host("127.0.0.1")); + assert!(is_loopback_bind_host("127.0.0.5")); + assert!(is_loopback_bind_host("::1")); + assert!(is_loopback_bind_host("localhost")); + assert!(is_loopback_bind_host("LOCALHOST")); + } + + #[test] + fn loopback_bind_host_rejects_non_loopback() { + assert!(!is_loopback_bind_host("0.0.0.0")); + assert!(!is_loopback_bind_host("192.168.1.10")); + assert!(!is_loopback_bind_host("8.8.8.8")); + assert!(!is_loopback_bind_host("::")); + // A hostname other than localhost is refused (not resolved). + assert!(!is_loopback_bind_host("example.com")); + assert!(!is_loopback_bind_host("")); + } } diff --git a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs index b8d8934ae..2e37e4bb4 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/tcp_listener.rs @@ -1,40 +1,129 @@ //! `HostTcpListener` impl — inbound TCP server hosting. //! -//! STUB SHELL — the bindings type exists in the WIT and the trait must -//! be implemented for the kernel to link. Every method returns -//! `CapabilityDenied` so capsules importing `bind-tcp` fail closed -//! rather than panic. Real impl lands alongside UDP in a follow-up. +//! The listener is created and capability-gated in +//! [`super::Host::bind_tcp`]; the `Resource` is a token over a +//! [`TcpListenerSlot`] holding the live `tokio` listener. `accept` / +//! `poll_accept` produce `TcpStream` resources that reuse the SAME +//! [`NetStream::Tcp`] representation as outbound `connect-tcp` streams, so +//! every existing read / write / peek / timeout host fn works on accepted +//! connections with no extra wiring. + +use std::sync::Arc; use wasmtime::component::Resource; use wasmtime_wasi::p2::DynPollable; -use super::{HostState, TcpListenerSlot}; +use super::{HostState, MAX_ACTIVE_STREAMS, TcpListenerSlot, map_io_err}; use crate::engine::wasm::bindings::astrid::net::host::{ ErrorCode, HostTcpListener, TcpListener, TcpStream, }; +use crate::engine::wasm::host::util; +use crate::engine::wasm::host_state::{NetStream, TcpStreamSlot}; + +impl HostState { + /// Clone the `Arc` out of the resource slot, + /// releasing the table borrow before any blocking accept. + fn tcp_listener_arc( + &self, + rep: u32, + ) -> Result, ErrorCode> { + let slot = self + .resource_table + .get::(&Resource::new_borrow(rep)) + .map_err(|_| ErrorCode::InvalidHandle)?; + Ok(Arc::clone(&slot.listener)) + } + + /// Register an accepted stream as a `NetStream::Tcp` resource, bumping the + /// per-capsule active-stream counter. Shared by `accept` / `poll_accept`. + fn register_accepted( + &mut self, + stream: tokio::net::TcpStream, + ) -> Result, ErrorCode> { + if self.net_stream_count >= MAX_ACTIVE_STREAMS { + drop(stream); + return Err(ErrorCode::Quota); + } + let net_stream = NetStream::Tcp(TcpStreamSlot { + stream: Arc::new(tokio::sync::Mutex::new(stream)), + read_timeout: None, + write_timeout: None, + }); + let res = self + .resource_table + .push(net_stream) + .map_err(|e| ErrorCode::Unknown(format!("resource table: {e}")))?; + self.net_stream_count += 1; + Ok(Resource::new_own(res.rep())) + } +} impl HostTcpListener for HostState { - fn accept(&mut self, _self_: Resource) -> Result, ErrorCode> { - Err(ErrorCode::CapabilityDenied) + fn accept(&mut self, self_: Resource) -> Result, ErrorCode> { + let listener = self.tcp_listener_arc(self_.rep())?; + if self.net_stream_count >= MAX_ACTIVE_STREAMS { + return Err(ErrorCode::Quota); + } + // Mark cooperative progress so a bound accept-loop is not mistaken for + // a no-yield spinner and epoch-trapped (parity with `ipc::recv`). + self.recv_yielded = true; + + let rt = self.runtime_handle.clone(); + let sem = self.blocking_semaphore.clone(); + let tok = self.effective_cancel_token(); + let accepted = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { + listener.accept().await + }); + let stream = match accepted { + Some(Ok((s, _addr))) => s, + Some(Err(e)) => return Err(map_io_err(e)), + None => return Err(ErrorCode::Closed), // cancelled (capsule unload) + }; + self.register_accepted(stream) } fn poll_accept( &mut self, - _self_: Resource, - _timeout_ms: u64, + self_: Resource, + timeout_ms: u64, ) -> Result>, ErrorCode> { - Err(ErrorCode::CapabilityDenied) + let listener = self.tcp_listener_arc(self_.rep())?; + if self.net_stream_count >= MAX_ACTIVE_STREAMS { + return Err(ErrorCode::Quota); + } + self.recv_yielded = true; + + let rt = self.runtime_handle.clone(); + let sem = self.blocking_semaphore.clone(); + let tok = self.effective_cancel_token(); + let timeout = std::time::Duration::from_millis(timeout_ms); + let accepted = util::bounded_block_on_cancellable(&rt, &sem, &tok, async move { + tokio::time::timeout(timeout, listener.accept()).await + }); + match accepted { + Some(Ok(Ok((s, _addr)))) => Ok(Some(self.register_accepted(s)?)), + Some(Ok(Err(e))) => Err(map_io_err(e)), + Some(Err(_elapsed)) => Ok(None), // no connection within the window + None => Err(ErrorCode::Closed), // cancelled (capsule unload) + } } - fn local_addr(&mut self, _self_: Resource) -> Result { - Err(ErrorCode::CapabilityDenied) + fn local_addr(&mut self, self_: Resource) -> Result { + let listener = self.tcp_listener_arc(self_.rep())?; + listener + .local_addr() + .map(|a| a.to_string()) + .map_err(map_io_err) } fn subscribe_readiness(&mut self, _self_: Resource) -> Resource { + // POC: always-ready. Guests use `poll_accept(timeout)` for bounded + // waits; a real readiness pollable over the listener fd is a follow-up. super::super::stubs::always_ready_pollable(&mut self.resource_table) } fn drop(&mut self, rep: Resource) -> wasmtime::Result<()> { + // Deleting the slot drops the Arc → closes the socket. let _ = self .resource_table .delete::(Resource::new_own(rep.rep())); diff --git a/crates/astrid-capsule/src/security/manifest_gate.rs b/crates/astrid-capsule/src/security/manifest_gate.rs index 322b07f81..1ceae125d 100644 --- a/crates/astrid-capsule/src/security/manifest_gate.rs +++ b/crates/astrid-capsule/src/security/manifest_gate.rs @@ -312,6 +312,33 @@ impl CapsuleSecurityGate for ManifestSecurityGate { } } + async fn check_net_tcp_bind( + &self, + capsule_id: &str, + host: &str, + port: u16, + ) -> Result<(), String> { + // Reuse the `net_bind` allowlist (its field documents "Unix/TCP socket + // bind addresses"). TCP entries are `host:port` / `host:*` patterns, + // matched with the SAME semantics as `net_connect`. Unix entries + // (`unix:*`) never match a TCP host:port, so the two socket families + // share the field without cross-authorizing. The host fn confines the + // bind to loopback after this gate returns Ok. + let allowed = self + .manifest + .capabilities + .net_bind + .iter() + .any(|entry| net_connect_pattern_matches(entry, host, port)); + if allowed { + Ok(()) + } else { + Err(format!( + "capsule '{capsule_id}' denied: TCP bind \"{host}:{port}\" not in net_bind allowlist" + )) + } + } + async fn check_identity( &self, capsule_id: &str, diff --git a/crates/astrid-capsule/src/security/manifest_gate_tests.rs b/crates/astrid-capsule/src/security/manifest_gate_tests.rs index 43193d79b..56fd35703 100644 --- a/crates/astrid-capsule/src/security/manifest_gate_tests.rs +++ b/crates/astrid-capsule/src/security/manifest_gate_tests.rs @@ -599,3 +599,43 @@ async fn check_net_connect_matches_allowlist_entry() { ); assert!(gate.check_net_connect("c", "evil.com", 443).await.is_err()); } + +#[tokio::test] +async fn check_net_tcp_bind_matches_net_bind_host_port() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["127.0.0.1:8799".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + // Exact host:port allowed. + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 8799).await.is_ok()); + // Wrong port denied. + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 9000).await.is_err()); + // Wrong host denied. + assert!(gate.check_net_tcp_bind("c", "0.0.0.0", 8799).await.is_err()); +} + +#[tokio::test] +async fn check_net_tcp_bind_wildcard_port() { + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["127.0.0.1:*".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 8799).await.is_ok()); + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 1234).await.is_ok()); +} + +#[tokio::test] +async fn check_net_tcp_bind_unix_entry_does_not_authorize_tcp() { + // The CLI proxy declares `net_bind = ["unix:*"]`. That entry must NEVER + // authorize an inbound TCP bind — the two socket families share the field + // without cross-authorizing. + let mut manifest = make_manifest(vec![], vec![], vec![]); + manifest.capabilities.net_bind = vec!["unix:*".to_string()]; + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 8799).await.is_err()); +} + +#[tokio::test] +async fn check_net_tcp_bind_empty_net_bind_denies() { + let manifest = make_manifest(vec![], vec![], vec![]); + let gate = ManifestSecurityGate::new(manifest, workspace_root(), None); + assert!(gate.check_net_tcp_bind("c", "127.0.0.1", 8799).await.is_err()); +} diff --git a/crates/astrid-capsule/src/security/mod.rs b/crates/astrid-capsule/src/security/mod.rs index 55b19dd97..6705d8663 100644 --- a/crates/astrid-capsule/src/security/mod.rs +++ b/crates/astrid-capsule/src/security/mod.rs @@ -151,6 +151,32 @@ pub trait CapsuleSecurityGate: Send + Sync { )) } + /// Check whether the capsule is allowed to bind an INBOUND TCP listener + /// on `host:port` (capsule-hosted server). + /// + /// Default denies (fail-closed). The manifest gate overrides this to match + /// `host:port` against the capsule's `net_bind` allowlist (whose field + /// documents "Unix/TCP socket bind addresses"). A `unix:*` entry never + /// matches a TCP `host:port`, so the unix-listener path + /// ([`check_net_bind`](Self::check_net_bind)) and this TCP path share the + /// `net_bind` field without cross-authorizing. + /// + /// SECURITY: this gate only enforces the manifest allowlist. The host fn + /// (`bind-tcp`) additionally confines the bind to loopback — the same + /// gate-then-airlock split `connect-tcp` uses (`check_net_connect` then + /// `is_safe_ip`). Exposing a capsule-hosted server beyond loopback is a + /// deliberate future opt-in, not reachable through this method today. + async fn check_net_tcp_bind( + &self, + capsule_id: &str, + _host: &str, + _port: u16, + ) -> Result<(), String> { + Err(format!( + "capsule '{capsule_id}' denied: net_tcp_bind not permitted (default)" + )) + } + /// Check whether the capsule is allowed to register a uplink. /// /// Default implementation permits all registrations. Override to enforce From 7f2da975c17155b5621ce7bba963972bff172564 Mon Sep 17 00:00:00 2001 From: Jamie Steiner Date: Tue, 14 Jul 2026 00:10:04 -0700 Subject: [PATCH 2/3] feat(capsule): deliver per-principal resources to run-loop capsules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An autostarted #[astrid::run] capsule drives its `run` export directly, bypassing invoke_interceptor, so it never received the per-invocation context that path installs: the operator env overlay, secret store, and home:// fs all failed (manifest-declared env keys arrived empty, home:// denied). Only bus-invoked capsules (carrying an inbound principal) got them. Install the owner (ctx.principal) resource context once on the run Store's HostState before the run task spawns — load_invocation_env_overlay + install_principal_overlays — mirroring what a bus invocation from the owner installs. caller_context is deliberately left None so an inbound ipc::recv can still scope per-publisher and effective_principal() keeps resolving the owner. No regression: absent config falls back to the neutral floor exactly as before. Fixes the run-loop half of #1224. Co-Authored-By: Claude Opus 4.8 --- crates/astrid-capsule/src/engine/wasm/mod.rs | 40 ++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/crates/astrid-capsule/src/engine/wasm/mod.rs b/crates/astrid-capsule/src/engine/wasm/mod.rs index eb5e01aa0..3a8d4b958 100644 --- a/crates/astrid-capsule/src/engine/wasm/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/mod.rs @@ -2197,6 +2197,46 @@ impl ExecutionEngine for WasmEngine { ); } + // Install the run loop's per-principal resource context. + // + // A run-loop capsule drives its `run` export directly (the task + // spawned further below) and never goes through + // `invoke_interceptor`, so the per-invocation overlays that path + // installs are never applied here: the env overlay, secret store, + // and `home://` all sit at the neutral deny-all floor. But `run()` + // reads `env::var` / secrets / `home://` for the WHOLE lifetime of + // its single long-lived invocation (e.g. a request loop calling + // `env::var` per request), so the owner context must be installed + // ONCE here, before the run task is spawned — not per message. + // + // Owner = `ctx.principal`, the shared-runtime load owner + // (`PrincipalId::default()` for a run-loop capsule). Scoping to it + // hands the loop ONLY the owner's config — exactly what a bus + // invocation from the owner would install, never another + // principal's data. `caller_context` is deliberately left `None` + // so an inbound `ipc::recv` can still install a per-publisher + // context and `effective_principal()` keeps resolving the owner + // for the loop's own autonomous work. + if has_run { + let mut s = store_arc.as_ref().expect("run-loop has store").lock().await; + let state = s.data_mut(); + state.invocation_env_overlay = + load_invocation_env_overlay(&ctx.principal, state.capsule_id.as_str()); + // Installs invocation_kv + invocation_secret_store + + // invocation_capsule_log + invocation_home/tmp scoped to the + // owner, or clears to the neutral fail-closed floor if the + // owner has no registered home / KV construction fails + // (identical to today's behavior — no regression). + install_principal_overlays(state, Some(&ctx.principal)).await; + tracing::debug!( + capsule = %manifest.package.name, + principal = %ctx.principal, + env_overlay = state.invocation_env_overlay.is_some(), + home = state.invocation_home.is_some(), + "Installed run-loop owner resource context" + ); + } + Ok::<_, CapsuleError>(( pool_opt, store_arc, From 411733e3ab90f2f54efa298c7ae39186ac653291 Mon Sep 17 00:00:00 2001 From: Jamie Steiner Date: Tue, 14 Jul 2026 07:01:15 -0700 Subject: [PATCH 3/3] feat(capsule): concurrent run-loop workers for TCP server capsules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run-loop (#[astrid::run]) capsule hosting a loopback TCP server was pinned to a single Store, so it handled requests serially — parallel clients (e.g. Claude Code subagents) queued behind each other. Add `bind_workers` (CapabilitiesDef): a run-loop capsule declaring net_bind and no host_process runs N worker Stores. Each executes `run()` and shares ONE bound listener via a shared registry (Approach B), blocking on accept() — the OS accept queue load-balances. SO_REUSEPORT was rejected: it does not load-balance on macOS (delivers every connection to the most-recent bind). - capabilities.rs: bind_workers: Option (default None => 1 worker). - discovery.rs: promote bind_workers in the component->root capability merge (the field-by-field merge silently dropped the new scalar field otherwise). - host_state.rs: shared_listeners registry, cloned into each worker HostState. - host/net/mod.rs: bind_tcp dedupes onto the shared Arc (first worker binds under the shard lock; siblings clone) — EADDRINUSE-safe on macOS. - mod.rs: build N worker Stores, per-worker context install (ready_tx / interceptor auto-subscribe / owner overlay), spawn N run tasks; run_handles and ready_rxs become Vecs; wait_ready awaits all N; unload aborts all N. Interceptors + workers>1 is forced to 1 with a warn (N subscriptions would double-process events). N=1 is byte-identical to prior behavior (no regression). Verified: 5 concurrent requests handled in parallel (~2.5s each, vs serial 2/4/6/8/10s); 8 workers spawn; clean teardown across daemon restarts. Co-Authored-By: Claude Opus 4.8 --- .../src/manifest/capabilities.rs | 8 + crates/astrid-capsule/src/discovery.rs | 6 + crates/astrid-capsule/src/engine/mcp_tests.rs | 1 + .../src/engine/wasm/host/net/mod.rs | 60 +- .../src/engine/wasm/host_state.rs | 12 + .../src/engine/wasm/host_state_connection.rs | 13 + .../src/engine/wasm/host_state_hook.rs | 3 + crates/astrid-capsule/src/engine/wasm/mod.rs | 615 ++++++++++-------- .../src/engine/wasm/test_fixtures.rs | 1 + .../src/security/manifest_gate_tests.rs | 1 + .../astrid-integration-tests/tests/mcp_e2e.rs | 1 + .../tests/wasm_e2e.rs | 1 + .../tests/wasm_env_e2e.rs | 1 + 13 files changed, 431 insertions(+), 292 deletions(-) diff --git a/crates/astrid-capsule-types/src/manifest/capabilities.rs b/crates/astrid-capsule-types/src/manifest/capabilities.rs index 76e495591..c594495b9 100644 --- a/crates/astrid-capsule-types/src/manifest/capabilities.rs +++ b/crates/astrid-capsule-types/src/manifest/capabilities.rs @@ -47,6 +47,13 @@ pub struct CapabilitiesDef { /// Unix/TCP socket bind addresses the capsule requires. #[serde(default)] pub net_bind: Vec, + /// Concurrent run-loop workers for a loopback TCP server capsule. Each worker + /// is an independent Store running `run()` and accepting on the shared bound + /// listener, so N workers serve N connections concurrently. None/1 = today's + /// single-instance behavior. Only honored for run-loop capsules that declare + /// net_bind and no host_process. + #[serde(default)] + pub bind_workers: Option, /// Outbound TCP destinations the capsule is allowed to connect to. /// /// Each entry is a `"host:port"` pattern. The `host` portion is a @@ -223,6 +230,7 @@ mod tests { host_process: vec!["bash".into()], allow_persistent: true, net_bind: vec!["127.0.0.1:0".into()], + bind_workers: None, net_connect: vec!["host:443".into()], identity: vec!["resolve".into()], allow_prompt_injection: true, diff --git a/crates/astrid-capsule/src/discovery.rs b/crates/astrid-capsule/src/discovery.rs index 582ad100b..0636e1130 100644 --- a/crates/astrid-capsule/src/discovery.rs +++ b/crates/astrid-capsule/src/discovery.rs @@ -280,6 +280,12 @@ pub fn load_manifest(path: &Path) -> CapsuleResult { .extend(caps.host_process.clone()); manifest.capabilities.net.extend(caps.net.clone()); manifest.capabilities.net_bind.extend(caps.net_bind.clone()); + // Scalar (not a list): a component declaring bind_workers wins over + // the root default. Single-component capsules (the common case) just + // promote it up so the run-loop worker count is honored. + if caps.bind_workers.is_some() { + manifest.capabilities.bind_workers = caps.bind_workers; + } } } diff --git a/crates/astrid-capsule/src/engine/mcp_tests.rs b/crates/astrid-capsule/src/engine/mcp_tests.rs index ab3e064a5..908042891 100644 --- a/crates/astrid-capsule/src/engine/mcp_tests.rs +++ b/crates/astrid-capsule/src/engine/mcp_tests.rs @@ -38,6 +38,7 @@ mod tests { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + bind_workers: None, net_connect: vec![], kv: vec![], fs_read: vec![], 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 f72fb3ff7..bf94ebaeb 100644 --- a/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/host/net/mod.rs @@ -341,29 +341,45 @@ impl net::Host for HostState { return Err(ErrorCode::AirlockRejected); } - // Bind a fresh tokio listener on the daemon runtime. Quick op — the - // non-cancellable bounded_block_on is fine (accept, which blocks - // indefinitely, uses the cancellable variant instead). - let rt = self.runtime_handle.clone(); - let sem = self.blocking_semaphore.clone(); - let host_owned = host.clone(); - let bind_result: Result = - util::bounded_block_on(&rt, &sem, async move { - tokio::net::TcpListener::bind((host_owned.as_str(), port)).await - }); - let listener = match bind_result { - Ok(l) => l, - Err(e) => { - let mapped = map_io_err(e); - let reason = format!("{mapped:?}"); - audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); - return Err(mapped); - }, - }; + // Resolve the listener via the shared registry so a run-loop capsule's + // N worker Stores dedupe onto ONE bound socket (Approach B): the first + // worker binds, the rest observe the Occupied entry and clone its + // `Arc`. All N then block on `accept()` against the single + // OS accept queue, which load-balances (SO_REUSEPORT does NOT on macOS). + // The bind runs UNDER the shard lock so racing workers serialize here — + // without it a second concurrent bind would fail EADDRINUSE (macOS sets + // no SO_REUSEADDR). Quick op — the non-cancellable bounded_block_on is + // fine (accept, which blocks indefinitely, uses the cancellable variant). + // For a non-run-loop pool `shared_listeners` starts empty and each + // instance binds its own address, unchanged from before. + let listener: Arc = + match self.shared_listeners.entry((host.clone(), port)) { + dashmap::mapref::entry::Entry::Occupied(existing) => Arc::clone(existing.get()), + dashmap::mapref::entry::Entry::Vacant(vacant) => { + let rt = self.runtime_handle.clone(); + let sem = self.blocking_semaphore.clone(); + let host_owned = host.clone(); + let bind_result: Result = + util::bounded_block_on(&rt, &sem, async move { + tokio::net::TcpListener::bind((host_owned.as_str(), port)).await + }); + match bind_result { + Ok(l) => { + let listener = Arc::new(l); + vacant.insert(Arc::clone(&listener)); + listener + }, + Err(e) => { + let mapped = map_io_err(e); + let reason = format!("{mapped:?}"); + audit_net_bind(self, &bind_addr, HostAuditOutcome::Failed(&reason)); + return Err(mapped); + }, + } + }, + }; - let slot = TcpListenerSlot { - listener: Arc::new(listener), - }; + let slot = TcpListenerSlot { listener }; let res = match self.resource_table.push(slot) { Ok(res) => res, Err(e) => { diff --git a/crates/astrid-capsule/src/engine/wasm/host_state.rs b/crates/astrid-capsule/src/engine/wasm/host_state.rs index 62568346f..4b949b50b 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state.rs @@ -720,6 +720,18 @@ pub struct HostState { /// anonymous fallback). `Arc` so the binding survives drop landing /// on a different pooled instance than the one that accepted. pub client_connections: Arc>, + /// Bound loopback TCP listeners shared across a run-loop capsule's worker + /// Stores, keyed by `(host, port)`. When `bind_workers > 1`, each of the N + /// worker Stores runs `run()` and calls `bind_tcp` for the same address; + /// the first worker binds the socket and the rest dedupe onto its + /// `Arc` here, so all N block on `accept()` against ONE OS + /// accept queue (which load-balances). `Arc` so the binding is + /// shared across the worker Stores, exactly like + /// [`connection_principals`](Self::connection_principals). Empty for the + /// single-worker default and for non-run-loop pools (each pooled instance + /// still binds independently, matching today's behavior). + pub shared_listeners: + Arc>>, /// Bound run-loop CPU-bound signal: set `true` by the ipc `recv` host fn /// each time the guest blocks on recv, read + cleared by the run-loop's /// epoch-deadline callback once per window. diff --git a/crates/astrid-capsule/src/engine/wasm/host_state_connection.rs b/crates/astrid-capsule/src/engine/wasm/host_state_connection.rs index 2193b0089..483b03f99 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state_connection.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state_connection.rs @@ -29,6 +29,19 @@ impl HostState { Arc::new(dashmap::DashMap::new()) } + /// Build a fresh, empty shared-listener registry. + /// + /// The pooled/run-loop path clones ONE registry into every worker Store so + /// `bind_workers > 1` capsules dedupe onto a single bound `TcpListener` + /// (Approach B). Out-of-pool constructors (lifecycle hooks, the hook + /// handler, tests) use this so they do not have to name the `dashmap` type; + /// they never bind a listener, so the registry stays empty. + #[must_use] + pub fn new_shared_listeners() + -> Arc>> { + Arc::new(dashmap::DashMap::new()) + } + /// Bind `principal` and the authenticating device `key_id` to the /// connection identified by stream resource `rep`. /// 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 b5c591fb5..f54ebfd53 100644 --- a/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs +++ b/crates/astrid-capsule/src/engine/wasm/host_state_hook.rs @@ -135,6 +135,9 @@ impl HostState { // the field. Keyed by the verified principal directly (distinct from // the device-aware `connection_principals` registry). client_connections: Self::new_client_connections(), + // Hooks never bind a listener; a throwaway empty registry satisfies + // the field. + shared_listeners: Self::new_shared_listeners(), // No client frame in flight; hooks never forward over publish-as, so // neither the ingress principal nor its device id / origin is set. ingress_principal: None, diff --git a/crates/astrid-capsule/src/engine/wasm/mod.rs b/crates/astrid-capsule/src/engine/wasm/mod.rs index 3a8d4b958..bed9c6137 100644 --- a/crates/astrid-capsule/src/engine/wasm/mod.rs +++ b/crates/astrid-capsule/src/engine/wasm/mod.rs @@ -155,7 +155,7 @@ pub struct WasmEngine { /// interceptors run concurrently instead of serialising through one Store /// (the throughput floor behind `astrid#813`; see `astrid#816` and /// [`pool`]). `None` for run-loop capsules — they keep one dedicated - /// Store owned by `run_handle` and never go through this pool. The pool is + /// Store(s) owned by `run_handles` and never go through this pool. The pool is /// dynamic: it warm-starts at `min_idle`, grows lazily toward the /// host-derived (operator-overridable) `instance_pool_size` max under load, /// and idle-evicts back down. Capsules carved out via the `host_process` @@ -163,14 +163,21 @@ pub struct WasmEngine { /// handles must never move to a second Store). pool: Option, inbound_rx: Option>, - run_handle: Option>, - /// Receiver for the readiness signal from the run loop. - /// Only set for capsules that have a `run()` export. + /// Background run-loop tasks. One entry for the single-worker default; N + /// entries when a loopback TCP server capsule declares `bind_workers > 1`, + /// each driving its own worker Store's `run()` export against the shared + /// bound listener. Empty for non-run-loop capsules. `unload` aborts every + /// handle after the shared `cancel_token` fires. + run_handles: Vec>, + /// Receivers for the readiness signal from the run loop — one per worker + /// Store (a single entry for the default, N for `bind_workers > 1`). + /// Only populated for capsules that have a `run()` export. /// The Mutex is required because `wait_ready` takes `&self` but we need - /// to clone the receiver (which marks the current value as seen). We + /// to clone each receiver (which marks the current value as seen). We /// clone inside the lock and immediately drop it, so concurrent - /// `wait_ready` calls each get their own independent receiver. - ready_rx: Option>>, + /// `wait_ready` calls each get their own independent receivers. + /// `wait_ready` reports Ready only once EVERY worker has signaled. + ready_rxs: Vec>>, /// Cancellation token for cooperative shutdown of blocking host functions. /// Triggered during `unload()` before aborting the run handle. cancel_token: Option, @@ -330,8 +337,8 @@ impl WasmEngine { wasmtime_engine: None, pool: None, inbound_rx: None, - run_handle: None, - ready_rx: None, + run_handles: Vec::new(), + ready_rxs: Vec::new(), cancel_token: None, principal_cancel_tokens: None, epoch_ticker: None, @@ -882,6 +889,51 @@ pub(crate) const fn exempt_epoch_action(window_ticks: u64) -> EpochAction { EpochAction::Yield(window_ticks) } +/// Apply a run-loop Store's CPU bound: a wasmtime EPOCH deadline plus interrupt +/// callback driven by the shared epoch ticker. Factored out of `load` so every +/// worker Store (one for the single-worker default, N for `bind_workers > 1`) +/// is configured identically. +/// +/// BOUND run-loop (`window_ticks = Some`): the callback runs the pure +/// [`epoch_decision`] each window — a recv/accept loop `Yield`s (never trapped), +/// a no-recv spinner accrues toward `Interrupt` after `MAX_NO_YIELD_WINDOWS` but +/// still `Yield`s during the grace windows, so it can never starve the daemon. +/// +/// EXEMPT run-loop (`window_ticks = None`): unbounded CPU — [`exempt_epoch_action`] +/// always `Yield`s, never `Interrupt`s — but still cooperatively yields the tokio +/// worker every window. `UpdateDeadline::Yield` is async-legal because the run +/// loop drives the guest via `call_async`. +fn configure_run_store(store: &mut Store, run_budget: &RunLoopBudget) { + if let Some(window_ticks) = run_budget.window_ticks { + store.set_epoch_deadline(window_ticks); + store.epoch_deadline_callback(move |mut store_ctx| { + let st = store_ctx.data_mut(); + let (action, recv_yielded, no_yield_windows) = epoch_decision( + st.recv_yielded, + st.no_yield_windows, + window_ticks, + MAX_NO_YIELD_WINDOWS, + ); + st.recv_yielded = recv_yielded; + st.no_yield_windows = no_yield_windows; + Ok(match action { + EpochAction::Yield(ticks) => wasmtime::UpdateDeadline::Yield(ticks), + EpochAction::Interrupt => wasmtime::UpdateDeadline::Interrupt, + }) + }); + } else { + let window_ticks = DEFAULT_RUN_LOOP_WINDOW_TICKS; + store.set_epoch_deadline(window_ticks); + store.epoch_deadline_callback(move |_store_ctx| { + Ok(match exempt_epoch_action(window_ticks) { + EpochAction::Yield(ticks) => wasmtime::UpdateDeadline::Yield(ticks), + // Unreachable: exempt is unbounded, so the policy never traps. + EpochAction::Interrupt => wasmtime::UpdateDeadline::Interrupt, + }) + }); + } +} + /// Build a minimal `WasiCtx` for capsule sandboxing. /// /// Only stderr is inherited so capsule panic messages reach the host. @@ -1441,11 +1493,10 @@ impl ExecutionEngine for WasmEngine { // for the duration of the engine build. let ( pool_opt, - store_arc, - run_instance, + run_stores, rx, has_run, - ready_rx, + ready_rxs, wt_engine, workspace_cow_backend, process_tracker_for_engine, @@ -1807,6 +1858,15 @@ impl ExecutionEngine for WasmEngine { let client_connections: Arc< dashmap::DashMap, > = Arc::new(dashmap::DashMap::new()); + // Bound loopback TCP listeners shared across a run-loop capsule's + // worker Stores. The first worker to `bind_tcp` an address binds the + // socket and inserts here; the rest dedupe onto its Arc so all N + // accept on ONE OS accept queue (Approach B). Cloned into every + // HostState below like `connection_principals`; starts empty and + // stays empty for the single-worker default and non-run-loop pools. + let shared_listeners: Arc< + dashmap::DashMap<(String, u16), Arc>, + > = Arc::new(dashmap::DashMap::new()); let make_state: Arc HostState + Send + Sync> = Arc::new(move || HostState { wasi_ctx: build_wasi_ctx(), resource_table: wasmtime::component::ResourceTable::new(), @@ -1903,6 +1963,7 @@ impl ExecutionEngine for WasmEngine { process_count_by_principal: std::collections::HashMap::new(), connection_principals: connection_principals.clone(), client_connections: client_connections.clone(), + shared_listeners: shared_listeners.clone(), // No frame in flight at construction; both the ingress // principal and its authenticating device key_id are set per // framed read. @@ -1988,6 +2049,38 @@ impl ExecutionEngine for WasmEngine { ) }; + // Concurrent run-loop workers (Approach B, shared listener). Only a + // loopback TCP server capsule (run-loop + `net_bind`, no + // `host_process`) is eligible; everyone else stays single-instance. + // Clamp to [1, instance_pool_size] so an operator misconfiguration + // can't spawn unbounded Stores. `None`/1 ⇒ today's single-worker + // behavior, byte-for-byte. A capsule that ALSO declares interceptors + // is forced back to 1: N auto-subscribed interceptor sets would + // double-process every matching event. + let worker_count = if has_run_export + && manifest.capabilities.host_process.is_empty() + && !manifest.capabilities.net_bind.is_empty() + { + let requested = manifest + .capabilities + .bind_workers + .unwrap_or(1) + .clamp(1, self.runtime_limits.instance_pool_size); + if requested > 1 && !manifest.effective_interceptors().is_empty() { + tracing::warn!( + capsule = %manifest.package.name, + requested, + "bind_workers > 1 ignored: capsule declares interceptors \ + (N subscriptions would double-process events); using 1 worker" + ); + 1 + } else { + requested + } + } else { + 1 + }; + // On-demand instance factory. The eager warm-start instances are // built through it too, so an eagerly-built and a lazily-grown // instance are identical (required for free checkout). The factory @@ -2000,9 +2093,18 @@ impl ExecutionEngine for WasmEngine { pool_epoch_deadline, INTERCEPTOR_FUEL_BUDGET, ); + // Run-loop capsules build one dedicated Store per worker (default 1); + // pools build their `min_idle` warm set. `worker_count == 1` for + // every non-eligible capsule, so this is `pool_min_idle` (today's + // behavior) unless a loopback TCP server opted into `bind_workers`. + let warm_count = if has_run_export { + worker_count + } else { + pool_min_idle + }; let mut initial_instances: Vec = - Vec::with_capacity(pool_min_idle); - for _ in 0..pool_min_idle { + Vec::with_capacity(warm_count); + for _ in 0..warm_count { initial_instances.push(builder.build().await?); } tracing::debug!( @@ -2016,92 +2118,38 @@ impl ExecutionEngine for WasmEngine { ); let has_run = has_run_export; - // Run-loop capsules pull one instance out as a dedicated, - // mutex-guarded Store owned by the run loop; pooled capsules keep - // the whole set for `invoke_interceptor` to lease from. + // Run-loop capsules pull their warm instances out as dedicated, + // mutex-guarded worker Stores owned by the run loop(s); pooled + // capsules keep the whole set for `invoke_interceptor` to lease from. let mut pool_opt: Option = None; - let mut store_arc: Option>>> = None; - let mut run_instance: Option = None; + // One `(Store, Instance)` per worker: a single entry for the + // single-worker default, N for a `bind_workers` loopback TCP server. + // All N share the bound listener via `shared_listeners` and each + // blocks on `accept()` against the one OS accept queue. + let mut run_stores: Vec<( + Arc>>, + wasmtime::component::Instance, + )> = Vec::new(); if has_run { - let mut pi = initial_instances - .pop() - .expect("min_idle >= 1, so the run-loop instance exists"); - // The run-loop Store's memory cap is already baked into - // `store_meter` by `make_state` (pool_size 1 ⇒ this IS the - // run-loop Store) and was enforced during `instantiate_async`. - // Fuel was seeded to INTERCEPTOR_FUEL_BUDGET above for - // instantiation; the run loop is NOT fuel-bound, so re-seed it - // to effectively-infinite (consume_fuel makes a 0-fuel Store - // trap, and we never want a run loop to fuel-out). CPU is - // bounded by the epoch interrupt below, not fuel. - pi.store.set_fuel(u64::MAX).map_err(|e| { - CapsuleError::UnsupportedEntryPoint(format!("Failed to set run-loop fuel: {e}")) - })?; - // Apply the CPU bound: a wasmtime EPOCH deadline + interrupt - // callback driven by the shared epoch ticker. - if let Some(window_ticks) = run_budget.window_ticks { - // BOUND run-loop. The epoch ticker fires the callback every - // `window_ticks` of wall-clock. The callback runs the pure - // `epoch_decision`: - // * a recv/accept loop sets `recv_yielded` (ipc recv host - // fn) → the window resets the counter and `Yield`s - // (cooperatively yields the worker, re-arms) — NEVER - // trapped; - // * a no-recv spinner accrues `no_yield_windows` and is - // `Interrupt`-trapped once it reaches - // MAX_NO_YIELD_WINDOWS — but still `Yield`s during the - // grace windows, so even a pure `loop {}` cooperatively - // yields the tokio worker every window and can NEVER - // starve the daemon (the real worker-starvation fix). - // - // DOCUMENTED RESIDUAL: a `loop { recv(0); burn() }` spammer - // sets `recv_yielded` every iteration, so it is never - // trapped — but because every recv yields the worker it - // also cannot starve the daemon; it can burn one core. An - // OS cgroup is the recommended backstop for that case. - // `UpdateDeadline::Yield` is async-legal here because the - // run loop drives the guest via `call_async` (verified - // against wasmtime 45). - pi.store.set_epoch_deadline(window_ticks); - pi.store.epoch_deadline_callback(move |mut store_ctx| { - let st = store_ctx.data_mut(); - let (action, recv_yielded, no_yield_windows) = epoch_decision( - st.recv_yielded, - st.no_yield_windows, - window_ticks, - MAX_NO_YIELD_WINDOWS, - ); - st.recv_yielded = recv_yielded; - st.no_yield_windows = no_yield_windows; - Ok(match action { - EpochAction::Yield(ticks) => wasmtime::UpdateDeadline::Yield(ticks), - EpochAction::Interrupt => wasmtime::UpdateDeadline::Interrupt, - }) - }); - } else { - // EXEMPT run-loop (CAP_RESOURCES_UNBOUNDED / CAP_NET_BIND / - // CAP_UPLINK on the owner principal): unbounded CPU — never - // `Interrupt`-trapped, never fuel-out — but it must STILL - // cooperatively yield the tokio worker every window so it - // can no longer starve the daemon/SIGTERM handler (the root - // of the wedge). Same finite-window deadline + epoch - // callback the bound run-loop uses, but with the - // yield-ALWAYS policy from `exempt_epoch_action` (never - // `Interrupt`). `UpdateDeadline::Yield` is async-legal here - // because the run loop drives the guest via `call_async`. - let window_ticks = DEFAULT_RUN_LOOP_WINDOW_TICKS; - pi.store.set_epoch_deadline(window_ticks); - pi.store.epoch_deadline_callback(move |_store_ctx| { - Ok(match exempt_epoch_action(window_ticks) { - EpochAction::Yield(ticks) => wasmtime::UpdateDeadline::Yield(ticks), - // Unreachable: exempt is unbounded, so the policy - // never traps. Mapped for exhaustiveness only. - EpochAction::Interrupt => wasmtime::UpdateDeadline::Interrupt, - }) - }); + // Each warm instance becomes a dedicated run-loop Store. The + // Store's memory cap is already baked into `store_meter` by + // `make_state` and was enforced during `instantiate_async`. Fuel + // was seeded to INTERCEPTOR_FUEL_BUDGET above for instantiation; + // the run loop is NOT fuel-bound, so re-seed it to + // effectively-infinite (a 0-fuel Store traps, and a run loop must + // never fuel-out). CPU is bounded by the epoch interrupt + // (`configure_run_store`), not fuel. For the single-worker + // default this drains exactly one instance — identical to the + // prior `pop()` path. + for mut pi in initial_instances.drain(..) { + pi.store.set_fuel(u64::MAX).map_err(|e| { + CapsuleError::UnsupportedEntryPoint(format!( + "Failed to set run-loop fuel: {e}" + )) + })?; + configure_run_store(&mut pi.store, &run_budget); + run_stores.push((Arc::new(AsyncMutex::new(pi.store)), pi.instance)); } - store_arc = Some(Arc::new(AsyncMutex::new(pi.store))); - run_instance = Some(pi.instance); } else { // Free-checkout pools tear down each returned instance's // resource table so a cancelled/panicked invocation can't leak @@ -2121,129 +2169,120 @@ impl ExecutionEngine for WasmEngine { )); } - // Only allocate the watch channel for run-loop capsules. - let ready_rx = if has_run { - let (ready_tx, ready_rx) = tokio::sync::watch::channel(false); - // Async-mutex `lock()` cannot fail (no poisoning) so the - // legacy poisoned-lock conversion is gone. The borrow is - // held synchronously across the small mutation below; - // no `.await` occurs while it is alive. - let mut s = store_arc.as_ref().expect("run-loop has store").lock().await; - s.data_mut().ready_tx = Some(ready_tx); - Some(ready_rx) - } else { - None - }; - - // Auto-subscribe interceptor topics for run-loop capsules. - // Events arrive via the IPC channel the run loop already reads from, - // avoiding mutex contention (no external invoke_interceptor calls). - // - // Note: subscriptions are created before the WASM guest starts, so - // events published between subscribe and the guest's first recv/poll - // call are buffered in the broadcast channel (same as normal IPC). - // RFC cargo-like-manifest: read interceptor bindings from - // [subscribe].handler (new) merged with [[interceptor]] (legacy). - let effective_interceptors = manifest.effective_interceptors(); - if has_run && !effective_interceptors.is_empty() { - // Cap auto-subscribed interceptors to leave headroom for - // guest-initiated subscriptions (shared 128-slot pool). - const MAX_AUTO_SUBSCRIBE: usize = 64; - if effective_interceptors.len() > MAX_AUTO_SUBSCRIBE { - return Err(CapsuleError::UnsupportedEntryPoint(format!( - "Capsule '{}' declares {} interceptors, exceeding the \ - auto-subscribe limit ({MAX_AUTO_SUBSCRIBE})", - manifest.package.name, - effective_interceptors.len() - ))); - } - - // Validate interceptor event patterns have well-formed segments - // (no empty segments, leading/trailing dots, or empty strings). - for interceptor in &effective_interceptors { - if !crate::topic::has_valid_segments(&interceptor.event) { + // Per-worker context install. Each run-loop worker Store gets, BEFORE + // its run task spawns: (a) its own readiness watch channel, (b) the + // auto-subscribed interceptor bindings (metadata under the new ABI), + // and (c) the owner's per-principal resource context. For the + // single-worker default this loops exactly once — behaviorally + // identical to the prior three separate single-store blocks. + // `worker_count > 1` is gated to net_bind capsules with NO + // interceptors (see `worker_count`), so the interceptor install is a + // no-op whenever N > 1. + let mut ready_rxs: Vec> = Vec::new(); + if has_run { + // Auto-subscribe interceptor topics for run-loop capsules. + // Events arrive via the IPC channel the run loop already reads + // from, avoiding mutex contention (no external invoke_interceptor + // calls). + // + // Note: subscriptions are created before the WASM guest starts, + // so events published between subscribe and the guest's first + // recv/poll call are buffered in the broadcast channel (same as + // normal IPC). RFC cargo-like-manifest: read interceptor bindings + // from [subscribe].handler (new) merged with [[interceptor]] + // (legacy). Validated ONCE below; installed per worker. + let effective_interceptors = manifest.effective_interceptors(); + if !effective_interceptors.is_empty() { + // Cap auto-subscribed interceptors to leave headroom for + // guest-initiated subscriptions (shared 128-slot pool). + const MAX_AUTO_SUBSCRIBE: usize = 64; + if effective_interceptors.len() > MAX_AUTO_SUBSCRIBE { return Err(CapsuleError::UnsupportedEntryPoint(format!( - "Interceptor event '{}' has invalid segment structure \ - (empty segments, leading/trailing dots, or empty string)", - interceptor.event + "Capsule '{}' declares {} interceptors, exceeding the \ + auto-subscribe limit ({MAX_AUTO_SUBSCRIBE})", + manifest.package.name, + effective_interceptors.len() ))); } + // Validate interceptor event patterns have well-formed + // segments (no empty segments, leading/trailing dots, or + // empty strings). + for interceptor in &effective_interceptors { + if !crate::topic::has_valid_segments(&interceptor.event) { + return Err(CapsuleError::UnsupportedEntryPoint(format!( + "Interceptor event '{}' has invalid segment structure \ + (empty segments, leading/trailing dots, or empty string)", + interceptor.event + ))); + } + } } - let mut s = store_arc.as_ref().expect("run-loop has store").lock().await; - let state = s.data_mut(); - // Interceptor bindings are metadata under the new - // ABI. The kernel dispatches matching IPC messages to - // `astrid-hook-trigger` directly (no capsule-side - // receiver poll), so we record the action / topic - // mapping but do not allocate an EventReceiver per - // interceptor. `handle-id` is informational only — - // capsules cannot convert it back to a - // `Resource`. - let count = effective_interceptors.len(); - for (idx, interceptor) in effective_interceptors.into_iter().enumerate() { - state - .interceptor_handles - .push(host_state::InterceptorHandle { - handle_id: idx as u64, - action: interceptor.action, - topic: interceptor.event, - }); + // Owner = `ctx.principal`, the shared-runtime load owner + // (`PrincipalId::default()` for a run-loop capsule) — the SAME + // owner for all N workers. A run-loop capsule drives its `run` + // export directly (the tasks spawned below) and never goes + // through `invoke_interceptor`, so the per-invocation overlays + // that path installs are never applied here: without this the env + // overlay, secret store, and `home://` sit at the neutral + // deny-all floor. But `run()` reads `env::var` / secrets / + // `home://` for the WHOLE lifetime of its single long-lived + // invocation, so the owner context is installed ONCE per worker + // here, before its run task spawns — not per message. + // `caller_context` is deliberately left `None` so an inbound + // `ipc::recv` can still install a per-publisher context and + // `effective_principal()` keeps resolving the owner for the + // loop's own autonomous work. + for (store_arc, _inst) in &run_stores { + let (ready_tx, ready_rx) = tokio::sync::watch::channel(false); + // Async-mutex `lock()` cannot fail (no poisoning). Held + // across the awaited overlay install (which needs `&mut + // state`) — safe on tokio's async mutex. + let mut s = store_arc.lock().await; + let state = s.data_mut(); + state.ready_tx = Some(ready_tx); + // Interceptor bindings are metadata under the new ABI. The + // kernel dispatches matching IPC messages to + // `astrid-hook-trigger` directly (no capsule-side receiver + // poll), so we record the action / topic mapping but allocate + // no EventReceiver. `handle-id` is informational only. + for (idx, interceptor) in effective_interceptors.iter().enumerate() { + state + .interceptor_handles + .push(host_state::InterceptorHandle { + handle_id: idx as u64, + action: interceptor.action.clone(), + topic: interceptor.event.clone(), + }); + } + state.invocation_env_overlay = load_invocation_env_overlay( + &ctx.principal, + state.capsule_id.as_str(), + ); + // Installs invocation_kv + invocation_secret_store + + // invocation_capsule_log + invocation_home/tmp scoped to the + // owner, or clears to the neutral fail-closed floor if the + // owner has no registered home / KV construction fails + // (identical to today's behavior — no regression). + install_principal_overlays(state, Some(&ctx.principal)).await; + drop(s); + ready_rxs.push(ready_rx); } - tracing::debug!( - capsule = %manifest.package.name, - count, - "Auto-subscribed interceptors for run-loop capsule" - ); - } - - // Install the run loop's per-principal resource context. - // - // A run-loop capsule drives its `run` export directly (the task - // spawned further below) and never goes through - // `invoke_interceptor`, so the per-invocation overlays that path - // installs are never applied here: the env overlay, secret store, - // and `home://` all sit at the neutral deny-all floor. But `run()` - // reads `env::var` / secrets / `home://` for the WHOLE lifetime of - // its single long-lived invocation (e.g. a request loop calling - // `env::var` per request), so the owner context must be installed - // ONCE here, before the run task is spawned — not per message. - // - // Owner = `ctx.principal`, the shared-runtime load owner - // (`PrincipalId::default()` for a run-loop capsule). Scoping to it - // hands the loop ONLY the owner's config — exactly what a bus - // invocation from the owner would install, never another - // principal's data. `caller_context` is deliberately left `None` - // so an inbound `ipc::recv` can still install a per-publisher - // context and `effective_principal()` keeps resolving the owner - // for the loop's own autonomous work. - if has_run { - let mut s = store_arc.as_ref().expect("run-loop has store").lock().await; - let state = s.data_mut(); - state.invocation_env_overlay = - load_invocation_env_overlay(&ctx.principal, state.capsule_id.as_str()); - // Installs invocation_kv + invocation_secret_store + - // invocation_capsule_log + invocation_home/tmp scoped to the - // owner, or clears to the neutral fail-closed floor if the - // owner has no registered home / KV construction fails - // (identical to today's behavior — no regression). - install_principal_overlays(state, Some(&ctx.principal)).await; tracing::debug!( capsule = %manifest.package.name, principal = %ctx.principal, - env_overlay = state.invocation_env_overlay.is_some(), - home = state.invocation_home.is_some(), - "Installed run-loop owner resource context" + workers = run_stores.len(), + interceptors = effective_interceptors.len(), + "Installed run-loop worker resource context(s)" ); } Ok::<_, CapsuleError>(( pool_opt, - store_arc, - run_instance, + run_stores, rx, has_run, - ready_rx, + ready_rxs, wt_engine, workspace_cow_backend, process_tracker_for_engine, @@ -2344,66 +2383,79 @@ impl ExecutionEngine for WasmEngine { } if has_run { - self.ready_rx = ready_rx.map(tokio::sync::Mutex::new); + self.ready_rxs = ready_rxs + .into_iter() + .map(tokio::sync::Mutex::new) + .collect(); - // The run loop holds the store mutex for its entire lifetime. - // We must NOT store the instance for direct invoke_interceptor use, - // because run-loop capsules receive events via auto-subscribed IPC - // channels instead — no external invoke_interceptor calls. - let capsule_name = self.manifest.package.name.clone(); - let run_store = Arc::clone(store_arc.as_ref().expect("run-loop has store")); - let run_inst = run_instance.expect("run-loop has instance"); - // Clone the instance cancel token so the run loop itself observes - // cancellation. `request_cancel()` (callable through a shared `&self` - // — no `&mut`/`Arc::get_mut` needed) cancels this token; racing it - // against `call_async` guarantees the loop stops even for a compute- - // bound guest that never touches a cancellable host call. Dropping - // the `call_async` future unwinds the wasmtime fiber (identical to - // the `handle.abort()` in `unload`), but here it is reachable while - // a dispatcher consumer still holds an `Arc` clone of this capsule — - // the mechanism a restart uses to fully tear the OLD instance down - // without exclusive ownership. It fires promptly because exempt - // run-loops now cooperatively `Yield` every epoch window (Fix 5), so - // the fiber reaches a yield point where the `select!` can preempt. - let run_cancel = cancel_token.clone(); - // With async wasmtime, `call_async` schedules guest execution - // on a fiber that yields back to the executor on every host - // import boundary. The spawned task no longer needs to be a - // blocking thread — it's an ordinary async task. - self.run_handle = Some(tokio::task::spawn(async move { - tracing::info!(capsule = %capsule_name, "Starting background WASM run loop"); - let mut s = run_store.lock().await; - let typed = match run_inst.get_typed_func::<(), ()>(&mut *s, "run") { - Ok(f) => f, - Err(e) => { - tracing::error!( - capsule = %capsule_name, - error = %e, - "WASM background loop missing `run` export" - ); - return; - }, - }; - tokio::select! { - biased; - () = run_cancel.cancelled() => { - tracing::info!( - capsule = %capsule_name, - "WASM background loop stopped (cancellation requested)" - ); - } - result = typed.call_async(&mut *s, ()) => { - if let Err(e) = result { + // Spawn one run task per worker Store. Each holds its Store's mutex + // for the worker's entire lifetime and drives the guest `run` export + // via `call_async`. We must NOT expose the instances for direct + // invoke_interceptor use, because run-loop capsules receive events + // via auto-subscribed IPC channels instead. For the single-worker + // default this spawns exactly one task — identical to before. + for (worker_idx, (run_store, run_inst)) in run_stores.into_iter().enumerate() { + let capsule_name = self.manifest.package.name.clone(); + // Clone the SHARED instance cancel token so every worker observes + // cancellation. `request_cancel()` (callable through a shared + // `&self` — no `&mut`/`Arc::get_mut` needed) cancels this token; + // racing it against `call_async` guarantees each loop stops even + // for a compute-bound guest that never touches a cancellable host + // call. Dropping the `call_async` future unwinds the wasmtime + // fiber (identical to the `handle.abort()` in `unload`), and it is + // reachable while a dispatcher consumer still holds an `Arc` clone + // of this capsule — the mechanism a restart uses to tear the OLD + // instance down without exclusive ownership. It fires promptly + // because exempt run-loops cooperatively `Yield` every epoch + // window, so the fiber reaches a yield point where the `select!` + // can preempt. + let run_cancel = cancel_token.clone(); + // With async wasmtime, `call_async` schedules guest execution on + // a fiber that yields back to the executor on every host import + // boundary, so each worker is an ordinary async task, not a + // blocking thread. + self.run_handles.push(tokio::task::spawn(async move { + tracing::info!( + capsule = %capsule_name, + worker = worker_idx, + "Starting background WASM run loop" + ); + let mut s = run_store.lock().await; + let typed = match run_inst.get_typed_func::<(), ()>(&mut *s, "run") { + Ok(f) => f, + Err(e) => { tracing::error!( capsule = %capsule_name, + worker = worker_idx, error = %e, - "WASM background loop failed" + "WASM background loop missing `run` export" + ); + return; + }, + }; + tokio::select! { + biased; + () = run_cancel.cancelled() => { + tracing::info!( + capsule = %capsule_name, + worker = worker_idx, + "WASM background loop stopped (cancellation requested)" ); } + result = typed.call_async(&mut *s, ()) => { + if let Err(e) = result { + tracing::error!( + capsule = %capsule_name, + worker = worker_idx, + error = %e, + "WASM background loop failed" + ); + } + } } - } - })); - // The run loop owns the Store via `run_store`; `self.pool` stays + })); + } + // The run loops own the Stores via `run_store`; `self.pool` stays // None so `invoke_interceptor` reports NotSupported for run-loop // capsules (they receive events through auto-subscribed IPC). } else { @@ -2440,21 +2492,24 @@ impl ExecutionEngine for WasmEngine { "Unloading WASM component" ); // Signal cooperative cancellation to unblock ipc_recv/elicit/net calls - // before aborting the run handle. + // before aborting the run handles. The single shared token fans out to + // every worker. if let Some(token) = self.cancel_token.take() { token.cancel(); } - if let Some(handle) = self.run_handle.take() { + // Abort every worker's run task (one for the default, N for + // `bind_workers > 1`). + for handle in self.run_handles.drain(..) { handle.abort(); } // Stop the epoch ticker thread (RAII guard joins on drop). drop(self.epoch_ticker.take()); // Drop the pool — releases every pooled Store's WASM memory. (Run-loop - // capsules have `pool == None`; their Store is owned by the aborted - // run_handle and dropped with it.) + // capsules have `pool == None`; their worker Stores are owned by the + // aborted run tasks and dropped with them.) self.pool = None; self.wasmtime_engine = None; - self.ready_rx = None; // Prevent stale channel observation post-unload + self.ready_rxs.clear(); // Prevent stale channel observation post-unload // Tear down the OS-level CoW working tree (unmount / remove the clone). // Explicit because there is no async Drop for the engine; the backend's // own Drop is a backstop. Uncommitted changes are discarded here — the @@ -2488,13 +2543,29 @@ impl ExecutionEngine for WasmEngine { async fn wait_ready(&self, timeout: std::time::Duration) -> crate::capsule::ReadyStatus { use crate::capsule::ReadyStatus; - let Some(rx_mutex) = &self.ready_rx else { + // No receivers ⇒ non-run-loop capsule (or already unloaded): Ready. + if self.ready_rxs.is_empty() { return ReadyStatus::Ready; + } + // Clone each worker's receiver under its lock (cloning marks the current + // value seen, so concurrent callers each get an independent receiver), + // then wait for EVERY worker to signal readiness within one shared + // timeout budget. Ready only once all workers signaled; Crashed if any + // worker's sender dropped first (its run task died before signaling). + let mut rxs = Vec::with_capacity(self.ready_rxs.len()); + for rx_mutex in &self.ready_rxs { + rxs.push(rx_mutex.lock().await.clone()); + } + let wait_all = async { + for rx in &mut rxs { + if rx.wait_for(|&v| v).await.is_err() { + return ReadyStatus::Crashed; // sender dropped before signaling + } + } + ReadyStatus::Ready }; - let mut rx = rx_mutex.lock().await.clone(); - match tokio::time::timeout(timeout, rx.wait_for(|&v| v)).await { - Ok(Ok(_)) => ReadyStatus::Ready, - Ok(Err(_)) => ReadyStatus::Crashed, // sender dropped before signaling + match tokio::time::timeout(timeout, wait_all).await { + Ok(status) => status, Err(_) => ReadyStatus::Timeout, } } @@ -2865,9 +2936,10 @@ impl ExecutionEngine for WasmEngine { } fn check_health(&self) -> crate::capsule::CapsuleState { - if let Some(handle) = &self.run_handle - && handle.is_finished() - { + // Any worker's run task finishing means its loop exited unexpectedly + // (they run forever until cancelled). One handle for the default, N for + // `bind_workers > 1`. + if self.run_handles.iter().any(|h| h.is_finished()) { return crate::capsule::CapsuleState::Failed( "WASM run loop exited unexpectedly".into(), ); @@ -3088,6 +3160,9 @@ async fn build_lifecycle_host_state( // Lifecycle hooks never accept inbound uplink connections; a throwaway // lifecycle registry satisfies the field. client_connections: Arc::new(dashmap::DashMap::new()), + // Lifecycle hooks never bind a listener; a throwaway empty registry + // satisfies the field. + shared_listeners: Arc::new(dashmap::DashMap::new()), // Lifecycle hooks never forward client frames; no in-flight principal // or authenticating device. ingress_principal: None, diff --git a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs index f5d44d9a3..807e1ab33 100644 --- a/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs +++ b/crates/astrid-capsule/src/engine/wasm/test_fixtures.rs @@ -151,6 +151,7 @@ pub(crate) fn minimal_host_state(rt: tokio::runtime::Handle) -> HostState { process_count_by_principal: HashMap::new(), connection_principals: Arc::new(dashmap::DashMap::new()), client_connections: Arc::new(dashmap::DashMap::new()), + shared_listeners: Arc::new(dashmap::DashMap::new()), ingress_principal: None, ingress_device_key_id: None, ingress_origin: None, diff --git a/crates/astrid-capsule/src/security/manifest_gate_tests.rs b/crates/astrid-capsule/src/security/manifest_gate_tests.rs index 56fd35703..f751c89e5 100644 --- a/crates/astrid-capsule/src/security/manifest_gate_tests.rs +++ b/crates/astrid-capsule/src/security/manifest_gate_tests.rs @@ -33,6 +33,7 @@ fn make_manifest(net: Vec<&str>, fs_read: Vec<&str>, fs_write: Vec<&str>) -> Cap capabilities: CapabilitiesDef { net: net.into_iter().map(String::from).collect(), net_bind: vec![], + bind_workers: None, net_connect: vec![], kv: vec![], fs_read: fs_read.into_iter().map(String::from).collect(), diff --git a/crates/astrid-integration-tests/tests/mcp_e2e.rs b/crates/astrid-integration-tests/tests/mcp_e2e.rs index 7562f419c..6e28a6626 100644 --- a/crates/astrid-integration-tests/tests/mcp_e2e.rs +++ b/crates/astrid-integration-tests/tests/mcp_e2e.rs @@ -36,6 +36,7 @@ async fn test_mcp_host_engine_capability_validation() { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + bind_workers: None, net_connect: vec![], kv: vec![], fs_read: vec![], diff --git a/crates/astrid-integration-tests/tests/wasm_e2e.rs b/crates/astrid-integration-tests/tests/wasm_e2e.rs index e116593ed..035f1bf4d 100644 --- a/crates/astrid-integration-tests/tests/wasm_e2e.rs +++ b/crates/astrid-integration-tests/tests/wasm_e2e.rs @@ -54,6 +54,7 @@ fn build_test_manifest( capabilities: CapabilitiesDef { net: net_caps, net_bind: vec![], + bind_workers: None, net_connect: vec![], kv: vec!["*".into()], fs_read: fs_read_caps, diff --git a/crates/astrid-integration-tests/tests/wasm_env_e2e.rs b/crates/astrid-integration-tests/tests/wasm_env_e2e.rs index c6bec7d45..95f8f4a72 100644 --- a/crates/astrid-integration-tests/tests/wasm_env_e2e.rs +++ b/crates/astrid-integration-tests/tests/wasm_env_e2e.rs @@ -87,6 +87,7 @@ async fn test_wasm_capsule_e2e_env_config_injection() { capabilities: CapabilitiesDef { net: vec![], net_bind: vec![], + bind_workers: None, net_connect: vec![], kv: vec!["*".into()], fs_read: vec![],