fix(native): stop a queued datagram being discarded as end of file

Both halves of a flow's socket pair decided end of file from a zero-byte read
plus a latched POLLHUP. POLLHUP latches when the peer closes and stays set while
its messages are still queued, so the rule was true in a case that is not end of
file.

The cost is a lost payload, not a lost empty datagram. A client that sends an
empty datagram, then a message, then closes leaves both queued. The zero-byte
read of the empty one satisfied every term of the rule, the daemon's drain loop
freed the flow, and the message behind it was never read. The client half lost
the mirror case the same way.

Measured on Linux 6.8 over an AF_UNIX SOCK_SEQPACKET pair. With an empty
datagram and then a three-byte datagram queued from a closed peer, revents is
0x0011, FIONREAD is 3 and recvmsg returns 0: every term of the old rule holds
while a real message waits.

End of file now also requires that nothing is queued behind the read. FIONREAD
answers that in one direction only, and it is the direction that matters: bytes
queued prove a further message is waiting, so it cannot be end of file. A zero
answer proves nothing, because a zero-length message contributes no bytes. That
one-directional reading is also what makes the check safe on every platform,
whichever socket type it chose and however its kernel signals a close.

One case therefore survives, and the documentation now says so rather than
asserting the opposite. A zero-length datagram that is the last message before a
close cannot be told from the close: reading it drains the queue, and the socket
is then identical to a drained one in revents, in FIONREAD, under MSG_PEEK and
in the recvmsg return. Separating those needs a payload that is never zero bytes
on the wire, which is a protocol change and is not made here.

The check is shared rather than written twice, because the end-of-file rule
belongs to the pair rather than to the end that reads it, and the two halves had
already drifted apart once.

Each guard is covered by a test that fails without it: the daemon half reports
end of file where a datagram is due, and the client half reports EPIPE.

CHANGELOG.md

@@ -89,7 +89,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   or FreeBSD `ECONNRESET` as end of file alongside the `POLLHUP` and
   zero-byte read that Linux gives. `EAGAIN` is deliberately not in that
   company: it means the socket is empty and the peer alive, so it stays an
-  error and the caller waits again.
+  error and the caller waits again. **A zero-length payload has one known
+  limitation.** A closed peer latches `POLLHUP` while its messages are still
+  queued, so that flag alone cannot say whether a zero-byte read is an empty
+  datagram or the close; the receive path also asks `FIONREAD`, and bytes still
+  queued prove a further message is waiting. That leaves one case unresolved:
+  a zero-length datagram that is the last message before a close is reported as
+  the close, because reading it drains the queue and a zero-length message
+  contributes no bytes to `FIONREAD`. A client should not give a zero-length
+  payload a meaning of its own, and should carry a one-byte discriminator
+  instead.
 
 #### OpenWrt mesh
 

src/native/client/mod.rs

@@ -475,11 +475,24 @@ impl FipsStream {
     /// exactly would be wrong: reading a zero-byte datagram as a close would let
     /// a peer tear down a live flow by sending nothing. A closed daemon half is
     /// `EPIPE` on every platform, but the platforms disagree on how the kernel
-    /// says so: Linux discriminates a zero-byte read with `POLLHUP`, and Darwin
-    /// returns `ECONNRESET` outright. Both are measured for this socket pair in
+    /// says so: Linux sets `POLLHUP` on a zero-byte read, and Darwin returns
+    /// `ECONNRESET` outright. Both are measured for this socket pair in
     /// [`seqpacket`](super::seqpacket), and both are translated to `EPIPE` here
     /// so a caller never sees the difference.
     ///
+    /// **`POLLHUP` alone is not the rule**, because it latches while messages
+    /// are still queued. A daemon half that wrote an empty datagram, then a
+    /// payload, then closed leaves all three facts true at once, and a read that
+    /// stopped at the flag would report the close and discard the payload. So a
+    /// hang-up is end of file only when
+    /// [`bytes_queued`](super::seqpacket::bytes_queued) reports nothing behind
+    /// it. The daemon half applies the identical rule, on the same pair.
+    ///
+    /// One case survives both checks: a zero-length datagram that is the last
+    /// message before the close is indistinguishable from the close, because
+    /// reading it drains the queue and a zero-length message contributes no
+    /// bytes. Do not give a zero-length payload a meaning of its own.
+    ///
     /// A datagram longer than `buf` is truncated and the remainder discarded,
     /// which is `SOCK_SEQPACKET` behaviour. Size `buf` at
     /// [`FipsStream::max_payload`] and it cannot happen.
@@ -496,7 +509,10 @@ impl FipsStream {
                 }
                 return Err(error);
             }
-            if received == 0 && seqpacket::peer_hung_up(self.fd.as_raw_fd()) {
+            if received == 0
+                && seqpacket::peer_hung_up(self.fd.as_raw_fd())
+                && seqpacket::bytes_queued(self.fd.as_raw_fd()) == 0
+            {
                 return Err(io::Error::from_raw_os_error(libc::EPIPE));
             }
             return Ok(received as usize);
@@ -1479,4 +1495,48 @@ mod tests {
         assert_eq!(addr.to_string(), format!("{PEER}:4242"));
         assert_eq!(addr.to_string().parse::<FipsAddr>().unwrap(), addr);
     }
+
+    #[test]
+    fn an_empty_datagram_before_the_daemon_closes_does_not_swallow_what_follows() {
+        // The mirror of the daemon-side test in `super::super::seqpacket`. The
+        // same end-of-file rule lives on both halves of the pair, so the same
+        // defect did: a `POLLHUP` latched while messages are still queued
+        // reported the close early, and the payload behind the empty datagram
+        // was never handed to the caller.
+        let (daemon, ours) = seqpacket::pair().unwrap();
+
+        // SAFETY: the descriptor is open and owned by `daemon`.
+        let sent = unsafe { libc::send(daemon.as_raw_fd(), std::ptr::null(), 0, 0) };
+        assert_eq!(sent, 0, "{}", io::Error::last_os_error());
+        // SAFETY: as above, and the pointer and length describe the literal.
+        let sent = unsafe { libc::send(daemon.as_raw_fd(), b"after".as_ptr().cast(), 5, 0) };
+        assert_eq!(sent, 5, "{}", io::Error::last_os_error());
+        drop(daemon);
+
+        let addr = FipsAddr::new(codec::pton(PEER).unwrap(), 4242);
+        let stream = FipsStream {
+            fd: ours,
+            peer: addr,
+            local: addr,
+            max: 1024,
+        };
+
+        let mut buf = [0u8; 64];
+        assert_eq!(
+            stream.recv(&mut buf).unwrap(),
+            0,
+            "the empty datagram was reported as the close"
+        );
+        assert_eq!(
+            stream.recv(&mut buf).unwrap(),
+            5,
+            "the message queued behind the empty datagram was lost"
+        );
+        assert_eq!(&buf[..5], b"after");
+        assert_eq!(
+            stream.recv(&mut buf).unwrap_err().raw_os_error(),
+            Some(libc::EPIPE),
+            "the close itself must still be reported once the queue is drained"
+        );
+    }
 }

src/native/seqpacket.rs

@@ -294,15 +294,29 @@ impl Seqpacket {
 /// socket returns `msg_flags == 0` for a normal message, for an empty message
 /// and at end of file alike, so the flag carries no information.
 ///
-/// `POLLHUP` does discriminate, measured the same way. After a zero-byte read,
-/// a queued empty datagram leaves the socket with no events pending, while a
-/// closed peer leaves `POLLHUP` set and latched. So a zero-byte read is end of
-/// file only when the peer has hung up.
+/// `POLLHUP` narrows the question and does not answer it, measured the same
+/// way. A live peer never sets it, so a zero-byte read without it is an empty
+/// datagram and nothing else. A closed peer does set it, **and it latches while
+/// messages are still queued**: a socket holding one empty datagram from a peer
+/// that has since closed reports what a drained socket reports, in `revents`,
+/// in `FIONREAD`, under `MSG_PEEK` and in the `recvmsg` return alike.
+///
+/// So the hang-up is one of two conditions rather than the whole rule. The other
+/// is [`bytes_queued`], which catches the case that costs a real payload: an
+/// empty datagram with a further message behind it, read after the peer closed.
+/// Without it that read reports end of file, the caller frees the flow, and the
+/// message behind it is never taken.
 ///
 /// This matters because reading an empty datagram as a close would let a client
 /// tear down its own flow by sending nothing, and the defect would present as a
 /// spurious disconnect.
 ///
+/// **One case survives and cannot be fixed here.** A zero-length datagram that
+/// is the last message before a close is indistinguishable from the close:
+/// reading it drains the queue, and every observation then matches a bare end of
+/// file. Separating those needs a payload that is never zero bytes on the wire,
+/// which is a protocol change rather than a receive-path one.
+///
 /// **`ECONNRESET` is treated as end of file too**, for the platform whose
 /// datagram sockets report a close that way rather than through `POLLHUP`. Both
 /// arms are compiled everywhere rather than split by `cfg`, because a rule that
@@ -320,12 +334,45 @@ fn recv_once(fd: RawFd, buf: &mut [u8]) -> io::Result<Received> {
         }
         return Err(err);
     }
-    if n == 0 && peer_hung_up(fd) {
+    if n == 0 && peer_hung_up(fd) && bytes_queued(fd) == 0 {
         return Ok(Received::Eof);
     }
     Ok(Received::Datagram(n as usize))
 }
 
+/// Bytes the receive queue still holds.
+///
+/// Only one direction of this is sound, and the rule that uses it depends on
+/// that asymmetry. **A non-zero answer proves a further message is waiting**, so
+/// whatever was just read was a datagram and not the close. A zero answer proves
+/// nothing: a zero-length datagram contributes no bytes, so a queue holding one
+/// is indistinguishable from a drained queue by this measure or any other.
+///
+/// The one-directional reading is what makes this safe on every platform the
+/// module compiles for, whatever socket type it chose and however its kernel
+/// signals a close. Bytes queued means the peer's data has not been consumed
+/// yet, and that cannot be end of file anywhere.
+///
+/// A failed `ioctl` reports zero, which leaves the caller with the rule that
+/// applied before this check existed rather than with an error on a path that
+/// has no way to report one.
+///
+/// Visible within [`super`] for the same reason [`peer_hung_up`] is: the
+/// end-of-file rule belongs to the pair, not to the end that reads it, and the
+/// client half at [`super::client::FipsStream::recv`] has to reach the same
+/// verdict on the same socket. One implementation is what stops the two halves
+/// drifting, which they had already done once.
+pub(super) fn bytes_queued(fd: RawFd) -> usize {
+    let mut queued: libc::c_int = 0;
+    // SAFETY: the descriptor is open and `queued` is a live `c_int`, which is
+    // what `FIONREAD` writes through the pointer.
+    let rc = unsafe { libc::ioctl(fd, libc::FIONREAD as _, &mut queued) };
+    if rc < 0 || queued < 0 {
+        return 0;
+    }
+    queued as usize
+}
+
 /// Translate a platform's spelling of "the peer is gone" into this API's.
 ///
 /// Darwin reports a closed `SOCK_DGRAM` peer as `ECONNRESET`, on the read path
@@ -591,6 +638,39 @@ mod tests {
         assert_eq!(recv_bounded(&daemon, &mut buf).await, Received::Datagram(5));
     }
 
+    #[tokio::test]
+    async fn an_empty_datagram_before_a_close_does_not_swallow_the_message_behind_it() {
+        // The case the test above does not reach. It keeps the client half open,
+        // so `POLLHUP` is never set and the end-of-file rule is never consulted.
+        // Here the client closes, which latches `POLLHUP` while both messages
+        // are still queued. Reading `POLLHUP` alone then reports the empty
+        // datagram as the close, `drain` frees the flow, and `after` is
+        // discarded without ever being read.
+        let (daemon, theirs) = pair().unwrap();
+        let daemon = Seqpacket::new(daemon).unwrap();
+        let mut theirs = client(theirs);
+
+        // SAFETY: the descriptor is open and owned by `theirs`.
+        let sent = unsafe { libc::send(theirs.as_raw_fd(), std::ptr::null(), 0, 0) };
+        assert_eq!(sent, 0, "{}", io::Error::last_os_error());
+        theirs.write_all(b"after").unwrap();
+        drop(theirs);
+
+        let mut buf = [0u8; 64];
+        assert_eq!(
+            recv_bounded(&daemon, &mut buf).await,
+            Received::Datagram(0),
+            "the empty datagram was reported as the close"
+        );
+        assert_eq!(
+            recv_bounded(&daemon, &mut buf).await,
+            Received::Datagram(5),
+            "the message queued behind the empty datagram was lost"
+        );
+        assert_eq!(&buf[..5], b"after");
+        assert_eq!(recv_bounded(&daemon, &mut buf).await, Received::Eof);
+    }
+
     #[test]
     fn both_halves_of_a_pair_are_close_on_exec() {
         // Asserted rather than assumed because the platforms disagree on how it