fix(sandbox): reclaim socket descriptors before exhaustion

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()
+        );
+    }
 }

crates/openshell-sandbox/src/network_broker.rs

@@ -28,6 +28,7 @@ use openshell_isolation_interface::linux::task_memory;
 use tokio::sync::{mpsc, oneshot};
 
 const SOCKET_CAPACITY: usize = 4_096;
+const SOCKET_FD_HEADROOM: usize = 64;
 const OPEN_QUEUE_CAPACITY: usize = 256;
 const ACCEPT_WORKER_CAPACITY: usize = 64;
 const DNS_QUEUE_CAPACITY: usize = 256;
@@ -196,6 +197,7 @@ struct NotificationQueues {
     dns_relay: DnsRelay,
     active_opens: Arc<AtomicUsize>,
     active_accepts: Arc<AtomicUsize>,
+    retained_socket_capacity: usize,
     decision_timeout: Duration,
 }
 
@@ -252,11 +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 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(),
@@ -265,6 +268,7 @@ impl NetworkBroker {
             dns_relay,
             active_opens,
             active_accepts,
+            retained_socket_capacity,
             decision_timeout,
         };
         let healthy = Arc::new(AtomicBool::new(true));
@@ -539,7 +543,12 @@ fn dispatch_notification(
         );
     }
     if syscall == libc::SYS_socket {
-        return create_socket(&registry, &listener, notification);
+        return create_socket(
+            &registry,
+            &listener,
+            notification,
+            queues.retained_socket_capacity,
+        );
     }
     if syscall == libc::SYS_connect {
         return connect_socket(registry, listener, notification, queues);
@@ -587,6 +596,7 @@ fn create_socket(
     registry: &Mutex<SocketRegistry>,
     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))?;
@@ -608,6 +618,12 @@ fn create_socket(
     } else {
         InetFamily::V6
     };
+    // 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
     // successful call returns one newly owned descriptor.
     let mut source = unsafe { libc::socket(domain, raw_kind, protocol) };
@@ -643,6 +659,44 @@ fn create_socket(
     Ok(())
 }
 
+fn retained_socket_capacity() -> io::Result<usize> {
+    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);
+    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, open_descriptors: usize) -> usize {
+    soft_limit
+        .saturating_sub(open_descriptors)
+        .saturating_sub(SOCKET_FD_HEADROOM)
+        .clamp(1, SOCKET_CAPACITY)
+}
+
+fn prepare_registry_for_socket(
+    registry: &Mutex<SocketRegistry>,
+    retained_socket_capacity: usize,
+) -> io::Result<()> {
+    let mut registry = lock(registry);
+    if registry.is_full() || registry.retained_preconnect_count() >= retained_socket_capacity {
+        collect_closed_socket_entries_locked(&mut registry)?;
+    }
+    if registry.is_full() || registry.retained_preconnect_count() >= retained_socket_capacity {
+        return Err(io::Error::from_raw_os_error(libc::EMFILE));
+    }
+    Ok(())
+}
+
 fn reject_protected_control_destination(
     destination: SocketAddr,
     protected_port: Option<u16>,
@@ -1763,6 +1817,88 @@ mod tests {
     use std::io::{Read as _, Write as _};
     use std::os::unix::net::{UnixListener, UnixStream};
 
+    #[test]
+    fn retained_socket_capacity_reserves_process_descriptor_headroom() {
+        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, 0),
+            SOCKET_CAPACITY
+        );
+    }
+
+    #[test]
+    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(
+                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(tentative).unwrap();
+        assert!(!registry.is_full());
+        assert_eq!(registry.retained_preconnect_count(), 1);
+
+        let registry = Mutex::new(registry);
+        prepare_registry_for_socket(&registry, 1).unwrap();
+
+        assert!(lock(&registry).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(&registry, 1).unwrap();
+
+        assert_eq!(lock(&registry).len(), 1);
+    }
+
     #[test]
     fn notification_receive_retries_interrupted_and_disappeared_targets() {
         assert!(retry_notification_receive(&io::Error::from(