From 2e64afaff2404310f0a6619aecc6471f4878325b Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 18:21:33 -0500 Subject: [PATCH 1/4] fix(h1): record when a file region has been sent A region's log ended at "sending a file region" whether the socket took it whole or the send never finished, so a server's record could not tell a body that went out from one that did not. The connection now records the region's end as well. Refs #159 --- h1/src/conn.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/h1/src/conn.rs b/h1/src/conn.rs index 41f8240..e552780 100644 --- a/h1/src/conn.rs +++ b/h1/src/conn.rs @@ -312,6 +312,9 @@ async fn send_region( tracing::trace!(offset, len, "sending a file region"); let BufResult(result, _file) = writer.send_file_all(file, offset, size).await; result?; + // With the record above, a log tells a region the socket took whole + // from one that never finished. + tracing::trace!(offset, len, "sent a file region"); outbox.extend(encoder.region_end()); Ok(()) } From a601ef875bf266f93e0c2b3d438f88417be0c87d Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 18:34:20 -0500 Subject: [PATCH 2/4] chore(io): check a compio stream's descriptor still names its socket A file region on compio once reached the peer as a 200 head, no body and a clean FIN. The head and the shutdown name the socket by its slot in the ring's file table, while the body's splice names it by descriptor number; both reaching the peer except the body fits a number closed and reused while the slot kept the socket open. Debug builds now record the socket's device and inode when a stream is made and compare them before every use of the number: each raw splice, FIONBIO, the AsFd handed to kTLS and to callers, and each operation submitted through the slot holder. A mismatch panics naming what the number now holds, so the next failure points at the reuse instead of losing bytes. Refs #159 --- io/src/compio.rs | 64 ++++++++++++++++++++++++++++----------- io/src/sys/unix.rs | 75 ++++++++++++++++++++++++++++++++++++++++++++++ io/tests/io.rs | 38 +++++++++++++++++++++++ 3 files changed, 160 insertions(+), 17 deletions(-) diff --git a/io/src/compio.rs b/io/src/compio.rs index 175ffd7..620b907 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -435,6 +435,10 @@ impl_socket!(net::UnixStream, zero_copy: false); /// A connected stream socket. pub struct Stream { inner: S, + /// The socket `inner`'s descriptor names, which every use of the + /// descriptor's number checks it still does (debug builds). + #[cfg(all(unix, debug_assertions))] + identity: crate::sys::unix::Identity, /// The pipe file transfers splice through, kept between transfers. A /// transfer takes it out while data may sit in it, so one that fails or /// is cancelled midway drops it instead of leaving stale bytes behind. @@ -472,6 +476,9 @@ impl Stream { #[cfg(not(target_os = "linux"))] let _ = io; Self { + #[cfg(all(unix, debug_assertions))] + identity: crate::sys::unix::Identity::of(inner.as_fd()) + .expect("a stream's socket is open when the stream is made"), #[cfg(target_os = "linux")] file: files::File::new(inner.shared_fd(), io.file_table()), inner, @@ -494,13 +501,37 @@ impl Stream { async fn submit(&self, op: T) -> BufResult { #[cfg(target_os = "linux")] { - self.file.submit(op).await + self.file().submit(op).await } #[cfg(not(target_os = "linux"))] { + self.fd(); ::compio::runtime::submit(op).await } } + + /// The socket's descriptor, for a call that names it by its number. + /// + /// Debug builds check first that the number still names the stream's + /// socket, and panic if it does not (zen-rs/http-kit#159). + #[cfg(unix)] + #[track_caller] + fn fd(&self) -> std::os::fd::BorrowedFd<'_> { + let fd = self.inner.as_fd(); + #[cfg(debug_assertions)] + self.identity.check(fd); + fd + } + + /// The socket's hold on its slot, through which every operation is + /// submitted: one naming the descriptor while the socket has no slot is + /// a use of its number, checked as [`fd`](Self::fd) checks it. + #[cfg(target_os = "linux")] + #[track_caller] + fn file(&self) -> &files::File { + self.fd(); + &self.file + } } impl Stream { @@ -533,7 +564,7 @@ impl Stream { if self.recv_empty.get() { op.poll_first(); } - let (BufResult(res, op), extra) = self.file.submit_with_extra(op).await; + let (BufResult(res, op), extra) = self.file().submit_with_extra(op).await; if let Some(Ok(nonempty)) = extra.map(|extra| extra.sock_nonempty()) { self.recv_empty.set(!nonempty); } @@ -550,7 +581,7 @@ impl Stream { use compio_buf::IntoInner as _; let op = Send::new(self.inner.shared_fd(), buf, SendFlags::empty()); - let BufResult(res, op) = self.file.submit(op).await; + let BufResult(res, op) = self.file().submit(op).await; BufResult(res, op.into_inner()) } @@ -559,7 +590,7 @@ impl Stream { use compio_buf::IntoInner as _; let op = SendVectored::new(self.inner.shared_fd(), buf, SendFlags::empty()); - let BufResult(res, op) = self.file.submit(op).await; + let BufResult(res, op) = self.file().submit(op).await; BufResult(res, op.into_inner()) } @@ -567,7 +598,7 @@ impl Stream { use ::compio::driver::op::ShutdownSocket; let op = ShutdownSocket::new(self.inner.shared_fd(), std::net::Shutdown::Write); - match self.file.submit(op).await.0 { + match self.file().submit(op).await.0 { // A socket that is not connected has nothing to shut down, as // compio's stream has it. Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()), @@ -638,7 +669,10 @@ impl fmt::Debug for Stream { #[cfg(unix)] impl std::os::fd::AsFd for Stream { fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> { - self.inner.as_fd() + let fd = self.inner.as_fd(); + #[cfg(debug_assertions)] + self.identity.check(fd); + fd } } @@ -651,7 +685,7 @@ impl SharedStream for Stream { async fn read(&self, buf: B) -> BufResult { #[cfg(target_os = "linux")] { - let buf = match self.recv.read_ahead(&self.inner, &self.file, buf).await { + let buf = match self.recv.read_ahead(&self.inner, self.file(), buf).await { recv::Copied::Done(result) => return result, recv::Copied::ReadInstead(buf) => buf, }; @@ -666,7 +700,7 @@ impl SharedStream for Stream { async fn read_owned(&self) -> io::Result> { #[cfg(target_os = "linux")] { - self.recv.read_owned(&self.inner, &self.file).await + self.recv.read_owned(&self.inner, self.file()).await } #[cfg(not(target_os = "linux"))] { @@ -917,11 +951,7 @@ mod splice { stream.pipe.set(Some(pipe)); return BufResult(Ok(filled), file); } - match crate::sys::pipe::splice( - pipe.read.as_fd(), - stream.inner.as_fd(), - filled - drained, - ) { + match crate::sys::pipe::splice(pipe.read.as_fd(), stream.fd(), filled - drained) { Ok(0) => { break io::Error::new(io::ErrorKind::WriteZero, "stream accepted no bytes"); } @@ -950,7 +980,7 @@ mod splice { /// Makes the socket `O_NONBLOCK`, once. fn nonblocking(stream: &Stream) -> io::Result<()> { if !stream.nonblocking.get() { - crate::sys::pipe::set_nonblocking(stream.inner.as_fd())?; + crate::sys::pipe::set_nonblocking(stream.fd())?; stream.nonblocking.set(true); } Ok(()) @@ -994,7 +1024,7 @@ mod splice { } let len = max.min(pipe.capacity()); loop { - match crate::sys::pipe::splice(stream.inner.as_fd(), pipe.write_fd(), len) { + match crate::sys::pipe::splice(stream.fd(), pipe.write_fd(), len) { Ok(0) => return Ok(Spliced::End(pipe)), Ok(n) => return Ok(Spliced::Filled(FilledPipe::new(pipe, n))), Err(e) if e.kind() == io::ErrorKind::WouldBlock => { @@ -1015,7 +1045,7 @@ mod splice { nonblocking(stream)?; let mut left = filled.len(); while left > 0 { - match crate::sys::pipe::splice(filled.pipe().read_fd(), stream.inner.as_fd(), left) { + match crate::sys::pipe::splice(filled.pipe().read_fd(), stream.fd(), left) { Ok(0) => { return Err(io::Error::new( io::ErrorKind::WriteZero, @@ -1515,7 +1545,7 @@ mod zero_copy { let rest = Shared(Rc::clone(&buf)).slice(written); let res = if zero_copy { let op = SendVectoredZc::new(stream.inner.shared_fd(), rest, flags); - match stream.file.address(op) { + match stream.file().address(op) { Ok(op) => send(op).await, Err((e, _)) => Err(e), } diff --git a/io/src/sys/unix.rs b/io/src/sys/unix.rs index 1618259..8972d32 100644 --- a/io/src/sys/unix.rs +++ b/io/src/sys/unix.rs @@ -32,3 +32,78 @@ pub(crate) fn local_addr(fd: impl AsFd) -> io::Result { pub(crate) fn peer_addr(fd: impl AsFd) -> io::Result { UnixStream::from(duplicate(fd)?).peer_addr() } + +/// The open file a descriptor names, as `fstat(2)` identifies it: its device +/// and inode, which stay the same for as long as the descriptor names that +/// file and differ for any other file open at the same time. +/// +/// A stream records its socket's at creation and compares before each use +/// of the descriptor's number, so that a number closed and reused behind the +/// stream's back (zen-rs/http-kit#159) fails at the use instead of moving +/// bytes to whatever file now holds it. It costs an `fstat` per use, so only +/// debug builds keep it. +#[cfg(all(feature = "compio", debug_assertions))] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Identity { + dev: libc::dev_t, + ino: libc::ino_t, + /// The file type bits of `st_mode`, which name what a reused number + /// now holds. + kind: libc::mode_t, +} + +#[cfg(all(feature = "compio", debug_assertions))] +impl Identity { + /// The identity of the file `fd` names. + pub(crate) fn of(fd: std::os::fd::BorrowedFd<'_>) -> io::Result { + use std::os::fd::AsRawFd as _; + + let mut stat = std::mem::MaybeUninit::::uninit(); + // SAFETY: `stat` is valid for the write of one `struct stat`, and + // `fd` is borrowed, so the call reads a descriptor that is open. + if unsafe { libc::fstat(fd.as_raw_fd(), stat.as_mut_ptr()) } != 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: `fstat` succeeded, so it filled `stat`. + let stat = unsafe { stat.assume_init() }; + Ok(Self { + dev: stat.st_dev, + ino: stat.st_ino, + kind: stat.st_mode & libc::S_IFMT, + }) + } + + /// Panics unless `fd` still names the file `self` identifies. + #[track_caller] + pub(crate) fn check(self, fd: std::os::fd::BorrowedFd<'_>) { + use std::os::fd::AsRawFd as _; + + match Self::of(fd) { + Ok(now) if now == self => {} + Ok(now) => panic!( + "descriptor {} no longer names the stream's socket ({self:?}): it names a {} \ + ({now:?}), so the number was closed and reused while the stream held it", + fd.as_raw_fd(), + now.kind_name(), + ), + Err(error) => panic!( + "descriptor {} no longer names the stream's socket ({self:?}): {error}, so the \ + number was closed while the stream held it", + fd.as_raw_fd(), + ), + } + } + + fn kind_name(self) -> &'static str { + match self.kind { + libc::S_IFSOCK => "socket", + libc::S_IFREG => "regular file", + libc::S_IFIFO => "pipe", + libc::S_IFCHR => "character device", + libc::S_IFDIR => "directory", + libc::S_IFBLK => "block device", + libc::S_IFLNK => "symbolic link", + _ => "file of an unknown type", + } + } +} diff --git a/io/tests/io.rs b/io/tests/io.rs index 089ea52..d329ba1 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -905,6 +905,44 @@ mod compio { server.join().expect("server thread"); } + /// A stream whose descriptor number is closed and reused behind its back + /// (here by `dup2` of a regular file over it, which does both at once) + /// fails its next use of the number in a debug build instead of moving + /// bytes into the file now holding it (zen-rs/http-kit#159). On Linux the + /// use is a file region's splice into the socket. + #[cfg(all(unix, debug_assertions))] + #[test] + #[should_panic(expected = "no longer names the stream's socket")] + fn a_reused_descriptor_number_fails_its_next_use() { + use http_kit_io::compio::Compio; + use std::os::fd::{AsFd as _, AsRawFd as _}; + + let listener = std::net::TcpListener::bind(super::loopback()).expect("bind"); + let client = std::net::TcpStream::connect(listener.local_addr().expect("local addr")) + .expect("connect"); + let (accepted, _) = listener.accept().expect("accept"); + let file = super::File::open(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/io.rs")) + .expect("open this test's source"); + run(async move { + let stream = Compio::default() + .tcp_stream(::compio::net::TcpStream::from_std(accepted).expect("adopt")); + let number = stream.as_fd().as_raw_fd(); + // SAFETY: both descriptors are open; `dup2` replaces the + // stream's number with a duplicate of the file's. + assert_eq!(unsafe { libc::dup2(file.as_raw_fd(), number) }, number); + #[cfg(target_os = "linux")] + { + let mut stream = stream; + let _ = stream.send_file(super::Rc::new(file), 0, 64).await; + } + #[cfg(not(target_os = "linux"))] + { + let _ = stream.as_fd(); + } + }); + drop(client); + } + /// How many submission queue entries the current thread's io_uring has /// been handed: the tail the kernel reports in the ring's fdinfo. #[cfg(target_os = "linux")] From ed6bdc63b9fca425ac64b305c3dcdde6af63a59e Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 18:35:41 -0500 Subject: [PATCH 3/4] test(io): bring SendFile into scope for the reused-descriptor test on Linux Refs #159 --- io/tests/io.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/io/tests/io.rs b/io/tests/io.rs index d329ba1..544d816 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -932,6 +932,8 @@ mod compio { assert_eq!(unsafe { libc::dup2(file.as_raw_fd(), number) }, number); #[cfg(target_os = "linux")] { + use super::SendFile as _; + let mut stream = stream; let _ = stream.send_file(super::Rc::new(file), 0, 64).await; } From b32959a6da71ae0ebc39f837b6f3b8c71046d9c9 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 18:37:44 -0500 Subject: [PATCH 4/4] fix(io): check the descriptor only where an operation names it An operation submitted while the stream holds its slot names the socket by the slot, so the number may legitimately point elsewhere then (the registered-slot test points it at /dev/null). The identity check moves into the slot holder and runs only for an operation that names the descriptor; raw syscalls and AsFd keep theirs. Refs #159 --- io/src/compio.rs | 66 +++++++++++++++++++++++++++------------------- io/src/sys/unix.rs | 23 +++++++--------- 2 files changed, 48 insertions(+), 41 deletions(-) diff --git a/io/src/compio.rs b/io/src/compio.rs index 620b907..b9ad0b7 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -477,8 +477,10 @@ impl Stream { let _ = io; Self { #[cfg(all(unix, debug_assertions))] - identity: crate::sys::unix::Identity::of(inner.as_fd()) - .expect("a stream's socket is open when the stream is made"), + identity: crate::sys::unix::Identity::of(std::os::fd::AsRawFd::as_raw_fd( + &inner.as_fd(), + )) + .expect("a stream's socket is open when the stream is made"), #[cfg(target_os = "linux")] file: files::File::new(inner.shared_fd(), io.file_table()), inner, @@ -501,7 +503,7 @@ impl Stream { async fn submit(&self, op: T) -> BufResult { #[cfg(target_os = "linux")] { - self.file().submit(op).await + self.file.submit(op).await } #[cfg(not(target_os = "linux"))] { @@ -519,19 +521,9 @@ impl Stream { fn fd(&self) -> std::os::fd::BorrowedFd<'_> { let fd = self.inner.as_fd(); #[cfg(debug_assertions)] - self.identity.check(fd); + self.identity.check(std::os::fd::AsRawFd::as_raw_fd(&fd)); fd } - - /// The socket's hold on its slot, through which every operation is - /// submitted: one naming the descriptor while the socket has no slot is - /// a use of its number, checked as [`fd`](Self::fd) checks it. - #[cfg(target_os = "linux")] - #[track_caller] - fn file(&self) -> &files::File { - self.fd(); - &self.file - } } impl Stream { @@ -564,7 +556,7 @@ impl Stream { if self.recv_empty.get() { op.poll_first(); } - let (BufResult(res, op), extra) = self.file().submit_with_extra(op).await; + let (BufResult(res, op), extra) = self.file.submit_with_extra(op).await; if let Some(Ok(nonempty)) = extra.map(|extra| extra.sock_nonempty()) { self.recv_empty.set(!nonempty); } @@ -581,7 +573,7 @@ impl Stream { use compio_buf::IntoInner as _; let op = Send::new(self.inner.shared_fd(), buf, SendFlags::empty()); - let BufResult(res, op) = self.file().submit(op).await; + let BufResult(res, op) = self.file.submit(op).await; BufResult(res, op.into_inner()) } @@ -590,7 +582,7 @@ impl Stream { use compio_buf::IntoInner as _; let op = SendVectored::new(self.inner.shared_fd(), buf, SendFlags::empty()); - let BufResult(res, op) = self.file().submit(op).await; + let BufResult(res, op) = self.file.submit(op).await; BufResult(res, op.into_inner()) } @@ -598,7 +590,7 @@ impl Stream { use ::compio::driver::op::ShutdownSocket; let op = ShutdownSocket::new(self.inner.shared_fd(), std::net::Shutdown::Write); - match self.file().submit(op).await.0 { + match self.file.submit(op).await.0 { // A socket that is not connected has nothing to shut down, as // compio's stream has it. Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()), @@ -671,7 +663,7 @@ impl std::os::fd::AsFd for Stream { fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> { let fd = self.inner.as_fd(); #[cfg(debug_assertions)] - self.identity.check(fd); + self.identity.check(std::os::fd::AsRawFd::as_raw_fd(&fd)); fd } } @@ -685,7 +677,7 @@ impl SharedStream for Stream { async fn read(&self, buf: B) -> BufResult { #[cfg(target_os = "linux")] { - let buf = match self.recv.read_ahead(&self.inner, self.file(), buf).await { + let buf = match self.recv.read_ahead(&self.inner, &self.file, buf).await { recv::Copied::Done(result) => return result, recv::Copied::ReadInstead(buf) => buf, }; @@ -700,7 +692,7 @@ impl SharedStream for Stream { async fn read_owned(&self) -> io::Result> { #[cfg(target_os = "linux")] { - self.recv.read_owned(&self.inner, self.file()).await + self.recv.read_owned(&self.inner, &self.file).await } #[cfg(not(target_os = "linux"))] { @@ -1218,6 +1210,10 @@ mod files { slot: Rc, /// The socket's descriptor, which the slot replaces in an entry. fd: RawFd, + /// The socket `fd` names, which an operation naming the descriptor + /// checks it still does (debug builds, zen-rs/http-kit#159). + #[cfg(debug_assertions)] + identity: crate::sys::unix::Identity, } /// What a stream and its slot's keeper share. @@ -1268,7 +1264,13 @@ mod files { } }); } - Self { slot, fd } + Self { + slot, + fd, + #[cfg(debug_assertions)] + identity: crate::sys::unix::Identity::of(fd) + .expect("a stream's socket is open when the stream is made"), + } } fn slot(&self) -> io::Result> { @@ -1288,12 +1290,22 @@ mod files { /// `op`, naming the socket by its slot when it has one; or, with /// `op` back, the error taking a slot failed with. + /// + /// Without a slot `op` names the descriptor's number, which debug + /// builds check still names the socket. + #[track_caller] pub(super) fn address(&self, op: T) -> Result, (io::Error, T)> { match self.slot() { - Ok(slot) => Ok(Addressed { - op, - slot: slot.map(|index| (index, self.fd)), - }), + Ok(slot) => { + #[cfg(debug_assertions)] + if slot.is_none() { + self.identity.check(self.fd); + } + Ok(Addressed { + op, + slot: slot.map(|index| (index, self.fd)), + }) + } Err(error) => Err((error, op)), } } @@ -1545,7 +1557,7 @@ mod zero_copy { let rest = Shared(Rc::clone(&buf)).slice(written); let res = if zero_copy { let op = SendVectoredZc::new(stream.inner.shared_fd(), rest, flags); - match stream.file().address(op) { + match stream.file.address(op) { Ok(op) => send(op).await, Err((e, _)) => Err(e), } diff --git a/io/src/sys/unix.rs b/io/src/sys/unix.rs index 8972d32..7692498 100644 --- a/io/src/sys/unix.rs +++ b/io/src/sys/unix.rs @@ -54,14 +54,13 @@ pub(crate) struct Identity { #[cfg(all(feature = "compio", debug_assertions))] impl Identity { - /// The identity of the file `fd` names. - pub(crate) fn of(fd: std::os::fd::BorrowedFd<'_>) -> io::Result { - use std::os::fd::AsRawFd as _; - + /// The identity of the file `fd` names. A number is all `fstat` needs: + /// one that names nothing is reported as `EBADF`. + pub(crate) fn of(fd: std::os::fd::RawFd) -> io::Result { let mut stat = std::mem::MaybeUninit::::uninit(); - // SAFETY: `stat` is valid for the write of one `struct stat`, and - // `fd` is borrowed, so the call reads a descriptor that is open. - if unsafe { libc::fstat(fd.as_raw_fd(), stat.as_mut_ptr()) } != 0 { + // SAFETY: `stat` is valid for the write of one `struct stat`; the + // kernel reads nothing through `fd` but its number. + if unsafe { libc::fstat(fd, stat.as_mut_ptr()) } != 0 { return Err(io::Error::last_os_error()); } // SAFETY: `fstat` succeeded, so it filled `stat`. @@ -75,21 +74,17 @@ impl Identity { /// Panics unless `fd` still names the file `self` identifies. #[track_caller] - pub(crate) fn check(self, fd: std::os::fd::BorrowedFd<'_>) { - use std::os::fd::AsRawFd as _; - + pub(crate) fn check(self, fd: std::os::fd::RawFd) { match Self::of(fd) { Ok(now) if now == self => {} Ok(now) => panic!( - "descriptor {} no longer names the stream's socket ({self:?}): it names a {} \ + "descriptor {fd} no longer names the stream's socket ({self:?}): it names a {} \ ({now:?}), so the number was closed and reused while the stream held it", - fd.as_raw_fd(), now.kind_name(), ), Err(error) => panic!( - "descriptor {} no longer names the stream's socket ({self:?}): {error}, so the \ + "descriptor {fd} no longer names the stream's socket ({self:?}): {error}, so the \ number was closed while the stream held it", - fd.as_raw_fd(), ), } }