diff --git a/DESIGN.md b/DESIGN.md index 066e847..e1ee007 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -437,6 +437,7 @@ Two backends, because both versions matter now: | Wake | eventfd | EVFILT_USER | PostQueuedCompletionStatus | same-thread flag | same-thread flag | `schedule_turn` import / Atomics.notify | | Timer wait precision | ns (epoll_pwait2 / timerfd) | ns (kevent timespec) | sub-ms via high-res waitable timer | host-dependent (Wasmtime ≈1 ms) | host-dependent (Wasmtime ≈1 ms) | host `setTimeout` (clamped by browser) | | TCP / UDP | non-blocking + readiness | non-blocking + readiness | overlapped Winsock | `wasi:sockets` | `wasi:sockets` | unsupported | +| Socket options (§7.7) | full setsockopt set | full setsockopt set, no IPv4 membership by interface index | full Winsock set, no IPv4 membership by interface index | keep-alive, buffer sizes, hop limit | keep-alive, buffer sizes, hop limit | unsupported | | Outbound HTTP | protocol crate | protocol crate | protocol crate | protocol crate or `wasi:http` | protocol crate or `wasi:http` | host `fetch` | | WebSocket | protocol crate | protocol crate | protocol crate | protocol crate | protocol crate | host `WebSocket` | | Local IPC | AF_UNIX | AF_UNIX | named pipes (overlapped) | unsupported | unsupported | `postMessage` | @@ -450,6 +451,39 @@ Two backends, because both versions matter now: | Host integration | epoll fd | kqueue fd | event HANDLE + helper thread | runtime-owned | runtime-owned | `HostCallback` | | Cost measurement | perf instructions:u/k | rusage ri_instructions | QueryProcessCycleTime | Wasmtime fuel / instruction counts | Wasmtime fuel | browser profiler (relative) | +### 7.7 Socket options + +Node exposes `socket.setNoDelay()`, `setKeepAlive()`, `setTTL()` and friends as +methods on a *connected* socket, and a server commonly configures every +connection it accepts, so a creation-time option struct cannot express what the +host needs. `Loop::set_option(handle, SocketOption)` and +`Loop::get_option(handle, SocketOptionKind)` are therefore part of the core API +and work on any live socket handle, including one produced by `accept`. + +- **Synchronous, not an operation.** Both run entirely inside the call: no + Request is accepted, no completion is produced, nothing is queued and nothing + is allocated. They are the same shape as `tty_set_mode` and `local_addr`. +- **No cache, ever.** `get_option` always asks the OS, because the OS is entitled + to round, clamp or double what it was given (`SO_RCVBUF` on Linux is the usual + example). Reporting the request back would be a lie the host acts on. +- **`Unsupported`, never silently ignored.** A backend that has no equivalent for + an option says so. "The option is on" is a latency and connection-lifetime + claim; a host that cannot distinguish an applied option from an ignored one has + no way to find the bug later. WASI 0.2/0.3 therefore refuse Nagle, linger, + IPv6-only, broadcast and multicast, because `wasi:sockets` has no interface for + them, and the web backend refuses all of them. +- **Bind-time options stay in the opts structs.** `SO_REUSEADDR`/`SO_REUSEPORT` + cannot be changed on a bound socket, so they belong to `ListenOpts`/`UdpOpts`, + not to `SocketOption`. `IPV6_V6ONLY` is readable on a live socket and kept in + the enum for that, but setting it after bind is refused by every OS. +- **`ListenOpts::accept_defaults`** carries the per-connection defaults a server + would otherwise apply by hand: `nodelay` and a keep-alive schedule, each at most + one `setsockopt` on the new socket. The accepting backend applies them after the + OS accept and **before** the `Accepted` completion is produced, so the host never + observes an unconfigured connection. A backend that cannot apply a requested + default rejects the *listener* when it is created, and an OS failure while + applying one fails that accept rather than handing up a half-configured socket. + ## 8. Loop liveness and ref/unref - A handle, pending operation or timer is **referenced** by default. `set_ref(h, false)` excludes it from `alive()`, but it still produces completions while something else keeps the loop running. diff --git a/crates/turnloop-contract/src/lib.rs b/crates/turnloop-contract/src/lib.rs index f4db02b..eb3e0dc 100644 --- a/crates/turnloop-contract/src/lib.rs +++ b/crates/turnloop-contract/src/lib.rs @@ -71,7 +71,7 @@ pub fn pair(l: &mut Driver) -> (Handle, Handle, Handle) { let client = l .tcp_connect( l.local_addr(server).expect("addr"), - &TcpOpts { nodelay: true }, + &TcpOpts::default(), Token(2), ) .expect("connect"); @@ -1609,6 +1609,7 @@ pub fn no_spin() { #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] pub mod filesystem; pub mod native_surface; +pub mod sockopts; #[cfg(feature = "executor")] pub mod executor_contract; diff --git a/crates/turnloop-contract/src/sockopts.rs b/crates/turnloop-contract/src/sockopts.rs new file mode 100644 index 0000000..7b353fd --- /dev/null +++ b/crates/turnloop-contract/src/sockopts.rs @@ -0,0 +1,661 @@ +//! Socket options on live handles (DESIGN §7.7; issue #34). +//! +//! Every assertion here reads the value back **from the operating system**: +//! `get_option` is a `getsockopt`/`wasi:sockets` call on the live socket, never a +//! cache of what `set_option` was given, so a backend that accepted an option and +//! ignored it fails these tests. Two of them go further and prove behaviour the +//! host can observe: `linger_zero_resets_the_connection` makes a peer's read fail +//! with `ConnectionReset` where a graceful close would have produced `Eof`, and +//! `multicast_membership_is_tracked` shows the kernel refusing to leave a group it +//! was never asked to join. +use super::*; +use turnloop::{KeepAlive, MulticastGroup, SocketOption, SocketOptionKind}; + +/// A connected triple whose client keeps the platform's own Nagle setting, so a +/// later `NoDelay(true)` is an observable transition rather than a no-op. +pub fn plain_pair(l: &mut Driver, listen: &ListenOpts) -> (Handle, Handle, Handle) { + let server = l.tcp_listen(localhost(), listen).expect("listen"); + l.accept(server, Token(1)).expect("accept"); + let client = l + .tcp_connect( + l.local_addr(server).expect("addr"), + &TcpOpts::default(), + Token(2), + ) + .expect("connect"); + let mut conn = None; + let mut connected = false; + let mut out = Completions::default(); + let until = l.now() + Duration::from_secs(5); + while conn.is_none() || !connected { + assert!(l.now() < until, "connect/accept timed out"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } => conn = Some(h), + OpResult::Connected => connected = true, + other => panic!("unexpected {other:?}"), + } + } + } + (server, client, conn.expect("accepted")) +} +/// Close every handle and run the loop until nothing is left, so a fixture cannot +/// leak a descriptor into the next one. +pub fn close_all(l: &mut Driver, handles: &[Handle]) { + let mut out = Completions::default(); + for (i, h) in handles.iter().enumerate() { + l.close(*h, Token(900 + i as u64)).expect("close"); + } + let until = l.now() + Duration::from_secs(5); + while l.alive() { + assert!(l.now() < until, "close never completed"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + out.drain(); + } +} +fn read_option(l: &Driver, h: Handle, kind: SocketOptionKind) -> SocketOption { + l.get_option(h, kind) + .unwrap_or_else(|e| panic!("get_option({kind:?}) must reach the OS: {e:?}")) +} +fn keep_alive_of(l: &Driver, h: Handle) -> Option { + match read_option(l, h, SocketOptionKind::KeepAlive) { + SocketOption::KeepAlive(value) => value, + other => panic!("KeepAlive kind answered with {other:?}"), + } +} +fn buffer_of(l: &Driver, h: Handle, send: bool) -> u32 { + let kind = if send { + SocketOptionKind::SendBufferSize + } else { + SocketOptionKind::RecvBufferSize + }; + match read_option(l, h, kind) { + SocketOption::SendBufferSize(n) | SocketOption::RecvBufferSize(n) => n, + other => panic!("buffer kind answered with {other:?}"), + } +} +/// The final buffer size is the kernel's, not ours: macOS keeps the request, +/// Linux doubles it and clamps it to `net.core.{r,w}mem_max`, and Windows rounds +/// up to its own granularity and may hold an auto-tuned window above it. The +/// contract is therefore "at least what was asked", and an upper bound would be +/// asserting one platform's policy rather than the API. +/// +/// This predicate alone cannot prove the write landed — a large enough default +/// satisfies it. The growth assertion at the call site is what does that. +fn honoured(actual: u32, requested: u32) -> bool { + actual >= requested +} + +/// Keep-alive is settable and readable on a connected client **and on an accepted +/// connection**, which is what `socket.setKeepAlive()` needs and what turnloop +/// could not do at all before issue #34. +pub fn keep_alive_round_trip() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &ListenOpts::default()); + let schedule = KeepAlive { + idle: Some(Duration::from_secs(7)), + interval: Some(Duration::from_secs(3)), + count: Some(4), + }; + for (name, h) in [("client", client), ("accepted", conn)] { + assert_eq!(keep_alive_of(&l, h), None, "{name} starts without probing"); + l.set_option(h, SocketOption::KeepAlive(Some(schedule))) + .unwrap_or_else(|e| panic!("{name} keep-alive: {e:?}")); + let read = keep_alive_of(&l, h).unwrap_or_else(|| panic!("{name} probing must be on")); + assert_eq!(read.idle, Some(Duration::from_secs(7)), "{name} idle"); + assert_eq!( + read.interval, + Some(Duration::from_secs(3)), + "{name} interval" + ); + assert_eq!(read.count, Some(4), "{name} count"); + l.set_option(h, SocketOption::KeepAlive(None)) + .unwrap_or_else(|e| panic!("{name} disable: {e:?}")); + assert_eq!( + keep_alive_of(&l, h), + None, + "{name} probing must be off again" + ); + } + // A zero idle time has no meaning to any OS and is rejected, not rounded to + // "immediately" or silently dropped. + assert_eq!( + l.set_option( + client, + SocketOption::KeepAlive(Some(KeepAlive { + idle: Some(Duration::ZERO), + ..KeepAlive::default() + })), + ) + .expect_err("zero idle") + .kind, + ErrorKind::InvalidInput + ); + close_all(&mut l, &[client, conn, server]); +} +/// Buffer sizes reach the kernel on a client, an accepted connection and a UDP +/// socket, and the value read back is the kernel's, not the request. +pub fn buffer_sizes_round_trip() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &ListenOpts::default()); + let udp = l.udp_bind(localhost(), &UdpOpts::default()).expect("bind"); + for (name, h) in [("client", client), ("accepted", conn), ("udp", udp)] { + for send in [false, true] { + let mut previous = 0; + for requested in [32u32 * 1024, 64 * 1024] { + let option = if send { + SocketOption::SendBufferSize(requested) + } else { + SocketOption::RecvBufferSize(requested) + }; + l.set_option(h, option) + .unwrap_or_else(|e| panic!("{name} send={send} {requested}: {e:?}")); + let actual = buffer_of(&l, h, send); + assert!( + honoured(actual, requested), + "{name} send={send}: asked {requested}, OS reports {actual}" + ); + // The subject proof: a second, larger request must move the + // kernel's own number. A `set_option` that did nothing reports + // the same size twice and fails here, whatever the rounding. + assert!( + actual > previous, + "{name} send={send}: {actual} did not grow past {previous}" + ); + previous = actual; + } + } + } + close_all(&mut l, &[client, conn, server, udp]); +} +/// The unicast hop limit is settable and readable on UDP (`dgram.setTTL`). +pub fn ttl_round_trip() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let udp = l.udp_bind(localhost(), &UdpOpts::default()).expect("bind"); + let before = match read_option(&l, udp, SocketOptionKind::Ttl) { + SocketOption::Ttl(hops) => hops, + other => panic!("Ttl kind answered with {other:?}"), + }; + assert!(before > 0, "a bound socket always has a hop limit"); + l.set_option(udp, SocketOption::Ttl(7)).expect("set TTL"); + assert!( + matches!( + read_option(&l, udp, SocketOptionKind::Ttl), + SocketOption::Ttl(7) + ), + "the OS must report the hop limit we set" + ); + close_all(&mut l, &[udp]); +} +/// A listener's `accept_defaults` reach every accepted connection before the host +/// sees it, so a server never has to configure each connection by hand. +pub fn accept_defaults_keep_alive() { + let schedule = KeepAlive { + idle: Some(Duration::from_secs(11)), + ..KeepAlive::default() + }; + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + keep_alive: Some(schedule), + ..AcceptDefaults::EMPTY + }, + ..ListenOpts::default() + }; + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &listen); + let read = keep_alive_of(&l, conn).expect("the accepted socket must have probing on"); + assert_eq!(read.idle, Some(Duration::from_secs(11))); + assert_eq!( + keep_alive_of(&l, client), + None, + "the default belongs to accepted connections, not to every socket" + ); + close_all(&mut l, &[client, conn, server]); +} +/// Handles that are not sockets, are closing, or never existed are rejected +/// rather than reaching a backend table with someone else's index. +pub fn option_handle_validation() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let timer = l + .timer(l.now() + Duration::from_secs(60), None, Token(1)) + .expect("timer"); + assert_eq!( + l.set_option(timer, SocketOption::RecvBufferSize(4096)) + .expect_err("a timer is not a socket") + .kind, + ErrorKind::InvalidInput + ); + assert_eq!( + l.get_option(timer, SocketOptionKind::RecvBufferSize) + .expect_err("a timer is not a socket") + .kind, + ErrorKind::InvalidInput + ); + let udp = l.udp_bind(localhost(), &UdpOpts::default()).expect("bind"); + l.close(udp, Token(2)).expect("close"); + assert_eq!( + l.set_option(udp, SocketOption::RecvBufferSize(4096)) + .expect_err("a closing socket takes no options") + .kind, + ErrorKind::InvalidInput + ); + close_all(&mut l, &[timer]); + assert_eq!( + l.set_option(udp, SocketOption::RecvBufferSize(4096)) + .expect_err("a released handle is gone") + .kind, + ErrorKind::NotFound + ); +} +/// `SO_LINGER` with a zero timeout is the one option whose effect a host can see +/// directly: the peer's pending read fails with `ConnectionReset` where the same +/// close without it delivers `Eof`. Both arms run, so neither verdict is vacuous. +pub fn linger_zero_resets_the_connection() { + for reset in [false, true] { + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &ListenOpts::default()); + if reset { + l.set_option(client, SocketOption::Linger(Some(Duration::ZERO))) + .expect("linger 0"); + assert!( + matches!( + read_option(&l, client, SocketOptionKind::Linger), + SocketOption::Linger(Some(Duration::ZERO)) + ), + "the OS must report the linger we set" + ); + } else { + assert!( + matches!( + read_option(&l, client, SocketOptionKind::Linger), + SocketOption::Linger(None) + ), + "lingering is off by default" + ); + } + l.read(conn, ReadBuf::Pooled, Token(10)).expect("read"); + l.close(client, Token(11)).expect("close client"); + let until = l.now() + Duration::from_secs(5); + let mut out = Completions::default(); + let mut observed = None; + while observed.is_none() { + assert!(l.now() < until, "the peer never noticed the close"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + if c.token == Token(10) { + observed = Some(c.result); + } + } + } + match (reset, observed.expect("read result")) { + (true, OpResult::Err(e)) => assert_eq!( + e.kind, + ErrorKind::ConnectionReset, + "linger 0 must reset, not close gracefully" + ), + (false, OpResult::Eof) => {} + (reset, other) => panic!("linger zero = {reset}: unexpected {other:?}"), + } + close_all(&mut l, &[conn, server]); + } +} +/// Broadcast and the multicast send options are settable on a UDP socket, and the +/// kernel reports each one back. +pub fn udp_broadcast_and_multicast_options() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let udp = l.udp_bind(localhost(), &UdpOpts::default()).expect("bind"); + assert!( + matches!( + read_option(&l, udp, SocketOptionKind::Broadcast), + SocketOption::Broadcast(false) + ), + "broadcast is off by default" + ); + l.set_option(udp, SocketOption::Broadcast(true)) + .expect("broadcast on"); + assert!(matches!( + read_option(&l, udp, SocketOptionKind::Broadcast), + SocketOption::Broadcast(true) + )); + l.set_option(udp, SocketOption::Broadcast(false)) + .expect("broadcast off"); + assert!(matches!( + read_option(&l, udp, SocketOptionKind::Broadcast), + SocketOption::Broadcast(false) + )); + l.set_option(udp, SocketOption::MulticastTtl(4)) + .expect("multicast ttl"); + assert!( + matches!( + read_option(&l, udp, SocketOptionKind::MulticastTtl), + SocketOption::MulticastTtl(4) + ), + "the OS must report the multicast hop limit we set" + ); + l.set_option(udp, SocketOption::MulticastLoop(false)) + .expect("multicast loop"); + assert!(matches!( + read_option(&l, udp, SocketOptionKind::MulticastLoop), + SocketOption::MulticastLoop(false) + )); + l.set_option(udp, SocketOption::MulticastLoop(true)) + .expect("multicast loop"); + assert!(matches!( + read_option(&l, udp, SocketOptionKind::MulticastLoop), + SocketOption::MulticastLoop(true) + )); + close_all(&mut l, &[udp]); +} +/// Group membership has no getter, so the proof is the kernel's own bookkeeping: +/// leaving a group it never joined fails, leaving one it joined succeeds, and +/// leaving that one twice fails again. A backend that dropped the join on the +/// floor cannot produce that sequence. +pub fn multicast_membership_is_tracked(group: std::net::IpAddr, bind: SocketAddr) { + let mut l = Driver::::new(Config::default()).expect("loop"); + let udp = l.udp_bind(bind, &UdpOpts::default()).expect("bind"); + let membership = MulticastGroup { + group, + interface: 0, + }; + assert!( + l.set_option(udp, SocketOption::MulticastLeave(membership)) + .is_err(), + "leaving a group that was never joined must fail" + ); + l.set_option(udp, SocketOption::MulticastJoin(membership)) + .expect("join"); + l.set_option(udp, SocketOption::MulticastLeave(membership)) + .expect("leave the group we joined"); + assert!( + l.set_option(udp, SocketOption::MulticastLeave(membership)) + .is_err(), + "the kernel forgot the leave" + ); + // A group of the other family is a caller error, not a kernel error. + let mismatched = MulticastGroup { + group: if group.is_ipv4() { + std::net::Ipv6Addr::new(0xff02, 0, 0, 0, 0, 0, 0, 1).into() + } else { + std::net::Ipv4Addr::new(224, 0, 0, 251).into() + }, + interface: 0, + }; + assert_eq!( + l.set_option(udp, SocketOption::MulticastJoin(mismatched)) + .expect_err("family mismatch") + .kind, + ErrorKind::InvalidInput + ); + close_all(&mut l, &[udp]); +} +/// `IPV6_V6ONLY` is readable on a live socket and, as documented, refused after +/// bind on every platform. The test exists so the documentation cannot drift away +/// from the behaviour. +pub fn ipv6_only_is_readable_and_bind_time() { + let mut l = Driver::::new(Config::default()).expect("loop"); + let Ok(udp) = l.udp_bind( + (std::net::Ipv6Addr::LOCALHOST, 0).into(), + &UdpOpts::default(), + ) else { + // A host without IPv6 cannot answer this question either way. + return; + }; + assert!( + matches!( + read_option(&l, udp, SocketOptionKind::Ipv6Only), + SocketOption::Ipv6Only(_) + ), + "a live IPv6 socket must answer the query" + ); + assert_eq!( + l.set_option(udp, SocketOption::Ipv6Only(true)) + .expect_err("bind-time only") + .kind, + ErrorKind::InvalidInput, + "setting IPV6_V6ONLY after bind must be reported, not quietly accepted" + ); + close_all(&mut l, &[udp]); +} +/// Nagle control is settable and readable on a connected client and on an +/// accepted connection, and a listener can apply it to every connection it +/// accepts. This is `socket.setNoDelay()` and the per-connection server default. +pub fn nodelay_round_trip_and_accept_default() { + let mut l = Driver::::new(Config::default()).expect("loop"); + // The creation-time hint still works, and is visible through the OS. + let eager = l + .tcp_connect( + (std::net::Ipv4Addr::LOCALHOST, 9).into(), + &TcpOpts { nodelay: true }, + Token(30), + ) + .expect("connecting socket"); + assert!( + matches!( + read_option(&l, eager, SocketOptionKind::NoDelay), + SocketOption::NoDelay(true) + ), + "TcpOpts::nodelay must reach the socket it created" + ); + // An IP-level option on a socket that has neither connected nor bound: the + // backend still has to learn its address family to pick IP_TTL over + // IPV6_UNICAST_HOPS. + assert!( + matches!(read_option(&l, eager, SocketOptionKind::Ttl), SocketOption::Ttl(hops) if hops > 0), + "an unconnected socket still knows its address family" + ); + close_all(&mut l, &[eager]); + + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &ListenOpts::default()); + for (name, h) in [("client", client), ("accepted", conn)] { + assert!( + matches!( + read_option(&l, h, SocketOptionKind::NoDelay), + SocketOption::NoDelay(false) + ), + "{name}: Nagle is on until a host turns it off" + ); + l.set_option(h, SocketOption::NoDelay(true)) + .unwrap_or_else(|e| panic!("{name} nodelay: {e:?}")); + assert!( + matches!( + read_option(&l, h, SocketOptionKind::NoDelay), + SocketOption::NoDelay(true) + ), + "{name}: the OS must report TCP_NODELAY" + ); + l.set_option(h, SocketOption::NoDelay(false)) + .expect("restore"); + assert!(matches!( + read_option(&l, h, SocketOptionKind::NoDelay), + SocketOption::NoDelay(false) + )); + } + close_all(&mut l, &[client, conn, server]); + + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + ..AcceptDefaults::EMPTY + }, + ..ListenOpts::default() + }; + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &listen); + assert!( + matches!( + read_option(&l, conn, SocketOptionKind::NoDelay), + SocketOption::NoDelay(true) + ), + "the accepted connection must arrive with the listener's default applied" + ); + assert!( + matches!( + read_option(&l, client, SocketOptionKind::NoDelay), + SocketOption::NoDelay(false) + ), + "the default belongs to accepted connections only" + ); + close_all(&mut l, &[client, conn, server]); +} +/// A small write still makes a full round trip with Nagle disabled on both +/// endpoints, including on the accepted side, which is the shape a request/ +/// response server uses. (Loopback acknowledges instantly, so the *latency* +/// difference Nagle causes is not reproducible here; the behaviour asserted is +/// that the configured connection still carries bytes in both directions.) +pub fn nodelay_small_write_round_trip() { + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + ..AcceptDefaults::EMPTY + }, + ..ListenOpts::default() + }; + let mut l = Driver::::new(Config::default()).expect("loop"); + let (server, client, conn) = plain_pair(&mut l, &listen); + l.set_option(client, SocketOption::NoDelay(true)) + .expect("client nodelay"); + let mut echoed = 0; + let mut received = 0; + for round in 0..8u64 { + l.read(conn, ReadBuf::Pooled, Token(20)) + .expect("server read"); + l.write(client, WriteBuf::Owned(vec![round as u8]), Token(21)) + .expect("client write"); + let until = l.now() + Duration::from_secs(5); + let mut out = Completions::default(); + let mut echo = None; + while echo.is_none() { + assert!(l.now() < until, "round {round} stalled"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + if let OpResult::Read { n, lease: Some(b) } = c.result { + assert_eq!(n, 1); + assert_eq!(b.as_slice(), &[round as u8]); + if c.token == Token(20) { + received += 1; + l.read(client, ReadBuf::Pooled, Token(22)) + .expect("read back"); + l.write(conn, WriteBuf::Owned(vec![round as u8]), Token(23)) + .expect("echo"); + } else { + echo = Some(()); + echoed += 1; + } + } + } + } + } + assert_eq!((received, echoed), (8, 8), "every round must complete"); + close_all(&mut l, &[client, conn, server]); +} +/// A listener's accept defaults belong to the listener, not to the loop that +/// created it: they travel with the transport through `detach`/`attach`, so a +/// listener handed to another agent keeps configuring its connections there. +pub fn accept_defaults_survive_transfer() { + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + ..AcceptDefaults::EMPTY + }, + ..ListenOpts::default() + }; + let mut source = Driver::::new(Config::default()).expect("source loop"); + let server = source.tcp_listen(localhost(), &listen).expect("listen"); + let address = source.local_addr(server).expect("address"); + let transport = source.detach(server).expect("quiescent detach"); + assert!(!source.alive(), "the source loop keeps nothing"); + + let mut l = Driver::::new(Config::default()).expect("destination loop"); + let server = l.attach(transport, Token(1)).expect("attach"); + l.accept(server, Token(2)).expect("accept"); + let client = l + .tcp_connect(address, &TcpOpts::default(), Token(3)) + .expect("connect"); + let mut out = Completions::default(); + let until = l.now() + Duration::from_secs(5); + let (mut conn, mut connected) = (None, false); + while conn.is_none() || !connected { + assert!(l.now() < until, "transferred listener never accepted"); + l.turn(Timeout::Until(until), &mut out).expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Accepted { conn: h, .. } => conn = Some(h), + OpResult::Connected => connected = true, + other => panic!("unexpected {other:?}"), + } + } + } + let conn = conn.expect("accepted"); + assert!( + matches!( + read_option(&l, conn, SocketOptionKind::NoDelay), + SocketOption::NoDelay(true) + ), + "the default did not travel with the listener" + ); + close_all(&mut l, &[client, conn, server]); +} +/// Options the platform cannot express are reported, never accepted and ignored. +/// `expected` names what this backend genuinely lacks. +pub fn unsupported_options_are_reported( + expected: &[(SocketOption, Option)], +) { + assert!(!expected.is_empty(), "an empty matrix proves nothing"); + let mut l = Driver::::new(Config::default()).expect("loop"); + let udp = l.udp_bind(localhost(), &UdpOpts::default()).expect("bind"); + for (option, kind) in expected { + let refused = l + .set_option(udp, *option) + .err() + .unwrap_or_else(|| panic!("{option:?} was accepted by a backend that cannot apply it")); + assert_eq!( + refused.kind, + ErrorKind::Unsupported, + "set {option:?} must report Unsupported" + ); + if let Some(kind) = kind { + assert_eq!( + l.get_option(udp, *kind) + .expect_err("unsupported getter") + .kind, + ErrorKind::Unsupported, + "get {kind:?} must report Unsupported" + ); + } + } + close_all(&mut l, &[udp]); +} +/// A backend without Nagle control refuses the creation-time hint as well, so a +/// host cannot tell itself the connection is configured when it is not. +pub fn unsupported_connect_nodelay_is_rejected() { + let mut l = Driver::::new(Config::default()).expect("loop"); + assert_eq!( + l.tcp_connect( + (std::net::Ipv4Addr::LOCALHOST, 9).into(), + &TcpOpts { nodelay: true }, + Token(1), + ) + .expect_err("no Nagle control here") + .kind, + ErrorKind::Unsupported + ); + assert!(!l.alive(), "a rejected socket retains nothing"); +} +/// A listener default this backend cannot apply is refused when the listener is +/// created, not ignored once per accepted connection. +pub fn unsupported_accept_default_rejects_the_listener(defaults: AcceptDefaults) { + let mut l = Driver::::new(Config::default()).expect("loop"); + let listen = ListenOpts { + accept_defaults: defaults, + ..ListenOpts::default() + }; + assert_eq!( + l.tcp_listen(localhost(), &listen) + .expect_err("unsupported accept default") + .kind, + ErrorKind::Unsupported + ); + assert!(!l.alive(), "a rejected listener retains nothing"); +} diff --git a/crates/turnloop-contract/tests/allocations.rs b/crates/turnloop-contract/tests/allocations.rs index e4b273b..47b2768 100644 --- a/crates/turnloop-contract/tests/allocations.rs +++ b/crates/turnloop-contract/tests/allocations.rs @@ -445,6 +445,80 @@ fn steady_udp_allocate_nothing() { assert_eq!(total, 0, "steady UDP allocations"); } +/// Setting and reading a socket option is a syscall and nothing else: no queue, +/// no completion, no allocation (DESIGN §10 rule 1; issue #34). The options used +/// here are the ones every backend with sockets supports, so the gate runs +/// identically on native and WASI. +#[test] +fn steady_socket_options_allocate_nothing() { + use turnloop::{KeepAlive, SocketOption, SocketOptionKind}; + let mut l = Loop::new(Config::default()).expect("loop"); + let udp = l + .udp_bind("127.0.0.1:0".parse().expect("address"), &UdpOpts::default()) + .expect("UDP"); + let mut total = 0; + let mut applied = 0; + for i in 0..101u32 { + ALLOCS.with(|n| n.set(0)); + ACTIVE.with(|v| v.set(i != 0)); + let size = 32 * 1024 + (i % 8) * 4096; + l.set_option(udp, SocketOption::RecvBufferSize(size)) + .expect("receive buffer"); + l.set_option(udp, SocketOption::SendBufferSize(size)) + .expect("send buffer"); + l.set_option(udp, SocketOption::Ttl(1 + i % 64)) + .expect("hop limit"); + let read = l.get_option(udp, SocketOptionKind::Ttl).expect("read back"); + assert!( + matches!(read, SocketOption::Ttl(hops) if hops == 1 + i % 64), + "the OS did not keep the hop limit: {read:?}" + ); + let _ = l + .get_option(udp, SocketOptionKind::RecvBufferSize) + .expect("read receive buffer"); + ACTIVE.with(|v| v.set(false)); + if i != 0 { + applied += 1; + total += ALLOCS.with(|n| n.get()); + } + } + assert_eq!(applied, 100, "the socket-option subject ran"); + assert_eq!(total, 0, "steady socket-option allocations"); + // Keep-alive writes three TCP-level values behind one option; a connected + // socket proves that path allocates nothing either. + let (server, client, conn) = + turnloop_contract::sockopts::plain_pair(&mut l, &ListenOpts::default()); + let schedule = KeepAlive { + idle: Some(Duration::from_secs(5)), + interval: Some(Duration::from_secs(2)), + count: Some(3), + }; + let mut probes = 0; + for i in 0..101 { + ALLOCS.with(|n| n.set(0)); + ACTIVE.with(|v| v.set(i != 0)); + l.set_option(conn, SocketOption::KeepAlive(Some(schedule))) + .expect("enable"); + let read = l + .get_option(conn, SocketOptionKind::KeepAlive) + .expect("read"); + l.set_option(conn, SocketOption::KeepAlive(None)) + .expect("disable"); + ACTIVE.with(|v| v.set(false)); + assert!( + matches!(read, SocketOption::KeepAlive(Some(_))), + "keep-alive was not actually on: {read:?}" + ); + if i != 0 { + probes += 1; + total += ALLOCS.with(|n| n.get()); + } + } + assert_eq!(probes, 100, "the keep-alive subject ran"); + assert_eq!(total, 0, "steady keep-alive allocations"); + turnloop_contract::sockopts::close_all(&mut l, &[client, conn, server, udp]); +} + #[test] fn steady_deadline_poll_allocate_nothing() { let mut l = Loop::new(Config::default()).expect("loop"); diff --git a/crates/turnloop-contract/tests/sockopts.rs b/crates/turnloop-contract/tests/sockopts.rs new file mode 100644 index 0000000..a49977d --- /dev/null +++ b/crates/turnloop-contract/tests/sockopts.rs @@ -0,0 +1,396 @@ +//! Socket options on live handles, on every native backend (issue #34). +//! +//! The shared contract functions already read every value back from the OS +//! through `get_option`. The probes in this file go around turnloop entirely: +//! they call `getsockopt` from the test process on a descriptor turnloop does not +//! know they hold, so a backend that answered its own getter from a cache of what +//! was set could not pass them. +#![deny(unsafe_op_in_unsafe_fn)] +#![cfg(all( + not(loom), + any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd", + windows + ) +))] +use std::net::{Ipv4Addr, SocketAddr}; +use turnloop::{backend::Platform, *}; +use turnloop_contract::sockopts as contract; + +macro_rules! contract { + ($($name:ident),+ $(,)?) => { $(#[test] fn $name() { contract::$name::(); })+ }; +} +contract!( + keep_alive_round_trip, + buffer_sizes_round_trip, + ttl_round_trip, + accept_defaults_keep_alive, + option_handle_validation, + linger_zero_resets_the_connection, + udp_broadcast_and_multicast_options, + ipv6_only_is_readable_and_bind_time, + nodelay_round_trip_and_accept_default, + nodelay_small_write_round_trip, + accept_defaults_survive_transfer, +); +#[test] +fn multicast_membership_is_tracked() { + contract::multicast_membership_is_tracked::( + Ipv4Addr::new(224, 0, 0, 251).into(), + (Ipv4Addr::UNSPECIFIED, 0).into(), + ); +} + +#[cfg(unix)] +mod probe { + use std::os::fd::{IntoRawFd, OwnedFd, RawFd}; + /// `getsockopt` straight from the test process, with no turnloop code involved. + pub fn int(fd: RawFd, level: i32, name: i32) -> i32 { + let mut value = 0i32; + let mut len = std::mem::size_of::() as libc::socklen_t; + // SAFETY: a live descriptor with writable integer output and its capacity. + let code = unsafe { + libc::getsockopt( + fd, + level, + name, + std::ptr::from_mut(&mut value).cast(), + &mut len, + ) + }; + assert_eq!(code, 0, "getsockopt: {}", std::io::Error::last_os_error()); + value + } + pub fn nodelay(fd: RawFd) -> bool { + int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY) != 0 + } + pub fn keep_alive(fd: RawFd) -> bool { + int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE) != 0 + } + fn endpoint(fd: RawFd, peer: bool) -> Option { + // SAFETY: plain writable address storage; zero is a valid initial value. + let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut len = std::mem::size_of::() as libc::socklen_t; + let out = std::ptr::from_mut(&mut storage).cast(); + // SAFETY: a possibly-unrelated descriptor with valid output storage; a + // non-socket simply fails and is skipped. + let code = unsafe { + if peer { + libc::getpeername(fd, out, &mut len) + } else { + libc::getsockname(fd, out, &mut len) + } + }; + if code != 0 || i32::from(storage.ss_family) != libc::AF_INET { + return None; + } + // SAFETY: the family was checked and the storage is aligned for sockaddr_in. + let addr = unsafe { &*std::ptr::from_ref(&storage).cast::() }; + Some( + ( + std::net::Ipv4Addr::from(addr.sin_addr.s_addr.to_ne_bytes()), + u16::from_be(addr.sin_port), + ) + .into(), + ) + } + /// Find the descriptor turnloop accepted, by its endpoints alone. The listener + /// shares the local address but has no peer, so the match is unique. + pub fn accepted(local: std::net::SocketAddr, peer: std::net::SocketAddr) -> RawFd { + let directory = if cfg!(any(target_os = "linux", target_os = "android")) { + "/proc/self/fd" + } else { + "/dev/fd" + }; + let mut found = None; + for entry in std::fs::read_dir(directory).expect("descriptor directory") { + let entry = entry.expect("descriptor entry"); + let Ok(fd) = entry.file_name().to_string_lossy().parse::() else { + continue; + }; + if endpoint(fd, false) == Some(local) && endpoint(fd, true) == Some(peer) { + assert!(found.is_none(), "two descriptors match {local} <- {peer}"); + found = Some(fd); + } + } + found.unwrap_or_else(|| panic!("no descriptor matches {local} <- {peer}")) + } + /// A second reference to one socket: `dup` shares the underlying socket, so an + /// option set through turnloop's handle is visible here. The clone is leaked + /// on purpose and reclaimed by `close_raw` once the loop has released its own. + pub fn shared(socket: impl Into) -> (OwnedFd, RawFd) { + let owned: OwnedFd = socket.into(); + let raw = owned.try_clone().expect("dup").into_raw_fd(); + (owned, raw) + } +} + +/// The accepted connection really carries `TCP_NODELAY` and `SO_KEEPALIVE`, +/// asserted with the test's own `getsockopt` on the accepted descriptor, found +/// without asking turnloop for it. +#[cfg(unix)] +#[test] +fn accepted_socket_options_are_visible_to_getsockopt() { + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + keep_alive: Some(KeepAlive { + idle: Some(std::time::Duration::from_secs(9)), + ..KeepAlive::default() + }), + }, + ..ListenOpts::default() + }; + let mut l = Loop::new(Config::default()).expect("loop"); + let (server, client, conn) = contract::plain_pair(&mut l, &listen); + let local = l.local_addr(conn).expect("accepted local address"); + let peer = l.local_addr(client).expect("client local address"); + let fd = probe::accepted(local, peer); + assert!( + probe::nodelay(fd), + "the listener's nodelay default never reached the accepted socket" + ); + assert!( + probe::keep_alive(fd), + "the listener's keep-alive default never reached the accepted socket" + ); + assert_eq!( + probe::int(fd, libc::IPPROTO_TCP, KEEPIDLE), + 9, + "the accepted socket kept a different idle time" + ); + // And a later per-connection change is equally real. + l.set_option(conn, SocketOption::NoDelay(false)) + .expect("clear nodelay"); + assert!( + !probe::nodelay(fd), + "set_option(false) did not reach the accepted socket" + ); + l.set_option(conn, SocketOption::KeepAlive(None)) + .expect("stop probing"); + assert!(!probe::keep_alive(fd), "keep-alive was not turned off"); + contract::close_all(&mut l, &[client, conn, server]); +} +#[cfg(all(unix, target_vendor = "apple"))] +const KEEPIDLE: i32 = libc::TCP_KEEPALIVE; +#[cfg(all(unix, not(target_vendor = "apple")))] +const KEEPIDLE: i32 = libc::TCP_KEEPIDLE; + +/// An adopted socket is configurable too, proven on a second reference to the +/// same socket that the loop knows nothing about. +#[test] +fn adopted_socket_options_reach_the_shared_socket() { + let listener = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).expect("listener"); + let address: SocketAddr = listener.local_addr().expect("address"); + let client = std::net::TcpStream::connect(address).expect("connect"); + let (accepted, _) = listener.accept().expect("accept"); + let mut l = Loop::new(Config::default()).expect("loop"); + + #[cfg(unix)] + let (owned, raw) = { + let (owned, raw) = probe::shared(accepted); + (Detached::from_fd(owned).expect("adopt"), raw) + }; + #[cfg(windows)] + let (owned, raw) = { + let shared = accepted.try_clone().expect("duplicate"); + let raw = windows_probe::leak(shared); + (Detached::from_socket(accepted.into()).expect("adopt"), raw) + }; + let h = l.attach(owned, Token(1)).expect("attach"); + assert!(!nodelay_of(raw), "the shared socket starts with Nagle on"); + l.set_option(h, SocketOption::NoDelay(true)) + .expect("nodelay"); + assert!( + nodelay_of(raw), + "set_option did not reach the shared socket" + ); + // A numeric option lands on the shared socket too. The hop limit carries that + // half of the proof because no kernel rounds it: the buffer sizes below are + // kernel-chosen (see `SocketOption::RecvBufferSize`), so they can only be + // checked for plausibility, never for the exact value asked. + let before = ttl_of(raw); + assert!( + before != 7 && before > 0, + "pick a hop limit the socket does not already have: {before}" + ); + l.set_option(h, SocketOption::Ttl(7)).expect("hop limit"); + assert_eq!( + ttl_of(raw), + 7, + "set_option did not reach the shared socket's hop limit" + ); + // Plausible honouring only: Linux doubles the request, macOS keeps it, and + // Windows may hold an auto-tuned receive window above it. Every platform + // gives at least what was asked for. + let requested = 48 * 1024; + l.set_option(h, SocketOption::RecvBufferSize(requested)) + .expect("receive buffer"); + let bytes = recv_buffer_of(raw); + assert!( + bytes >= requested, + "the OS reports {bytes} bytes for a {requested}-byte request" + ); + contract::close_all(&mut l, &[h]); + drop(client); + close_raw(raw); +} +#[cfg(unix)] +fn nodelay_of(raw: std::os::fd::RawFd) -> bool { + probe::nodelay(raw) +} +#[cfg(unix)] +fn recv_buffer_of(raw: std::os::fd::RawFd) -> u32 { + probe::int(raw, libc::SOL_SOCKET, libc::SO_RCVBUF).max(0) as u32 +} +#[cfg(unix)] +fn ttl_of(raw: std::os::fd::RawFd) -> u32 { + probe::int(raw, libc::IPPROTO_IP, libc::IP_TTL).max(0) as u32 +} +#[cfg(unix)] +fn close_raw(raw: std::os::fd::RawFd) { + // SAFETY: this descriptor was leaked by `probe::shared` and has no other owner. + drop(unsafe { ::from_raw_fd(raw) }); +} +#[cfg(windows)] +mod windows_probe { + use std::os::windows::io::{IntoRawSocket, OwnedSocket, RawSocket}; + use windows_sys::Win32::Networking::WinSock::*; + pub fn leak(socket: std::net::TcpStream) -> RawSocket { + let owned: OwnedSocket = socket.into(); + owned.into_raw_socket() + } + pub fn int(raw: RawSocket, level: i32, name: i32) -> i32 { + let mut value = 0i32; + let mut len = std::mem::size_of::() as i32; + // SAFETY: a live socket with writable integer output and its capacity. + let code = unsafe { + getsockopt( + raw as usize, + level, + name, + std::ptr::from_mut(&mut value).cast(), + &mut len, + ) + }; + assert_eq!(code, 0, "getsockopt: {}", std::io::Error::last_os_error()); + value + } + pub fn close(raw: RawSocket) { + // SAFETY: this socket was leaked by `leak` and has no other owner. + drop(unsafe { ::from_raw_socket(raw) }); + } +} +#[cfg(windows)] +fn nodelay_of(raw: std::os::windows::io::RawSocket) -> bool { + windows_probe::int( + raw, + windows_sys::Win32::Networking::WinSock::IPPROTO_TCP, + windows_sys::Win32::Networking::WinSock::TCP_NODELAY, + ) != 0 +} +#[cfg(windows)] +fn recv_buffer_of(raw: std::os::windows::io::RawSocket) -> u32 { + windows_probe::int( + raw, + windows_sys::Win32::Networking::WinSock::SOL_SOCKET, + windows_sys::Win32::Networking::WinSock::SO_RCVBUF, + ) + .max(0) as u32 +} +#[cfg(windows)] +fn ttl_of(raw: std::os::windows::io::RawSocket) -> u32 { + windows_probe::int( + raw, + windows_sys::Win32::Networking::WinSock::IPPROTO_IP, + windows_sys::Win32::Networking::WinSock::IP_TTL, + ) + .max(0) as u32 +} +#[cfg(windows)] +fn close_raw(raw: std::os::windows::io::RawSocket) { + windows_probe::close(raw); +} + +/// Handles that are not sockets report `Unsupported` instead of reaching a +/// backend table with a foreign index. +#[test] +fn non_socket_handles_report_unsupported() { + let mut l = Loop::new(Config::default()).expect("loop"); + let stdio = l.open_stdio(Stdio::Stderr).expect("stderr"); + assert_eq!( + l.set_option(stdio, SocketOption::NoDelay(true)) + .expect_err("stderr is not a socket") + .kind, + ErrorKind::Unsupported + ); + assert_eq!( + l.get_option(stdio, SocketOptionKind::NoDelay) + .expect_err("stderr is not a socket") + .kind, + ErrorKind::Unsupported + ); + contract::close_all(&mut l, &[stdio]); +} +/// A local (AF_UNIX / named pipe) listener cannot apply TCP accept defaults, so +/// it refuses them when it is created rather than per connection. +#[test] +fn local_listener_refuses_tcp_accept_defaults() { + let mut l = Loop::new(Config::default()).expect("loop"); + let name = local_name(); + let listen = ListenOpts { + accept_defaults: AcceptDefaults { + nodelay: true, + ..AcceptDefaults::EMPTY + }, + ..ListenOpts::default() + }; + assert_eq!( + l.pipe_listen(&name, &listen) + .expect_err("TCP defaults on a local listener") + .kind, + ErrorKind::Unsupported + ); + assert!(!l.alive(), "a rejected listener retains nothing"); +} +#[cfg(unix)] +fn local_name() -> PipeName { + PipeName(std::env::temp_dir().join(format!("tl-sockopt-{}.sock", std::process::id()))) +} +#[cfg(windows)] +fn local_name() -> PipeName { + PipeName(format!(r"\\.\pipe\turnloop-sockopt-{}", std::process::id()).into()) +} +/// An IPv4 membership keyed by interface index exists only on Linux; elsewhere it +/// is reported rather than quietly applied to the default interface. +#[test] +fn ipv4_membership_by_interface_index() { + let mut l = Loop::new(Config::default()).expect("loop"); + let udp = l + .udp_bind((Ipv4Addr::UNSPECIFIED, 0).into(), &UdpOpts::default()) + .expect("bind"); + let group = MulticastGroup { + group: Ipv4Addr::new(224, 0, 0, 251).into(), + interface: 1, + }; + let result = l.set_option(udp, SocketOption::MulticastJoin(group)); + if cfg!(any(target_os = "linux", target_os = "android")) { + // Interface 1 is the loopback index on Linux; either it joins or the + // kernel explains why, but the request is never silently redirected. + if let Err(e) = result { + assert_ne!(e.kind, ErrorKind::Unsupported, "Linux supports ip_mreqn"); + } else { + l.set_option(udp, SocketOption::MulticastLeave(group)) + .expect("leave"); + } + } else { + assert_eq!( + result.expect_err("no ip_mreqn here").kind, + ErrorKind::Unsupported + ); + } + contract::close_all(&mut l, &[udp]); +} diff --git a/crates/turnloop-contract/tests/wasi.rs b/crates/turnloop-contract/tests/wasi.rs index 999f818..535f5aa 100644 --- a/crates/turnloop-contract/tests/wasi.rs +++ b/crates/turnloop-contract/tests/wasi.rs @@ -137,6 +137,78 @@ fn timer_liveness() { fn timers_io_posts() { contract::io_and_posts_progress_with_repeating_timers::(); } +/// Socket options (issue #34). `wasi:sockets` exposes keep-alive, buffer sizes +/// and the hop limit and nothing else, so those round-trip through the OS and the +/// rest must report Unsupported instead of being accepted and dropped. +#[test] +fn socket_option_keep_alive() { + contract::sockopts::keep_alive_round_trip::(); +} +#[test] +fn socket_option_buffer_sizes() { + contract::sockopts::buffer_sizes_round_trip::(); +} +#[test] +fn socket_option_ttl() { + contract::sockopts::ttl_round_trip::(); +} +#[test] +fn socket_option_accept_defaults() { + contract::sockopts::accept_defaults_keep_alive::(); +} +#[test] +fn socket_option_handle_validation() { + contract::sockopts::option_handle_validation::(); +} +#[test] +fn socket_options_without_a_wasi_interface_are_unsupported() { + use std::time::Duration; + use turnloop::{MulticastGroup, SocketOption, SocketOptionKind}; + let group = MulticastGroup { + group: std::net::Ipv4Addr::new(224, 0, 0, 251).into(), + interface: 0, + }; + contract::sockopts::unsupported_options_are_reported::(&[ + (SocketOption::NoDelay(true), Some(SocketOptionKind::NoDelay)), + ( + SocketOption::Linger(Some(Duration::ZERO)), + Some(SocketOptionKind::Linger), + ), + ( + SocketOption::Ipv6Only(true), + Some(SocketOptionKind::Ipv6Only), + ), + ( + SocketOption::Broadcast(true), + Some(SocketOptionKind::Broadcast), + ), + ( + SocketOption::MulticastTtl(4), + Some(SocketOptionKind::MulticastTtl), + ), + ( + SocketOption::MulticastLoop(false), + Some(SocketOptionKind::MulticastLoop), + ), + (SocketOption::MulticastJoin(group), None), + (SocketOption::MulticastLeave(group), None), + ]); +} +/// `wasi:sockets` has no Nagle control, so a listener asking for it as a +/// per-connection default is refused when it is created. +#[test] +fn nodelay_connect_hint_is_rejected() { + contract::sockopts::unsupported_connect_nodelay_is_rejected::(); +} +#[test] +fn nodelay_accept_default_rejects_the_listener() { + contract::sockopts::unsupported_accept_default_rejects_the_listener::( + turnloop::AcceptDefaults { + nodelay: true, + ..turnloop::AcceptDefaults::EMPTY + }, + ); +} #[test] fn no_spin() { contract::no_spin::(); diff --git a/crates/turnloop-contract/tests/web/web_contract.rs b/crates/turnloop-contract/tests/web/web_contract.rs index 96a5178..2b81573 100644 --- a/crates/turnloop-contract/tests/web/web_contract.rs +++ b/crates/turnloop-contract/tests/web/web_contract.rs @@ -100,6 +100,73 @@ fn now_only_and_unsupported() { .kind, ErrorKind::Unsupported ); + // A timer is not a socket on any backend, so the core rejects it before the + // browser adapter is asked anything (issue #34). + let timer = l + .timer(l.now() + Duration::from_secs(60), None, Token(2)) + .expect("timer"); + assert_eq!( + l.set_option(timer, SocketOption::NoDelay(true)) + .expect_err("a timer is not a socket") + .kind, + ErrorKind::InvalidInput + ); + assert_eq!( + l.get_option(timer, SocketOptionKind::NoDelay) + .expect_err("a timer is not a socket") + .kind, + ErrorKind::InvalidInput + ); + l.close(timer, Token(3)).expect("close"); + l.turn(Timeout::Now, &mut out).expect("drain"); +} +/// A browser host has no socket behind its WebSocket, so every option is +/// reported `Unsupported` rather than accepted and ignored (DESIGN §7.5). +#[wasm_bindgen_test(async)] +async fn socket_options_are_unsupported_on_a_host_stream() { + let mut l = Loop::new(Config::default()).expect("loop"); + let h = websocket(&mut l).await; + for option in [ + SocketOption::NoDelay(true), + SocketOption::KeepAlive(None), + SocketOption::Linger(Some(Duration::ZERO)), + SocketOption::RecvBufferSize(4096), + SocketOption::SendBufferSize(4096), + SocketOption::Ttl(4), + SocketOption::Ipv6Only(true), + SocketOption::Broadcast(true), + SocketOption::MulticastTtl(4), + SocketOption::MulticastLoop(true), + ] { + assert_eq!( + l.set_option(h, option).expect_err("no socket here").kind, + ErrorKind::Unsupported, + "set {option:?}" + ); + } + for kind in [ + SocketOptionKind::NoDelay, + SocketOptionKind::KeepAlive, + SocketOptionKind::Linger, + SocketOptionKind::RecvBufferSize, + SocketOptionKind::SendBufferSize, + SocketOptionKind::Ttl, + SocketOptionKind::Ipv6Only, + SocketOptionKind::Broadcast, + SocketOptionKind::MulticastTtl, + SocketOptionKind::MulticastLoop, + ] { + assert_eq!( + l.get_option(h, kind).expect_err("no socket here").kind, + ErrorKind::Unsupported, + "get {kind:?}" + ); + } + let mut out = Completions::default(); + l.close(h, Token(93)).expect("close websocket"); + l.turn(Timeout::Now, &mut out).expect("drain close"); + assert_eq!(out.len(), 1); + assert!(matches!(out[0].result, OpResult::Closed)); } #[wasm_bindgen_test(async)] async fn queued_post_with_idle_callback_io_never_waits() { diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index b2dbe51..53550d1 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -9,6 +9,7 @@ mod process; mod services; mod signals; mod socket; +mod sockopt; mod sync_io; mod timer; mod watch; @@ -87,6 +88,10 @@ pub struct Detached { port: Option>, mode: Option, console_input: bool, + /// A listener's per-connection defaults, applied to each socket it accepts. + /// It travels with the listener across detach/attach, so a listener handed to + /// another loop keeps configuring its connections there. + accept_defaults: AcceptDefaults, } impl Detached { fn new(native: Native, kind: Kind, routed: bool) -> Self { @@ -97,6 +102,7 @@ impl Detached { port: None, mode: None, console_input: false, + accept_defaults: AcceptDefaults::EMPTY, } } } @@ -210,6 +216,18 @@ impl Iocp { .as_ref() .is_some_and(|p| p.request.op == op && p.waiting) } + /// The Winsock socket behind a handle. Process, signal and watch handles, named + /// pipes and adopted files/consoles are not sockets and say so. + fn socket_of(&self, h: Handle) -> Result { + if self.services.contains(h) || self.watches.contains(h) { + return Err(unsupported()); + } + let r = self.get(h)?; + match r.transport.native { + Native::Socket(_) => Ok(r.transport.native.raw() as usize), + _ => Err(unsupported()), + } + } fn get(&self, h: Handle) -> Result<&Resource> { self.resources .get(h.index()) @@ -469,6 +487,7 @@ impl Iocp { } let raw = r.transport.native.raw(); let kind = r.transport.kind; + let defaults = r.transport.accept_defaults; if let Some(result) = p.completion.take() { let bytes = match result { Ok(bytes) => bytes as usize, @@ -513,6 +532,10 @@ impl Iocp { size_of::() as i32, ) })?; + // Before the connection becomes visible to the host, and + // after the accept context makes the socket queryable. A + // rejected default fails this accept and drops the socket. + sockopt::apply_accept_defaults(transport.native.raw() as usize, defaults)?; let peer = socket::address(transport.native.raw() as usize, true)?; Outcome::Accepted { transport, peer } } else { @@ -1211,17 +1234,16 @@ unsafe impl Backend for Iocp { if opts.reuse_port { return Err(unsupported()); } + sockopt::validate_accept_defaults(opts.accept_defaults, Kind::Listener)?; let socket = socket::create(addr.is_ipv6(), false)?; let raw = socket.as_raw_socket() as usize; socket::option(raw, SO_EXCLUSIVEADDRUSE, 1)?; socket::bind_to(raw, addr)?; // SAFETY: bound socket; backlog is bounded to the signed API range. socket::check(unsafe { listen(raw, opts.backlog.min(i32::MAX as u32) as i32) })?; - ( - Detached::new(Native::Socket(socket), Kind::Listener, false), - None, - None, - ) + let mut transport = Detached::new(Native::Socket(socket), Kind::Listener, false); + transport.accept_defaults = opts.accept_defaults; + (transport, None, None) } Open::Udp { addr, opts } => { if opts.reuse_port { @@ -1239,6 +1261,7 @@ unsafe impl Backend for Iocp { if opts.reuse_port { return Err(unsupported()); } + sockopt::validate_accept_defaults(opts.accept_defaults, Kind::PipeListener)?; let name = pipes::name(&name)?; let key = self.next_listener_key; self.next_listener_key = key @@ -1285,6 +1308,12 @@ unsafe impl Backend for Iocp { } socket::address(r.transport.native.raw() as usize, false) } + fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + sockopt::set(self.socket_of(h)?, option) + } + fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { + sockopt::get(self.socket_of(h)?, kind) + } fn submit(&mut self, request: Request) -> Result<()> { if self.services.contains(request.handle) { return self.services.submit(request); diff --git a/crates/turnloop/src/backend/iocp/sockopt.rs b/crates/turnloop/src/backend/iocp/sockopt.rs new file mode 100644 index 0000000..77a9f6e --- /dev/null +++ b/crates/turnloop/src/backend/iocp/sockopt.rs @@ -0,0 +1,283 @@ +//! Typed socket options on live Winsock sockets (DESIGN §7.3, §7.7). +//! +//! Synchronous, allocation-free and uncached, exactly like the Unix module: a +//! getter always asks Winsock, so a buffer size the stack rounded is visible. +//! Winsock reports a wrong-level option itself (`WSAENOPROTOOPT`), and that error +//! is returned unchanged rather than guessed at. +use super::{Kind, invalid, unsupported}; +use crate::{ + AcceptDefaults, Error, KeepAlive, MulticastGroup, Result, SocketOption, SocketOptionKind, +}; +use std::{mem::size_of, net::IpAddr, ptr, time::Duration}; +use windows_sys::Win32::Networking::WinSock::*; + +fn set_int(socket: usize, level: i32, name: i32, value: i32) -> Result<()> { + // SAFETY: value is an initialized integer of exactly the length passed. + super::socket::check(unsafe { + setsockopt( + socket, + level, + name, + ptr::from_ref(&value).cast(), + size_of::() as i32, + ) + }) +} +fn get_int(socket: usize, level: i32, name: i32) -> Result { + let mut value = 0i32; + let mut len = size_of::() as i32; + // SAFETY: writable initialized integer output with its exact capacity. + super::socket::check(unsafe { + getsockopt( + socket, + level, + name, + ptr::from_mut(&mut value).cast(), + &mut len, + ) + })?; + Ok(value) +} +fn bounded(value: u32) -> Result { + i32::try_from(value).map_err(|_| invalid()) +} +/// One-second granularity, rounded up: a caller asking for 1.2 s must not get 1 s. +fn whole_seconds(value: Duration) -> Result { + let seconds = value + .as_secs() + .checked_add(u64::from(value.subsec_nanos() != 0)) + .ok_or_else(invalid)?; + u32::try_from(seconds).map_err(|_| invalid()) +} +fn positive_seconds(value: Duration) -> Result { + match whole_seconds(value)? { + 0 => Err(invalid()), + seconds => Ok(seconds), + } +} +/// True for an IPv6 socket, false for IPv4; anything else has no IP options. +fn ipv6(socket: usize) -> Result { + // SAFETY: plain C address storage; zero is a valid initial value. + let mut storage: SOCKADDR_STORAGE = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as i32; + // SAFETY: live socket with writable address storage and its capacity. + super::socket::check(unsafe { + getsockname(socket, ptr::from_mut(&mut storage).cast(), &mut len) + })?; + match storage.ss_family { + AF_INET => Ok(false), + AF_INET6 => Ok(true), + _ => Err(unsupported()), + } +} +fn ip_level(socket: usize, v4: i32, v6: i32) -> Result<(i32, i32)> { + if ipv6(socket)? { + Ok((IPPROTO_IPV6, v6)) + } else { + Ok((IPPROTO_IP, v4)) + } +} +fn set_linger(socket: usize, value: Option) -> Result<()> { + let seconds = value.map_or(Ok(0), whole_seconds)?; + let value = LINGER { + l_onoff: u16::from(value.is_some()), + l_linger: u16::try_from(seconds).map_err(|_| invalid())?, + }; + // SAFETY: a fully initialized LINGER of exactly the length passed. + super::socket::check(unsafe { + setsockopt( + socket, + SOL_SOCKET, + SO_LINGER, + ptr::from_ref(&value).cast(), + size_of::() as i32, + ) + }) +} +fn get_linger(socket: usize) -> Result> { + let mut value = LINGER::default(); + let mut len = size_of::() as i32; + // SAFETY: writable LINGER output with its exact capacity. + super::socket::check(unsafe { + getsockopt( + socket, + SOL_SOCKET, + SO_LINGER, + ptr::from_mut(&mut value).cast(), + &mut len, + ) + })?; + Ok((value.l_onoff != 0).then(|| Duration::from_secs(u64::from(value.l_linger)))) +} +/// `SO_KEEPALIVE` plus the three TCP schedule values Windows 10 1709 added. +/// Disabling clears the master switch and leaves the schedule alone, as the OS does. +fn set_keep_alive(socket: usize, value: Option) -> Result<()> { + let Some(schedule) = value else { + return set_int(socket, SOL_SOCKET, SO_KEEPALIVE, 0); + }; + let idle = schedule.idle.map(positive_seconds).transpose()?; + let interval = schedule.interval.map(positive_seconds).transpose()?; + let count = schedule.count.map(bounded).transpose()?; + if count == Some(0) { + return Err(invalid()); + } + set_int(socket, SOL_SOCKET, SO_KEEPALIVE, 1)?; + if let Some(idle) = idle { + set_int(socket, IPPROTO_TCP, TCP_KEEPIDLE, bounded(idle)?)?; + } + if let Some(interval) = interval { + set_int(socket, IPPROTO_TCP, TCP_KEEPINTVL, bounded(interval)?)?; + } + if let Some(count) = count { + set_int(socket, IPPROTO_TCP, TCP_KEEPCNT, count)?; + } + Ok(()) +} +fn get_keep_alive(socket: usize) -> Result> { + if get_int(socket, SOL_SOCKET, SO_KEEPALIVE)? == 0 { + return Ok(None); + } + Ok(Some(KeepAlive { + idle: Some(Duration::from_secs( + get_int(socket, IPPROTO_TCP, TCP_KEEPIDLE)?.max(0) as u64, + )), + interval: Some(Duration::from_secs( + get_int(socket, IPPROTO_TCP, TCP_KEEPINTVL)?.max(0) as u64, + )), + count: Some(get_int(socket, IPPROTO_TCP, TCP_KEEPCNT)?.max(0) as u32), + })) +} +fn membership(socket: usize, group: MulticastGroup, join: bool) -> Result<()> { + match (group.group, ipv6(socket)?) { + (IpAddr::V6(address), true) => { + let mut request = IPV6_MREQ::default(); + request.ipv6mr_multiaddr.u.Byte = address.octets(); + request.ipv6mr_interface = group.interface; + let name = if join { + IPV6_ADD_MEMBERSHIP + } else { + IPV6_DROP_MEMBERSHIP + }; + // SAFETY: a fully initialized request of exactly the length passed. + super::socket::check(unsafe { + setsockopt( + socket, + IPPROTO_IPV6, + name, + ptr::from_ref(&request).cast(), + size_of::() as i32, + ) + }) + } + // Winsock's IPv4 membership names the interface by address, not index, so + // a nonzero index is reported instead of being applied to the default one. + (IpAddr::V4(_), false) if group.interface != 0 => Err(unsupported()), + (IpAddr::V4(address), false) => { + let mut request = IP_MREQ::default(); + request.imr_multiaddr.S_un.S_addr = u32::from_ne_bytes(address.octets()); + request.imr_interface.S_un.S_addr = 0; + let name = if join { + IP_ADD_MEMBERSHIP + } else { + IP_DROP_MEMBERSHIP + }; + // SAFETY: a fully initialized request of exactly the length passed. + super::socket::check(unsafe { + setsockopt( + socket, + IPPROTO_IP, + name, + ptr::from_ref(&request).cast(), + size_of::() as i32, + ) + }) + } + _ => Err(invalid()), + } +} + +/// Apply one option to a live socket. +pub(super) fn set(socket: usize, option: SocketOption) -> Result<()> { + match option { + SocketOption::NoDelay(on) => set_int(socket, IPPROTO_TCP, TCP_NODELAY, i32::from(on)), + SocketOption::KeepAlive(schedule) => set_keep_alive(socket, schedule), + SocketOption::Linger(value) => set_linger(socket, value), + SocketOption::RecvBufferSize(bytes) => { + set_int(socket, SOL_SOCKET, SO_RCVBUF, bounded(bytes)?) + } + SocketOption::SendBufferSize(bytes) => { + set_int(socket, SOL_SOCKET, SO_SNDBUF, bounded(bytes)?) + } + SocketOption::Ttl(hops) => { + let (level, name) = ip_level(socket, IP_TTL, IPV6_UNICAST_HOPS)?; + set_int(socket, level, name, bounded(hops)?) + } + SocketOption::Ipv6Only(on) => set_int(socket, IPPROTO_IPV6, IPV6_V6ONLY, i32::from(on)), + SocketOption::Broadcast(on) => set_int(socket, SOL_SOCKET, SO_BROADCAST, i32::from(on)), + SocketOption::MulticastTtl(hops) => { + let (level, name) = ip_level(socket, IP_MULTICAST_TTL, IPV6_MULTICAST_HOPS)?; + set_int(socket, level, name, bounded(hops)?) + } + SocketOption::MulticastLoop(on) => { + let (level, name) = ip_level(socket, IP_MULTICAST_LOOP, IPV6_MULTICAST_LOOP)?; + set_int(socket, level, name, i32::from(on)) + } + SocketOption::MulticastJoin(group) => membership(socket, group, true), + SocketOption::MulticastLeave(group) => membership(socket, group, false), + } +} +/// Read one option back from Winsock. +pub(super) fn get(socket: usize, kind: SocketOptionKind) -> Result { + Ok(match kind { + SocketOptionKind::NoDelay => { + SocketOption::NoDelay(get_int(socket, IPPROTO_TCP, TCP_NODELAY)? != 0) + } + SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(socket)?), + SocketOptionKind::Linger => SocketOption::Linger(get_linger(socket)?), + SocketOptionKind::RecvBufferSize => { + SocketOption::RecvBufferSize(get_int(socket, SOL_SOCKET, SO_RCVBUF)?.max(0) as u32) + } + SocketOptionKind::SendBufferSize => { + SocketOption::SendBufferSize(get_int(socket, SOL_SOCKET, SO_SNDBUF)?.max(0) as u32) + } + SocketOptionKind::Ttl => { + let (level, name) = ip_level(socket, IP_TTL, IPV6_UNICAST_HOPS)?; + SocketOption::Ttl(get_int(socket, level, name)?.max(0) as u32) + } + SocketOptionKind::Ipv6Only => { + SocketOption::Ipv6Only(get_int(socket, IPPROTO_IPV6, IPV6_V6ONLY)? != 0) + } + SocketOptionKind::Broadcast => { + SocketOption::Broadcast(get_int(socket, SOL_SOCKET, SO_BROADCAST)? != 0) + } + SocketOptionKind::MulticastTtl => { + let (level, name) = ip_level(socket, IP_MULTICAST_TTL, IPV6_MULTICAST_HOPS)?; + SocketOption::MulticastTtl(get_int(socket, level, name)?.max(0) as u32) + } + SocketOptionKind::MulticastLoop => { + let (level, name) = ip_level(socket, IP_MULTICAST_LOOP, IPV6_MULTICAST_LOOP)?; + SocketOption::MulticastLoop(get_int(socket, level, name)? != 0) + } + }) +} +/// Apply a listener's accepted-socket defaults to one accepted socket, after +/// `SO_UPDATE_ACCEPT_CONTEXT` (which is what makes the socket queryable) and +/// before the `Accepted` outcome exists. +pub(super) fn apply_accept_defaults(socket: usize, defaults: AcceptDefaults) -> Result<()> { + if defaults.nodelay { + set(socket, SocketOption::NoDelay(true))?; + } + if let Some(schedule) = defaults.keep_alive { + set(socket, SocketOption::KeepAlive(Some(schedule)))?; + } + Ok(()) +} +/// Reject at listener creation what this backend could not apply per connection. +/// Both native defaults are TCP options, so a named-pipe listener refuses them. +pub(super) fn validate_accept_defaults(defaults: AcceptDefaults, kind: Kind) -> Result<()> { + if kind == Kind::Listener || defaults.is_empty() { + Ok(()) + } else { + Err(Error::new(crate::ErrorKind::Unsupported)) + } +} diff --git a/crates/turnloop/src/backend/mod.rs b/crates/turnloop/src/backend/mod.rs index 5c386f1..0424721 100644 --- a/crates/turnloop/src/backend/mod.rs +++ b/crates/turnloop/src/backend/mod.rs @@ -80,6 +80,18 @@ //! Windows' explicitly requested Integration::Event helper is the D7 exception. //! * Unsupported platforms/capabilities return errors; no fake successful I/O. //! +//! # Socket options (§7.7) +//! +//! * `set_option`/`get_option` are synchronous and produce no Event. They run +//! entirely inside the call, touch no operation storage, and must not allocate. +//! * A backend reports `Unsupported` for an option its platform lacks. It must +//! never accept and ignore one: a host reads a lie back from `get_option`, and +//! "the option is on" is a security- and latency-relevant claim. +//! * `ListenOpts::accept_defaults` is applied by the accepting backend to each +//! accepted transport before its `Accepted`/`PipeAccepted` Outcome is produced, +//! so the core hands the host an already-configured connection. A backend that +//! cannot apply a requested default rejects the listener in `open`. +//! //! # Filesystem (§7.6 Files) //! //! * `FILESYSTEM` selects where typed `FsRequest`s run. `Pool` (native default): @@ -345,6 +357,23 @@ pub unsafe trait Backend: Sized + 'static { fn tty_set_mode(&mut self, _handle: Handle, _mode: crate::TtyMode) -> Result<()> { Err(Error::new(crate::ErrorKind::Unsupported)) } + /// Apply one socket option to a live resource, synchronously, through the OS. + /// No Request is accepted and no Event is produced. A platform without an + /// equivalent returns Unsupported; silently ignoring an option is forbidden, + /// because a host cannot tell an ignored option from an applied one. + fn set_option(&mut self, _handle: Handle, _option: crate::SocketOption) -> Result<()> { + Err(Error::new(crate::ErrorKind::Unsupported)) + } + /// Read one socket option from the OS. Backends never answer from a cache of + /// what was set: the kernel may round, clamp or double a request, and the + /// caller is entitled to the value the kernel actually holds. + fn get_option( + &self, + _handle: Handle, + _kind: crate::SocketOptionKind, + ) -> Result { + Err(Error::new(crate::ErrorKind::Unsupported)) + } /// Query current terminal dimensions. fn tty_window_size(&self, _handle: Handle) -> Result { Err(Error::new(crate::ErrorKind::Unsupported)) @@ -409,6 +438,8 @@ mod poller; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] mod socket; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] +mod sockopt; +#[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] pub mod unix; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] pub use unix::Unix as Platform; diff --git a/crates/turnloop/src/backend/sockopt.rs b/crates/turnloop/src/backend/sockopt.rs new file mode 100644 index 0000000..afc69e7 --- /dev/null +++ b/crates/turnloop/src/backend/sockopt.rs @@ -0,0 +1,406 @@ +//! Typed socket options on live Unix descriptors (DESIGN §7.1, §7.2, §7.7). +//! +//! Every entry point is synchronous, allocation-free and talks to the kernel on +//! the calling turn. Nothing is cached: a getter always asks the OS, so a value +//! the kernel rounded, clamped or doubled is reported as the kernel holds it. +//! +//! An option the platform lacks is reported, never ignored. Where the wrong +//! level would be used for a socket (`TCP_NODELAY` on UDP, an IP option on a +//! Unix-domain socket) the kernel's own `ENOPROTOOPT`/`EOPNOTSUPP` is returned +//! unchanged rather than being translated into a guess. +use super::poller::last_error; +use crate::{ + AcceptDefaults, Error, ErrorKind, KeepAlive, MulticastGroup, Result, SocketOption, + SocketOptionKind, +}; +use std::{mem::size_of, net::IpAddr, os::fd::RawFd, time::Duration}; + +/// Idle time before the first keep-alive probe. Apple spells it `TCP_KEEPALIVE`. +#[cfg(target_vendor = "apple")] +const KEEPIDLE: i32 = libc::TCP_KEEPALIVE; +#[cfg(not(target_vendor = "apple"))] +const KEEPIDLE: i32 = libc::TCP_KEEPIDLE; + +#[cfg(any(target_os = "linux", target_os = "android"))] +const JOIN_V6: i32 = libc::IPV6_ADD_MEMBERSHIP; +#[cfg(any(target_os = "linux", target_os = "android"))] +const LEAVE_V6: i32 = libc::IPV6_DROP_MEMBERSHIP; +#[cfg(not(any(target_os = "linux", target_os = "android")))] +const JOIN_V6: i32 = libc::IPV6_JOIN_GROUP; +#[cfg(not(any(target_os = "linux", target_os = "android")))] +const LEAVE_V6: i32 = libc::IPV6_LEAVE_GROUP; + +fn invalid() -> Error { + Error::new(ErrorKind::InvalidInput) +} +fn unsupported() -> Error { + Error::new(ErrorKind::Unsupported) +} +fn check(code: i32) -> Result<()> { + if code < 0 { Err(last_error()) } else { Ok(()) } +} +fn set_int(fd: RawFd, level: i32, name: i32, value: i32) -> Result<()> { + // SAFETY: value is an initialized integer of exactly the length passed. + check(unsafe { + libc::setsockopt( + fd, + level, + name, + std::ptr::from_ref(&value).cast(), + size_of::() as libc::socklen_t, + ) + }) +} +fn get_int(fd: RawFd, level: i32, name: i32) -> Result { + let mut value = 0i32; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: writable initialized integer output with its exact capacity. + check(unsafe { + libc::getsockopt( + fd, + level, + name, + std::ptr::from_mut(&mut value).cast(), + &mut len, + ) + })?; + Ok(value) +} +/// IPv4 multicast TTL/loop take a `u_char` on BSD and an `int` on Linux; both +/// widths are written and read explicitly instead of relying on byte order. +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn set_v4_multicast(fd: RawFd, name: i32, value: u32) -> Result<()> { + let value = u8::try_from(value).map_err(|_| invalid())?; + // SAFETY: value is an initialized byte of exactly the length passed. + check(unsafe { + libc::setsockopt( + fd, + libc::IPPROTO_IP, + name, + std::ptr::from_ref(&value).cast(), + size_of::() as libc::socklen_t, + ) + }) +} +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn get_v4_multicast(fd: RawFd, name: i32) -> Result { + let mut value = 0u8; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: writable initialized byte output with its exact capacity. + check(unsafe { + libc::getsockopt( + fd, + libc::IPPROTO_IP, + name, + std::ptr::from_mut(&mut value).cast(), + &mut len, + ) + })?; + Ok(u32::from(value)) +} +#[cfg(any(target_os = "linux", target_os = "android"))] +fn set_v4_multicast(fd: RawFd, name: i32, value: u32) -> Result<()> { + set_int(fd, libc::IPPROTO_IP, name, bounded(value)?) +} +#[cfg(any(target_os = "linux", target_os = "android"))] +fn get_v4_multicast(fd: RawFd, name: i32) -> Result { + Ok(get_int(fd, libc::IPPROTO_IP, name)?.max(0) as u32) +} +fn bounded(value: u32) -> Result { + i32::try_from(value).map_err(|_| invalid()) +} +/// One-second granularity, rounded up: a caller asking for 1.2 s must not get 1 s. +fn whole_seconds(value: Duration) -> Result { + let seconds = value + .as_secs() + .checked_add(u64::from(value.subsec_nanos() != 0)) + .ok_or_else(invalid)?; + i32::try_from(seconds).map_err(|_| invalid()) +} +/// True for an IPv6 socket, false for IPv4; anything else has no IP options. +fn ipv6(fd: RawFd) -> Result { + // SAFETY: sockaddr_storage is plain C integer/padding storage; zero is valid. + let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: live descriptor with writable address storage and its capacity. + check(unsafe { libc::getsockname(fd, std::ptr::from_mut(&mut storage).cast(), &mut len) })?; + match i32::from(storage.ss_family) { + libc::AF_INET => Ok(false), + libc::AF_INET6 => Ok(true), + _ => Err(unsupported()), + } +} +fn ip_level(fd: RawFd, v4: i32, v6: i32) -> Result<(i32, i32)> { + if ipv6(fd)? { + Ok((libc::IPPROTO_IPV6, v6)) + } else { + Ok((libc::IPPROTO_IP, v4)) + } +} +fn set_linger(fd: RawFd, value: Option) -> Result<()> { + let value = libc::linger { + l_onoff: i32::from(value.is_some()), + l_linger: value.map_or(Ok(0), whole_seconds)?, + }; + // SAFETY: a fully initialized `linger` of exactly the length passed. + check(unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_LINGER, + std::ptr::from_ref(&value).cast(), + size_of::() as libc::socklen_t, + ) + }) +} +fn get_linger(fd: RawFd) -> Result> { + // SAFETY: `linger` is two plain C integers; zero is a valid initial value. + let mut value: libc::linger = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: writable `linger` output with its exact capacity. + check(unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_LINGER, + std::ptr::from_mut(&mut value).cast(), + &mut len, + ) + })?; + Ok((value.l_onoff != 0).then(|| Duration::from_secs(value.l_linger.max(0) as u64))) +} +/// Keep-alive is `SO_KEEPALIVE` plus up to three TCP-level schedule values. +/// Disabling clears the master switch and leaves the schedule untouched, exactly +/// as the OS does; the schedule is only written when probing is enabled. +fn set_keep_alive(fd: RawFd, value: Option) -> Result<()> { + let Some(schedule) = value else { + return set_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE, 0); + }; + let idle = schedule.idle.map(positive_seconds).transpose()?; + let interval = schedule.interval.map(positive_seconds).transpose()?; + let count = schedule.count.map(bounded).transpose()?; + if count == Some(0) { + return Err(invalid()); + } + set_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE, 1)?; + if let Some(idle) = idle { + set_int(fd, libc::IPPROTO_TCP, KEEPIDLE, idle)?; + } + if let Some(interval) = interval { + set_int(fd, libc::IPPROTO_TCP, libc::TCP_KEEPINTVL, interval)?; + } + if let Some(count) = count { + set_int(fd, libc::IPPROTO_TCP, libc::TCP_KEEPCNT, count)?; + } + Ok(()) +} +fn positive_seconds(value: Duration) -> Result { + match whole_seconds(value)? { + 0 => Err(invalid()), + seconds => Ok(seconds), + } +} +fn get_keep_alive(fd: RawFd) -> Result> { + if get_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE)? == 0 { + return Ok(None); + } + Ok(Some(KeepAlive { + idle: Some(Duration::from_secs( + get_int(fd, libc::IPPROTO_TCP, KEEPIDLE)?.max(0) as u64, + )), + interval: Some(Duration::from_secs( + get_int(fd, libc::IPPROTO_TCP, libc::TCP_KEEPINTVL)?.max(0) as u64, + )), + count: Some(get_int(fd, libc::IPPROTO_TCP, libc::TCP_KEEPCNT)?.max(0) as u32), + })) +} +fn membership(fd: RawFd, group: MulticastGroup, join: bool) -> Result<()> { + match (group.group, ipv6(fd)?) { + (IpAddr::V6(address), true) => { + // SAFETY: `ipv6_mreq` is a plain address plus an index; zero is valid. + let mut request: libc::ipv6_mreq = unsafe { std::mem::zeroed() }; + request.ipv6mr_multiaddr.s6_addr = address.octets(); + request.ipv6mr_interface = group.interface as _; + let name = if join { JOIN_V6 } else { LEAVE_V6 }; + // SAFETY: a fully initialized request of exactly the length passed. + check(unsafe { + libc::setsockopt( + fd, + libc::IPPROTO_IPV6, + name, + std::ptr::from_ref(&request).cast(), + size_of::() as libc::socklen_t, + ) + }) + } + (IpAddr::V4(address), false) => { + let name = if join { + libc::IP_ADD_MEMBERSHIP + } else { + libc::IP_DROP_MEMBERSHIP + }; + if group.interface == 0 { + // SAFETY: `ip_mreq` is two plain addresses; zero is valid. + let mut request: libc::ip_mreq = unsafe { std::mem::zeroed() }; + request.imr_multiaddr.s_addr = u32::from_ne_bytes(address.octets()); + request.imr_interface.s_addr = libc::INADDR_ANY.to_be(); + // SAFETY: a fully initialized request of exactly the length passed. + return check(unsafe { + libc::setsockopt( + fd, + libc::IPPROTO_IP, + name, + std::ptr::from_ref(&request).cast(), + size_of::() as libc::socklen_t, + ) + }); + } + interface_membership(fd, address, group.interface, name) + } + _ => Err(invalid()), + } +} +/// Only Linux accepts an IPv4 membership keyed by interface index (`ip_mreqn`). +/// Elsewhere the request is reported, not silently applied to the default route. +#[cfg(any(target_os = "linux", target_os = "android"))] +fn interface_membership( + fd: RawFd, + address: std::net::Ipv4Addr, + interface: u32, + name: i32, +) -> Result<()> { + // SAFETY: `ip_mreqn` is two plain addresses and an index; zero is valid. + let mut request: libc::ip_mreqn = unsafe { std::mem::zeroed() }; + request.imr_multiaddr.s_addr = u32::from_ne_bytes(address.octets()); + request.imr_ifindex = bounded(interface)?; + // SAFETY: a fully initialized request of exactly the length passed. + check(unsafe { + libc::setsockopt( + fd, + libc::IPPROTO_IP, + name, + std::ptr::from_ref(&request).cast(), + size_of::() as libc::socklen_t, + ) + }) +} +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn interface_membership( + _fd: RawFd, + _address: std::net::Ipv4Addr, + _interface: u32, + _name: i32, +) -> Result<()> { + Err(unsupported()) +} + +/// Apply one option to a live descriptor. +pub(super) fn set(fd: RawFd, option: SocketOption) -> Result<()> { + match option { + SocketOption::NoDelay(on) => { + set_int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY, i32::from(on)) + } + SocketOption::KeepAlive(schedule) => set_keep_alive(fd, schedule), + SocketOption::Linger(value) => set_linger(fd, value), + SocketOption::RecvBufferSize(bytes) => { + set_int(fd, libc::SOL_SOCKET, libc::SO_RCVBUF, bounded(bytes)?) + } + SocketOption::SendBufferSize(bytes) => { + set_int(fd, libc::SOL_SOCKET, libc::SO_SNDBUF, bounded(bytes)?) + } + SocketOption::Ttl(hops) => { + let (level, name) = ip_level(fd, libc::IP_TTL, libc::IPV6_UNICAST_HOPS)?; + set_int(fd, level, name, bounded(hops)?) + } + SocketOption::Ipv6Only(on) => { + set_int(fd, libc::IPPROTO_IPV6, libc::IPV6_V6ONLY, i32::from(on)) + } + SocketOption::Broadcast(on) => { + set_int(fd, libc::SOL_SOCKET, libc::SO_BROADCAST, i32::from(on)) + } + SocketOption::MulticastTtl(hops) => { + if ipv6(fd)? { + set_int( + fd, + libc::IPPROTO_IPV6, + libc::IPV6_MULTICAST_HOPS, + bounded(hops)?, + ) + } else { + set_v4_multicast(fd, libc::IP_MULTICAST_TTL, hops) + } + } + SocketOption::MulticastLoop(on) => { + if ipv6(fd)? { + set_int( + fd, + libc::IPPROTO_IPV6, + libc::IPV6_MULTICAST_LOOP, + i32::from(on), + ) + } else { + set_v4_multicast(fd, libc::IP_MULTICAST_LOOP, u32::from(on)) + } + } + SocketOption::MulticastJoin(group) => membership(fd, group, true), + SocketOption::MulticastLeave(group) => membership(fd, group, false), + } +} +/// Read one option back from the kernel. +pub(super) fn get(fd: RawFd, kind: SocketOptionKind) -> Result { + Ok(match kind { + SocketOptionKind::NoDelay => { + SocketOption::NoDelay(get_int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY)? != 0) + } + SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(fd)?), + SocketOptionKind::Linger => SocketOption::Linger(get_linger(fd)?), + SocketOptionKind::RecvBufferSize => SocketOption::RecvBufferSize( + get_int(fd, libc::SOL_SOCKET, libc::SO_RCVBUF)?.max(0) as u32, + ), + SocketOptionKind::SendBufferSize => SocketOption::SendBufferSize( + get_int(fd, libc::SOL_SOCKET, libc::SO_SNDBUF)?.max(0) as u32, + ), + SocketOptionKind::Ttl => { + let (level, name) = ip_level(fd, libc::IP_TTL, libc::IPV6_UNICAST_HOPS)?; + SocketOption::Ttl(get_int(fd, level, name)?.max(0) as u32) + } + SocketOptionKind::Ipv6Only => { + SocketOption::Ipv6Only(get_int(fd, libc::IPPROTO_IPV6, libc::IPV6_V6ONLY)? != 0) + } + SocketOptionKind::Broadcast => { + SocketOption::Broadcast(get_int(fd, libc::SOL_SOCKET, libc::SO_BROADCAST)? != 0) + } + SocketOptionKind::MulticastTtl => SocketOption::MulticastTtl(if ipv6(fd)? { + get_int(fd, libc::IPPROTO_IPV6, libc::IPV6_MULTICAST_HOPS)?.max(0) as u32 + } else { + get_v4_multicast(fd, libc::IP_MULTICAST_TTL)? + }), + SocketOptionKind::MulticastLoop => SocketOption::MulticastLoop(if ipv6(fd)? { + get_int(fd, libc::IPPROTO_IPV6, libc::IPV6_MULTICAST_LOOP)? != 0 + } else { + get_v4_multicast(fd, libc::IP_MULTICAST_LOOP)? != 0 + }), + }) +} +/// Apply a listener's accepted-socket defaults to one freshly accepted socket. +/// Called before the `Accepted` outcome exists, so a failure fails that accept +/// and closes the connection instead of handing up a half-configured socket. +pub(super) fn apply_accept_defaults(fd: RawFd, defaults: AcceptDefaults) -> Result<()> { + if defaults.nodelay { + set(fd, SocketOption::NoDelay(true))?; + } + if let Some(schedule) = defaults.keep_alive { + set(fd, SocketOption::KeepAlive(Some(schedule)))?; + } + Ok(()) +} +/// Reject at listener creation what this backend could not apply per connection. +/// Both native defaults are TCP options, so only a local (AF_UNIX) listener has +/// to refuse them. +pub(super) fn validate_accept_defaults(defaults: AcceptDefaults, tcp: bool) -> Result<()> { + if tcp || defaults.is_empty() { + Ok(()) + } else { + Err(unsupported()) + } +} diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index 67c2c0d..12e0d12 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -36,6 +36,10 @@ pub struct Detached { pub(super) kind: Kind, pub(super) original_flags: Option, original_mode: Option, + /// A listener's per-connection defaults, applied to each socket it accepts. + /// It travels with the listener across detach/attach, so a listener handed to + /// another loop keeps configuring its connections there. + accept_defaults: AcceptDefaults, } impl std::fmt::Debug for Detached { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -57,6 +61,7 @@ impl Detached { kind, original_flags: None, original_mode, + accept_defaults: AcceptDefaults::EMPTY, } } /// Adopt an owned Unix descriptor, classifying stream/file/TTY or socket. @@ -128,6 +133,18 @@ impl Unix { .filter(|r| r.handle == h) .ok_or(Error::new(ErrorKind::NotFound)) } + /// The descriptor behind a socket handle. Process, signal and watch handles + /// and adopted files/terminals are not sockets and say so. + fn socket_fd(&self, h: Handle) -> Result { + if self.services.contains(h) || self.watches.contains(h) { + return Err(Error::new(ErrorKind::Unsupported)); + } + let r = self.get(h)?; + if matches!(r.transport.kind, Kind::File | Kind::Stream) { + return Err(Error::new(ErrorKind::Unsupported)); + } + Ok(r.transport.fd.as_raw_fd()) + } fn install(&mut self, h: Handle, transport: Detached, connect: Option) -> Result<()> { if self.resources.get(h.index()).is_none_or(Option::is_some) { return Err(Error::new(ErrorKind::InvalidInput)); @@ -432,6 +449,7 @@ unsafe impl Backend for Unix { return self.install(h, transport, Some(addr)); } Open::PipeListener { name, opts } => { + super::sockopt::validate_accept_defaults(opts.accept_defaults, false)?; let (transport, _) = super::ipc::open(&name, Some(opts))?; return self.install(h, transport, None); } @@ -446,14 +464,34 @@ unsafe impl Backend for Unix { } other => other, }; - let (addr, kind, reuse, backlog, nodelay) = match spec { - Open::Tcp { addr, opts } => (addr, Kind::Tcp, false, 0, opts.nodelay), - Open::Listener { addr, opts } => { - (addr, Kind::Listener, opts.reuse_port, opts.backlog, false) - } - Open::Udp { addr, opts } => (addr, Kind::Udp, opts.reuse_port, 0, false), + let (addr, kind, reuse, backlog, nodelay, accept_defaults) = match spec { + Open::Tcp { addr, opts } => ( + addr, + Kind::Tcp, + false, + 0, + opts.nodelay, + AcceptDefaults::EMPTY, + ), + Open::Listener { addr, opts } => ( + addr, + Kind::Listener, + opts.reuse_port, + opts.backlog, + false, + opts.accept_defaults, + ), + Open::Udp { addr, opts } => ( + addr, + Kind::Udp, + opts.reuse_port, + 0, + false, + AcceptDefaults::EMPTY, + ), _ => unreachable!("native open handled above"), }; + super::sockopt::validate_accept_defaults(accept_defaults, kind == Kind::Listener)?; if backlog > i32::MAX as u32 { return Err(Error::new(ErrorKind::InvalidInput)); } @@ -483,15 +521,19 @@ unsafe impl Backend for Unix { if nodelay { socket::option(fd.as_raw_fd(), libc::IPPROTO_TCP, libc::TCP_NODELAY, 1)?; } - self.install( - h, - Detached::new(fd, kind), - (kind == Kind::Tcp).then(|| Addr::new(addr)), - ) + let mut transport = Detached::new(fd, kind); + transport.accept_defaults = accept_defaults; + self.install(h, transport, (kind == Kind::Tcp).then(|| Addr::new(addr))) } fn local_addr(&self, h: Handle) -> Result { socket::local_addr(self.get(h)?.transport.fd.as_raw_fd()) } + fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + super::sockopt::set(self.socket_fd(h)?, option) + } + fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { + super::sockopt::get(self.socket_fd(h)?, kind) + } fn submit(&mut self, request: Request) -> Result<()> { let h = request.handle; if self.services.contains(h) { @@ -757,6 +799,10 @@ fn execute( ))); } let (fd, peer) = socket::accept(fd)?; + // Before the connection becomes visible to the host. A rejected + // default fails this accept and drops the socket; it is never + // reported as an accepted-but-unconfigured connection. + super::sockopt::apply_accept_defaults(fd.as_raw_fd(), r.transport.accept_defaults)?; Ok(Some(( Outcome::Accepted { transport: Detached::new(fd, Kind::Tcp), diff --git a/crates/turnloop/src/backend/wasi_p2.rs b/crates/turnloop/src/backend/wasi_p2.rs index 4a6b3a5..0d8917e 100644 --- a/crates/turnloop/src/backend/wasi_p2.rs +++ b/crates/turnloop/src/backend/wasi_p2.rs @@ -2,6 +2,7 @@ //! lists, and synchronous nonblocking I/O with generational cancellation. mod abi; mod fs; +mod sockopt; use crate::{ backend::{Backend, Event, Filesystem, Operation, Outcome, PollInfo, Request, Wake}, *, @@ -69,6 +70,8 @@ pub struct Detached { poll: Option, socket: Socket, kind: Kind, + /// A listener's per-connection defaults, applied to each socket it accepts. + accept_defaults: AcceptDefaults, } struct Resource { handle: Handle, @@ -414,7 +417,7 @@ unsafe impl Backend for WasiP2 { self.wake.clone() } fn open(&mut self, h: Handle, spec: Open) -> Result<()> { - let (addr, kind, reuse, backlog) = match spec { + let (addr, kind, reuse, backlog, accept_defaults) = match spec { Open::Pipe(_) | Open::PipeListener { .. } => { return Err(Error::new(ErrorKind::Unsupported)); } @@ -438,17 +441,35 @@ unsafe impl Backend for WasiP2 { poll: None, socket: Socket::Stdio, kind: Kind::Stdio(which), + accept_defaults: AcceptDefaults::EMPTY, }, None, ); } - Open::Tcp { addr, .. } => (addr, Kind::Tcp, false, 0), - Open::Listener { addr, opts } => (addr, Kind::Listener, opts.reuse_port, opts.backlog), - Open::Udp { addr, opts } => (addr, Kind::Udp, opts.reuse_port, 0), + Open::Tcp { addr, opts } => { + // wasi:sockets has no Nagle control. Reporting it is the only + // honest answer; a dropped hint is indistinguishable from an + // applied one (see Backend::set_option). + if opts.nodelay { + return Err(Error::new(ErrorKind::Unsupported)); + } + (addr, Kind::Tcp, false, 0, AcceptDefaults::EMPTY) + } + Open::Listener { addr, opts } => ( + addr, + Kind::Listener, + opts.reuse_port, + opts.backlog, + opts.accept_defaults, + ), + Open::Udp { addr, opts } => { + (addr, Kind::Udp, opts.reuse_port, 0, AcceptDefaults::EMPTY) + } }; if reuse { return Err(Error::new(ErrorKind::Unsupported)); } + sockopt::validate_accept_defaults(accept_defaults)?; let family = if addr.is_ipv4() { IpAddressFamily::Ipv4 } else { @@ -475,6 +496,7 @@ unsafe impl Backend for WasiP2 { datagrams: Some(datagrams), socket: Socket::Udp(socket), kind, + accept_defaults, } } else { let socket = create_tcp_socket(family).map_err(error)?; @@ -495,12 +517,20 @@ unsafe impl Backend for WasiP2 { datagrams: None, socket: Socket::Tcp(socket), kind, + accept_defaults, } }; // Keep all child fields established before installation. transport.kind = kind; self.install(h, transport, (kind == Kind::Tcp).then_some(addr)) } + fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + let socket = &self.get(h)?.transport.socket; + sockopt::set(socket, option) + } + fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { + sockopt::get(&self.get(h)?.transport.socket, kind) + } fn local_addr(&self, h: Handle) -> Result { let a = match &self.get(h)?.transport.socket { Socket::Tcp(s) => s.local_address(), @@ -757,12 +787,16 @@ fn execute( }; let (socket, input, output) = socket.accept().map_err(error)?; let peer = native(socket.remote_address().map_err(error)?); + // Before the connection becomes visible to the host. A rejected + // default fails this accept and drops the socket. + sockopt::apply_accept_defaults(&socket, t.accept_defaults)?; let transport = Detached { streams: Some(streams(input, output)), datagrams: None, poll: Some(socket.subscribe()), socket: Socket::Tcp(socket), kind: Kind::Tcp, + accept_defaults: AcceptDefaults::EMPTY, }; Ok(Some((Outcome::Accepted { transport, peer }, !*multishot))) } diff --git a/crates/turnloop/src/backend/wasi_p2/sockopt.rs b/crates/turnloop/src/backend/wasi_p2/sockopt.rs new file mode 100644 index 0000000..ddc8308 --- /dev/null +++ b/crates/turnloop/src/backend/wasi_p2/sockopt.rs @@ -0,0 +1,171 @@ +//! Typed socket options over `wasi:sockets` 0.2 (DESIGN §7.4, §7.7). +//! +//! WASI exposes a deliberately small option surface. Everything outside it — +//! Nagle, linger, IPv6-only, broadcast and multicast membership — has no +//! interface in `wasi:sockets@0.2.9`, so it is reported `Unsupported` rather than +//! accepted and dropped on the floor. Nothing is cached: each getter is an +//! import call, so a value the host clamped is the value reported back. +use super::{Socket, error}; +use crate::{AcceptDefaults, Error, ErrorKind, KeepAlive, Result, SocketOption, SocketOptionKind}; +use std::time::Duration; +use wasip2::sockets::tcp::TcpSocket; + +fn unsupported() -> Result { + Err(Error::new(ErrorKind::Unsupported)) +} +fn invalid() -> Result { + Err(Error::new(ErrorKind::InvalidInput)) +} +fn hops(value: u32) -> Result { + match u8::try_from(value) { + Ok(0) | Err(_) => invalid(), + Ok(hops) => Ok(hops), + } +} +fn tcp(socket: &Socket) -> Result<&TcpSocket> { + match socket { + Socket::Tcp(socket) => Ok(socket), + _ => unsupported(), + } +} +/// WASI takes the schedule as a duration, so no rounding is needed; a zero +/// duration is still rejected, matching the native backends. +fn nonzero(value: Duration) -> Result { + match u64::try_from(value.as_nanos()) { + Ok(0) | Err(_) => invalid(), + Ok(nanos) => Ok(nanos), + } +} +fn set_keep_alive(socket: &TcpSocket, value: Option) -> Result<()> { + let Some(schedule) = value else { + return socket.set_keep_alive_enabled(false).map_err(error); + }; + let idle = schedule.idle.map(nonzero).transpose()?; + let interval = schedule.interval.map(nonzero).transpose()?; + if schedule.count == Some(0) { + return invalid(); + } + socket.set_keep_alive_enabled(true).map_err(error)?; + if let Some(idle) = idle { + socket.set_keep_alive_idle_time(idle).map_err(error)?; + } + if let Some(interval) = interval { + socket.set_keep_alive_interval(interval).map_err(error)?; + } + if let Some(count) = schedule.count { + socket.set_keep_alive_count(count).map_err(error)?; + } + Ok(()) +} +fn get_keep_alive(socket: &TcpSocket) -> Result> { + if !socket.keep_alive_enabled().map_err(error)? { + return Ok(None); + } + Ok(Some(KeepAlive { + idle: Some(Duration::from_nanos( + socket.keep_alive_idle_time().map_err(error)?, + )), + interval: Some(Duration::from_nanos( + socket.keep_alive_interval().map_err(error)?, + )), + count: Some(socket.keep_alive_count().map_err(error)?), + })) +} +fn bytes(value: u64) -> u32 { + u32::try_from(value).unwrap_or(u32::MAX) +} + +/// Apply one option to a live WASI socket. +pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { + match option { + SocketOption::KeepAlive(schedule) => set_keep_alive(tcp(socket)?, schedule), + SocketOption::RecvBufferSize(size) => match socket { + Socket::Tcp(socket) => socket.set_receive_buffer_size(u64::from(size)), + Socket::Udp(socket) => socket.set_receive_buffer_size(u64::from(size)), + Socket::Stdio => return unsupported(), + } + .map_err(error), + SocketOption::SendBufferSize(size) => match socket { + Socket::Tcp(socket) => socket.set_send_buffer_size(u64::from(size)), + Socket::Udp(socket) => socket.set_send_buffer_size(u64::from(size)), + Socket::Stdio => return unsupported(), + } + .map_err(error), + SocketOption::Ttl(value) => { + let value = hops(value)?; + match socket { + Socket::Tcp(socket) => socket.set_hop_limit(value), + Socket::Udp(socket) => socket.set_unicast_hop_limit(value), + Socket::Stdio => return unsupported(), + } + .map_err(error) + } + // No interface in wasi:sockets@0.2.9. + SocketOption::NoDelay(_) + | SocketOption::Linger(_) + | SocketOption::Ipv6Only(_) + | SocketOption::Broadcast(_) + | SocketOption::MulticastTtl(_) + | SocketOption::MulticastLoop(_) + | SocketOption::MulticastJoin(_) + | SocketOption::MulticastLeave(_) => unsupported(), + } +} +/// Read one option back through `wasi:sockets`. +pub(super) fn get(socket: &Socket, kind: SocketOptionKind) -> Result { + Ok(match kind { + SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(tcp(socket)?)?), + SocketOptionKind::RecvBufferSize => SocketOption::RecvBufferSize(bytes( + match socket { + Socket::Tcp(socket) => socket.receive_buffer_size(), + Socket::Udp(socket) => socket.receive_buffer_size(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::SendBufferSize => SocketOption::SendBufferSize(bytes( + match socket { + Socket::Tcp(socket) => socket.send_buffer_size(), + Socket::Udp(socket) => socket.send_buffer_size(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::Ttl => SocketOption::Ttl(u32::from( + match socket { + Socket::Tcp(socket) => socket.hop_limit(), + Socket::Udp(socket) => socket.unicast_hop_limit(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::NoDelay + | SocketOptionKind::Linger + | SocketOptionKind::Ipv6Only + | SocketOptionKind::Broadcast + | SocketOptionKind::MulticastTtl + | SocketOptionKind::MulticastLoop => return unsupported(), + }) +} +/// Apply a listener's accepted-socket defaults to one accepted socket, before +/// the `Accepted` outcome exists. +pub(super) fn apply_accept_defaults(socket: &TcpSocket, defaults: AcceptDefaults) -> Result<()> { + // `open` already refused this, but an accept must never quietly hand up a + // connection whose requested default was not applied. + if defaults.nodelay { + return unsupported(); + } + if let Some(schedule) = defaults.keep_alive { + set_keep_alive(socket, Some(schedule))?; + } + Ok(()) +} +/// Reject at listener creation what this backend could not apply per connection. +/// `wasi:sockets` has no Nagle control, so a `nodelay` default is refused here +/// instead of being ignored once per accepted connection. +pub(super) fn validate_accept_defaults(defaults: AcceptDefaults) -> Result<()> { + if defaults.nodelay { + return unsupported(); + } + Ok(()) +} diff --git a/crates/turnloop/src/backend/wasi_p3.rs b/crates/turnloop/src/backend/wasi_p3.rs index 5e24d98..297c822 100644 --- a/crates/turnloop/src/backend/wasi_p3.rs +++ b/crates/turnloop/src/backend/wasi_p3.rs @@ -5,6 +5,7 @@ mod abi; mod fs; mod return_storage; +mod sockopt; mod wait_set; use crate::{ backend::{Backend, Event, Filesystem, Operation, Outcome, PollInfo, Request, Wake}, @@ -85,6 +86,8 @@ pub struct Detached { incoming: Option>, socket: Socket, kind: Kind, + /// A listener's per-connection defaults, applied to each socket it accepts. + accept_defaults: AcceptDefaults, } struct Resource { _udp_return: Option, @@ -368,7 +371,7 @@ unsafe impl Backend for WasiP3 { self.wake.clone() } fn open(&mut self, h: Handle, spec: Open) -> Result<()> { - let (addr, kind, reuse, backlog) = match spec { + let (addr, kind, reuse, backlog, accept_defaults) = match spec { Open::Pipe(_) | Open::PipeListener { .. } => { return Err(Error::new(ErrorKind::Unsupported)); } @@ -410,17 +413,35 @@ unsafe impl Backend for WasiP3 { incoming: None, socket: Socket::Stdio, kind: Kind::Stdio(which), + accept_defaults: AcceptDefaults::EMPTY, }, None, ); } - Open::Tcp { addr, .. } => (addr, Kind::Tcp, false, 0), - Open::Listener { addr, opts } => (addr, Kind::Listener, opts.reuse_port, opts.backlog), - Open::Udp { addr, opts } => (addr, Kind::Udp, opts.reuse_port, 0), + Open::Tcp { addr, opts } => { + // wasi:sockets has no Nagle control. Reporting it is the only + // honest answer; a dropped hint is indistinguishable from an + // applied one (see Backend::set_option). + if opts.nodelay { + return Err(Error::new(ErrorKind::Unsupported)); + } + (addr, Kind::Tcp, false, 0, AcceptDefaults::EMPTY) + } + Open::Listener { addr, opts } => ( + addr, + Kind::Listener, + opts.reuse_port, + opts.backlog, + opts.accept_defaults, + ), + Open::Udp { addr, opts } => { + (addr, Kind::Udp, opts.reuse_port, 0, AcceptDefaults::EMPTY) + } }; if reuse { return Err(Error::new(ErrorKind::Unsupported)); } + sockopt::validate_accept_defaults(accept_defaults)?; let family = if addr.is_ipv4() { IpAddressFamily::Ipv4 } else { @@ -434,6 +455,7 @@ unsafe impl Backend for WasiP3 { incoming: None, socket: Socket::Udp(s), kind, + accept_defaults, } } else { let s = TcpSocket::create(family).map_err(error)?; @@ -450,10 +472,17 @@ unsafe impl Backend for WasiP3 { incoming, socket: Socket::Tcp(s), kind, + accept_defaults, } }; self.install(h, transport, (kind == Kind::Tcp).then_some(addr)) } + fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + sockopt::set(&self.get(h)?.transport.socket, option) + } + fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { + sockopt::get(&self.get(h)?.transport.socket, kind) + } fn local_addr(&self, h: Handle) -> Result { match &self.get(h)?.transport.socket { Socket::Tcp(s) => s.get_local_address(), @@ -767,11 +796,15 @@ fn execute( // SAFETY: one canonical owned socket handle was transferred into area[0]. let socket = unsafe { TcpSocket::from_handle(p.area[0]) }; let peer = native(socket.get_remote_address().map_err(error)?); + // Before the connection becomes visible to the host. A rejected + // default fails this accept and drops the socket. + sockopt::apply_accept_defaults(&socket, r.transport.accept_defaults)?; let transport = Detached { streams: Some(streams(&socket)), incoming: None, socket: Socket::Tcp(socket), kind: Kind::Tcp, + accept_defaults: AcceptDefaults::EMPTY, }; let multishot = matches!(p.request.operation, Operation::Accept { multishot: true }); diff --git a/crates/turnloop/src/backend/wasi_p3/sockopt.rs b/crates/turnloop/src/backend/wasi_p3/sockopt.rs new file mode 100644 index 0000000..b8a45f3 --- /dev/null +++ b/crates/turnloop/src/backend/wasi_p3/sockopt.rs @@ -0,0 +1,171 @@ +//! Typed socket options over `wasi:sockets` 0.3 (DESIGN §7.4, §7.7). +//! +//! WASI exposes a deliberately small option surface. Everything outside it — +//! Nagle, linger, IPv6-only, broadcast and multicast membership — has no +//! interface in `wasi:sockets` 0.3, so it is reported `Unsupported` rather than +//! accepted and dropped on the floor. Nothing is cached: each getter is an +//! import call, so a value the host clamped is the value reported back. +use super::{Socket, error}; +use crate::{AcceptDefaults, Error, ErrorKind, KeepAlive, Result, SocketOption, SocketOptionKind}; +use std::time::Duration; +use wasip3::sockets::types::TcpSocket; + +fn unsupported() -> Result { + Err(Error::new(ErrorKind::Unsupported)) +} +fn invalid() -> Result { + Err(Error::new(ErrorKind::InvalidInput)) +} +fn hops(value: u32) -> Result { + match u8::try_from(value) { + Ok(0) | Err(_) => invalid(), + Ok(hops) => Ok(hops), + } +} +fn tcp(socket: &Socket) -> Result<&TcpSocket> { + match socket { + Socket::Tcp(socket) => Ok(socket), + _ => unsupported(), + } +} +/// WASI takes the schedule as a duration, so no rounding is needed; a zero +/// duration is still rejected, matching the native backends. +fn nonzero(value: Duration) -> Result { + match u64::try_from(value.as_nanos()) { + Ok(0) | Err(_) => invalid(), + Ok(nanos) => Ok(nanos), + } +} +fn set_keep_alive(socket: &TcpSocket, value: Option) -> Result<()> { + let Some(schedule) = value else { + return socket.set_keep_alive_enabled(false).map_err(error); + }; + let idle = schedule.idle.map(nonzero).transpose()?; + let interval = schedule.interval.map(nonzero).transpose()?; + if schedule.count == Some(0) { + return invalid(); + } + socket.set_keep_alive_enabled(true).map_err(error)?; + if let Some(idle) = idle { + socket.set_keep_alive_idle_time(idle).map_err(error)?; + } + if let Some(interval) = interval { + socket.set_keep_alive_interval(interval).map_err(error)?; + } + if let Some(count) = schedule.count { + socket.set_keep_alive_count(count).map_err(error)?; + } + Ok(()) +} +fn get_keep_alive(socket: &TcpSocket) -> Result> { + if !socket.get_keep_alive_enabled().map_err(error)? { + return Ok(None); + } + Ok(Some(KeepAlive { + idle: Some(Duration::from_nanos( + socket.get_keep_alive_idle_time().map_err(error)?, + )), + interval: Some(Duration::from_nanos( + socket.get_keep_alive_interval().map_err(error)?, + )), + count: Some(socket.get_keep_alive_count().map_err(error)?), + })) +} +fn bytes(value: u64) -> u32 { + u32::try_from(value).unwrap_or(u32::MAX) +} + +/// Apply one option to a live WASI socket. +pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { + match option { + SocketOption::KeepAlive(schedule) => set_keep_alive(tcp(socket)?, schedule), + SocketOption::RecvBufferSize(size) => match socket { + Socket::Tcp(socket) => socket.set_receive_buffer_size(u64::from(size)), + Socket::Udp(socket) => socket.set_receive_buffer_size(u64::from(size)), + Socket::Stdio => return unsupported(), + } + .map_err(error), + SocketOption::SendBufferSize(size) => match socket { + Socket::Tcp(socket) => socket.set_send_buffer_size(u64::from(size)), + Socket::Udp(socket) => socket.set_send_buffer_size(u64::from(size)), + Socket::Stdio => return unsupported(), + } + .map_err(error), + SocketOption::Ttl(value) => { + let value = hops(value)?; + match socket { + Socket::Tcp(socket) => socket.set_hop_limit(value), + Socket::Udp(socket) => socket.set_unicast_hop_limit(value), + Socket::Stdio => return unsupported(), + } + .map_err(error) + } + // No interface in wasi:sockets@0.2.9. + SocketOption::NoDelay(_) + | SocketOption::Linger(_) + | SocketOption::Ipv6Only(_) + | SocketOption::Broadcast(_) + | SocketOption::MulticastTtl(_) + | SocketOption::MulticastLoop(_) + | SocketOption::MulticastJoin(_) + | SocketOption::MulticastLeave(_) => unsupported(), + } +} +/// Read one option back through `wasi:sockets`. +pub(super) fn get(socket: &Socket, kind: SocketOptionKind) -> Result { + Ok(match kind { + SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(tcp(socket)?)?), + SocketOptionKind::RecvBufferSize => SocketOption::RecvBufferSize(bytes( + match socket { + Socket::Tcp(socket) => socket.get_receive_buffer_size(), + Socket::Udp(socket) => socket.get_receive_buffer_size(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::SendBufferSize => SocketOption::SendBufferSize(bytes( + match socket { + Socket::Tcp(socket) => socket.get_send_buffer_size(), + Socket::Udp(socket) => socket.get_send_buffer_size(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::Ttl => SocketOption::Ttl(u32::from( + match socket { + Socket::Tcp(socket) => socket.get_hop_limit(), + Socket::Udp(socket) => socket.get_unicast_hop_limit(), + Socket::Stdio => return unsupported(), + } + .map_err(error)?, + )), + SocketOptionKind::NoDelay + | SocketOptionKind::Linger + | SocketOptionKind::Ipv6Only + | SocketOptionKind::Broadcast + | SocketOptionKind::MulticastTtl + | SocketOptionKind::MulticastLoop => return unsupported(), + }) +} +/// Apply a listener's accepted-socket defaults to one accepted socket, before +/// the `Accepted` outcome exists. +pub(super) fn apply_accept_defaults(socket: &TcpSocket, defaults: AcceptDefaults) -> Result<()> { + // `open` already refused this, but an accept must never quietly hand up a + // connection whose requested default was not applied. + if defaults.nodelay { + return unsupported(); + } + if let Some(schedule) = defaults.keep_alive { + set_keep_alive(socket, Some(schedule))?; + } + Ok(()) +} +/// Reject at listener creation what this backend could not apply per connection. +/// `wasi:sockets` has no Nagle control, so a `nodelay` default is refused here +/// instead of being ignored once per accepted connection. +pub(super) fn validate_accept_defaults(defaults: AcceptDefaults) -> Result<()> { + if defaults.nodelay { + return unsupported(); + } + Ok(()) +} diff --git a/crates/turnloop/src/backend/web.rs b/crates/turnloop/src/backend/web.rs index 560b28a..c8ac1b3 100644 --- a/crates/turnloop/src/backend/web.rs +++ b/crates/turnloop/src/backend/web.rs @@ -1,5 +1,11 @@ //! Browser/Node host callbacks. The owner calls only turn(Now). See //! docs/wasm.md for scheduling, cancellation, and the host allocation boundary. +//! +//! Socket options are `Unsupported` here, from the Backend trait's own defaults: +//! a browser host exposes `fetch` and `WebSocket`, not a socket, so there is no +//! `TCP_NODELAY`, keep-alive schedule, linger, buffer size or group membership to +//! set or read. `ListenOpts` never reaches this backend either, because listening +//! sockets are themselves unsupported (DESIGN §7.5). use crate::{ backend::{Backend, Event, Operation, Outcome, PollInfo, Request, Wake}, *, diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index 926a9d8..1b3edf4 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -663,6 +663,35 @@ impl Driver { self.resource(h)?; self.backend.local_addr(h) } + /// Apply a socket option to a live socket handle, including an accepted one. + /// + /// The option is applied immediately, inside this call, through the OS: no + /// operation is submitted, no completion is produced, nothing is queued and + /// nothing is allocated. A handle that is closing, or that is not a socket, + /// is `InvalidInput`; a platform with no equivalent for the option is + /// `Unsupported`. A backend never accepts an option it cannot apply. + /// + /// Bind-time-only options are not reachable here: `SO_REUSEPORT` and + /// `SO_REUSEADDR` belong to [`ListenOpts`]/[`UdpOpts`], and a listener's + /// per-connection defaults belong to [`AcceptDefaults`]. See [`SocketOption`]. + pub fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + let r = self.resource(h)?; + if r.closing.is_some() || !matches!(r.kind, Kind::Socket) { + return Err(Error::new(ErrorKind::InvalidInput)); + } + self.backend.set_option(h, option) + } + /// Read a socket option back from the OS. + /// + /// Always a fresh kernel query, never a cache of what was set, so a value the + /// OS rounded, clamped or doubled is visible as the OS holds it. + pub fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { + let r = self.resource(h)?; + if !matches!(r.kind, Kind::Socket) { + return Err(Error::new(ErrorKind::InvalidInput)); + } + self.backend.get_option(h, kind) + } fn submit(&mut self, h: Handle, operation: Operation, token: Token) -> Result { let r = self.resource(h)?; if r.closing.is_some() || !matches!(r.kind, Kind::Socket) { diff --git a/crates/turnloop/src/types.rs b/crates/turnloop/src/types.rs index 466e3d8..adfa2ae 100644 --- a/crates/turnloop/src/types.rs +++ b/crates/turnloop/src/types.rs @@ -1,6 +1,6 @@ //! Backend-neutral identifiers, errors, deadlines and socket options. use crate::Instant; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::time::Duration; #[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] @@ -177,29 +177,193 @@ impl Default for Config { } } #[derive(Clone, Copy, Debug, Default)] -/// TCP connection options applied when creating the socket. +/// TCP connection options applied when creating the socket. Anything a host +/// needs to change later goes through `Loop::set_option` and +/// [`SocketOption`] instead. pub struct TcpOpts { /// Disable the TCP Nagle algorithm for latency-sensitive small writes. pub nodelay: bool, } #[derive(Clone, Copy, Debug)] -/// Listener backlog and optional kernel reuse-port configuration. +/// Listener backlog, kernel reuse-port configuration and accepted-socket defaults. +/// +/// Address reuse is bind-time only and stays here rather than in +/// [`SocketOption`]: `SO_REUSEPORT` is `reuse_port`, and `SO_REUSEADDR` is applied +/// by the backend to every TCP listener it binds (TIME_WAIT rebinding). Neither +/// can be changed on a socket that is already bound, so neither is an option. pub struct ListenOpts { /// Enable SO_REUSEPORT when supported; macOS does not promise balanced accepts. pub reuse_port: bool, /// Maximum pending connection backlog requested from the OS. pub backlog: u32, + /// Options applied to every connection this listener accepts. + pub accept_defaults: AcceptDefaults, } impl Default for ListenOpts { fn default() -> Self { Self { reuse_port: false, backlog: 128, + accept_defaults: AcceptDefaults::EMPTY, } } } +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +/// Options a listener applies to every connection it accepts. +/// +/// **When they are applied:** by the accepting loop, on the accepted socket, +/// after the OS accept succeeds and *before* the `Accepted` completion reaches +/// the host. A host therefore never observes an unconfigured connection, and +/// never needs a `set_option` round trip per connection. +/// +/// **What "cheaply" means:** each field costs at most one `setsockopt` on the new +/// socket. Anything that would need a syscall per accepted byte, a kernel query +/// or a second handle is not a default and belongs in [`Loop::set_option`]. +/// +/// A backend that cannot apply a requested default rejects the *listener* when it +/// is created, rather than ignoring the request once per connection. If the OS +/// rejects a default while applying it to a live accepted socket, that accept +/// operation fails with the OS error and the connection is closed. +/// +/// [`Loop::set_option`]: crate::Driver::set_option +pub struct AcceptDefaults { + /// Disable Nagle coalescing on each accepted connection (`TCP_NODELAY`). + pub nodelay: bool, + /// Enable keep-alive probes on each accepted connection, with this schedule. + pub keep_alive: Option, +} +impl AcceptDefaults { + /// Apply nothing: every accepted socket keeps the platform defaults. + pub const EMPTY: Self = Self { + nodelay: false, + keep_alive: None, + }; + /// Whether any default would need to be applied to an accepted socket. + pub const fn is_empty(self) -> bool { + !self.nodelay && self.keep_alive.is_none() + } +} +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +/// TCP keep-alive probe schedule. Each `None` keeps the platform default. +/// +/// Native platforms express the schedule in whole seconds, so a duration is +/// rounded **up** to the next whole second and a zero duration is rejected as +/// `InvalidInput`. WASI takes the duration unrounded. +pub struct KeepAlive { + /// Connection idle time before the first probe is sent. + pub idle: Option, + /// Interval between probes once probing has started. + pub interval: Option, + /// Unanswered probes before the connection is dropped. + pub count: Option, +} +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +/// A multicast group and the local interface that should carry it. +pub struct MulticastGroup { + /// Group address. Its family must match the socket's own family. + pub group: IpAddr, + /// Local interface index, or zero for the kernel's default interface. + /// + /// An IPv6 group accepts any index on every native platform. An IPv4 group + /// accepts a nonzero index only on Linux (`ip_mreqn`); elsewhere a nonzero + /// index is reported `Unsupported` rather than silently ignored. + pub interface: u32, +} +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +/// A socket option that can be changed after the socket exists. +/// +/// Bind-time-only options are deliberately absent: `SO_REUSEPORT`/`SO_REUSEADDR` +/// live in [`ListenOpts`] and [`UdpOpts`], because the OS accepts them only on an +/// unbound socket. [`Ipv6Only`](Self::Ipv6Only) is the borderline case: it is +/// kept here so it can be *read* on any socket, but every platform rejects +/// setting it after bind, and turnloop binds listeners and UDP sockets when they +/// are created. +/// +/// A backend with no equivalent for a variant reports `Unsupported`. It never +/// accepts the call and ignores it. +pub enum SocketOption { + /// `TCP_NODELAY`: send small writes immediately instead of coalescing them. + NoDelay(bool), + /// `SO_KEEPALIVE` and its probe schedule; `None` disables probing. + /// + /// The switch and each schedule value are separate kernel settings, and every + /// value is validated before any of them is written. If the OS still rejects + /// one after the switch was set, the error is reported and the socket keeps + /// whatever the OS left: read it back rather than assuming a rollback. + /// Disabling clears the switch and leaves the schedule alone, as the OS does. + KeepAlive(Option), + /// `SO_LINGER`: `Some(d)` blocks the close until queued data is delivered or + /// `d` elapses, and `Some(Duration::ZERO)` discards it and resets the + /// connection. `None` restores the platform default (a graceful background + /// close). The duration has one-second granularity, rounded up. + Linger(Option), + /// `SO_RCVBUF`: requested receive-buffer bytes. + /// + /// **The final size is the kernel's choice, not the request.** The contract + /// is only that the socket ends up with *at least* what was asked for, so + /// read it back rather than assuming an exact value: + /// + /// * **Linux** doubles the request and clamps it to `net.core.rmem_max` + /// (asking 262144 on a stock kernel reports 524288). + /// * **macOS/BSD** normally keep the request exactly, but start much higher + /// than Linux: an accepted loopback socket defaults to around 408300 bytes. + /// * **Windows** rounds up to its own granularity and may keep an auto-tuned + /// receive window that is larger than the request. + /// * **WASI** forwards to the host socket, so it inherits that host's policy. + RecvBufferSize(u32), + /// `SO_SNDBUF`: requested send-buffer bytes. The final size is the kernel's + /// choice with the same per-platform rounding as + /// [`RecvBufferSize`](Self::RecvBufferSize); read it back. + SendBufferSize(u32), + /// `IP_TTL` / `IPV6_UNICAST_HOPS`: hop limit for outgoing unicast packets. + Ttl(u32), + /// `IPV6_V6ONLY`: refuse IPv4-mapped peers on an IPv6 socket. Bind-time on + /// every platform, so setting it on a bound socket fails. + Ipv6Only(bool), + /// `SO_BROADCAST`: permit datagrams addressed to a broadcast address. + Broadcast(bool), + /// `IP_MULTICAST_TTL` / `IPV6_MULTICAST_HOPS`: outgoing multicast hop limit. + MulticastTtl(u32), + /// `IP_MULTICAST_LOOP` / `IPV6_MULTICAST_LOOP`: deliver this socket's own + /// multicast sends back to local members. + MulticastLoop(bool), + /// `IP_ADD_MEMBERSHIP` / `IPV6_JOIN_GROUP`: start receiving this group. + MulticastJoin(MulticastGroup), + /// `IP_DROP_MEMBERSHIP` / `IPV6_LEAVE_GROUP`: stop receiving this group. + MulticastLeave(MulticastGroup), +} +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +/// Names a readable socket option for `Loop::get_option`. +/// +/// Group membership has no getter: the OS exposes no per-socket membership +/// query, so [`SocketOption::MulticastJoin`] and +/// [`SocketOption::MulticastLeave`] have no kind here rather than a kind that +/// would have to answer `Unsupported` everywhere. +pub enum SocketOptionKind { + /// Read [`SocketOption::NoDelay`]. + NoDelay, + /// Read [`SocketOption::KeepAlive`], including the schedule the OS holds. + KeepAlive, + /// Read [`SocketOption::Linger`]. + Linger, + /// Read [`SocketOption::RecvBufferSize`] as the OS actually kept it. + RecvBufferSize, + /// Read [`SocketOption::SendBufferSize`] as the OS actually kept it. + SendBufferSize, + /// Read [`SocketOption::Ttl`]. + Ttl, + /// Read [`SocketOption::Ipv6Only`]. + Ipv6Only, + /// Read [`SocketOption::Broadcast`]. + Broadcast, + /// Read [`SocketOption::MulticastTtl`]. + MulticastTtl, + /// Read [`SocketOption::MulticastLoop`]. + MulticastLoop, +} #[derive(Clone, Copy, Debug, Default)] -/// UDP binding options. +/// UDP binding options. Reuse is bind-time only and stays here, not in +/// [`SocketOption`]; everything changeable on a live socket is an option. pub struct UdpOpts { /// Enable SO_REUSEPORT when supported; macOS does not promise balanced accepts. /// Defaults to false: a live UDP endpoint cannot be shared by another bind. diff --git a/docs/lanes/sockopts.md b/docs/lanes/sockopts.md new file mode 100644 index 0000000..0bf8f2b --- /dev/null +++ b/docs/lanes/sockopts.md @@ -0,0 +1,296 @@ +# sockopts — socket options on live handles (issue #34) + +Base: `09d205f`. Branch `lane/sockopts` (PR #36), clone +`/Users/amlug/projects/perry/windlass-lanes/sockopts`. Linux coverage ran in +`/root/claude-sockopts` on the shared build box. Implementation and verification +complete: CI run 34991952873 is green on every platform arm. + +Perry's `node:net` migration could not implement `socket.setNoDelay()`, +`setKeepAlive()` or `dgram.setTTL()`: turnloop took `nodelay` only in `ConnectOpts` +at creation, `ListenOpts` had no such field, and there was no way to change an +option on a live handle at all — so an **accepted** socket, which is what a server +configures, was unconfigurable. + +## API + +```rust +impl Loop { + pub fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()>; + pub fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result; +} + +pub enum SocketOption { + NoDelay(bool), // TCP_NODELAY + KeepAlive(Option), // SO_KEEPALIVE + schedule; None disables + Linger(Option), // SO_LINGER; Some(ZERO) resets on close + RecvBufferSize(u32), // SO_RCVBUF + SendBufferSize(u32), // SO_SNDBUF + Ttl(u32), // IP_TTL / IPV6_UNICAST_HOPS + Ipv6Only(bool), // IPV6_V6ONLY (bind-time; see below) + Broadcast(bool), // SO_BROADCAST + MulticastTtl(u32), // IP_MULTICAST_TTL / IPV6_MULTICAST_HOPS + MulticastLoop(bool), // IP_MULTICAST_LOOP / IPV6_MULTICAST_LOOP + MulticastJoin(MulticastGroup), // IP_ADD_MEMBERSHIP / IPV6_JOIN_GROUP + MulticastLeave(MulticastGroup), // IP_DROP_MEMBERSHIP / IPV6_LEAVE_GROUP +} +pub struct KeepAlive { idle: Option, interval: Option, count: Option } +pub struct MulticastGroup { group: IpAddr, interface: u32 } + +pub enum SocketOptionKind { NoDelay, KeepAlive, Linger, RecvBufferSize, + SendBufferSize, Ttl, Ipv6Only, Broadcast, + MulticastTtl, MulticastLoop } + +pub struct ListenOpts { reuse_port: bool, backlog: u32, accept_defaults: AcceptDefaults } +pub struct AcceptDefaults { nodelay: bool, keep_alive: Option } +``` + +Decisions, all of them written into the rustdoc and DESIGN §7.7 so they cannot +drift: + +- **Synchronous, not an operation.** Both calls run inside the call: no Request is + accepted, no completion is produced, nothing is queued, nothing is allocated. + Same shape as `tty_set_mode`/`local_addr`. +- **`get_option` never answers from a cache.** Every call is a `getsockopt` (or the + matching `wasi:sockets` import), because the OS is entitled to round, clamp or + double a request — `SO_RCVBUF` on Linux doubles it. Reporting the *request* back + would be a lie the host acts on. +- **A getter key, not a value, for reads.** `MulticastJoin`/`MulticastLeave` have no + `SocketOptionKind`: no OS exposes a per-socket membership query, and a kind that + answered `Unsupported` everywhere would be worse than its absence. +- **Bind-time options stay in the opts structs.** `SO_REUSEPORT` is + `ListenOpts::reuse_port`/`UdpOpts::reuse_port`, and `SO_REUSEADDR` is applied by + the backend to every TCP listener it binds. Neither can be changed on a bound + socket, so neither is an option. `Ipv6Only` is the borderline case: it is kept in + the enum so it can be *read* on a live socket, and setting it after bind is + refused by the OS (asserted, not just documented). +- **`accept_defaults` applies before the host sees the connection.** The accepting + backend applies them to the accepted socket after the OS accept and before the + `Accepted` completion is produced. A backend that cannot apply a requested + default rejects the **listener** when it is created rather than ignoring the + request once per connection; an OS failure while applying one fails that accept + and drops the socket instead of handing up a half-configured connection. Each + field is at most one `setsockopt` — that is what "cheaply" is allowed to mean. +- **Granularity.** Native keep-alive and linger schedules are whole seconds, so a + `Duration` is rounded **up** and a zero keep-alive interval is `InvalidInput` + rather than silently becoming "immediately". WASI takes the duration unrounded. +- **Buffer sizes are kernel-chosen; the contract is a floor, not an equality.** + Measured on an accepted loopback socket: macOS defaults to **408300** bytes and + keeps a request exactly; Linux defaults to **87380**, doubles the request and + clamps it to `net.core.rmem_max`; Windows rounds up to its own granularity and + refused to shrink an auto-tuned window from 131072 to a 49152 request. There is + therefore no portable exact expectation, and no portable *direction* either, so + the API documents "at least what you asked for, read it back" and the tests + assert that. Recorded on `SocketOption::RecvBufferSize`. +- **Partial failure is reported, not hidden.** Keep-alive is a switch plus up to + three separate kernel settings. Every value is validated before any is written, + but if the OS still rejects one after the switch was set, the error is returned + and the socket keeps whatever the OS left — documented on the variant, because a + silent rollback that itself failed would be worse. +- **Handle rules.** A timer handle is `InvalidInput`; a closing handle is + `InvalidInput`; a released handle is `NotFound`; stdio/file/TTY, process, signal + and fs-watch handles are `Unsupported`. + +### One behaviour change outside the strict ask + +`Open::Tcp` on WASI 0.2/0.3 previously **dropped `TcpOpts::nodelay` on the floor** +(`Open::Tcp { addr, .. }`), because `wasi:sockets` has no Nagle control. Leaving +that while making `SocketOption::NoDelay` and `AcceptDefaults::nodelay` report +`Unsupported` on the same backend would have been indefensible, so the +creation-time hint is now reported too, and the shared `pair()` fixture in +`turnloop-contract` stops requesting `nodelay: true` (native coverage of that path +moved into `nodelay_round_trip_and_accept_default`, which asserts it through the +OS instead of assuming it). Revert this hunk if the spec owner prefers the hint. + +## Per-backend support matrix + +`Unsupported` in this table means the call is **reported**, never accepted and +ignored. + +| Option | Linux (epoll) | macOS / BSD (kqueue) | Windows (IOCP) | WASI 0.2 | WASI 0.3 | Web | +|---|---|---|---|---|---|---| +| `NoDelay` | ✓ `TCP_NODELAY` | ✓ | ✓ | **Unsupported** | **Unsupported** | **Unsupported** | +| `KeepAlive` | ✓ `SO_KEEPALIVE` + `TCP_KEEPIDLE`/`_KEEPINTVL`/`_KEEPCNT` | ✓ (`TCP_KEEPALIVE` is Apple's idle name) | ✓ (schedule needs Win10 1709+; older Winsock returns its own error) | ✓ `keep-alive-{enabled,idle-time,interval,count}` | ✓ (`get_`-prefixed getters) | **Unsupported** | +| `Linger` | ✓ `SO_LINGER` | ✓ | ✓ (`LINGER`, `u16` fields) | **Unsupported** | **Unsupported** | **Unsupported** | +| `RecvBufferSize` / `SendBufferSize` | ✓ (kernel doubles) | ✓ | ✓ | ✓ TCP and UDP | ✓ | **Unsupported** | +| `Ttl` | ✓ `IP_TTL` / `IPV6_UNICAST_HOPS` | ✓ | ✓ | ✓ `hop-limit` / `unicast-hop-limit` | ✓ | **Unsupported** | +| `Ipv6Only` | read ✓; set refused after bind by the OS | same | same | **Unsupported** | **Unsupported** | **Unsupported** | +| `Broadcast` | ✓ `SO_BROADCAST` | ✓ | ✓ | **Unsupported** | **Unsupported** | **Unsupported** | +| `MulticastTtl` / `MulticastLoop` | ✓ (`int`) | ✓ (`u_char` for IPv4, `int` for IPv6 — written and read at the right width, not left to byte order) | ✓ (`DWORD`) | **Unsupported** | **Unsupported** | **Unsupported** | +| `MulticastJoin` / `MulticastLeave`, IPv6 | ✓ `ipv6_mreq`, any interface index | ✓ | ✓ | **Unsupported** | **Unsupported** | **Unsupported** | +| `MulticastJoin` / `MulticastLeave`, IPv4 | ✓ `ip_mreq` (index 0) or `ip_mreqn` (index *n*) | ✓ index 0; **Unsupported** for a nonzero index | ✓ index 0; **Unsupported** for a nonzero index | **Unsupported** | **Unsupported** | **Unsupported** | +| `AcceptDefaults::nodelay` | ✓ | ✓ | ✓ | listener rejected at creation | listener rejected at creation | no listeners at all | +| `AcceptDefaults::keep_alive` | ✓ | ✓ | ✓ | ✓ | ✓ | no listeners at all | + +Why each `Unsupported` is real, not laziness: + +- **WASI 0.2/0.3.** `wasi:sockets@0.2.9` / `0.3.0` expose exactly keep-alive + (enabled + idle + interval + count), send/receive buffer size and the hop limit + on `tcp-socket`, and unicast hop limit + buffer sizes on `udp-socket`. There is no + interface for Nagle, linger, IPv6-only, broadcast or group membership; there is + nothing to call. +- **Web.** The backend's handles are a host `fetch` and a host `WebSocket`; there is + no socket underneath to configure, and listening sockets are themselves + unsupported (DESIGN §7.5). It inherits the `Backend` trait's `Unsupported` + defaults and adds no code. +- **IPv4 membership by interface index off Linux.** BSD and Winsock name the + interface by *address* in `ip_mreq`, not by index. Applying an index-keyed request + to the default interface would be exactly the silent lie this API exists to + prevent, so it is reported. + +## Files + +| File | What | +|---|---| +| `crates/turnloop/src/types.rs` | `SocketOption`, `SocketOptionKind`, `KeepAlive`, `MulticastGroup`, `AcceptDefaults`; `ListenOpts::accept_defaults` | +| `crates/turnloop/src/driver.rs` | `Loop::set_option` / `Loop::get_option`, handle validation | +| `crates/turnloop/src/backend/mod.rs` | `Backend::set_option` / `get_option` (default `Unsupported`) and the rules in the module docs | +| `crates/turnloop/src/backend/sockopt.rs` | new; epoll + kqueue implementation | +| `crates/turnloop/src/backend/unix.rs` | wiring; accept applies the listener's defaults; `socket_fd` rejects non-sockets | +| `crates/turnloop/src/backend/iocp/sockopt.rs` | new; Winsock implementation | +| `crates/turnloop/src/backend/iocp/mod.rs` | wiring; defaults applied after `SO_UPDATE_ACCEPT_CONTEXT` | +| `crates/turnloop/src/backend/wasi_p2/sockopt.rs`, `wasi_p3/sockopt.rs` | new; `wasi:sockets` implementation | +| `crates/turnloop/src/backend/wasi_p2.rs`, `wasi_p3.rs` | wiring; `nodelay` reported instead of dropped | +| `crates/turnloop/src/backend/web.rs` | doc note only; trait defaults already report `Unsupported` | +| `crates/turnloop-contract/src/sockopts.rs` | new; backend-generic contracts | +| `crates/turnloop-contract/tests/sockopts.rs` | new; native arm + the independent `getsockopt` probes | +| `crates/turnloop-contract/tests/wasi.rs`, `tests/web/web_contract.rs` | WASI and web arms | +| `crates/turnloop-contract/tests/allocations.rs` | steady-state allocation gate | +| `DESIGN.md` | §7.7 and a §7.6 matrix row | +| `docs/wasm.md` | WASI/web contract-family rows | + +## Tests — and how each one proves its subject + +Every contract assertion reads the value back **through the OS**, because +`get_option` is a `getsockopt`. Three tests go further and do not trust turnloop's +own getter at all. + +**Independent OS probes** (`crates/turnloop-contract/tests/sockopts.rs`): + +- `accepted_socket_options_are_visible_to_getsockopt` — finds the descriptor + turnloop accepted *without asking turnloop for it*: it walks `/proc/self/fd` + (Linux) or `/dev/fd` (macOS/BSD) and matches the one whose `getsockname` is the + listener's address and whose `getpeername` is the client's, which the listener + itself cannot satisfy. It then calls `libc::getsockopt` from the test process and + asserts `TCP_NODELAY`, `SO_KEEPALIVE` and the idle time the *listener default* + asked for, and that a later `set_option(NoDelay(false))`/`KeepAlive(None)` is + visible there too. +- `adopted_socket_options_reach_the_shared_socket` — `dup`s a socket the test owns, + adopts one reference through `Detached::from_fd` / `from_socket`, and asserts + through `getsockopt` on the reference turnloop never saw. This is the portable + probe and the one that runs on Windows. Its subject proof is carried by the two + options no kernel rounds — `TCP_NODELAY` (false → true) and `IP_TTL` (its + default → exactly 7) — because the buffer sizes cannot carry it: see the + kernel-chosen note above. `SO_RCVBUF` is still checked, as a floor. +- `linger_zero_resets_the_connection` — the observable-behaviour test. Both arms + run in one test: with `Linger(Some(ZERO))` the peer's pending read completes with + `ConnectionReset`; without it the same close delivers `Eof`. The only difference + between the arms is the option, so neither verdict is vacuous. +- `multicast_membership_is_tracked` — membership has no getter, so the proof is the + kernel's own bookkeeping: leaving a group that was never joined **must fail**, + leaving the group that was joined must succeed, and leaving it a second time must + fail again. A backend that dropped the join cannot produce that sequence. + +**Coverage by endpoint**: connected client, **accepted connection**, listener +(through `accept_defaults` and `ipv4_membership_by_interface_index`), UDP socket, +adopted/attached socket, a listener moved between loops +(`accept_defaults_survive_transfer` — the defaults live on the transport, so they +travel through `detach`/`attach` and still configure connections on the new loop), +unconnected TCP socket (where address-family detection for IP-level options is +non-obvious), and non-socket handles. + +**`Unsupported` paths**: `socket_options_without_a_wasi_interface_are_unsupported` +(8 options + 6 getter kinds on WASI), `socket_options_are_unsupported_on_a_host_stream` +(10 options + 10 getter kinds on a live web WebSocket handle), +`non_socket_handles_report_unsupported` (stderr), +`local_listener_refuses_tcp_accept_defaults`, +`nodelay_accept_default_rejects_the_listener` and `nodelay_connect_hint_is_rejected` +(WASI), and `ipv4_membership_by_interface_index` (non-Linux). + +**Allocation gate**: `steady_socket_options_allocate_nothing` in +`tests/allocations.rs` runs 100 measured iterations of +`set_option`×3 + `get_option`×2 on UDP and 100 more of the three-syscall keep-alive +path on a connected socket, asserting `0` allocations *and* that the OS kept each +value — a gate that cannot pass having done nothing. It runs on native and WASI. + +**Sabotage checks (not committed).** With `apply_accept_defaults` neutered and +`set_linger` turned into `Ok(())`, exactly the three tests that should fail did: +`accepted_socket_options_are_visible_to_getsockopt`, +`nodelay_round_trip_and_accept_default` and `linger_zero_resets_the_connection`. +The other tests stayed green, which is the right blast radius. After the buffer +assertions were relaxed to a floor, both relaxed sites were re-sabotaged: stubbing +`Ttl` fails the adopted probe, and stubbing the two buffer setters fails +`buffer_sizes_round_trip` on its growth assertion. Neither relaxation made a test +unable to fail. + +## Verification + +macOS (kqueue) is this machine, `aarch64-apple-darwin`. Linux (epoll) is +`x86_64-unknown-linux-gnu` on the shared build box, clone `/root/claude-sockopts` +with an `OWNER` file naming this lane. + +| Command | Result | +| --- | --- | +| `cargo +nightly-2026-08-20 fmt --all --check` | PASS | +| `python3 scripts/ci/check-paths.py` | PASS — 1501 tracked files, 264 references | +| `python3 scripts/ci/feature_modes.py` | PASS — 6 public features, 18 required arms | +| `cargo +nightly-2026-08-20 clippy --locked --workspace --all-targets --all-features -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS (macOS) | +| same, `--target x86_64-unknown-linux-gnu`, `-p turnloop -p turnloop-contract -p turnloop-io` | PASS | +| same, `--target x86_64-pc-windows-msvc` | PASS | +| same, `--target wasm32-wasip2 --all-features` | PASS | +| same, `--target wasm32-unknown-unknown --all-features` | PASS | +| `cargo +nightly-2026-09-07 clippy … --target wasm32-wasip3 --all-features` | PASS | +| `RUSTDOCFLAGS='-D warnings' cargo +nightly-2026-08-20 doc -p turnloop --all-features --no-deps` | PASS | +| `cargo +stable check --locked --workspace --all-targets --all-features` | PASS | +| `cargo test --workspace --no-fail-fast -- --test-threads=1` (macOS) | PASS — 65 test binaries ok, 0 failed | +| `cargo test -p turnloop-contract --test sockopts -- --test-threads=1` (macOS) | PASS — 17/17 | +| `cargo test -p turnloop-contract --test sockopts -- --test-threads=1` (Linux, epoll) | PASS — 17/17 | +| Linux six required modes: `default`, `epoll-timerfd`, `process-sigchld`, `fallbacks`, `executor`, `all-features`, each `cargo +nightly-2026-08-20 test --locked --workspace --no-fail-fast … -- --test-threads=1 --skip permission_denied_is_reported` | PASS — all six modes, 406 `test result: ok` lines, 0 FAILED; the `getsockopt`, multicast, transfer and allocation tests each ran once per mode | +| `python3 scripts/ci/run-tests.py wasi --target wasm32-wasip2` (Wasmtime 46.0.0) | PASS — 42 contract + 13 allocation tests | +| `python3 scripts/ci/run-tests.py wasi --target wasm32-wasip3` (nightly-2026-09-07) | PASS — 42 contract + 14 allocation tests | +| `python3 scripts/ci/run-tests.py node` (web backend under Node 26.5.1) | PASS — 16/16, including `socket_options_are_unsupported_on_a_host_stream` | +| `bash scripts/ci/no-tokio.sh` | PASS — 5 graphs, zero runtime crates | +| `python3 scripts/ci/soak.py` | PASS — 251 locked versions, 1 pre-existing rustls exception | +| `cargo +nightly-2026-08-20 deny --locked check` | PASS — advisories, bans, licenses, sources | +| `python3 scripts/ci/lint-workflows.py` (actionlint + zizmor + shellcheck) | PASS — no workflow files were changed | +| `python3 -m unittest discover -s scripts/ci -p 'test_*.py'` | PASS — 98 tests | +| `python3 scripts/ci/run-tests.py web` (headless Chromium/Firefox) | PASS **in CI** (run 34991952873); UNRUN locally, no browsers on this machine | +| Windows `cargo test` | PASS **in CI** on all three `windows-2025` modes (run 34991952873) | + +### CI + +| Run | SHA | Result | +| --- | --- | --- | +| [34990033482](https://github.com/PerryTS/turnloop/actions/runs/34990033482) | `7d97b2a` | FAIL — every job green except the three `windows-2025` arms, each failing only `adopted_socket_options_reach_the_shared_socket` (15/16), all with `the OS reports 131072 bytes for a 49152-byte request` | +| [34991952873](https://github.com/PerryTS/turnloop/actions/runs/34991952873) | `b87bae6` | **PASS — every job, `ci-gate` green.** Zero `test result: FAILED` lines in the whole log; the previously failing probe now passes six times on `windows-2025` (three modes × workspace and per-member runs) | + +That first run is the evidence for the buffer-size decision above: it is also the +first execution of `backend/iocp/sockopt.rs` anywhere, and everything else in it +passed on Windows first time — the `LINGER` layout, the Win10-1709 keep-alive +schedule, the accept default applied after `SO_UPDATE_ACCEPT_CONTEXT`, and the +`linger 0` → `ConnectionReset` behavioural test. + +### Pre-existing failures seen on the build box (not this lane) + +- `turnloop-contract --test filesystem permission_denied_is_reported` fails there + because the box runs as **root**, and root ignores a read-only file's mode. It + fails identically on the base commit `09d205f`, which was checked before skipping + it in the mode matrix. CI runs unprivileged. +- `cargo check -p turnloop --target aarch64-linux-android` fails on the base commit + too (two errors in the inotify watch code). Android is not in the CI matrix. + +## What still needs CI + +Nothing. Every platform arm has executed: Linux x86_64 and arm64 (six modes each), +macOS, Windows (three modes), WASI 0.2 and 0.3, and the headless-browser web arm. + +## Follow-ups this lane deliberately did not take + +- `turnloop-http`'s server still configures nothing per connection. Now that + `ListenOpts::accept_defaults` exists, `nodelay: true` is the obvious default for + an HTTP/1.1 and HTTP/2 server; that is a protocol-crate decision, not a core one. +- `TcpOpts` gained no new fields. Anything a host wants to change after connect + goes through `set_option`, so the connect-time struct stays minimal. +- No `SocketOption` variant was added for `SO_REUSEADDR`, `SO_REUSEPORT`, + `IP_MULTICAST_IF` or `SO_BINDTODEVICE`: the first two are bind-time (already in + the opts structs) and the last two have no portable shape worth guessing at + before a host asks for them. diff --git a/docs/wasm.md b/docs/wasm.md index aa56807..80d4aff 100644 --- a/docs/wasm.md +++ b/docs/wasm.md @@ -322,7 +322,8 @@ and release allocation subjects remain mandatory. | Contract family | WASI p2/p3 | Web/Node | | --- | --- | --- | -| TCP connect/listen/accept, UDP, writev, shutdown | Exercised; reuse-port Unsupported, nodelay remains a hint because bindings lack a setter | Native socket cases excluded: platform lacks raw sockets. Actual Unsupported results tested; fetch/WebSocket byte paths replace transport workloads | +| TCP connect/listen/accept, UDP, writev, shutdown | Exercised; reuse-port Unsupported. `wasi:sockets` has no Nagle control, so `TcpOpts::nodelay`, `AcceptDefaults::nodelay` and `SocketOption::NoDelay` are all reported Unsupported rather than accepted and dropped | Native socket cases excluded: platform lacks raw sockets. Actual Unsupported results tested; fetch/WebSocket byte paths replace transport workloads | +| Socket options on live handles (issue #34) | Keep-alive (enable + idle/interval/count), send/receive buffer sizes and the unicast hop limit round-trip through `wasi:sockets` on clients, accepted connections and UDP; `ListenOpts::accept_defaults.keep_alive` reaches each accepted socket. Nagle, linger, IPv6-only, broadcast and multicast have no interface and report Unsupported for both set and get | Excluded: a host `fetch`/`WebSocket` has no socket behind it. Every option and every getter kind is asserted Unsupported on a live WebSocket handle | | Blocking pool / native cross-thread wake and 8-peer posting | Excluded: these single-agent targets cannot spawn OS threads; Running-notify/post behavior exercised | Native threads/pool excluded and Unsupported tested; two actual JS Workers exercise SAB posting | | Native detach/attach transfer | Excluded: WASI resource transfer Unsupported (accept attaches owned transport internally) | Excluded: browser transport transfer Unsupported, tested | | Integration fd/event, POSIX signal EINTR, kqueue/epoll/IOCP specifics | Excluded: runtime-owned integration, no native fd | Excluded: HostCallback and zero OS waits |