use alloc::{boxed::Box, vec::Vec}; use core::{cmp, net::SocketAddr}; use thiserror::Error; use tracing::{debug, trace}; use super::{ PathStats, SpaceKind, mtud::MtuDiscovery, pacing::Pacer, spaces::{PacketNumberSpace, SentPacket}, }; use crate::{ ConnectionId, Duration, FourTuple, Instant, IntMap, TIMER_GRANULARITY, TransportConfig, TransportErrorCode, VarInt, coding::{self, Decodable, Encodable}, congestion, connection::{MAX_BACKOFF_EXPONENT, MAX_PTO_INTERVAL}, frame::ObservedAddr, }; #[cfg(feature = "qlog")] use qlog::events::quic::RecoveryMetricsUpdated; /// Id representing different paths when using multipath extension #[cfg_attr(test, derive(test_strategy::Arbitrary))] #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Default)] pub struct PathId(pub(crate) u32); impl core::hash::Hash for PathId { fn hash(&self, state: &mut H) { state.write_u32(self.0); } } impl Decodable for PathId { fn decode(r: &mut B) -> coding::Result { let v = VarInt::decode(r)?; let v = u32::try_from(v.0).map_err(|_| coding::UnexpectedEnd)?; Ok(Self(v)) } } impl Encodable for PathId { fn encode(&self, w: &mut B) { VarInt(self.0.into()).encode(w) } } impl PathId { /// The maximum path ID allowed. pub const MAX: Self = Self(u32::MAX); /// The 0 path id. pub const ZERO: Self = Self(0); /// The number of bytes this [`PathId`] uses when encoded as a [`VarInt`] pub(crate) const fn size(&self) -> usize { VarInt(self.0 as u64).size() } /// Saturating integer addition. Computes self + rhs, saturating at the numeric bounds instead /// of overflowing. pub fn saturating_add(self, rhs: impl Into) -> Self { let rhs = rhs.into(); let inner = self.0.saturating_add(rhs.0); Self(inner) } /// Saturating integer subtraction. Computes self - rhs, saturating at the numeric bounds /// instead of overflowing. pub fn saturating_sub(self, rhs: impl Into) -> Self { let rhs = rhs.into(); let inner = self.0.saturating_sub(rhs.0); Self(inner) } /// Get the next [`PathId`] pub(crate) fn next(&self) -> Self { self.saturating_add(Self(1)) } /// Get the underlying u32 pub(crate) fn as_u32(&self) -> u32 { self.0 } } impl core::fmt::Display for PathId { fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { self.0.fmt(f) } } impl> From for PathId { fn from(source: T) -> Self { Self(source.into()) } } /// State needed for a single path ID. /// /// A single path ID can migrate according to the rules in RFC9000 §9, either voluntary or /// involuntary. We need to keep the [`PathData`] of the previously used such path available /// in order to defend against migration attacks (see RFC9000 §9.3.1, §9.3.2 and §9.3.3) as /// well as to support path probing (RFC9000 §9.1). #[derive(Debug)] pub(super) struct PathState { pub(super) data: PathData, pub(super) prev: Option<(ConnectionId, PathData)>, } impl PathState { /// Update counters to account for a packet becoming acknowledged, lost, or abandoned pub(super) fn remove_in_flight(&mut self, packet: &SentPacket) { // Visit known paths from newest to oldest to find the one `pn` was sent on for path_data in [&mut self.data] .into_iter() .chain(self.prev.as_mut().map(|(_, data)| data)) { if path_data.remove_in_flight(packet) { return; } } } } #[derive(Debug)] pub(super) struct SentChallengeInfo { /// When was the challenge sent on the wire. pub(super) sent_instant: Instant, /// The 4-tuple on which this path challenge was sent. pub(super) network_path: FourTuple, } /// State of particular network path 4-tuple within a [`PacketNumberSpace`]. /// /// With QUIC-Multipath a path is identified by a [`PathId`] and it is possible to have /// multiple paths on the same 4-tuple. Furthermore a single QUIC-Multipath path can migrate /// to a different 4-tuple, in a similar manner as an RFC9000 connection can use "path /// migration" to move to a different 4-tuple. There are thus two states we keep for paths: /// /// - [`PacketNumberSpace`]: The state for a single packet number space, i.e. [`PathId`], which /// remains in place across path migrations to different 4-tuples. /// /// This is stored in [`PacketSpace::number_spaces`] indexed on [`PathId`]. /// /// - [`PathData`]: The state we keep for each unique 4-tuple within a space. Of note is that a /// single [`PathData`] can never belong to a different [`PacketNumberSpace`]. /// /// This is stored in [`Connection::paths`] indexed by the current [`PathId`] for which /// space it exists. Either as the primary 4-tuple or as the previous 4-tuple just after a /// migration. /// /// It follows that there might be several [`PathData`] structs for the same 4-tuple if /// several spaces are sharing the same 4-tuple. Note that during the handshake, the /// Initial, Handshake and Data spaces for [`PathId::ZERO`] all share the same [`PathData`]. /// /// [`PacketSpace::number_spaces`]: super::spaces::PacketSpace::number_spaces /// [`Connection::paths`]: super::Connection::paths #[derive(Debug)] pub(super) struct PathData { pub(super) network_path: FourTuple, pub(super) rtt: RttEstimator, /// Whether we're enabling ECN on outgoing packets pub(super) sending_ecn: bool, /// Congestion controller state pub(super) congestion: Box, /// Pacing state pub(super) pacing: Pacer, /// Whether the last `poll_transmit_on_path` call yielded no data because there was /// no outgoing application data. /// /// The RFC writes: /// > When bytes in flight is smaller than the congestion window and sending is not pacing /// > limited, the congestion window is underutilized. This can happen due to insufficient /// > application data or flow control limits. When this occurs, the congestion window SHOULD /// > NOT be increased in either slow start or congestion avoidance. /// /// (RFC9002, section 7.8) /// /// I.e. when app_limited is true, the congestion controller doesn't increase the congestion /// window. pub(super) app_limited: bool, /// Whether to trigger sending another PATH_CHALLENGE in the next poll_transmit. /// /// This is picked up by [`super::Connection::space_can_send`]. These are **not** /// retransmittable, which is why they are not part of the `PathRetransmits`. /// /// Only used for **on-path** challenges, like RFC9000-style path migration and /// multipath path validation (for opening). /// /// This is **not used** for n0 nat traversal challenge sending, which is off-path. pub(super) pending_challenge: bool, /// On-path path challenges sent that we didn't receive a path response for yet. unconfirmed_challenges: IntMap, /// How often we've deemed a path challenge to be lost. /// /// Similar to [`Self::pto_count`], but for on-path path challenges. /// Used to calculate exponential backoff for retrying path challenges. pub(super) lost_challenge_count: u32, /// Whether we're certain the peer can both send and receive on this address /// /// Initially equal to `use_stateless_retry` for servers, and becomes false again on every /// migration. Always true for clients. pub(super) validated: bool, /// Total size of all UDP datagrams sent on this path pub(super) total_sent: u64, /// Total size of all UDP datagrams received on this path pub(super) total_recvd: u64, /// The state of the MTU discovery process pub(super) mtud: MtuDiscovery, /// Packet number of the first packet sent after an RTT sample was collected on this path /// /// Used in persistent congestion determination. pub(super) first_packet_after_rtt_sample: Option<(SpaceKind, u64)>, /// The in-flight packets and bytes /// /// Note that this is across all spaces on this path pub(super) in_flight: InFlight, /// Queue of data that must be sent over this specific [`PathData::generation`] path. pub(super) pending: PathRetransmits, /// Observed address frame with the largest sequence number received from the peer on this /// path. pub(super) last_observed_addr_report: Option, /// The QUIC-MULTIPATH path status pub(super) status: PathStatusState, /// Number of the first packet sent on this path /// /// With RFC9000 §9 style migration (i.e. not multipath) the PathId does not change and /// hence packet numbers continue. This is used to determine whether a packet was sent /// on such an earlier path. Insufficient to determine if a packet was sent on a later /// path. first_packet: Option, /// The number of times a tail-loss probe has been sent without receiving an ack. /// /// This is incremented by one every time the [`LossDetection`] timer fires because a /// tail-loss probe needs to be sent. Once an acknowledgement for a packet is received /// again it is reset to 0. Used to compute the PTO duration. /// /// [`LossDetection`]: super::timer::PathTimer::LossDetection pub(super) pto_count: u32, // // Per-path idle & keep alive /// Idle timeout for the path /// /// If expired, the path will be abandoned. This is different from the connection-wide /// idle timeout which closes the connection if expired. pub(super) idle_timeout: Option, /// Keep alives to send on this path /// /// There is also a connection-level keep alive configured in the /// [`TransportParameters`]. This triggers activity on any path which can keep the /// connection alive. /// /// [`TransportParameters`]: crate::transport_parameters::TransportParameters pub(super) keep_alive: Option, /// Whether to reset the idle timer when the next ack-eliciting packet is sent. /// /// Whenever we receive an authenticated packet the connection and path idle timers are /// reset if a maximum idle timeout was negotiated. However on the first ack-eliciting /// packet *sent* after this the idle timer also needs to be reset to avoid the idle /// timer firing while the sent packet is in-fight. See /// . pub(super) permit_idle_reset: bool, /// Whether we're currently draining the path after having abandoned it. /// /// This should only be true when a path discard timer is armed, and after the path was /// abandoned (and added to the abandoned_paths set). /// /// This will only ever be set from false to true. pub(super) draining: bool, /// Snapshot of the qlog recovery metrics #[cfg(feature = "qlog")] recovery_metrics: RecoveryMetrics, /// Tag uniquely identifying a path in a connection. /// /// When a migration happens on the same [`PathId`] we still detect a change in the /// 4-tuple and generate a new [`PathData`] for it. Each such generation has a unique /// value to keep track of which 4-tuple a packet belonged to. generation: u64, } impl PathData { pub(super) fn new( network_path: FourTuple, allow_mtud: bool, peer_max_udp_payload_size: Option, generation: u64, now: Instant, config: &TransportConfig, ) -> Self { let congestion = config .congestion_controller_factory .clone() .build(now, config.get_initial_mtu()); Self { network_path, rtt: RttEstimator::new(config.initial_rtt), sending_ecn: true, pacing: Pacer::new( config.initial_rtt, congestion.initial_window(), config.get_initial_mtu(), config.max_outgoing_bytes_per_second, now, ), congestion, app_limited: false, unconfirmed_challenges: Default::default(), lost_challenge_count: 0, pending_challenge: false, validated: false, total_sent: 0, total_recvd: 0, mtud: config .mtu_discovery_config .as_ref() .filter(|_| allow_mtud) .map_or_else( || MtuDiscovery::disabled(config.get_initial_mtu(), config.min_mtu), |mtud_config| { MtuDiscovery::new( config.get_initial_mtu(), config.min_mtu, peer_max_udp_payload_size, mtud_config.clone(), ) }, ), first_packet_after_rtt_sample: None, in_flight: InFlight::new(), pending: PathRetransmits::default(), last_observed_addr_report: None, status: Default::default(), first_packet: None, pto_count: 0, idle_timeout: config.default_path_max_idle_timeout, keep_alive: config.default_path_keep_alive_interval, permit_idle_reset: true, draining: false, #[cfg(feature = "qlog")] recovery_metrics: RecoveryMetrics::default(), generation, } } /// Create a new path from a previous one. /// /// This should only be called when migrating paths. pub(super) fn from_previous( network_path: FourTuple, prev: &Self, generation: u64, now: Instant, ) -> Self { let congestion = prev.congestion.clone_box(); let smoothed_rtt = prev.rtt.get(); Self { network_path, rtt: prev.rtt, pacing: Pacer::new( smoothed_rtt, congestion.window(), prev.current_mtu(), prev.pacing.max_bytes_per_second(), now, ), sending_ecn: true, congestion, app_limited: false, unconfirmed_challenges: Default::default(), lost_challenge_count: 0, pending_challenge: false, validated: false, total_sent: 0, total_recvd: 0, mtud: prev.mtud.clone(), first_packet_after_rtt_sample: prev.first_packet_after_rtt_sample, in_flight: InFlight::new(), pending: PathRetransmits::default(), last_observed_addr_report: None, status: prev.status.clone(), first_packet: None, pto_count: 0, idle_timeout: prev.idle_timeout, keep_alive: prev.keep_alive, permit_idle_reset: true, draining: false, #[cfg(feature = "qlog")] recovery_metrics: prev.recovery_metrics.clone(), generation, } } /// Whether we're in the process of validating this path with PATH_CHALLENGEs pub(super) fn is_validating_path(&self) -> bool { !self.unconfirmed_challenges.is_empty() || self.pending_challenge } /// Indicates whether we're a server that hasn't validated the peer's address and hasn't /// received enough data from the peer to permit sending `bytes_to_send` additional bytes pub(super) fn anti_amplification_blocked(&self, bytes_to_send: u64) -> bool { !self.validated && self.total_recvd * 3 < self.total_sent + bytes_to_send } /// Returns the path's current MTU pub(super) fn current_mtu(&self) -> u16 { self.mtud.current_mtu() } /// Account for transmission of `packet` with number `pn` in `space` pub(super) fn sent(&mut self, pn: u64, packet: SentPacket, space: &mut PacketNumberSpace) { self.in_flight.insert(&packet); if self.first_packet.is_none() { self.first_packet = Some(pn); } if let Some(forgotten) = space.sent(pn, packet) { self.remove_in_flight(&forgotten); } } pub(super) fn record_path_challenge_sent( &mut self, now: Instant, token: u64, network_path: FourTuple, ) { let info = SentChallengeInfo { sent_instant: now, network_path, }; debug_assert_eq!(network_path, self.network_path); self.unconfirmed_challenges.insert(token, info); } /// Remove `packet` with number `pn` from this path's congestion control counters, or return /// `false` if `pn` was sent before this path was established. pub(super) fn remove_in_flight(&mut self, packet: &SentPacket) -> bool { if packet.path_generation != self.generation { return false; } self.in_flight.remove(packet); true } /// Increment the total size of sent UDP datagrams pub(super) fn inc_total_sent(&mut self, inc: u64) { self.total_sent = self.total_sent.saturating_add(inc); if !self.validated { trace!( network_path = %self.network_path, anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent), "anti amplification budget decreased" ); } } /// Increment the total size of received UDP datagrams pub(super) fn inc_total_recvd(&mut self, inc: u64) { self.total_recvd = self.total_recvd.saturating_add(inc); if !self.validated { trace!( network_path = %self.network_path, anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent), "anti amplification budget increased" ); } } /// The earliest time at which an on-path challenge we sent is considered lost. pub(super) fn earliest_on_path_expiring_challenge(&self) -> Option { if self.unconfirmed_challenges.is_empty() { return None; } let duration = self.on_path_challenge_pto(); self.unconfirmed_challenges .values() .map(|info| info.sent_instant + duration) .min() } /// The duration after which a PTO expires for an on-path challenge, if sent now. /// /// Since challenges need an on-path response rather than just an ACK that can be sent /// on any path they need a different timer from the /// [`PathTimer::LossDetection`]. Functionally this behaves as the probe timeout /// however. /// /// [`PathTimer::LossDetection`]: super::timer::PathTimer::LossDetection pub(super) fn on_path_challenge_pto(&self) -> Duration { let backoff = 2u32.pow(self.lost_challenge_count.min(MAX_BACKOFF_EXPONENT)); let duration = self.rtt.pto_base() * backoff; duration.min(MAX_PTO_INTERVAL) } /// Handle receiving a PATH_RESPONSE. pub(super) fn on_path_response_received( &mut self, now: Instant, token: u64, ) -> OnPathResponseReceived { // > § 8.2.3 // > Path validation succeeds when a PATH_RESPONSE frame is received that contains the // > data that was sent in a previous PATH_CHALLENGE frame. A PATH_RESPONSE frame // > received on any network path validates the path on which the PATH_CHALLENGE was // > sent. // // At this point we have three potentially different network paths: // - current network path (`Self::network_path`) // - network path used to send the path challenge (`SentChallengeInfo::network_path`) // - network path over which the response arrived (not needed) // // As per above spec quote, this only validates the network path on which this was // *sent*, regardless of the path on which it was received in order to protect // against off-path packet forwarding attacks. match self.unconfirmed_challenges.remove(&token) { // Response to an on-path PathChallenge that validates this path. // The sent path should match the current path. However, it's possible that the // challenge was sent when no local_ip was known. This case is allowed as well. Some(info) if info.network_path.is_probably_same_path(&self.network_path) => { // Do not update or set the self.network_path.local_ip: // Connection::process_payload handles this later when required. We can mark // the path as validated though, because for a change in local_ip only we do // not need to re-validate the path. let sent_instant = info.sent_instant; if !core::mem::replace(&mut self.validated, true) { trace!("new path validated"); } // Clear any other on-path sent challenges and stop sending new ones. self.reset_on_path_challenges(); // This RTT can only be used for the initial RTT, not as a normal // sample: https://www.rfc-editor.org/rfc/rfc9002#section-6.2.2-2. let rtt = now.saturating_duration_since(sent_instant); self.rtt.reset_initial_rtt(rtt); OnPathResponseReceived::OnPath } // Response to an on-path PathChallenge that does not validate this path. Some(info) => { // This is a valid path response, but this validates a 4-tuple we no longer // have in use. Keep only sent challenges for the current path. self.unconfirmed_challenges .retain(|_token, i| i.network_path == self.network_path); // If there are no challenges for the current path, schedule one if !self.unconfirmed_challenges.is_empty() { self.pending_challenge = true; } OnPathResponseReceived::Ignored { sent_on: info.network_path, current_path: self.network_path, } } None => { // Response to an unknown PathChallenge. Does not indicate failure. OnPathResponseReceived::Unknown } } } /// Removes all on-path challenges we remember and cancels sending new on-path challenges. pub(super) fn reset_on_path_challenges(&mut self) { self.unconfirmed_challenges.clear(); self.pending_challenge = false; self.lost_challenge_count = 0; } #[cfg(feature = "qlog")] pub(super) fn qlog_recovery_metrics( &mut self, path_id: PathId, ) -> Option { let controller_metrics = self.congestion.metrics(); let metrics = RecoveryMetrics { min_rtt: Some(self.rtt.min), smoothed_rtt: Some(self.rtt.get()), latest_rtt: Some(self.rtt.latest), rtt_variance: Some(self.rtt.var), pto_count: Some(self.pto_count), bytes_in_flight: Some(self.in_flight.bytes), packets_in_flight: Some(self.in_flight.ack_eliciting), congestion_window: Some(controller_metrics.congestion_window), ssthresh: controller_metrics.ssthresh, pacing_rate: controller_metrics.pacing_rate, }; let event = metrics.to_qlog_event(path_id, &self.recovery_metrics); self.recovery_metrics = metrics; event } /// Return how long we need to wait before sending `bytes_to_send` /// /// See [`Pacer::delay`]. pub(super) fn pacing_delay(&mut self, bytes_to_send: u64, now: Instant) -> Option { let smoothed_rtt = self.rtt.get(); let metrics = self.congestion.metrics(); self.pacing.delay( smoothed_rtt, bytes_to_send, self.current_mtu(), now, &metrics, ) } /// Updates the last observed address report received on this path. /// /// If the address was updated, it's returned to be informed to the application. #[must_use = "updated observed address must be reported to the application"] pub(super) fn update_observed_addr_report( &mut self, observed: ObservedAddr, ) -> Option { match self.last_observed_addr_report.as_mut() { Some(prev) => { if prev.seq_no >= observed.seq_no { // frames that do not increase the sequence number on this path are ignored None } else if prev.ip == observed.ip && prev.port == observed.port { // keep track of the last seq_no but do not report the address as updated prev.seq_no = observed.seq_no; None } else { let addr = observed.socket_addr(); self.last_observed_addr_report = Some(observed); Some(addr) } } None => { let addr = observed.socket_addr(); self.last_observed_addr_report = Some(observed); Some(addr) } } } pub(crate) fn remote_status(&self) -> Option { self.status.remote_status.map(|(_seq, status)| status) } pub(crate) fn local_status(&self) -> PathStatus { self.status.local_status } /// Tag uniquely identifying a path in a connection. /// /// When a migration happens on the same [`PathId`] we still detect a change in the /// 4-tuple and generate a new [`PathData`] for it. Each such generation has a unique /// value to keep track of which 4-tuple a packet belonged to. pub(super) fn generation(&self) -> u64 { self.generation } } pub(super) enum OnPathResponseReceived { /// This response validates the path on its current remote address. OnPath, /// The received token is unknown. Unknown, /// The response is valid but it's not usable for path validation. Ignored { sent_on: FourTuple, current_path: FourTuple, }, } /// Congestion metrics as described in [`recovery_metrics_updated`]. /// /// [`recovery_metrics_updated`]: https://datatracker.ietf.org/doc/html/draft-ietf-quic-qlog-quic-events.html#name-recovery_metrics_updated #[cfg(feature = "qlog")] #[derive(Default, Clone, PartialEq, Debug)] #[non_exhaustive] struct RecoveryMetrics { pub min_rtt: Option, pub smoothed_rtt: Option, pub latest_rtt: Option, pub rtt_variance: Option, pub pto_count: Option, pub bytes_in_flight: Option, pub packets_in_flight: Option, pub congestion_window: Option, pub ssthresh: Option, pub pacing_rate: Option, } #[cfg(feature = "qlog")] impl RecoveryMetrics { /// Retain only values that have been updated since the last snapshot. fn retain_updated(&self, previous: &Self) -> Self { macro_rules! keep_if_changed { ($name:ident) => { if previous.$name == self.$name { None } else { self.$name } }; } Self { min_rtt: keep_if_changed!(min_rtt), smoothed_rtt: keep_if_changed!(smoothed_rtt), latest_rtt: keep_if_changed!(latest_rtt), rtt_variance: keep_if_changed!(rtt_variance), pto_count: keep_if_changed!(pto_count), bytes_in_flight: keep_if_changed!(bytes_in_flight), packets_in_flight: keep_if_changed!(packets_in_flight), congestion_window: keep_if_changed!(congestion_window), ssthresh: keep_if_changed!(ssthresh), pacing_rate: keep_if_changed!(pacing_rate), } } /// Emit a `MetricsUpdated` event containing only updated values fn to_qlog_event(&self, path_id: PathId, previous: &Self) -> Option { let updated = self.retain_updated(previous); if updated == Self::default() { return None; } Some(RecoveryMetricsUpdated { min_rtt: updated.min_rtt.map(|rtt| rtt.as_micros() as f32 / 1000.0), smoothed_rtt: updated .smoothed_rtt .map(|rtt| rtt.as_micros() as f32 / 1000.0), latest_rtt: updated .latest_rtt .map(|rtt| rtt.as_micros() as f32 / 1000.0), rtt_variance: updated .rtt_variance .map(|rtt| rtt.as_micros() as f32 / 1000.0), pto_count: updated .pto_count .map(|count| count.try_into().unwrap_or(u16::MAX)), bytes_in_flight: updated.bytes_in_flight, packets_in_flight: updated.packets_in_flight, congestion_window: updated.congestion_window, ssthresh: updated.ssthresh, pacing_rate: updated.pacing_rate, path_id: Some(path_id.as_u32() as u64), ex_data: Default::default(), }) } } /// RTT estimation for a particular network path #[derive(Copy, Clone, Debug)] pub struct RttEstimator { /// The most recent RTT measurement made when receiving an ack for a previously unacked packet latest: Duration, /// The smoothed RTT of the connection, computed as described in RFC6298 smoothed: Option, /// The RTT variance, computed as described in RFC6298 var: Duration, /// The minimum RTT seen in the connection, ignoring ack delay. min: Duration, } impl RttEstimator { pub(crate) fn new(initial_rtt: Duration) -> Self { Self { latest: initial_rtt, smoothed: None, var: initial_rtt / 2, min: initial_rtt, } } /// Resets the estimator using a new initial_rtt value. /// /// This only resets the initial_rtt **if** no samples have been recorded yet. If there /// are any recorded samples the initial estimate can not be adjusted after the fact. /// /// This is useful when you receive a PATH_RESPONSE in the first packet received on a /// new path. In this case you can use the delay of the PATH_CHALLENGE-PATH_RESPONSE as /// the initial RTT to get a better expected estimation. /// /// A PATH_CHALLENGE-PATH_RESPONSE pair later in the connection should not be used /// explicitly as an estimation since PATH_CHALLENGE is an ACK-eliciting packet itself /// already. pub(crate) fn reset_initial_rtt(&mut self, initial_rtt: Duration) { if self.smoothed.is_none() { self.latest = initial_rtt; self.var = initial_rtt / 2; self.min = initial_rtt; } } /// The current best RTT estimation. pub fn get(&self) -> Duration { self.smoothed.unwrap_or(self.latest) } /// Conservative estimate of RTT /// /// Takes the maximum of smoothed and latest RTT, as recommended /// in 6.1.2 of the recovery spec (draft 29). pub fn conservative(&self) -> Duration { self.get().max(self.latest) } /// Minimum RTT registered so far for this estimator. pub fn min(&self) -> Duration { self.min } /// PTO computed as described in RFC9002#6.2.1. pub(crate) fn pto_base(&self) -> Duration { self.get() + cmp::max(4 * self.var, TIMER_GRANULARITY) } /// Records an RTT sample. pub(crate) fn update(&mut self, ack_delay: Duration, rtt: Duration) { self.latest = rtt; // https://www.rfc-editor.org/rfc/rfc9002.html#section-5.2-3: // min_rtt does not adjust for ack_delay to avoid underestimating. self.min = cmp::min(self.min, self.latest); // Based on RFC6298. if let Some(smoothed) = self.smoothed { let adjusted_rtt = if self.min + ack_delay <= self.latest { self.latest - ack_delay } else { self.latest }; let var_sample = smoothed.abs_diff(adjusted_rtt); self.var = (3 * self.var + var_sample) / 4; self.smoothed = Some((7 * smoothed + adjusted_rtt) / 8); } else { self.smoothed = Some(self.latest); self.var = self.latest / 2; self.min = self.latest; } } } #[derive(Default, Debug)] pub(crate) struct PathResponses { pending: Vec, } impl PathResponses { pub(crate) fn push(&mut self, packet: u64, token: u64, network_path: FourTuple) { /// An arbitrary permissive limit to prevent abuse. /// /// If we've negotiated the n0 NAT Traversal extension, and one user might have a lot /// of addresses, e.g. because of having lots of interfaces (we've seen >25 interfaces /// on Macs with docker and other things), then we need to be able to process at least /// as many PATH_CHALLENGE frames as there are interfaces. /// On top of that, there are retries, which make it possible that we need to process /// even more. /// /// Considering that there can be up to 2 `PathData`s per active `PathId`, and /// reasonable default values for maximum concurrent multipath paths are ~8 and each /// `PathResponse` struct takes up 72 bytes at the moment this, means an attacker can /// cause us to keep `32 * 2 * 8 * 72 = ~37KB` of data around. const MAX_PATH_RESPONSES: usize = 32; let response = PathResponse { packet, token, network_path, }; let existing = self .pending .iter_mut() .find(|x| x.network_path.remote == network_path.remote); if let Some(existing) = existing { // Update a queued response if existing.packet <= packet { *existing = response; } return; } if self.pending.len() < MAX_PATH_RESPONSES { self.pending.push(response); } else { // We don't expect to ever hit this with well-behaved peers, so we don't bother dropping // older challenges. trace!("ignoring excessive PATH_CHALLENGE"); } } pub(crate) fn pop_off_path(&mut self, network_path: FourTuple) -> Option<(u64, FourTuple)> { let response = *self.pending.last()?; // We use an exact comparison here, because once we've received for the first time, // we really should either already have a local_ip, or we will never get one // (because our OS doesn't support it). And even if we get it wrong we are only // slightly less efficient and would not include other on-path data in the packet. if response.network_path == network_path { // We don't bother searching further because we expect that the on-path response will // get drained in the immediate future by a call to `pop_on_path` return None; } self.pending.pop(); Some((response.token, response.network_path)) } pub(crate) fn pop_on_path(&mut self, network_path: FourTuple) -> Option { let response = *self.pending.last()?; // Using an exact comparison. See explanation in `pop_off_path`. if response.network_path != network_path { // We don't bother searching further because we expect that the off-path response will // get drained in the immediate future by a call to `pop_off_path` return None; } self.pending.pop(); Some(response.token) } /// Whether the next [`Self::pop_on_path`] will return something to send. pub(crate) fn has_pending_on_path(&self, network_path: FourTuple) -> bool { self.pending .last() .is_some_and(|response| response.network_path == network_path) } pub(crate) fn is_empty(&self) -> bool { self.pending.is_empty() } } #[derive(Copy, Clone, Debug)] struct PathResponse { /// The packet number the corresponding PATH_CHALLENGE was received in packet: u64, /// The token of the PATH_CHALLENGE token: u64, /// The path the corresponding PATH_CHALLENGE was received from network_path: FourTuple, } /// Summary statistics of packets that have been sent on a particular path, but which have not yet /// been acked or deemed lost #[derive(Debug)] pub(super) struct InFlight { /// Sum of the sizes of all sent packets considered "in flight" by congestion control /// /// The size does not include IP or UDP overhead. Packets only containing ACK frames do not /// count towards this to ensure congestion control does not impede congestion feedback. pub(super) bytes: u64, /// Number of packets in flight containing frames other than ACK and PADDING /// /// This can be 0 even when bytes is not 0 because PADDING frames cause a packet to be /// considered "in flight" by congestion control. However, if this is nonzero, bytes will /// always also be nonzero. pub(super) ack_eliciting: u64, } impl InFlight { fn new() -> Self { Self { bytes: 0, ack_eliciting: 0, } } fn insert(&mut self, packet: &SentPacket) { self.bytes += u64::from(packet.size); self.ack_eliciting += u64::from(packet.ack_eliciting); } /// Update counters to account for a packet becoming acknowledged, lost, or abandoned fn remove(&mut self, packet: &SentPacket) { self.bytes -= u64::from(packet.size); self.ack_eliciting -= u64::from(packet.ack_eliciting); } } /// 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 { if self.local_status == status { return None; } self.local_seq = self.local_seq.saturating_add(1u8); Some(core::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": /// #[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, } /// Application events about paths #[derive(Debug, Clone, PartialEq, Eq)] #[non_exhaustive] pub enum PathEvent { /// A new path has established connection with the peer. #[non_exhaustive] Established { /// The path which can now be used for application data. id: PathId, }, /// A path was abandoned and is no longer usable. /// /// Note that this may be the first event for a path: If a path is abandoned /// before having been established, no [`Self::Established`] event is emitted. /// /// This event will always be followed by [`Self::Discarded`] after some time. #[non_exhaustive] Abandoned { /// The path that was abandoned. id: PathId, /// Reason why this path was abandoned. reason: PathAbandonReason, }, /// A path was discarded and all remaining state for it has been removed. /// /// This event is the last event for a path, and is always emitted after [`Self::Abandoned`]. #[non_exhaustive] Discarded { /// Which path had its state dropped id: PathId, /// The final path stats, they are no longer available via [`Connection::stats`] /// /// [`Connection::stats`]: super::Connection::stats path_stats: Box, }, /// The remote changed the status of the path /// /// The local status is not changed because of this event. It is up to the application /// to update the local status, which is used for packet scheduling, when the remote /// changes the status. #[non_exhaustive] RemoteStatus { /// Path which has changed status id: PathId, /// The new status set by the remote status: PathStatus, }, /// Received an observation of our external address from the peer. #[non_exhaustive] ObservedAddr { /// Path over which the observed address was reported, [`PathId::ZERO`] when multipath is /// not negotiated id: PathId, /// The address observed by the remote over this path addr: SocketAddr, }, } /// Reason for why a path was abandoned. #[derive(Debug, Clone, Eq, PartialEq)] pub enum PathAbandonReason { /// The path was closed locally by the application. ApplicationClosed { /// The error code to be sent with the abandon frame. error_code: VarInt, }, /// We didn't receive a path response in time after opening this path. /// /// This event is no longer emitted, when validation fails a path is only abandoned once /// there's a path timeout and the [`Self::TimedOut`] event will be emitted instead. #[deprecated( since = "1.1.0", note = "This event is no longer emitted, TimedOut will be emitted instead" )] ValidationFailed, /// We didn't receive any data from the remote within the path's idle timeout. TimedOut, /// The path became unusable after a local network change. UnusableAfterNetworkChange, /// The remote closed the path. RemoteAbandoned { /// The error that was sent with the abandon frame. error_code: VarInt, }, } impl PathAbandonReason { /// Whether this abandon was initiated by the remote peer. pub(crate) fn is_remote(&self) -> bool { matches!(self, Self::RemoteAbandoned { .. }) } /// Returns the error code to send with a PATH_ABANDON frame. pub(crate) fn error_code(&self) -> TransportErrorCode { match self { Self::ApplicationClosed { error_code } => (*error_code).into(), #[allow(deprecated)] Self::ValidationFailed | Self::TimedOut | Self::UnusableAfterNetworkChange => { TransportErrorCode::PATH_UNSTABLE_OR_POOR } Self::RemoteAbandoned { error_code } => (*error_code).into(), } } } /// Error from setting path status #[derive(Debug, Error, Clone, PartialEq, Eq)] pub enum SetPathStatusError { /// Error indicating that a path has not been opened or has already been abandoned #[error("closed path")] ClosedPath, /// Error indicating that this operation requires multipath to be negotiated whereas it hasn't /// been #[error("multipath not negotiated")] MultipathNotNegotiated, } /// Error indicating that a path has not been opened or has already been abandoned #[derive(Debug, Default, Error, Clone, PartialEq, Eq)] #[error("closed path")] pub struct ClosedPath { pub(super) _private: (), } /// Retransmittable data specific to a [`PathData::generation`]. #[derive(Debug, Default, Clone)] pub(super) struct PathRetransmits { /// Whether this path needs to report its remote address back to the peer. /// /// This only happens if both peers agree to do so based on their transport parameters. pub(super) observed_address: bool, } impl PathRetransmits { pub(super) fn is_empty(&self) -> bool { let Self { observed_address } = self; !observed_address } } impl core::ops::BitOrAssign for PathRetransmits { fn bitor_assign(&mut self, rhs: Self) { let Self { observed_address } = rhs; self.observed_address |= observed_address; } } #[cfg(test)] mod tests { use super::*; #[test] fn test_path_id_saturating_add() { // add within range behaves normally let large: PathId = u16::MAX.into(); let next = u32::from(u16::MAX) + 1; assert_eq!(large.saturating_add(1u8), PathId::from(next)); // outside range saturates assert_eq!(PathId::MAX.saturating_add(1u8), PathId::MAX) } }