feat: More ECN and DSCP stats (#2505)

* feat: More ECN stats

Inspired by @martinduke's [IETF presentation on
ECN](https://datatracker.ietf.org/meeting/122/materials/slides-122-tsvwg-udp-ecn-00),
add some more stats around what packet types we send and receive with different
ECN codepoints.

This will require additional glue code to ship these stats back as telemetry.

* fmt

* Improve coverage?

* Record ECN mark transitions

* Tweak output

* Output tweaks

* Add DSCP stats

* Update neqo-transport/src/stats.rs

Co-authored-by: Max Inden <[email protected]>
Signed-off-by: Lars Eggert <[email protected]>

* Fix rebase

* Update neqo-transport/src/stats.rs

Co-authored-by: Max Inden <[email protected]>
Signed-off-by: Lars Eggert <[email protected]>

* Suggestions from @mxinden

* Fixes and simplifications

* Update neqo-transport/src/connection/mod.rs

Co-authored-by: Max Inden <[email protected]>
Signed-off-by: Lars Eggert <[email protected]>

* Fix

* Update neqo-transport/src/connection/mod.rs

Co-authored-by: Max Inden <[email protected]>
Signed-off-by: Lars Eggert <[email protected]>

* Fixes

* Fix DSCP tracking and add some tests

---------

Signed-off-by: Lars Eggert <[email protected]>
Co-authored-by: Max Inden <[email protected]>

Author: larseggert

neqo-common/src/tos.rs

@@ -7,12 +7,12 @@
 use std::fmt::Debug;
 
 use enum_map::Enum;
-use strum::FromRepr;
+use strum::{EnumIter, FromRepr};
 
 /// ECN (Explicit Congestion Notification) codepoints mapped to the
 /// lower 2 bits of the TOS field.
 /// <https://www.iana.org/assignments/dscp-registry/dscp-registry.xhtml>
-#[derive(Copy, Clone, PartialEq, Eq, Enum, Default, Debug, FromRepr)]
+#[derive(Copy, Clone, PartialEq, Eq, Enum, Default, Debug, FromRepr, EnumIter)]
 #[repr(u8)]
 pub enum IpTosEcn {
     #[default]

neqo-transport/src/connection/mod.rs

@@ -20,7 +20,7 @@ use std::{
 
 use neqo_common::{
     event::Provider as EventProvider, hex, hex_snip_middle, hrtime, qdebug, qerror, qinfo,
-    qlog::NeqoQlog, qtrace, qwarn, Datagram, Decoder, Encoder, IpTos, Role,
+    qlog::NeqoQlog, qtrace, qwarn, Datagram, Decoder, Encoder, IpTos, IpTosEcn, Role,
 };
 use neqo_crypto::{
     agent::CertificateInfo, Agent, AntiReplay, AuthenticationStatus, Cipher, Client, Group,
@@ -1528,20 +1528,32 @@ impl Connection {
     }
 
     /// After a Initial, Handshake, `ZeroRtt`, or Short packet is successfully processed.
+    #[expect(clippy::too_many_arguments, reason = "Yes, but they're needed.")]
     fn postprocess_packet(
         &mut self,
         path: &PathRef,
         tos: IpTos,
         remote: SocketAddr,
         packet: &PublicPacket,
+        packet_number: PacketNumber,
         migrate: bool,
         now: Instant,
     ) {
+        let ecn_mark = IpTosEcn::from(tos);
+        let mut stats = self.stats.borrow_mut();
+        stats.ecn_rx[packet.packet_type()] += ecn_mark;
+        if let Some(last_ecn_mark) = stats.ecn_last_mark.filter(|&last_ecn_mark| {
+            last_ecn_mark != ecn_mark && stats.ecn_rx_transition[last_ecn_mark][ecn_mark].is_none()
+        }) {
+            stats.ecn_rx_transition[last_ecn_mark][ecn_mark] =
+                Some((packet.packet_type(), packet_number));
+        }
+
+        stats.ecn_last_mark = Some(ecn_mark);
+        drop(stats);
         let space = PacketNumberSpace::from(packet.packet_type());
         if let Some(space) = self.acks.get_mut(space) {
-            let space_ecn_marks = space.ecn_marks();
-            *space_ecn_marks += tos.into();
-            self.stats.borrow_mut().ecn_rx = *space_ecn_marks;
+            *space.ecn_marks() += ecn_mark;
         } else {
             qtrace!("Not tracking ECN for dropped packet number space");
         }
@@ -1627,6 +1639,7 @@ impl Connection {
                         break;
                     }
                 };
+            self.stats.borrow_mut().dscp_rx[tos.into()] += 1;
             match self.preprocess_packet(&packet, path, dcid.as_ref(), now)? {
                 PreprocessResult::Continue => (),
                 PreprocessResult::Next => break,
@@ -1639,6 +1652,7 @@ impl Connection {
             match packet.decrypt(&mut self.crypto.states, now + pto) {
                 Ok(payload) => {
                     // OK, we have a valid packet.
+                    let pn = payload.pn();
                     self.idle_timeout.on_packet_received(now);
                     self.log_packet(
                         packet::MetaData::new_in(path, tos, packet_len, &payload),
@@ -1657,14 +1671,14 @@ impl Connection {
 
                     let space = PacketNumberSpace::from(payload.packet_type());
                     if let Some(space) = self.acks.get_mut(space) {
-                        if space.is_duplicate(payload.pn()) {
-                            qdebug!("Duplicate packet {space}-{}", payload.pn());
+                        if space.is_duplicate(pn) {
+                            qdebug!("Duplicate packet {space}-{}", pn);
                             self.stats.borrow_mut().dups_rx += 1;
                         } else {
                             match self.process_packet(path, &payload, now) {
                                 Ok(migrate) => {
                                     self.postprocess_packet(
-                                        path, tos, remote, &packet, migrate, now,
+                                        path, tos, remote, &packet, pn, migrate, now,
                                     );
                                 }
                                 Err(e) => {
@@ -2493,14 +2507,17 @@ impl Connection {
                 continue;
             }
 
+            // If we don't have a TOS for this UDP datagram yet (i.e. `tos` is `None`), get it,
+            // adding a `RecoveryToken::EcnEct0` to `tokens` in case of loss.
+            let tos = packet_tos.get_or_insert_with(|| path.borrow().tos(&mut tokens));
             self.log_packet(
                 packet::MetaData::new_out(
                     path,
                     pt,
                     pn,
                     builder.len() + aead_expansion,
                     &builder.as_ref()[payload_start..],
-                    *packet_tos.get_or_insert(path.borrow().tos(&mut tokens)),
+                    *tos,
                 ),
                 now,
             );
@@ -2540,6 +2557,12 @@ impl Connection {
                 self.loss_recovery.on_packet_sent(path, sent, now);
             }
 
+            // Track which packet types are sent with which ECN codepoints. For
+            // coalesced packets, this increases the counts for each packet type
+            // contained in the coalesced packet. This is per Section 13.4.1 of
+            // RFC 9000.
+            self.stats.borrow_mut().ecn_tx[pt] += IpTosEcn::from(*tos);
+
             if space == PacketNumberSpace::Handshake
                 && self.role == Role::Server
                 && self.state == State::Confirmed

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

@@ -7,6 +7,7 @@
 use std::time::Duration;
 
 use neqo_common::{Datagram, IpTos, IpTosEcn};
+use strum::IntoEnumIterator as _;
 use test_fixture::{
     assertions::{assert_v4_path, assert_v6_path},
     fixture_init, now, DEFAULT_ADDR_V4,
@@ -20,6 +21,7 @@ use crate::{
         send_with_modifier_and_receive, DEFAULT_RTT,
     },
     ecn,
+    packet::PacketType,
     path::MAX_PATH_PROBES,
     ConnectionId, ConnectionParameters, StreamType,
 };
@@ -141,6 +143,44 @@ fn migration_delay_to_ecn_blackhole() {
     }
 }
 
+#[test]
+fn debug() {
+    let stats = crate::Stats::default();
+    assert_eq!(
+        format!("{stats:?}"),
+        "stats for\u{0020}
+  rx: 0 drop 0 dup 0 saved 0
+  tx: 0 lost 0 lateack 0 ptoack 0
+  pmtud: 0 sent 0 acked 0 lost 0 change 0 iface_mtu 0 pmtu
+  resumed: false
+  frames rx:
+    crypto 0 done 0 token 0 close 0
+    ack 0 (max 0) ping 0 padding 0
+    stream 0 reset 0 stop 0
+    max: stream 0 data 0 stream_data 0
+    blocked: stream 0 data 0 stream_data 0
+    datagram 0
+    ncid 0 rcid 0 pchallenge 0 presponse 0
+    ack_frequency 0
+  frames tx:
+    crypto 0 done 0 token 0 close 0
+    ack 0 (max 0) ping 0 padding 0
+    stream 0 reset 0 stop 0
+    max: stream 0 data 0 stream_data 0
+    blocked: stream 0 data 0 stream_data 0
+    datagram 0
+    ncid 0 rcid 0 pchallenge 0 presponse 0
+    ack_frequency 0
+  ecn:
+    tx:
+    acked:
+    rx:
+    path validation outcomes: ValidationCount({Capable: 0, NotCapable(BlackHole): 0, NotCapable(Bleaching): 0, NotCapable(ReceivedUnsentECT1): 0})
+    mark transitions:
+  dscp: \n"
+    );
+}
+
 #[test]
 fn stats() {
     let now = now();
@@ -166,14 +206,33 @@ fn stats() {
             }
         }
 
-        for codepoint in [IpTosEcn::Ect1, IpTosEcn::Ce] {
-            assert_eq!(stats.ecn_tx[codepoint], 0);
-            assert_eq!(stats.ecn_rx[codepoint], 0);
+        for packet_type in PacketType::iter() {
+            for codepoint in [IpTosEcn::Ect1, IpTosEcn::Ce] {
+                assert_eq!(stats.ecn_tx[packet_type][codepoint], 0);
+                assert_eq!(stats.ecn_tx_acked[packet_type][codepoint], 0);
+                assert_eq!(stats.ecn_rx[packet_type][codepoint], 0);
+            }
         }
     }
 
-    assert!(client.stats().ecn_tx[IpTosEcn::Ect0] <= server.stats().ecn_rx[IpTosEcn::Ect0]);
-    assert!(server.stats().ecn_tx[IpTosEcn::Ect0] <= client.stats().ecn_rx[IpTosEcn::Ect0]);
+    for packet_type in PacketType::iter() {
+        assert!(
+            client.stats().ecn_tx_acked[packet_type][IpTosEcn::Ect0]
+                <= server.stats().ecn_rx[packet_type][IpTosEcn::Ect0]
+        );
+        assert!(
+            server.stats().ecn_tx_acked[packet_type][IpTosEcn::Ect0]
+                <= client.stats().ecn_rx[packet_type][IpTosEcn::Ect0]
+        );
+        assert_eq!(
+            client.stats().ecn_tx[packet_type][IpTosEcn::Ect0],
+            server.stats().ecn_rx[packet_type][IpTosEcn::Ect0]
+        );
+        assert_eq!(
+            server.stats().ecn_tx[packet_type][IpTosEcn::Ect0],
+            client.stats().ecn_rx[packet_type][IpTosEcn::Ect0]
+        );
+    }
 }
 
 #[test]

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

@@ -6,7 +6,7 @@
 
 use std::time::Duration;
 
-use neqo_common::{event::Provider as _, Decoder, Encoder};
+use neqo_common::{event::Provider as _, Decoder, Encoder, IpTosDscp};
 use test_fixture::{assertions, datagram, now};
 
 use super::{
@@ -17,12 +17,20 @@ use super::{
 use crate::{
     packet::PACKET_BIT_LONG,
     tparams::{TransportParameter, TransportParameterId::*},
-    ConnectionParameters, Error, Version, MIN_INITIAL_PACKET_SIZE,
+    ConnectionParameters, Error, Stats, Version, MIN_INITIAL_PACKET_SIZE,
 };
 
 // The expected PTO duration after the first Initial is sent.
 const INITIAL_PTO: Duration = Duration::from_millis(300);
 
+/// # Panics
+///
+/// When the count of received packets doesn't match the count of received packets with the
+/// (default) DSCP.
+pub fn assert_dscp(stats: &Stats) {
+    assert_eq!(stats.dscp_rx[IpTosDscp::Cs0], stats.packets_rx);
+}
+
 #[test]
 fn unknown_version() {
     let mut client = default_client();
@@ -33,6 +41,7 @@ fn unknown_version() {
     unknown_version_packet.resize(MIN_INITIAL_PACKET_SIZE, 0x0);
     drop(client.process(Some(datagram(unknown_version_packet)), now()));
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -48,6 +57,7 @@ fn server_receive_unknown_first_packet() {
     );
 
     assert_eq!(1, server.stats().dropped_rx);
+    assert_dscp(&server.stats());
 }
 
 fn create_vn(initial_pkt: &[u8], versions: &[u32]) -> Vec<u8> {
@@ -89,6 +99,7 @@ fn version_negotiation_current_version() {
     assert_eq!(delay, INITIAL_PTO);
     assert_eq!(*client.state(), State::WaitInitial);
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -110,6 +121,7 @@ fn version_negotiation_version0() {
     assert_eq!(delay, INITIAL_PTO);
     assert_eq!(*client.state(), State::WaitInitial);
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -132,6 +144,7 @@ fn version_negotiation_only_reserved() {
         }
         _ => panic!("Invalid client state"),
     }
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -153,6 +166,7 @@ fn version_negotiation_corrupted() {
     assert_eq!(delay, INITIAL_PTO);
     assert_eq!(*client.state(), State::WaitInitial);
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -174,6 +188,7 @@ fn version_negotiation_empty() {
     assert_eq!(delay, INITIAL_PTO);
     assert_eq!(*client.state(), State::WaitInitial);
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -195,6 +210,7 @@ fn version_negotiation_not_supported() {
         }
         _ => panic!("Invalid client state"),
     }
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -217,6 +233,7 @@ fn version_negotiation_bad_cid() {
     assert_eq!(delay, INITIAL_PTO);
     assert_eq!(*client.state(), State::WaitInitial);
     assert_eq!(1, client.stats().dropped_rx);
+    assert_dscp(&client.stats());
 }
 
 #[test]
@@ -227,6 +244,8 @@ fn compatible_upgrade() {
     connect(&mut client, &mut server);
     assert_eq!(client.version(), Version::Version2);
     assert_eq!(server.version(), Version::Version2);
+    assert_dscp(&client.stats());
+    assert_dscp(&server.stats());
 }
 
 /// When the first packet from the client is gigantic, the server might generate acknowledgment
@@ -264,6 +283,8 @@ fn compatible_upgrade_large_initial() {
     // Only handshake padding is "dropped".
     assert_eq!(client.stats().dropped_rx, 1);
     assert!(matches!(server.stats().dropped_rx, 2 | 3));
+    assert_dscp(&client.stats());
+    assert_dscp(&server.stats());
 }
 
 /// A server that supports versions 1 and 2 might prefer version 1 and that's OK.