diff --git a/ql-fsm/src/fsm.rs b/ql-fsm/src/fsm.rs index a532c9bd..456cb499 100644 --- a/ql-fsm/src/fsm.rs +++ b/ql-fsm/src/fsm.rs @@ -132,6 +132,9 @@ pub fn receive( let (decrypt_len, seq) = { let record = wire::QlSessionRecord::decode(&mut reader) .map_err(|error| ReceiveError::wire(ReceiveStage::SessionRecord, error))?; + if conn.session.is_replay(record.header.seq) { + return Ok(()); + } let payload = wire::decrypt_record( crypto, &header, diff --git a/ql-fsm/src/handshake/mod.rs b/ql-fsm/src/handshake/mod.rs index e70c2936..f10cf5ee 100644 --- a/ql-fsm/src/handshake/mod.rs +++ b/ql-fsm/src/handshake/mod.rs @@ -9,7 +9,7 @@ use ql_wire::{ use crate::{ fsm::emit_peer_status, - session::{SessionConfig, SessionFsm, StreamParity}, + session::{SessionFsm, SessionParams, StreamParity}, state::{ConnectedState, LinkState, SessionTransport}, Event, NoPeerError, QlFsm, ReceiveError, }; @@ -50,7 +50,7 @@ pub fn handle_disarm_pairing(fsm: &mut QlFsm) { fn local_transport_params(fsm: &QlFsm) -> wire::TransportParams { wire::TransportParams { - initial_stream_receive_window: fsm.config.session_stream_receive_buffer_size, + initial_stream_receive_window: fsm.config.session.stream_receive_buffer_size, } } @@ -110,20 +110,11 @@ pub fn finish_handshake( fsm.events.push_back(Event::NewPeer); } - let config = &fsm.config; let session = SessionFsm::new( - SessionConfig { + fsm.config.session, + SessionParams { local_parity: StreamParity::for_local(fsm.identity.qid, qid), - record_max_size: config.session_record_max_size, - ack_delay: config.session_record_ack_delay, - retransmit_timeout: config.session_record_retransmit_timeout, - keepalive_interval: config.session_keepalive_interval, - peer_timeout: config.session_peer_timeout, - stream_send_buffer_size: config.session_stream_send_buffer_size, - stream_receive_buffer_size: config.session_stream_receive_buffer_size, - accepted_record_window: config.session_accepted_record_window, - pending_ack_range_limit: config.session_pending_ack_range_limit, - initial_peer_stream_receive_window: transport + initial_stream_receive_window: transport .remote_transport_params .initial_stream_receive_window, }, diff --git a/ql-fsm/src/lib.rs b/ql-fsm/src/lib.rs index 02964bdb..d55200e4 100644 --- a/ql-fsm/src/lib.rs +++ b/ql-fsm/src/lib.rs @@ -37,7 +37,7 @@ pub use error::*; pub use pairing::PairingInvite; use ql_common::{ResetCode, StreamId}; use ql_wire::{PairingToken, PeerBundle, QlCrypto, QlIdentity, SessionClose, SessionCloseCode}; -pub use session::{SessionEvent, StreamReadIter, StreamWriter}; +pub use session::{SessionConfig, SessionEvent, StreamReadIter, StreamWriter}; use crate::state::{LinkState, QlFsmState}; @@ -170,40 +170,14 @@ impl StreamOps<'_> { pub struct QlFsmConfig { /// overall time limit for one handshake attempt pub handshake_timeout: Duration, - /// delay before sending a pure record ack - pub session_record_ack_delay: Duration, - /// initial wait before resending unacked session records - pub session_record_retransmit_timeout: Duration, - /// idle delay before sending a keepalive ping - pub session_keepalive_interval: Duration, - /// how long to wait before declaring the peer dead - pub session_peer_timeout: Duration, - /// maximum total wire size for one session record, including header and auth tag - pub session_record_max_size: usize, - /// maximum bytes buffered locally for one stream send side - pub session_stream_send_buffer_size: usize, - /// maximum bytes buffered locally for one stream receive side - pub session_stream_receive_buffer_size: u32, - /// how many accepted record sequence numbers to retain for duplicate detection - pub session_accepted_record_window: u64, - /// maximum disjoint pending ACK ranges to retain before dropping the oldest low ranges - pub session_pending_ack_range_limit: usize, + pub session: SessionConfig, } impl Default for QlFsmConfig { fn default() -> Self { - let s = session::SessionConfig::default(); Self { handshake_timeout: Duration::from_secs(5), - session_record_ack_delay: s.ack_delay, - session_record_retransmit_timeout: s.retransmit_timeout, - session_keepalive_interval: s.keepalive_interval, - session_peer_timeout: s.peer_timeout, - session_record_max_size: s.record_max_size, - session_stream_send_buffer_size: s.stream_send_buffer_size, - session_stream_receive_buffer_size: s.stream_receive_buffer_size, - session_accepted_record_window: s.accepted_record_window, - session_pending_ack_range_limit: s.pending_ack_range_limit, + session: SessionConfig::default(), } } } diff --git a/ql-fsm/src/session/ack_tracker.rs b/ql-fsm/src/session/ack_tracker.rs index 7f947d2d..da8d0d83 100644 --- a/ql-fsm/src/session/ack_tracker.rs +++ b/ql-fsm/src/session/ack_tracker.rs @@ -2,14 +2,12 @@ use std::{ops::RangeInclusive, time::Instant}; use ql_wire::{RecordAck, RecordAckBuilder, RecordSeq}; -use super::range_set::RangeSet; +use super::range_set::{single_range, RangeSet}; #[derive(Debug, Clone)] pub struct AckTracker { - accepted_records: RangeSet, pending_ack: RangeSet, - ack_state: AckState, - accepted_record_window: u64, + ack_due_at: Option, pending_ack_range_limit: usize, } @@ -20,74 +18,36 @@ pub struct PendingAck { pub includes_all_pending: bool, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ReceiveOutcome { - New, - Duplicate, - TooOld, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum AckState { - Idle, - Dirty { due_at: Instant }, -} - impl AckTracker { - pub fn new(accepted_record_window: u64, pending_ack_range_limit: usize) -> Self { + pub fn new(pending_ack_range_limit: usize) -> Self { Self { - accepted_records: RangeSet::new(), pending_ack: RangeSet::new(), - ack_state: AckState::Idle, - accepted_record_window: accepted_record_window.max(1), + ack_due_at: None, pending_ack_range_limit: pending_ack_range_limit.max(1), } } - pub fn insert(&mut self, seq: RecordSeq) -> ReceiveOutcome { - let seq = seq.0; - let largest_accepted = self.accepted_records.max(); - if largest_accepted.is_some_and(|largest| seq < self.accepted_cutoff(largest)) { - return ReceiveOutcome::TooOld; - } - if self.accepted_records.contains(seq) { - self.pending_ack.insert(single_range(seq)); - self.trim_pending_ack_ranges(); - return ReceiveOutcome::Duplicate; - } - - self.accepted_records.insert(single_range(seq)); - self.trim_accepted_records(); - - self.pending_ack.insert(single_range(seq)); + /// queues an ack for `seq` + pub fn push(&mut self, seq: RecordSeq) { + self.pending_ack.insert(single_range(seq.0)); self.trim_pending_ack_ranges(); - - ReceiveOutcome::New } pub fn ack_deadline(&self) -> Option { - match self.ack_state { - AckState::Idle => None, - AckState::Dirty { due_at } => Some(due_at), - } + self.ack_due_at } pub fn schedule_ack(&mut self, due_at: Instant) { - self.ack_state = match self.ack_state { - AckState::Dirty { due_at: old } => AckState::Dirty { - due_at: due_at.min(old), - }, - AckState::Idle => AckState::Dirty { due_at }, - }; + self.ack_due_at = Some(self.ack_due_at.map_or(due_at, |old| old.min(due_at))); } pub fn pending_ack(&self, max_wire_size: usize) -> Option { let due_at = self.ack_deadline()?; - if max_wire_size == 0 || self.pending_ack.range_count() == 0 { + let total_range_count = self.pending_ack.range_count(); + if max_wire_size == 0 || total_range_count == 0 { return None; } - let total_range_count = self.pending_ack.range_count(); let mut ack = RecordAckBuilder::new(); let mut selected_range_count = 0usize; @@ -109,23 +69,16 @@ impl AckTracker { } pub fn on_ack_emitted(&mut self, pending_ack: &PendingAck) { - self.retire_acked_ranges(&pending_ack.ack); - if pending_ack.includes_all_pending || self.pending_ack.range_count() == 0 { - self.ack_state = AckState::Idle; - } - } - - pub fn retire_acked_ranges(&mut self, ack: &RecordAck) { - for range in ack.ranges() { + for range in pending_ack.ack.ranges() { self.pending_ack.remove(from_ack_range(range)); } - if self.pending_ack.range_count() == 0 { - self.ack_state = AckState::Idle; + if pending_ack.includes_all_pending || self.pending_ack.range_count() == 0 { + self.ack_due_at = None; } } pub fn clear_ack_state(&mut self) { - self.ack_state = AckState::Idle; + self.ack_due_at = None; } pub fn restore_acked_ranges(&mut self, ack: &RecordAck, due_at: Instant) { @@ -136,20 +89,6 @@ impl AckTracker { self.schedule_ack(due_at); } - fn accepted_cutoff(&self, largest_accepted: u64) -> u64 { - largest_accepted - .saturating_add(1) - .saturating_sub(self.accepted_record_window) - } - - fn trim_accepted_records(&mut self) { - let Some(largest_accepted) = self.accepted_records.max() else { - return; - }; - let cutoff = self.accepted_cutoff(largest_accepted); - self.accepted_records.remove(0..cutoff); - } - fn trim_pending_ack_ranges(&mut self) { while self.pending_ack.range_count() > self.pending_ack_range_limit { self.pending_ack.pop_min(); @@ -157,10 +96,6 @@ impl AckTracker { } } -fn single_range(seq: u64) -> std::ops::Range { - seq..seq.checked_add(1).unwrap() -} - fn to_ack_range(range: std::ops::Range) -> RangeInclusive { let end = range.end.checked_sub(1).unwrap(); RecordSeq(range.start)..=RecordSeq(end) @@ -178,7 +113,7 @@ mod tests { use ql_wire::RecordSeq; - use super::{AckTracker, PendingAck, ReceiveOutcome}; + use super::{AckTracker, PendingAck}; fn ack_ranges(pending_ack: &PendingAck) -> Vec<(u64, u64)> { pending_ack @@ -191,11 +126,11 @@ mod tests { #[test] fn contiguous_records_emit_one_ack_range() { let now = Instant::now(); - let mut ack_tracker = AckTracker::new(128, 8); + let mut ack_tracker = AckTracker::new(8); - assert_eq!(ack_tracker.insert(RecordSeq(10)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(11)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(12)), ReceiveOutcome::New); + ack_tracker.push(RecordSeq(10)); + ack_tracker.push(RecordSeq(11)); + ack_tracker.push(RecordSeq(12)); ack_tracker.schedule_ack(now); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); @@ -205,36 +140,26 @@ mod tests { #[test] fn sparse_records_emit_descending_ack_ranges() { let now = Instant::now(); - let mut ack_tracker = AckTracker::new(128, 8); + let mut ack_tracker = AckTracker::new(8); - assert_eq!(ack_tracker.insert(RecordSeq(10)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(15)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(16)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(12)), ReceiveOutcome::New); + ack_tracker.push(RecordSeq(10)); + ack_tracker.push(RecordSeq(15)); + ack_tracker.push(RecordSeq(16)); + ack_tracker.push(RecordSeq(12)); ack_tracker.schedule_ack(now + Duration::from_millis(5)); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); assert_eq!(ack_ranges(&pending_ack), vec![(15, 16), (12, 12), (10, 10)]); } - #[test] - fn accepted_record_window_evicts_old_sequences() { - let mut ack_tracker = AckTracker::new(4, 8); - - assert_eq!(ack_tracker.insert(RecordSeq(10)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(15)), ReceiveOutcome::New); - - assert_eq!(ack_tracker.insert(RecordSeq(10)), ReceiveOutcome::TooOld); - } - #[test] fn pending_ack_range_limit_drops_oldest_low_ranges() { let now = Instant::now(); - let mut ack_tracker = AckTracker::new(128, 2); + let mut ack_tracker = AckTracker::new(2); - assert_eq!(ack_tracker.insert(RecordSeq(1)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(3)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(5)), ReceiveOutcome::New); + ack_tracker.push(RecordSeq(1)); + ack_tracker.push(RecordSeq(3)); + ack_tracker.push(RecordSeq(5)); ack_tracker.schedule_ack(now); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); @@ -242,19 +167,18 @@ mod tests { } #[test] - fn retire_acked_ranges_removes_only_exact_snapshot() { + fn emitting_an_ack_retires_only_its_own_ranges() { let now = Instant::now(); - let mut ack_tracker = AckTracker::new(128, 8); + let mut ack_tracker = AckTracker::new(8); - assert_eq!(ack_tracker.insert(RecordSeq(1)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(3)), ReceiveOutcome::New); - assert_eq!(ack_tracker.insert(RecordSeq(5)), ReceiveOutcome::New); + ack_tracker.push(RecordSeq(1)); + ack_tracker.push(RecordSeq(3)); + ack_tracker.push(RecordSeq(5)); ack_tracker.schedule_ack(now); let first_ack = ack_tracker.pending_ack(4).unwrap(); assert_eq!(ack_ranges(&first_ack), vec![(5, 5)]); ack_tracker.on_ack_emitted(&first_ack); - ack_tracker.retire_acked_ranges(&first_ack.ack); let second_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); assert_eq!(ack_ranges(&second_ack), vec![(3, 3), (1, 1)]); diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index a24da78f..4eb12db1 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -3,6 +3,7 @@ pub use self::{state::TerminalFrame, stream_ops::*, stream_parity::*, stream_rx: mod ack_tracker; mod range_set; mod remote_stream_history; +mod replay_window; mod state; mod stream_ops; mod stream_parity; @@ -25,8 +26,9 @@ use ql_wire::{ }; use self::{ - ack_tracker::{AckTracker, PendingAck, ReceiveOutcome}, + ack_tracker::{AckTracker, PendingAck}, remote_stream_history::RemoteStreamHistory, + replay_window::ReplayWindow, state::{InboundState, OutboundState, SessionPhase, SessionState, StreamRole, StreamState}, stream_tx::StreamTxRange, tracked::{LossRecovery, TrackedFrame, TrackedRecord, TrackedStreamData}, @@ -35,23 +37,29 @@ use crate::{NoSessionError, StreamError, StreamResetEvent, StreamResetTarget}; #[derive(Debug, Clone, Copy)] pub struct SessionConfig { - pub local_parity: StreamParity, + /// maximum total wire size for one session record, including header and auth tag pub record_max_size: usize, + /// delay before sending a pure record ack pub ack_delay: Duration, + /// initial wait before resending unacked session records pub retransmit_timeout: Duration, + /// idle delay before sending a keepalive ping pub keepalive_interval: Duration, + /// how long to wait before declaring the peer dead pub peer_timeout: Duration, + /// maximum bytes buffered locally for one stream send side pub stream_send_buffer_size: usize, + /// maximum bytes buffered locally for one stream receive side pub stream_receive_buffer_size: u32, - pub initial_peer_stream_receive_window: u32, + /// how many accepted record sequence numbers to retain for replay detection pub accepted_record_window: u64, + /// maximum disjoint pending ACK ranges to retain before dropping the oldest low ranges pub pending_ack_range_limit: usize, } impl Default for SessionConfig { fn default() -> Self { Self { - local_parity: StreamParity::Even, record_max_size: 8 * 1024, ack_delay: Duration::from_millis(5), retransmit_timeout: Duration::from_secs(1), @@ -59,13 +67,28 @@ impl Default for SessionConfig { peer_timeout: Duration::from_secs(30), stream_send_buffer_size: 16 * 1024, stream_receive_buffer_size: 16 * 1024, - initial_peer_stream_receive_window: 16 * 1024, accepted_record_window: 4096, pending_ack_range_limit: 64, } } } +/// per-session values settled by the handshake +#[derive(Debug, Clone, Copy)] +pub struct SessionParams { + pub local_parity: StreamParity, + pub initial_stream_receive_window: u32, +} + +impl Default for SessionParams { + fn default() -> Self { + Self { + local_parity: StreamParity::Even, + initial_stream_receive_window: 16 * 1024, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum SessionEvent { Opened(StreamId), @@ -93,20 +116,20 @@ where pub struct SessionFsm { config: SessionConfig, + params: SessionParams, state: SessionState, } impl SessionFsm { - pub fn new(mut config: SessionConfig, now: Instant) -> Self { + pub fn new(mut config: SessionConfig, params: SessionParams, now: Instant) -> Self { config.record_max_size = config .record_max_size .max(SessionRecordBuilder::MIN_CAPACITY); config.stream_send_buffer_size = config.stream_send_buffer_size.max(1); config.stream_receive_buffer_size = config.stream_receive_buffer_size.max(1); - config.accepted_record_window = config.accepted_record_window.max(1); - config.pending_ack_range_limit = config.pending_ack_range_limit.max(1); Self { config, + params, state: SessionState { last_activity_at: now, last_inbound_at: now, @@ -116,14 +139,12 @@ impl SessionFsm { next_write_id: 0, tracked_records: IndexMap::default(), loss_recovery: LossRecovery::new(config.retransmit_timeout), - ack_tracker: AckTracker::new( - config.accepted_record_window, - config.pending_ack_range_limit, - ), + replay_window: ReplayWindow::new(config.accepted_record_window), + ack_tracker: AckTracker::new(config.pending_ack_range_limit), pending_ping: false, streams: IndexMap::default(), next_stream_index: 0, - remote_stream_history: RemoteStreamHistory::new(config.local_parity.remote()), + remote_stream_history: RemoteStreamHistory::new(params.local_parity.remote()), }, } } @@ -138,7 +159,7 @@ impl SessionFsm { { self.ensure_session_open()?; let stream_id = self - .config + .params .local_parity .make_stream_id(self.state.next_stream_ordinal); self.state.next_stream_ordinal = self.state.next_stream_ordinal.saturating_add(1); @@ -148,7 +169,7 @@ impl SessionFsm { StreamRole::Initiator, Some(Bytes::from(header)), self.config.stream_receive_buffer_size, - self.config.initial_peer_stream_receive_window, + self.params.initial_stream_receive_window, ), ); let stream_index = self.state.streams.len() - 1; @@ -202,6 +223,10 @@ impl SessionFsm { self.state.phase == SessionPhase::Closed } + pub fn is_replay(&self, seq: RecordSeq) -> bool { + self.state.replay_window.is_replay(seq) + } + pub fn receive(&mut self, now: Instant, seq: RecordSeq, frames: I, sink: &mut impl EventSink) where I: IntoIterator, ql_wire::Error>>, @@ -210,18 +235,13 @@ impl SessionFsm { return; } - self.state.last_activity_at = now; - self.state.last_inbound_at = now; self.collect_timeouts(now); - match self.state.ack_tracker.insert(seq) { - ReceiveOutcome::TooOld => return, - ReceiveOutcome::Duplicate => { - self.schedule_ack(now, true); - return; - } - ReceiveOutcome::New => {} - } + self.state.replay_window.accept(seq); + self.state.ack_tracker.push(seq); + + self.state.last_activity_at = now; + self.state.last_inbound_at = now; let mut ack_eliciting = false; @@ -244,7 +264,7 @@ impl SessionFsm { return; } } - SessionFrame::StreamWindow(frame) => self.handle_stream_window(&frame, sink), + SessionFrame::StreamWindow(frame) => self.handle_stream_window(&frame), SessionFrame::StreamReset(frame) => { if self.handle_stream_reset(&frame, sink).is_err() { self.close(SessionCloseCode::PROTOCOL, sink); @@ -259,7 +279,7 @@ impl SessionFsm { } if ack_eliciting { - self.schedule_ack(now, false); + self.schedule_ack(now); } } @@ -293,7 +313,7 @@ impl SessionFsm { &mut self.state.ack_tracker, &mut self.state.pending_ping, &mut self.state.streams, - &record, + record, ); } } @@ -393,9 +413,7 @@ impl SessionFsm { let (builder, outbound) = self.build_next_record(now)?; - let should_track = outbound.ping_included - || !outbound.window_updates.is_empty() - || !outbound.frames.is_empty(); + let should_track = outbound.ping_included || !outbound.frames.is_empty(); let write_id = should_track.then(|| { debug_assert!(self.state.tracked_records.len() < TRACKED_RECORD_LIMIT); let write_id = self.state.next_write_id; @@ -415,7 +433,6 @@ impl SessionFsm { frames: Vec::new(), ack: None, ping_included: false, - window_updates: Vec::new(), sent_at: None, }; @@ -513,8 +530,8 @@ impl SessionFsm { stream.pending_window = false; stream.advertised_max_offset = *frame.maximum_offset; outbound - .window_updates - .push((stream_id, *frame.maximum_offset)); + .frames + .push(TrackedFrame::StreamWindow(stream_id, *frame.maximum_offset)); } } @@ -595,9 +612,10 @@ impl SessionFsm { let stream_send_buffer_size = self.config.stream_send_buffer_size; let mut latest_sent_at = None; let state = &mut self.state; - for (_, record) in state.tracked_records.extract_if(.., |_, record| { - record.sent_at.is_some() && ack.contains(record.seq.0) - }) { + for (_, record) in state + .tracked_records + .extract_if(.., |_, record| ack.contains(record.seq.0)) + { latest_sent_at = latest_sent_at.max(record.sent_at); for frame in &record.frames { acknowledge_tracked_frame(&mut state.streams, stream_send_buffer_size, frame, sink); @@ -611,12 +629,10 @@ impl SessionFsm { self.reap_reapable_streams(); } - fn schedule_ack(&mut self, now: Instant, immediate: bool) { - self.state.ack_tracker.schedule_ack(if immediate { - now - } else { - now + self.config.ack_delay - }); + fn schedule_ack(&mut self, now: Instant) { + self.state + .ack_tracker + .schedule_ack(now + self.config.ack_delay); } fn pending_ack(&self, remaining_capacity: usize) -> Option { @@ -636,7 +652,7 @@ impl SessionFsm { &mut state.ack_tracker, &mut state.pending_ping, &mut state.streams, - &record, + record, ); timed_out = true; } @@ -738,19 +754,15 @@ impl SessionFsm { Ok(()) } - fn handle_stream_window(&mut self, frame: &StreamWindow, sink: &mut impl EventSink) { + fn handle_stream_window(&mut self, frame: &StreamWindow) { let Some(stream) = self.state.streams.get_mut(&frame.stream_id) else { return; }; - let was_full = stream.send_capacity(self.config.stream_send_buffer_size) == 0; let maximum_offset = *frame.maximum_offset; if maximum_offset > stream.peer_max_offset { stream.peer_max_offset = maximum_offset; } - if was_full && stream.send_capacity(self.config.stream_send_buffer_size) > 0 { - sink.emit(SessionEvent::Writable(frame.stream_id)); - } } fn handle_stream_reset( @@ -820,67 +832,41 @@ impl SessionFsm { matches!(target, ResetTarget::Both) || role.outbound_target() == target } - fn stream_is_reapable(&self, stream_id: StreamId, stream: &StreamState) -> bool { - let tracked_refs_stream = self.state.tracked_records.values().any(|record| { - record.window_updates.iter().any(|(id, _)| *id == stream_id) - || record.frames.iter().any(|frame| match frame { - TrackedFrame::StreamData(frame) => frame.stream_id == stream_id, - TrackedFrame::StreamReset(frame) => frame.stream_id == stream_id, - }) - }); - if tracked_refs_stream { - return false; - } - - if !stream.tx.is_empty() - || stream.pending_reset.is_some() - || stream.pending_window - || stream.readable_bytes() > 0 - || stream.rx.buffered_end_offset() > stream.rx.start_offset() - { - return false; - } - - matches!( - stream.inbound_state, - InboundState::Finished | InboundState::Reset(_) | InboundState::Discarding - ) && matches!( - stream.outbound_state, - OutboundState::Finished | OutboundState::Closed - ) - } - fn reap_reapable_streams(&mut self) { - let mut index = 0usize; - while index < self.state.streams.len() { - let stream_id = *self.state.streams.get_index(index).unwrap().0; - let len_before = self.state.streams.len(); - self.try_reap_stream(stream_id); - if self.state.streams.len() == len_before { - index += 1; + let SessionState { + tracked_records, + streams, + next_stream_index, + .. + } = &mut self.state; + let old_start = *next_stream_index; + let mut old_index = 0usize; + let mut retained = 0usize; + let mut new_start = None; + + streams.retain(|&stream_id, stream| { + let keep = !stream_is_reapable(tracked_records, stream_id, stream); + if keep { + if old_index >= old_start && new_start.is_none() { + new_start = Some(retained); + } + retained += 1; } - } + old_index += 1; + keep + }); + + *next_stream_index = new_start.unwrap_or(0); } fn try_reap_stream(&mut self, stream_id: StreamId) { let Some(index) = self.state.streams.get_index_of(&stream_id) else { return; }; - self.try_reap_stream_at(stream_id, index); - } - - fn try_reap_stream_at(&mut self, stream_id: StreamId, index: usize) { - let Some((indexed_stream_id, stream)) = self.state.streams.get_index(index) else { - return; - }; - debug_assert_eq!(*indexed_stream_id, stream_id); - if !self.stream_is_reapable(stream_id, stream) { + let stream = &self.state.streams[index]; + if !stream_is_reapable(&self.state.tracked_records, stream_id, stream) { return; } - self.reap_stream_at(index); - } - - fn reap_stream_at(&mut self, index: usize) { self.state.streams.shift_remove_index(index); if self.state.streams.is_empty() { @@ -905,7 +891,7 @@ impl SessionFsm { stream_id: StreamId, ) -> Result, ()> { match classify_missing_stream( - self.config.local_parity, + self.params.local_parity, self.state.next_stream_ordinal, stream_id, &mut self.state.remote_stream_history, @@ -925,7 +911,7 @@ impl SessionFsm { StreamRole::Responder, None, self.config.stream_receive_buffer_size, - self.config.initial_peer_stream_receive_window, + self.params.initial_stream_receive_window, )); Ok(Some(stream.into_mut())) @@ -974,53 +960,77 @@ fn restore_tracked_record( ack_tracker: &mut AckTracker, pending_ping: &mut bool, streams: &mut IndexMap, - record: &TrackedRecord, + record: TrackedRecord, ) { - if let Some(ack) = &record.ack { - ack_tracker.restore_acked_ranges(ack, now); + if let Some(ack) = record.ack { + ack_tracker.restore_acked_ranges(&ack, now); } if record.ping_included { *pending_ping = true; } - for &(stream_id, maximum_offset) in &record.window_updates { - if let Some(stream) = streams.get_mut(&stream_id) { - if stream.recv_limit() >= maximum_offset { - stream.pending_window = true; + for frame in record.frames { + match frame { + TrackedFrame::StreamReset(reset) => { + if let Some(stream) = streams.get_mut(&reset.stream_id) { + stream.pending_reset = Some(reset); + } + } + TrackedFrame::StreamData(frame) => { + let Some(stream) = streams.get_mut(&frame.stream_id) else { + continue; + }; + if matches!(stream.outbound_state, OutboundState::Closed) { + continue; + } + stream.tx.retransmit(StreamTxRange { + offset: frame.offset, + len: frame.len, + fin: frame.fin, + }); + if frame.fin && matches!(stream.outbound_state, OutboundState::Finished) { + stream.outbound_state = OutboundState::FinQueued; + } + } + TrackedFrame::StreamWindow(stream_id, maximum_offset) => { + if let Some(stream) = streams.get_mut(&stream_id) { + stream.pending_window |= stream.recv_limit() >= maximum_offset; + } } } } - for frame in &record.frames { - requeue_tracked_frame(streams, frame); - } } -fn requeue_tracked_frame(streams: &mut IndexMap, frame: &TrackedFrame) { - match frame { - TrackedFrame::StreamReset(reset) => restore_stream_reset(streams, reset.clone()), - TrackedFrame::StreamData(frame) => restore_stream_data(streams, *frame), +fn stream_is_reapable( + tracked_records: &IndexMap, + stream_id: StreamId, + stream: &StreamState, +) -> bool { + let tracked_refs_stream = tracked_records.values().any(|record| { + record + .frames + .iter() + .any(|frame| frame.references_stream(stream_id)) + }); + if tracked_refs_stream { + return false; } -} -fn restore_stream_reset(streams: &mut IndexMap, reset: StreamReset) { - if let Some(stream) = streams.get_mut(&reset.stream_id) { - stream.pending_reset = Some(reset); + if !stream.tx.is_empty() + || stream.pending_reset.is_some() + || stream.pending_window + || stream.readable_bytes() > 0 + || stream.rx.buffered_end_offset() > stream.rx.start_offset() + { + return false; } -} -fn restore_stream_data(streams: &mut IndexMap, frame: TrackedStreamData) { - if let Some(stream) = streams.get_mut(&frame.stream_id) { - if matches!(stream.outbound_state, OutboundState::Closed) { - return; - } - stream.tx.retransmit(stream_tx::StreamTxRange { - offset: frame.offset, - len: frame.len, - fin: frame.fin, - }); - if frame.fin && matches!(stream.outbound_state, OutboundState::Finished) { - stream.outbound_state = OutboundState::FinQueued; - } - } + matches!( + (&stream.inbound_state, &stream.outbound_state), + ( + InboundState::Finished | InboundState::Reset(_) | InboundState::Discarding, + OutboundState::Finished | OutboundState::Closed, + ) + ) } fn acknowledge_tracked_frame( @@ -1030,7 +1040,7 @@ fn acknowledge_tracked_frame( sink: &mut impl EventSink, ) { match frame { - TrackedFrame::StreamReset(_) => {} + TrackedFrame::StreamReset(_) | TrackedFrame::StreamWindow(..) => {} TrackedFrame::StreamData(frame) => { let stream_id = frame.stream_id; if let Some(stream) = streams.get_mut(&stream_id) { diff --git a/ql-fsm/src/session/range_set.rs b/ql-fsm/src/session/range_set.rs index 53d66269..06ec3e30 100644 --- a/ql-fsm/src/session/range_set.rs +++ b/ql-fsm/src/session/range_set.rs @@ -11,6 +11,10 @@ use std::{ #[derive(Debug, Default, Clone, PartialEq, Eq)] pub struct RangeSet(BTreeMap); +pub fn single_range(value: u64) -> Range { + value..value.checked_add(1).unwrap() +} + impl RangeSet { pub fn new() -> Self { Self::default() diff --git a/ql-fsm/src/session/replay_window.rs b/ql-fsm/src/session/replay_window.rs new file mode 100644 index 00000000..2ff333bc --- /dev/null +++ b/ql-fsm/src/session/replay_window.rs @@ -0,0 +1,62 @@ +use ql_wire::RecordSeq; + +use super::range_set::{single_range, RangeSet}; + +#[derive(Debug, Clone)] +pub struct ReplayWindow { + accepted: RangeSet, + window: u64, +} + +impl ReplayWindow { + pub fn new(window: u64) -> Self { + Self { + accepted: RangeSet::new(), + window: window.max(1), + } + } + + /// returns true when `seq` was already accepted or has aged out of the window + pub fn is_replay(&self, seq: RecordSeq) -> bool { + let Some(cutoff) = self.cutoff() else { + return false; + }; + seq.0 < cutoff || self.accepted.contains(seq.0) + } + + /// records `seq` as accepted + /// + /// callers must reject replays with `is_replay` first + pub fn accept(&mut self, seq: RecordSeq) { + debug_assert!(!self.is_replay(seq), "accepting a replayed seq"); + self.accepted.insert(single_range(seq.0)); + self.trim(); + } + + fn cutoff(&self) -> Option { + Some(self.accepted.max()?.saturating_sub(self.window - 1)) + } + + fn trim(&mut self) { + if let Some(cutoff) = self.cutoff() { + self.accepted.remove(0..cutoff); + } + } +} + +#[cfg(test)] +mod tests { + use ql_wire::RecordSeq; + + use super::ReplayWindow; + + #[test] + fn window_evicts_old_sequences() { + let mut window = ReplayWindow::new(4); + + window.accept(RecordSeq(10)); + window.accept(RecordSeq(15)); + + assert!(window.is_replay(RecordSeq(10))); + } +} diff --git a/ql-fsm/src/session/state.rs b/ql-fsm/src/session/state.rs index 7e2e8a07..98bf19aa 100644 --- a/ql-fsm/src/session/state.rs +++ b/ql-fsm/src/session/state.rs @@ -8,6 +8,7 @@ use ql_wire::{RecordSeq, ResetTarget, SessionClose, StreamReset}; use super::{ ack_tracker::AckTracker, remote_stream_history::RemoteStreamHistory, + replay_window::ReplayWindow, stream_rx::StreamRx, stream_tx::StreamTx, tracked::{LossRecovery, TrackedRecord}, @@ -22,6 +23,7 @@ pub struct SessionState { pub next_write_id: u64, pub tracked_records: IndexMap, pub loss_recovery: LossRecovery, + pub replay_window: ReplayWindow, pub ack_tracker: AckTracker, pub pending_ping: bool, pub streams: IndexMap, diff --git a/ql-fsm/src/session/stream_ops.rs b/ql-fsm/src/session/stream_ops.rs index 5ab1c1b8..9ecd549b 100644 --- a/ql-fsm/src/session/stream_ops.rs +++ b/ql-fsm/src/session/stream_ops.rs @@ -13,7 +13,6 @@ pub struct StreamOps<'a, E> { emit: E, stream_id: StreamId, stream_index: usize, - reap_on_drop: bool, } impl<'a, E: EventSink> StreamOps<'a, E> { @@ -28,7 +27,6 @@ impl<'a, E: EventSink> StreamOps<'a, E> { emit, stream_id, stream_index, - reap_on_drop: false, } } @@ -73,7 +71,6 @@ impl<'a, E: EventSink> StreamOps<'a, E> { if emit_finished { self.emit.emit(SessionEvent::Finished(stream_id)); } - self.reap_on_drop = true; Ok(()) } @@ -102,7 +99,6 @@ impl<'a, E: EventSink> StreamOps<'a, E> { target: wire_target, code, }); - self.reap_on_drop = true; } #[inline] @@ -116,17 +112,6 @@ impl<'a, E: EventSink> StreamOps<'a, E> { } } -impl Drop for StreamOps<'_, E> { - fn drop(&mut self) { - if !self.reap_on_drop { - return; - } - - self.session - .try_reap_stream_at(self.stream_id, self.stream_index); - } -} - pub struct StreamWriter<'a> { stream: &'a mut StreamState, send_buffer_size: usize, diff --git a/ql-fsm/src/session/tests.rs b/ql-fsm/src/session/tests.rs index 6599ce8d..dd4becfb 100644 --- a/ql-fsm/src/session/tests.rs +++ b/ql-fsm/src/session/tests.rs @@ -8,7 +8,10 @@ use ql_wire::{ SessionRecordBuilder, StreamData, StreamReset, }; -use super::{SessionConfig, SessionEvent, SessionFsm}; +use super::{ + state::{InboundState, OutboundState}, + SessionConfig, SessionEvent, SessionFsm, SessionParams, +}; use crate::{session::stream_parity::StreamParity, StreamResetEvent}; const REFUSED: ResetCode = ResetCode(1); @@ -94,7 +97,7 @@ fn receive_events( #[test] fn outbound_record_seq_increments_monotonically() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = open_stream_id(&mut fsm); assert_eq!(write_stream_bytes(&mut fsm, stream_id, b"one"), 3); @@ -107,6 +110,43 @@ fn outbound_record_seq_increments_monotonically() { assert_eq!(second_seq, RecordSeq(1)); } +#[test] +fn stream_data_is_scheduled_round_robin() { + let now = Instant::now(); + let config = SessionConfig::default(); + let payload = vec![b'x'; config.stream_send_buffer_size]; + let mut fsm = SessionFsm::new(config, SessionParams::default(), now); + let a = open_stream_id(&mut fsm); + let b = open_stream_id(&mut fsm); + let c = open_stream_id(&mut fsm); + let d = open_stream_id(&mut fsm); + for stream_id in [a, c, d] { + assert_eq!( + write_stream_bytes(&mut fsm, stream_id, &payload), + payload.len() + ); + } + let next_stream_id = |fsm: &mut SessionFsm| { + let (_, frames) = next_outbound(fsm, now).unwrap(); + let [SessionFrame::StreamData(frame)] = frames.as_slice() else { + panic!("expected one stream data frame, got {frames:?}"); + }; + frame.stream_id + }; + + assert_eq!(next_stream_id(&mut fsm), a); + + let stream = fsm.state.streams.get_mut(&b).unwrap(); + stream.inbound_state = InboundState::Finished; + stream.outbound_state = OutboundState::Finished; + fsm.reap_reapable_streams(); + assert!(fsm.stream(b, |_| {}).is_err()); + + for expected in [c, d, a] { + assert_eq!(next_stream_id(&mut fsm), expected); + } +} + #[test] fn retransmit_uses_new_record_seq() { let now = Instant::now(); @@ -115,6 +155,7 @@ fn retransmit_uses_new_record_seq() { retransmit_timeout: Duration::from_millis(100), ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -139,6 +180,7 @@ fn retransmitted_record_ack_releases_stream_data() { stream_send_buffer_size: 4, ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -173,6 +215,7 @@ fn acknowledged_rtt_updates_retransmit_timeout() { retransmit_timeout: Duration::from_millis(100), ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -207,6 +250,7 @@ fn retransmit_timeout_backs_off() { retransmit_timeout: Duration::from_millis(20), ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -237,9 +281,12 @@ fn tracked_record_count_is_bounded() { + StreamData::>::MAX_WIRE_OVERHEAD + 1, stream_send_buffer_size: PAYLOAD_LEN, - initial_peer_stream_receive_window: PAYLOAD_LEN as u32, ..SessionConfig::default() }, + SessionParams { + initial_stream_receive_window: PAYLOAD_LEN as u32, + ..SessionParams::default() + }, now, ); let stream_id = open_stream_id(&mut fsm); @@ -271,6 +318,7 @@ fn lost_record_on_one_stream_does_not_block_another_stream() { + PAYLOAD_LEN, ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id_a = open_stream_id(&mut fsm); @@ -315,6 +363,7 @@ fn ack_reopens_write_capacity() { stream_send_buffer_size: 4, ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -340,7 +389,7 @@ fn ack_reopens_write_capacity() { #[test] fn ack_of_fin_emits_outbound_finished_once() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = open_stream_id(&mut fsm); assert_eq!(write_stream_bytes(&mut fsm, stream_id, b"done"), 4); @@ -393,10 +442,13 @@ fn commit_stream_read_is_what_advances_stream_window() { let now = Instant::now(); let mut fsm = SessionFsm::new( SessionConfig { - local_parity: StreamParity::Even, ack_delay: Duration::ZERO, ..SessionConfig::default() }, + SessionParams { + local_parity: StreamParity::Even, + ..SessionParams::default() + }, now, ); let stream_id = StreamId(1); @@ -442,6 +494,71 @@ fn commit_stream_read_is_what_advances_stream_window() { )); } +#[test] +fn lost_stream_window_is_resent_after_a_timeout() { + let now = Instant::now(); + let mut fsm = SessionFsm::new( + SessionConfig { + ack_delay: Duration::ZERO, + retransmit_timeout: Duration::from_millis(20), + ..SessionConfig::default() + }, + SessionParams::default(), + now, + ); + let stream_id = StreamId(1); + let data = vec![SessionFrame::StreamData(StreamData { + stream_id, + offset: Varint(0), + header: Some(vec![1_u8]), + fin: false, + bytes: b"hi".to_vec(), + })]; + receive_events(&mut fsm, now, RecordSeq(7), &data); + next_outbound(&mut fsm, now + Duration::from_millis(1)).unwrap(); + + read_stream_all(&mut fsm, stream_id); + let (_first_seq, first) = next_outbound(&mut fsm, now + Duration::from_millis(2)).unwrap(); + let [SessionFrame::StreamWindow(sent)] = first.as_slice() else { + panic!("expected a window update, got {first:?}"); + }; + let sent_offset = *sent.maximum_offset; + + let mut emit = |_| {}; + fsm.on_timer(now + Duration::from_millis(23), &mut emit); + + let (_resent_seq, resent) = next_outbound(&mut fsm, now + Duration::from_millis(23)).unwrap(); + assert!(matches!( + resent.as_slice(), + [SessionFrame::StreamWindow(window)] + if window.stream_id == stream_id && *window.maximum_offset == sent_offset + )); +} + +#[test] +fn lost_ping_is_resent_after_a_timeout() { + let now = Instant::now(); + let mut fsm = SessionFsm::new( + SessionConfig { + keepalive_interval: Duration::ZERO, + retransmit_timeout: Duration::from_millis(20), + ..SessionConfig::default() + }, + SessionParams::default(), + now, + ); + + fsm.queue_ping().unwrap(); + let (_first_seq, first) = next_outbound(&mut fsm, now).unwrap(); + assert!(matches!(first.as_slice(), [SessionFrame::Ping])); + + let mut emit = |_| {}; + fsm.on_timer(now + Duration::from_millis(21), &mut emit); + + let (_resent_seq, resent) = next_outbound(&mut fsm, now + Duration::from_millis(21)).unwrap(); + assert!(matches!(resent.as_slice(), [SessionFrame::Ping])); +} + #[test] fn pure_ack_only_records_are_fire_and_forget() { let now = Instant::now(); @@ -450,7 +567,7 @@ fn pure_ack_only_records_are_fire_and_forget() { ..SessionConfig::default() }; let retransmit_timeout = config.retransmit_timeout; - let mut fsm = SessionFsm::new(config, now); + let mut fsm = SessionFsm::new(config, SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(StreamData { stream_id, @@ -480,7 +597,7 @@ fn pure_ack_only_records_are_fire_and_forget() { #[test] fn inbound_stream_data_emits_opened_and_readable() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(ql_wire::StreamData { stream_id, @@ -509,7 +626,7 @@ fn inbound_stream_data_emits_opened_and_readable() { #[test] fn inbound_empty_fin_emits_finished_immediately() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(StreamData { stream_id, @@ -537,6 +654,7 @@ fn remote_stream_reset_is_reliable_and_retried() { retransmit_timeout: Duration::from_millis(100), ..SessionConfig::default() }, + SessionParams::default(), now, ); let stream_id = open_stream_id(&mut fsm); @@ -567,9 +685,10 @@ fn stream_ids_follow_even_odd_xid_ordering() { let odd = StreamParity::for_local(QID([2; QID::SIZE]), QID([1; QID::SIZE])); let even_id = SessionFsm::new( - SessionConfig { + SessionConfig::default(), + SessionParams { local_parity: even, - ..SessionConfig::default() + ..SessionParams::default() }, now, ) @@ -577,9 +696,10 @@ fn stream_ids_follow_even_odd_xid_ordering() { .unwrap() .stream_id(); let odd_id = SessionFsm::new( - SessionConfig { + SessionConfig::default(), + SessionParams { local_parity: odd, - ..SessionConfig::default() + ..SessionParams::default() }, now, ) @@ -594,7 +714,7 @@ fn stream_ids_follow_even_odd_xid_ordering() { #[test] fn duplicate_stream_data_is_not_redelivered() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(StreamData { stream_id, @@ -617,7 +737,7 @@ fn duplicate_stream_data_is_not_redelivered() { #[test] fn duplicate_remote_reset_after_reap_is_ignored() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let reset = StreamReset { stream_id: StreamId(1), target: ResetTarget::Both, @@ -647,7 +767,7 @@ fn duplicate_remote_reset_after_reap_is_ignored() { #[test] fn late_remote_stream_data_after_reset_is_ignored() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let reset = vec![SessionFrame::StreamReset(StreamReset { stream_id, @@ -684,7 +804,7 @@ fn late_remote_stream_data_after_reset_is_ignored() { #[test] fn duplicate_finished_remote_data_after_reap_is_ignored() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(StreamData { stream_id, @@ -721,7 +841,7 @@ fn duplicate_finished_remote_data_after_reap_is_ignored() { #[test] fn duplicate_finished_remote_data_before_read_is_ignored() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let stream_id = StreamId(1); let record = vec![SessionFrame::StreamData(StreamData { stream_id, @@ -758,7 +878,7 @@ fn duplicate_finished_remote_data_before_read_is_ignored() { #[test] fn out_of_order_remote_stream_first_observations_still_open_once_each() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let reset3 = vec![SessionFrame::StreamReset(StreamReset { stream_id: StreamId(3), target: ResetTarget::Both, @@ -807,7 +927,7 @@ fn out_of_order_remote_stream_first_observations_still_open_once_each() { #[test] fn invalid_remote_stream_reset_closes_session() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let invalid = vec![SessionFrame::StreamReset(StreamReset { stream_id: StreamId(0), @@ -832,6 +952,7 @@ fn close_does_not_ack_rejected_record_seq() { ack_delay: Duration::ZERO, ..SessionConfig::default() }, + SessionParams::default(), now, ); @@ -866,7 +987,7 @@ fn close_does_not_ack_rejected_record_seq() { #[test] fn inbound_unpair_emits_final_unpair_frame() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let events = receive_events(&mut fsm, now, RecordSeq(1), &[SessionFrame::Unpair]); assert_eq!(events, vec![SessionEvent::Unpaired]); @@ -880,7 +1001,7 @@ fn inbound_unpair_emits_final_unpair_frame() { #[test] fn terminating_session_ignores_inbound_frames() { let now = Instant::now(); - let mut fsm = SessionFsm::new(SessionConfig::default(), now); + let mut fsm = SessionFsm::new(SessionConfig::default(), SessionParams::default(), now); let mut events = Vec::new(); fsm.unpair(&mut |event| events.push(event)); @@ -903,9 +1024,10 @@ fn terminating_session_ignores_inbound_frames() { fn initial_peer_stream_receive_window_limits_first_send() { let now = Instant::now(); let mut fsm = SessionFsm::new( - SessionConfig { - initial_peer_stream_receive_window: 3, - ..SessionConfig::default() + SessionConfig::default(), + SessionParams { + initial_stream_receive_window: 3, + ..SessionParams::default() }, now, ); @@ -945,7 +1067,6 @@ fn initial_peer_stream_receive_window_limits_first_send() { fn sparse_out_of_order_ack_ranges_page_and_quiesce() { let now = Instant::now(); let sender_config = SessionConfig { - local_parity: StreamParity::Even, record_max_size: SessionRecordBuilder::MIN_CAPACITY + 1 // discriminator byte + StreamData::>::MAX_WIRE_OVERHEAD @@ -953,20 +1074,21 @@ fn sparse_out_of_order_ack_ranges_page_and_quiesce() { ack_delay: Duration::from_millis(5), retransmit_timeout: Duration::from_millis(25), stream_send_buffer_size: 8 * 1024, - initial_peer_stream_receive_window: 8 * 1024, ..SessionConfig::default() }; let receiver_config = SessionConfig { - local_parity: StreamParity::Odd, record_max_size: SessionRecordBuilder::MIN_CAPACITY + 10, ack_delay: Duration::from_millis(1), retransmit_timeout: Duration::from_millis(25), pending_ack_range_limit: 512, - initial_peer_stream_receive_window: 8 * 1024, ..SessionConfig::default() }; - let mut sender = SessionFsm::new(sender_config, now); - let mut receiver = SessionFsm::new(receiver_config, now); + let params = |local_parity| SessionParams { + local_parity, + initial_stream_receive_window: 8 * 1024, + }; + let mut sender = SessionFsm::new(sender_config, params(StreamParity::Even), now); + let mut receiver = SessionFsm::new(receiver_config, params(StreamParity::Odd), now); let stream_id = open_stream_id(&mut sender); let payload = vec![b'x'; 1200]; @@ -1032,6 +1154,7 @@ fn stream_header_larger_than_the_record_budget_does_not_panic() { record_max_size, ..SessionConfig::default() }, + SessionParams::default(), now, ); diff --git a/ql-fsm/src/session/tracked.rs b/ql-fsm/src/session/tracked.rs index f8a234ca..e03716fd 100644 --- a/ql-fsm/src/session/tracked.rs +++ b/ql-fsm/src/session/tracked.rs @@ -9,9 +9,9 @@ use ql_wire::{RecordAck, RecordSeq, StreamReset}; pub struct TrackedRecord { pub seq: RecordSeq, pub frames: Vec, + /// separate from `frames`, which only holds ack-eliciting ones pub ack: Option, pub ping_included: bool, - pub window_updates: Vec<(StreamId, u64)>, pub sent_at: Option, } @@ -19,6 +19,17 @@ pub struct TrackedRecord { pub enum TrackedFrame { StreamData(TrackedStreamData), StreamReset(StreamReset), + StreamWindow(StreamId, u64), +} + +impl TrackedFrame { + pub fn references_stream(&self, stream_id: StreamId) -> bool { + match self { + Self::StreamData(frame) => frame.stream_id == stream_id, + Self::StreamReset(frame) => frame.stream_id == stream_id, + Self::StreamWindow(id, _) => *id == stream_id, + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/ql-fsm/src/tests/handshake.rs b/ql-fsm/src/tests/handshake.rs index 3ffe083a..2b6f20aa 100644 --- a/ql-fsm/src/tests/handshake.rs +++ b/ql-fsm/src/tests/handshake.rs @@ -55,11 +55,17 @@ fn xx_connect_round_trip_establishes_transport_when_armed() { fn ik_connect_learns_remote_initial_stream_receive_window() { let mut harness = Harness::paired_known_with_configs( QlFsmConfig { - session_stream_receive_buffer_size: 9, + session: SessionConfig { + stream_receive_buffer_size: 9, + ..SessionConfig::default() + }, ..QlFsmConfig::default() }, QlFsmConfig { - session_stream_receive_buffer_size: 3, + session: SessionConfig { + stream_receive_buffer_size: 3, + ..SessionConfig::default() + }, ..QlFsmConfig::default() }, ); diff --git a/ql-fsm/src/tests/mod.rs b/ql-fsm/src/tests/mod.rs index 60986ae3..fed9db6b 100644 --- a/ql-fsm/src/tests/mod.rs +++ b/ql-fsm/src/tests/mod.rs @@ -11,7 +11,7 @@ use ql_wire::{ }; use crate::{ - session::{SessionConfig, SessionFsm, StreamParity}, + session::{SessionConfig, SessionFsm, SessionParams, StreamParity}, state::{ConnectedState, LinkState, SessionTransport}, Event, NoPeerError, OutboundWrite, PairingInvite, QlFsm, QlFsmConfig, WriteId, }; @@ -111,10 +111,15 @@ impl Harness { .b .fsm .config - .session_stream_receive_buffer_size, + .session + .stream_receive_buffer_size, }, }, - session: SessionFsm::new(session_config(&harness, true), harness.now), + session: SessionFsm::new( + harness.a.fsm.config.session, + session_params(&harness, true), + harness.now, + ), }); harness.b.fsm.state.link = LinkState::Connected(ConnectedState { handshake_id: HandshakeId(0), @@ -127,10 +132,15 @@ impl Harness { .a .fsm .config - .session_stream_receive_buffer_size, + .session + .stream_receive_buffer_size, }, }, - session: SessionFsm::new(session_config(&harness, false), harness.now), + session: SessionFsm::new( + harness.b.fsm.config.session, + session_params(&harness, false), + harness.now, + ), }); harness } @@ -296,36 +306,25 @@ fn pairing_token(byte: u8) -> PairingToken { PairingToken([byte; PairingToken::SIZE]) } -fn session_config(harness: &Harness, a: bool) -> SessionConfig { - let (local, peer, config) = if a { +fn session_params(harness: &Harness, a: bool) -> SessionParams { + let (local, remote) = if a { ( harness.a.fsm.identity.qid, harness.a.fsm.state.peer.as_ref().unwrap().qid, - harness.a.fsm.config, ) } else { ( harness.b.fsm.identity.qid, harness.b.fsm.state.peer.as_ref().unwrap().qid, - harness.b.fsm.config, ) }; - SessionConfig { - local_parity: StreamParity::for_local(local, peer), - record_max_size: config.session_record_max_size, - ack_delay: config.session_record_ack_delay, - retransmit_timeout: config.session_record_retransmit_timeout, - keepalive_interval: config.session_keepalive_interval, - peer_timeout: config.session_peer_timeout, - stream_send_buffer_size: config.session_stream_send_buffer_size, - stream_receive_buffer_size: config.session_stream_receive_buffer_size, - accepted_record_window: config.session_accepted_record_window, - pending_ack_range_limit: config.session_pending_ack_range_limit, - initial_peer_stream_receive_window: if a { - harness.b.fsm.config.session_stream_receive_buffer_size + SessionParams { + local_parity: StreamParity::for_local(local, remote), + initial_stream_receive_window: if a { + harness.b.fsm.config.session.stream_receive_buffer_size } else { - harness.a.fsm.config.session_stream_receive_buffer_size + harness.a.fsm.config.session.stream_receive_buffer_size }, } } diff --git a/ql-fsm/src/tests/proptest.rs b/ql-fsm/src/tests/proptest.rs index 3e397b2d..b932a1fb 100644 --- a/ql-fsm/src/tests/proptest.rs +++ b/ql-fsm/src/tests/proptest.rs @@ -151,11 +151,13 @@ struct Runner { impl Runner { fn handshake() -> Self { let config = QlFsmConfig { + session: SessionConfig { + ack_delay: Duration::from_millis(5), + retransmit_timeout: Duration::from_millis(15), + peer_timeout: Duration::from_millis(80), + ..SessionConfig::default() + }, handshake_timeout: Duration::from_millis(60), - session_record_ack_delay: Duration::from_millis(5), - session_record_retransmit_timeout: Duration::from_millis(15), - session_peer_timeout: Duration::from_millis(80), - ..QlFsmConfig::default() }; Self { @@ -175,9 +177,12 @@ impl Runner { fn connected() -> Self { let config = QlFsmConfig { - session_record_ack_delay: Duration::from_millis(5), - session_record_retransmit_timeout: Duration::from_millis(15), - session_peer_timeout: Duration::from_secs(5), + session: SessionConfig { + ack_delay: Duration::from_millis(5), + retransmit_timeout: Duration::from_millis(15), + peer_timeout: Duration::from_secs(5), + ..SessionConfig::default() + }, ..QlFsmConfig::default() }; Self::connected_with_config(config) @@ -366,8 +371,9 @@ impl Runner { .a .fsm .config - .session_record_retransmit_timeout - .max(self.harness.a.fsm.config.session_record_ack_delay) + .session + .retransmit_timeout + .max(self.harness.a.fsm.config.session.ack_delay) + Duration::from_millis(1); self.reject_all_taken(); @@ -965,13 +971,7 @@ proptest_crate::proptest! { payload in vec(any::(), 512..2048), actions in vec(packet_loss_recovery_action_strategy(), 1..96), ) { - let config = QlFsmConfig { - session_record_ack_delay: Duration::from_millis(1), - session_record_retransmit_timeout: Duration::from_millis(10), - session_record_max_size: ql_wire::SessionRecordBuilder::MIN_CAPACITY + 94, - session_pending_ack_range_limit: 512, - ..QlFsmConfig::default() - }; + let config = QlFsmConfig { session: SessionConfig { ack_delay: Duration::from_millis(1), retransmit_timeout: Duration::from_millis(10), record_max_size: ql_wire::SessionRecordBuilder::MIN_CAPACITY + 94, pending_ack_range_limit: 512, ..SessionConfig::default() }, ..QlFsmConfig::default() }; let mut runner = Runner::connected_with_config(config); runner.apply(&Action::open_stream(Side::A, 0)); diff --git a/ql-fsm/src/tests/session.rs b/ql-fsm/src/tests/session.rs index 2f7386a6..79cebec6 100644 --- a/ql-fsm/src/tests/session.rs +++ b/ql-fsm/src/tests/session.rs @@ -74,7 +74,7 @@ fn connected_fsms_deliver_stream_data() { harness.take_event(Side::B), Some(Event::Finished(stream_id)) ); - harness.advance(QlFsmConfig::default().session_record_ack_delay); + harness.advance(QlFsmConfig::default().session.ack_delay); harness.on_timer(Side::B); harness.pump(); assert_eq!( @@ -96,7 +96,7 @@ fn session_retransmit_uses_new_record_seq() { let first = harness.next_decoded_outbound(Side::A).unwrap(); - harness.advance(config.session_record_retransmit_timeout + Duration::from_millis(1)); + harness.advance(config.session.retransmit_timeout + Duration::from_millis(1)); harness.on_timer(Side::A); let retried = harness.next_decoded_outbound(Side::A).unwrap(); @@ -105,7 +105,7 @@ fn session_retransmit_uses_new_record_seq() { assert_eq!(retried.frames, first.frames); harness.deliver(Side::B, retried.record); - harness.advance(config.session_record_ack_delay); + harness.advance(config.session.ack_delay); harness.on_timer(Side::A); harness.on_timer(Side::B); harness.pump(); @@ -120,7 +120,7 @@ fn session_retransmit_uses_new_record_seq() { b"retry".to_vec() ); - harness.advance(config.session_record_retransmit_timeout + Duration::from_millis(1)); + harness.advance(config.session.retransmit_timeout + Duration::from_millis(1)); harness.on_timer(Side::A); assert!(harness.next_outbound(Side::A).is_none()); } @@ -293,12 +293,12 @@ fn unconfirmed_session_write_does_not_start_retransmit_timer() { let first = harness.next_decoded_write(Side::A).unwrap(); let id = first.write_id.expect("expected session write"); - harness.advance(config.session_record_retransmit_timeout + Duration::from_millis(1)); + harness.advance(config.session.retransmit_timeout + Duration::from_millis(1)); harness.on_timer(Side::A); assert!(harness.next_write(Side::A).is_none()); harness.confirm_write(Side::A, id); - harness.advance(config.session_record_retransmit_timeout + Duration::from_millis(1)); + harness.advance(config.session.retransmit_timeout + Duration::from_millis(1)); harness.on_timer(Side::A); let retried = harness.next_decoded_write(Side::A).unwrap(); @@ -310,7 +310,10 @@ fn unconfirmed_session_write_does_not_start_retransmit_timer() { #[test] fn ack_frame_releases_stream_capacity_and_emits_writable() { let config = QlFsmConfig { - session_stream_send_buffer_size: 4, + session: SessionConfig { + stream_send_buffer_size: 4, + ..SessionConfig::default() + }, ..QlFsmConfig::default() }; let mut harness = Harness::connected(config); @@ -327,7 +330,7 @@ fn ack_frame_releases_stream_capacity_and_emits_writable() { let record = harness.next_outbound(Side::A).unwrap(); harness.deliver(Side::B, record); - harness.advance(config.session_record_ack_delay); + harness.advance(config.session.ack_delay); harness.on_timer(Side::A); harness.on_timer(Side::B); harness.pump(); @@ -451,7 +454,7 @@ fn session_records_contain_ack_frames_after_delivery() { let data = harness.next_outbound(Side::A).unwrap(); harness.deliver(Side::B, data); - harness.advance(config.session_record_ack_delay); + harness.advance(config.session.ack_delay); harness.on_timer(Side::B); let ack = harness.next_decoded_outbound(Side::B).unwrap(); @@ -461,15 +464,84 @@ fn session_records_contain_ack_frames_after_delivery() { )); } +#[test] +fn duplicate_record_is_dropped_and_not_acked() { + let config = QlFsmConfig::default(); + let mut harness = Harness::connected(config); + + let stream_id = open_stream_id(&mut harness.a.fsm); + assert_eq!( + write_stream_bytes(&mut harness.a.fsm, stream_id, b"x").unwrap(), + 1 + ); + + let data = harness.next_outbound(Side::A).unwrap(); + harness.deliver(Side::B, data.clone()); + harness.advance(config.session.ack_delay); + harness.on_timer(Side::B); + harness.next_decoded_outbound(Side::B).unwrap(); + + harness.deliver(Side::B, data); + harness.advance(config.session.ack_delay); + harness.on_timer(Side::B); + assert!(harness.next_outbound(Side::B).is_none()); +} + +#[test] +fn replayed_record_does_not_renew_the_peer_timeout() { + let config = QlFsmConfig { + session: SessionConfig { + accepted_record_window: 1, + peer_timeout: Duration::from_millis(30), + ..SessionConfig::default() + }, + ..QlFsmConfig::default() + }; + let mut harness = Harness::connected(config); + + let stream_id = open_stream_id(&mut harness.a.fsm); + assert_eq!( + write_stream_bytes(&mut harness.a.fsm, stream_id, b"x").unwrap(), + 1 + ); + let first = harness.next_outbound(Side::A).unwrap(); + assert_eq!( + write_stream_bytes(&mut harness.a.fsm, stream_id, b"y").unwrap(), + 1 + ); + let second = harness.next_outbound(Side::A).unwrap(); + + harness.deliver(Side::B, first.clone()); + harness.deliver(Side::B, second); + harness.drain_events(Side::B); + + harness.advance(config.session.peer_timeout); + harness.deliver(Side::B, first); + harness.on_timer(Side::B); + + assert_eq!( + harness.drain_events(Side::B), + vec![Event::SessionClosed(SessionClose { + code: ql_wire::SessionCloseCode::TIMEOUT, + })] + ); +} + #[test] fn first_stream_data_uses_negotiated_initial_peer_credit() { let mut harness = Harness::paired_known_with_configs( QlFsmConfig { - session_stream_receive_buffer_size: 8, + session: SessionConfig { + stream_receive_buffer_size: 8, + ..SessionConfig::default() + }, ..QlFsmConfig::default() }, QlFsmConfig { - session_stream_receive_buffer_size: 3, + session: SessionConfig { + stream_receive_buffer_size: 3, + ..SessionConfig::default() + }, ..QlFsmConfig::default() }, ); @@ -495,12 +567,15 @@ fn first_stream_data_uses_negotiated_initial_peer_credit() { #[test] fn session_timeout_emits_close_before_disconnect() { let config = QlFsmConfig { - session_peer_timeout: Duration::from_millis(30), + session: SessionConfig { + peer_timeout: Duration::from_millis(30), + ..SessionConfig::default() + }, ..QlFsmConfig::default() }; let mut harness = Harness::connected(config); - harness.advance(config.session_peer_timeout); + harness.advance(config.session.peer_timeout); harness.on_timer(Side::A); assert_eq!( diff --git a/ql-runtime/src/tests/mod.rs b/ql-runtime/src/tests/mod.rs index e356d7f9..5160c8a9 100644 --- a/ql-runtime/src/tests/mod.rs +++ b/ql-runtime/src/tests/mod.rs @@ -13,7 +13,7 @@ use async_channel::{Receiver, Sender}; use futures_lite::Stream; use ql_codec::Decode; use ql_common::{StreamInfo, QID}; -use ql_fsm::PeerStatus; +use ql_fsm::{PeerStatus, SessionConfig}; use ql_wire::{ generate_identity, test_identities, PairingToken, PeerBundle, QlIdentity, RecordHeader, RecordType, SoftwareCrypto, @@ -616,11 +616,13 @@ async fn next_chunk(stream: &mut crate::StreamReader) -> Result>, fn default_runtime_config() -> RuntimeConfig { RuntimeConfig { fsm: QlFsmConfig { + session: SessionConfig { + retransmit_timeout: Duration::from_millis(30), + keepalive_interval: Duration::ZERO, + peer_timeout: Duration::ZERO, + ..SessionConfig::default() + }, handshake_timeout: Duration::from_millis(300), - session_record_retransmit_timeout: Duration::from_millis(30), - session_keepalive_interval: Duration::ZERO, - session_peer_timeout: Duration::ZERO, - ..Default::default() }, ..Default::default() } diff --git a/ql-runtime/src/tests/session.rs b/ql-runtime/src/tests/session.rs index 89066e00..4421be11 100644 --- a/ql-runtime/src/tests/session.rs +++ b/ql-runtime/src/tests/session.rs @@ -164,8 +164,11 @@ async fn session_timeout_disconnects_and_fails_pending_open() { run_local_test(async { let config_a = RuntimeConfig { fsm: QlFsmConfig { - session_keepalive_interval: Duration::from_millis(40), - session_peer_timeout: Duration::from_millis(60), + session: SessionConfig { + keepalive_interval: Duration::from_millis(40), + peer_timeout: Duration::from_millis(60), + ..default_runtime_config().fsm.session + }, ..default_runtime_config().fsm }, ..default_runtime_config() diff --git a/ql-runtime/src/tests/stream.rs b/ql-runtime/src/tests/stream.rs index af94b641..af8ab090 100644 --- a/ql-runtime/src/tests/stream.rs +++ b/ql-runtime/src/tests/stream.rs @@ -308,7 +308,10 @@ async fn stream_round_trip_survives_encrypted_packet_drops() { run_local_test(async { let config = RuntimeConfig { fsm: QlFsmConfig { - session_record_retransmit_timeout: Duration::from_millis(20), + session: SessionConfig { + retransmit_timeout: Duration::from_millis(20), + ..default_runtime_config().fsm.session + }, ..default_runtime_config().fsm }, ..default_runtime_config() @@ -387,13 +390,16 @@ async fn multi_megabyte_stream_survives_asymmetric_loss_and_delay() { let expected = payload.clone(); let config = RuntimeConfig { fsm: QlFsmConfig { - session_record_max_size: 16 * 1024, - session_record_ack_delay: Duration::from_millis(2), - session_record_retransmit_timeout: Duration::from_millis(25), - session_stream_send_buffer_size: 4 * 1024 * 1024, - session_stream_receive_buffer_size: 4 * 1024 * 1024, - session_accepted_record_window: 16 * 1024, - session_pending_ack_range_limit: 4 * 1024, + session: SessionConfig { + record_max_size: 16 * 1024, + ack_delay: Duration::from_millis(2), + retransmit_timeout: Duration::from_millis(25), + stream_send_buffer_size: 4 * 1024 * 1024, + stream_receive_buffer_size: 4 * 1024 * 1024, + accepted_record_window: 16 * 1024, + pending_ack_range_limit: 4 * 1024, + ..default_runtime_config().fsm.session + }, ..default_runtime_config().fsm }, ..default_runtime_config() @@ -506,13 +512,16 @@ async fn reproducer_writer_stalls_after_reverse_path_impairment() { .collect(); let config = RuntimeConfig { fsm: QlFsmConfig { - session_record_max_size: 16 * 1024, - session_record_ack_delay: Duration::from_millis(2), - session_record_retransmit_timeout: Duration::from_millis(25), - session_stream_send_buffer_size: 4 * 1024 * 1024, - session_stream_receive_buffer_size: 4 * 1024 * 1024, - session_accepted_record_window: 16 * 1024, - session_pending_ack_range_limit: 4 * 1024, + session: SessionConfig { + record_max_size: 16 * 1024, + ack_delay: Duration::from_millis(2), + retransmit_timeout: Duration::from_millis(25), + stream_send_buffer_size: 4 * 1024 * 1024, + stream_receive_buffer_size: 4 * 1024 * 1024, + accepted_record_window: 16 * 1024, + pending_ack_range_limit: 4 * 1024, + ..default_runtime_config().fsm.session + }, ..default_runtime_config().fsm }, ..default_runtime_config()