fix(endpoint): preserve FIN received before stream creation

Accept established connections whose peer has already closed its write
direction. Initialize their receive task with the saved FIN so empty
requests produce EOF and can still receive a response. Apply the same
handling when an empty response arrives before connect resumes.

Add deterministic tests for FIN preceding accept and for SYN-ACK plus
FIN arriving before the connecting future resumes. The delayed accept
case times out before this fix and returns EOF plus its response after.

Validation: all 9 tests and both example targets pass, with formatting
and strict all-targets clippy checks passing.

src/endpoint.rs

@@ -685,8 +685,12 @@ impl KcpEndpoint {
         });
     }
 
-    fn add_conn(&self, conn_id: ConnId) -> Result<(), Error> {
+    fn add_conn(&self, conn_id: ConnId, peer_closed: bool) -> Result<(), Error> {
         let mut conn = KcpConnection::new(conn_id)?;
+        // A peer may send FIN before the application claims the connection.
+        if peer_closed {
+            conn.close_recv(false);
+        }
         conn.run(self.output_sender.clone());
 
         let data = self.data.clone();
@@ -803,8 +807,11 @@ impl KcpEndpoint {
 
         if let Some(state) = self.data.state_map.get(&conn_id) {
             tracing::debug!(?conn_id, ?state, "connect done, checkin state");
-            if matches!(state.fsm, KcpConnectionFSM::Established) {
-                self.add_conn(conn_id)?;
+            if matches!(
+                state.fsm,
+                KcpConnectionFSM::Established | KcpConnectionFSM::PeerClosed
+            ) {
+                self.add_conn(conn_id, state.is_peer_closed())?;
                 return Ok(conn_id);
             } else {
                 drop(state);
@@ -849,8 +856,11 @@ impl KcpEndpoint {
                 continue;
             }
             state.pending_accept = false;
-            if state.is_established() {
-                self.add_conn(conn_id)?;
+            if matches!(
+                state.fsm,
+                KcpConnectionFSM::Established | KcpConnectionFSM::PeerClosed
+            ) {
+                self.add_conn(conn_id, state.is_peer_closed())?;
                 return Ok(conn_id);
             }
         }
@@ -1141,4 +1151,70 @@ mod tests {
             assert_eq!(error, std::io::ErrorKind::BrokenPipe);
         }
     }
+    #[tokio::test]
+    async fn test_accept_after_empty_request_half_close() {
+        use crate::stream::KcpStream;
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let (client, server, _tasks) = prepare_test().await;
+        let id = client
+            .connect(std::time::Duration::from_secs(1), 1, 2, Bytes::new())
+            .await
+            .unwrap();
+        let mut client_stream = KcpStream::new(&client, id).unwrap();
+        client_stream.shutdown().await.unwrap();
+        timeout(std::time::Duration::from_secs(1), async {
+            while !server.data.state_map.get(&id).unwrap().is_peer_closed() {
+                tokio::task::yield_now().await;
+            }
+            assert_eq!(server.accept().await.unwrap(), id);
+            let mut server_stream = KcpStream::new(&server, id).unwrap();
+            assert_eq!(server_stream.read(&mut [0; 1]).await.unwrap(), 0);
+            server_stream
+                .write_all(b"empty request response")
+                .await
+                .unwrap();
+            server_stream.shutdown().await.unwrap();
+            let mut response = Vec::new();
+            client_stream.read_to_end(&mut response).await.unwrap();
+            assert_eq!(response, b"empty request response");
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_connect_after_peer_half_close() {
+        use crate::stream::KcpStream;
+        use tokio::io::AsyncReadExt;
+        let mut client = KcpEndpoint::new();
+        client.run().await;
+        let mut output = client.output_receiver().unwrap();
+        let mut connecting =
+            Box::pin(client.connect(std::time::Duration::from_secs(1), 1, 2, Bytes::new()));
+        let syn = tokio::select! {
+            _ = &mut connecting => panic!("connect completed before SYN-ACK"),
+            packet = output.recv() => packet.unwrap(),
+        };
+        let id = ConnId::from(&syn);
+        let mut syn_ack = KcpPacket::new(0);
+        id.fill_packet_header(&mut syn_ack);
+        syn_ack.mut_header().set_syn(true).set_ack(true);
+        let mut fin = KcpPacket::new(0);
+        id.fill_packet_header(&mut fin);
+        fin.mut_header().set_fin(true);
+        client.input_sender().send(syn_ack).await.unwrap();
+        client.input_sender().send(fin).await.unwrap();
+        timeout(std::time::Duration::from_secs(1), async {
+            // Hold the connect future after its handshake notification until
+            // FIN has also arrived, modelling an immediately empty response.
+            while !client.data.state_map.get(&id).unwrap().is_peer_closed() {
+                tokio::task::yield_now().await;
+            }
+            assert_eq!(connecting.await.unwrap(), id);
+            let mut stream = KcpStream::new(&client, id).unwrap();
+            assert_eq!(stream.read(&mut [0; 1]).await.unwrap(), 0);
+        })
+        .await
+        .unwrap();
+    }
 }