From a700fab03ef30dc3522df1157da07ac1fc1e4c71 Mon Sep 17 00:00:00 2001 From: Alex's Bot Date: Thu, 30 Jul 2026 12:45:50 +0000 Subject: [PATCH 1/8] ql-fsm: treat duplicate records as replays The record seq is the GCM nonce, so no retransmission can reuse one: every transmission carries a fresh seq. A repeated seq is therefore network duplication or a replay, never evidence that our ack was lost, which made the immediate ack it triggered pure reflection. Lost acks already recover through restore_acked_ranges when our own record times out. A replay is also no proof the peer is alive, so it must not refresh the liveness timers: one captured record replayed every 20s was enough to hold a session open against a peer that had gone away. Rejecting replays before decrypt_record saves spending an AES-GCM on them. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016YFf732Q7Rq4bf2PJR83LN --- ql-fsm/src/fsm.rs | 4 ++ ql-fsm/src/session/ack_tracker.rs | 68 +++++++++++++++---------------- ql-fsm/src/session/mod.rs | 33 ++++++++------- ql-fsm/src/tests/session.rs | 60 +++++++++++++++++++++++++++ 4 files changed, 112 insertions(+), 53 deletions(-) diff --git a/ql-fsm/src/fsm.rs b/ql-fsm/src/fsm.rs index a532c9bd..603d1bd9 100644 --- a/ql-fsm/src/fsm.rs +++ b/ql-fsm/src/fsm.rs @@ -132,6 +132,10 @@ pub fn receive( let (decrypt_len, seq) = { let record = wire::QlSessionRecord::decode(&mut reader) .map_err(|error| ReceiveError::wire(ReceiveStage::SessionRecord, error))?; + // checked here too, or else a replay costs a full AES-GCM for nothing + if conn.session.is_replay(record.header.seq) { + return Ok(()); + } let payload = wire::decrypt_record( crypto, &header, diff --git a/ql-fsm/src/session/ack_tracker.rs b/ql-fsm/src/session/ack_tracker.rs index 7f947d2d..6c787ba6 100644 --- a/ql-fsm/src/session/ack_tracker.rs +++ b/ql-fsm/src/session/ack_tracker.rs @@ -20,13 +20,6 @@ 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, @@ -44,25 +37,28 @@ impl AckTracker { } } - pub fn insert(&mut self, seq: RecordSeq) -> ReceiveOutcome { + /// returns true when `seq` was already accepted or has aged out of the window + pub fn is_replay(&self, seq: RecordSeq) -> bool { 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; + let Some(largest_accepted) = self.accepted_records.max() else { + return false; + }; + seq < self.accepted_cutoff(largest_accepted) || self.accepted_records.contains(seq) + } + + /// records `seq` and queues an ack for it, or returns false when the record must be dropped + pub fn accept(&mut self, seq: RecordSeq) -> bool { + if self.is_replay(seq) { + return false; } - self.accepted_records.insert(single_range(seq)); + self.accepted_records.insert(single_range(seq.0)); self.trim_accepted_records(); - self.pending_ack.insert(single_range(seq)); + self.pending_ack.insert(single_range(seq.0)); self.trim_pending_ack_ranges(); - ReceiveOutcome::New + true } pub fn ack_deadline(&self) -> Option { @@ -178,7 +174,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 @@ -193,9 +189,9 @@ mod tests { let now = Instant::now(); let mut ack_tracker = AckTracker::new(128, 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); + assert!(ack_tracker.accept(RecordSeq(10))); + assert!(ack_tracker.accept(RecordSeq(11))); + assert!(ack_tracker.accept(RecordSeq(12))); ack_tracker.schedule_ack(now); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); @@ -207,10 +203,10 @@ mod tests { let now = Instant::now(); let mut ack_tracker = AckTracker::new(128, 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); + assert!(ack_tracker.accept(RecordSeq(10))); + assert!(ack_tracker.accept(RecordSeq(15))); + assert!(ack_tracker.accept(RecordSeq(16))); + assert!(ack_tracker.accept(RecordSeq(12))); ack_tracker.schedule_ack(now + Duration::from_millis(5)); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); @@ -221,10 +217,10 @@ mod tests { 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!(ack_tracker.accept(RecordSeq(10))); + assert!(ack_tracker.accept(RecordSeq(15))); - assert_eq!(ack_tracker.insert(RecordSeq(10)), ReceiveOutcome::TooOld); + assert!(!ack_tracker.accept(RecordSeq(10))); } #[test] @@ -232,9 +228,9 @@ mod tests { let now = Instant::now(); let mut ack_tracker = AckTracker::new(128, 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); + assert!(ack_tracker.accept(RecordSeq(1))); + assert!(ack_tracker.accept(RecordSeq(3))); + assert!(ack_tracker.accept(RecordSeq(5))); ack_tracker.schedule_ack(now); let pending_ack = ack_tracker.pending_ack(usize::MAX).unwrap(); @@ -246,9 +242,9 @@ mod tests { let now = Instant::now(); let mut ack_tracker = AckTracker::new(128, 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); + assert!(ack_tracker.accept(RecordSeq(1))); + assert!(ack_tracker.accept(RecordSeq(3))); + assert!(ack_tracker.accept(RecordSeq(5))); ack_tracker.schedule_ack(now); let first_ack = ack_tracker.pending_ack(4).unwrap(); diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index a24da78f..babc739d 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -25,7 +25,7 @@ use ql_wire::{ }; use self::{ - ack_tracker::{AckTracker, PendingAck, ReceiveOutcome}, + ack_tracker::{AckTracker, PendingAck}, remote_stream_history::RemoteStreamHistory, state::{InboundState, OutboundState, SessionPhase, SessionState, StreamRole, StreamState}, stream_tx::StreamTxRange, @@ -202,6 +202,10 @@ impl SessionFsm { self.state.phase == SessionPhase::Closed } + pub fn is_replay(&self, seq: RecordSeq) -> bool { + self.state.ack_tracker.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,19 +214,16 @@ 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 => {} + // retransmissions carry a fresh seq, so a repeat is duplication or replay, never a lost ack + if !self.state.ack_tracker.accept(seq) { + return; } + self.state.last_activity_at = now; + self.state.last_inbound_at = now; + let mut ack_eliciting = false; for frame in frames { @@ -259,7 +260,7 @@ impl SessionFsm { } if ack_eliciting { - self.schedule_ack(now, false); + self.schedule_ack(now); } } @@ -611,12 +612,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 { diff --git a/ql-fsm/src/tests/session.rs b/ql-fsm/src/tests/session.rs index 2f7386a6..d57fb805 100644 --- a/ql-fsm/src/tests/session.rs +++ b/ql-fsm/src/tests/session.rs @@ -461,6 +461,66 @@ 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_record_ack_delay); + harness.on_timer(Side::B); + harness.next_decoded_outbound(Side::B).unwrap(); + + harness.deliver(Side::B, data); + harness.advance(config.session_record_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_accepted_record_window: 1, + session_peer_timeout: Duration::from_millis(30), + ..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( From ffc5ead9925e8767089a153c783c754e3c163924 Mon Sep 17 00:00:00 2001 From: Alex's Bot Date: Thu, 30 Jul 2026 13:28:12 +0000 Subject: [PATCH 2/8] ql-fsm: extract the replay window from the ack tracker Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016YFf732Q7Rq4bf2PJR83LN --- ql-fsm/src/fsm.rs | 1 - ql-fsm/src/session/ack_tracker.rs | 136 +++++++--------------------- ql-fsm/src/session/mod.rs | 18 ++-- ql-fsm/src/session/range_set.rs | 4 + ql-fsm/src/session/replay_window.rs | 62 +++++++++++++ ql-fsm/src/session/state.rs | 2 + 6 files changed, 107 insertions(+), 116 deletions(-) create mode 100644 ql-fsm/src/session/replay_window.rs diff --git a/ql-fsm/src/fsm.rs b/ql-fsm/src/fsm.rs index 603d1bd9..456cb499 100644 --- a/ql-fsm/src/fsm.rs +++ b/ql-fsm/src/fsm.rs @@ -132,7 +132,6 @@ pub fn receive( let (decrypt_len, seq) = { let record = wire::QlSessionRecord::decode(&mut reader) .map_err(|error| ReceiveError::wire(ReceiveStage::SessionRecord, error))?; - // checked here too, or else a replay costs a full AES-GCM for nothing if conn.session.is_replay(record.header.seq) { return Ok(()); } diff --git a/ql-fsm/src/session/ack_tracker.rs b/ql-fsm/src/session/ack_tracker.rs index 6c787ba6..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,70 +18,36 @@ pub struct PendingAck { pub includes_all_pending: bool, } -#[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), } } - /// returns true when `seq` was already accepted or has aged out of the window - pub fn is_replay(&self, seq: RecordSeq) -> bool { - let seq = seq.0; - let Some(largest_accepted) = self.accepted_records.max() else { - return false; - }; - seq < self.accepted_cutoff(largest_accepted) || self.accepted_records.contains(seq) - } - - /// records `seq` and queues an ack for it, or returns false when the record must be dropped - pub fn accept(&mut self, seq: RecordSeq) -> bool { - if self.is_replay(seq) { - return false; - } - - self.accepted_records.insert(single_range(seq.0)); - self.trim_accepted_records(); - + /// 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(); - - true } 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; @@ -105,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) { @@ -132,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(); @@ -153,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) @@ -187,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!(ack_tracker.accept(RecordSeq(10))); - assert!(ack_tracker.accept(RecordSeq(11))); - assert!(ack_tracker.accept(RecordSeq(12))); + 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(); @@ -201,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!(ack_tracker.accept(RecordSeq(10))); - assert!(ack_tracker.accept(RecordSeq(15))); - assert!(ack_tracker.accept(RecordSeq(16))); - assert!(ack_tracker.accept(RecordSeq(12))); + 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!(ack_tracker.accept(RecordSeq(10))); - assert!(ack_tracker.accept(RecordSeq(15))); - - assert!(!ack_tracker.accept(RecordSeq(10))); - } - #[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!(ack_tracker.accept(RecordSeq(1))); - assert!(ack_tracker.accept(RecordSeq(3))); - assert!(ack_tracker.accept(RecordSeq(5))); + 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(); @@ -238,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!(ack_tracker.accept(RecordSeq(1))); - assert!(ack_tracker.accept(RecordSeq(3))); - assert!(ack_tracker.accept(RecordSeq(5))); + 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 babc739d..af0d770d 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; @@ -27,6 +28,7 @@ use ql_wire::{ use self::{ 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}, @@ -103,8 +105,6 @@ impl SessionFsm { .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, state: SessionState { @@ -116,10 +116,8 @@ 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, @@ -203,7 +201,7 @@ impl SessionFsm { } pub fn is_replay(&self, seq: RecordSeq) -> bool { - self.state.ack_tracker.is_replay(seq) + self.state.replay_window.is_replay(seq) } pub fn receive(&mut self, now: Instant, seq: RecordSeq, frames: I, sink: &mut impl EventSink) @@ -216,10 +214,8 @@ impl SessionFsm { self.collect_timeouts(now); - // retransmissions carry a fresh seq, so a repeat is duplication or replay, never a lost ack - if !self.state.ack_tracker.accept(seq) { - return; - } + self.state.replay_window.accept(seq); + self.state.ack_tracker.push(seq); self.state.last_activity_at = now; self.state.last_inbound_at = now; 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, From 39f663d33e55b9a55fd5726f28dcc485e46ac70b Mon Sep 17 00:00:00 2001 From: Alex's Bot Date: Thu, 30 Jul 2026 15:27:17 +0000 Subject: [PATCH 3/8] ql-fsm: nest session config in QlFsmConfig The session_* fields were a field-for-field mirror of SessionConfig, so every new knob had to be declared in five places and copied across by hand at handshake time. local_parity and the peer's initial stream receive window can't come along: the handshake settles them per session, so a nested config would let a caller set values we then silently overwrite. They move to SessionParams instead. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016YFf732Q7Rq4bf2PJR83LN --- ql-fsm/src/handshake/mod.rs | 19 +++----- ql-fsm/src/lib.rs | 32 ++------------ ql-fsm/src/session/mod.rs | 43 +++++++++++++----- ql-fsm/src/session/tests.rs | 78 ++++++++++++++++++++------------- ql-fsm/src/tests/handshake.rs | 10 ++++- ql-fsm/src/tests/mod.rs | 45 ++++++++++--------- ql-fsm/src/tests/proptest.rs | 32 +++++++------- ql-fsm/src/tests/session.rs | 51 +++++++++++++-------- ql-runtime/src/tests/mod.rs | 12 ++--- ql-runtime/src/tests/session.rs | 7 ++- ql-runtime/src/tests/stream.rs | 39 ++++++++++------- 11 files changed, 204 insertions(+), 164 deletions(-) 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/mod.rs b/ql-fsm/src/session/mod.rs index af0d770d..3784862f 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -37,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), @@ -61,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), @@ -95,11 +116,12 @@ 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); @@ -107,6 +129,7 @@ impl SessionFsm { config.stream_receive_buffer_size = config.stream_receive_buffer_size.max(1); Self { config, + params, state: SessionState { last_activity_at: now, last_inbound_at: now, @@ -121,7 +144,7 @@ impl SessionFsm { 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()), }, } } @@ -136,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); @@ -146,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; @@ -900,7 +923,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, @@ -920,7 +943,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())) diff --git a/ql-fsm/src/session/tests.rs b/ql-fsm/src/session/tests.rs index 6599ce8d..e89d9acd 100644 --- a/ql-fsm/src/session/tests.rs +++ b/ql-fsm/src/session/tests.rs @@ -8,7 +8,7 @@ use ql_wire::{ SessionRecordBuilder, StreamData, StreamReset, }; -use super::{SessionConfig, SessionEvent, SessionFsm}; +use super::{SessionConfig, SessionEvent, SessionFsm, SessionParams}; use crate::{session::stream_parity::StreamParity, StreamResetEvent}; const REFUSED: ResetCode = ResetCode(1); @@ -94,7 +94,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); @@ -115,6 +115,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 +140,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 +175,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 +210,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 +241,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 +278,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 +323,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 +349,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 +402,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); @@ -450,7 +462,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 +492,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 +521,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 +549,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 +580,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 +591,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 +609,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 +632,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 +662,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 +699,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 +736,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 +773,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 +822,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 +847,7 @@ fn close_does_not_ack_rejected_record_seq() { ack_delay: Duration::ZERO, ..SessionConfig::default() }, + SessionParams::default(), now, ); @@ -866,7 +882,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 +896,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 +919,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 +962,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 +969,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 +1049,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/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 d57fb805..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(); @@ -474,12 +477,12 @@ fn duplicate_record_is_dropped_and_not_acked() { let data = harness.next_outbound(Side::A).unwrap(); harness.deliver(Side::B, data.clone()); - harness.advance(config.session_record_ack_delay); + 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_record_ack_delay); + harness.advance(config.session.ack_delay); harness.on_timer(Side::B); assert!(harness.next_outbound(Side::B).is_none()); } @@ -487,8 +490,11 @@ fn duplicate_record_is_dropped_and_not_acked() { #[test] fn replayed_record_does_not_renew_the_peer_timeout() { let config = QlFsmConfig { - session_accepted_record_window: 1, - session_peer_timeout: Duration::from_millis(30), + session: SessionConfig { + accepted_record_window: 1, + peer_timeout: Duration::from_millis(30), + ..SessionConfig::default() + }, ..QlFsmConfig::default() }; let mut harness = Harness::connected(config); @@ -509,7 +515,7 @@ fn replayed_record_does_not_renew_the_peer_timeout() { harness.deliver(Side::B, second); harness.drain_events(Side::B); - harness.advance(config.session_peer_timeout); + harness.advance(config.session.peer_timeout); harness.deliver(Side::B, first); harness.on_timer(Side::B); @@ -525,11 +531,17 @@ fn replayed_record_does_not_renew_the_peer_timeout() { 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() }, ); @@ -555,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() From f8c9e7eaae9faf9637a8e482f53962de0a0827f8 Mon Sep 17 00:00:00 2001 From: Alex's Bot Date: Thu, 30 Jul 2026 20:14:04 +0000 Subject: [PATCH 4/8] ql-fsm: track window updates and pings as frames TrackedRecord kept four parallel records of what a record carried, so restoring one meant three ad-hoc blocks plus a loop, and asking whether a record still referenced a stream meant two separate queries. Window updates and pings are ack-eliciting like the other two, so they belong in frames, and should_track is then just a non-empty check. Acking a record no longer requires it to be confirmed sent. An ack that overtook its own write completion was dropped on the floor, and the record went on to be retransmitted at the RTO even though the peer already had it. The Writable emit in handle_stream_window could never fire, since nothing between the two send_capacity calls changes the send buffer. collect_timeouts drains into a Vec so the restore can hang off SessionState. That costs an allocation, but only on the loss path: collect sizes from a lower bound of zero. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016YFf732Q7Rq4bf2PJR83LN --- ql-fsm/src/session/mod.rs | 127 ++++++++-------------------------- ql-fsm/src/session/state.rs | 47 ++++++++++++- ql-fsm/src/session/tests.rs | 65 +++++++++++++++++ ql-fsm/src/session/tracked.rs | 16 ++++- 4 files changed, 152 insertions(+), 103 deletions(-) diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index 3784862f..df320907 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -264,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); @@ -308,13 +308,7 @@ impl SessionFsm { let Some(record) = self.state.tracked_records.shift_remove(&write_id) else { return; }; - restore_tracked_record( - now, - &mut self.state.ack_tracker, - &mut self.state.pending_ping, - &mut self.state.streams, - &record, - ); + self.state.restore_tracked_record(now, &record); } } @@ -413,9 +407,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.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; @@ -434,8 +426,6 @@ impl SessionFsm { seq, frames: Vec::new(), ack: None, - ping_included: false, - window_updates: Vec::new(), sent_at: None, }; @@ -443,7 +433,7 @@ impl SessionFsm { if self.state.pending_ping && builder.push_ping() { self.state.pending_ping = false; - outbound.ping_included = true; + outbound.frames.push(TrackedFrame::Ping); } self.push_next_pending_stream_window(&mut builder, &mut outbound); @@ -533,8 +523,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)); } } @@ -615,9 +605,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); @@ -644,22 +635,19 @@ impl SessionFsm { fn collect_timeouts(&mut self, now: Instant) { let rto = self.state.loss_recovery.rto(); - let mut timed_out = false; - let state = &mut self.state; - for (_, record) in state.tracked_records.extract_if(.., |_, record| { - record.sent_at.is_some_and(|sent_at| sent_at + rto <= now) - }) { - restore_tracked_record( - now, - &mut state.ack_tracker, - &mut state.pending_ping, - &mut state.streams, - &record, - ); - timed_out = true; + let timed_out: Vec<_> = self + .state + .tracked_records + .extract_if(.., |_, record| { + record.sent_at.is_some_and(|sent_at| sent_at + rto <= now) + }) + .map(|(_, record)| record) + .collect(); + for record in &timed_out { + self.state.restore_tracked_record(now, record); } - if timed_out { - state.loss_recovery.on_timeout(); + if !timed_out.is_empty() { + self.state.loss_recovery.on_timeout(); } self.reap_reapable_streams(); } @@ -756,19 +744,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( @@ -840,11 +824,10 @@ impl SessionFsm { 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, - }) + record + .frames + .iter() + .any(|frame| frame.references_stream(stream_id)) }); if tracked_refs_stream { return false; @@ -987,60 +970,6 @@ fn local_stream_was_opened( && stream_id.0 < local_parity.make_stream_id(next_stream_ordinal).0 } -fn restore_tracked_record( - now: Instant, - ack_tracker: &mut AckTracker, - pending_ping: &mut bool, - streams: &mut IndexMap, - record: &TrackedRecord, -) { - 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 { - 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 restore_stream_reset(streams: &mut IndexMap, reset: StreamReset) { - if let Some(stream) = streams.get_mut(&reset.stream_id) { - stream.pending_reset = Some(reset); - } -} - -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; - } - } -} - fn acknowledge_tracked_frame( streams: &mut IndexMap, stream_send_buffer_size: usize, @@ -1048,7 +977,7 @@ fn acknowledge_tracked_frame( sink: &mut impl EventSink, ) { match frame { - TrackedFrame::StreamReset(_) => {} + TrackedFrame::StreamReset(_) | TrackedFrame::StreamWindow(..) | TrackedFrame::Ping => {} 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/state.rs b/ql-fsm/src/session/state.rs index 98bf19aa..a408991d 100644 --- a/ql-fsm/src/session/state.rs +++ b/ql-fsm/src/session/state.rs @@ -10,8 +10,8 @@ use super::{ remote_stream_history::RemoteStreamHistory, replay_window::ReplayWindow, stream_rx::StreamRx, - stream_tx::StreamTx, - tracked::{LossRecovery, TrackedRecord}, + stream_tx::{StreamTx, StreamTxRange}, + tracked::{LossRecovery, TrackedFrame, TrackedRecord}, }; pub struct SessionState { @@ -31,6 +31,49 @@ pub struct SessionState { pub remote_stream_history: RemoteStreamHistory, } +impl SessionState { + pub fn restore_tracked_record(&mut self, now: Instant, record: &TrackedRecord) { + if let Some(ack) = &record.ack { + self.ack_tracker.restore_acked_ranges(ack, now); + } + for frame in &record.frames { + self.requeue_tracked_frame(frame); + } + } + + fn requeue_tracked_frame(&mut self, frame: &TrackedFrame) { + match frame { + TrackedFrame::StreamReset(reset) => { + if let Some(stream) = self.streams.get_mut(&reset.stream_id) { + stream.pending_reset = Some(reset.clone()); + } + } + TrackedFrame::StreamData(frame) => { + let Some(stream) = self.streams.get_mut(&frame.stream_id) else { + return; + }; + if matches!(stream.outbound_state, OutboundState::Closed) { + return; + } + 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) = self.streams.get_mut(stream_id) { + stream.pending_window |= stream.recv_limit() >= *maximum_offset; + } + } + TrackedFrame::Ping => self.pending_ping = true, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum SessionPhase { Open, diff --git a/ql-fsm/src/session/tests.rs b/ql-fsm/src/session/tests.rs index e89d9acd..1194461d 100644 --- a/ql-fsm/src/session/tests.rs +++ b/ql-fsm/src/session/tests.rs @@ -454,6 +454,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(); diff --git a/ql-fsm/src/session/tracked.rs b/ql-fsm/src/session/tracked.rs index f8a234ca..d9231af8 100644 --- a/ql-fsm/src/session/tracked.rs +++ b/ql-fsm/src/session/tracked.rs @@ -9,9 +9,8 @@ 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 +18,19 @@ pub struct TrackedRecord { pub enum TrackedFrame { StreamData(TrackedStreamData), StreamReset(StreamReset), + StreamWindow(StreamId, u64), + Ping, +} + +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, + Self::Ping => false, + } + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] From fe9c65b5a5b22229c2d88a031cff9c2cc3cf7756 Mon Sep 17 00:00:00 2001 From: Alex's Bot Date: Thu, 30 Jul 2026 20:51:56 +0000 Subject: [PATCH 5/8] ql-fsm: reap streams by id instead of by index The round-robin cursor was an index into the stream map, so every removal had to shift it, and StreamOps had to carry the index alongside the id in order to reap on drop. Keying the cursor on a stream id costs a lookup per record and leaves reaping as a retain plus a shift_remove. The drop reap goes with it: reset() sets pending_reset and commit_read() sets pending_window, and either one disqualifies a stream, so it never reaped anything. Asserting that in Drop held across the suite. The reaps in handle_stream_data and handle_stream_reset stay. They look redundant next to the sweep in collect_timeouts, but receive sweeps before it processes frames, so a stream that one frame finishes is still around for the next frame in the same record, and a stale frame there fails as a protocol error instead of reaching classify_missing_stream. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016ZEW7twTbm7F1dAvfESBNi --- ql-fsm/src/session/mod.rs | 132 +++++++++++-------------------- ql-fsm/src/session/state.rs | 35 +++++++- ql-fsm/src/session/stream_ops.rs | 28 +------ 3 files changed, 85 insertions(+), 110 deletions(-) diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index df320907..25085002 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -143,7 +143,7 @@ impl SessionFsm { ack_tracker: AckTracker::new(config.pending_ack_range_limit), pending_ping: false, streams: IndexMap::default(), - next_stream_index: 0, + next_stream_id: None, remote_stream_history: RemoteStreamHistory::new(params.local_parity.remote()), }, } @@ -172,8 +172,7 @@ impl SessionFsm { self.params.initial_stream_receive_window, ), ); - let stream_index = self.state.streams.len() - 1; - Ok(StreamOps::new(self, stream_id, stream_index, sink)) + Ok(StreamOps::new(self, stream_id, sink)) } pub fn stream( @@ -185,16 +184,18 @@ impl SessionFsm { E: EventSink, { self.ensure_session_open()?; - let Some(stream_index) = (|| { - let index = self.state.streams.get_index_of(&stream_id)?; - // Event::Opened only fires after we receive the first frame of a stream - // prevent early access to streams - let _ = self.state.streams[index].header.as_ref()?; - Some(index) - })() else { + // Event::Opened only fires after we receive the first frame of a stream + // prevent early access to streams + if self + .state + .streams + .get(&stream_id) + .and_then(|stream| stream.header.as_ref()) + .is_none() + { return Err(StreamError::MissingStream); - }; - Ok(StreamOps::new(self, stream_id, stream_index, sink)) + } + Ok(StreamOps::new(self, stream_id, sink)) } pub fn queue_ping(&mut self) -> Result<(), NoSessionError> { @@ -478,7 +479,7 @@ impl SessionFsm { return; } - let start = self.state.next_stream_index % len; + let start = self.state.round_robin_start(); for offset in 0..len { let index = (start + offset) % len; let stream = self.state.streams.get_index_mut(index).unwrap().1; @@ -505,7 +506,7 @@ impl SessionFsm { return; } - let start = self.state.next_stream_index % len; + let start = self.state.round_robin_start(); for offset in 0..len { let index = (start + offset) % len; let (&stream_id, stream) = self.state.streams.get_index_mut(index).unwrap(); @@ -540,7 +541,7 @@ impl SessionFsm { return; } - let start = self.state.next_stream_index % len; + let start = self.state.round_robin_start(); let mut next_index = start; for offset in 0..len { @@ -590,7 +591,7 @@ impl SessionFsm { next_index = (index + 1) % len; } - self.state.next_stream_index = next_index; + self.state.set_round_robin_start(next_index); } fn ensure_session_open(&self) -> Result<(), NoSessionError> { @@ -740,6 +741,8 @@ impl SessionFsm { sink.emit(SessionEvent::Finished(stream_id)); } + // Drop it here, or a later frame in this record fails as a protocol error instead of + // being ignored. self.try_reap_stream(stream_id); Ok(()) } @@ -799,6 +802,8 @@ impl SessionFsm { target, })); } + // Drop it here, or a later frame in this record fails as a protocol error instead of + // being ignored. self.try_reap_stream(frame.stream_id); Ok(()) } @@ -822,82 +827,27 @@ 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 - .frames - .iter() - .any(|frame| frame.references_stream(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, + .. + } = &mut self.state; + streams + .retain(|&stream_id, stream| !stream_is_reapable(tracked_records, stream_id, stream)); } 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) { - 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() { - self.state.next_stream_index = 0; - return; - } - if index < self.state.next_stream_index { - self.state.next_stream_index -= 1; - } - if self.state.next_stream_index >= self.state.streams.len() { - self.state.next_stream_index %= self.state.streams.len(); + let reapable = self.state.streams.get(&stream_id).is_some_and(|stream| { + stream_is_reapable(&self.state.tracked_records, stream_id, stream) + }); + if reapable { + self.state.streams.shift_remove(&stream_id); } } fn clear_streams(&mut self) { - self.state.next_stream_index = 0; + self.state.next_stream_id = None; self.state.streams.clear(); } @@ -970,6 +920,20 @@ fn local_stream_was_opened( && stream_id.0 < local_parity.make_stream_id(next_stream_ordinal).0 } +fn stream_is_reapable( + tracked_records: &IndexMap, + stream_id: StreamId, + stream: &StreamState, +) -> bool { + stream.is_done() + && !tracked_records.values().any(|record| { + record + .frames + .iter() + .any(|frame| frame.references_stream(stream_id)) + }) +} + fn acknowledge_tracked_frame( streams: &mut IndexMap, stream_send_buffer_size: usize, diff --git a/ql-fsm/src/session/state.rs b/ql-fsm/src/session/state.rs index a408991d..a5ff802b 100644 --- a/ql-fsm/src/session/state.rs +++ b/ql-fsm/src/session/state.rs @@ -27,11 +27,24 @@ pub struct SessionState { pub ack_tracker: AckTracker, pub pending_ping: bool, pub streams: IndexMap, - pub next_stream_index: usize, + pub next_stream_id: Option, pub remote_stream_history: RemoteStreamHistory, } impl SessionState { + pub fn round_robin_start(&self) -> usize { + self.next_stream_id + .and_then(|stream_id| self.streams.get_index_of(&stream_id)) + .unwrap_or(0) + } + + pub fn set_round_robin_start(&mut self, index: usize) { + self.next_stream_id = self + .streams + .get_index(index) + .map(|(&stream_id, _)| stream_id); + } + pub fn restore_tracked_record(&mut self, now: Instant, record: &TrackedRecord) { if let Some(ack) = &record.ack { self.ack_tracker.restore_acked_ranges(ack, now); @@ -150,6 +163,26 @@ impl StreamState { pub fn reset_recv(&mut self) { self.rx = StreamRx::with_start_offset(self.rx.start_offset(), self.rx.max_buffered()); } + + /// Both halves are terminated and nothing is buffered or pending. + pub fn is_done(&self) -> bool { + if !self.tx.is_empty() + || self.pending_reset.is_some() + || self.pending_window + || self.readable_bytes() > 0 + || self.rx.buffered_end_offset() > self.rx.start_offset() + { + return false; + } + + matches!( + self.inbound_state, + InboundState::Finished | InboundState::Reset(_) | InboundState::Discarding + ) && matches!( + self.outbound_state, + OutboundState::Finished | OutboundState::Closed + ) + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/ql-fsm/src/session/stream_ops.rs b/ql-fsm/src/session/stream_ops.rs index 5ab1c1b8..1fbb40b7 100644 --- a/ql-fsm/src/session/stream_ops.rs +++ b/ql-fsm/src/session/stream_ops.rs @@ -12,23 +12,14 @@ pub struct StreamOps<'a, E> { session: &'a mut SessionFsm, emit: E, stream_id: StreamId, - stream_index: usize, - reap_on_drop: bool, } impl<'a, E: EventSink> StreamOps<'a, E> { - pub(super) fn new( - session: &'a mut SessionFsm, - stream_id: StreamId, - stream_index: usize, - emit: E, - ) -> Self { + pub(super) fn new(session: &'a mut SessionFsm, stream_id: StreamId, emit: E) -> Self { Self { session, emit, stream_id, - stream_index, - reap_on_drop: false, } } @@ -73,7 +64,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,28 +92,16 @@ impl<'a, E: EventSink> StreamOps<'a, E> { target: wire_target, code, }); - self.reap_on_drop = true; } #[inline] fn stream(&self) -> &StreamState { - &self.session.state.streams[self.stream_index] + &self.session.state.streams[&self.stream_id] } #[inline] fn stream_mut(&mut self) -> &mut StreamState { - &mut self.session.state.streams[self.stream_index] - } -} - -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); + &mut self.session.state.streams[&self.stream_id] } } From 03815d91b9c60b9908232dd1ae00d47a0e5da779 Mon Sep 17 00:00:00 2001 From: Nico Burniske Date: Mon, 3 Aug 2026 09:05:09 -0400 Subject: [PATCH 6/8] ql-fsm: trackedrecord ping as bool field + no intermediate collection --- ql-fsm/src/session/mod.rs | 87 ++++++++++++++++++++++++++++------- ql-fsm/src/session/state.rs | 45 +----------------- ql-fsm/src/session/tracked.rs | 3 +- 3 files changed, 74 insertions(+), 61 deletions(-) diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index 25085002..98e3fd88 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -309,7 +309,13 @@ impl SessionFsm { let Some(record) = self.state.tracked_records.shift_remove(&write_id) else { return; }; - self.state.restore_tracked_record(now, &record); + restore_tracked_record( + now, + &mut self.state.ack_tracker, + &mut self.state.pending_ping, + &mut self.state.streams, + record, + ); } } @@ -408,7 +414,7 @@ impl SessionFsm { let (builder, outbound) = self.build_next_record(now)?; - let should_track = !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; @@ -427,6 +433,7 @@ impl SessionFsm { seq, frames: Vec::new(), ack: None, + ping_included: false, sent_at: None, }; @@ -434,7 +441,7 @@ impl SessionFsm { if self.state.pending_ping && builder.push_ping() { self.state.pending_ping = false; - outbound.frames.push(TrackedFrame::Ping); + outbound.ping_included = true; } self.push_next_pending_stream_window(&mut builder, &mut outbound); @@ -636,19 +643,22 @@ impl SessionFsm { fn collect_timeouts(&mut self, now: Instant) { let rto = self.state.loss_recovery.rto(); - let timed_out: Vec<_> = self - .state - .tracked_records - .extract_if(.., |_, record| { - record.sent_at.is_some_and(|sent_at| sent_at + rto <= now) - }) - .map(|(_, record)| record) - .collect(); - for record in &timed_out { - self.state.restore_tracked_record(now, record); + let mut timed_out = false; + let state = &mut self.state; + for (_, record) in state.tracked_records.extract_if(.., |_, record| { + record.sent_at.is_some_and(|sent_at| sent_at + rto <= now) + }) { + restore_tracked_record( + now, + &mut state.ack_tracker, + &mut state.pending_ping, + &mut state.streams, + record, + ); + timed_out = true; } - if !timed_out.is_empty() { - self.state.loss_recovery.on_timeout(); + if timed_out { + state.loss_recovery.on_timeout(); } self.reap_reapable_streams(); } @@ -920,6 +930,51 @@ fn local_stream_was_opened( && stream_id.0 < local_parity.make_stream_id(next_stream_ordinal).0 } +fn restore_tracked_record( + now: Instant, + ack_tracker: &mut AckTracker, + pending_ping: &mut bool, + streams: &mut IndexMap, + record: TrackedRecord, +) { + if let Some(ack) = record.ack { + ack_tracker.restore_acked_ranges(&ack, now); + } + if record.ping_included { + *pending_ping = 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; + } + } + } + } +} + fn stream_is_reapable( tracked_records: &IndexMap, stream_id: StreamId, @@ -941,7 +996,7 @@ fn acknowledge_tracked_frame( sink: &mut impl EventSink, ) { match frame { - TrackedFrame::StreamReset(_) | TrackedFrame::StreamWindow(..) | TrackedFrame::Ping => {} + 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/state.rs b/ql-fsm/src/session/state.rs index a5ff802b..c75765b8 100644 --- a/ql-fsm/src/session/state.rs +++ b/ql-fsm/src/session/state.rs @@ -10,8 +10,8 @@ use super::{ remote_stream_history::RemoteStreamHistory, replay_window::ReplayWindow, stream_rx::StreamRx, - stream_tx::{StreamTx, StreamTxRange}, - tracked::{LossRecovery, TrackedFrame, TrackedRecord}, + stream_tx::StreamTx, + tracked::{LossRecovery, TrackedRecord}, }; pub struct SessionState { @@ -44,47 +44,6 @@ impl SessionState { .get_index(index) .map(|(&stream_id, _)| stream_id); } - - pub fn restore_tracked_record(&mut self, now: Instant, record: &TrackedRecord) { - if let Some(ack) = &record.ack { - self.ack_tracker.restore_acked_ranges(ack, now); - } - for frame in &record.frames { - self.requeue_tracked_frame(frame); - } - } - - fn requeue_tracked_frame(&mut self, frame: &TrackedFrame) { - match frame { - TrackedFrame::StreamReset(reset) => { - if let Some(stream) = self.streams.get_mut(&reset.stream_id) { - stream.pending_reset = Some(reset.clone()); - } - } - TrackedFrame::StreamData(frame) => { - let Some(stream) = self.streams.get_mut(&frame.stream_id) else { - return; - }; - if matches!(stream.outbound_state, OutboundState::Closed) { - return; - } - 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) = self.streams.get_mut(stream_id) { - stream.pending_window |= stream.recv_limit() >= *maximum_offset; - } - } - TrackedFrame::Ping => self.pending_ping = true, - } - } } #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/ql-fsm/src/session/tracked.rs b/ql-fsm/src/session/tracked.rs index d9231af8..e03716fd 100644 --- a/ql-fsm/src/session/tracked.rs +++ b/ql-fsm/src/session/tracked.rs @@ -11,6 +11,7 @@ pub struct TrackedRecord { pub frames: Vec, /// separate from `frames`, which only holds ack-eliciting ones pub ack: Option, + pub ping_included: bool, pub sent_at: Option, } @@ -19,7 +20,6 @@ pub enum TrackedFrame { StreamData(TrackedStreamData), StreamReset(StreamReset), StreamWindow(StreamId, u64), - Ping, } impl TrackedFrame { @@ -28,7 +28,6 @@ impl TrackedFrame { Self::StreamData(frame) => frame.stream_id == stream_id, Self::StreamReset(frame) => frame.stream_id == stream_id, Self::StreamWindow(id, _) => *id == stream_id, - Self::Ping => false, } } } From a9eb088a545bd76aa8f553f4c1c9e554713d9e56 Mon Sep 17 00:00:00 2001 From: Nico Burniske Date: Mon, 3 Aug 2026 09:08:59 -0400 Subject: [PATCH 7/8] Revert "ql-fsm: reap streams by id instead of by index" This reverts commit fe9c65b5a5b22229c2d88a031cff9c2cc3cf7756. --- ql-fsm/src/session/mod.rs | 132 ++++++++++++++++++++----------- ql-fsm/src/session/state.rs | 37 +-------- ql-fsm/src/session/stream_ops.rs | 28 ++++++- 3 files changed, 110 insertions(+), 87 deletions(-) diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index 98e3fd88..c9a8ec58 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -143,7 +143,7 @@ impl SessionFsm { ack_tracker: AckTracker::new(config.pending_ack_range_limit), pending_ping: false, streams: IndexMap::default(), - next_stream_id: None, + next_stream_index: 0, remote_stream_history: RemoteStreamHistory::new(params.local_parity.remote()), }, } @@ -172,7 +172,8 @@ impl SessionFsm { self.params.initial_stream_receive_window, ), ); - Ok(StreamOps::new(self, stream_id, sink)) + let stream_index = self.state.streams.len() - 1; + Ok(StreamOps::new(self, stream_id, stream_index, sink)) } pub fn stream( @@ -184,18 +185,16 @@ impl SessionFsm { E: EventSink, { self.ensure_session_open()?; - // Event::Opened only fires after we receive the first frame of a stream - // prevent early access to streams - if self - .state - .streams - .get(&stream_id) - .and_then(|stream| stream.header.as_ref()) - .is_none() - { + let Some(stream_index) = (|| { + let index = self.state.streams.get_index_of(&stream_id)?; + // Event::Opened only fires after we receive the first frame of a stream + // prevent early access to streams + let _ = self.state.streams[index].header.as_ref()?; + Some(index) + })() else { return Err(StreamError::MissingStream); - } - Ok(StreamOps::new(self, stream_id, sink)) + }; + Ok(StreamOps::new(self, stream_id, stream_index, sink)) } pub fn queue_ping(&mut self) -> Result<(), NoSessionError> { @@ -486,7 +485,7 @@ impl SessionFsm { return; } - let start = self.state.round_robin_start(); + let start = self.state.next_stream_index % len; for offset in 0..len { let index = (start + offset) % len; let stream = self.state.streams.get_index_mut(index).unwrap().1; @@ -513,7 +512,7 @@ impl SessionFsm { return; } - let start = self.state.round_robin_start(); + let start = self.state.next_stream_index % len; for offset in 0..len { let index = (start + offset) % len; let (&stream_id, stream) = self.state.streams.get_index_mut(index).unwrap(); @@ -548,7 +547,7 @@ impl SessionFsm { return; } - let start = self.state.round_robin_start(); + let start = self.state.next_stream_index % len; let mut next_index = start; for offset in 0..len { @@ -598,7 +597,7 @@ impl SessionFsm { next_index = (index + 1) % len; } - self.state.set_round_robin_start(next_index); + self.state.next_stream_index = next_index; } fn ensure_session_open(&self) -> Result<(), NoSessionError> { @@ -751,8 +750,6 @@ impl SessionFsm { sink.emit(SessionEvent::Finished(stream_id)); } - // Drop it here, or a later frame in this record fails as a protocol error instead of - // being ignored. self.try_reap_stream(stream_id); Ok(()) } @@ -812,8 +809,6 @@ impl SessionFsm { target, })); } - // Drop it here, or a later frame in this record fails as a protocol error instead of - // being ignored. self.try_reap_stream(frame.stream_id); Ok(()) } @@ -837,27 +832,82 @@ 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 + .frames + .iter() + .any(|frame| frame.references_stream(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 SessionState { - tracked_records, - streams, - .. - } = &mut self.state; - streams - .retain(|&stream_id, stream| !stream_is_reapable(tracked_records, stream_id, stream)); + 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; + } + } } fn try_reap_stream(&mut self, stream_id: StreamId) { - let reapable = self.state.streams.get(&stream_id).is_some_and(|stream| { - stream_is_reapable(&self.state.tracked_records, stream_id, stream) - }); - if reapable { - self.state.streams.shift_remove(&stream_id); + 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) { + 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() { + self.state.next_stream_index = 0; + return; + } + if index < self.state.next_stream_index { + self.state.next_stream_index -= 1; + } + if self.state.next_stream_index >= self.state.streams.len() { + self.state.next_stream_index %= self.state.streams.len(); } } fn clear_streams(&mut self) { - self.state.next_stream_id = None; + self.state.next_stream_index = 0; self.state.streams.clear(); } @@ -975,20 +1025,6 @@ fn restore_tracked_record( } } -fn stream_is_reapable( - tracked_records: &IndexMap, - stream_id: StreamId, - stream: &StreamState, -) -> bool { - stream.is_done() - && !tracked_records.values().any(|record| { - record - .frames - .iter() - .any(|frame| frame.references_stream(stream_id)) - }) -} - fn acknowledge_tracked_frame( streams: &mut IndexMap, stream_send_buffer_size: usize, diff --git a/ql-fsm/src/session/state.rs b/ql-fsm/src/session/state.rs index c75765b8..98bf19aa 100644 --- a/ql-fsm/src/session/state.rs +++ b/ql-fsm/src/session/state.rs @@ -27,25 +27,10 @@ pub struct SessionState { pub ack_tracker: AckTracker, pub pending_ping: bool, pub streams: IndexMap, - pub next_stream_id: Option, + pub next_stream_index: usize, pub remote_stream_history: RemoteStreamHistory, } -impl SessionState { - pub fn round_robin_start(&self) -> usize { - self.next_stream_id - .and_then(|stream_id| self.streams.get_index_of(&stream_id)) - .unwrap_or(0) - } - - pub fn set_round_robin_start(&mut self, index: usize) { - self.next_stream_id = self - .streams - .get_index(index) - .map(|(&stream_id, _)| stream_id); - } -} - #[derive(Debug, Clone, PartialEq, Eq)] pub enum SessionPhase { Open, @@ -122,26 +107,6 @@ impl StreamState { pub fn reset_recv(&mut self) { self.rx = StreamRx::with_start_offset(self.rx.start_offset(), self.rx.max_buffered()); } - - /// Both halves are terminated and nothing is buffered or pending. - pub fn is_done(&self) -> bool { - if !self.tx.is_empty() - || self.pending_reset.is_some() - || self.pending_window - || self.readable_bytes() > 0 - || self.rx.buffered_end_offset() > self.rx.start_offset() - { - return false; - } - - matches!( - self.inbound_state, - InboundState::Finished | InboundState::Reset(_) | InboundState::Discarding - ) && matches!( - self.outbound_state, - OutboundState::Finished | OutboundState::Closed - ) - } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/ql-fsm/src/session/stream_ops.rs b/ql-fsm/src/session/stream_ops.rs index 1fbb40b7..5ab1c1b8 100644 --- a/ql-fsm/src/session/stream_ops.rs +++ b/ql-fsm/src/session/stream_ops.rs @@ -12,14 +12,23 @@ pub struct StreamOps<'a, E> { session: &'a mut SessionFsm, emit: E, stream_id: StreamId, + stream_index: usize, + reap_on_drop: bool, } impl<'a, E: EventSink> StreamOps<'a, E> { - pub(super) fn new(session: &'a mut SessionFsm, stream_id: StreamId, emit: E) -> Self { + pub(super) fn new( + session: &'a mut SessionFsm, + stream_id: StreamId, + stream_index: usize, + emit: E, + ) -> Self { Self { session, emit, stream_id, + stream_index, + reap_on_drop: false, } } @@ -64,6 +73,7 @@ impl<'a, E: EventSink> StreamOps<'a, E> { if emit_finished { self.emit.emit(SessionEvent::Finished(stream_id)); } + self.reap_on_drop = true; Ok(()) } @@ -92,16 +102,28 @@ impl<'a, E: EventSink> StreamOps<'a, E> { target: wire_target, code, }); + self.reap_on_drop = true; } #[inline] fn stream(&self) -> &StreamState { - &self.session.state.streams[&self.stream_id] + &self.session.state.streams[self.stream_index] } #[inline] fn stream_mut(&mut self) -> &mut StreamState { - &mut self.session.state.streams[&self.stream_id] + &mut self.session.state.streams[self.stream_index] + } +} + +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); } } From 676e35340e2b740b0615e2f1c452e3e8af11032b Mon Sep 17 00:00:00 2001 From: Nico Burniske Date: Mon, 3 Aug 2026 09:14:22 -0400 Subject: [PATCH 8/8] ql-fsm: compact reapable streams in one pass --- ql-fsm/src/session/mod.rs | 108 +++++++++++++++++-------------- ql-fsm/src/session/stream_ops.rs | 15 ----- ql-fsm/src/session/tests.rs | 42 +++++++++++- 3 files changed, 99 insertions(+), 66 deletions(-) diff --git a/ql-fsm/src/session/mod.rs b/ql-fsm/src/session/mod.rs index c9a8ec58..4eb12db1 100644 --- a/ql-fsm/src/session/mod.rs +++ b/ql-fsm/src/session/mod.rs @@ -832,66 +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 - .frames - .iter() - .any(|frame| frame.references_stream(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() { @@ -1025,6 +1000,39 @@ fn restore_tracked_record( } } +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; + } + + 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, &stream.outbound_state), + ( + InboundState::Finished | InboundState::Reset(_) | InboundState::Discarding, + OutboundState::Finished | OutboundState::Closed, + ) + ) +} + fn acknowledge_tracked_frame( streams: &mut IndexMap, stream_send_buffer_size: usize, 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 1194461d..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, SessionParams}; +use super::{ + state::{InboundState, OutboundState}, + SessionConfig, SessionEvent, SessionFsm, SessionParams, +}; use crate::{session::stream_parity::StreamParity, StreamResetEvent}; const REFUSED: ResetCode = ResetCode(1); @@ -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();