From 7b9ae1276fcfb259769598f6af40e06b45d09eb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Sun, 8 Nov 2026 07:27:46 +0100 Subject: [PATCH 1/4] feat(turnloop): bound TCP connect, receive shutdown, UDP datagrams and shared stdio Upstreams the provider repairs Perry has been carrying as a patched copy of alpha.10, so its WASI provider, native async and concurrent HTTP paths can run on a published turnloop. - ExecutorHandle::connect_bound and Backend::open_bound_tcp: connect an explicitly bound, not-yet-connected native TCP socket through the driver (Unix and IOCP), keeping its local address. - AsyncIo::shutdown_receive: shut down the receive side of a stream without disturbing writes in flight; refused while a read is pending. - AsyncDatagram and ExecutorHandle::datagram: an executor adapter that keeps UDP packet boundaries and source addresses. - backend::stdio: overlapping handles on one Unix stdio descriptor share its ownership, so closing one does not reset the settings another relies on. - Keepalive parameter setters on Unix, IOCP and WASI use the shared validation helpers without changing whether keepalive is enabled. The IOCP unit tests move to backend/iocp/tests.rs; no test is dropped. Supersedes #119, #120, #121, #122 and #123. --- crates/turnloop/src/backend/iocp/mod.rs | 189 +++++----------- crates/turnloop/src/backend/iocp/sockopt.rs | 18 ++ crates/turnloop/src/backend/iocp/tests.rs | 127 +++++++++++ crates/turnloop/src/backend/ipc.rs | 8 +- crates/turnloop/src/backend/mod.rs | 23 ++ crates/turnloop/src/backend/sockopt.rs | 18 ++ crates/turnloop/src/backend/stdio.rs | 91 ++++++++ crates/turnloop/src/backend/unix.rs | 69 +++++- crates/turnloop/src/backend/wasi_p2.rs | 17 ++ .../turnloop/src/backend/wasi_p2/sockopt.rs | 34 ++- crates/turnloop/src/backend/wasi_p3.rs | 5 + .../turnloop/src/backend/wasi_p3/sockopt.rs | 34 ++- crates/turnloop/src/driver.rs | 208 ++++++++++++------ crates/turnloop/src/executor.rs | 81 ++++++- crates/turnloop/src/executor/datagram.rs | 167 ++++++++++++++ crates/turnloop/src/lib.rs | 4 +- crates/turnloop/src/types.rs | 23 +- 17 files changed, 896 insertions(+), 220 deletions(-) create mode 100644 crates/turnloop/src/backend/iocp/tests.rs create mode 100644 crates/turnloop/src/backend/stdio.rs create mode 100644 crates/turnloop/src/executor/datagram.rs diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index b9851b03..6bcdc3d9 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -1467,11 +1467,56 @@ unsafe impl Backend for Iocp { sockopt::shutdown_read(self.socket_of(h)?) } fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + if matches!(option, SocketOption::ListenBacklog(_)) + && self.get(h)?.transport.kind != Kind::Listener + { + return Err(invalid()); + } sockopt::set(self.socket_of(h)?, option) } + fn open_bound_tcp( + &mut self, + h: Handle, + transport: Detached, + peer: SocketAddr, + opts: TcpOpts, + ) -> Result<()> { + if transport.kind != Kind::Tcp { + return Err(invalid()); + } + let Native::Socket(socket) = &transport.native else { + return Err(invalid()); + }; + let raw = socket.as_raw_socket() as usize; + let local = socket::address(raw, false)?; + if local.port() == 0 || local.is_ipv4() != peer.is_ipv4() { + return Err(invalid()); + } + match socket::address(raw, true) { + Ok(_) => return Err(invalid()), + Err(error) if error.os == Some(WSAENOTCONN) => {} + Err(error) => return Err(error), + } + if opts.nodelay { + sockopt::set(raw, SocketOption::NoDelay(true))?; + } + if peer.ip().is_loopback() { + socket::loopback_connect(raw); + } + self.install(h, transport, Some(peer), None) + } fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { sockopt::get(self.socket_of(h)?, kind) } + fn shutdown_receive(&mut self, h: Handle) -> Result<()> { + let transport = &self.get(h)?.transport; + if transport.kind != Kind::Tcp { + return Err(unsupported()); + } + // SAFETY: live owned TCP socket. SD_RECEIVE preserves sends and does + // not release OVERLAPPED storage; outstanding receives still complete. + socket::check(unsafe { shutdown(transport.native.raw() as usize, SD_RECEIVE) }) + } fn submit(&mut self, request: Request) -> Result<()> { if self.services.contains(request.handle) { return self.services.submit(request); @@ -1872,136 +1917,24 @@ fn write_buffers(op: &Operation, mut offset: usize, out: &mut [WSABUF; MAX_IOV]) } #[cfg(all(test, not(loom)))] -mod tests { - use super::*; +mod tests; - #[test] - fn direct_discovery_and_event_queue_draining_have_distinct_counts() { - use windows_sys::Win32::{ - Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject, - }; - let mut backend = Iocp::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); - let mut events = Vec::with_capacity(1); - for wake in [false, true] { - if wake { - backend.port.post(WAKE, 0).expect("wake"); - } - let info = backend - .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events) - .expect("discovery"); - assert_eq!( - (info.waits, info.discovery_polls, info.zero_event_waits), - (0, 1, u32::from(!wake)) - ); - assert!(events.is_empty()); - } - let Integration::Event(event) = backend.integration().expect("event helper") else { - panic!("Event required"); - }; - backend.port.post(WAKE, 0).expect("helper packet"); - // SAFETY: backend owns the event throughout this bounded external wait. - let result = unsafe { WaitForSingleObject(event as _, 2000) }; - assert_eq!(result, WAIT_OBJECT_0, "helper must forward a real packet"); - let info = backend - .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events) - .expect("helper drain"); - assert_eq!( - (info.waits, info.discovery_polls, info.zero_event_waits), - (0, 0, 0) - ); - assert!(events.is_empty()); - } +/// Construct an explicitly bound socket for the portable adoption contract. +#[cfg(test)] +pub(crate) fn bound_tcp_for_test(address: SocketAddr) -> (Detached, SocketAddr) { + let socket = socket::create(address.is_ipv6(), false).unwrap(); + let raw = socket.as_raw_socket() as usize; + socket::bind_to(raw, address).unwrap(); + let local = socket::address(raw, false).unwrap(); + (Detached::from_socket(socket).unwrap(), local) +} - // These operations model already-dequeued kernel completions. No native I/O - // is submitted, so unwind must discard their metadata rather than wait on it. - pub(super) struct Synthetic { - pub backend: Iocp, - pub entries: [Entry; 2], - pub ops: [OpId; 2], - } - impl Synthetic { - pub fn new() -> Self { - let config = Config { - max_handles: 2, - max_operations: 2, - ..Config::default() - }; - let mut backend = Iocp::new(&config, BufferPool::new(2, 64)).expect("backend"); - let handle = Handle { owner: 1, key: 0 }; - backend - .open( - handle, - Open::Tcp { - addr: ([127, 0, 0, 1], 9).into(), - opts: TcpOpts::default(), - }, - ) - .expect("socket resource"); - let ops = [OpId { owner: 1, key: 0 }, OpId { owner: 1, key: 1 }]; - let entries = std::array::from_fn(|i| { - backend - .submit(Request { - op: ops[i], - handle, - operation: Operation::Write(WriteBuf::Owned(vec![0; i + 1])), - }) - .expect("synthetic request"); - let pending = backend.ops[i].as_mut().expect("op"); - pending.waiting = true; - pending.queued = false; - pending.stage = Stage::Io; - Entry { - key: 1, - overlapped: backend.kernel_ptr(i) as usize, - bytes: (i + 1) as u32, - status: 0, - } - }); - backend.ready.clear(); - Self { - backend, - entries, - ops, - } - } - pub fn assert_delivered(&mut self) { - let mut out = Vec::with_capacity(1); - for (i, op) in self.ops.into_iter().enumerate() { - let info = self - .backend - .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut out) - .expect("drain valid packet"); - assert_eq!((info.waits, info.discovery_polls), (0, 0)); - assert_eq!(out.len(), 1); - assert_eq!(out[0].op, op); - assert!(out[0].terminal); - assert!(matches!(out[0].result, Ok(Outcome::Wrote(n)) if n == i + 1)); - out.clear(); - } - assert!(self.backend.ops.iter().all(Option::is_none)); - assert!(!self.backend.has_work()); - } - } - impl Drop for Synthetic { - fn drop(&mut self) { - self.backend.ops.iter_mut().for_each(|op| *op = None); - self.backend.ready.clear(); - } - } - #[test] - fn invalid_entry_does_not_discard_later_completions() { - let mut fixture = Synthetic::new(); - let batch = [fixture.entries[0], Entry::default(), fixture.entries[1]]; - assert_eq!( - fixture - .backend - .entries(&batch) - .expect_err("invalid packet") - .kind, - ErrorKind::InvalidInput - ); - fixture.assert_delivered(); - } +/// Unbound socket with a non-default kernel option for adoption tests. +#[cfg(test)] +pub(crate) fn unbound_tcp_for_test(address: SocketAddr) -> Detached { + let stream = std::net::TcpStream::from(socket::create(address.is_ipv6(), false).unwrap()); + stream.set_nodelay(true).unwrap(); + Detached::from_socket(stream.into()).unwrap() } /// Construct an explicitly bound socket for the portable adoption contract. diff --git a/crates/turnloop/src/backend/iocp/sockopt.rs b/crates/turnloop/src/backend/iocp/sockopt.rs index 796f8e3a..0c55b30a 100644 --- a/crates/turnloop/src/backend/iocp/sockopt.rs +++ b/crates/turnloop/src/backend/iocp/sockopt.rs @@ -235,6 +235,18 @@ 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::KeepAliveEnabled(on) => { + set_int(socket, SOL_SOCKET, SO_KEEPALIVE, i32::from(on)) + } + SocketOption::KeepAliveSettings(schedule) => set_keepalive_parameters(socket, schedule), + SocketOption::ListenBacklog(value) => { + let value = bounded(value)?; + if value == 0 { + return Err(invalid()); + } + // SAFETY: the backend verified this is a live TCP listener; no ownership transfer. + super::socket::check(unsafe { listen(socket, value) }) + } SocketOption::Linger(value) => set_linger(socket, value), SocketOption::RecvBufferSize(bytes) => { set_int(socket, SOL_SOCKET, SO_RCVBUF, bounded(bytes)?) @@ -267,6 +279,12 @@ pub(super) fn get(socket: usize, kind: SocketOptionKind) -> Result SocketOption::NoDelay(get_int(socket, IPPROTO_TCP, TCP_NODELAY)? != 0) } SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(socket)?), + SocketOptionKind::KeepAliveEnabled => { + SocketOption::KeepAliveEnabled(get_int(socket, SOL_SOCKET, SO_KEEPALIVE)? != 0) + } + SocketOptionKind::KeepAliveSettings => { + SocketOption::KeepAliveSettings(keepalive_parameters(socket)?) + } SocketOptionKind::Linger => SocketOption::Linger(get_linger(socket)?), SocketOptionKind::RecvBufferSize => { SocketOption::RecvBufferSize(get_int(socket, SOL_SOCKET, SO_RCVBUF)?.max(0) as u32) diff --git a/crates/turnloop/src/backend/iocp/tests.rs b/crates/turnloop/src/backend/iocp/tests.rs new file mode 100644 index 00000000..a52655da --- /dev/null +++ b/crates/turnloop/src/backend/iocp/tests.rs @@ -0,0 +1,127 @@ +use super::*; + +#[test] +fn direct_discovery_and_event_queue_draining_have_distinct_counts() { + use windows_sys::Win32::{Foundation::WAIT_OBJECT_0, System::Threading::WaitForSingleObject}; + let mut backend = Iocp::new(&Config::default(), BufferPool::new(2, 64)).expect("backend"); + let mut events = Vec::with_capacity(1); + for wake in [false, true] { + if wake { + backend.port.post(WAKE, 0).expect("wake"); + } + let info = backend + .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events) + .expect("discovery"); + assert_eq!( + (info.waits, info.discovery_polls, info.zero_event_waits), + (0, 1, u32::from(!wake)) + ); + assert!(events.is_empty()); + } + let Integration::Event(event) = backend.integration().expect("event helper") else { + panic!("Event required"); + }; + backend.port.post(WAKE, 0).expect("helper packet"); + // SAFETY: backend owns the event throughout this bounded external wait. + let result = unsafe { WaitForSingleObject(event as _, 2000) }; + assert_eq!(result, WAIT_OBJECT_0, "helper must forward a real packet"); + let info = backend + .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut events) + .expect("helper drain"); + assert_eq!( + (info.waits, info.discovery_polls, info.zero_event_waits), + (0, 0, 0) + ); + assert!(events.is_empty()); +} + +// These operations model already-dequeued kernel completions. No native I/O +// is submitted, so unwind must discard their metadata rather than wait on it. +pub(super) struct Synthetic { + pub backend: Iocp, + pub entries: [Entry; 2], + pub ops: [OpId; 2], +} +impl Synthetic { + pub fn new() -> Self { + let config = Config { + max_handles: 2, + max_operations: 2, + ..Config::default() + }; + let mut backend = Iocp::new(&config, BufferPool::new(2, 64)).expect("backend"); + let handle = Handle { owner: 1, key: 0 }; + backend + .open( + handle, + Open::Tcp { + addr: ([127, 0, 0, 1], 9).into(), + opts: TcpOpts::default(), + }, + ) + .expect("socket resource"); + let ops = [OpId { owner: 1, key: 0 }, OpId { owner: 1, key: 1 }]; + let entries = std::array::from_fn(|i| { + backend + .submit(Request { + op: ops[i], + handle, + operation: Operation::Write(WriteBuf::Owned(vec![0; i + 1])), + }) + .expect("synthetic request"); + let pending = backend.ops[i].as_mut().expect("op"); + pending.waiting = true; + pending.queued = false; + pending.stage = Stage::Io; + Entry { + key: 1, + overlapped: backend.kernel_ptr(i) as usize, + bytes: (i + 1) as u32, + status: 0, + } + }); + backend.ready.clear(); + Self { + backend, + entries, + ops, + } + } + pub fn assert_delivered(&mut self) { + let mut out = Vec::with_capacity(1); + for (i, op) in self.ops.into_iter().enumerate() { + let info = self + .backend + .poll(Some(Duration::ZERO), Budget::UNLIMITED, &mut out) + .expect("drain valid packet"); + assert_eq!((info.waits, info.discovery_polls), (0, 0)); + assert_eq!(out.len(), 1); + assert_eq!(out[0].op, op); + assert!(out[0].terminal); + assert!(matches!(out[0].result, Ok(Outcome::Wrote(n)) if n == i + 1)); + out.clear(); + } + assert!(self.backend.ops.iter().all(Option::is_none)); + assert!(!self.backend.has_work()); + } +} +impl Drop for Synthetic { + fn drop(&mut self) { + self.backend.ops.iter_mut().for_each(|op| *op = None); + self.backend.ready.clear(); + } +} +#[test] +fn invalid_entry_does_not_discard_later_completions() { + let mut fixture = Synthetic::new(); + let batch = [fixture.entries[0], Entry::default(), fixture.entries[1]]; + assert_eq!( + fixture + .backend + .entries(&batch) + .expect_err("invalid packet") + .kind, + ErrorKind::InvalidInput + ); + fixture.assert_delivered(); +} diff --git a/crates/turnloop/src/backend/ipc.rs b/crates/turnloop/src/backend/ipc.rs index a9a6b2b7..54b1f541 100644 --- a/crates/turnloop/src/backend/ipc.rs +++ b/crates/turnloop/src/backend/ipc.rs @@ -90,12 +90,16 @@ pub(super) fn accept(fd: RawFd) -> Result { pub(super) fn stdio(raw: RawFd) -> Result { // SAFETY: integer fcntl command duplicates the borrowed host fd with CLOEXEC. - let dup = unsafe { libc::fcntl(raw, libc::F_DUPFD_CLOEXEC, 0) }; + let dup = unsafe { libc::fcntl(raw, libc::F_DUPFD_CLOEXEC, 3) }; if dup < 0 { return Err(last_error()); } // SAFETY: successful fcntl returns a fresh owned descriptor. - classify(unsafe { OwnedFd::from_raw_fd(dup) }) + let fd = unsafe { OwnedFd::from_raw_fd(dup) }; + let (lease, mode) = super::stdio::Lease::acquire(raw)?; + let mut transport = classify(fd)?; + transport.share_stdio(lease, mode); + Ok(transport) } pub(super) fn classify(fd: OwnedFd) -> Result { diff --git a/crates/turnloop/src/backend/mod.rs b/crates/turnloop/src/backend/mod.rs index 98e6f304..95896e2e 100644 --- a/crates/turnloop/src/backend/mod.rs +++ b/crates/turnloop/src/backend/mod.rs @@ -357,6 +357,20 @@ pub unsafe trait Backend: Sized + 'static { }; /// Construct backend storage and native wait resources with the supplied shared read pool. fn new(config: &Config, pool: BufferPool) -> Result; + /// Adopt an exclusively owned, bound but unconnected TCP transport and + /// configure its peer before a Connect request. Preserve the local binding + /// and existing socket options; apply nodelay when requested. On error, + /// retain no resource or native registration. Backends without native + /// transport adoption return Unsupported. + fn open_bound_tcp( + &mut self, + _handle: Handle, + _transport: Self::Detached, + _peer: SocketAddr, + _opts: crate::TcpOpts, + ) -> Result<()> { + Err(Error::new(crate::ErrorKind::Unsupported)) + } /// Accept a typed filesystem request (only with `Filesystem::Backend`). `handle` /// is the request's file handle, or the new handle for an open. fn fs(&mut self, _op: OpId, _handle: Option, _request: crate::FsRequest) -> Result<()> { @@ -414,6 +428,13 @@ pub unsafe trait Backend: Sized + 'static { fn set_option(&mut self, _handle: Handle, _option: crate::SocketOption) -> Result<()> { Err(Error::new(crate::ErrorKind::Unsupported)) } + /// Synchronously request native receive-side shutdown without closing the + /// send direction. Existing accepted requests retain their normal buffer + /// ownership and terminal acknowledgement requirements. Unsupported transports + /// must fail without changing either direction. + fn shutdown_receive(&mut self, _handle: Handle) -> 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. @@ -580,6 +601,8 @@ pub use wasi_p3::WasiP3 as Platform; mod ipc; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] mod process; +#[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] +mod stdio; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] mod services; diff --git a/crates/turnloop/src/backend/sockopt.rs b/crates/turnloop/src/backend/sockopt.rs index f6d58a33..83590132 100644 --- a/crates/turnloop/src/backend/sockopt.rs +++ b/crates/turnloop/src/backend/sockopt.rs @@ -325,6 +325,18 @@ pub(super) fn set(fd: RawFd, option: SocketOption) -> Result<()> { set_int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY, i32::from(on)) } SocketOption::KeepAlive(schedule) => set_keep_alive(fd, schedule), + SocketOption::KeepAliveEnabled(on) => { + set_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE, i32::from(on)) + } + SocketOption::KeepAliveSettings(schedule) => set_keepalive_parameters(fd, schedule), + SocketOption::ListenBacklog(value) => { + let value = bounded(value)?; + if value == 0 { + return Err(invalid()); + } + // SAFETY: the backend verified this is a live TCP listener; no ownership transfer. + check(unsafe { libc::listen(fd, value) }) + } SocketOption::Linger(value) => set_linger(fd, value), SocketOption::RecvBufferSize(bytes) => { set_int(fd, libc::SOL_SOCKET, libc::SO_RCVBUF, bounded(bytes)?) @@ -377,6 +389,12 @@ pub(super) fn get(fd: RawFd, kind: SocketOptionKind) -> Result { SocketOption::NoDelay(get_int(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY)? != 0) } SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(fd)?), + SocketOptionKind::KeepAliveEnabled => { + SocketOption::KeepAliveEnabled(get_int(fd, libc::SOL_SOCKET, libc::SO_KEEPALIVE)? != 0) + } + SocketOptionKind::KeepAliveSettings => { + SocketOption::KeepAliveSettings(keepalive_parameters(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, diff --git a/crates/turnloop/src/backend/stdio.rs b/crates/turnloop/src/backend/stdio.rs new file mode 100644 index 00000000..b0e52b1e --- /dev/null +++ b/crates/turnloop/src/backend/stdio.rs @@ -0,0 +1,91 @@ +//! Process stdio duplicates share open-file descriptions, even across loops. +//! Snapshot all three before changing any: stdout and stderr may be aliases. +use super::poller::last_error; +use crate::{Error, ErrorKind, Result}; +use std::{ + os::fd::{AsRawFd, FromRawFd, OwnedFd}, + sync::Mutex, +}; + +static STATE: Mutex> = Mutex::new(None); + +struct Saved { + fd: OwnedFd, + flags: i32, + mode: Option, +} +struct State { + users: usize, + saved: [Option; 3], +} + +/// Held by the native resource until all its operations have quiesced. +pub(super) struct Lease; +impl Lease { + pub(super) fn acquire(raw: i32) -> Result<(Self, Option)> { + let mut state = STATE.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(state) = state.as_mut() { + state.users = state + .users + .checked_add(1) + .ok_or_else(|| Error::new(ErrorKind::ResourceLimit))?; + } else { + let mut saved = [None, None, None]; + for (raw, slot) in saved.iter_mut().enumerate() { + // SAFETY: duplicates a borrowed descriptor. Start above stdio so + // a closed standard descriptor is never accidentally populated. + let dup = unsafe { libc::fcntl(raw as i32, libc::F_DUPFD_CLOEXEC, 3) }; + if dup < 0 { + let error = std::io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::EBADF) { + continue; + } + return Err(error.into()); + } + // SAFETY: successful fcntl returned a fresh owned descriptor. + let fd = unsafe { OwnedFd::from_raw_fd(dup) }; + // SAFETY: integer-only command on a live descriptor. + let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) }; + if flags < 0 { + return Err(last_error()); + } + // SAFETY: termios is plain C output storage. + let mut mode = unsafe { std::mem::zeroed() }; + // SAFETY: live descriptor and writable termios storage. + let mode = + (unsafe { libc::tcgetattr(fd.as_raw_fd(), &mut mode) } == 0).then_some(mode); + *slot = Some(Saved { fd, flags, mode }); + } + *state = Some(State { users: 1, saved }); + } + let mode = state + .as_ref() + .and_then(|s| s.saved[raw as usize].as_ref()) + .and_then(|s| s.mode); + Ok((Self, mode)) + } +} +impl Drop for Lease { + fn drop(&mut self) { + let mut state = STATE.lock().unwrap_or_else(|e| e.into_inner()); + let Some(active) = state.as_mut() else { + return; + }; + active.users -= 1; + if active.users != 0 { + return; + } + // Serialize final restoration with the next acquisition. An Arc/Weak + // registry alone would let a new cohort snapshot flags while the last + // old owner is still restoring them in its destructor. + for saved in active.saved.iter().flatten() { + if let Some(mode) = &saved.mode { + // SAFETY: retained descriptor and previously captured termios. + unsafe { libc::tcsetattr(saved.fd.as_raw_fd(), libc::TCSANOW, mode) }; + } + // SAFETY: retained descriptor and flags from F_GETFL. + unsafe { libc::fcntl(saved.fd.as_raw_fd(), libc::F_SETFL, saved.flags) }; + } + *state = None; + } +} diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index 9aa09b08..bc281c0f 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -37,6 +37,7 @@ pub struct Detached { pub(super) kind: Kind, pub(super) original_flags: Option, original_mode: Option, + stdio_lease: 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. @@ -63,21 +64,29 @@ impl Detached { /// settings captured when the descriptor was adopted are restored first, /// exactly as they would be on close, and then the close is *not* performed. /// + /// For standard I/O, restoration waits until the last managed standard-I/O + /// transport is released, including duplicates held by other loops. A handed + /// off duplicate shares those flags and must not reconfigure them meanwhile. + /// /// A descriptor turnloop created itself was created non-blocking with /// `FD_CLOEXEC`, and it is handed over that way: nothing is restored, /// because nothing was changed. Call `fcntl(F_SETFL)` (or /// `TcpStream::set_nonblocking(false)`) if the receiving code wants blocking /// I/O, which is what a synchronous TLS handshake on the descriptor needs. - pub fn into_fd(self) -> OwnedFd { + pub fn into_fd(mut self) -> OwnedFd { self.restore(); + drop(self.stdio_lease.take()); let this = std::mem::ManuallyDrop::new(self); // SAFETY: `this` is never dropped, so this move of the single owning - // field cannot be observed twice; the remaining fields are Copy/plain - // data whose Drop is a no-op. `restore` already ran, and it is the only - // thing this type's Drop does besides releasing `fd`. + // field cannot be observed twice; the lease was released above and + // all other remaining fields are Copy/plain data whose Drop is a no-op. + // Restoration and lease release have already run. unsafe { std::ptr::read(&this.fd) } } fn restore(&self) { + if self.stdio_lease.is_some() { + return; + } if let Some(mode) = &self.original_mode { // SAFETY: descriptor is still owned; restore before OwnedFd drops. unsafe { @@ -102,9 +111,15 @@ impl Detached { kind, original_flags: None, original_mode, + stdio_lease: None, accept_defaults: AcceptDefaults::EMPTY, } } + pub(super) fn share_stdio(&mut self, lease: super::stdio::Lease, mode: Option) { + self.original_flags = None; + self.original_mode = mode; + self.stdio_lease = Some(lease); + } /// Adopt an owned Unix descriptor, classifying stream/file/TTY or socket. /// The descriptor must have no concurrent I/O users. Status flags and terminal /// settings are restored when this transport is closed or dropped. @@ -711,6 +726,35 @@ unsafe impl Backend for Unix { fn local_addr(&self, h: Handle) -> Result { socket::local_addr(self.get(h)?.transport.fd.as_raw_fd()) } + fn open_bound_tcp( + &mut self, + h: Handle, + transport: Detached, + peer: SocketAddr, + opts: TcpOpts, + ) -> Result<()> { + if transport.kind != Kind::Tcp { + return Err(Error::new(ErrorKind::InvalidInput)); + } + let fd = transport.fd.as_raw_fd(); + let local = socket::local_addr(fd)?; + if local.port() == 0 || local.is_ipv4() != peer.is_ipv4() { + return Err(Error::new(ErrorKind::InvalidInput)); + } + let mut address = Addr::empty(); + // SAFETY: exclusively owned socket and initialized writable address storage. + if unsafe { libc::getpeername(fd, address.mut_ptr(), &mut address.len) } == 0 { + return Err(Error::new(ErrorKind::InvalidInput)); + } + let error = last_error(); + if error.os != Some(libc::ENOTCONN) { + return Err(error); + } + if opts.nodelay { + socket::option(fd, libc::IPPROTO_TCP, libc::TCP_NODELAY, 1)?; + } + self.install(h, transport, Some(Addr::new(peer))) + } fn set_keepalive_parameters(&mut self, h: Handle, value: crate::KeepAlive) -> Result<()> { super::sockopt::set_keepalive_parameters(self.socket_fd(h)?, value) } @@ -736,11 +780,28 @@ unsafe impl Backend for Unix { super::sockopt::shutdown_read(self.socket_fd(h)?) } fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + if matches!(option, SocketOption::ListenBacklog(_)) + && self.get(h)?.transport.kind != Kind::Listener + { + return Err(Error::new(ErrorKind::InvalidInput)); + } 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 shutdown_receive(&mut self, h: Handle) -> Result<()> { + let transport = &self.get(h)?.transport; + if !matches!(transport.kind, Kind::Tcp | Kind::Pipe) { + return Err(Error::new(ErrorKind::Unsupported)); + } + // SAFETY: live owned TCP/Unix stream socket; receive shutdown does not + // close it or release any buffer held by accepted native requests. + if unsafe { libc::shutdown(transport.fd.as_raw_fd(), libc::SHUT_RD) } < 0 { + return Err(last_error()); + } + Ok(()) + } fn submit(&mut self, request: Request) -> Result<()> { let h = request.handle; if self.services.contains(h) { diff --git a/crates/turnloop/src/backend/wasi_p2.rs b/crates/turnloop/src/backend/wasi_p2.rs index ba2e8b42..33bb3805 100644 --- a/crates/turnloop/src/backend/wasi_p2.rs +++ b/crates/turnloop/src/backend/wasi_p2.rs @@ -575,12 +575,29 @@ unsafe impl Backend for WasiP2 { self.install(h, transport, (kind == Kind::Tcp).then_some(addr)) } fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + if matches!(option, SocketOption::ListenBacklog(_)) + && self.get(h)?.transport.kind != Kind::Listener + { + return Err(Error::new(ErrorKind::InvalidInput)); + } 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 shutdown_receive(&mut self, h: Handle) -> Result<()> { + let transport = &self.get(h)?.transport; + if transport.kind != Kind::Tcp { + return Err(Error::new(ErrorKind::Unsupported)); + } + let Socket::Tcp(socket) = &transport.socket else { + return Err(Error::new(ErrorKind::Unsupported)); + }; + socket + .shutdown(wasip2::sockets::tcp::ShutdownType::Receive) + .map_err(error) + } fn local_addr(&self, h: Handle) -> Result { let a = match &self.get(h)?.transport.socket { Socket::Tcp(s) => s.local_address(), diff --git a/crates/turnloop/src/backend/wasi_p2/sockopt.rs b/crates/turnloop/src/backend/wasi_p2/sockopt.rs index ddc8308a..51db225e 100644 --- a/crates/turnloop/src/backend/wasi_p2/sockopt.rs +++ b/crates/turnloop/src/backend/wasi_p2/sockopt.rs @@ -40,12 +40,17 @@ fn set_keep_alive(socket: &TcpSocket, value: Option) -> Result<()> { let Some(schedule) = value else { return socket.set_keep_alive_enabled(false).map_err(error); }; + set_keep_alive_settings(socket, schedule, true) +} +fn set_keep_alive_settings(socket: &TcpSocket, schedule: KeepAlive, enable: bool) -> Result<()> { 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 enable { + socket.set_keep_alive_enabled(true).map_err(error)?; + } if let Some(idle) = idle { socket.set_keep_alive_idle_time(idle).map_err(error)?; } @@ -61,7 +66,10 @@ fn get_keep_alive(socket: &TcpSocket) -> Result> { if !socket.keep_alive_enabled().map_err(error)? { return Ok(None); } - Ok(Some(KeepAlive { + get_keep_alive_settings(socket).map(Some) +} +fn get_keep_alive_settings(socket: &TcpSocket) -> Result { + Ok(KeepAlive { idle: Some(Duration::from_nanos( socket.keep_alive_idle_time().map_err(error)?, )), @@ -69,7 +77,7 @@ fn get_keep_alive(socket: &TcpSocket) -> Result> { 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) @@ -79,6 +87,20 @@ fn bytes(value: u64) -> u32 { pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { match option { SocketOption::KeepAlive(schedule) => set_keep_alive(tcp(socket)?, schedule), + SocketOption::KeepAliveEnabled(on) => { + tcp(socket)?.set_keep_alive_enabled(on).map_err(error) + } + SocketOption::KeepAliveSettings(schedule) => { + set_keep_alive_settings(tcp(socket)?, schedule, false) + } + SocketOption::ListenBacklog(value) => { + if value == 0 || value > i32::MAX as u32 { + return invalid(); + } + tcp(socket)? + .set_listen_backlog_size(u64::from(value)) + .map_err(error) + } 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)), @@ -115,6 +137,12 @@ pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { pub(super) fn get(socket: &Socket, kind: SocketOptionKind) -> Result { Ok(match kind { SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(tcp(socket)?)?), + SocketOptionKind::KeepAliveEnabled => { + SocketOption::KeepAliveEnabled(tcp(socket)?.keep_alive_enabled().map_err(error)?) + } + SocketOptionKind::KeepAliveSettings => { + SocketOption::KeepAliveSettings(get_keep_alive_settings(tcp(socket)?)?) + } SocketOptionKind::RecvBufferSize => SocketOption::RecvBufferSize(bytes( match socket { Socket::Tcp(socket) => socket.receive_buffer_size(), diff --git a/crates/turnloop/src/backend/wasi_p3.rs b/crates/turnloop/src/backend/wasi_p3.rs index 82b5843b..985b8bf5 100644 --- a/crates/turnloop/src/backend/wasi_p3.rs +++ b/crates/turnloop/src/backend/wasi_p3.rs @@ -537,6 +537,11 @@ unsafe impl Backend for WasiP3 { self.install(h, transport, (kind == Kind::Tcp).then_some(addr)) } fn set_option(&mut self, h: Handle, option: SocketOption) -> Result<()> { + if matches!(option, SocketOption::ListenBacklog(_)) + && self.get(h)?.transport.kind != Kind::Listener + { + return Err(Error::new(ErrorKind::InvalidInput)); + } sockopt::set(&self.get(h)?.transport.socket, option) } fn get_option(&self, h: Handle, kind: SocketOptionKind) -> Result { diff --git a/crates/turnloop/src/backend/wasi_p3/sockopt.rs b/crates/turnloop/src/backend/wasi_p3/sockopt.rs index b8a45f32..4b7ab101 100644 --- a/crates/turnloop/src/backend/wasi_p3/sockopt.rs +++ b/crates/turnloop/src/backend/wasi_p3/sockopt.rs @@ -40,12 +40,17 @@ fn set_keep_alive(socket: &TcpSocket, value: Option) -> Result<()> { let Some(schedule) = value else { return socket.set_keep_alive_enabled(false).map_err(error); }; + set_keep_alive_settings(socket, schedule, true) +} +fn set_keep_alive_settings(socket: &TcpSocket, schedule: KeepAlive, enable: bool) -> Result<()> { 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 enable { + socket.set_keep_alive_enabled(true).map_err(error)?; + } if let Some(idle) = idle { socket.set_keep_alive_idle_time(idle).map_err(error)?; } @@ -61,7 +66,10 @@ fn get_keep_alive(socket: &TcpSocket) -> Result> { if !socket.get_keep_alive_enabled().map_err(error)? { return Ok(None); } - Ok(Some(KeepAlive { + get_keep_alive_settings(socket).map(Some) +} +fn get_keep_alive_settings(socket: &TcpSocket) -> Result { + Ok(KeepAlive { idle: Some(Duration::from_nanos( socket.get_keep_alive_idle_time().map_err(error)?, )), @@ -69,7 +77,7 @@ fn get_keep_alive(socket: &TcpSocket) -> Result> { 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) @@ -79,6 +87,20 @@ fn bytes(value: u64) -> u32 { pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { match option { SocketOption::KeepAlive(schedule) => set_keep_alive(tcp(socket)?, schedule), + SocketOption::KeepAliveEnabled(on) => { + tcp(socket)?.set_keep_alive_enabled(on).map_err(error) + } + SocketOption::KeepAliveSettings(schedule) => { + set_keep_alive_settings(tcp(socket)?, schedule, false) + } + SocketOption::ListenBacklog(value) => { + if value == 0 || value > i32::MAX as u32 { + return invalid(); + } + tcp(socket)? + .set_listen_backlog_size(u64::from(value)) + .map_err(error) + } 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)), @@ -115,6 +137,12 @@ pub(super) fn set(socket: &Socket, option: SocketOption) -> Result<()> { pub(super) fn get(socket: &Socket, kind: SocketOptionKind) -> Result { Ok(match kind { SocketOptionKind::KeepAlive => SocketOption::KeepAlive(get_keep_alive(tcp(socket)?)?), + SocketOptionKind::KeepAliveEnabled => { + SocketOption::KeepAliveEnabled(tcp(socket)?.get_keep_alive_enabled().map_err(error)?) + } + SocketOptionKind::KeepAliveSettings => { + SocketOption::KeepAliveSettings(get_keep_alive_settings(tcp(socket)?)?) + } SocketOptionKind::RecvBufferSize => SocketOption::RecvBufferSize(bytes( match socket { Socket::Tcp(socket) => socket.get_receive_buffer_size(), diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index ffd85f7b..3e94a33f 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -664,6 +664,12 @@ impl Driver { Ok(()) } /// Duplicate standard input, output or error, classifying pipe/file/terminal. + /// + /// Unix duplicates share nonblocking flags and terminal settings. Overlapping + /// managed standard-I/O transports, even on different loops, keep those + /// settings until the last transport is released; then the original settings + /// are restored. Do not replace or reconfigure process standard descriptors + /// (or their aliases) while any managed standard-I/O transport is live. pub fn open_stdio(&mut self, which: Stdio) -> Result { self.open(Open::Stdio(which)) } @@ -865,6 +871,39 @@ impl Driver { return Err(Error::new(ErrorKind::InvalidInput)); } let h = self.open(Open::Tcp { addr, opts: *opts })?; + self.start_tcp_connect(h, opts, token) + } + /// Connect an exclusively owned, already-bound native TCP socket. + /// + /// The binding (including an OS-assigned ephemeral port) and socket options + /// survive adoption. This supports options which affect the handshake, + /// such as the receive buffer size, and hosts which reserve a local port + /// before choosing a peer. `nodelay: true` enables TCP_NODELAY; otherwise + /// the existing setting is preserved. The peer must have the same address + /// family. Connected sockets, listeners and unbound sockets are rejected. + /// + /// Unix and Windows native backends support this operation. Others report + /// Unsupported. Ownership is consumed even on setup failure. Once accepted, + /// completion, timeout and close/cancellation follow [`Self::tcp_connect`]. + pub fn tcp_connect_bound_with_options( + &mut self, + transport: B::Detached, + addr: SocketAddr, + opts: &TcpOpts, + token: Token, + ) -> Result { + if opts.connect_timeout == Some(Duration::ZERO) { + return Err(Error::new(ErrorKind::InvalidInput)); + } + let h = self.new_handle(Kind::Socket)?; + if let Err(error) = self.backend.open_bound_tcp(h, transport, addr, *opts) { + self.handles.remove(h.key); + self.refs -= 1; + return Err(error); + } + self.start_tcp_connect(h, opts, token) + } + fn start_tcp_connect(&mut self, h: Handle, opts: &TcpOpts, token: Token) -> Result { match self.submit(h, Operation::Connect, token) { Ok(op) => { if let Some(at) = opts.connect_timeout.and_then(|t| self.now().checked_add(t)) { @@ -1154,6 +1193,22 @@ impl Driver { pub fn shutdown(&mut self, h: Handle, token: Token) -> Result { self.submit(h, Operation::Shutdown, token) } + /// Request native receive-side shutdown while preserving the send direction. + /// + /// This synchronous control supports TCP on Unix, Windows and WASI 0.2, + /// and Unix-domain stream sockets. Other transports/backends return + /// Unsupported. No completion is generated for the control itself. + /// Outstanding reads still require terminal acknowledgement before their + /// buffers can be reused; queued or already-completed data is not revoked. + /// Use the executor adapter's `shutdown_receive` to also discard its pending + /// read and buffered bytes and make all subsequent adapter reads return EOF. + pub fn shutdown_receive(&mut self, h: Handle) -> Result<()> { + let resource = self.resource(h)?; + if resource.closing.is_some() { + return Err(Error::new(ErrorKind::InvalidInput)); + } + self.backend.shutdown_receive(h) + } /// Request cancellation; terminal acknowledgement must precede memory reuse. pub fn cancel(&mut self, id: OpId) -> bool { self.cancel_inner(id, false) @@ -2149,7 +2204,14 @@ mod clock_contract { self.changes += 1; } fn open(&mut self, _: Handle, spec: Open) -> Result<()> { - if self.accept_connect && matches!(spec, Open::Pipe(_)) { + if self.accept_connect && matches!(spec, Open::Pipe(_) | Open::Tcp { .. }) { + Ok(()) + } else { + Err(Error::new(ErrorKind::Unsupported)) + } + } + fn open_bound_tcp(&mut self, _: Handle, _: (), _: SocketAddr, _: TcpOpts) -> Result<()> { + if self.accept_connect { Ok(()) } else { Err(Error::new(ErrorKind::Unsupported)) @@ -2365,76 +2427,86 @@ mod clock_contract { #[test] fn connection_deadline_waits_for_acknowledgement_and_retains_cancellation_errors() { - let mut driver = Driver::::new(Config::default()).expect("host loop"); - driver.backend.accept_connect = true; - let at = driver.now() + Duration::from_millis(10); - let h = driver - .pipe_connect_until(&PipeName("synthetic".into()), at, Token(1)) + for kind in 0..3 { + let mut driver = Driver::::new(Config::default()).expect("host loop"); + driver.backend.accept_connect = true; + let at = driver.now() + Duration::from_millis(10); + let opts = TcpOpts { + connect_timeout: Some(Duration::from_millis(10)), + ..TcpOpts::default() + }; + let peer = "127.0.0.1:9".parse().unwrap(); + let h = match kind { + 0 => driver.pipe_connect_until(&PipeName("synthetic".into()), at, Token(1)), + 1 => driver.tcp_connect(peer, &opts, Token(1)), + _ => driver.tcp_connect_bound_with_options((), peer, &opts, Token(1)), + } .expect("connect"); - let op = driver - .backend - .pending - .as_ref() - .expect("submitted connect") - .op; - let mut out = Completions::with_capacity(1); - driver - .turn(Timeout::Now, &mut out) - .expect("before deadline"); - assert!(out.is_empty()); - assert_eq!(driver.backend.cancel_calls, 0); - assert_eq!(driver.next_deadline(), Some(at)); - let failure = Error { - kind: ErrorKind::Other, - os: Some(12345), - }; - driver.backend.cancel_error = Some(failure); - driver.backend.now = at; - assert_eq!( + let op = driver + .backend + .pending + .as_ref() + .expect("submitted connect") + .op; + let mut out = Completions::with_capacity(1); driver .turn(Timeout::Now, &mut out) - .expect_err("injected cancellation error"), - failure - ); - assert_eq!(driver.backend.cancel_calls, 1); - assert_eq!(driver.next_deadline(), Some(at)); - assert!(out.is_empty() && driver.backend.pending.is_some()); - driver.backend.cancel_error = None; - driver - .turn(Timeout::Now, &mut out) - .expect("retry cancellation"); - assert_eq!(driver.backend.cancel_calls, 2); - assert_eq!(driver.next_deadline(), None); - assert!( - out.is_empty() && driver.backend.pending.is_some(), - "timeout cannot retire before native acknowledgement" - ); - driver.backend.acknowledge = true; - driver - .turn(Timeout::Now, &mut out) - .expect("acknowledge timeout"); - assert_eq!(out.len(), 1); - assert_eq!( - (out[0].handle, out[0].op, out[0].token, out[0].terminal), - (Some(h), Some(op), Token(1), true) - ); - assert!(matches!( - out[0].result, - OpResult::Err(Error { - kind: ErrorKind::TimedOut, - .. - }) - )); - assert!(driver.backend.pending.is_none()); - assert_eq!(driver.backend.cancel_calls, 2); - driver.close(h, Token(2)).expect("close timed-out stream"); - driver.turn(Timeout::Now, &mut out).expect("closed"); - assert_eq!(out.len(), 1); - assert!(matches!(out[0].result, OpResult::Closed)); - driver - .turn(Timeout::Now, &mut out) - .expect("no duplicate timeout"); - assert!(out.is_empty() && !driver.alive()); + .expect("before deadline"); + assert!(out.is_empty()); + assert_eq!(driver.backend.cancel_calls, 0); + assert_eq!(driver.next_deadline(), Some(at)); + let failure = Error { + kind: ErrorKind::Other, + os: Some(12345), + }; + driver.backend.cancel_error = Some(failure); + driver.backend.now = at; + assert_eq!( + driver + .turn(Timeout::Now, &mut out) + .expect_err("injected cancellation error"), + failure + ); + assert_eq!(driver.backend.cancel_calls, 1); + assert_eq!(driver.next_deadline(), Some(at)); + assert!(out.is_empty() && driver.backend.pending.is_some()); + driver.backend.cancel_error = None; + driver + .turn(Timeout::Now, &mut out) + .expect("retry cancellation"); + assert_eq!(driver.backend.cancel_calls, 2); + assert_eq!(driver.next_deadline(), None); + assert!( + out.is_empty() && driver.backend.pending.is_some(), + "timeout cannot retire before native acknowledgement" + ); + driver.backend.acknowledge = true; + driver + .turn(Timeout::Now, &mut out) + .expect("acknowledge timeout"); + assert_eq!(out.len(), 1); + assert_eq!( + (out[0].handle, out[0].op, out[0].token, out[0].terminal), + (Some(h), Some(op), Token(1), true) + ); + assert!(matches!( + out[0].result, + OpResult::Err(Error { + kind: ErrorKind::TimedOut, + .. + }) + )); + assert!(driver.backend.pending.is_none()); + assert_eq!(driver.backend.cancel_calls, 2); + driver.close(h, Token(2)).expect("close timed-out stream"); + driver.turn(Timeout::Now, &mut out).expect("closed"); + assert_eq!(out.len(), 1); + assert!(matches!(out[0].result, OpResult::Closed)); + driver + .turn(Timeout::Now, &mut out) + .expect("no duplicate timeout"); + assert!(out.is_empty() && !driver.alive()); + } } #[test] fn optional_native_capabilities_reject_without_leaking_core_reservations() { diff --git a/crates/turnloop/src/executor.rs b/crates/turnloop/src/executor.rs index f7f9dcc7..c62bb6d6 100644 --- a/crates/turnloop/src/executor.rs +++ b/crates/turnloop/src/executor.rs @@ -43,6 +43,9 @@ use std::{ task::{Context, Poll, Wake, Waker}, time::Duration, }; +mod datagram; +pub use datagram::AsyncDatagram; + const TAG: u64 = 1 << 63; /// Token for executor-owned closes that no future awaits by token. It carries the /// executor's tag, so its completion is claimed rather than handed to the host, @@ -491,6 +494,26 @@ impl ExecutorHandle { opts, key: None, handle: None, + transport: None, + } + } + /// Connect an already-bound native TCP transport without replacing its + /// socket, local endpoint or pre-handshake options. Dropping the future + /// before polling drops the transport; dropping a pending attempt closes + /// it through the driver. See [`Driver::tcp_connect_bound_with_options`]. + pub fn connect_bound( + &self, + transport: B::Detached, + addr: SocketAddr, + opts: TcpOpts, + ) -> Connect { + Connect { + executor: self.clone(), + address: addr, + opts, + key: None, + handle: None, + transport: Some(transport), } } /// Apply an absolute deadline without resetting it between I/O operations. @@ -509,6 +532,7 @@ impl ExecutorHandle { read: None, write: None, shutdown: None, + read_shut: false, write_shut: false, closed: false, } @@ -687,6 +711,7 @@ pub struct AsyncIo { read: Option, write: Option, shutdown: Option, + read_shut: bool, write_shut: bool, closed: bool, } @@ -708,6 +733,41 @@ impl AsyncIo { pub fn is_write_shut(&self) -> bool { self.write_shut } + /// Whether receive shutdown completed for this adapter. + pub fn is_read_shut(&self) -> bool { + self.read_shut + } + /// Stop receiving, discard this adapter's buffered input, and preserve sends. + /// + /// Calls native receive shutdown first; an unsupported transport leaves the + /// adapter unchanged. After success all reads return EOF, including when a + /// previous read was Pending or retained a partially consumed completion. + /// Pending native access is cancelled, but its internal buffer remains owned + /// by the executor until terminal acknowledgement. Repeated calls succeed. + /// Supports TCP on Unix, Windows and WASI 0.2 and Unix-domain streams. WASI + /// 0.3 and web currently return Unsupported; they require asynchronous stream + /// cancellation before dropping the receive endpoint. + pub fn shutdown_receive(&mut self) -> io::Result<()> { + if self.read_shut { + return Ok(()); + } + if self.closed { + return Err(io::ErrorKind::NotConnected.into()); + } + if self.peer.is_some() { + return Err(io::ErrorKind::Unsupported.into()); + } + self.shared + .driver + .borrow_mut() + .shutdown_receive(self.handle) + .map_err(shutdown_error)?; + self.read_shut = true; + if let Some(key) = self.read.take() { + self.shared.abandon(key); + } + Ok(()) + } /// Gracefully end the write direction without closing the handle. /// /// Flushes accepted writes first, then submits the backend's stream shutdown @@ -832,7 +892,7 @@ impl AsyncRead for AsyncIo { buf: &mut [u8], ) -> Poll> { let this = self.get_mut(); - if buf.is_empty() { + if this.read_shut || buf.is_empty() { return Poll::Ready(Ok(0)); } let key = match this.read { @@ -1094,6 +1154,7 @@ pub struct Connect { opts: TcpOpts, key: Option, handle: Option, + transport: Option, } /// A cancellable asynchronous name lookup. @@ -1279,11 +1340,19 @@ impl Future for Connect { Some(key) => key, None => { let key = shared.reserve(cx)?; - let result = - shared - .driver - .borrow_mut() - .tcp_connect(this.address, &this.opts, key.token()); + let result = match this.transport.take() { + Some(transport) => shared.driver.borrow_mut().tcp_connect_bound_with_options( + transport, + this.address, + &this.opts, + key.token(), + ), + None => shared.driver.borrow_mut().tcp_connect( + this.address, + &this.opts, + key.token(), + ), + }; match result { Ok(handle) => { this.handle = Some(handle); diff --git a/crates/turnloop/src/executor/datagram.rs b/crates/turnloop/src/executor/datagram.rs new file mode 100644 index 00000000..44a4e4cd --- /dev/null +++ b/crates/turnloop/src/executor/datagram.rs @@ -0,0 +1,167 @@ +//! Datagram-preserving operations on executor-owned staging slots. +use super::*; + +/// An owned UDP socket with independent receive and send operation slots. +/// +/// Packet boundaries and source addresses are preserved, including empty +/// packets. Pending operations borrow no caller memory. The executor's +/// `buffer_size` bounds a packet: larger sends fail before submission; receives +/// may be truncated by the backend. Configure at least 65,535 bytes to receive +/// full ordinary UDP packets. Dropping this adapter cancels pending operations +/// and closes the socket, retaining native buffers until terminal acknowledgement. +pub struct AsyncDatagram { + io: AsyncIo, +} +impl ExecutorHandle { + /// Adopt an existing loop-owned UDP socket. Do not give the same handle to + /// another adapter or close it through the driver while this owner is live. + pub fn datagram(&self, handle: Handle) -> AsyncDatagram { + AsyncDatagram { + io: self.io(handle), + } + } +} +impl AsyncDatagram { + /// The owned handle, for driver address queries and socket controls. + pub fn handle(&self) -> Handle { + self.io.handle() + } + + /// Receive one packet and its source address. Empty packets return length + /// zero, never EOF. If `buf` is smaller than the completed packet, the tail + /// is discarded, not returned as a second packet. An empty `buf` therefore + /// consumes one packet. Pending polls retain only executor-owned storage; + /// the next poll may supply a different buffer. + pub fn poll_recv_from( + &mut self, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + let this = &mut self.io; + if this.closed { + return Poll::Ready(Err(io::ErrorKind::NotConnected.into())); + } + let key = match this.read { + Some(key) => key, + None => { + let key = this.shared.reserve(cx).map_err(io_error)?; + let buffer = { + let mut slots = this.shared.slots.borrow_mut(); + let bytes = &mut slots[key.index].bytes; + // SAFETY: fixed slot storage outlives native access, including + // abandoned requests awaiting terminal acknowledgement. + ReadBuf::Provided(unsafe { + IoBufMut::from_raw_parts(bytes.as_mut_ptr(), bytes.len()) + }) + }; + let result = this + .shared + .driver + .borrow_mut() + .recv(this.handle, buffer, key.token()); + match result { + Ok(op) => { + this.shared.slots.borrow_mut()[key.index].op = Some(op); + this.read = Some(key); + key + } + Err(error) => { + this.shared.free(key); + return Poll::Ready(Err(io_error(error))); + } + } + } + }; + let Some(result) = this.shared.result(key, cx) else { + return Poll::Pending; + }; + let result = match result { + OpResult::RecvFrom { n, from, .. } => { + let slots = this.shared.slots.borrow(); + let bytes = &slots[key.index].bytes; + let take = buf.len().min(n); + buf[..take].copy_from_slice(&bytes[..take]); + Ok((take, from)) + } + OpResult::Err(error) => Err(io_error(error)), + OpResult::Cancelled => Err(io_error(Error::new(ErrorKind::Cancelled))), + _ => Err(io::Error::other("unexpected datagram receive completion")), + }; + this.shared.free(key); + this.read = None; + Poll::Ready(result) + } + + /// Accept exactly one packet for sending, including an empty packet. A + /// successful return means the adapter owns a copy; call `poll_flush` to + /// observe native completion/errors. A pending return consumes no bytes or + /// destination from this call: it waits for an earlier accepted packet. + /// Packets larger than the executor buffer are rejected without splitting. + pub fn poll_send_to( + &mut self, + cx: &mut Context<'_>, + buf: &[u8], + to: SocketAddr, + ) -> Poll> { + if self.io.closed { + return Poll::Ready(Err(io::ErrorKind::NotConnected.into())); + } + std::task::ready!(self.poll_flush(cx))?; + let this = &mut self.io; + let key = this.shared.reserve(cx).map_err(io_error)?; + let buffer = { + let mut slots = this.shared.slots.borrow_mut(); + let bytes = &mut slots[key.index].bytes; + if buf.len() > bytes.len() { + drop(slots); + this.shared.free(key); + return Poll::Ready(Err(io::Error::new( + io::ErrorKind::InvalidInput, + "datagram exceeds executor buffer", + ))); + } + bytes[..buf.len()].copy_from_slice(buf); + // SAFETY: copied bytes remain immutable and owned by this slot until + // native terminal completion, even if this adapter is dropped. + WriteBuf::Provided(unsafe { IoBuf::from_raw_parts(bytes.as_ptr(), buf.len()) }) + }; + let result = this + .shared + .driver + .borrow_mut() + .send_to(this.handle, buffer, to, key.token()); + match result { + Ok(op) => { + this.shared.slots.borrow_mut()[key.index].op = Some(op); + this.write = Some(key); + Poll::Ready(Ok(buf.len())) + } + Err(error) => { + this.shared.free(key); + Poll::Ready(Err(io_error(error))) + } + } + } + + /// Wait for the last accepted packet's native completion. This does not + /// imply delivery to the remote application; UDP has no acknowledgement. + pub fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.io).poll_flush(cx) + } + + /// Cancel and discard this adapter's pending receive. It may already have + /// consumed a packet. Native storage remains alive until cancellation is + /// acknowledged; a subsequent receive uses a different slot. Sends survive. + pub fn cancel_receive(&mut self) { + if let Some(key) = self.io.read.take() { + self.io.shared.abandon(key); + } + } + + /// Flush accepted sends, then close the socket and cancel pending receives. + pub fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll> { + std::task::ready!(self.poll_flush(cx))?; + self.cancel_receive(); + Pin::new(&mut self.io).poll_close(cx) + } +} diff --git a/crates/turnloop/src/lib.rs b/crates/turnloop/src/lib.rs index 1275fd66..c8b9f491 100644 --- a/crates/turnloop/src/lib.rs +++ b/crates/turnloop/src/lib.rs @@ -98,8 +98,8 @@ pub mod executor; #[cfg(feature = "executor")] pub use executor::{ - AsyncIo, Blocking, Close, ExecutorConfig, ExecutorHandle, JoinError, JoinHandle, LocalExecutor, - Sleep, + AsyncDatagram, AsyncIo, Blocking, Close, ExecutorConfig, ExecutorHandle, JoinError, JoinHandle, + LocalExecutor, Sleep, }; #[cfg(all(test, not(loom)))] diff --git a/crates/turnloop/src/types.rs b/crates/turnloop/src/types.rs index 7f9b5329..2faa3440 100644 --- a/crates/turnloop/src/types.rs +++ b/crates/turnloop/src/types.rs @@ -271,9 +271,10 @@ impl Config { /// receive buffer set that way is applied after the handshake has started, so it /// cannot change the window scale the peers negotiate. /// -/// Not offered at all, on any backend: binding the connecting socket to a -/// chosen local address or port before it connects. `IPV6_V6ONLY` is not needed -/// here: a connecting socket's family is the destination's. +/// To preserve a local address/port or pre-handshake options set by the host, +/// use [`crate::Driver::tcp_connect_bound`] with a bound native transport. +/// `IPV6_V6ONLY` is not needed for fresh connections here: their socket family +/// is the destination's. /// /// [`Loop::tcp_connect`]: crate::Driver::tcp_connect /// [`Loop::set_option`]: crate::Driver::set_option @@ -421,7 +422,7 @@ impl AcceptDefaults { } } #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] -/// TCP keep-alive probe schedule. Each `None` keeps the platform default. +/// TCP keep-alive probe schedule. Each `None` leaves that kernel setting unchanged. /// /// 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 @@ -469,6 +470,16 @@ pub enum SocketOption { /// 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), + /// Change only the keep-alive switch, preserving the kernel's probe schedule. + KeepAliveEnabled(bool), + /// Change the provided schedule fields without enabling/disabling probing. + /// Omitted fields remain unchanged; getters return all three native values, + /// including while probing is disabled. Validation precedes every write. + KeepAliveSettings(KeepAlive), + /// Update an existing TCP listener's backlog without replacing its socket. + /// Zero, values above i32::MAX, and non-listener handles are InvalidInput. + /// This option has no getter: kernels do not expose a portable backlog query. + ListenBacklog(u32), /// `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 @@ -521,6 +532,10 @@ pub enum SocketOptionKind { NoDelay, /// Read [`SocketOption::KeepAlive`], including the schedule the OS holds. KeepAlive, + /// Read only the keep-alive switch. + KeepAliveEnabled, + /// Read the probe schedule, even when keep-alive is disabled. + KeepAliveSettings, /// Read [`SocketOption::Linger`]. Linger, /// Read [`SocketOption::RecvBufferSize`] as the OS actually kept it. From 57588d97e25b99c60e0abf8ad218738726324039 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Sun, 8 Nov 2026 07:27:46 +0100 Subject: [PATCH 2/4] feat(turnloop-http): preserve header spelling; HTTP/2 PUSH_PROMISE, named SETTINGS and PING - Header::new_preserving_case: HTTP/1 keeps the field and trailer name spelling the caller gave, without weakening validation. - HTTP/2 SETTINGS carry only the options the caller named, and any HEADER_TABLE_SIZE the peer advertises is honoured (hpack set_allowed). - PUSH_PROMISE is handled in the core; promised stream ids are validated. - SETTINGS and PING acknowledgements report what they measured; ping() validates its arguments, and an unsolicited PING acknowledgement is an INTERNAL_ERROR. Supersedes #124. --- protocols/turnloop-http/src/hpack.rs | 5 + protocols/turnloop-http/src/http1.rs | 55 +++-- protocols/turnloop-http/src/http2.rs | 227 ++++++++++++++++-- protocols/turnloop-http/tests/codecs.rs | 5 + .../turnloop-http/tests/http2_settings.rs | 122 +++++++++- 5 files changed, 357 insertions(+), 57 deletions(-) diff --git a/protocols/turnloop-http/src/hpack.rs b/protocols/turnloop-http/src/hpack.rs index 254fd879..f932e52c 100644 --- a/protocols/turnloop-http/src/hpack.rs +++ b/protocols/turnloop-http/src/hpack.rs @@ -295,6 +295,11 @@ impl Decoder { failed: false, } } + /// The largest dynamic table size the peer's encoder may signal: the + /// HEADER_TABLE_SIZE this endpoint has advertised. + pub fn set_allowed(&mut self, size: usize) { + self.allowed = size; + } pub fn decode(&mut self, input: &[u8], out: &mut Vec
) -> Result<()> { if self.failed { return Err(bad()); diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index 181ac7bf..ccd2b58e 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -30,6 +30,14 @@ impl Header { value: value.as_ref().to_vec(), } } + /// Preserve an HTTP/1 field's original spelling. Lookups remain case-insensitive. + /// HTTP/2 callers must continue to supply lowercase names. + pub fn new_preserving_case(name: &str, value: impl AsRef<[u8]>) -> Self { + Self { + name: name.to_owned(), + value: value.as_ref().to_vec(), + } + } } #[derive(Debug, Clone)] pub struct Head { @@ -224,7 +232,10 @@ fn headers(raw: &[httparse::Header<'_>], limits: Limits) -> Result> if raw.len() > limits.headers { return Err(Error::new("HPE_HEADER_OVERFLOW", "header count limit")); } - Ok(raw.iter().map(|h| Header::new(h.name, h.value)).collect()) + Ok(raw + .iter() + .map(|h| Header::new_preserving_case(h.name, h.value)) + .collect()) } fn lengths(head: &Head) -> Result<(Option, bool)> { let mut cl = None; @@ -397,7 +408,12 @@ impl Decoder { }; if self.mode == Mode::Request && head.version == 1 - && (head.headers.iter().filter(|h| h.name == "host").count() != 1 + && (head + .headers + .iter() + .filter(|h| h.name.eq_ignore_ascii_case("host")) + .count() + != 1 || head.get("host").is_none_or(|v| v.is_empty())) { return Err(invalid("HTTP/1.1 requires exactly one Host")); @@ -538,17 +554,7 @@ impl Decoder { return Err(invalid("incomplete trailers")); }; let trailers = headers(raw, self.limits)?; - if trailers.iter().any(|h| { - matches!( - h.name.as_str(), - "content-length" - | "transfer-encoding" - | "host" - | "connection" - | "trailer" - | "upgrade" - ) - }) { + if trailers.iter().any(|h| forbidden_trailer(&h.name)) { return Err(invalid("forbidden trailer")); } step.consumed = n; @@ -749,15 +755,7 @@ impl Encoder { return Err(invalid("trailers require chunked body")); } for h in trailers { - if matches!( - h.name.as_str(), - "content-length" - | "transfer-encoding" - | "host" - | "connection" - | "trailer" - | "upgrade" - ) { + if forbidden_trailer(&h.name) { return Err(invalid("forbidden trailer")); } validate_field(h)?; @@ -777,6 +775,19 @@ impl Encoder { } } +fn forbidden_trailer(name: &str) -> bool { + [ + "content-length", + "transfer-encoding", + "host", + "connection", + "trailer", + "upgrade", + ] + .iter() + .any(|forbidden| name.eq_ignore_ascii_case(forbidden)) +} + /// RFC token grammar, shared with HPACK field validation without temporary types. pub(crate) fn valid_token(value: &[u8]) -> bool { !value.is_empty() diff --git a/protocols/turnloop-http/src/http2.rs b/protocols/turnloop-http/src/http2.rs index f5b1d02b..c0bdc6b1 100644 --- a/protocols/turnloop-http/src/http2.rs +++ b/protocols/turnloop-http/src/http2.rs @@ -98,15 +98,16 @@ impl Settings { payload } /// Refuse what this endpoint cannot honour once the peer acts on it. - fn validate(&self) -> Result<()> { + fn validate(&self, role: Role) -> Result<()> { for (id, value) in self.iter() { let valid = match id { - // The HPACK decoder's dynamic table is fixed at 4096 octets; - // advertising more would let the peer's encoder outgrow it. - Self::HEADER_TABLE_SIZE => value <= 4096, - // Server push is not implemented: PUSH_PROMISE is always a - // connection error, so neither side may invite one. - Self::ENABLE_PUSH => value == 0, + Self::HEADER_TABLE_SIZE => true, + // Only a client may invite pushes (RFC 9113 section 6.5.2); a + // server never sends a non-zero ENABLE_PUSH. + Self::ENABLE_PUSH => match role { + Role::Client => value <= 1, + Role::Server => value == 0, + }, Self::INITIAL_WINDOW_SIZE => value <= 0x7fffffff, Self::MAX_FRAME_SIZE => (16384..=0xffffff).contains(&value), Self::ENABLE_CONNECT_PROTOCOL => value <= 1, @@ -238,6 +239,17 @@ pub enum Event<'a> { bytes: &'a [u8], end_stream: bool, }, + /// The server promised a response to the request `headers` (RFC 9113 + /// section 8.4), on `stream`, a stream this endpoint opened. `promised` is + /// the server-initiated stream that will carry the response: its HEADERS + /// and DATA arrive as ordinary events, and [`Connection::reset`] declines + /// it. Only a client receives this, and only while it has not disabled push + /// with ENABLE_PUSH = 0. + PushPromise { + stream: u32, + promised: u32, + headers: Vec
, + }, /// The stream is over. It may name a stream the host never saw opened: a /// stream past the concurrent-stream limit is refused with /// RST_STREAM(REFUSED_STREAM) before any [`Event::Headers`] for it is @@ -398,6 +410,11 @@ pub struct Connection { acked_limits: Limits, /// Our SETTINGS_INITIAL_WINDOW_SIZE as acknowledged, and as enforced. acked_recv_initial: i64, + acked_table: usize, + /// Whether this endpoint accepts PUSH_PROMISE: ENABLE_PUSH as acknowledged, + /// loosened by every outstanding frame, defaulting to enabled. + acked_push: bool, + push_allowed: bool, recv_initial: i64, decoder: hpack::Decoder, encoder: hpack::Encoder, @@ -408,6 +425,13 @@ pub struct Connection { /// belong to a live stream. HPACK is connection state, so such a block /// still has to advance the decoder before it is thrown away. disposal: Option, + /// PINGs sent and not yet acknowledged. An acknowledgement with none + /// outstanding answers nothing this endpoint sent. + pings: usize, + /// The PUSH_PROMISE whose header block is being assembled: the promised + /// stream and whether the block is only to be decoded and dropped (the + /// associated stream is gone). + promise: Option<(u32, bool)>, output: Vec, output_pos: usize, draining: bool, @@ -417,7 +441,7 @@ pub struct Connection { impl Connection { /// A connection whose initial SETTINGS advertises `limits`: the /// concurrent-stream, frame-size and header-list limits, in that - /// (ascending) identifier order, plus ENABLE_PUSH = 0 for a client. + /// (ascending) identifier order. pub fn new(role: Role, limits: Limits) -> Result { Self::with_settings(role, limits, &Settings::from_limits(&limits)) } @@ -428,10 +452,9 @@ impl Connection { /// `limits` is what this endpoint enforces on its own. A parameter in /// `settings` that corresponds to a limit replaces it once the peer /// acknowledges the frame (and at once, if that relaxes it: the peer may - /// act on the new value before its ack reaches us). One parameter is - /// added: a client that does not name ENABLE_PUSH sends ENABLE_PUSH = 0, - /// because push defaults to enabled and this crate treats PUSH_PROMISE - /// as a connection error. + /// act on the new value before its ack reaches us). A client that + /// does not name ENABLE_PUSH leaves push enabled, as the protocol defaults + /// it, and surfaces each promise as [`Event::PushPromise`]. pub fn with_settings(role: Role, limits: Limits, settings: &Settings) -> Result { if !(16384..=0xffffff).contains(&limits.frame_size) || limits.streams == 0 @@ -439,11 +462,8 @@ impl Connection { { return Err(protocol("invalid limits")); } - let mut settings = settings.clone(); - if role == Role::Client && settings.get(Settings::ENABLE_PUSH).is_none() { - settings.set(Settings::ENABLE_PUSH, 0); - } - settings.validate()?; + settings.validate(role)?; + let settings = settings.clone(); let mut result = Self { role, limits, @@ -463,6 +483,9 @@ impl Connection { remote_settings: Settings::new(), acked_limits: limits, acked_recv_initial: 65535, + acked_table: 4096, + acked_push: true, + push_allowed: true, recv_initial: 65535, decoder: hpack::Decoder::new(4096, limits.header_list), encoder: hpack::Encoder::new(4096), @@ -470,6 +493,8 @@ impl Connection { scratch: Vec::new(), continuation: None, disposal: None, + pings: 0, + promise: None, output: Vec::new(), output_pos: 0, draining: false, @@ -495,13 +520,13 @@ impl Connection { /// peer sent under the old value are never treated as errors. /// /// Values this endpoint could not honour are refused before anything is - /// sent: ENABLE_PUSH other than 0, HEADER_TABLE_SIZE above 4096, and the + /// sent: ENABLE_PUSH above 1 from a client, anything but 0 from a server, and the /// out-of-range values RFC 9113 section 6.5.2 forbids. pub fn settings(&mut self, settings: &Settings) -> Result<()> { if self.failed { return Err(protocol("failed connection")); } - settings.validate()?; + settings.validate(self.role)?; self.send_settings(settings.clone()) } fn send_settings(&mut self, settings: Settings) -> Result<()> { @@ -519,6 +544,8 @@ impl Connection { fn enforce_settings(&mut self) { let mut limits = self.acked_limits; let mut recv_initial = self.acked_recv_initial; + let mut table = self.acked_table; + let mut push = self.acked_push; for settings in &self.settings_pending { for (id, value) in settings.iter() { let value = value as usize; @@ -529,6 +556,8 @@ impl Connection { Settings::MAX_HEADER_LIST_SIZE => { limits.header_list = limits.header_list.max(value) } + Settings::HEADER_TABLE_SIZE => table = table.max(value), + Settings::ENABLE_PUSH => push |= value != 0, _ => {} } } @@ -543,6 +572,8 @@ impl Connection { self.recv_initial = recv_initial; } self.decoder.max_list_size = limits.header_list; + self.decoder.set_allowed(table); + self.push_allowed = push; self.limits = limits; } /// What this endpoint enforces now, including the effect of SETTINGS it @@ -741,6 +772,66 @@ impl Connection { self.send_headers(id, headers, end_stream)?; Ok(id) } + /// Promise a response to the request `headers` on `stream`, a request + /// stream the client opened (RFC 9113 section 8.4). Returns the promised + /// stream, which then carries the response through + /// [`send_headers`](Self::send_headers) and [`send_data`](Self::send_data). + /// Refused when the peer disabled push with ENABLE_PUSH = 0. + pub fn push(&mut self, stream: u32, headers: &[Header]) -> Result { + if self.role != Role::Server || self.draining || self.failed { + return Err(protocol("cannot push")); + } + if self.remote_settings.get(Settings::ENABLE_PUSH) == Some(0) { + return Err(protocol("push disabled by peer")); + } + let i = self.index(stream).map_err(|_| self.gone(stream))?; + if self.streams[i].local_end { + return Err(error("STREAM_CLOSED", "local stream closed")); + } + if self.active() >= self.peer_streams { + return Err(error("REFUSED_STREAM", "peer stream limit")); + } + validate_headers(headers, true, false)?; + let id = self.next_id; + let next = id + .checked_add(2) + .filter(|n| *n <= 0x7fffffff) + .ok_or(protocol("stream IDs exhausted"))?; + self.scratch.clear(); + self.encoder.encode(headers, &mut self.scratch); + let first = self.scratch.len().min(self.peer_frame - 4); + let mut payload = id.to_be_bytes().to_vec(); + payload.extend_from_slice(&self.scratch[..first]); + let j = self.add_stream(id)?; + self.next_id = next; + encode_frame( + 5, + if first == self.scratch.len() { 4 } else { 0 }, + stream, + &payload, + &mut self.output, + )?; + let mut start = first; + while start < self.scratch.len() { + let end = (start + self.peer_frame).min(self.scratch.len()); + encode_frame( + 9, + if end == self.scratch.len() { 4 } else { 0 }, + stream, + &self.scratch[start..end], + &mut self.output, + )?; + start = end; + } + let s = &mut self.streams[j]; + // The promised request has no body and the client sends nothing on it. + s.remote_end = true; + s.received_head = true; + s.head_request = headers + .iter() + .any(|h| h.name == ":method" && h.value == b"HEAD"); + Ok(id) + } pub fn send_headers(&mut self, id: u32, headers: &[Header], end_stream: bool) -> Result<()> { if self.failed { return Err(protocol("failed connection")); @@ -902,7 +993,9 @@ impl Connection { self.frame(3, 0, id, &code.to_be_bytes()) } pub fn ping(&mut self, data: [u8; 8]) -> Result<()> { - self.frame(6, 0, 0, &data) + self.frame(6, 0, 0, &data)?; + self.pings += 1; + Ok(()) } /// Graceful close: GOAWAY(NO_ERROR) naming the last stream this connection /// has processed, then drain. A stream the peer opens before it sees the @@ -942,6 +1035,7 @@ impl Connection { self.failed = true; let code = match e.code { "FLOW_CONTROL_ERROR" => 3, + "INTERNAL_ERROR" => 2, "FRAME_SIZE_ERROR" => 6, "COMPRESSION_ERROR" => 9, "ENHANCE_YOUR_CALM" => 11, @@ -1178,6 +1272,8 @@ impl Connection { Settings::MAX_HEADER_LIST_SIZE => { self.acked_limits.header_list = value as usize } + Settings::HEADER_TABLE_SIZE => self.acked_table = value as usize, + Settings::ENABLE_PUSH => self.acked_push = value != 0, _ => {} } } @@ -1196,7 +1292,10 @@ impl Connection { match id { 1 => self.encoder.set_table_size((value as usize).min(4096)), 2 => { - if value > 1 || self.role == Role::Client { + if match self.role { + Role::Server => value > 1, + Role::Client => value != 0, + } { return Err(protocol("invalid ENABLE_PUSH")); } } @@ -1233,7 +1332,43 @@ impl Connection { step.event = Some(Event::Settings(settings)); } } - 5 => return Err(protocol("server push disabled")), + 5 => { + if self.role != Role::Client || !self.push_allowed { + return Err(protocol("server push disabled")); + } + if f.stream == 0 { + return Err(protocol("PUSH_PROMISE on connection")); + } + let payload = unpadded(f)?; + if payload.len() < 4 { + return Err(frame_error()); + } + let promised = u32be(payload) & 0x7fffffff; + if promised == 0 || !promised.is_multiple_of(2) || promised <= self.highest_remote { + return Err(protocol("invalid promised stream")); + } + // A promise rides on a stream we opened. One we have already + // reset may still have a promise in flight: decode it to keep + // HPACK in step and cancel the promised stream. + let discard = match self.index(f.stream) { + Ok(i) if self.streams[i].remote_end => { + return Err(error("STREAM_CLOSED", "PUSH_PROMISE after END_STREAM")); + } + Ok(_) => false, + Err(_) if self.seen(f.stream) => true, + Err(_) => return Err(protocol("PUSH_PROMISE on idle stream")), + }; + self.highest_remote = promised; + self.disposal = None; + self.block.clear(); + self.append_block(&payload[4..])?; + self.promise = Some((promised, discard)); + if f.flags & 4 != 0 { + step.event = self.finish_block(f.stream, false)?; + } else { + self.continuation = Some((f.stream, false, 0)); + } + } 6 => { if f.stream != 0 { return Err(protocol("PING on stream")); @@ -1245,6 +1380,12 @@ impl Connection { let ack = f.flags & 1 != 0; if !ack { self.frame(6, 1, 0, f.payload)?; + } else if self.pings == 0 { + // Node answers GOAWAY(INTERNAL_ERROR) and drops the + // connection (measured against a raw peer). + return Err(error("INTERNAL_ERROR", "unsolicited PING acknowledgement")); + } else { + self.pings -= 1; } step.event = Some(Event::Ping { ack, data }); } @@ -1319,6 +1460,9 @@ impl Connection { } /// Complete a header block: the stream's own, or one being thrown away. fn finish_block(&mut self, id: u32, end: bool) -> Result>> { + if let Some((promised, discard)) = self.promise.take() { + return self.finish_promise(id, promised, discard); + } let Some(disposal) = self.disposal.take() else { return self.finish_headers(id, end).map(Some); }; @@ -1335,6 +1479,45 @@ impl Connection { Disposal::Discard => Ok(None), } } + /// Complete a PUSH_PROMISE: reserve the promised stream, or decline it. + fn finish_promise( + &mut self, + id: u32, + promised: u32, + discard: bool, + ) -> Result>> { + let mut headers = Vec::new(); + self.decoder.decode(&self.block, &mut headers)?; + if discard { + self.frame(3, 0, promised, &8u32.to_be_bytes())?; + return Ok(None); + } + validate_headers(&headers, true, false)?; + match self.add_stream(promised) { + Ok(i) => { + let s = &mut self.streams[i]; + // The client never sends on a pushed stream. + s.local_end = true; + s.head_request = headers + .iter() + .any(|h| h.name == ":method" && h.value == b"HEAD"); + self.last_remote = promised; + Ok(Some(Event::PushPromise { + stream: id, + promised, + headers, + })) + } + Err(e) if e.code == "REFUSED_STREAM" => { + self.frame(3, 0, promised, &7u32.to_be_bytes())?; + Ok(Some(Event::Reset { + stream: promised, + code: 7, + })) + } + Err(e) => Err(e), + } + } fn finish_headers(&mut self, id: u32, end: bool) -> Result> { let mut headers = Vec::new(); self.decoder.decode(&self.block, &mut headers)?; diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index e61bea26..749e6d61 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -768,6 +768,11 @@ fn drive(to: &mut http2::Connection, input: &mut Vec) -> Result, http2::Event::Goaway { code, .. } => format!("Goaway code={code}"), http2::Event::Ping { ack, .. } => format!("Ping ack={ack}"), http2::Event::WindowUpdate { stream } => format!("WindowUpdate s={stream}"), + http2::Event::PushPromise { + stream, promised, .. + } => { + format!("PushPromise s={stream} p={promised}") + } }); } input.drain(..consumed); diff --git a/protocols/turnloop-http/tests/http2_settings.rs b/protocols/turnloop-http/tests/http2_settings.rs index 6daff431..804b3744 100644 --- a/protocols/turnloop-http/tests/http2_settings.rs +++ b/protocols/turnloop-http/tests/http2_settings.rs @@ -14,6 +14,7 @@ enum Seen { Headers(u32), Data(u32, usize), Reset(u32, u32), + Push(u32, u32), Goaway(u32, Option>), Other, } @@ -32,6 +33,9 @@ fn drive(to: &mut Connection, input: &mut Vec) -> Result, &'static Event::Headers { stream, .. } => Seen::Headers(stream), Event::Data { stream, bytes, .. } => Seen::Data(stream, bytes.len()), Event::Reset { stream, code } => Seen::Reset(stream, code), + Event::PushPromise { + stream, promised, .. + } => Seen::Push(stream, promised), Event::Goaway { code, debug, .. } => Seen::Goaway(code, debug.map(<[u8]>::to_vec)), _ => Seen::Other, }); @@ -90,12 +94,7 @@ fn settings_acks_are_an_ordered_queue() { let seen = drive(&mut server, &mut ship(&mut client)).unwrap(); assert_eq!( seen, - vec![Seen::Settings(vec![ - (2, 0), - (3, 100), - (5, 16384), - (6, 32768) - ])] + vec![Seen::Settings(vec![(3, 100), (5, 16384), (6, 32768)])] ); let seen = drive(&mut client, &mut ship(&mut server)).unwrap(); assert_eq!( @@ -104,7 +103,7 @@ fn settings_acks_are_an_ordered_queue() { Seen::Settings(DEFAULT_ADVERTISED.to_vec()), Seen::Settings(vec![(3, 7)]), Seen::Settings(vec![(6, 65536)]), - Seen::SettingsAck(vec![(2, 0), (3, 100), (5, 16384), (6, 32768)]), + Seen::SettingsAck(vec![(3, 100), (5, 16384), (6, 32768)]), ] ); // The client acked all three; the server pairs them with what it sent. @@ -243,20 +242,21 @@ fn initial_settings_are_host_chosen_and_ordered() { .unwrap(); assert_eq!(ordered.output(), expected); - // `new` still advertises its limits, now in ascending order for a client - // too: ENABLE_PUSH (2) first rather than appended. + // `new` still advertises its limits, in ascending order. A client leaves + // push enabled, as the protocol defaults it. let client = Connection::new(Role::Client, Limits::default()).unwrap(); let settings = &client.output()[http2::PREFACE.len()..]; let frame = http2::decode_frame(settings, 16384).unwrap().unwrap(); let ids: Vec = frame.payload.chunks(6).map(|s| s[1]).collect(); - assert_eq!(ids, [2, 3, 5, 6]); + assert_eq!(ids, [3, 5, 6]); - // A client that names no ENABLE_PUSH still refuses push: it defaults to on. + // A client that names no ENABLE_PUSH sends an empty SETTINGS and so keeps + // the protocol default, push on. let client = Connection::with_settings(Role::Client, Limits::default(), &Settings::new()).unwrap(); assert_eq!( &client.output()[http2::PREFACE.len()..], - [0, 0, 6, 4, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 0] + [0, 0, 0, 4, 0, 0, 0, 0, 0] ); // An empty-SETTINGS server still serves a request end to end. @@ -274,7 +274,6 @@ fn settings_it_cannot_honour_are_refused() { let (_, mut server) = handshake(); for (id, value) in [ (Settings::ENABLE_PUSH, 1), - (Settings::HEADER_TABLE_SIZE, 8192), (Settings::INITIAL_WINDOW_SIZE, 0x8000_0000), (Settings::MAX_FRAME_SIZE, 16383), (Settings::MAX_FRAME_SIZE, 0x0100_0000), @@ -292,6 +291,18 @@ fn settings_it_cannot_honour_are_refused() { assert_eq!(server.pending_settings(), 0); } +/// The HPACK decoder honours any advertised HEADER_TABLE_SIZE, so a larger +/// table is accepted and sent, not refused. +#[test] +fn a_larger_header_table_is_advertised() { + let (_, mut server) = handshake(); + let mut big = Settings::new(); + big.set(Settings::HEADER_TABLE_SIZE, 8192); + assert!(server.settings(&big).is_ok()); + assert_eq!(server.pending_settings(), 1); + assert!(Connection::with_settings(Role::Server, Limits::default(), &big).is_ok()); +} + /// Item 6. A live change applies to the running connection: a loosening at /// once (the peer may use it before its ack arrives), a tightening on the ack. #[test] @@ -418,3 +429,88 @@ fn live_initial_window_change_adjusts_streams() { http2::encode_frame(0, 0, a, &[0], &mut wire).unwrap(); assert_eq!(drive(&mut server, &mut wire), Err("FLOW_CONTROL_ERROR")); } + +fn promised_request() -> Vec
{ + vec![ + Header::new(":method", "GET"), + Header::new(":scheme", "http"), + Header::new(":path", "/pushed"), + Header::new(":authority", "localhost"), + ] +} +fn ok() -> Vec
{ + vec![Header::new(":status", "200")] +} + +/// A server promises a response on a client's request; the client sees the +/// promise first, then the pushed response on the promised stream. +#[test] +fn a_client_receives_a_pushed_response() { + let (mut client, mut server) = handshake(); + let id = client.open(&request(), true).unwrap(); + assert_eq!( + drive(&mut server, &mut ship(&mut client)).unwrap(), + vec![Seen::Headers(id)] + ); + let promised = server.push(id, &promised_request()).unwrap(); + assert_eq!(promised, 2); + server.send_headers(promised, &ok(), true).unwrap(); + server.send_headers(id, &ok(), true).unwrap(); + assert_eq!( + drive(&mut client, &mut ship(&mut server)).unwrap(), + vec![ + Seen::Push(id, promised), + Seen::Headers(promised), + Seen::Headers(id) + ] + ); +} + +/// A client may decline a promise; the server sees the reset. +#[test] +fn a_client_can_decline_a_promise() { + let (mut client, mut server) = handshake(); + let id = client.open(&request(), true).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + let promised = server.push(id, &promised_request()).unwrap(); + assert_eq!( + drive(&mut client, &mut ship(&mut server)).unwrap(), + vec![Seen::Push(id, promised)] + ); + client.reset(promised, 8).unwrap(); + assert_eq!( + drive(&mut server, &mut ship(&mut client)).unwrap(), + vec![Seen::Reset(promised, 8)] + ); +} + +/// A client that sent ENABLE_PUSH = 0 cannot be promised anything. +#[test] +fn push_is_refused_once_the_client_disabled_it() { + let mut off = Settings::new(); + off.set(Settings::ENABLE_PUSH, 0); + let mut client = Connection::with_settings(Role::Client, Limits::default(), &off).unwrap(); + let mut server = Connection::new(Role::Server, Limits::default()).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + drive(&mut client, &mut ship(&mut server)).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + let id = client.open(&request(), true).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + assert!(server.push(id, &promised_request()).is_err()); +} + +/// A PING acknowledgement with no PING outstanding is an INTERNAL_ERROR. +#[test] +fn an_unsolicited_ping_ack_fails_the_connection() { + let (mut client, mut server) = handshake(); + client.ping([7; 8]).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + let wire = ship(&mut server); + drive(&mut client, &mut wire.clone()).unwrap(); + assert_eq!( + drive(&mut client, &mut wire.clone()).unwrap_err(), + "INTERNAL_ERROR" + ); + let goaway = ship(&mut client); + assert_eq!(goaway[goaway.len() - 4..], 2u32.to_be_bytes()); +} From e23c6fa0946ecfa5c308262dbfe7cdf86ae41dd5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Sun, 8 Nov 2026 07:39:50 +0100 Subject: [PATCH 3/4] fix(turnloop): drop the IOCP test helpers the merge duplicated --- crates/turnloop/src/backend/iocp/mod.rs | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index 6bcdc3d9..cc2dbd16 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -1936,21 +1936,3 @@ pub(crate) fn unbound_tcp_for_test(address: SocketAddr) -> Detached { stream.set_nodelay(true).unwrap(); Detached::from_socket(stream.into()).unwrap() } - -/// Construct an explicitly bound socket for the portable adoption contract. -#[cfg(test)] -pub(crate) fn bound_tcp_for_test(address: SocketAddr) -> (Detached, SocketAddr) { - let socket = socket::create(address.is_ipv6(), false).unwrap(); - let raw = socket.as_raw_socket() as usize; - socket::bind_to(raw, address).unwrap(); - let local = socket::address(raw, false).unwrap(); - (Detached::from_socket(socket).unwrap(), local) -} - -/// Unbound socket with a non-default kernel option for adoption tests. -#[cfg(test)] -pub(crate) fn unbound_tcp_for_test(address: SocketAddr) -> Detached { - let stream = std::net::TcpStream::from(socket::create(address.is_ipv6(), false).unwrap()); - stream.set_nodelay(true).unwrap(); - Detached::from_socket(stream.into()).unwrap() -} From 38a8c3cdc240dc738c34fa46094401fa09972851 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Sun, 8 Nov 2026 07:42:00 +0100 Subject: [PATCH 4/4] fix(turnloop-http): reject an unsolicited PING acknowledgement only on request RFC 9113 requires no reaction to a PING ACK that answers nothing this endpoint sent, and h2spec's strict 6.7/2 requires the connection to survive it. The default reports it as Event::Ping again, as main did; Connection::set_reject_unsolicited_ping_ack(true) keeps Node's policy of failing the connection with GOAWAY(INTERNAL_ERROR). --- protocols/turnloop-http/src/http2.rs | 16 ++++++++++--- .../turnloop-http/tests/http2_settings.rs | 23 +++++++++++++++++-- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/protocols/turnloop-http/src/http2.rs b/protocols/turnloop-http/src/http2.rs index c0bdc6b1..38ab9ba8 100644 --- a/protocols/turnloop-http/src/http2.rs +++ b/protocols/turnloop-http/src/http2.rs @@ -428,6 +428,8 @@ pub struct Connection { /// PINGs sent and not yet acknowledged. An acknowledgement with none /// outstanding answers nothing this endpoint sent. pings: usize, + /// Whether an acknowledgement with no PING outstanding fails the connection. + reject_unsolicited_ping_ack: bool, /// The PUSH_PROMISE whose header block is being assembled: the promised /// stream and whether the block is only to be decoded and dropped (the /// associated stream is gone). @@ -494,6 +496,7 @@ impl Connection { continuation: None, disposal: None, pings: 0, + reject_unsolicited_ping_ack: false, promise: None, output: Vec::new(), output_pos: 0, @@ -992,6 +995,13 @@ impl Connection { self.streams.swap_remove(i); self.frame(3, 0, id, &code.to_be_bytes()) } + /// RFC 9113 leaves a PING acknowledgement that answers nothing this + /// endpoint sent without a required reaction, and by default it is only + /// reported as [`Event::Ping`]. With `reject` set it fails the connection + /// with GOAWAY(INTERNAL_ERROR), as Node does: no legitimate peer sends one. + pub fn set_reject_unsolicited_ping_ack(&mut self, reject: bool) { + self.reject_unsolicited_ping_ack = reject; + } pub fn ping(&mut self, data: [u8; 8]) -> Result<()> { self.frame(6, 0, 0, &data)?; self.pings += 1; @@ -1380,12 +1390,12 @@ impl Connection { let ack = f.flags & 1 != 0; if !ack { self.frame(6, 1, 0, f.payload)?; - } else if self.pings == 0 { + } else if self.pings > 0 { + self.pings -= 1; + } else if self.reject_unsolicited_ping_ack { // Node answers GOAWAY(INTERNAL_ERROR) and drops the // connection (measured against a raw peer). return Err(error("INTERNAL_ERROR", "unsolicited PING acknowledgement")); - } else { - self.pings -= 1; } step.event = Some(Event::Ping { ack, data }); } diff --git a/protocols/turnloop-http/tests/http2_settings.rs b/protocols/turnloop-http/tests/http2_settings.rs index 804b3744..89f721d9 100644 --- a/protocols/turnloop-http/tests/http2_settings.rs +++ b/protocols/turnloop-http/tests/http2_settings.rs @@ -499,14 +499,33 @@ fn push_is_refused_once_the_client_disabled_it() { assert!(server.push(id, &promised_request()).is_err()); } -/// A PING acknowledgement with no PING outstanding is an INTERNAL_ERROR. +/// By default a PING acknowledgement with no PING outstanding is only +/// reported (RFC 9113 ยง6.7; h2spec 6.7/2 requires the connection to survive). #[test] -fn an_unsolicited_ping_ack_fails_the_connection() { +fn an_unsolicited_ping_ack_is_reported_by_default() { let (mut client, mut server) = handshake(); client.ping([7; 8]).unwrap(); drive(&mut server, &mut ship(&mut client)).unwrap(); let wire = ship(&mut server); drive(&mut client, &mut wire.clone()).unwrap(); + let events = drive(&mut client, &mut wire.clone()).unwrap(); + assert_eq!(events, [Seen::Other], "only the PING event"); + assert!(ship(&mut client).is_empty(), "no GOAWAY and no reply"); + client.ping([8; 8]).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + drive(&mut client, &mut ship(&mut server)).unwrap(); +} + +/// With the Node policy on, a PING acknowledgement with no PING outstanding is +/// an INTERNAL_ERROR. +#[test] +fn an_unsolicited_ping_ack_fails_the_connection_when_rejected() { + let (mut client, mut server) = handshake(); + client.set_reject_unsolicited_ping_ack(true); + client.ping([7; 8]).unwrap(); + drive(&mut server, &mut ship(&mut client)).unwrap(); + let wire = ship(&mut server); + drive(&mut client, &mut wire.clone()).unwrap(); assert_eq!( drive(&mut client, &mut wire.clone()).unwrap_err(), "INTERNAL_ERROR"