fix: No time-based loss detection before largest_acked & fixes

neqo-transport/src/cc/classic_cc.rs

@@ -337,6 +337,41 @@ impl<T: WindowAdjustment> CongestionControl for ClassicCongestionControl<T> {
         congestion || persistent_congestion
     }
 
+    /// Handle a congestion event.
+    /// Returns true if this was a true congestion event.
+    fn on_congestion_event(&mut self, last_packet: &SentPacket) -> bool {
+        // Start a new congestion event if lost or ECN CE marked packet was sent
+        // after the start of the previous congestion recovery period.
+        if !self.after_recovery_start(last_packet) {
+            return false;
+        }
+
+        let (cwnd, acked_bytes) = self.cc_algorithm.reduce_cwnd(
+            self.congestion_window,
+            self.acked_bytes,
+            self.max_datagram_size(),
+        );
+        self.congestion_window = max(cwnd, self.cwnd_min());
+        self.acked_bytes = acked_bytes;
+        self.ssthresh = self.congestion_window;
+        qdebug!(
+            [self],
+            "Cong event -> recovery; cwnd {}, ssthresh {}",
+            self.congestion_window,
+            self.ssthresh
+        );
+        qlog::metrics_updated(
+            &self.qlog,
+            &[
+                QlogMetric::CongestionWindow(self.congestion_window),
+                QlogMetric::SsThresh(self.ssthresh),
+                QlogMetric::InRecovery(true),
+            ],
+        );
+        self.set_state(State::RecoveryStart);
+        true
+    }
+
     /// Report received ECN CE mark(s) to the congestion controller as a
     /// congestion event.
     ///
@@ -537,41 +572,6 @@ impl<T: WindowAdjustment> ClassicCongestionControl<T> {
         !self.state.transient() && self.recovery_start.map_or(true, |pn| packet.pn() >= pn)
     }
 
-    /// Handle a congestion event.
-    /// Returns true if this was a true congestion event.
-    fn on_congestion_event(&mut self, last_packet: &SentPacket) -> bool {
-        // Start a new congestion event if lost or ECN CE marked packet was sent
-        // after the start of the previous congestion recovery period.
-        if !self.after_recovery_start(last_packet) {
-            return false;
-        }
-
-        let (cwnd, acked_bytes) = self.cc_algorithm.reduce_cwnd(
-            self.congestion_window,
-            self.acked_bytes,
-            self.max_datagram_size(),
-        );
-        self.congestion_window = max(cwnd, self.cwnd_min());
-        self.acked_bytes = acked_bytes;
-        self.ssthresh = self.congestion_window;
-        qdebug!(
-            [self],
-            "Cong event -> recovery; cwnd {}, ssthresh {}",
-            self.congestion_window,
-            self.ssthresh
-        );
-        qlog::metrics_updated(
-            &self.qlog,
-            &[
-                QlogMetric::CongestionWindow(self.congestion_window),
-                QlogMetric::SsThresh(self.ssthresh),
-                QlogMetric::InRecovery(true),
-            ],
-        );
-        self.set_state(State::RecoveryStart);
-        true
-    }
-
     fn app_limited(&self) -> bool {
         if self.bytes_in_flight >= self.congestion_window {
             false

neqo-transport/src/cc/mod.rs

@@ -62,6 +62,11 @@ pub trait CongestionControl: Display + Debug {
         lost_packets: &[SentPacket],
     ) -> bool;
 
+    /// Initiate a congestion response.
+    ///
+    /// Returns true if the congestion window was reduced.
+    fn on_congestion_event(&mut self, last_packet: &SentPacket) -> bool;
+
     /// Returns true if the congestion window was reduced.
     fn on_ecn_ce_received(&mut self, largest_acked_pkt: &SentPacket) -> bool;
 

neqo-transport/src/connection/tests/handshake.rs

@@ -18,9 +18,7 @@ use neqo_crypto::{
 };
 #[cfg(not(feature = "disable-encryption"))]
 use test_fixture::datagram;
-use test_fixture::{
-    assertions, assertions::assert_coalesced_0rtt, fixture_init, now, split_datagram, DEFAULT_ADDR,
-};
+use test_fixture::{assertions, fixture_init, now, split_datagram, DEFAULT_ADDR};
 
 use super::{
     super::{Connection, Output, State},
@@ -401,10 +399,10 @@ fn reorder_05rtt_with_0rtt() {
     // Now PTO at the client and cause the server to re-send handshake packets.
     now += AT_LEAST_PTO;
     let c3 = client.process(None, now).dgram();
-    assert_coalesced_0rtt(c3.as_ref().unwrap());
 
     now += RTT / 2;
     let s3 = server.process(c3.as_ref(), now).dgram().unwrap();
+    assertions::assert_no_1rtt(&s3[..]);
 
     // The client should be able to process the 0.5 RTT now.
     // This should contain an ACK, so we are processing an ACK from the past.

neqo-transport/src/connection/tests/recovery.rs

@@ -174,10 +174,12 @@ fn pto_initial() {
     assert_eq!(delay, INITIAL_PTO);
 
     // Resend initial after PTO.
+    let cwnd_prior: usize = cwnd(&client);
     now += delay;
     let pkt2 = client.process(None, now).dgram();
     assert!(pkt2.is_some());
     assert_eq!(pkt2.unwrap().len(), client.plpmtu());
+    assert_eq!(cwnd_prior, 2 * cwnd(&client)); // cwnd has halved
 
     let delay = client.process(None, now).callback();
     // PTO has doubled.
@@ -294,21 +296,16 @@ fn pto_handshake_complete() {
     // Check that the other packets (pkt2, pkt3) are Handshake packets.
     // The server discarded the Handshake keys already, therefore they are dropped.
     // Note that these don't include 1-RTT packets, because 1-RTT isn't send on PTO.
-    let (pkt2_hs, pkt2_1rtt) = split_datagram(&pkt2.unwrap());
-    assert_handshake(&pkt2_hs);
-    assert!(pkt2_1rtt.is_some());
     let dropped_before1 = server.stats().dropped_rx;
     let server_frames = server.stats().frame_rx.all;
-    server.process_input(&pkt2_hs, now);
+    server.process_input(&pkt2.unwrap(), now);
     assert_eq!(1, server.stats().dropped_rx - dropped_before1);
     assert_eq!(server.stats().frame_rx.all, server_frames);
 
-    server.process_input(&pkt2_1rtt.unwrap(), now);
-    let server_frames2 = server.stats().frame_rx.all;
     let dropped_before2 = server.stats().dropped_rx;
     server.process_input(&pkt3_hs, now);
     assert_eq!(1, server.stats().dropped_rx - dropped_before2);
-    assert_eq!(server.stats().frame_rx.all, server_frames2);
+    assert_eq!(server.stats().frame_rx.all, server_frames);
 
     now += HALF_RTT;
 

neqo-transport/src/path.rs

@@ -1018,6 +1018,13 @@ impl Path {
         }
     }
 
+    /// Initiate a congestion response.
+    ///
+    /// Returns true if the congestion window was reduced.
+    pub fn on_congestion_event(&mut self, last_packet: &SentPacket) -> bool {
+        self.sender.on_congestion_event(last_packet)
+    }
+
     /// Determine whether we should be setting a PTO for this path. This is true when either the
     /// path is valid or when there is enough remaining in the amplification limit to fit a
     /// full-sized path (i.e., the path MTU).

neqo-transport/src/recovery/mod.rs

@@ -321,13 +321,15 @@ impl LossRecoverySpace {
         );
         self.first_ooo_time = None;
 
-        let largest_acked = self.largest_acked;
+        let Some(largest_acked) = self.largest_acked else {
+            return;
+        };
 
         for packet in self
             .sent_packets
             .iter_mut()
             // BTreeMap iterates in order of ascending PN
-            .take_while(|p| p.pn() < largest_acked.unwrap_or(PacketNumber::MAX))
+            .take_while(|p| p.pn() < largest_acked)
         {
             // Packets sent before now - loss_delay are deemed lost.
             if packet.time_sent() + loss_delay <= now {
@@ -337,17 +339,15 @@ impl LossRecoverySpace {
                     packet.time_sent(),
                     loss_delay
                 );
-            } else if largest_acked >= Some(packet.pn() + PACKET_THRESHOLD) {
+            } else if largest_acked >= packet.pn() + PACKET_THRESHOLD {
                 qtrace!(
                     "lost={}, is >= {} from largest acked {:?}",
                     packet.pn(),
                     PACKET_THRESHOLD,
                     largest_acked
                 );
             } else {
-                if largest_acked.is_some() {
-                    self.first_ooo_time = Some(packet.time_sent());
-                }
+                self.first_ooo_time = Some(packet.time_sent());
                 // No more packets can be declared lost after this one.
                 break;
             };
@@ -840,34 +840,49 @@ impl LossRecovery {
     /// When it has, mark a few packets as "lost" for the purposes of having frames
     /// regenerated in subsequent packets.  The packets aren't truly lost, so
     /// we have to clone the `SentPacket` instance.
-    fn maybe_fire_pto(&mut self, rtt: &RttEstimate, now: Instant, lost: &mut Vec<SentPacket>) {
+    fn maybe_fire_pto(&mut self, path: &PathRef, now: Instant, lost: &mut Vec<SentPacket>) {
         let mut pto_space = None;
         // The spaces in which we will allow probing.
         let mut allow_probes = PacketNumberSpaceSet::default();
         for pn_space in PacketNumberSpace::iter() {
-            if let Some(t) = self.pto_time(rtt, *pn_space) {
-                allow_probes[*pn_space] = true;
-                if t <= now {
-                    qdebug!([self], "PTO timer fired for {}", pn_space);
-                    let space = self.spaces.get_mut(*pn_space).unwrap();
-                    lost.extend(
-                        space
-                            .pto_packets(PtoState::pto_packet_count(
-                                *pn_space,
-                                self.stats.borrow().packets_rx,
-                            ))
-                            .cloned(),
-                    );
-
-                    pto_space = pto_space.or(Some(*pn_space));
-                }
+            if self
+                .pto_time(path.borrow().rtt(), *pn_space)
+                .map_or(true, |t| now < t)
+            {
+                continue;
             }
+            allow_probes[*pn_space] = true;
+            qdebug!([self], "PTO timer fired for {}", pn_space);
+            let space = self.spaces.get_mut(*pn_space).unwrap();
+            lost.extend(
+                space
+                    .pto_packets(PtoState::pto_packet_count(
+                        *pn_space,
+                        self.stats.borrow().packets_rx,
+                    ))
+                    .cloned(),
+            );
+
+            pto_space = pto_space.or(Some(*pn_space));
         }
 
         // This has to happen outside the loop. Increasing the PTO count here causes the
         // pto_time to increase which might cause PTO for later packet number spaces to not fire.
         if let Some(pn_space) = pto_space {
             qtrace!([self], "PTO {}, probing {:?}", pn_space, allow_probes);
+            // Packets are only declared as lost, relative to the
+            // `largest_acked`. If we hit a PTO while we don't have a
+            // largest_acked yet, also do a congestion control reaction (because
+            // otherwise none would happen).
+            if self
+                .spaces
+                .get(pn_space)
+                .map_or(false, |space| space.largest_acked.is_none())
+            {
+                if let Some(last) = lost.last() {
+                    path.borrow_mut().on_congestion_event(last);
+                }
+            }
             self.fire_pto(pn_space, allow_probes);
         }
     }
@@ -898,7 +913,7 @@ impl LossRecovery {
         }
         self.stats.borrow_mut().lost += lost_packets.len();
 
-        self.maybe_fire_pto(primary_path.borrow().rtt(), now, &mut lost_packets);
+        self.maybe_fire_pto(primary_path, now, &mut lost_packets);
         lost_packets
     }
 
@@ -967,7 +982,6 @@ mod tests {
         ecn::EcnCount,
         packet::{PacketNumber, PacketType},
         path::{Path, PathRef},
-        rtt::RttEstimate,
         stats::{Stats, StatsCell},
     };
 
@@ -978,8 +992,8 @@ mod tests {
 
     const ON_SENT_SIZE: usize = 100;
     /// An initial RTT for using with `setup_lr`.
-    const TEST_RTT: Duration = ms(80);
-    const TEST_RTTVAR: Duration = ms(40);
+    const TEST_RTT: Duration = ms(7000);
+    const TEST_RTTVAR: Duration = ms(3500);
 
     struct Fixture {
         lr: LossRecovery,
@@ -1050,6 +1064,7 @@ mod tests {
                 ConnectionIdEntry::new(0, ConnectionId::from(&[1, 2, 3]), [0; 16]),
             );
             path.set_primary(true);
+            path.rtt_mut().set_initial(TEST_RTT);
             Self {
                 lr: LossRecovery::new(StatsCell::default(), FAST_PTO_SCALE),
                 path: Rc::new(RefCell::new(path)),
@@ -1540,7 +1555,11 @@ mod tests {
 
         // Expiring state after the PTO on the ApplicationData space has
         // expired should result in setting a PTO state.
-        let default_pto = RttEstimate::default().pto(PacketNumberSpace::ApplicationData);
+        let default_pto = lr
+            .path
+            .borrow()
+            .rtt()
+            .pto(PacketNumberSpace::ApplicationData);
         let expected_pto = pn_time(2) + default_pto;
         lr.discard(PacketNumberSpace::Handshake, expected_pto);
         let profile = lr.send_profile(now());
@@ -1572,7 +1591,7 @@ mod tests {
             ON_SENT_SIZE,
         ));
 
-        let handshake_pto = RttEstimate::default().pto(PacketNumberSpace::Handshake);
+        let handshake_pto = lr.path.borrow().rtt().pto(PacketNumberSpace::Handshake);
         let expected_pto = now() + handshake_pto;
         assert_eq!(lr.pto_time(PacketNumberSpace::Initial), Some(expected_pto));
         let profile = lr.send_profile(now());

neqo-transport/src/sender.rs

@@ -113,6 +113,13 @@ impl PacketSender {
         self.maybe_update_pacer_mtu();
     }
 
+    /// Initiate a congestion response.
+    ///
+    /// Returns true if the congestion window was reduced.
+    pub fn on_congestion_event(&mut self, last_packet: &SentPacket) -> bool {
+        self.cc.on_congestion_event(last_packet)
+    }
+
     /// Called when packets are lost.  Returns true if the congestion window was reduced.
     pub fn on_packets_lost(
         &mut self,

test-fixture/src/lib.rs

@@ -53,7 +53,7 @@ pub fn fixture_init() {
 
 // This needs to be > 2ms to avoid it being rounded to zero.
 // NSS operates in milliseconds and halves any value it is provided.
-pub const ANTI_REPLAY_WINDOW: Duration = Duration::from_millis(10);
+pub const ANTI_REPLAY_WINDOW: Duration = Duration::from_millis(1000);
 
 /// A baseline time for all tests.  This needs to be earlier than what `now()` produces
 /// because of the need to have a span of time elapse for anti-replay purposes.