1 files changed, 47 insertions, 114 deletions

diff --git a/src/main.rs b/src/main.rs
index f16138e..a10eb73 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,4 +1,5 @@
-//! continuum: a replicated Automerge key-value store synchronized over Mycelium.
+//! continuum: a replicated Automerge document, read and written by path, synchronized
+//! over Mycelium, libp2p or Tor.
 //!
 //! main.rs only wires things together: arguments, the starting document, the node
 //! (`node`), bootstrap and the HTTP server.
@@ -8,7 +9,6 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
 use anyhow::{Context, Result};
-use automerge::AutoCommit;
 use tokio::sync::watch;
 use tracing::{debug, error, info, warn};
 
@@ -34,10 +34,10 @@ mod transport;
 mod units;
 
 use api::auth::AuthState;
+use document::Document;
 use document::replica::Epoch;
 use swarm::bootstrap;
 use transport::Transport;
-use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig};
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -52,6 +52,19 @@ async fn main() -> Result<()> {
         return Ok(());
     }
     let args = &config.settings;
+    if let Some(format) = logging::LogFormat::parse(&args.log_format) {
+        logging::set_format(format);
+    }
+    for (transport, built, options) in config.unused_transport_options() {
+        warn!(
+            transport,
+            built,
+            options = ?options,
+            "Options of a transport this node doesn't use are ignored"
+        );
+    }
+    // Only with libp2p: without, `config` refuses `relay-only` already.
+    #[cfg(feature = "libp2p")]
     if args.relay_only {
         return relay_only(args).await;
     }
@@ -63,11 +76,14 @@ async fn main() -> Result<()> {
         .context("node-key-file is required")?;
     let node_key = Arc::new(identity::NodeKey::load(
         node_key_file,
-        args.create_node_key,
+        args.node_key_create,
     )?);
     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)?)),
+        Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(
+            path,
+            args.swarm_key_create,
+        )?)),
         None => {
             warn!("No swarm key: any node that can reach a member may join");
             None
@@ -75,34 +91,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,
@@ -158,7 +151,7 @@ async fn main() -> Result<()> {
             Some((starting.swarm_id, starting.origin)),
             starting.earlier_actors,
         ),
-        None => (AutoCommit::new(), None, None, Vec::new()),
+        None => (Document::new(), None, None, Vec::new()),
     };
     let mut initial_doc = initial_doc;
     if let Some(spec) = &args.remake {
@@ -176,46 +169,12 @@ async fn main() -> Result<()> {
     };
     debug!(epoch = %initial_epoch, "Local epoch");
 
-    // In the order of preference (`--transports`, checked when loading).
+    // The enabled, in the order of preference (`--transport-order`, 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,
@@ -229,6 +188,12 @@ async fn main() -> Result<()> {
         .map(|(_, transport)| transport.delivery_limit())
         .max()
         .unwrap_or_default();
+    // An answer may come over any of them.
+    let sync_reply_timeout = transports
+        .iter()
+        .map(|(_, transport)| transport.sync_reply_timeout())
+        .max()
+        .unwrap_or(state::SYNC_REPLY_TIMEOUT);
     let mut transports = transports.into_iter();
     let (scheme, transport) = transports.next().context("No transports")?;
 
@@ -250,10 +215,13 @@ async fn main() -> Result<()> {
         initial_epoch,
         initial_doc,
         earlier_actors,
-        timing: state::Timing::from_keepalive(args.keepalive, delivery_limit),
+        timing: state::Timing::new(args.keepalive, sync_reply_timeout, delivery_limit)
+            .with_probe_max(args.dormant_probe_max),
         neighbours_k: args.neighbours,
         alias: args.alias.clone(),
         electorate_window: args.electorate_window,
+        clock_skew_warn: args.clock_skew_warn,
+        superseded_writes: args.superseded_writes,
         // At most 4MiB (`Settings::check`).
         blob_chunk_bytes: args.blob_chunk as u32,
         swarm_key,
@@ -298,15 +266,14 @@ async fn main() -> Result<()> {
 
     info!(address = %listener.local_addr()?, "HTTP API listening");
 
-    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
-
+    // The node shutting down (`AppState::closing`): the server drains.
+    let closing = state.closing.subscribe();
     let server_builder = axum::serve(listener, app).with_graceful_shutdown(async move {
-        let _ = shutdown_rx.await;
+        stopped(closing).await;
         info!("Shutdown signal received. Draining API connections...")
     });
 
     let server_fut = server_builder.into_future();
-    let stopping = state.clone();
     tokio::pin!(server_fut);
 
     // Joins, then runs until stopped; returns only if joining fails.
@@ -328,14 +295,14 @@ async fn main() -> Result<()> {
 
     tokio::select! {
         failed = joining.as_mut() => {
-            let _ = shutdown_tx.send(());
+            state.closing.send_replace(true);
             let _ = server_fut.as_mut().await;
             return failed;
         }
 
         exit_status = state.transport.wait_for_exit() => {
             error!(status = ?exit_status, "A transport stopped");
-            let _ = shutdown_tx.send(());
+            state.closing.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
 
@@ -348,7 +315,7 @@ async fn main() -> Result<()> {
         }
 
         _ = stopped(stop.clone()) => {
-            let _ = shutdown_tx.send(());
+            state.closing.send_replace(true);
             let _ = server_fut.as_mut().await;
         }
     }
@@ -358,41 +325,12 @@ async fn main() -> Result<()> {
         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;
+    swarm::peer::farewell(&state).await;
+    swarm::state_file::update(&state, true).await;
+    swarm::document_file::flush(&state).await;
     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,
-        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")]
@@ -400,7 +338,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)");
 
@@ -435,11 +373,6 @@ async fn relay_only(args: &config::Settings) -> Result<()> {
     Ok(())
 }
 
-#[cfg(not(feature = "libp2p"))]
-async fn relay_only(_args: &config::Settings) -> Result<()> {
-    anyhow::bail!("relay-only relays over libp2p; this build has no libp2p")
-}
-
 /// Turns SIGINT and SIGTERM into a shutdown request. A second signal exits at once (the
 /// supervised Mycelium still gets its parent-death signal then).
 fn watch_for_shutdown() -> Result<watch::Receiver<bool>> {