spurious loss detection and recovery

Author: stormshield-fabs

quinn-proto/src/congestion.rs

@@ -56,9 +56,16 @@ pub trait Controller: Send + Sync {
         now: Instant,
         sent: Instant,
         is_persistent_congestion: bool,
+        is_ecn: bool,
         lost_bytes: u64,
     );
 
+    /// Packets were incorrectly deemed lost
+    ///
+    /// This function is called when all packets that were deemed lost (for instance because
+    /// of packet reordering) are acknowledged after the congestion event was raised.
+    fn on_spurious_congestion_event(&mut self) {}
+
     /// The known MTU for the current network path has been updated
     fn on_mtu_update(&mut self, new_mtu: u16);
 

quinn-proto/src/congestion/bbr/mod.rs

@@ -465,6 +465,7 @@ impl Controller for Bbr {
         _now: Instant,
         _sent: Instant,
         _is_persistent_congestion: bool,
+        _is_ecn: bool,
         lost_bytes: u64,
     ) {
         self.loss_state.lost_bytes += lost_bytes;

quinn-proto/src/congestion/cubic.rs

@@ -75,8 +75,10 @@ impl State {
 #[derive(Debug, Clone)]
 pub struct Cubic {
     config: Arc<CubicConfig>,
-    state: State,
     current_mtu: u64,
+    state: State,
+    /// Copy of the controller state to restore when a spurious congestion event is detected.
+    pre_congestion_state: Option<State>,
 }
 
 impl Cubic {
@@ -89,6 +91,7 @@ impl Cubic {
                 ..Default::default()
             },
             current_mtu: current_mtu as u64,
+            pre_congestion_state: None,
             config,
         }
     }
@@ -176,6 +179,7 @@ impl Controller for Cubic {
         now: Instant,
         sent: Instant,
         is_persistent_congestion: bool,
+        is_ecn: bool,
         _lost_bytes: u64,
     ) {
         if self
@@ -187,6 +191,11 @@ impl Controller for Cubic {
             return;
         }
 
+        // Save state in case this event ends up being spurious
+        if !is_ecn {
+            self.pre_congestion_state = Some(self.state.clone());
+        }
+
         self.state.recovery_start_time = Some(now);
 
         // Fast convergence
@@ -221,6 +230,14 @@ impl Controller for Cubic {
         }
     }
 
+    fn on_spurious_congestion_event(&mut self) {
+        if let Some(prior_state) = self.pre_congestion_state.take() {
+            if self.state.window < prior_state.window {
+                self.state = prior_state;
+            }
+        }
+    }
+
     fn on_mtu_update(&mut self, new_mtu: u16) {
         self.current_mtu = new_mtu as u64;
         self.state.window = self.state.window.max(self.minimum_window());

quinn-proto/src/congestion/new_reno.rs

@@ -87,6 +87,7 @@ impl Controller for NewReno {
         now: Instant,
         sent: Instant,
         is_persistent_congestion: bool,
+        _is_ecn: bool,
         _lost_bytes: u64,
     ) {
         if sent <= self.recovery_start_time {

quinn-proto/src/connection/mod.rs

@@ -22,6 +22,7 @@ use crate::{
     cid_queue::CidQueue,
     coding::BufMutExt,
     config::{ServerConfig, TransportConfig},
+    connection::spaces::LostPacket,
     crypto::{self, KeyPair, Keys, PacketKey},
     frame::{self, Close, Datagram, FrameStruct, NewConnectionId, NewToken},
     packet::{
@@ -1461,6 +1462,10 @@ impl Connection {
             }
         };
 
+        if self.detect_spurious_loss(&ack, space) {
+            self.path.congestion.on_spurious_congestion_event();
+        }
+
         // Avoid DoS from unreasonably huge ack ranges by filtering out just the new acks.
         let mut newly_acked = ArrayRangeSet::new();
         for range in ack.iter() {
@@ -1555,6 +1560,43 @@ impl Connection {
         Ok(())
     }
 
+    fn detect_spurious_loss(&mut self, ack: &frame::Ack, space: SpaceId) -> bool {
+        let lost_packets = &mut self.spaces[space].lost_packets;
+
+        if lost_packets.is_empty() {
+            return false;
+        }
+
+        for range in ack.iter() {
+            let spurious_losses: Vec<u64> = lost_packets
+                .range(range.clone())
+                .map(|(pn, _info)| pn)
+                .copied()
+                .collect();
+
+            for pn in spurious_losses {
+                lost_packets.remove(&pn);
+            }
+        }
+
+        // If this ACK frame acknowledged all deemed lost packets,
+        // then we have raised a spurious congestion event in the past.
+        // We cannot conclude when there are remaining packets,
+        // but future ACK frames might indicate a spurious loss detection.
+        lost_packets.is_empty()
+    }
+
+    /// Drain lost packets that we reasonably think will never arrive
+    ///
+    /// The current criterion is copied from `msquic`:
+    /// discard packets that were sent earlier than 2 probe timeouts ago.
+    fn drain_lost_packets(&mut self, now: Instant, space: SpaceId) {
+        let two_pto = 2 * self.path.rtt.pto_base();
+
+        let lost_packets = &mut self.spaces[space].lost_packets;
+        lost_packets.retain(|_pn, info| now.saturating_duration_since(info.time_sent) <= two_pto);
+    }
+
     /// Process a new ECN block from an in-order ACK
     fn process_ecn(
         &mut self,
@@ -1577,7 +1619,7 @@ impl Connection {
                 self.stats.path.congestion_events += 1;
                 self.path
                     .congestion
-                    .on_congestion_event(now, largest_sent_time, false, 0);
+                    .on_congestion_event(now, largest_sent_time, false, true, 0);
             }
         }
     }
@@ -1735,6 +1777,8 @@ impl Connection {
             prev_packet = Some(packet);
         }
 
+        self.drain_lost_packets(now, pn_space);
+
         // OnPacketsLost
         if let Some(largest_lost) = lost_packets.last().cloned() {
             let old_bytes_in_flight = self.path.in_flight.bytes;
@@ -1756,12 +1800,20 @@ impl Connection {
                     now,
                     self.orig_rem_cid,
                 );
+
                 self.remove_in_flight(&info);
                 for frame in info.stream_frames {
                     self.streams.retransmit(frame);
                 }
                 self.spaces[pn_space].pending |= info.retransmits;
                 self.path.mtud.on_non_probe_lost(packet, info.size);
+
+                self.spaces[pn_space].lost_packets.insert(
+                    packet,
+                    LostPacket {
+                        time_sent: info.time_sent,
+                    },
+                );
             }
 
             if self.path.mtud.black_hole_detected(now) {
@@ -1783,6 +1835,7 @@ impl Connection {
                     now,
                     largest_lost_sent,
                     in_persistent_congestion,
+                    false,
                     size_of_lost_packets,
                 );
             }

quinn-proto/src/connection/spaces.rs

@@ -39,6 +39,9 @@ pub(super) struct PacketSpace {
     /// Transmitted but not acked
     // We use a BTreeMap here so we can efficiently query by range on ACK and for loss detection
     pub(super) sent_packets: BTreeMap<u64, SentPacket>,
+    /// Packets that were deemed lost
+    // Older packets are regularly removed in `Connection::drain_lost_packets`.
+    pub(super) lost_packets: BTreeMap<u64, LostPacket>,
     /// Number of explicit congestion notification codepoints seen on incoming packets
     pub(super) ecn_counters: frame::EcnCounts,
     /// Recent ECN counters sent by the peer in ACK frames
@@ -84,6 +87,7 @@ impl PacketSpace {
             largest_ack_eliciting_sent: 0,
             unacked_non_ack_eliciting_tail: 0,
             sent_packets: BTreeMap::new(),
+            lost_packets: BTreeMap::new(),
             ecn_counters: frame::EcnCounts::ZERO,
             ecn_feedback: frame::EcnCounts::ZERO,
 
@@ -302,6 +306,13 @@ pub(super) struct SentPacket {
     pub(super) stream_frames: frame::StreamMetaVec,
 }
 
+/// Represents one or more packets that are deemed lost.
+#[derive(Debug)]
+pub(super) struct LostPacket {
+    /// The time the packet was sent.
+    pub(super) time_sent: Instant,
+}
+
 /// Retransmittable data queue
 #[allow(unreachable_pub)] // fuzzing only
 #[derive(Debug, Default, Clone)]