refactor transports to start themselves

4 files changed, 156 insertions, 8 deletions

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"),