fix: stop relay actor on endpoint close (#3601)

## Description

Since the transport abstraction was introduced, we no longer shut down
the relay actor when calling `Endpoint::close`. The relay actor is
currently only stopped when the last clone of the `iroh::Endpoint` is
dropped. This is surprising and wasteful. This PR fixes this so that the
relay actor is stopped when calling `Endpoint::close`.

Background: Since introducing the transport abstraction, the relay actor
handle is now owned by `Transports` which is owned by `MagicUdpSocket`
which is owned by the `quinn::Endpoint` as its `AsyncUdpSocket`. The
`AsyncUdpSocket` trait has no notification when the endpoint is closing,
thus currently the relay transport (and the relay actor, which it owns)
is not notified at all when the endpoint is shutting down. Therefore, it
keeps running (and potentially reconnecting to the relay or receiving
messages) until it is stopped when the last clone of the endpoint is
dropped and thus the `AbortOnDropHandle` owning the relay actor tasks is
dropped, finally stopping the relay actor tasks.

This PR exposes an existing cancellation token within the relay actor to
the magicsocket, allowing it to cancel the relay actors when calling
`Endpoint::close`.

## Breaking Changes

<!-- Optional, if there are any breaking changes document them,
including how to migrate older code. -->

## Notes & open questions

I tested this by looking at logs after closing an endpoint. Not sure if
there is a good way to test this. I'd vote to merge this even without a
test because it fixes an issue, unless someone has a good idea on how to
test this.

## Change checklist
<!-- Remove any that are not relevant. -->
- [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.
- [ ] Tests if relevant.
- [x] All breaking changes documented.
- [ ] List all breaking changes in the above "Breaking Changes" section.
- [ ] Open an issue or PR on any number0 repos that are affected by this
breaking change. Give guidance on how the updates should be handled or
do the actual updates themselves. The major ones are:
    - [ ] [`quic-rpc`](https://github.com/n0-computer/quic-rpc)
    - [ ] [`iroh-gossip`](https://github.com/n0-computer/iroh-gossip)
    - [ ] [`iroh-blobs`](https://github.com/n0-computer/iroh-blobs)
    - [ ] [`dumbpipe`](https://github.com/n0-computer/dumbpipe)
    - [ ] [`sendme`](https://github.com/n0-computer/sendme)

Author: Frando

iroh/src/magicsock.rs

@@ -149,8 +149,8 @@ pub(crate) struct Handle {
     msock: Arc<MagicSock>,
     // empty when shutdown
     actor_task: Arc<Mutex<Option<AbortOnDropHandle<()>>>>,
-    /// Token to cancel the actor task.
-    actor_token: CancellationToken,
+    /// Token to cancel the actor task and shutdown the relay transport.
+    shutdown_token: CancellationToken,
     // quinn endpoint
     endpoint: quinn::Endpoint,
 }
@@ -1392,17 +1392,22 @@ impl Handle {
         let my_relay = Watchable::new(None);
         let ipv6_reported = Arc::new(AtomicBool::new(ipv6_reported));
 
-        let relay_transport = RelayTransport::new(RelayActorConfig {
-            my_relay: my_relay.clone(),
-            secret_key: secret_key.clone(),
-            #[cfg(not(wasm_browser))]
-            dns_resolver: dns_resolver.clone(),
-            proxy_url: proxy_url.clone(),
-            ipv6_reported: ipv6_reported.clone(),
-            #[cfg(any(test, feature = "test-utils"))]
-            insecure_skip_relay_cert_verify,
-            metrics: metrics.magicsock.clone(),
-        });
+        let shutdown_token = CancellationToken::new();
+
+        let relay_transport = RelayTransport::new(
+            RelayActorConfig {
+                my_relay: my_relay.clone(),
+                secret_key: secret_key.clone(),
+                #[cfg(not(wasm_browser))]
+                dns_resolver: dns_resolver.clone(),
+                proxy_url: proxy_url.clone(),
+                ipv6_reported: ipv6_reported.clone(),
+                #[cfg(any(test, feature = "test-utils"))]
+                insecure_skip_relay_cert_verify,
+                metrics: metrics.magicsock.clone(),
+            },
+            shutdown_token.child_token(),
+        );
         let relay_transports = vec![relay_transport];
 
         let secret_encryption_key = secret_ed_box(&secret_key);
@@ -1531,12 +1536,9 @@ impl Handle {
         #[cfg(not(wasm_browser))]
         actor.update_direct_addresses(None);
 
-        let actor_token = CancellationToken::new();
-        let token = actor_token.clone();
-
         let actor_task = task::spawn(
             actor
-                .run(token, local_addrs_watch, sender)
+                .run(shutdown_token.child_token(), local_addrs_watch, sender)
                 .instrument(info_span!("actor")),
         );
 
@@ -1546,7 +1548,7 @@ impl Handle {
             msock,
             actor_task,
             endpoint,
-            actor_token,
+            shutdown_token,
         })
     }
 
@@ -1587,7 +1589,7 @@ impl Handle {
             return;
         }
         self.msock.closing.store(true, Ordering::Relaxed);
-        self.actor_token.cancel();
+        self.shutdown_token.cancel();
 
         // MutexGuard is not held across await points
         let task = self.actor_task.lock().expect("poisoned").take();

iroh/src/magicsock/transports/relay.rs

@@ -13,7 +13,7 @@ use n0_future::{
 };
 use n0_watcher::{Watchable, Watcher as _};
 use tokio::sync::mpsc;
-use tokio_util::sync::PollSender;
+use tokio_util::sync::{CancellationToken, PollSender};
 use tracing::{Instrument, error, info_span, trace, warn};
 
 use super::{Addr, Transmit};
@@ -38,7 +38,7 @@ pub(crate) struct RelayTransport {
 }
 
 impl RelayTransport {
-    pub(crate) fn new(config: RelayActorConfig) -> Self {
+    pub(crate) fn new(config: RelayActorConfig, cancel_token: CancellationToken) -> Self {
         let (relay_datagram_send_tx, relay_datagram_send_rx) = mpsc::channel(256);
 
         let (relay_datagram_recv_tx, relay_datagram_recv_rx) = mpsc::channel(512);
@@ -48,7 +48,7 @@ impl RelayTransport {
         let my_endpoint_id = config.secret_key.public();
         let my_relay = config.my_relay.clone();
 
-        let relay_actor = RelayActor::new(config, relay_datagram_recv_tx);
+        let relay_actor = RelayActor::new(config, relay_datagram_recv_tx, cancel_token);
 
         let actor_handle = AbortOnDropHandle::new(task::spawn(
             async move {

iroh/src/magicsock/transports/relay/actor.rs

@@ -855,8 +855,8 @@ impl RelayActor {
     pub(super) fn new(
         config: Config,
         relay_datagram_recv_queue: mpsc::Sender<RelayRecvDatagram>,
+        cancel_token: CancellationToken,
     ) -> Self {
-        let cancel_token = CancellationToken::new();
         Self {
             config,
             relay_datagram_recv_queue,