fix: remove problematic usage of else branches in tokio::selects (#2940)

## Description

Fixed some more cases of `else` branches in `tokio::select!`s that would
never actually run. I noticed because the relay actor would start
logging `recv error Receive(shut down)` when I dropped the `Endpoint`.

## Notes & open questions

There's 2 cases where the `else` branch still would never run but I
wasn't sure of what the intended behaviour was, so I left a todo
comment.

In the `RelayActor` it seems like it would make sense to exit in both
cases - the `ping_tasks` `JoinSet` being empty and the `receiver` being
closed.

The magicsock actor I'm much less sure of - I suspect the `else` branch
could just be removed?

## Change checklist

- [ ] Self-review.
- [ ] Tests if relevant.

Author: PaulOlteanu

iroh-net/src/magicsock.rs

@@ -1767,10 +1767,22 @@ impl Actor {
                 discovery_events = events;
             }
         }
+
+        let mut receiver_closed = false;
+        let mut portmap_watcher_closed = false;
+        let mut link_change_closed = false;
         loop {
             inc!(Metrics, actor_tick_main);
             tokio::select! {
-                Some(msg) = self.msg_receiver.recv() => {
+                msg = self.msg_receiver.recv(), if !receiver_closed => {
+                    let Some(msg) = msg else {
+                        trace!("tick: magicsock receiver closed");
+                        inc!(Metrics, actor_tick_other);
+
+                        receiver_closed = true;
+                        continue;
+                    };
+
                     trace!(?msg, "tick: msg");
                     inc!(Metrics, actor_tick_msg);
                     if self.handle_actor_message(msg).await {
@@ -1782,7 +1794,15 @@ impl Actor {
                     inc!(Metrics, actor_tick_re_stun);
                     self.msock.re_stun("periodic");
                 }
-                Ok(()) = portmap_watcher.changed() => {
+                change = portmap_watcher.changed(), if !portmap_watcher_closed => {
+                    if change.is_err() {
+                        trace!("tick: portmap watcher closed");
+                        inc!(Metrics, actor_tick_other);
+
+                        portmap_watcher_closed = true;
+                        continue;
+                    }
+
                     trace!("tick: portmap changed");
                     inc!(Metrics, actor_tick_portmap_changed);
                     let new_external_address = *portmap_watcher.borrow();
@@ -1809,22 +1829,29 @@ impl Actor {
                         self.refresh_direct_addrs(reason).await;
                     }
                 }
-                Some(is_major) = link_change_r.recv() => {
+                is_major = link_change_r.recv(), if !link_change_closed => {
+                    let Some(is_major) = is_major else {
+                        trace!("tick: link change receiver closed");
+                        inc!(Metrics, actor_tick_other);
+
+                        link_change_closed = true;
+                        continue;
+                    };
+
                     trace!("tick: link change {}", is_major);
                     inc!(Metrics, actor_link_change);
                     self.handle_network_change(is_major).await;
                 }
+                // Even if `discovery_events` yields `None`, it could begin to yield
+                // `Some` again in the future, so we don't want to disable this branch
+                // forever like we do with the other branches that yield `Option`s
                 Some(discovery_item) = discovery_events.next() => {
                     trace!("tick: discovery event, address discovered: {discovery_item:?}");
                     let node_addr = NodeAddr {node_id: discovery_item.node_id, info: discovery_item.addr_info};
                     if let Err(e) = self.msock.add_node_addr(node_addr.clone(), Source::Discovery { name: discovery_item.provenance.into() }) {
                         warn!(?node_addr, "unable to add discovered node address to the node map: {e:?}");
                     }
                 }
-                else => {
-                    trace!("tick: other");
-                    inc!(Metrics, actor_tick_other);
-                }
             }
         }
     }

iroh-net/src/magicsock/relay_actor.rs

@@ -111,7 +111,12 @@ impl ActiveRelay {
                 self.relay_client.connect().await.context("keepalive")?;
             }
             tokio::select! {
-                Some(msg) = inbox.recv() => {
+                msg = inbox.recv() => {
+                    let Some(msg) = msg else {
+                        debug!("all clients closed");
+                        break;
+                    };
+
                     trace!("tick: inbox: {:?}", msg);
                     match msg {
                         ActiveRelayMessage::GetLastWrite(r) => {
@@ -144,6 +149,7 @@ impl ActiveRelay {
                         }
                     }
                 }
+
                 msg = self.relay_client_receiver.recv() => {
                     trace!("tick: relay_client_receiver");
                     if let Some(msg) = msg {
@@ -153,10 +159,6 @@ impl ActiveRelay {
                         }
                     }
                 }
-                else => {
-                    debug!("all clients closed");
-                    break;
-                }
             }
         }
         debug!("exiting");
@@ -301,25 +303,37 @@ impl RelayActor {
                     trace!("shutting down");
                     break;
                 }
-                Some(Ok((url, ping_success))) = self.ping_tasks.join_next() => {
-                    if !ping_success {
-                        with_cancel(
-                            self.cancel_token.child_token(),
-                            self.close_or_reconnect_relay(&url, "rebind-ping-fail")
-                        ).await;
+                // `ping_tasks` being empty is a normal situation - in fact it starts empty
+                // until a `MaybeCloseRelaysOnRebind` message is received.
+                Some(task_result) = self.ping_tasks.join_next() => {
+                    match task_result {
+                        Ok((url, ping_success)) => {
+                            if !ping_success {
+                                with_cancel(
+                                    self.cancel_token.child_token(),
+                                    self.close_or_reconnect_relay(&url, "rebind-ping-fail")
+                                ).await;
+                            }
+                        }
+
+                        Err(err) => {
+                            warn!("ping task error: {:?}", err);
+                        }
                     }
                 }
-                Some(msg) = receiver.recv() => {
+
+                msg = receiver.recv() => {
+                    let Some(msg) = msg else {
+                        trace!("shutting down relay recv loop");
+                        break;
+                    };
+
                     with_cancel(self.cancel_token.child_token(), self.handle_msg(msg)).await;
                 }
                 _ = cleanup_timer.tick() => {
                     trace!("tick: cleanup");
                     with_cancel(self.cancel_token.child_token(), self.clean_stale_relay()).await;
                 }
-                else => {
-                    trace!("shutting down relay recv loop");
-                    break;
-                }
             }
         }
 

iroh-relay/src/client.rs

@@ -506,7 +506,13 @@ impl Actor {
                     }
                     msg_sender.send(res).await.ok();
                 }
-                Some(msg) = inbox.recv() => {
+                msg = inbox.recv() => {
+                    let Some(msg) = msg else {
+                        // Shutting down
+                        self.close().await;
+                        break;
+                    };
+
                     match msg {
                         ActorMessage::Connect(s) => {
                             let res = self.connect("actor msg").await.map(|(client, _)| (client));
@@ -546,11 +552,6 @@ impl Actor {
                         },
                     }
                 }
-                else => {
-                    // Shutting down
-                    self.close().await;
-                    break;
-                }
             }
         }
     }

iroh-router/src/router.rs

@@ -117,7 +117,11 @@ impl RouterBuilder {
                         break;
                     },
                     // handle incoming p2p connections.
-                    Some(incoming) = endpoint.accept() => {
+                    incoming = endpoint.accept() => {
+                        let Some(incoming) = incoming else {
+                            break;
+                        };
+
                         let protocols = protocols.clone();
                         join_set.spawn(async move {
                             handle_connection(incoming, protocols).await;
@@ -144,7 +148,6 @@ impl RouterBuilder {
                             _ => {}
                         }
                     },
-                    else => break,
                 }
             }
 

iroh/src/node.rs

@@ -433,7 +433,6 @@ impl<D: iroh_blobs::store::Store> NodeInner<D> {
                         _ => {}
                     }
                 },
-                else => break,
             }
         }