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

neqo-transport/src/cc/classic_cc.rs

@@ -347,8 +347,7 @@ impl<T: WindowAdjustment> CongestionControl for ClassicCongestionControl<T> {
 
     fn discard(&mut self, pkt: &SentPacket) {
         if pkt.cc_outstanding() {
-            assert!(self.bytes_in_flight >= pkt.len());
-            self.bytes_in_flight -= pkt.len();
+            self.bytes_in_flight = self.bytes_in_flight.saturating_sub(pkt.len());
             qlog::metrics_updated(
                 &self.qlog,
                 &[QlogMetric::BytesInFlight(self.bytes_in_flight)],

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/recovery/mod.rs

@@ -327,7 +327,7 @@
             .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| Some(p.pn()) < largest_acked)
         {
             // Packets sent before now - loss_delay are deemed lost.
             if packet.time_sent() + loss_delay <= now {
@@ -835,14 +835,14 @@
     /// 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 let Some(t) = self.pto_time(path.borrow().rtt(), *pn_space) {
                 if t <= now {
+                    allow_probes[*pn_space] = true;
                     qdebug!([self], "PTO timer fired for {}", pn_space);
                     let space = self.spaces.get_mut(*pn_space).unwrap();
                     lost.extend(
@@ -863,6 +863,19 @@
         // 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);
+            // 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 let Some(space) = self.spaces.get(pn_space) {
+                if space.largest_acked.is_none() {
+                    path.borrow_mut().on_packets_lost(
+                        space.largest_acked_sent_time,
+                        pn_space,
+                        lost,
+                        &mut self.stats.borrow_mut(),
+                        now,
+                    );
+                }
+            }
             self.fire_pto(pn_space, allow_probes);
         }
     }
@@ -893,7 +906,7 @@
         }
         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
     }
 
@@ -962,7 +975,6 @@
         ecn::EcnCount,
         packet::{PacketNumber, PacketType},
         path::{Path, PathRef},
-        rtt::RttEstimate,
         stats::{Stats, StatsCell},
     };
 
@@ -973,8 +985,8 @@
 
     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,
@@ -1045,6 +1057,7 @@
                 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)),
@@ -1535,7 +1548,11 @@
 
         // 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());
@@ -1567,7 +1584,7 @@
             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());

test-fixture/src/lib.rs

@@ -52,7 +52,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.