chore(io): check a compio stream's descriptor before each raw use

h1/src/conn.rs

@@ -312,6 +312,9 @@ async fn send_region<W: Write + SendFile>(
     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(())
 }

io/src/compio.rs

@@ -435,6 +435,10 @@ impl_socket!(net::UnixStream, zero_copy: false);
 /// A connected stream socket.
 pub struct Stream<S> {
     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,11 @@ impl<S: Socket> Stream<S> {
         #[cfg(not(target_os = "linux"))]
         let _ = io;
         Self {
+            #[cfg(all(unix, debug_assertions))]
+            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,
@@ -498,9 +507,23 @@ impl<S: Socket> Stream<S> {
         }
         #[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(std::os::fd::AsRawFd::as_raw_fd(&fd));
+        fd
+    }
 }
 
 impl<S> Stream<S> {
@@ -638,7 +661,10 @@ impl<S: fmt::Debug> fmt::Debug for Stream<S> {
 #[cfg(unix)]
 impl<S: std::os::fd::AsFd> std::os::fd::AsFd for Stream<S> {
     fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
-        self.inner.as_fd()
+        let fd = self.inner.as_fd();
+        #[cfg(debug_assertions)]
+        self.identity.check(std::os::fd::AsRawFd::as_raw_fd(&fd));
+        fd
     }
 }
 
@@ -917,11 +943,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 +972,7 @@ mod splice {
     /// Makes the socket `O_NONBLOCK`, once.
     fn nonblocking<S: Socket + AsFd>(stream: &Stream<S>) -> 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 +1016,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 +1037,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,
@@ -1188,6 +1210,10 @@ mod files {
         slot: Rc<Slot>,
         /// 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.
@@ -1238,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<Option<u32>> {
@@ -1258,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<T>(&self, op: T) -> Result<Addressed<T>, (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)),
             }
         }

io/src/sys/unix.rs

@@ -32,3 +32,73 @@ pub(crate) fn local_addr(fd: impl AsFd) -> io::Result<SocketAddr> {
 pub(crate) fn peer_addr(fd: impl AsFd) -> io::Result<SocketAddr> {
     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. 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<Self> {
+        let mut stat = std::mem::MaybeUninit::<libc::stat>::uninit();
+        // 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`.
+        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::RawFd) {
+        match Self::of(fd) {
+            Ok(now) if now == self => {}
+            Ok(now) => panic!(
+                "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",
+                now.kind_name(),
+            ),
+            Err(error) => panic!(
+                "descriptor {fd} no longer names the stream's socket ({self:?}): {error}, so the \
+                 number was closed while the stream held it",
+            ),
+        }
+    }
+
+    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",
+        }
+    }
+}

io/tests/io.rs

@@ -905,6 +905,46 @@ 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")]
+            {
+                use super::SendFile as _;
+
+                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")]