1use std::{
2 cmp,
3 collections::{BTreeMap, BTreeSet, VecDeque},
4 mem,
5 ops::{Bound, Index, IndexMut},
6};
7
8use rand::{CryptoRng, RngExt};
9use rustc_hash::{FxHashMap, FxHashSet};
10use sorted_index_buffer::SortedIndexBuffer;
11use tracing::trace;
12
13use super::{PathId, paths::PathResponses, paths::PathRetransmits};
14use crate::{
15 Dir, Duration, FourTuple, Instant, StreamId, TransportError, TransportErrorCode, VarInt,
16 connection::StreamsState,
17 frame::{self, AddAddress, RemoveAddress},
18 packet::SpaceId,
19 range_set::ArrayRangeSet,
20 shared::IssuedCid,
21};
22
23pub(super) struct PacketSpace {
24 pub(super) pending: Retransmits,
26
27 pub(super) number_spaces: BTreeMap<PathId, PacketNumberSpace>,
33}
34
35impl PacketSpace {
36 pub(super) fn new(now: Instant, space: SpaceId, rng: &mut (impl CryptoRng + ?Sized)) -> Self {
37 let number_space_0 = PacketNumberSpace::new(now, space, rng);
38 Self {
39 pending: Retransmits::default(),
40 number_spaces: BTreeMap::from([(PathId::ZERO, number_space_0)]),
41 }
42 }
43
44 #[cfg(test)]
45 pub(super) fn new_deterministic(now: Instant, space: SpaceId) -> Self {
46 let number_space_0 = PacketNumberSpace::new_deterministic(now, space);
47 Self {
48 pending: Retransmits::default(),
49 number_spaces: BTreeMap::from([(PathId::ZERO, number_space_0)]),
50 }
51 }
52
53 pub(super) fn path_space(&self, path_id: PathId) -> Option<&PacketNumberSpace> {
57 self.number_spaces.get(&path_id)
58 }
59
60 pub(super) fn path_space_mut(&mut self, path_id: PathId) -> Option<&mut PacketNumberSpace> {
64 self.number_spaces.get_mut(&path_id)
65 }
66
67 pub(super) fn for_path(&mut self, path: PathId) -> &mut PacketNumberSpace {
76 self.number_spaces
77 .get_mut(&path)
78 .unwrap_or_else(|| panic!("PacketNumberSpace missing for {path}"))
79 }
80
81 pub(super) fn iter_paths_mut(&mut self) -> impl Iterator<Item = &mut PacketNumberSpace> {
82 self.number_spaces.values_mut()
83 }
84
85 pub(super) fn queue_tail_loss_probe(
98 &mut self,
99 path_id: PathId,
100 request_immediate_ack: bool,
101 streams: &StreamsState,
102 ) {
103 if request_immediate_ack {
104 self.for_path(path_id).pending_immediate_ack = true;
107 }
108
109 if !self.pending.is_empty(streams) {
111 return;
113 }
114
115 for packet in self
117 .number_spaces
118 .values_mut()
119 .flat_map(|s| s.sent_packets.values_mut())
120 {
121 if !packet.retransmits.is_empty(streams) {
122 self.pending |= mem::take(&mut packet.retransmits);
125 return;
126 }
127 }
128
129 if !self.for_path(path_id).pending_immediate_ack {
133 self.for_path(path_id).pending_ping = true;
134 }
135 }
136
137 pub(super) fn can_send(&self, path_id: PathId, streams: &StreamsState) -> SendableFrames {
145 let acks = self
146 .number_spaces
147 .values()
148 .any(|pns| pns.pending_acks.can_send());
149 let space_specific = self.number_spaces.get(&path_id).is_some_and(|s| {
150 s.pending_ping || s.pending_immediate_ack || !s.pending_path_responses.is_empty()
151 });
152 let other = !self.pending.is_empty(streams);
153 SendableFrames {
154 acks,
155 close: false,
156 space_specific,
157 other,
158 }
159 }
160}
161
162impl Index<SpaceId> for [PacketSpace; 3] {
163 type Output = PacketSpace;
164 fn index(&self, space: SpaceId) -> &PacketSpace {
165 &self.as_ref()[space as usize]
166 }
167}
168
169impl IndexMut<SpaceId> for [PacketSpace; 3] {
170 fn index_mut(&mut self, space: SpaceId) -> &mut PacketSpace {
171 &mut self.as_mut()[space as usize]
172 }
173}
174
175#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
180pub(crate) enum SpaceKind {
181 Initial = 0,
183 Handshake = 1,
185 Data = 2,
187}
188
189impl SpaceKind {
190 pub(crate) fn encryption_level(self) -> super::EncryptionLevel {
192 match self {
193 Self::Initial => super::EncryptionLevel::Initial,
194 Self::Handshake => super::EncryptionLevel::Handshake,
195 Self::Data => super::EncryptionLevel::OneRtt,
196 }
197 }
198}
199
200impl Index<SpaceKind> for [PacketSpace; 3] {
201 type Output = PacketSpace;
202 fn index(&self, space: SpaceKind) -> &PacketSpace {
203 &self.as_ref()[space as usize]
204 }
205}
206
207impl IndexMut<SpaceKind> for [PacketSpace; 3] {
208 fn index_mut(&mut self, space: SpaceKind) -> &mut PacketSpace {
209 &mut self.as_mut()[space as usize]
210 }
211}
212
213pub(super) struct PacketNumberSpace {
230 pub(super) open_status: OpenStatus,
241
242 pub(super) largest_received_packet_number: Option<u64>,
244 pub(super) next_packet_number: u64,
247 pub(super) largest_acked_packet_pn: Option<u64>,
249 pub(super) largest_acked_packet_send_time: Instant,
250 pub(super) largest_ack_eliciting_sent: u64,
252 pub(super) unacked_non_ack_eliciting_tail: u64,
254 pub(super) sent_packets: SortedIndexBuffer<SentPacket>,
257 pub(super) lost_packets: SortedIndexBuffer<LostPacket>,
260 pub(super) ecn_counters: frame::EcnCounts,
262 pub(super) ecn_feedback: frame::EcnCounts,
269 pub(super) pending_ping: bool,
271 pub(super) pending_acks: PendingAcks,
273 pub(super) pending_immediate_ack: bool,
275 pub(super) pending_path_responses: PathResponses,
281 pub(super) dedup: Dedup,
283
284 pub(super) time_of_last_ack_eliciting_packet: Option<Instant>,
288 pub(super) loss_time: Option<Instant>,
294 pub(super) loss_probes: u32,
296
297 pn_filter: Option<PacketNumberFilter>,
299}
300
301impl PacketNumberSpace {
302 pub(super) fn new(now: Instant, space: SpaceId, rng: &mut (impl CryptoRng + ?Sized)) -> Self {
303 let pn_filter = match space {
304 SpaceId::Initial | SpaceId::Handshake => None,
305 SpaceId::Data => Some(PacketNumberFilter::new(rng)),
306 };
307 Self {
308 open_status: OpenStatus::default(),
309 largest_received_packet_number: None,
310 next_packet_number: 0,
311 largest_acked_packet_pn: None,
312 largest_acked_packet_send_time: now,
313 largest_ack_eliciting_sent: 0,
314 unacked_non_ack_eliciting_tail: 0,
315 sent_packets: SortedIndexBuffer::new(),
316 lost_packets: SortedIndexBuffer::new(),
317 ecn_counters: frame::EcnCounts::ZERO,
318 ecn_feedback: frame::EcnCounts::ZERO,
319 pending_ping: false,
320 pending_acks: PendingAcks::new(),
321 pending_immediate_ack: false,
322 pending_path_responses: PathResponses::default(),
323 dedup: Default::default(),
324 time_of_last_ack_eliciting_packet: None,
325 loss_time: None,
326 loss_probes: 0,
327 pn_filter,
328 }
329 }
330
331 #[cfg(test)]
332 fn new_deterministic(now: Instant, space: SpaceId) -> Self {
333 let pn_filter = match space {
334 SpaceId::Initial | SpaceId::Handshake => None,
335 SpaceId::Data => Some(PacketNumberFilter::disabled()),
336 };
337 Self {
338 open_status: OpenStatus::default(),
339 largest_received_packet_number: None,
340 next_packet_number: 0,
341 largest_acked_packet_pn: None,
342 largest_acked_packet_send_time: now,
343 largest_ack_eliciting_sent: 0,
344 unacked_non_ack_eliciting_tail: 0,
345 sent_packets: SortedIndexBuffer::new(),
346 lost_packets: SortedIndexBuffer::new(),
347 ecn_counters: frame::EcnCounts::ZERO,
348 ecn_feedback: frame::EcnCounts::ZERO,
349 pending_ping: false,
350 pending_acks: PendingAcks::new(),
351 pending_immediate_ack: false,
352 pending_path_responses: PathResponses::default(),
353 dedup: Default::default(),
354 time_of_last_ack_eliciting_packet: None,
355 loss_time: None,
356 loss_probes: 0,
357 pn_filter,
358 }
359 }
360
361 pub(super) fn get_tx_number(&mut self, rng: &mut (impl CryptoRng + ?Sized)) -> u64 {
366 assert!(self.next_packet_number < 2u64.pow(62));
368 let mut pn = self.next_packet_number;
369 self.next_packet_number += 1;
370
371 if let Some(ref mut filter) = self.pn_filter
373 && filter.skip_pn(pn, rng)
374 {
375 pn = self.next_packet_number;
376 self.next_packet_number += 1;
377 }
378 pn
379 }
380
381 pub(super) fn peek_tx_number(&mut self) -> u64 {
382 let pn = self.next_packet_number;
383 if let Some(ref filter) = self.pn_filter
384 && pn == filter.next_skipped_packet_number
385 {
386 return pn + 1;
387 }
388 pn
389 }
390
391 pub(super) fn check_ack(&self, range: std::ops::Range<u64>) -> Result<(), TransportError> {
393 if let Some(ref filter) = self.pn_filter
394 && filter
395 .prev_skipped_packet_number
396 .is_some_and(|pn| range.contains(&pn))
397 {
398 return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked"));
399 }
400 Ok(())
401 }
402
403 pub(super) fn detect_ecn(
405 &mut self,
406 newly_acked: u64,
407 ecn: frame::EcnCounts,
408 ) -> Result<bool, &'static str> {
409 let ect0_increase = ecn
410 .ect0
411 .checked_sub(self.ecn_feedback.ect0)
412 .ok_or("peer ECT(0) count regression")?;
413 let ect1_increase = ecn
414 .ect1
415 .checked_sub(self.ecn_feedback.ect1)
416 .ok_or("peer ECT(1) count regression")?;
417 let ce_increase = ecn
418 .ce
419 .checked_sub(self.ecn_feedback.ce)
420 .ok_or("peer CE count regression")?;
421 let total_increase = ect0_increase + ect1_increase + ce_increase;
422 if total_increase < newly_acked {
423 return Err("ECN bleaching");
424 }
425 if (ect0_increase + ce_increase) < newly_acked || ect1_increase != 0 {
426 return Err("ECN corruption");
427 }
428 self.ecn_feedback = ecn;
433 Ok(ce_increase != 0)
434 }
435
436 pub(super) fn take(&mut self, number: u64) -> Option<SentPacket> {
438 let packet = self.sent_packets.remove(number)?;
439 if !packet.ack_eliciting && number > self.largest_ack_eliciting_sent {
440 self.unacked_non_ack_eliciting_tail =
441 self.unacked_non_ack_eliciting_tail.checked_sub(1).unwrap();
442 }
443 Some(packet)
444 }
445
446 pub(super) fn sent(&mut self, number: u64, packet: SentPacket) -> Option<SentPacket> {
448 const MAX_UNACKED_NON_ACK_ELICTING_TAIL: u64 = 1_000;
455
456 let mut forgotten = None;
457 if packet.ack_eliciting {
458 self.unacked_non_ack_eliciting_tail = 0;
459 self.largest_ack_eliciting_sent = number;
460 } else if self.unacked_non_ack_eliciting_tail > MAX_UNACKED_NON_ACK_ELICTING_TAIL {
461 let oldest_after_ack_eliciting = self
462 .sent_packets
463 .keys_range((
464 Bound::Excluded(self.largest_ack_eliciting_sent),
465 Bound::Unbounded,
466 ))
467 .next()
468 .unwrap();
469 let packet = self
474 .sent_packets
475 .remove(oldest_after_ack_eliciting)
476 .unwrap();
477 debug_assert!(!packet.ack_eliciting);
478 forgotten = Some(packet);
479 } else {
480 self.unacked_non_ack_eliciting_tail += 1;
481 }
482
483 self.sent_packets.insert(number, packet);
484 forgotten
485 }
486
487 pub(super) fn has_in_flight(&self) -> bool {
489 self.sent_packets.values().any(|x| x.size != 0)
493 }
494}
495
496#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
497pub(super) enum OpenStatus {
498 #[default]
500 Pending,
501 Informed,
503}
504
505#[derive(Debug, Clone)]
507pub(super) struct SentPacket {
508 pub(super) path_generation: u64,
511 pub(super) time_sent: Instant,
513 pub(super) size: u16,
517 pub(super) ack_eliciting: bool,
519 pub(super) largest_acked: FxHashMap<PathId, u64>,
521 pub(super) retransmits: ThinRetransmits,
525 pub(super) path_retransmits: PathRetransmits,
527 pub(super) stream_frames: frame::StreamMetaVec,
531}
532
533#[derive(Debug)]
535pub(super) struct LostPacket {
536 pub(super) time_sent: Instant,
538}
539
540#[allow(unreachable_pub)] #[derive(Debug, Default, Clone)]
545pub struct Retransmits {
546 pub(super) max_data: bool,
547 pub(super) max_stream_id: [bool; 2],
548 pub(super) streams_blocked: [bool; 2],
549 pub(super) reset_stream: Vec<(StreamId, VarInt)>,
550 pub(super) stop_sending: Vec<frame::StopSending>,
551 pub(super) max_stream_data: FxHashSet<StreamId>,
552 pub(super) crypto: VecDeque<frame::Crypto>,
553 pub(super) new_cids: PendingNewCids,
554 pub(super) retire_cids: Vec<(PathId, u64)>,
555 pub(super) ack_frequency: bool,
556 pub(super) handshake_done: bool,
557 pub(super) max_path_id: bool,
559 pub(super) paths_blocked: Option<PathId>,
567 pub(super) new_tokens: Vec<FourTuple>,
584 pub(super) path_abandon: BTreeMap<PathId, TransportErrorCode>,
586 pub(super) path_status: BTreeSet<PathId>,
589 pub(super) path_cids_blocked: BTreeMap<PathId, VarInt>,
595
596 pub(super) add_address: BTreeSet<AddAddress>,
599 pub(super) remove_address: BTreeSet<RemoveAddress>,
601 pub(super) reach_out: PendingReachOutFrames,
603}
604
605impl Retransmits {
606 pub(super) fn is_empty(&self, streams: &StreamsState) -> bool {
607 let Self {
608 max_data,
609 max_stream_id,
610 streams_blocked,
611 reset_stream,
612 stop_sending,
613 max_stream_data,
614 crypto,
615 new_cids,
616 retire_cids,
617 ack_frequency,
618 handshake_done,
619 max_path_id,
620 paths_blocked,
621 new_tokens,
622 path_abandon,
623 path_status,
624 path_cids_blocked,
625 add_address,
626 remove_address,
627 reach_out,
628 } = &self;
629 !max_data
630 && !max_stream_id.iter().any(|x| *x)
631 && !streams_blocked.iter().any(|x| *x)
632 && reset_stream.is_empty()
633 && stop_sending.is_empty()
634 && max_stream_data
635 .iter()
636 .all(|&id| !streams.can_send_flow_control(id))
637 && crypto.is_empty()
638 && new_cids.is_empty()
639 && retire_cids.is_empty()
640 && !ack_frequency
641 && !handshake_done
642 && !max_path_id
643 && paths_blocked.is_none()
644 && new_tokens.is_empty()
645 && path_abandon.is_empty()
646 && path_status.is_empty()
647 && path_cids_blocked.is_empty()
648 && add_address.is_empty()
649 && remove_address.is_empty()
650 && reach_out.is_empty()
651 }
652}
653
654impl ::std::ops::BitOrAssign for Retransmits {
655 fn bitor_assign(&mut self, rhs: Self) {
656 let Self {
657 max_data,
658 max_stream_id,
659 streams_blocked,
660 reset_stream,
661 stop_sending,
662 max_stream_data,
663 crypto,
664 new_cids,
665 retire_cids,
666 ack_frequency,
667 handshake_done,
668 max_path_id,
669 paths_blocked,
670 new_tokens,
671 mut path_abandon,
672 mut path_status,
673 mut path_cids_blocked,
674 add_address,
675 remove_address,
676 mut reach_out,
677 } = rhs;
678
679 self.max_data |= max_data;
682 for dir in Dir::iter() {
683 self.max_stream_id[dir as usize] |= max_stream_id[dir as usize];
684 self.streams_blocked[dir as usize] |= streams_blocked[dir as usize];
685 }
686 self.reset_stream.extend_from_slice(&reset_stream);
687 self.stop_sending.extend_from_slice(&stop_sending);
688 self.max_stream_data.extend(&max_stream_data);
689 for crypto in crypto.into_iter().rev() {
690 self.crypto.push_front(crypto);
691 }
692 self.new_cids.extend(&new_cids);
693 self.retire_cids.extend(retire_cids);
694 self.ack_frequency |= ack_frequency;
695 self.handshake_done |= handshake_done;
696 self.max_path_id |= max_path_id;
697 self.paths_blocked = cmp::max(self.paths_blocked, paths_blocked);
698 self.new_tokens.extend_from_slice(&new_tokens);
699 self.path_abandon.append(&mut path_abandon);
700 self.path_status.append(&mut path_status);
701 self.path_cids_blocked.append(&mut path_cids_blocked);
702 self.add_address.extend(add_address.iter().copied());
703 self.remove_address.extend(remove_address.iter().copied());
704 self.reach_out.append(&mut reach_out);
705 }
706}
707
708impl ::std::ops::BitOrAssign<ThinRetransmits> for Retransmits {
709 fn bitor_assign(&mut self, rhs: ThinRetransmits) {
710 let ThinRetransmits { retransmits } = rhs;
711 if let Some(retransmits) = retransmits {
712 self.bitor_assign(*retransmits)
713 }
714 }
715}
716
717impl ::std::iter::FromIterator<Self> for Retransmits {
718 fn from_iter<T>(iter: T) -> Self
719 where
720 T: IntoIterator<Item = Self>,
721 {
722 let mut result = Self::default();
723 for packet in iter {
724 result |= packet;
725 }
726 result
727 }
728}
729
730#[derive(Clone, Debug, Default)]
737pub(super) struct PendingNewCids {
738 cids: Vec<IssuedCid>,
740 sorted: bool,
742}
743
744impl PendingNewCids {
745 pub(super) fn push(&mut self, cid: IssuedCid) {
747 self.cids.push(cid);
748 self.sorted = false;
749 }
750
751 pub(super) fn pop(&mut self) -> Option<IssuedCid> {
753 if !mem::replace(&mut self.sorted, true) {
754 self.cids
755 .sort_by_key(|cid| cmp::Reverse((cid.path_id, cid.sequence)));
756 }
757 self.cids.pop()
758 }
759
760 pub(super) fn is_empty(&self) -> bool {
761 self.cids.is_empty()
762 }
763
764 pub(super) fn extend(&mut self, other: &Self) {
765 self.cids.extend(&other.cids);
766 self.sorted = false;
767 }
768
769 pub(super) fn retain<F>(&mut self, f: F)
770 where
771 F: FnMut(&IssuedCid) -> bool,
772 {
773 self.cids.retain(f);
774 }
775}
776
777#[derive(Debug, Default, Clone)]
784pub(crate) struct PendingReachOutFrames {
785 round: VarInt,
787 frames: Vec<frame::ReachOut>,
789}
790
791impl PendingReachOutFrames {
792 pub(crate) fn len(&self) -> usize {
793 self.frames.len()
794 }
795
796 pub(crate) fn is_empty(&self) -> bool {
797 self.frames.is_empty()
798 }
799
800 pub(crate) fn push(&mut self, frame: frame::ReachOut) {
801 if frame.round < self.round {
802 return;
803 } else if frame.round > self.round {
804 self.round = frame.round;
805 self.frames.clear();
806 }
807 self.frames.push(frame);
808 }
809
810 pub(crate) fn append(&mut self, other: &mut Self) {
811 if other.round < self.round {
812 other.frames.clear();
813 return;
814 } else if other.round > self.round {
815 self.round = other.round;
816 self.frames.clear();
817 }
818 self.frames.append(&mut other.frames);
819 }
820
821 pub(crate) fn pop_if(
822 &mut self,
823 predicate: impl FnOnce(&mut frame::ReachOut) -> bool,
824 ) -> Option<frame::ReachOut> {
825 self.frames.pop_if(predicate)
826 }
827}
828
829impl FromIterator<frame::ReachOut> for PendingReachOutFrames {
830 fn from_iter<T: IntoIterator<Item = frame::ReachOut>>(iter: T) -> Self {
831 let iter = iter.into_iter();
832 let size_hint = iter.size_hint();
833 let mut this = Self {
834 round: Default::default(),
835 frames: Vec::with_capacity(size_hint.1.unwrap_or(size_hint.0)),
836 };
837 for frame in iter {
838 this.push(frame);
839 }
840 this
841 }
842}
843
844#[derive(Debug, Default, Clone)]
846pub(super) struct ThinRetransmits {
847 retransmits: Option<Box<Retransmits>>,
848}
849
850impl ThinRetransmits {
851 pub(super) fn is_empty(&self, streams: &StreamsState) -> bool {
853 match &self.retransmits {
854 Some(retransmits) => retransmits.is_empty(streams),
855 None => true,
856 }
857 }
858
859 pub(super) fn get(&self) -> Option<&Retransmits> {
861 self.retransmits.as_deref()
862 }
863
864 pub(super) fn get_mut(&mut self) -> Option<&mut Retransmits> {
866 self.retransmits.as_deref_mut()
867 }
868
869 pub(super) fn get_or_create(&mut self) -> &mut Retransmits {
873 if self.retransmits.is_none() {
874 self.retransmits = Some(Box::default());
875 }
876 self.retransmits.as_deref_mut().unwrap()
877 }
878}
879
880#[derive(Debug, Default)]
892pub(super) struct Dedup {
893 window: Window,
894 next: u64,
896}
897
898impl Dedup {
899 #[cfg(test)]
901 pub(super) fn new() -> Self {
902 Self { window: 0, next: 0 }
903 }
904
905 fn highest(&self) -> u64 {
907 self.next - 1
908 }
909
910 pub(super) fn insert(&mut self, packet: u64) -> bool {
914 if let Some(diff) = packet.checked_sub(self.next) {
915 self.window = ((self.window << 1) | 1)
917 .checked_shl(cmp::min(diff, u64::from(u32::MAX)) as u32)
918 .unwrap_or(0);
919 self.next = packet + 1;
920 false
921 } else if self.highest() - packet < WINDOW_SIZE {
922 if let Some(bit) = (self.highest() - packet).checked_sub(1) {
924 let mask = 1 << bit;
926 let duplicate = self.window & mask != 0;
927 self.window |= mask;
928 duplicate
929 } else {
930 true
932 }
933 } else {
934 true
936 }
937 }
938
939 fn smallest_missing_in_interval(&self, lower_bound: u64, upper_bound: u64) -> Option<u64> {
943 debug_assert!(lower_bound <= upper_bound);
944 debug_assert!(upper_bound <= self.highest());
945 const BITFIELD_SIZE: u64 = Window::BITS as u64;
946
947 let lower_bound = lower_bound + 1;
951 let upper_bound = upper_bound.saturating_sub(1);
952
953 let start_offset = (self.highest() - upper_bound).max(1) - 1;
956 if start_offset >= BITFIELD_SIZE {
957 return None;
960 }
961
962 let end_offset_exclusive = self.highest().saturating_sub(lower_bound);
963
964 let range_len = end_offset_exclusive
967 .saturating_sub(start_offset)
968 .min(BITFIELD_SIZE);
969 if range_len == 0 {
970 return None;
971 }
972
973 let mask = if range_len == BITFIELD_SIZE {
976 u128::MAX
977 } else {
978 ((1u128 << range_len) - 1) << start_offset
979 };
980 let gaps = !self.window & mask;
981
982 let smallest_missing_offset = 128 - gaps.leading_zeros() as u64;
983 let smallest_missing_packet = self.highest() - smallest_missing_offset;
984
985 if smallest_missing_packet <= upper_bound {
986 Some(smallest_missing_packet)
987 } else {
988 None
989 }
990 }
991
992 fn missing_in_interval(&self, lower_bound: u64, upper_bound: u64) -> bool {
996 self.smallest_missing_in_interval(lower_bound, upper_bound)
997 .is_some()
998 }
999}
1000
1001type Window = u128;
1006
1007const WINDOW_SIZE: u64 = 1 + size_of::<Window>() as u64 * 8;
1009#[derive(Clone, Copy, PartialEq, Eq, Debug)]
1013pub(super) struct SendableFrames {
1014 pub(super) acks: bool,
1016 pub(super) close: bool,
1018 pub(super) space_specific: bool,
1026 pub(super) other: bool,
1028}
1029
1030impl SendableFrames {
1031 pub(super) fn empty() -> Self {
1033 Self {
1034 acks: false,
1035 close: false,
1036 space_specific: false,
1037 other: false,
1038 }
1039 }
1040
1041 pub(super) fn is_ack_eliciting(&self) -> bool {
1043 let Self {
1044 acks: _,
1045 close,
1046 space_specific,
1047 other,
1048 } = *self;
1049 if close {
1050 return false;
1052 }
1053 space_specific || other
1054 }
1055
1056 pub(super) fn is_empty(&self) -> bool {
1058 let Self {
1059 acks,
1060 close,
1061 space_specific,
1062 other,
1063 } = *self;
1064 !acks && !close && !space_specific && !other
1065 }
1066}
1067
1068impl ::std::ops::BitOrAssign for SendableFrames {
1069 fn bitor_assign(&mut self, rhs: Self) {
1070 let Self {
1071 acks,
1072 close,
1073 space_specific,
1074 other,
1075 } = rhs;
1076
1077 self.acks |= acks;
1078 self.close |= close;
1079 self.space_specific |= space_specific;
1080 self.other |= other;
1081 }
1082}
1083
1084#[derive(Debug)]
1085pub(super) struct PendingAcks {
1086 immediate_ack_required: bool,
1091 ack_eliciting_since_last_ack_sent: u64,
1095 non_ack_eliciting_since_last_ack_sent: u64,
1096 ack_eliciting_threshold: u64,
1097 reordering_threshold: u64,
1106 earliest_ack_eliciting_since_last_ack_sent: Option<Instant>,
1109 ranges: ArrayRangeSet,
1115 largest_packet: Option<(u64, Instant)>,
1119 largest_ack_eliciting_packet: Option<u64>,
1121 largest_acked: Option<u64>,
1123}
1124
1125impl PendingAcks {
1126 fn new() -> Self {
1127 Self {
1128 immediate_ack_required: false,
1129 ack_eliciting_since_last_ack_sent: 0,
1130 non_ack_eliciting_since_last_ack_sent: 0,
1131 ack_eliciting_threshold: 1,
1132 reordering_threshold: 1,
1133 earliest_ack_eliciting_since_last_ack_sent: None,
1134 ranges: Default::default(),
1135 largest_packet: Default::default(),
1136 largest_ack_eliciting_packet: Default::default(),
1137 largest_acked: Default::default(),
1138 }
1139 }
1140
1141 pub(super) fn set_ack_frequency_params(&mut self, frame: &frame::AckFrequency) {
1142 self.ack_eliciting_threshold = frame.ack_eliciting_threshold.into_inner();
1143 self.reordering_threshold = frame.reordering_threshold.into_inner();
1144 }
1145
1146 pub(super) fn set_immediate_ack_required(&mut self) {
1147 self.immediate_ack_required = true;
1148 }
1149
1150 pub(super) fn on_max_ack_delay_timeout(&mut self) {
1151 self.immediate_ack_required = self.ack_eliciting_since_last_ack_sent > 0;
1152 }
1153
1154 pub(super) fn max_ack_delay_timeout(&self, max_ack_delay: Duration) -> Option<Instant> {
1155 self.earliest_ack_eliciting_since_last_ack_sent
1156 .map(|earliest_unacked| earliest_unacked + max_ack_delay)
1157 }
1158
1159 pub(super) fn can_send(&self) -> bool {
1168 self.immediate_ack_required && !self.ranges.is_empty()
1169 }
1170
1171 pub(super) fn ack_delay(&self, now: Instant) -> Duration {
1173 self.largest_packet
1174 .map_or_else(Duration::default, |(_, received)| now - received)
1175 }
1176
1177 pub(super) fn packet_received(
1181 &mut self,
1182 now: Instant,
1183 packet_number: u64,
1184 ack_eliciting: bool,
1185 dedup: &Dedup,
1186 ) -> bool {
1187 if !ack_eliciting {
1188 self.non_ack_eliciting_since_last_ack_sent += 1;
1189 return false;
1190 }
1191
1192 let prev_largest_ack_eliciting = self.largest_ack_eliciting_packet.unwrap_or(0);
1193
1194 self.largest_ack_eliciting_packet = self
1196 .largest_ack_eliciting_packet
1197 .map(|pn| pn.max(packet_number))
1198 .or(Some(packet_number));
1199
1200 self.ack_eliciting_since_last_ack_sent += 1;
1202 self.immediate_ack_required |=
1203 self.ack_eliciting_since_last_ack_sent > self.ack_eliciting_threshold;
1204
1205 self.immediate_ack_required |=
1207 self.is_out_of_order(packet_number, prev_largest_ack_eliciting, dedup);
1208
1209 if self.earliest_ack_eliciting_since_last_ack_sent.is_none() && !self.can_send() {
1211 self.earliest_ack_eliciting_since_last_ack_sent = Some(now);
1212 return true;
1213 }
1214
1215 false
1216 }
1217
1218 fn is_out_of_order(
1219 &self,
1220 packet_number: u64,
1221 prev_largest_ack_eliciting: u64,
1222 dedup: &Dedup,
1223 ) -> bool {
1224 match self.reordering_threshold {
1225 0 => false,
1226 1 => {
1227 packet_number < prev_largest_ack_eliciting
1229 || dedup.missing_in_interval(prev_largest_ack_eliciting, packet_number)
1230 }
1231 _ => {
1232 let Some((largest_acked, largest_unacked)) =
1235 self.largest_acked.zip(self.largest_ack_eliciting_packet)
1236 else {
1237 return false;
1238 };
1239 if self.reordering_threshold > largest_acked {
1240 return false;
1241 }
1242 let largest_reported = largest_acked - self.reordering_threshold + 1;
1245 let Some(smallest_missing_unreported) =
1246 dedup.smallest_missing_in_interval(largest_reported, largest_unacked)
1247 else {
1248 return false;
1249 };
1250 largest_unacked - smallest_missing_unreported >= self.reordering_threshold
1251 }
1252 }
1253 }
1254
1255 pub(super) fn acks_sent(&mut self) {
1259 self.immediate_ack_required = false;
1269 self.ack_eliciting_since_last_ack_sent = 0;
1270 self.non_ack_eliciting_since_last_ack_sent = 0;
1271 self.earliest_ack_eliciting_since_last_ack_sent = None;
1272 self.largest_acked = self.largest_ack_eliciting_packet;
1273 }
1274
1275 pub(super) fn insert_one(&mut self, packet: u64, now: Instant) {
1277 self.ranges.insert_one(packet);
1278
1279 if self.largest_packet.is_none_or(|(pn, _)| packet > pn) {
1280 self.largest_packet = Some((packet, now));
1281 }
1282
1283 if self.ranges.range_count() > MAX_ACK_BLOCKS {
1284 self.ranges.pop_min();
1285 }
1286 }
1287
1288 pub(super) fn subtract_below(&mut self, max: u64) {
1290 self.ranges.remove(0..(max + 1));
1291 }
1292
1293 pub(super) fn ranges(&self) -> &ArrayRangeSet {
1295 &self.ranges
1296 }
1297
1298 pub(super) fn maybe_ack_non_eliciting(&mut self) {
1304 const LAZY_ACK_THRESHOLD: u64 = 10;
1309 if self.non_ack_eliciting_since_last_ack_sent > LAZY_ACK_THRESHOLD {
1310 self.immediate_ack_required = true;
1311 }
1312 }
1313}
1314
1315pub(super) struct PacketNumberFilter {
1332 next_skipped_packet_number: u64,
1334 prev_skipped_packet_number: Option<u64>,
1336 exponent: u32,
1338}
1339
1340impl PacketNumberFilter {
1341 pub(super) fn new(rng: &mut (impl CryptoRng + ?Sized)) -> Self {
1342 let exponent = 6;
1344 Self {
1345 next_skipped_packet_number: rng.random_range(0..2u64.saturating_pow(exponent)),
1346 prev_skipped_packet_number: None,
1347 exponent,
1348 }
1349 }
1350
1351 #[cfg(test)]
1352 pub(super) fn disabled() -> Self {
1353 Self {
1354 next_skipped_packet_number: u64::MAX,
1355 prev_skipped_packet_number: None,
1356 exponent: u32::MAX,
1357 }
1358 }
1359
1360 pub(super) fn skip_pn(&mut self, n: u64, rng: &mut (impl CryptoRng + ?Sized)) -> bool {
1362 if n != self.next_skipped_packet_number {
1363 return false;
1364 }
1365
1366 trace!("skipping pn {n}");
1367 self.prev_skipped_packet_number = Some(self.next_skipped_packet_number);
1369 let next_exponent = self.exponent.saturating_add(1);
1370 self.next_skipped_packet_number = rng
1371 .random_range(2u64.saturating_pow(self.exponent)..2u64.saturating_pow(next_exponent));
1372 self.exponent = next_exponent;
1373 true
1374 }
1375}
1376
1377const MAX_ACK_BLOCKS: usize = 64;
1379
1380#[cfg(test)]
1381mod test {
1382 use rand::Rng;
1383 use rand::seq::SliceRandom;
1384
1385 use crate::token::ResetToken;
1386 use crate::{ConnectionIdGenerator, RandomConnectionIdGenerator};
1387
1388 use super::*;
1389
1390 #[test]
1391 fn sanity() {
1392 let mut dedup = Dedup::new();
1393 assert!(!dedup.insert(0));
1394 assert_eq!(dedup.next, 1);
1395 assert_eq!(dedup.window, 0b1);
1396 assert!(dedup.insert(0));
1397 assert_eq!(dedup.next, 1);
1398 assert_eq!(dedup.window, 0b1);
1399 assert!(!dedup.insert(1));
1400 assert_eq!(dedup.next, 2);
1401 assert_eq!(dedup.window, 0b11);
1402 assert!(!dedup.insert(2));
1403 assert_eq!(dedup.next, 3);
1404 assert_eq!(dedup.window, 0b111);
1405 assert!(!dedup.insert(4));
1406 assert_eq!(dedup.next, 5);
1407 assert_eq!(dedup.window, 0b11110);
1408 assert!(!dedup.insert(7));
1409 assert_eq!(dedup.next, 8);
1410 assert_eq!(dedup.window, 0b1111_0100);
1411 assert!(dedup.insert(4));
1412 assert!(!dedup.insert(3));
1413 assert_eq!(dedup.next, 8);
1414 assert_eq!(dedup.window, 0b1111_1100);
1415 assert!(!dedup.insert(6));
1416 assert_eq!(dedup.next, 8);
1417 assert_eq!(dedup.window, 0b1111_1101);
1418 assert!(!dedup.insert(5));
1419 assert_eq!(dedup.next, 8);
1420 assert_eq!(dedup.window, 0b1111_1111);
1421 }
1422
1423 #[test]
1424 fn happypath() {
1425 let mut dedup = Dedup::new();
1426 for i in 0..(2 * WINDOW_SIZE) {
1427 assert!(!dedup.insert(i));
1428 for j in 0..=i {
1429 assert!(dedup.insert(j));
1430 }
1431 }
1432 }
1433
1434 #[test]
1435 fn jump() {
1436 let mut dedup = Dedup::new();
1437 dedup.insert(2 * WINDOW_SIZE);
1438 assert!(dedup.insert(WINDOW_SIZE));
1439 assert_eq!(dedup.next, 2 * WINDOW_SIZE + 1);
1440 assert_eq!(dedup.window, 0);
1441 assert!(!dedup.insert(WINDOW_SIZE + 1));
1442 assert_eq!(dedup.next, 2 * WINDOW_SIZE + 1);
1443 assert_eq!(dedup.window, 1 << (WINDOW_SIZE - 2));
1444 }
1445
1446 #[test]
1447 fn dedup_has_missing() {
1448 let mut dedup = Dedup::new();
1449
1450 dedup.insert(0);
1451 assert!(!dedup.missing_in_interval(0, 0));
1452
1453 dedup.insert(1);
1454 assert!(!dedup.missing_in_interval(0, 1));
1455
1456 dedup.insert(3);
1457 assert!(dedup.missing_in_interval(1, 3));
1458
1459 dedup.insert(4);
1460 assert!(!dedup.missing_in_interval(3, 4));
1461 assert!(dedup.missing_in_interval(0, 4));
1462
1463 dedup.insert(2);
1464 assert!(!dedup.missing_in_interval(0, 4));
1465 }
1466
1467 #[test]
1468 fn dedup_outside_of_window_has_missing() {
1469 let mut dedup = Dedup::new();
1470
1471 for i in 0..140 {
1472 dedup.insert(i);
1473 }
1474
1475 assert!(!dedup.missing_in_interval(0, 4));
1477 dedup.insert(160);
1478 assert!(!dedup.missing_in_interval(0, 4));
1479 assert!(!dedup.missing_in_interval(0, 140));
1480 assert!(dedup.missing_in_interval(0, 160));
1481 }
1482
1483 #[test]
1484 fn dedup_smallest_missing() {
1485 let mut dedup = Dedup::new();
1486
1487 dedup.insert(0);
1488 assert_eq!(dedup.smallest_missing_in_interval(0, 0), None);
1489
1490 dedup.insert(1);
1491 assert_eq!(dedup.smallest_missing_in_interval(0, 1), None);
1492
1493 dedup.insert(5);
1494 dedup.insert(7);
1495 assert_eq!(dedup.smallest_missing_in_interval(0, 7), Some(2));
1496 assert_eq!(dedup.smallest_missing_in_interval(5, 7), Some(6));
1497
1498 dedup.insert(2);
1499 assert_eq!(dedup.smallest_missing_in_interval(1, 7), Some(3));
1500
1501 dedup.insert(170);
1502 dedup.insert(172);
1503 dedup.insert(300);
1504 assert_eq!(dedup.smallest_missing_in_interval(170, 172), None);
1505
1506 dedup.insert(500);
1507 assert_eq!(dedup.smallest_missing_in_interval(0, 500), Some(372));
1508 assert_eq!(dedup.smallest_missing_in_interval(0, 373), Some(372));
1509 assert_eq!(dedup.smallest_missing_in_interval(0, 372), None);
1510 }
1511
1512 #[test]
1513 fn pending_acks_first_packet_is_not_considered_reordered() {
1514 let mut acks = PendingAcks::new();
1515 let mut dedup = Dedup::new();
1516 dedup.insert(0);
1517 acks.packet_received(Instant::now(), 0, true, &dedup);
1518 assert!(!acks.immediate_ack_required);
1519 }
1520
1521 #[test]
1522 fn pending_acks_after_immediate_ack_set() {
1523 let mut acks = PendingAcks::new();
1524 let mut dedup = Dedup::new();
1525
1526 dedup.insert(0);
1528 let now = Instant::now();
1529 acks.insert_one(0, now);
1530 acks.packet_received(now, 0, true, &dedup);
1531
1532 assert!(!acks.ranges.is_empty());
1534 assert!(!acks.can_send());
1535
1536 acks.set_immediate_ack_required();
1538 assert!(acks.can_send());
1539 }
1540
1541 #[test]
1542 fn pending_acks_ack_delay() {
1543 let mut acks = PendingAcks::new();
1544 let mut dedup = Dedup::new();
1545
1546 let t1 = Instant::now();
1547 let t2 = t1 + Duration::from_millis(2);
1548 let t3 = t2 + Duration::from_millis(5);
1549 assert_eq!(acks.ack_delay(t1), Duration::from_millis(0));
1550 assert_eq!(acks.ack_delay(t2), Duration::from_millis(0));
1551 assert_eq!(acks.ack_delay(t3), Duration::from_millis(0));
1552
1553 dedup.insert(0);
1555 acks.insert_one(0, t1);
1556 acks.packet_received(t1, 0, true, &dedup);
1557 assert_eq!(acks.ack_delay(t1), Duration::from_millis(0));
1558 assert_eq!(acks.ack_delay(t2), Duration::from_millis(2));
1559 assert_eq!(acks.ack_delay(t3), Duration::from_millis(7));
1560
1561 dedup.insert(3);
1563 acks.insert_one(3, t2);
1564 acks.packet_received(t2, 3, true, &dedup);
1565 assert_eq!(acks.ack_delay(t2), Duration::from_millis(0));
1566 assert_eq!(acks.ack_delay(t3), Duration::from_millis(5));
1567
1568 dedup.insert(2);
1570 acks.insert_one(2, t3);
1571 acks.packet_received(t3, 2, true, &dedup);
1572 assert_eq!(acks.ack_delay(t3), Duration::from_millis(5));
1573 }
1574
1575 #[test]
1576 fn sent_packet_size() {
1577 assert!(size_of::<SentPacket>() <= 128);
1580 }
1581
1582 #[test]
1583 fn pending_new_cids() {
1584 #[cfg(all(feature = "aws-lc-rs", not(feature = "ring")))]
1585 use aws_lc_rs::hmac;
1586 #[cfg(feature = "ring")]
1587 use ring::hmac;
1588
1589 let mut cid_generator = RandomConnectionIdGenerator::new(8);
1590 let mut reset_key = [0; 64];
1591 rand::rng().fill_bytes(&mut reset_key);
1592 let hmac = hmac::Key::new(hmac::HMAC_SHA256, &reset_key);
1593
1594 let cid_a = cid_generator.generate_cid();
1595 let a = IssuedCid {
1596 path_id: PathId::ZERO,
1597 sequence: 1,
1598 id: cid_a,
1599 reset_token: ResetToken::new(&hmac, cid_a),
1600 };
1601 let cid_b = cid_generator.generate_cid();
1602 let b = IssuedCid {
1603 path_id: PathId::ZERO,
1604 sequence: 2,
1605 id: cid_b,
1606 reset_token: ResetToken::new(&hmac, cid_b),
1607 };
1608 let cid_c = cid_generator.generate_cid();
1609 let c = IssuedCid {
1610 path_id: PathId(1),
1611 sequence: 1,
1612 id: cid_c,
1613 reset_token: ResetToken::new(&hmac, cid_c),
1614 };
1615
1616 let mut pending_cids = PendingNewCids::default();
1617
1618 for _ in 0..9 {
1619 let mut input = vec![a, b, c];
1621 input.shuffle(&mut rand::rng());
1622 for cid in input {
1623 pending_cids.push(cid);
1624 }
1625
1626 assert_eq!(pending_cids.pop().map(|i| i.id), Some(a.id));
1628 assert_eq!(pending_cids.pop().map(|i| i.id), Some(b.id));
1629 assert_eq!(pending_cids.pop().map(|i| i.id), Some(c.id));
1630 assert!(pending_cids.pop().is_none());
1631 }
1632 }
1633}