diff --git a/AGENTS.md b/AGENTS.md index 53b4e80496..b982dbf89a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -156,6 +156,19 @@ ocsf_emit!(event); - If you change sandbox infrastructure, ensure the relevant sandbox e2e path succeeds. +## Network Sockets + +- On latency-sensitive TCP streams, disable Nagle's algorithm so small + request/response frames don't stall on delayed ACKs. Use + `openshell_core::net::set_tcp_nodelay_best_effort` on an accepted or + already-connected stream, or `openshell_core::net::connect_tcp_nodelay_best_effort` + when dialing. +- This applies to loopback/localhost TCP too — the delayed-ACK stall is a timer + behavior, not wire latency. +- You should skip it for unix domain sockets (no Nagle). It's not critical for + test-only connections, though using it on any non-UDS TCP stream — tests + included — is fine and preferred. + ## Commits - Always use [Conventional Commits](https://www.conventionalcommits.org/) format for commit messages diff --git a/crates/openshell-cli/src/edge_tunnel.rs b/crates/openshell-cli/src/edge_tunnel.rs index 814e245f3c..99f1f44f63 100644 --- a/crates/openshell-cli/src/edge_tunnel.rs +++ b/crates/openshell-cli/src/edge_tunnel.rs @@ -26,6 +26,7 @@ use futures::stream::{SplitSink, SplitStream}; use futures::{SinkExt, StreamExt}; use miette::{IntoDiagnostic, Result}; +use openshell_core::net::set_tcp_nodelay_best_effort; use std::net::SocketAddr; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -101,6 +102,7 @@ async fn accept_loop(listener: TcpListener, config: Arc) { match listener.accept().await { Ok((stream, peer)) => { debug!(peer = %peer, "accepted local tunnel connection"); + set_tcp_nodelay_best_effort(&stream); let config = Arc::clone(&config); tokio::spawn(async move { if let Err(e) = handle_connection(stream, &config).await { @@ -174,6 +176,15 @@ async fn open_ws(config: &TunnelConfig) -> Result Some(tcp), + MaybeTlsStream::Rustls(tls) => Some(tls.get_ref().0), + _ => None, + }; + if let Some(tcp) = tcp { + set_tcp_nodelay_best_effort(tcp); + } + debug!( status = %response.status(), "WebSocket connected to edge" diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index 0182e1b668..2145ddfc23 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -27,6 +27,7 @@ use openshell_bootstrap::{ use openshell_bootstrap::{ GatewayMetadataSource, ListedGateway, gateway_metadata_source, list_gateways_with_source, }; +use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::progress::{ PROGRESS_ACTIVE_DETAIL_KEY, PROGRESS_ACTIVE_STEP_KEY, PROGRESS_COMPLETE_LABEL_KEY, PROGRESS_COMPLETE_STEP_KEY, PROGRESS_STEP_PULLING_IMAGE, PROGRESS_STEP_REQUESTING_SANDBOX, @@ -2891,6 +2892,7 @@ pub async fn service_forward_tcp( let (socket, peer) = accepted .into_diagnostic() .wrap_err("failed to accept local forward connection")?; + set_tcp_nodelay_best_effort(&socket); let mut client = client.clone(); let sandbox_id = sandbox_id.clone(); let target_host = target_host.to_string(); diff --git a/crates/openshell-cli/src/tls.rs b/crates/openshell-cli/src/tls.rs index 10df401a5b..2eadafc71a 100644 --- a/crates/openshell-cli/src/tls.rs +++ b/crates/openshell-cli/src/tls.rs @@ -3,6 +3,7 @@ use miette::{IntoDiagnostic, Result, WrapErr}; use openshell_core::auth::EdgeAuthInterceptor; +use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::proto::inference_client::InferenceClient; use openshell_core::proto::open_shell_client::OpenShellClient; use rustls::{ @@ -295,6 +296,7 @@ impl tower::Service for InsecureTlsConnector { let port = uri.port_u16().unwrap_or(443); let addr = format!("{host}:{port}"); let tcp = tokio::net::TcpStream::connect(addr).await?; + set_tcp_nodelay_best_effort(&tcp); let server_name = ServerName::try_from(host)?; let tls_stream = tls_connector.connect(server_name, tcp).await?; Ok(hyper_util::rt::TokioIo::new(tls_stream)) diff --git a/crates/openshell-core/src/net.rs b/crates/openshell-core/src/net.rs index 59bfdfc5d5..db5e5e4d2c 100644 --- a/crates/openshell-core/src/net.rs +++ b/crates/openshell-core/src/net.rs @@ -1,16 +1,23 @@ // SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Network IP classification utilities shared across `OpenShell` crates. +//! Shared networking utilities for `OpenShell` crates. //! -//! These helpers enforce the always-blocked IP invariant (loopback, link-local, -//! unspecified) and the broader internal-IP classification (adds RFC 1918 and -//! ULA). They are used by: +//! The IP-classification helpers enforce the always-blocked IP invariant +//! (loopback, link-local, unspecified) and the broader internal-IP +//! classification (adds RFC 1918 and ULA). They are used by: //! - The sandbox proxy for runtime SSRF enforcement //! - The mechanistic mapper for proposal filtering //! - The gateway server for defense-in-depth validation on approval +//! +//! The socket tuning helpers ([`set_tcp_nodelay_best_effort`], +//! [`connect_tcp_nodelay_best_effort`]) help avoid the known latency +//! imposed by conflict between Nagle's algorithm and delayed ACK behaviors. +//! + +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; -use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use tokio::net::TcpStream; /// Check if a hostname is a known cloud metadata hostname that resolves to an /// always-blocked metadata service. @@ -221,6 +228,27 @@ fn is_internal_v4(v4: Ipv4Addr) -> bool { false } +/// Enable `TCP_NODELAY` on a stream, logging (not returning) any failure. +/// +/// Disabling Nagle's algorithm keeps small writes from waiting on delayed ACKs. +/// It's a latency optimization: if it fails the connection still works, just a +/// bit slower, so there is nothing for the caller to act on — we log and move on. +pub fn set_tcp_nodelay_best_effort(stream: &TcpStream) { + if let Err(e) = stream.set_nodelay(true) { + tracing::debug!(error = %e, "failed to set TCP_NODELAY"); + } +} + +/// Connect to `addrs`, then enable `TCP_NODELAY` on a best-effort basis, propagating +/// any errors from `TcpStream::connect`. +/// +/// The returned stream is not *guaranteed* to have `TCP_NODELAY` set. +pub async fn connect_tcp_nodelay_best_effort(addrs: &[SocketAddr]) -> std::io::Result { + let stream = TcpStream::connect(addrs).await?; + set_tcp_nodelay_best_effort(&stream); + Ok(stream) +} + #[cfg(test)] mod tests { use super::*; @@ -584,4 +612,31 @@ mod tests { let v6 = Ipv4Addr::new(100, 64, 0, 1).to_ipv6_mapped(); assert!(is_internal_ip(IpAddr::V6(v6))); } + + // -- tcp_nodelay helpers -- + + #[tokio::test] + async fn set_tcp_nodelay_best_effort_enables_nodelay() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local addr"); + let stream = TcpStream::connect(addr).await.expect("connect"); + + set_tcp_nodelay_best_effort(&stream); + assert!(stream.nodelay().expect("query TCP_NODELAY")); + } + + #[tokio::test] + async fn connect_tcp_nodelay_best_effort_sets_nodelay() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local addr"); + + let stream = connect_tcp_nodelay_best_effort(&[addr]) + .await + .expect("connect"); + assert!(stream.nodelay().expect("query TCP_NODELAY")); + } } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 6462ccbbf3..ac31684cb2 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -52,6 +52,7 @@ pub mod tracing_bus; mod ws_tunnel; use metrics_exporter_prometheus::PrometheusBuilder; +use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::{ComputeDriverKind, Config, Error, Result}; use std::collections::HashMap; use std::io::ErrorKind; @@ -549,6 +550,8 @@ async fn serve_gateway_listener( } }; + set_tcp_nodelay_best_effort(&stream); + spawn_gateway_connection( stream, addr, diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index db6315dd86..ea32f9ece5 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -10,7 +10,10 @@ use crate::policy_local::{POLICY_LOCAL_HOST, PolicyLocalContext}; use miette::{IntoDiagnostic, Result}; use openshell_core::activity::{ActivitySender, try_record_activity}; use openshell_core::denial::DenialEvent; -use openshell_core::net::{is_always_blocked_ip, is_internal_ip, is_link_local_ip}; +use openshell_core::net::{ + connect_tcp_nodelay_best_effort, is_always_blocked_ip, is_internal_ip, is_link_local_ip, + set_tcp_nodelay_best_effort, +}; use openshell_core::policy::ProxyPolicy; use openshell_core::provider_credentials::ProviderCredentialState; use openshell_core::secrets::{self, SecretResolver, rewrite_header_line_checked}; @@ -268,6 +271,7 @@ impl ProxyHandle { loop { match listener.accept().await { Ok((stream, _addr)) => { + set_tcp_nodelay_best_effort(&stream); let opa = opa_engine.clone(); let cache = identity_cache.clone(); let spid = entrypoint_pid.clone(); @@ -1076,7 +1080,7 @@ async fn handle_tcp_connection( return Ok(()); } - let mut upstream = TcpStream::connect(validated_addrs.as_slice()) + let mut upstream = connect_tcp_nodelay_best_effort(validated_addrs.as_slice()) .await .into_diagnostic()?; @@ -4337,7 +4341,7 @@ async fn handle_forward_proxy( } // 6. Connect upstream - let mut upstream = match TcpStream::connect(addrs.as_slice()).await { + let mut upstream = match connect_tcp_nodelay_best_effort(addrs.as_slice()).await { Ok(s) => s, Err(e) => { let event = HttpActivityBuilder::new(openshell_ocsf::ctx::ctx()) diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index f5a3ee0793..3c24efe5ec 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -11,6 +11,7 @@ use crate::sandbox; use miette::{IntoDiagnostic, Result}; use nix::pty::{Winsize, openpty}; use nix::unistd::setsid; +use openshell_core::net::set_tcp_nodelay_best_effort; use openshell_core::policy::SandboxPolicy; use openshell_core::provider_credentials::ProviderCredentialState; use openshell_ocsf::{ @@ -651,13 +652,17 @@ pub async fn connect_in_netns( .await .map_err(|_| std::io::Error::other("netns connect thread panicked"))??; std_stream.set_nonblocking(true)?; - return tokio::net::TcpStream::from_std(std_stream); + let stream = tokio::net::TcpStream::from_std(std_stream)?; + set_tcp_nodelay_best_effort(&stream); + return Ok(stream); } #[cfg(not(target_os = "linux"))] let _ = netns_fd; - tokio::net::TcpStream::connect(addr).await + let stream = tokio::net::TcpStream::connect(addr).await?; + set_tcp_nodelay_best_effort(&stream); + Ok(stream) } #[derive(Clone)] @@ -1324,6 +1329,20 @@ mod tests { use super::*; use std::process::Stdio; + /// Regression test: the direct-tcpip connect path sets `TCP_NODELAY`. + #[tokio::test] + async fn connect_in_netns_sets_tcp_nodelay() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listener"); + let addr = listener.local_addr().expect("local addr"); + + let stream = connect_in_netns(&addr.to_string(), None) + .await + .expect("connect"); + assert!(stream.nodelay().expect("query TCP_NODELAY")); + } + #[cfg(unix)] fn file_mode(path: &Path) -> u32 { use std::os::unix::fs::PermissionsExt; diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index 594d865599..734b658108 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -31,6 +31,7 @@ use tokio_stream::StreamExt; use tracing::{debug, warn}; use openshell_core::grpc_client; +use openshell_core::net::set_tcp_nodelay_best_effort; const INITIAL_BACKOFF: Duration = Duration::from_secs(1); const MAX_BACKOFF: Duration = Duration::from_secs(30); @@ -663,10 +664,14 @@ async fn connect_tcp_target( .await .map_err(|_| "netns tcp connect thread panicked")??; stream.set_nonblocking(true)?; - return Ok(tokio::net::TcpStream::from_std(stream)?); + let stream = tokio::net::TcpStream::from_std(stream)?; + set_tcp_nodelay_best_effort(&stream); + return Ok(stream); } - Ok(tokio::net::TcpStream::connect((host.as_str(), port)).await?) + let stream = tokio::net::TcpStream::connect((host.as_str(), port)).await?; + set_tcp_nodelay_best_effort(&stream); + Ok(stream) } #[cfg(not(target_os = "linux"))] @@ -675,7 +680,9 @@ async fn connect_tcp_target( port: u16, _netns_fd: Option, ) -> Result> { - Ok(tokio::net::TcpStream::connect((host.as_str(), port)).await?) + let stream = tokio::net::TcpStream::connect((host.as_str(), port)).await?; + set_tcp_nodelay_best_effort(&stream); + Ok(stream) } #[cfg(test)] @@ -717,6 +724,20 @@ mod target_tests { } } + /// Regression test: the TCP relay connect path sets `TCP_NODELAY`. + #[tokio::test] + async fn connect_tcp_target_sets_tcp_nodelay() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listener"); + let addr = listener.local_addr().expect("local addr"); + + let stream = connect_tcp_target(addr.ip().to_string(), addr.port(), None) + .await + .expect("connect"); + assert!(stream.nodelay().expect("query TCP_NODELAY")); + } + #[test] fn tcp_target_allows_loopback_hosts() { validate_tcp_target(&tcp("127.0.0.1", 8080)).expect("ipv4 loopback");