fix(proto): Use correct info in retransmission of *_BLOCKED frames

noq-proto/src/connection/mod.rs

@@ -563,9 +563,9 @@ impl Connection {
         initial_status: PathStatus,
         now: Instant,
     ) -> Result<PathId, PathError> {
-        if !self.is_multipath_negotiated() {
+        let Some(max_path_id) = self.max_path_id() else {
             return Err(PathError::MultipathNotNegotiated);
-        }
+        };
         if self.side().is_server() {
             return Err(PathError::ServerSideNotAllowed);
         }
@@ -577,23 +577,15 @@ impl Connection {
             .unwrap_or(PathId::ZERO)
             .saturating_add(1u8);
 
-        if Some(path_id) > self.max_path_id() {
+        if path_id > max_path_id {
+            self.spaces[SpaceId::Data].pending.paths_blocked = Some(self.remote_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
-            .remote_cids
-            .get(&path_id)
-            .map(CidQueue::active)
-            .is_none()
-        {
+        if !self.remote_cids.contains_key(&path_id) {
             self.spaces[SpaceId::Data]
                 .pending
                 .path_cids_blocked
-                .insert(path_id);
+                .insert(path_id, VarInt(0));
             return Err(PathError::RemoteCidsExhausted);
         }
 
@@ -691,15 +683,13 @@ impl Connection {
 
         // Remove pending NEW CIDs for this path
         pending_space.new_cids.retain(|cid| cid.path_id != path_id);
-        pending_space.path_cids_blocked.retain(|&id| id != path_id);
         pending_space.path_status.retain(|&id| id != path_id);
 
         // Cleanup retransmits across ALL paths (CIDs for path_id may have been transmitted on other paths)
         for space in self.spaces[SpaceId::Data].iter_paths_mut() {
             for sent_packet in space.sent_packets.values_mut() {
                 if let Some(retransmits) = sent_packet.retransmits.get_mut() {
                     retransmits.new_cids.retain(|cid| cid.path_id != path_id);
-                    retransmits.path_cids_blocked.retain(|&id| id != path_id);
                     retransmits.path_status.retain(|&id| id != path_id);
                 }
             }
@@ -1002,7 +992,7 @@ impl Connection {
             self.spaces[SpaceId::Data]
                 .pending
                 .path_cids_blocked
-                .insert(path_id);
+                .insert(path_id, VarInt(0));
             // Do not abandon this path right away. CIDs might be in-flight still and arrive
             // soon. It is up to the remote to handle this situation.
         }
@@ -5403,8 +5393,6 @@ impl Connection {
                                 "PATHS_BLOCKED maximum path identifier was larger than local maximum",
                             ));
                         }
-                        debug!("received PATHS_BLOCKED({:?})", max_path_id);
-                        // TODO(@divma): ensure max concurrent paths
                     } else {
                         return Err(TransportError::PROTOCOL_VIOLATION(
                             "received PATHS_BLOCKED frame when not multipath was not negotiated",
@@ -6445,12 +6433,11 @@ impl Connection {
         if space_id == SpaceId::Data
             && !scheduling_info.is_abandoned
             && scheduling_info.may_send_data
-            && space.pending.paths_blocked
             && frame::PathsBlocked::SIZE_BOUND <= builder.frame_space_remaining()
+            && let Some(remote_max_path_id) = space.pending.paths_blocked.take()
         {
-            let frame = frame::PathsBlocked(self.remote_max_path_id);
+            let frame = frame::PathsBlocked(remote_max_path_id);
             builder.write_frame(frame, stats);
-            space.pending.paths_blocked = false;
         }
 
         // PATH_CIDS_BLOCKED
@@ -6459,13 +6446,9 @@ impl Connection {
             && scheduling_info.may_send_data
             && frame::PathCidsBlocked::SIZE_BOUND <= builder.frame_space_remaining()
         {
-            let Some(path_id) = space.pending.path_cids_blocked.pop_first() else {
+            let Some((path_id, next_seq)) = space.pending.path_cids_blocked.pop_first() else {
                 break;
             };
-            let next_seq = match self.remote_cids.get(&path_id) {
-                Some(cid_queue) => VarInt(cid_queue.active_seq() + 1),
-                None => VarInt(0),
-            };
             let frame = frame::PathCidsBlocked { path_id, next_seq };
             builder.write_frame(frame, stats);
         }
@@ -7714,11 +7697,18 @@ impl SentFrames {
                 self.retransmits_mut().path_status.insert(path_id);
             }
             MaxPathId(_) => self.retransmits_mut().max_path_id = true,
-            PathsBlocked(_) => self.retransmits_mut().paths_blocked = true,
+            PathsBlocked(frame::PathsBlocked(path_id)) => {
+                let paths_blocked = &mut self.retransmits_mut().paths_blocked;
+                *paths_blocked = cmp::max(*paths_blocked, Some(path_id));
+            }
             PathCidsBlocked(path_cids_blocked) => {
                 self.retransmits_mut()
                     .path_cids_blocked
-                    .insert(path_cids_blocked.path_id);
+                    .entry(path_cids_blocked.path_id)
+                    .and_modify(|next_seq| {
+                        *next_seq = cmp::max(*next_seq, path_cids_blocked.next_seq);
+                    })
+                    .or_insert(path_cids_blocked.next_seq);
             }
             ResetStream(reset) => self
                 .retransmits_mut()

noq-proto/src/connection/spaces.rs

@@ -558,8 +558,12 @@ pub struct Retransmits {
     pub(super) max_path_id: bool,
     /// Whether we should inform the peer that their max [`PathId`] is blocking our attempt to open
     /// new paths.
-    // TODO(@divma): we need logic to prevent sending this more than once after being ack-d once
-    pub(super) paths_blocked: bool,
+    ///
+    /// Stores the remote_max_path_id at the time this was generated.
+    /// This frame is entirely informational, so when it's retransmitted, the remote_max_path_id is
+    /// intentionally not updated to preserve the fact that this was the state of the client at some
+    /// point.
+    pub(super) paths_blocked: Option<PathId>,
     /// For each enqueued NEW_TOKEN frame, a copy of the path's remote address
     ///
     /// There are 2 reasons this is unusual:
@@ -581,8 +585,12 @@ pub struct Retransmits {
     pub(super) path_abandon: BTreeMap<PathId, TransportErrorCode>,
     /// If a [`frame::PathStatusAvailable`] and [`frame::PathStatusBackup`] need to be sent for a path
     pub(super) path_status: BTreeSet<PathId>,
-    /// If a PATH_CIDS_BLOCKED frame needs to be sent for a path
-    pub(super) path_cids_blocked: BTreeSet<PathId>,
+    /// Whether a PATH_CIDS_BLOCKED frame needs to be sent for a path.
+    ///
+    /// Stores the next_seq number for the blocked path. This number can be "outdated" at the time of
+    /// sending when this is a retransmission. This is intentional, as this frame is purely
+    /// informational, and this would preserve this information.
+    pub(super) path_cids_blocked: BTreeMap<PathId, VarInt>,
 
     // Nat traversal data
     /// Addresses to report in `ADD_ADDRESS` frames
@@ -633,7 +641,7 @@ impl Retransmits {
             && !handshake_done
             && !observed_addr
             && !max_path_id
-            && !paths_blocked
+            && paths_blocked.is_none()
             && new_tokens.is_empty()
             && path_abandon.is_empty()
             && path_status.is_empty()
@@ -689,7 +697,7 @@ impl ::std::ops::BitOrAssign for Retransmits {
         self.handshake_done |= handshake_done;
         self.observed_addr |= observed_addr;
         self.max_path_id |= max_path_id;
-        self.paths_blocked |= paths_blocked;
+        self.paths_blocked = cmp::max(self.paths_blocked, paths_blocked);
         self.new_tokens.extend_from_slice(&new_tokens);
         self.path_abandon.append(&mut path_abandon);
         self.path_status.append(&mut path_status);

noq-proto/src/tests/multipath.rs

@@ -2124,6 +2124,35 @@ fn on_path_challenge_lost_backoff() {
     assert_eq!(duration, Duration::from_millis(1));
 }
 
+#[test]
+fn paths_blocked_retransmission() -> TestResult {
+    let _guard = subscribe();
+    let mut pair = ConnPair::builder().enable_multipath().connect();
+
+    let server_addr = pair.routes.public_server_addr();
+    for _ in 1..MAX_PATHS {
+        pair.open_path(
+            Client,
+            FourTuple::from_remote(server_addr),
+            PathStatus::Available,
+        )?;
+    }
+    pair.drive_client(); // Open all the paths we are allowed to open
+    pair.drive_server(); // Let the server process all these newly opened paths
+    pair.open_path(
+        Client,
+        FourTuple::from_remote(server_addr),
+        PathStatus::Available,
+    )
+    .expect_err("expected PathError::MaxPathIdReached");
+    pair.drive_client(); // Let the client produce the PATHS_BLOCKED frame
+    assert_eq!(pair.stats(Client).frame_tx.paths_blocked, 1);
+    pair.server.inbound.clear(); // We drop the PATHS_BLOCKED frame
+    pair.drive();
+    assert_eq!(pair.stats(Client).frame_tx.paths_blocked, 2);
+    Ok(())
+}
+
 /// This test used to generate a PROTOCOL_VIOLATION error from just packet loss and delayed packets.
 ///
 /// The problem was receiving a PATH_CIDS_BLOCKED frame with path_id=1 and next_seq=1 when the server