diff --git a/crates/op-collab-transport/src/tcp.rs b/crates/op-collab-transport/src/tcp.rs index 08c6a66fe..c8199ef61 100644 --- a/crates/op-collab-transport/src/tcp.rs +++ b/crates/op-collab-transport/src/tcp.rs @@ -341,19 +341,38 @@ pub fn prepare_tcp_stream(stream: &TcpStream, config: TransportConfig) -> Result Ok(()) } +/// Smallest keepalive period the OS socket knobs can express. +/// +/// `TCP_KEEPIDLE` / `TCP_KEEPINTVL` on Linux and `TCP_KEEPALIVE` on macOS are +/// whole-second options, so socket2 hands them `Duration::as_secs()`. Anything +/// under a second therefore reaches the kernel as `0`, and the two platforms +/// disagree about what that means: Linux rejects it with `EINVAL`, which fails +/// every connect and accept before a single protocol byte moves, while macOS +/// accepts the call and silently keeps its 2-hour default instead of the +/// requested period. Neither is the configured behaviour. +const MIN_OS_KEEPALIVE_PERIOD: Duration = Duration::from_secs(1); + +/// Rounds a configured period up to the whole-second granularity the OS +/// keepalive options accept, never below [`MIN_OS_KEEPALIVE_PERIOD`]. +/// +/// `TransportConfig::validate` accepts any non-zero `timeouts.heartbeat`, and +/// the protocol's own heartbeat and idle deadlines keep that full `Duration` +/// precision in `SecureConnection` and `ConnectionDriver`. The kernel keepalive +/// is only a coarse backstop underneath them, so rounding up here costs nothing +/// while keeping the config-to-socket mapping total on every platform. +fn os_keepalive_period(period: Duration) -> Duration { + let whole_seconds = period + .as_secs() + .saturating_add(u64::from(period.subsec_nanos() > 0)); + Duration::from_secs(whole_seconds.max(MIN_OS_KEEPALIVE_PERIOD.as_secs())) +} + fn configure_tcp_common(stream: &TcpStream, config: TransportConfig) -> Result<(), RuntimeError> { stream.set_nodelay(true)?; let socket = SockRef::from(stream); socket.set_keepalive(true)?; - // Linux TCP_KEEPIDLE/TCP_KEEPINTVL have whole-second granularity and - // reject zero, so a sub-second heartbeat (test configs use 50ms) must - // not truncate to 0 or setsockopt fails with EINVAL. Application-level - // heartbeats still run at the configured cadence; only the kernel - // keepalive probes are clamped. - let keepalive_cadence = config.timeouts.heartbeat.max(Duration::from_secs(1)); - let keepalive = TcpKeepalive::new() - .with_time(keepalive_cadence) - .with_interval(keepalive_cadence); + let period = os_keepalive_period(config.timeouts.heartbeat); + let keepalive = TcpKeepalive::new().with_time(period).with_interval(period); socket.set_tcp_keepalive(&keepalive)?; Ok(()) } @@ -434,6 +453,10 @@ impl Write for DeadlineTcp<'_> { } } +#[cfg(test)] +#[path = "tcp_keepalive_tests.rs"] +mod keepalive_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/op-collab-transport/src/tcp_keepalive_tests.rs b/crates/op-collab-transport/src/tcp_keepalive_tests.rs new file mode 100644 index 000000000..4ee58c6e4 --- /dev/null +++ b/crates/op-collab-transport/src/tcp_keepalive_tests.rs @@ -0,0 +1,181 @@ +//! Guards for the config-to-socket keepalive mapping. +//! +//! `timeouts.heartbeat` is an arbitrary-precision `Duration` that +//! `TransportConfig::validate` accepts as long as it is non-zero, but the OS +//! keepalive options behind it are whole-second knobs. Feeding them the raw +//! duration made every sub-second heartbeat unusable: Linux failed each connect +//! and accept with `EINVAL` before any protocol byte moved, and macOS quietly +//! discarded the request and kept its own default. These tests pin both ends of +//! that mapping — the rounding itself, and what the kernel ends up holding. + +use super::*; +use std::net::TcpListener; + +#[test] +fn sub_second_heartbeats_round_up_to_the_os_keepalive_granularity() { + assert_eq!( + os_keepalive_period(Duration::from_millis(1)), + Duration::from_secs(1) + ); + assert_eq!( + os_keepalive_period(Duration::from_millis(50)), + Duration::from_secs(1) + ); + assert_eq!( + os_keepalive_period(Duration::from_millis(999)), + Duration::from_secs(1) + ); +} + +#[test] +fn whole_second_heartbeats_reach_the_socket_unchanged() { + assert_eq!( + os_keepalive_period(Duration::from_secs(1)), + Duration::from_secs(1) + ); + assert_eq!( + os_keepalive_period(Duration::from_secs(30)), + Duration::from_secs(30) + ); +} + +#[test] +fn fractional_heartbeats_round_up_rather_than_truncating_toward_zero() { + // Truncation is what socket2 does with `Duration::as_secs()`, so a period + // that lands between two seconds must be lifted before it gets there. + assert_eq!( + os_keepalive_period(Duration::from_millis(1_900)), + Duration::from_secs(2) + ); + assert_eq!( + os_keepalive_period(Duration::from_millis(30_001)), + Duration::from_secs(31) + ); +} + +/// The OS keepalive option names differ per platform but share the same +/// whole-second contract. +#[cfg(any(target_os = "linux", target_os = "macos"))] +mod socket_readback { + use super::*; + use std::os::fd::AsRawFd; + + #[cfg(target_os = "linux")] + const KEEPALIVE_TIME: libc::c_int = libc::TCP_KEEPIDLE; + #[cfg(target_os = "macos")] + const KEEPALIVE_TIME: libc::c_int = libc::TCP_KEEPALIVE; + + fn keepalive_seconds(stream: &TcpStream, option: libc::c_int) -> libc::c_int { + let mut value: libc::c_int = -1; + let mut length = std::mem::size_of::() as libc::socklen_t; + let result = unsafe { + libc::getsockopt( + stream.as_raw_fd(), + libc::IPPROTO_TCP, + option, + (&raw mut value).cast(), + &raw mut length, + ) + }; + assert_eq!( + result, + 0, + "getsockopt failed: {}", + std::io::Error::last_os_error() + ); + value + } + + fn config_with_heartbeat(heartbeat: Duration) -> TransportConfig { + // `idle` must stay above `heartbeat` for the config to validate. + let window = heartbeat.saturating_mul(8); + TransportConfig { + timeouts: crate::TimeoutConfig { + heartbeat, + idle: window, + read_write: window, + admission: window, + ..crate::TimeoutConfig::default() + }, + ..TransportConfig::default() + } + } + + /// The value the kernel actually holds is the only honest evidence here. + /// Linux rejects a zero-second keepalive outright, so the bug surfaced there + /// as a failed `configure_tcp_common`; macOS accepted the call and kept its + /// 7200 s default, so the requested period vanished without any error. This + /// asserts the applied period on both. + #[test] + fn a_sub_second_heartbeat_still_installs_a_usable_keepalive_period() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let stream = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + + for (heartbeat, expected_seconds) in [ + (Duration::from_millis(50), 1), + (Duration::from_millis(999), 1), + (Duration::from_millis(1_900), 2), + (Duration::from_secs(30), 30), + ] { + let config = config_with_heartbeat(heartbeat).validate().unwrap(); + configure_tcp_common(&stream, config) + .unwrap_or_else(|error| panic!("{heartbeat:?} heartbeat was rejected: {error:?}")); + assert_eq!( + keepalive_seconds(&stream, KEEPALIVE_TIME), + expected_seconds, + "{heartbeat:?} heartbeat produced the wrong keepalive idle period" + ); + assert_eq!( + keepalive_seconds(&stream, libc::TCP_KEEPINTVL), + expected_seconds, + "{heartbeat:?} heartbeat produced the wrong keepalive probe interval" + ); + } + } +} + +/// End-to-end shape of the original failure: with a sub-second heartbeat every +/// `connect_secure_tcp` and `accept_secure_tcp` on Linux returned +/// `Io(Os { code: 22, kind: InvalidInput })` out of the socket setup, long +/// before the Noise handshake could run. +#[test] +fn a_sub_second_heartbeat_config_still_completes_the_noise_handshake() { + let config = TransportConfig { + timeouts: crate::TimeoutConfig { + heartbeat: Duration::from_millis(20), + idle: Duration::from_millis(400), + read_write: Duration::from_millis(400), + admission: Duration::from_millis(400), + ..crate::TimeoutConfig::default() + }, + ..TransportConfig::default() + }; + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let owner_key = DeviceStaticKey::from_private([41_u8; 32]).unwrap(); + let prelude = ServerPrelude::new( + "00112233445566778899aabbccddeeff".to_owned(), + op_collab::SessionId::from("session"), + op_collab::Epoch(1), + ) + .unwrap(); + + let owner = std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + accept_secure_tcp(stream, &owner_key, &prelude, config).map(|_| ()) + }); + let guest = connect_secure_tcp( + address, + &DeviceStaticKey::from_private([42_u8; 32]).unwrap(), + None, + config, + ); + + assert!(guest.is_ok(), "the initiator failed: {:?}", guest.err()); + let accepted = owner.join().unwrap(); + assert!( + accepted.is_ok(), + "the responder failed: {:?}", + accepted.err() + ); +}