Retransmit stream EOF controls

src/bin/dosh-client.rs

@@ -242,6 +242,12 @@ struct PendingStreamOpen {
     attempts: u32,
 }
 
+#[derive(Clone)]
+struct PendingStreamControl {
+    last_sent: Instant,
+    attempts: u32,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum StartupGateMode {
     HoldAll,
@@ -5905,6 +5911,7 @@ async fn run_terminal(
     let mut stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>> = HashMap::new();
     let mut stream_local_eof_sent: HashSet<u64> = HashSet::new();
     let mut stream_remote_eof_received: HashSet<u64> = HashSet::new();
+    let mut stream_eof_retransmit: HashMap<u64, PendingStreamControl> = HashMap::new();
     let mut stream_retired: HashSet<u64> = HashSet::new();
     let mut stream_retired_order: VecDeque<u64> = VecDeque::new();
     let mut pending_user_input: VecDeque<PendingUserInput> = VecDeque::new();
@@ -6808,6 +6815,13 @@ async fn run_terminal(
                         if stream_local_eof_sent.contains(&ok.stream_id) {
                             send_stream_eof(&socket, addr, &cred, &mut send_seq, ok.stream_id)
                                 .await?;
+                            stream_eof_retransmit.insert(
+                                ok.stream_id,
+                                PendingStreamControl {
+                                    last_sent: Instant::now(),
+                                    attempts: 1,
+                                },
+                            );
                         }
                     }
                     PacketKind::StreamOpenReject => {
@@ -6836,6 +6850,7 @@ async fn run_terminal(
                             &mut stream_recv_pending,
                             &mut stream_local_eof_sent,
                             &mut stream_remote_eof_received,
+                            &mut stream_eof_retransmit,
                             &mut stream_retired,
                             &mut stream_retired_order,
                         );
@@ -6946,6 +6961,7 @@ async fn run_terminal(
                                 &mut stream_recv_pending,
                                 &mut stream_local_eof_sent,
                                 &mut stream_remote_eof_received,
+                                &mut stream_eof_retransmit,
                                 &mut stream_retired,
                                 &mut stream_retired_order,
                             );
@@ -6975,6 +6991,7 @@ async fn run_terminal(
                             &mut stream_recv_pending,
                             &mut stream_local_eof_sent,
                             &mut stream_remote_eof_received,
+                            &mut stream_eof_retransmit,
                             &mut stream_retired,
                             &mut stream_retired_order,
                         );
@@ -7031,6 +7048,13 @@ async fn run_terminal(
                         stream_local_eof_sent.insert(stream_id);
                         if opened_streams.contains(&stream_id) {
                             send_stream_eof(&socket, addr, &cred, &mut send_seq, stream_id).await?;
+                            stream_eof_retransmit.insert(
+                                stream_id,
+                                PendingStreamControl {
+                                    last_sent: Instant::now(),
+                                    attempts: 1,
+                                },
+                            );
                         }
                         if stream_eof_complete(
                             stream_id,
@@ -7050,6 +7074,7 @@ async fn run_terminal(
                                 &mut stream_recv_pending,
                                 &mut stream_local_eof_sent,
                                 &mut stream_remote_eof_received,
+                                &mut stream_eof_retransmit,
                                 &mut stream_retired,
                                 &mut stream_retired_order,
                             );
@@ -7072,6 +7097,7 @@ async fn run_terminal(
                             &mut stream_recv_pending,
                             &mut stream_local_eof_sent,
                             &mut stream_remote_eof_received,
+                            &mut stream_eof_retransmit,
                             &mut stream_retired,
                             &mut stream_retired_order,
                         );
@@ -7238,6 +7264,16 @@ async fn run_terminal(
                     &mut stream_pending_opens,
                     stream_retransmit_srtt,
                 ).await?;
+                retransmit_stream_eofs(
+                    &socket,
+                    addr,
+                    &cred,
+                    &mut send_seq,
+                    &opened_streams,
+                    &stream_local_eof_sent,
+                    &mut stream_eof_retransmit,
+                    stream_retransmit_srtt,
+                ).await?;
             }
         }
     }
@@ -8252,6 +8288,44 @@ async fn retransmit_stream_opens(
     Ok(())
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn retransmit_stream_eofs(
+    socket: &UdpSocket,
+    addr: SocketAddr,
+    cred: &CachedCredential,
+    send_seq: &mut u64,
+    opened_streams: &HashSet<u64>,
+    stream_local_eof_sent: &HashSet<u64>,
+    stream_eof_retransmit: &mut HashMap<u64, PendingStreamControl>,
+    stream_retransmit_srtt: Option<Duration>,
+) -> Result<()> {
+    let now = Instant::now();
+    let retransmit_after = dosh::transport::adaptive_retransmit_after(
+        dosh::transport::DEFAULT_RETRANSMIT_AFTER,
+        stream_retransmit_srtt,
+    );
+    let stream_ids: Vec<u64> = stream_local_eof_sent
+        .iter()
+        .copied()
+        .filter(|stream_id| opened_streams.contains(stream_id))
+        .collect();
+    for stream_id in stream_ids {
+        let pending = stream_eof_retransmit
+            .entry(stream_id)
+            .or_insert(PendingStreamControl {
+                last_sent: now - retransmit_after,
+                attempts: 0,
+            });
+        if now.duration_since(pending.last_sent) < retransmit_after {
+            continue;
+        }
+        pending.last_sent = now;
+        pending.attempts = pending.attempts.saturating_add(1);
+        send_stream_eof(socket, addr, cred, send_seq, stream_id).await?;
+    }
+    Ok(())
+}
+
 fn accept_stream_data(
     stream_next_recv_offset: &mut HashMap<u64, u64>,
     stream_recv_pending: &mut HashMap<u64, BTreeMap<u64, Vec<u8>>>,
@@ -8280,6 +8354,7 @@ fn cleanup_stream_state(
     stream_recv_pending: &mut HashMap<u64, BTreeMap<u64, Vec<u8>>>,
     stream_local_eof_sent: &mut HashSet<u64>,
     stream_remote_eof_received: &mut HashSet<u64>,
+    stream_eof_retransmit: &mut HashMap<u64, PendingStreamControl>,
 ) {
     opened_streams.remove(&stream_id);
     stream_send_credit.remove(&stream_id);
@@ -8291,6 +8366,7 @@ fn cleanup_stream_state(
     stream_recv_pending.remove(&stream_id);
     stream_local_eof_sent.remove(&stream_id);
     stream_remote_eof_received.remove(&stream_id);
+    stream_eof_retransmit.remove(&stream_id);
 }
 
 fn stream_eof_complete(
@@ -8314,6 +8390,7 @@ fn stream_state_exists(
     stream_recv_pending: &HashMap<u64, BTreeMap<u64, Vec<u8>>>,
     stream_local_eof_sent: &HashSet<u64>,
     stream_remote_eof_received: &HashSet<u64>,
+    stream_eof_retransmit: &HashMap<u64, PendingStreamControl>,
 ) -> bool {
     opened_streams.contains(&stream_id)
         || stream_send_credit.contains_key(&stream_id)
@@ -8325,6 +8402,7 @@ fn stream_state_exists(
         || stream_recv_pending.contains_key(&stream_id)
         || stream_local_eof_sent.contains(&stream_id)
         || stream_remote_eof_received.contains(&stream_id)
+        || stream_eof_retransmit.contains_key(&stream_id)
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -8340,6 +8418,7 @@ fn retire_stream_state(
     stream_recv_pending: &mut HashMap<u64, BTreeMap<u64, Vec<u8>>>,
     stream_local_eof_sent: &mut HashSet<u64>,
     stream_remote_eof_received: &mut HashSet<u64>,
+    stream_eof_retransmit: &mut HashMap<u64, PendingStreamControl>,
     stream_retired: &mut HashSet<u64>,
     stream_retired_order: &mut VecDeque<u64>,
 ) {
@@ -8355,6 +8434,7 @@ fn retire_stream_state(
         stream_recv_pending,
         stream_local_eof_sent,
         stream_remote_eof_received,
+        stream_eof_retransmit,
     );
     cleanup_stream_state(
         stream_id,
@@ -8368,6 +8448,7 @@ fn retire_stream_state(
         stream_recv_pending,
         stream_local_eof_sent,
         stream_remote_eof_received,
+        stream_eof_retransmit,
     );
     if !had_state || STREAM_RETIRED_TOMBSTONES == 0 {
         return;
@@ -9780,33 +9861,33 @@ mod tests {
         FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LOCAL_SLEEP_REPAINT_AFTER,
         LOCAL_SLEEP_REPAINT_RETRY_WINDOW, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
         NativeIdentityContext, POST_RECONNECT_STALE_INPUT_GRACE, POST_SUBMIT_ALL_INPUT_HOLD,
-        PendingStreamOpen, PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward,
+        PendingStreamControl, PendingStreamOpen, PredictMode, Predictor, RESTART_STATUS_SCRIPT,
-        STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_INITIAL_WINDOW, SshConfig,
+        RemoteForward, STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_INITIAL_WINDOW,
-        SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_SNAPSHOT_RESET, UpdateOptions,
+        SshConfig, SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_SNAPSHOT_RESET,
-        UpdateRole, auth_allows, cache_key, cache_server_prefix, cleanup_stream_state,
+        UpdateOptions, UpdateRole, auth_allows, cache_key, cache_server_prefix,
-        clear_cached_credentials, effective_update_artifact_tag, ensure_tui_safe_status_overlay,
+        cleanup_stream_state, clear_cached_credentials, effective_update_artifact_tag,
-        expand_ssh_path_tokens, input_contains_focus_in, input_matches_escape,
+        ensure_tui_safe_status_overlay, expand_ssh_path_tokens, input_contains_focus_in,
-        is_local_status_target, is_resume_response_for_client, latest_release_download_url,
+        input_matches_escape, is_local_status_target, is_resume_response_for_client,
-        load_first_native_identity_with_prompt, native_proxy_udp_warning,
+        latest_release_download_url, load_first_native_identity_with_prompt,
-        newest_client_trace_path_from, parse_dynamic_forward, parse_escape_key,
+        native_proxy_udp_warning, newest_client_trace_path_from, parse_dynamic_forward,
-        parse_local_forward, parse_remote_forward, parse_single_remote_path, parse_ssh_config,
+        parse_escape_key, parse_local_forward, parse_remote_forward, parse_single_remote_path,
-        parse_trace_line, parse_trace_options, parse_trace_report_options, parse_trace_summary,
+        parse_ssh_config, parse_trace_line, parse_trace_options, parse_trace_report_options,
-        parse_update_options, post_submit_hold_duration, queue_or_send_stream_data,
+        parse_trace_summary, parse_update_options, post_submit_hold_duration,
-        queue_pending_user_input, queue_stale_pending_user_input, raw_contains_host_table,
+        queue_or_send_stream_data, queue_pending_user_input, queue_stale_pending_user_input,
-        recv_response_until, refresh_live_addr, release_tag_download_url,
+        raw_contains_host_table, recv_response_until, refresh_live_addr, release_tag_download_url,
         release_tag_from_effective_url, release_version_from_tag, render_frame_bytes,
         render_status_clear, render_status_overlay, requested_env, resolved_startup_command,
-        retire_stream_state, retransmit_stream_opens, rewrite_forward_command, sanitize_trace_name,
+        retire_stream_state, retransmit_stream_eofs, retransmit_stream_opens,
-        selected_predict_mode, selected_udp_host, send_stream_eof, server_version_mismatch,
+        rewrite_forward_command, sanitize_trace_name, selected_predict_mode, selected_udp_host,
-        should_flush_terminal_input_after_contact, should_health_log_client_start,
+        send_stream_eof, server_version_mismatch, should_flush_terminal_input_after_contact,
-        should_hold_during_startup_gate, should_hold_post_submit_input,
+        should_health_log_client_start, should_hold_during_startup_gate,
-        should_reconnect_before_input_for_local_sleep, should_repaint_idle_terminal,
+        should_hold_post_submit_input, should_reconnect_before_input_for_local_sleep,
-        should_strip_unowned_terminal_reports, split_after_command_submit, split_trace_tokens,
+        should_repaint_idle_terminal, should_strip_unowned_terminal_reports,
-        ssh_command_target, ssh_config_uses_proxy, ssh_destination_host, ssh_username,
+        split_after_command_submit, split_trace_tokens, ssh_command_target, ssh_config_uses_proxy,
-        ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports,
+        ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
-        strip_terminal_focus_reports, strip_unowned_terminal_reports, summarize_trace_file,
+        strip_stale_mouse_reports, strip_terminal_focus_reports, strip_unowned_terminal_reports,
-        summarize_trace_file_with_mode, terminal_private_mode_transition, toml_bare_key_or_quoted,
+        summarize_trace_file, summarize_trace_file_with_mode, terminal_private_mode_transition,
-        top_trace_events, trace_report_warnings, unix_update_script,
+        toml_bare_key_or_quoted, top_trace_events, trace_report_warnings, unix_update_script,
         update_binary_version_for_installer, update_installer_url, update_version_status,
         upsert_managed_block, valid_forward_host, vscode_safe_alias, wake_repaint_retry_deadline,
         windows_update_script,
@@ -9926,6 +10007,22 @@ mod tests {
         ]);
         let mut stream_local_eof_sent = HashSet::from([stream_id, other_stream_id]);
         let mut stream_remote_eof_received = HashSet::from([stream_id, other_stream_id]);
+        let mut stream_eof_retransmit = HashMap::from([
+            (
+                stream_id,
+                PendingStreamControl {
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            ),
+            (
+                other_stream_id,
+                PendingStreamControl {
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            ),
+        ]);
 
         cleanup_stream_state(
             stream_id,
@@ -9939,6 +10036,7 @@ mod tests {
             &mut stream_recv_pending,
             &mut stream_local_eof_sent,
             &mut stream_remote_eof_received,
+            &mut stream_eof_retransmit,
         );
 
         assert!(!opened_streams.contains(&stream_id));
@@ -9951,6 +10049,7 @@ mod tests {
         assert!(!stream_recv_pending.contains_key(&stream_id));
         assert!(!stream_local_eof_sent.contains(&stream_id));
         assert!(!stream_remote_eof_received.contains(&stream_id));
+        assert!(!stream_eof_retransmit.contains_key(&stream_id));
 
         assert!(opened_streams.contains(&other_stream_id));
         assert!(stream_send_credit.contains_key(&other_stream_id));
@@ -9962,6 +10061,7 @@ mod tests {
         assert!(stream_recv_pending.contains_key(&other_stream_id));
         assert!(stream_local_eof_sent.contains(&other_stream_id));
         assert!(stream_remote_eof_received.contains(&other_stream_id));
+        assert!(stream_eof_retransmit.contains_key(&other_stream_id));
     }
 
     #[test]
@@ -9977,6 +10077,7 @@ mod tests {
         let mut stream_recv_pending = HashMap::new();
         let mut stream_local_eof_sent = HashSet::new();
         let mut stream_remote_eof_received = HashSet::new();
+        let mut stream_eof_retransmit = HashMap::new();
         let mut stream_retired = HashSet::new();
         let mut stream_retired_order = VecDeque::new();
 
@@ -9992,6 +10093,7 @@ mod tests {
             &mut stream_recv_pending,
             &mut stream_local_eof_sent,
             &mut stream_remote_eof_received,
+            &mut stream_eof_retransmit,
             &mut stream_retired,
             &mut stream_retired_order,
         );
@@ -10007,6 +10109,7 @@ mod tests {
             &mut stream_recv_pending,
             &mut stream_local_eof_sent,
             &mut stream_remote_eof_received,
+            &mut stream_eof_retransmit,
             &mut stream_retired,
             &mut stream_retired_order,
         );
@@ -11184,6 +11287,65 @@ mod tests {
         assert_eq!(send_seq, 11);
     }
 
+    #[tokio::test]
+    async fn client_retransmits_pending_stream_eof_for_open_streams() {
+        let receiver = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let addr = receiver.local_addr().unwrap();
+        let cred = CachedCredential {
+            server: "host".to_string(),
+            session: "term".to_string(),
+            mode: "forward-only".to_string(),
+            udp_host: "127.0.0.1".to_string(),
+            udp_port: addr.port(),
+            client_id: [1; 16],
+            session_key: [2; 32],
+            session_key_id: protocol::session_key_id(&[2; 32]),
+            attach_ticket: Vec::new(),
+            attach_ticket_psk: [3; 32],
+            last_rendered_seq: 0,
+        };
+        let mut send_seq = 10;
+        let opened_streams = HashSet::from([44]);
+        let local_eof_sent = HashSet::from([44, 99]);
+        let mut eof_retransmit = HashMap::from([(
+            44,
+            PendingStreamControl {
+                last_sent: Instant::now() - Duration::from_millis(250),
+                attempts: 1,
+            },
+        )]);
+
+        retransmit_stream_eofs(
+            &sender,
+            addr,
+            &cred,
+            &mut send_seq,
+            &opened_streams,
+            &local_eof_sent,
+            &mut eof_retransmit,
+            None,
+        )
+        .await
+        .unwrap();
+
+        let mut buf = [0u8; 2048];
+        let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
+            .await
+            .unwrap()
+            .unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        assert_eq!(packet.header.kind, PacketKind::StreamEof);
+        assert_eq!(packet.header.seq, 10);
+        let plain =
+            protocol::decrypt_body(&packet, &cred.session_key, protocol::CLIENT_TO_SERVER).unwrap();
+        let eof: protocol::StreamEof = protocol::from_body(&plain).unwrap();
+        assert_eq!(eof.stream_id, 44);
+        assert_eq!(send_seq, 11);
+        assert_eq!(eof_retransmit[&44].attempts, 2);
+        assert!(!eof_retransmit.contains_key(&99));
+    }
+
     #[tokio::test]
     async fn client_stream_retransmit_uses_observed_rtt() {
         let receiver = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();

src/bin/dosh-server.rs

@@ -435,6 +435,7 @@ struct ClientState {
     stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
     stream_local_eof_sent: HashSet<u64>,
     stream_remote_eof_received: HashSet<u64>,
+    stream_eof_retransmit: HashMap<u64, PendingStreamControl>,
     stream_retired: HashSet<u64>,
     stream_retired_order: VecDeque<u64>,
     stream_retransmit_srtt: Option<Duration>,
@@ -473,6 +474,12 @@ struct PendingStreamOpen {
     attempts: u32,
 }
 
+#[derive(Clone)]
+struct PendingStreamControl {
+    last_sent: Instant,
+    attempts: u32,
+}
+
 #[derive(Clone)]
 struct PendingNativeAuth {
     client: dosh::native::NativeClientHello,
@@ -1273,6 +1280,7 @@ async fn handle_native_user_auth(
                     stream_recv_pending: HashMap::new(),
                     stream_local_eof_sent: HashSet::new(),
                     stream_remote_eof_received: HashSet::new(),
+                    stream_eof_retransmit: HashMap::new(),
                     stream_retired: HashSet::new(),
                     stream_retired_order: VecDeque::new(),
                     stream_retransmit_srtt: None,
@@ -1425,6 +1433,7 @@ async fn handle_bootstrap_attach(
                     stream_recv_pending: HashMap::new(),
                     stream_local_eof_sent: HashSet::new(),
                     stream_remote_eof_received: HashSet::new(),
+                    stream_eof_retransmit: HashMap::new(),
                     stream_retired: HashSet::new(),
                     stream_retired_order: VecDeque::new(),
                     stream_retransmit_srtt: None,
@@ -1577,6 +1586,7 @@ async fn handle_ticket_attach(
                     stream_recv_pending: HashMap::new(),
                     stream_local_eof_sent: HashSet::new(),
                     stream_remote_eof_received: HashSet::new(),
+                    stream_eof_retransmit: HashMap::new(),
                     stream_retired: HashSet::new(),
                     stream_retired_order: VecDeque::new(),
                     stream_retransmit_srtt: None,
@@ -2334,6 +2344,18 @@ async fn handle_stream_open_ok(
         let body = protocol::to_body(&StreamEof {
             stream_id: ok.stream_id,
         })?;
+        {
+            let mut locked = state.lock().expect("server state poisoned");
+            if let Some(client) = locked.client_mut(&packet.header.conn_id) {
+                client.stream_eof_retransmit.insert(
+                    ok.stream_id,
+                    PendingStreamControl {
+                        last_sent: Instant::now(),
+                        attempts: 1,
+                    },
+                );
+            }
+        }
         send_stream_packet_to_client(
             state,
             socket,
@@ -3631,6 +3653,7 @@ fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) {
     client.stream_recv_pending.remove(&stream_id);
     client.stream_local_eof_sent.remove(&stream_id);
     client.stream_remote_eof_received.remove(&stream_id);
+    client.stream_eof_retransmit.remove(&stream_id);
 }
 
 fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool {
@@ -3645,6 +3668,7 @@ fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool {
         || client.stream_recv_pending.contains_key(&stream_id)
         || client.stream_local_eof_sent.contains(&stream_id)
         || client.stream_remote_eof_received.contains(&stream_id)
+        || client.stream_eof_retransmit.contains_key(&stream_id)
 }
 
 fn client_stream_eof_complete(client: &ClientState, stream_id: u64) -> bool {
@@ -3755,6 +3779,13 @@ async fn send_stream_eof_to_client(
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {
             client.stream_local_eof_sent.insert(stream_id);
+            client.stream_eof_retransmit.insert(
+                stream_id,
+                PendingStreamControl {
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            );
         }
     }
     let body = protocol::to_body(&StreamEof { stream_id })?;
@@ -4030,6 +4061,37 @@ async fn retransmit_pending(
                     )?;
                     sends.push((client.endpoint, packet));
                 }
+                let stream_eof_ids: Vec<u64> = client
+                    .stream_local_eof_sent
+                    .iter()
+                    .copied()
+                    .filter(|stream_id| client.opened_streams.contains(stream_id))
+                    .collect();
+                for stream_id in stream_eof_ids {
+                    let pending = client.stream_eof_retransmit.entry(stream_id).or_insert(
+                        PendingStreamControl {
+                            last_sent: now - stream_retransmit_after,
+                            attempts: 0,
+                        },
+                    );
+                    if now.duration_since(pending.last_sent) < stream_retransmit_after {
+                        continue;
+                    }
+                    pending.last_sent = now;
+                    pending.attempts = pending.attempts.saturating_add(1);
+                    client.send_seq += 1;
+                    let body = protocol::to_body(&StreamEof { stream_id })?;
+                    let packet = protocol::encode_encrypted(
+                        PacketKind::StreamEof,
+                        *client_id,
+                        client.send_seq,
+                        client.last_acked,
+                        &client.session_key,
+                        SERVER_TO_CLIENT,
+                        &body,
+                    )?;
+                    sends.push((client.endpoint, packet));
+                }
                 let stream_ids: Vec<u64> = client.stream_sent_data.keys().copied().collect();
                 let mut retransmit_chunks = Vec::new();
                 for stream_id in stream_ids {
@@ -4626,6 +4688,7 @@ mod tests {
             stream_recv_pending: HashMap::new(),
             stream_local_eof_sent: HashSet::new(),
             stream_remote_eof_received: HashSet::new(),
+            stream_eof_retransmit: HashMap::new(),
             stream_retired: HashSet::new(),
             stream_retired_order: VecDeque::new(),
             stream_retransmit_srtt: None,
@@ -5292,6 +5355,7 @@ mod tests {
                         stream_recv_pending: HashMap::new(),
                         stream_local_eof_sent: HashSet::new(),
                         stream_remote_eof_received: HashSet::new(),
+                        stream_eof_retransmit: HashMap::new(),
                         stream_retired: HashSet::new(),
                         stream_retired_order: VecDeque::new(),
                         stream_retransmit_srtt: None,
@@ -5393,6 +5457,7 @@ mod tests {
                         stream_recv_pending: HashMap::new(),
                         stream_local_eof_sent: HashSet::new(),
                         stream_remote_eof_received: HashSet::new(),
+                        stream_eof_retransmit: HashMap::new(),
                         stream_retired: HashSet::new(),
                         stream_retired_order: VecDeque::new(),
                         stream_retransmit_srtt: None,
@@ -5439,6 +5504,66 @@ mod tests {
         assert_eq!(eof.stream_id, 42);
     }
 
+    #[tokio::test]
+    async fn pending_server_stream_eof_is_retransmitted() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        let client_id = [17u8; 16];
+        let session_key = [18u8; 32];
+        let stream_id = 125;
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
+        let mut client = test_client_state(session_key);
+        client.endpoint = receiver.local_addr().unwrap();
+        client.opened_streams.insert(stream_id);
+        client.stream_local_eof_sent.insert(stream_id);
+        client.stream_local_eof_sent.insert(999);
+        client.stream_eof_retransmit.insert(
+            stream_id,
+            PendingStreamControl {
+                last_sent: Instant::now() - Duration::from_millis(250),
+                attempts: 1,
+            },
+        );
+        state.sessions.insert(
+            "test".to_string(),
+            Session {
+                pty: None,
+                parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
+                clients: HashMap::from([(client_id, client)]),
+                output_seq: 0,
+                recent: VecDeque::new(),
+                empty_since: None,
+                holder_control: None,
+                persistent: false,
+                bytes_since_persist: 0,
+                last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
+            },
+        );
+        state.client_index.insert(client_id, "test".to_string());
+        let state = Arc::new(Mutex::new(state));
+
+        retransmit_pending(&state, &sender).await.unwrap();
+
+        let mut buf = [0u8; 2048];
+        let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
+            .await
+            .unwrap()
+            .unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        assert_eq!(packet.header.kind, PacketKind::StreamEof);
+        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
+        let eof: StreamEof = protocol::from_body(&plain).unwrap();
+        assert_eq!(eof.stream_id, stream_id);
+
+        let locked = state.lock().unwrap();
+        let client = locked.sessions["test"].clients.get(&client_id).unwrap();
+        assert_eq!(client.stream_eof_retransmit[&stream_id].attempts, 2);
+        assert!(!client.stream_eof_retransmit.contains_key(&999));
+    }
+
     #[tokio::test]
     async fn server_stream_send_splits_large_writes() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -5479,6 +5604,7 @@ mod tests {
                         stream_recv_pending: HashMap::new(),
                         stream_local_eof_sent: HashSet::new(),
                         stream_remote_eof_received: HashSet::new(),
+                        stream_eof_retransmit: HashMap::new(),
                         stream_retired: HashSet::new(),
                         stream_retired_order: VecDeque::new(),
                         stream_retransmit_srtt: None,
@@ -5583,6 +5709,7 @@ mod tests {
                         stream_recv_pending: HashMap::new(),
                         stream_local_eof_sent: HashSet::new(),
                         stream_remote_eof_received: HashSet::new(),
+                        stream_eof_retransmit: HashMap::new(),
                         stream_retired: HashSet::new(),
                         stream_retired_order: VecDeque::new(),
                         stream_retransmit_srtt: None,