@@ -28,6 +28,7 @@ use openshell_isolation_interface::linux::task_memory;
2828use tokio:: sync:: { mpsc, oneshot} ;
2929
3030const SOCKET_CAPACITY : usize = 4_096 ;
31+ const SOCKET_FD_HEADROOM : usize = 64 ;
3132const OPEN_QUEUE_CAPACITY : usize = 256 ;
3233const ACCEPT_WORKER_CAPACITY : usize = 64 ;
3334const DNS_QUEUE_CAPACITY : usize = 256 ;
@@ -196,6 +197,7 @@ struct NotificationQueues {
196197 dns_relay : DnsRelay ,
197198 active_opens : Arc < AtomicUsize > ,
198199 active_accepts : Arc < AtomicUsize > ,
200+ retained_socket_capacity : usize ,
199201 decision_timeout : Duration ,
200202}
201203
@@ -252,11 +254,12 @@ impl NetworkBroker {
252254 } ) ?) ;
253255 let ( pending_tx, pending_rx) = mpsc:: channel ( OPEN_QUEUE_CAPACITY ) ;
254256 let ( pending_dns_tx, pending_dns_rx) = mpsc:: channel ( DNS_QUEUE_CAPACITY ) ;
255- let registry = Arc :: new ( Mutex :: new ( SocketRegistry :: new ( 1 , SOCKET_CAPACITY ) ?) ) ;
256257 let active_opens = Arc :: new ( AtomicUsize :: new ( 0 ) ) ;
257258 let active_accepts = Arc :: new ( AtomicUsize :: new ( 0 ) ) ;
258259 let dns_relay = start_dns_relay ( dns_address, pending_dns_tx) ?;
259260 let dns_address = dns_relay. address ;
261+ let retained_socket_capacity = retained_socket_capacity ( ) ?;
262+ let registry = Arc :: new ( Mutex :: new ( SocketRegistry :: new ( 1 , SOCKET_CAPACITY ) ?) ) ;
260263 let queues = NotificationQueues {
261264 protected_control_port,
262265 accept_registrar : accept_monitor. registrar ( ) ,
@@ -265,6 +268,7 @@ impl NetworkBroker {
265268 dns_relay,
266269 active_opens,
267270 active_accepts,
271+ retained_socket_capacity,
268272 decision_timeout,
269273 } ;
270274 let healthy = Arc :: new ( AtomicBool :: new ( true ) ) ;
@@ -539,7 +543,12 @@ fn dispatch_notification(
539543 ) ;
540544 }
541545 if syscall == libc:: SYS_socket {
542- return create_socket ( & registry, & listener, notification) ;
546+ return create_socket (
547+ & registry,
548+ & listener,
549+ notification,
550+ queues. retained_socket_capacity ,
551+ ) ;
543552 }
544553 if syscall == libc:: SYS_connect {
545554 return connect_socket ( registry, listener, notification, queues) ;
@@ -587,6 +596,7 @@ fn create_socket(
587596 registry : & Mutex < SocketRegistry > ,
588597 listener : & NotificationListener ,
589598 notification : Notification ,
599+ retained_socket_capacity : usize ,
590600) -> io:: Result < ( ) > {
591601 let domain = i32:: try_from ( notification. args [ 0 ] )
592602 . map_err ( |_| io:: Error :: from_raw_os_error ( libc:: EAFNOSUPPORT ) ) ?;
@@ -608,6 +618,12 @@ fn create_socket(
608618 } else {
609619 InetFamily :: V6
610620 } ;
621+ // Reclaim stale descriptors before the broker exhausts its process limit.
622+ // Connected sockets remain in the metadata registry without consuming
623+ // this broker-owned descriptor budget.
624+ if let Err ( error) = prepare_registry_for_socket ( registry, retained_socket_capacity) {
625+ return listener. respond_errno ( notification. id , error_to_errno ( & error) ) ;
626+ }
611627 // SAFETY: arguments were reduced to the supported native INET matrix. A
612628 // successful call returns one newly owned descriptor.
613629 let mut source = unsafe { libc:: socket ( domain, raw_kind, protocol) } ;
@@ -643,6 +659,44 @@ fn create_socket(
643659 Ok ( ( ) )
644660}
645661
662+ fn retained_socket_capacity ( ) -> io:: Result < usize > {
663+ let mut limit = libc:: rlimit {
664+ rlim_cur : 0 ,
665+ rlim_max : 0 ,
666+ } ;
667+ // SAFETY: limit points to writable storage for one rlimit value.
668+ if unsafe { libc:: getrlimit ( libc:: RLIMIT_NOFILE , & raw mut limit) } < 0 {
669+ return Err ( io:: Error :: last_os_error ( ) ) ;
670+ }
671+ let soft_limit = usize:: try_from ( limit. rlim_cur ) . unwrap_or ( usize:: MAX ) ;
672+ let open_descriptors = std:: fs:: read_dir ( "/proc/self/fd" ) ?. count ( ) ;
673+ Ok ( retained_socket_capacity_for_limit (
674+ soft_limit,
675+ open_descriptors,
676+ ) )
677+ }
678+
679+ fn retained_socket_capacity_for_limit ( soft_limit : usize , open_descriptors : usize ) -> usize {
680+ soft_limit
681+ . saturating_sub ( open_descriptors)
682+ . saturating_sub ( SOCKET_FD_HEADROOM )
683+ . clamp ( 1 , SOCKET_CAPACITY )
684+ }
685+
686+ fn prepare_registry_for_socket (
687+ registry : & Mutex < SocketRegistry > ,
688+ retained_socket_capacity : usize ,
689+ ) -> io:: Result < ( ) > {
690+ let mut registry = lock ( registry) ;
691+ if registry. is_full ( ) || registry. retained_preconnect_count ( ) >= retained_socket_capacity {
692+ collect_closed_socket_entries_locked ( & mut registry) ?;
693+ }
694+ if registry. is_full ( ) || registry. retained_preconnect_count ( ) >= retained_socket_capacity {
695+ return Err ( io:: Error :: from_raw_os_error ( libc:: EMFILE ) ) ;
696+ }
697+ Ok ( ( ) )
698+ }
699+
646700fn reject_protected_control_destination (
647701 destination : SocketAddr ,
648702 protected_port : Option < u16 > ,
@@ -1763,6 +1817,88 @@ mod tests {
17631817 use std:: io:: { Read as _, Write as _} ;
17641818 use std:: os:: unix:: net:: { UnixListener , UnixStream } ;
17651819
1820+ #[ test]
1821+ fn retained_socket_capacity_reserves_process_descriptor_headroom ( ) {
1822+ assert_eq ! ( retained_socket_capacity_for_limit( 1_024 , 24 ) , 936 ) ;
1823+ assert_eq ! ( retained_socket_capacity_for_limit( 128 , 32 ) , 32 ) ;
1824+ assert_eq ! ( retained_socket_capacity_for_limit( 64 , 0 ) , 1 ) ;
1825+ assert_eq ! (
1826+ retained_socket_capacity_for_limit( usize :: MAX , 0 ) ,
1827+ SOCKET_CAPACITY
1828+ ) ;
1829+ }
1830+
1831+ #[ test]
1832+ fn retained_socket_limit_reclaims_stale_entry_before_opening_another_socket ( ) {
1833+ // SAFETY: socket returns one newly owned descriptor on success.
1834+ let fd = unsafe {
1835+ libc:: socket (
1836+ libc:: AF_INET ,
1837+ libc:: SOCK_STREAM | libc:: SOCK_CLOEXEC ,
1838+ libc:: IPPROTO_TCP ,
1839+ )
1840+ } ;
1841+ assert ! ( fd >= 0 , "socket: {}" , io:: Error :: last_os_error( ) ) ;
1842+ // SAFETY: successful socket returned one owned descriptor.
1843+ let socket = unsafe { OwnedFd :: from_raw_fd ( fd) } ;
1844+ let metadata = SocketMetadata {
1845+ family : InetFamily :: V4 ,
1846+ kind : InetKind :: Tcp ,
1847+ close_on_exec : true ,
1848+ nonblocking : false ,
1849+ creator_generation : 1 ,
1850+ } ;
1851+ let mut registry = SocketRegistry :: new ( 1 , 2 ) . unwrap ( ) ;
1852+ let tentative = registry. stage ( socket, metadata) . unwrap ( ) ;
1853+ registry. commit ( tentative) . unwrap ( ) ;
1854+ assert ! ( !registry. is_full( ) ) ;
1855+ assert_eq ! ( registry. retained_preconnect_count( ) , 1 ) ;
1856+
1857+ let registry = Mutex :: new ( registry) ;
1858+ prepare_registry_for_socket ( & registry, 1 ) . unwrap ( ) ;
1859+
1860+ assert ! ( lock( & registry) . is_empty( ) ) ;
1861+ }
1862+
1863+ #[ test]
1864+ fn retained_socket_limit_does_not_cap_connected_metadata ( ) {
1865+ // SAFETY: socket returns one newly owned descriptor on success.
1866+ let fd = unsafe {
1867+ libc:: socket (
1868+ libc:: AF_INET ,
1869+ libc:: SOCK_STREAM | libc:: SOCK_CLOEXEC ,
1870+ libc:: IPPROTO_TCP ,
1871+ )
1872+ } ;
1873+ assert ! ( fd >= 0 , "socket: {}" , io:: Error :: last_os_error( ) ) ;
1874+ // SAFETY: successful socket returned one owned descriptor.
1875+ let socket = unsafe { OwnedFd :: from_raw_fd ( fd) } ;
1876+ let metadata = SocketMetadata {
1877+ family : InetFamily :: V4 ,
1878+ kind : InetKind :: Tcp ,
1879+ close_on_exec : true ,
1880+ nonblocking : false ,
1881+ creator_generation : 1 ,
1882+ } ;
1883+ let mut registry = SocketRegistry :: new ( 1 , 2 ) . unwrap ( ) ;
1884+ let tentative = registry. stage ( socket, metadata) . unwrap ( ) ;
1885+ registry
1886+ . commit_with_state (
1887+ tentative,
1888+ SocketState :: Connected {
1889+ original_peer : "127.0.0.1:443" . parse ( ) . unwrap ( ) ,
1890+ } ,
1891+ )
1892+ . unwrap ( ) ;
1893+ assert_eq ! ( registry. len( ) , 1 ) ;
1894+ assert_eq ! ( registry. retained_preconnect_count( ) , 0 ) ;
1895+
1896+ let registry = Mutex :: new ( registry) ;
1897+ prepare_registry_for_socket ( & registry, 1 ) . unwrap ( ) ;
1898+
1899+ assert_eq ! ( lock( & registry) . len( ) , 1 ) ;
1900+ }
1901+
17661902 #[ test]
17671903 fn notification_receive_retries_interrupted_and_disappeared_targets ( ) {
17681904 assert ! ( retry_notification_receive( & io:: Error :: from(
0 commit comments