send and receive PATH_AVAILABLE/PATH_BACKUP

quinn-proto/src/connection/mod.rs

@@ -24,7 +24,7 @@ use crate::{
     config::{ServerConfig, TransportConfig},
     congestion::Controller,
     crypto::{self, KeyPair, Keys, PacketKey},
-    frame::{self, Close, Datagram, FrameStruct, NewToken, ObservedAddr, PathAvailable},
+    frame::{self, Close, Datagram, FrameStruct, NewToken, ObservedAddr},
     packet::{
         FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, LongType, Packet,
         PacketNumber, PartialDecode, SpaceId,
@@ -62,7 +62,7 @@ use packet_crypto::{PrevCrypto, ZeroRttCrypto};
 
 mod paths;
 use paths::PathData;
-pub use paths::{PathEvent, PathId, PathStatus, RttEstimator};
+pub use paths::{ClosedPath, PathEvent, PathId, PathStatus, RttEstimator};
 
 mod send_buffer;
 
@@ -543,11 +543,41 @@ impl Connection {
         &self.paths.get(&path_id).expect("known path").data
     }
 
-    /// Gets the [`PathStatus`] for a known [`PathId`].
+    /// Gets the [`PathStatus`] for a known [`PathId`]
+    pub fn path_status(&self, path_id: PathId) -> Result<PathStatus, ClosedPath> {
+        match self.paths.get(&path_id) {
+            Some(path) => Ok(path.data.status.local_status),
+            None => Err(ClosedPath { _private: () }),
+        }
+    }
+
+    /// Sets the [`PathStatus`] for a known [`PathId`]
     ///
-    /// Will panic if the path_id does not reference any known path.
-    pub fn path_status(&self, path_id: PathId) -> PathStatus {
-        self.path_data(path_id).status
+    /// Returns the previous path status on success.
+    pub fn set_path_status(
+        &mut self,
+        path_id: PathId,
+        status: PathStatus,
+    ) -> Result<PathStatus, ClosedPath> {
+        match self.paths.get_mut(&path_id) {
+            Some(path) => {
+                let prev_status = path.data.status.local_status;
+                path.data.status.local_status = status;
+                self.spaces[SpaceId::Data].pending.path_status.push(path_id);
+                Ok(prev_status)
+            }
+            None => Err(ClosedPath { _private: () }),
+        }
+    }
+
+    /// Returns the remote path status
+    // TODO(flub): Probably should also be some kind of path event?  Not even sure if I like
+    //    this as an API, but for now it allows me to write a test easily.
+    // TODO(flub): Technically this should be a Result<Option<PathSTatus>>?
+    pub fn remote_path_status(&self, path_id: PathId) -> Option<PathStatus> {
+        self.paths
+            .get(&path_id)
+            .and_then(|path| path.data.status.remote_status)
     }
 
     /// Gets the [`PathData`] for a known [`PathId`].
@@ -587,12 +617,10 @@ impl Connection {
         };
 
         let mut path = PathData::new(remote, self.allow_mtud, None, now, &self.config);
-        path.status = status;
+        path.status.local_status = status;
 
         let challenge = self.rng.random();
         path.challenge = Some(challenge);
-        let status_seq_no = path.local_status_seq_no;
-        path.local_status_seq_no = path.local_status_seq_no.saturating_add(1u8);
         self.paths.insert(
             path_id,
             PathState {
@@ -601,15 +629,10 @@ impl Connection {
             },
         );
 
-        // regardles of what's the status we inform it to the remote
-        let frames = vec![
-            Frame::PathChallenge(challenge),
-            Frame::PathAvailable(PathAvailable {
-                is_backup: status == PathStatus::Backup,
-                path_id,
-                status_seq_no,
-            }),
-        ];
+        // Inform the remote of the path status.
+        self.spaces[SpaceId::Data].pending.path_status.push(path_id);
+
+        let frames = vec![Frame::PathChallenge(challenge)];
         Some((Some((path_id, remote_cid)), frames))
     }
 
@@ -702,12 +725,12 @@ impl Connection {
 
         let mut path_id = *self.paths.first_key_value().expect("one path must exist").0;
 
-        // If there is any available path we only want to send frames to a backup path that
-        // must be sent on that path.
+        // If there is any available path we only want to send frames to any backup path
+        // that must be sent on that backup path exclusively.
         let have_available_path = self
             .paths
             .values()
-            .any(|path| path.data.status == PathStatus::Available);
+            .any(|path| path.data.status.local_status == PathStatus::Available);
 
         // Setup for the first path_id
         let mut transmit = TransmitBuf::new(
@@ -737,7 +760,7 @@ impl Connection {
             let path_should_send = {
                 let path_exclusive_only = space_id == SpaceId::Data
                     && have_available_path
-                    && self.path_data(path_id).status == PathStatus::Backup;
+                    && self.path_data(path_id).status.local_status == PathStatus::Backup;
                 let path_should_send = if path_exclusive_only {
                     can_send.path_exclusive
                 } else {
@@ -748,7 +771,9 @@ impl Connection {
             };
 
             if !path_should_send && space_id < SpaceId::Data {
-                trace!(?space_id, ?path_id, "everything sent in space");
+                if self.spaces[space_id].crypto.is_some() {
+                    trace!(?space_id, ?path_id, "nothing to send in space");
+                }
                 space_id = space_id.next();
                 continue;
             }
@@ -781,7 +806,12 @@ impl Connection {
                 match self.paths.keys().find(|&&next| next > path_id) {
                     Some(next_path_id) => {
                         // See if this next path can send anything.
-                        trace!(?space_id, ?path_id, ?next_path_id, "trying next path");
+                        trace!(
+                            ?space_id,
+                            ?path_id,
+                            ?next_path_id,
+                            "nothing to send on path"
+                        );
                         path_id = *next_path_id;
                         space_id = SpaceId::Data;
 
@@ -797,7 +827,11 @@ impl Connection {
                     }
                     None => {
                         // Nothing more to send.
-                        trace!(?space_id, ?path_id, "no higher path id to send on");
+                        trace!(
+                            ?space_id,
+                            ?path_id,
+                            "nothing to send on path, no more paths"
+                        );
                         break;
                     }
                 }
@@ -995,8 +1029,8 @@ impl Connection {
             }
 
             let sent_frames = {
-                let path_exclusive_only =
-                    have_available_path && self.path_data(path_id).status == PathStatus::Backup;
+                let path_exclusive_only = have_available_path
+                    && self.path_data(path_id).status.local_status == PathStatus::Backup;
                 let pn = builder.exact_number;
                 self.populate_packet(
                     now,
@@ -3589,10 +3623,23 @@ impl Connection {
                 }
                 Frame::PathAvailable(info) => {
                     if self.is_multipath_negotiated() {
-                        self.on_path_available(info.path_id, info.is_backup, info.status_seq_no);
+                        self.on_path_status(
+                            info.path_id,
+                            PathStatus::Available,
+                            info.status_seq_no,
+                        );
+                    } else {
+                        return Err(TransportError::PROTOCOL_VIOLATION(
+                            "received PATH_AVAILABLE frame when multipath was not negotiated",
+                        ));
+                    }
+                }
+                Frame::PathBackup(info) => {
+                    if self.is_multipath_negotiated() {
+                        self.on_path_status(info.path_id, PathStatus::Backup, info.status_seq_no);
                     } else {
                         return Err(TransportError::PROTOCOL_VIOLATION(
-                            "received PATH_AVAILABLE frame when not multipath was not negotiated",
+                            "received PATH_BACKUP frame when multipath was not negotiated",
                         ));
                     }
                 }
@@ -4082,6 +4129,7 @@ impl Connection {
             }
         }
 
+        // TODO(flub): maybe this is much higher priority?
         // PATH_ABANDON
         while !path_exclusive_only
             && space_id == SpaceId::Data
@@ -4098,6 +4146,38 @@ impl Connection {
             // TODO(flub): frame stats?
         }
 
+        // PATH_AVAILABLE & PATH_BACKUP
+        while !path_exclusive_only
+            && space_id == SpaceId::Data
+            && frame::PathAvailable::SIZE_BOUND <= buf.remaining_mut()
+        {
+            let Some(path_id) = space.pending.path_status.pop() else {
+                break;
+            };
+            let status_state = &mut path.status;
+            let seq = status_state.local_seq;
+            status_state.local_seq.saturating_add(1u8);
+            match status_state.local_status {
+                PathStatus::Available => {
+                    frame::PathAvailable {
+                        path_id,
+                        status_seq_no: seq,
+                    }
+                    .encode(buf);
+                    trace!(?path_id, %seq, "PATH_AVAILABLE")
+                }
+                PathStatus::Backup => {
+                    frame::PathBackup {
+                        path_id,
+                        status_seq_no: seq,
+                    }
+                    .encode(buf);
+                    trace!(?path_id, %seq, "PATH_BACKUP")
+                }
+            }
+            // TODO(flub): frame stats?
+        }
+
         // RESET_STREAM, STOP_SENDING, MAX_DATA, MAX_STREAM_DATA, MAX_STREAMS
         if space_id == SpaceId::Data {
             self.streams.write_control_frames(
@@ -4727,19 +4807,10 @@ impl Connection {
         }
     }
 
-    /// Handle new path availability information
-    fn on_path_available(&mut self, path_id: PathId, is_backup: bool, status_seq_no: VarInt) {
-        if let Some(path_data) = self.paths.get_mut(&path_id) {
-            let data = &mut path_data.data;
-            // If a newer sequence no was sent, update the information.
-            if data.status_seq_no < Some(status_seq_no) {
-                data.status = if is_backup {
-                    PathStatus::Backup
-                } else {
-                    PathStatus::Available
-                };
-                data.status_seq_no.replace(status_seq_no);
-            }
+    /// Handle new path status information: PATH_AVAILABLE, PATH_BACKUP
+    fn on_path_status(&mut self, path_id: PathId, status: PathStatus, status_seq_no: VarInt) {
+        if let Some(path) = self.paths.get_mut(&path_id) {
+            path.data.status.on_path_status(status, status_seq_no);
         } else {
             debug!("PATH_AVAILABLE received unknown path {:?}", path_id);
         }

quinn-proto/src/connection/paths.rs

@@ -1,6 +1,7 @@
 use std::{cmp, net::SocketAddr};
 
-use tracing::trace;
+use thiserror::Error;
+use tracing::{debug, trace};
 
 use super::{
     OpenPathError,
@@ -98,11 +99,7 @@ pub(super) struct PathData {
     /// Observed address frame with the largest sequence number received from the peer on this path.
     pub(super) last_observed_addr_report: Option<ObservedAddr>,
     /// The QUIC-MULTIPATH path status
-    pub(super) status: PathStatus,
-    /// Next status sequence number to use to inform the remote of status updates.
-    pub(super) local_status_seq_no: VarInt,
-    /// The sequence number of the received PATH_AVAILABLE and PATH_BACKUP frames.
-    pub(super) status_seq_no: Option<VarInt>,
+    pub(super) status: PathStatusState,
     /// Number of the first packet sent on this path
     ///
     /// Used to determine whether a packet was sent on an earlier path. Insufficient to determine if
@@ -161,8 +158,6 @@ impl PathData {
             observed_addr_sent: false,
             last_observed_addr_report: None,
             status: Default::default(),
-            local_status_seq_no: VarInt::default(),
-            status_seq_no: None,
             first_packet: None,
             pto_count: 0,
         }
@@ -191,9 +186,7 @@ impl PathData {
             in_flight: InFlight::new(),
             observed_addr_sent: false,
             last_observed_addr_report: None,
-            status: prev.status,
-            local_status_seq_no: prev.local_status_seq_no,
-            status_seq_no: prev.status_seq_no,
+            status: prev.status.clone(),
             first_packet: None,
             pto_count: 0,
         }
@@ -480,6 +473,40 @@ impl InFlight {
     }
 }
 
+/// State for QUIC-MULTIPATH PATH_AVAILABLE and PATH_BACKUP frames
+// TODO(flub): We might want to keep track of whether the local state was set explicitly.
+//    OTOH our open path API allows explicitly setting the path status, which means it is
+//    always set explicitly.  So basically we currently just use the local state and only
+//    keep track of the remote state.
+#[derive(Debug, Clone, Default)]
+pub(super) struct PathStatusState {
+    /// The local status
+    pub(super) local_status: PathStatus,
+    /// Local sequence number, for both PATH_AVAIALABLE and PATH_BACKUP
+    ///
+    /// This is the number of the *next* path status frame to be sent.
+    pub(super) local_seq: VarInt,
+    /// The status set by the remote
+    pub(super) remote_status: Option<PathStatus>,
+    /// Remote sequence number, for both PATH_AVAIALABLE and PATH_BACKUP
+    pub(super) remote_seq: Option<VarInt>,
+}
+
+impl PathStatusState {
+    /// To be called on received PATH_AVAILABLE/PATH_BACKUP frames
+    pub(super) fn on_path_status(&mut self, status: PathStatus, seq: VarInt) {
+        if Some(seq) <= self.remote_seq {
+            tracing::warn!("whoops");
+            return;
+        }
+        self.remote_seq = Some(seq);
+        let prev = self.remote_status.replace(status);
+        if prev != Some(status) {
+            debug!(?status, ?seq, "remote changed path status");
+        }
+    }
+}
+
 /// The QUIC-MULTIPATH path status
 ///
 /// See section "3.3 Path Status Management":
@@ -525,6 +552,13 @@ pub enum PathEvent {
     },
 }
 
+/// 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: (),
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

quinn-proto/src/connection/spaces.rs

@@ -520,6 +520,8 @@ pub struct Retransmits {
     pub(super) new_tokens: Vec<SocketAddr>,
     /// Paths which need to be abandoned
     pub(super) path_abandon: Vec<(PathId, TransportErrorCode)>,
+    /// If a PATH_AVAILABLE and PATH_BACKUP frame needs to be sent for a path
+    pub(super) path_status: Vec<PathId>,
 }
 
 impl Retransmits {
@@ -540,6 +542,7 @@ impl Retransmits {
             && !self.observed_addr
             && self.new_tokens.is_empty()
             && self.path_abandon.is_empty()
+            && self.path_status.is_empty()
     }
 }