fix(endpoint): order EOF checks and atomically expire state

Read receive-close state before peeking at buffered KCP data so a FIN
arriving after an empty peek cannot turn that stale result into EOF.
Evaluate expiration and remove state under the same state-map lock so
a concurrent pong cannot revive an entry selected for later deletion.
Remove corresponding connections after releasing state-map locks.

Run buffered half-close and simultaneous-close tests on a two-worker
runtime. All 9 tests and example targets pass, with 100 additional
multithreaded FIN drain repetitions, formatting and strict clippy.

src/endpoint.rs

@@ -219,11 +219,14 @@ impl KcpConnection {
             async move {
                 let mut buf = BytesMut::new();
                 loop {
+                    // Observe FIN before inspecting the queue: a FIN arriving
+                    // after an empty peek must not make that stale peek EOF.
+                    let state = recv_state.load(Ordering::Acquire);
                     let peeksize = kcp.lock().peeksize();
                     if peeksize <= 0 {
                         // FIN is sent only after KCP acknowledges all bytes. Drain
                         // those bytes before publishing EOF to the stream.
-                        if recv_state.load(Ordering::Acquire) != RECV_OPEN {
+                        if state != RECV_OPEN {
                             let _ = recv_state.compare_exchange(
                                 RECV_FIN,
                                 RECV_EOF,
@@ -628,22 +631,22 @@ impl KcpEndpoint {
         let data = self.data.clone();
         self.tasks.spawn(async move {
             loop {
-                let expired = data
-                    .state_map
-                    .iter()
-                    .filter_map(|state| {
-                        let conn_id = *state.key();
-                        let closed = state.is_closed()
-                            && data
-                                .conn_map
-                                .get(&conn_id)
-                                .map(|conn| conn.recv_drained())
-                                .unwrap_or(true);
-                        (closed || state.is_pong_timeout()).then_some(conn_id)
-                    })
-                    .collect::<Vec<_>>();
+                let mut expired = Vec::new();
+                data.state_map.retain(|conn_id, state| {
+                    let closed = state.is_closed()
+                        && data
+                            .conn_map
+                            .get(conn_id)
+                            .map(|conn| conn.recv_drained())
+                            .unwrap_or(true);
+                    if closed || state.is_pong_timeout() {
+                        expired.push(*conn_id);
+                        false
+                    } else {
+                        true
+                    }
+                });
                 for conn_id in expired {
-                    data.state_map.remove(&conn_id);
                     data.conn_map.remove(&conn_id);
                 }
                 data.state_map.shrink_to_fit();
@@ -1006,7 +1009,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_stream_half_close_drains_buffered_data_and_response() {
         use crate::stream::KcpStream;
         use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -1044,7 +1047,7 @@ mod tests {
         .unwrap();
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_simultaneous_shutdown_drains_both_directions() {
         use crate::stream::KcpStream;
         use tokio::io::{AsyncReadExt, AsyncWriteExt};