diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index 9fa3677..bc88acf 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -249,6 +249,14 @@ struct PendingStreamControl { attempts: u32, } +#[derive(Clone)] +struct PendingWindowAdjust { + received_offset: u64, + bytes: usize, + last_sent: Instant, + attempts: u32, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum StartupGateMode { HoldAll, @@ -5914,6 +5922,7 @@ async fn run_terminal( let mut stream_remote_eof_received: HashSet = HashSet::new(); let mut stream_eof_retransmit: HashMap = HashMap::new(); let mut stream_close_retransmit: HashMap = HashMap::new(); + let mut stream_window_adjust_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(); @@ -6853,6 +6862,7 @@ async fn run_terminal( &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -6895,6 +6905,17 @@ async fn run_terminal( consumed, received_offset, ).await?; + if consumed > 0 { + stream_window_adjust_retransmit.insert( + stream_id, + PendingWindowAdjust { + received_offset, + bytes: consumed, + last_sent: Instant::now(), + attempts: 1, + }, + ); + } } PacketKind::StreamWindowAdjust => { let Ok(plain) = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT) else { @@ -6964,6 +6985,7 @@ async fn run_terminal( &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -6995,6 +7017,7 @@ async fn run_terminal( &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -7078,6 +7101,7 @@ async fn run_terminal( &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -7101,6 +7125,7 @@ async fn run_terminal( &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -7292,6 +7317,14 @@ async fn run_terminal( &mut stream_close_retransmit, stream_retransmit_srtt, ).await?; + retransmit_stream_window_adjusts( + &socket, + addr, + &cred, + &mut send_seq, + &mut stream_window_adjust_retransmit, + stream_retransmit_srtt, + ).await?; } } } @@ -8376,6 +8409,47 @@ async fn retransmit_stream_closes( Ok(()) } +async fn retransmit_stream_window_adjusts( + socket: &UdpSocket, + addr: SocketAddr, + cred: &CachedCredential, + send_seq: &mut u64, + stream_window_adjust_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_window_adjust_retransmit.keys().copied().collect(); + for stream_id in stream_ids { + let Some(pending) = stream_window_adjust_retransmit.get_mut(&stream_id) else { + continue; + }; + if pending.attempts >= STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS { + stream_window_adjust_retransmit.remove(&stream_id); + continue; + } + if now.duration_since(pending.last_sent) < retransmit_after { + continue; + } + pending.last_sent = now; + pending.attempts = pending.attempts.saturating_add(1); + send_stream_window_adjust( + socket, + addr, + cred, + send_seq, + stream_id, + pending.bytes, + pending.received_offset, + ) + .await?; + } + Ok(()) +} + fn accept_stream_data( stream_next_recv_offset: &mut HashMap, stream_recv_pending: &mut HashMap>>, @@ -8405,6 +8479,7 @@ fn cleanup_stream_state( stream_local_eof_sent: &mut HashSet, stream_remote_eof_received: &mut HashSet, stream_eof_retransmit: &mut HashMap, + stream_window_adjust_retransmit: &mut HashMap, ) { opened_streams.remove(&stream_id); stream_send_credit.remove(&stream_id); @@ -8417,6 +8492,7 @@ fn cleanup_stream_state( stream_local_eof_sent.remove(&stream_id); stream_remote_eof_received.remove(&stream_id); stream_eof_retransmit.remove(&stream_id); + stream_window_adjust_retransmit.remove(&stream_id); } fn stream_eof_complete( @@ -8441,6 +8517,7 @@ fn stream_state_exists( stream_local_eof_sent: &HashSet, stream_remote_eof_received: &HashSet, stream_eof_retransmit: &HashMap, + stream_window_adjust_retransmit: &HashMap, ) -> bool { opened_streams.contains(&stream_id) || stream_send_credit.contains_key(&stream_id) @@ -8453,6 +8530,7 @@ fn stream_state_exists( || stream_local_eof_sent.contains(&stream_id) || stream_remote_eof_received.contains(&stream_id) || stream_eof_retransmit.contains_key(&stream_id) + || stream_window_adjust_retransmit.contains_key(&stream_id) } #[allow(clippy::too_many_arguments)] @@ -8469,6 +8547,7 @@ fn retire_stream_state( stream_local_eof_sent: &mut HashSet, stream_remote_eof_received: &mut HashSet, stream_eof_retransmit: &mut HashMap, + stream_window_adjust_retransmit: &mut HashMap, stream_retired: &mut HashSet, stream_retired_order: &mut VecDeque, ) { @@ -8485,6 +8564,7 @@ fn retire_stream_state( stream_local_eof_sent, stream_remote_eof_received, stream_eof_retransmit, + stream_window_adjust_retransmit, ); cleanup_stream_state( stream_id, @@ -8499,6 +8579,7 @@ fn retire_stream_state( stream_local_eof_sent, stream_remote_eof_received, stream_eof_retransmit, + stream_window_adjust_retransmit, ); if !had_state || STREAM_RETIRED_TOMBSTONES == 0 { return; @@ -9911,8 +9992,8 @@ 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, - PendingStreamControl, PendingStreamOpen, PredictMode, Predictor, RESTART_STATUS_SCRIPT, - RemoteForward, STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, + PendingStreamControl, PendingStreamOpen, PendingWindowAdjust, PredictMode, Predictor, + RESTART_STATUS_SCRIPT, RemoteForward, STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS, STREAM_INITIAL_WINDOW, SshConfig, SshPathTokenContext, StartupGateMode, StatusAction, TERMINAL_SNAPSHOT_RESET, UpdateOptions, UpdateRole, auth_allows, cache_key, cache_server_prefix, cleanup_stream_state, @@ -9929,17 +10010,17 @@ mod tests { 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_closes, 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, + retransmit_stream_opens, retransmit_stream_window_adjusts, 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, @@ -10075,6 +10156,26 @@ mod tests { }, ), ]); + let mut stream_window_adjust_retransmit = HashMap::from([ + ( + stream_id, + PendingWindowAdjust { + received_offset: 77, + bytes: 5, + last_sent: Instant::now(), + attempts: 1, + }, + ), + ( + other_stream_id, + PendingWindowAdjust { + received_offset: 88, + bytes: 6, + last_sent: Instant::now(), + attempts: 1, + }, + ), + ]); cleanup_stream_state( stream_id, @@ -10089,6 +10190,7 @@ mod tests { &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, ); assert!(!opened_streams.contains(&stream_id)); @@ -10102,6 +10204,7 @@ mod tests { 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!(!stream_window_adjust_retransmit.contains_key(&stream_id)); assert!(opened_streams.contains(&other_stream_id)); assert!(stream_send_credit.contains_key(&other_stream_id)); @@ -10114,6 +10217,7 @@ mod tests { 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)); + assert!(stream_window_adjust_retransmit.contains_key(&other_stream_id)); } #[test] @@ -10130,6 +10234,7 @@ mod tests { 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_window_adjust_retransmit = HashMap::new(); let mut stream_retired = HashSet::new(); let mut stream_retired_order = VecDeque::new(); @@ -10146,6 +10251,7 @@ mod tests { &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -10162,6 +10268,7 @@ mod tests { &mut stream_local_eof_sent, &mut stream_remote_eof_received, &mut stream_eof_retransmit, + &mut stream_window_adjust_retransmit, &mut stream_retired, &mut stream_retired_order, ); @@ -11494,6 +11601,94 @@ mod tests { assert!(close_retransmit.is_empty()); } + #[tokio::test] + async fn client_retransmits_pending_stream_window_adjust() { + 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 mut pending = HashMap::from([( + 44, + PendingWindowAdjust { + received_offset: 1234, + bytes: 99, + last_sent: Instant::now() - Duration::from_millis(250), + attempts: 1, + }, + )]); + + retransmit_stream_window_adjusts(&sender, addr, &cred, &mut send_seq, &mut pending, 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::StreamWindowAdjust); + assert_eq!(packet.header.seq, 10); + let plain = + protocol::decrypt_body(&packet, &cred.session_key, protocol::CLIENT_TO_SERVER).unwrap(); + let adjust: protocol::StreamWindowAdjust = protocol::from_body(&plain).unwrap(); + assert_eq!(adjust.stream_id, 44); + assert_eq!(adjust.received_offset, 1234); + assert_eq!(adjust.bytes, 99); + assert_eq!(send_seq, 11); + assert_eq!(pending[&44].attempts, 2); + } + + #[tokio::test] + async fn client_expires_stream_window_adjust_after_attempt_cap() { + 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 mut pending = HashMap::from([( + 44, + PendingWindowAdjust { + received_offset: 1234, + bytes: 99, + last_sent: Instant::now() - Duration::from_millis(250), + attempts: STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS, + }, + )]); + + retransmit_stream_window_adjusts(&sender, addr, &cred, &mut send_seq, &mut pending, None) + .await + .unwrap(); + + assert_eq!(send_seq, 10); + assert!(pending.is_empty()); + } + #[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 ad1929c..f93443d 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -438,6 +438,7 @@ struct ClientState { stream_remote_eof_received: HashSet, stream_eof_retransmit: HashMap, stream_close_retransmit: HashMap, + stream_window_adjust_retransmit: HashMap, stream_retired: HashSet, stream_retired_order: VecDeque, stream_retransmit_srtt: Option, @@ -482,6 +483,14 @@ struct PendingStreamControl { attempts: u32, } +#[derive(Clone)] +struct PendingWindowAdjust { + received_offset: u64, + bytes: usize, + last_sent: Instant, + attempts: u32, +} + #[derive(Clone)] struct PendingNativeAuth { client: dosh::native::NativeClientHello, @@ -1284,6 +1293,7 @@ async fn handle_native_user_auth( stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -1438,6 +1448,7 @@ async fn handle_bootstrap_attach( stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -1592,6 +1603,7 @@ async fn handle_ticket_attach( stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -3660,6 +3672,7 @@ fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) { client.stream_local_eof_sent.remove(&stream_id); client.stream_remote_eof_received.remove(&stream_id); client.stream_eof_retransmit.remove(&stream_id); + client.stream_window_adjust_retransmit.remove(&stream_id); } fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool { @@ -3675,6 +3688,9 @@ fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool { || client.stream_local_eof_sent.contains(&stream_id) || client.stream_remote_eof_received.contains(&stream_id) || client.stream_eof_retransmit.contains_key(&stream_id) + || client + .stream_window_adjust_retransmit + .contains_key(&stream_id) } fn client_stream_eof_complete(client: &ClientState, stream_id: u64) -> bool { @@ -3744,6 +3760,20 @@ async fn send_stream_window_adjust_to_client( bytes: usize, received_offset: u64, ) -> Result<()> { + if bytes > 0 { + let mut locked = state.lock().expect("server state poisoned"); + if let Some(client) = locked.client_mut(&client_id) { + client.stream_window_adjust_retransmit.insert( + stream_id, + PendingWindowAdjust { + received_offset, + bytes, + last_sent: Instant::now(), + attempts: 1, + }, + ); + } + } let body = protocol::to_body(&StreamWindowAdjust { stream_id, received_offset, @@ -4133,6 +4163,42 @@ async fn retransmit_pending( )?; sends.push((client.endpoint, packet)); } + let stream_adjust_ids: Vec = client + .stream_window_adjust_retransmit + .keys() + .copied() + .collect(); + for stream_id in stream_adjust_ids { + let Some(pending) = client.stream_window_adjust_retransmit.get_mut(&stream_id) + else { + continue; + }; + if pending.attempts >= STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS { + client.stream_window_adjust_retransmit.remove(&stream_id); + continue; + } + 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(&StreamWindowAdjust { + stream_id, + received_offset: pending.received_offset, + bytes: pending.bytes.min(u32::MAX as usize) as u32, + })?; + let packet = protocol::encode_encrypted( + PacketKind::StreamWindowAdjust, + *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 { @@ -4731,6 +4797,7 @@ mod tests { stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -4810,6 +4877,24 @@ mod tests { client .stream_recv_pending .insert(other_stream_id, BTreeMap::from([(88, vec![6])])); + client.stream_window_adjust_retransmit.insert( + stream_id, + PendingWindowAdjust { + received_offset: 77, + bytes: 5, + last_sent: Instant::now(), + attempts: 1, + }, + ); + client.stream_window_adjust_retransmit.insert( + other_stream_id, + PendingWindowAdjust { + received_offset: 88, + bytes: 6, + last_sent: Instant::now(), + attempts: 1, + }, + ); cleanup_client_stream(&mut client, stream_id); @@ -4822,6 +4907,11 @@ mod tests { assert!(!client.stream_sent_data.contains_key(&stream_id)); assert!(!client.stream_next_recv_offset.contains_key(&stream_id)); assert!(!client.stream_recv_pending.contains_key(&stream_id)); + assert!( + !client + .stream_window_adjust_retransmit + .contains_key(&stream_id) + ); assert!(client.opened_streams.contains(&other_stream_id)); assert!(client.stream_send_credit.contains_key(&other_stream_id)); @@ -4839,6 +4929,11 @@ mod tests { .contains_key(&other_stream_id) ); assert!(client.stream_recv_pending.contains_key(&other_stream_id)); + assert!( + client + .stream_window_adjust_retransmit + .contains_key(&other_stream_id) + ); } #[test] @@ -5399,6 +5494,7 @@ mod tests { stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -5502,6 +5598,7 @@ mod tests { stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -5709,6 +5806,120 @@ mod tests { assert!(!client.stream_close_retransmit.contains_key(&stream_id)); } + #[tokio::test] + async fn pending_server_stream_window_adjust_is_retransmitted() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx); + let client_id = [23u8; 16]; + let session_key = [24u8; 32]; + let stream_id = 128; + 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.stream_window_adjust_retransmit.insert( + stream_id, + PendingWindowAdjust { + received_offset: 4567, + bytes: 123, + 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::StreamWindowAdjust); + let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap(); + let adjust: StreamWindowAdjust = protocol::from_body(&plain).unwrap(); + assert_eq!(adjust.stream_id, stream_id); + assert_eq!(adjust.received_offset, 4567); + assert_eq!(adjust.bytes, 123); + + let locked = state.lock().unwrap(); + let client = locked.sessions["test"].clients.get(&client_id).unwrap(); + assert_eq!( + client.stream_window_adjust_retransmit[&stream_id].attempts, + 2 + ); + } + + #[tokio::test] + async fn pending_server_stream_window_adjust_expires_after_attempt_cap() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx); + let client_id = [25u8; 16]; + let session_key = [26u8; 32]; + let stream_id = 129; + 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.stream_window_adjust_retransmit.insert( + stream_id, + PendingWindowAdjust { + received_offset: 4567, + bytes: 123, + last_sent: Instant::now() - Duration::from_millis(250), + attempts: STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS, + }, + ); + 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 locked = state.lock().unwrap(); + let client = locked.sessions["test"].clients.get(&client_id).unwrap(); + assert!( + !client + .stream_window_adjust_retransmit + .contains_key(&stream_id) + ); + } + #[tokio::test] async fn server_stream_send_splits_large_writes() { let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); @@ -5751,6 +5962,7 @@ mod tests { stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None, @@ -5857,6 +6069,7 @@ mod tests { stream_remote_eof_received: HashSet::new(), stream_eof_retransmit: HashMap::new(), stream_close_retransmit: HashMap::new(), + stream_window_adjust_retransmit: HashMap::new(), stream_retired: HashSet::new(), stream_retired_order: VecDeque::new(), stream_retransmit_srtt: None,