feat(snownet): only keep the best possible candidate pair alive (#3792)

This took me a while to figure out but I think the solution is quite
neat. We are using ICE trickle which means there could be new candidates
at any point in time. Thus, there really is never a good time to say
"ICE is finished" and clean-up all other candidates (that is what
non-trickle ICE would want you to do:
https://datatracker.ietf.org/doc/html/rfc8445#section-8.3). But what we
can do is, upon each nomination, look at our local candidates and
invalidate all that are of the same priority or less.

For example, if we start with a connection via a relay, discard all
other relay candidates but keep the host and server-reflexive ones. If
the ICE agent then figures out a better path, it will give us a new
nomination and we can discard even more candidates.

On the other hand, if hole-punching fails, str0m will eventually give up
on certain candidate pairs because it is not receiving replies and
consider them failed.

Thus, the behaviour that we are getting with this PR is: Try all
possible candidate pairs but settle on the best possible one.

What is kind of neat is that, because we are still in ICE trickle mode,
receiving a new candidate could still upgrade existing relayed
connections to direct ones if the new candidate allows it.

The other side of this coin is that we won't have a fallback any more to
other pairs if the current one fails. In that case, we will consider the
entire connection failed, remove it and create a new one on the next
connection intent.

Resolves: #3789.

rust/connlib/snownet/src/node.rs

@@ -30,6 +30,7 @@ use std::borrow::Cow;
 use std::iter;
 use std::ops::ControlFlow;
 use stun_codec::rfc5389::attributes::{Realm, Username};
+use tracing::{field, Span};
 
 // Note: Taken from boringtun
 const HANDSHAKE_RATE_LIMIT: u64 = 100;
@@ -372,10 +373,7 @@ where
                         ..
                     } => {
                         let candidate = conn
-                            .agent
-                            .local_candidates()
-                            .iter()
-                            .find(|c| c.addr() == source)
+                            .local_candidate(source)
                             .expect("to only nominate existing candidates");
 
                         let remote_socket = match candidate.kind() {
@@ -415,6 +413,8 @@ where
                             tracing::info!(old = ?conn.peer_socket, new = ?remote_socket, "Updating remote socket");
                             conn.peer_socket = Some(remote_socket);
 
+                            conn.invalidate_candiates();
+
                             if is_first_connection {
                                 tracing::info!(%id, "Starting wireguard handshake");
 
@@ -1292,6 +1292,15 @@ enum PeerSocket {
     },
 }
 
+impl PeerSocket {
+    fn our_socket(&self) -> SocketAddr {
+        match self {
+            PeerSocket::Direct { source, .. } => *source,
+            PeerSocket::Relay { relay, .. } => *relay,
+        }
+    }
+}
+
 impl Connection {
     /// Checks if we want to accept a packet from a certain address.
     ///
@@ -1464,4 +1473,40 @@ impl Connection {
 
         self.encapsulate(bytes, allocations, now)
     }
+
+    /// Invalidates all local candidates with a lower or equal priority compared to the nominated one.
+    ///
+    /// Each time we nominate a candidate pair, we don't really want to keep all the others active because it creates a lot of noise.
+    /// At the same time, we want to retain trickle ICE and allow the ICE agent to find a _better_ pair, hence we invalidate by priority.
+    #[tracing::instrument(level = "debug", skip_all, fields(nominated_prio))]
+    fn invalidate_candiates(&mut self) {
+        let Some(socket) = self.peer_socket else {
+            return;
+        };
+
+        let Some(nominated) = self.local_candidate(socket.our_socket()).cloned() else {
+            return;
+        };
+
+        Span::current().record("nominated_prio", field::display(&nominated.prio()));
+
+        let irrelevant_candidates = self
+            .agent
+            .local_candidates()
+            .iter()
+            .filter(|c| c.prio() <= nominated.prio() && c != &&nominated)
+            .cloned()
+            .collect::<Vec<_>>();
+
+        for candidate in irrelevant_candidates {
+            self.agent.invalidate_candidate(&candidate);
+        }
+    }
+
+    fn local_candidate(&self, source: SocketAddr) -> Option<&Candidate> {
+        self.agent
+            .local_candidates()
+            .iter()
+            .find(|c| c.addr() == source)
+    }
 }