1 files changed, 58 insertions, 26 deletions

diff --git a/src/main.rs b/src/main.rs
index f2494d8..f174c4e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -5,6 +5,7 @@
 
 use std::future::IntoFuture;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use anyhow::{Context, Result};
 use automerge::AutoCommit;
@@ -53,7 +54,10 @@ async fn main() -> Result<()> {
     let args = &config.settings;
 
     // Read before anything starts, so a bad key file fails right away.
-    let node_key = identity::NodeKey::load(&args.node_key_file, args.create_node_key)?;
+    let node_key = Arc::new(identity::NodeKey::load(
+        &args.node_key_file,
+        args.create_node_key,
+    )?);
     info!(node_id = %node_key.id(), "Node identity");
     let swarm_key = match &args.swarm_key_file {
         Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)),
@@ -205,11 +209,12 @@ async fn main() -> Result<()> {
                             .with_context(|| format!("libp2p-relays {address:?}"))
                     })
                     .collect::<Result<Vec<_>>>()?;
-                let key = transport::p2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?;
+                let key =
+                    transport::libp2p::Libp2pConfig::load_key(args.libp2p_key_file.as_deref())?;
                 (
                     "libp2p",
-                    Arc::new(transport::p2p::Libp2pTransport::start(
-                        transport::p2p::Libp2pConfig {
+                    Arc::new(transport::libp2p::Libp2pTransport::start(
+                        transport::libp2p::Libp2pConfig {
                             key,
                             listen,
                             announce,
@@ -222,9 +227,25 @@ async fn main() -> Result<()> {
                     )?),
                 )
             }
+            #[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(()),
+                };
+                ("tor", Arc::new(transport) as Arc<dyn Transport>)
+            }
             other => anyhow::bail!("transports: no {other} in this build"),
         };
-        debug!(
+        info!(
             transport = transport.0,
             address = transport.1.local_id(),
             "Transport ready"
@@ -250,7 +271,7 @@ async fn main() -> Result<()> {
     let node = node::Node::new(node::NodeConfig {
         transport,
         identity: Some(transport::identity::IdentityConfig {
-            key: Arc::new(node_key),
+            key: Arc::clone(&node_key),
             scheme,
             more: transports.collect(),
         }),
@@ -294,24 +315,9 @@ async fn main() -> Result<()> {
 
     node.spawn_workers();
 
-    let joining = bootstrap::bootstrap(&state, &seeds, start);
-    tokio::select! {
-        joined = joining => joined.context("Joining swarm")?,
-        _ = stopped(stop.clone()) => {
-            info!("Stopped while joining the swarm");
-            return Ok(());
-        }
-    }
-
-    node::spawn_ready_workers(&state);
-
-    info!(
-        swarm_id = ?state.swarm_id().await,
-        ready = state.is_ready().await,
-        "Bootstrap completed"
-    );
-    swarm::state_file::update(&state, true).await;
-
+    // The API is up while the node joins: what it serves then is the handlers' business
+    // (`docs/lifecycle.md`), and /peers and /admin/stats show how joining goes.
+    let joined = AtomicBool::new(false);
     let app = api::router(state.clone(), auth_state, args.debug_api);
 
     let listener = tokio::net::TcpListener::bind(args.bind_addr)
@@ -331,9 +337,32 @@ async fn main() -> Result<()> {
     let stopping = state.clone();
     tokio::pin!(server_fut);
 
+    // Joins, then runs until stopped; returns only if joining fails.
+    let joining = async {
+        bootstrap::bootstrap(&state, &seeds, start)
+            .await
+            .context("Joining swarm")?;
+        node::spawn_ready_workers(&state);
+        info!(
+            swarm_id = ?state.swarm_id().await,
+            ready = state.is_ready().await,
+            "Bootstrap completed"
+        );
+        swarm::state_file::update(&state, true).await;
+        joined.store(true, Ordering::Release);
+        std::future::pending::<Result<()>>().await
+    };
+    tokio::pin!(joining);
+
     tokio::select! {
+        failed = joining.as_mut() => {
+            let _ = shutdown_tx.send(());
+            let _ = server_fut.as_mut().await;
+            return failed;
+        }
+
         exit_status = state.transport.wait_for_exit() => {
-            error!("Mycelium runtime process exited. Status: {:?}", exit_status);
+            error!(status = ?exit_status, "A transport stopped");
             let _ = shutdown_tx.send(());
             let _ = server_fut.as_mut().await;
         }
@@ -344,7 +373,6 @@ async fn main() -> Result<()> {
             } else {
                 info!("Axum HTTP engine shutdown cleanly.");
             }
-            std::mem::drop(state);
         }
 
         _ = stopped(stop.clone()) => {
@@ -354,6 +382,10 @@ async fn main() -> Result<()> {
     }
 
     info!("HTTP API server stopped");
+    if !joined.load(Ordering::Acquire) {
+        info!("Stopped while joining the swarm");
+        return Ok(());
+    }
     swarm::peer::farewell(&stopping).await;
     swarm::state_file::update(&stopping, true).await;
     document::file::flush(&stopping).await;