virtio-devices: vsock: Handle guest half-close

When the guest did a half-close (shutting down only its send side) the
connection state was updated but the write half of the host Unix socket
was never closed so the host peer never saw an EOF. This caused issues
with newer systemd (v256+) as it now half closes its socket and waits
for the host side to react and fully close the connection.

Propagate the guest's half-close to the host by shutting down the write
half of the backing stream. This is deferred until any buffered guest
data has been flushed so that no data is lost, and the connection is
left open so that host-to-guest data keeps flowing.

Assisted-by: Claude:claude-opus-4-8
Signed-off-by: Rob Bradford <rbradford@meta.com>

virtio-devices/src/seccomp_filters.rs

@@ -293,6 +293,7 @@ fn virtio_vsock_thread_rules() -> Vec<(i64, Vec<SeccompRule>)> {
         (libc::SYS_ioctl, create_vsock_ioctl_seccomp_rule()),
         (libc::SYS_recvfrom, vec![]),
         (libc::SYS_sendto, vec![]),
+        (libc::SYS_shutdown, vec![]),
         (libc::SYS_socket, vec![]),
     ]
 }

virtio-devices/src/vsock/csm/connection.rs

@@ -137,6 +137,9 @@ pub struct VsockConnection<S: Read + ReadVolatile + Write + WriteVolatile + AsRa
     /// Instant when this connection should be scheduled for immediate termination, due to some
     /// timeout condition having been fulfilled.
     expiry: Option<Instant>,
+    /// Whether we've already shut down the write half of the host stream, after the guest
+    /// half-closed its send side.
+    host_write_shutdown: bool,
 }
 
 impl<S> VsockChannel for VsockConnection<S>
@@ -350,18 +353,22 @@ where
                             Instant::now() + Duration::from_millis(defs::CONN_SHUTDOWN_TIMEOUT_MS),
                         );
                     }
+                } else if send_off {
+                    // The guest half-closed its send side; surface that as an EOF to the host.
+                    self.shutdown_host_write_side();
                 }
             }
 
             // The peer wants to update a shutdown request, with more receive/send indications.
             // The same logic as above applies.
-            ConnState::PeerClosed(ref mut recv_off, ref mut send_off)
-                if pkt.op() == uapi::VSOCK_OP_SHUTDOWN =>
-            {
-                *recv_off = *recv_off || (pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV != 0);
-                *send_off = *send_off || (pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND != 0);
-                if *recv_off && *send_off && self.tx_buf.is_empty() {
+            ConnState::PeerClosed(recv_off, send_off) if pkt.op() == uapi::VSOCK_OP_SHUTDOWN => {
+                let recv_off = recv_off || (pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_RCV != 0);
+                let new_send_off = send_off || (pkt.flags() & uapi::VSOCK_FLAGS_SHUTDOWN_SEND != 0);
+                self.state = ConnState::PeerClosed(recv_off, new_send_off);
+                if recv_off && new_send_off && self.tx_buf.is_empty() {
                     self.pending_rx.insert(PendingRx::Rst);
+                } else if new_send_off && !send_off {
+                    self.shutdown_host_write_side();
                 }
             }
 
@@ -476,6 +483,11 @@ where
             // before forceful termination, the wait might be over.
             if self.state == ConnState::PeerClosed(true, true) && self.tx_buf.is_empty() {
                 self.pending_rx.insert(PendingRx::Rst);
+            } else if matches!(self.state, ConnState::PeerClosed(_, true)) && self.tx_buf.is_empty()
+            {
+                // A deferred guest send-side half-close: now that the TX buffer is drained, we can
+                // surface the EOF to the host.
+                self.shutdown_host_write_side();
             } else if self.peer_needs_credit_update() {
                 // If we've freed up some more buffer space, we may need to let the peer know it
                 // can safely send more data our way.
@@ -514,6 +526,7 @@ where
             last_fwd_cnt_to_peer: Wrapping(0),
             pending_rx: PendingRxSet::from(PendingRx::Response),
             expiry: None,
+            host_write_shutdown: false,
         }
     }
 
@@ -541,6 +554,7 @@ where
             last_fwd_cnt_to_peer: Wrapping(0),
             pending_rx: PendingRxSet::from(PendingRx::Request),
             expiry: None,
+            host_write_shutdown: false,
         }
     }
 
@@ -641,6 +655,26 @@ where
         self.tx_buf.push_from(pkt, offset, len)
     }
 
+    /// Propagate a guest send-side half-close to the host stream, so the host peer reads an EOF.
+    /// This is deferred while the TX buffer still holds guest data, since shutting down the write
+    /// half now would drop those not-yet-flushed bytes; `notify()` retries once the buffer drains.
+    fn shutdown_host_write_side(&mut self) {
+        if self.host_write_shutdown || !self.tx_buf.is_empty() {
+            return;
+        }
+        // SAFETY: the stream owns the socket fd by construction and `shutdown` doesn't touch process
+        // memory
+        if unsafe { libc::shutdown(self.stream.as_raw_fd(), libc::SHUT_WR) } != 0 {
+            warn!(
+                "vsock: error shutting down host write side (lp={}, pp={}): {:?}",
+                self.local_port,
+                self.peer_port,
+                std::io::Error::last_os_error()
+            );
+        }
+        self.host_write_shutdown = true;
+    }
+
     /// Check if the credit information the peer has last received from us is outdated.
     ///
     fn peer_needs_credit_update(&self) -> bool {
@@ -1116,6 +1150,33 @@ mod unit_tests {
         }
     }
 
+    #[test]
+    fn test_peer_send_shutdown_defers_host_write_until_drained() {
+        // If the TX buffer still holds guest data when the guest half-closes its send side, the
+        // host write-side shutdown must be deferred until the buffer drains, so no data is lost.
+        let mut ctx = CsmTestContext::new_established();
+        let mut stream = TestStream::new();
+        stream.write_state = StreamState::WouldBlock;
+        ctx.set_stream(stream);
+
+        let data = &[1, 2, 3, 4];
+        ctx.init_data_pkt(data);
+        ctx.send();
+        assert!(!ctx.conn.tx_buf.is_empty());
+
+        ctx.init_pkt(uapi::VSOCK_OP_SHUTDOWN, 0)
+            .set_flags(uapi::VSOCK_FLAGS_SHUTDOWN_SEND);
+        ctx.send();
+        // Not shut down yet: the TX buffer still holds data.
+        assert!(!ctx.conn.host_write_shutdown);
+
+        // Once the buffer drains, the shutdown is issued.
+        ctx.set_stream(TestStream::new());
+        ctx.notify_epollout();
+        assert!(ctx.conn.tx_buf.is_empty());
+        assert!(ctx.conn.host_write_shutdown);
+    }
+
     #[test]
     fn test_local_read_error() {
         let mut ctx = CsmTestContext::new_established();

virtio-devices/src/vsock/unix/muxer.rs

@@ -1323,6 +1323,46 @@ mod unit_tests {
         assert_eq!(stream.read(buf.as_mut_slice()).unwrap(), 0);
     }
 
+    #[test]
+    fn test_peer_send_half_close() {
+        // Regression test for the systemd sd_notify (vsock-stream) deadlock: the guest writes its
+        // message, half-closes its send side, then waits for the host to close. The muxer must
+        // surface the guest's half-close as an EOF on the host stream (while keeping the connection
+        // alive), otherwise both ends block forever.
+        let peer_port = 1025;
+        let local_port = 1026;
+        let mut ctx = MuxerTestContext::new("peer_send_half_close");
+
+        let mut sock = ctx.create_local_listener(local_port);
+        ctx.init_pkt(local_port, peer_port, uapi::VSOCK_OP_REQUEST);
+        ctx.send();
+        let mut stream = sock.accept();
+
+        assert!(ctx.muxer.has_pending_rx());
+        ctx.recv();
+        assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RESPONSE);
+
+        // The guest sends its message, then half-closes its send side.
+        let data = &[1, 2, 3, 4];
+        ctx.init_data_pkt(local_port, peer_port, data);
+        ctx.send();
+        ctx.init_pkt(local_port, peer_port, uapi::VSOCK_OP_SHUTDOWN)
+            .set_flag(uapi::VSOCK_FLAGS_SHUTDOWN_SEND);
+        ctx.send();
+
+        // The host should read the message followed by an EOF, and the connection should still be
+        // alive (the muxer did not tear it down).
+        let mut buf = vec![0u8; 16];
+        assert_eq!(stream.read(buf.as_mut_slice()).unwrap(), data.len());
+        assert_eq!(&buf[..data.len()], data);
+        assert_eq!(stream.read(buf.as_mut_slice()).unwrap(), 0);
+        let key = ConnMapKey {
+            local_port,
+            peer_port,
+        };
+        assert!(ctx.muxer.conn_map.contains_key(&key));
+    }
+
     #[test]
     fn test_muxer_rxq() {
         let mut ctx = MuxerTestContext::new("muxer_rxq");