From 4894930ec467dd0a45cec35f33a454c1d840806d Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 20 Jul 2026 12:09:58 -0400 Subject: [PATCH 1/8] fix(sandbox): terminate sandbox when proxy accept loop exits unexpectedly Add a oneshot notification channel to ProxyHandle that fires when the proxy task exits for any reason (panic, abort, or loop break). The sandbox main loop now races this signal alongside the entrypoint process and shutdown signals, terminating with a clear OCSF error if the proxy dies. This prevents the sandbox from continuing to report Ready with a dead proxy. Follows the existing sidecar control channel pattern. All five wait paths (process+sidecar, process-only, network+sidecar, network-only on Linux, and non-Linux) now monitor proxy liveness. Refs #2337 Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 104 +++++++++++++++++- .../openshell-supervisor-network/src/proxy.rs | 16 ++- 2 files changed, 113 insertions(+), 7 deletions(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index d96f141cc..4ba03dbb7 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -16,6 +16,7 @@ mod sidecar_control; use miette::{IntoDiagnostic, Result, WrapErr}; use std::future::Future; +use std::pin::Pin; use std::sync::Arc; #[cfg(target_os = "linux")] use std::sync::atomic::Ordering; @@ -514,7 +515,7 @@ pub async fn run_sandbox( // API read the current value so proposals target the correct workspace. let (workspace_tx, workspace_rx) = tokio::sync::watch::channel(String::new()); - let networking = if network_enabled { + let mut networking = if network_enabled { #[cfg(target_os = "linux")] let proxy_bind_ip = netns .as_ref() @@ -823,6 +824,18 @@ pub async fn run_sandbox( .zip(bootstrap.proxy_ca_bundle_path.clone()) }); + let proxy_exited: Pin + Send>> = + if let Some(rx) = networking + .as_mut() + .and_then(|n| n.proxy.as_mut()) + .map(|p| p.take_exit_receiver()) + { + Box::pin(async { let _ = rx.await; }) + } else { + Box::pin(std::future::pending()) + }; + tokio::pin!(proxy_exited); + let exit_code = if process_enabled { let ca_file_paths = networking .as_ref() @@ -935,9 +948,41 @@ pub async fn run_sandbox( "authoritative network-sidecar control channel closed" )); } + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } } } else { - process.await? + tokio::select! { + result = process => result?, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } } else { // Network-only sidecar mode: keep the proxy and its background @@ -953,15 +998,62 @@ pub async fn run_sandbox( warn!(?result, "Authoritative sidecar control channel exited; restarting sidecar"); 1 } + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } } } else { - wait_for_shutdown_signal().await; - 0 + tokio::select! { + () = wait_for_shutdown_signal() => 0, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } #[cfg(not(target_os = "linux"))] { - wait_for_shutdown_signal().await; - 0 + tokio::select! { + () = wait_for_shutdown_signal() => 0, + _ = &mut proxy_exited => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Proxy accept loop exited unexpectedly; terminating sandbox" + ) + .build() + ); + return Err(miette::miette!( + "proxy accept loop exited unexpectedly" + )); + } + } } }; diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index bc3ad8b3d..82b29b634 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -228,11 +228,11 @@ impl InferenceContext { } } -#[derive(Debug)] pub struct ProxyHandle { #[allow(dead_code)] http_addr: Option, join: JoinHandle<()>, + exited_rx: tokio::sync::oneshot::Receiver<()>, } impl ProxyHandle { @@ -337,7 +337,13 @@ impl ProxyHandle { ocsf_emit!(event); } + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); let join = tokio::spawn(async move { + // Hold the sender for the lifetime of this task — when the task + // exits (panic, abort, or loop break), the sender drops and the + // receiver fires, notifying the sandbox that the proxy is gone. + let _proxy_exit_guard = exited_tx; + // Wait for the OPA engine's symlink resolution reload to complete // before accepting connections. This prevents requests from // observing a generation transition mid-flight, which would cause @@ -448,6 +454,7 @@ impl ProxyHandle { Ok(Self { http_addr: Some(local_addr), join, + exited_rx, }) } @@ -455,6 +462,13 @@ impl ProxyHandle { pub const fn http_addr(&self) -> Option { self.http_addr } + + /// Take the exit notification receiver for health monitoring. + /// Resolves when the proxy accept loop task exits for any reason. + pub fn take_exit_receiver(&mut self) -> tokio::sync::oneshot::Receiver<()> { + let (_tx, dummy_rx) = tokio::sync::oneshot::channel(); + std::mem::replace(&mut self.exited_rx, dummy_rx) + } } impl Drop for ProxyHandle { From 355499881626c19d65b198537fe9c624a37c8d95 Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 20 Jul 2026 12:51:51 -0400 Subject: [PATCH 2/8] test(proxy): add unit tests for exit detection drop-guard and take_exit_receiver Verify the oneshot drop-guard pattern fires on normal task exit and abort, and document the take_exit_receiver contract including the double-call hazard. Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 82b29b634..15e84f2a6 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -13054,6 +13054,65 @@ network_policies: assert_eq!(res, 3); assert_eq!(unk, 2); } + + #[tokio::test] + async fn test_exit_receiver_fires_when_task_exits() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + }); + handle.await.unwrap(); + // The sender was dropped when the task completed, so the receiver + // should resolve immediately with an Err (sender dropped). + assert!(exited_rx.await.is_err()); + } + + #[tokio::test] + async fn test_exit_receiver_fires_when_task_is_aborted() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + std::future::pending::<()>().await; + }); + handle.abort(); + // Abort drops the task's locals, including the sender guard. + assert!(exited_rx.await.is_err()); + } + + #[tokio::test] + async fn test_take_exit_receiver_returns_real_receiver() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let mut taken = handle.take_exit_receiver(); + // Sender still alive — receiver should not be ready yet. + assert!(taken.try_recv().is_err()); + // Drop the original sender — the taken receiver should resolve. + drop(tx); + assert!(taken.await.is_err()); + } + + #[tokio::test] + async fn test_take_exit_receiver_second_call_returns_instantly() { + let (_tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let _first = handle.take_exit_receiver(); + let second = handle.take_exit_receiver(); + // The dummy's sender was dropped inside take_exit_receiver, so + // the second receiver resolves immediately — this demonstrates + // the double-call hazard. + assert!(second.await.is_err()); + } + #[path = "compatibility.rs"] mod compatibility; } From 2231caf555689903c0c21495a1e4fb1e410e79f9 Mon Sep 17 00:00:00 2001 From: politerealism Date: Tue, 21 Jul 2026 11:29:57 -0400 Subject: [PATCH 3/8] fix(proxy): classify terminal vs transient accept errors and exit on unrecoverable failures Extract accept-loop error handling into classify_accept_error() with AcceptAction enum so the decision logic is independently testable. Terminal errors (EBADF/EINVAL/ENOTSOCK) now exit the loop immediately, allowing the supervisor to detect proxy death. Unknown errors exit after 10 consecutive failures. FD exhaustion retries with exponential backoff capped at 5s (fix unreachable cap from min(6) to min(7)). Adds 9 unit tests covering classification, counter interactions, backoff progression, and bounded exit behavior. Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 21 +- .../openshell-supervisor-network/src/proxy.rs | 738 +++++++----------- 2 files changed, 283 insertions(+), 476 deletions(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 4ba03dbb7..ff4be1f64 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -824,16 +824,17 @@ pub async fn run_sandbox( .zip(bootstrap.proxy_ca_bundle_path.clone()) }); - let proxy_exited: Pin + Send>> = - if let Some(rx) = networking - .as_mut() - .and_then(|n| n.proxy.as_mut()) - .map(|p| p.take_exit_receiver()) - { - Box::pin(async { let _ = rx.await; }) - } else { - Box::pin(std::future::pending()) - }; + let proxy_exited: Pin + Send>> = if let Some(rx) = networking + .as_mut() + .and_then(|n| n.proxy.as_mut()) + .map(|p| p.take_exit_receiver()) + { + Box::pin(async { + let _ = rx.await; + }) + } else { + Box::pin(std::future::pending()) + }; tokio::pin!(proxy_exited); let exit_code = if process_enabled { diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 15e84f2a6..e3b786395 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -368,12 +368,12 @@ impl ProxyHandle { } } - let mut consecutive_resource_errors: u32 = 0; + let mut consecutive_fd_errors: u32 = 0; let mut consecutive_unknown_errors: u32 = 0; loop { match listener.accept().await { Ok((stream, _addr)) => { - consecutive_resource_errors = 0; + consecutive_fd_errors = 0; consecutive_unknown_errors = 0; set_tcp_nodelay_best_effort(&stream); let opa = opa_engine.clone(); @@ -428,23 +428,36 @@ impl ProxyHandle { }); } Err(err) => { - let outcome = handle_accept_error( + match classify_accept_error( &err, - &mut consecutive_resource_errors, + &mut consecutive_fd_errors, &mut consecutive_unknown_errors, - ); - - let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) - .activity(ActivityId::Fail) - .severity(outcome.severity) - .status(StatusId::Failure) - .message(outcome.message) - .build(); - ocsf_emit!(event); - - match outcome.backoff { - Some(backoff) => tokio::time::sleep(backoff).await, - None => break, + ) { + AcceptAction::Terminal => { + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message(format!( + "Proxy accept loop exiting on terminal error: {err}", + )) + .build(); + ocsf_emit!(event); + break; + } + AcceptAction::Retry { backoff, severity } => { + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Fail) + .severity(severity) + .status(StatusId::Failure) + .message(format!( + "Proxy accept error (retrying in {}ms): {err}", + backoff.as_millis(), + )) + .build(); + ocsf_emit!(event); + tokio::time::sleep(backoff).await; + } } } } @@ -986,142 +999,56 @@ fn emit_transparent_policy_denial( ); } -fn emit_activity(tx: &Option, denied: bool, deny_group: &'static str) { - if let Some(tx) = tx { - let _ = try_record_activity(tx, denied, deny_group); - } -} +const MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS: u32 = 10; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum AcceptErrorClass { - Transient, +#[derive(Debug, PartialEq)] +enum AcceptAction { Terminal, - Unknown, -} - -#[cfg(unix)] -fn classify_accept_error(err: &std::io::Error) -> AcceptErrorClass { - match err.raw_os_error() { - 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 - | libc::ENOSR - | libc::ESOCKTNOSUPPORT - | libc::EPROTONOSUPPORT - | libc::ETIMEDOUT, - ) => AcceptErrorClass::Transient, - #[cfg(target_os = "linux")] - Some(libc::ENONET) => AcceptErrorClass::Transient, - Some(libc::EBADF | libc::EINVAL | libc::ENOTSOCK) => AcceptErrorClass::Terminal, - _ => AcceptErrorClass::Unknown, - } -} - -#[cfg(not(unix))] -fn classify_accept_error(_err: &std::io::Error) -> AcceptErrorClass { - AcceptErrorClass::Unknown -} - -#[cfg(unix)] -fn is_resource_pressure_error(err: &std::io::Error) -> bool { - matches!( - err.raw_os_error(), - Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM | libc::ENOSR) - ) -} - -#[cfg(not(unix))] -fn is_resource_pressure_error(_err: &std::io::Error) -> bool { - false + Retry { + backoff: std::time::Duration, + severity: SeverityId, + }, } -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); - let ms = ACCEPT_BACKOFF_BASE_MS - .saturating_mul(1u64 << exponent) - .min(ACCEPT_BACKOFF_MAX_MS); - std::time::Duration::from_millis(ms) -} +fn classify_accept_error( + err: &std::io::Error, + consecutive_fd_errors: &mut u32, + consecutive_unknown_errors: &mut u32, +) -> AcceptAction { + // EBADF (9) / EINVAL (22) / ENOTSOCK (88) — the listener socket is + // permanently unusable; retrying will never succeed. EINVAL from + // accept() means the socket is no longer listening (shut down or + // corrupted), not a generic invalid-argument condition. + if matches!(err.raw_os_error(), Some(9 | 22 | 88)) { + return AcceptAction::Terminal; + } + + // EMFILE (24) / ENFILE (23) — process or system FD table is full. + if matches!(err.raw_os_error(), Some(24 | 23)) { + *consecutive_unknown_errors = 0; + *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); + let backoff_ms = 100u64 + .saturating_mul(1u64 << (*consecutive_fd_errors).min(7).saturating_sub(1)) + .min(5_000); + return AcceptAction::Retry { + backoff: std::time::Duration::from_millis(backoff_ms), + severity: SeverityId::Medium, + }; + } -struct AcceptErrorOutcome { - severity: SeverityId, - message: String, - backoff: Option, + *consecutive_unknown_errors = consecutive_unknown_errors.saturating_add(1); + if *consecutive_unknown_errors >= MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + return AcceptAction::Terminal; + } + AcceptAction::Retry { + backoff: std::time::Duration::from_millis(100), + severity: SeverityId::Low, + } } -fn handle_accept_error( - err: &std::io::Error, - consecutive_resource_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_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!( - "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 emit_activity(tx: &Option, denied: bool, deny_group: &'static str) { + if let Some(tx) = tx { + let _ = try_record_activity(tx, denied, deny_group); } } @@ -12725,392 +12652,271 @@ 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_resource_pressure_detects_emfile() { - let err = std::io::Error::from_raw_os_error(libc::EMFILE); - assert!(is_resource_pressure_error(&err)); + #[tokio::test] + async fn test_exit_receiver_fires_when_task_exits() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + }); + handle.await.unwrap(); + // The sender was dropped when the task completed, so the receiver + // should resolve immediately with an Err (sender dropped). + assert!(exited_rx.await.is_err()); } - #[cfg(unix)] - #[test] - fn is_resource_pressure_detects_enfile() { - let err = std::io::Error::from_raw_os_error(libc::ENFILE); - assert!(is_resource_pressure_error(&err)); + #[tokio::test] + async fn test_exit_receiver_fires_when_task_is_aborted() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let handle = tokio::spawn(async move { + let _guard = exited_tx; + std::future::pending::<()>().await; + }); + handle.abort(); + // Abort drops the task's locals, including the sender guard. + assert!(exited_rx.await.is_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) - )); - assert!(is_resource_pressure_error( - &std::io::Error::from_raw_os_error(libc::ENOSR) - )); + #[tokio::test] + async fn test_take_exit_receiver_returns_real_receiver() { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let mut taken = handle.take_exit_receiver(); + // Sender still alive — receiver should not be ready yet. + assert!(taken.try_recv().is_err()); + // Drop the original sender — the taken receiver should resolve. + drop(tx); + assert!(taken.await.is_err()); } - #[cfg(unix)] - #[test] - fn is_resource_pressure_rejects_other_errors() { - let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); - assert!(!is_resource_pressure_error(&err)); + #[tokio::test] + async fn test_take_exit_receiver_second_call_returns_instantly() { + let (_tx, rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(std::future::pending::<()>()); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: rx, + }; + let _first = handle.take_exit_receiver(); + let second = handle.take_exit_receiver(); + // The dummy's sender was dropped inside take_exit_receiver, so + // the second receiver resolves immediately — this demonstrates + // the double-call hazard. + assert!(second.await.is_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, - ); - } + // --- classify_accept_error tests --- - #[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, - ); + fn test_classify_terminal_error_ebadf() { + let err = std::io::Error::from_raw_os_error(9); // EBADF + let mut fd = 0; + let mut unk = 0; assert_eq!( - classify_accept_error(&std::io::Error::from_raw_os_error(libc::EINTR)), - AcceptErrorClass::Transient, + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal ); } - #[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, - ); + fn test_classify_terminal_error_einval() { + let err = std::io::Error::from_raw_os_error(22); // EINVAL + let mut fd = 0; + let mut unk = 0; assert_eq!( - classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOTSOCK)), - AcceptErrorClass::Terminal, + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal ); } - #[cfg(unix)] + #[cfg(target_os = "linux")] #[test] - fn classify_accept_error_unrecognized_errno_is_unknown() { + fn test_classify_terminal_error_enotsock() { + let err = std::io::Error::from_raw_os_error(88); // ENOTSOCK (Linux) + let mut fd = 0; + let mut unk = 0; assert_eq!( - classify_accept_error(&std::io::Error::from_raw_os_error(libc::EPERM)), - AcceptErrorClass::Unknown, + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal ); } - #[cfg(unix)] #[test] - fn handle_accept_error_terminal_exits_immediately() { - let mut res = 0; + fn test_classify_fd_exhaustion_returns_retry_medium() { + let err = std::io::Error::from_raw_os_error(24); // EMFILE + 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 res, &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 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 res, &mut unk); - assert!(outcome.backoff.is_some(), "should retry on attempt {i}"); - assert_eq!(outcome.severity, SeverityId::Medium); - } - assert_eq!(res, 20); + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!( + action, + AcceptAction::Retry { + severity: SeverityId::Medium, + .. + } + ), + "expected Retry/Medium for EMFILE, got {action:?}", + ); } - #[cfg(unix)] #[test] - fn handle_accept_error_unknown_exits_after_limit() { - let mut res = 0; + fn test_classify_unknown_error_returns_retry_low() { + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + 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 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 res, &mut unk); + let action = classify_accept_error(&err, &mut fd, &mut unk); assert!( - outcome.backoff.is_none(), - "should exit after limit exceeded" + matches!( + action, + AcceptAction::Retry { + severity: SeverityId::Low, + .. + } + ), + "expected Retry/Low for unknown error, got {action:?}", ); - assert_eq!(outcome.severity, SeverityId::High); } - #[cfg(unix)] #[test] - fn handle_accept_error_transient_resets_unknown_counter() { - let mut res = 0; + fn test_classify_fd_exhaustion_backoff_increases_and_caps() { + 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 res, &mut unk); + let err = std::io::Error::from_raw_os_error(24); // EMFILE + + let mut prev_backoff = std::time::Duration::ZERO; + for _ in 0..6 { + match classify_accept_error(&err, &mut fd, &mut unk) { + AcceptAction::Retry { backoff, .. } => { + assert!( + backoff > prev_backoff, + "backoff should increase: {backoff:?} <= {prev_backoff:?}", + ); + prev_backoff = backoff; + } + AcceptAction::Terminal => panic!("expected Retry, got Terminal"), + } } - assert_eq!(unk, MAX_CONSECUTIVE_UNKNOWN_ERRORS); - // A transient error resets the unknown counter. - 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 res, &mut unk); - assert!(outcome.backoff.is_some()); - assert_eq!(unk, 1); + // After enough consecutive errors the backoff should hit the 5s cap. + for _ in 6..12 { + classify_accept_error(&err, &mut fd, &mut unk); + } + match classify_accept_error(&err, &mut fd, &mut unk) { + AcceptAction::Retry { backoff, .. } => { + assert_eq!( + backoff, + std::time::Duration::from_secs(5), + "backoff should cap at 5000ms", + ); + } + AcceptAction::Terminal => panic!("expected Retry, got Terminal"), + } } - #[cfg(unix)] #[test] - fn handle_accept_error_fd_exhaustion_uses_exponential_backoff() { - let mut res = 0; + fn test_classify_unknown_errors_exit_after_threshold() { + let mut fd = 0; let mut unk = 0; - let err = std::io::Error::from_raw_os_error(libc::EMFILE); + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED - 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 classify_accept_error_network_errors_are_transient() { - for errno in [ - libc::ENETDOWN, - libc::EPROTO, - libc::ENOPROTOOPT, - libc::EHOSTDOWN, - 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)), - AcceptErrorClass::Transient, - "errno {errno} should be transient", + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "call {i} should be Retry, got {action:?}", ); } - } - - #[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, - ); + let final_action = classify_accept_error(&err, &mut fd, &mut unk); assert_eq!( - classify_accept_error(&std::io::Error::from_raw_os_error(libc::ENOSR)), - AcceptErrorClass::Transient, + final_action, + AcceptAction::Terminal, + "call {MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS} should be Terminal", ); } - #[cfg(unix)] #[test] - fn handle_accept_error_non_resource_transient_uses_fixed_backoff() { - let mut res = 0; + fn test_classify_success_resets_counters() { + let mut fd = 0; let mut unk = 0; - let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED - 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); - } + // Accumulate 9 unknown errors (one below threshold). + for _ in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + classify_accept_error(&err, &mut fd, &mut unk); + } - #[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); + // Simulate a successful accept — production loop resets both counters. + fd = 0; + unk = 0; - 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); + // Another 9 unknowns should NOT trigger Terminal. + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "after reset, call {i} should be Retry, got {action:?}", + ); + } } - #[cfg(unix)] #[test] - fn handle_accept_error_resource_counter_persists_across_mixed_transient() { - let mut res = 0; + fn test_classify_fd_error_resets_unknown_counter() { + let mut fd = 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 unknown_err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let fd_err = std::io::Error::from_raw_os_error(24); // EMFILE - let o1 = handle_accept_error(&resource_err, &mut res, &mut unk); - assert_eq!(res, 1); - assert_eq!(o1.backoff.unwrap().as_millis(), 100); + // Accumulate 5 unknown errors. + for _ in 0..5 { + classify_accept_error(&unknown_err, &mut fd, &mut unk); + } - let o2 = handle_accept_error(&transient_err, &mut res, &mut unk); - assert_eq!(res, 1); - assert_eq!(o2.backoff.unwrap().as_millis(), 100); + // One FD error resets the unknown counter. + classify_accept_error(&fd_err, &mut fd, &mut unk); - let o3 = handle_accept_error(&resource_err, &mut res, &mut unk); - assert_eq!(res, 2); - assert_eq!(o3.backoff.unwrap().as_millis(), 200); + // Another 9 unknowns should NOT trigger Terminal (counter was reset). + for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { + let action = classify_accept_error(&unknown_err, &mut fd, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "after FD reset, call {i} should be Retry, got {action:?}", + ); + } } - #[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); - } - - #[tokio::test] - async fn test_exit_receiver_fires_when_task_exits() { - let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); - let handle = tokio::spawn(async move { - let _guard = exited_tx; - }); - handle.await.unwrap(); - // The sender was dropped when the task completed, so the receiver - // should resolve immediately with an Err (sender dropped). - assert!(exited_rx.await.is_err()); - } - - #[tokio::test] - async fn test_exit_receiver_fires_when_task_is_aborted() { - let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); - let handle = tokio::spawn(async move { - let _guard = exited_tx; - std::future::pending::<()>().await; - }); - handle.abort(); - // Abort drops the task's locals, including the sender guard. - assert!(exited_rx.await.is_err()); - } + fn test_classify_fd_exhaustion_and_terminal_are_disjoint() { + let mut fd = 0; + let mut unk = 0; - #[tokio::test] - async fn test_take_exit_receiver_returns_real_receiver() { - let (tx, rx) = tokio::sync::oneshot::channel::<()>(); - let join = tokio::spawn(std::future::pending::<()>()); - let mut handle = ProxyHandle { - http_addr: None, - join, - exited_rx: rx, - }; - let mut taken = handle.take_exit_receiver(); - // Sender still alive — receiver should not be ready yet. - assert!(taken.try_recv().is_err()); - // Drop the original sender — the taken receiver should resolve. - drop(tx); - assert!(taken.await.is_err()); - } + for errno in [24, 23] { + // EMFILE, ENFILE + let err = std::io::Error::from_raw_os_error(errno); + assert!( + matches!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Retry { .. } + ), + "errno {errno} should be Retry", + ); + fd = 0; + unk = 0; + } - #[tokio::test] - async fn test_take_exit_receiver_second_call_returns_instantly() { - let (_tx, rx) = tokio::sync::oneshot::channel::<()>(); - let join = tokio::spawn(std::future::pending::<()>()); - let mut handle = ProxyHandle { - http_addr: None, - join, - exited_rx: rx, - }; - let _first = handle.take_exit_receiver(); - let second = handle.take_exit_receiver(); - // The dummy's sender was dropped inside take_exit_receiver, so - // the second receiver resolves immediately — this demonstrates - // the double-call hazard. - assert!(second.await.is_err()); + for errno in [9, 22] { + // EBADF, EINVAL (portable values) + let err = std::io::Error::from_raw_os_error(errno); + assert_eq!( + classify_accept_error(&err, &mut fd, &mut unk), + AcceptAction::Terminal, + "errno {errno} should be Terminal", + ); + } } #[path = "compatibility.rs"] From 8c1d3c95981c18689b16b630b10f85ac07414939 Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 3 Aug 2026 14:07:05 -0400 Subject: [PATCH 4/8] refactor(proxy): use libc constants and Option-based exit receiver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace raw errno numbers with libc:: constants behind #[cfg(unix)] gates in classify_accept_error and its tests. Refactor ProxyHandle::exited_rx to Option> so take_exit_receiver uses .take() instead of allocating a dummy sender. Add a ProxyHandle-level test that verifies the full drop → abort → receiver fires path. Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 2 +- .../openshell-supervisor-network/src/proxy.rs | 105 ++++++++++-------- 2 files changed, 60 insertions(+), 47 deletions(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index ff4be1f64..f794a41ed 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -827,7 +827,7 @@ pub async fn run_sandbox( let proxy_exited: Pin + Send>> = if let Some(rx) = networking .as_mut() .and_then(|n| n.proxy.as_mut()) - .map(|p| p.take_exit_receiver()) + .and_then(|p| p.take_exit_receiver()) { Box::pin(async { let _ = rx.await; diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index e3b786395..655df0035 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -232,7 +232,7 @@ pub struct ProxyHandle { #[allow(dead_code)] http_addr: Option, join: JoinHandle<()>, - exited_rx: tokio::sync::oneshot::Receiver<()>, + exited_rx: Option>, } impl ProxyHandle { @@ -467,7 +467,7 @@ impl ProxyHandle { Ok(Self { http_addr: Some(local_addr), join, - exited_rx, + exited_rx: Some(exited_rx), }) } @@ -476,11 +476,8 @@ impl ProxyHandle { self.http_addr } - /// Take the exit notification receiver for health monitoring. - /// Resolves when the proxy accept loop task exits for any reason. - pub fn take_exit_receiver(&mut self) -> tokio::sync::oneshot::Receiver<()> { - let (_tx, dummy_rx) = tokio::sync::oneshot::channel(); - std::mem::replace(&mut self.exited_rx, dummy_rx) + pub fn take_exit_receiver(&mut self) -> Option> { + self.exited_rx.take() } } @@ -1015,16 +1012,20 @@ fn classify_accept_error( consecutive_fd_errors: &mut u32, consecutive_unknown_errors: &mut u32, ) -> AcceptAction { - // EBADF (9) / EINVAL (22) / ENOTSOCK (88) — the listener socket is - // permanently unusable; retrying will never succeed. EINVAL from - // accept() means the socket is no longer listening (shut down or - // corrupted), not a generic invalid-argument condition. - if matches!(err.raw_os_error(), Some(9 | 22 | 88)) { + // Terminal: the listener socket is permanently unusable; retrying will + // never succeed. EINVAL from accept() means the socket is no longer + // listening (shut down or corrupted). + #[cfg(unix)] + if matches!( + err.raw_os_error(), + Some(libc::EBADF | libc::EINVAL | libc::ENOTSOCK) + ) { return AcceptAction::Terminal; } - // EMFILE (24) / ENFILE (23) — process or system FD table is full. - if matches!(err.raw_os_error(), Some(24 | 23)) { + // FD exhaustion: process or system FD table is full. + #[cfg(unix)] + if matches!(err.raw_os_error(), Some(libc::EMFILE | libc::ENFILE)) { *consecutive_unknown_errors = 0; *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); let backoff_ms = 100u64 @@ -12683,38 +12684,50 @@ network_policies: let mut handle = ProxyHandle { http_addr: None, join, - exited_rx: rx, + exited_rx: Some(rx), }; - let mut taken = handle.take_exit_receiver(); - // Sender still alive — receiver should not be ready yet. + let mut taken = handle.take_exit_receiver().expect("first take should return Some"); assert!(taken.try_recv().is_err()); - // Drop the original sender — the taken receiver should resolve. drop(tx); assert!(taken.await.is_err()); } #[tokio::test] - async fn test_take_exit_receiver_second_call_returns_instantly() { + async fn test_take_exit_receiver_second_call_returns_none() { let (_tx, rx) = tokio::sync::oneshot::channel::<()>(); let join = tokio::spawn(std::future::pending::<()>()); let mut handle = ProxyHandle { http_addr: None, join, - exited_rx: rx, + exited_rx: Some(rx), }; let _first = handle.take_exit_receiver(); - let second = handle.take_exit_receiver(); - // The dummy's sender was dropped inside take_exit_receiver, so - // the second receiver resolves immediately — this demonstrates - // the double-call hazard. - assert!(second.await.is_err()); + assert!(handle.take_exit_receiver().is_none()); + } + + #[tokio::test] + async fn test_proxy_handle_drop_fires_exit_receiver() { + let (exited_tx, exited_rx) = tokio::sync::oneshot::channel::<()>(); + let join = tokio::spawn(async move { + let _guard = exited_tx; + std::future::pending::<()>().await; + }); + let mut handle = ProxyHandle { + http_addr: None, + join, + exited_rx: Some(exited_rx), + }; + let rx = handle.take_exit_receiver().expect("should return Some"); + drop(handle); + assert!(rx.await.is_err()); } // --- classify_accept_error tests --- + #[cfg(unix)] #[test] fn test_classify_terminal_error_ebadf() { - let err = std::io::Error::from_raw_os_error(9); // EBADF + let err = std::io::Error::from_raw_os_error(libc::EBADF); let mut fd = 0; let mut unk = 0; assert_eq!( @@ -12723,9 +12736,10 @@ network_policies: ); } + #[cfg(unix)] #[test] fn test_classify_terminal_error_einval() { - let err = std::io::Error::from_raw_os_error(22); // EINVAL + let err = std::io::Error::from_raw_os_error(libc::EINVAL); let mut fd = 0; let mut unk = 0; assert_eq!( @@ -12734,10 +12748,10 @@ network_policies: ); } - #[cfg(target_os = "linux")] + #[cfg(unix)] #[test] fn test_classify_terminal_error_enotsock() { - let err = std::io::Error::from_raw_os_error(88); // ENOTSOCK (Linux) + let err = std::io::Error::from_raw_os_error(libc::ENOTSOCK); let mut fd = 0; let mut unk = 0; assert_eq!( @@ -12746,9 +12760,10 @@ network_policies: ); } + #[cfg(unix)] #[test] fn test_classify_fd_exhaustion_returns_retry_medium() { - let err = std::io::Error::from_raw_os_error(24); // EMFILE + let err = std::io::Error::from_raw_os_error(libc::EMFILE); let mut fd = 0; let mut unk = 0; let action = classify_accept_error(&err, &mut fd, &mut unk); @@ -12764,9 +12779,10 @@ network_policies: ); } + #[cfg(unix)] #[test] fn test_classify_unknown_error_returns_retry_low() { - let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let err = std::io::Error::from_raw_os_error(libc::ECONNREFUSED); let mut fd = 0; let mut unk = 0; let action = classify_accept_error(&err, &mut fd, &mut unk); @@ -12782,11 +12798,12 @@ network_policies: ); } + #[cfg(unix)] #[test] fn test_classify_fd_exhaustion_backoff_increases_and_caps() { let mut fd = 0; let mut unk = 0; - let err = std::io::Error::from_raw_os_error(24); // EMFILE + let err = std::io::Error::from_raw_os_error(libc::EMFILE); let mut prev_backoff = std::time::Duration::ZERO; for _ in 0..6 { @@ -12818,11 +12835,12 @@ network_policies: } } + #[cfg(unix)] #[test] fn test_classify_unknown_errors_exit_after_threshold() { let mut fd = 0; let mut unk = 0; - let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let err = std::io::Error::from_raw_os_error(libc::ECONNREFUSED); for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { let action = classify_accept_error(&err, &mut fd, &mut unk); @@ -12839,22 +12857,20 @@ network_policies: ); } + #[cfg(unix)] #[test] fn test_classify_success_resets_counters() { let mut fd = 0; let mut unk = 0; - let err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED + let err = std::io::Error::from_raw_os_error(libc::ECONNREFUSED); - // Accumulate 9 unknown errors (one below threshold). for _ in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { classify_accept_error(&err, &mut fd, &mut unk); } - // Simulate a successful accept — production loop resets both counters. fd = 0; unk = 0; - // Another 9 unknowns should NOT trigger Terminal. for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { let action = classify_accept_error(&err, &mut fd, &mut unk); assert!( @@ -12864,22 +12880,20 @@ network_policies: } } + #[cfg(unix)] #[test] fn test_classify_fd_error_resets_unknown_counter() { let mut fd = 0; let mut unk = 0; - let unknown_err = std::io::Error::from_raw_os_error(111); // ECONNREFUSED - let fd_err = std::io::Error::from_raw_os_error(24); // EMFILE + let unknown_err = std::io::Error::from_raw_os_error(libc::ECONNREFUSED); + let fd_err = std::io::Error::from_raw_os_error(libc::EMFILE); - // Accumulate 5 unknown errors. for _ in 0..5 { classify_accept_error(&unknown_err, &mut fd, &mut unk); } - // One FD error resets the unknown counter. classify_accept_error(&fd_err, &mut fd, &mut unk); - // Another 9 unknowns should NOT trigger Terminal (counter was reset). for i in 1..MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS { let action = classify_accept_error(&unknown_err, &mut fd, &mut unk); assert!( @@ -12889,13 +12903,13 @@ network_policies: } } + #[cfg(unix)] #[test] fn test_classify_fd_exhaustion_and_terminal_are_disjoint() { let mut fd = 0; let mut unk = 0; - for errno in [24, 23] { - // EMFILE, ENFILE + for errno in [libc::EMFILE, libc::ENFILE] { let err = std::io::Error::from_raw_os_error(errno); assert!( matches!( @@ -12908,8 +12922,7 @@ network_policies: unk = 0; } - for errno in [9, 22] { - // EBADF, EINVAL (portable values) + for errno in [libc::EBADF, libc::EINVAL, libc::ENOTSOCK] { let err = std::io::Error::from_raw_os_error(errno); assert_eq!( classify_accept_error(&err, &mut fd, &mut unk), From 5346bca9e4c4c73efa768bb2959f08136ed1de17 Mon Sep 17 00:00:00 2001 From: politerealism Date: Mon, 3 Aug 2026 14:21:51 -0400 Subject: [PATCH 5/8] fix(proxy): restore full transient errno classification from main The refactored classifier only recognized EMFILE/ENFILE as retryable, routing all other documented transient accept errors (ENOBUFS, ENOMEM, ECONNABORTED, ECONNRESET, EINTR, ENETDOWN, etc.) through the unknown- error budget which terminates the proxy after 10 consecutive failures. Restore the complete transient errno table from main so these errors retry indefinitely with appropriate backoff. Signed-off-by: politerealism --- .../openshell-supervisor-network/src/proxy.rs | 119 ++++++++++++++---- 1 file changed, 98 insertions(+), 21 deletions(-) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 655df0035..d5f6d7904 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -368,12 +368,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; set_tcp_nodelay_best_effort(&stream); let opa = opa_engine.clone(); @@ -430,7 +430,7 @@ impl ProxyHandle { Err(err) => { match classify_accept_error( &err, - &mut consecutive_fd_errors, + &mut consecutive_resource_errors, &mut consecutive_unknown_errors, ) { AcceptAction::Terminal => { @@ -1009,12 +1009,9 @@ enum AcceptAction { fn classify_accept_error( err: &std::io::Error, - consecutive_fd_errors: &mut u32, + consecutive_resource_errors: &mut u32, consecutive_unknown_errors: &mut u32, ) -> AcceptAction { - // Terminal: the listener socket is permanently unusable; retrying will - // never succeed. EINVAL from accept() means the socket is no longer - // listening (shut down or corrupted). #[cfg(unix)] if matches!( err.raw_os_error(), @@ -1023,17 +1020,66 @@ fn classify_accept_error( return AcceptAction::Terminal; } - // FD exhaustion: process or system FD table is full. #[cfg(unix)] - if matches!(err.raw_os_error(), Some(libc::EMFILE | libc::ENFILE)) { + if matches!( + err.raw_os_error(), + 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 + | libc::ENOSR + | libc::ESOCKTNOSUPPORT + | libc::EPROTONOSUPPORT + | libc::ETIMEDOUT + ) + ) { *consecutive_unknown_errors = 0; - *consecutive_fd_errors = consecutive_fd_errors.saturating_add(1); - let backoff_ms = 100u64 - .saturating_mul(1u64 << (*consecutive_fd_errors).min(7).saturating_sub(1)) - .min(5_000); + + #[cfg(unix)] + let is_resource_pressure = matches!( + err.raw_os_error(), + Some(libc::EMFILE | libc::ENFILE | libc::ENOBUFS | libc::ENOMEM | libc::ENOSR) + ); + #[cfg(not(unix))] + let is_resource_pressure = false; + + if is_resource_pressure { + *consecutive_resource_errors = consecutive_resource_errors.saturating_add(1); + let backoff_ms = 100u64 + .saturating_mul(1u64 << (*consecutive_resource_errors).min(7).saturating_sub(1)) + .min(5_000); + return AcceptAction::Retry { + backoff: std::time::Duration::from_millis(backoff_ms), + severity: SeverityId::Medium, + }; + } + + *consecutive_resource_errors = 0; return AcceptAction::Retry { - backoff: std::time::Duration::from_millis(backoff_ms), - severity: SeverityId::Medium, + backoff: std::time::Duration::from_millis(100), + severity: SeverityId::Low, + }; + } + + #[cfg(unix)] + #[cfg(target_os = "linux")] + if matches!(err.raw_os_error(), Some(libc::ENONET)) { + *consecutive_unknown_errors = 0; + *consecutive_resource_errors = 0; + return AcceptAction::Retry { + backoff: std::time::Duration::from_millis(100), + severity: SeverityId::Low, }; } @@ -12905,33 +12951,64 @@ network_policies: #[cfg(unix)] #[test] - fn test_classify_fd_exhaustion_and_terminal_are_disjoint() { - let mut fd = 0; + fn test_classify_transient_and_terminal_are_disjoint() { + let mut res = 0; let mut unk = 0; - for errno in [libc::EMFILE, libc::ENFILE] { + for errno in [ + libc::EMFILE, + libc::ENFILE, + libc::ENOBUFS, + libc::ENOMEM, + libc::ECONNABORTED, + libc::ECONNRESET, + libc::EINTR, + libc::ENETDOWN, + libc::EHOSTDOWN, + libc::EHOSTUNREACH, + libc::EOPNOTSUPP, + libc::ENETUNREACH, + libc::ENOSR, + libc::ETIMEDOUT, + ] { let err = std::io::Error::from_raw_os_error(errno); assert!( matches!( - classify_accept_error(&err, &mut fd, &mut unk), + classify_accept_error(&err, &mut res, &mut unk), AcceptAction::Retry { .. } ), "errno {errno} should be Retry", ); - fd = 0; + res = 0; unk = 0; } for errno in [libc::EBADF, libc::EINVAL, libc::ENOTSOCK] { let err = std::io::Error::from_raw_os_error(errno); assert_eq!( - classify_accept_error(&err, &mut fd, &mut unk), + classify_accept_error(&err, &mut res, &mut unk), AcceptAction::Terminal, "errno {errno} should be Terminal", ); } } + #[cfg(unix)] + #[test] + fn test_transient_errors_never_hit_unknown_budget() { + let mut res = 0; + let mut unk = 0; + let err = std::io::Error::from_raw_os_error(libc::ECONNABORTED); + + for i in 0..(MAX_CONSECUTIVE_UNKNOWN_ACCEPT_ERRORS + 5) { + let action = classify_accept_error(&err, &mut res, &mut unk); + assert!( + matches!(action, AcceptAction::Retry { .. }), + "transient error on call {i} should always Retry, got {action:?}", + ); + } + } + #[path = "compatibility.rs"] mod compatibility; } From 2a9b58202ee241ed3bc85c7912b38543be4382be Mon Sep 17 00:00:00 2001 From: politerealism Date: Thu, 6 Aug 2026 13:43:19 -0400 Subject: [PATCH 6/8] fix(sandbox): resolve Clippy warnings in proxy-exit detection code Replace redundant closure |p| p.take_exit_receiver() with the method reference ProxyHandle::take_exit_receiver, and change all five select! arms from _ = &mut proxy_exited to () = &mut proxy_exited to satisfy clippy::ignored_unit_patterns. Also run cargo fmt on the new proxy test in proxy.rs. Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 13 +++++++------ crates/openshell-supervisor-network/src/proxy.rs | 4 +++- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index f794a41ed..de2b4ee7d 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -63,6 +63,7 @@ use openshell_core::policy::{NetworkMode, NetworkPolicy, ProxyPolicy, SandboxPol use openshell_core::proposals::AgentProposals; use openshell_core::provider_credentials::ProviderCredentialState; use openshell_supervisor_network::opa::OpaEngine; +use openshell_supervisor_network::proxy::ProxyHandle; use openshell_supervisor_process::process::ProcessEnforcementMode; pub use openshell_supervisor_process::process::{ProcessHandle, ProcessStatus}; use openshell_supervisor_process::skills; @@ -827,7 +828,7 @@ pub async fn run_sandbox( let proxy_exited: Pin + Send>> = if let Some(rx) = networking .as_mut() .and_then(|n| n.proxy.as_mut()) - .and_then(|p| p.take_exit_receiver()) + .and_then(ProxyHandle::take_exit_receiver) { Box::pin(async { let _ = rx.await; @@ -949,7 +950,7 @@ pub async fn run_sandbox( "authoritative network-sidecar control channel closed" )); } - _ = &mut proxy_exited => { + () = &mut proxy_exited => { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) @@ -968,7 +969,7 @@ pub async fn run_sandbox( } else { tokio::select! { result = process => result?, - _ = &mut proxy_exited => { + () = &mut proxy_exited => { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) @@ -999,7 +1000,7 @@ pub async fn run_sandbox( warn!(?result, "Authoritative sidecar control channel exited; restarting sidecar"); 1 } - _ = &mut proxy_exited => { + () = &mut proxy_exited => { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) @@ -1018,7 +1019,7 @@ pub async fn run_sandbox( } else { tokio::select! { () = wait_for_shutdown_signal() => 0, - _ = &mut proxy_exited => { + () = &mut proxy_exited => { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) @@ -1039,7 +1040,7 @@ pub async fn run_sandbox( { tokio::select! { () = wait_for_shutdown_signal() => 0, - _ = &mut proxy_exited => { + () = &mut proxy_exited => { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index d5f6d7904..dc2736a4e 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -12732,7 +12732,9 @@ network_policies: join, exited_rx: Some(rx), }; - let mut taken = handle.take_exit_receiver().expect("first take should return Some"); + let mut taken = handle + .take_exit_receiver() + .expect("first take should return Some"); assert!(taken.try_recv().is_err()); drop(tx); assert!(taken.await.is_err()); From 4dbcc2bf6cc48dade449272e3bd14af1145922cc Mon Sep 17 00:00:00 2001 From: politerealism Date: Tue, 11 Aug 2026 13:02:19 -0400 Subject: [PATCH 7/8] ci: trigger required check re-run Signed-off-by: politerealism From 21901608e727c479879f791d961b27e40199fe4c Mon Sep 17 00:00:00 2001 From: politerealism Date: Wed, 19 Aug 2026 11:45:58 -0400 Subject: [PATCH 8/8] fix(sandbox): remove unnecessary std::pin qualification Signed-off-by: politerealism --- crates/openshell-sandbox/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index de2b4ee7d..b4c2cfd9d 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -3251,7 +3251,7 @@ type MiddlewareConnector = Arc< dyn Fn( Vec, MiddlewareAuthentication, - ) -> std::pin::Pin< + ) -> Pin< Box< dyn std::future::Future< Output = Result,