refactor(multipath): Stop inactive endpoint actors (#3643)

## Description

Fixes #3638 (partially)

This is a first, small solution to stop inactive endpoint actors after
an idle timeout. I implemented it such that the *actor* decides once to
stop, while making sure that we *never* create senders to actors that
are shutting down.

Logic in the actor:
* The actor enters an idle timeout (set to 60 seconds) once it has no
active connections, an empty inbox, and no inbox senders
* Once the timeout expires, it is rechecked that the idle conditions
hold, and if so the actor exits
* Once any of the idle conditions don't hold anymore, the idle timeout
is deactivated and restarted once the conditions are met again

The actor checks if the inbox's sender strong count equals 1, which
means that no senders exist apart from the one held in the endpoint map.
This check is protected with a mutex, to enter a critical section for
closing the inbox while the lock is held in case the conditions are met.
This is to ensure that there cannot be a race condition where a sender
is cloned out right after the check in the actor returns true, but
before the inbox is closed.

Logic in the endpoint map:
* When handing out senders, we acquire the shared lock, and check that
the channel is not closed while the lock is held. This ensures that the
actor never closes while a sender is alive. If the actor is closed, we
remove the handle to the dead actor and create a new actor.
* On regular intervals (set to 60 seconds) the magicsock actor removes
handles to dead actors.


## Breaking Changes

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

## Notes & open questions

* I *think* my logic around the critical section and ensuring that we
never close the actor while senders exist is sound. However, it needs
careful review and tests. I'll do some thinking on how to best test
this.
* Instead of employing an interval to remove dead actor handles, we
could use a channel where the actor informs an outside-task which
endpoint actors terminated, so that the outside-task can then lock the
endpoint map and remove just those. Not sure if that's worth it.
* Another solution here might be to spawn the actor tasks into a join
set in the magicsock actor. However this would need further refactoring
and would likely make spawning actors async. I think I'd prefer to keep
that sync because it makes the surrounding code a lot simpler.
* This does not yet implement some of the more advanced reasoning that
#3638 proposes. I think we should start with something simple that
prevents memory exhaustion and tweak as needed. However, it could also
be argued that we should start with a more featureful design right away.

## Change checklist
<!-- Remove any that are not relevant. -->
- [ ] Self-review.
- [ ] 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.
- [ ] 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)

iroh/src/magicsock.rs

@@ -1388,6 +1388,9 @@ impl Actor {
         // ensure we are doing an initial publish of our addresses
         self.msock.publish_my_addr();
 
+        // Interval timer to remove closed `EndpointStateActor` handles from the endpoint map.
+        let mut endpoint_map_gc = time::interval(endpoint_map::ENDPOINT_MAP_GC_INTERVAL);
+
         loop {
             self.msock.metrics.magicsock.actor_tick_main.inc();
             #[cfg(not(wasm_browser))]
@@ -1499,6 +1502,9 @@ impl Actor {
                         warn!(%dst, endpoint = %dst_key.fmt_short(), ?err, "failed to send disco message (UDP)");
                     }
                 }
+                _ = endpoint_map_gc.tick() => {
+                    self.msock.endpoint_map.remove_closed_endpoint_state_actors();
+                }
             }
         }
     }

iroh/src/magicsock/endpoint_map.rs

@@ -1,8 +1,9 @@
 use std::{
-    collections::BTreeSet,
+    collections::{BTreeSet, hash_map},
     hash::Hash,
     net::{IpAddr, SocketAddr},
     sync::{Arc, Mutex},
+    time::Duration,
 };
 
 use iroh_base::{EndpointAddr, EndpointId, RelayUrl};
@@ -27,6 +28,9 @@ use crate::disco;
 mod endpoint_state;
 mod path_state;
 
+/// Interval in which handles to closed [`EndpointStateActor`]s should be removed.
+pub(super) const ENDPOINT_MAP_GC_INTERVAL: Duration = Duration::from_secs(60);
+
 // TODO: use this
 // /// Number of endpoints that are inactive for which we keep info about. This limit is enforced
 // /// periodically via [`NodeMap::prune_inactive`].
@@ -106,6 +110,15 @@ impl EndpointMap {
         self.endpoint_mapped_addrs.get(&eid)
     }
 
+    /// Removes the handles for terminated [`EndpointStateActor`]s from the endpoint map.
+    ///
+    /// This should be called periodically to remove handles to endpoint state actors
+    /// that have shutdown after their idle timeout expired.
+    pub(super) fn remove_closed_endpoint_state_actors(&self) {
+        let mut handles = self.actor_handles.lock().expect("poisoned");
+        handles.retain(|_eid, handle| !handle.sender.is_closed())
+    }
+
     /// Returns the sender for the [`EndpointStateActor`].
     ///
     /// If needed a new actor is started on demand.
@@ -116,33 +129,48 @@ impl EndpointMap {
         eid: EndpointId,
     ) -> mpsc::Sender<EndpointStateMessage> {
         let mut handles = self.actor_handles.lock().expect("poisoned");
-        match handles.get(&eid) {
-            Some(handle) => handle.sender.clone(),
-            None => {
-                // Create a new EndpointStateActor and insert it into the endpoint map.
-                let local_addrs = self.local_addrs.clone();
-                let disco = self.disco.clone();
-                let metrics = self.metrics.clone();
-                let actor = EndpointStateActor::new(
-                    eid,
-                    self.local_endpoint_id,
-                    local_addrs,
-                    disco,
-                    self.relay_mapped_addrs.clone(),
-                    metrics,
-                    self.sender.clone(),
-                );
-                let handle = actor.start();
-                let sender = handle.sender.clone();
-                handles.insert(eid, handle);
-
-                // Ensure there is a EndpointMappedAddr for this EndpointId.
-                self.endpoint_mapped_addrs.get(&eid);
+        match handles.entry(eid) {
+            hash_map::Entry::Occupied(mut entry) => {
+                if let Some(sender) = entry.get().sender.get() {
+                    sender
+                } else {
+                    // The actor is dead: Start a new actor.
+                    let (handle, sender) = self.start_endpoint_state_actor(eid);
+                    entry.insert(handle);
+                    sender
+                }
+            }
+            hash_map::Entry::Vacant(entry) => {
+                let (handle, sender) = self.start_endpoint_state_actor(eid);
+                entry.insert(handle);
                 sender
             }
         }
     }
 
+    /// Starts a new endpoint state actor and returns a handle and a sender.
+    ///
+    /// The handle is not inserted into the endpoint map, this must be done by the caller of this function.
+    fn start_endpoint_state_actor(
+        &self,
+        eid: EndpointId,
+    ) -> (EndpointStateHandle, mpsc::Sender<EndpointStateMessage>) {
+        // Ensure there is a EndpointMappedAddr for this EndpointId.
+        self.endpoint_mapped_addrs.get(&eid);
+        let handle = EndpointStateActor::new(
+            eid,
+            self.local_endpoint_id,
+            self.local_addrs.clone(),
+            self.disco.clone(),
+            self.relay_mapped_addrs.clone(),
+            self.metrics.clone(),
+            self.sender.clone(),
+        )
+        .start();
+        let sender = handle.sender.get().expect("just created");
+        (handle, sender)
+    }
+
     pub(super) fn handle_ping(&self, msg: disco::Ping, sender: EndpointId, src: transports::Addr) {
         if msg.endpoint_key != sender {
             warn!("DISCO Ping EndpointId mismatch, ignoring ping");

iroh/src/magicsock/endpoint_map/endpoint_state.rs

@@ -17,14 +17,12 @@ use quinn::{PathStats, WeakConnectionHandle};
 use quinn_proto::{PathError, PathEvent, PathId, PathStatus};
 use rustc_hash::FxHashMap;
 use smallvec::SmallVec;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
 use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
 use tracing::{Instrument, Level, debug, error, event, info_span, instrument, trace, warn};
 
+use self::guarded_channel::{GuardedReceiver, GuardedSender, guarded_channel};
 use super::{Source, path_state::PathState};
-// TODO: Use this
-// #[cfg(any(test, feature = "test-utils"))]
-// use crate::endpoint::PathSelection;
 use crate::{
     disco::{self},
     endpoint::DirectAddr,
@@ -37,6 +35,12 @@ use crate::{
     util::MaybeFuture,
 };
 
+// TODO: Use this
+// #[cfg(any(test, feature = "test-utils"))]
+// use crate::endpoint::PathSelection;
+
+mod guarded_channel;
+
 // TODO: use this
 // /// Number of addresses that are not active that we keep around per endpoint.
 // ///
@@ -67,6 +71,14 @@ use crate::{
 // TODO: Quinn should just do this.  Also, I made this value up.
 const APPLICATION_ABANDON_PATH: u8 = 30;
 
+/// The time after which an idle [`EndpointStateActor`] stops.
+///
+/// The actor only enters the idle state if no connections are active and no inbox senders exist
+/// apart from the one stored in the endpoint map. Stopping and restarting the actor in this state
+/// is not an issue; a timeout here serves the purpose of not stopping-and-recreating actors
+/// in a high frequency, and to keep data about previous path around for subsequent connections.
+const ACTOR_MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(60);
+
 /// A stream of events from all paths for all connections.
 ///
 /// The connection is identified using [`ConnId`].  The event `Err` variant happens when the
@@ -172,7 +184,7 @@ impl EndpointStateActor {
     }
 
     pub(super) fn start(mut self) -> EndpointStateHandle {
-        let (tx, rx) = mpsc::channel(16);
+        let (tx, rx) = guarded_channel(16);
         let me = self.local_endpoint_id;
         let endpoint_id = self.endpoint_id;
 
@@ -207,9 +219,11 @@ impl EndpointStateActor {
     /// discipline is needed to not turn pending for a long time.
     async fn run(
         &mut self,
-        mut inbox: mpsc::Receiver<EndpointStateMessage>,
+        mut inbox: GuardedReceiver<EndpointStateMessage>,
     ) -> n0_error::Result<()> {
         trace!("actor started");
+        let idle_timeout = MaybeFuture::None;
+        tokio::pin!(idle_timeout);
         loop {
             let scheduled_path_open = match self.scheduled_open_path {
                 Some(when) => MaybeFuture::Some(time::sleep_until(when)),
@@ -252,6 +266,22 @@ impl EndpointStateActor {
                     self.scheduled_holepunch = None;
                     self.trigger_holepunching().await;
                 }
+                _ = &mut idle_timeout => {
+                    if self.connections.is_empty() && inbox.close_if_idle() {
+                        trace!("idle timeout expired and still idle: terminate actor");
+                        break;
+                    }
+                }
+            }
+
+            if self.connections.is_empty() && inbox.is_idle() && idle_timeout.is_none() {
+                trace!("start idle timeout");
+                idle_timeout
+                    .as_mut()
+                    .set_future(time::sleep(ACTOR_MAX_IDLE_TIMEOUT));
+            } else if idle_timeout.is_some() {
+                trace!("abort idle timeout");
+                idle_timeout.as_mut().set_none()
             }
         }
         trace!("actor terminating");
@@ -1028,11 +1058,16 @@ pub(crate) enum EndpointStateMessage {
 
 /// A handle to a [`EndpointStateActor`].
 ///
-/// Dropping this will stop the actor.
+/// Dropping this will stop the actor. The actor will also stop after an idle timeout
+/// if it has no connections, an empty inbox, and no other senders than the one stored
+/// in the endpoint map exist.
 #[derive(Debug)]
 pub(super) struct EndpointStateHandle {
     /// Sender for the channel into the [`EndpointStateActor`].
-    pub(super) sender: mpsc::Sender<EndpointStateMessage>,
+    ///
+    /// This is a [`GuardedSender`], from which we can get a sender but only if the receiver
+    /// hasn't been closed.
+    pub(super) sender: GuardedSender<EndpointStateMessage>,
     _task: AbortOnDropHandle<()>,
 }
 

iroh/src/magicsock/endpoint_map/endpoint_state/guarded_channel.rs

@@ -0,0 +1,78 @@
+use std::sync::{Arc, Mutex};
+
+use tokio::sync::mpsc;
+
+/// Creates a new [`mpsc`] channel where the receiver can only close if there are no active senders.
+pub(super) fn guarded_channel<T>(cap: usize) -> (GuardedSender<T>, GuardedReceiver<T>) {
+    let (tx, rx) = mpsc::channel(cap);
+    let tx = Arc::new(Mutex::new(Some(tx)));
+    (GuardedSender { tx: tx.clone() }, GuardedReceiver { tx, rx })
+}
+
+#[derive(Debug)]
+pub(crate) struct GuardedSender<T> {
+    tx: Arc<Mutex<Option<mpsc::Sender<T>>>>,
+}
+
+impl<T> GuardedSender<T> {
+    /// Returns a sender to the channel.
+    ///
+    /// Returns a new sender if the channel is not closed. It is guaranteed that
+    /// [`GuardedReceiver::close_if_idle`] will not return `true` until the sender is dropped.
+    /// Returns `None` if the channel has been closed.
+    pub(crate) fn get(&self) -> Option<mpsc::Sender<T>> {
+        self.tx.lock().expect("poisoned").clone()
+    }
+
+    /// Returns `true` if the channel has been closed.
+    pub(crate) fn is_closed(&self) -> bool {
+        self.tx.lock().expect("poisoned").is_none()
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct GuardedReceiver<T> {
+    rx: mpsc::Receiver<T>,
+    tx: Arc<Mutex<Option<mpsc::Sender<T>>>>,
+}
+
+impl<T> GuardedReceiver<T> {
+    /// Receives the next value for this receiver.
+    ///
+    /// See [`mpsc::Receiver::recv`].
+    pub(super) async fn recv(&mut self) -> Option<T> {
+        self.rx.recv().await
+    }
+
+    /// Returns `true` if the inbox is empty and no senders to the inbox exist.
+    pub(super) fn is_idle(&self) -> bool {
+        self.rx.is_empty() && self.rx.sender_strong_count() <= 1
+    }
+
+    /// Closes the channel if the channel is idle.
+    ///
+    /// Returns `true` if the channel is idle and has now been closed, and `false` if the channel
+    /// is not idle and therefore has not been not closed.
+    ///
+    /// Uses a lock internally to make sure that there cannot be a race condition between
+    /// calling this and a new sender being created.
+    pub(super) fn close_if_idle(&mut self) -> bool {
+        let mut guard = self.tx.lock().expect("poisoned");
+        if self.is_idle() {
+            *guard = None;
+            self.rx.close();
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl<T> Drop for GuardedReceiver<T> {
+    fn drop(&mut self) {
+        let mut guard = self.tx.lock().expect("poisoned");
+        *guard = None;
+        self.rx.close();
+        drop(guard)
+    }
+}