noq_proto/connection/

spaces.rs

use std::{
    cmp,
    collections::{BTreeMap, BTreeSet, VecDeque},
    mem,
    ops::{Bound, Index, IndexMut},
};

use rand::{CryptoRng, RngExt};
use rustc_hash::{FxHashMap, FxHashSet};
use sorted_index_buffer::SortedIndexBuffer;
use tracing::{debug, trace};

use super::{PathId, paths::PathResponses, paths::PathRetransmits};
use crate::{
    Dir, Duration, FourTuple, Instant, StreamId, TransportError, TransportErrorCode, VarInt,
    connection::StreamsState,
    frame::{self, AddAddress, RemoveAddress},
    packet::SpaceId,
    range_set::ArrayRangeSet,
    shared::IssuedCid,
};

pub(super) struct PacketSpace {
    /// Data to send
    pub(super) pending: Retransmits,

    /// Multipath packet number spaces
    ///
    /// Each [`PathId`] has it's own [`PacketNumberSpace`].  Only the [`SpaceId::Data`] can
    /// have multiple packet number spaces, the other spaces only have a number space for
    /// `PathId::ZERO`, which is populated at creation.
    pub(super) number_spaces: BTreeMap<PathId, PacketNumberSpace>,
}

impl PacketSpace {
    pub(super) fn new(now: Instant, space: SpaceId, rng: &mut (impl CryptoRng + ?Sized)) -> Self {
        let number_space_0 = PacketNumberSpace::new(now, space, rng);
        Self {
            pending: Retransmits::default(),
            number_spaces: BTreeMap::from([(PathId::ZERO, number_space_0)]),
        }
    }

    #[cfg(test)]
    pub(super) fn new_deterministic(now: Instant, space: SpaceId) -> Self {
        let number_space_0 = PacketNumberSpace::new_deterministic(now, space);
        Self {
            pending: Retransmits::default(),
            number_spaces: BTreeMap::from([(PathId::ZERO, number_space_0)]),
        }
    }

    /// Returns the [`PacketNumberSpace`] for a path
    ///
    /// When multipath is disabled use [`PathId::ZERO`].
    pub(super) fn path_space(&self, path_id: PathId) -> Option<&PacketNumberSpace> {
        self.number_spaces.get(&path_id)
    }

    /// Returns a mutable reference to the [`PacketNumberSpace`] for a path
    ///
    /// When multipath is disabled use [`PathId::ZERO`].
    pub(super) fn path_space_mut(&mut self, path_id: PathId) -> Option<&mut PacketNumberSpace> {
        self.number_spaces.get_mut(&path_id)
    }

    /// Returns the [`PacketNumberSpace`] for a path
    ///
    /// When multipath is disabled use `PathId::ZERO`.
    // TODO(flub): Note that this only exists as `&mut self` because it creates a new
    //    [`PacketNumberSpace`] if one is not yet available for a path.  This forces a few
    //    more `&mut` references to users than strictly needed.  An alternative would be to
    //    return an Option but that would need to be handled for all callers.  This could be
    //    worth exploring once we have all the main multipath bits fitted.
    pub(super) fn for_path(&mut self, path: PathId) -> &mut PacketNumberSpace {
        self.number_spaces
            .get_mut(&path)
            .unwrap_or_else(|| panic!("PacketNumberSpace missing for {path}"))
    }

    pub(super) fn iter_paths_mut(&mut self) -> impl Iterator<Item = &mut PacketNumberSpace> {
        self.number_spaces.values_mut()
    }

    /// Queue data for a tail loss probe (or anti-amplification deadlock prevention) packet
    ///
    /// Probes are sent similarly to normal packets when an expected ACK has not arrived. We never
    /// deem a packet lost until we receive an ACK that should have included it, but if a trailing
    /// run of packets (or their ACKs) are lost, this might not happen in a timely fashion. We send
    /// probe packets to force an ACK, and exempt them from congestion control to prevent a deadlock
    /// when the congestion window is filled with lost tail packets.
    ///
    /// We prefer to send new data, to make the most efficient use of bandwidth. If there's no data
    /// waiting to be sent, then we retransmit in-flight data to reduce odds of loss. If there's no
    /// in-flight data either, we're probably a client guarding against a handshake
    /// anti-amplification deadlock and we just make something up.
    pub(super) fn queue_tail_loss_probe(
        &mut self,
        path_id: PathId,
        request_immediate_ack: bool,
        streams: &StreamsState,
    ) {
        if request_immediate_ack {
            // The probe should be ACKed without delay (should only be used in the Data space and
            // when the peer supports the acknowledgement frequency extension)
            self.for_path(path_id).pending_immediate_ack = true;
        }

        // We prefer to send new data to make most efficient use of bandwidth.
        if !self.pending.is_empty(streams) {
            // There's real data to send here, no need to make something up
            return;
        }

        // Retransmit data from the oldest in-flight data from any path
        for packet in self
            .number_spaces
            .values_mut()
            .flat_map(|s| s.sent_packets.values_mut())
        {
            if !packet.retransmits.is_empty(streams) {
                // Remove retransmitted data from the old packet so we don't end up retransmitting
                // it *again* even if the copy we're sending now gets acknowledged.
                self.pending |= mem::take(&mut packet.retransmits);
                return;
            }
        }

        // Nothing new to send and nothing to retransmit, so fall back on a ping. This should only
        // happen in rare cases during the handshake when the server becomes blocked by
        // anti-amplification.
        if !self.for_path(path_id).pending_immediate_ack {
            self.for_path(path_id).pending_ping = true;
        }
    }

    /// Whether there is anything to send in this space
    ///
    /// For the data space [`Connection::can_send_1rtt`] also needs to be consulted. Prefer
    /// to use [`Connection::space_can_send`] which handles this.
    ///
    /// [`Connection::can_send_1rtt`]: super::Connection::can_send_1rtt
    /// [`Connection::space_can_send`]: super::Connection::space_can_send
    pub(super) fn can_send(&self, path_id: PathId, streams: &StreamsState) -> SendableFrames {
        let acks = self
            .number_spaces
            .values()
            .any(|pns| pns.pending_acks.can_send());
        let space_specific = self.number_spaces.get(&path_id).is_some_and(|s| {
            s.pending_ping || s.pending_immediate_ack || !s.pending_path_responses.is_empty()
        });
        let other = !self.pending.is_empty(streams);
        SendableFrames {
            acks,
            close: false,
            space_specific,
            other,
        }
    }
}

impl Index<SpaceId> for [PacketSpace; 3] {
    type Output = PacketSpace;
    fn index(&self, space: SpaceId) -> &PacketSpace {
        &self.as_ref()[space as usize]
    }
}

impl IndexMut<SpaceId> for [PacketSpace; 3] {
    fn index_mut(&mut self, space: SpaceId) -> &mut PacketSpace {
        &mut self.as_mut()[space as usize]
    }
}

/// The three QUIC packet number space kinds
///
/// Unlike [`SpaceId`], this always has exactly three variants — it represents the
/// encryption level / space kind, not a specific packet number space identity.
#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub(crate) enum SpaceKind {
    /// Initial packets (client and server).
    Initial = 0,
    /// Handshake packets.
    Handshake = 1,
    /// Data (1-RTT and 0-RTT)
    Data = 2,
}

impl SpaceKind {
    /// Returns the encryption level for this space kind.
    pub(crate) fn encryption_level(self) -> super::EncryptionLevel {
        match self {
            Self::Initial => super::EncryptionLevel::Initial,
            Self::Handshake => super::EncryptionLevel::Handshake,
            Self::Data => super::EncryptionLevel::OneRtt,
        }
    }
}

impl Index<SpaceKind> for [PacketSpace; 3] {
    type Output = PacketSpace;
    fn index(&self, space: SpaceKind) -> &PacketSpace {
        &self.as_ref()[space as usize]
    }
}

impl IndexMut<SpaceKind> for [PacketSpace; 3] {
    fn index_mut(&mut self, space: SpaceKind) -> &mut PacketSpace {
        &mut self.as_mut()[space as usize]
    }
}

/// The state of a single packet number space.
///
/// In RFC9000 there are 3 packet number spaces: Initial, Handshake and Data. In QUIC
/// Multipath there are multiple packet number spaces for Data, each identified by a
/// [`PathId`].
///
/// This contains the state for a packet number space which is not specific to the 4-tuple
/// this space is currently using. The 4-tuple specific state, like congestion controller,
/// pacing, ECN, MTU etc, is stored in [`PathData`].
///
/// Note that the `Initial`, `Handshake` and `Data(PathId(0))` space all share the same
/// [`PathData`].
///
/// You should access this via [`PacketSpace::for_path`].
///
/// [`PathData`]: super::paths::PathData
pub(super) struct PacketNumberSpace {
    /// Whether the path has already been considered opened from an application perspective.
    ///
    /// This means, for paths other than the original [`PathId::ZERO`], a first path
    /// challenge has been responded to, regardless of the initial validation status of the
    /// path. This state is irreversible, since it's not affected by the path being closed.
    ///
    /// Sending a PATH_CHALLENGE and receiving a valid response before the application is
    /// informed of the path, is a way to ensure the path is usable before it is
    /// reported. This is not required by the spec, and in the future might be changed for
    /// simply requiring a first ack'd packet.
    pub(super) open_status: OpenStatus,
    /// The QUIC-MULTIPATH path status.
    ///
    /// This field is unused for the Initial and Handshake spaces, and when multipath is not
    /// negotiated.
    pub(super) status: PathStatusState,

    /// Highest received packet number, if any
    pub(super) largest_received_packet_number: Option<u64>,
    /// The packet number of the next packet that will be sent, if any. In the Data space, the
    /// packet number stored here is sometimes skipped by [`PacketNumberFilter`] logic.
    pub(super) next_packet_number: u64,
    /// The largest packet number the remote peer acknowledged in an ACK frame.
    pub(super) largest_acked_packet_pn: Option<u64>,
    pub(super) largest_acked_packet_send_time: Instant,
    /// The highest-numbered ACK-eliciting packet we've sent
    pub(super) largest_ack_eliciting_sent: u64,
    /// Number of packets in `sent_packets` with numbers above `largest_ack_eliciting_sent`
    pub(super) unacked_non_ack_eliciting_tail: u64,
    /// Transmitted but not acked
    // We use a BTreeMap here so we can efficiently query by range on ACK and for loss detection
    pub(super) sent_packets: SortedIndexBuffer<SentPacket>,
    /// Packets that were deemed lost
    // Older packets are regularly removed in `Connection::drain_lost_packets`.
    pub(super) lost_packets: SortedIndexBuffer<LostPacket>,
    /// Number of explicit congestion notification codepoints seen on incoming packets
    pub(super) ecn_counters: frame::EcnCounts,
    /// Recent ECN counters sent by the peer in ACK frames
    ///
    /// Updated (and inspected) whenever we receive an ACK with a new highest acked packet
    /// number. Stored per-space to simplify verification, which would otherwise have difficulty
    /// distinguishing between ECN bleaching and counts having been updated by a near-simultaneous
    /// ACK already processed in another space.
    pub(super) ecn_feedback: frame::EcnCounts,
    /// A PING frame needs to be sent on this path.
    pub(super) pending_ping: bool,
    /// Packet numbers to acknowledge.
    pub(super) pending_acks: PendingAcks,
    /// An IMMEDIATE_ACK (draft-ietf-quic-ack-frequency) frame needs to be sent on this path.
    pub(super) pending_immediate_ack: bool,
    /// Responses to path challenges that need to be sent, on and off-path.
    ///
    /// Responses are only tied to the 4-tuple they were received on, not to a specific path
    /// generation. Whether the response is on or off-path only depends on the path's
    /// 4-tuple at sending time.
    pub(super) pending_path_responses: PathResponses,
    /// Packet deduplicator
    pub(super) dedup: Dedup,

    //
    // Loss Detection
    /// The time the most recently sent retransmittable packet was sent.
    pub(super) time_of_last_ack_eliciting_packet: Option<Instant>,
    /// Earliest time when we might declare a packet lost.
    ///
    /// The time at which the earliest sent packet in this space will be considered lost
    /// based on exceeding the reordering window in time. Only set for packets numbered
    /// prior to a packet that has been acknowledged.
    pub(super) loss_time: Option<Instant>,
    /// Number of tail loss probes to send
    pub(super) loss_probes: u32,

    /// Packet numbers to skip, only used in the data package space.
    pn_filter: Option<PacketNumberFilter>,
}

impl PacketNumberSpace {
    pub(super) fn new(now: Instant, space: SpaceId, rng: &mut (impl CryptoRng + ?Sized)) -> Self {
        let pn_filter = match space {
            SpaceId::Initial | SpaceId::Handshake => None,
            SpaceId::Data => Some(PacketNumberFilter::new(rng)),
        };
        Self {
            open_status: OpenStatus::default(),
            status: PathStatusState::default(),
            largest_received_packet_number: None,
            next_packet_number: 0,
            largest_acked_packet_pn: None,
            largest_acked_packet_send_time: now,
            largest_ack_eliciting_sent: 0,
            unacked_non_ack_eliciting_tail: 0,
            sent_packets: SortedIndexBuffer::new(),
            lost_packets: SortedIndexBuffer::new(),
            ecn_counters: frame::EcnCounts::ZERO,
            ecn_feedback: frame::EcnCounts::ZERO,
            pending_ping: false,
            pending_acks: PendingAcks::new(),
            pending_immediate_ack: false,
            pending_path_responses: PathResponses::default(),
            dedup: Default::default(),
            time_of_last_ack_eliciting_packet: None,
            loss_time: None,
            loss_probes: 0,
            pn_filter,
        }
    }

    #[cfg(test)]
    fn new_deterministic(now: Instant, space: SpaceId) -> Self {
        let pn_filter = match space {
            SpaceId::Initial | SpaceId::Handshake => None,
            SpaceId::Data => Some(PacketNumberFilter::disabled()),
        };
        Self {
            open_status: OpenStatus::default(),
            status: PathStatusState::default(),
            largest_received_packet_number: None,
            next_packet_number: 0,
            largest_acked_packet_pn: None,
            largest_acked_packet_send_time: now,
            largest_ack_eliciting_sent: 0,
            unacked_non_ack_eliciting_tail: 0,
            sent_packets: SortedIndexBuffer::new(),
            lost_packets: SortedIndexBuffer::new(),
            ecn_counters: frame::EcnCounts::ZERO,
            ecn_feedback: frame::EcnCounts::ZERO,
            pending_ping: false,
            pending_acks: PendingAcks::new(),
            pending_immediate_ack: false,
            pending_path_responses: PathResponses::default(),
            dedup: Default::default(),
            time_of_last_ack_eliciting_packet: None,
            loss_time: None,
            loss_probes: 0,
            pn_filter,
        }
    }

    pub(crate) fn remote_status(&self) -> Option<PathStatus> {
        self.status.remote_status.map(|(_seq, status)| status)
    }

    pub(crate) fn local_status(&self) -> PathStatus {
        self.status.local_status
    }

    /// Get the next outgoing packet number in this space
    ///
    /// In the Data space, the connection's [`PacketNumberFilter`] must be used rather than calling
    /// this directly.
    pub(super) fn get_tx_number(&mut self, rng: &mut (impl CryptoRng + ?Sized)) -> u64 {
        // TODO: Handle packet number overflow gracefully
        assert!(self.next_packet_number < 2u64.pow(62));
        let mut pn = self.next_packet_number;
        self.next_packet_number += 1;

        // Skip this number if the filter says so, only enabled in the data space
        if let Some(ref mut filter) = self.pn_filter
            && filter.skip_pn(pn, rng)
        {
            pn = self.next_packet_number;
            self.next_packet_number += 1;
        }
        pn
    }

    pub(super) fn peek_tx_number(&mut self) -> u64 {
        let pn = self.next_packet_number;
        if let Some(ref filter) = self.pn_filter
            && pn == filter.next_skipped_packet_number
        {
            return pn + 1;
        }
        pn
    }

    /// Checks whether a skipped packet number was ACKed.
    pub(super) fn check_ack(&self, range: std::ops::Range<u64>) -> Result<(), TransportError> {
        if let Some(ref filter) = self.pn_filter
            && filter
                .prev_skipped_packet_number
                .is_some_and(|pn| range.contains(&pn))
        {
            return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked"));
        }
        Ok(())
    }

    /// Verifies sanity of an ECN block and returns whether congestion was encountered.
    pub(super) fn detect_ecn(
        &mut self,
        newly_acked: u64,
        ecn: frame::EcnCounts,
    ) -> Result<bool, &'static str> {
        let ect0_increase = ecn
            .ect0
            .checked_sub(self.ecn_feedback.ect0)
            .ok_or("peer ECT(0) count regression")?;
        let ect1_increase = ecn
            .ect1
            .checked_sub(self.ecn_feedback.ect1)
            .ok_or("peer ECT(1) count regression")?;
        let ce_increase = ecn
            .ce
            .checked_sub(self.ecn_feedback.ce)
            .ok_or("peer CE count regression")?;
        let total_increase = ect0_increase + ect1_increase + ce_increase;
        if total_increase < newly_acked {
            return Err("ECN bleaching");
        }
        if (ect0_increase + ce_increase) < newly_acked || ect1_increase != 0 {
            return Err("ECN corruption");
        }
        // If total_increase > newly_acked (which happens when ACKs are lost), this is required by
        // the draft so that long-term drift does not occur. If =, then the only question is whether
        // to count CE packets as CE or ECT0. Recording them as CE is more consistent and keeps the
        // congestion check obvious.
        self.ecn_feedback = ecn;
        Ok(ce_increase != 0)
    }

    /// Stop tracking sent packet `number`, and return what we knew about it
    pub(super) fn take(&mut self, number: u64) -> Option<SentPacket> {
        let packet = self.sent_packets.remove(number)?;
        if !packet.ack_eliciting && number > self.largest_ack_eliciting_sent {
            self.unacked_non_ack_eliciting_tail =
                self.unacked_non_ack_eliciting_tail.checked_sub(1).unwrap();
        }
        Some(packet)
    }

    /// May return a packet that should be forgotten
    pub(super) fn sent(&mut self, number: u64, packet: SentPacket) -> Option<SentPacket> {
        // Retain state for at most this many non-ACK-eliciting packets sent after the most recently
        // sent ACK-eliciting packet. We're never guaranteed to receive an ACK for those, and we
        // can't judge them as lost without an ACK, so to limit memory in applications which receive
        // packets but don't send ACK-eliciting data for long periods use we must eventually start
        // forgetting about them, although it might also be reasonable to just kill the connection
        // due to weird peer behavior.
        const MAX_UNACKED_NON_ACK_ELICTING_TAIL: u64 = 1_000;

        let mut forgotten = None;
        if packet.ack_eliciting {
            self.unacked_non_ack_eliciting_tail = 0;
            self.largest_ack_eliciting_sent = number;
        } else if self.unacked_non_ack_eliciting_tail > MAX_UNACKED_NON_ACK_ELICTING_TAIL {
            let oldest_after_ack_eliciting = self
                .sent_packets
                .keys_range((
                    Bound::Excluded(self.largest_ack_eliciting_sent),
                    Bound::Unbounded,
                ))
                .next()
                .unwrap();
            // Per https://www.rfc-editor.org/rfc/rfc9000.html#name-frames-and-frame-types,
            // non-ACK-eliciting packets must only contain PADDING, ACK, and CONNECTION_CLOSE
            // frames, which require no special handling on ACK or loss beyond removal from
            // in-flight counters if padded.
            let packet = self
                .sent_packets
                .remove(oldest_after_ack_eliciting)
                .unwrap();
            debug_assert!(!packet.ack_eliciting);
            forgotten = Some(packet);
        } else {
            self.unacked_non_ack_eliciting_tail += 1;
        }

        self.sent_packets.insert(number, packet);
        forgotten
    }

    /// Whether any congestion-controlled packets in this space are not yet acknowledged or lost
    pub(super) fn has_in_flight(&self) -> bool {
        // The number of non-congestion-controlled (i.e. size == 0) packets in flight at a time
        // should be small, since otherwise congestion control wouldn't be effective. Therefore,
        // this shouldn't need to visit many packets before finishing one way or another.
        self.sent_packets.values().any(|x| x.size != 0)
    }
}

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(super) enum OpenStatus {
    /// The application has not yet been informed of this path.
    #[default]
    Pending,
    /// The application has been informed of this path.
    Informed,
}

/// State for QUIC-MULTIPATH PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP frames
#[derive(Debug, Clone, Default)]
pub(super) struct PathStatusState {
    /// The local status
    local_status: PathStatus,
    /// Local sequence number, for both PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP
    ///
    /// This is the number of the *next* path status frame to be sent.
    local_seq: VarInt,
    /// The status set by the remote
    remote_status: Option<(VarInt, PathStatus)>,
}

impl PathStatusState {
    /// To be called on received PATH_STATUS_AVAILABLE/PATH_STATUS_BACKUP frames
    pub(super) fn remote_update(&mut self, status: PathStatus, seq: VarInt) {
        if self.remote_status.is_some_and(|(curr, _)| curr >= seq) {
            return trace!(%seq, "ignoring path status update");
        }

        let prev = self.remote_status.replace((seq, status)).map(|(_, s)| s);
        if prev != Some(status) {
            debug!(?status, ?seq, "remote changed path status");
        }
    }

    /// Updates the local status
    ///
    /// If the local status changed, the previous value is returned
    pub(super) fn local_update(&mut self, status: PathStatus) -> Option<PathStatus> {
        if self.local_status == status {
            return None;
        }

        self.local_seq = self.local_seq.saturating_add(1u8);
        Some(std::mem::replace(&mut self.local_status, status))
    }

    pub(crate) fn seq(&self) -> VarInt {
        self.local_seq
    }
}

/// The QUIC-MULTIPATH path status
///
/// See section "3.3 Path Status Management":
/// <https://quicwg.org/multipath/draft-ietf-quic-multipath.html#name-path-status-management>
#[cfg_attr(test, derive(test_strategy::Arbitrary))]
#[derive(Debug, Copy, Clone, Default, PartialEq, Eq)]
pub enum PathStatus {
    /// Paths marked with as available will be used when scheduling packets
    ///
    /// If multiple paths are available, packets will be scheduled on whichever has
    /// capacity.
    #[default]
    Available,
    /// Paths marked as backup will only be used if there are no available paths
    ///
    /// If the max_idle_timeout is specified the path will be kept alive so that it does not
    /// expire.
    Backup,
}

/// Represents one or more packets subject to retransmission
#[derive(Debug, Clone)]
pub(super) struct SentPacket {
    /// [`PathData::generation`](super::PathData::generation) of the path on which this packet was
    /// sent
    pub(super) path_generation: u64,
    /// The time the packet was sent.
    pub(super) time_sent: Instant,
    /// The number of bytes sent in the packet, not including UDP or IP overhead, but including
    /// QUIC framing overhead. Zero if this packet is not counted towards congestion control,
    /// i.e. not an "in flight" packet.
    pub(super) size: u16,
    /// Whether an acknowledgement is expected directly in response to this packet.
    pub(super) ack_eliciting: bool,
    /// The largest packet number acknowledged by this packet
    pub(super) largest_acked: FxHashMap<PathId, u64>,
    /// Data which needs to be retransmitted in case the packet is lost.
    ///
    /// These might be retransmitted over any available path in the same [`SpaceKind`].
    pub(super) retransmits: ThinRetransmits,
    /// Retransmittable data specific to a path generation.
    pub(super) path_retransmits: PathRetransmits,
    /// Metadata for stream frames in a packet
    ///
    /// The actual application data is stored with the stream state.
    pub(super) stream_frames: frame::StreamMetaVec,
}

/// Represents one or more packets that are deemed lost.
#[derive(Debug)]
pub(super) struct LostPacket {
    /// The time the packet was sent.
    pub(super) time_sent: Instant,
}

/// Retransmittable data queue.
///
/// Data in this queue must be retransmittable over any path in the same [`SpaceKind`].
#[allow(unreachable_pub)] // fuzzing only
#[derive(Debug, Default, Clone)]
pub struct Retransmits {
    pub(super) max_data: bool,
    pub(super) max_stream_id: [bool; 2],
    pub(super) streams_blocked: [bool; 2],
    pub(super) reset_stream: Vec<(StreamId, VarInt)>,
    pub(super) stop_sending: Vec<frame::StopSending>,
    pub(super) max_stream_data: FxHashSet<StreamId>,
    pub(super) crypto: VecDeque<frame::Crypto>,
    pub(super) new_cids: PendingNewCids,
    pub(super) retire_cids: Vec<(PathId, u64)>,
    pub(super) ack_frequency: bool,
    pub(super) handshake_done: bool,
    /// Whether we should inform the peer we will allow higher [`PathId`]s.
    pub(super) max_path_id: bool,
    /// Whether we should inform the peer that their max [`PathId`] is blocking our attempt to open
    /// new paths.
    ///
    /// Stores the remote_max_path_id at the time this was generated.
    /// This frame is entirely informational, so when it's retransmitted, the remote_max_path_id is
    /// intentionally not updated to preserve the fact that this was the state of the client at
    /// some point.
    pub(super) paths_blocked: Option<PathId>,
    /// For each enqueued NEW_TOKEN frame, a copy of the path's remote address
    ///
    /// There are 2 reasons this is unusual:
    ///
    /// - If the path changes, NEW_TOKEN frames bound for the old path are not retransmitted on the
    ///   new path. That is why this field stores the remote address: so that ones for old paths
    ///   can be filtered out.
    /// - If a token is lost, a new randomly generated token is re-transmitted, rather than the
    ///   original. This is so that if both transmissions are received, the client won't risk
    ///   sending the same token twice. That is why this field does _not_ store any actual token.
    ///
    /// It is true that a QUIC endpoint will only want to effectively have NEW_TOKEN frames
    /// enqueued for its current path at a given point in time. Based on that, we could conceivably
    /// change this from a vector to an `Option<(FourTuple, usize)>` or just a `usize` or
    /// something. However, due to the architecture of noq, it is considerably simpler to not do
    /// that; consider what such a change would mean for implementing `BitOrAssign` on Self.
    pub(super) new_tokens: Vec<FourTuple>,
    /// Paths which need to be abandoned
    pub(super) path_abandon: BTreeMap<PathId, TransportErrorCode>,
    /// If a [`frame::PathStatusAvailable`] and [`frame::PathStatusBackup`] need to be sent for a
    /// path
    pub(super) path_status: BTreeSet<PathId>,
    /// Whether a PATH_CIDS_BLOCKED frame needs to be sent for a path.
    ///
    /// Stores the next_seq number for the blocked path. This number can be "outdated" at the time
    /// of sending when this is a retransmission. This is intentional, as this frame is purely
    /// informational, and this would preserve this information.
    pub(super) path_cids_blocked: BTreeMap<PathId, VarInt>,

    // Nat traversal data
    /// Addresses to report in `ADD_ADDRESS` frames
    pub(super) add_address: BTreeSet<AddAddress>,
    /// Address IDs to remove in `REMOVE_ADDRESS` frames
    pub(super) remove_address: BTreeSet<RemoveAddress>,
    /// Round and local addresses to advertise in `REACH_OUT` frames
    pub(super) reach_out: PendingReachOutFrames,
}

impl Retransmits {
    pub(super) fn is_empty(&self, streams: &StreamsState) -> bool {
        let Self {
            max_data,
            max_stream_id,
            streams_blocked,
            reset_stream,
            stop_sending,
            max_stream_data,
            crypto,
            new_cids,
            retire_cids,
            ack_frequency,
            handshake_done,
            max_path_id,
            paths_blocked,
            new_tokens,
            path_abandon,
            path_status,
            path_cids_blocked,
            add_address,
            remove_address,
            reach_out,
        } = &self;
        !max_data
            && !max_stream_id.iter().any(|x| *x)
            && !streams_blocked.iter().any(|x| *x)
            && reset_stream.is_empty()
            && stop_sending.is_empty()
            && max_stream_data
                .iter()
                .all(|&id| !streams.can_send_flow_control(id))
            && crypto.is_empty()
            && new_cids.is_empty()
            && retire_cids.is_empty()
            && !ack_frequency
            && !handshake_done
            && !max_path_id
            && paths_blocked.is_none()
            && new_tokens.is_empty()
            && path_abandon.is_empty()
            && path_status.is_empty()
            && path_cids_blocked.is_empty()
            && add_address.is_empty()
            && remove_address.is_empty()
            && reach_out.is_empty()
    }
}

impl ::std::ops::BitOrAssign for Retransmits {
    fn bitor_assign(&mut self, rhs: Self) {
        let Self {
            max_data,
            max_stream_id,
            streams_blocked,
            reset_stream,
            stop_sending,
            max_stream_data,
            crypto,
            new_cids,
            retire_cids,
            ack_frequency,
            handshake_done,
            max_path_id,
            paths_blocked,
            new_tokens,
            mut path_abandon,
            mut path_status,
            mut path_cids_blocked,
            add_address,
            remove_address,
            mut reach_out,
        } = rhs;

        // We reduce in-stream head-of-line blocking by queueing retransmits before other data for
        // STREAM and CRYPTO frames.
        self.max_data |= max_data;
        for dir in Dir::iter() {
            self.max_stream_id[dir as usize] |= max_stream_id[dir as usize];
            self.streams_blocked[dir as usize] |= streams_blocked[dir as usize];
        }
        self.reset_stream.extend_from_slice(&reset_stream);
        self.stop_sending.extend_from_slice(&stop_sending);
        self.max_stream_data.extend(&max_stream_data);
        for crypto in crypto.into_iter().rev() {
            self.crypto.push_front(crypto);
        }
        self.new_cids.extend(&new_cids);
        self.retire_cids.extend(retire_cids);
        self.ack_frequency |= ack_frequency;
        self.handshake_done |= handshake_done;
        self.max_path_id |= max_path_id;
        self.paths_blocked = cmp::max(self.paths_blocked, paths_blocked);
        self.new_tokens.extend_from_slice(&new_tokens);
        self.path_abandon.append(&mut path_abandon);
        self.path_status.append(&mut path_status);
        self.path_cids_blocked.append(&mut path_cids_blocked);
        self.add_address.extend(add_address.iter().copied());
        self.remove_address.extend(remove_address.iter().copied());
        self.reach_out.append(&mut reach_out);
    }
}

impl ::std::ops::BitOrAssign<ThinRetransmits> for Retransmits {
    fn bitor_assign(&mut self, rhs: ThinRetransmits) {
        let ThinRetransmits { retransmits } = rhs;
        if let Some(retransmits) = retransmits {
            self.bitor_assign(*retransmits)
        }
    }
}

impl ::std::iter::FromIterator<Self> for Retransmits {
    fn from_iter<T>(iter: T) -> Self
    where
        T: IntoIterator<Item = Self>,
    {
        let mut result = Self::default();
        for packet in iter {
            result |= packet;
        }
        result
    }
}

/// The queue of new CIDs to be transmitted to the peer.
///
/// This queue is always sorted, so that popping off the last item is always the lowest
/// sequence number of the lowest path ID. Which is the CID you want to be issued next.
///
/// This is but a newtype over a `Vec` to enforce the sorted invariant.
#[derive(Clone, Debug, Default)]
pub(super) struct PendingNewCids {
    /// The CIDs themselves.
    cids: Vec<IssuedCid>,
    /// Whether [`Self::cids`] is sorted or not.
    sorted: bool,
}

impl PendingNewCids {
    /// Inserts an issued CID into the queue.
    pub(super) fn push(&mut self, cid: IssuedCid) {
        self.cids.push(cid);
        self.sorted = false;
    }

    /// Pops the next issued CID to transmit from the queue.
    pub(super) fn pop(&mut self) -> Option<IssuedCid> {
        if !mem::replace(&mut self.sorted, true) {
            self.cids
                .sort_by_key(|cid| cmp::Reverse((cid.path_id, cid.sequence)));
        }
        self.cids.pop()
    }

    pub(super) fn is_empty(&self) -> bool {
        self.cids.is_empty()
    }

    pub(super) fn extend(&mut self, other: &Self) {
        self.cids.extend(&other.cids);
        self.sorted = false;
    }

    pub(super) fn retain<F>(&mut self, f: F)
    where
        F: FnMut(&IssuedCid) -> bool,
    {
        self.cids.retain(f);
    }
}

/// Logically a Vec of REACH_OUT frames queued for transmit.
///
/// This keeps track of the highest round ID and automatically drops frames with a lower
/// round ID.
///
/// The API is directly modelled on [`Vec`].
#[derive(Debug, Default, Clone)]
pub(crate) struct PendingReachOutFrames {
    /// The round ID of the REACH_OUT frames currently pending.
    round: VarInt,
    /// The REACH_OUT frames, always all having the same round ID.
    frames: Vec<frame::ReachOut>,
}

impl PendingReachOutFrames {
    pub(crate) fn len(&self) -> usize {
        self.frames.len()
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.frames.is_empty()
    }

    pub(crate) fn push(&mut self, frame: frame::ReachOut) {
        if frame.round < self.round {
            return;
        } else if frame.round > self.round {
            self.round = frame.round;
            self.frames.clear();
        }
        self.frames.push(frame);
    }

    pub(crate) fn append(&mut self, other: &mut Self) {
        if other.round < self.round {
            other.frames.clear();
            return;
        } else if other.round > self.round {
            self.round = other.round;
            self.frames.clear();
        }
        self.frames.append(&mut other.frames);
    }

    pub(crate) fn pop_if(
        &mut self,
        predicate: impl FnOnce(&mut frame::ReachOut) -> bool,
    ) -> Option<frame::ReachOut> {
        self.frames.pop_if(predicate)
    }
}

impl FromIterator<frame::ReachOut> for PendingReachOutFrames {
    fn from_iter<T: IntoIterator<Item = frame::ReachOut>>(iter: T) -> Self {
        let iter = iter.into_iter();
        let size_hint = iter.size_hint();
        let mut this = Self {
            round: Default::default(),
            frames: Vec::with_capacity(size_hint.1.unwrap_or(size_hint.0)),
        };
        for frame in iter {
            this.push(frame);
        }
        this
    }
}

/// A variant of `Retransmits` which only allocates storage when required
#[derive(Debug, Default, Clone)]
pub(super) struct ThinRetransmits {
    retransmits: Option<Box<Retransmits>>,
}

impl ThinRetransmits {
    /// Returns `true` if no retransmits are necessary
    pub(super) fn is_empty(&self, streams: &StreamsState) -> bool {
        match &self.retransmits {
            Some(retransmits) => retransmits.is_empty(streams),
            None => true,
        }
    }

    /// Returns a reference to the retransmits stored in this box
    pub(super) fn get(&self) -> Option<&Retransmits> {
        self.retransmits.as_deref()
    }

    /// Returns a mutable reference to the retransmits stored in this box
    pub(super) fn get_mut(&mut self) -> Option<&mut Retransmits> {
        self.retransmits.as_deref_mut()
    }

    /// Returns a mutable reference to the stored retransmits
    ///
    /// This function will allocate a backing storage if required.
    pub(super) fn get_or_create(&mut self) -> &mut Retransmits {
        if self.retransmits.is_none() {
            self.retransmits = Some(Box::default());
        }
        self.retransmits.as_deref_mut().unwrap()
    }
}

/// RFC4303-style sliding window packet number deduplicator.
///
/// A contiguous bitfield, where each bit corresponds to a packet number and the rightmost bit is
/// always set. A set bit represents a packet that has been successfully authenticated. Bits left of
/// the window are assumed to be set.
///
/// ```text
/// ...xxxxxxxxx 1 0
///     ^        ^ ^
/// window highest next
/// ```
#[derive(Debug, Default)]
pub(super) struct Dedup {
    window: Window,
    /// Lowest packet number higher than all yet authenticated.
    next: u64,
}

impl Dedup {
    /// Construct an empty window positioned at the start.
    #[cfg(test)]
    pub(super) fn new() -> Self {
        Self { window: 0, next: 0 }
    }

    /// Highest packet number authenticated.
    fn highest(&self) -> u64 {
        self.next - 1
    }

    /// Record a newly authenticated packet number.
    ///
    /// Returns whether the packet might be a duplicate.
    pub(super) fn insert(&mut self, packet: u64) -> bool {
        if let Some(diff) = packet.checked_sub(self.next) {
            // Right of window
            self.window = ((self.window << 1) | 1)
                .checked_shl(cmp::min(diff, u64::from(u32::MAX)) as u32)
                .unwrap_or(0);
            self.next = packet + 1;
            false
        } else if self.highest() - packet < WINDOW_SIZE {
            // Within window
            if let Some(bit) = (self.highest() - packet).checked_sub(1) {
                // < highest
                let mask = 1 << bit;
                let duplicate = self.window & mask != 0;
                self.window |= mask;
                duplicate
            } else {
                // == highest
                true
            }
        } else {
            // Left of window
            true
        }
    }

    /// Returns the packet number of the smallest packet missing between the provided interval
    ///
    /// If there are no missing packets, returns `None`
    fn smallest_missing_in_interval(&self, lower_bound: u64, upper_bound: u64) -> Option<u64> {
        debug_assert!(lower_bound <= upper_bound);
        debug_assert!(upper_bound <= self.highest());
        const BITFIELD_SIZE: u64 = Window::BITS as u64;

        // Since we already know the packets at the boundaries have been received, we only need to
        // check those in between them (this removes the necessity of extra logic to deal with the
        // highest packet, which is stored outside the bitfield)
        let lower_bound = lower_bound + 1;
        let upper_bound = upper_bound.saturating_sub(1);

        // Note: the offsets are counted from the right
        // The highest packet is not included in the bitfield, so we subtract 1 to account for that
        let start_offset = (self.highest() - upper_bound).max(1) - 1;
        if start_offset >= BITFIELD_SIZE {
            // The start offset is outside of the window. All packets outside of the window are
            // considered to be received.
            return None;
        }

        let end_offset_exclusive = self.highest().saturating_sub(lower_bound);

        // The range is clamped at the edge of the window, because any earlier packets are
        // considered to be received
        let range_len = end_offset_exclusive
            .saturating_sub(start_offset)
            .min(BITFIELD_SIZE);
        if range_len == 0 {
            return None;
        }

        // Ensure the shift is within bounds (we already know start_offset < BITFIELD_SIZE,
        // because of the early return)
        let mask = if range_len == BITFIELD_SIZE {
            u128::MAX
        } else {
            ((1u128 << range_len) - 1) << start_offset
        };
        let gaps = !self.window & mask;

        let smallest_missing_offset = 128 - gaps.leading_zeros() as u64;
        let smallest_missing_packet = self.highest() - smallest_missing_offset;

        if smallest_missing_packet <= upper_bound {
            Some(smallest_missing_packet)
        } else {
            None
        }
    }

    /// Returns true if there are any missing packets between the provided interval
    ///
    /// The provided packet numbers must have been received before calling this function
    fn missing_in_interval(&self, lower_bound: u64, upper_bound: u64) -> bool {
        self.smallest_missing_in_interval(lower_bound, upper_bound)
            .is_some()
    }
}

/// Inner bitfield type.
///
/// Because QUIC never reuses packet numbers, this only needs to be large enough to deal with
/// packets that are reordered but still delivered in a timely manner.
type Window = u128;

/// Number of packets tracked by `Dedup`.
const WINDOW_SIZE: u64 = 1 + size_of::<Window>() as u64 * 8;
/// Indicates which data is available for sending
///
/// This applies to a particular space ID that was queried and all refers to on-path data.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(super) struct SendableFrames {
    /// Whether there are ACK frames to send, these are not ack-eliciting.
    pub(super) acks: bool,
    /// Whether there is a CONNECTION_CLOSE to send, this is not ack-eliciting.
    pub(super) close: bool,
    /// Whether there are any frames that must be sent on this specific space.
    ///
    /// A space here in the sense of a QUIC Multipath packet number space: `Initial`,
    /// `Handshake` and all `Data(PathId)` spaces.
    ///
    /// These are ack-eliciting. Some frames are scheduled per path, e.g. PING,
    /// IMMEDIATE_ACK, PATH_CHALLENGE or PATH_RESPONSE.
    pub(super) space_specific: bool,
    /// Whether there are any other frames to send, these are ack-eliciting.
    pub(super) other: bool,
}

impl SendableFrames {
    /// Returns that no data is available for sending
    pub(super) fn empty() -> Self {
        Self {
            acks: false,
            close: false,
            space_specific: false,
            other: false,
        }
    }

    /// Whether an ack-eliciting packet will be sent.
    pub(super) fn is_ack_eliciting(&self) -> bool {
        let Self {
            acks: _,
            close,
            space_specific,
            other,
        } = *self;
        if close {
            // No ack-eliciting frames are included with a CONNECTION_CLOSE, only acks.
            return false;
        }
        space_specific || other
    }

    /// Whether no data is sendable.
    pub(super) fn is_empty(&self) -> bool {
        let Self {
            acks,
            close,
            space_specific,
            other,
        } = *self;
        !acks && !close && !space_specific && !other
    }
}

impl ::std::ops::BitOrAssign for SendableFrames {
    fn bitor_assign(&mut self, rhs: Self) {
        let Self {
            acks,
            close,
            space_specific,
            other,
        } = rhs;

        self.acks |= acks;
        self.close |= close;
        self.space_specific |= space_specific;
        self.other |= other;
    }
}

#[derive(Debug)]
pub(super) struct PendingAcks {
    /// Whether we should send an ACK immediately, even if that means sending an ACK-only packet
    ///
    /// When `immediate_ack_required` is false, the normal behavior is to send ACK frames only when
    /// there is other data to send, or when the `MaxAckDelay` timer expires.
    immediate_ack_required: bool,
    /// The number of ack-eliciting packets received since the last ACK frame was sent
    ///
    /// Once the count _exceeds_ `ack_eliciting_threshold`, an immediate ACK is required
    ack_eliciting_since_last_ack_sent: u64,
    non_ack_eliciting_since_last_ack_sent: u64,
    ack_eliciting_threshold: u64,
    /// The reordering threshold, controlling how we respond to out-of-order ack-eliciting packets
    ///
    /// Different values enable different behavior:
    ///
    /// * `0`: no special action is taken
    /// * `1`: an ACK is immediately sent if it is out-of-order according to RFC 9000
    /// * `>1`: an ACK is immediately sent if it is out-of-order according to the ACK frequency
    ///   draft
    reordering_threshold: u64,
    /// The earliest ack-eliciting packet since the last ACK was sent, used to calculate the moment
    /// upon which `max_ack_delay` elapses
    earliest_ack_eliciting_since_last_ack_sent: Option<Instant>,
    /// Packet number ranges for which to still send acknowledgements.
    ///
    /// These are packet number ranges of ack-eliciting packets the peer has sent and which
    /// need to be acknowledged.  Packet numbers are only removed from here once the peer has
    /// acknowledged the ACKs for them.
    ranges: ArrayRangeSet,
    /// The largest packet number received and the time it was received
    ///
    /// Used to calculate ACK delay in [`PendingAcks::ack_delay`].
    largest_packet: Option<(u64, Instant)>,
    /// The ack-eliciting packet we have received with the largest packet number
    largest_ack_eliciting_packet: Option<u64>,
    /// The largest acknowledged packet number sent in an ACK frame
    largest_acked: Option<u64>,
}

impl PendingAcks {
    fn new() -> Self {
        Self {
            immediate_ack_required: false,
            ack_eliciting_since_last_ack_sent: 0,
            non_ack_eliciting_since_last_ack_sent: 0,
            ack_eliciting_threshold: 1,
            reordering_threshold: 1,
            earliest_ack_eliciting_since_last_ack_sent: None,
            ranges: Default::default(),
            largest_packet: Default::default(),
            largest_ack_eliciting_packet: Default::default(),
            largest_acked: Default::default(),
        }
    }

    pub(super) fn set_ack_frequency_params(&mut self, frame: &frame::AckFrequency) {
        self.ack_eliciting_threshold = frame.ack_eliciting_threshold.into_inner();
        self.reordering_threshold = frame.reordering_threshold.into_inner();
    }

    pub(super) fn set_immediate_ack_required(&mut self) {
        self.immediate_ack_required = true;
    }

    pub(super) fn on_max_ack_delay_timeout(&mut self) {
        self.immediate_ack_required = self.ack_eliciting_since_last_ack_sent > 0;
    }

    pub(super) fn max_ack_delay_timeout(&self, max_ack_delay: Duration) -> Option<Instant> {
        self.earliest_ack_eliciting_since_last_ack_sent
            .map(|earliest_unacked| earliest_unacked + max_ack_delay)
    }

    /// Whether any ACK frames SHOULD be sent
    ///
    /// This is used in the top-level [`Connection::space_can_send`], so determines if a
    /// packet will be built. It is often possible to construct new ACK ranges to send
    /// before this returns `true`. This results in more ACK frames being sent, and
    /// processing those at the receiver costs CPU for very little improvements.
    ///
    /// [`Connection::space_can_send`]: super::Connection::space_can_send
    pub(super) fn can_send(&self) -> bool {
        self.immediate_ack_required && !self.ranges.is_empty()
    }

    /// Returns the delay since the packet with the largest packet number was received
    pub(super) fn ack_delay(&self, now: Instant) -> Duration {
        self.largest_packet
            .map_or_else(Duration::default, |(_, received)| now - received)
    }

    /// Handle receipt of a new packet
    ///
    /// Returns true if the max ack delay timer should be armed
    pub(super) fn packet_received(
        &mut self,
        now: Instant,
        packet_number: u64,
        ack_eliciting: bool,
        dedup: &Dedup,
    ) -> bool {
        if !ack_eliciting {
            self.non_ack_eliciting_since_last_ack_sent += 1;
            return false;
        }

        let prev_largest_ack_eliciting = self.largest_ack_eliciting_packet.unwrap_or(0);

        // Track largest ack-eliciting packet
        self.largest_ack_eliciting_packet = self
            .largest_ack_eliciting_packet
            .map(|pn| pn.max(packet_number))
            .or(Some(packet_number));

        // Handle ack_eliciting_threshold
        self.ack_eliciting_since_last_ack_sent += 1;
        self.immediate_ack_required |=
            self.ack_eliciting_since_last_ack_sent > self.ack_eliciting_threshold;

        // Handle out-of-order packets
        self.immediate_ack_required |=
            self.is_out_of_order(packet_number, prev_largest_ack_eliciting, dedup);

        // Arm max_ack_delay timer if necessary
        if self.earliest_ack_eliciting_since_last_ack_sent.is_none() && !self.can_send() {
            self.earliest_ack_eliciting_since_last_ack_sent = Some(now);
            return true;
        }

        false
    }

    fn is_out_of_order(
        &self,
        packet_number: u64,
        prev_largest_ack_eliciting: u64,
        dedup: &Dedup,
    ) -> bool {
        match self.reordering_threshold {
            0 => false,
            1 => {
                // From https://www.rfc-editor.org/rfc/rfc9000#section-13.2.1-7
                packet_number < prev_largest_ack_eliciting
                    || dedup.missing_in_interval(prev_largest_ack_eliciting, packet_number)
            }
            _ => {
                // From acknowledgement frequency draft, section 6.1: send an ACK immediately if
                // doing so would cause the sender to detect a new packet loss
                let Some((largest_acked, largest_unacked)) =
                    self.largest_acked.zip(self.largest_ack_eliciting_packet)
                else {
                    return false;
                };
                if self.reordering_threshold > largest_acked {
                    return false;
                }
                // The largest packet number that could be declared lost without a new ACK being
                // sent
                let largest_reported = largest_acked - self.reordering_threshold + 1;
                let Some(smallest_missing_unreported) =
                    dedup.smallest_missing_in_interval(largest_reported, largest_unacked)
                else {
                    return false;
                };
                largest_unacked - smallest_missing_unreported >= self.reordering_threshold
            }
        }
    }

    /// Should be called whenever ACKs have been sent
    ///
    /// This will suppress sending further ACKs until additional ACK eliciting frames arrive
    pub(super) fn acks_sent(&mut self) {
        // It is possible (though unlikely) that the ACKs we just sent do not cover all the
        // ACK-eliciting packets we have received (e.g. if there is not enough room in the packet to
        // fit all the ranges). To keep things simple, however, we assume they do. If there are
        // indeed some ACKs that weren't covered, the packets might be ACKed later anyway, because
        // they are still contained in `self.ranges`. If we somehow fail to send the ACKs at a later
        // moment, the peer will assume the packets got lost and will retransmit their frames in a
        // new packet, which is suboptimal, because we already received them. Our assumption here is
        // that simplicity results in code that is more performant, even in the presence of
        // occasional redundant retransmits.
        self.immediate_ack_required = false;
        self.ack_eliciting_since_last_ack_sent = 0;
        self.non_ack_eliciting_since_last_ack_sent = 0;
        self.earliest_ack_eliciting_since_last_ack_sent = None;
        self.largest_acked = self.largest_ack_eliciting_packet;
    }

    /// Insert one packet that needs to be acknowledged
    pub(super) fn insert_one(&mut self, packet: u64, now: Instant) {
        self.ranges.insert_one(packet);

        if self.largest_packet.is_none_or(|(pn, _)| packet > pn) {
            self.largest_packet = Some((packet, now));
        }

        if self.ranges.range_count() > MAX_ACK_BLOCKS {
            self.ranges.pop_min();
        }
    }

    /// Remove ACKs of packets numbered at or below `max` from the set of pending ACKs
    pub(super) fn subtract_below(&mut self, max: u64) {
        self.ranges.remove(0..(max + 1));
    }

    /// Returns the set of currently pending ACK ranges
    pub(super) fn ranges(&self) -> &ArrayRangeSet {
        &self.ranges
    }

    /// Queue an ACK if a significant number of non-ACK-eliciting packets have not yet been
    /// acknowledged
    ///
    /// Should be called immediately before a non-probing packet is composed, when we've already
    /// committed to sending a packet regardless.
    pub(super) fn maybe_ack_non_eliciting(&mut self) {
        // If we're going to send a packet anyway, and we've received a significant number of
        // non-ACK-eliciting packets, then include an ACK to help the peer perform timely loss
        // detection even if they're not sending any ACK-eliciting packets themselves. Exact
        // threshold chosen somewhat arbitrarily.
        const LAZY_ACK_THRESHOLD: u64 = 10;
        if self.non_ack_eliciting_since_last_ack_sent > LAZY_ACK_THRESHOLD {
            self.immediate_ack_required = true;
        }
    }
}

/// Helper for mitigating [optimistic ACK attacks]
///
/// A malicious peer could prompt the local application to begin a large data transfer, and then
/// send ACKs without first waiting for data to be received. This could defeat congestion control,
/// allowing the connection to consume disproportionate resources. We therefore occasionally skip
/// packet numbers, and classify any ACK referencing a skipped packet number as a transport error.
///
/// Skipped packet numbers occur only in the application data space (where costly transfers might
/// take place) and are distributed exponentially to reflect the reduced likelihood and impact of
/// bad behavior from a peer that has been well-behaved for an extended period.
///
/// ACKs for packet numbers that have not yet been allocated are also a transport error, but an
/// attacker with knowledge of the congestion control algorithm in use could time falsified ACKs to
/// arrive after the packets they reference are sent.
///
/// [optimistic ACK attacks]: https://www.rfc-editor.org/rfc/rfc9000.html#name-optimistic-ack-attack
pub(super) struct PacketNumberFilter {
    /// Next outgoing packet number to skip
    next_skipped_packet_number: u64,
    /// Most recently skipped packet number
    prev_skipped_packet_number: Option<u64>,
    /// Next packet number to skip is randomly selected from 2^n..2^n+1
    exponent: u32,
}

impl PacketNumberFilter {
    pub(super) fn new(rng: &mut (impl CryptoRng + ?Sized)) -> Self {
        // First skipped PN is in 0..64
        let exponent = 6;
        Self {
            next_skipped_packet_number: rng.random_range(0..2u64.saturating_pow(exponent)),
            prev_skipped_packet_number: None,
            exponent,
        }
    }

    #[cfg(test)]
    pub(super) fn disabled() -> Self {
        Self {
            next_skipped_packet_number: u64::MAX,
            prev_skipped_packet_number: None,
            exponent: u32::MAX,
        }
    }

    /// Whether to use the provided packet number (false) or to skip it (true)
    pub(super) fn skip_pn(&mut self, n: u64, rng: &mut (impl CryptoRng + ?Sized)) -> bool {
        if n != self.next_skipped_packet_number {
            return false;
        }

        trace!("skipping pn {n}");
        // Skip this packet number, and choose the next one to skip
        self.prev_skipped_packet_number = Some(self.next_skipped_packet_number);
        let next_exponent = self.exponent.saturating_add(1);
        self.next_skipped_packet_number = rng
            .random_range(2u64.saturating_pow(self.exponent)..2u64.saturating_pow(next_exponent));
        self.exponent = next_exponent;
        true
    }
}

/// Ensures we can always fit all our ACKs in a single minimum-MTU packet with room to spare
const MAX_ACK_BLOCKS: usize = 64;

#[cfg(test)]
mod test {
    use rand::Rng;
    use rand::seq::SliceRandom;

    use crate::token::ResetToken;
    use crate::{ConnectionIdGenerator, RandomConnectionIdGenerator};

    use super::*;

    #[test]
    fn sanity() {
        let mut dedup = Dedup::new();
        assert!(!dedup.insert(0));
        assert_eq!(dedup.next, 1);
        assert_eq!(dedup.window, 0b1);
        assert!(dedup.insert(0));
        assert_eq!(dedup.next, 1);
        assert_eq!(dedup.window, 0b1);
        assert!(!dedup.insert(1));
        assert_eq!(dedup.next, 2);
        assert_eq!(dedup.window, 0b11);
        assert!(!dedup.insert(2));
        assert_eq!(dedup.next, 3);
        assert_eq!(dedup.window, 0b111);
        assert!(!dedup.insert(4));
        assert_eq!(dedup.next, 5);
        assert_eq!(dedup.window, 0b11110);
        assert!(!dedup.insert(7));
        assert_eq!(dedup.next, 8);
        assert_eq!(dedup.window, 0b1111_0100);
        assert!(dedup.insert(4));
        assert!(!dedup.insert(3));
        assert_eq!(dedup.next, 8);
        assert_eq!(dedup.window, 0b1111_1100);
        assert!(!dedup.insert(6));
        assert_eq!(dedup.next, 8);
        assert_eq!(dedup.window, 0b1111_1101);
        assert!(!dedup.insert(5));
        assert_eq!(dedup.next, 8);
        assert_eq!(dedup.window, 0b1111_1111);
    }

    #[test]
    fn happypath() {
        let mut dedup = Dedup::new();
        for i in 0..(2 * WINDOW_SIZE) {
            assert!(!dedup.insert(i));
            for j in 0..=i {
                assert!(dedup.insert(j));
            }
        }
    }

    #[test]
    fn jump() {
        let mut dedup = Dedup::new();
        dedup.insert(2 * WINDOW_SIZE);
        assert!(dedup.insert(WINDOW_SIZE));
        assert_eq!(dedup.next, 2 * WINDOW_SIZE + 1);
        assert_eq!(dedup.window, 0);
        assert!(!dedup.insert(WINDOW_SIZE + 1));
        assert_eq!(dedup.next, 2 * WINDOW_SIZE + 1);
        assert_eq!(dedup.window, 1 << (WINDOW_SIZE - 2));
    }

    #[test]
    fn dedup_has_missing() {
        let mut dedup = Dedup::new();

        dedup.insert(0);
        assert!(!dedup.missing_in_interval(0, 0));

        dedup.insert(1);
        assert!(!dedup.missing_in_interval(0, 1));

        dedup.insert(3);
        assert!(dedup.missing_in_interval(1, 3));

        dedup.insert(4);
        assert!(!dedup.missing_in_interval(3, 4));
        assert!(dedup.missing_in_interval(0, 4));

        dedup.insert(2);
        assert!(!dedup.missing_in_interval(0, 4));
    }

    #[test]
    fn dedup_outside_of_window_has_missing() {
        let mut dedup = Dedup::new();

        for i in 0..140 {
            dedup.insert(i);
        }

        // 0 and 4 are outside of the window
        assert!(!dedup.missing_in_interval(0, 4));
        dedup.insert(160);
        assert!(!dedup.missing_in_interval(0, 4));
        assert!(!dedup.missing_in_interval(0, 140));
        assert!(dedup.missing_in_interval(0, 160));
    }

    #[test]
    fn dedup_smallest_missing() {
        let mut dedup = Dedup::new();

        dedup.insert(0);
        assert_eq!(dedup.smallest_missing_in_interval(0, 0), None);

        dedup.insert(1);
        assert_eq!(dedup.smallest_missing_in_interval(0, 1), None);

        dedup.insert(5);
        dedup.insert(7);
        assert_eq!(dedup.smallest_missing_in_interval(0, 7), Some(2));
        assert_eq!(dedup.smallest_missing_in_interval(5, 7), Some(6));

        dedup.insert(2);
        assert_eq!(dedup.smallest_missing_in_interval(1, 7), Some(3));

        dedup.insert(170);
        dedup.insert(172);
        dedup.insert(300);
        assert_eq!(dedup.smallest_missing_in_interval(170, 172), None);

        dedup.insert(500);
        assert_eq!(dedup.smallest_missing_in_interval(0, 500), Some(372));
        assert_eq!(dedup.smallest_missing_in_interval(0, 373), Some(372));
        assert_eq!(dedup.smallest_missing_in_interval(0, 372), None);
    }

    #[test]
    fn pending_acks_first_packet_is_not_considered_reordered() {
        let mut acks = PendingAcks::new();
        let mut dedup = Dedup::new();
        dedup.insert(0);
        acks.packet_received(Instant::now(), 0, true, &dedup);
        assert!(!acks.immediate_ack_required);
    }

    #[test]
    fn pending_acks_after_immediate_ack_set() {
        let mut acks = PendingAcks::new();
        let mut dedup = Dedup::new();

        // Receive ack-eliciting packet
        dedup.insert(0);
        let now = Instant::now();
        acks.insert_one(0, now);
        acks.packet_received(now, 0, true, &dedup);

        // Sanity check
        assert!(!acks.ranges.is_empty());
        assert!(!acks.can_send());

        // Can send ACK after max_ack_delay exceeded
        acks.set_immediate_ack_required();
        assert!(acks.can_send());
    }

    #[test]
    fn pending_acks_ack_delay() {
        let mut acks = PendingAcks::new();
        let mut dedup = Dedup::new();

        let t1 = Instant::now();
        let t2 = t1 + Duration::from_millis(2);
        let t3 = t2 + Duration::from_millis(5);
        assert_eq!(acks.ack_delay(t1), Duration::from_millis(0));
        assert_eq!(acks.ack_delay(t2), Duration::from_millis(0));
        assert_eq!(acks.ack_delay(t3), Duration::from_millis(0));

        // In-order packet
        dedup.insert(0);
        acks.insert_one(0, t1);
        acks.packet_received(t1, 0, true, &dedup);
        assert_eq!(acks.ack_delay(t1), Duration::from_millis(0));
        assert_eq!(acks.ack_delay(t2), Duration::from_millis(2));
        assert_eq!(acks.ack_delay(t3), Duration::from_millis(7));

        // Out of order (higher than expected)
        dedup.insert(3);
        acks.insert_one(3, t2);
        acks.packet_received(t2, 3, true, &dedup);
        assert_eq!(acks.ack_delay(t2), Duration::from_millis(0));
        assert_eq!(acks.ack_delay(t3), Duration::from_millis(5));

        // Out of order (lower than expected, so previous instant is kept)
        dedup.insert(2);
        acks.insert_one(2, t3);
        acks.packet_received(t3, 2, true, &dedup);
        assert_eq!(acks.ack_delay(t3), Duration::from_millis(5));
    }

    #[test]
    fn sent_packet_size() {
        // The tracking state of sent packets should be minimal, and not grow
        // over time.
        assert!(size_of::<SentPacket>() <= 128);
    }

    #[test]
    fn pending_new_cids() {
        #[cfg(all(feature = "aws-lc-rs", not(feature = "ring")))]
        use aws_lc_rs::hmac;
        #[cfg(feature = "ring")]
        use ring::hmac;

        let mut cid_generator = RandomConnectionIdGenerator::new(8);
        let mut reset_key = [0; 64];
        rand::rng().fill_bytes(&mut reset_key);
        let hmac = hmac::Key::new(hmac::HMAC_SHA256, &reset_key);

        let cid_a = cid_generator.generate_cid();
        let a = IssuedCid {
            path_id: PathId::ZERO,
            sequence: 1,
            id: cid_a,
            reset_token: ResetToken::new(&hmac, cid_a),
        };
        let cid_b = cid_generator.generate_cid();
        let b = IssuedCid {
            path_id: PathId::ZERO,
            sequence: 2,
            id: cid_b,
            reset_token: ResetToken::new(&hmac, cid_b),
        };
        let cid_c = cid_generator.generate_cid();
        let c = IssuedCid {
            path_id: PathId(1),
            sequence: 1,
            id: cid_c,
            reset_token: ResetToken::new(&hmac, cid_c),
        };

        let mut pending_cids = PendingNewCids::default();

        for _ in 0..9 {
            // Push CIDs in a random order
            let mut input = vec![a, b, c];
            input.shuffle(&mut rand::rng());
            for cid in input {
                pending_cids.push(cid);
            }

            // Pop order is always the same
            assert_eq!(pending_cids.pop().map(|i| i.id), Some(a.id));
            assert_eq!(pending_cids.pop().map(|i| i.id), Some(b.id));
            assert_eq!(pending_cids.pop().map(|i| i.id), Some(c.id));
            assert!(pending_cids.pop().is_none());
        }
    }
}