fix(iroh): Correctly abandon paths when new are opened with worse RTT (#4296)

## Description

Locally, `test_paths_watcher` and `endpoint_two_direct_add_relay` fairly
consistently failed for me.
I dug into `test_paths_watcher`, this was what was going on:
1. A new connection is established via IP path
2. The test waits for both relay and IP path to be established with a 1s
timeout
3. The connection immediately tries to add a relay path after it was
established via IP path, but it can't yet because of missing remote
CIDs, so it schedules a later path open for the relay path.
4. <lots and lots of path open events, many many new paths opened and
immedaitely closed>
5. Scheduled path open for the relay path triggers, but it fails because
MaxPathIdReached. Unlike RemoteCidsExhausted, it doesn't re-schedule
this path open.
6. 1s timeout triggers

The weird thing was that the max path ID limit wasn't increased. This
was because the hole-punched paths that were opened were never cleaned
up - no path was ever abandoned.

Digging deeper I found out that our logic for abandoning redundant IP
paths wasn't triggered when the newly hole-punched path wasn't better
than our current paths.

This PR fixes that.

During investigation I also found out that we would never re-try opening
the relay path in case you get a `MaxPathIdReached` error, which seems
bad, so now we explicitly handle this case as well.

## Breaking Changes

None

## Notes & open questions

Arguably, we could also decide to not match on the particular error that
causes us to not be able to open a path and always re-schedule path
opening on error?

## Change checklist
<!-- Remove any that are not relevant. -->
- [x] Self-review.

Author: matheus23

iroh/src/endpoint/connection.rs

@@ -1503,7 +1503,7 @@ mod tests {
     async fn test_paths_watcher() -> Result {
         const ALPN: &[u8] = b"test";
         let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
-        let (relay_map, _relay_map, _guard) = run_relay_server().await?;
+        let (relay_map, _relay_url, _guard) = run_relay_server().await?;
         let server = Endpoint::builder(presets::Minimal)
             .relay_mode(RelayMode::Custom(relay_map.clone()))
             .secret_key(SecretKey::from_bytes(&rng.random()))
@@ -1532,7 +1532,7 @@ mod tests {
         let mut paths_client = conn_client.paths_stream();
         let mut paths_server = conn_server.paths_stream();
 
-        /// Advances the path stream until at least one IP and one relay paths are available.
+        /// Advances the path stream until at least one IP and one relay path is available.
         ///
         /// Panics if the path stream finishes before that happens.
         async fn wait_for_paths(mut stream: impl Send + Unpin + Stream<Item = PathList<'_>>) {

iroh/src/socket/remote_map/remote_state.rs

@@ -666,10 +666,11 @@ impl RemoteStateActor {
                 network_path = %addr,
                 prev_network_path = prev_remote.map(|p| format!("{p}")).unwrap_or("None".to_string()),
             );
-            self.apply_selected_change(&addr);
         } else {
             trace!(?current_path, "keeping current path");
         }
+
+        self.apply_selected_path();
     }
 
     /// Propagates a change of [`State::selected_path`] to noq.
@@ -679,15 +680,21 @@ impl RemoteStateActor {
     /// - Sets all non-selected paths to [`PathStatus::Backup`]
     /// - Opens the selected path if it does not exist on the connection
     /// - Sets the selected path to [`PathStatus::Available`]
-    fn apply_selected_change(&mut self, selected: &transports::FourTuple) {
+    fn apply_selected_path(&mut self) {
+        let Some(selected) = self.state.selected_path.clone() else {
+            // We can't open the selected path on all paths if we don't have one yet.
+            // And we can't close all "unselected" paths either, because we don't know which one is selected.
+            return;
+        };
+
         for (conn_id, conn_state) in self.connections.iter() {
             let Some(conn) = conn_state.handle.upgrade() else {
                 continue;
             };
 
             // Open path if it doesn't exist yet.
             self.state
-                .open_path_on_conn(*conn_id, conn_state, &conn, selected);
+                .open_path_on_conn(*conn_id, conn_state, &conn, &selected);
 
             for (path_id, path_remote) in conn_state.paths.iter() {
                 let Some(path) = conn.path(*path_id) else {
@@ -701,7 +708,7 @@ impl RemoteStateActor {
                 // and racing to abandon the last one.
                 if conn.side().is_client()
                     && path_remote.is_ip()
-                    && path_remote != selected
+                    && path_remote != &selected
                     && conn_state.paths.values().filter(|a| a.is_ip()).count() > 1
                 {
                     trace!(?path_remote, %conn_id, %path_id, "closing direct path");
@@ -725,7 +732,7 @@ impl RemoteStateActor {
             }
 
             // Record the new selected path in the path watcher.
-            conn_state.path_state.record_selected(selected);
+            conn_state.path_state.record_selected(&selected);
         }
     }
 
@@ -1045,11 +1052,12 @@ impl State {
             None => {
                 let ret = now_or_never(fut);
                 match ret {
-                    Some(Err(PathError::RemoteCidsExhausted)) => {
+                    Some(Err(PathError::RemoteCidsExhausted))
+                    | Some(Err(PathError::MaxPathIdReached)) => {
                         self.scheduled_open_path =
                             Some(Instant::now() + Duration::from_millis(333));
                         self.pending_open_paths.push_back(open_addr.clone());
-                        trace!(?open_addr, "scheduling open_path");
+                        trace!(?open_addr, ?ret, "scheduling open_path");
                     }
                     _ => warn!(?ret, "Opening path failed"),
                 }