refactor: Track 4-tuples instead of only the remote per path.

Author: matheus23

quinn-proto/src/connection/mod.rs

@@ -11,7 +11,7 @@ use super::{
     spaces::{PacketNumberSpace, SentPacket},
 };
 use crate::{
-    ConnectionId, Duration, Instant, TIMER_GRANULARITY, TransportConfig, VarInt, coding,
+    ConnectionId, Duration, FourTuple, Instant, TIMER_GRANULARITY, TransportConfig, VarInt, coding,
     congestion, frame::ObservedAddr, packet::SpaceId,
 };
 
@@ -124,14 +124,14 @@ impl PathState {
 pub(super) struct SentChallengeInfo {
     /// When was the challenge sent on the wire.
     pub(super) sent_instant: Instant,
-    /// The remote to which this path challenge was sent.
-    pub(super) remote: SocketAddr,
+    /// The 4-tuple on which this path challenge was sent.
+    pub(super) addresses: FourTuple,
 }
 
 /// Description of a particular network path
 #[derive(Debug)]
 pub(super) struct PathData {
-    pub(super) remote: SocketAddr,
+    pub(super) addresses: FourTuple,
     pub(super) rtt: RttEstimator,
     /// Whether we're enabling ECN on outgoing packets
     pub(super) sending_ecn: bool,
@@ -223,7 +223,7 @@ pub(super) struct PathData {
 
 impl PathData {
     pub(super) fn new(
-        remote: SocketAddr,
+        addresses: FourTuple,
         allow_mtud: bool,
         peer_max_udp_payload_size: Option<u16>,
         generation: u64,
@@ -235,7 +235,7 @@ impl PathData {
             .clone()
             .build(now, config.get_initial_mtu());
         Self {
-            remote,
+            addresses,
             rtt: RttEstimator::new(config.initial_rtt),
             sending_ecn: true,
             pacing: Pacer::new(
@@ -287,15 +287,15 @@ impl PathData {
     ///
     /// This should only be called when migrating paths.
     pub(super) fn from_previous(
-        remote: SocketAddr,
+        addresses: FourTuple,
         prev: &Self,
         generation: u64,
         now: Instant,
     ) -> Self {
         let congestion = prev.congestion.clone_box();
         let smoothed_rtt = prev.rtt.get();
         Self {
-            remote,
+            addresses,
             rtt: prev.rtt,
             pacing: Pacer::new(smoothed_rtt, congestion.window(), prev.current_mtu(), now),
             sending_ecn: true,
@@ -366,7 +366,7 @@ impl PathData {
         self.total_sent = self.total_sent.saturating_add(inc);
         if !self.validated {
             trace!(
-                remote = %self.remote,
+                addresses = %self.addresses,
                 anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent),
                 "anti amplification budget decreased"
             );
@@ -378,7 +378,7 @@ impl PathData {
         self.total_recvd = self.total_recvd.saturating_add(inc);
         if !self.validated {
             trace!(
-                remote = %self.remote,
+                addresses = %self.addresses,
                 anti_amplification_budget = %(self.total_recvd * 3).saturating_sub(self.total_sent),
                 "anti amplification budget increased"
             );
@@ -638,15 +638,18 @@ pub(crate) struct PathResponses {
 }
 
 impl PathResponses {
-    pub(crate) fn push(&mut self, packet: u64, token: u64, remote: SocketAddr) {
+    pub(crate) fn push(&mut self, packet: u64, token: u64, addresses: FourTuple) {
         /// Arbitrary permissive limit to prevent abuse
         const MAX_PATH_RESPONSES: usize = 16;
         let response = PathResponse {
             packet,
             token,
-            remote,
+            addresses,
         };
-        let existing = self.pending.iter_mut().find(|x| x.remote == remote);
+        let existing = self
+            .pending
+            .iter_mut()
+            .find(|x| x.addresses.remote == addresses.remote);
         if let Some(existing) = existing {
             // Update a queued response
             if existing.packet <= packet {
@@ -663,20 +666,20 @@ impl PathResponses {
         }
     }
 
-    pub(crate) fn pop_off_path(&mut self, remote: SocketAddr) -> Option<(u64, SocketAddr)> {
+    pub(crate) fn pop_off_path(&mut self, addresses: FourTuple) -> Option<(u64, FourTuple)> {
         let response = *self.pending.last()?;
-        if response.remote == remote {
+        if response.addresses.is_same_path(&addresses) {
             // 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.remote))
+        Some((response.token, response.addresses))
     }
 
-    pub(crate) fn pop_on_path(&mut self, remote: SocketAddr) -> Option<u64> {
+    pub(crate) fn pop_on_path(&mut self, addresses: FourTuple) -> Option<u64> {
         let response = *self.pending.last()?;
-        if response.remote != remote {
+        if !response.addresses.is_same_path(&addresses) {
             // 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;
@@ -696,8 +699,8 @@ struct PathResponse {
     packet: u64,
     /// The token of the PATH_CHALLENGE
     token: u64,
-    /// The address the corresponding PATH_CHALLENGE was received from
-    remote: SocketAddr,
+    /// The path the corresponding PATH_CHALLENGE was received from
+    addresses: FourTuple,
 }
 
 /// Summary statistics of packets that have been sent on a particular path, but which have not yet

quinn-proto/src/connection/paths.rs

@@ -36,7 +36,7 @@ use qlog::{
 use tracing::warn;
 
 use crate::{
-    Connection, ConnectionId, Frame, Instant, PathId,
+    Connection, ConnectionId, FourTuple, Frame, Instant, PathId,
     connection::{PathData, SentPacket, timer::Timer},
     frame::{EcnCounts, StreamMeta},
     packet::{Header, SpaceId},
@@ -104,10 +104,10 @@ impl QlogStream {
         self.emit_event_with_tuple_id(event, now, None);
     }
 
-    fn emit_event_with_tuple_id(&self, event: EventData, now: Instant, tuple: Option<String>) {
+    fn emit_event_with_tuple_id(&self, event: EventData, now: Instant, addresses: Option<String>) {
         // Time will be overwritten by `add_event_with_instant`
         let mut event = Event::with_time(0.0, event);
-        event.tuple = tuple;
+        event.tuple = addresses;
         let mut qlog_streamer = self.0.lock().unwrap();
         if let Err(e) = qlog_streamer.add_event_with_instant(event, now) {
             warn!("could not emit qlog event: {e}");
@@ -242,7 +242,7 @@ impl QlogSink {
         }
     }
 
-    pub(super) fn emit_tuple_assigned(&self, path_id: PathId, remote: SocketAddr, now: Instant) {
+    pub(super) fn emit_tuple_assigned(&self, path_id: PathId, tuple: FourTuple, now: Instant) {
         #[cfg(feature = "qlog")]
         {
             let Some(stream) = self.stream.as_ref() else {
@@ -251,10 +251,12 @@ impl QlogSink {
             let tuple_id = fmt_tuple_id(path_id.as_u32() as u64);
             let event = TupleAssigned {
                 tuple_id,
-                tuple_local: None,
+                tuple_local: tuple
+                    .local_ip
+                    .map(|local_ip| tuple_endpoint_info(Some(local_ip), None, None)),
                 tuple_remote: Some(tuple_endpoint_info(
-                    Some(remote.ip()),
-                    Some(remote.port()),
+                    Some(tuple.remote.ip()),
+                    Some(tuple.remote.port()),
                     None,
                 )),
             };

quinn-proto/src/connection/qlog.rs

@@ -13,7 +13,7 @@ use tracing::trace;
 
 use super::{PathId, assembler::Assembler};
 use crate::{
-    Dir, Duration, Instant, SocketAddr, StreamId, TransportError, TransportErrorCode, VarInt,
+    Dir, Duration, FourTuple, Instant, StreamId, TransportError, TransportErrorCode, VarInt,
     connection::StreamsState,
     crypto::Keys,
     frame::{self, AddAddress, RemoveAddress},
@@ -525,10 +525,10 @@ pub struct Retransmits {
     ///
     /// 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<(SocketAddr, usize)>` or just a `usize` or
+    /// change this from a vector to an `Option<(FourTuple, usize)>` or just a `usize` or
     /// something. However, due to the architecture of Quinn, 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<SocketAddr>,
+    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

quinn-proto/src/connection/spaces.rs

@@ -15,8 +15,8 @@ use thiserror::Error;
 use tracing::{debug, error, trace, warn};
 
 use crate::{
-    Duration, INITIAL_MTU, Instant, MAX_CID_SIZE, MIN_INITIAL_SIZE, PathId, RESET_TOKEN_SIZE,
-    ResetToken, Side, Transmit, TransportConfig, TransportError,
+    Duration, FourTuple, INITIAL_MTU, Instant, MAX_CID_SIZE, MIN_INITIAL_SIZE, PathId,
+    RESET_TOKEN_SIZE, ResetToken, Side, Transmit, TransportConfig, TransportError,
     cid_generator::ConnectionIdGenerator,
     coding::BufMutExt,
     config::{ClientConfig, EndpointConfig, ServerConfig},
@@ -152,8 +152,7 @@ impl Endpoint {
     pub fn handle(
         &mut self,
         now: Instant,
-        remote: SocketAddr,
-        local_ip: Option<IpAddr>,
+        addresses: FourTuple,
         ecn: Option<EcnCodepoint>,
         data: BytesMut,
         buf: &mut Vec<u8>,
@@ -168,7 +167,7 @@ impl Endpoint {
         ) {
             Ok((first_decode, remaining)) => DatagramConnectionEvent {
                 now,
-                remote,
+                addresses,
                 path_id: PathId::ZERO, // Corrected later for existing paths
                 ecn,
                 first_decode,
@@ -200,11 +199,11 @@ impl Endpoint {
                     buf.write(version);
                 }
                 return Some(DatagramEvent::Response(Transmit {
-                    destination: remote,
+                    destination: addresses.remote,
                     ecn: None,
                     size: buf.len(),
                     segment_size: None,
-                    src_ip: local_ip,
+                    src_ip: addresses.local_ip,
                 }));
             }
             Err(e) => {
@@ -213,7 +212,6 @@ impl Endpoint {
             }
         };
 
-        let addresses = FourTuple { remote, local_ip };
         let dst_cid = event.first_decode.dst_cid();
 
         if let Some(route_to) = self.index.get(&addresses, &event.first_decode) {
@@ -663,7 +661,7 @@ impl Endpoint {
 
         match conn.handle_first_packet(
             incoming.received_at,
-            incoming.addresses.remote,
+            incoming.addresses,
             incoming.ecn,
             packet_number,
             incoming.packet,
@@ -848,8 +846,7 @@ impl Endpoint {
             init_cid,
             loc_cid,
             rem_cid,
-            addresses.remote,
-            addresses.local_ip,
+            addresses,
             tls,
             self.local_cid_generator.as_ref(),
             now,
@@ -1023,7 +1020,7 @@ struct ConnectionIndex {
     /// Identifies outgoing connections with zero-length CIDs
     ///
     /// We don't yet support explicit source addresses for client connections, and zero-length CIDs
-    /// require a unique four-tuple, so at most one client connection with zero-length local CIDs
+    /// require a unique 4-tuple, so at most one client connection with zero-length local CIDs
     /// may be established per remote. We must omit the local address from the key because we don't
     /// necessarily know what address we're sending from, and hence receiving at.
     ///
@@ -1386,14 +1383,3 @@ impl ResetTokenTable {
         self.0.get(&remote)?.get(&token)
     }
 }
-
-/// Identifies a connection by the combination of remote and local addresses
-///
-/// Including the local ensures good behavior when the host has multiple IP addresses on the same
-/// subnet and zero-length connection IDs are in use.
-#[derive(Hash, Eq, PartialEq, Debug, Copy, Clone)]
-struct FourTuple {
-    remote: SocketAddr,
-    // A single socket can only listen on a single port, so no need to store it explicitly
-    local_ip: Option<IpAddr>,
-}

quinn-proto/src/endpoint.rs

@@ -338,3 +338,69 @@ const MAX_UDP_PAYLOAD: u16 = 65527;
 const TIMER_GRANULARITY: Duration = Duration::from_millis(1);
 /// Maximum number of streams that can be uniquely identified by a stream ID
 const MAX_STREAM_COUNT: u64 = 1 << 60;
+
+/// Identifies a network path by the combination of remote and local addresses
+///
+/// Including the local ensures good behavior when the host has multiple IP addresses on the same
+/// subnet and zero-length connection IDs are in use or when multipath is enabled and multiple
+/// paths exist with the same remote, but different local IP interfaces.
+#[derive(Hash, Eq, PartialEq, Copy, Clone)]
+pub struct FourTuple {
+    /// The remote side of this tuple
+    pub remote: SocketAddr,
+    /// The local side of this tuple.
+    // A single socket can only listen on a single port, so no need to store it explicitly
+    // TODO(matheus23): add back the port. The comment above is wrong. There *can* be multiple sockets!
+    // Even ignoring multiple sockets behind a single abstract `AsyncUdpSocket`, there can
+    // be multiple `AsyncUdpSocket`s when you use `rebind`!
+    pub local_ip: Option<IpAddr>,
+}
+
+impl FourTuple {
+    /// Returns whether we know that these two tuples represent the same path.
+    ///
+    /// If any of these tuples doesn't have a local address, then we cannot know
+    /// if the two tuples represent the same path, even if the remote is the same
+    /// address, because the local address might be different.
+    // TODO(matheus23): Should this be the `PartialEq` definition on FourTuple? It might be too hidden.
+    pub fn is_same_path(&self, other: &Self) -> bool {
+        self.local_ip.is_some() && other.local_ip.is_some() && self == other
+    }
+
+    /// Updates this tuple's local address iff
+    /// - it was unset before,
+    /// - the other tuple has the same remote, and
+    /// - the other tuple has a local address set.
+    ///
+    /// Returns whether this and the other remote are now fully equal.
+    pub fn update_local_if_same_remote(&mut self, other: &Self) -> bool {
+        if self.remote != other.remote {
+            return false;
+        }
+        if self.local_ip.is_some() && self.local_ip != other.local_ip {
+            return false;
+        }
+        self.local_ip = other.local_ip;
+        true
+    }
+}
+
+impl fmt::Display for FourTuple {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("(")?;
+        if let Some(local_ip) = &self.local_ip {
+            local_ip.fmt(f)?;
+            f.write_str(":<port>, ")?;
+        } else {
+            f.write_str("<unknown>:<port>, ")?;
+        }
+        self.remote.fmt(f)?;
+        f.write_str(")")
+    }
+}
+
+impl fmt::Debug for FourTuple {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&self, f)
+    }
+}

quinn-proto/src/lib.rs

@@ -2,6 +2,7 @@ use std::{fmt, net::SocketAddr};
 
 use bytes::{Buf, BufMut, BytesMut};
 
+use crate::FourTuple;
 use crate::PathId;
 use crate::{Duration, Instant, MAX_CID_SIZE, ResetToken, coding::BufExt, packet::PartialDecode};
 
@@ -21,7 +22,7 @@ pub(crate) enum ConnectionEventInner {
 #[derive(Debug)]
 pub(crate) struct DatagramConnectionEvent {
     pub(crate) now: Instant,
-    pub(crate) remote: SocketAddr,
+    pub(crate) addresses: FourTuple,
     pub(crate) path_id: PathId,
     pub(crate) ecn: Option<EcnCodepoint>,
     pub(crate) first_decode: PartialDecode,