diff options
45 files changed, 5666 insertions, 756 deletions
@@ -1,7 +1,10 @@ [package] name = "continuum" version = "1.0.0" -edition = "2021" +edition = "2024" + +[features] +tokio-console = ["dep:console-subscriber"] [dependencies] anyhow = "1.0" @@ -11,7 +14,7 @@ async-stream = "0.3" async-trait = "0.1" automerge = "0.12" axum = "0.8" -console-subscriber = "0.5.0" +console-subscriber = { version = "0.5.0", optional = true } base64 = "0.23" blake3 = "1.8" bytes = "1.12" diff --git a/src/api/auth.rs b/src/api/auth.rs index 3926c70..70b1af5 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -1,14 +1,14 @@ -use argon2::{password_hash::PasswordVerifier, Argon2}; +use argon2::{Argon2, password_hash::PasswordVerifier}; use axum::extract::State; use axum::{ body::Body, http::{ - header::{AUTHORIZATION, WWW_AUTHENTICATE}, HeaderMap, Request, StatusCode, + header::{AUTHORIZATION, WWW_AUTHENTICATE}, }, response::{IntoResponse, Response}, }; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use dashmap::DashMap; use password_hash::phc::PasswordHash; use std::{ diff --git a/src/api/debug.rs b/src/api/debug.rs index 929d665..6a56a9c 100644 --- a/src/api/debug.rs +++ b/src/api/debug.rs @@ -2,7 +2,7 @@ use std::collections::HashSet; -use axum::{body::Bytes, extract::State, http::StatusCode, Json}; +use axum::{Json, body::Bytes, extract::State, http::StatusCode}; use tracing::{info, warn}; use crate::state::AppState; diff --git a/src/api/document.rs b/src/api/document.rs new file mode 100644 index 0000000..303f5c9 --- /dev/null +++ b/src/api/document.rs @@ -0,0 +1,77 @@ +//! `GET /admin/document`: the local document as Automerge bytes, for backups and as +//! input to an `import` remake. +//! +//! The whole document is saved: history, and protocol data under `_continuum`. Where +//! it came from is in headers: `X-Continuum-Swarm`, `X-Continuum-Epoch` +//! (`<generation>:<id hex>`) and `X-Continuum-Heads` (hex, comma-separated). + +use axum::{ + extract::State, + http::{HeaderName, HeaderValue, StatusCode, header}, + response::{IntoResponse, Response}, +}; +use tokio::time::Instant; +use tracing::{info, warn}; + +use crate::state::AppState; + +pub(crate) async fn get_document(State(state): State<AppState>) -> Response { + if !state.is_ready().await { + return (StatusCode::SERVICE_UNAVAILABLE, "node is not ready\n").into_response(); + } + let swarm_id = state.swarm_id().await.unwrap_or_default(); + + let start = Instant::now(); + let doc = std::sync::Arc::clone(&state.doc); + // Both steps take time proportional to the document: off the async threads, and + // the lock only for the copy. + let saved = tokio::task::spawn_blocking(move || { + let (mut copy, epoch) = { + let replica = doc.blocking_read(); + (replica.doc.clone(), replica.epoch) + }; + let heads = copy.get_heads(); + (copy.save(), heads, epoch) + }) + .await; + let (bytes, heads, epoch) = match saved { + Ok(saved) => saved, + Err(error) => { + warn!(error = %error, "Saving the document for export failed"); + return (StatusCode::INTERNAL_SERVER_ERROR, "saving failed\n").into_response(); + } + }; + let took = start.elapsed(); + state.stats.saved_size_measured(bytes.len() as u64, took); + info!( + bytes = bytes.len(), + took_ms = took.as_millis() as u64, + "Exported the document" + ); + + let heads = heads + .iter() + .map(|head| head.to_string()) + .collect::<Vec<_>>() + .join(","); + let filename = format!( + "attachment; filename=\"continuum-{swarm_id}-{}.automerge\"", + epoch.generation + ); + let mut response = bytes.into_response(); + let headers = response.headers_mut(); + let mut set = |name: HeaderName, value: String| { + if let Ok(value) = HeaderValue::from_str(&value) { + headers.insert(name, value); + } + }; + set(header::CONTENT_TYPE, "application/octet-stream".to_owned()); + set(header::CONTENT_DISPOSITION, filename); + set(HeaderName::from_static("x-continuum-swarm"), swarm_id); + set( + HeaderName::from_static("x-continuum-epoch"), + format!("{}:{}", epoch.generation, epoch.id_hex()), + ); + set(HeaderName::from_static("x-continuum-heads"), heads); + response +} diff --git a/src/api/log.rs b/src/api/log.rs index 4881b38..5d44926 100644 --- a/src/api/log.rs +++ b/src/api/log.rs @@ -18,7 +18,12 @@ pub(crate) async fn get_log_filter() -> Result<String, StatusCode> { /// 400 (with the reason) if the body doesn't parse, and the old filter stays. pub(crate) async fn put_log_filter(body: Bytes) -> Result<String, (StatusCode, String)> { let directives = std::str::from_utf8(&body) - .map_err(|_| (StatusCode::BAD_REQUEST, "The filter must be UTF-8\n".to_owned()))? + .map_err(|_| { + ( + StatusCode::BAD_REQUEST, + "The filter must be UTF-8\n".to_owned(), + ) + })? .trim(); let applied = logging::set_filter(directives) .map_err(|error| (StatusCode::BAD_REQUEST, format!("{error:#}\n")))?; diff --git a/src/api/mod.rs b/src/api/mod.rs index 7328059..3233b16 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -2,18 +2,20 @@ pub(crate) mod auth; mod debug; +mod document; mod log; mod peers; +mod stats; mod store; use axum::{ + Router, http::{ - header::{CONTENT_TYPE, ETAG}, HeaderValue, StatusCode, + header::{CONTENT_TYPE, ETAG}, }, response::{IntoResponse, Response}, routing::{delete, get, put}, - Router, }; use serde_json::Value as JsonValue; use sha2::{Digest, Sha256}; @@ -21,7 +23,7 @@ use tracing::{error, warn}; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; -use crate::api::auth::{basic_auth, AuthState}; +use crate::api::auth::{AuthState, basic_auth}; use crate::document::store::StoreError; use crate::state::AppState; @@ -38,7 +40,12 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> .route("/stream/{*key}", get(store::stream_store_entry)) .route("/peers", get(peers::list_peers)) .route("/peers/{peer}", delete(peers::delete_peer)) - .route("/admin/log", get(log::get_log_filter).put(log::put_log_filter)) + .route( + "/admin/log", + get(log::get_log_filter).put(log::put_log_filter), + ) + .route("/admin/stats", get(stats::get_stats)) + .route("/admin/document", get(document::get_document)) .merge(crate::remake::router::<AppState>()); if debug_api { @@ -150,6 +157,11 @@ fn response_for_value( peers::PeerListResponse, peers::PeerLivenessStatus, peers::EpochInfo, + crate::transport::traffic::PeerTraffic, + crate::transport::traffic::ByKind, + crate::transport::traffic::Counter, + crate::state::Incidents, + peers::ThroughputInfo, ) ), tags( diff --git a/src/api/peers.rs b/src/api/peers.rs index d60e7b3..7caaaf3 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -3,9 +3,9 @@ use std::sync::Arc; use axum::{ + Json, extract::{Path, Query, State}, http::StatusCode, - Json, }; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; @@ -16,8 +16,9 @@ use utoipa::ToSchema; use super::store_status; use crate::document::replica::Epoch; use crate::document::store; -use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState}; +use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState, unix_millis}; use crate::swarm::peer::forgotten_path; +use crate::transport::traffic::PeerTraffic; #[derive(Debug, Serialize, ToSchema)] #[serde(rename_all = "lowercase")] @@ -47,6 +48,8 @@ impl From<Epoch> for EpochInfo { #[serde(rename_all = "camelCase")] pub(crate) struct PeerStatus { public_key: String, + /// The name the peer gave itself in the member list, if any (not unique). + alias: Option<String>, validated_swarm: Option<String>, status: PeerLivenessStatus, handshake_complete: bool, @@ -57,12 +60,41 @@ pub(crate) struct PeerStatus { epoch: Option<EpochInfo>, /// Whether that epoch equals ours, i.e. whether we sync with this peer. same_epoch: Option<bool>, + /// Number of heads in its last keepalive; absent before one, or when it had more + /// than 32 (then only their digest is sent). + heads: Option<usize>, + /// We keep it alive and sync with it on our own; other peers we only answer. + neighbour: bool, + /// Effective round-trip time in milliseconds (smoothed, raised by requests still + /// unanswered); absent without a recent sample. + rtt_ms: Option<f64>, + /// Messages and payload bytes exchanged with it, by kind, since we started (or + /// since it was last listed). + traffic: PeerTraffic, + incidents: Incidents, + /// Smoothed sync throughput (`swarm::throughput`); absent before the first large + /// exchange. + sync_throughput: Option<ThroughputInfo>, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ThroughputInfo { + bytes_per_sec: f64, + samples: u32, + /// Age of the last sample. + secs_ago: u64, } #[derive(Debug, Serialize, ToSchema)] #[serde(rename_all = "camelCase")] pub(crate) struct PeerListResponse { + /// Our own alias (`--alias`). + local_alias: Option<String>, local_epoch: EpochInfo, + /// Number of heads of our document. Several heads are concurrent changes no later + /// change has merged yet. + local_heads: usize, peers: Vec<PeerStatus>, } @@ -97,9 +129,26 @@ pub(crate) async fn list_peers( let now = Instant::now(); let local_epoch = state.local_epoch(); + let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc); let mut peers = Vec::with_capacity(snapshot.len()); for (public_key, peer_state_mutex) in snapshot { + let neighbour = crate::swarm::topology::is_neighbour(&state, &public_key); + let rtt_ms = state + .lock_rtt() + .effective(&public_key, now, state.timing.rtt_fresh_for()) + .map(|rtt| rtt.as_secs_f64() * 1000.0); + let traffic = state.traffic.peer(&public_key).unwrap_or_default(); + let sync_throughput = state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&public_key) + .map(|estimate| ThroughputInfo { + bytes_per_sec: estimate.bytes_per_sec.round(), + samples: estimate.samples, + secs_ago: now.saturating_duration_since(estimate.at).as_secs(), + }); let ps = peer_state_mutex.lock().await; let (status, dormant_for_secs) = match ps.liveness { @@ -111,26 +160,33 @@ pub(crate) async fn list_peers( }; peers.push(PeerStatus { + alias: aliases.remove(&public_key), public_key, - validated_swarm: match &ps.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { swarm_id } => Some(swarm_id.clone()), - }, + validated_swarm: ps.standing.swarm_id().map(str::to_owned), status, handshake_complete: ps.hello_state == HelloState::Established, last_seen_secs_ago: ps .last_seen .map(|t| now.saturating_duration_since(t).as_secs()), dormant_for_secs, - has_pending_delivery: ps.pending_outbound_sync.is_some(), + has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(), epoch: ps.epoch.map(EpochInfo::from), same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), + heads: ps.reported_heads, + neighbour, + rtt_ms, + traffic, + incidents: ps.incidents, + sync_throughput, }); } peers.sort_by(|a, b| a.public_key.cmp(&b.public_key)); + let local_heads = state.doc.write().await.doc.get_heads().len(); Ok(Json(PeerListResponse { + local_alias: state.alias.clone(), local_epoch: local_epoch.into(), + local_heads, peers, })) } @@ -178,7 +234,10 @@ pub(crate) async fn delete_peer( let peer_clone = peer.clone(); state - .mutate_doc(move |doc| store::write_path(doc, &forgotten_path(&peer_clone), &record)) + .mutate_doc(move |doc| { + store::write_path(doc, &forgotten_path(&peer_clone), &record)?; + crate::swarm::members::unregister(doc, &peer_clone) + }) .await .map_err(|e| store_status(e, "Failed to write forget record"))?; diff --git a/src/api/stats.rs b/src/api/stats.rs new file mode 100644 index 0000000..bc925ff --- /dev/null +++ b/src/api/stats.rs @@ -0,0 +1,199 @@ +//! /admin/stats: how this node is doing, in numbers. JSON, or the Prometheus text +//! format when the client asks for it (`prometheus`). + +mod prometheus; + +use axum::{ + Json, + extract::State, + http::{HeaderMap, header}, + response::{IntoResponse, Response}, +}; +use serde::Serialize; + +use super::peers::EpochInfo; +use crate::state::{AppState, Incidents}; +use crate::swarm::{members, stability, topology}; +use crate::transport::traffic::PeerTraffic; + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct StatsResponse { + public_key: String, + alias: Option<String>, + uptime_secs: u64, + ready: bool, + epoch: EpochInfo, + document: DocumentStats, + catch_up: CatchUpStats, + topology: TopologyStats, + stability: StabilityStats, + /// HELLO_REPLYs to an earlier run of ours, ignored. + stale_hello_replies: u64, + /// All peers together, including ones no longer listed in /peers. + traffic: PeerTraffic, + /// The peers listed in /peers together (a forgotten peer's are gone). + incidents: Incidents, + propagation: PropagationStats, +} + +/// How long changes written elsewhere took from their commit there until applied here, +/// once caught up. Includes the clock skew between the writer and us. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationStats { + count: u64, + sum_ms: u64, + /// Changes stamped later than our clock on arrival (writer's clock ahead). + ahead: u64, + /// Not cumulative; the last bucket has no bound. + buckets: Vec<PropagationBucket>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationBucket { + le_ms: Option<u64>, + count: u64, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct DocumentStats { + heads: usize, + /// Most heads after applying a sync message (concurrent changes not yet merged). + max_heads: usize, + /// Size as saved (compressed): about what a joining node downloads. Measured in the + /// background, at least a minute apart (longer for large documents); absent until + /// first measured. + saved_bytes: Option<u64>, + /// How long ago `saved_bytes` was measured. + saved_bytes_secs_ago: Option<u64>, + /// How long the last measurement took (copy and save), in milliseconds. + saved_bytes_took_ms: Option<u64>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct CatchUpStats { + /// Catch-up sources chosen (more than one: it switched). + sources: u32, + /// Seconds from start until the initial sync was done; absent until then (and on + /// the genesis node). + done_after_secs: Option<f64>, + /// Sync payload received until then. + sync_bytes: Option<u64>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct TopologyStats { + k: usize, + /// Absent until ready with a member list. + view: Option<TopologyView>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct TopologyView { + /// Registered members, us included. + members: usize, + /// Members other than us not judged unreachable. + reachable: usize, + stable: usize, + neighbours: usize, + stable_neighbours: usize, + full_mesh: bool, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct StabilityStats { + /// As published in the member registry. + tier: &'static str, + /// Fraction of the observed time (up to a day) with a live peer. + connected_fraction: f64, + observed_secs: u64, +} + +/// `GET /admin/stats`. +pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap) -> Response { + let stats = collect(&state).await; + let accept = headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()); + if prometheus::wanted(accept) { + let body = prometheus::render(&state, &stats).await; + ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response() + } else { + Json(stats).into_response() + } +} + +async fn collect(state: &AppState) -> StatsResponse { + let node = state.stats.snapshot(); + + let (heads, tier) = { + let mut replica = state.doc.write().await; + let heads = replica.doc.get_heads().len(); + (heads, members::tier(&replica.doc, state.local_id())) + }; + let saved = state.stats.saved_size(); + + let (connected_fraction, observed) = stability::measured(state); + + let mut incidents = Incidents::default(); + let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect(); + for peer_state in peers { + incidents.add(&peer_state.lock().await.incidents); + } + + StatsResponse { + public_key: state.local_id().to_owned(), + alias: state.alias.clone(), + uptime_secs: node.uptime.as_secs(), + ready: state.is_ready().await, + epoch: state.local_epoch().into(), + document: DocumentStats { + heads, + max_heads: node.max_heads.max(heads), + saved_bytes: saved.map(|s| s.bytes), + saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()), + saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64), + }, + catch_up: CatchUpStats { + sources: node.catch_up_sources, + done_after_secs: node.caught_up.map(|c| c.after.as_secs_f64()), + sync_bytes: node.caught_up.map(|c| c.sync_bytes), + }, + topology: TopologyStats { + k: state.neighbours_k, + view: topology::summary(state).map(|s| TopologyView { + members: s.members, + reachable: s.reachable, + stable: s.stable, + neighbours: s.neighbours, + stable_neighbours: s.stable_neighbours, + full_mesh: s.full_mesh, + }), + }, + stability: StabilityStats { + tier: tier.as_str(), + connected_fraction, + observed_secs: observed.as_secs(), + }, + stale_hello_replies: node.stale_hello_replies, + traffic: state.traffic.total(), + incidents, + propagation: PropagationStats { + count: node.propagation.count, + sum_ms: node.propagation.sum_ms, + ahead: node.propagation.ahead, + buckets: node + .propagation + .buckets() + .map(|(le_ms, count)| PropagationBucket { le_ms, count }) + .collect(), + }, + } +} diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs new file mode 100644 index 0000000..d29207c --- /dev/null +++ b/src/api/stats/prometheus.rs @@ -0,0 +1,521 @@ +//! /admin/stats in the Prometheus text exposition format (version 0.0.4), for a +//! scraper asking for it. Adds per-peer series (the figures /peers shows), labelled +//! with the peer's public key. + +use std::fmt::Write; + +use tokio::time::Instant; + +use super::StatsResponse; +use crate::state::{AppState, Incidents, Liveness}; +use crate::transport::traffic::{ByKind, Counter, PeerTraffic}; + +pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8"; + +/// Whether the client prefers this format to JSON: Prometheus asks for OpenMetrics or +/// plain text; browsers and curl don't name either. +pub(super) fn wanted(accept: Option<&str>) -> bool { + accept.is_some_and(|accept| { + accept.split(',').any(|item| { + let media = item.split(';').next().unwrap_or("").trim(); + media == "application/openmetrics-text" || media == "text/plain" + }) + }) +} + +#[derive(Default)] +struct Out(String); + +impl Out { + fn family(&mut self, name: &str, kind: &str, help: &str) { + let _ = writeln!(self.0, "# HELP continuum_{name} {help}"); + let _ = writeln!(self.0, "# TYPE continuum_{name} {kind}"); + } + + fn sample(&mut self, name: &str, labels: &[(&str, &str)], value: f64) { + let _ = write!(self.0, "continuum_{name}"); + if !labels.is_empty() { + self.0.push('{'); + for (i, (key, value)) in labels.iter().enumerate() { + if i > 0 { + self.0.push(','); + } + let _ = write!(self.0, "{key}=\"{}\"", escape(value)); + } + self.0.push('}'); + } + let _ = writeln!(self.0, " {value}"); + } + + /// A family with one unlabelled sample. + fn single(&mut self, name: &str, kind: &str, help: &str, value: f64) { + self.family(name, kind, help); + self.sample(name, &[], value); + } +} + +fn escape(value: &str) -> String { + value + .replace('\\', "\\\\") + .replace('"', "\\\"") + .replace('\n', "\\n") +} + +fn flag(value: bool) -> f64 { + if value { 1.0 } else { 0.0 } +} + +/// What we show per peer. +struct Peer { + key: String, + alias: Option<String>, + neighbour: bool, + dormant: bool, + last_seen_secs: Option<f64>, + traffic: PeerTraffic, + incidents: Incidents, + rtt_secs: Option<f64>, + throughput: Option<f64>, +} + +async fn peers(state: &AppState) -> Vec<Peer> { + let now = Instant::now(); + let snapshot: Vec<_> = state + .peer_states + .read() + .await + .iter() + .map(|(key, peer_state)| (key.clone(), peer_state.clone())) + .collect(); + let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc); + let mut peers = Vec::with_capacity(snapshot.len()); + for (key, peer_state) in snapshot { + let neighbour = crate::swarm::topology::is_neighbour(state, &key); + let rtt_secs = state + .lock_rtt() + .effective(&key, now, state.timing.rtt_fresh_for()) + .map(|rtt| rtt.as_secs_f64()); + let throughput = state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&key) + .map(|estimate| estimate.bytes_per_sec); + let traffic = state.traffic.peer(&key).unwrap_or_default(); + let peer_state = peer_state.lock().await; + peers.push(Peer { + neighbour, + dormant: matches!(peer_state.liveness, Liveness::Dormant { .. }), + last_seen_secs: peer_state + .last_seen + .map(|t| now.saturating_duration_since(t).as_secs_f64()), + traffic, + incidents: peer_state.incidents, + rtt_secs, + throughput, + alias: aliases.remove(&key), + key, + }); + } + peers.sort_by(|a, b| a.key.cmp(&b.key)); + peers +} + +/// Every (direction, kind, counter) of `traffic`. +fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Counter)> { + let by_kind = |direction: &'static str, by: &ByKind| { + [ + (direction, "hello", by.hello), + (direction, "hello_reply", by.hello_reply), + (direction, "sync", by.sync), + (direction, "ping", by.ping), + (direction, "pong", by.pong), + ] + }; + let mut series = Vec::new(); + series.extend(by_kind("sent", &traffic.sent)); + series.extend(by_kind("received", &traffic.received)); + series.extend(by_kind("refused", &traffic.refused)); + series +} + +pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { + let mut out = Out::default(); + + // Names for joining (`* on(peer) group_left(alias) continuum_peer_info`), kept out + // of the other series so a rename doesn't start new ones. + out.family( + "info", + "gauge", + "This node's public key and alias (always 1).", + ); + out.sample( + "info", + &[ + ("public_key", &stats.public_key), + ("alias", stats.alias.as_deref().unwrap_or("")), + ], + 1.0, + ); + + out.single( + "uptime_seconds", + "gauge", + "Seconds since this node started.", + stats.uptime_secs as f64, + ); + out.single( + "ready", + "gauge", + "1 once the node holds the swarm's document.", + flag(stats.ready), + ); + out.single( + "epoch_generation", + "gauge", + "Generation of the local epoch (remakes so far).", + state.local_epoch().generation as f64, + ); + + let document = &stats.document; + out.single( + "document_heads", + "gauge", + "Heads of the local document.", + document.heads as f64, + ); + out.single( + "document_max_heads", + "gauge", + "Most heads after applying a sync message, since start.", + document.max_heads as f64, + ); + if let Some(bytes) = document.saved_bytes { + out.single( + "document_saved_bytes", + "gauge", + "Size of the document as saved (compressed); measured in the background.", + bytes as f64, + ); + } + + let catch_up = &stats.catch_up; + out.single( + "catch_up_sources_total", + "counter", + "Catch-up sources chosen (more than one: switched).", + catch_up.sources as f64, + ); + if let Some(secs) = catch_up.done_after_secs { + out.single( + "catch_up_duration_seconds", + "gauge", + "Seconds from start until the initial sync was done.", + secs, + ); + } + if let Some(bytes) = catch_up.sync_bytes { + out.single( + "catch_up_sync_bytes", + "gauge", + "Sync payload received until the initial sync was done.", + bytes as f64, + ); + } + + let topology = &stats.topology; + out.single( + "neighbours_target", + "gauge", + "Target number of neighbours (k).", + topology.k as f64, + ); + if let Some(view) = &topology.view { + out.single( + "members", + "gauge", + "Registered members, this node included.", + view.members as f64, + ); + out.single( + "reachable_members", + "gauge", + "Members other than this node not judged unreachable.", + view.reachable as f64, + ); + out.single( + "stable_members", + "gauge", + "Members with the stable tier.", + view.stable as f64, + ); + out.single( + "neighbours", + "gauge", + "Current neighbours.", + view.neighbours as f64, + ); + out.single( + "stable_neighbours", + "gauge", + "Current neighbours with the stable tier.", + view.stable_neighbours as f64, + ); + out.single( + "full_mesh", + "gauge", + "1 while every reachable member is a neighbour.", + flag(view.full_mesh), + ); + } + + let stability = &stats.stability; + out.single( + "stable", + "gauge", + "1 while this node's published tier is stable.", + flag(stability.tier == "stable"), + ); + out.single( + "connected_ratio", + "gauge", + "Fraction of the observed time (up to a day) with a live peer.", + stability.connected_fraction, + ); + out.single( + "connectivity_observed_seconds", + "gauge", + "Time the connected ratio is based on.", + stability.observed_secs as f64, + ); + + out.single( + "stale_hello_replies_total", + "counter", + "HELLO_REPLYs to an earlier run of this node, ignored.", + stats.stale_hello_replies as f64, + ); + + let propagation = &stats.propagation; + out.family( + "propagation_delay_seconds", + "histogram", + "Time from a change's commit elsewhere until applied here (includes clock skew).", + ); + let mut cumulative = 0; + for bucket in &propagation.buckets { + cumulative += bucket.count; + let le = match bucket.le_ms { + Some(ms) => (ms as f64 / 1000.0).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "propagation_delay_seconds_bucket", + &[("le", &le)], + cumulative as f64, + ); + } + out.sample( + "propagation_delay_seconds_sum", + &[], + propagation.sum_ms as f64 / 1000.0, + ); + out.sample( + "propagation_delay_seconds_count", + &[], + propagation.count as f64, + ); + out.single( + "propagation_clock_ahead_total", + "counter", + "Changes stamped later than this node's clock on arrival.", + propagation.ahead as f64, + ); + + // Node totals, including peers no longer listed. + let totals = traffic_series(&stats.traffic); + out.family( + "messages_total", + "counter", + "Messages by direction and kind, all peers.", + ); + for (direction, kind, counter) in &totals { + out.sample( + "messages_total", + &[("direction", direction), ("kind", kind)], + counter.messages as f64, + ); + } + out.family( + "payload_bytes_total", + "counter", + "Payload bytes by direction and kind, all peers.", + ); + for (direction, kind, counter) in &totals { + out.sample( + "payload_bytes_total", + &[("direction", direction), ("kind", kind)], + counter.bytes as f64, + ); + } + + let peers = peers(state).await; + + out.family( + "peer_info", + "gauge", + "A peer's alias from the member list (always 1).", + ); + for peer in &peers { + out.sample( + "peer_info", + &[ + ("peer", &peer.key), + ("alias", peer.alias.as_deref().unwrap_or("")), + ], + 1.0, + ); + } + + out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour."); + for peer in &peers { + out.sample( + "peer_neighbour", + &[("peer", &peer.key)], + flag(peer.neighbour), + ); + } + out.family("peer_dormant", "gauge", "1 while the peer is dormant."); + for peer in &peers { + out.sample("peer_dormant", &[("peer", &peer.key)], flag(peer.dormant)); + } + out.family( + "peer_last_seen_seconds", + "gauge", + "Seconds since the peer's last message.", + ); + for peer in &peers { + if let Some(secs) = peer.last_seen_secs { + out.sample("peer_last_seen_seconds", &[("peer", &peer.key)], secs); + } + } + out.family( + "peer_rtt_seconds", + "gauge", + "Effective round-trip time (smoothed, raised by unanswered requests).", + ); + for peer in &peers { + if let Some(rtt) = peer.rtt_secs { + out.sample("peer_rtt_seconds", &[("peer", &peer.key)], rtt); + } + } + out.family( + "peer_sync_throughput_bytes_per_second", + "gauge", + "Smoothed sync throughput in large exchanges.", + ); + for peer in &peers { + if let Some(throughput) = peer.throughput { + out.sample( + "peer_sync_throughput_bytes_per_second", + &[("peer", &peer.key)], + throughput, + ); + } + } + + out.family( + "peer_messages_total", + "counter", + "Messages by direction and kind.", + ); + for peer in &peers { + for (direction, kind, counter) in traffic_series(&peer.traffic) { + out.sample( + "peer_messages_total", + &[ + ("peer", &peer.key), + ("direction", direction), + ("kind", kind), + ], + counter.messages as f64, + ); + } + } + out.family( + "peer_payload_bytes_total", + "counter", + "Payload bytes by direction and kind.", + ); + for peer in &peers { + for (direction, kind, counter) in traffic_series(&peer.traffic) { + out.sample( + "peer_payload_bytes_total", + &[ + ("peer", &peer.key), + ("direction", direction), + ("kind", kind), + ], + counter.bytes as f64, + ); + } + } + + let incidents: [(&str, &str, fn(&Incidents) -> f64); 4] = [ + ( + "peer_reply_timeouts_total", + "Sync messages unanswered until the reply timeout.", + |i| i.reply_timeouts as f64, + ), + ( + "peer_heals_total", + "Re-handshakes because documents differed while sync was idle.", + |i| i.heals as f64, + ), + ( + "peer_dormancies_total", + "Times the peer went dormant.", + |i| i.dormancies as f64, + ), + ( + "peer_dormant_seconds_total", + "Time spent dormant, up to the last return.", + |i| i.dormant_ms as f64 / 1000.0, + ), + ]; + for (name, help, value) in incidents { + out.family(name, "counter", help); + for peer in &peers { + out.sample(name, &[("peer", &peer.key)], value(&peer.incidents)); + } + } + + out.0 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn negotiation() { + let prometheus = "application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1"; + assert!(wanted(Some(prometheus))); + assert!(wanted(Some("text/plain"))); + assert!(!wanted(Some("*/*"))); + assert!(!wanted(Some("application/json"))); + assert!(!wanted(Some("text/html,application/xhtml+xml,*/*;q=0.8"))); + assert!(!wanted(None)); + } + + #[test] + fn sample_lines() { + let mut out = Out::default(); + out.single("x_total", "counter", "An x.", 3.0); + out.sample("y", &[("peer", "a\"b"), ("kind", "sync")], 0.25); + assert_eq!( + out.0, + "# HELP continuum_x_total An x.\n# TYPE continuum_x_total counter\ncontinuum_x_total 3\ncontinuum_y{peer=\"a\\\"b\",kind=\"sync\"} 0.25\n" + ); + } +} diff --git a/src/api/store.rs b/src/api/store.rs index 409a530..d6ff306 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -7,8 +7,8 @@ use axum::{ body::{Body, Bytes}, extract::{Path, Query, State}, http::{ - header::{ACCEPT, CONTENT_TYPE}, HeaderMap, HeaderValue, StatusCode, + header::{ACCEPT, CONTENT_TYPE}, }, response::{IntoResponse, Response}, }; diff --git a/src/backoff.rs b/src/backoff.rs new file mode 100644 index 0000000..1f87b8d --- /dev/null +++ b/src/backoff.rs @@ -0,0 +1,58 @@ +//! Exponential backoff: a delay that doubles, up to a limit, while something keeps +//! failing, and starts over when it succeeds. Only the delay; each user keeps its own +//! point in time (next attempt, deadline). + +use std::time::Duration; + +#[derive(Clone, Copy, Debug)] +pub(crate) struct Backoff { + initial: Duration, + max: Duration, + current: Duration, +} + +impl Backoff { + pub(crate) const fn new(initial: Duration, max: Duration) -> Self { + Self { + initial, + max, + current: initial, + } + } + + /// The delay to use now. + pub(crate) fn current(&self) -> Duration { + self.current + } + + /// Doubles the delay, up to the limit. + pub(crate) fn grow(&mut self) { + self.current = self.current.saturating_mul(2).min(self.max); + } + + /// The delay to use now; the next call gives twice as much (up to the limit). + pub(crate) fn next(&mut self) -> Duration { + let delay = self.current; + self.grow(); + delay + } + + /// Back to the initial delay. + pub(crate) fn reset(&mut self) { + self.current = self.initial; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn doubles_up_to_the_limit_and_resets() { + let mut backoff = Backoff::new(Duration::from_secs(1), Duration::from_secs(5)); + let delays: Vec<_> = (0..5).map(|_| backoff.next().as_secs()).collect(); + assert_eq!(delays, [1, 2, 4, 5, 5]); + backoff.reset(); + assert_eq!(backoff.current(), Duration::from_secs(1)); + } +} diff --git a/src/document/store.rs b/src/document/store.rs index 9db8300..9e18693 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -2,9 +2,9 @@ use anyhow::anyhow; use automerge::{ - transaction::Transactable, ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, + ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, transaction::Transactable, }; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use serde::Serialize; use serde_json::Value as JsonValue; use std::collections::BTreeSet; @@ -97,7 +97,7 @@ fn resolved_to_json( let maps = match resolved { Resolved::Maps(maps) => maps, Resolved::Other(Value::Scalar(scalar), _) => { - return scalar_to_json(scalar.clone().into_owned()) + return scalar_to_json(scalar.clone().into_owned()); } Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log), }; diff --git a/src/logging.rs b/src/logging.rs index c87837c..4496125 100644 --- a/src/logging.rs +++ b/src/logging.rs @@ -6,8 +6,8 @@ use std::sync::OnceLock; -use anyhow::{anyhow, Result}; -use tracing_subscriber::{prelude::*, reload, EnvFilter, Registry}; +use anyhow::{Result, anyhow}; +use tracing_subscriber::{EnvFilter, Registry, prelude::*, reload}; type FilterHandle = reload::Handle<EnvFilter, Registry>; @@ -18,14 +18,15 @@ pub(crate) fn init() { let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let (filter, handle) = reload::Layer::new(filter); - let registry = tracing_subscriber::registry() - .with(tracing_subscriber::fmt::layer().compact().with_filter(filter)); + let registry = tracing_subscriber::registry().with( + tracing_subscriber::fmt::layer() + .compact() + .with_filter(filter), + ); #[cfg(feature = "tokio-console")] - let registry = registry.with( - console_subscriber::spawn() - .with_filter(EnvFilter::new("tokio=trace,runtime=trace")), - ); + let registry = registry + .with(console_subscriber::spawn().with_filter(EnvFilter::new("tokio=trace,runtime=trace"))); registry.init(); let _ = FILTER.set(handle); diff --git a/src/main.rs b/src/main.rs index 67cb366..65134d8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,18 +7,19 @@ 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 document; mod logging; mod remake; @@ -30,9 +31,10 @@ 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; +use transport::traffic::{CountingTransport, Traffic}; #[derive(Parser, Debug)] #[command(version, about = "continuum messaging system", long_about = None)] @@ -90,12 +92,37 @@ struct Args { #[arg(long, value_name = "SECONDS", default_value_t = 60)] keepalive_secs: u64, + /// Neighbours each node keeps (at least 2). Swarms of up to this many + 1 nodes are + /// a full mesh; larger ones sync over several hops. + #[arg(long, value_name = "COUNT", default_value_t = 12)] + neighbours: usize, + + /// A name for this node, shown next to its public key (/peers, metrics, logs). + /// Published in the swarm's member list; not unique, never an identity. + #[arg(long, value_name = "NAME", value_parser = parse_alias)] + alias: Option<String>, + /// 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>, } +/// Up to 64 characters, no control characters, not blank. +fn parse_alias(value: &str) -> Result<String, String> { + let alias = value.trim(); + if alias.is_empty() { + return Err("must not be blank".to_owned()); + } + if alias.chars().count() > 64 { + return Err("must be at most 64 characters".to_owned()); + } + if alias.chars().any(char::is_control) { + return Err("must not contain control characters".to_owned()); + } + Ok(alias.to_owned()) +} + #[tokio::main] async fn main() -> Result<()> { logging::init(); @@ -161,6 +188,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 +200,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 +219,17 @@ 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.max(2), + alias: args.alias, + neighbours: Default::default(), + pulling: 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 +239,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 +332,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())?; diff --git a/src/remake/api.rs b/src/remake/api.rs index 5267e48..8467429 100644 --- a/src/remake/api.rs +++ b/src/remake/api.rs @@ -24,22 +24,35 @@ //! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs` //! below one keepalive interval: for tests and debugging, not for real swarms. //! -//! `kind` is `compact` (no spec) or `replace`, whose spec is +//! `kind` is `compact` (no spec), `replace`, whose spec is //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }` -//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`). +//! (see `kind::Replace`), or `import`, whose spec is +//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see +//! `kind::Import`; `GET /admin/document` exports one). Reports leave the document out +//! (`contentBytes`). Requests may be up to 64 MiB. +//! +//! An import can also be uploaded as is, with `Content-Type: application/octet-stream`; +//! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and +//! `switchDelaySecs` go in the query. Restoring a backup, keeping writes made since at +//! their paths: +//! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes` use std::time::Duration; use automerge::ChangeHash; use axum::{ + Json, body::Bytes, extract::{Path, Query, State}, - http::{header::LOCATION, StatusCode}, + http::{ + HeaderMap, StatusCode, + header::{CONTENT_TYPE, LOCATION}, + }, response::{IntoResponse, Response}, - Json, }; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; -use serde_json::{json, Value as JsonValue}; +use serde_json::{Value as JsonValue, json}; use tokio::sync::broadcast; use tracing::{error, info, warn}; use uuid::Uuid; @@ -47,9 +60,9 @@ use uuid::Uuid; use super::coordinator; use super::local::LocalStatus; use super::records::{ - self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, + self, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, unix_millis, }; -use super::{kind, Host}; +use super::{Host, kind}; use crate::document::store::StoreError; #[derive(Debug, Default, Deserialize)] @@ -59,12 +72,64 @@ pub(crate) struct WaitQuery { } #[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase")] pub(crate) struct StartQuery { #[serde(default)] wait: bool, /// Allows a settle period shorter than the dormancy threshold. #[serde(default)] force: bool, + // Only for an import uploaded as `application/octet-stream`, which has no JSON + // body to carry them: + /// The rules, as JSON (as in the spec). + rules: Option<String>, + /// Comma-separated hex. + base_heads: Option<String>, + settle_secs: Option<u64>, + switch_delay_secs: Option<u64>, +} + +impl StartQuery { + fn has_request_fields(&self) -> bool { + self.rules.is_some() + || self.base_heads.is_some() + || self.settle_secs.is_some() + || self.switch_delay_secs.is_some() + } + + /// An import request for a document uploaded as is. + fn import_request(&self, document: &[u8]) -> Result<StartRequest, String> { + let rules = match &self.rules { + None => JsonValue::Array(Vec::new()), + Some(rules) => serde_json::from_str(rules) + .map_err(|error| format!("rules are not valid JSON: {error}"))?, + }; + Ok(StartRequest { + kind: Some("import".to_owned()), + spec: Some(json!({ + "document": BASE64.encode(document), + "rules": rules, + })), + base_heads: self.base_heads.as_ref().map(|heads| { + heads + .split(',') + .filter(|head| !head.trim().is_empty()) + .map(str::to_owned) + .collect() + }), + settle_secs: self.settle_secs, + switch_delay_secs: self.switch_delay_secs, + ..StartRequest::default() + }) + } +} + +fn is_octet_stream(headers: &HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .is_some_and(|media| media.trim() == "application/octet-stream") } #[derive(Debug, Default, Deserialize)] @@ -186,23 +251,26 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> { pub(crate) async fn start<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, None).await + start_kind(host, query, &headers, body, None).await } /// `POST /admin/compaction`: the original endpoint, for compacting remakes only. pub(crate) async fn start_compaction<H: Host>( State(host): State<H>, Query(query): Query<StartQuery>, + headers: HeaderMap, body: Bytes, ) -> Response { - start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await + start_kind(host, query, &headers, body, Some(kind::DEFAULT_KIND)).await } async fn start_kind<H: Host>( host: H, query: StartQuery, + headers: &HeaderMap, body: Bytes, only: Option<&'static str>, ) -> Response { @@ -210,7 +278,24 @@ async fn start_kind<H: Host>( return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None); } - let request: StartRequest = if body.is_empty() { + let request: StartRequest = if is_octet_stream(headers) { + match query.import_request(&body) { + Ok(request) => request, + Err(reason) => { + return error_response( + StatusCode::BAD_REQUEST, + format!("invalid request: {reason}"), + None, + ); + } + } + } else if query.has_request_fields() { + return error_response( + StatusCode::BAD_REQUEST, + "rules, baseHeads, settleSecs and switchDelaySecs go in the query only for an application/octet-stream import; put them in the JSON body", + None, + ); + } else if body.is_empty() { StartRequest::default() } else { match serde_json::from_slice(&body) { @@ -220,7 +305,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid request: {error}"), None, - ) + ); } } }; @@ -247,7 +332,7 @@ async fn start_kind<H: Host>( StatusCode::BAD_REQUEST, format!("invalid {kind_name} spec: {reason}"), None, - ) + ); } }; @@ -559,14 +644,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response { StatusCode::CONFLICT, "already committed; the switch cannot be undone", Some(json!({ "remake": current })), - ) + ); } Phase::Aborted | Phase::Expired => { return error_response( StatusCode::CONFLICT, "already over", Some(json!({ "remake": current })), - ) + ); } } diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 4aa66e7..db25251 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -31,13 +31,13 @@ use automerge::{AutoCommit, ChangeHash}; use tokio::time::Instant; use tracing::{error, info, warn}; +use super::Host; use super::delta::{self, ReplayStats}; use super::kind::Translator; use super::local::{CatchUp, LocalStatus, Rejoin}; use super::quarantine; use super::records::unix_millis; use super::snapshot::{self, EpochMetadata}; -use super::Host; use crate::document::replica::Epoch; /// Pause before trying the same target again after a failed rejoin. @@ -444,9 +444,9 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) { #[cfg(test)] mod tests { use super::*; - use crate::remake::snapshot::{build, SnapshotParams}; - use automerge::{transaction::Transactable, ActorId, ROOT}; - use serde_json::{json, Value as JsonValue}; + use crate::remake::snapshot::{SnapshotParams, build}; + use automerge::{ActorId, ROOT, transaction::Transactable}; + use serde_json::{Value as JsonValue, json}; /// An old-epoch document with one own change after the base, and the swarm's /// document of the next epoch (snapshot plus one foreign change). diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index d1b3397..28140cb 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -20,9 +20,9 @@ use std::time::Duration; use tracing::{info, warn}; +use super::Host; use super::records::{self, Decision, Record}; use super::snapshot; -use super::Host; use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); diff --git a/src/remake/delta.rs b/src/remake/delta.rs index c161e2c..d318c00 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -268,7 +268,7 @@ fn patch_path(patch: &automerge::Patch) -> Vec<String> { #[cfg(test)] mod tests { use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT}; + use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, transaction::Transactable}; use serde_json::json; fn view(doc: &AutoCommit) -> serde_json::Value { diff --git a/src/remake/kind.rs b/src/remake/kind.rs index b709a05..685fa28 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -10,17 +10,19 @@ //! * where a write made in the old epoch after the base goes in the new one //! ([`RemakeKind::translate`]). //! -//! Kinds: [`Compact`] (same content, no history) and [`Replace`] (an operator's -//! document, with rules for where later writes go). +//! Kinds: [`Compact`] (same content, no history), [`Replace`] (an operator's document +//! in store form, with rules for where later writes go) and [`Import`] (the same with +//! an Automerge document, e.g. one exported with `GET /admin/document`). //! //! Protocol data under `_continuum` is not a kind's business: the generic part carries //! it forward and replays it unchanged. use automerge::AutoCommit; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; -use super::snapshot::{canonical, META_ROOT}; +use super::snapshot::{META_ROOT, canonical}; use crate::document::store::{self, StoreError}; /// Name of the kind used when a request doesn't name one. @@ -75,6 +77,7 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { match name { "compact" => Some(&Compact), "replace" => Some(&Replace), + "import" => Some(&Import), _ => None, } } @@ -196,33 +199,9 @@ impl RemakeKind for Replace { )); } - let mut rules = Vec::with_capacity(spec.rules.len()); - for rule in spec.rules { - let from = rule.from.segments()?; - let to = match (rule.to, rule.drop) { - (Some(to), false) => Some(to.segments()?), - (None, true) => None, - _ => { - return Err(format!( - "rule for {from:?} needs exactly one of \"to\" and \"drop\"" - )) - } - }; - if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) { - return Err(format!( - "rules must not touch the reserved key {META_ROOT:?}" - )); - } - if rules.iter().any(|existing: &Rule| existing.from == from) { - return Err(format!("two rules for {from:?}")); - } - rules.push(Rule { from, to }); - } - Ok(Prepared { content: canonical(spec.document), - translation: serde_json::to_value(Rules { rules }) - .map_err(|error| error.to_string())?, + translation: prepare_rules(spec.rules)?, }) } @@ -246,31 +225,132 @@ impl RemakeKind for Replace { } fn translate(&self, path: &[String], translation: &JsonValue) -> Translation { - let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else { - return Translation::Quarantine("unreadable replacement rules".to_owned()); - }; - let Some(rule) = rules - .iter() - .filter(|rule| path.starts_with(&rule.from)) - .max_by_key(|rule| rule.from.len()) - else { - return Translation::Quarantine("no replacement rule covers this path".to_owned()); + translate_by_rules(path, translation) + } +} + +/// Validates rules from a request; the translation to store. +fn prepare_rules(specs: Vec<RuleSpec>) -> Result<JsonValue, String> { + let mut rules = Vec::with_capacity(specs.len()); + for rule in specs { + let from = rule.from.segments()?; + let to = match (rule.to, rule.drop) { + (Some(to), false) => Some(to.segments()?), + (None, true) => None, + _ => { + return Err(format!( + "rule for {from:?} needs exactly one of \"to\" and \"drop\"" + )); + } }; - match &rule.to { - None => Translation::Drop, - Some(to) => { - let mut target = to.clone(); - target.extend_from_slice(&path[rule.from.len()..]); - if target.as_slice() == path { - Translation::Same - } else { - Translation::To(target) - } + if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) { + return Err(format!( + "rules must not touch the reserved key {META_ROOT:?}" + )); + } + if rules.iter().any(|existing: &Rule| existing.from == from) { + return Err(format!("two rules for {from:?}")); + } + rules.push(Rule { from, to }); + } + serde_json::to_value(Rules { rules }).map_err(|error| error.to_string()) +} + +/// Where `path` goes under the stored rules: the longest matching `from` wins. +fn translate_by_rules(path: &[String], translation: &JsonValue) -> Translation { + let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else { + return Translation::Quarantine("unreadable replacement rules".to_owned()); + }; + let Some(rule) = rules + .iter() + .filter(|rule| path.starts_with(&rule.from)) + .max_by_key(|rule| rule.from.len()) + else { + return Translation::Quarantine("no replacement rule covers this path".to_owned()); + }; + match &rule.to { + None => Translation::Drop, + Some(to) => { + let mut target = to.clone(); + target.extend_from_slice(&path[rule.from.len()..]); + if target.as_slice() == path { + Translation::Same + } else { + Translation::To(target) } } } } +/// Import: like [`Replace`], but the new document is an Automerge document (bytes, as +/// saved), so Automerge types the store's JSON can't express survive: text, counters, +/// bytes, timestamps, floats. Its `_continuum` (protocol data of whatever swarm it came +/// from) is left out, as is its history: the content is the merged view. +/// +/// Spec: `{ "document": "<base64 of the saved document>", "rules": […] }`, rules as for +/// [`Replace`]. To restore a backup of this swarm with every later write kept where it +/// was: `"rules": [{"from": "", "to": ""}]`. +pub(crate) struct Import; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ImportSpec { + document: String, + #[serde(default)] + rules: Vec<RuleSpec>, +} + +/// The document in `content` (base64 of a saved Automerge document). +fn load_content(content: &JsonValue) -> Result<AutoCommit, String> { + let JsonValue::String(encoded) = content else { + return Err("the document must be a base64 string".to_owned()); + }; + let bytes = BASE64 + .decode(encoded.trim()) + .map_err(|error| format!("the document is not valid base64: {error}"))?; + AutoCommit::load(&bytes) + .map_err(|error| format!("the document is not an Automerge document: {error}")) +} + +impl RemakeKind for Import { + fn name(&self) -> &'static str { + "import" + } + + fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> { + let spec: ImportSpec = + serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?; + let imported = load_content(&JsonValue::String(spec.document))?; + + // Keep only what goes into the new epoch: the merged view without protocol data + // and history. Usually much smaller than what was uploaded, and it is what gets + // replicated with the proposal. + let mut content = AutoCommit::new(); + store::materialize_into(&imported, &mut content, &is_reserved) + .map_err(|error| format!("the document can't be read: {error}"))?; + + Ok(Prepared { + content: JsonValue::String(BASE64.encode(content.save())), + translation: prepare_rules(spec.rules)?, + }) + } + + fn build( + &self, + _old_at_base: &AutoCommit, + content: &JsonValue, + target: &mut AutoCommit, + ) -> Result<(), StoreError> { + let imported = load_content(content).map_err(|reason| anyhow::anyhow!(reason))?; + // Deterministic: the same bytes give the same view, copied in a fixed order. + store::materialize_into(&imported, target, &is_reserved) + } + + fn translate(&self, path: &[String], translation: &JsonValue) -> Translation { + translate_by_rules(path, translation) + } +} + /// The translations of several consecutive remakes, applied in order: a node that /// rejoins several epochs later carries its writes through every remake in between. #[derive(Default, Clone)] @@ -399,10 +479,10 @@ mod tests { assert!(replace(json!({ "document": [] })).is_err()); assert!(replace(json!({ "document": { META_ROOT: {} } })).is_err()); assert!(replace(json!({ "document": {}, "rules": [{ "from": "a" }] })).is_err()); - assert!(replace( - json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] }) - ) - .is_err()); + assert!( + replace(json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] })) + .is_err() + ); assert!( replace(json!({ "document": {}, "rules": [{ "from": "a//b", "to": "c" }] })).is_err() ); @@ -448,6 +528,63 @@ mod tests { } #[test] + fn import_keeps_types_and_drops_protocol_data() { + use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable}; + use serde_json::json; + + let mut source = AutoCommit::new(); + let text = source.put_object(ROOT, "notes", ObjType::Text).unwrap(); + source.splice_text(&text, 0, 0, "hello").unwrap(); + source + .put(ROOT, "hits", automerge::ScalarValue::counter(3)) + .unwrap(); + let meta = source.put_object(ROOT, META_ROOT, ObjType::Map).unwrap(); + source.put(&meta, "members", "elsewhere").unwrap(); + let encoded = BASE64.encode(source.save()); + + assert!( + Import + .propose(&json!({ "document": "not base64!" })) + .is_err() + ); + assert!( + Import + .propose(&json!({ "document": BASE64.encode(b"junk") })) + .is_err() + ); + + let prepared = Import + .propose(&json!({ "document": encoded, "rules": [{ "from": "", "to": "" }] })) + .unwrap(); + let build = || { + let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); + Import + .build(&AutoCommit::new(), &prepared.content, &mut target) + .unwrap(); + target.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); + target + }; + let mut a = build(); + let mut b = build(); + assert_eq!(a.get_heads(), b.get_heads()); + + assert!(a.get(ROOT, META_ROOT).unwrap().is_none()); + let (notes, notes_id) = a.get(ROOT, "notes").unwrap().unwrap(); + assert!(matches!(notes, automerge::Value::Object(ObjType::Text))); + assert_eq!(a.text(¬es_id).unwrap(), "hello"); + assert!(matches!( + a.get(ROOT, "hits").unwrap().unwrap().0, + automerge::Value::Scalar(ref s) if matches!(s.as_ref(), automerge::ScalarValue::Counter(_)) + )); + + // Writes after the base keep their paths. + assert_eq!( + Translator::single(&Import, prepared.translation).translate(&path(&["x", "y"])), + Translation::Same + ); + } + + #[test] fn replace_builds_the_document_deterministically() { use serde_json::json; let build = |document: serde_json::Value| { diff --git a/src/remake/local.rs b/src/remake/local.rs index 35a4ff1..d2ea2c2 100644 --- a/src/remake/local.rs +++ b/src/remake/local.rs @@ -12,7 +12,7 @@ use automerge::AutoCommit; use serde::Serialize; use tokio::time::Instant; -use super::records::{unix_millis, Retention}; +use super::records::{Retention, unix_millis}; use super::snapshot::Snapshot; use crate::document::replica::Epoch; diff --git a/src/remake/mod.rs b/src/remake/mod.rs index ef036a6..6a9c6f9 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -37,8 +37,9 @@ use std::time::Duration; use automerge::AutoCommit; use axum::{ - routing::{get, post}, Router, + extract::DefaultBodyLimit, + routing::{get, post}, }; use tokio::sync::broadcast; @@ -122,13 +123,20 @@ pub(crate) struct PeerSummary { } /// Admin routes. Merged into the authenticated part of the node's router. +/// Largest remake request (bytes). The proposal carries the new document to every +/// node in one sync message, which Mycelium must deliver within its deadline, so +/// documents near this size are already a stretch. +const REMAKE_REQUEST_LIMIT: usize = 64 * 1024 * 1024; + pub(crate) fn router<H: Host>() -> Router<H> { Router::new() .route( "/admin/remakes", get(api::list::<H>) .post(api::start::<H>) - .delete(api::abort_current::<H>), + .delete(api::abort_current::<H>) + // A replace or import request carries a whole document. + .layer(DefaultBodyLimit::max(REMAKE_REQUEST_LIMIT)), ) // The original endpoint, kept as a shortcut for a compacting remake. .route("/admin/compaction", post(api::start_compaction::<H>)) diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs index deed8cd..1090e2c 100644 --- a/src/remake/quarantine.rs +++ b/src/remake/quarantine.rs @@ -16,22 +16,22 @@ use std::convert::Infallible; use automerge::AutoCommit; use axum::{ + Json, extract::{Path, State}, - http::{header, StatusCode}, + http::{StatusCode, header}, response::{IntoResponse, Response}, - Json, }; use serde::Serialize; -use serde_json::{json, Value as JsonValue}; +use serde_json::{Value as JsonValue, json}; use tracing::{info, warn}; use uuid::Uuid; +use super::Host; use super::api::error_response; use super::delta; use super::kind::Translator; use super::local::Quarantined; use super::records::unix_millis; -use super::Host; use crate::document::replica::Epoch; use crate::document::store; diff --git a/src/remake/records.rs b/src/remake/records.rs index 600d4dc..5c844f1 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -21,7 +21,7 @@ use serde_json::Value as JsonValue; use tracing::warn; use super::kind::DEFAULT_KIND; -use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES}; +use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index bd99854..efe7e37 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -24,11 +24,11 @@ use std::collections::BTreeMap; -use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash}; -use serde_json::{json, Map as JsonMap, Value as JsonValue}; +use automerge::{ActorId, AutoCommit, ChangeHash, transaction::CommitOptions}; +use serde_json::{Map as JsonMap, Value as JsonValue, json}; use sha2::{Digest, Sha256}; -use super::kind::{by_name, RemakeKind, Translator, DEFAULT_KIND}; +use super::kind::{DEFAULT_KIND, RemakeKind, Translator, by_name}; use super::records::{EpochRecord, Policy}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -353,7 +353,7 @@ pub(super) fn canonical(value: JsonValue) -> JsonValue { #[cfg(test)] mod tests { use super::*; - use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT}; + use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, Value, transaction::Transactable}; fn doc(actor: &[u8]) -> AutoCommit { AutoCommit::new().with_actor(ActorId::from(actor)) diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 2670f14..f4a018a 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -24,14 +24,14 @@ use automerge::{ActorId, AutoCommit, ChangeHash}; use tokio::sync::broadcast; use tracing::{error, info, warn}; +use super::Host; use super::catchup; use super::delta::{self, ReplayStats}; use super::kind::Translator; use super::local::LocalStatus; use super::quarantine; -use super::records::{self, unix_millis, EpochRecord, Record}; +use super::records::{self, EpochRecord, Record, unix_millis}; use super::snapshot::{self, META_ROOT, REMAKES}; -use super::Host; use crate::document::replica::Epoch; use crate::document::store; use crate::swarm::sync::paths_related; diff --git a/src/state.rs b/src/state.rs index dd360de..7517b72 100644 --- a/src/state.rs +++ b/src/state.rs @@ -2,19 +2,21 @@ use std::collections::HashMap; use std::future::Future; -use std::sync::atomic::AtomicBool; use std::sync::Arc; +use std::sync::atomic::AtomicBool; use std::time::Duration; -use anyhow::{bail, Result}; -use automerge::{sync, AutoCommit, ChangeHash}; -use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock}; +use anyhow::{Result, bail}; +use automerge::{AutoCommit, ChangeHash, sync}; +use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch}; use tokio::time::Instant; use tracing::{debug, error, warn}; +use crate::backoff::Backoff; use crate::document::replica::{Epoch, Replica}; use crate::remake::{Host, PeerSummary, RemakeState}; use crate::swarm::sync::{ChangeSet, SyncEvent}; +use crate::swarm::throughput::{SyncBytes, ThroughputTable}; use crate::transport; use crate::transport::firewall::Firewall; @@ -25,13 +27,32 @@ pub(crate) enum NodeLifecycle { Ready { swarm_id: String }, } +/// How far a peer got into the swarm, as far as we're concerned. #[derive(Debug, Clone, Default, PartialEq, Eq)] -pub(crate) enum PeerMembership { +pub(crate) enum Standing { + /// Nothing proven yet. With a swarm key, everything but a handshake from it is + /// ignored (`swarm::peer::admit`). #[default] - Unknown, - Member { - swarm_id: String, - }, + Stranger, + /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key. + Admitted, + /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it. + Member { swarm_id: String }, +} + +impl Standing { + /// The swarm a member belongs to; None before its HELLO_REPLY. + pub(crate) fn swarm_id(&self) -> Option<&str> { + match self { + Self::Member { swarm_id } => Some(swarm_id), + Self::Stranger | Self::Admitted => None, + } + } + + /// Proved the swarm key (meaningful only with one). + pub(crate) fn is_admitted(&self) -> bool { + !matches!(self, Self::Stranger) + } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -47,72 +68,135 @@ pub(crate) enum Liveness { pub(crate) enum HelloState { #[default] NotSent, - /// A send is in flight. `next_delay` is how long to wait after it before retrying. - Sending { next_delay: Duration }, - /// HELLO was sent (or the send failed); no HELLO_REPLY yet. - AwaitingReply { - next_attempt: Instant, - next_delay: Duration, - }, + /// A send is in flight. + Sending, + /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at + /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant). + AwaitingReply { next_attempt: Instant }, /// HELLO_REPLY received. No more retries. Established, } +/// A sync message on its way to our Mycelium (the HTTP request is in progress). #[derive(Clone)] -pub(crate) struct PendingOutbound { - pub(crate) message: transport::EncodedMessage, - pub(crate) retry_delay: Duration, - pub(crate) next_attempt: Instant, +pub(crate) struct Sending { + /// Changes it carries for the first time; taken back if the send fails. + pub(crate) sent: Vec<automerge::ChangeHash>, + /// A sync pass was requested meanwhile. Once this one is handed over, the peer must + /// be enqueued again so the deferred changes go out. + pub(crate) regenerate: bool, } -#[derive(Clone)] -pub(crate) struct PeerState { - pub(crate) membership: PeerMembership, - pub(crate) sync_state: sync::State, - pub(crate) hello_state: HelloState, +/// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none +/// comes. +const HELLO_RETRY: Duration = Duration::from_secs(1); +const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); - /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet, - /// or sending failed). - pub(crate) reply_owed: bool, +/// First wait before generating again after our Mycelium refused a message; doubles +/// while it keeps refusing. +const SEND_RETRY: Duration = Duration::from_millis(500); +const SEND_RETRY_MAX: Duration = Duration::from_secs(30); - /// Received from this peer, waiting to be processed. - pub(crate) pending_inbound_sync: Vec<transport::SyncEnvelope>, +#[derive(Clone)] +pub(crate) struct PeerState { + /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for + /// the pair, so computed once. + pub(crate) leads: bool, - /// Generated for this peer, waiting for delivery. - pub(crate) pending_outbound_sync: Option<PendingOutbound>, + pub(crate) standing: Standing, + pub(crate) hello_state: HelloState, + /// Wait between unanswered HELLOs; reset whenever a new handshake starts. + pub(crate) hello_backoff: Backoff, - /// A sync pass was requested while an envelope was pending. Once that envelope - /// is delivered, the peer must be re-enqueued so the deferred changes go out. - pub(crate) needs_generate: bool, + /// Everything about the current Automerge sync with this peer; a reset replaces it. + pub(crate) sync: SyncSession, pub(crate) liveness: Liveness, pub(crate) added_at: Instant, pub(crate) last_seen: Option<Instant>, pub(crate) last_ping_at: Option<Instant>, - /// Last time a sync message from this peer was applied. + /// Last time a sync message from this peer was applied (survives resets: the + /// keepalive heal looks at it). pub(crate) last_sync_rx: Option<Instant>, /// Last self-healing re-handshake (cooldown). pub(crate) last_resync_at: Option<Instant>, + /// Since when keepalives show the peer's heads differ from ours with no sync message + /// from it in between (None while they match). + pub(crate) diverged_since: Option<Instant>, + /// Epoch the peer advertised in its most recent message. pub(crate) epoch: Option<Epoch>, - /// Consecutive sync exchanges that changed nothing (livelock detection). - pub(crate) stalled_exchanges: u32, + /// Whether the peer was ready according to its most recent sync message or keepalive. + pub(crate) reports_ready: Option<bool>, + + /// Whether the peer said, in its most recent keepalive, that it would sync with us. + pub(crate) syncs_with_us: Option<bool>, + + /// How many heads its most recent keepalive listed (None if it sent a digest). + pub(crate) reported_heads: Option<usize>, + + /// HELLOs sent in the current handshake attempt without hearing anything back. + pub(crate) hello_attempts: u32, + /// Since when we consider the peer unreachable (three unanswered HELLOs, or + /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the + /// peer becomes a neighbour candidate again and gets tried once more. + pub(crate) unreachable_since: Option<Instant>, + + /// What went wrong with this peer so far (counted only, for /peers and /admin/stats). + pub(crate) incidents: Incidents, +} + +/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused +/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING. +#[derive(Debug, Default, Clone, Copy, serde::Serialize, utoipa::ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Incidents { + /// Our sync messages left unanswered until the reply timeout (then sent again). + pub(crate) reply_timeouts: u64, + /// Re-handshakes because our documents differed while sync was idle. + pub(crate) heals: u64, + /// Times the peer went dormant. + pub(crate) dormancies: u64, + /// Time spent dormant, in milliseconds, up to its last return (the current + /// dormancy, if any, isn't included). + pub(crate) dormant_ms: u64, +} + +impl Incidents { + pub(crate) fn add(&mut self, other: &Incidents) { + self.reply_timeouts += other.reply_timeouts; + self.heals += other.heals; + self.dormancies += other.dormancies; + self.dormant_ms += other.dormant_ms; + } +} + +/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart). +const UNREACHABLE_AFTER_HELLOS: u32 = 3; + +/// One Automerge sync with a peer, from a reset to the next. Everything a reset must +/// forget lives here, so a reset replaces it whole and can't miss a field. +#[derive(Clone)] +pub(crate) struct SyncSession { + pub(crate) state: sync::State, + + /// A message being handed to our Mycelium. + pub(crate) sending: Option<Sending>, - /// The peer proved the swarm key in its HELLO or HELLO_REPLY (always false and - /// irrelevant without a swarm key). - pub(crate) authenticated: bool, + /// After our Mycelium refused a message: no new one before this. The refused one is + /// not kept; the next one is generated afresh and carries everything pending. + pub(crate) retry_at: Option<Instant>, + /// The wait after the next refusal. + pub(crate) send_backoff: Backoff, /// The answer we expect to our last sync message(s), if any. - pub(crate) sync_reply: Option<SyncReplyWait>, + pub(crate) reply: Option<SyncReplyWait>, /// How long to wait for it; doubles while answers stay out. - pub(crate) sync_reply_timeout: Duration, - - /// When sync with this peer last started from scratch. - pub(crate) sync_reset_at: Instant, + pub(crate) reply_backoff: Backoff, } /// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery: @@ -121,6 +205,9 @@ pub(crate) struct PeerState { #[derive(Clone, Debug)] pub(crate) struct SyncReplyWait { pub(crate) deadline: Instant, + /// When the first of the unanswered messages was generated, and the sync bytes + /// exchanged with the peer by then (`swarm::throughput`). + pub(crate) started: (Instant, SyncBytes), /// Changes first sent in the unanswered messages. pub(crate) sent: Vec<automerge::ChangeHash>, } @@ -131,110 +218,145 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10); /// Upper bound (Mycelium gives up on a message after five minutes). const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300); -/// How long the side that doesn't open a fresh sync waits for the other's opening -/// before opening itself (the other one may not have reset, or its message got lost). -const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3); - -impl Default for PeerState { - fn default() -> Self { +impl SyncSession { + pub(crate) fn new() -> Self { Self { - membership: PeerMembership::default(), - sync_state: sync::State::new(), - hello_state: HelloState::default(), - reply_owed: false, - pending_inbound_sync: Vec::new(), - pending_outbound_sync: None, - needs_generate: false, - liveness: Liveness::Active, - added_at: Instant::now(), - last_seen: None, - last_ping_at: None, - last_sync_rx: None, - last_resync_at: None, - stalled_exchanges: 0, - authenticated: false, - sync_reply: None, - sync_reply_timeout: SYNC_REPLY_TIMEOUT, - sync_reset_at: Instant::now(), - epoch: None, + state: sync::State::new(), + sending: None, + retry_at: None, + send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), + reply: None, + reply_backoff: Backoff::new(SYNC_REPLY_TIMEOUT, SYNC_REPLY_TIMEOUT_MAX), } } -} -impl PeerState { - /// Start Automerge sync with this peer from scratch. Any envelope generated against - /// the old state is discarded; a stale RetryDelivery event is then a no-op. - pub(crate) fn reset_sync(&mut self) { - self.sync_state = sync::State::new(); - self.pending_outbound_sync = None; - self.needs_generate = false; - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; - self.sync_reset_at = Instant::now(); + /// Whether we leave the first message of this session to the peer. + /// + /// After a reset, both sides usually open at once, and the openings cross. Between + /// peers that share history that costs nothing. But a peer with an empty document + /// answers our opening with empty heads, which Automerge takes for "lost + /// everything", so the whole document would go out a second time. So a ready node + /// never opens towards a peer that isn't ready (one that is still catching up); that + /// peer opens itself. + pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool { + self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) } - /// After a reset, usually both sides do it and both open the sync at once. The - /// openings cross, and each side answers the other's: harmless between peers that - /// share history, but an empty peer answers with empty heads, which Automerge takes - /// for "lost everything", so the whole document goes out a second time. Only the - /// side with the smaller key opens; the other waits for its opening, up to - /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all. - pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> { - if we_open || self.sync_state.their_heads.is_some() { - return None; + /// Our Mycelium took the message. + pub(crate) fn send_succeeded(&mut self) -> bool { + self.retry_at = None; + self.send_backoff.reset(); + self.sending + .take() + .is_some_and(|sending| sending.regenerate) + } + + /// Our Mycelium refused the message: as far as the peer is concerned, it was never + /// sent. Takes back its changes (they go out with the next message), clears + /// `in_flight`, and returns how long to wait before generating again. None if a reset + /// already discarded what was being sent. + pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> { + let sending = self.sending.take()?; + for hash in &sending.sent { + self.state.sent_hashes.remove(hash); } - SYNC_OPENING_GRACE - .checked_sub(now.saturating_duration_since(self.sync_reset_at)) - .filter(|wait| !wait.is_zero()) + if let Some(wait) = &mut self.reply { + wait.sent.retain(|hash| !sending.sent.contains(hash)); + } + self.state.in_flight = false; + let delay = self.send_backoff.next(); + self.retry_at = Some(now + delay); + Some(delay) } /// A sync message went out; `sent` are the changes it carried for the first time. - pub(crate) fn expect_sync_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { - match &mut self.sync_reply { + /// `bytes` are the sync bytes exchanged with the peer before it. + pub(crate) fn expect_reply( + &mut self, + sent: Vec<automerge::ChangeHash>, + now: Instant, + bytes: SyncBytes, + ) { + match &mut self.reply { // Still waiting for an earlier one: keep its deadline, remember these too. Some(wait) => wait.sent.extend(sent), None => { - self.sync_reply = Some(SyncReplyWait { - deadline: now + self.sync_reply_timeout, + self.reply = Some(SyncReplyWait { + deadline: now + self.reply_backoff.current(), + started: (now, bytes), sent, }); } } } - /// Keepalives show both documents are the same: stop waiting for an answer. - pub(crate) fn sync_in_step(&mut self) { - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; - } - - /// The peer answered. - pub(crate) fn sync_reply_received(&mut self) { - self.sync_reply = None; - self.sync_reply_timeout = SYNC_REPLY_TIMEOUT; + /// No answer is due any more: the peer answered, or keepalives show both documents + /// are the same. + pub(crate) fn reply_settled(&mut self) { + self.reply = None; + self.reply_backoff.reset(); } /// If the answer is overdue: forgets that the unanswered changes were sent and that /// a message is in flight, so the next sync pass sends them again. Returns how many /// changes that concerns. - pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { - if self.liveness != Liveness::Active - || self.hello_state != HelloState::Established - || self - .sync_reply - .as_ref() - .is_none_or(|wait| now < wait.deadline) - { + fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> { + if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) { return None; } - let wait = self.sync_reply.take()?; + let wait = self.reply.take()?; for hash in &wait.sent { - self.sync_state.sent_hashes.remove(hash); + self.state.sent_hashes.remove(hash); } - self.sync_state.in_flight = false; - self.sync_reply_timeout = (self.sync_reply_timeout * 2).min(SYNC_REPLY_TIMEOUT_MAX); + self.state.in_flight = false; + self.reply_backoff.grow(); Some(wait.sent.len()) } +} + +impl PeerState { + pub(crate) fn new(leads: bool) -> Self { + Self { + leads, + standing: Standing::default(), + hello_state: HelloState::default(), + hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX), + sync: SyncSession::new(), + liveness: Liveness::Active, + added_at: Instant::now(), + last_seen: None, + last_ping_at: None, + last_sync_rx: None, + last_resync_at: None, + diverged_since: None, + epoch: None, + reports_ready: None, + syncs_with_us: None, + reported_heads: None, + hello_attempts: 0, + unreachable_since: None, + incidents: Incidents::default(), + } + } + + /// Start Automerge sync with this peer from scratch. A message still being handed to + /// Mycelium was generated against the old state; its outcome is then ignored. + pub(crate) fn reset_sync(&mut self) { + self.sync = SyncSession::new(); + } + + /// An overdue sync answer from an established, active peer (see + /// `SyncSession::take_overdue_reply`). + pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> { + if self.liveness != Liveness::Active || self.hello_state != HelloState::Established { + return None; + } + let overdue = self.sync.take_overdue_reply(now); + if overdue.is_some() { + self.incidents.reply_timeouts += 1; + } + overdue + } fn quiet_for(&self, now: Instant) -> Duration { now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at)) @@ -243,11 +365,14 @@ impl PeerState { /// Records inbound activity. Returns true if the peer was dormant. pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { self.last_seen = Some(now); + self.hello_attempts = 0; + self.unreachable_since = None; - if matches!(self.liveness, Liveness::Dormant { .. }) { + if let Liveness::Dormant { since } = self.liveness { + self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64; self.liveness = Liveness::Active; // Force a fresh handshake; its reply resets sync on both sides. - if !matches!(self.hello_state, HelloState::Sending { .. }) { + if self.hello_state != HelloState::Sending { self.hello_state = HelloState::NotSent; } return true; @@ -255,6 +380,29 @@ impl PeerState { false } + /// A HELLO went out (or failed to). After the third unanswered one, the peer counts + /// as unreachable. + pub(crate) fn hello_sent(&mut self, now: Instant) { + self.hello_attempts += 1; + if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS { + self.unreachable_since.get_or_insert(now); + } + } + + /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). + pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool { + self.unreachable_since + .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe) + } + + /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight + /// finishes; a later handshake starts over. + pub(crate) fn stop_hello_retries(&mut self) { + if matches!(self.hello_state, HelloState::AwaitingReply { .. }) { + self.hello_state = HelloState::NotSent; + } + } + /// Moves a silent peer to Dormant. Returns true on the transition. pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool { if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after { @@ -262,12 +410,13 @@ impl PeerState { } self.liveness = Liveness::Dormant { since: now }; + self.incidents.dormancies += 1; + self.unreachable_since.get_or_insert(now); self.reset_sync(); - if !matches!(self.hello_state, HelloState::Sending { .. }) { + if self.hello_state != HelloState::Sending { self.hello_state = HelloState::AwaitingReply { next_attempt: now + timing.dormant_probe, - next_delay: timing.dormant_probe, }; } true @@ -276,19 +425,14 @@ impl PeerState { /// True (and records the ping) if an established, quiet peer needs a keepalive. /// /// One PING and its PONG tell both sides that the other is alive, so only one side of - /// a pair needs to ping: the one with the smaller key (`we_ping`). Each of its PINGs + /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs /// makes the other side's peer look recently heard from, so the other side never /// pings while it does. If the pings stop, the other side pings once it has heard /// nothing for half an interval more. Without that, both sides would get due at about /// the same time (each one's quiet time starts with the same exchange) and their /// PINGs would often cross. - pub(crate) fn take_keepalive_due( - &mut self, - timing: &Timing, - we_ping: bool, - now: Instant, - ) -> bool { - let quiet_enough = if we_ping { + pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool { + let quiet_enough = if self.leads { timing.keepalive } else { timing.keepalive + timing.keepalive / 2 @@ -307,43 +451,48 @@ impl PeerState { due } - /// Schedules a fresh handshake, which resets sync on both sides: the peer resets on - /// our HELLO, we reset on its reply. The peer worker sends the HELLO on its next tick. - /// Usable without awaiting (e.g. under a blocking lock). - pub(crate) fn request_resync(&mut self) { - if matches!(self.hello_state, HelloState::Sending { .. }) { - return; // a HELLO is already on its way - } - self.hello_state = HelloState::AwaitingReply { - next_attempt: Instant::now(), - next_delay: Duration::from_millis(500), - }; - } - /// Self-healing check on PING/PONG. Returns true if a re-handshake should start. pub(crate) fn needs_heal( &mut self, - local: &transport::HeadsDigest, - remote: &transport::HeadsDigest, + local: &transport::HeadsSummary, + remote: &transport::HeadsSummary, timing: &Timing, now: Instant, ) -> bool { - if local == remote - || self.liveness != Liveness::Active + if local == remote { + self.diverged_since = None; + return false; + } + + // How long the heads have differed without sync making progress. Measured from + // when we noticed, not from the last sync message: in a quiet swarm that one is + // long past, and the first keepalive after a lost message would heal at once, + // before the sender's reply timeout could resend it. Any sync message restarts + // the clock. + let noticed = match self.diverged_since { + Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since, + _ => *self.diverged_since.insert(now), + }; + + // While an answer to our last sync message is due, a lost message is the reply + // timeout's to resend; healing is for when neither side has anything outstanding. + if self.liveness != Liveness::Active || self.hello_state != HelloState::Established + || self.sync.reply.is_some() { return false; } - let sync_idle = now.saturating_duration_since(self.last_sync_rx.unwrap_or(self.added_at)) - >= timing.resync_after; + let persisted = now.saturating_duration_since(noticed) >= timing.resync_after; let cooled_down = self .last_resync_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after); - if !(sync_idle && cooled_down) { + if !(persisted && cooled_down) { return false; } + self.diverged_since = None; + self.incidents.heals += 1; self.last_resync_at = Some(now); self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply @@ -366,9 +515,17 @@ pub(crate) struct Timing { pub(crate) dormant_after: Duration, pub(crate) resync_after: Duration, pub(crate) dormant_probe: Duration, + /// While catching up: PING the source after this much silence (`swarm::source`). + pub(crate) catch_up_probe: Duration, } impl Timing { + /// How long a round-trip sample stays meaningful (`swarm::rtt`): a few keepalive + /// intervals, since nodes move. + pub(crate) fn rtt_fresh_for(&self) -> Duration { + self.keepalive * 5 + } + // Recommended values // Environment | Keepalive | Notes // --------------------------------------------+-----------+------------------------------------------------------------ @@ -380,8 +537,13 @@ impl Timing { Self { keepalive, dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row - resync_after: keepalive * 2, // resync_after should be >= 2* keepalive to guarantee at least one digest exchange while sync is idle + // >= 2 keepalives, so a second exchange confirms the difference; and >= twice + // the first sync reply timeout, so a lost message's sender resends it first. + resync_after: (keepalive * 2).max(SYNC_REPLY_TIMEOUT * 2), dormant_probe: keepalive * 10, // >= 10*: dormant peers cost almost nothing, but still rejoin within minutes + // Seconds, not a keepalive interval: a dead source should cost little time. + // Only the source gets these, and only while catching up. + catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), } } } @@ -402,6 +564,10 @@ pub(crate) struct AppState { pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>, pub(crate) transport: Arc<dyn transport::Transport>, pub(crate) firewall: Arc<Firewall>, + /// Messages and bytes exchanged, per peer (`transport::traffic`). + pub(crate) traffic: Arc<transport::traffic::Traffic>, + /// Node-wide figures for /admin/stats (`swarm::stats`). + pub(crate) stats: Arc<crate::swarm::stats::NodeStats>, pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>, @@ -421,9 +587,45 @@ pub(crate) struct AppState { /// knows the swarm ID. pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, + /// Random per run of this node; our HELLOs carry it and replies must echo it. + pub(crate) run_id: transport::RunId, + + /// Target number of neighbours (`--neighbours`); up to this many + 1 members the + /// swarm is a full mesh. + pub(crate) neighbours_k: usize, + /// Our alias (`--alias`), published in the member list (`swarm::members`). + pub(crate) alias: Option<String>, + /// Our own connectivity record and stability tier (`swarm::stability`). + pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, + /// Round-trip times to peers (`swarm::rtt`). + pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>, + /// Sync throughput with peers (`swarm::throughput`). + pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>, + /// The neighbour we are pulling changes from, and since when (`swarm::relay`). + pub(crate) pulling: Arc<std::sync::Mutex<Option<(String, Instant)>>>, + /// Our neighbours, once known (`swarm::topology`). + pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>, + /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>, + + /// While the initial sync isn't done: the one peer we catch up from. + pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>, + /// Since when a closer peer has had a head start to become the source. + pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>, +} + +/// The peer a catching-up node syncs with first (`swarm::source`). +#[derive(Clone, Debug)] +pub(crate) struct CatchUpSource { + pub(crate) peer: String, + /// When it was chosen or last sent us anything. + pub(crate) last_heard: Instant, + /// Our last probe PING to it. + pub(crate) last_probe: Option<Instant>, + /// Its keepalives in a row saying it wouldn't sync with us. + pub(crate) refusals: u32, } impl AppState { @@ -431,6 +633,12 @@ impl AppState { self.transport.local_id() } + pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> { + self.rtt + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + /// The local epoch. Code that labels or applies sync messages must use the epoch /// of the locked `Replica` instead, so label and document can't disagree. pub(crate) fn local_epoch(&self) -> Epoch { @@ -539,10 +747,16 @@ impl AppState { Arc::clone( peers .entry(pk.to_owned()) - .or_insert_with(|| Arc::new(Mutex::new(PeerState::default()))), + .or_insert_with(|| self.new_peer_state(pk)), ) } + /// State for a peer we haven't known so far (not inserted). + pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> { + let leads = crate::swarm::pair::leads(self.local_id(), pk); + Arc::new(Mutex::new(PeerState::new(leads))) + } + pub(crate) fn notify_store_changed(&self) { self.change_tx.send_modify(|version| { *version = version.wrapping_add(1); @@ -604,7 +818,12 @@ impl AppState { match &result { // One API write = one Automerge change. Ok(_) => { - replica.doc.commit(); + // Stamped with our clock (milliseconds), so receivers can tell + // how long the change took to reach them (`swarm::stats`). + replica.doc.commit_with( + automerge::transaction::CommitOptions::default() + .with_time(unix_millis() as i64), + ); } // Discard partial ops (e.g. old value deleted, new one not written). Err(_) => { @@ -649,15 +868,31 @@ impl AppState { self.publish_change(); // Tell peers about the new epoch now instead of with the next keepalive (up to a - // keepalive interval later): sync with peers already on it resumes right away, - // and peers that never saw the commit start catching up. - let keepalive = crate::swarm::peer::local_keepalive(self).await; - for (id, _) in &peers { - if let Err(error) = self - .transport - .send_message(id, &transport::ParsedMessage::Ping(keepalive.clone())) - .await - { + // keepalive interval later). Ready peers already on it learn it from our sync + // opening, which goes out right away; everyone else gets a PING, so peers that + // never saw the commit start catching up (we don't sync with them, so nothing + // else would tell them). + let local = self.local_epoch(); + let mut announce = Vec::new(); + for (id, peer_state_mutex) in &peers { + // Others learn it from their own neighbours. + if !crate::swarm::topology::is_neighbour(self, id) { + continue; + } + let mut peer_state = peer_state_mutex.lock().await; + // What it said about syncing with us was about our old epoch. + peer_state.syncs_with_us = None; + let gets_our_opening = peer_state.liveness == Liveness::Active + && matches!(peer_state.standing, Standing::Member { .. }) + && peer_state.epoch == Some(local) + && peer_state.reports_ready == Some(true); + if !gets_our_opening { + announce.push(id); + } + } + + for id in announce { + if let Err(error) = crate::swarm::peer::send_ping(self, id).await { debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed"); } } @@ -860,6 +1095,14 @@ impl Host for AppState { fn reset_initial_sync(&self) { self.initial_sync_done .store(false, std::sync::atomic::Ordering::Release); + *self + .catch_up_source + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + *self + .catch_up_preference + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; } fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> { diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs index be748e6..31230a5 100644 --- a/src/swarm/auth.rs +++ b/src/swarm/auth.rs @@ -8,8 +8,8 @@ //! The tag is `HMAC-SHA256(key, label ‖ swarm_id ‖ from_pk ‖ to_pk)`. It needs no //! challenge: it is bound to both keys, and only `from_pk` can send as `from_pk`, so a //! captured tag is useless to anyone else. Different labels for HELLO and HELLO_REPLY -//! keep one from being passed off as the other. A HELLO has no swarm ID yet (a new node -//! doesn't know it), so its tag binds an empty one. +//! keep one from being passed off as the other. A HELLO binds the swarm ID its sender +//! claims, or an empty one from a new node that doesn't know it yet. //! //! Limits of a shared key: every member holds the same secret, so a single member can't //! be revoked (forgetting it lasts until it says HELLO again), and changing the key @@ -19,7 +19,7 @@ use std::path::Path; use std::sync::Mutex; use std::time::{Duration, Instant}; -use anyhow::{bail, Context, Result}; +use anyhow::{Context, Result, bail}; use sha2::{Digest, Sha256}; const HELLO_LABEL: &[u8] = b"continuum/hello/1"; @@ -82,9 +82,9 @@ impl SwarmKey { Some(std::mem::take(&mut rejections.suppressed)) } - /// Tag for a HELLO from `from` to `to`. - pub(crate) fn hello_tag(&self, from: &str, to: &str) -> Vec<u8> { - self.tag(HELLO_LABEL, "", from, to).to_vec() + /// Tag for a HELLO from `from` to `to`, claiming `swarm_id` ("" if unknown). + pub(crate) fn hello_tag(&self, swarm_id: &str, from: &str, to: &str) -> Vec<u8> { + self.tag(HELLO_LABEL, swarm_id, from, to).to_vec() } /// Tag for a HELLO_REPLY from `from` to `to`, naming the swarm. @@ -92,8 +92,8 @@ impl SwarmKey { self.tag(HELLO_REPLY_LABEL, swarm_id, from, to).to_vec() } - pub(crate) fn verify_hello(&self, tag: &[u8], from: &str, to: &str) -> bool { - constant_time_eq(tag, &self.tag(HELLO_LABEL, "", from, to)) + pub(crate) fn verify_hello(&self, tag: &[u8], swarm_id: &str, from: &str, to: &str) -> bool { + constant_time_eq(tag, &self.tag(HELLO_LABEL, swarm_id, from, to)) } pub(crate) fn verify_hello_reply( @@ -176,18 +176,19 @@ mod tests { let key = SwarmKey::new(vec![1; 32]); let other = SwarmKey::new(vec![2; 32]); - let hello = key.hello_tag("A", "B"); - assert!(key.verify_hello(&hello, "A", "B")); - assert!(!other.verify_hello(&hello, "A", "B")); - assert!(!key.verify_hello(&hello, "B", "A")); - assert!(!key.verify_hello(&hello, "A", "C")); - assert!(!key.verify_hello(&[], "A", "B")); + let hello = key.hello_tag("swarm", "A", "B"); + assert!(key.verify_hello(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&hello, "", "A", "B")); + assert!(!other.verify_hello(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&hello, "swarm", "B", "A")); + assert!(!key.verify_hello(&hello, "swarm", "A", "C")); + assert!(!key.verify_hello(&[], "swarm", "A", "B")); let reply = key.hello_reply_tag("swarm", "A", "B"); assert!(key.verify_hello_reply(&reply, "swarm", "A", "B")); assert!(!key.verify_hello_reply(&reply, "other-swarm", "A", "B")); // A HELLO tag is no HELLO_REPLY tag, and the other way round. - assert!(!key.verify_hello_reply(&hello, "", "A", "B")); - assert!(!key.verify_hello(&key.hello_reply_tag("", "A", "B"), "A", "B")); + assert!(!key.verify_hello_reply(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&key.hello_reply_tag("swarm", "A", "B"), "swarm", "A", "B")); } } diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 7290212..27ffdd8 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -3,11 +3,11 @@ use std::sync::atomic::Ordering; use std::time::Duration; -use anyhow::{bail, Result}; +use anyhow::{Result, bail}; use tracing::{debug, error, info, warn}; use uuid::Uuid; -use crate::state::{AppState, PeerMembership}; +use crate::state::AppState; use crate::swarm::peer::{remember_peer, send_hello_once}; const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300); @@ -68,15 +68,10 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> for (peer, peer_state_mutex) in &peer_states { let peer_state = peer_state_mutex.lock().await; - let validated_swarm = match &peer_state.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { swarm_id } => Some(swarm_id.as_str()), - }; + let validated_swarm = peer_state.standing.swarm_id(); peer_diagnostics.push(format!( - "{peer}: validated_swarm={:?}, epoch={:?}, pending_inbound_sync={}", - validated_swarm, - peer_state.epoch, - peer_state.pending_inbound_sync.len() + "{peer}: validated_swarm={:?}, epoch={:?}", + validated_swarm, peer_state.epoch, )) } let local_swarm = state.swarm_id().await; diff --git a/src/swarm/members.rs b/src/swarm/members.rs new file mode 100644 index 0000000..e02dba4 --- /dev/null +++ b/src/swarm/members.rs @@ -0,0 +1,169 @@ +//! The member registry: every node that has joined the swarm, under +//! `_continuum/members/<pk>`, replicated with the document (and carried into new +//! epochs with the rest of `_continuum`). +//! +//! A pure registry: it decides nothing about admission (that's the swarm key). Each +//! node registers itself once it is ready, and re-registers if its entry disappears +//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry +//! together with writing the forget record; a forgotten node that turns out to be alive +//! cancels the record and registers again. +//! +//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name +//! for people reading /peers, metrics and logs. Not unique and never an identity: the +//! public key stays that. + +use std::collections::HashMap; + +use automerge::AutoCommit; +use serde_json::{Value as JsonValue, json}; +use tracing::{info, warn}; + +use crate::document::store::{self, StoreError}; +use crate::state::{AppState, unix_millis}; + +const META_ROOT: &str = "_continuum"; + +fn members_dir() -> Vec<String> { + vec![META_ROOT.to_owned(), "members".to_owned()] +} + +pub(crate) fn member_path(pk: &str) -> Vec<String> { + let mut path = members_dir(); + path.push(pk.to_owned()); + path +} + +/// Public keys of all registered members, sorted. +pub(crate) fn members(doc: &AutoCommit) -> Vec<String> { + let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default(); + members.sort(); + members +} + +/// A member's stability tier, judged by itself (`swarm::stability`). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum Tier { + Stable, + #[default] + Transient, +} + +impl Tier { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Stable => "stable", + Self::Transient => "transient", + } + } +} + +fn tier_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("tier".to_owned()); + path +} + +/// `pk`'s tier; transient unless its entry says stable. +pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier { + match store::read_path(doc, &tier_path(pk)) { + Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable, + _ => Tier::Transient, + } +} + +pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> { + store::write_path(doc, &tier_path(pk), &json!(tier.as_str())) +} + +fn alias_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("alias".to_owned()); + path +} + +/// `pk`'s alias, if it gave itself one. +pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> { + match store::read_path(doc, &alias_path(pk)) { + Ok(Some(JsonValue::String(alias))) => Some(alias), + _ => None, + } +} + +/// The aliases of all members that have one. +pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> { + members(doc) + .into_iter() + .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias))) + .collect() +} + +fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> { + match alias { + Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)), + None => match store::delete_path(doc, &alias_path(pk), false) { + Err(StoreError::NotFound) => Ok(()), + result => result, + }, + } +} + +/// Registers this node if it is ready and its entry is missing, and keeps its alias +/// as configured (set, changed or removed after a restart with another `--alias`). +pub(crate) async fn ensure_registered(state: &AppState) { + if !state.is_ready().await { + return; + } + let local = state.local_id().to_owned(); + let own = member_path(&local); + let (registered, recorded_alias) = { + let replica = state.doc.read().await; + let registered = store::read_path(&replica.doc, &own) + .ok() + .flatten() + .is_some(); + (registered, alias(&replica.doc, &local)) + }; + let wanted = state.alias.clone(); + + if !registered { + let mut entry = json!({ "since": unix_millis() }); + if let Some(alias) = &wanted { + entry["alias"] = json!(alias); + } + match state + .mutate_doc(move |doc| store::write_path(doc, &own, &entry)) + .await + { + Ok(()) => info!( + alias = wanted.as_deref(), + "Registered in the swarm's member list" + ), + Err(error) => warn!(error = %error, "Failed to register in the member list"), + } + return; + } + + if recorded_alias == wanted { + return; + } + let alias = wanted.clone(); + match state + .mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref())) + .await + { + Ok(()) => info!( + alias = wanted.as_deref(), + previous = recorded_alias.as_deref(), + "Alias updated in the member list" + ), + Err(error) => warn!(error = %error, "Failed to record our alias"), + } +} + +/// Removes `pk`'s entry (an operator forgot it). Absent is fine. +pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> { + match store::delete_path(doc, &member_path(pk), false) { + Err(StoreError::NotFound) => Ok(()), + result => result, + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index a4277c7..6f194dc 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -3,5 +3,14 @@ pub(crate) mod auth; pub(crate) mod bootstrap; +pub(crate) mod members; +pub(crate) mod pair; pub(crate) mod peer; +pub(crate) mod relay; +pub(crate) mod rtt; +pub(crate) mod source; +pub(crate) mod stability; +pub(crate) mod stats; pub(crate) mod sync; +pub(crate) mod throughput; +pub(crate) mod topology; diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs new file mode 100644 index 0000000..c1a196b --- /dev/null +++ b/src/swarm/pair.rs @@ -0,0 +1,78 @@ +//! Which side of a pair of nodes leads, for duties only one side needs to do (sending +//! keepalive PINGs). +//! +//! Both sides decide on their own, from the two keys alone, and agree. The leader is +//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a +//! global order would make the node with the smallest key lead every one of its pairs +//! and the one with the largest lead none, so every such duty would pile up on the same +//! few nodes. With the hash, each node leads about half of its pairs. + +use sha2::{Digest, Sha256}; + +const LABEL: &[u8] = b"continuum/pair-leader/1"; + +/// Whether `local` leads its pair with `peer`. Exactly one of `leads(a, b)` and +/// `leads(b, a)` is true for any two different keys. +pub(crate) fn leads(local: &str, peer: &str) -> bool { + let (low, high) = if local <= peer { + (local, peer) + } else { + (peer, local) + }; + let mut hash = Sha256::new(); + // Length-prefixed fields, so no two different pairs hash the same input. + for field in [LABEL, low.as_bytes(), high.as_bytes()] { + hash.update((field.len() as u32).to_be_bytes()); + hash.update(field); + } + let leader = if hash.finalize()[0] & 1 == 0 { + low + } else { + high + }; + leader == local +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(n: u32) -> String { + format!("{:064x}", u64::from(n).wrapping_mul(0x9e37_79b9_7f4a_7c15)) + } + + #[test] + fn exactly_one_side_leads() { + for a in 0..50 { + for b in 0..50 { + if a != b { + assert_ne!(leads(&key(a), &key(b)), leads(&key(b), &key(a))); + } + } + } + } + + #[test] + fn leading_is_spread_across_nodes() { + let nodes: Vec<String> = (0..64).map(key).collect(); + let mut smallest_key_leads = 0; + for (i, node) in nodes.iter().enumerate() { + let led = nodes + .iter() + .filter(|peer| *peer != node && leads(node, peer)) + .count(); + // 63 pairs per node: about half each; far from all or none. + assert!((10..=53).contains(&led), "node {i} leads {led} of 63 pairs"); + for peer in &nodes { + if node < peer && leads(node, peer) { + smallest_key_leads += 1; + } + } + } + // Not simply the smaller key of each pair (64 * 63 / 2 = 2016 pairs). + assert!( + (800..=1216).contains(&smallest_key_leads), + "{smallest_key_leads}" + ); + } +} diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 5928ac2..3b85539 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -2,24 +2,22 @@ //! swarm-wide forget records, and dispatch of inbound messages. use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; -use anyhow::{anyhow, bail, Context, Result}; +use anyhow::{Context, Result, anyhow, bail}; use automerge::AutoCommit; -use sha2::{Digest, Sha256}; use tokio::sync::Mutex; use tokio::time::{Instant, MissedTickBehavior}; use tracing::{debug, info, instrument, trace, warn}; use crate::document::replica::{Epoch, EpochRelation}; use crate::document::store::{self, StoreError}; -use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState}; -use crate::swarm::sync::{handle_sync, SyncEvent}; +use crate::state::{AppState, HelloState, Liveness, PeerState, Standing}; +use crate::swarm::sync::{SyncEvent, handle_sync}; +use crate::swarm::topology; use crate::transport::{self, ParsedMessage}; -const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1); -const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); - const PEER_TICK: Duration = Duration::from_millis(500); const META_ROOT: &str = "_continuum"; @@ -98,24 +96,63 @@ async fn apply_forget_records(state: &AppState) { // --- keepalive and self-healing ---------------------------------------------- -fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest { - let mut heads = doc.get_heads(); - heads.sort_unstable(); - let mut hasher = Sha256::new(); - for head in &heads { - hasher.update(head.0); - } - hasher.finalize().into() +/// Sends `peer` a PING with our keepalive, noting the time for its round trip. +pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { + let keepalive = local_keepalive(state, peer).await; + state.lock_rtt().ping_sent(peer, Instant::now()); + state + .transport + .send_message(peer, &ParsedMessage::Ping(keepalive)) + .await } -pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive { +/// Our keepalive for `peer`. +pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { // AutoCommit::get_heads takes &mut self (it closes the pending transaction). - // Digest and epoch come from the same locked replica, so they always match. - let mut replica = state.doc.write().await; + // Heads and epoch come from the same locked replica, so they always match. + let (heads, epoch) = { + let mut replica = state.doc.write().await; + ( + transport::HeadsSummary::of(replica.doc.get_heads()), + replica.epoch, + ) + }; + let syncs_with_you = match state.get_peer_state(peer).await { + Some(peer_state) => { + let peer_state = peer_state.lock().await; + peer_state.liveness == Liveness::Active + && matches!(peer_state.standing, Standing::Member { .. }) + && peer_state.epoch == Some(epoch) + } + None => false, + }; transport::Keepalive { - heads_digest: heads_digest(&mut replica.doc), - epoch: replica.epoch, + heads, + epoch, + ready: state.is_ready().await, + syncs_with_you, + } +} + +/// What a peer's keepalive says about it. +async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + let became_ready = remote.ready && peer_state.reports_ready != Some(true); + peer_state.reports_ready = Some(remote.ready); + peer_state.syncs_with_us = Some(remote.syncs_with_you); + peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); + drop(peer_state); + crate::swarm::relay::pull_if_behind(state, peer, remote).await; + if became_ready { + // We may have left the opening to it while it wasn't ready, and nothing else + // would make us look again. + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())); + } } + crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you)); } async fn heal_if_diverged( @@ -124,8 +161,9 @@ async fn heal_if_diverged( local: &transport::Keepalive, remote: &transport::Keepalive, ) { - // Digests of documents from different epochs are never equal and never converge. - if local.epoch != remote.epoch { + // Heads of documents from different epochs are never equal and never converge. + // And a peer that isn't our neighbour isn't ours to keep in step. + if local.epoch != remote.epoch || !topology::is_neighbour(state, peer) { return; } @@ -135,18 +173,17 @@ async fn heal_if_diverged( let heal = { let mut peer_state = peer_state_mutex.lock().await; - if local.heads_digest == remote.heads_digest { + if local.heads == remote.heads { // Same document: whatever answer we were waiting for, there is nothing left // to send. (Automerge answers only when it has something to say, so a sync // message to a peer that already has everything gets no reply at all.) - peer_state.sync_in_step(); + peer_state.sync.reply_settled(); } - peer_state.needs_heal( - &local.heads_digest, - &remote.heads_digest, - &state.timing, - Instant::now(), - ) + // While we catch up, we hold back sync with everyone but the source on purpose: + // differing heads and idle sync are expected then, not a reason to heal. + let catching_up = !state.initial_sync_done.load(Ordering::Acquire); + !catching_up + && peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now()) }; if heal { @@ -171,7 +208,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) { let mut peer_states = state.peer_states.write().await; if !peer_states.contains_key(peer) { - peer_states.entry(peer.to_owned()).or_default(); + peer_states.insert(peer.to_owned(), state.new_peer_state(peer)); info!( peer = %peer, peer_count = peer_states.len(), @@ -191,9 +228,10 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> let claimed = { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::NotSent { - peer_state.hello_state = HelloState::Sending { - next_delay: HELLO_RETRY_INITIAL, - }; + // A new handshake: retries start over. + peer_state.hello_state = HelloState::Sending; + peer_state.hello_backoff.reset(); + peer_state.hello_attempts = 0; true } else { trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending"); @@ -206,7 +244,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> } debug!(to = %peer, "Sending HELLO"); - transmit_hello(state, peer, HELLO_RETRY_INITIAL).await + transmit_hello(state, peer).await } async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { @@ -214,43 +252,41 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { return Ok(()); }; - let delay = { + { let mut peer_state = peer_state_mutex.lock().await; match peer_state.hello_state { - HelloState::AwaitingReply { - next_attempt, - next_delay, - } if Instant::now() >= next_attempt => { - peer_state.hello_state = HelloState::Sending { next_delay }; - next_delay + HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => { + peer_state.hello_state = HelloState::Sending; } _ => return Ok(()), } - }; + } - debug!( - peer = %peer, - next_retry_ms = delay.as_millis(), - "No HELLO_REPLY yet; re-sending HELLO" - ); - transmit_hello(state, peer, delay).await + debug!(peer = %peer, "No HELLO_REPLY yet; re-sending HELLO"); + transmit_hello(state, peer).await } /// Sends a HELLO without holding any lock, then schedules the next retry. /// Success and failure both end in AwaitingReply: Mycelium accepting the message /// says nothing about whether the peer received it. -async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> { +async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { + let swarm_id = state.swarm_id().await; + let auth = state + .swarm_key + .as_ref() + .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) + .unwrap_or_default(); + state.lock_rtt().hello_sent(peer, Instant::now()); let result = state .transport .send_message( peer, &ParsedMessage::Hello(transport::HelloRequest { + swarm_id, epoch: state.local_epoch(), - auth: state - .swarm_key - .as_ref() - .map(|key| key.hello_tag(state.local_id(), peer)) - .unwrap_or_default(), + auth, + ready: state.is_ready().await, + run_id: state.run_id, }), ) .await; @@ -258,16 +294,14 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result if let Some(peer_state_mutex) = state.get_peer_state(peer).await { let mut peer_state = peer_state_mutex.lock().await; - if matches!(peer_state.hello_state, HelloState::Sending { .. }) { - let (wait, next_delay) = match peer_state.liveness { - Liveness::Dormant { .. } => { - (state.timing.dormant_probe, state.timing.dormant_probe) - } - Liveness::Active => (delay, delay.saturating_mul(2).min(HELLO_RETRY_MAX)), + if peer_state.hello_state == HelloState::Sending { + peer_state.hello_sent(Instant::now()); + let wait = match peer_state.liveness { + Liveness::Dormant { .. } => state.timing.dormant_probe, + Liveness::Active => peer_state.hello_backoff.next(), }; peer_state.hello_state = HelloState::AwaitingReply { next_attempt: Instant::now() + wait, - next_delay, }; } } @@ -281,36 +315,14 @@ async fn ensure_handshake(state: &AppState, peer: &str) { } } -/// Sends a HELLO_REPLY if one is owed and our swarm ID is known. -/// Otherwise it stays owed and the peer worker tries again. -async fn send_owed_hello_reply(state: &AppState, peer: &str) { - if state.swarm_id().await.is_none() { - return; - } - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return; - }; - - let owed = std::mem::take(&mut peer_state_mutex.lock().await.reply_owed); - if !owed { - return; - } - - if let Err(error) = send_hello_reply(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed; will retry"); - peer_state_mutex.lock().await.reply_owed = true; - } -} - -async fn flush_owed_hello_replies(state: &AppState) { - let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); - for peer in &peers { - send_owed_hello_reply(state, peer).await; - } -} - -async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { +/// Answers a HELLO that carried `run_id`; `away` is how long before it we had last heard +/// from the peer. +async fn send_hello_reply( + state: &AppState, + peer: &str, + run_id: transport::RunId, + away: Option<Duration>, +) -> Result<()> { let swarm_id = state .swarm_id() .await @@ -330,6 +342,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { let peer_state = peer_state_mutex.lock().await; if peer_state.liveness == Liveness::Active && peer_state.hello_state == HelloState::Established + && topology::is_neighbour(state, &known) { peers.push(known); } @@ -340,15 +353,17 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); + let ready = state.is_ready().await; let reply = transport::HelloReply { swarm_id, peers, epoch: state.local_epoch(), auth, + ready, + run_id, + away_ms: away.map(|away| away.as_millis() as u64), }; - let ready = state.is_ready().await; - debug!(to = %peer, ready, "Sending HELLO_REPLY"); state @@ -370,6 +385,9 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S remember_peer(state, discovered_peer).await; + if !topology::is_neighbour(state, discovered_peer) { + continue; + } if let Err(error) = send_hello_once(state, discovered_peer).await { warn!( peer = %discovered_peer, @@ -380,6 +398,26 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S } } +/// New neighbours get our HELLO (we may not have known them before: the member list is +/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives +/// or sync openings from us; their session is reset, so we don't keep pushing changes +/// to them either. They are still answered if they contact us. +async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { + for peer in &change.added { + remember_peer(state, peer).await; + if let Err(error) = send_hello_once(state, peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry"); + } + } + for peer in &change.removed { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + peer_state.stop_hello_retries(); + peer_state.reset_sync(); + } + } +} + // --- peer worker ------------------------------------------------------------- pub(crate) async fn run_peer_worker(state: AppState) { @@ -406,12 +444,17 @@ pub(crate) async fn run_peer_worker(state: AppState) { for (peer, peer_state_mutex) in &peers { let now = Instant::now(); + // Only neighbours are kept alive and judged; others we only answer. + let neighbour = topology::is_neighbour(&state, peer); let (became_dormant, ping_due, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; - let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); - let we_ping = state.local_id() < peer.as_str(); - let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now); + let became_dormant = + ready && neighbour && peer_state.check_dormancy(&state.timing, now); + let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now); + if !neighbour { + peer_state.stop_hello_retries(); + } let unanswered = peer_state.take_overdue_sync_reply(now); (became_dormant, ping_due, unanswered) }; @@ -432,13 +475,10 @@ pub(crate) async fn run_peer_worker(state: AppState) { ); } - if ping_due { - let local = local_keepalive(&state).await; - if let Err(error) = state - .transport - .send_message(peer, &ParsedMessage::Ping(local)) - .await - { + let probe_due = + !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now()); + if ping_due || probe_due { + if let Err(error) = send_ping(&state, peer).await { debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); } } @@ -446,12 +486,25 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Err(error) = retry_hello_if_due(&state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed"); } - - send_owed_hello_reply(&state, peer).await; } if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) { apply_forget_records(&state).await; + crate::swarm::members::ensure_registered(&state).await; + crate::swarm::stability::tick(&state).await; + crate::swarm::stats::measure_saved_size(&state); + { + let known = state.peer_states.read().await; + state.traffic.retain(|peer| known.contains_key(peer)); + state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .retain(|peer| known.contains_key(peer)); + } + if let Some(change) = topology::refresh(&state).await { + apply_neighbour_change(&state, change).await; + } last_forget_check = Some(Instant::now()); } } @@ -459,11 +512,17 @@ pub(crate) async fn run_peer_worker(state: AppState) { // --- peer membership --------------------------------------------------------- +/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or +/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to +/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's +/// key), so neither side needs a HELLO of its own. `ready` is what the message said. async fn establish_peer( state: &AppState, peer: &str, swarm_id: String, -) -> Result<Vec<transport::SyncEnvelope>> { + from_hello: bool, + ready: bool, +) -> Result<()> { let expected_swarm = state.swarm_id().await; if let Some(expected) = &expected_swarm { @@ -479,7 +538,7 @@ async fn establish_peer( let peer_state_mutex = state.get_or_create_peer_state(peer).await; let mut peer_state = peer_state_mutex.lock().await; - if let PeerMembership::Member { swarm_id: existing } = &peer_state.membership { + if let Standing::Member { swarm_id: existing } = &peer_state.standing { if existing != &swarm_id { bail!( "Peer {peer} changed swarm identity from {} to {}", @@ -494,34 +553,30 @@ async fn establish_peer( state.set_expected_swarm_id(swarm_id.clone()).await?; } - if matches!(peer_state.membership, PeerMembership::Unknown) { - peer_state.membership = PeerMembership::Member { + if !matches!(peer_state.standing, Standing::Member { .. }) { + peer_state.standing = Standing::Member { swarm_id: swarm_id.clone(), }; } - // First reply completes the handshake: stop retries and sync from a clean slate. - // Duplicate replies (caused by our own retries) don't reset again. - if peer_state.hello_state != HelloState::Established { + // Stop our retries and sync from a clean slate. A HELLO always resets: the peer + // starts over with us. Duplicate replies (caused by our own retries) don't. + peer_state.reports_ready = Some(ready); + if from_hello || peer_state.hello_state != HelloState::Established { peer_state.hello_state = HelloState::Established; peer_state.reset_sync(); - info!(peer = %peer, "Handshake complete; sync state reset"); + info!(peer = %peer, from_hello, "Handshake complete; sync state reset"); } - let pending = std::mem::take(&mut peer_state.pending_inbound_sync); - - info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established"); - Ok(pending) + info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established"); + Ok(()) } #[instrument(skip(state), fields(peer = %peer))] pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> { let peer_state_mutex = state.get_peer_state(peer).await?; let peer_state = peer_state_mutex.lock().await; - match &peer_state.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { swarm_id } => Some(swarm_id.clone()), - } + peer_state.standing.swarm_id().map(str::to_owned) } // --- epochs ------------------------------------------------------------------ @@ -545,6 +600,8 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool // (resetting both sides again) while this fresh sync is under way. peer_state.reset_sync(); peer_state.last_sync_rx = Some(Instant::now()); + // Whether it would sync with us was said about its old epoch. + peer_state.syncs_with_us = None; } previous }; @@ -597,7 +654,12 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { let local = state.local_id(); let proof = match message { - ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)), + ParsedMessage::Hello(hello) => Some(key.verify_hello( + &hello.auth, + hello.swarm_id.as_deref().unwrap_or(""), + peer, + local, + )), ParsedMessage::HelloReply(reply) => { Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local)) } @@ -609,8 +671,8 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { remember_peer(state, peer).await; if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; - if !peer_state.authenticated { - peer_state.authenticated = true; + if peer_state.standing == Standing::Stranger { + peer_state.standing = Standing::Admitted; debug!(peer = %peer, "Peer proved the swarm key"); } } @@ -622,11 +684,10 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { // we stop syncing our document to it, not just listening to it. if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; - if peer_state.authenticated { + if peer_state.standing.is_admitted() { warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership"); } - peer_state.authenticated = false; - peer_state.membership = PeerMembership::Unknown; + peer_state.standing = Standing::Stranger; peer_state.hello_state = HelloState::NotSent; peer_state.reset_sync(); } @@ -643,7 +704,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { } None => { let admitted = match state.get_peer_state(peer).await { - Some(peer_state) => peer_state.lock().await.authenticated, + Some(peer_state) => peer_state.lock().await.standing.is_admitted(), None => false, }; if !admitted { @@ -666,19 +727,31 @@ pub(crate) async fn handle_message( return Ok(()); } + // A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while + // we were down) says nothing about now; the peer answers our current HELLO. + if let transport::ParsedMessage::HelloReply(reply) = &message { + if reply.run_id != state.run_id { + state.stats.stale_hello_reply(); + debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run"); + return Ok(()); + } + } + if !admit(state, &peer, &message).await { return Ok(()); } remember_peer(state, &peer).await; - // Any inbound message proves liveness. - let reactivated = state - .get_or_create_peer_state(&peer) - .await - .lock() - .await - .mark_seen(Instant::now()); + // Any inbound message proves liveness. When we heard from it before tells a + // restarted peer how long it was away (in our HELLO_REPLY). + let (reactivated, previously_seen) = { + let peer_state_mutex = state.get_or_create_peer_state(&peer).await; + let mut peer_state = peer_state_mutex.lock().await; + let previously_seen = peer_state.last_seen; + (peer_state.mark_seen(Instant::now()), previously_seen) + }; + crate::swarm::source::heard_from(state, &peer); if reactivated { info!(peer = %peer, "Dormant peer is active again; re-handshaking"); @@ -689,50 +762,38 @@ pub(crate) async fn handle_message( //debug!(from = %peer, message = ?message, "Received message"); match message { - transport::ParsedMessage::Hello(_) => { + transport::ParsedMessage::Hello(hello) => { // A HELLO means the peer is (re)starting its handshake with us, e.g. after a - // restart. Whatever our sync state believed about it is stale. - { - let peer_state_mutex = state.get_or_create_peer_state(&peer).await; - let mut peer_state = peer_state_mutex.lock().await; - peer_state.reset_sync(); - peer_state.reply_owed = true; + // restart. It becomes a member right away; our reply makes us one for it. + let Some(swarm_id) = state.swarm_id().await else { + // We are joining ourselves; the peer retries until we know our swarm. + debug!(peer = %peer, "HELLO before we know our swarm; not answering yet"); + return Ok(()); + }; + if let Some(theirs) = &hello.swarm_id { + if theirs != &swarm_id { + bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}"); + } } + establish_peer(state, &peer, swarm_id, true, hello.ready).await?; - // Sends now if we know our swarm ID; otherwise stays owed. - send_owed_hello_reply(state, &peer).await; - - // Our own HELLO goes out regardless of whether the reply could be sent. - if let Err(error) = send_hello_once(state, &peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); + let away = previously_seen.map(|seen| seen.elapsed()); + if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await { + // The peer retries its HELLO. + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed"); } - // Restart sync with the fresh state (no-op until the peer is a validated member). let _ = state .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); + .send(SyncEvent::PeerValidated(peer.clone())); } transport::ParsedMessage::HelloReply(reply) => { - let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?; + state.lock_rtt().hello_reply_received(&peer, Instant::now()); + crate::swarm::stability::note_away(state, reply.away_ms.map(Duration::from_millis)); + establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?; discover_peers(state, &peer, &reply.peers).await; - // Our swarm ID may have just become known: pay any replies we owe. - flush_owed_hello_replies(state).await; - - if !pending.is_empty() { - debug!( - to = %peer, - pending_messages = pending.len(), - "Replaying pending sync messages" - ); - for envelope in pending { - if let Err(error) = handle_sync(state, &peer, envelope).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Discarding invalid pending sync envelope"); - } - } - } - state .sync_events .send(SyncEvent::PeerValidated(peer.clone())) @@ -752,7 +813,8 @@ pub(crate) async fn handle_message( if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; if let Err(error) = state .transport @@ -763,10 +825,12 @@ pub(crate) async fn handle_message( } } transport::ParsedMessage::Pong(remote) => { + state.lock_rtt().pong_received(&peer, Instant::now()); if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; } } diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs new file mode 100644 index 0000000..454a46a --- /dev/null +++ b/src/swarm/relay.rs @@ -0,0 +1,121 @@ +//! Passing changes on in a partial mesh (`swarm::topology`). +//! +//! A node pushes its own changes to its neighbours, but nothing it receives goes any +//! further by itself. So after applying changes from one neighbour, a node *announces* +//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100 +//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the +//! announced heads as the peer's heads in its sync state, Automerge asks for what's +//! missing (`need`), and the announcer answers with exactly those changes. Each node +//! receives a change about once; the rest is announcements. Regular keepalives work the +//! same way, so any gap is noticed within a keepalive interval. +//! +//! In a full mesh the writer reaches everyone itself, so nothing is announced. + +use std::sync::atomic::Ordering; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::debug; + +use crate::state::{AppState, Standing}; +use crate::swarm::peer::send_ping; +use crate::swarm::sync::SyncEvent; +use crate::swarm::topology; +use crate::transport::{self, HeadsSummary}; + +/// A pull that brought nothing for this long no longer blocks pulling from another peer. +const PULL_TIMEOUT: Duration = Duration::from_secs(15); + +/// After changes from `from` were applied and our heads changed: tell our other +/// neighbours. +pub(crate) async fn announce(state: &AppState, from: &str) { + if !state.initial_sync_done.load(Ordering::Acquire) || topology::is_full_mesh(state) { + return; + } + let neighbours: Vec<String> = state + .peer_states + .read() + .await + .keys() + .filter(|peer| peer.as_str() != from && topology::is_neighbour(state, peer)) + .cloned() + .collect(); + for peer in neighbours { + if let Err(error) = send_ping(state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); + } + } +} + +/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time). +pub(crate) async fn pull_if_behind(state: &AppState, peer: &str, remote: &transport::Keepalive) { + if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) { + return; + } + let missing = match &remote.heads { + HeadsSummary::Heads(heads) => { + // AutoCommit's lookups take &mut self. + let mut replica = state.doc.write().await; + let missing = heads + .iter() + .any(|head| replica.doc.get_change_by_hash(head).is_none()); + if !missing { + return; + } + Some(heads.clone()) + } + // Can't tell which heads are missing; the digest differs from ours, though. + HeadsSummary::Digest(_) => None, + }; + + let now = Instant::now(); + { + let mut pulling = state + .pulling + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let busy = matches!( + &*pulling, + Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT + ); + if busy { + return; + } + *pulling = Some((peer.to_owned(), now)); + } + + let Some(peer_state) = state.get_peer_state(peer).await else { + return; + }; + { + let mut peer_state = peer_state.lock().await; + if !matches!(peer_state.standing, Standing::Member { .. }) { + return; + } + match missing { + Some(heads) => { + // Automerge asks for whatever these heads need that we don't have. + peer_state.sync.state.their_heads = Some(heads); + peer_state.sync.state.in_flight = false; + } + // A fresh session's opening carries our heads and bloom filter; the answer + // brings what we lack. + None => peer_state.reset_sync(), + } + } + debug!(peer = %peer, "Neighbour has changes we lack; pulling"); + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())); +} + +/// A sync message from `peer` was applied: a pull from it is answered. +pub(crate) fn pulled(state: &AppState, peer: &str) { + let mut pulling = state + .pulling + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if pulling.as_ref().is_some_and(|(p, _)| p == peer) { + *pulling = None; + } +} diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs new file mode 100644 index 0000000..ed6cc73 --- /dev/null +++ b/src/swarm/rtt.rs @@ -0,0 +1,179 @@ +//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and +//! HELLO → HELLO_REPLY. No probes of our own. +//! +//! A sample only counts when exactly one request was outstanding: with two, we can't tell +//! which one the answer belongs to (an answer to the older one would look too fast). +//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter. +//! +//! What we use is the *effective* round-trip time: the smoothed value, raised to the age +//! of the oldest request still unanswered (the round trip takes at least that long), so +//! a peer that stops answering drifts away by itself. Without a recent sample (nodes +//! move; an old value says little) it is unknown. Silence alone doesn't count: we don't +//! ask non-neighbours anything, so their quiet is ours. + +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::Instant; + +#[derive(Debug, Default)] +pub(crate) struct RttTable { + pings: HashMap<String, Outstanding>, + hellos: HashMap<String, Outstanding>, + smoothed: HashMap<String, Sample>, +} + +#[derive(Debug, Clone, Copy)] +struct Sample { + rtt: Duration, + at: Instant, +} + +#[derive(Debug, Clone, Copy)] +struct Outstanding { + sent: Instant, + count: u32, +} + +impl RttTable { + pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) { + Self::sent(&mut self.pings, peer, now); + } + + pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) { + if let Some(sample) = Self::answered(&mut self.pings, peer, now) { + self.record(peer, sample, now); + } + } + + pub(crate) fn hello_sent(&mut self, peer: &str, now: Instant) { + Self::sent(&mut self.hellos, peer, now); + } + + pub(crate) fn hello_reply_received(&mut self, peer: &str, now: Instant) { + if let Some(sample) = Self::answered(&mut self.hellos, peer, now) { + self.record(peer, sample, now); + } + } + + /// The effective round-trip time: the smoothed value if sampled within `fresh_for`, + /// raised to the age of the oldest unanswered request; None without a recent sample. + pub(crate) fn effective( + &self, + peer: &str, + now: Instant, + fresh_for: Duration, + ) -> Option<Duration> { + let sample = self + .smoothed + .get(peer) + .filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?; + let waiting = [self.pings.get(peer), self.hellos.get(peer)] + .into_iter() + .flatten() + .map(|outstanding| now.saturating_duration_since(outstanding.sent)) + .max() + .unwrap_or_default(); + Some(sample.rtt.max(waiting)) + } + + /// The effective round-trip time of every peer that has one. + pub(crate) fn all_effective( + &self, + now: Instant, + fresh_for: Duration, + ) -> impl Iterator<Item = (&str, Duration)> { + self.smoothed.keys().filter_map(move |peer| { + self.effective(peer, now, fresh_for) + .map(|rtt| (peer.as_str(), rtt)) + }) + } + + fn sent(map: &mut HashMap<String, Outstanding>, peer: &str, now: Instant) { + map.entry(peer.to_owned()) + .and_modify(|outstanding| outstanding.count += 1) + .or_insert(Outstanding { + sent: now, + count: 1, + }); + } + + fn answered( + map: &mut HashMap<String, Outstanding>, + peer: &str, + now: Instant, + ) -> Option<Duration> { + let outstanding = map.remove(peer)?; + (outstanding.count == 1).then(|| now.saturating_duration_since(outstanding.sent)) + } + + fn record(&mut self, peer: &str, sample: Duration, now: Instant) { + let rtt = match self.smoothed.get(peer) { + Some(old) => (old.rtt * 3 + sample) / 4, + None => sample, + }; + self.smoothed + .insert(peer.to_owned(), Sample { rtt, at: now }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const FRESH: Duration = Duration::from_secs(300); + + #[test] + fn samples_only_unambiguous_answers_and_smooths() { + let mut table = RttTable::default(); + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + table.ping_sent("a", start); + table.pong_received("a", at(100)); + assert_eq!( + table.effective("a", at(100), FRESH), + Some(Duration::from_millis(100)) + ); + + table.ping_sent("a", at(100)); + table.pong_received("a", at(300)); + assert_eq!( + table.effective("a", at(300), FRESH), + Some(Duration::from_millis(125)) + ); + + // Two PINGs out: the PONG can't be attributed. + table.ping_sent("a", at(300)); + table.ping_sent("a", at(400)); + table.pong_received("a", at(450)); + assert_eq!( + table.effective("a", at(450), FRESH), + Some(Duration::from_millis(125)) + ); + + // An unsolicited PONG changes nothing. + table.pong_received("b", start); + assert_eq!(table.effective("b", start, FRESH), None); + } + + #[test] + fn unanswered_requests_raise_it_and_old_samples_expire() { + let mut table = RttTable::default(); + let start = Instant::now(); + table.ping_sent("a", start); + table.pong_received("a", start + Duration::from_millis(50)); + + // A PING unanswered for 40 s: at least 40 s. + let sent = start + Duration::from_secs(60); + table.ping_sent("a", sent); + let later = sent + Duration::from_secs(40); + assert_eq!( + table.effective("a", later, FRESH), + Some(Duration::from_secs(40)) + ); + + // No sample within the freshness window: unknown. + let stale = start + FRESH + Duration::from_secs(1); + assert_eq!(table.effective("a", stale, FRESH), None); + } +} diff --git a/src/swarm/source.rs b/src/swarm/source.rs new file mode 100644 index 0000000..4025f0e --- /dev/null +++ b/src/swarm/source.rs @@ -0,0 +1,210 @@ +//! The catch-up source: while a node catches up (joining, or rejoining a newer epoch, +//! until its initial sync is done), it syncs with one peer only. +//! +//! Every peer would otherwise send us what we lack, all at once: a node with an empty +//! document gets the whole document from each of its peers, and a node that is merely +//! behind gets the missing changes from each. Automerge can't deduplicate across peers. +//! So one suitable peer becomes the source, preferably the closest by round-trip time +//! (a clearly closer one gets a 1 s head start to claim it, else the first to ask); +//! the others wait until the initial sync is done, and then only exchange heads. Messages from waiting +//! peers are still applied; we just don't answer them yet. +//! +//! A source must be ready and willing to sync with us, as far as we know (its sync +//! messages and keepalives say so). It counts as alive while any message from it +//! arrives, so a slow transfer of a large document never looks like a stall. When it +//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or +//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the +//! first time may be a handshake still under way), the next peer takes over. + +use std::sync::atomic::Ordering; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::{debug, info}; + +use crate::state::{AppState, CatchUpSource, PeerState}; +use crate::swarm::sync::SyncEvent; + +/// How long a clearly closer peer (by round-trip time) gets to claim the source role +/// before whoever asks next takes it. +const CLOSER_PEER_HEAD_START: Duration = Duration::from_secs(1); + +/// Probes without an answer before the source is given up. +const UNANSWERED_PROBES: u32 = 3; + +/// Keepalives in a row saying "I wouldn't sync with you" before the source is given up. +const REFUSALS: u32 = 2; + +fn lock_preference(state: &AppState) -> std::sync::MutexGuard<'_, Option<Instant>> { + state + .catch_up_preference + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn catching_up(state: &AppState) -> bool { + !state.initial_sync_done.load(Ordering::Acquire) +} + +fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Option<CatchUpSource>> { + state + .catch_up_source + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// Returns how long to hold back sync with `peer`, if at all: while catching up, with +/// everyone but the source. Picks the source if there is none (or it went quiet). +pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Option<Duration> { + if !catching_up(state) { + return None; + } + let stall = state.timing.catch_up_probe * UNANSWERED_PROBES; + if peer_state.reports_ready == Some(false) || peer_state.syncs_with_us == Some(false) { + // Not a candidate; its next message re-evaluates this, the timer is a fallback. + return Some(stall); + } + + let now = Instant::now(); + let mut source = lock(state); + let quiet = match &*source { + Some(current) if current.peer == peer => return None, + Some(current) if now < current.last_heard + stall => { + return Some(current.last_heard + stall - now); + } + Some(current) => Some(current.peer.clone()), + None => None, + }; + + // Prefer the closest peer: the source sends us everything we lack, so its distance + // matters most. If another peer we measured is clearly closer (by a fifth), it gets + // a short head start to claim the role; it is asked in the same sync pass. After + // that, whoever asks first takes it (the closer one may not be suitable after all). + let rtt = { + let rtts = state.lock_rtt(); + let fresh_for = state.timing.rtt_fresh_for(); + let mine = rtts.effective(peer, now, fresh_for); + let closer = rtts.all_effective(now, fresh_for).any(|(other, rtt)| { + other != peer + && quiet.as_deref() != Some(other) + && mine.is_none_or(|mine| rtt * 5 < mine * 4) + }); + if closer { + let mut preference = lock_preference(state); + let since = *preference.get_or_insert(now); + if now < since + CLOSER_PEER_HEAD_START { + return Some(since + CLOSER_PEER_HEAD_START - now); + } + } + mine + }; + *lock_preference(state) = None; + let rtt_ms = rtt.map(|rtt| rtt.as_millis() as u64); + + match quiet { + Some(previous) => info!( + peer = %peer, + previous = %previous, + rtt_ms, + quiet_secs = stall.as_secs_f32(), + "Catch-up source stopped answering; catching up from another peer" + ), + None => debug!(peer = %peer, rtt_ms, "Catching up from this peer first"), + } + state.stats.catch_up_source_chosen(); + *source = Some(CatchUpSource { + peer: peer.to_owned(), + last_heard: now, + last_probe: None, + refusals: 0, + }); + None +} + +/// Any message from `peer` arrived. +pub(crate) fn heard_from(state: &AppState, peer: &str) { + if let Some(current) = lock(state).as_mut().filter(|current| current.peer == peer) { + current.last_heard = Instant::now(); + } +} + +/// What `peer` said about itself, in a sync message (`syncs_with_us: None`) or a +/// keepalive. Gives up the source if that rules it out. +pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us: Option<bool>) { + if !catching_up(state) { + return; + } + let reason = { + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return; + }; + let reason = if !ready { + Some("it isn't ready itself") + } else { + match syncs_with_us { + Some(false) => { + current.refusals += 1; + (current.refusals >= REFUSALS).then_some("it wouldn't sync with us") + } + Some(true) => { + current.refusals = 0; + None + } + None => None, + } + }; + if reason.is_some() { + *source = None; + } + reason + }; + if let Some(reason) = reason { + debug!(peer = %peer, reason, "Giving up the catch-up source; trying another peer"); + resync_everyone(state); + } +} + +/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval. +pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool { + if !catching_up(state) { + return false; + } + let probe = state.timing.catch_up_probe; + let mut source = lock(state); + let Some(current) = source.as_mut().filter(|current| current.peer == peer) else { + return false; + }; + let due = now >= current.last_heard + probe + && current.last_probe.is_none_or(|last| now >= last + probe); + if due { + current.last_probe = Some(now); + } + due +} + +/// The initial sync is done: no source any more, and the peers we held back only need +/// to learn our heads now. +pub(crate) fn finished(state: &AppState, from: &str) { + *lock(state) = None; + *lock_preference(state) = None; + let state = state.clone(); + let from = from.to_owned(); + tokio::spawn(async move { + let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); + for other in peers.into_iter().filter(|other| *other != from) { + let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other)); + } + }); +} + +/// After giving up a source: let every peer be considered again. +fn resync_everyone(state: &AppState) { + let state = state.clone(); + tokio::spawn(async move { + let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); + for other in peers { + let _ = state.sync_events.send(SyncEvent::PeerNeedsSync(other)); + } + }); +} diff --git a/src/swarm/stability.rs b/src/swarm/stability.rs new file mode 100644 index 0000000..9d79b17 --- /dev/null +++ b/src/swarm/stability.rs @@ -0,0 +1,270 @@ +//! Whether this node is a stable member of the swarm, judged by itself. +//! +//! Long-running, well-connected nodes make a good backbone; laptops that sleep and +//! phones that lose signal don't. Each node measures what fraction of the last 24 hours +//! it had at least one live peer, and publishes only a coarse tier in its registry entry +//! (`_continuum/members/<pk>/tier`): `stable` or `transient`, written only when it +//! changes, with hysteresis. So the swarm learns about stability from the replicated +//! document alone, without anyone measuring anyone else, and the tier changes rarely. +//! +//! * transient → stable: at least 24 h observed, connected at least 95% of it; +//! * stable → transient: connected less than 80% (over at least an hour observed). +//! +//! After a restart a node has no history. If its entry says `stable` and a peer reports +//! (in its HELLO_REPLY) that it last heard from us less than an hour ago, the node was +//! only briefly away and keeps its tier, starting with a full day's credit. Otherwise it +//! starts over as transient. Time the process didn't run (a laptop asleep) counts as +//! disconnected. + +use std::collections::VecDeque; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::{info, warn}; + +use crate::state::{AppState, HelloState}; +use crate::swarm::members::{self, Tier}; + +const WINDOW: Duration = Duration::from_secs(24 * 3600); +const BUCKET: Duration = Duration::from_secs(5 * 60); +const BECOME_STABLE: f64 = 0.95; +const BECOME_TRANSIENT: f64 = 0.80; +/// A day observed (the oldest bucket may be partly outside the window). +const FULL_DAY: Duration = WINDOW.saturating_sub(BUCKET); +/// Less than this observed is too little to demote a stable node. +const MIN_TO_DEMOTE: Duration = Duration::from_secs(3600); +/// An absence shorter than this keeps a stable node stable across a restart. +const BRIEF_ABSENCE: Duration = Duration::from_secs(3600); +/// How long after becoming ready we wait for a peer to tell us how long we were away. +const AWAY_REPORT_GRACE: Duration = Duration::from_secs(30); +/// A gap between two measurements longer than this means we didn't run (asleep). +const MAX_TICK_GAP: Duration = Duration::from_secs(30); + +#[derive(Debug, Default)] +pub(crate) struct Stability { + buckets: VecDeque<Bucket>, + last_tick: Option<Instant>, + /// The shortest absence any peer reported to us in this run. + away: Option<Duration>, + ready_since: Option<Instant>, + /// The decision about the tier we found after a restart is made. + settled: bool, +} + +#[derive(Debug, Clone, Copy)] +struct Bucket { + start: Instant, + observed: Duration, + connected: Duration, +} + +impl Stability { + fn record(&mut self, now: Instant, connected: bool) { + let Some(last) = self.last_tick.replace(now) else { + return; + }; + let elapsed = now.saturating_duration_since(last).min(WINDOW); + let connected = connected && elapsed <= MAX_TICK_GAP; + if self + .buckets + .back() + .is_none_or(|bucket| now.saturating_duration_since(bucket.start) >= BUCKET) + { + self.buckets.push_back(Bucket { + start: now, + observed: Duration::ZERO, + connected: Duration::ZERO, + }); + } + let bucket = self.buckets.back_mut().expect("just ensured"); + bucket.observed += elapsed; + if connected { + bucket.connected += elapsed; + } + while self + .buckets + .front() + .is_some_and(|bucket| now.saturating_duration_since(bucket.start) > WINDOW) + { + self.buckets.pop_front(); + } + } + + /// A full day of connectivity, for a stable node that was only briefly away. + fn credit(&mut self, now: Instant) { + self.buckets.push_front(Bucket { + start: now, + observed: WINDOW, + connected: WINDOW, + }); + } + + fn observed(&self) -> Duration { + self.buckets.iter().map(|bucket| bucket.observed).sum() + } + + fn fraction(&self) -> f64 { + let observed = self.observed().as_secs_f64(); + if observed == 0.0 { + return 0.0; + } + let connected: f64 = self.buckets.iter().map(|b| b.connected.as_secs_f64()).sum(); + connected / observed + } + + /// The tier we should have, given the one we have now. + fn assess(&self, current: Tier) -> Tier { + match current { + Tier::Transient if self.observed() >= FULL_DAY && self.fraction() >= BECOME_STABLE => { + Tier::Stable + } + Tier::Stable + if self.observed() >= MIN_TO_DEMOTE && self.fraction() < BECOME_TRANSIENT => + { + Tier::Transient + } + tier => tier, + } + } +} + +fn lock(state: &AppState) -> std::sync::MutexGuard<'_, Stability> { + state + .stability + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// The fraction of the observed time (up to a day) we had a live peer, and how much +/// time was observed. +pub(crate) fn measured(state: &AppState) -> (f64, Duration) { + let stability = lock(state); + (stability.fraction(), stability.observed()) +} + +/// A HELLO_REPLY told us how long ago that peer last heard from us before this run. +pub(crate) fn note_away(state: &AppState, away: Option<Duration>) { + if let Some(away) = away { + let mut stability = lock(state); + stability.away = Some(stability.away.map_or(away, |known| known.min(away))); + } +} + +/// Measures, and updates our tier in the registry when it should change. Called every +/// few seconds by the peer worker. +pub(crate) async fn tick(state: &AppState) { + let now = Instant::now(); + let mut connected = false; + let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect(); + for peer_state in peers { + let peer_state = peer_state.lock().await; + if peer_state.hello_state == HelloState::Established + && peer_state.is_alive(&state.timing, now) + { + connected = true; + break; + } + } + lock(state).record(now, connected); + + if !state.is_ready().await { + return; + } + let local = state.local_id().to_owned(); + let current = members::tier(&state.doc.read().await.doc, &local); + + let desired = { + let mut stability = lock(state); + let ready_since = *stability.ready_since.get_or_insert(now); + if stability.settled { + stability.assess(current) + } else { + let brief = stability.away.is_some_and(|away| away < BRIEF_ABSENCE); + if current == Tier::Stable && !brief && now < ready_since + AWAY_REPORT_GRACE { + return; // a peer may still tell us we were only briefly away + } + stability.settled = true; + match current { + Tier::Stable if brief => { + stability.credit(now); + info!("Back after a brief absence; staying a stable member"); + Tier::Stable + } + // Restarted without a report that we were only briefly away. + Tier::Stable => Tier::Transient, + Tier::Transient => Tier::Transient, + } + } + }; + if desired == current { + return; + } + + let fraction = lock(state).fraction(); + match state + .mutate_doc(move |doc| members::set_tier(doc, &local, desired)) + .await + { + Ok(()) => info!( + tier = desired.as_str(), + connected_percent = (fraction * 100.0).round(), + "Stability tier changed" + ), + Err(error) => warn!(error = %error, "Failed to record our stability tier"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn run(stability: &mut Stability, start: Instant, hours: u64, connected: impl Fn(u64) -> bool) { + let step = Duration::from_secs(5); + let steps = hours * 3600 / 5; + for i in 0..=steps { + stability.record(start + step * i as u32, connected(i)); + } + } + + #[test] + fn becomes_stable_after_a_well_connected_day() { + let mut stability = Stability::default(); + let start = Instant::now(); + run(&mut stability, start, 12, |_| true); + assert_eq!(stability.assess(Tier::Transient), Tier::Transient); + run( + &mut stability, + start + Duration::from_secs(12 * 3600), + 13, + |_| true, + ); + assert_eq!(stability.assess(Tier::Transient), Tier::Stable); + } + + #[test] + fn hysteresis_between_the_thresholds() { + let mut stability = Stability::default(); + // Connected 9 of every 10 measurements: 90%. + run(&mut stability, Instant::now(), 25, |i| i % 10 != 0); + assert_eq!(stability.assess(Tier::Transient), Tier::Transient); + assert_eq!(stability.assess(Tier::Stable), Tier::Stable); + } + + #[test] + fn falls_back_to_transient_when_often_disconnected() { + let mut stability = Stability::default(); + // Connected half the time. + run(&mut stability, Instant::now(), 2, |i| i % 2 == 0); + assert_eq!(stability.assess(Tier::Stable), Tier::Transient); + } + + #[test] + fn time_asleep_counts_as_disconnected() { + let mut stability = Stability::default(); + let start = Instant::now(); + run(&mut stability, start, 1, |_| true); + // The process didn't run for two hours, then one more measurement. + stability.record(start + Duration::from_secs(3 * 3600), true); + assert!(stability.fraction() < 0.5); + } +} diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs new file mode 100644 index 0000000..786734a --- /dev/null +++ b/src/swarm/stats.rs @@ -0,0 +1,262 @@ +//! Node-wide figures for /admin/stats that no other part keeps. Recorded where things +//! happen anyway; the one exception is the document's saved size (see +//! `measure_saved_size`). + +use std::sync::{Mutex, MutexGuard}; +use std::time::Duration; + +use tokio::time::Instant; +use tracing::{debug, warn}; + +use crate::state::AppState; + +/// Saving takes time proportional to the document; we spend at most about 1% of it on +/// measuring, and measure at least this long apart. +const SAVED_SIZE_MIN_INTERVAL: Duration = Duration::from_secs(60); +const SAVED_SIZE_DUTY: u32 = 100; + +#[derive(Debug)] +pub(crate) struct NodeStats { + started: Instant, + inner: Mutex<Inner>, +} + +#[derive(Debug, Default)] +struct Inner { + caught_up: Option<CaughtUp>, + catch_up_sources: u32, + max_heads: usize, + stale_hello_replies: u64, + saved_size: Option<SavedSize>, + measuring: bool, + propagation: Propagation, +} + +/// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last +/// bucket takes the rest. +const PROPAGATION_BUCKETS_MS: [u64; 11] = [ + 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000, 60_000, 300_000, 3_600_000, +]; + +/// How long changes written elsewhere took to reach us: from the writer's commit (its +/// clock) to applying them here (ours), so clock skew between the two is part of it. +#[derive(Debug, Clone, Copy, Default)] +pub(crate) struct Propagation { + /// Per bucket of `PROPAGATION_BUCKETS_MS`, then the rest; not cumulative. + pub(crate) buckets: [u64; PROPAGATION_BUCKETS_MS.len() + 1], + pub(crate) count: u64, + pub(crate) sum_ms: u64, + /// Changes stamped later than our clock said on arrival (counted in the first + /// bucket): the writer's clock is ahead of ours. + pub(crate) ahead: u64, +} + +impl Propagation { + fn record(&mut self, delay_ms: i64) { + if delay_ms < 0 { + self.ahead += 1; + } + let delay_ms = delay_ms.max(0) as u64; + let bucket = PROPAGATION_BUCKETS_MS + .iter() + .position(|&bound| delay_ms <= bound) + .unwrap_or(PROPAGATION_BUCKETS_MS.len()); + self.buckets[bucket] += 1; + self.count += 1; + self.sum_ms += delay_ms; + } + + /// Each bucket's upper bound (None for the last) and count. + pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ { + PROPAGATION_BUCKETS_MS + .iter() + .copied() + .map(Some) + .chain([None]) + .zip(self.buckets.iter().copied()) + } +} + +/// The document's size as saved (compressed), when it was measured and how long that +/// took (copy and save). +#[derive(Debug, Clone, Copy)] +pub(crate) struct SavedSize { + pub(crate) bytes: u64, + pub(crate) at: Instant, + pub(crate) took: Duration, +} + +/// When the initial sync finished, and what it took. +#[derive(Debug, Clone, Copy)] +pub(crate) struct CaughtUp { + /// Since the process started (handshakes included). + pub(crate) after: Duration, + /// Sync payload received until then, from all peers. + pub(crate) sync_bytes: u64, +} + +/// A copy of the figures. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Snapshot { + pub(crate) uptime: Duration, + pub(crate) caught_up: Option<CaughtUp>, + pub(crate) catch_up_sources: u32, + pub(crate) max_heads: usize, + pub(crate) stale_hello_replies: u64, + pub(crate) propagation: Propagation, +} + +impl Default for NodeStats { + fn default() -> Self { + Self { + started: Instant::now(), + inner: Mutex::default(), + } + } +} + +impl NodeStats { + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub(crate) fn caught_up(&self, sync_bytes: u64) { + let after = self.started.elapsed(); + self.lock() + .caught_up + .get_or_insert(CaughtUp { after, sync_bytes }); + } + + /// A catch-up source was chosen (the first, or a replacement). + pub(crate) fn catch_up_source_chosen(&self) { + self.lock().catch_up_sources += 1; + } + + /// Our document has `count` heads (after applying a sync message). + pub(crate) fn heads_seen(&self, count: usize) { + let mut inner = self.lock(); + inner.max_heads = inner.max_heads.max(count); + } + + /// Changes from elsewhere were applied, `delays_ms` after they were committed. + pub(crate) fn propagated(&self, delays_ms: impl IntoIterator<Item = i64>) { + let mut inner = self.lock(); + for delay_ms in delays_ms { + inner.propagation.record(delay_ms); + } + } + + pub(crate) fn stale_hello_reply(&self) { + self.lock().stale_hello_replies += 1; + } + + /// The last measurement of the saved size, if any. + pub(crate) fn saved_size(&self) -> Option<SavedSize> { + self.lock().saved_size + } + + /// Whether a measurement is due; if so, it counts as started. + fn start_measuring(&self, now: Instant) -> bool { + let mut inner = self.lock(); + let due = !inner.measuring + && inner.saved_size.is_none_or(|last| { + now >= last.at + SAVED_SIZE_MIN_INTERVAL.max(last.took * SAVED_SIZE_DUTY) + }); + if due { + inner.measuring = true; + } + due + } + + /// The document was saved for another reason (an export): a measurement for free. + pub(crate) fn saved_size_measured(&self, bytes: u64, took: Duration) { + self.lock().saved_size = Some(SavedSize { + bytes, + at: Instant::now(), + took, + }); + } + + fn finish_measuring(&self, measured: Option<SavedSize>) { + let mut inner = self.lock(); + inner.measuring = false; + if let Some(measured) = measured { + inner.saved_size = Some(measured); + } + } + + pub(crate) fn snapshot(&self) -> Snapshot { + let inner = self.lock(); + Snapshot { + uptime: self.started.elapsed(), + caught_up: inner.caught_up, + catch_up_sources: inner.catch_up_sources, + max_heads: inner.max_heads, + stale_hello_replies: inner.stale_hello_replies, + propagation: inner.propagation, + } + } +} + +/// Measures the document's saved size in the background when due: the first time, then +/// at least a minute apart and spending no more than about 1% of the time on it (a +/// document that takes 2 s to copy and save is measured every 200 s). Called by the +/// peer worker every few seconds. +/// +/// The copy holds the read lock (writes wait for it); the save runs without any lock. +pub(crate) fn measure_saved_size(state: &AppState) { + if !state.stats.start_measuring(Instant::now()) { + return; + } + let state = state.clone(); + tokio::spawn(async move { + let start = Instant::now(); + let doc = std::sync::Arc::clone(&state.doc); + // Both steps take time proportional to the document: off the async threads. + let measured = match tokio::task::spawn_blocking(move || { + let mut copy = doc.blocking_read().doc.clone(); + copy.save().len() as u64 + }) + .await + { + Ok(bytes) => { + let took = start.elapsed(); + debug!( + bytes, + took_ms = took.as_millis() as u64, + "Measured the saved document size" + ); + Some(SavedSize { + bytes, + at: Instant::now(), + took, + }) + } + Err(error) => { + warn!(error = %error, "Measuring the saved document size failed"); + None + } + }; + state.stats.finish_measuring(measured); + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn propagation_buckets() { + let mut propagation = Propagation::default(); + for delay_ms in [-20, 100, 101, 7_200_000] { + propagation.record(delay_ms); + } + let buckets: Vec<_> = propagation.buckets().filter(|(_, n)| *n > 0).collect(); + assert_eq!(buckets, [(Some(100), 2), (Some(250), 1), (None, 1)]); + assert_eq!(propagation.ahead, 1); + assert_eq!(propagation.count, 4); + assert_eq!(propagation.sum_ms, 7_200_201); + } +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 15b4ced..3284015 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -2,30 +2,27 @@ //! messages, and the change publisher that feeds incremental streams. use std::collections::HashSet; -use std::sync::atomic::Ordering; use std::sync::Arc; -use std::time::Duration; +use std::sync::atomic::Ordering; -use anyhow::{anyhow, bail, Context, Result}; +use anyhow::{Context, Result, anyhow, bail}; use automerge::sync::SyncDoc; use tokio::sync::mpsc; use tokio::time::Instant; use tracing::{debug, info, trace, warn}; -use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound}; +use crate::state::{AppState, Liveness, Sending, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; +use crate::swarm::source; +use crate::swarm::throughput::SyncBytes; use crate::transport::{self, ParsedMessage}; -const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500); -const MAX_SYNC_RETRY: Duration = Duration::from_secs(30); - #[derive(Debug)] pub(crate) enum SyncEvent { PeerValidated(String), DocumentChanged, MessageReceived(String), PeerNeedsSync(String), - RetryDelivery(String), } /// Store paths touched by one batch of document changes. @@ -34,12 +31,14 @@ pub(crate) struct ChangeSet { pub(crate) paths: Vec<Vec<String>>, } -async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> { +async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { let Some(peer_state_mutex) = state.get_peer_state(peer).await else { trace!("Peer was forgotten; skipping synchronization"); return Ok(()); }; + let we_ready = state.is_ready().await; + let validated_swarm = { let peer_state = peer_state_mutex.lock().await; @@ -58,25 +57,41 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool return Ok(()); } - let PeerMembership::Member { swarm_id } = &peer_state.membership else { + let Standing::Member { swarm_id } = &peer_state.standing else { return Ok(()); }; - if !retry_delivery { - let we_open = state.local_id() < peer; - if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) { - trace!( - wait_ms = wait.as_millis(), - "Leaving the opening sync message to the peer" - ); - let tx = state.sync_events.clone(); - let peer = peer.to_owned(); - tokio::spawn(async move { - tokio::time::sleep(wait).await; - let _ = tx.send(SyncEvent::PeerNeedsSync(peer)); - }); - return Ok(()); - } + if let Some(wait) = source::defer(state, peer, &peer_state) { + trace!( + wait_ms = wait.as_millis(), + "Catching up from another peer first; not syncing with this one yet" + ); + // Done earlier when the initial sync completes (every peer is enqueued + // then); this only matters if the source goes quiet. + let tx = state.sync_events.clone(); + let peer = peer.to_owned(); + tokio::spawn(async move { + tokio::time::sleep(wait).await; + let _ = tx.send(SyncEvent::PeerNeedsSync(peer)); + }); + return Ok(()); + } + + if peer_state + .sync + .leave_opening(we_ready, peer_state.reports_ready) + { + // Its own opening (or anything else from it) brings us back here. + trace!("Leaving the opening sync message to the peer, which isn't ready"); + return Ok(()); + } + + if peer_state.sync.state.their_heads.is_none() + && !crate::swarm::topology::is_neighbour(state, peer) + { + // We only answer peers that aren't our neighbours. + trace!("Not opening sync with a peer that isn't our neighbour"); + return Ok(()); } swarm_id.clone() @@ -91,104 +106,113 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool bail!("Peer {peer} belongs to a different swarm"); } - let existing = { - let mut guard = peer_state_mutex.lock().await; - let peer_state = &mut *guard; - match &peer_state.pending_outbound_sync { - Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)), - Some(_) => { - peer_state.needs_generate = true; - return Ok(()); - } - None => None, + { + let mut peer_state = peer_state_mutex.lock().await; + if let Some(sending) = &mut peer_state.sync.sending { + sending.regenerate = true; + return Ok(()); } - }; - - let encoded = match existing { - Some((message, next_attempt)) => { - if Instant::now() < next_attempt { - return Ok(()); - } - message + if peer_state + .sync + .retry_at + .is_some_and(|at| Instant::now() < at) + { + // The retry timer enqueues the peer once it's due. + return Ok(()); } - None if retry_delivery => return Ok(()), - None => { - let doc = Arc::clone(&state.doc); - let psm = Arc::clone(&peer_state_mutex); - let transport = Arc::clone(&state.transport); - let peer_owned = peer.to_owned(); + } - let generated = tokio::task::spawn_blocking( - move || -> Result<Option<transport::EncodedMessage>> { - // Label the message with the epoch of the document it was generated - // from, read under the same lock. - let (epoch, message) = { - let mut replica = doc.blocking_write(); - let mut peer_state = psm.blocking_lock(); - peer_state.needs_generate = false; - let epoch = replica.epoch; - let sent_before = peer_state.sync_state.sent_hashes.clone(); - let message = { - let sync_doc = replica.doc.sync(); - sync_doc.generate_sync_message(&mut peer_state.sync_state) - }; - // A peer whose heads are ours already has everything; Automerge - // answers only when it has something to say, so it will stay - // silent. Waiting for an answer would only end in a pointless - // resend. - let in_step = peer_state.sync_state.their_heads.as_ref() - == Some(&replica.doc.get_heads()); - if in_step { - peer_state.sync_in_step(); - } else if message.is_some() { - let sent = peer_state - .sync_state - .sent_hashes - .difference(&sent_before) - .copied() - .collect(); - peer_state.expect_sync_reply(sent, Instant::now()); - } - (epoch, message) - }; - let Some(message) = message else { - return Ok(None); - }; - let envelope = transport::SyncEnvelope { - swarm_id, - epoch, - message, + let encoded = { + let doc = Arc::clone(&state.doc); + let psm = Arc::clone(&peer_state_mutex); + let transport = Arc::clone(&state.transport); + let peer_owned = peer.to_owned(); + let ready = state.is_ready().await; + // Before this message is counted. + let bytes_before = SyncBytes::of(&state.traffic, peer); + + let generated = + tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> { + // Label the message with the epoch of the document it was generated + // from, read under the same lock. + let (epoch, message) = { + let mut replica = doc.blocking_write(); + let mut peer_state = psm.blocking_lock(); + let epoch = replica.epoch; + let sent_before = peer_state.sync.state.sent_hashes.clone(); + let message = { + let sync_doc = replica.doc.sync(); + sync_doc.generate_sync_message(&mut peer_state.sync.state) }; - transport - .encode(&peer_owned, &ParsedMessage::Sync(envelope)) - .map(Some) - }, - ) + // A peer whose heads are ours already has everything; Automerge + // answers only when it has something to say, so it will stay + // silent. Waiting for an answer would only end in a pointless + // resend. + let in_step = peer_state.sync.state.their_heads.as_ref() + == Some(&replica.doc.get_heads()); + let sent: Vec<_> = peer_state + .sync + .state + .sent_hashes + .difference(&sent_before) + .copied() + .collect(); + if in_step { + peer_state.sync.reply_settled(); + } else if message.is_some() { + peer_state + .sync + .expect_reply(sent.clone(), Instant::now(), bytes_before); + } + if message.is_some() { + // Marked under the same lock as the generation, so no second + // message is generated while this one is handed over. + peer_state.sync.sending = Some(Sending { + sent, + regenerate: false, + }); + } + (epoch, message) + }; + let Some(message) = message else { + return Ok(None); + }; + let envelope = transport::SyncEnvelope { + swarm_id, + epoch, + message, + ready, + }; + transport + .encode(&peer_owned, &ParsedMessage::Sync(envelope)) + .map(Some) + }) .await - .context("Sync generation task panicked")??; - - let Some(encoded) = generated else { - return Ok(()); - }; + .context("Sync generation task panicked")?; + let generated = match generated { + Ok(generated) => generated, + Err(error) => { + // Not encodable, so nothing went out: take back what it would have carried. + let _ = peer_state_mutex + .lock() + .await + .sync + .send_failed(Instant::now()); + return Err(error); + } + }; - peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound { - message: encoded.clone(), - retry_delay: INITIAL_SYNC_RETRY, - next_attempt: Instant::now(), - }); + let Some(encoded) = generated else { + return Ok(()); + }; - encoded - } + encoded }; // No state locks are held across the send. match state.transport.send_encoded(&encoded).await { Ok(()) => { - let regenerate = { - let mut peer_state = peer_state_mutex.lock().await; - peer_state.pending_outbound_sync = None; - std::mem::take(&mut peer_state.needs_generate) - }; + let regenerate = peer_state_mutex.lock().await.sync.send_succeeded(); trace!(regenerate, "Outbound sync message accepted by Mycelium"); @@ -202,30 +226,27 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool Ok(()) } Err(error) => { - let delay = { - let mut peer_state = peer_state_mutex.lock().await; - - let Some(ref mut entry) = peer_state.pending_outbound_sync else { - debug!("Pending envelope discarded by a sync reset during delivery"); - return Ok(()); - }; - - entry.retry_delay = entry.retry_delay.saturating_mul(2).min(MAX_SYNC_RETRY); - entry.next_attempt = Instant::now() + entry.retry_delay; - entry.retry_delay + let Some(delay) = peer_state_mutex + .lock() + .await + .sync + .send_failed(Instant::now()) + else { + debug!("Sync message refused, but a reset already discarded it"); + return Ok(()); }; warn!( retry_ms = delay.as_millis(), error = %format!("{error:#}"), - "Sync delivery failed; retaining the same outbound envelope" + "Sync delivery failed; generating a fresh message later" ); let tx = state.sync_events.clone(); let peer = peer.to_owned(); tokio::spawn(async move { tokio::time::sleep(delay).await; - let _ = tx.send(SyncEvent::RetryDelivery(peer)); + let _ = tx.send(SyncEvent::PeerNeedsSync(peer)); }); Ok(()) @@ -239,38 +260,28 @@ pub(crate) async fn handle_sync( envelope: transport::SyncEnvelope, ) -> Result<()> { debug!(peer = %peer, "Processing sync envelope"); + // This message is counted already (`transport::traffic`). + let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer)); let peer_state_mutex = state.get_or_create_peer_state(peer).await; let needs_hello = { - let mut peer_state = peer_state_mutex.lock().await; + let peer_state = peer_state_mutex.lock().await; - match &peer_state.membership { - PeerMembership::Unknown => { - let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| { - pending.message == envelope.message - && pending.swarm_id == envelope.swarm_id - && pending.epoch == envelope.epoch - }); - if !duplicate { - peer_state.pending_inbound_sync.push(envelope.clone()); - info!( - peer = %peer, - pending_count = peer_state.pending_inbound_sync.len(), - "Queued sync envelope pending peer membership" - ); - } else { - trace!(peer = %peer, "Ignoring duplicate pending sync envelope"); - } + match &peer_state.standing { + Standing::Stranger | Standing::Admitted => { + // Not processed: our handshake reply resets the session anyway, and the + // sender's reply timeout resends what still matters. + debug!(peer = %peer, "Dropping sync message from a peer that isn't a member yet"); true } - PeerMembership::Member { swarm_id } if swarm_id != &envelope.swarm_id => { + Standing::Member { swarm_id } if swarm_id != &envelope.swarm_id => { bail!( "Sync swarm mismatch for peer {peer}: validated_swarm={swarm_id}, envelope_swarm={}", envelope.swarm_id ); } - PeerMembership::Member { .. } => false, + Standing::Member { .. } => false, } }; @@ -296,17 +307,26 @@ pub(crate) async fn handle_sync( } let epoch_label = envelope.epoch; + let from_ready = envelope.ready; let applied = { let doc = Arc::clone(&state.doc); let peer_state_mutex = Arc::clone(&peer_state_mutex); + let stats = Arc::clone(&state.stats); + let throughput = Arc::clone(&state.throughput); + let rtt = state + .lock_rtt() + .effective(peer, Instant::now(), state.timing.rtt_fresh_for()); let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire); + // While catching up, everything arrives late; that says nothing about propagation. + let measure_propagation = !check_caught_up; let epoch = envelope.epoch; let message = envelope.message; let peer = peer.to_owned(); - // Some(caught_up) when applied, None when the message belongs to another epoch. - tokio::task::spawn_blocking(move || -> Result<Option<bool>> { + // Some((caught_up, heads changed)) when applied, None when the message belongs + // to another epoch. + tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> { let mut replica = doc.blocking_write(); // Authoritative check, under the same lock as the apply: the document may @@ -316,28 +336,49 @@ pub(crate) async fn handle_sync( } let mut peer_state = peer_state_mutex.blocking_lock(); - + peer_state.reports_ready = Some(from_ready); let heads_before = replica.doc.get_heads(); - let their_heads_before = peer_state.sync_state.their_heads.clone(); replica .doc .sync() - .receive_sync_message(&mut peer_state.sync_state, message) + .receive_sync_message(&mut peer_state.sync.state, message) .context("Automerge synchronization failed")?; peer_state.last_sync_rx = Some(Instant::now()); - peer_state.sync_reply_received(); + if let Some(wait) = &peer_state.sync.reply { + throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .exchange(&peer, wait.started, arrived, rtt); + } + peer_state.sync.reply_settled(); + resend_if_needed(&peer, &mut peer_state.sync.state); let heads_after = replica.doc.get_heads(); - let unproductive = heads_before == heads_after - && peer_state.sync_state.their_heads == their_heads_before - && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after); - break_sync_stall(&peer, &mut peer_state, unproductive); + stats.heads_seen(heads_after.len()); + let changed = heads_after != heads_before; + if changed && measure_propagation { + let now_ms = crate::state::unix_millis() as i64; + // Time 0: not stamped (snapshots, older nodes). + let delays: Vec<i64> = replica + .doc + .get_changes(&heads_before) + .iter() + .map(|change| change.timestamp()) + .filter(|&time| time > 0) + .map(|time| now_ms - time) + .collect(); + stats.propagated(delays); + } - Ok(Some( + Ok(Some(( + // Only from a ready peer: a peer still catching up has empty heads, which + // every document "contains". check_caught_up + && from_ready && peer_state - .sync_state + .sync + .state .their_heads .as_ref() .is_some_and(|heads| { @@ -345,23 +386,35 @@ pub(crate) async fn handle_sync( .iter() .all(|h| replica.doc.get_change_by_hash(h).is_some()) }), - )) + changed, + ))) }) .await .context("Sync task panicked")?? }; - let Some(caught_up) = applied else { + let Some((caught_up, changed)) = applied else { debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); return Ok(()); }; + source::reported(state, peer, from_ready, None); + if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { info!(peer = %peer, "Initial sync complete"); + state + .stats + .caught_up(state.traffic.total().received.sync.bytes); + source::finished(state, peer); } state.notify_store_changed(); + crate::swarm::relay::pulled(state, peer); + if changed { + crate::swarm::relay::announce(state, peer).await; + } + state .sync_events .send(SyncEvent::MessageReceived(peer.to_owned())) @@ -370,75 +423,38 @@ pub(crate) async fn handle_sync( Ok(()) } -/// Exchanges in a row that changed nothing on either side before `sent_hashes` is -/// cleared, and before a full two-sided reset (re-handshake). -const STALL_RESEND_NEEDED: u32 = 2; -const STALL_CLEAR_SENT: u32 = 8; -const STALL_RESET: u32 = 32; - -/// Breaks sync livelocks. -/// /// Automerge never resends a change it believes the peer has received (`sent_hashes`). -/// If that change was lost on the way (a dropped or undeliverable message, an epoch -/// check, a reset of only one side's state), the two sides keep exchanging messages -/// with differing heads and no changes, as fast as the transport allows. The keepalive -/// heal never fires then, because sync traffic keeps flowing. -/// -/// * A change the peer keeps asking for is sent again. -/// * After a few unproductive exchanges, forget what we sent: the next message carries -/// everything the peer's bloom filter lacks. -/// * If that doesn't help either, reset both sides through a fresh handshake. -fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) { - if !unproductive { - peer_state.stalled_exchanges = 0; +/// If the message carrying it was lost while later ones arrive, the peer keeps asking for +/// it (`need`, from the heads we advertise) and we keep not sending it: a livelock the +/// reply timeout can't see (answers keep coming) and the keepalive heal can't either +/// (sync never goes idle). So when the peer needs a change we think it has, what we +/// think it has is wrong: forget all of it, and the next message carries everything its +/// bloom filter lacks. (A reordered message can make that send a change twice; cheap.) +fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) { + let Some(need) = &sync_state.their_need else { return; - } - peer_state.stalled_exchanges += 1; - - let sync_state = &mut peer_state.sync_state; - - // Not on the first unproductive exchange: when messages cross, the peer asks for a - // change that is merely still in flight, and resending it would only duplicate it. - if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED { - if let Some(need) = &sync_state.their_need { - let before = sync_state.sent_hashes.len(); - sync_state.sent_hashes.retain(|hash| !need.contains(hash)); - let resend = before - sync_state.sent_hashes.len(); - if resend > 0 { - debug!(peer = %peer, resend, "Peer still needs changes we had sent; sending them again"); - } - } - } - - match peer_state.stalled_exchanges { - STALL_CLEAR_SENT => { - warn!( - peer = %peer, - sent = sync_state.sent_hashes.len(), - their_heads = ?sync_state.their_heads, - shared_heads = ?sync_state.shared_heads, - "Sync is stalled; forgetting which changes were sent" - ); - sync_state.sent_hashes.clear(); - } - STALL_RESET => { - warn!(peer = %peer, "Sync is still stalled; resetting it on both sides"); - peer_state.stalled_exchanges = 0; - peer_state.request_resync(); - } - _ => {} + }; + if need + .iter() + .any(|hash| sync_state.sent_hashes.contains(hash)) + { + debug!( + peer = %peer, + sent = sync_state.sent_hashes.len(), + "Peer needs changes we had sent; forgetting what was sent" + ); + sync_state.sent_hashes.clear(); } } pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) { let mut dirty_peers = HashSet::<String>::new(); - let mut retry_peers = HashSet::<String>::new(); let mut document_dirty = false; info!("Starting synchronization worker"); loop { - if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty { + if dirty_peers.is_empty() && !document_dirty { match rx.recv().await { Some(event) => match event { SyncEvent::PeerValidated(peer) @@ -446,9 +462,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece | SyncEvent::PeerNeedsSync(peer) => { dirty_peers.insert(peer); } - SyncEvent::RetryDelivery(peer) => { - retry_peers.insert(peer); - } SyncEvent::DocumentChanged => { document_dirty = true; } @@ -468,9 +481,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece | SyncEvent::PeerNeedsSync(peer) => { dirty_peers.insert(peer); } - SyncEvent::RetryDelivery(peer) => { - retry_peers.insert(peer); - } SyncEvent::DocumentChanged => { document_dirty = true; } @@ -489,30 +499,21 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece peers_guard.keys().cloned().collect() }; + // Our own changes go to our neighbours; they pass them on. Others only get + // answers to what they send us. for peer in peer_keys { - if peer_swarm(&state, &peer).await.is_some() { + if peer_swarm(&state, &peer).await.is_some() + && crate::swarm::topology::is_neighbour(&state, &peer) + { dirty_peers.insert(peer); } } } - // Retry existing envelopes separately from generating new ones. - let retries: Vec<String> = retry_peers.drain().collect(); - - for peer in retries { - if let Err(error) = deliver_sync_to_peer(&state, &peer, true).await { - warn!( - peer = %peer, - error = %format!("{error:#}"), - "Sync delivery retry failed" - ); - } - } - let peers: Vec<String> = dirty_peers.drain().collect(); for peer in peers { - if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await { + if let Err(error) = deliver_sync_to_peer(&state, &peer).await { warn!( peer = %peer, error = %format!("{error:#}"), diff --git a/src/swarm/throughput.rs b/src/swarm/throughput.rs new file mode 100644 index 0000000..f1b9b14 --- /dev/null +++ b/src/swarm/throughput.rs @@ -0,0 +1,168 @@ +//! Sync throughput with peers, measured from sync exchanges we have anyway: from sending +//! a sync message until the peer's answer arrives, the sync bytes that went either way +//! with that peer. No transfers of our own. +//! +//! Only exchanges of at least `MIN_BYTES` count; smaller ones measure the round trip, +//! not the throughput. The round trip (`swarm::rtt`) is taken off the elapsed time, but +//! never more than half of it, so a stale round trip can't inflate a sample. What +//! remains also includes the time the peer took to apply and answer: this is the rate a +//! peer delivers at in practice, not the raw capacity of the path. +//! +//! Samples are smoothed (exponential moving average, weight 1/4). Large exchanges are +//! rare (catching up, big writes), so estimates are kept however old; they come with +//! their age. + +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::Instant; + +use crate::transport::traffic::Traffic; + +/// Smallest exchange that counts: some dozens of Mycelium chunks. +const MIN_BYTES: u64 = 32 * 1024; +const WEIGHT: f64 = 0.25; + +/// Sync bytes exchanged with a peer so far (from the traffic counters). +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct SyncBytes { + sent: u64, + received: u64, +} + +impl SyncBytes { + pub(crate) fn of(traffic: &Traffic, peer: &str) -> Self { + traffic + .peer(peer) + .map(|t| Self { + sent: t.sent.sync.bytes, + received: t.received.sync.bytes, + }) + .unwrap_or_default() + } + + /// Bytes either way since `earlier` (0 if the counters were dropped meanwhile). + fn since(&self, earlier: &SyncBytes) -> u64 { + self.sent.saturating_sub(earlier.sent) + self.received.saturating_sub(earlier.received) + } +} + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Estimate { + pub(crate) bytes_per_sec: f64, + pub(crate) samples: u32, + /// When the last sample was taken. + pub(crate) at: Instant, +} + +#[derive(Debug, Default)] +pub(crate) struct ThroughputTable { + peers: HashMap<String, Estimate>, +} + +impl ThroughputTable { + /// A sync exchange with `peer` finished: it started at `started` with the counters + /// at `before`, and its answer arrived at `arrived` with the counters at `after`. + pub(crate) fn exchange( + &mut self, + peer: &str, + (started, before): (Instant, SyncBytes), + (arrived, after): (Instant, SyncBytes), + rtt: Option<Duration>, + ) { + let bytes = after.since(&before); + if bytes < MIN_BYTES { + return; + } + let elapsed = arrived.saturating_duration_since(started); + let transfer = elapsed + .saturating_sub(rtt.unwrap_or_default()) + .max(elapsed / 2); + if transfer.is_zero() { + return; + } + let sample = bytes as f64 / transfer.as_secs_f64(); + let estimate = self.peers.entry(peer.to_owned()).or_insert(Estimate { + bytes_per_sec: sample, + samples: 0, + at: arrived, + }); + if estimate.samples > 0 { + estimate.bytes_per_sec += WEIGHT * (sample - estimate.bytes_per_sec); + } + estimate.samples += 1; + estimate.at = arrived; + } + + pub(crate) fn get(&self, peer: &str) -> Option<Estimate> { + self.peers.get(peer).copied() + } + + pub(crate) fn retain(&mut self, mut keep: impl FnMut(&str) -> bool) { + self.peers.retain(|peer, _| keep(peer)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn bytes(sent: u64, received: u64) -> SyncBytes { + SyncBytes { sent, received } + } + + #[test] + fn small_exchanges_dont_count() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + table.exchange( + "a", + (t0, bytes(0, 0)), + (t0 + Duration::from_millis(200), bytes(500, 10_000)), + None, + ); + assert!(table.get("a").is_none()); + } + + #[test] + fn round_trip_is_taken_off_but_at_most_half() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + // 1 MB in 2 s, of which 1 s round trip: 1 MB/s. + table.exchange( + "a", + (t0, bytes(100, 0)), + (t0 + Duration::from_secs(2), bytes(200, 999_900)), + Some(Duration::from_secs(1)), + ); + let a = table.get("a").unwrap(); + assert!((a.bytes_per_sec - 1_000_000.0).abs() < 1.0); + + // A round trip longer than the exchange takes off only half. + table.exchange( + "b", + (t0, bytes(0, 0)), + (t0 + Duration::from_secs(2), bytes(0, 1_000_000)), + Some(Duration::from_secs(5)), + ); + assert!((table.get("b").unwrap().bytes_per_sec - 1_000_000.0).abs() < 1.0); + } + + #[test] + fn smoothed() { + let mut table = ThroughputTable::default(); + let t0 = Instant::now(); + for received in [500_000u64, 1_000_000] { + table.exchange( + "a", + (t0, bytes(0, 0)), + (t0 + Duration::from_secs(1), bytes(0, received)), + None, + ); + } + // Samples of 500 kB/s, then 1 MB/s: 500 kB/s + (1 MB/s - 500 kB/s) / 4. + let a = table.get("a").unwrap(); + assert_eq!(a.samples, 2); + assert!((a.bytes_per_sec - 625_000.0).abs() < 1.0); + } +} diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs new file mode 100644 index 0000000..c5a02f1 --- /dev/null +++ b/src/swarm/topology.rs @@ -0,0 +1,479 @@ +//! Which members a node syncs with: its neighbours. +//! +//! In a full mesh every change reaches every node from every other node (Automerge +//! decides per pair, so nobody knows that a third peer already passed it on), so the +//! copies of one write grow with N². Above `k` members, each node keeps only a few +//! neighbours, and changes travel over several hops. +//! +//! Every node computes the graph from the member list and which members it can reach, +//! with no coordination: +//! * a ring ordered by key hash: each member's nearest reachable member in both +//! directions, which keeps the reachable members connected; +//! * plus, for each member, the `k - 2` reachable others with the lowest pair score (a +//! hash of both keys, so both ends agree on it), which keeps paths short. +//! +//! A link exists if either end chose it, so links are symmetric and a node has at least +//! `k` neighbours. With at most `k + 1` reachable members the graph is the full mesh: +//! small swarms behave exactly as before. A member joining, leaving or going offline +//! changes only links near it. + +use std::collections::{BTreeMap, BTreeSet}; + +use sha2::{Digest, Sha256}; +use tokio::time::Instant; +use tracing::info; + +use crate::state::AppState; +use crate::swarm::members; + +/// This node's current neighbours, and the member list they were computed from. +#[derive(Debug, Default)] +pub(crate) struct NeighbourView { + members: Vec<String>, + k: usize, + unreachable: BTreeSet<String>, + stable: BTreeSet<String>, + neighbours: BTreeSet<String>, + /// Every reachable member is a neighbour. + full_mesh: bool, +} + +/// Whether every reachable member is our neighbour (also while the view is unknown). +pub(crate) fn is_full_mesh(state: &AppState) -> bool { + state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .is_none_or(|view| view.full_mesh) +} + +/// The current view in numbers, once known. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Summary { + /// Registered members, us included. + pub(crate) members: usize, + /// Members other than us not currently judged unreachable. + pub(crate) reachable: usize, + pub(crate) stable: usize, + pub(crate) neighbours: usize, + pub(crate) stable_neighbours: usize, + pub(crate) full_mesh: bool, +} + +pub(crate) fn summary(state: &AppState) -> Option<Summary> { + let local = state.local_id(); + state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .as_ref() + .map(|view| Summary { + members: view.members.len(), + reachable: view + .members + .iter() + .filter(|m| m.as_str() != local && !view.unreachable.contains(*m)) + .count(), + stable: view.stable.len(), + neighbours: view.neighbours.len(), + stable_neighbours: view + .neighbours + .iter() + .filter(|m| view.stable.contains(*m)) + .count(), + full_mesh: view.full_mesh, + }) +} + +/// Whether we initiate contact with `peer`: HELLOs, keepalives, dormancy checks, sync +/// openings. Everyone is a neighbour until the view is known (while joining, or before +/// any member registered), so joining works as in a full mesh. We always *answer* +/// anyone who contacts us. +pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { + match &*state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + { + Some(view) => view.neighbours.contains(peer), + None => true, + } +} + +/// How the neighbours changed. +pub(crate) struct NeighbourChange { + pub(crate) added: Vec<String>, + pub(crate) removed: Vec<String>, +} + +/// Recomputes the neighbours once ready, if the member list, the reachable members or +/// `k` changed. +pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> { + if !state.is_ready().await { + return None; + } + let (current, stable) = { + let replica = state.doc.read().await; + let current = members::members(&replica.doc); + let stable: BTreeSet<String> = current + .iter() + .filter(|m| members::tier(&replica.doc, m) == members::Tier::Stable) + .cloned() + .collect(); + (current, stable) + }; + let k = state.neighbours_k; + let now = Instant::now(); + let mut unreachable = BTreeSet::new(); + for member in ¤t { + if let Some(peer_state) = state.get_peer_state(member).await { + if peer_state.lock().await.is_unreachable(&state.timing, now) { + unreachable.insert(member.clone()); + } + } + } + let previous = { + let view = state + .neighbours + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if view.as_ref().is_some_and(|view| { + view.members == current + && view.k == k + && view.unreachable == unreachable + && view.stable == stable + }) { + return None; + } + view.as_ref().map(|view| view.neighbours.clone()) + }; + + // Before the first view everyone known counted as a neighbour. + let previous = match previous { + Some(previous) => previous, + None => state.peer_states.read().await.keys().cloned().collect(), + }; + let computed = neighbours( + ¤t, + state.local_id(), + k, + &|m| unreachable.contains(m), + &|m| stable.contains(m), + ); + let change = NeighbourChange { + added: computed + .iter() + .filter(|m| !previous.contains(*m)) + .cloned() + .collect(), + removed: previous + .iter() + .filter(|m| !computed.contains(*m)) + .cloned() + .collect(), + }; + info!( + members = current.len(), + unreachable = unreachable.len(), + stable = stable.len(), + k, + neighbours = computed.len(), + added = change.added.len(), + removed = change.removed.len(), + "Neighbours computed from the member list" + ); + let reachable_others = current + .iter() + .filter(|m| m.as_str() != state.local_id() && !unreachable.contains(*m)) + .count(); + *state + .neighbours + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(NeighbourView { + full_mesh: computed.len() >= reachable_others, + members: current, + k, + unreachable, + stable, + neighbours: computed, + }); + Some(change) +} + +const RING_LABEL: &[u8] = b"continuum/topology-ring/1"; +const PAIR_LABEL: &[u8] = b"continuum/topology-pair/1"; + +/// The neighbours of `local` among `members` (which may or may not contain `local`), +/// skipping members for which `unreachable` holds. +pub(crate) fn neighbours( + members: &[String], + local: &str, + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, +) -> BTreeSet<String> { + let mut all: Vec<String> = members.to_vec(); + all.push(local.to_owned()); + graph(&all, k, &|m: &str| m != local && unreachable(m), stable) + .remove(local) + .unwrap_or_default() +} + +/// Every reachable member's neighbours. Costs about N² hashes (a few milliseconds for +/// hundreds of members). `stable` members (`swarm::stability`) are preferred for half of +/// each member's scored slots. +/// +/// Each reachable member walks the ring (by key hash) both ways to the first reachable +/// member, which gives a ring over the reachable members, and adds the `k - 2` +/// reachable members with the lowest pair score. A link exists if either end chose it. +/// With at most `k + 1` reachable members, the reachable members form a full mesh. +/// +/// Each node judges reachability by itself, so nodes may briefly disagree; that only +/// adds links (whoever chose a link initiates, the other answers), it never cuts one. +pub(crate) fn graph( + members: &[String], + k: usize, + unreachable: &dyn Fn(&str) -> bool, + stable: &dyn Fn(&str) -> bool, +) -> BTreeMap<String, BTreeSet<String>> { + let k = k.max(2); + let all: BTreeSet<&str> = members.iter().map(String::as_str).collect(); + let live: BTreeSet<&str> = all.iter().copied().filter(|m| !unreachable(m)).collect(); + let mut links: BTreeMap<String, BTreeSet<String>> = live + .iter() + .map(|m| ((*m).to_owned(), BTreeSet::new())) + .collect(); + + let mut link = |a: &str, b: &str| { + if a != b { + links.get_mut(a).expect("live member").insert(b.to_owned()); + links.get_mut(b).expect("live member").insert(a.to_owned()); + } + }; + + if live.len() <= k + 1 { + for a in &live { + for b in &live { + link(a, b); + } + } + return links; + } + + // The ring over all members, so its order doesn't change when one goes offline. + let mut ring: Vec<([u8; 32], &str)> = + all.iter().map(|m| (hash(RING_LABEL, &[*m]), *m)).collect(); + ring.sort_unstable(); + let n = ring.len(); + let position: BTreeMap<&str, usize> = + ring.iter().enumerate().map(|(i, (_, m))| (*m, i)).collect(); + let first_live = |from: usize, step: usize| { + (1..n) + .map(|d| ring[(from + d * step) % n].1) + .find(|m| live.contains(m)) + .expect("more than k + 1 live members") + }; + + for member in &live { + let at = position[member]; + let next = first_live(at, 1); + let prev = first_live(at, n - 1); + link(member, next); + link(member, prev); + let mut candidates: Vec<([u8; 32], &str)> = live + .iter() + .copied() + .filter(|other| other != member && *other != next && *other != prev) + .map(|other| (pair_score(member, other), other)) + .collect(); + candidates.sort_unstable(); + // Half the slots (rounded up) prefer stable members, so long-running nodes form + // a backbone; the rest go by pair score alone, keeping paths short and the load + // on stable members bounded. + let stable_slots = (k - 2).div_ceil(2); + let preferred: Vec<&str> = candidates + .iter() + .map(|(_, other)| *other) + .filter(|other| stable(other)) + .take(stable_slots) + .collect(); + let rest = candidates + .iter() + .map(|(_, other)| *other) + .filter(|other| !preferred.contains(other)) + .take(k - 2 - preferred.len()); + for partner in preferred.iter().copied().chain(rest) { + link(member, partner); + } + } + links +} + +/// Symmetric: the same for (a, b) and (b, a). +fn pair_score(a: &str, b: &str) -> [u8; 32] { + let (low, high) = if a <= b { (a, b) } else { (b, a) }; + hash(PAIR_LABEL, &[low, high]) +} + +fn hash(label: &[u8], fields: &[&str]) -> [u8; 32] { + let mut hasher = Sha256::new(); + for field in [label] + .into_iter() + .chain(fields.iter().map(|f| f.as_bytes())) + { + hasher.update((field.len() as u32).to_be_bytes()); + hasher.update(field); + } + hasher.finalize().into() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::VecDeque; + + fn members(n: usize) -> Vec<String> { + (0..n).map(|i| format!("{:064x}", i * 7919 + 13)).collect() + } + + fn all_reachable(_: &str) -> bool { + false + } + + fn none_stable(_: &str) -> bool { + false + } + + fn graph(members: &[String], k: usize) -> BTreeMap<String, BTreeSet<String>> { + super::graph(members, k, &all_reachable, &none_stable) + } + + fn connected(g: &BTreeMap<String, BTreeSet<String>>) -> bool { + let Some(start) = g.keys().next() else { + return true; + }; + let mut seen = BTreeSet::from([start.clone()]); + let mut queue = VecDeque::from([start.clone()]); + while let Some(m) = queue.pop_front() { + for next in &g[&m] { + if seen.insert(next.clone()) { + queue.push_back(next.clone()); + } + } + } + seen.len() == g.len() + } + + fn edges(graph: &BTreeMap<String, BTreeSet<String>>) -> BTreeSet<(String, String)> { + graph + .iter() + .flat_map(|(a, ns)| { + ns.iter().map(move |b| { + if a < b { + (a.clone(), b.clone()) + } else { + (b.clone(), a.clone()) + } + }) + }) + .collect() + } + + #[test] + fn small_swarms_are_a_full_mesh() { + let ms = members(13); + for m in &ms { + assert_eq!( + neighbours(&ms, m, 12, &all_reachable, &none_stable).len(), + 12 + ); + } + } + + #[test] + fn links_are_symmetric_and_degree_at_least_k() { + let ms = members(250); + let g = graph(&ms, 8); + for (a, ns) in &g { + assert!(ns.len() >= 8, "{a} has {} neighbours", ns.len()); + assert!(!ns.contains(a)); + for b in ns { + assert!(g[b].contains(a), "{a} -> {b} not symmetric"); + } + } + } + + #[test] + fn graph_is_connected() { + for (n, k) in [(250, 8), (250, 2), (60, 3)] { + assert!(connected(&graph(&members(n), k)), "n={n} k={k}"); + } + } + + #[test] + fn one_member_joining_changes_few_links() { + let ms = members(200); + let before = edges(&graph(&ms, 8)); + let mut more = ms.clone(); + more.push(format!("{:064x}", 999_999)); + let after = edges(&graph(&more, 8)); + let changed = before.symmetric_difference(&after).count(); + assert!(changed <= 4 * 8, "{changed} links changed"); + } + + #[test] + fn reachable_members_stay_connected_when_most_are_offline() { + let ms = members(100); + // Every tenth member is online. + let online: BTreeSet<String> = ms.iter().step_by(10).cloned().collect(); + let offline = |m: &str| !online.contains(m); + let g = super::graph(&ms, 12, &offline, &none_stable); + assert_eq!(g.len(), 10); + // Ten reachable members and k = 12: a full mesh among them. + assert!(g.values().all(|ns| ns.len() == 9)); + + let online: BTreeSet<String> = ms.iter().step_by(4).cloned().collect(); + let offline = |m: &str| !online.contains(m); + let g = super::graph(&ms, 8, &offline, &none_stable); + assert_eq!(g.len(), 25); + assert!(connected(&g)); + assert!( + g.values() + .all(|ns| ns.len() >= 8 && ns.iter().all(|n| online.contains(n))) + ); + } + + #[test] + fn stable_members_become_hubs_with_bounded_degree() { + let ms = members(250); + let stable: BTreeSet<String> = ms.iter().step_by(12).take(20).cloned().collect(); + let is_stable = |m: &str| stable.contains(m); + let g = super::graph(&ms, 12, &all_reachable, &is_stable); + assert!(connected(&g)); + + let degree = |pred: &dyn Fn(&String) -> bool| -> (usize, f64) { + let degrees: Vec<usize> = g + .iter() + .filter(|(m, _)| pred(m)) + .map(|(_, ns)| ns.len()) + .collect(); + let max = *degrees.iter().max().unwrap(); + ( + max, + degrees.iter().sum::<usize>() as f64 / degrees.len() as f64, + ) + }; + let (stable_max, stable_mean) = degree(&|m| stable.contains(m)); + let (_, transient_mean) = degree(&|m| !stable.contains(m)); + // 230 transient members put about 5 of their 10 scored slots on 20 stable ones. + assert!( + stable_mean > 3.0 * transient_mean, + "{stable_mean} vs {transient_mean}" + ); + assert!( + stable_max <= 100, + "a stable member has {stable_max} neighbours" + ); + } +} diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs index 84fdc59..b8e05cd 100644 --- a/src/transport/firewall.rs +++ b/src/transport/firewall.rs @@ -89,12 +89,12 @@ impl Transport for FirewallTransport { let firewall = Arc::clone(&self.firewall); self.inner - .start_consumer(Box::new(move |sender, message| { + .start_consumer(Box::new(move |sender, message, payload_len| { if firewall.blocks_inbound(&sender) { tracing::trace!(peer = %sender, "Firewall dropped inbound message"); return Box::pin(async { Ok(()) }); } - handler(sender, message) + handler(sender, message, payload_len) })) .await } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 69d9e70..b2f5c9d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1,6 +1,8 @@ use anyhow::Result; use async_trait::async_trait; +use automerge::ChangeHash; use automerge::sync::Message; +use sha2::{Digest, Sha256}; use std::boxed::Box; use std::future::Future; use std::pin::Pin; @@ -9,14 +11,86 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +pub mod traffic; -pub type HeadsDigest = [u8; 32]; +/// Random identifier of one run of a node (from start to exit). +pub type RunId = [u8; 16]; +/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on +/// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see +/// the size test in `mycelium`). There are rarely more heads than nodes that wrote +/// concurrently since the last merging write. +pub const MAX_LISTED_HEADS: usize = 32; + +/// A document's heads, compactly: the heads themselves while there are few, otherwise a +/// digest of them. Two summaries are equal exactly when the heads are, whatever form +/// each took (nodes with different limits still agree). +#[derive(Debug, Clone)] +pub enum HeadsSummary { + /// Sorted. + Heads(Vec<ChangeHash>), + /// SHA-256 over the sorted heads. + Digest([u8; 32]), +} + +impl HeadsSummary { + pub fn of(mut heads: Vec<ChangeHash>) -> Self { + heads.sort_unstable(); + if heads.len() <= MAX_LISTED_HEADS { + Self::Heads(heads) + } else { + Self::Digest(digest_of(&heads)) + } + } + + /// The digest of the heads, computed if they are listed. + pub fn digest(&self) -> [u8; 32] { + match self { + Self::Heads(heads) => digest_of(heads), + Self::Digest(digest) => *digest, + } + } + + /// The heads, if listed. + pub fn heads(&self) -> Option<&[ChangeHash]> { + match self { + Self::Heads(heads) => Some(heads), + Self::Digest(_) => None, + } + } +} + +impl PartialEq for HeadsSummary { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (Self::Heads(a), Self::Heads(b)) => a == b, + _ => self.digest() == other.digest(), + } + } +} + +impl Eq for HeadsSummary {} + +/// SHA-256 over sorted heads. +fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { + let mut hasher = Sha256::new(); + for head in sorted_heads { + hasher.update(head.0); + } + hasher.finalize().into() +} + +/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient. #[derive(Debug, Clone)] pub struct Keepalive { - pub heads_digest: HeadsDigest, - /// The digest is only comparable between nodes on the same epoch. + /// Only comparable between nodes on the same epoch. + pub heads: HeadsSummary, pub epoch: Epoch, + /// The sender is ready (same meaning as `SyncEnvelope::ready`). + pub ready: bool, + /// The sender would sync with the recipient: counts it as an active member on its + /// own epoch. + pub syncs_with_you: bool, } #[derive(Debug)] @@ -28,7 +102,27 @@ pub enum ParsedMessage { Pong(Keepalive), } +/// What a message is, without its contents. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MessageKind { + Hello, + HelloReply, + Sync, + Ping, + Pong, +} + impl ParsedMessage { + pub fn kind(&self) -> MessageKind { + match self { + ParsedMessage::Hello(_) => MessageKind::Hello, + ParsedMessage::HelloReply(_) => MessageKind::HelloReply, + ParsedMessage::Sync(_) => MessageKind::Sync, + ParsedMessage::Ping(_) => MessageKind::Ping, + ParsedMessage::Pong(_) => MessageKind::Pong, + } + } + /// Every message carries the sender's epoch. pub fn epoch(&self) -> Epoch { match self { @@ -42,9 +136,16 @@ impl ParsedMessage { #[derive(Debug, Clone)] pub struct HelloRequest { + /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in + /// another swarm doesn't answer. + pub swarm_id: Option<String>, pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready (same meaning as `SyncEnvelope::ready`). + pub ready: bool, + /// Random per run of the sender; the reply echoes it. + pub run_id: RunId, } #[derive(Debug, Clone)] @@ -54,6 +155,14 @@ pub struct HelloReply { pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready. + pub ready: bool, + /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the + /// recipient (delivered late, or queued while it was down) is ignored. + pub run_id: RunId, + /// How long before this HELLO the sender had last heard from the recipient, if ever + /// (in milliseconds): tells a restarted node how long it was away. + pub away_ms: Option<u64>, } #[derive(Debug, Clone)] @@ -62,6 +171,10 @@ pub struct SyncEnvelope { /// Epoch of the document the message was generated from. pub epoch: Epoch, pub message: Message, + /// The sender was ready when it generated the message: its document is the swarm's, + /// not one it is still filling (joining, rejoining). Only such a peer can tell a + /// catching-up node that it has everything; an empty document says nothing else. + pub ready: bool, } /// A message serialized for the wire. Cloning is cheap (shared buffer). @@ -69,9 +182,13 @@ pub struct SyncEnvelope { pub struct EncodedMessage { pub peer: String, pub body: bytes::Bytes, + pub kind: MessageKind, + /// Bytes the message takes on the wire (without the transport's own framing). + pub payload_len: usize, } -pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> +/// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`). +pub type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send + Sync + 'static; diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index a41d509..cc7f25f 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -1,6 +1,7 @@ -use anyhow::{anyhow, bail, Context, Result}; +use anyhow::{Context, Result, anyhow, bail}; use async_trait::async_trait; -use base64::engine::{general_purpose::STANDARD as BASE64, Engine}; +use automerge::ChangeHash; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use reqwest::StatusCode; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_bytes::{ByteBuf, Bytes}; @@ -18,8 +19,8 @@ use tokio::process::{Child, Command}; use url::Url; use crate::transport::EncodedMessage; -use crate::transport::Keepalive; use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; +use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS}; use crate::transport::{MessageHandler, ParsedMessage, Transport}; const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; @@ -253,15 +254,14 @@ impl SupervisedMycelium { let (key_file_handle, key_file_path, node_fd) = Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; - let (net_key_file_handle, net_key_file_path, net_fd) = - match private_network { - Some(net_cfg) => { - let (h, p, fd) = - Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; - (h, Some(p), fd) - } - None => (None, None, None), - }; + let (net_key_file_handle, net_key_file_path, net_fd) = match private_network { + Some(net_cfg) => { + let (h, p, fd) = + Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; + (h, Some(p), fd) + } + None => (None, None, None), + }; let api_addr = format!("127.0.0.1:{}", config.api_port); let rpc_addr = format!("127.0.0.1:{}", config.rpc_port); @@ -456,14 +456,60 @@ impl MyceliumTransport { .await .context("Querying local Mycelium identity")?; - Ok(Self { + let transport = Self { setup: tokio::sync::Mutex::new(setup), client, poll_client, messages_endpoint: rest_url.join("/api/v1/messages")?, pubkey_basepath: rest_url.join("/api/v1/pubkey/")?, local_id: admin.node_pubkey, - }) + }; + transport.discard_backlog().await; + Ok(transport) + } + + /// Drops what Mycelium queued for us before we started. A Mycelium that outlives us + /// (an external one, or across our restart) keeps accepting messages meant for us: + /// HELLO retries, keepalives, sync messages of sessions that ended with our previous + /// run. Processing them would only replay stale handshakes. Everything that still + /// matters is resent: peers retry their HELLOs, and sync starts over after the + /// handshake. Runs before we send anything, so no answer to us is dropped. + /// + /// Bounded in time, not count: the backlog is whatever piled up while we were away, + /// which a local loop drains in well under a second even for large swarms. Whatever + /// is still arriving after that is current traffic (peers that don't know we're + /// back), and it is processed normally, exactly as without this step. + async fn discard_backlog(&self) { + const DRAIN_FOR: Duration = Duration::from_secs(2); + let deadline = std::time::Instant::now() + DRAIN_FOR; + let mut discarded = 0u64; + while std::time::Instant::now() < deadline { + let response = self + .client + .get(self.messages_endpoint.clone()) + .query(&[("timeout", 0)]) + .query(&[("topic", encode_topic(TOPIC_PREFIX))]) + .send() + .await; + match response { + Ok(response) if response.status() == StatusCode::NO_CONTENT => break, + Ok(response) if response.status().is_success() => discarded += 1, + Ok(response) => { + tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox"); + break; + } + Err(error) => { + tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox"); + break; + } + } + } + if discarded > 0 { + tracing::info!( + discarded, + "Discarded messages Mycelium queued for us before we started" + ); + } } async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> { @@ -513,10 +559,12 @@ impl Transport for MyceliumTransport { } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { - let outbound = format_message(peer, message)?; + let (outbound, payload_len) = format_message(peer, message)?; Ok(EncodedMessage { peer: peer.to_owned(), body: serde_json::to_vec(&outbound)?.into(), + kind: message.kind(), + payload_len, }) } @@ -639,7 +687,8 @@ impl Transport for MyceliumTransport { //payload_bytes = raw_bytes.len(), "Received message" ); - if let Err(error) = handler(sender, parsed).await { + let payload_len = base64_decoded_len(&inbound_message.payload); + if let Err(error) = handler(sender, parsed, payload_len).await { tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); } } @@ -674,6 +723,12 @@ fn encode_payload(payload: &Vec<u8>) -> String { BASE64.encode(payload) } +/// Length of what padded base64 `encoded` decodes to. +fn base64_decoded_len(encoded: &str) -> usize { + let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count(); + (encoded.len() / 4 * 3).saturating_sub(padding) +} + fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> { let topic: &str = &decode_topic(topic_encoded)?; let bytes = BASE64.decode(payload)?; @@ -708,10 +763,11 @@ fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), Tr Ok((sender, parsed)) } +/// The message for Mycelium's API, and the length of its payload. fn format_message( recipient: &str, message: &ParsedMessage, -) -> Result<OutboundMessage, TransportError> { +) -> Result<(OutboundMessage, usize), TransportError> { let (topic, payload_bytes) = match message { ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?), ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?), @@ -719,13 +775,14 @@ fn format_message( ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?), ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?), }; - Ok(OutboundMessage { + let outbound = OutboundMessage { dst: Destination { pk: recipient.to_owned(), }, topic: encode_topic(topic), payload: encode_payload(&payload_bytes), - }) + }; + Ok((outbound, payload_bytes.len())) } // ===================================================================== @@ -765,39 +822,73 @@ fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { }) } +/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set. #[derive(Serialize, Deserialize)] struct KeepaliveShadow { - heads_digest: ByteBuf, - #[serde(default)] + heads: Vec<ByteBuf>, + digest: ByteBuf, epoch: EpochShadow, + ready: bool, + syncs_with_you: bool, } fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let (heads, digest) = match &msg.heads { + HeadsSummary::Heads(heads) => ( + heads + .iter() + .map(|head| ByteBuf::from(head.0.to_vec())) + .collect(), + ByteBuf::new(), + ), + HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())), + }; rmp_serde::to_vec(&KeepaliveShadow { - heads_digest: ByteBuf::from(msg.heads_digest.to_vec()), + heads, + digest, epoch: epoch_to_shadow(&msg.epoch), + ready: msg.ready, + syncs_with_you: msg.syncs_with_you, }) } fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?; - let heads_digest = shadow - .heads_digest - .as_slice() - .try_into() - .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?; + let heads = if shadow.digest.is_empty() { + if shadow.heads.len() > MAX_LISTED_HEADS { + return Err(TransportError::InvalidPayload(format!( + "more than {MAX_LISTED_HEADS} heads listed" + ))); + } + let heads = shadow + .heads + .iter() + .map(|head| ChangeHash::try_from(head.as_slice())) + .collect::<Result<Vec<_>, _>>() + .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?; + HeadsSummary::Heads(heads) + } else { + let digest = + shadow.digest.as_slice().try_into().map_err(|_| { + TransportError::InvalidPayload("heads digest must be 32 bytes".into()) + })?; + HeadsSummary::Digest(digest) + }; Ok(Keepalive { - heads_digest, + heads, epoch: epoch_from_shadow(shadow.epoch)?, + ready: shadow.ready, + syncs_with_you: shadow.syncs_with_you, }) } #[derive(Serialize, Deserialize)] pub struct HelloRequestShadow { - #[serde(default)] + swarm_id: Option<String>, epoch: EpochShadow, - #[serde(default)] auth: ByteBuf, + ready: bool, + run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -810,6 +901,9 @@ pub struct HelloReplyShadow<'a> { epoch: EpochShadow, #[serde(default)] auth: ByteBuf, + ready: bool, + run_id: ByteBuf, + away_ms: Option<u64>, } #[derive(Serialize, Deserialize)] @@ -819,6 +913,9 @@ pub struct SyncEnvelopeShadow<'a> { pub message: automerge::sync::Message, #[serde(default)] epoch: EpochShadow, + /// Absent from nodes that predate it: never counts as ready. + #[serde(default)] + ready: bool, } // ===================================================================== @@ -853,18 +950,31 @@ mod automerge_msg_codec { // --- HelloRequest --- pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { + swarm_id: msg.swarm_id.clone(), epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), + ready: msg.ready, + run_id: ByteBuf::from(msg.run_id.to_vec()), }) } pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; Ok(HelloRequest { + swarm_id: shadow.swarm_id, epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), + ready: shadow.ready, + run_id: run_id_from_shadow(&shadow.run_id)?, }) } +fn run_id_from_shadow(bytes: &ByteBuf) -> Result<crate::transport::RunId, TransportError> { + bytes + .as_slice() + .try_into() + .map_err(|_| TransportError::InvalidPayload("run id must be 16 bytes".into())) +} + // --- HelloReply --- pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> { let shadow = HelloReplyShadow { @@ -872,6 +982,9 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), + ready: msg.ready, + run_id: ByteBuf::from(msg.run_id.to_vec()), + away_ms: msg.away_ms, }; rmp_serde::to_vec(&shadow) } @@ -882,6 +995,9 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> peers: shadow.peers, epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), + ready: shadow.ready, + run_id: run_id_from_shadow(&shadow.run_id)?, + away_ms: shadow.away_ms, }) } @@ -891,6 +1007,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde:: swarm_id: Cow::Borrowed(&msg.swarm_id), message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules epoch: epoch_to_shadow(&msg.epoch), + ready: msg.ready, }; rmp_serde::to_vec(&shadow) } @@ -900,6 +1017,7 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr swarm_id: shadow.swarm_id.into_owned(), epoch: epoch_from_shadow(shadow.epoch)?, message: shadow.message, + ready: shadow.ready, }) } @@ -935,3 +1053,45 @@ pub enum TransportError { status: reqwest::StatusCode, }, } + +#[cfg(test)] +mod tests { + use super::*; + + /// Mycelium splits messages into chunks of this size. + const MYCELIUM_CHUNK: usize = 1300; + + #[test] + fn largest_keepalive_fits_in_one_chunk() { + let keepalive = Keepalive { + heads: HeadsSummary::Heads( + (0..MAX_LISTED_HEADS as u8) + .map(|n| ChangeHash([n; 32])) + .collect(), + ), + epoch: Epoch { + generation: u64::MAX, + id: [0xff; 32], + }, + ready: true, + syncs_with_you: true, + }; + let payload = keepalive_to_bytes(&keepalive).unwrap(); + // Mycelium's message data: a length byte, the topic, then our payload. + let size = 1 + TOPIC_PONG.len().max(TOPIC_PING.len()) + payload.len(); + assert!( + size + 50 <= MYCELIUM_CHUNK, + "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes" + ); + assert_eq!(bytes_to_keepalive(&payload).unwrap().heads, keepalive.heads); + } + + #[test] + fn listed_heads_equal_their_digest() { + let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect(); + let listed = HeadsSummary::of(heads.clone()); + let digest = HeadsSummary::Digest(listed.digest()); + assert_eq!(listed, digest); + assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec())); + } +} diff --git a/src/transport/mycelium.rs.orig b/src/transport/mycelium.rs.orig new file mode 100644 index 0000000..9d1fb90 --- /dev/null +++ b/src/transport/mycelium.rs.orig @@ -0,0 +1,975 @@ +use anyhow::{Context, Result, anyhow, bail}; +use async_trait::async_trait; +use automerge::ChangeHash; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; +use reqwest::StatusCode; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_bytes::{ByteBuf, Bytes}; +use std::borrow::Cow; +use std::boxed::Box; +use std::fs::File; +use std::io::Write; +use std::net::TcpListener; +use std::os::unix::io::FromRawFd; +use std::path::PathBuf; +use std::process::Stdio; +use std::time::Duration; +use thiserror::Error; +use tokio::process::{Child, Command}; +use url::Url; + +use crate::transport::EncodedMessage; +use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; +use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS}; +use crate::transport::{MessageHandler, ParsedMessage, Transport}; + +const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; + +const TOPIC_PREFIX: &str = "continuum.v1."; +const TOPIC_HELLO: &str = "continuum.v1.hello"; +const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply"; +const TOPIC_SYNC: &str = "continuum.v1.sync"; +const TOPIC_PING: &str = "continuum.v1.ping"; +const TOPIC_PONG: &str = "continuum.v1.pong"; + +const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25); + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct InboundMessage { + id: String, + src_pk: String, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct OutboundMessage { + dst: Destination, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct Destination { + pk: String, +} + +/// Represents where a 32-byte cryptographic key originates. +#[derive(Debug, Clone)] +pub enum KeySource { + Raw([u8; 32]), + Path(PathBuf), +} + +/// Atomic structure ensuring private network credentials are always handled together. +#[derive(Debug, Clone)] +pub struct PrivateNetworkConfig { + pub name: String, + pub key: KeySource, +} + +/// Comprehensive parameters required to spawn or target an instance. +#[derive(Debug, Clone)] +pub struct MyceliumSpawnConfig { + pub rest_url_opt: Option<Url>, + pub key_file_opt: Option<PathBuf>, + pub network_key_file: Option<PathBuf>, + pub network_name: String, + pub tcp_port: Option<u16>, + pub api_port: Option<u16>, + pub rpc_port: Option<u16>, + pub discovery_port: Option<u16>, + pub peers: Vec<String>, +} + +impl Default for MyceliumSpawnConfig { + fn default() -> Self { + Self { + rest_url_opt: None, + key_file_opt: None, + network_key_file: None, + network_name: "something".to_string(), + tcp_port: None, + api_port: None, + rpc_port: None, + discovery_port: None, + peers: vec!["tls://127.0.0.1:9651".to_string()], + } + } +} + +pub enum MyceliumRuntime { + Supervised(Box<SupervisedMycelium>), + External, +} + +pub struct MyceliumSetup { + pub runtime: MyceliumRuntime, + pub rest_url: Url, +} + +impl MyceliumSetup { + /// Evaluates the configuration payload and initializes the chosen execution strategy. + pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> { + // TODO warn for all ignored options if rest_url_opt is set + if let Some(url) = config.rest_url_opt { + tracing::info!( + "Configured to interface with an external standalone Mycelium instance at: {}", + url + ); + return Ok(Self { + runtime: MyceliumRuntime::External, + rest_url: url, + }); + } + + tracing::info!( + "Configured to initialize an embedded user-space Mycelium background worker" + ); + + let key = match config.key_file_opt { + Some(key_file) => KeySource::Path(key_file), + None => { + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes)?; + KeySource::Raw(bytes) + } + }; + + let private_net_config = config + .network_key_file + .map(|net_key_path| PrivateNetworkConfig { + name: config.network_name.clone(), + key: KeySource::Path(net_key_path), + }); + + let supervised_node = SupervisedMycelium::spawn( + key, + private_net_config, + config.tcp_port, + config.api_port, + config.rpc_port, + config.discovery_port, + &config.peers, + ) + .await + .context("Failed to securely bootstrap embedded Mycelium background worker")?; + + let resolved_url = Url::parse(&format!( + "http://127.0.0.1:{}", + supervised_node.config.api_port + ))?; + + Ok(Self { + runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)), + rest_url: resolved_url, + }) + } + + /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely. + pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> { + match &mut self.runtime { + MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, + MyceliumRuntime::External => { + std::future::pending::<std::io::Result<std::process::ExitStatus>>().await + } + } + } +} + +/// The part of Mycelium's `/api/v1/admin` answer we use. +#[derive(Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct MyceliumAdminInfo { + pub node_pubkey: String, +} + +#[derive(Debug, Clone)] +pub struct MyceliumNodeConfig { + pub tcp_port: u16, + pub api_port: u16, + pub rpc_port: u16, + pub discovery_port: Option<u16>, + pub private_network: Option<String>, +} + +pub struct SupervisedMycelium { + child: Child, + pub config: MyceliumNodeConfig, + _key_file_handle: Option<File>, + _net_key_file_handle: Option<File>, +} + +impl SupervisedMycelium { + async fn spawn( + node_key: KeySource, + private_network: Option<PrivateNetworkConfig>, + explicit_tcp: Option<u16>, + explicit_api: Option<u16>, + explicit_rpc: Option<u16>, + explicit_discovery_port: Option<u16>, + peers: &[String], + ) -> Result<Self> { + // Linux only: keys are handed over in memory (memfd, never on disk), and the + // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself. + if cfg!(not(target_os = "linux")) { + bail!( + "Supervising Mycelium is only supported on Linux; run myceliumd yourself \ + and pass --mycelium-rest-url" + ); + } + + let tcp_port = match explicit_tcp { + Some(p) => p, + None => Self::find_available_port().context("Failed to allocate TCP wire port")?, + }; + let api_port = match explicit_api { + Some(p) => p, + None => Self::find_available_port().context("Failed to allocate API port")?, + }; + let rpc_port = match explicit_rpc { + Some(p) => p, + None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?, + }; + let discovery_port = match explicit_discovery_port { + Some(0) => Some( + Self::find_available_port().context("Failed to allocate peer discovery port")?, + ), + p => p, + }; + + let config = MyceliumNodeConfig { + tcp_port, + api_port, + rpc_port, + discovery_port, + private_network: private_network.as_ref().map(|pn| pn.name.clone()), + }; + + let parent_pid = std::process::id(); + + let (key_file_handle, key_file_path, node_fd) = + Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; + + let (net_key_file_handle, net_key_file_path, net_fd) = match private_network { + Some(net_cfg) => { + let (h, p, fd) = + Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; + (h, Some(p), fd) + } + None => (None, None, None), + }; + + let api_addr = format!("127.0.0.1:{}", config.api_port); + let rpc_addr = format!("127.0.0.1:{}", config.rpc_port); + + let mut cmd = Command::new("mycelium"); + + let fds_to_inherit: Vec<i32> = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); + // The child must never outlive us. An orphaned Mycelium keeps running with our + // key; the next one we start then shares its address, and replies (including + // Mycelium's own message ACKs) go to whichever instance the routers picked, + // leaving our messages pending forever. Drop only runs on a clean shutdown, so + // let the kernel signal the child when we die, however we die. + let parent_pid = std::process::id() as libc::pid_t; + unsafe { + cmd.pre_exec(move || { + for &fd in &fds_to_inherit { + libc::fcntl(fd, libc::F_SETFD, 0); + } + // Own process group: a Ctrl-C in the terminal goes to continuum only, + // which then stops Mycelium itself, instead of both exiting at once (which + // shows up as "Mycelium runtime process exited"). + if libc::setpgid(0, 0) != 0 { + return Err(std::io::Error::last_os_error()); + } + #[cfg(target_os = "linux")] + { + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { + return Err(std::io::Error::last_os_error()); + } + // We may have died between fork and prctl. + if libc::getppid() != parent_pid { + libc::_exit(1); + } + } + Ok(()) + }); + } + + cmd.arg("--key-file") + .arg(&key_file_path) + .arg("--tcp-listen-port") + .arg(config.tcp_port.to_string()) + .arg("--disable-quic") + .arg("--api-addr") + .arg(&api_addr) + .arg("--jsonrpc-addr") + .arg(&rpc_addr) + .arg("--no-tun"); + + if let Some(ref discovery_port) = config.discovery_port { + cmd.arg("--peer-discovery-port") + .arg(discovery_port.to_string()); + } else { + cmd.arg("--disable-peer-discovery"); + } + + if let Some(ref net_name) = config.private_network { + cmd.arg("--network-name").arg(net_name); + } + if let Some(ref net_key_path) = net_key_file_path { + cmd.arg("--network-key-file").arg(net_key_path); + } + + if !peers.is_empty() { + cmd.arg("--peers").arg(peers.join(" ")); + } + + cmd.env("RUST_LOG", "info"); + + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + + let mut child = cmd + .spawn() + .context("Failed to execute mycelium process daemon")?; + + // 1. FIXED: Pass-through streaming for child process STDOUT + if let Some(mut stdout) = child.stdout.take() { + tokio::spawn(async move { + let mut host_stdout = tokio::io::stdout(); + let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await; + }); + } + + // 2. FIXED: Pass-through streaming for child process STDERR + if let Some(mut stderr) = child.stderr.take() { + tokio::spawn(async move { + let mut host_stderr = tokio::io::stderr(); + let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await; + }); + } + + let node = Self { + child, + config, + _key_file_handle: key_file_handle, + _net_key_file_handle: net_key_file_handle, + }; + + Ok(node) + } + + fn resolve_key_strategy( + desc_tag: &str, + source: KeySource, + parent_pid: u32, + ) -> Result<(Option<File>, PathBuf, Option<i32>)> { + match source { + KeySource::Raw(raw_bytes) => { + let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?; + let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd)); + Ok((Some(handle), pseudo_path, Some(fd))) + } + KeySource::Path(existing_path) => Ok((None, existing_path, None)), + } + } + + fn find_available_port() -> Result<u16> { + let listener = TcpListener::bind("127.0.0.1:0")?; + let addr = listener.local_addr()?; + Ok(addr.port()) + } + + fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> { + #[cfg(target_os = "linux")] + { + let memfd_name = std::ffi::CString::new(desc_tag)?; + let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) }; + if fd < 0 { + return Err(anyhow!( + "memfd_create failed natively inside Linux kernel for tag: {}", + desc_tag + )); + } + let mut file = unsafe { File::from_raw_fd(fd) }; + file.write_all(&raw_key)?; + file.flush()?; + Ok((file, fd)) + } + + // Unreachable: `spawn` refuses to run elsewhere. + #[cfg(not(target_os = "linux"))] + { + let _ = (desc_tag, raw_key); + bail!("In-memory key files need Linux") + } + } + + pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> { + self.child.wait().await + } +} + +impl Drop for SupervisedMycelium { + fn drop(&mut self) { + // The memfd key files go away with their handles. + let _ = self.child.start_kill(); + } +} + +fn is_pubkey(s: &str) -> bool { + s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) +} + +pub struct MyceliumTransport { + // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane + pub setup: tokio::sync::Mutex<MyceliumSetup>, + pub client: reqwest::Client, + pub poll_client: reqwest::Client, + pub messages_endpoint: Url, + pubkey_basepath: Url, + local_id: String, +} + +impl MyceliumTransport { + pub async fn connect(setup: MyceliumSetup) -> Result<Self> { + let rest_url = setup.rest_url.clone(); + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()?; + let poll_client = reqwest::Client::builder() + .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")?; + + Ok(Self { + setup: tokio::sync::Mutex::new(setup), + client, + poll_client, + messages_endpoint: rest_url.join("/api/v1/messages")?, + pubkey_basepath: rest_url.join("/api/v1/pubkey/")?, + local_id: admin.node_pubkey, + }) + } + + async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> { + #[derive(Deserialize)] + struct PublicKeyResponse { + #[serde(rename = "publicKey")] + public_key: String, + } + + let response: PublicKeyResponse = self + .client + .get(self.pubkey_basepath.join(&ip.to_string())?) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(response.public_key) + } +} + +#[async_trait] +impl Transport for MyceliumTransport { + fn local_id(&self) -> &str { + &self.local_id + } + + async fn resolve_seed(&self, hint: &str) -> Result<String> { + if is_pubkey(hint) { + return Ok(hint.to_owned()); + } + + if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() { + return self.lookup_pubkey(&ip).await; + } + + let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0))) + .await + .context("DNS lookup timed out")?? + .find_map(|addr| match addr.ip() { + std::net::IpAddr::V6(ip) => Some(ip), + _ => None, + }) + .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?; + + self.lookup_pubkey(&ip).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let outbound = format_message(peer, message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + body: serde_json::to_vec(&outbound)?.into(), + }) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + tracing::trace!(payload_bytes = message.body.len(), "Sending message"); + let response = self + .client + .post(self.messages_endpoint.clone()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(message.body.clone()) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + tracing::error!(status = %status, "Mycelium rejected outbound message"); + return Err(TransportError::MyceliumRejected { + peer: message.peer.clone(), + status, + } + .into()); + } + + // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing + // a message that never arrived. + if tracing::enabled!(tracing::Level::DEBUG) { + let Ok(body) = response + .bytes() + .await + .map_err(anyhow::Error::from) + .and_then(|bytes| Ok(serde_json::from_slice::<serde_json::Value>(&bytes)?)) + else { + tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); + return Ok(()); + }; + let Ok(outbound) = serde_json::from_slice::<OutboundMessage>(&message.body) else { + tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); + return Ok(()); + }; + let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { + tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); + return Ok(()); + }; + let Some(id) = body.get("id").and_then(|id| id.as_str()) else { + tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); + return Ok(()); + }; + tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); + } + Ok(()) + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + tracing::info!( + endpoint = self.messages_endpoint.as_str(), + "Starting Mycelium message consumer" + ); + + loop { + tracing::trace!("Polling Mycelium inbox"); + + let response = match self + .poll_client + .get(self.messages_endpoint.clone()) + .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)]) + .query(&[("topic", encode_topic(TOPIC_PREFIX))]) + .send() + .await + { + Ok(resp) => resp, + Err(error) => { + tracing::warn!(error = %error, "Mycelium inbox poll failed"); + tokio::time::sleep(Duration::from_millis(250)).await; + continue; + } + }; + + let status = response.status(); + + if status == StatusCode::NO_CONTENT { + tracing::trace!("Mycelium inbox is empty"); + tokio::time::sleep(EMPTY_QUEUE_DELAY).await; + continue; + } + + if !status.is_success() { + tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status"); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + + let raw_bytes = match response.bytes().await { + Ok(bytes) => bytes, + Err(error) => { + tracing::warn!(error = %error, "Failed to read raw Mycelium response stream"); + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + }; + + let inbound_message = match serde_json::from_slice::<InboundMessage>(&raw_bytes) { + Ok(msg) => msg, + Err(error) => { + tracing::error!( + error = %error, + raw_payload = %String::from_utf8_lossy(&raw_bytes), + "Failed to decode inbound Mycelium message structure" + ); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + }; + + match parse_message(&inbound_message) { + Ok((sender, parsed)) => { + tracing::debug!( + from = %inbound_message.src_pk, + message = ?parsed, + id = %inbound_message.id, + //payload_bytes = raw_bytes.len(), + "Received message" + ); + if let Err(error) = handler(sender, parsed).await { + tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Err(transport_error) => { + tracing::error!( + error = %transport_error, + topic = %inbound_message.topic, + "Dropping poisoned packet: parsing failed gracefully" + ); + // No sleep required: we successfully popped a malformed packet and want to process the next one immediately + } + } + } + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + let mut guard = self.setup.lock().await; + guard.wait_for_exit().await + } +} + +fn encode_topic(topic: &str) -> String { + BASE64.encode(topic.as_bytes()) +} + +fn decode_topic(topic: &str) -> Result<String, TransportError> { + let bytes = BASE64.decode(topic)?; + Ok(String::from_utf8(bytes)?) +} + +fn encode_payload(payload: &Vec<u8>) -> String { + BASE64.encode(payload) +} + +fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> { + let topic: &str = &decode_topic(topic_encoded)?; + let bytes = BASE64.decode(payload)?; + match topic { + TOPIC_HELLO => { + let msg = bytes_to_hello_request(&bytes)?; + Ok(ParsedMessage::Hello(msg)) + } + TOPIC_HELLO_REPLY => { + let msg = bytes_to_hello_reply(&bytes)?; + Ok(ParsedMessage::HelloReply(msg)) + } + TOPIC_SYNC => { + let msg = bytes_to_sync_envelope(&bytes)?; + Ok(ParsedMessage::Sync(msg)) + } + TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)), + TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)), + t => Err(TransportError::UnknownTopic(t.to_string())), + } +} + +fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> { + let sender = message.src_pk.clone(); + + if sender.is_empty() { + return Err(TransportError::NoSender); + } + + let parsed = decode_payload(&message.topic, &message.payload)?; + + Ok((sender, parsed)) +} + +fn format_message( + recipient: &str, + message: &ParsedMessage, +) -> Result<OutboundMessage, TransportError> { + let (topic, payload_bytes) = match message { + ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?), + ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?), + ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?), + ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?), + ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?), + }; + Ok(OutboundMessage { + dst: Destination { + pk: recipient.to_owned(), + }, + topic: encode_topic(topic), + payload: encode_payload(&payload_bytes), + }) +} + +// ===================================================================== +// 1. WIRE SHADOWS +// ===================================================================== +// +// rmp_serde encodes structs as arrays. Fields added later are appended at the end and +// marked #[serde(default)], so a message from a node that predates them still decodes +// (an absent epoch is the genesis epoch). + +/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch. +#[derive(Serialize, Deserialize, Default)] +struct EpochShadow { + generation: u64, + id: ByteBuf, +} + +fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { + EpochShadow { + generation: epoch.generation, + id: ByteBuf::from(epoch.id.to_vec()), + } +} + +fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> { + if shadow.generation == 0 && shadow.id.is_empty() { + return Ok(Epoch::GENESIS); + } + let id = shadow + .id + .as_slice() + .try_into() + .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?; + Ok(Epoch { + generation: shadow.generation, + id, + }) +} + +/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set. +#[derive(Serialize, Deserialize)] +struct KeepaliveShadow { + heads: Vec<ByteBuf>, + digest: ByteBuf, + epoch: EpochShadow, + ready: bool, + syncs_with_you: bool, +} + +fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let (heads, digest) = match &msg.heads { + HeadsSummary::Heads(heads) => ( + heads + .iter() + .map(|head| ByteBuf::from(head.0.to_vec())) + .collect(), + ByteBuf::new(), + ), + HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())), + }; + rmp_serde::to_vec(&KeepaliveShadow { + heads, + digest, + epoch: epoch_to_shadow(&msg.epoch), + ready: msg.ready, + syncs_with_you: msg.syncs_with_you, + }) +} + +fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { + let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?; + let heads = if shadow.digest.is_empty() { + if shadow.heads.len() > MAX_LISTED_HEADS { + return Err(TransportError::InvalidPayload(format!( + "more than {MAX_LISTED_HEADS} heads listed" + ))); + } + let heads = shadow + .heads + .iter() + .map(|head| ChangeHash::try_from(head.as_slice())) + .collect::<Result<Vec<_>, _>>() + .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?; + HeadsSummary::Heads(heads) + } else { + let digest = + shadow.digest.as_slice().try_into().map_err(|_| { + TransportError::InvalidPayload("heads digest must be 32 bytes".into()) + })?; + HeadsSummary::Digest(digest) + }; + Ok(Keepalive { + heads, + epoch: epoch_from_shadow(shadow.epoch)?, + ready: shadow.ready, + syncs_with_you: shadow.syncs_with_you, + }) +} + +#[derive(Serialize, Deserialize)] +pub struct HelloRequestShadow { + #[serde(default)] + epoch: EpochShadow, + #[serde(default)] + auth: ByteBuf, +} + +#[derive(Serialize, Deserialize)] +pub struct HelloReplyShadow<'a> { + // Cow allows zero-copy references when saving, but creates an owned String when loading + pub swarm_id: Cow<'a, str>, + #[serde(default)] + pub peers: Vec<String>, + #[serde(default)] + epoch: EpochShadow, + #[serde(default)] + auth: ByteBuf, +} + +#[derive(Serialize, Deserialize)] +pub struct SyncEnvelopeShadow<'a> { + pub swarm_id: Cow<'a, str>, + #[serde(with = "automerge_msg_codec")] + pub message: automerge::sync::Message, + #[serde(default)] + epoch: EpochShadow, + /// Absent from nodes that predate it: never counts as ready. + #[serde(default)] + ready: bool, +} + +// ===================================================================== +// 2. CUSTOM CODEC +// ===================================================================== +mod automerge_msg_codec { + use super::*; + use automerge::sync::Message; + + pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error> + where + S: Serializer, + { + let bytes = msg.clone().encode(); + Bytes::new(&bytes).serialize(serializer) + } + + pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error> + where + D: Deserializer<'de>, + { + let byte_buf = ByteBuf::deserialize(deserializer)?; + Message::decode(&byte_buf) + .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e))) + } +} + +// ===================================================================== +// 3. CLEAN CONVERSION INTERFACE +// ===================================================================== + +// --- HelloRequest --- +pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { + rmp_serde::to_vec(&HelloRequestShadow { + epoch: epoch_to_shadow(&msg.epoch), + auth: ByteBuf::from(msg.auth.clone()), + }) +} +pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { + let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; + Ok(HelloRequest { + epoch: epoch_from_shadow(shadow.epoch)?, + auth: shadow.auth.into_vec(), + }) +} + +// --- HelloReply --- +pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let shadow = HelloReplyShadow { + swarm_id: Cow::Borrowed(&msg.swarm_id), + peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching + epoch: epoch_to_shadow(&msg.epoch), + auth: ByteBuf::from(msg.auth.clone()), + }; + rmp_serde::to_vec(&shadow) +} +pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> { + let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?; + Ok(HelloReply { + swarm_id: shadow.swarm_id.into_owned(), + peers: shadow.peers, + epoch: epoch_from_shadow(shadow.epoch)?, + auth: shadow.auth.into_vec(), + }) +} + +// --- SyncEnvelope --- +pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> { + let shadow = SyncEnvelopeShadow { + swarm_id: Cow::Borrowed(&msg.swarm_id), + message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules + epoch: epoch_to_shadow(&msg.epoch), + ready: msg.ready, + }; + rmp_serde::to_vec(&shadow) +} +pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> { + let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; + Ok(SyncEnvelope { + swarm_id: shadow.swarm_id.into_owned(), + epoch: epoch_from_shadow(shadow.epoch)?, + message: shadow.message, + ready: shadow.ready, + }) +} + +#[derive(Error, Debug)] +pub enum TransportError { + #[error("Mycelium delivered a message without a sender public key")] + NoSender, + + #[error("Base64 decoding failed: {0}")] + Base64Decode(#[from] base64::DecodeError), + + #[error("UTF-8 decoding failed: {0}")] + Utf8Decode(#[from] std::string::FromUtf8Error), + + #[error("MessagePack serialization failed: {0}")] + MessagePackEncode(#[from] rmp_serde::encode::Error), + + #[error("MessagePack deserialization failed: {0}")] + MessagePackDecode(#[from] rmp_serde::decode::Error), + + #[error("Unknown topic: {0}")] + UnknownTopic(String), + + #[error("Invalid payload: {0}")] + InvalidPayload(String), + + #[error("Network HTTP request failed: {0}")] + Http(#[from] reqwest::Error), + + #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")] + MyceliumRejected { + peer: String, + status: reqwest::StatusCode, + }, +} diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs new file mode 100644 index 0000000..bb93d48 --- /dev/null +++ b/src/transport/traffic.rs @@ -0,0 +1,199 @@ +//! Messages and bytes exchanged with each peer, by kind: counted where every message +//! passes anyway, at the transport. No messages of our own. +//! +//! Bytes are payload bytes, what the Mycelium message carries, without its framing, +//! the topic, or the base64 of its HTTP API. Counters only grow (from the start of the +//! process); rates are for whoever reads them. +//! +//! A message dropped by the debug firewall still counts as sent (the firewall plays a +//! lossy network) but not as received. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; + +use anyhow::Result; +use async_trait::async_trait; +use serde::Serialize; +use utoipa::ToSchema; + +use crate::transport::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport}; + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +pub struct Counter { + pub messages: u64, + pub bytes: u64, +} + +impl Counter { + fn add(&mut self, bytes: usize) { + self.messages += 1; + self.bytes += bytes as u64; + } +} + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ByKind { + pub hello: Counter, + pub hello_reply: Counter, + pub sync: Counter, + pub ping: Counter, + pub pong: Counter, +} + +impl ByKind { + fn get_mut(&mut self, kind: MessageKind) -> &mut Counter { + match kind { + MessageKind::Hello => &mut self.hello, + MessageKind::HelloReply => &mut self.hello_reply, + MessageKind::Sync => &mut self.sync, + MessageKind::Ping => &mut self.ping, + MessageKind::Pong => &mut self.pong, + } + } +} + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct PeerTraffic { + /// Handed to Mycelium (it accepted them). + pub sent: ByKind, + pub received: ByKind, + /// Sends Mycelium refused (e.g. no route to the peer); not counted as sent. + pub refused: ByKind, +} + +#[derive(Debug, Default)] +struct Table { + peers: HashMap<String, PeerTraffic>, + /// All peers, including ones no longer listed. + total: PeerTraffic, +} + +#[derive(Debug, Default)] +pub struct Traffic { + table: Mutex<Table>, +} + +impl Traffic { + fn lock(&self) -> MutexGuard<'_, Table> { + self.table + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn count(&self, peer: &str, bytes: usize, pick: impl Fn(&mut PeerTraffic) -> &mut Counter) { + let mut guard = self.lock(); + let table = &mut *guard; + pick(&mut table.total).add(bytes); + match table.peers.get_mut(peer) { + Some(traffic) => pick(traffic).add(bytes), + None => pick(table.peers.entry(peer.to_owned()).or_default()).add(bytes), + } + } + + fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.sent.get_mut(kind)); + } + + fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.refused.get_mut(kind)); + } + + fn received(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.received.get_mut(kind)); + } + + pub fn peer(&self, peer: &str) -> Option<PeerTraffic> { + self.lock().peers.get(peer).copied() + } + + pub fn total(&self) -> PeerTraffic { + self.lock().total + } + + /// Drops the counters of peers not kept (anyone can send us messages; the totals + /// keep their share). + pub fn retain(&self, mut keep: impl FnMut(&str) -> bool) { + self.lock().peers.retain(|peer, _| keep(peer)); + } +} + +/// Counts what passes through `inner`. +pub struct CountingTransport { + inner: Arc<dyn Transport>, + traffic: Arc<Traffic>, +} + +impl CountingTransport { + pub fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self { + Self { inner, traffic } + } +} + +#[async_trait] +impl Transport for CountingTransport { + fn local_id(&self) -> &str { + self.inner.local_id() + } + + async fn resolve_seed(&self, seed: &str) -> Result<String> { + self.inner.resolve_seed(seed).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + self.inner.encode(peer, message) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let result = self.inner.send_encoded(message).await; + if result.is_ok() { + self.traffic + .sent(&message.peer, message.kind, message.payload_len); + } else { + self.traffic + .refused(&message.peer, message.kind, message.payload_len); + } + result + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let traffic = Arc::clone(&self.traffic); + self.inner + .start_consumer(Box::new(move |sender, message, payload_len| { + traffic.received(&sender, message.kind(), payload_len); + handler(sender, message, payload_len) + })) + .await + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + self.inner.wait_for_exit().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn totals_survive_retain() { + let traffic = Traffic::default(); + traffic.sent("a", MessageKind::Ping, 100); + traffic.sent("b", MessageKind::Sync, 1000); + traffic.received("b", MessageKind::Pong, 90); + traffic.refused("a", MessageKind::Sync, 500); + + let b = traffic.peer("b").unwrap(); + assert_eq!(b.sent.sync.messages, 1); + assert_eq!(b.received.pong.bytes, 90); + + traffic.retain(|peer| peer == "b"); + assert!(traffic.peer("a").is_none()); + + let total = traffic.total(); + assert_eq!(total.sent.ping.bytes, 100); + assert_eq!(total.sent.sync.bytes, 1000); + assert_eq!(total.refused.sync.bytes, 500); + } +} |