diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index c38b150..5962639 100644 --- a/src/bin/dosh-client.rs +++ b/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>> = HashMap::new(); let mut stream_local_eof_sent: HashSet = HashSet::new(); let mut stream_remote_eof_received: HashSet = HashSet::new(); + let mut stream_eof_retransmit: HashMap = HashMap::new(); let mut stream_retired: HashSet = HashSet::new(); let mut stream_retired_order: VecDeque = VecDeque::new(); let mut pending_user_input: VecDeque = 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, + stream_local_eof_sent: &HashSet, + stream_eof_retransmit: &mut HashMap, + stream_retransmit_srtt: Option, +) -> 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 = 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, stream_recv_pending: &mut HashMap>>, @@ -8280,6 +8354,7 @@ fn cleanup_stream_state( stream_recv_pending: &mut HashMap>>, stream_local_eof_sent: &mut HashSet, stream_remote_eof_received: &mut HashSet, + stream_eof_retransmit: &mut HashMap, ) { 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>>, stream_local_eof_sent: &HashSet, stream_remote_eof_received: &HashSet, + stream_eof_retransmit: &HashMap, ) -> 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>>, stream_local_eof_sent: &mut HashSet, stream_remote_eof_received: &mut HashSet, + stream_eof_retransmit: &mut HashMap, stream_retired: &mut HashSet, stream_retired_order: &mut VecDeque, ) { @@ -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, - STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_INITIAL_WINDOW, SshConfig, - SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_SNAPSHOT_RESET, UpdateOptions, - UpdateRole, auth_allows, cache_key, cache_server_prefix, cleanup_stream_state, - clear_cached_credentials, effective_update_artifact_tag, ensure_tui_safe_status_overlay, - expand_ssh_path_tokens, input_contains_focus_in, input_matches_escape, - is_local_status_target, is_resume_response_for_client, latest_release_download_url, - load_first_native_identity_with_prompt, native_proxy_udp_warning, - newest_client_trace_path_from, parse_dynamic_forward, parse_escape_key, - parse_local_forward, parse_remote_forward, parse_single_remote_path, parse_ssh_config, - parse_trace_line, parse_trace_options, parse_trace_report_options, parse_trace_summary, - parse_update_options, post_submit_hold_duration, queue_or_send_stream_data, - queue_pending_user_input, queue_stale_pending_user_input, raw_contains_host_table, - recv_response_until, refresh_live_addr, release_tag_download_url, + PendingStreamControl, PendingStreamOpen, PredictMode, Predictor, RESTART_STATUS_SCRIPT, + RemoteForward, STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_INITIAL_WINDOW, + SshConfig, SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_SNAPSHOT_RESET, + UpdateOptions, UpdateRole, auth_allows, cache_key, cache_server_prefix, + cleanup_stream_state, clear_cached_credentials, effective_update_artifact_tag, + ensure_tui_safe_status_overlay, expand_ssh_path_tokens, input_contains_focus_in, + input_matches_escape, is_local_status_target, is_resume_response_for_client, + latest_release_download_url, load_first_native_identity_with_prompt, + native_proxy_udp_warning, newest_client_trace_path_from, parse_dynamic_forward, + parse_escape_key, parse_local_forward, parse_remote_forward, parse_single_remote_path, + parse_ssh_config, parse_trace_line, parse_trace_options, parse_trace_report_options, + parse_trace_summary, parse_update_options, post_submit_hold_duration, + queue_or_send_stream_data, queue_pending_user_input, queue_stale_pending_user_input, + 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, - selected_predict_mode, selected_udp_host, send_stream_eof, server_version_mismatch, - should_flush_terminal_input_after_contact, should_health_log_client_start, - should_hold_during_startup_gate, should_hold_post_submit_input, - should_reconnect_before_input_for_local_sleep, should_repaint_idle_terminal, - should_strip_unowned_terminal_reports, split_after_command_submit, split_trace_tokens, - ssh_command_target, ssh_config_uses_proxy, ssh_destination_host, ssh_username, - ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports, - strip_terminal_focus_reports, strip_unowned_terminal_reports, summarize_trace_file, - summarize_trace_file_with_mode, terminal_private_mode_transition, toml_bare_key_or_quoted, - top_trace_events, trace_report_warnings, unix_update_script, + retire_stream_state, retransmit_stream_eofs, retransmit_stream_opens, + rewrite_forward_command, sanitize_trace_name, selected_predict_mode, selected_udp_host, + send_stream_eof, server_version_mismatch, should_flush_terminal_input_after_contact, + should_health_log_client_start, should_hold_during_startup_gate, + should_hold_post_submit_input, should_reconnect_before_input_for_local_sleep, + should_repaint_idle_terminal, should_strip_unowned_terminal_reports, + split_after_command_submit, split_trace_tokens, ssh_command_target, ssh_config_uses_proxy, + ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target, + strip_stale_mouse_reports, strip_terminal_focus_reports, strip_unowned_terminal_reports, + summarize_trace_file, summarize_trace_file_with_mode, terminal_private_mode_transition, + 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(); diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index ae65627..d46b906 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -435,6 +435,7 @@ struct ClientState { stream_recv_pending: HashMap>>, stream_local_eof_sent: HashSet, stream_remote_eof_received: HashSet, + stream_eof_retransmit: HashMap, stream_retired: HashSet, stream_retired_order: VecDeque, stream_retransmit_srtt: Option, @@ -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 = 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 = 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,