fix(proto): Avoid double-emitting Draining event, causing an active_connections underflow (#748)

## Description

- Fixes a bug where the `Draining` event was emitted twice when a
stateless reset token was duplicated and received twice
- Adds a regression test for the above scenario

This is one of those bugs that triggers an underflow in
`active_connections` in noq.

## Breaking Changes

None

## Notes & open questions

I'm still investigating whether there are further such cases.

## Change checklist

- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

Author: matheus23

noq-proto/src/connection/mod.rs

@@ -4408,7 +4408,7 @@ impl Connection {
                     ..
                 }) => {
                     let was_draining = self.state.move_to_drained(Some(conn_err));
-                    if !was_draining {
+                    if !was_draining && !was_drained {
                         self.endpoint_events.push_back(EndpointEventInner::Draining);
                     }
                 }

noq-proto/src/tests/mod.rs

@@ -309,6 +309,77 @@ fn stateless_reset_limit() {
     assert!(matches!(event, Some(DatagramEvent::Response(_))));
 }
 
+/// Regression test to ensure a connection that is already `Drained` doesn't emit a
+/// duplicate `Draining` endpoint event when a second stateless-reset datagram is processed.
+#[test]
+fn duplicate_stateless_reset_emits_single_draining() {
+    let _guard = subscribe();
+    let mut key_material = vec![0; 64];
+    let mut rng = rand::rng();
+    rng.fill_bytes(&mut key_material);
+    let reset_key = hmac::Key::new(hmac::HMAC_SHA256, &key_material);
+    rng.fill_bytes(&mut key_material);
+
+    let mut endpoint_config = EndpointConfig::new(Arc::new(reset_key));
+    endpoint_config.cid_generator(Arc::new(move || {
+        Box::new(HashedConnectionIdGenerator::from_key(0))
+    }));
+    let endpoint_config = Arc::new(endpoint_config);
+
+    let mut pair = Pair::new(endpoint_config.clone(), server_config());
+    let (client_ch, _) = pair.connect();
+    pair.drive(); // Flush any post-handshake frames
+
+    // Recreate the server endpoint so it loses all connection state but keeps the same
+    // reset key, causing it to respond to the client's packets with stateless resets.
+    pair.server.endpoint = Endpoint::new(endpoint_config, Some(Arc::new(server_config())), true);
+    // Force the server to generate the smallest possible stateless reset
+    pair.client.connections.get_mut(&client_ch).unwrap().ping();
+    pair.drive_client();
+    pair.drive_server();
+
+    // Capture the stateless reset datagram delivered to the client before it is processed.
+    let (_, captured_stateless_reset) = pair
+        .client
+        .inbound
+        .pop_first()
+        .expect("server should have sent a stateless reset");
+    pair.client.inbound.clear();
+
+    let now = pair.time;
+
+    // Duplicate the captured stateless reset token:
+    pair.client
+        .inbound
+        .push(now, captured_stateless_reset.clone());
+    pair.client.inbound.push(now, captured_stateless_reset);
+
+    // Only call drive_incoming instead of drive_client so we can manually count
+    // endpoint events below.
+    pair.client.drive_incoming(now);
+
+    // Apply the connection events produced by the endpoint and collect all endpoint events
+    // emitted by the connection.
+    let conn = pair.client.connections.get_mut(&client_ch).unwrap();
+    for (_, mut events) in pair.client.conn_events.drain() {
+        for event in events.drain(..) {
+            conn.handle_event(event);
+        }
+    }
+
+    // A drained connection receiving a second stateless reset must not emit a duplicate
+    // `Draining` event.
+    let mut draining = 0;
+    let mut drained = 0;
+    while let Some(event) = conn.poll_endpoint_events() {
+        draining += event.is_draining() as usize;
+        drained += event.is_drained() as usize;
+    }
+
+    assert_eq!(draining, 1, "expected exactly one Draining event");
+    assert_eq!(drained, 1, "expected exactly one Drained event");
+}
+
 #[test]
 fn export_keying_material() {
     let _guard = subscribe();

noq-proto/src/tests/util.rs

@@ -1097,15 +1097,15 @@ pub(super) struct TestEndpoint {
     pub(super) accepted: Option<Result<ConnectionHandle, ConnectionError>>,
     pub(super) connections: HashMap<ConnectionHandle, Connection>,
     pub(super) draining_connections: HashSet<ConnectionHandle>,
-    conn_events: HashMap<ConnectionHandle, VecDeque<ConnectionEvent>>,
+    pub(super) conn_events: HashMap<ConnectionHandle, VecDeque<ConnectionEvent>>,
     pub(super) captured_packets: Vec<Vec<u8>>,
     pub(super) capture_inbound_packets: bool,
     #[debug("handle_incoming")]
     pub(super) handle_incoming: Box<dyn FnMut(&Incoming) -> IncomingConnectionBehavior>,
     pub(super) waiting_incoming: Vec<Incoming>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub(super) struct Inbound {
     pub(super) ecn: Option<EcnCodepoint>,
     pub(super) packet: BytesMut,

noq-udp/build.rs

@@ -1,5 +1,9 @@
 use cfg_aliases::cfg_aliases;
 
+#[allow(
+    semicolon_in_expressions_from_macros,
+    reason = "cfg_aliases needs an update: https://github.com/katharostech/cfg_aliases/pull/15"
+)]
 fn main() {
     // Setup cfg aliases
     cfg_aliases! {

noq/build.rs

@@ -1,5 +1,9 @@
 use cfg_aliases::cfg_aliases;
 
+#[allow(
+    semicolon_in_expressions_from_macros,
+    reason = "cfg_aliases needs an update: https://github.com/katharostech/cfg_aliases/pull/15"
+)]
 fn main() {
     // Setup cfg aliases
     cfg_aliases! {