summaryrefslogtreecommitdiffstats
path: root/src/main.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/main.rs')
-rw-r--r--src/main.rs128
1 files changed, 47 insertions, 81 deletions
diff --git a/src/main.rs b/src/main.rs
index 67cb366..217a9a4 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -5,20 +5,18 @@
use std::collections::HashMap;
use std::future::IntoFuture;
-use std::net::SocketAddr;
-use std::path::PathBuf;
-use std::sync::atomic::AtomicBool;
use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
use std::time::Duration;
use anyhow::{Context, Result};
use automerge::AutoCommit;
-use clap::Parser;
-use tokio::sync::{broadcast, mpsc, watch, RwLock};
+use tokio::sync::{RwLock, broadcast, mpsc, watch};
use tracing::{debug, error, info, trace, warn};
-use url::Url;
mod api;
+mod backoff;
+mod config;
mod document;
mod logging;
mod remake;
@@ -30,77 +28,21 @@ use api::auth::AuthState;
use document::replica::{Epoch, Replica};
use state::{AppState, NodeLifecycle, Timing};
use swarm::{bootstrap, peer, sync};
+use transport::Transport;
use transport::firewall::{Firewall, FirewallTransport};
use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig};
-use transport::Transport;
-
-#[derive(Parser, Debug)]
-#[command(version, about = "continuum messaging system", long_about = None)]
-struct Args {
- #[arg(long)]
- debug_api: bool,
-
- #[arg(long, value_name = "HOSTNAME:PORT", default_value = "127.0.0.1:3000")]
- bind_addr: SocketAddr,
-
- #[arg(
- long,
- value_name = "SCHEME://HOSTNAME:PORT",
- default_value = None,
- conflicts_with_all = &["mycelium_key_file", "mycelium_network_key_file", "mycelium_network_name"],
- )]
- mycelium_rest_url: Option<Url>,
-
- #[arg(long)]
- pub mycelium_key_file: Option<PathBuf>,
-
- #[arg(long)]
- pub mycelium_network_key_file: Option<PathBuf>,
-
- #[arg(long)]
- pub mycelium_network_name: Option<String>,
-
- #[arg(long)]
- pub mycelium_tcp_listen_port: Option<u16>,
-
- #[arg(
- long,
- use_value_delimiter = true,
- default_value = "tls://127.0.0.1:9651"
- )]
- /// Peers of the supervised Mycelium daemon (overlay links such as
- /// `tls://host:9651`), not continuum peers.
- pub mycelium_peers: Vec<String>,
-
- /// Seeds: how to reach a node of the swarm to join, as a Mycelium public key, a
- /// Mycelium IPv6 address, or a host name with an AAAA record for one. Without seeds,
- /// the node starts a new swarm (genesis).
- #[arg(long, value_name = "STRING,[STRING...]", value_delimiter = ',')]
- seeds: Option<Vec<String>>,
-
- #[arg(long, value_name = "PATH")]
- htpasswd: String,
-
- /// Pre-shared swarm key: only nodes that know it may join (raw bytes, at least 16;
- /// e.g. `head -c 32 /dev/urandom`). Without it, anyone who knows the swarm ID can.
- #[arg(long, value_name = "PATH")]
- swarm_key_file: Option<PathBuf>,
-
- /// Keepalive interval; dormancy, resync and probe timings derive from it.
- #[arg(long, value_name = "SECONDS", default_value_t = 60)]
- keepalive_secs: u64,
-
- /// Test hook: start on a non-genesis epoch of this generation (with a fixed dummy
- /// id), to check that nodes on different epochs refuse to sync.
- #[arg(long, hide = true, requires = "debug_api", value_name = "GENERATION")]
- debug_epoch: Option<u64>,
-}
+use transport::traffic::{CountingTransport, Traffic};
#[tokio::main]
async fn main() -> Result<()> {
logging::init();
- let args = Args::parse();
+ let config = config::load()?;
+ if config.print_only {
+ println!("{}", serde_json::to_string_pretty(&config.report())?);
+ return Ok(());
+ }
+ let args = &config.settings;
// Read before anything starts, so a bad key file fails right away.
let swarm_key = match &args.swarm_key_file {
@@ -117,13 +59,14 @@ async fn main() -> Result<()> {
let mycelium_setup = {
let mut spawn_config = MyceliumSpawnConfig {
- rest_url_opt: args.mycelium_rest_url,
- key_file_opt: args.mycelium_key_file,
- network_key_file: args.mycelium_network_key_file,
- peers: args.mycelium_peers,
+ rest_url_opt: args.mycelium_rest_url(),
+ key_file_opt: args.mycelium_key_file.clone(),
+ network_key_file: args.mycelium_network_key_file.clone(),
+ tcp_port: args.mycelium_tcp_listen_port,
+ peers: args.mycelium_peers.clone(),
..Default::default()
};
- if let Some(net_name) = args.mycelium_network_name {
+ if let Some(net_name) = args.mycelium_network_name.clone() {
spawn_config.network_name = net_name;
}
tokio::select! {
@@ -143,7 +86,7 @@ async fn main() -> Result<()> {
"Using Mycelium REST API"
);
- let seeds: Vec<String> = args.seeds.unwrap_or_default();
+ let seeds: Vec<String> = args.seeds.clone();
debug!(seed_count = seeds.len(), "Loaded bootstrap configuration");
let (sync_tx, sync_rx) = mpsc::unbounded_channel();
@@ -161,6 +104,7 @@ async fn main() -> Result<()> {
debug!(epoch = %initial_epoch, "Local epoch");
let firewall = Arc::new(Firewall::default());
+ let traffic = Arc::new(Traffic::default());
let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?);
debug!(local_id = mycelium_transport.local_id(), "Transport ready");
@@ -172,11 +116,18 @@ async fn main() -> Result<()> {
discovered_swarm_id: None,
})),
peer_states: Arc::new(RwLock::new(HashMap::new())),
- transport: Arc::new(FirewallTransport::new(
- mycelium_transport,
- Arc::clone(&firewall),
+ // Counted outside the firewall: what it drops outbound counts as sent (it plays a
+ // lossy network), what it drops inbound never arrived.
+ transport: Arc::new(CountingTransport::new(
+ Arc::new(FirewallTransport::new(
+ mycelium_transport,
+ Arc::clone(&firewall),
+ )),
+ Arc::clone(&traffic),
)),
firewall,
+ traffic,
+ stats: Default::default(),
sync_events: sync_tx,
change_tx,
initial_sync_done: Arc::new(AtomicBool::new(false)),
@@ -184,7 +135,22 @@ async fn main() -> Result<()> {
timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)),
remake: Arc::new(remake::RemakeState::default()),
swarm_key,
+ run_id: uuid::Uuid::new_v4().into_bytes(),
+ neighbours_k: args.neighbours,
+ alias: args.alias.clone(),
+ electorate_window: Duration::from_secs(args.electorate_window_days * 24 * 3600),
+ blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)),
+ protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()),
+ config: Arc::new(config.report()),
+ neighbours: Default::default(),
+ pulling: Default::default(),
+ announce_pending: Default::default(),
+ rtt: Default::default(),
+ throughput: Default::default(),
+ stability: Default::default(),
stale: Default::default(),
+ catch_up_source: Default::default(),
+ catch_up_preference: Default::default(),
};
trace!("Application state initialized");
@@ -194,7 +160,7 @@ async fn main() -> Result<()> {
let inner_state = consumer_state.clone();
let _ = consumer_state
.transport
- .start_consumer(Box::new(move |sender, msg| {
+ .start_consumer(Box::new(move |sender, msg, _payload_len| {
let state_clone = inner_state.clone();
Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await })
}))
@@ -287,7 +253,7 @@ async fn main() -> Result<()> {
/// 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>> {
- use tokio::signal::unix::{signal, SignalKind};
+ use tokio::signal::unix::{SignalKind, signal};
let mut interrupt = signal(SignalKind::interrupt())?;
let mut terminate = signal(SignalKind::terminate())?;
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 20:19:19 +0000