Start sending PATH_CIDS_BLOCKED (#106)

Sends and receives PATH_CIDS_BLOCKED and PATHS_BLOCKED correctly

quinn-proto/src/connection/cid_state.rs

@@ -195,7 +195,6 @@ impl CidState {
         self.retire_seq
     }
 
-    #[cfg(test)]
     pub(crate) fn active_seq(&self) -> (u64, u64) {
         let mut min = u64::MAX;
         let mut max = u64::MIN;

quinn-proto/src/connection/mod.rs

@@ -545,11 +545,17 @@ impl Connection {
         if Some(path_id) > self.max_path_id() {
             return Err(PathError::MaxPathIdReached);
         }
-
         if path_id > self.remote_max_path_id {
             self.spaces[SpaceId::Data].pending.paths_blocked = true;
             return Err(PathError::MaxPathIdReached);
         }
+        if self.rem_cids.get(&path_id).map(CidQueue::active).is_none() {
+            self.spaces[SpaceId::Data]
+                .pending
+                .path_cids_blocked
+                .push(path_id);
+            return Err(PathError::RemoteCidsExhausted);
+        }
 
         // Create PathData, schedule PATH_CHALLENGE to be sent.
         // TODO(flub): Not sure if we need to send a PATH_CHALLENGE in all situations?
@@ -845,18 +851,14 @@ impl Connection {
                     id: path_id,
                     error: err,
                 }));
-                // this allows us to safely consume the path_id due to the failed attempt,
-                // otherwise we would need to keep track of holes in the range and revert the
-                // max_path_id_in_use
+                // Locally we should have refused to open this path, the remote should have
+                // given us CIDs for this path before opening it.
+                self.close_path(path_id, TransportErrorCode::NO_CID_AVAILABLE.into())
+                    .ok();
                 self.spaces[SpaceId::Data]
                     .pending
-                    .path_abandon
-                    .push((path_id, TransportErrorCode::NO_CID_AVAILABLE));
-                // TODO(@divma): we here need to set the path as abandoned to avoid using it
-                // TODO(@divma): missing logic to remove the path, most likely not here tho. Note
-                // that this includes "considering all associated CIDs retired"
-                // TODO(@divma): here we could send PATH_CIDS_BLOCKED frame, just to play all the
-                // spec's frames, given our range maintenance strategy it's otherwise useless
+                    .path_cids_blocked
+                    .push(path_id);
 
                 match self.paths.keys().find(|&&next| next > path_id) {
                     Some(next_path_id) => {
@@ -3918,7 +3920,7 @@ impl Connection {
                         ));
                     }
                 }
-                Frame::PathCidsBlocked(path_id) => {
+                Frame::PathCidsBlocked(frame::PathCidsBlocked { path_id, next_seq }) => {
                     // Nothing to do.  This is recorded in the frame stats, but otherwise we
                     // always issue all CIDs we're allowed to issue, so either this is an
                     // impatient peer or a bug on our side.
@@ -3932,7 +3934,18 @@ impl Connection {
                                 "PATH_CIDS_BLOCKED path identifier was larger than local maximum",
                             ));
                         }
-                        debug!("received PATH_CIDS_BLOCKED({:?})", path_id);
+                        if next_seq.0
+                            > self
+                                .local_cid_state
+                                .get(&path_id)
+                                .map(|cid_state| cid_state.active_seq().1 + 1)
+                                .unwrap_or_default()
+                        {
+                            return Err(TransportError::PROTOCOL_VIOLATION(
+                                "PATH_CIDS_BLOCKED next sequence number larger than in local state",
+                            ));
+                        }
+                        debug!(?path_id, %next_seq, "received PATH_CIDS_BLOCKED");
                     } else {
                         return Err(TransportError::PROTOCOL_VIOLATION(
                             "received PATH_CIDS_BLOCKED frame when not multipath was not negotiated",
@@ -4435,11 +4448,39 @@ impl Connection {
             self.stats.frame_tx.max_path_id += 1;
         }
 
-        // TODO(@divma): missing size bound checks and potentially other checks
-        if space_id == SpaceId::Data && space.pending.paths_blocked {
+        // PATHS_BLOCKED
+        if space_id == SpaceId::Data
+            && space.pending.paths_blocked
+            && frame::PathsBlocked::SIZE_BOUND <= buf.remaining_mut()
+        {
             frame::PathsBlocked(self.remote_max_path_id).encode(buf);
             space.pending.paths_blocked = false;
             sent.retransmits.get_or_create().paths_blocked = true;
+            trace!(max_path_id = ?self.remote_max_path_id, "PATHS_BLOCKED");
+            self.stats.frame_tx.paths_blocked += 1;
+        }
+
+        // PATH_CIDS_BLOCKED
+        while space_id == SpaceId::Data && frame::PathCidsBlocked::SIZE_BOUND <= buf.remaining_mut()
+        {
+            let Some(path_id) = space.pending.path_cids_blocked.pop() else {
+                break;
+            };
+            let next_seq = match self.rem_cids.get(&path_id) {
+                Some(cid_queue) => cid_queue.active_seq() + 1,
+                None => 0,
+            };
+            frame::PathCidsBlocked {
+                path_id,
+                next_seq: VarInt(next_seq),
+            }
+            .encode(buf);
+            sent.retransmits
+                .get_or_create()
+                .path_cids_blocked
+                .push(path_id);
+            trace!(?path_id, next_seq, "PATH_CIDS_BLOCKED");
+            self.stats.frame_tx.path_cids_blocked += 1;
         }
 
         // RESET_STREAM, STOP_SENDING, MAX_DATA, MAX_STREAM_DATA, MAX_STREAMS

quinn-proto/src/connection/spaces.rs

@@ -546,6 +546,8 @@ pub struct Retransmits {
     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>,
+    /// If a PATH_CIDS_BLOCKED frame needs to be sent for a path
+    pub(super) path_cids_blocked: Vec<PathId>,
 }
 
 impl Retransmits {

quinn-proto/src/frame.rs

@@ -202,10 +202,7 @@ pub(crate) enum Frame {
     PathBackup(PathBackup),
     MaxPathId(MaxPathId),
     PathsBlocked(PathsBlocked),
-    // TODO(flub): We should send this to be spec-compliant, but for ourselves we don't
-    // really care because we always issue CIDs.  Perhaps we can get this frame removed
-    // again from the spec: https://github.com/quicwg/multipath/issues/500
-    PathCidsBlocked(PathId),
+    PathCidsBlocked(PathCidsBlocked),
 }
 
 impl Frame {
@@ -751,10 +748,13 @@ impl MaxPathId {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct PathsBlocked(pub(crate) PathId);
 
 impl PathsBlocked {
+    pub(crate) const SIZE_BOUND: usize =
+        VarInt(FrameType::PATHS_BLOCKED.0).size() + VarInt(u32::MAX as u64).size();
+
     pub(crate) fn decode<B: Buf>(buf: &mut B) -> coding::Result<Self> {
         Ok(Self(buf.get()?))
     }
@@ -765,6 +765,31 @@ impl PathsBlocked {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct PathCidsBlocked {
+    pub(crate) path_id: PathId,
+    pub(crate) next_seq: VarInt,
+}
+
+impl PathCidsBlocked {
+    pub(crate) const SIZE_BOUND: usize = VarInt(FrameType::PATH_CIDS_BLOCKED.0).size()
+        + VarInt(u32::MAX as u64).size()
+        + VarInt::MAX.size();
+
+    pub(crate) fn decode<R: Buf>(buf: &mut R) -> coding::Result<Self> {
+        Ok(Self {
+            path_id: buf.get()?,
+            next_seq: buf.get()?,
+        })
+    }
+
+    pub(crate) fn encode<W: BufMut>(&self, buf: &mut W) {
+        buf.write(FrameType::PATH_CIDS_BLOCKED);
+        buf.write(self.path_id);
+        buf.write(self.next_seq);
+    }
+}
+
 pub(crate) struct Iter {
     bytes: Bytes,
     last_ty: Option<FrameType>,
@@ -932,7 +957,9 @@ impl Iter {
             FrameType::PATH_BACKUP => Frame::PathBackup(PathBackup::decode(&mut self.bytes)?),
             FrameType::MAX_PATH_ID => Frame::MaxPathId(MaxPathId::decode(&mut self.bytes)?),
             FrameType::PATHS_BLOCKED => Frame::PathsBlocked(PathsBlocked::decode(&mut self.bytes)?),
-            FrameType::PATH_CIDS_BLOCKED => Frame::PathCidsBlocked(self.bytes.get()?),
+            FrameType::PATH_CIDS_BLOCKED => {
+                Frame::PathCidsBlocked(PathCidsBlocked::decode(&mut self.bytes)?)
+            }
             _ => {
                 if let Some(s) = ty.stream() {
                     Frame::Stream(Stream {
@@ -1603,4 +1630,28 @@ mod test {
             x => panic!("incorrect frame {x:?}"),
         }
     }
+
+    #[test]
+    fn test_paths_blocked_path_cids_blocked_roundtrip() {
+        let mut buf = Vec::new();
+
+        let frame0 = PathsBlocked(PathId(22));
+        frame0.encode(&mut buf);
+        let frame1 = PathCidsBlocked {
+            path_id: PathId(23),
+            next_seq: VarInt(32),
+        };
+        frame1.encode(&mut buf);
+
+        let mut decoded = frames(buf);
+        assert_eq!(decoded.len(), 2);
+        match decoded.pop().expect("non empty") {
+            Frame::PathCidsBlocked(decoded) => assert_eq!(decoded, frame1),
+            x => panic!("incorrect frame {x:?}"),
+        }
+        match decoded.pop().expect("non empty") {
+            Frame::PathsBlocked(decoded) => assert_eq!(decoded, frame0),
+            x => panic!("incorrect frame {x:?}"),
+        }
+    }
 }