From 54a8d7b3409f6a5857b6fc9a964a36b66c28ae25 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 30 Oct 2026 15:36:47 +0100 Subject: refactor transports to start themselves --- src/main.rs | 102 ++---------------------------------------- src/transport/libp2p.rs | 35 +++++++++++++++ src/transport/mod.rs | 34 ++++++++++++++ src/transport/mycelium/mod.rs | 73 ++++++++++++++++++++++++++---- src/transport/tor.rs | 22 +++++++++ 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)> = Vec::new(); for name in &args.transports { - let transport: (&'static str, Arc) = 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, - ) - } - 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 { - let parse = |option: &str, addresses: &[String]| { - addresses - .iter() - .map(|address| { - address - .parse::() - .with_context(|| format!("{option} {address:?}")) - }) - .collect::>>() - }; - 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::>>()?, - 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 = 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 { + let parse = |option: &str, addresses: &[String]| { + addresses + .iter() + .map(|address| { + address + .parse::() + .with_context(|| format!("{option} {address:?}")) + }) + .collect::>>() + }; + 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::>>()?, + 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, + stop: tokio::sync::watch::Receiver, +) -> Result)>> { + use std::sync::Arc; + let _ = key; + let started: Option<(&'static str, Arc)> = match name { + "mycelium" => mycelium::start(settings, stop) + .await? + .map(|transport| (mycelium::SCHEME, Arc::new(transport) as Arc)), + #[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, + )) + } + #[cfg(feature = "tor")] + "tor" => tor::TorTransport::from_settings(settings, key, stop) + .await? + .map(|transport| (tor::SCHEME, Arc::new(transport) as Arc)), + 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, +) -> Result> { + 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 { + 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 { 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, + mut stop: watch::Receiver, + ) -> Result> { + 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 { let tor_config = TorClientConfigBuilder::from_directories( config.state_dir.join("state"), -- cgit v1.3.1