feat: Define a path migration event (#3598)

* Define a path migration event

* Sort order fix

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Signed-off-by: Martin Thomson <mt@lowentropy.net>

* Factor event generation

Which should save one borrow each time.

* Interior mutability

* Better factoring of path response

* fmt

---------

Signed-off-by: Martin Thomson <mt@lowentropy.net>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: martinthomson

neqo-bin/src/server/http09.rs

@@ -232,7 +232,8 @@ impl super::HttpServer for HttpServer {
                     }
                     ConnectionEvent::StateChange(_)
                     | ConnectionEvent::SendStreamCreatable { .. }
-                    | ConnectionEvent::SendStreamComplete { .. } => (),
+                    | ConnectionEvent::SendStreamComplete { .. }
+                    | ConnectionEvent::PathMigrated { .. } => (),
                     e => qwarn!("unhandled event {e:?}"),
                 }
             }

neqo-http3/src/connection_client.rs

@@ -1170,7 +1170,8 @@ impl Http3Client {
                 ConnectionEvent::SendStreamComplete { .. }
                 | ConnectionEvent::OutgoingDatagramOutcome { .. }
                 | ConnectionEvent::IncomingDatagramDropped
-                | ConnectionEvent::SconeUpdated(_) => {}
+                | ConnectionEvent::SconeUpdated(_)
+                | ConnectionEvent::PathMigrated { .. } => {}
             }
         }
         Ok(())

neqo-http3/src/connection_server.rs

@@ -377,7 +377,8 @@ impl Http3ServerHandler {
                 | ConnectionEvent::SendStreamCreatable { .. }
                 | ConnectionEvent::OutgoingDatagramOutcome { .. }
                 | ConnectionEvent::IncomingDatagramDropped
-                | ConnectionEvent::SconeUpdated(_) => {}
+                | ConnectionEvent::SconeUpdated(_)
+                | ConnectionEvent::PathMigrated { .. } => {}
             }
         }
         Ok(())

neqo-transport/src/connection/mod.rs

@@ -2096,10 +2096,17 @@ impl Connection {
             .migrate(&path, force, now, &mut self.stats.borrow_mut())
         {
             self.loss_recovery.migrate();
+            self.path_migrated(&path);
         }
         Ok(())
     }
 
+    fn path_migrated(&self, path: &PathRef) {
+        let p = path.borrow();
+        self.events
+            .path_migrated(p.local_address(), p.remote_address());
+    }
+
     fn migrate_to_preferred_address(&mut self, now: Instant) -> Res<()> {
         let spa: Option<(tparams::PreferredAddress, ConnectionIdEntry<Srt>)> = if matches!(
             self.conn_params.get_preferred_address(),
@@ -2164,8 +2171,12 @@ impl Connection {
         }
 
         if self.ensure_permanent(path, now).is_ok() {
+            let was_primary = path.borrow().is_primary();
             self.paths
                 .handle_migration(path, remote, now, &mut self.stats.borrow_mut());
+            if !was_primary {
+                self.path_migrated(path);
+            }
         } else {
             qinfo!(
                 "[{self}] {} Peer migrated, but no connection ID available",
@@ -3376,11 +3387,11 @@ impl Connection {
             }
             Frame::PathResponse { data } => {
                 self.stats.borrow_mut().frame_rx.path_response += 1;
-                if self
-                    .paths
-                    .path_response(data, now, &mut self.stats.borrow_mut())
+                if let Some(primary) =
+                    self.paths
+                        .path_response(data, now, &mut self.stats.borrow_mut())
                 {
-                    // This PATH_RESPONSE enabled migration; tell loss recovery.
+                    self.path_migrated(&primary);
                     self.loss_recovery.migrate();
                 }
             }

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

@@ -17,7 +17,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use neqo_common::{Datagram, Decoder, qdebug};
+use neqo_common::{Datagram, Decoder, event::Provider as _, qdebug};
 use test_fixture::{
     DEFAULT_ADDR, DEFAULT_ADDR_V4,
     assertions::{assert_v4_path, assert_v6_path},
@@ -31,8 +31,9 @@ use super::{
     zero_len_cid_client,
 };
 use crate::{
-    CloseReason, ConnectionId, ConnectionIdDecoder as _, ConnectionIdGenerator, ConnectionIdRef,
-    ConnectionParameters, EmptyConnectionIdGenerator, Error, MIN_INITIAL_PACKET_SIZE,
+    CloseReason, ConnectionEvent, ConnectionId, ConnectionIdDecoder as _, ConnectionIdGenerator,
+    ConnectionIdRef, ConnectionParameters, EmptyConnectionIdGenerator, Error,
+    MIN_INITIAL_PACKET_SIZE,
     cid::ConnectionIdManager,
     connection::tests::{
         assert_path_challenge_min_len, connect, send_something_paced, send_with_extra,
@@ -434,6 +435,11 @@ fn migrate_immediate() {
     client
         .migrate(Some(DEFAULT_ADDR_V4), Some(DEFAULT_ADDR_V4), true, now)
         .unwrap();
+    assert!(client.events().any(|e| matches!(
+        e,
+        ConnectionEvent::PathMigrated { local, remote }
+        if local == DEFAULT_ADDR_V4 && remote == DEFAULT_ADDR_V4
+    )));
 
     let client1 = send_something(&mut client, now);
     assert_v4_path(&client1, true); // Contains PATH_CHALLENGE.
@@ -447,6 +453,11 @@ fn migrate_immediate() {
     // The server accepts the first packet and migrates (but probes).
     let server1 = server.process(Some(client1), now).dgram().unwrap();
     assert_v4_path(&server1, true);
+    assert!(
+        server
+            .events()
+            .any(|e| matches!(e, ConnectionEvent::PathMigrated { .. }))
+    );
     let server2 = server.process_output(now).dgram().unwrap();
     assert_v6_path(&server2, true);
 
@@ -655,6 +666,11 @@ fn migration(mut client: Connection) {
 
     // Once the client receives the probe response, it migrates to the new path.
     client.process_input(resp, now);
+    assert!(client.events().any(|e| matches!(
+        e,
+        ConnectionEvent::PathMigrated { local, remote }
+        if local == DEFAULT_ADDR_V4 && remote == DEFAULT_ADDR_V4
+    )));
     assert_eq!(client.stats().frame_rx.path_challenge, 1);
     let migrate_client = send_something(&mut client, now);
     assert_v4_path(&migrate_client, true); // Responds to server probe.

neqo-transport/src/events.rs

@@ -6,7 +6,7 @@
 
 // Collecting a list of events relevant to whoever is using the Connection.
 
-use std::{cell::RefCell, collections::VecDeque, num::NonZeroU64, rc::Rc};
+use std::{cell::RefCell, collections::VecDeque, net::SocketAddr, num::NonZeroU64, rc::Rc};
 
 use neqo_common::event::Provider as EventProvider;
 use nss::ResumptionToken;
@@ -82,6 +82,11 @@ pub enum ConnectionEvent {
     /// An update was received to SCONE throughput advice.
     /// The value is the approximate rate in bits per second; None = unknown.
     SconeUpdated(Option<NonZeroU64>),
+    /// A path migration completed; the connection is now sending on this path.
+    PathMigrated {
+        local: SocketAddr,
+        remote: SocketAddr,
+    },
 }
 
 #[derive(Debug, Default, Clone)]
@@ -216,6 +221,10 @@ impl ConnectionEvents {
         }
     }
 
+    pub fn path_migrated(&self, local: SocketAddr, remote: SocketAddr) {
+        self.insert(ConnectionEvent::PathMigrated { local, remote });
+    }
+
     fn insert(&self, event: ConnectionEvent) {
         let mut q = self.events.borrow_mut();
 

neqo-transport/src/path.rs

@@ -322,10 +322,15 @@ impl Paths {
     }
 
     /// A `PATH_RESPONSE` was received.
-    /// Returns `true` if migration occurred.
+    /// Returns `Some` with the new primary path if migration occurred.
     /// If PMTUD is enabled and migration occurs, it will be started on the new primary path.
     #[must_use]
-    pub fn path_response(&mut self, response: [u8; 8], now: Instant, stats: &mut Stats) -> bool {
+    pub fn path_response(
+        &mut self,
+        response: [u8; 8],
+        now: Instant,
+        stats: &mut Stats,
+    ) -> Option<PathRef> {
         // TODO(mt) consider recording an RTT measurement here as we don't train
         // RTT for non-primary paths.
         for p in &self.paths {
@@ -340,12 +345,12 @@ impl Paths {
                     if self.pmtud {
                         primary.borrow_mut().pmtud_mut().start(now, stats);
                     }
-                    return true;
+                    return Some(primary);
                 }
                 break;
             }
         }
-        false
+        None
     }
 
     /// Retire all of the connection IDs prior to the indicated sequence number.