refactor transports to start themselves

5 files changed, 160 insertions, 106 deletions

diff --git a/src/main.rs b/src/main.rs
index 7481f38..1908775 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -37,7 +37,6 @@ use api::auth::AuthState;
 use document::replica::Epoch;
 use swarm::bootstrap;
 use transport::Transport;
-use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig};
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -78,34 +77,11 @@ async fn main() -> Result<()> {
     };
 
     // First thing: until a handler is installed, SIGINT and SIGTERM kill the process on
-    // the spot, and startup (spawning Mycelium, bootstrap) can take minutes.
+    // the spot, and startup (starting transports, bootstrap) can take minutes.
     let stop = watch_for_shutdown().context("Installing signal handlers")?;
     // A broken htpasswd will abort startup.
     let auth_state = AuthState::new(args.htpasswd.as_deref().context("htpasswd is required")?)?;
 
-    let uses = |name: &str| args.transports.iter().any(|transport| transport == name);
-    let mut mycelium_setup = if uses("mycelium") {
-        let spawn_config = MyceliumSpawnConfig {
-            rest_url_opt: args.mycelium_rest_url(),
-            key_file_opt: args.mycelium_key_file.clone(),
-            network_key_file: args.mycelium_network_key_file.clone(),
-            network_name: args.mycelium_network_name.clone(),
-            tcp_port: args.mycelium_tcp_listen_port,
-            peers: args.mycelium_peers.clone(),
-        };
-        let setup = tokio::select! {
-            setup = MyceliumSetup::bootstrap(spawn_config) => setup?,
-            _ = stopped(stop.clone()) => return Ok(()),
-        };
-        // Without a password it may carry (the option is secret).
-        let mut shown_url = setup.rest_url.clone();
-        let _ = shown_url.set_password(None);
-        debug!(url = shown_url.to_string(), "Using Mycelium REST API");
-        Some(setup)
-    } else {
-        None
-    };
-
     info!(
         bind_addr = args.bind_addr.to_string(),
         transports = ?args.transports,
@@ -182,43 +158,8 @@ async fn main() -> Result<()> {
     // In the order of preference (`--transports`, checked when loading).
     let mut transports: Vec<(&'static str, Arc<dyn Transport>)> = Vec::new();
     for name in &args.transports {
-        let transport: (&'static str, Arc<dyn Transport>) = match name.as_str() {
-            "mycelium" => {
-                let setup = mycelium_setup
-                    .take()
-                    .context("Mycelium named twice in transports")?;
-                (
-                    transport::mycelium::SCHEME,
-                    Arc::new(mycelium::MyceliumTransport::connect(setup).await?),
-                )
-            }
-            #[cfg(feature = "libp2p")]
-            "libp2p" => (
-                transport::libp2p::SCHEME,
-                Arc::new(transport::libp2p::Libp2pTransport::start(libp2p_config(
-                    args, false,
-                )?)?),
-            ),
-            #[cfg(feature = "tor")]
-            "tor" => {
-                let state_dir = args
-                    .tor_state_dir
-                    .clone()
-                    .context("tor needs tor-state-dir")?;
-                let transport = tokio::select! {
-                    transport = transport::tor::TorTransport::start(transport::tor::TorConfig {
-                        key: Arc::clone(&node_key),
-                        state_dir,
-                        port: args.tor_port,
-                    }) => transport?,
-                    _ = stopped(stop.clone()) => return Ok(()),
-                };
-                (
-                    transport::tor::SCHEME,
-                    Arc::new(transport) as Arc<dyn Transport>,
-                )
-            }
-            other => anyhow::bail!("transports: no {other} in this build"),
+        let Some(transport) = transport::start(name, args, &node_key, stop.clone()).await? else {
+            return Ok(());
         };
         info!(
             transport = transport.0,
@@ -368,41 +309,6 @@ async fn main() -> Result<()> {
     Ok(())
 }
 
-/// The libp2p transport's configuration; `relay_only`: a relay that is no node.
-#[cfg(feature = "libp2p")]
-fn libp2p_config(
-    args: &config::Settings,
-    relay_only: bool,
-) -> Result<transport::libp2p::Libp2pConfig> {
-    let parse = |option: &str, addresses: &[String]| {
-        addresses
-            .iter()
-            .map(|address| {
-                address
-                    .parse::<libp2p::Multiaddr>()
-                    .with_context(|| format!("{option} {address:?}"))
-            })
-            .collect::<Result<Vec<_>>>()
-    };
-    Ok(transport::libp2p::Libp2pConfig {
-        key: transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?,
-        listen: parse("libp2p-listen", &args.libp2p_listen)?,
-        announce: parse("libp2p-announce", &args.libp2p_announce)?,
-        relays: parse("libp2p-relays", &args.libp2p_relays)?,
-        relay: args.libp2p_relay || relay_only,
-        continuum: !relay_only,
-        mdns: args.libp2p_mdns,
-        upnp: args.libp2p_upnp,
-        ignore: args
-            .libp2p_ignore
-            .iter()
-            .map(|range| range.parse().context("libp2p-ignore"))
-            .collect::<Result<Vec<_>>>()?,
-        public: args.libp2p_public,
-        memory: false,
-    })
-}
-
 /// `--relay-only`: libp2p relaying for nodes of swarms, and the HTTP API's relay part
 /// (with `--htpasswd`), until stopped.
 #[cfg(feature = "libp2p")]
@@ -410,7 +316,7 @@ async fn relay_only(args: &config::Settings) -> Result<()> {
     let stop = watch_for_shutdown().context("Installing signal handlers")?;
     let auth_state = args.htpasswd.as_deref().map(AuthState::new).transpose()?;
     let transport: Arc<dyn Transport> = Arc::new(transport::libp2p::Libp2pTransport::start(
-        libp2p_config(args, true)?,
+        transport::libp2p::Libp2pConfig::from_settings(args, true)?,
     )?);
     info!(peer_id = transport.local_id(), "Relaying only (no node)");
 
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 610b94e..2d67d2c 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -205,6 +205,41 @@ pub(crate) struct Libp2pConfig {
 }
 
 impl Libp2pConfig {
+    /// The configuration `settings` give; `relay_only`: a relay that is no node
+    /// (`--relay-only`).
+    pub(crate) fn from_settings(
+        settings: &crate::config::Settings,
+        relay_only: bool,
+    ) -> Result<Self> {
+        let parse = |option: &str, addresses: &[String]| {
+            addresses
+                .iter()
+                .map(|address| {
+                    address
+                        .parse::<Multiaddr>()
+                        .with_context(|| format!("{option} {address:?}"))
+                })
+                .collect::<Result<Vec<_>>>()
+        };
+        Ok(Self {
+            key: Self::load_key(settings.libp2p_key_file.as_deref())?,
+            listen: parse("libp2p-listen", &settings.libp2p_listen)?,
+            announce: parse("libp2p-announce", &settings.libp2p_announce)?,
+            relays: parse("libp2p-relays", &settings.libp2p_relays)?,
+            relay: settings.libp2p_relay || relay_only,
+            continuum: !relay_only,
+            mdns: settings.libp2p_mdns,
+            upnp: settings.libp2p_upnp,
+            ignore: settings
+                .libp2p_ignore
+                .iter()
+                .map(|range| range.parse().context("libp2p-ignore"))
+                .collect::<Result<Vec<_>>>()?,
+            public: settings.libp2p_public,
+            memory: false,
+        })
+    }
+
     /// The key in `path` (32 raw bytes, an Ed25519 secret key), or a new one each start
     /// without a file: the peer ID only says where a node is, so a new one costs nothing
     /// but the WHO that finds it again (`transport::identity`).
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b10c799..9d43e5a 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -28,6 +28,40 @@ pub(crate) mod tor;
 pub(crate) mod traffic;
 pub(crate) mod wire;
 
+/// Starts the transport `name` (one of `config::TRANSPORTS`) as `settings` configure
+/// it: its scheme and the transport, or None if `stop` came first (some take a while to
+/// start: a supervised Mycelium daemon, arti). `key`: the node key, which some prove
+/// themselves.
+pub(crate) async fn start(
+    name: &str,
+    settings: &crate::config::Settings,
+    key: &std::sync::Arc<crate::identity::NodeKey>,
+    stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<(&'static str, std::sync::Arc<dyn Transport>)>> {
+    use std::sync::Arc;
+    let _ = key;
+    let started: Option<(&'static str, Arc<dyn Transport>)> = match name {
+        "mycelium" => mycelium::start(settings, stop)
+            .await?
+            .map(|transport| (mycelium::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+        #[cfg(feature = "libp2p")]
+        "libp2p" => {
+            let _ = stop;
+            let config = libp2p::Libp2pConfig::from_settings(settings, false)?;
+            Some((
+                libp2p::SCHEME,
+                Arc::new(libp2p::Libp2pTransport::start(config)?) as Arc<dyn Transport>,
+            ))
+        }
+        #[cfg(feature = "tor")]
+        "tor" => tor::TorTransport::from_settings(settings, key, stop)
+            .await?
+            .map(|transport| (tor::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+        other => anyhow::bail!("transports: no {other} in this build"),
+    };
+    Ok(started)
+}
+
 /// Random identifier of one run of a node (from start to exit).
 pub(crate) type RunId = [u8; 16];
 
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index d33e323..836eca9 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -91,6 +91,70 @@ pub(crate) struct MyceliumTransport {
     local_id: String,
 }
 
+/// The transport as `settings` configure it (`transport::start`): its daemon supervised
+/// (started, which can take a while) or one running elsewhere. None if `stop` came
+/// first.
+pub(crate) async fn start(
+    settings: &crate::config::Settings,
+    mut stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<MyceliumTransport>> {
+    let spawn_config = MyceliumSpawnConfig {
+        rest_url_opt: settings.mycelium_rest_url(),
+        key_file_opt: settings.mycelium_key_file.clone(),
+        network_key_file: settings.mycelium_network_key_file.clone(),
+        network_name: settings.mycelium_network_name.clone(),
+        tcp_port: settings.mycelium_tcp_listen_port,
+        peers: settings.mycelium_peers.clone(),
+    };
+    let setup = tokio::select! {
+        setup = MyceliumSetup::bootstrap(spawn_config) => setup?,
+        _ = stop.wait_for(|stop| *stop) => return Ok(None),
+    };
+    // Without a password it may carry (the option is secret).
+    let mut shown_url = setup.rest_url.clone();
+    let _ = shown_url.set_password(None);
+    tracing::debug!(url = shown_url.to_string(), "Using Mycelium REST API");
+    tokio::select! {
+        transport = MyceliumTransport::connect(setup) => Ok(Some(transport?)),
+        _ = stop.wait_for(|stop| *stop) => Ok(None),
+    }
+}
+
+/// How long the daemon's API may take to come up: a daemon just started listens a
+/// moment later (`admin_info`).
+const API_READY_WITHIN: Duration = Duration::from_secs(30);
+
+/// The daemon's identity, from its API at `url` (`/api/v1/admin`), once the API answers:
+/// a daemon we just started doesn't listen yet, so refused connections are tried again
+/// (50 ms, doubling up to a second) for `API_READY_WITHIN`; anything else is an error.
+async fn admin_info(client: &reqwest::Client, url: Url) -> Result<MyceliumAdminInfo> {
+    let deadline = tokio::time::Instant::now() + API_READY_WITHIN;
+    let mut wait = Duration::from_millis(50);
+    loop {
+        match client.get(url.clone()).send().await {
+            Ok(response) => {
+                return response
+                    .error_for_status()?
+                    .json()
+                    .await
+                    .context("Querying local Mycelium identity");
+            }
+            Err(error) if error.is_connect() && tokio::time::Instant::now() + wait < deadline => {
+                tracing::debug!(url = %url, error = %error, "Mycelium's API doesn't answer yet; trying again");
+                tokio::time::sleep(wait).await;
+                wait = (wait * 2).min(Duration::from_secs(1));
+            }
+            Err(error) => {
+                return Err(error).with_context(|| {
+                    format!(
+                        "Querying local Mycelium identity (no answer within {API_READY_WITHIN:?})"
+                    )
+                });
+            }
+        }
+    }
+}
+
 impl MyceliumTransport {
     pub(crate) async fn connect(setup: MyceliumSetup) -> Result<Self> {
         let rest_url = setup.rest_url.clone();
@@ -102,14 +166,7 @@ impl MyceliumTransport {
             .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
             .build()?;
 
-        let admin: MyceliumAdminInfo = client
-            .get(rest_url.join("/api/v1/admin")?)
-            .send()
-            .await?
-            .error_for_status()?
-            .json()
-            .await
-            .context("Querying local Mycelium identity")?;
+        let admin = admin_info(&client, rest_url.join("/api/v1/admin")?).await?;
 
         let transport = Self {
             setup: tokio::sync::Mutex::new(setup),
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index 2cf3249..8eb1d02 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -198,6 +198,28 @@ impl TorTransport {
     /// while: it fetches what it needs to know of the Tor network). Until it is done,
     /// dials wait for it (arti bootstraps on demand), and the onion service is published
     /// once it is: the node starts without waiting for Tor.
+    /// The transport as `settings` configure it (`transport::start`), proving `key`.
+    /// None if `stop` came first (bootstrapping arti can take a while).
+    pub(crate) async fn from_settings(
+        settings: &crate::config::Settings,
+        key: &Arc<NodeKey>,
+        mut stop: watch::Receiver<bool>,
+    ) -> Result<Option<Self>> {
+        let state_dir = settings
+            .tor_state_dir
+            .clone()
+            .context("tor needs tor-state-dir")?;
+        let config = TorConfig {
+            key: Arc::clone(key),
+            state_dir,
+            port: settings.tor_port,
+        };
+        tokio::select! {
+            transport = Self::start(config) => transport.map(Some),
+            _ = stop.wait_for(|stop| *stop) => Ok(None),
+        }
+    }
+
     pub(crate) async fn start(config: TorConfig) -> Result<Self> {
         let tor_config = TorClientConfigBuilder::from_directories(
             config.state_dir.join("state"),