fix(proto): send a path's acknowledgements on that path

noq-proto/src/connection/mod.rs

@@ -2206,6 +2206,10 @@ impl Connection {
         packet_size: usize,
         connection_close_pending: bool,
     ) -> SendableFrames {
+        let (paths, remote_cids, abandoned) =
+            (&self.paths, &self.remote_cids, &self.abandoned_paths);
+        let carries_own_acks =
+            |p: PathId| Self::path_carries_own_acks(paths, remote_cids, abandoned, p);
         let space = &mut self.spaces[space_id];
         let space_has_crypto = self.crypto_state.has_keys(space_id.encryption_level());
 
@@ -2218,7 +2222,7 @@ impl Connection {
             return SendableFrames::empty();
         }
 
-        let mut can_send = space.can_send(path_id, &self.streams);
+        let mut can_send = space.can_send(path_id, &self.streams, carries_own_acks);
 
         // Check for 1RTT space.
         if space_id == SpaceId::Data {
@@ -6099,6 +6103,27 @@ impl Connection {
         let is_0rtt = space_id == SpaceId::Data && !space_has_keys;
         let stats = &mut self.path_stats.get_mut(path_id).frame_tx;
         let space = &mut self.spaces[space_id];
+        // Other paths whose acks must not ride this packet because they can carry their own
+        // (see the ACK section below). Only relevant with more than one path; the list is at
+        // most the number of paths.
+        let other_own_carriers: Vec<PathId> = if space.number_spaces.len() > 1 {
+            space
+                .number_spaces
+                .keys()
+                .copied()
+                .filter(|&p| {
+                    p != path_id
+                        && Self::path_carries_own_acks(
+                            &self.paths,
+                            &self.remote_cids,
+                            &self.abandoned_paths,
+                            p,
+                        )
+                })
+                .collect()
+        } else {
+            Vec::new()
+        };
         let path = &mut self.paths.get_mut(&path_id).expect("known path").data;
         space
             .for_path(path_id)
@@ -6134,11 +6159,19 @@ impl Connection {
         }
 
         // ACK
+        //
+        // A path's acknowledgements are sent on that path whenever it can carry them, and only
+        // piggybacked onto another path's packet when it cannot (not yet validated, abandoned,
+        // no CIDs). Sending path B's PATH_ACKs on path A ties B's loss recovery and congestion
+        // feedback to A's fate: if A degrades, B's RTT and delivery-rate estimates inflate with
+        // A's queue even though B itself is healthy.
         if !scheduling_info.is_abandoned && scheduling_info.may_send_data {
             for path_id in space
                 .number_spaces
                 .iter_mut()
-                .filter(|(_, pns)| pns.pending_acks.can_send())
+                .filter(|(pid, pns)| {
+                    pns.pending_acks.can_send() && !other_own_carriers.contains(pid)
+                })
                 .map(|(&path_id, _)| path_id)
                 .collect::<Vec<_>>()
             {
@@ -6946,6 +6979,22 @@ impl Connection {
         }
     }
 
+    /// Whether path `p` can currently carry its own acknowledgements: known CIDs, not abandoned
+    /// and validated. Acks for such a path are sent on the path itself rather than piggybacked on
+    /// whichever path happens to transmit first (see the ACK section of `populate_packet`).
+    ///
+    /// Takes the fields rather than `&self` so callers can hold `&mut self.spaces` meanwhile.
+    fn path_carries_own_acks(
+        paths: &BTreeMap<PathId, PathState>,
+        remote_cids: &FxHashMap<PathId, CidQueue>,
+        abandoned: &AbandonedPaths,
+        p: PathId,
+    ) -> bool {
+        remote_cids.contains_key(&p)
+            && !abandoned.contains(&p)
+            && paths.get(&p).is_some_and(|path| path.data.validated)
+    }
+
     /// Whether we have **on-path** 1-RTT data to send.
     ///
     /// This checks for frames that can only be sent in the data space (1-RTT):

noq-proto/src/connection/spaces.rs

@@ -141,11 +141,17 @@ impl PacketSpace {
     ///
     /// [`Connection::can_send_1rtt`]: super::Connection::can_send_1rtt
     /// [`Connection::space_can_send`]: super::Connection::space_can_send
-    pub(super) fn can_send(&self, path_id: PathId, streams: &StreamsState) -> SendableFrames {
-        let acks = self
-            .number_spaces
-            .values()
-            .any(|pns| pns.pending_acks.can_send());
+    /// `ack_on_own_path(p)` says whether path `p` can currently carry its own acknowledgements;
+    /// acks for such paths are only sent on that path (see `Connection::ack_carrier_paths`).
+    pub(super) fn can_send(
+        &self,
+        path_id: PathId,
+        streams: &StreamsState,
+        ack_on_own_path: impl Fn(PathId) -> bool,
+    ) -> SendableFrames {
+        let acks = self.number_spaces.iter().any(|(&pid, pns)| {
+            pns.pending_acks.can_send() && (pid == path_id || !ack_on_own_path(pid))
+        });
         let space_specific = self.number_spaces.get(&path_id).is_some_and(|s| {
             s.pending_ping || s.pending_immediate_ack || !s.pending_path_responses.is_empty()
         });

noq-proto/src/tests/multipath.rs

@@ -2299,3 +2299,55 @@ fn regression_discarded_path_stats_are_up_to_date() -> TestResult {
 
     Ok(())
 }
+
+/// A path's acknowledgements are sent on that path, not on whichever path happens to transmit
+/// first. With two validated paths and data flowing on both, the server must transmit PATH_ACK
+/// frames on path 1 as well as on path 0; before the fix every path's acks were coalesced into
+/// path 0's packets, which ties path 1's loss recovery and RTT/delivery samples to path 0's fate.
+#[test]
+fn path_acks_sent_on_own_path() -> TestResult {
+    let _guard = subscribe();
+    let mut pair = ConnPair::builder().enable_multipath().connect();
+    let server_addr = pair.routes.public_server_addr();
+    let path_1 = pair.open_path(
+        Client,
+        FourTuple::from_remote(server_addr),
+        PathStatus::Available,
+    )?;
+    pair.drive();
+    while pair.poll(Client).is_some() {}
+    while pair.poll(Server).is_some() {}
+
+    let before_0 = pair
+        .path_stats(Server, PathId::ZERO)
+        .unwrap()
+        .frame_tx
+        .path_acks;
+    let before_1 = pair.path_stats(Server, path_1).unwrap().frame_tx.path_acks;
+
+    // Client data on both paths: path 0 first, then path 1 as the only available path.
+    let s = pair.streams(Client).open(Dir::Uni).unwrap();
+    pair.send_stream(Client, s).write(&[1u8; 20_000]).unwrap();
+    pair.drive();
+    pair.set_path_status(Client, PathId::ZERO, PathStatus::Backup)?;
+    pair.send_stream(Client, s).write(&[2u8; 20_000]).unwrap();
+    pair.drive();
+    pair.set_path_status(Client, PathId::ZERO, PathStatus::Available)?;
+    pair.send_stream(Client, s).write(&[3u8; 20_000]).unwrap();
+    pair.drive();
+
+    let acks_0 = pair
+        .path_stats(Server, PathId::ZERO)
+        .unwrap()
+        .frame_tx
+        .path_acks
+        - before_0;
+    let acks_1 = pair.path_stats(Server, path_1).unwrap().frame_tx.path_acks - before_1;
+    info!("server PATH_ACKs transmitted: path 0 = {acks_0}, path 1 = {acks_1}");
+    assert!(acks_0 > 0, "path 0 carried data, its acks go on path 0");
+    assert!(
+        acks_1 > 0,
+        "path 1 carried data, its acks must go on path 1 (got {acks_1}; path 0 carried {acks_0})"
+    );
+    Ok(())
+}