diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 8ebd7c9d31..e89d893968 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -563,9 +563,9 @@ impl Connection { initial_status: PathStatus, now: Instant, ) -> Result { - 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,7 +683,6 @@ 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) @@ -699,7 +690,6 @@ impl Connection { 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() diff --git a/noq-proto/src/connection/spaces.rs b/noq-proto/src/connection/spaces.rs index 69e2d1738c..8885828366 100644 --- a/noq-proto/src/connection/spaces.rs +++ b/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, /// 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, /// If a [`frame::PathStatusAvailable`] and [`frame::PathStatusBackup`] need to be sent for a path pub(super) path_status: BTreeSet, - /// If a PATH_CIDS_BLOCKED frame needs to be sent for a path - pub(super) path_cids_blocked: BTreeSet, + /// 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, // 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); diff --git a/noq-proto/src/tests/multipath.rs b/noq-proto/src/tests/multipath.rs index d93d299416..5fb4b2b6a2 100644 --- a/noq-proto/src/tests/multipath.rs +++ b/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