diff --git a/src/endpoint.rs b/src/endpoint.rs
index 5c1b755..9e95c5f 100644
--- a/src/endpoint.rs
+++ b/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};