From 8d0099f141e9a50e84ac5541e4cbc1d093c633ea Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 20 Jul 2026 11:42:23 -0400 Subject: [PATCH 1/7] fix(proxy): retry with backoff on transient accept errors instead of exiting The proxy accept loop unconditionally broke on any accept() error, permanently killing the proxy while the sandbox continued to report Ready. Replace the break with a sleep-and-continue pattern: EMFILE/ENFILE errors get exponential backoff (100ms to 3.2s) to let file descriptors drain, and all other accept errors get a fixed 100ms backoff matching the existing metadata_server and edge_tunnel patterns. Closes #2337 Signed-off-by: Sean Burdine Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 28 +++++++++++++++++-- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index ab2313b890..a96b56e8c6 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -265,9 +265,11 @@ impl ProxyHandle { } } + let mut consecutive_fd_errors: u32 = 0; loop { match listener.accept().await { Ok((stream, _addr)) => { + consecutive_fd_errors = 0; let opa = opa_engine.clone(); let cache = identity_cache.clone(); let spid = entrypoint_pid.clone(); @@ -314,14 +316,34 @@ impl ProxyHandle { }); } Err(err) => { + // EMFILE (24) / ENFILE (23) indicate FD exhaustion — + // back off exponentially to let connections drain. + let is_fd_exhaustion = matches!( + err.raw_os_error(), + Some(24) | Some(23) + ); + let (severity, backoff) = if is_fd_exhaustion { + consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); + let backoff_ms = 100u64 + .saturating_mul( + 1u64 << consecutive_fd_errors.min(6).saturating_sub(1), + ) + .min(5_000); + (SeverityId::Medium, std::time::Duration::from_millis(backoff_ms)) + } else { + (SeverityId::Low, std::time::Duration::from_millis(100)) + }; let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) - .severity(SeverityId::Low) + .severity(severity) .status(StatusId::Failure) - .message(format!("Proxy accept error: {err}")) + .message(format!( + "Proxy accept error (retrying in {}ms): {err}", + backoff.as_millis(), + )) .build(); ocsf_emit!(event); - break; + tokio::time::sleep(backoff).await; } } } From 629fd7d4cef2348aea55394598c47d2600df1ccc Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 20 Jul 2026 12:23:53 -0400 Subject: [PATCH 2/7] fix(proxy): address review feedback on accept retry backoff - Use libc::EMFILE / libc::ENFILE instead of raw errno values; promote libc from dev-dependency to regular dependency - Extract accept_backoff() and is_fd_exhaustion_error() into testable helpers - Fix unreachable 5s cap: bump exponent limit from min(6) to min(7) so the 5_000ms ceiling is reachable (100 * 2^6 = 6400, capped to 5000) - Add 6 unit tests covering exponential progression, counter reset, saturation at cap, EMFILE/ENFILE detection, and non-FD error rejection Signed-off-by: Sean Burdine Signed-off-by: politerealism --- .../openshell-supervisor-network/Cargo.toml | 4 +- .../openshell-supervisor-network/src/proxy.rs | 80 ++++++++++++++++--- 2 files changed, 71 insertions(+), 13 deletions(-) diff --git a/crates/openshell-supervisor-network/Cargo.toml b/crates/openshell-supervisor-network/Cargo.toml index 1449276f4f..979097f92b 100644 --- a/crates/openshell-supervisor-network/Cargo.toml +++ b/crates/openshell-supervisor-network/Cargo.toml @@ -59,8 +59,10 @@ tokio-tungstenite = { workspace = true } futures = { workspace = true } tracing-subscriber = { workspace = true } -[target.'cfg(unix)'.dev-dependencies] +[target.'cfg(unix)'.dependencies] libc = "0.2" +[target.'cfg(unix)'.dev-dependencies] + [lints] workspace = true diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index a96b56e8c6..39a2e4db97 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -316,20 +316,10 @@ impl ProxyHandle { }); } Err(err) => { - // EMFILE (24) / ENFILE (23) indicate FD exhaustion — - // back off exponentially to let connections drain. - let is_fd_exhaustion = matches!( - err.raw_os_error(), - Some(24) | Some(23) - ); + let is_fd_exhaustion = is_fd_exhaustion_error(&err); let (severity, backoff) = if is_fd_exhaustion { consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); - let backoff_ms = 100u64 - .saturating_mul( - 1u64 << consecutive_fd_errors.min(6).saturating_sub(1), - ) - .min(5_000); - (SeverityId::Medium, std::time::Duration::from_millis(backoff_ms)) + (SeverityId::Medium, accept_backoff(consecutive_fd_errors)) } else { (SeverityId::Low, std::time::Duration::from_millis(100)) }; @@ -373,6 +363,27 @@ fn emit_activity(tx: &Option, denied: bool, deny_group: &'static } } +#[cfg(unix)] +fn is_fd_exhaustion_error(err: &std::io::Error) -> bool { + matches!(err.raw_os_error(), Some(libc::EMFILE) | Some(libc::ENFILE)) +} + +#[cfg(not(unix))] +fn is_fd_exhaustion_error(_err: &std::io::Error) -> bool { + false +} + +const ACCEPT_BACKOFF_BASE_MS: u64 = 100; +const ACCEPT_BACKOFF_MAX_MS: u64 = 5_000; + +fn accept_backoff(consecutive_errors: u32) -> std::time::Duration { + let exponent = consecutive_errors.saturating_sub(1).min(7); + let ms = ACCEPT_BACKOFF_BASE_MS + .saturating_mul(1u64 << exponent) + .min(ACCEPT_BACKOFF_MAX_MS); + std::time::Duration::from_millis(ms) +} + fn l7_inspection_active(l7_route: Option<&L7RouteSnapshot>) -> bool { l7_route.is_some_and(|route| !route.configs.is_empty()) } @@ -9849,4 +9860,49 @@ network_policies: } } } + + #[test] + fn accept_backoff_exponential_progression() { + let ms = |n| accept_backoff(n).as_millis(); + assert_eq!(ms(1), 100); + assert_eq!(ms(2), 200); + assert_eq!(ms(3), 400); + assert_eq!(ms(4), 800); + assert_eq!(ms(5), 1_600); + assert_eq!(ms(6), 3_200); + assert_eq!(ms(7), 5_000); // 6400 capped to 5000 + assert_eq!(ms(8), 5_000); // stays at cap + } + + #[test] + fn accept_backoff_zero_consecutive_errors() { + assert_eq!(accept_backoff(0).as_millis(), 100); + } + + #[test] + fn accept_backoff_saturates_at_cap() { + assert_eq!(accept_backoff(100).as_millis(), 5_000); + assert_eq!(accept_backoff(u32::MAX).as_millis(), 5_000); + } + + #[cfg(unix)] + #[test] + fn is_fd_exhaustion_detects_emfile() { + let err = std::io::Error::from_raw_os_error(libc::EMFILE); + assert!(is_fd_exhaustion_error(&err)); + } + + #[cfg(unix)] + #[test] + fn is_fd_exhaustion_detects_enfile() { + let err = std::io::Error::from_raw_os_error(libc::ENFILE); + assert!(is_fd_exhaustion_error(&err)); + } + + #[cfg(unix)] + #[test] + fn is_fd_exhaustion_rejects_other_errors() { + let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + assert!(!is_fd_exhaustion_error(&err)); + } } From abfe555af0e45201a1d52e57643622aae401ceb8 Mon Sep 17 00:00:00 2001 From: politerealism Date: Thu, 23 Jul 2026 10:51:50 -0400 Subject: [PATCH 3/7] fix(proxy): classify accept errors and exit on terminal failures Three-way error classification for the accept loop: transient errors (EMFILE, ECONNABORTED) retry with backoff, terminal errors (EBADF, EINVAL, ENOTSOCK) exit immediately, and unknown errors exit after 5 consecutive failures. Adds unit tests for the classifier and error handler, plus a subprocess-isolated integration test that validates EMFILE recovery by lowering RLIMIT_NOFILE. Signed-off-by: Quinn Burdine Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 268 +++++++++++++++++- .../tests/accept_fd_exhaustion.rs | 104 +++++++ 2 files changed, 359 insertions(+), 13 deletions(-) create mode 100644 crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 39a2e4db97..fd325c1d10 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -266,10 +266,12 @@ impl ProxyHandle { } let mut consecutive_fd_errors: u32 = 0; + let mut consecutive_unknown_errors: u32 = 0; loop { match listener.accept().await { Ok((stream, _addr)) => { consecutive_fd_errors = 0; + consecutive_unknown_errors = 0; let opa = opa_engine.clone(); let cache = identity_cache.clone(); let spid = entrypoint_pid.clone(); @@ -316,24 +318,24 @@ impl ProxyHandle { }); } Err(err) => { - let is_fd_exhaustion = is_fd_exhaustion_error(&err); - let (severity, backoff) = if is_fd_exhaustion { - consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); - (SeverityId::Medium, accept_backoff(consecutive_fd_errors)) - } else { - (SeverityId::Low, std::time::Duration::from_millis(100)) - }; + let outcome = handle_accept_error( + &err, + &mut consecutive_fd_errors, + &mut consecutive_unknown_errors, + ); + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) .activity(ActivityId::Fail) - .severity(severity) + .severity(outcome.severity) .status(StatusId::Failure) - .message(format!( - "Proxy accept error (retrying in {}ms): {err}", - backoff.as_millis(), - )) + .message(outcome.message) .build(); ocsf_emit!(event); - tokio::time::sleep(backoff).await; + + match outcome.backoff { + Some(backoff) => tokio::time::sleep(backoff).await, + None => break, + } } } } @@ -363,6 +365,29 @@ fn emit_activity(tx: &Option, denied: bool, deny_group: &'static } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AcceptErrorClass { + Transient, + Terminal, + Unknown, +} + +#[cfg(unix)] +fn classify_accept_error(err: &std::io::Error) -> AcceptErrorClass { + match err.raw_os_error() { + Some(libc::EMFILE) | Some(libc::ENFILE) => AcceptErrorClass::Transient, + Some(libc::ECONNABORTED) | Some(libc::ECONNRESET) => AcceptErrorClass::Transient, + Some(libc::EINTR) => AcceptErrorClass::Transient, + Some(libc::EBADF) | Some(libc::EINVAL) | Some(libc::ENOTSOCK) => AcceptErrorClass::Terminal, + _ => AcceptErrorClass::Unknown, + } +} + +#[cfg(not(unix))] +fn classify_accept_error(_err: &std::io::Error) -> AcceptErrorClass { + AcceptErrorClass::Unknown +} + #[cfg(unix)] fn is_fd_exhaustion_error(err: &std::io::Error) -> bool { matches!(err.raw_os_error(), Some(libc::EMFILE) | Some(libc::ENFILE)) @@ -375,6 +400,7 @@ fn is_fd_exhaustion_error(_err: &std::io::Error) -> bool { const ACCEPT_BACKOFF_BASE_MS: u64 = 100; const ACCEPT_BACKOFF_MAX_MS: u64 = 5_000; +const MAX_CONSECUTIVE_UNKNOWN_ERRORS: u32 = 5; fn accept_backoff(consecutive_errors: u32) -> std::time::Duration { let exponent = consecutive_errors.saturating_sub(1).min(7); @@ -384,6 +410,72 @@ fn accept_backoff(consecutive_errors: u32) -> std::time::Duration { std::time::Duration::from_millis(ms) } +struct AcceptErrorOutcome { + severity: SeverityId, + message: String, + backoff: Option, +} + +fn handle_accept_error( + err: &std::io::Error, + consecutive_fd_errors: &mut u32, + consecutive_unknown_errors: &mut u32, +) -> AcceptErrorOutcome { + let class = classify_accept_error(err); + + match class { + AcceptErrorClass::Terminal => AcceptErrorOutcome { + severity: SeverityId::High, + message: format!("Proxy accept error (terminal, exiting): {err}"), + backoff: None, + }, + AcceptErrorClass::Unknown => { + *consecutive_unknown_errors = consecutive_unknown_errors.saturating_add(1); + if *consecutive_unknown_errors > MAX_CONSECUTIVE_UNKNOWN_ERRORS { + AcceptErrorOutcome { + severity: SeverityId::High, + message: format!( + "Proxy accept error (exceeded {MAX_CONSECUTIVE_UNKNOWN_ERRORS} retries, exiting): {err}" + ), + backoff: None, + } + } else { + let backoff = accept_backoff(*consecutive_unknown_errors); + AcceptErrorOutcome { + severity: SeverityId::Medium, + message: format!( + "Proxy accept error (retry {}/{MAX_CONSECUTIVE_UNKNOWN_ERRORS} in {}ms): {err}", + *consecutive_unknown_errors, + backoff.as_millis(), + ), + backoff: Some(backoff), + } + } + } + AcceptErrorClass::Transient => { + *consecutive_unknown_errors = 0; + if is_fd_exhaustion_error(err) { + *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); + let backoff = accept_backoff(*consecutive_fd_errors); + AcceptErrorOutcome { + severity: SeverityId::Medium, + message: format!( + "Proxy accept error (retrying in {}ms): {err}", + backoff.as_millis(), + ), + backoff: Some(backoff), + } + } else { + AcceptErrorOutcome { + severity: SeverityId::Low, + message: format!("Proxy accept error (retrying in 100ms): {err}"), + backoff: Some(std::time::Duration::from_millis(100)), + } + } + } + } +} + fn l7_inspection_active(l7_route: Option<&L7RouteSnapshot>) -> bool { l7_route.is_some_and(|route| !route.configs.is_empty()) } @@ -9905,4 +9997,154 @@ network_policies: let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); assert!(!is_fd_exhaustion_error(&err)); } + + #[cfg(unix)] + #[test] + fn classify_accept_error_fd_exhaustion_is_transient() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::EMFILE)), + AcceptErrorClass::Transient, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENFILE)), + AcceptErrorClass::Transient, + ); + } + + #[cfg(unix)] + #[test] + fn classify_accept_error_connection_errors_are_transient() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ECONNABORTED)), + AcceptErrorClass::Transient, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ECONNRESET)), + AcceptErrorClass::Transient, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::EINTR)), + AcceptErrorClass::Transient, + ); + } + + #[cfg(unix)] + #[test] + fn classify_accept_error_broken_listener_is_terminal() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::EBADF)), + AcceptErrorClass::Terminal, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::EINVAL)), + AcceptErrorClass::Terminal, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOTSOCK)), + AcceptErrorClass::Terminal, + ); + } + + #[cfg(unix)] + #[test] + fn classify_accept_error_unrecognized_errno_is_unknown() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::EPERM)), + AcceptErrorClass::Unknown, + ); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_terminal_exits_immediately() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::EBADF); + let outcome = handle_accept_error(&err, &mut fd, &mut unk); + assert!(outcome.backoff.is_none()); + assert_eq!(outcome.severity, SeverityId::High); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_transient_retries_indefinitely() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::EMFILE); + for i in 1..=20 { + let outcome = handle_accept_error(&err, &mut fd, &mut unk); + assert!(outcome.backoff.is_some(), "should retry on attempt {i}"); + assert_eq!(outcome.severity, SeverityId::Medium); + } + assert_eq!(fd, 20); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_unknown_exits_after_limit() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::EPERM); + for i in 1..=MAX_CONSECUTIVE_UNKNOWN_ERRORS { + let outcome = handle_accept_error(&err, &mut fd, &mut unk); + assert!( + outcome.backoff.is_some(), + "should retry on attempt {i}/{MAX_CONSECUTIVE_UNKNOWN_ERRORS}", + ); + assert_eq!(outcome.severity, SeverityId::Medium); + } + let outcome = handle_accept_error(&err, &mut fd, &mut unk); + assert!( + outcome.backoff.is_none(), + "should exit after limit exceeded" + ); + assert_eq!(outcome.severity, SeverityId::High); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_transient_resets_unknown_counter() { + let mut fd = 0; + let mut unk = 0; + let unknown_err = std::io::Error::from_raw_os_error(libc::EPERM); + let transient_err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + + // Accumulate unknowns up to the limit. + for _ in 1..=MAX_CONSECUTIVE_UNKNOWN_ERRORS { + handle_accept_error(&unknown_err, &mut fd, &mut unk); + } + assert_eq!(unk, MAX_CONSECUTIVE_UNKNOWN_ERRORS); + + // A transient error resets the unknown counter. + let outcome = handle_accept_error(&transient_err, &mut fd, &mut unk); + assert!(outcome.backoff.is_some()); + assert_eq!(unk, 0); + + // Unknown errors can retry again from zero. + let outcome = handle_accept_error(&unknown_err, &mut fd, &mut unk); + assert!(outcome.backoff.is_some()); + assert_eq!(unk, 1); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_fd_exhaustion_uses_exponential_backoff() { + let mut fd = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::EMFILE); + + let b1 = handle_accept_error(&err, &mut fd, &mut unk) + .backoff + .unwrap(); + let b2 = handle_accept_error(&err, &mut fd, &mut unk) + .backoff + .unwrap(); + let b3 = handle_accept_error(&err, &mut fd, &mut unk) + .backoff + .unwrap(); + + assert_eq!(b1.as_millis(), 100); + assert_eq!(b2.as_millis(), 200); + assert_eq!(b3.as_millis(), 400); + } } diff --git a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs new file mode 100644 index 0000000000..b22ef249f6 --- /dev/null +++ b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs @@ -0,0 +1,104 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Subprocess-isolated integration test: lowers `RLIMIT_NOFILE`, induces +//! EMFILE on `accept()`, releases file descriptors, and verifies the +//! listener recovers and accepts a subsequent connection. +//! +//! Uses blocking I/O in the child to surface EMFILE reliably across +//! platforms (tokio's async accept may swallow EMFILE on kqueue-based +//! systems). The proxy's retry logic is validated by the unit tests for +//! `handle_accept_error`; this test validates the OS-level precondition +//! that recovery is possible after FD exhaustion clears. +//! +//! Runs in a child process so the test runner's process-wide resource +//! limits are not affected. + +#![cfg(unix)] +#![allow(unsafe_code, reason = "setrlimit requires unsafe")] + +use std::env; +use std::io::Read; +use std::io::Write; +use std::process::Command; + +const CHILD_SENTINEL: &str = "__ACCEPT_FD_EXHAUSTION_CHILD"; + +#[test] +fn accept_recovers_after_fd_exhaustion() { + let exe = env::current_exe().expect("current_exe"); + let output = Command::new(exe) + .env(CHILD_SENTINEL, "1") + .arg("--test-threads=1") + .arg("accept_fd_exhaustion_child") + .output() + .expect("spawn child"); + + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + output.status.success(), + "child failed (exit {}):\nstdout: {stdout}\nstderr: {stderr}", + output.status, + ); +} + +#[test] +fn accept_fd_exhaustion_child() { + if env::var(CHILD_SENTINEL).is_err() { + return; + } + + let limit = libc::rlimit { + rlim_cur: 32, + rlim_max: 32, + }; + assert_eq!(unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &limit) }, 0); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let addr = listener.local_addr().unwrap(); + + // Place a connection in the kernel backlog before exhausting FDs. + let backlog_conn = std::net::TcpStream::connect(addr).expect("backlog connect"); + + // Exhaust remaining FDs. + let mut held_fds = Vec::new(); + loop { + match std::fs::File::open("/dev/null") { + Ok(f) => held_fds.push(f), + Err(_) => break, + } + } + + // accept() should fail with EMFILE: there's a connection in the + // backlog but no FD available for the accepted socket. + let first = listener.accept(); + match first { + Err(ref e) if e.raw_os_error() == Some(libc::EMFILE) => {} + Err(ref e) => panic!("expected EMFILE, got: {e}"), + Ok(_) => { + // OS found a spare slot — skip gracefully. + return; + } + } + + // Release FDs. + held_fds.clear(); + drop(backlog_conn); + + // Make a fresh connection now that FDs are available. + let client = std::net::TcpStream::connect(addr).expect("connect after FD release"); + + // Retry: accept should now succeed, proving that an accept loop + // retrying on EMFILE (like the proxy does) will recover once FDs + // are available again. + let (accepted, _peer) = listener + .accept() + .expect("accept should succeed after FD release"); + + // Verify the connection is functional. + (&client).write_all(b"ping").expect("write"); + let mut buf = [0u8; 4]; + (&accepted).read_exact(&mut buf).expect("read"); + assert_eq!(&buf, b"ping"); +} From 39705af9f2d2b7732c13a0b606be0898082badba Mon Sep 17 00:00:00 2001 From: politerealism Date: Thu, 23 Jul 2026 17:48:48 -0400 Subject: [PATCH 4/7] fix(proxy): use nested or-patterns to satisfy clippy lint Signed-off-by: Quinn Burdine Signed-off-by: politerealism --- crates/openshell-supervisor-network/src/proxy.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index fd325c1d10..a748f7c3da 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -375,10 +375,10 @@ enum AcceptErrorClass { #[cfg(unix)] fn classify_accept_error(err: &std::io::Error) -> AcceptErrorClass { match err.raw_os_error() { - Some(libc::EMFILE) | Some(libc::ENFILE) => AcceptErrorClass::Transient, - Some(libc::ECONNABORTED) | Some(libc::ECONNRESET) => AcceptErrorClass::Transient, - Some(libc::EINTR) => AcceptErrorClass::Transient, - Some(libc::EBADF) | Some(libc::EINVAL) | Some(libc::ENOTSOCK) => AcceptErrorClass::Terminal, + Some(libc::EMFILE | libc::ENFILE | libc::ECONNABORTED | libc::ECONNRESET | libc::EINTR) => { + AcceptErrorClass::Transient + } + Some(libc::EBADF | libc::EINVAL | libc::ENOTSOCK) => AcceptErrorClass::Terminal, _ => AcceptErrorClass::Unknown, } } @@ -390,7 +390,7 @@ fn classify_accept_error(_err: &std::io::Error) -> AcceptErrorClass { #[cfg(unix)] fn is_fd_exhaustion_error(err: &std::io::Error) -> bool { - matches!(err.raw_os_error(), Some(libc::EMFILE) | Some(libc::ENFILE)) + matches!(err.raw_os_error(), Some(libc::EMFILE | libc::ENFILE)) } #[cfg(not(unix))] From ddae69bebea51b6dbd0f28afe8d3134effe388ac Mon Sep 17 00:00:00 2001 From: politerealism Date: Fri, 24 Jul 2026 17:57:50 -0400 Subject: [PATCH 5/7] fix(proxy): expand transient error allowlist and fix clippy/test issues Widen the accept-error classification to cover all Linux accept(2) transient errnos (ENETDOWN, EPROTO, ENOPROTOOPT, EHOSTDOWN, EHOSTUNREACH, EOPNOTSUPP, ENETUNREACH, ENONET behind cfg gate). Rename fd-exhaustion helpers to resource-pressure to reflect ENOBUFS and ENOMEM coverage. Fix clippy lints in the integration test (borrow_as_ptr, collection_is_never_read, while_let_loop) and make post-EMFILE recovery verification cross-platform by handling Linux's stale backlog behavior. Signed-off-by: Quinn Burdine Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 213 +++++++++++++++--- .../tests/accept_fd_exhaustion.rs | 34 ++- 2 files changed, 200 insertions(+), 47 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index a748f7c3da..129a050227 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -265,12 +265,12 @@ impl ProxyHandle { } } - let mut consecutive_fd_errors: u32 = 0; + let mut consecutive_resource_errors: u32 = 0; let mut consecutive_unknown_errors: u32 = 0; loop { match listener.accept().await { Ok((stream, _addr)) => { - consecutive_fd_errors = 0; + consecutive_resource_errors = 0; consecutive_unknown_errors = 0; let opa = opa_engine.clone(); let cache = identity_cache.clone(); @@ -320,7 +320,7 @@ impl ProxyHandle { Err(err) => { let outcome = handle_accept_error( &err, - &mut consecutive_fd_errors, + &mut consecutive_resource_errors, &mut consecutive_unknown_errors, ); @@ -375,9 +375,24 @@ enum AcceptErrorClass { #[cfg(unix)] fn classify_accept_error(err: &std::io::Error) -> AcceptErrorClass { match err.raw_os_error() { - Some(libc::EMFILE | libc::ENFILE | libc::ECONNABORTED | libc::ECONNRESET | libc::EINTR) => { - AcceptErrorClass::Transient - } + Some( + libc::EMFILE + | libc::ENFILE + | libc::ENOBUFS + | libc::ENOMEM + | libc::ECONNABORTED + | libc::ECONNRESET + | libc::EINTR + | libc::ENETDOWN + | libc::EPROTO + | libc::ENOPROTOOPT + | libc::EHOSTDOWN + | libc::EHOSTUNREACH + | libc::EOPNOTSUPP + | libc::ENETUNREACH, + ) => AcceptErrorClass::Transient, + #[cfg(target_os = "linux")] + Some(libc::ENONET) => AcceptErrorClass::Transient, Some(libc::EBADF | libc::EINVAL | libc::ENOTSOCK) => AcceptErrorClass::Terminal, _ => AcceptErrorClass::Unknown, } @@ -389,12 +404,15 @@ fn classify_accept_error(_err: &std::io::Error) -> AcceptErrorClass { } #[cfg(unix)] -fn is_fd_exhaustion_error(err: &std::io::Error) -> bool { - matches!(err.raw_os_error(), Some(libc::EMFILE | libc::ENFILE)) +fn is_resource_pressure_error(err: &std::io::Error) -> bool { + matches!( + err.raw_os_error(), + Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM) + ) } #[cfg(not(unix))] -fn is_fd_exhaustion_error(_err: &std::io::Error) -> bool { +fn is_resource_pressure_error(_err: &std::io::Error) -> bool { false } @@ -418,7 +436,7 @@ struct AcceptErrorOutcome { fn handle_accept_error( err: &std::io::Error, - consecutive_fd_errors: &mut u32, + consecutive_resource_errors: &mut u32, consecutive_unknown_errors: &mut u32, ) -> AcceptErrorOutcome { let class = classify_accept_error(err); @@ -454,9 +472,9 @@ fn handle_accept_error( } AcceptErrorClass::Transient => { *consecutive_unknown_errors = 0; - if is_fd_exhaustion_error(err) { - *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); - let backoff = accept_backoff(*consecutive_fd_errors); + if is_resource_pressure_error(err) { + *consecutive_resource_errors = consecutive_resource_errors.saturating_add(1); + let backoff = accept_backoff(*consecutive_resource_errors); AcceptErrorOutcome { severity: SeverityId::Medium, message: format!( @@ -9979,23 +9997,34 @@ network_policies: #[cfg(unix)] #[test] - fn is_fd_exhaustion_detects_emfile() { + fn is_resource_pressure_detects_emfile() { let err = std::io::Error::from_raw_os_error(libc::EMFILE); - assert!(is_fd_exhaustion_error(&err)); + assert!(is_resource_pressure_error(&err)); } #[cfg(unix)] #[test] - fn is_fd_exhaustion_detects_enfile() { + fn is_resource_pressure_detects_enfile() { let err = std::io::Error::from_raw_os_error(libc::ENFILE); - assert!(is_fd_exhaustion_error(&err)); + assert!(is_resource_pressure_error(&err)); + } + + #[cfg(unix)] + #[test] + fn is_resource_pressure_detects_memory_pressure() { + assert!(is_resource_pressure_error( + &std::io::Error::from_raw_os_error(libc::ENOBUFS) + )); + assert!(is_resource_pressure_error( + &std::io::Error::from_raw_os_error(libc::ENOMEM) + )); } #[cfg(unix)] #[test] - fn is_fd_exhaustion_rejects_other_errors() { + fn is_resource_pressure_rejects_other_errors() { let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); - assert!(!is_fd_exhaustion_error(&err)); + assert!(!is_resource_pressure_error(&err)); } #[cfg(unix)] @@ -10057,10 +10086,10 @@ network_policies: #[cfg(unix)] #[test] fn handle_accept_error_terminal_exits_immediately() { - let mut fd = 0; + let mut res = 0; let mut unk = 0; let err = std::io::Error::from_raw_os_error(libc::EBADF); - let outcome = handle_accept_error(&err, &mut fd, &mut unk); + let outcome = handle_accept_error(&err, &mut res, &mut unk); assert!(outcome.backoff.is_none()); assert_eq!(outcome.severity, SeverityId::High); } @@ -10068,32 +10097,32 @@ network_policies: #[cfg(unix)] #[test] fn handle_accept_error_transient_retries_indefinitely() { - let mut fd = 0; + let mut res = 0; let mut unk = 0; let err = std::io::Error::from_raw_os_error(libc::EMFILE); for i in 1..=20 { - let outcome = handle_accept_error(&err, &mut fd, &mut unk); + let outcome = handle_accept_error(&err, &mut res, &mut unk); assert!(outcome.backoff.is_some(), "should retry on attempt {i}"); assert_eq!(outcome.severity, SeverityId::Medium); } - assert_eq!(fd, 20); + assert_eq!(res, 20); } #[cfg(unix)] #[test] fn handle_accept_error_unknown_exits_after_limit() { - let mut fd = 0; + let mut res = 0; let mut unk = 0; let err = std::io::Error::from_raw_os_error(libc::EPERM); for i in 1..=MAX_CONSECUTIVE_UNKNOWN_ERRORS { - let outcome = handle_accept_error(&err, &mut fd, &mut unk); + let outcome = handle_accept_error(&err, &mut res, &mut unk); assert!( outcome.backoff.is_some(), "should retry on attempt {i}/{MAX_CONSECUTIVE_UNKNOWN_ERRORS}", ); assert_eq!(outcome.severity, SeverityId::Medium); } - let outcome = handle_accept_error(&err, &mut fd, &mut unk); + let outcome = handle_accept_error(&err, &mut res, &mut unk); assert!( outcome.backoff.is_none(), "should exit after limit exceeded" @@ -10104,24 +10133,24 @@ network_policies: #[cfg(unix)] #[test] fn handle_accept_error_transient_resets_unknown_counter() { - let mut fd = 0; + let mut res = 0; let mut unk = 0; let unknown_err = std::io::Error::from_raw_os_error(libc::EPERM); let transient_err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); // Accumulate unknowns up to the limit. for _ in 1..=MAX_CONSECUTIVE_UNKNOWN_ERRORS { - handle_accept_error(&unknown_err, &mut fd, &mut unk); + handle_accept_error(&unknown_err, &mut res, &mut unk); } assert_eq!(unk, MAX_CONSECUTIVE_UNKNOWN_ERRORS); // A transient error resets the unknown counter. - let outcome = handle_accept_error(&transient_err, &mut fd, &mut unk); + let outcome = handle_accept_error(&transient_err, &mut res, &mut unk); assert!(outcome.backoff.is_some()); assert_eq!(unk, 0); // Unknown errors can retry again from zero. - let outcome = handle_accept_error(&unknown_err, &mut fd, &mut unk); + let outcome = handle_accept_error(&unknown_err, &mut res, &mut unk); assert!(outcome.backoff.is_some()); assert_eq!(unk, 1); } @@ -10129,17 +10158,17 @@ network_policies: #[cfg(unix)] #[test] fn handle_accept_error_fd_exhaustion_uses_exponential_backoff() { - let mut fd = 0; + let mut res = 0; let mut unk = 0; let err = std::io::Error::from_raw_os_error(libc::EMFILE); - let b1 = handle_accept_error(&err, &mut fd, &mut unk) + let b1 = handle_accept_error(&err, &mut res, &mut unk) .backoff .unwrap(); - let b2 = handle_accept_error(&err, &mut fd, &mut unk) + let b2 = handle_accept_error(&err, &mut res, &mut unk) .backoff .unwrap(); - let b3 = handle_accept_error(&err, &mut fd, &mut unk) + let b3 = handle_accept_error(&err, &mut res, &mut unk) .backoff .unwrap(); @@ -10147,4 +10176,118 @@ network_policies: assert_eq!(b2.as_millis(), 200); assert_eq!(b3.as_millis(), 400); } + + #[cfg(unix)] + #[test] + fn classify_accept_error_network_errors_are_transient() { + for errno in [ + libc::ENETDOWN, + libc::EPROTO, + libc::ENOPROTOOPT, + libc::EHOSTDOWN, + libc::EHOSTUNREACH, + libc::EOPNOTSUPP, + libc::ENETUNREACH, + ] { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(errno)), + AcceptErrorClass::Transient, + "errno {errno} should be transient", + ); + } + } + + #[cfg(target_os = "linux")] + #[test] + fn classify_accept_error_enonet_is_transient() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENONET)), + AcceptErrorClass::Transient, + ); + } + + #[cfg(unix)] + #[test] + fn classify_accept_error_resource_pressure_is_transient() { + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOBUFS)), + AcceptErrorClass::Transient, + ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOMEM)), + AcceptErrorClass::Transient, + ); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_non_resource_transient_uses_fixed_backoff() { + let mut res = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + + let o1 = handle_accept_error(&err, &mut res, &mut unk); + let o2 = handle_accept_error(&err, &mut res, &mut unk); + + assert_eq!(o1.severity, SeverityId::Low); + assert_eq!(o1.backoff.unwrap().as_millis(), 100); + assert_eq!(o2.backoff.unwrap().as_millis(), 100); + assert_eq!(res, 0); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_unknown_uses_exponential_backoff() { + let mut res = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::EPERM); + + let b1 = handle_accept_error(&err, &mut res, &mut unk) + .backoff + .unwrap(); + let b2 = handle_accept_error(&err, &mut res, &mut unk) + .backoff + .unwrap(); + let b3 = handle_accept_error(&err, &mut res, &mut unk) + .backoff + .unwrap(); + + assert_eq!(b1.as_millis(), 100); + assert_eq!(b2.as_millis(), 200); + assert_eq!(b3.as_millis(), 400); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_resource_counter_persists_across_mixed_transient() { + let mut res = 0; + let mut unk = 0; + let resource_err = std::io::Error::from_raw_os_error(libc::EMFILE); + let transient_err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + + let o1 = handle_accept_error(&resource_err, &mut res, &mut unk); + assert_eq!(res, 1); + assert_eq!(o1.backoff.unwrap().as_millis(), 100); + + let o2 = handle_accept_error(&transient_err, &mut res, &mut unk); + assert_eq!(res, 1); + assert_eq!(o2.backoff.unwrap().as_millis(), 100); + + let o3 = handle_accept_error(&resource_err, &mut res, &mut unk); + assert_eq!(res, 2); + assert_eq!(o3.backoff.unwrap().as_millis(), 200); + } + + #[cfg(unix)] + #[test] + fn handle_accept_error_terminal_leaves_counters_unchanged() { + let mut res = 3; + let mut unk = 2; + let err = std::io::Error::from_raw_os_error(libc::EBADF); + + let outcome = handle_accept_error(&err, &mut res, &mut unk); + assert!(outcome.backoff.is_none()); + assert_eq!(res, 3); + assert_eq!(unk, 2); + } } diff --git a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs index b22ef249f6..e2c1aaac84 100644 --- a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs +++ b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs @@ -53,7 +53,10 @@ fn accept_fd_exhaustion_child() { rlim_cur: 32, rlim_max: 32, }; - assert_eq!(unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &limit) }, 0); + assert_eq!( + unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, std::ptr::from_ref(&limit)) }, + 0, + ); let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); let addr = listener.local_addr().unwrap(); @@ -62,12 +65,10 @@ fn accept_fd_exhaustion_child() { let backlog_conn = std::net::TcpStream::connect(addr).expect("backlog connect"); // Exhaust remaining FDs. + #[allow(clippy::collection_is_never_read)] let mut held_fds = Vec::new(); - loop { - match std::fs::File::open("/dev/null") { - Ok(f) => held_fds.push(f), - Err(_) => break, - } + while let Ok(f) = std::fs::File::open("/dev/null") { + held_fds.push(f); } // accept() should fail with EMFILE: there's a connection in the @@ -82,23 +83,32 @@ fn accept_fd_exhaustion_child() { } } - // Release FDs. - held_fds.clear(); + // Release FDs and close the backlog connection. + drop(held_fds); drop(backlog_conn); // Make a fresh connection now that FDs are available. let client = std::net::TcpStream::connect(addr).expect("connect after FD release"); - // Retry: accept should now succeed, proving that an accept loop - // retrying on EMFILE (like the proxy does) will recover once FDs - // are available again. + // Accept should succeed, proving that an accept loop retrying on + // EMFILE (like the proxy does) will recover once FDs are available. + // On Linux the closed backlog connection may still be queued ahead + // of the fresh client (EMFILE fires before dequeue), so use a read + // timeout to detect and drain it. let (accepted, _peer) = listener .accept() .expect("accept should succeed after FD release"); // Verify the connection is functional. (&client).write_all(b"ping").expect("write"); + accepted + .set_read_timeout(Some(std::time::Duration::from_secs(1))) + .expect("set timeout"); let mut buf = [0u8; 4]; - (&accepted).read_exact(&mut buf).expect("read"); + if (&accepted).read_exact(&mut buf).is_err() { + // Got the stale backlog socket; accept the fresh connection. + let (fresh, _) = listener.accept().expect("accept fresh connection"); + (&fresh).read_exact(&mut buf).expect("read"); + } assert_eq!(&buf, b"ping"); } From e8b4aabe434d2e7736415fa1845a63ded3bd8f07 Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 27 Jul 2026 10:45:36 -0400 Subject: [PATCH 6/7] fix(proxy): add remaining Linux accept(2) transient errnos and make test skip explicit Classify ENOSR, ESOCKTNOSUPPORT, EPROTONOSUPPORT, and ETIMEDOUT as transient accept errors per Linux accept(2) documentation. Treat ENOSR as resource pressure (exponential backoff). Make the EMFILE integration test skip path emit a diagnostic instead of silently returning success. Signed-off-by: Quinn Burdine Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 18 ++++++++++++++++-- .../tests/accept_fd_exhaustion.rs | 4 +++- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 129a050227..aa9cf00c6c 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -389,7 +389,11 @@ fn classify_accept_error(err: &std::io::Error) -> AcceptErrorClass { | libc::EHOSTDOWN | libc::EHOSTUNREACH | libc::EOPNOTSUPP - | libc::ENETUNREACH, + | libc::ENETUNREACH + | libc::ENOSR + | libc::ESOCKTNOSUPPORT + | libc::EPROTONOSUPPORT + | libc::ETIMEDOUT, ) => AcceptErrorClass::Transient, #[cfg(target_os = "linux")] Some(libc::ENONET) => AcceptErrorClass::Transient, @@ -407,7 +411,7 @@ fn classify_accept_error(_err: &std::io::Error) -> AcceptErrorClass { fn is_resource_pressure_error(err: &std::io::Error) -> bool { matches!( err.raw_os_error(), - Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM) + Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM | libc::ENOSR) ) } @@ -10018,6 +10022,9 @@ network_policies: assert!(is_resource_pressure_error( &std::io::Error::from_raw_os_error(libc::ENOMEM) )); + assert!(is_resource_pressure_error( + &std::io::Error::from_raw_os_error(libc::ENOSR) + )); } #[cfg(unix)] @@ -10188,6 +10195,9 @@ network_policies: libc::EHOSTUNREACH, libc::EOPNOTSUPP, libc::ENETUNREACH, + libc::ESOCKTNOSUPPORT, + libc::EPROTONOSUPPORT, + libc::ETIMEDOUT, ] { assert_eq!( classify_accept_error(&std::io::Error::from_raw_os_error(errno)), @@ -10217,6 +10227,10 @@ network_policies: classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOMEM)), AcceptErrorClass::Transient, ); + assert_eq!( + classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOSR)), + AcceptErrorClass::Transient, + ); } #[cfg(unix)] diff --git a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs index e2c1aaac84..0bcb7d2193 100644 --- a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs +++ b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs @@ -78,7 +78,9 @@ fn accept_fd_exhaustion_child() { Err(ref e) if e.raw_os_error() == Some(libc::EMFILE) => {} Err(ref e) => panic!("expected EMFILE, got: {e}"), Ok(_) => { - // OS found a spare slot — skip gracefully. + // OS found a spare FD slot despite exhaustion attempt — platform + // does not reliably deliver EMFILE under these conditions. + eprintln!("SKIP: EMFILE not triggered (OS found a spare FD slot)"); return; } } From f13c4544607e226ae0f3ebd8b2497e4f510109f6 Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 27 Jul 2026 13:16:32 -0400 Subject: [PATCH 7/7] fix(proxy): panic instead of silent skip when EMFILE is not triggered The integration test child runs inside Command::output() which captures stderr. A silent return on the Ok path meant CI would report a passing test without any diagnostic. Panic instead so the parent test fails visibly if the platform cannot induce EMFILE. Signed-off-by: Quinn Burdine Signed-off-by: politerealism --- .../tests/accept_fd_exhaustion.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs index 0bcb7d2193..3fd09f9c89 100644 --- a/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs +++ b/crates/openshell-supervisor-network/tests/accept_fd_exhaustion.rs @@ -78,10 +78,7 @@ fn accept_fd_exhaustion_child() { Err(ref e) if e.raw_os_error() == Some(libc::EMFILE) => {} Err(ref e) => panic!("expected EMFILE, got: {e}"), Ok(_) => { - // OS found a spare FD slot despite exhaustion attempt — platform - // does not reliably deliver EMFILE under these conditions. - eprintln!("SKIP: EMFILE not triggered (OS found a spare FD slot)"); - return; + panic!("EMFILE not triggered: OS found a spare FD slot despite RLIMIT_NOFILE=32"); } }