diff --git a/crates/openshell-supervisor-network/Cargo.toml b/crates/openshell-supervisor-network/Cargo.toml index 1449276f4..979097f92 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 ab2313b89..39a2e4db9 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,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 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; } } } @@ -351,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()) } @@ -9827,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)); + } }