noq_proto/connection/
paths.rs

1use std::{cmp, net::SocketAddr};
2
3use identity_hash::IntMap;
4use thiserror::Error;
5use tracing::trace;
6
7use super::{
8    PathStats, PathStatus, SpaceKind,
9    mtud::MtuDiscovery,
10    pacing::Pacer,
11    spaces::{PacketNumberSpace, SentPacket},
12};
13use crate::{
14    ConnectionId, Duration, FourTuple, Instant, TIMER_GRANULARITY, TransportConfig,
15    TransportErrorCode, VarInt,
16    coding::{self, Decodable, Encodable},
17    congestion,
18    connection::{MAX_BACKOFF_EXPONENT, MAX_PTO_INTERVAL},
19    frame::ObservedAddr,
20};
21
22#[cfg(feature = "qlog")]
23use qlog::events::quic::RecoveryMetricsUpdated;
24
25/// Id representing different paths when using multipath extension
26#[cfg_attr(test, derive(test_strategy::Arbitrary))]
27#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy, Default)]
28pub struct PathId(pub(crate) u32);
29
30impl std::hash::Hash for PathId {
31    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
32        state.write_u32(self.0);
33    }
34}
35
36impl Decodable for PathId {
37    fn decode<B: bytes::Buf>(r: &mut B) -> coding::Result<Self> {
38        let v = VarInt::decode(r)?;
39        let v = u32::try_from(v.0).map_err(|_| coding::UnexpectedEnd)?;
40        Ok(Self(v))
41    }
42}
43
44impl Encodable for PathId {
45    fn encode<B: bytes::BufMut>(&self, w: &mut B) {
46        VarInt(self.0.into()).encode(w)
47    }
48}
49
50impl PathId {
51    /// The maximum path ID allowed.
52    pub const MAX: Self = Self(u32::MAX);
53
54    /// The 0 path id.
55    pub const ZERO: Self = Self(0);
56
57    /// The number of bytes this [`PathId`] uses when encoded as a [`VarInt`]
58    pub(crate) const fn size(&self) -> usize {
59        VarInt(self.0 as u64).size()
60    }
61
62    /// Saturating integer addition. Computes self + rhs, saturating at the numeric bounds instead
63    /// of overflowing.
64    pub fn saturating_add(self, rhs: impl Into<Self>) -> Self {
65        let rhs = rhs.into();
66        let inner = self.0.saturating_add(rhs.0);
67        Self(inner)
68    }
69
70    /// Saturating integer subtraction. Computes self - rhs, saturating at the numeric bounds
71    /// instead of overflowing.
72    pub fn saturating_sub(self, rhs: impl Into<Self>) -> Self {
73        let rhs = rhs.into();
74        let inner = self.0.saturating_sub(rhs.0);
75        Self(inner)
76    }
77
78    /// Get the next [`PathId`]
79    pub(crate) fn next(&self) -> Self {
80        self.saturating_add(Self(1))
81    }
82
83    /// Get the underlying u32
84    pub(crate) fn as_u32(&self) -> u32 {
85        self.0
86    }
87}
88
89impl std::fmt::Display for PathId {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        self.0.fmt(f)
92    }
93}
94
95impl<T: Into<u32>> From<T> for PathId {
96    fn from(source: T) -> Self {
97        Self(source.into())
98    }
99}
100
101/// State needed for a single path ID.
102///
103/// A single path ID can migrate according to the rules in RFC9000 §9, either voluntary or
104/// involuntary. We need to keep the [`PathData`] of the previously used such path available
105/// in order to defend against migration attacks (see RFC9000 §9.3.1, §9.3.2 and §9.3.3) as
106/// well as to support path probing (RFC9000 §9.1).
107#[derive(Debug)]
108pub(super) struct PathState {
109    pub(super) data: PathData,
110    pub(super) prev: Option<(ConnectionId, PathData)>,
111}
112
113impl PathState {
114    /// Update counters to account for a packet becoming acknowledged, lost, or abandoned
115    pub(super) fn remove_in_flight(&mut self, packet: &SentPacket) {
116        // Visit known paths from newest to oldest to find the one `pn` was sent on
117        for path_data in [&mut self.data]
118            .into_iter()
119            .chain(self.prev.as_mut().map(|(_, data)| data))
120        {
121            if path_data.remove_in_flight(packet) {
122                return;
123            }
124        }
125    }
126}
127
128#[derive(Debug)]
129pub(super) struct SentChallengeInfo {
130    /// When was the challenge sent on the wire.
131    pub(super) sent_instant: Instant,
132    /// The 4-tuple on which this path challenge was sent.
133    pub(super) network_path: FourTuple,
134}
135
136/// State of particular network path 4-tuple within a [`PacketNumberSpace`].
137///
138/// With QUIC-Multipath a path is identified by a [`PathId`] and it is possible to have
139/// multiple paths on the same 4-tuple. Furthermore a single QUIC-Multipath path can migrate
140/// to a different 4-tuple, in a similar manner as an RFC9000 connection can use "path
141/// migration" to move to a different 4-tuple. There are thus two states we keep for paths:
142///
143/// - [`PacketNumberSpace`]: The state for a single packet number space, i.e. [`PathId`], which
144///   remains in place across path migrations to different 4-tuples.
145///
146///   This is stored in [`PacketSpace::number_spaces`] indexed on [`PathId`].
147///
148/// - [`PathData`]: The state we keep for each unique 4-tuple within a space. Of note is that a
149///   single [`PathData`] can never belong to a different [`PacketNumberSpace`].
150///
151///   This is stored in [`Connection::paths`] indexed by the current [`PathId`] for which
152///   space it exists. Either as the primary 4-tuple or as the previous 4-tuple just after a
153///   migration.
154///
155/// It follows that there might be several [`PathData`] structs for the same 4-tuple if
156/// several spaces are sharing the same 4-tuple. Note that during the handshake, the
157/// Initial, Handshake and Data spaces for [`PathId::ZERO`] all share the same [`PathData`].
158///
159/// [`PacketSpace::number_spaces`]: super::spaces::PacketSpace::number_spaces
160/// [`Connection::paths`]: super::Connection::paths
161#[derive(Debug)]
162pub(super) struct PathData {
163    pub(super) network_path: FourTuple,
164    pub(super) rtt: RttEstimator,
165    /// Whether we're enabling ECN on outgoing packets
166    pub(super) sending_ecn: bool,
167    /// Congestion controller state
168    pub(super) congestion: Box<dyn congestion::Controller>,
169    /// Pacing state
170    pub(super) pacing: Pacer,
171    /// Whether the last `poll_transmit_on_path` call yielded no data because there was
172    /// no outgoing application data.
173    ///
174    /// The RFC writes:
175    /// > When bytes in flight is smaller than the congestion window and sending is not pacing
176    /// > limited, the congestion window is underutilized. This can happen due to insufficient
177    /// > application data or flow control limits. When this occurs, the congestion window SHOULD
178    /// > NOT be increased in either slow start or congestion avoidance.
179    ///
180    /// (RFC9002, section 7.8)
181    ///
182    /// I.e. when app_limited is true, the congestion controller doesn't increase the congestion
183    /// window.
184    pub(super) app_limited: bool,
185
186    /// Whether to trigger sending another PATH_CHALLENGE in the next poll_transmit.
187    ///
188    /// This is picked up by [`super::Connection::space_can_send`]. These are **not**
189    /// retransmittable, which is why they are not part of the `PathRetransmits`.
190    ///
191    /// Only used for **on-path** challenges, like RFC9000-style path migration and
192    /// multipath path validation (for opening).
193    ///
194    /// This is **not used** for n0 nat traversal challenge sending, which is off-path.
195    pub(super) pending_challenge: bool,
196    /// On-path path challenges sent that we didn't receive a path response for yet.
197    unconfirmed_challenges: IntMap<u64, SentChallengeInfo>,
198    /// How often we've deemed a path challenge to be lost.
199    ///
200    /// Similar to [`Self::pto_count`], but for on-path path challenges.
201    /// Used to calculate exponential backoff for retrying path challenges.
202    pub(super) lost_challenge_count: u32,
203    /// Whether we're certain the peer can both send and receive on this address
204    ///
205    /// Initially equal to `use_stateless_retry` for servers, and becomes false again on every
206    /// migration. Always true for clients.
207    pub(super) validated: bool,
208    /// Total size of all UDP datagrams sent on this path
209    pub(super) total_sent: u64,
210    /// Total size of all UDP datagrams received on this path
211    pub(super) total_recvd: u64,
212    /// The state of the MTU discovery process
213    pub(super) mtud: MtuDiscovery,
214    /// Packet number of the first packet sent after an RTT sample was collected on this path
215    ///
216    /// Used in persistent congestion determination.
217    pub(super) first_packet_after_rtt_sample: Option<(SpaceKind, u64)>,
218    /// The in-flight packets and bytes
219    ///
220    /// Note that this is across all spaces on this path
221    pub(super) in_flight: InFlight,
222    /// Queue of data that must be sent over this specific [`PathData::generation`] path.
223    pub(super) pending: PathRetransmits,
224    /// Observed address frame with the largest sequence number received from the peer on this
225    /// path.
226    pub(super) last_observed_addr_report: Option<ObservedAddr>,
227    /// Number of the first packet sent on this path
228    ///
229    /// With RFC9000 §9 style migration (i.e. not multipath) the PathId does not change and
230    /// hence packet numbers continue. This is used to determine whether a packet was sent
231    /// on such an earlier path. Insufficient to determine if a packet was sent on a later
232    /// path.
233    first_packet: Option<u64>,
234    /// The number of times a tail-loss probe has been sent without receiving an ack.
235    ///
236    /// This is incremented by one every time the [`LossDetection`] timer fires because a
237    /// tail-loss probe needs to be sent. Once an acknowledgement for a packet is received
238    /// again it is reset to 0. Used to compute the PTO duration.
239    ///
240    /// [`LossDetection`]: super::timer::PathTimer::LossDetection
241    pub(super) pto_count: u32,
242
243    //
244    // Per-path idle & keep alive
245    /// Idle timeout for the path
246    ///
247    /// If expired, the path will be abandoned.  This is different from the connection-wide
248    /// idle timeout which closes the connection if expired.
249    pub(super) idle_timeout: Option<Duration>,
250    /// Keep alives to send on this path
251    ///
252    /// There is also a connection-level keep alive configured in the
253    /// [`TransportParameters`].  This triggers activity on any path which can keep the
254    /// connection alive.
255    ///
256    /// [`TransportParameters`]: crate::transport_parameters::TransportParameters
257    pub(super) keep_alive: Option<Duration>,
258    /// Whether to reset the idle timer when the next ack-eliciting packet is sent.
259    ///
260    /// Whenever we receive an authenticated packet the connection and path idle timers are
261    /// reset if a maximum idle timeout was negotiated. However on the first ack-eliciting
262    /// packet *sent* after this the idle timer also needs to be reset to avoid the idle
263    /// timer firing while the sent packet is in-fight. See
264    /// <https://www.rfc-editor.org/rfc/rfc9000.html#section-10.1>.
265    pub(super) permit_idle_reset: bool,
266
267    /// Whether we're currently draining the path after having abandoned it.
268    ///
269    /// This should only be true when a path discard timer is armed, and after the path was
270    /// abandoned (and added to the abandoned_paths set).
271    ///
272    /// This will only ever be set from false to true.
273    pub(super) draining: bool,
274
275    /// Snapshot of the qlog recovery metrics
276    #[cfg(feature = "qlog")]
277    recovery_metrics: RecoveryMetrics,
278
279    /// Tag uniquely identifying a path in a connection.
280    ///
281    /// When a migration happens on the same [`PathId`] we still detect a change in the
282    /// 4-tuple and generate a new [`PathData`] for it. Each such generation has a unique
283    /// value to keep track of which 4-tuple a packet belonged to.
284    generation: u64,
285}
286
287impl PathData {
288    pub(super) fn new(
289        network_path: FourTuple,
290        allow_mtud: bool,
291        peer_max_udp_payload_size: Option<u16>,
292        generation: u64,
293        now: Instant,
294        config: &TransportConfig,
295    ) -> Self {
296        let congestion = config
297            .congestion_controller_factory
298            .clone()
299            .build(now, config.get_initial_mtu());
300        Self {
301            network_path,
302            rtt: RttEstimator::new(config.initial_rtt),
303            sending_ecn: true,
304            pacing: Pacer::new(
305                config.initial_rtt,
306                congestion.initial_window(),
307                config.get_initial_mtu(),
308                config.max_outgoing_bytes_per_second,
309                now,
310            ),
311            congestion,
312            app_limited: false,
313            unconfirmed_challenges: Default::default(),
314            lost_challenge_count: 0,
315            pending_challenge: false,
316            validated: false,
317            total_sent: 0,
318            total_recvd: 0,
319            mtud: config
320                .mtu_discovery_config
321                .as_ref()
322                .filter(|_| allow_mtud)
323                .map_or_else(
324                    || MtuDiscovery::disabled(config.get_initial_mtu(), config.min_mtu),
325                    |mtud_config| {
326                        MtuDiscovery::new(
327                            config.get_initial_mtu(),
328                            config.min_mtu,
329                            peer_max_udp_payload_size,
330                            mtud_config.clone(),
331                        )
332                    },
333                ),
334            first_packet_after_rtt_sample: None,
335            in_flight: InFlight::new(),
336            pending: PathRetransmits::default(),
337            last_observed_addr_report: None,
338            first_packet: None,
339            pto_count: 0,
340            idle_timeout: config.default_path_max_idle_timeout,
341            keep_alive: config.default_path_keep_alive_interval,
342            permit_idle_reset: true,
343            draining: false,
344            #[cfg(feature = "qlog")]
345            recovery_metrics: RecoveryMetrics::default(),
346            generation,
347        }
348    }
349
350    /// Create a new path from a previous one.
351    ///
352    /// This should only be called when migrating paths.
353    pub(super) fn from_previous(
354        network_path: FourTuple,
355        prev: &Self,
356        generation: u64,
357        now: Instant,
358    ) -> Self {
359        let congestion = prev.congestion.clone_box();
360        let smoothed_rtt = prev.rtt.get();
361        Self {
362            network_path,
363            rtt: prev.rtt,
364            pacing: Pacer::new(
365                smoothed_rtt,
366                congestion.window(),
367                prev.current_mtu(),
368                prev.pacing.max_bytes_per_second(),
369                now,
370            ),
371            sending_ecn: true,
372            congestion,
373            app_limited: false,
374            unconfirmed_challenges: Default::default(),
375            lost_challenge_count: 0,
376            pending_challenge: false,
377            validated: false,
378            total_sent: 0,
379            total_recvd: 0,
380            mtud: prev.mtud.clone(),
381            first_packet_after_rtt_sample: prev.first_packet_after_rtt_sample,
382            in_flight: InFlight::new(),
383            pending: PathRetransmits::default(),
384            last_observed_addr_report: None,
385            first_packet: None,
386            pto_count: 0,
387            idle_timeout: prev.idle_timeout,
388            keep_alive: prev.keep_alive,
389            permit_idle_reset: true,
390            draining: false,
391            #[cfg(feature = "qlog")]
392            recovery_metrics: prev.recovery_metrics.clone(),
393            generation,
394        }
395    }
396
397    /// Whether we're in the process of validating this path with PATH_CHALLENGEs
398    pub(super) fn is_validating_path(&self) -> bool {
399        !self.unconfirmed_challenges.is_empty() || self.pending_challenge
400    }
401
402    /// Indicates whether we're a server that hasn't validated the peer's address and hasn't
403    /// received enough data from the peer to permit sending `bytes_to_send` additional bytes
404    pub(super) fn anti_amplification_blocked(&self, bytes_to_send: u64) -> bool {
405        !self.validated && self.total_recvd * 3 < self.total_sent + bytes_to_send
406    }
407
408    /// Returns the path's current MTU
409    pub(super) fn current_mtu(&self) -> u16 {
410        self.mtud.current_mtu()
411    }
412
413    /// Account for transmission of `packet` with number `pn` in `space`
414    pub(super) fn sent(&mut self, pn: u64, packet: SentPacket, space: &mut PacketNumberSpace) {
415        self.in_flight.insert(&packet);
416        if self.first_packet.is_none() {
417            self.first_packet = Some(pn);
418        }
419        if let Some(forgotten) = space.sent(pn, packet) {
420            self.remove_in_flight(&forgotten);
421        }
422    }
423
424    pub(super) fn record_path_challenge_sent(
425        &mut self,
426        now: Instant,
427        token: u64,
428        network_path: FourTuple,
429    ) {
430        let info = SentChallengeInfo {
431            sent_instant: now,
432            network_path,
433        };
434        debug_assert_eq!(network_path, self.network_path);
435        self.unconfirmed_challenges.insert(token, info);
436    }
437
438    /// Remove `packet` with number `pn` from this path's congestion control counters, or return
439    /// `false` if `pn` was sent before this path was established.
440    pub(super) fn remove_in_flight(&mut self, packet: &SentPacket) -> bool {
441        if packet.path_generation != self.generation {
442            return false;
443        }
444        self.in_flight.remove(packet);
445        true
446    }
447
448    /// Increment the total size of sent UDP datagrams
449    pub(super) fn inc_total_sent(&mut self, inc: u64) {
450        self.total_sent = self.total_sent.saturating_add(inc);
451        if !self.validated {
452            trace!(
453                network_path = %self.network_path,
454                anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent),
455                "anti amplification budget decreased"
456            );
457        }
458    }
459
460    /// Increment the total size of received UDP datagrams
461    pub(super) fn inc_total_recvd(&mut self, inc: u64) {
462        self.total_recvd = self.total_recvd.saturating_add(inc);
463        if !self.validated {
464            trace!(
465                network_path = %self.network_path,
466                anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent),
467                "anti amplification budget increased"
468            );
469        }
470    }
471
472    /// The earliest time at which an on-path challenge we sent is considered lost.
473    pub(super) fn earliest_on_path_expiring_challenge(&self) -> Option<Instant> {
474        if self.unconfirmed_challenges.is_empty() {
475            return None;
476        }
477        let duration = self.on_path_challenge_pto();
478        self.unconfirmed_challenges
479            .values()
480            .map(|info| info.sent_instant + duration)
481            .min()
482    }
483
484    /// The duration after which a PTO expires for an on-path challenge, if sent now.
485    ///
486    /// Since challenges need an on-path response rather than just an ACK that can be sent
487    /// on any path they need a different timer from the
488    /// [`PathTimer::LossDetection`]. Functionally this behaves as the probe timeout
489    /// however.
490    ///
491    /// [`PathTimer::LossDetection`]: super::timer::PathTimer::LossDetection
492    pub(super) fn on_path_challenge_pto(&self) -> Duration {
493        let backoff = 2u32.pow(self.lost_challenge_count.min(MAX_BACKOFF_EXPONENT));
494        let duration = self.rtt.pto_base() * backoff;
495        duration.min(MAX_PTO_INTERVAL)
496    }
497
498    /// Handle receiving a PATH_RESPONSE.
499    pub(super) fn on_path_response_received(
500        &mut self,
501        now: Instant,
502        token: u64,
503    ) -> OnPathResponseReceived {
504        // > § 8.2.3
505        // > Path validation succeeds when a PATH_RESPONSE frame is received that contains the
506        // > data that was sent in a previous PATH_CHALLENGE frame. A PATH_RESPONSE frame
507        // > received on any network path validates the path on which the PATH_CHALLENGE was
508        // > sent.
509        //
510        // At this point we have three potentially different network paths:
511        // - current network path (`Self::network_path`)
512        // - network path used to send the path challenge (`SentChallengeInfo::network_path`)
513        // - network path over which the response arrived (not needed)
514        //
515        // As per above spec quote, this only validates the network path on which this was
516        // *sent*, regardless of the path on which it was received in order to protect
517        // against off-path packet forwarding attacks.
518        match self.unconfirmed_challenges.remove(&token) {
519            // Response to an on-path PathChallenge that validates this path.
520            // The sent path should match the current path. However, it's possible that the
521            // challenge was sent when no local_ip was known. This case is allowed as well.
522            Some(info) if info.network_path.is_probably_same_path(&self.network_path) => {
523                // Do not update or set the self.network_path.local_ip:
524                // Connection::process_payload handles this later when required. We can mark
525                // the path as validated though, because for a change in local_ip only we do
526                // not need to re-validate the path.
527                let sent_instant = info.sent_instant;
528                if !std::mem::replace(&mut self.validated, true) {
529                    trace!("new path validated");
530                }
531                // Clear any other on-path sent challenges and stop sending new ones.
532                self.reset_on_path_challenges();
533
534                // This RTT can only be used for the initial RTT, not as a normal
535                // sample: https://www.rfc-editor.org/rfc/rfc9002#section-6.2.2-2.
536                let rtt = now.saturating_duration_since(sent_instant);
537                self.rtt.reset_initial_rtt(rtt);
538
539                OnPathResponseReceived::OnPath
540            }
541            // Response to an on-path PathChallenge that does not validate this path.
542            Some(info) => {
543                // This is a valid path response, but this validates a 4-tuple we no longer
544                // have in use. Keep only sent challenges for the current path.
545                self.unconfirmed_challenges
546                    .retain(|_token, i| i.network_path == self.network_path);
547
548                // If there are no challenges for the current path, schedule one
549                if !self.unconfirmed_challenges.is_empty() {
550                    self.pending_challenge = true;
551                }
552                OnPathResponseReceived::Ignored {
553                    sent_on: info.network_path,
554                    current_path: self.network_path,
555                }
556            }
557            None => {
558                // Response to an unknown PathChallenge. Does not indicate failure.
559                OnPathResponseReceived::Unknown
560            }
561        }
562    }
563
564    /// Removes all on-path challenges we remember and cancels sending new on-path challenges.
565    pub(super) fn reset_on_path_challenges(&mut self) {
566        self.unconfirmed_challenges.clear();
567        self.pending_challenge = false;
568        self.lost_challenge_count = 0;
569    }
570
571    #[cfg(feature = "qlog")]
572    pub(super) fn qlog_recovery_metrics(
573        &mut self,
574        path_id: PathId,
575    ) -> Option<RecoveryMetricsUpdated> {
576        let controller_metrics = self.congestion.metrics();
577
578        let metrics = RecoveryMetrics {
579            min_rtt: Some(self.rtt.min),
580            smoothed_rtt: Some(self.rtt.get()),
581            latest_rtt: Some(self.rtt.latest),
582            rtt_variance: Some(self.rtt.var),
583            pto_count: Some(self.pto_count),
584            bytes_in_flight: Some(self.in_flight.bytes),
585            packets_in_flight: Some(self.in_flight.ack_eliciting),
586
587            congestion_window: Some(controller_metrics.congestion_window),
588            ssthresh: controller_metrics.ssthresh,
589            pacing_rate: controller_metrics.pacing_rate,
590        };
591
592        let event = metrics.to_qlog_event(path_id, &self.recovery_metrics);
593        self.recovery_metrics = metrics;
594        event
595    }
596
597    /// Return how long we need to wait before sending `bytes_to_send`
598    ///
599    /// See [`Pacer::delay`].
600    pub(super) fn pacing_delay(&mut self, bytes_to_send: u64, now: Instant) -> Option<Duration> {
601        let smoothed_rtt = self.rtt.get();
602        let metrics = self.congestion.metrics();
603        self.pacing.delay(
604            smoothed_rtt,
605            bytes_to_send,
606            self.current_mtu(),
607            now,
608            &metrics,
609        )
610    }
611
612    /// Updates the last observed address report received on this path.
613    ///
614    /// If the address was updated, it's returned to be informed to the application.
615    #[must_use = "updated observed address must be reported to the application"]
616    pub(super) fn update_observed_addr_report(
617        &mut self,
618        observed: ObservedAddr,
619    ) -> Option<SocketAddr> {
620        match self.last_observed_addr_report.as_mut() {
621            Some(prev) => {
622                if prev.seq_no >= observed.seq_no {
623                    // frames that do not increase the sequence number on this path are ignored
624                    None
625                } else if prev.ip == observed.ip && prev.port == observed.port {
626                    // keep track of the last seq_no but do not report the address as updated
627                    prev.seq_no = observed.seq_no;
628                    None
629                } else {
630                    let addr = observed.socket_addr();
631                    self.last_observed_addr_report = Some(observed);
632                    Some(addr)
633                }
634            }
635            None => {
636                let addr = observed.socket_addr();
637                self.last_observed_addr_report = Some(observed);
638                Some(addr)
639            }
640        }
641    }
642
643    /// Tag uniquely identifying a path in a connection.
644    ///
645    /// When a migration happens on the same [`PathId`] we still detect a change in the
646    /// 4-tuple and generate a new [`PathData`] for it. Each such generation has a unique
647    /// value to keep track of which 4-tuple a packet belonged to.
648    pub(super) fn generation(&self) -> u64 {
649        self.generation
650    }
651}
652
653pub(super) enum OnPathResponseReceived {
654    /// This response validates the path on its current remote address.
655    OnPath,
656    /// The received token is unknown.
657    Unknown,
658    /// The response is valid but it's not usable for path validation.
659    Ignored {
660        sent_on: FourTuple,
661        current_path: FourTuple,
662    },
663}
664
665/// Congestion metrics as described in [`recovery_metrics_updated`].
666///
667/// [`recovery_metrics_updated`]: https://datatracker.ietf.org/doc/html/draft-ietf-quic-qlog-quic-events.html#name-recovery_metrics_updated
668#[cfg(feature = "qlog")]
669#[derive(Default, Clone, PartialEq, Debug)]
670#[non_exhaustive]
671struct RecoveryMetrics {
672    pub min_rtt: Option<Duration>,
673    pub smoothed_rtt: Option<Duration>,
674    pub latest_rtt: Option<Duration>,
675    pub rtt_variance: Option<Duration>,
676    pub pto_count: Option<u32>,
677    pub bytes_in_flight: Option<u64>,
678    pub packets_in_flight: Option<u64>,
679    pub congestion_window: Option<u64>,
680    pub ssthresh: Option<u64>,
681    pub pacing_rate: Option<u64>,
682}
683
684#[cfg(feature = "qlog")]
685impl RecoveryMetrics {
686    /// Retain only values that have been updated since the last snapshot.
687    fn retain_updated(&self, previous: &Self) -> Self {
688        macro_rules! keep_if_changed {
689            ($name:ident) => {
690                if previous.$name == self.$name {
691                    None
692                } else {
693                    self.$name
694                }
695            };
696        }
697
698        Self {
699            min_rtt: keep_if_changed!(min_rtt),
700            smoothed_rtt: keep_if_changed!(smoothed_rtt),
701            latest_rtt: keep_if_changed!(latest_rtt),
702            rtt_variance: keep_if_changed!(rtt_variance),
703            pto_count: keep_if_changed!(pto_count),
704            bytes_in_flight: keep_if_changed!(bytes_in_flight),
705            packets_in_flight: keep_if_changed!(packets_in_flight),
706            congestion_window: keep_if_changed!(congestion_window),
707            ssthresh: keep_if_changed!(ssthresh),
708            pacing_rate: keep_if_changed!(pacing_rate),
709        }
710    }
711
712    /// Emit a `MetricsUpdated` event containing only updated values
713    fn to_qlog_event(&self, path_id: PathId, previous: &Self) -> Option<RecoveryMetricsUpdated> {
714        let updated = self.retain_updated(previous);
715
716        if updated == Self::default() {
717            return None;
718        }
719
720        Some(RecoveryMetricsUpdated {
721            min_rtt: updated.min_rtt.map(|rtt| rtt.as_micros() as f32 / 1000.0),
722            smoothed_rtt: updated
723                .smoothed_rtt
724                .map(|rtt| rtt.as_micros() as f32 / 1000.0),
725            latest_rtt: updated
726                .latest_rtt
727                .map(|rtt| rtt.as_micros() as f32 / 1000.0),
728            rtt_variance: updated
729                .rtt_variance
730                .map(|rtt| rtt.as_micros() as f32 / 1000.0),
731            pto_count: updated
732                .pto_count
733                .map(|count| count.try_into().unwrap_or(u16::MAX)),
734            bytes_in_flight: updated.bytes_in_flight,
735            packets_in_flight: updated.packets_in_flight,
736            congestion_window: updated.congestion_window,
737            ssthresh: updated.ssthresh,
738            pacing_rate: updated.pacing_rate,
739            path_id: Some(path_id.as_u32() as u64),
740            ex_data: Default::default(),
741        })
742    }
743}
744
745/// RTT estimation for a particular network path
746#[derive(Copy, Clone, Debug)]
747pub struct RttEstimator {
748    /// The most recent RTT measurement made when receiving an ack for a previously unacked packet
749    latest: Duration,
750    /// The smoothed RTT of the connection, computed as described in RFC6298
751    smoothed: Option<Duration>,
752    /// The RTT variance, computed as described in RFC6298
753    var: Duration,
754    /// The minimum RTT seen in the connection, ignoring ack delay.
755    min: Duration,
756}
757
758impl RttEstimator {
759    pub(crate) fn new(initial_rtt: Duration) -> Self {
760        Self {
761            latest: initial_rtt,
762            smoothed: None,
763            var: initial_rtt / 2,
764            min: initial_rtt,
765        }
766    }
767
768    /// Resets the estimator using a new initial_rtt value.
769    ///
770    /// This only resets the initial_rtt **if** no samples have been recorded yet. If there
771    /// are any recorded samples the initial estimate can not be adjusted after the fact.
772    ///
773    /// This is useful when you receive a PATH_RESPONSE in the first packet received on a
774    /// new path. In this case you can use the delay of the PATH_CHALLENGE-PATH_RESPONSE as
775    /// the initial RTT to get a better expected estimation.
776    ///
777    /// A PATH_CHALLENGE-PATH_RESPONSE pair later in the connection should not be used
778    /// explicitly as an estimation since PATH_CHALLENGE is an ACK-eliciting packet itself
779    /// already.
780    pub(crate) fn reset_initial_rtt(&mut self, initial_rtt: Duration) {
781        if self.smoothed.is_none() {
782            self.latest = initial_rtt;
783            self.var = initial_rtt / 2;
784            self.min = initial_rtt;
785        }
786    }
787
788    /// The current best RTT estimation.
789    pub fn get(&self) -> Duration {
790        self.smoothed.unwrap_or(self.latest)
791    }
792
793    /// Conservative estimate of RTT
794    ///
795    /// Takes the maximum of smoothed and latest RTT, as recommended
796    /// in 6.1.2 of the recovery spec (draft 29).
797    pub fn conservative(&self) -> Duration {
798        self.get().max(self.latest)
799    }
800
801    /// Minimum RTT registered so far for this estimator.
802    pub fn min(&self) -> Duration {
803        self.min
804    }
805
806    /// PTO computed as described in RFC9002#6.2.1.
807    pub(crate) fn pto_base(&self) -> Duration {
808        self.get() + cmp::max(4 * self.var, TIMER_GRANULARITY)
809    }
810
811    /// Records an RTT sample.
812    pub(crate) fn update(&mut self, ack_delay: Duration, rtt: Duration) {
813        self.latest = rtt;
814        // https://www.rfc-editor.org/rfc/rfc9002.html#section-5.2-3:
815        // min_rtt does not adjust for ack_delay to avoid underestimating.
816        self.min = cmp::min(self.min, self.latest);
817        // Based on RFC6298.
818        if let Some(smoothed) = self.smoothed {
819            let adjusted_rtt = if self.min + ack_delay <= self.latest {
820                self.latest - ack_delay
821            } else {
822                self.latest
823            };
824            let var_sample = smoothed.abs_diff(adjusted_rtt);
825            self.var = (3 * self.var + var_sample) / 4;
826            self.smoothed = Some((7 * smoothed + adjusted_rtt) / 8);
827        } else {
828            self.smoothed = Some(self.latest);
829            self.var = self.latest / 2;
830            self.min = self.latest;
831        }
832    }
833}
834
835#[derive(Default, Debug)]
836pub(crate) struct PathResponses {
837    pending: Vec<PathResponse>,
838}
839
840impl PathResponses {
841    pub(crate) fn push(&mut self, packet: u64, token: u64, network_path: FourTuple) {
842        /// An arbitrary permissive limit to prevent abuse.
843        ///
844        /// If we've negotiated the n0 NAT Traversal extension, and one user might have a lot
845        /// of addresses, e.g. because of having lots of interfaces (we've seen >25 interfaces
846        /// on Macs with docker and other things), then we need to be able to process at least
847        /// as many PATH_CHALLENGE frames as there are interfaces.
848        /// On top of that, there are retries, which make it possible that we need to process
849        /// even more.
850        ///
851        /// Considering that there can be up to 2 `PathData`s per active `PathId`, and
852        /// reasonable default values for maximum concurrent multipath paths are ~8 and each
853        /// `PathResponse` struct takes up 72 bytes at the moment this, means an attacker can
854        /// cause us to keep `32 * 2 * 8 * 72 = ~37KB` of data around.
855        const MAX_PATH_RESPONSES: usize = 32;
856        let response = PathResponse {
857            packet,
858            token,
859            network_path,
860        };
861        let existing = self
862            .pending
863            .iter_mut()
864            .find(|x| x.network_path.remote == network_path.remote);
865        if let Some(existing) = existing {
866            // Update a queued response
867            if existing.packet <= packet {
868                *existing = response;
869            }
870            return;
871        }
872        if self.pending.len() < MAX_PATH_RESPONSES {
873            self.pending.push(response);
874        } else {
875            // We don't expect to ever hit this with well-behaved peers, so we don't bother dropping
876            // older challenges.
877            trace!("ignoring excessive PATH_CHALLENGE");
878        }
879    }
880
881    pub(crate) fn pop_off_path(&mut self, network_path: FourTuple) -> Option<(u64, FourTuple)> {
882        let response = *self.pending.last()?;
883        // We use an exact comparison here, because once we've received for the first time,
884        // we really should either already have a local_ip, or we will never get one
885        // (because our OS doesn't support it). And even if we get it wrong we are only
886        // slightly less efficient and would not include other on-path data in the packet.
887        if response.network_path == network_path {
888            // We don't bother searching further because we expect that the on-path response will
889            // get drained in the immediate future by a call to `pop_on_path`
890            return None;
891        }
892        self.pending.pop();
893        Some((response.token, response.network_path))
894    }
895
896    pub(crate) fn pop_on_path(&mut self, network_path: FourTuple) -> Option<u64> {
897        let response = *self.pending.last()?;
898        // Using an exact comparison. See explanation in `pop_off_path`.
899        if response.network_path != network_path {
900            // We don't bother searching further because we expect that the off-path response will
901            // get drained in the immediate future by a call to `pop_off_path`
902            return None;
903        }
904        self.pending.pop();
905        Some(response.token)
906    }
907
908    /// Whether the next [`Self::pop_on_path`] will return something to send.
909    pub(crate) fn has_pending_on_path(&self, network_path: FourTuple) -> bool {
910        self.pending
911            .last()
912            .is_some_and(|response| response.network_path == network_path)
913    }
914
915    pub(crate) fn is_empty(&self) -> bool {
916        self.pending.is_empty()
917    }
918}
919
920#[derive(Copy, Clone, Debug)]
921struct PathResponse {
922    /// The packet number the corresponding PATH_CHALLENGE was received in
923    packet: u64,
924    /// The token of the PATH_CHALLENGE
925    token: u64,
926    /// The path the corresponding PATH_CHALLENGE was received from
927    network_path: FourTuple,
928}
929
930/// Summary statistics of packets that have been sent on a particular path, but which have not yet
931/// been acked or deemed lost
932#[derive(Debug)]
933pub(super) struct InFlight {
934    /// Sum of the sizes of all sent packets considered "in flight" by congestion control
935    ///
936    /// The size does not include IP or UDP overhead. Packets only containing ACK frames do not
937    /// count towards this to ensure congestion control does not impede congestion feedback.
938    pub(super) bytes: u64,
939    /// Number of packets in flight containing frames other than ACK and PADDING
940    ///
941    /// This can be 0 even when bytes is not 0 because PADDING frames cause a packet to be
942    /// considered "in flight" by congestion control. However, if this is nonzero, bytes will
943    /// always also be nonzero.
944    pub(super) ack_eliciting: u64,
945}
946
947impl InFlight {
948    fn new() -> Self {
949        Self {
950            bytes: 0,
951            ack_eliciting: 0,
952        }
953    }
954
955    fn insert(&mut self, packet: &SentPacket) {
956        self.bytes += u64::from(packet.size);
957        self.ack_eliciting += u64::from(packet.ack_eliciting);
958    }
959
960    /// Update counters to account for a packet becoming acknowledged, lost, or abandoned
961    fn remove(&mut self, packet: &SentPacket) {
962        self.bytes -= u64::from(packet.size);
963        self.ack_eliciting -= u64::from(packet.ack_eliciting);
964    }
965}
966
967/// Application events about paths
968#[derive(Debug, Clone, PartialEq, Eq)]
969#[non_exhaustive]
970pub enum PathEvent {
971    /// A new path has established connection with the peer.
972    #[non_exhaustive]
973    Established {
974        /// The path which can now be used for application data.
975        id: PathId,
976    },
977    /// A path was abandoned and is no longer usable.
978    ///
979    /// Note that this may be the first event for a path: If a path is abandoned
980    /// before having been established, no [`Self::Established`] event is emitted.
981    ///
982    /// This event will always be followed by [`Self::Discarded`] after some time.
983    #[non_exhaustive]
984    Abandoned {
985        /// The path that was abandoned.
986        id: PathId,
987        /// Reason why this path was abandoned.
988        reason: PathAbandonReason,
989    },
990    /// A path was discarded and all remaining state for it has been removed.
991    ///
992    /// This event is the last event for a path, and is always emitted after [`Self::Abandoned`].
993    #[non_exhaustive]
994    Discarded {
995        /// Which path had its state dropped
996        id: PathId,
997        /// The final path stats, they are no longer available via [`Connection::stats`]
998        ///
999        /// [`Connection::stats`]: super::Connection::stats
1000        path_stats: Box<PathStats>,
1001    },
1002    /// The remote changed the status of the path
1003    ///
1004    /// The local status is not changed because of this event. It is up to the application
1005    /// to update the local status, which is used for packet scheduling, when the remote
1006    /// changes the status.
1007    #[non_exhaustive]
1008    RemoteStatus {
1009        /// Path which has changed status
1010        id: PathId,
1011        /// The new status set by the remote
1012        status: PathStatus,
1013    },
1014    /// Received an observation of our external address from the peer.
1015    #[non_exhaustive]
1016    ObservedAddr {
1017        /// Path over which the observed address was reported, [`PathId::ZERO`] when multipath is
1018        /// not negotiated
1019        id: PathId,
1020        /// The address observed by the remote over this path
1021        addr: SocketAddr,
1022    },
1023}
1024
1025/// Reason for why a path was abandoned.
1026#[derive(Debug, Clone, Eq, PartialEq)]
1027pub enum PathAbandonReason {
1028    /// The path was closed locally by the application.
1029    ApplicationClosed {
1030        /// The error code to be sent with the abandon frame.
1031        error_code: VarInt,
1032    },
1033    /// We didn't receive a path response in time after opening this path.
1034    ///
1035    /// This event is no longer emitted, when validation fails a path is only abandoned once
1036    /// there's a path timeout and the [`Self::TimedOut`] event will be emitted instead.
1037    #[deprecated(
1038        since = "1.1.0",
1039        note = "This event is no longer emitted, TimedOut will be emitted instead"
1040    )]
1041    ValidationFailed,
1042    /// We didn't receive any data from the remote within the path's idle timeout.
1043    TimedOut,
1044    /// The path became unusable after a local network change.
1045    UnusableAfterNetworkChange,
1046    /// The remote closed the path.
1047    RemoteAbandoned {
1048        /// The error that was sent with the abandon frame.
1049        error_code: VarInt,
1050    },
1051}
1052
1053impl PathAbandonReason {
1054    /// Whether this abandon was initiated by the remote peer.
1055    pub(crate) fn is_remote(&self) -> bool {
1056        matches!(self, Self::RemoteAbandoned { .. })
1057    }
1058
1059    /// Returns the error code to send with a PATH_ABANDON frame.
1060    pub(crate) fn error_code(&self) -> TransportErrorCode {
1061        match self {
1062            Self::ApplicationClosed { error_code } => (*error_code).into(),
1063            #[allow(deprecated)]
1064            Self::ValidationFailed | Self::TimedOut | Self::UnusableAfterNetworkChange => {
1065                TransportErrorCode::PATH_UNSTABLE_OR_POOR
1066            }
1067            Self::RemoteAbandoned { error_code } => (*error_code).into(),
1068        }
1069    }
1070}
1071
1072/// Error from setting path status
1073#[derive(Debug, Error, Clone, PartialEq, Eq)]
1074pub enum SetPathStatusError {
1075    /// Error indicating that a path has not been opened or has already been abandoned
1076    #[error("closed path")]
1077    ClosedPath,
1078    /// Error indicating that this operation requires multipath to be negotiated whereas it hasn't
1079    /// been
1080    #[error("multipath not negotiated")]
1081    MultipathNotNegotiated,
1082}
1083
1084/// Error indicating that a path has not been opened or has already been abandoned
1085#[derive(Debug, Default, Error, Clone, PartialEq, Eq)]
1086#[error("closed path")]
1087pub struct ClosedPath {
1088    pub(super) _private: (),
1089}
1090
1091/// Retransmittable data specific to a [`PathData::generation`].
1092#[derive(Debug, Default, Clone)]
1093pub(super) struct PathRetransmits {
1094    /// Whether this path needs to report its remote address back to the peer.
1095    ///
1096    /// This only happens if both peers agree to do so based on their transport parameters.
1097    pub(super) observed_address: bool,
1098}
1099
1100impl PathRetransmits {
1101    pub(super) fn is_empty(&self) -> bool {
1102        let Self { observed_address } = self;
1103        !observed_address
1104    }
1105}
1106
1107impl std::ops::BitOrAssign for PathRetransmits {
1108    fn bitor_assign(&mut self, rhs: Self) {
1109        let Self { observed_address } = rhs;
1110        self.observed_address |= observed_address;
1111    }
1112}
1113
1114#[cfg(test)]
1115mod tests {
1116    use super::*;
1117
1118    #[test]
1119    fn test_path_id_saturating_add() {
1120        // add within range behaves normally
1121        let large: PathId = u16::MAX.into();
1122        let next = u32::from(u16::MAX) + 1;
1123        assert_eq!(large.saturating_add(1u8), PathId::from(next));
1124
1125        // outside range saturates
1126        assert_eq!(PathId::MAX.saturating_add(1u8), PathId::MAX)
1127    }
1128}