From 3ab314698b8ad8e1e36c7caea960462f35129a9c Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Sun, 12 Jul 2026 23:59:42 -0400 Subject: [PATCH] Split stream writes into packet chunks --- src/bin/dosh-client.rs | 184 +++++++++++++++++++++++++-------- src/bin/dosh-server.rs | 228 +++++++++++++++++++++++++++++++---------- src/transport.rs | 134 ++++++++++++++++++++---- 3 files changed, 429 insertions(+), 117 deletions(-) diff --git a/src/bin/dosh-client.rs b/src/bin/dosh-client.rs index a7cd71c..d913b6d 100644 --- a/src/bin/dosh-client.rs +++ b/src/bin/dosh-client.rs @@ -7966,31 +7966,34 @@ async fn queue_or_send_stream_data( stream_next_send_offset: &mut HashMap, stream_sent_data: &mut HashMap>, ) -> Result<()> { - if !opened_streams.contains(&stream_id) - || stream_send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len() - || stream_pending_data - .get(&stream_id) - .is_some_and(|pending| !pending.is_empty()) - { - stream_pending_data - .entry(stream_id) - .or_default() - .push_back(bytes); - return Ok(()); - } - *stream_send_credit.entry(stream_id).or_default() -= bytes.len(); - let offset = *stream_next_send_offset.entry(stream_id).or_default(); - stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64)); - stream_sent_data.entry(stream_id).or_default().insert( - offset, - PendingStreamChunk { + for chunk in dosh::transport::split_stream_data_bytes(bytes, STREAM_INITIAL_WINDOW) { + if !opened_streams.contains(&stream_id) + || stream_send_credit.get(&stream_id).copied().unwrap_or(0) < chunk.len() + || stream_pending_data + .get(&stream_id) + .is_some_and(|pending| !pending.is_empty()) + { + stream_pending_data + .entry(stream_id) + .or_default() + .push_back(chunk); + continue; + } + *stream_send_credit.entry(stream_id).or_default() -= chunk.len(); + let offset = *stream_next_send_offset.entry(stream_id).or_default(); + stream_next_send_offset.insert(stream_id, offset.saturating_add(chunk.len() as u64)); + stream_sent_data.entry(stream_id).or_default().insert( offset, - bytes: bytes.clone(), - last_sent: Instant::now(), - attempts: 1, - }, - ); - send_stream_data(socket, addr, cred, send_seq, stream_id, offset, bytes).await + PendingStreamChunk { + offset, + bytes: chunk.clone(), + last_sent: Instant::now(), + attempts: 1, + }, + ); + send_stream_data(socket, addr, cred, send_seq, stream_id, offset, chunk).await?; + } + Ok(()) } #[allow(clippy::too_many_arguments)] @@ -8008,9 +8011,22 @@ async fn flush_stream_pending_data( let Some(pending) = stream_pending_data.get_mut(&stream_id) else { return Ok(()); }; - while let Some(bytes) = pending.front() { + let chunk_limit = dosh::transport::stream_data_chunk_limit(STREAM_INITIAL_WINDOW); + while let Some(front_len) = pending.front().map(Vec::len) { + if front_len > chunk_limit { + let Some(bytes) = pending.pop_front() else { + break; + }; + for chunk in dosh::transport::split_stream_data_bytes(bytes, STREAM_INITIAL_WINDOW) + .into_iter() + .rev() + { + pending.push_front(chunk); + } + continue; + } let credit = stream_send_credit.get(&stream_id).copied().unwrap_or(0); - if credit < bytes.len() { + if credit < front_len { break; } let Some(bytes) = pending.pop_front() else { @@ -9605,9 +9621,9 @@ mod tests { 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, SshConfig, SshPathTokenContext, - StartupGateMode, StatusAction, UpdateOptions, UpdateRole, auth_allows, cache_key, - cache_server_prefix, cleanup_stream_state, clear_cached_credentials, + STALE_TERMINAL_INPUT_AFTER, STARTUP_INPUT_HOLD, STREAM_INITIAL_WINDOW, SshConfig, + SshPathTokenContext, StartupGateMode, StatusAction, 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, @@ -9615,21 +9631,22 @@ mod tests { 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_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_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, - 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, + 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_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, 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, @@ -10860,6 +10877,85 @@ mod tests { assert_eq!(pending[&44].attempts, 2); } + #[tokio::test] + async fn client_stream_send_splits_large_writes() { + 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 stream_id = 44; + let opened_streams = HashSet::from([stream_id]); + let mut stream_send_credit = HashMap::from([(stream_id, STREAM_INITIAL_WINDOW)]); + let mut stream_pending_data = HashMap::new(); + let mut stream_next_send_offset = HashMap::new(); + let mut stream_sent_data = HashMap::new(); + let bytes = vec![9; dosh::transport::MAX_STREAM_DATA_BYTES * 2 + 11]; + + queue_or_send_stream_data( + &sender, + addr, + &cred, + &mut send_seq, + stream_id, + bytes, + &opened_streams, + &mut stream_send_credit, + &mut stream_pending_data, + &mut stream_next_send_offset, + &mut stream_sent_data, + ) + .await + .unwrap(); + + assert!(stream_pending_data.is_empty()); + assert_eq!(stream_sent_data[&stream_id].len(), 3); + assert_eq!( + stream_next_send_offset[&stream_id], + (dosh::transport::MAX_STREAM_DATA_BYTES * 2 + 11) as u64 + ); + assert_eq!(send_seq, 13); + + let mut seen = Vec::new(); + let mut buf = vec![0u8; u16::MAX as usize]; + for _ in 0..3 { + 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::StreamData); + let plain = + protocol::decrypt_body(&packet, &cred.session_key, protocol::CLIENT_TO_SERVER) + .unwrap(); + let data: protocol::StreamData = protocol::from_body(&plain).unwrap(); + seen.push((data.offset, data.bytes.len())); + } + assert_eq!( + seen, + vec![ + (0, dosh::transport::MAX_STREAM_DATA_BYTES), + ( + dosh::transport::MAX_STREAM_DATA_BYTES as u64, + dosh::transport::MAX_STREAM_DATA_BYTES + ), + ((dosh::transport::MAX_STREAM_DATA_BYTES * 2) as u64, 11) + ] + ); + } + #[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 1bd36e3..305653c 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -3363,67 +3363,69 @@ async fn queue_or_send_stream_data_to_client( stream_id: u64, bytes: Vec, ) -> Result<()> { - let send = { + let sends = { let mut locked = state.lock().expect("server state poisoned"); - let mut send = None; + let mut sends = Vec::new(); if let Some(client) = locked.client_mut(&client_id) { - if !client.opened_streams.contains(&stream_id) - || client - .stream_send_credit - .get(&stream_id) - .copied() - .unwrap_or(0) - < bytes.len() - || client - .stream_pending_data - .get(&stream_id) - .is_some_and(|pending| !pending.is_empty()) - { + for chunk in dosh::transport::split_stream_data_bytes(bytes, STREAM_INITIAL_WINDOW) { + if !client.opened_streams.contains(&stream_id) + || client + .stream_send_credit + .get(&stream_id) + .copied() + .unwrap_or(0) + < chunk.len() + || client + .stream_pending_data + .get(&stream_id) + .is_some_and(|pending| !pending.is_empty()) + { + client + .stream_pending_data + .entry(stream_id) + .or_default() + .push_back(chunk); + continue; + } + *client.stream_send_credit.entry(stream_id).or_default() -= chunk.len(); + let offset = *client.stream_next_send_offset.entry(stream_id).or_default(); client - .stream_pending_data + .stream_next_send_offset + .insert(stream_id, offset.saturating_add(chunk.len() as u64)); + client.send_seq += 1; + let body = protocol::to_body(&StreamData { + stream_id, + offset, + bytes: chunk.clone(), + })?; + let packet = protocol::encode_encrypted( + PacketKind::StreamData, + client_id, + client.send_seq, + client.last_acked, + &client.session_key, + SERVER_TO_CLIENT, + &body, + )?; + client + .stream_sent_data .entry(stream_id) .or_default() - .push_back(bytes); - return Ok(()); - } - *client.stream_send_credit.entry(stream_id).or_default() -= bytes.len(); - let offset = *client.stream_next_send_offset.entry(stream_id).or_default(); - client - .stream_next_send_offset - .insert(stream_id, offset.saturating_add(bytes.len() as u64)); - client.send_seq += 1; - let body = protocol::to_body(&StreamData { - stream_id, - offset, - bytes: bytes.clone(), - })?; - let packet = protocol::encode_encrypted( - PacketKind::StreamData, - client_id, - client.send_seq, - client.last_acked, - &client.session_key, - SERVER_TO_CLIENT, - &body, - )?; - client - .stream_sent_data - .entry(stream_id) - .or_default() - .insert( - offset, - PendingStreamChunk { + .insert( offset, - bytes, - last_sent: Instant::now(), - attempts: 1, - }, - ); - send = Some((client.endpoint, packet)); + PendingStreamChunk { + offset, + bytes: chunk, + last_sent: Instant::now(), + attempts: 1, + }, + ); + sends.push((client.endpoint, packet)); + } } - send + sends }; - if let Some((endpoint, packet)) = send { + for (endpoint, packet) in sends { let _ = send_udp(socket, &packet, endpoint).await?; } Ok(()) @@ -3443,16 +3445,30 @@ async fn flush_stream_pending_data_to_client( let Some(pending) = client.stream_pending_data.get_mut(&stream_id) else { return Ok(()); }; - let Some(bytes) = pending.front() else { + let Some(front_len) = pending.front().map(Vec::len) else { client.stream_pending_data.remove(&stream_id); return Ok(()); }; + if front_len > dosh::transport::stream_data_chunk_limit(STREAM_INITIAL_WINDOW) { + let Some(bytes) = pending.pop_front() else { + client.stream_pending_data.remove(&stream_id); + return Ok(()); + }; + for chunk in + dosh::transport::split_stream_data_bytes(bytes, STREAM_INITIAL_WINDOW) + .into_iter() + .rev() + { + pending.push_front(chunk); + } + continue; + } let credit = client .stream_send_credit .get(&stream_id) .copied() .unwrap_or(0); - if credit < bytes.len() { + if credit < front_len { return Ok(()); } let Some(bytes) = pending.pop_front() else { @@ -5259,6 +5275,108 @@ mod tests { assert_eq!(data.bytes, b"hello"); } + #[tokio::test] + async fn server_stream_send_splits_large_writes() { + let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); + let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx); + let client_id = [11u8; 16]; + let session_key = [12u8; 32]; + 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 endpoint = receiver.local_addr().unwrap(); + state.sessions.insert( + "test".to_string(), + Session { + pty: None, + parser: vt100::Parser::new(24, 80, 100), + restored_screen: None, + clients: HashMap::from([( + client_id, + ClientState { + endpoint, + mode: "forward-only".to_string(), + session_key, + last_acked: 0, + replay: ReplayWindow::default(), + send_seq: 1, + cols: 80, + rows: 24, + last_seen: Instant::now(), + pending: VecDeque::new(), + allowed_forwardings: Vec::new(), + stream_writers: HashMap::new(), + opened_streams: HashSet::from([42]), + stream_send_credit: HashMap::from([(42, STREAM_INITIAL_WINDOW)]), + stream_pending_data: HashMap::new(), + stream_pending_opens: HashMap::new(), + stream_next_send_offset: HashMap::new(), + stream_sent_data: HashMap::new(), + stream_next_recv_offset: HashMap::new(), + stream_recv_pending: HashMap::new(), + stream_retired: HashSet::new(), + stream_retired_order: VecDeque::new(), + stream_retransmit_srtt: None, + epoch: 0, + epoch_started: Instant::now(), + epoch_packets: 0, + previous_session_key: None, + }, + )]), + 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)); + let bytes = vec![3; dosh::transport::MAX_STREAM_DATA_BYTES * 2 + 17]; + + queue_or_send_stream_data_to_client(&state, &sender, client_id, 42, bytes) + .await + .unwrap(); + + { + let locked = state.lock().unwrap(); + let client = locked.sessions["test"].clients.get(&client_id).unwrap(); + assert!(client.stream_pending_data.is_empty()); + assert_eq!(client.stream_sent_data[&42].len(), 3); + assert_eq!( + client.stream_next_send_offset[&42], + (dosh::transport::MAX_STREAM_DATA_BYTES * 2 + 17) as u64 + ); + } + + let mut seen = Vec::new(); + let mut buf = vec![0u8; u16::MAX as usize]; + for _ in 0..3 { + 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::StreamData); + let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap(); + let data: StreamData = protocol::from_body(&plain).unwrap(); + seen.push((data.offset, data.bytes.len())); + } + assert_eq!( + seen, + vec![ + (0, dosh::transport::MAX_STREAM_DATA_BYTES), + ( + dosh::transport::MAX_STREAM_DATA_BYTES as u64, + dosh::transport::MAX_STREAM_DATA_BYTES + ), + ((dosh::transport::MAX_STREAM_DATA_BYTES * 2) as u64, 17) + ] + ); + } + #[tokio::test] async fn blocked_stream_data_does_not_block_terminal_frames() { let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); diff --git a/src/transport.rs b/src/transport.rs index 5934315..4ac66f5 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -34,6 +34,7 @@ pub const ADAPTIVE_RETRANSMIT_MIN: Duration = Duration::from_millis(10); pub const DEFAULT_KEEPALIVE_AFTER: Duration = Duration::from_secs(2); pub const DEFAULT_RETIRED_STREAM_TOMBSTONES: usize = 16 * 1024; pub const SERVICE_TARGET_PREFIX: &str = "@dosh-"; +pub const MAX_STREAM_DATA_BYTES: usize = 60 * 1024; #[derive(Debug, Clone, PartialEq, Eq)] pub struct TransportConfig { @@ -362,29 +363,35 @@ impl StreamMux { if bytes.is_empty() { return Ok(Vec::new()); } - if !self.opened_streams.contains(&stream_id) { - if self.pending_opens.contains_key(&stream_id) { + let chunks = split_stream_data_bytes(bytes, self.config.initial_window); + let mut out = Vec::new(); + if !self.opened_streams.contains(&stream_id) && !self.pending_opens.contains_key(&stream_id) + { + bail!("stream {stream_id} is not open"); + } + for chunk in chunks { + if !self.opened_streams.contains(&stream_id) { self.pending_data .entry(stream_id) .or_default() - .push_back(bytes); - return Ok(Vec::new()); + .push_back(chunk); + continue; } - bail!("stream {stream_id} is not open"); + if self.send_credit.get(&stream_id).copied().unwrap_or(0) < chunk.len() + || self + .pending_data + .get(&stream_id) + .is_some_and(|pending| !pending.is_empty()) + { + self.pending_data + .entry(stream_id) + .or_default() + .push_back(chunk); + continue; + } + out.push(self.send_data_now(stream_id, chunk)?); } - if self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len() - || self - .pending_data - .get(&stream_id) - .is_some_and(|pending| !pending.is_empty()) - { - self.pending_data - .entry(stream_id) - .or_default() - .push_back(bytes); - return Ok(Vec::new()); - } - Ok(vec![self.send_data_now(stream_id, bytes)?]) + Ok(out) } pub fn handle_data(&mut self, data: StreamData) -> Result> { @@ -600,11 +607,30 @@ impl StreamMux { fn flush_pending_data(&mut self, stream_id: u64) -> Result> { let mut out = Vec::new(); + let chunk_limit = stream_data_chunk_limit(self.config.initial_window); while let Some(front_len) = self .pending_data .get(&stream_id) .and_then(|pending| pending.front().map(Vec::len)) { + if front_len > chunk_limit { + let Some(bytes) = self + .pending_data + .get_mut(&stream_id) + .and_then(VecDeque::pop_front) + else { + break; + }; + if let Some(pending) = self.pending_data.get_mut(&stream_id) { + for chunk in split_stream_data_bytes(bytes, self.config.initial_window) + .into_iter() + .rev() + { + pending.push_front(chunk); + } + } + continue; + } if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len { break; } @@ -796,6 +822,21 @@ pub fn stream_data_within_receive_window( end <= expected_offset.saturating_add(window) } +pub fn stream_data_chunk_limit(receive_window: usize) -> usize { + MAX_STREAM_DATA_BYTES.min(receive_window.max(1)) +} + +pub fn split_stream_data_bytes(bytes: Vec, receive_window: usize) -> Vec> { + if bytes.is_empty() { + return Vec::new(); + } + let limit = stream_data_chunk_limit(receive_window); + if bytes.len() <= limit { + return vec![bytes]; + } + bytes.chunks(limit).map(Vec::from).collect() +} + pub struct DoshTransport { socket: Arc, role: SessionRole, @@ -1328,6 +1369,63 @@ mod tests { assert_eq!(data.bytes, b"!"); } + #[test] + fn send_data_splits_large_writes_into_packet_sized_chunks() { + let mut mux = StreamMux::new(TransportConfig::default()); + mux.open_stream(1, "@dosh-test", 0).unwrap(); + mux.handle_open_ok(StreamOpenOk { stream_id: 1 }) + .unwrap() + .unwrap(); + + let bytes = vec![7; MAX_STREAM_DATA_BYTES * 2 + 13]; + let sent = mux.send_data(1, bytes).unwrap(); + + assert_eq!(sent.len(), 3); + let first: StreamData = decode(&sent[0]); + let second: StreamData = decode(&sent[1]); + let third: StreamData = decode(&sent[2]); + assert_eq!(first.offset, 0); + assert_eq!(first.bytes.len(), MAX_STREAM_DATA_BYTES); + assert_eq!(second.offset, MAX_STREAM_DATA_BYTES as u64); + assert_eq!(second.bytes.len(), MAX_STREAM_DATA_BYTES); + assert_eq!(third.offset, (MAX_STREAM_DATA_BYTES * 2) as u64); + assert_eq!(third.bytes.len(), 13); + } + + #[test] + fn queued_large_write_flushes_in_window_sized_chunks_after_open() { + let mut mux = StreamMux::new(TransportConfig { + initial_window: 5, + ..TransportConfig::default() + }); + mux.open_stream(1, "@dosh-test", 0).unwrap(); + assert!(mux.send_data(1, b"hello!".to_vec()).unwrap().is_empty()); + + let flushed = mux + .handle_open_ok(StreamOpenOk { stream_id: 1 }) + .unwrap() + .unwrap(); + + assert_eq!(flushed.len(), 1); + let first: StreamData = decode(&flushed[0]); + assert_eq!(first.offset, 0); + assert_eq!(first.bytes, b"hello"); + assert_eq!(mux.pending_bytes(1), 1); + + let flushed = mux + .handle_window_adjust(StreamWindowAdjust { + stream_id: 1, + received_offset: 5, + bytes: 5, + }) + .unwrap() + .unwrap(); + assert_eq!(flushed.len(), 1); + let second: StreamData = decode(&flushed[0]); + assert_eq!(second.offset, 5); + assert_eq!(second.bytes, b"!"); + } + #[test] fn stream_open_ack_lowers_retransmit_timeout_from_observed_rtt() { let mut mux = StreamMux::new(TransportConfig {