test(zerocopy): the fd must actually cross the socket, not just claim to

SCM_RIGHTS descriptor passing is the mechanism the whole worker isolation
design rests on, and no test verified it end to end: both suites asserted
only the has_fd boolean parsed from the JSON body. Setting the send-site's
descriptor to None — zero-copy broken in production — stayed green; so did
dropping the received fd in the worker's dispatch loop.

Now the client-side scripted peer records the st_ino of every descriptor
that arrives and the tests assert the sequence against dmabuf_key() of the
sent plane (SCM_RIGHTS re-numbers the fd but preserves the open file
description, so the inode is the identity the worker keys its cache on);
the worker dispatch test sends one import with a live fd and asserts the
backend received a descriptor with the sender's identity. (T1 from
design/pf-zerocopy-sweep-handoff.md)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/pf-zerocopy/src/imp/client.rs

@@ -627,8 +627,16 @@ mod tests {
         assert_eq!(status.code(), Some(42));
     }
 
+    /// A request as the scripted peer saw it, paired with the identity (`st_ino`) of the
+    /// descriptor that actually arrived via SCM_RIGHTS — the `has_fd` boolean in the JSON body is
+    /// a *claim*; the received fd is the mechanism the whole worker design rests on, so tests
+    /// assert on it directly.
+    type SeenRequest = (Request, Option<u64>);
+
     /// A scripted peer: answers the handshake, then serves canned replies per request.
-    fn scripted_server(replies: Vec<Reply>) -> (RemoteImporter, thread::JoinHandle<Vec<Request>>) {
+    fn scripted_server(
+        replies: Vec<Reply>,
+    ) -> (RemoteImporter, thread::JoinHandle<Vec<SeenRequest>>) {
         let (host, worker) = proto::socketpair_seqpacket().unwrap();
         proto::send(
             worker.as_fd(),
@@ -642,9 +650,12 @@ mod tests {
             let mut buf = Vec::new();
             let mut seen = Vec::new();
             let mut replies = replies.into_iter();
-            while let Ok((req, _fd)) = proto::recv::<Request>(worker.as_fd(), &mut buf) {
+            while let Ok((req, fd)) = proto::recv::<Request>(worker.as_fd(), &mut buf) {
                 let needs_reply = matches!(req, Request::Modifiers { .. } | Request::Import { .. });
-                seen.push(req);
+                let ino = fd
+                    .as_ref()
+                    .map(|f| dmabuf_key(f.as_raw_fd()).expect("fstat received fd"));
+                seen.push((req, ino));
                 if needs_reply {
                     match replies.next() {
                         Some(r) => proto::send(worker.as_fd(), &r, None).unwrap(),
@@ -669,9 +680,12 @@ mod tests {
         let seen = join.join().unwrap();
         assert_eq!(
             seen,
-            vec![Request::Modifiers {
+            vec![(
-                fourcc: 0x3432_5258
+                Request::Modifiers {
-            }]
+                    fourcc: 0x3432_5258
+                },
+                None
+            )]
         );
     }
 
@@ -698,17 +712,26 @@ mod tests {
         // Second import: no fd (already sent) → worker answers NeedFd → one retry WITH the fd.
         assert!(imp.import(&plane, 64, 64, 1, Some(2)).is_err());
         assert!(!imp.dead(), "NeedFd handling must not mark the worker dead");
+        // The identity the passed descriptor must carry — SCM_RIGHTS re-numbers the fd but
+        // preserves the open file description, so st_ino survives the crossing.
+        let key = dmabuf_key(plane.fd).unwrap();
         drop(imp);
-        let fd_flags: Vec<bool> = join
+        let fd_sends: Vec<(bool, Option<u64>)> = join
             .join()
             .unwrap()
             .iter()
-            .map(|r| match r {
+            .map(|(r, ino)| match r {
-                Request::Import { has_fd, .. } => *has_fd,
+                Request::Import { has_fd, .. } => (*has_fd, *ino),
                 other => panic!("unexpected request {other:?}"),
             })
             .collect();
-        assert_eq!(fd_flags, vec![true, false, true]);
+        // Not just the has_fd *claim* — the descriptor itself must have crossed, with the same
+        // identity the worker will key its cache on (`pass = None` at the send site would leave
+        // has_fd=true with no actual fd, which only this assertion catches).
+        assert_eq!(
+            fd_sends,
+            vec![(true, Some(key)), (false, None), (true, Some(key))]
+        );
     }
 
     #[test]
@@ -735,18 +758,23 @@ mod tests {
         };
         assert!(format!("{err:#}").contains("died"), "{err:#}");
         assert!(imp.dead());
+        let key = dmabuf_key(plane.fd).unwrap();
         drop(imp);
         let seen = join.join().unwrap();
-        // First import carried the fd (first sight of the key); the retry didn't re-send it.
+        // First import carried the fd (first sight of the key — and the DESCRIPTOR arrived, with
+        // the sender's identity); the retry didn't re-send it.
         match (&seen[0], &seen[1]) {
             (
-                Request::Import {
+                (
-                    has_fd: true,
+                    Request::Import {
-                    kind: ImportKind::Tiled,
+                        has_fd: true,
-                    ..
+                        kind: ImportKind::Tiled,
-                },
+                        ..
-                Request::Import { has_fd: false, .. },
+                    },
-            ) => {}
+                    Some(ino),
+                ),
+                (Request::Import { has_fd: false, .. }, None),
+            ) => assert_eq!(*ino, key),
             other => panic!("unexpected requests {other:?}"),
         }
     }

crates/pf-zerocopy/src/imp/worker.rs

@@ -359,6 +359,21 @@ mod tests {
     use super::*;
     use std::sync::mpsc;
 
+    /// Identity (`st_ino`) of an open fd — what SCM_RIGHTS preserves across the socket while
+    /// re-numbering the descriptor. Lets the dispatch test assert the fd that ARRIVED is the one
+    /// the host sent, not merely that the JSON body claimed one.
+    fn fd_ino(fd: impl AsRawFd) -> u64 {
+        // SAFETY: `libc::stat` is plain-old-data for which all-zero is a valid value, so
+        // `mem::zeroed()` is a sound initializer. `fd` is a live descriptor owned by the caller;
+        // `fstat` writes into the live, correctly-sized `&mut st`, and `st_ino` is read only
+        // after the return value is checked.
+        unsafe {
+            let mut st: libc::stat = std::mem::zeroed();
+            assert_eq!(libc::fstat(fd.as_raw_fd(), &mut st), 0, "fstat");
+            st.st_ino
+        }
+    }
+
     /// Records calls; import behavior is scripted per key.
     struct MockBackend {
         calls: mpsc::Sender<String>,
@@ -371,11 +386,13 @@ mod tests {
             vec![7, 8, 9]
         }
         fn import(&mut self, req: &ImportReq, fd: Option<OwnedFd>) -> Reply {
+            let received = match &fd {
+                Some(f) => format!("ino:{}", fd_ino(f.as_raw_fd())),
+                None => "none".into(),
+            };
             let _ = self.calls.send(format!(
-                "import:key={} kind={:?} fd={}",
+                "import:key={} kind={:?} fd={received}",
-                req.key,
+                req.key, req.kind,
-                req.kind,
-                fd.is_some()
             ));
             if req.key == 0xbad {
                 return Reply::Err {
@@ -460,6 +477,15 @@ mod tests {
         let (reply, _) = proto::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
         assert_eq!(reply, Reply::Frame { id: 1, desc: None });
 
+        // The descriptor itself must cross the socket: an import WITH an fd rides SCM_RIGHTS and
+        // the backend receives a live descriptor with the sender's identity — `serve` dropping the
+        // received fd (e.g. `backend.import(&req, None)`) would only be caught here.
+        let (pr, _pw) = std::io::pipe().unwrap();
+        let sent_ino = fd_ino(pr.as_fd().as_raw_fd());
+        proto::send(host.as_fd(), &import_req(3, true), Some(pr.as_fd())).unwrap();
+        let (reply, _) = proto::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
+        assert_eq!(reply, Reply::Frame { id: 2, desc: None });
+
         // A missing worker-side fd is a NeedFd reply (host resends), not a failure.
         proto::send(host.as_fd(), &import_req(0xfeed, false), None).unwrap();
         let (reply, _) = proto::recv::<Reply>(host.as_fd(), &mut buf).unwrap();
@@ -485,13 +511,14 @@ mod tests {
         assert_eq!(
             calls,
             vec![
-                "modifiers:42",
+                "modifiers:42".to_string(),
-                "import:key=1 kind=Tiled fd=false",
+                "import:key=1 kind=Tiled fd=none".to_string(),
-                "import:key=1 kind=Tiled fd=false",
+                "import:key=1 kind=Tiled fd=none".to_string(),
-                "import:key=65261 kind=Tiled fd=false", // 0xfeed
+                format!("import:key=3 kind=Tiled fd=ino:{sent_ino}"),
-                "import:key=2989 kind=Tiled fd=false",  // 0xbad
+                "import:key=65261 kind=Tiled fd=none".to_string(), // 0xfeed
-                "release:0",
+                "import:key=2989 kind=Tiled fd=none".to_string(),  // 0xbad
-                "clear",
+                "release:0".to_string(),
+                "clear".to_string(),
             ]
         );
     }