From 74d56decaddbd1025a2f96403b0e8c1cc4de0795 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Mon, 21 Sep 2026 14:58:11 -0700 Subject: [PATCH 1/3] fix(sandbox): reclaim socket descriptors before exhaustion Signed-off-by: Piotr Mlocek --- .../openshell-sandbox/src/network_broker.rs | 87 ++++++++++++++++++- 1 file changed, 86 insertions(+), 1 deletion(-) diff --git a/crates/openshell-sandbox/src/network_broker.rs b/crates/openshell-sandbox/src/network_broker.rs index d4972465c6..c3c6fe8bcd 100644 --- a/crates/openshell-sandbox/src/network_broker.rs +++ b/crates/openshell-sandbox/src/network_broker.rs @@ -28,6 +28,8 @@ use openshell_isolation_interface::linux::task_memory; use tokio::sync::{mpsc, oneshot}; const SOCKET_CAPACITY: usize = 4_096; +const SOCKET_FD_HEADROOM_DIVISOR: usize = 2; +const SOCKET_FD_MIN_HEADROOM: usize = 64; const OPEN_QUEUE_CAPACITY: usize = 256; const ACCEPT_WORKER_CAPACITY: usize = 64; const DNS_QUEUE_CAPACITY: usize = 256; @@ -252,7 +254,10 @@ impl NetworkBroker { })?); let (pending_tx, pending_rx) = mpsc::channel(OPEN_QUEUE_CAPACITY); let (pending_dns_tx, pending_dns_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); - let registry = Arc::new(Mutex::new(SocketRegistry::new(1, SOCKET_CAPACITY)?)); + let registry = Arc::new(Mutex::new(SocketRegistry::new( + 1, + socket_registry_capacity()?, + )?)); let active_opens = Arc::new(AtomicUsize::new(0)); let active_accepts = Arc::new(AtomicUsize::new(0)); let dns_relay = start_dns_relay(dns_address, pending_dns_tx)?; @@ -608,6 +613,13 @@ fn create_socket( } else { InetFamily::V6 }; + // Reclaim stale retained descriptors before opening another socket. The + // registry capacity leaves descriptor headroom below RLIMIT_NOFILE so the + // procfs scan can still open directories while it collects closed + // workload sockets. + if let Err(error) = prepare_registry_for_socket(registry) { + return listener.respond_errno(notification.id, error_to_errno(&error)); + } // SAFETY: arguments were reduced to the supported native INET matrix. A // successful call returns one newly owned descriptor. let mut source = unsafe { libc::socket(domain, raw_kind, protocol) }; @@ -643,6 +655,37 @@ fn create_socket( Ok(()) } +fn socket_registry_capacity() -> io::Result { + let mut limit = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + // SAFETY: limit points to writable storage for one rlimit value. + if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) } < 0 { + return Err(io::Error::last_os_error()); + } + let soft_limit = usize::try_from(limit.rlim_cur).unwrap_or(usize::MAX); + Ok(socket_registry_capacity_for_limit(soft_limit)) +} + +fn socket_registry_capacity_for_limit(soft_limit: usize) -> usize { + let headroom = (soft_limit / SOCKET_FD_HEADROOM_DIVISOR).max(SOCKET_FD_MIN_HEADROOM); + soft_limit + .saturating_sub(headroom) + .clamp(1, SOCKET_CAPACITY) +} + +fn prepare_registry_for_socket(registry: &Mutex) -> io::Result<()> { + let mut registry = lock(registry); + if registry.is_full() { + collect_closed_socket_entries_locked(&mut registry)?; + } + if registry.is_full() { + return Err(io::Error::from_raw_os_error(libc::EMFILE)); + } + Ok(()) +} + fn reject_protected_control_destination( destination: SocketAddr, protected_port: Option, @@ -1763,6 +1806,48 @@ mod tests { use std::io::{Read as _, Write as _}; use std::os::unix::net::{UnixListener, UnixStream}; + #[test] + fn socket_registry_capacity_reserves_process_descriptor_headroom() { + assert_eq!(socket_registry_capacity_for_limit(1_024), 512); + assert_eq!(socket_registry_capacity_for_limit(128), 64); + assert_eq!(socket_registry_capacity_for_limit(64), 1); + assert_eq!( + socket_registry_capacity_for_limit(usize::MAX), + SOCKET_CAPACITY + ); + } + + #[test] + fn socket_registry_reclaims_stale_entry_before_opening_another_socket() { + // SAFETY: socket returns one newly owned descriptor on success. + let fd = unsafe { + libc::socket( + libc::AF_INET, + libc::SOCK_STREAM | libc::SOCK_CLOEXEC, + libc::IPPROTO_TCP, + ) + }; + assert!(fd >= 0, "socket: {}", io::Error::last_os_error()); + // SAFETY: successful socket returned one owned descriptor. + let socket = unsafe { OwnedFd::from_raw_fd(fd) }; + let metadata = SocketMetadata { + family: InetFamily::V4, + kind: InetKind::Tcp, + close_on_exec: true, + nonblocking: false, + creator_generation: 1, + }; + let mut registry = SocketRegistry::new(1, 1).unwrap(); + let tentative = registry.stage(socket, metadata).unwrap(); + registry.commit(tentative).unwrap(); + assert!(registry.is_full()); + + let registry = Mutex::new(registry); + prepare_registry_for_socket(®istry).unwrap(); + + assert!(lock(®istry).is_empty()); + } + #[test] fn notification_receive_retries_interrupted_and_disappeared_targets() { assert!(retry_notification_receive(&io::Error::from( From b64fd22c9d5561b96fbc817d7812ae67769c8742 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Mon, 21 Sep 2026 16:25:00 -0700 Subject: [PATCH 2/3] fix(sandbox): separate socket and descriptor limits Signed-off-by: Piotr Mlocek --- .../src/linux/socket_registry.rs | 36 ++++++ .../openshell-sandbox/src/network_broker.rs | 104 +++++++++++++----- 2 files changed, 111 insertions(+), 29 deletions(-) diff --git a/crates/openshell-isolation-interface/src/linux/socket_registry.rs b/crates/openshell-isolation-interface/src/linux/socket_registry.rs index 81a71920e3..7a245d1eac 100644 --- a/crates/openshell-isolation-interface/src/linux/socket_registry.rs +++ b/crates/openshell-isolation-interface/src/linux/socket_registry.rs @@ -205,6 +205,15 @@ impl SocketRegistry { self.entries.is_empty() } + /// Number of entries that still retain a broker-owned descriptor. + #[must_use] + pub fn retained_preconnect_count(&self) -> usize { + self.entries + .values() + .filter(|entry| entry.retained_preconnect.is_some()) + .count() + } + /// Whether another socket would exceed the configured bound. #[must_use] pub fn is_full(&self) -> bool { @@ -458,4 +467,31 @@ mod tests { assert!(registry.remove_inode(first.inode)); assert!(!registry.remove_inode(second.inode)); } + + #[test] + fn retained_count_excludes_connected_metadata() { + let mut registry = SocketRegistry::new(13, 2).unwrap(); + let retained = registry + .commit(registry.stage(tcp_socket(), metadata()).unwrap()) + .unwrap(); + let connected = registry + .commit(registry.stage(tcp_socket(), metadata()).unwrap()) + .unwrap(); + + assert_eq!(registry.retained_preconnect_count(), 2); + + let entry = registry.entries.get_mut(&connected.inode).unwrap(); + entry.set_state(SocketState::Connected { + original_peer: "127.0.0.1:443".parse().unwrap(), + }); + entry.release_preconnect(); + + assert_eq!(registry.len(), 2); + assert_eq!(registry.retained_preconnect_count(), 1); + assert!( + registry.entries[&retained.inode] + .retained_preconnect + .is_some() + ); + } } diff --git a/crates/openshell-sandbox/src/network_broker.rs b/crates/openshell-sandbox/src/network_broker.rs index c3c6fe8bcd..4e5b1c6fe2 100644 --- a/crates/openshell-sandbox/src/network_broker.rs +++ b/crates/openshell-sandbox/src/network_broker.rs @@ -28,8 +28,7 @@ use openshell_isolation_interface::linux::task_memory; use tokio::sync::{mpsc, oneshot}; const SOCKET_CAPACITY: usize = 4_096; -const SOCKET_FD_HEADROOM_DIVISOR: usize = 2; -const SOCKET_FD_MIN_HEADROOM: usize = 64; +const SOCKET_FD_HEADROOM: usize = 64; const OPEN_QUEUE_CAPACITY: usize = 256; const ACCEPT_WORKER_CAPACITY: usize = 64; const DNS_QUEUE_CAPACITY: usize = 256; @@ -198,6 +197,7 @@ struct NotificationQueues { dns_relay: DnsRelay, active_opens: Arc, active_accepts: Arc, + retained_socket_capacity: usize, decision_timeout: Duration, } @@ -254,10 +254,8 @@ impl NetworkBroker { })?); let (pending_tx, pending_rx) = mpsc::channel(OPEN_QUEUE_CAPACITY); let (pending_dns_tx, pending_dns_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); - let registry = Arc::new(Mutex::new(SocketRegistry::new( - 1, - socket_registry_capacity()?, - )?)); + let retained_socket_capacity = retained_socket_capacity()?; + let registry = Arc::new(Mutex::new(SocketRegistry::new(1, SOCKET_CAPACITY)?)); let active_opens = Arc::new(AtomicUsize::new(0)); let active_accepts = Arc::new(AtomicUsize::new(0)); let dns_relay = start_dns_relay(dns_address, pending_dns_tx)?; @@ -270,6 +268,7 @@ impl NetworkBroker { dns_relay, active_opens, active_accepts, + retained_socket_capacity, decision_timeout, }; let healthy = Arc::new(AtomicBool::new(true)); @@ -544,7 +543,12 @@ fn dispatch_notification( ); } if syscall == libc::SYS_socket { - return create_socket(®istry, &listener, notification); + return create_socket( + ®istry, + &listener, + notification, + queues.retained_socket_capacity, + ); } if syscall == libc::SYS_connect { return connect_socket(registry, listener, notification, queues); @@ -592,6 +596,7 @@ fn create_socket( registry: &Mutex, listener: &NotificationListener, notification: Notification, + retained_socket_capacity: usize, ) -> io::Result<()> { let domain = i32::try_from(notification.args[0]) .map_err(|_| io::Error::from_raw_os_error(libc::EAFNOSUPPORT))?; @@ -613,11 +618,10 @@ fn create_socket( } else { InetFamily::V6 }; - // Reclaim stale retained descriptors before opening another socket. The - // registry capacity leaves descriptor headroom below RLIMIT_NOFILE so the - // procfs scan can still open directories while it collects closed - // workload sockets. - if let Err(error) = prepare_registry_for_socket(registry) { + // Reclaim stale descriptors before the broker exhausts its process limit. + // Connected sockets remain in the metadata registry without consuming + // this broker-owned descriptor budget. + if let Err(error) = prepare_registry_for_socket(registry, retained_socket_capacity) { return listener.respond_errno(notification.id, error_to_errno(&error)); } // SAFETY: arguments were reduced to the supported native INET matrix. A @@ -655,7 +659,7 @@ fn create_socket( Ok(()) } -fn socket_registry_capacity() -> io::Result { +fn retained_socket_capacity() -> io::Result { let mut limit = libc::rlimit { rlim_cur: 0, rlim_max: 0, @@ -665,22 +669,24 @@ fn socket_registry_capacity() -> io::Result { return Err(io::Error::last_os_error()); } let soft_limit = usize::try_from(limit.rlim_cur).unwrap_or(usize::MAX); - Ok(socket_registry_capacity_for_limit(soft_limit)) + Ok(retained_socket_capacity_for_limit(soft_limit)) } -fn socket_registry_capacity_for_limit(soft_limit: usize) -> usize { - let headroom = (soft_limit / SOCKET_FD_HEADROOM_DIVISOR).max(SOCKET_FD_MIN_HEADROOM); +fn retained_socket_capacity_for_limit(soft_limit: usize) -> usize { soft_limit - .saturating_sub(headroom) + .saturating_sub(SOCKET_FD_HEADROOM) .clamp(1, SOCKET_CAPACITY) } -fn prepare_registry_for_socket(registry: &Mutex) -> io::Result<()> { +fn prepare_registry_for_socket( + registry: &Mutex, + retained_socket_capacity: usize, +) -> io::Result<()> { let mut registry = lock(registry); - if registry.is_full() { + if registry.is_full() || registry.retained_preconnect_count() >= retained_socket_capacity { collect_closed_socket_entries_locked(&mut registry)?; } - if registry.is_full() { + if registry.is_full() || registry.retained_preconnect_count() >= retained_socket_capacity { return Err(io::Error::from_raw_os_error(libc::EMFILE)); } Ok(()) @@ -1807,18 +1813,18 @@ mod tests { use std::os::unix::net::{UnixListener, UnixStream}; #[test] - fn socket_registry_capacity_reserves_process_descriptor_headroom() { - assert_eq!(socket_registry_capacity_for_limit(1_024), 512); - assert_eq!(socket_registry_capacity_for_limit(128), 64); - assert_eq!(socket_registry_capacity_for_limit(64), 1); + fn retained_socket_capacity_reserves_process_descriptor_headroom() { + assert_eq!(retained_socket_capacity_for_limit(1_024), 960); + assert_eq!(retained_socket_capacity_for_limit(128), 64); + assert_eq!(retained_socket_capacity_for_limit(64), 1); assert_eq!( - socket_registry_capacity_for_limit(usize::MAX), + retained_socket_capacity_for_limit(usize::MAX), SOCKET_CAPACITY ); } #[test] - fn socket_registry_reclaims_stale_entry_before_opening_another_socket() { + fn retained_socket_limit_reclaims_stale_entry_before_opening_another_socket() { // SAFETY: socket returns one newly owned descriptor on success. let fd = unsafe { libc::socket( @@ -1837,17 +1843,57 @@ mod tests { nonblocking: false, creator_generation: 1, }; - let mut registry = SocketRegistry::new(1, 1).unwrap(); + let mut registry = SocketRegistry::new(1, 2).unwrap(); let tentative = registry.stage(socket, metadata).unwrap(); registry.commit(tentative).unwrap(); - assert!(registry.is_full()); + assert!(!registry.is_full()); + assert_eq!(registry.retained_preconnect_count(), 1); let registry = Mutex::new(registry); - prepare_registry_for_socket(®istry).unwrap(); + prepare_registry_for_socket(®istry, 1).unwrap(); assert!(lock(®istry).is_empty()); } + #[test] + fn retained_socket_limit_does_not_cap_connected_metadata() { + // SAFETY: socket returns one newly owned descriptor on success. + let fd = unsafe { + libc::socket( + libc::AF_INET, + libc::SOCK_STREAM | libc::SOCK_CLOEXEC, + libc::IPPROTO_TCP, + ) + }; + assert!(fd >= 0, "socket: {}", io::Error::last_os_error()); + // SAFETY: successful socket returned one owned descriptor. + let socket = unsafe { OwnedFd::from_raw_fd(fd) }; + let metadata = SocketMetadata { + family: InetFamily::V4, + kind: InetKind::Tcp, + close_on_exec: true, + nonblocking: false, + creator_generation: 1, + }; + let mut registry = SocketRegistry::new(1, 2).unwrap(); + let tentative = registry.stage(socket, metadata).unwrap(); + registry + .commit_with_state( + tentative, + SocketState::Connected { + original_peer: "127.0.0.1:443".parse().unwrap(), + }, + ) + .unwrap(); + assert_eq!(registry.len(), 1); + assert_eq!(registry.retained_preconnect_count(), 0); + + let registry = Mutex::new(registry); + prepare_registry_for_socket(®istry, 1).unwrap(); + + assert_eq!(lock(®istry).len(), 1); + } + #[test] fn notification_receive_retries_interrupted_and_disappeared_targets() { assert!(retry_notification_receive(&io::Error::from( From 2b2b7996962e581c294ed48b64c08645b2d5e87a Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Mon, 21 Sep 2026 16:31:09 -0700 Subject: [PATCH 3/3] fix(sandbox): account for existing broker descriptors Signed-off-by: Piotr Mlocek --- .../openshell-sandbox/src/network_broker.rs | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/crates/openshell-sandbox/src/network_broker.rs b/crates/openshell-sandbox/src/network_broker.rs index 4e5b1c6fe2..d093ebf199 100644 --- a/crates/openshell-sandbox/src/network_broker.rs +++ b/crates/openshell-sandbox/src/network_broker.rs @@ -254,12 +254,12 @@ impl NetworkBroker { })?); let (pending_tx, pending_rx) = mpsc::channel(OPEN_QUEUE_CAPACITY); let (pending_dns_tx, pending_dns_rx) = mpsc::channel(DNS_QUEUE_CAPACITY); - let retained_socket_capacity = retained_socket_capacity()?; - let registry = Arc::new(Mutex::new(SocketRegistry::new(1, SOCKET_CAPACITY)?)); let active_opens = Arc::new(AtomicUsize::new(0)); let active_accepts = Arc::new(AtomicUsize::new(0)); let dns_relay = start_dns_relay(dns_address, pending_dns_tx)?; let dns_address = dns_relay.address; + let retained_socket_capacity = retained_socket_capacity()?; + let registry = Arc::new(Mutex::new(SocketRegistry::new(1, SOCKET_CAPACITY)?)); let queues = NotificationQueues { protected_control_port, accept_registrar: accept_monitor.registrar(), @@ -669,11 +669,16 @@ fn retained_socket_capacity() -> io::Result { return Err(io::Error::last_os_error()); } let soft_limit = usize::try_from(limit.rlim_cur).unwrap_or(usize::MAX); - Ok(retained_socket_capacity_for_limit(soft_limit)) + let open_descriptors = std::fs::read_dir("/proc/self/fd")?.count(); + Ok(retained_socket_capacity_for_limit( + soft_limit, + open_descriptors, + )) } -fn retained_socket_capacity_for_limit(soft_limit: usize) -> usize { +fn retained_socket_capacity_for_limit(soft_limit: usize, open_descriptors: usize) -> usize { soft_limit + .saturating_sub(open_descriptors) .saturating_sub(SOCKET_FD_HEADROOM) .clamp(1, SOCKET_CAPACITY) } @@ -1814,11 +1819,11 @@ mod tests { #[test] fn retained_socket_capacity_reserves_process_descriptor_headroom() { - assert_eq!(retained_socket_capacity_for_limit(1_024), 960); - assert_eq!(retained_socket_capacity_for_limit(128), 64); - assert_eq!(retained_socket_capacity_for_limit(64), 1); + assert_eq!(retained_socket_capacity_for_limit(1_024, 24), 936); + assert_eq!(retained_socket_capacity_for_limit(128, 32), 32); + assert_eq!(retained_socket_capacity_for_limit(64, 0), 1); assert_eq!( - retained_socket_capacity_for_limit(usize::MAX), + retained_socket_capacity_for_limit(usize::MAX, 0), SOCKET_CAPACITY ); }