noq_proto/connection/
spaces.rs

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    /// Data to send
25    pub(super) pending: Retransmits,
26
27    /// Multipath packet number spaces
28    ///
29    /// Each [`PathId`] has it's own [`PacketNumberSpace`].  Only the [`SpaceId::Data`] can
30    /// have multiple packet number spaces, the other spaces only have a number space for
31    /// `PathId::ZERO`, which is populated at creation.
32    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    /// Returns the [`PacketNumberSpace`] for a path
54    ///
55    /// When multipath is disabled use [`PathId::ZERO`].
56    pub(super) fn path_space(&self, path_id: PathId) -> Option<&PacketNumberSpace> {
57        self.number_spaces.get(&path_id)
58    }
59
60    /// Returns a mutable reference to the [`PacketNumberSpace`] for a path
61    ///
62    /// When multipath is disabled use [`PathId::ZERO`].
63    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    /// Returns the [`PacketNumberSpace`] for a path
68    ///
69    /// When multipath is disabled use `PathId::ZERO`.
70    // TODO(flub): Note that this only exists as `&mut self` because it creates a new
71    //    [`PacketNumberSpace`] if one is not yet available for a path.  This forces a few
72    //    more `&mut` references to users than strictly needed.  An alternative would be to
73    //    return an Option but that would need to be handled for all callers.  This could be
74    //    worth exploring once we have all the main multipath bits fitted.
75    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    /// Queue data for a tail loss probe (or anti-amplification deadlock prevention) packet
86    ///
87    /// Probes are sent similarly to normal packets when an expected ACK has not arrived. We never
88    /// deem a packet lost until we receive an ACK that should have included it, but if a trailing
89    /// run of packets (or their ACKs) are lost, this might not happen in a timely fashion. We send
90    /// probe packets to force an ACK, and exempt them from congestion control to prevent a deadlock
91    /// when the congestion window is filled with lost tail packets.
92    ///
93    /// We prefer to send new data, to make the most efficient use of bandwidth. If there's no data
94    /// waiting to be sent, then we retransmit in-flight data to reduce odds of loss. If there's no
95    /// in-flight data either, we're probably a client guarding against a handshake
96    /// anti-amplification deadlock and we just make something up.
97    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            // The probe should be ACKed without delay (should only be used in the Data space and
105            // when the peer supports the acknowledgement frequency extension)
106            self.for_path(path_id).pending_immediate_ack = true;
107        }
108
109        // We prefer to send new data to make most efficient use of bandwidth.
110        if !self.pending.is_empty(streams) {
111            // There's real data to send here, no need to make something up
112            return;
113        }
114
115        // Retransmit data from the oldest in-flight data from any path
116        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                // Remove retransmitted data from the old packet so we don't end up retransmitting
123                // it *again* even if the copy we're sending now gets acknowledged.
124                self.pending |= mem::take(&mut packet.retransmits);
125                return;
126            }
127        }
128
129        // Nothing new to send and nothing to retransmit, so fall back on a ping. This should only
130        // happen in rare cases during the handshake when the server becomes blocked by
131        // anti-amplification.
132        if !self.for_path(path_id).pending_immediate_ack {
133            self.for_path(path_id).pending_ping = true;
134        }
135    }
136
137    /// Whether there is anything to send in this space
138    ///
139    /// For the data space [`Connection::can_send_1rtt`] also needs to be consulted. Prefer
140    /// to use [`Connection::space_can_send`] which handles this.
141    ///
142    /// [`Connection::can_send_1rtt`]: super::Connection::can_send_1rtt
143    /// [`Connection::space_can_send`]: super::Connection::space_can_send
144    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/// The three QUIC packet number space kinds
176///
177/// Unlike [`SpaceId`], this always has exactly three variants — it represents the
178/// encryption level / space kind, not a specific packet number space identity.
179#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
180pub(crate) enum SpaceKind {
181    /// Initial packets (client and server).
182    Initial = 0,
183    /// Handshake packets.
184    Handshake = 1,
185    /// Data (1-RTT and 0-RTT)
186    Data = 2,
187}
188
189impl SpaceKind {
190    /// Returns the encryption level for this space kind.
191    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
213/// The state of a single packet number space.
214///
215/// In RFC9000 there are 3 packet number spaces: Initial, Handshake and Data. In QUIC
216/// Multipath there are multiple packet number spaces for Data, each identified by a
217/// [`PathId`].
218///
219/// This contains the state for a packet number space which is not specific to the 4-tuple
220/// this space is currently using. The 4-tuple specific state, like congestion controller,
221/// pacing, ECN, MTU etc, is stored in [`PathData`].
222///
223/// Note that the `Initial`, `Handshake` and `Data(PathId(0))` space all share the same
224/// [`PathData`].
225///
226/// You should access this via [`PacketSpace::for_path`].
227///
228/// [`PathData`]: super::paths::PathData
229pub(super) struct PacketNumberSpace {
230    /// Whether the path has already been considered opened from an application perspective.
231    ///
232    /// This means, for paths other than the original [`PathId::ZERO`], a first path
233    /// challenge has been responded to, regardless of the initial validation status of the
234    /// path. This state is irreversible, since it's not affected by the path being closed.
235    ///
236    /// Sending a PATH_CHALLENGE and receiving a valid response before the application is
237    /// informed of the path, is a way to ensure the path is usable before it is
238    /// reported. This is not required by the spec, and in the future might be changed for
239    /// simply requiring a first ack'd packet.
240    pub(super) open_status: OpenStatus,
241
242    /// Highest received packet number, if any
243    pub(super) largest_received_packet_number: Option<u64>,
244    /// The packet number of the next packet that will be sent, if any. In the Data space, the
245    /// packet number stored here is sometimes skipped by [`PacketNumberFilter`] logic.
246    pub(super) next_packet_number: u64,
247    /// The largest packet number the remote peer acknowledged in an ACK frame.
248    pub(super) largest_acked_packet_pn: Option<u64>,
249    pub(super) largest_acked_packet_send_time: Instant,
250    /// The highest-numbered ACK-eliciting packet we've sent
251    pub(super) largest_ack_eliciting_sent: u64,
252    /// Number of packets in `sent_packets` with numbers above `largest_ack_eliciting_sent`
253    pub(super) unacked_non_ack_eliciting_tail: u64,
254    /// Transmitted but not acked
255    // We use a BTreeMap here so we can efficiently query by range on ACK and for loss detection
256    pub(super) sent_packets: SortedIndexBuffer<SentPacket>,
257    /// Packets that were deemed lost
258    // Older packets are regularly removed in `Connection::drain_lost_packets`.
259    pub(super) lost_packets: SortedIndexBuffer<LostPacket>,
260    /// Number of explicit congestion notification codepoints seen on incoming packets
261    pub(super) ecn_counters: frame::EcnCounts,
262    /// Recent ECN counters sent by the peer in ACK frames
263    ///
264    /// Updated (and inspected) whenever we receive an ACK with a new highest acked packet
265    /// number. Stored per-space to simplify verification, which would otherwise have difficulty
266    /// distinguishing between ECN bleaching and counts having been updated by a near-simultaneous
267    /// ACK already processed in another space.
268    pub(super) ecn_feedback: frame::EcnCounts,
269    /// A PING frame needs to be sent on this path.
270    pub(super) pending_ping: bool,
271    /// Packet numbers to acknowledge.
272    pub(super) pending_acks: PendingAcks,
273    /// An IMMEDIATE_ACK (draft-ietf-quic-ack-frequency) frame needs to be sent on this path.
274    pub(super) pending_immediate_ack: bool,
275    /// Responses to path challenges that need to be sent, on and off-path.
276    ///
277    /// Responses are only tied to the 4-tuple they were received on, not to a specific path
278    /// generation. Whether the response is on or off-path only depends on the path's
279    /// 4-tuple at sending time.
280    pub(super) pending_path_responses: PathResponses,
281    /// Packet deduplicator
282    pub(super) dedup: Dedup,
283
284    //
285    // Loss Detection
286    /// The time the most recently sent retransmittable packet was sent.
287    pub(super) time_of_last_ack_eliciting_packet: Option<Instant>,
288    /// Earliest time when we might declare a packet lost.
289    ///
290    /// The time at which the earliest sent packet in this space will be considered lost
291    /// based on exceeding the reordering window in time. Only set for packets numbered
292    /// prior to a packet that has been acknowledged.
293    pub(super) loss_time: Option<Instant>,
294    /// Number of tail loss probes to send
295    pub(super) loss_probes: u32,
296
297    /// Packet numbers to skip, only used in the data package space.
298    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    /// Get the next outgoing packet number in this space
362    ///
363    /// In the Data space, the connection's [`PacketNumberFilter`] must be used rather than calling
364    /// this directly.
365    pub(super) fn get_tx_number(&mut self, rng: &mut (impl CryptoRng + ?Sized)) -> u64 {
366        // TODO: Handle packet number overflow gracefully
367        assert!(self.next_packet_number < 2u64.pow(62));
368        let mut pn = self.next_packet_number;
369        self.next_packet_number += 1;
370
371        // Skip this number if the filter says so, only enabled in the data space
372        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    /// Checks whether a skipped packet number was ACKed.
392    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    /// Verifies sanity of an ECN block and returns whether congestion was encountered.
404    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        // If total_increase > newly_acked (which happens when ACKs are lost), this is required by
429        // the draft so that long-term drift does not occur. If =, then the only question is whether
430        // to count CE packets as CE or ECT0. Recording them as CE is more consistent and keeps the
431        // congestion check obvious.
432        self.ecn_feedback = ecn;
433        Ok(ce_increase != 0)
434    }
435
436    /// Stop tracking sent packet `number`, and return what we knew about it
437    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    /// May return a packet that should be forgotten
447    pub(super) fn sent(&mut self, number: u64, packet: SentPacket) -> Option<SentPacket> {
448        // Retain state for at most this many non-ACK-eliciting packets sent after the most recently
449        // sent ACK-eliciting packet. We're never guaranteed to receive an ACK for those, and we
450        // can't judge them as lost without an ACK, so to limit memory in applications which receive
451        // packets but don't send ACK-eliciting data for long periods use we must eventually start
452        // forgetting about them, although it might also be reasonable to just kill the connection
453        // due to weird peer behavior.
454        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            // Per https://www.rfc-editor.org/rfc/rfc9000.html#name-frames-and-frame-types,
470            // non-ACK-eliciting packets must only contain PADDING, ACK, and CONNECTION_CLOSE
471            // frames, which require no special handling on ACK or loss beyond removal from
472            // in-flight counters if padded.
473            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    /// Whether any congestion-controlled packets in this space are not yet acknowledged or lost
488    pub(super) fn has_in_flight(&self) -> bool {
489        // The number of non-congestion-controlled (i.e. size == 0) packets in flight at a time
490        // should be small, since otherwise congestion control wouldn't be effective. Therefore,
491        // this shouldn't need to visit many packets before finishing one way or another.
492        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    /// The application has not yet been informed of this path.
499    #[default]
500    Pending,
501    /// The application has been informed of this path.
502    Informed,
503}
504
505/// Represents one or more packets subject to retransmission
506#[derive(Debug, Clone)]
507pub(super) struct SentPacket {
508    /// [`PathData::generation`](super::PathData::generation) of the path on which this packet was
509    /// sent
510    pub(super) path_generation: u64,
511    /// The time the packet was sent.
512    pub(super) time_sent: Instant,
513    /// The number of bytes sent in the packet, not including UDP or IP overhead, but including
514    /// QUIC framing overhead. Zero if this packet is not counted towards congestion control,
515    /// i.e. not an "in flight" packet.
516    pub(super) size: u16,
517    /// Whether an acknowledgement is expected directly in response to this packet.
518    pub(super) ack_eliciting: bool,
519    /// The largest packet number acknowledged by this packet
520    pub(super) largest_acked: FxHashMap<PathId, u64>,
521    /// Data which needs to be retransmitted in case the packet is lost.
522    ///
523    /// These might be retransmitted over any available path in the same [`SpaceKind`].
524    pub(super) retransmits: ThinRetransmits,
525    /// Retransmittable data specific to a path generation.
526    pub(super) path_retransmits: PathRetransmits,
527    /// Metadata for stream frames in a packet
528    ///
529    /// The actual application data is stored with the stream state.
530    pub(super) stream_frames: frame::StreamMetaVec,
531}
532
533/// Represents one or more packets that are deemed lost.
534#[derive(Debug)]
535pub(super) struct LostPacket {
536    /// The time the packet was sent.
537    pub(super) time_sent: Instant,
538}
539
540/// Retransmittable data queue.
541///
542/// Data in this queue must be retransmittable over any path in the same [`SpaceKind`].
543#[allow(unreachable_pub)] // fuzzing only
544#[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    /// Whether we should inform the peer we will allow higher [`PathId`]s.
558    pub(super) max_path_id: bool,
559    /// Whether we should inform the peer that their max [`PathId`] is blocking our attempt to open
560    /// new paths.
561    ///
562    /// Stores the remote_max_path_id at the time this was generated.
563    /// This frame is entirely informational, so when it's retransmitted, the remote_max_path_id is
564    /// intentionally not updated to preserve the fact that this was the state of the client at
565    /// some point.
566    pub(super) paths_blocked: Option<PathId>,
567    /// For each enqueued NEW_TOKEN frame, a copy of the path's remote address
568    ///
569    /// There are 2 reasons this is unusual:
570    ///
571    /// - If the path changes, NEW_TOKEN frames bound for the old path are not retransmitted on the
572    ///   new path. That is why this field stores the remote address: so that ones for old paths
573    ///   can be filtered out.
574    /// - If a token is lost, a new randomly generated token is re-transmitted, rather than the
575    ///   original. This is so that if both transmissions are received, the client won't risk
576    ///   sending the same token twice. That is why this field does _not_ store any actual token.
577    ///
578    /// It is true that a QUIC endpoint will only want to effectively have NEW_TOKEN frames
579    /// enqueued for its current path at a given point in time. Based on that, we could conceivably
580    /// change this from a vector to an `Option<(FourTuple, usize)>` or just a `usize` or
581    /// something. However, due to the architecture of noq, it is considerably simpler to not do
582    /// that; consider what such a change would mean for implementing `BitOrAssign` on Self.
583    pub(super) new_tokens: Vec<FourTuple>,
584    /// Paths which need to be abandoned
585    pub(super) path_abandon: BTreeMap<PathId, TransportErrorCode>,
586    /// If a [`frame::PathStatusAvailable`] and [`frame::PathStatusBackup`] need to be sent for a
587    /// path
588    pub(super) path_status: BTreeSet<PathId>,
589    /// Whether a PATH_CIDS_BLOCKED frame needs to be sent for a path.
590    ///
591    /// Stores the next_seq number for the blocked path. This number can be "outdated" at the time
592    /// of sending when this is a retransmission. This is intentional, as this frame is purely
593    /// informational, and this would preserve this information.
594    pub(super) path_cids_blocked: BTreeMap<PathId, VarInt>,
595
596    // Nat traversal data
597    /// Addresses to report in `ADD_ADDRESS` frames
598    pub(super) add_address: BTreeSet<AddAddress>,
599    /// Address IDs to remove in `REMOVE_ADDRESS` frames
600    pub(super) remove_address: BTreeSet<RemoveAddress>,
601    /// Round and local addresses to advertise in `REACH_OUT` frames
602    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        // We reduce in-stream head-of-line blocking by queueing retransmits before other data for
680        // STREAM and CRYPTO frames.
681        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/// The queue of new CIDs to be transmitted to the peer.
731///
732/// This queue is always sorted, so that popping off the last item is always the lowest
733/// sequence number of the lowest path ID. Which is the CID you want to be issued next.
734///
735/// This is but a newtype over a `Vec` to enforce the sorted invariant.
736#[derive(Clone, Debug, Default)]
737pub(super) struct PendingNewCids {
738    /// The CIDs themselves.
739    cids: Vec<IssuedCid>,
740    /// Whether [`Self::cids`] is sorted or not.
741    sorted: bool,
742}
743
744impl PendingNewCids {
745    /// Inserts an issued CID into the queue.
746    pub(super) fn push(&mut self, cid: IssuedCid) {
747        self.cids.push(cid);
748        self.sorted = false;
749    }
750
751    /// Pops the next issued CID to transmit from the queue.
752    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/// Logically a Vec of REACH_OUT frames queued for transmit.
778///
779/// This keeps track of the highest round ID and automatically drops frames with a lower
780/// round ID.
781///
782/// The API is directly modelled on [`Vec`].
783#[derive(Debug, Default, Clone)]
784pub(crate) struct PendingReachOutFrames {
785    /// The round ID of the REACH_OUT frames currently pending.
786    round: VarInt,
787    /// The REACH_OUT frames, always all having the same round ID.
788    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/// A variant of `Retransmits` which only allocates storage when required
845#[derive(Debug, Default, Clone)]
846pub(super) struct ThinRetransmits {
847    retransmits: Option<Box<Retransmits>>,
848}
849
850impl ThinRetransmits {
851    /// Returns `true` if no retransmits are necessary
852    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    /// Returns a reference to the retransmits stored in this box
860    pub(super) fn get(&self) -> Option<&Retransmits> {
861        self.retransmits.as_deref()
862    }
863
864    /// Returns a mutable reference to the retransmits stored in this box
865    pub(super) fn get_mut(&mut self) -> Option<&mut Retransmits> {
866        self.retransmits.as_deref_mut()
867    }
868
869    /// Returns a mutable reference to the stored retransmits
870    ///
871    /// This function will allocate a backing storage if required.
872    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/// RFC4303-style sliding window packet number deduplicator.
881///
882/// A contiguous bitfield, where each bit corresponds to a packet number and the rightmost bit is
883/// always set. A set bit represents a packet that has been successfully authenticated. Bits left of
884/// the window are assumed to be set.
885///
886/// ```text
887/// ...xxxxxxxxx 1 0
888///     ^        ^ ^
889/// window highest next
890/// ```
891#[derive(Debug, Default)]
892pub(super) struct Dedup {
893    window: Window,
894    /// Lowest packet number higher than all yet authenticated.
895    next: u64,
896}
897
898impl Dedup {
899    /// Construct an empty window positioned at the start.
900    #[cfg(test)]
901    pub(super) fn new() -> Self {
902        Self { window: 0, next: 0 }
903    }
904
905    /// Highest packet number authenticated.
906    fn highest(&self) -> u64 {
907        self.next - 1
908    }
909
910    /// Record a newly authenticated packet number.
911    ///
912    /// Returns whether the packet might be a duplicate.
913    pub(super) fn insert(&mut self, packet: u64) -> bool {
914        if let Some(diff) = packet.checked_sub(self.next) {
915            // Right of window
916            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            // Within window
923            if let Some(bit) = (self.highest() - packet).checked_sub(1) {
924                // < highest
925                let mask = 1 << bit;
926                let duplicate = self.window & mask != 0;
927                self.window |= mask;
928                duplicate
929            } else {
930                // == highest
931                true
932            }
933        } else {
934            // Left of window
935            true
936        }
937    }
938
939    /// Returns the packet number of the smallest packet missing between the provided interval
940    ///
941    /// If there are no missing packets, returns `None`
942    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        // Since we already know the packets at the boundaries have been received, we only need to
948        // check those in between them (this removes the necessity of extra logic to deal with the
949        // highest packet, which is stored outside the bitfield)
950        let lower_bound = lower_bound + 1;
951        let upper_bound = upper_bound.saturating_sub(1);
952
953        // Note: the offsets are counted from the right
954        // The highest packet is not included in the bitfield, so we subtract 1 to account for that
955        let start_offset = (self.highest() - upper_bound).max(1) - 1;
956        if start_offset >= BITFIELD_SIZE {
957            // The start offset is outside of the window. All packets outside of the window are
958            // considered to be received.
959            return None;
960        }
961
962        let end_offset_exclusive = self.highest().saturating_sub(lower_bound);
963
964        // The range is clamped at the edge of the window, because any earlier packets are
965        // considered to be received
966        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        // Ensure the shift is within bounds (we already know start_offset < BITFIELD_SIZE,
974        // because of the early return)
975        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    /// Returns true if there are any missing packets between the provided interval
993    ///
994    /// The provided packet numbers must have been received before calling this function
995    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
1001/// Inner bitfield type.
1002///
1003/// Because QUIC never reuses packet numbers, this only needs to be large enough to deal with
1004/// packets that are reordered but still delivered in a timely manner.
1005type Window = u128;
1006
1007/// Number of packets tracked by `Dedup`.
1008const WINDOW_SIZE: u64 = 1 + size_of::<Window>() as u64 * 8;
1009/// Indicates which data is available for sending
1010///
1011/// This applies to a particular space ID that was queried and all refers to on-path data.
1012#[derive(Clone, Copy, PartialEq, Eq, Debug)]
1013pub(super) struct SendableFrames {
1014    /// Whether there are ACK frames to send, these are not ack-eliciting.
1015    pub(super) acks: bool,
1016    /// Whether there is a CONNECTION_CLOSE to send, this is not ack-eliciting.
1017    pub(super) close: bool,
1018    /// Whether there are any frames that must be sent on this specific space.
1019    ///
1020    /// A space here in the sense of a QUIC Multipath packet number space: `Initial`,
1021    /// `Handshake` and all `Data(PathId)` spaces.
1022    ///
1023    /// These are ack-eliciting. Some frames are scheduled per path, e.g. PING,
1024    /// IMMEDIATE_ACK, PATH_CHALLENGE or PATH_RESPONSE.
1025    pub(super) space_specific: bool,
1026    /// Whether there are any other frames to send, these are ack-eliciting.
1027    pub(super) other: bool,
1028}
1029
1030impl SendableFrames {
1031    /// Returns that no data is available for sending
1032    pub(super) fn empty() -> Self {
1033        Self {
1034            acks: false,
1035            close: false,
1036            space_specific: false,
1037            other: false,
1038        }
1039    }
1040
1041    /// Whether an ack-eliciting packet will be sent.
1042    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            // No ack-eliciting frames are included with a CONNECTION_CLOSE, only acks.
1051            return false;
1052        }
1053        space_specific || other
1054    }
1055
1056    /// Whether no data is sendable.
1057    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    /// Whether we should send an ACK immediately, even if that means sending an ACK-only packet
1087    ///
1088    /// When `immediate_ack_required` is false, the normal behavior is to send ACK frames only when
1089    /// there is other data to send, or when the `MaxAckDelay` timer expires.
1090    immediate_ack_required: bool,
1091    /// The number of ack-eliciting packets received since the last ACK frame was sent
1092    ///
1093    /// Once the count _exceeds_ `ack_eliciting_threshold`, an immediate ACK is required
1094    ack_eliciting_since_last_ack_sent: u64,
1095    non_ack_eliciting_since_last_ack_sent: u64,
1096    ack_eliciting_threshold: u64,
1097    /// The reordering threshold, controlling how we respond to out-of-order ack-eliciting packets
1098    ///
1099    /// Different values enable different behavior:
1100    ///
1101    /// * `0`: no special action is taken
1102    /// * `1`: an ACK is immediately sent if it is out-of-order according to RFC 9000
1103    /// * `>1`: an ACK is immediately sent if it is out-of-order according to the ACK frequency
1104    ///   draft
1105    reordering_threshold: u64,
1106    /// The earliest ack-eliciting packet since the last ACK was sent, used to calculate the moment
1107    /// upon which `max_ack_delay` elapses
1108    earliest_ack_eliciting_since_last_ack_sent: Option<Instant>,
1109    /// Packet number ranges for which to still send acknowledgements.
1110    ///
1111    /// These are packet number ranges of ack-eliciting packets the peer has sent and which
1112    /// need to be acknowledged.  Packet numbers are only removed from here once the peer has
1113    /// acknowledged the ACKs for them.
1114    ranges: ArrayRangeSet,
1115    /// The largest packet number received and the time it was received
1116    ///
1117    /// Used to calculate ACK delay in [`PendingAcks::ack_delay`].
1118    largest_packet: Option<(u64, Instant)>,
1119    /// The ack-eliciting packet we have received with the largest packet number
1120    largest_ack_eliciting_packet: Option<u64>,
1121    /// The largest acknowledged packet number sent in an ACK frame
1122    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    /// Whether any ACK frames SHOULD be sent
1160    ///
1161    /// This is used in the top-level [`Connection::space_can_send`], so determines if a
1162    /// packet will be built. It is often possible to construct new ACK ranges to send
1163    /// before this returns `true`. This results in more ACK frames being sent, and
1164    /// processing those at the receiver costs CPU for very little improvements.
1165    ///
1166    /// [`Connection::space_can_send`]: super::Connection::space_can_send
1167    pub(super) fn can_send(&self) -> bool {
1168        self.immediate_ack_required && !self.ranges.is_empty()
1169    }
1170
1171    /// Returns the delay since the packet with the largest packet number was received
1172    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    /// Handle receipt of a new packet
1178    ///
1179    /// Returns true if the max ack delay timer should be armed
1180    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        // Track largest ack-eliciting packet
1195        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        // Handle ack_eliciting_threshold
1201        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        // Handle out-of-order packets
1206        self.immediate_ack_required |=
1207            self.is_out_of_order(packet_number, prev_largest_ack_eliciting, dedup);
1208
1209        // Arm max_ack_delay timer if necessary
1210        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                // From https://www.rfc-editor.org/rfc/rfc9000#section-13.2.1-7
1228                packet_number < prev_largest_ack_eliciting
1229                    || dedup.missing_in_interval(prev_largest_ack_eliciting, packet_number)
1230            }
1231            _ => {
1232                // From acknowledgement frequency draft, section 6.1: send an ACK immediately if
1233                // doing so would cause the sender to detect a new packet loss
1234                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                // The largest packet number that could be declared lost without a new ACK being
1243                // sent
1244                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    /// Should be called whenever ACKs have been sent
1256    ///
1257    /// This will suppress sending further ACKs until additional ACK eliciting frames arrive
1258    pub(super) fn acks_sent(&mut self) {
1259        // It is possible (though unlikely) that the ACKs we just sent do not cover all the
1260        // ACK-eliciting packets we have received (e.g. if there is not enough room in the packet to
1261        // fit all the ranges). To keep things simple, however, we assume they do. If there are
1262        // indeed some ACKs that weren't covered, the packets might be ACKed later anyway, because
1263        // they are still contained in `self.ranges`. If we somehow fail to send the ACKs at a later
1264        // moment, the peer will assume the packets got lost and will retransmit their frames in a
1265        // new packet, which is suboptimal, because we already received them. Our assumption here is
1266        // that simplicity results in code that is more performant, even in the presence of
1267        // occasional redundant retransmits.
1268        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    /// Insert one packet that needs to be acknowledged
1276    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    /// Remove ACKs of packets numbered at or below `max` from the set of pending ACKs
1289    pub(super) fn subtract_below(&mut self, max: u64) {
1290        self.ranges.remove(0..(max + 1));
1291    }
1292
1293    /// Returns the set of currently pending ACK ranges
1294    pub(super) fn ranges(&self) -> &ArrayRangeSet {
1295        &self.ranges
1296    }
1297
1298    /// Queue an ACK if a significant number of non-ACK-eliciting packets have not yet been
1299    /// acknowledged
1300    ///
1301    /// Should be called immediately before a non-probing packet is composed, when we've already
1302    /// committed to sending a packet regardless.
1303    pub(super) fn maybe_ack_non_eliciting(&mut self) {
1304        // If we're going to send a packet anyway, and we've received a significant number of
1305        // non-ACK-eliciting packets, then include an ACK to help the peer perform timely loss
1306        // detection even if they're not sending any ACK-eliciting packets themselves. Exact
1307        // threshold chosen somewhat arbitrarily.
1308        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
1315/// Helper for mitigating [optimistic ACK attacks]
1316///
1317/// A malicious peer could prompt the local application to begin a large data transfer, and then
1318/// send ACKs without first waiting for data to be received. This could defeat congestion control,
1319/// allowing the connection to consume disproportionate resources. We therefore occasionally skip
1320/// packet numbers, and classify any ACK referencing a skipped packet number as a transport error.
1321///
1322/// Skipped packet numbers occur only in the application data space (where costly transfers might
1323/// take place) and are distributed exponentially to reflect the reduced likelihood and impact of
1324/// bad behavior from a peer that has been well-behaved for an extended period.
1325///
1326/// ACKs for packet numbers that have not yet been allocated are also a transport error, but an
1327/// attacker with knowledge of the congestion control algorithm in use could time falsified ACKs to
1328/// arrive after the packets they reference are sent.
1329///
1330/// [optimistic ACK attacks]: https://www.rfc-editor.org/rfc/rfc9000.html#name-optimistic-ack-attack
1331pub(super) struct PacketNumberFilter {
1332    /// Next outgoing packet number to skip
1333    next_skipped_packet_number: u64,
1334    /// Most recently skipped packet number
1335    prev_skipped_packet_number: Option<u64>,
1336    /// Next packet number to skip is randomly selected from 2^n..2^n+1
1337    exponent: u32,
1338}
1339
1340impl PacketNumberFilter {
1341    pub(super) fn new(rng: &mut (impl CryptoRng + ?Sized)) -> Self {
1342        // First skipped PN is in 0..64
1343        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    /// Whether to use the provided packet number (false) or to skip it (true)
1361    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        // Skip this packet number, and choose the next one to skip
1368        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
1377/// Ensures we can always fit all our ACKs in a single minimum-MTU packet with room to spare
1378const 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        // 0 and 4 are outside of the window
1476        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        // Receive ack-eliciting packet
1527        dedup.insert(0);
1528        let now = Instant::now();
1529        acks.insert_one(0, now);
1530        acks.packet_received(now, 0, true, &dedup);
1531
1532        // Sanity check
1533        assert!(!acks.ranges.is_empty());
1534        assert!(!acks.can_send());
1535
1536        // Can send ACK after max_ack_delay exceeded
1537        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        // In-order packet
1554        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        // Out of order (higher than expected)
1562        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        // Out of order (lower than expected, so previous instant is kept)
1569        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        // The tracking state of sent packets should be minimal, and not grow
1578        // over time.
1579        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            // Push CIDs in a random order
1620            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            // Pop order is always the same
1627            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}