[PHP] fix: preserve orderly TCP close across bridges

abi/snapshot.json

@@ -1050,6 +1050,11 @@
       "name": "kernel_pipe_close_write",
       "signature": "(i32,i32) -> (i32)"
     },
+    {
+      "kind": "func",
+      "name": "kernel_pipe_has_readers",
+      "signature": "(i32,i32) -> (i32)"
+    },
     {
       "kind": "func",
       "name": "kernel_pipe_is_read_open",

crates/kernel/src/pipe.rs

@@ -60,6 +60,11 @@ pub struct PipeBuffer {
     len: usize,
     read_count: u32,
     write_count: u32,
+    /// The receive half of a normally closed TCP endpoint remains as an
+    /// orphaned discard sink until the peer closes its write half. This models
+    /// TCP's simplex FIN without inventing a fixed number of successful writes
+    /// after EOF.
+    orphaned_read: bool,
     /// Index of this pipe in the PipeTable (for wakeup events).
     pipe_idx: u32,
     /// Ancillary data queue for SCM_RIGHTS FD passing.
@@ -79,6 +84,7 @@ impl PipeBuffer {
             len: 0,
             read_count: 1,
             write_count: 1,
+            orphaned_read: false,
             pipe_idx: 0,
             ancillary_fds: VecDeque::new(),
         }
@@ -109,6 +115,9 @@ impl PipeBuffer {
     /// Performs a partial write if the buffer does not have enough free space
     /// for all of `data`. Returns 0 if the buffer is full.
     pub fn write(&mut self, data: &[u8]) -> usize {
+        if self.read_count == 0 {
+            return if self.orphaned_read { data.len() } else { 0 };
+        }
         let cap = self.capacity();
         let n = data.len().min(self.free_space());
         if n == 0 {
@@ -178,19 +187,46 @@ impl PipeBuffer {
     /// Close one read end of the pipe. Decrements the read reference count.
     pub fn close_read_end(&mut self) {
         self.read_count = self.read_count.saturating_sub(1);
+        if self.read_count == 0 {
+            self.orphaned_read = false;
+            self.head = 0;
+            self.tail = 0;
+            self.len = 0;
+        }
         // Read end closed → pipe became writable (writers get EPIPE/SIGPIPE)
         crate::wakeup::push(self.pipe_idx, crate::wakeup::WAKE_WRITABLE);
     }
 
+    /// Close one TCP read end with orderly-close semantics.
+    ///
+    /// The last real reader becomes an orphaned discard sink while a writer is
+    /// still open. This is the pipe-backed equivalent of an operating system
+    /// retaining a TCP control block after the application closes its socket.
+    /// Explicit read shutdown uses `close_read_end` instead.
+    pub fn close_read_end_orderly(&mut self) {
+        self.read_count = self.read_count.saturating_sub(1);
+        if self.read_count == 0 {
+            self.head = 0;
+            self.tail = 0;
+            self.len = 0;
+            self.orphaned_read = self.write_count > 0;
+        }
+        crate::wakeup::push(self.pipe_idx, crate::wakeup::WAKE_WRITABLE);
+    }
+
     /// Close one write end of the pipe. Decrements the write reference count.
     pub fn close_write_end(&mut self) {
         self.write_count = self.write_count.saturating_sub(1);
+        if self.write_count == 0 {
+            self.orphaned_read = false;
+        }
         // Write end closed → pipe became readable (readers get EOF)
         crate::wakeup::push(self.pipe_idx, crate::wakeup::WAKE_READABLE);
     }
 
     /// Add a reader reference (e.g., after fork or dup).
     pub fn add_reader(&mut self) {
+        self.orphaned_read = false;
         self.read_count += 1;
     }
 
@@ -201,6 +237,14 @@ impl PipeBuffer {
 
     /// Returns true if the read end is still open (any readers remain).
     pub fn is_read_end_open(&self) -> bool {
+        self.read_count > 0 || self.orphaned_read
+    }
+
+    /// Returns true if an application-owned reader remains.
+    ///
+    /// Unlike `is_read_end_open`, this excludes TCP's orphaned discard sink so
+    /// host bridges can distinguish SHUT_WR from a final close.
+    pub fn has_readers(&self) -> bool {
         self.read_count > 0
     }
 
@@ -211,7 +255,7 @@ impl PipeBuffer {
 
     /// Returns true if both endpoints are closed and the pipe can be freed.
     pub fn is_fully_closed(&self) -> bool {
-        self.read_count == 0 && self.write_count == 0
+        self.read_count == 0 && self.write_count == 0 && !self.orphaned_read
     }
 
     /// Push ancillary FDs (SCM_RIGHTS) to be delivered with the next recvmsg.
@@ -499,6 +543,43 @@ mod tests {
         assert!(pipe.is_fully_closed());
     }
 
+    #[test]
+    fn test_orderly_read_close_discards_until_last_writer_closes() {
+        let mut pipe = PipeBuffer::new(8);
+
+        pipe.close_read_end_orderly();
+        assert!(pipe.is_read_end_open());
+        assert!(!pipe.has_readers());
+        assert_eq!(pipe.write(b"first"), 5);
+        assert_eq!(pipe.write(b"larger than capacity"), 20);
+        assert_eq!(pipe.available(), 0);
+        assert!(!pipe.is_fully_closed());
+
+        pipe.close_write_end();
+        assert!(!pipe.is_read_end_open());
+        assert!(pipe.is_fully_closed());
+    }
+
+    #[test]
+    fn test_orderly_read_close_preserves_other_real_readers() {
+        let mut pipe = PipeBuffer::new(8);
+        pipe.add_reader();
+
+        pipe.close_read_end_orderly();
+        assert!(pipe.has_readers());
+        assert_eq!(pipe.write(b"live"), 4);
+        let mut buf = [0u8; 4];
+        assert_eq!(pipe.read(&mut buf), 4);
+        assert_eq!(&buf, b"live");
+
+        pipe.close_read_end_orderly();
+        assert!(!pipe.has_readers());
+        assert_eq!(pipe.write(b"discarded"), 9);
+        assert_eq!(pipe.available(), 0);
+        pipe.close_write_end();
+        assert!(pipe.is_fully_closed());
+    }
+
     #[test]
     fn test_pipe_table_alloc_and_free() {
         let mut table = PipeTable::new();

crates/kernel/src/process_table.rs

@@ -366,7 +366,9 @@ impl ProcessTable {
             }
         }
 
-        // Clean up socket OFDs: close pipe endpoints so peers get EOF/EPIPE.
+        // Clean up socket OFDs. Active TCP streams use the same orderly FIN
+        // and orphaned receive state as close(2); other socket kinds close
+        // their pipe endpoints directly.
         // Without this, a peer process reading from a connected socket would
         // block forever instead of getting EOF when this process exits.
         //
@@ -380,6 +382,14 @@ impl ProcessTable {
                 let sock_idx = (-(ofd.host_handle + 1)) as usize;
                 if let Some(sock) = proc.sockets.get(sock_idx) {
                     if sock.global_pipes {
+                        let orderly_tcp_close = matches!(
+                            (sock.domain, sock.sock_type),
+                            (
+                                crate::socket::SocketDomain::Inet
+                                    | crate::socket::SocketDomain::Inet6,
+                                crate::socket::SocketType::Stream,
+                            )
+                        );
                         // Cross-process socket: close pipe ends in global table
                         if let Some(send_idx) = sock.send_buf_idx {
                             if let Some(pipe) = pipe_table.get_mut(send_idx) {
@@ -389,7 +399,11 @@ impl ProcessTable {
                         }
                         if let Some(recv_idx) = sock.recv_buf_idx {
                             if let Some(pipe) = pipe_table.get_mut(recv_idx) {
-                                pipe.close_read_end();
+                                if orderly_tcp_close {
+                                    pipe.close_read_end_orderly();
+                                } else {
+                                    pipe.close_read_end();
+                                }
                             }
                             pipe_table.free_if_closed(recv_idx);
                         }
@@ -1073,6 +1087,53 @@ pub fn current_pid() -> u32 {
 mod tests {
     use super::*;
 
+    #[test]
+    fn process_exit_closes_tcp_pipes_orderly() {
+        use crate::pipe::{global_pipe_table, PipeBuffer, DEFAULT_PIPE_CAPACITY};
+        use crate::socket::{SocketDomain, SocketInfo, SocketState, SocketType};
+
+        let pipe_table = unsafe { global_pipe_table() };
+        let send_idx = pipe_table.alloc(PipeBuffer::new(DEFAULT_PIPE_CAPACITY));
+        let recv_idx = pipe_table.alloc(PipeBuffer::new(DEFAULT_PIPE_CAPACITY));
+
+        let mut table = ProcessTable::new();
+        table.create_process(950_001).unwrap();
+        let proc = table.processes.get_mut(&950_001).unwrap();
+        let mut socket = SocketInfo::new(SocketDomain::Inet, SocketType::Stream, 6);
+        socket.state = SocketState::Connected;
+        socket.send_buf_idx = Some(send_idx);
+        socket.recv_buf_idx = Some(recv_idx);
+        socket.global_pipes = true;
+        let sock_idx = proc.sockets.alloc(socket);
+        let ofd_idx = proc.ofd_table.create(
+            FileType::Socket,
+            wasm_posix_shared::flags::O_RDWR,
+            -((sock_idx as i64) + 1),
+            Vec::new(),
+        );
+        proc.fd_table
+            .alloc(crate::fd::OpenFileDescRef(ofd_idx), 0)
+            .unwrap();
+
+        table.remove_process(950_001).unwrap();
+
+        let send_pipe = pipe_table.get_mut(send_idx).unwrap();
+        assert!(!send_pipe.is_write_end_open());
+        assert!(send_pipe.is_read_end_open());
+        let recv_pipe = pipe_table.get_mut(recv_idx).unwrap();
+        assert!(recv_pipe.is_read_end_open());
+        assert_eq!(recv_pipe.write(b"after-exit-one"), 14);
+        assert_eq!(recv_pipe.write(b"after-exit-two"), 14);
+        assert_eq!(recv_pipe.available(), 0);
+
+        pipe_table.get_mut(send_idx).unwrap().close_read_end();
+        pipe_table.free_if_closed(send_idx);
+        pipe_table.get_mut(recv_idx).unwrap().close_write_end();
+        pipe_table.free_if_closed(recv_idx);
+        assert!(pipe_table.get(send_idx).is_none());
+        assert!(pipe_table.get(recv_idx).is_none());
+    }
+
     fn install_bound_udp4_socket(table: &mut ProcessTable, pid: u32, port: u16) -> usize {
         use crate::socket::{SocketDomain, SocketInfo, SocketState, SocketType};