From 2c2bcf761fda888844a8538e756c8dbb9986a51f Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 21 Oct 2026 03:22:55 +0200 Subject: reorganize all the modules --- src/api/auth.rs | 141 ++++++ src/api/debug.rs | 2 +- src/api/mod.rs | 5 +- src/api/peers.rs | 6 +- src/api/store.rs | 4 +- src/auth.rs | 141 ------ src/bootstrap.rs | 104 ----- src/compaction/api.rs | 2 +- src/compaction/catchup.rs | 2 +- src/compaction/coordinator.rs | 2 +- src/compaction/delta.rs | 2 +- src/compaction/local.rs | 2 +- src/compaction/mod.rs | 4 +- src/compaction/quarantine.rs | 2 +- src/compaction/records.rs | 4 +- src/compaction/snapshot.rs | 4 +- src/compaction/switch.rs | 6 +- src/document/mod.rs | 5 + src/document/replica.rs | 130 ++++++ src/document/store.rs | 696 ++++++++++++++++++++++++++++++ src/firewall.rs | 105 ----- src/main.rs | 19 +- src/mycelium.rs | 964 ------------------------------------------ src/peer.rs | 675 ----------------------------- src/replica.rs | 130 ------ src/state.rs | 8 +- src/store.rs | 696 ------------------------------ src/swarm/bootstrap.rs | 104 +++++ src/swarm/mod.rs | 6 + src/swarm/peer.rs | 675 +++++++++++++++++++++++++++++ src/swarm/sync.rs | 564 ++++++++++++++++++++++++ src/sync.rs | 564 ------------------------ src/transport.rs | 95 ----- src/transport/firewall.rs | 105 +++++ src/transport/mod.rs | 98 +++++ src/transport/mycelium.rs | 964 ++++++++++++++++++++++++++++++++++++++++++ 36 files changed, 3523 insertions(+), 3513 deletions(-) create mode 100644 src/api/auth.rs delete mode 100644 src/auth.rs delete mode 100644 src/bootstrap.rs create mode 100644 src/document/mod.rs create mode 100644 src/document/replica.rs create mode 100644 src/document/store.rs delete mode 100644 src/firewall.rs delete mode 100644 src/mycelium.rs delete mode 100644 src/peer.rs delete mode 100644 src/replica.rs delete mode 100644 src/store.rs create mode 100644 src/swarm/bootstrap.rs create mode 100644 src/swarm/mod.rs create mode 100644 src/swarm/peer.rs create mode 100644 src/swarm/sync.rs delete mode 100644 src/sync.rs delete mode 100644 src/transport.rs create mode 100644 src/transport/firewall.rs create mode 100644 src/transport/mod.rs create mode 100644 src/transport/mycelium.rs (limited to 'src') diff --git a/src/api/auth.rs b/src/api/auth.rs new file mode 100644 index 0000000..3926c70 --- /dev/null +++ b/src/api/auth.rs @@ -0,0 +1,141 @@ +use argon2::{password_hash::PasswordVerifier, Argon2}; +use axum::extract::State; +use axum::{ + body::Body, + http::{ + header::{AUTHORIZATION, WWW_AUTHENTICATE}, + HeaderMap, Request, StatusCode, + }, + response::{IntoResponse, Response}, +}; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use dashmap::DashMap; +use password_hash::phc::PasswordHash; +use std::{ + collections::HashMap, + fs::File, + io::{BufRead, BufReader}, + sync::Arc, + time::{Duration, Instant}, +}; +use tokio::sync::Semaphore; +use tokio::time::sleep; + +fn load_users(path: &str) -> std::io::Result> { + let file = File::open(path)?; + let reader = BufReader::new(file); + + let mut users = HashMap::new(); + + for line in reader.lines() { + let line = line?; + + if let Some((user, hash)) = line.split_once(':') { + users.insert(user.to_owned(), hash.to_owned()); + } + } + + Ok(users) +} + +#[derive(Clone)] +pub struct AuthState { + idle_timeout: Duration, + + users: Arc>, // user -> PHC hash string + cache: Arc>, + /// Bounds concurrent Argon2 verifications (CPU and ~19 MiB each). + verify_slots: Arc, +} + +impl AuthState { + pub fn new(path: &str) -> Self { + let users = load_users(path).unwrap_or_default(); + + Self { + idle_timeout: Duration::from_secs(300), + cache: Arc::new(DashMap::new()), + users: Arc::new(users), + verify_slots: Arc::new(Semaphore::new(2)), + } + } +} + +fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> { + let encoded = headers + .get(AUTHORIZATION)? + .to_str() + .ok()? + .strip_prefix("Basic ")?; + let decoded = String::from_utf8(BASE64.decode(encoded).ok()?).ok()?; + let (user, pass) = decoded.split_once(':')?; + Some((user.to_owned(), pass.to_owned())) +} + +async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool { + let Some((user, pass)) = parse_basic(headers) else { + return false; + }; + let Some(hash) = state.users.get(&user).cloned() else { + return false; + }; + + let key = *blake3::hash(format!("{user}\0{pass}\0{hash}").as_bytes()).as_bytes(); + + // Fast path. The shard guard lives only until the end of this `match`; + // never call into the map while holding it. + let fresh = match state.cache.get_mut(&key) { + Some(mut last_seen) if last_seen.elapsed() < state.idle_timeout => { + *last_seen = Instant::now(); // slide + true + } + _ => false, + }; + if fresh { + return true; + } + + // Slow path: Argon2 is deliberately expensive, so keep it off the runtime + // threads and bound how many run at once. + let Ok(_permit) = state.verify_slots.acquire().await else { + return false; + }; + + let verified = tokio::task::spawn_blocking(move || { + PasswordHash::new(&hash) + .map(|ph| { + Argon2::default() + .verify_password(pass.as_bytes(), &ph) + .is_ok() + }) + .unwrap_or(false) + }) + .await + .unwrap_or(false); + + if verified { + // Overwrites an expired entry; no separate remove needed. + state.cache.insert(key, Instant::now()); + } + + verified +} + +pub async fn basic_auth( + State(state): State, + headers: HeaderMap, + request: Request, + next: axum::middleware::Next, +) -> Response { + if !authenticate(&state, &headers).await { + sleep(Duration::from_millis(1500)).await; + return ( + StatusCode::UNAUTHORIZED, + [(WWW_AUTHENTICATE, r#"Basic realm="Restricted""#)], + "Unauthorized", + ) + .into_response(); + } + + next.run(request).await +} diff --git a/src/api/debug.rs b/src/api/debug.rs index 3c24f62..929d665 100644 --- a/src/api/debug.rs +++ b/src/api/debug.rs @@ -5,8 +5,8 @@ use std::collections::HashSet; use axum::{body::Bytes, extract::State, http::StatusCode, Json}; use tracing::{info, warn}; -use crate::firewall::FirewallRules; use crate::state::AppState; +use crate::transport::firewall::FirewallRules; pub(crate) async fn get_debug_firewall(State(state): State) -> Json { Json(state.firewall.rules()) diff --git a/src/api/mod.rs b/src/api/mod.rs index 98400c7..ac6f2d0 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,5 +1,6 @@ //! HTTP API: router, OpenAPI document and shared response helpers. +pub(crate) mod auth; mod debug; mod peers; mod store; @@ -19,9 +20,9 @@ use tracing::{error, warn}; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; -use crate::auth::{basic_auth, AuthState}; +use crate::api::auth::{basic_auth, AuthState}; +use crate::document::store::StoreError; use crate::state::AppState; -use crate::store::StoreError; /// Builds the complete HTTP router. Everything except Swagger is authenticated. pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router { diff --git a/src/api/peers.rs b/src/api/peers.rs index b12e3bf..5308eb9 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -14,10 +14,10 @@ use tracing::warn; use utoipa::ToSchema; use super::store_status; -use crate::peer::forgotten_path; -use crate::replica::Epoch; +use crate::document::replica::Epoch; +use crate::document::store; use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState}; -use crate::store; +use crate::swarm::peer::forgotten_path; #[derive(Debug, Serialize, ToSchema)] #[serde(rename_all = "lowercase")] diff --git a/src/api/store.rs b/src/api/store.rs index e991005..8ea0202 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -17,9 +17,9 @@ use tokio::sync::broadcast; use tracing::{debug, error, instrument}; use super::{parse_key_path, response_for_value, store_status}; +use crate::document::store::{self, StoreError}; use crate::state::{AppState, ReadSource}; -use crate::store::{self, StoreError}; -use crate::sync::paths_related; +use crate::swarm::sync::paths_related; #[derive(Debug, Default, Deserialize)] pub(crate) struct ReadQuery { diff --git a/src/auth.rs b/src/auth.rs deleted file mode 100644 index 3926c70..0000000 --- a/src/auth.rs +++ /dev/null @@ -1,141 +0,0 @@ -use argon2::{password_hash::PasswordVerifier, Argon2}; -use axum::extract::State; -use axum::{ - body::Body, - http::{ - header::{AUTHORIZATION, WWW_AUTHENTICATE}, - HeaderMap, Request, StatusCode, - }, - response::{IntoResponse, Response}, -}; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; -use dashmap::DashMap; -use password_hash::phc::PasswordHash; -use std::{ - collections::HashMap, - fs::File, - io::{BufRead, BufReader}, - sync::Arc, - time::{Duration, Instant}, -}; -use tokio::sync::Semaphore; -use tokio::time::sleep; - -fn load_users(path: &str) -> std::io::Result> { - let file = File::open(path)?; - let reader = BufReader::new(file); - - let mut users = HashMap::new(); - - for line in reader.lines() { - let line = line?; - - if let Some((user, hash)) = line.split_once(':') { - users.insert(user.to_owned(), hash.to_owned()); - } - } - - Ok(users) -} - -#[derive(Clone)] -pub struct AuthState { - idle_timeout: Duration, - - users: Arc>, // user -> PHC hash string - cache: Arc>, - /// Bounds concurrent Argon2 verifications (CPU and ~19 MiB each). - verify_slots: Arc, -} - -impl AuthState { - pub fn new(path: &str) -> Self { - let users = load_users(path).unwrap_or_default(); - - Self { - idle_timeout: Duration::from_secs(300), - cache: Arc::new(DashMap::new()), - users: Arc::new(users), - verify_slots: Arc::new(Semaphore::new(2)), - } - } -} - -fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> { - let encoded = headers - .get(AUTHORIZATION)? - .to_str() - .ok()? - .strip_prefix("Basic ")?; - let decoded = String::from_utf8(BASE64.decode(encoded).ok()?).ok()?; - let (user, pass) = decoded.split_once(':')?; - Some((user.to_owned(), pass.to_owned())) -} - -async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool { - let Some((user, pass)) = parse_basic(headers) else { - return false; - }; - let Some(hash) = state.users.get(&user).cloned() else { - return false; - }; - - let key = *blake3::hash(format!("{user}\0{pass}\0{hash}").as_bytes()).as_bytes(); - - // Fast path. The shard guard lives only until the end of this `match`; - // never call into the map while holding it. - let fresh = match state.cache.get_mut(&key) { - Some(mut last_seen) if last_seen.elapsed() < state.idle_timeout => { - *last_seen = Instant::now(); // slide - true - } - _ => false, - }; - if fresh { - return true; - } - - // Slow path: Argon2 is deliberately expensive, so keep it off the runtime - // threads and bound how many run at once. - let Ok(_permit) = state.verify_slots.acquire().await else { - return false; - }; - - let verified = tokio::task::spawn_blocking(move || { - PasswordHash::new(&hash) - .map(|ph| { - Argon2::default() - .verify_password(pass.as_bytes(), &ph) - .is_ok() - }) - .unwrap_or(false) - }) - .await - .unwrap_or(false); - - if verified { - // Overwrites an expired entry; no separate remove needed. - state.cache.insert(key, Instant::now()); - } - - verified -} - -pub async fn basic_auth( - State(state): State, - headers: HeaderMap, - request: Request, - next: axum::middleware::Next, -) -> Response { - if !authenticate(&state, &headers).await { - sleep(Duration::from_millis(1500)).await; - return ( - StatusCode::UNAUTHORIZED, - [(WWW_AUTHENTICATE, r#"Basic realm="Restricted""#)], - "Unauthorized", - ) - .into_response(); - } - - next.run(request).await -} diff --git a/src/bootstrap.rs b/src/bootstrap.rs deleted file mode 100644 index d3ac94d..0000000 --- a/src/bootstrap.rs +++ /dev/null @@ -1,104 +0,0 @@ -//! Joining a cluster (or creating one as the genesis node). - -use std::sync::atomic::Ordering; -use std::time::Duration; - -use anyhow::{bail, Result}; -use tracing::{debug, error, info, warn}; -use uuid::Uuid; - -use crate::peer::{remember_peer, send_hello_once}; -use crate::state::{AppState, PeerMembership}; - -const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300); -const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500); - -pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> { - if seeds.is_empty() { - info!("Initializing genesis node (no seed peers configured)"); - state.initial_sync_done.store(true, Ordering::Release); - state.mark_ready(Uuid::new_v4().to_string()).await?; - return Ok(()); - } - - info!( - timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), - "Joining existing cluster" - ); - - let mut peers = Vec::new(); - - for seed in seeds { - match state.transport.resolve_peer(seed).await { - Ok(peer) => { - debug!(seed = %seed, peer = %peer, "Resolved seed to peer"); - peers.push(peer) - } - Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"), - } - } - - if peers.is_empty() { - bail!("None of the configured seeds could be resolved"); - } - - for peer in peers { - remember_peer(state, &peer).await; - if let Err(error) = send_hello_once(state, &peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry"); - } - } - - let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT; - loop { - if let Some(cluster_id) = state.cluster_id().await { - // A node joining a compacted cluster rejoins the newer epoch first; it is - // ready only once that rejoin verified the epoch. - if state.initial_sync_done.load(Ordering::Acquire) && !state.compaction.is_rejoining() { - state.mark_ready(cluster_id.clone()).await?; - info!(cluster_id = %cluster_id, "Node is ready"); - return Ok(()); - } - } - - if tokio::time::Instant::now() >= deadline { - let mut peer_diagnostics: Vec = Vec::new(); - - let peer_states = state.peer_states.read().await.clone(); - - for (peer, peer_state_mutex) in &peer_states { - let peer_state = peer_state_mutex.lock().await; - let validated_cluster = match &peer_state.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()), - }; - peer_diagnostics.push(format!( - "{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}", - validated_cluster, - peer_state.epoch, - peer_state.pending_inbound_sync.len() - )) - } - let local_cluster = state.cluster_id().await; - let local_epoch = state.local_epoch(); - let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire); - error!( - local_cluster = ?local_cluster, - local_epoch = %local_epoch, - initial_sync_done, - peers = %peer_diagnostics.join("; "), - timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), - "Bootstrap timed out" - ); - bail!( - "Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.", - BOOTSTRAP_TIMEOUT, - local_cluster, - local_epoch, - initial_sync_done, - peer_diagnostics.join("; ") - ); - } - tokio::time::sleep(BOOTSTRAP_RETRY).await; - } -} diff --git a/src/compaction/api.rs b/src/compaction/api.rs index 7f423da..9fae7c9 100644 --- a/src/compaction/api.rs +++ b/src/compaction/api.rs @@ -40,7 +40,7 @@ use super::records::{ self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention, }; use super::Host; -use crate::store::StoreError; +use crate::document::store::StoreError; #[derive(Debug, Default, Deserialize)] pub(crate) struct WaitQuery { diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs index d8bac74..5085669 100644 --- a/src/compaction/catchup.rs +++ b/src/compaction/catchup.rs @@ -37,7 +37,7 @@ use super::local::{CatchUp, LocalStatus, Quarantined, Rejoin}; use super::records::unix_millis; use super::snapshot::{self, EpochMetadata}; use super::Host; -use crate::replica::Epoch; +use crate::document::replica::Epoch; /// Pause before trying the same target again after a failed rejoin. const RETRY: Duration = Duration::from_secs(30); diff --git a/src/compaction/coordinator.rs b/src/compaction/coordinator.rs index 1b41faf..8ce3928 100644 --- a/src/compaction/coordinator.rs +++ b/src/compaction/coordinator.rs @@ -23,7 +23,7 @@ use tracing::{info, warn}; use super::records::{self, Decision, Record}; use super::snapshot; use super::Host; -use crate::store::StoreError; +use crate::document::store::StoreError; const TICK: Duration = Duration::from_secs(1); diff --git a/src/compaction/delta.rs b/src/compaction/delta.rs index ba04799..cf556b2 100644 --- a/src/compaction/delta.rs +++ b/src/compaction/delta.rs @@ -23,7 +23,7 @@ use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop}; use tracing::warn; use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT}; -use crate::store::{self, StoreError}; +use crate::document::store::{self, StoreError}; #[derive(Debug, Default, Clone, Copy)] pub(crate) struct ReplayStats { diff --git a/src/compaction/local.rs b/src/compaction/local.rs index c798a4c..ec40b06 100644 --- a/src/compaction/local.rs +++ b/src/compaction/local.rs @@ -14,7 +14,7 @@ use tokio::time::Instant; use super::records::{unix_millis, Retention}; use super::snapshot::Snapshot; -use crate::replica::Epoch; +use crate::document::replica::Epoch; #[derive(Default)] pub(crate) struct Compaction { diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs index 6de22c5..41f011f 100644 --- a/src/compaction/mod.rs +++ b/src/compaction/mod.rs @@ -38,8 +38,8 @@ use axum::{ }; use tokio::sync::broadcast; -use crate::replica::Epoch; -use crate::sync::ChangeSet; +use crate::document::replica::Epoch; +use crate::swarm::sync::ChangeSet; /// What compaction needs from the node. Implemented by `AppState`. /// diff --git a/src/compaction/quarantine.rs b/src/compaction/quarantine.rs index da39ec0..ad3c616 100644 --- a/src/compaction/quarantine.rs +++ b/src/compaction/quarantine.rs @@ -28,7 +28,7 @@ use super::api::error_response; use super::delta; use super::local::Quarantined; use super::Host; -use crate::store; +use crate::document::store; #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] diff --git a/src/compaction/records.rs b/src/compaction/records.rs index ccd52c3..f533ee2 100644 --- a/src/compaction/records.rs +++ b/src/compaction/records.rs @@ -20,8 +20,8 @@ use serde::{Deserialize, Serialize}; use tracing::warn; use super::snapshot::{SnapshotRequest, COMPACTIONS, META_ROOT}; -use crate::replica::Epoch; -use crate::store::{self, StoreError}; +use crate::document::replica::Epoch; +use crate::document::store::{self, StoreError}; /// One day, for retention defaults. const DAY_SECS: u64 = 24 * 60 * 60; diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs index 0a89e23..0d49d48 100644 --- a/src/compaction/snapshot.rs +++ b/src/compaction/snapshot.rs @@ -29,8 +29,8 @@ use serde_json::{json, Map as JsonMap, Value as JsonValue}; use sha2::{Digest, Sha256}; use super::records::{EpochRecord, Policy}; -use crate::replica::Epoch; -use crate::store::{self, StoreError}; +use crate::document::replica::Epoch; +use crate::document::store::{self, StoreError}; pub(crate) const META_ROOT: &str = "_continuum"; pub(crate) const COMPACTIONS: &str = "compactions"; diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs index 24bcee0..67ff471 100644 --- a/src/compaction/switch.rs +++ b/src/compaction/switch.rs @@ -30,9 +30,9 @@ use super::local::LocalStatus; use super::records::{self, unix_millis, EpochRecord, Record}; use super::snapshot::{self, COMPACTIONS, META_ROOT}; use super::Host; -use crate::replica::Epoch; -use crate::store; -use crate::sync::paths_related; +use crate::document::replica::Epoch; +use crate::document::store; +use crate::swarm::sync::paths_related; const TICK: Duration = Duration::from_secs(1); diff --git a/src/document/mod.rs b/src/document/mod.rs new file mode 100644 index 0000000..ce24815 --- /dev/null +++ b/src/document/mod.rs @@ -0,0 +1,5 @@ +//! The replicated document: the replica with its epoch, and the store semantics +//! (merged-view paths over JSON) the HTTP API and remakes work with. + +pub mod replica; +pub mod store; diff --git a/src/document/replica.rs b/src/document/replica.rs new file mode 100644 index 0000000..4579029 --- /dev/null +++ b/src/document/replica.rs @@ -0,0 +1,130 @@ +//! The local replica: the Automerge document together with the epoch it belongs to. +//! +//! Compaction (later steps) replaces the document with a snapshot whose history is +//! unrelated to the old one. Syncing two unrelated histories would make Automerge merge +//! them, turning every key into a conflict with its own copy, so nodes on different +//! epochs must never sync. Keeping the epoch and the document under one lock makes it +//! impossible to label a sync message from one document with the other's epoch. + +use std::fmt; +use std::ops::{Deref, DerefMut}; + +use automerge::AutoCommit; + +/// One generation of the cluster document. +/// +/// * `generation` orders epochs: every compaction increments it, so a node can tell +/// whether a peer is behind or ahead. +/// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same +/// generation apart, which only happens if two compactions committed concurrently. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct Epoch { + pub generation: u64, + pub id: [u8; 32], +} + +/// How a peer's epoch relates to ours. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EpochRelation { + Same, + /// The peer is on an older generation and has to migrate. + PeerBehind, + /// The peer is on a newer generation; we have to migrate. + PeerAhead, + /// Same generation, different snapshot: competing compactions. Never sync. + Diverged, +} + +impl Epoch { + /// Clusters that have never been compacted. + pub const GENESIS: Epoch = Epoch { + generation: 0, + id: [0; 32], + }; + + pub fn relation_to(&self, peer: &Epoch) -> EpochRelation { + use std::cmp::Ordering; + match peer.generation.cmp(&self.generation) { + Ordering::Less => EpochRelation::PeerBehind, + Ordering::Greater => EpochRelation::PeerAhead, + Ordering::Equal if peer.id == self.id => EpochRelation::Same, + Ordering::Equal => EpochRelation::Diverged, + } + } + + pub fn id_hex(&self) -> String { + self.id.iter().map(|byte| format!("{byte:02x}")).collect() + } +} + +impl fmt::Display for Epoch { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + if *self == Epoch::GENESIS { + write!(f, "0:genesis") + } else { + write!(f, "{}:{}", self.generation, &self.id_hex()[..12]) + } + } +} + +impl fmt::Debug for Epoch { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, f) + } +} + +/// The document and its epoch, always locked and swapped together. +/// +/// Derefs to the document, so code that only needs the `AutoCommit` keeps working +/// with a guard of `Replica`. +pub struct Replica { + pub epoch: Epoch, + pub doc: AutoCommit, +} + +impl Replica { + pub fn new(epoch: Epoch, doc: AutoCommit) -> Self { + Self { epoch, doc } + } +} + +impl Deref for Replica { + type Target = AutoCommit; + + fn deref(&self) -> &AutoCommit { + &self.doc + } +} + +impl DerefMut for Replica { + fn deref_mut(&mut self) -> &mut AutoCommit { + &mut self.doc + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn epoch(generation: u64, byte: u8) -> Epoch { + Epoch { + generation, + id: [byte; 32], + } + } + + #[test] + fn relation() { + let local = epoch(3, 1); + assert_eq!(local.relation_to(&epoch(3, 1)), EpochRelation::Same); + assert_eq!(local.relation_to(&epoch(2, 9)), EpochRelation::PeerBehind); + assert_eq!(local.relation_to(&epoch(4, 9)), EpochRelation::PeerAhead); + assert_eq!(local.relation_to(&epoch(3, 2)), EpochRelation::Diverged); + } + + #[test] + fn display() { + assert_eq!(Epoch::GENESIS.to_string(), "0:genesis"); + assert_eq!(epoch(2, 0xab).to_string(), "2:abababababab"); + } +} diff --git a/src/document/store.rs b/src/document/store.rs new file mode 100644 index 0000000..1815635 --- /dev/null +++ b/src/document/store.rs @@ -0,0 +1,696 @@ +//! JSON view over the Automerge document with conflict-aware merging. + +use anyhow::anyhow; +use automerge::{ + transaction::Transactable, ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, +}; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use serde::Serialize; +use serde_json::Value as JsonValue; +use std::collections::BTreeSet; + +#[derive(Debug, thiserror::Error)] +pub enum StoreError { + #[error("path crosses a non-map value")] + Conflict, + #[error("value does not exist")] + NotFound, + #[error(transparent)] + Doc(#[from] automerge::AutomergeError), + #[error(transparent)] + Other(#[from] anyhow::Error), +} + +/// Automerge's total order on operations: Lamport counter, then actor. +fn op_order(id: &ObjId) -> (u64, Option<&ActorId>) { + match id { + ObjId::Root => (0, None), + ObjId::Id(counter, actor, _) => (*counter, Some(actor)), + } +} + +#[derive(Debug, Clone)] +enum Resolved { + /// Concurrent maps presented as one. Newest first; `[0]` is canonical for writes. + Maps(Vec), + /// Any other value: the newest concurrent candidate. + Other(Value<'static>, ObjId), +} + +/// All concurrent values of `key` across a merged set of parent maps, newest first. +fn candidates( + doc: &AutoCommit, + parents: &[ObjId], + key: &str, +) -> Result, ObjId)>, StoreError> { + let mut all = Vec::new(); + for parent in parents { + for (value, id) in doc.get_all(parent, key)? { + all.push((value.into_owned(), id)); + } + } + all.sort_by(|a, b| op_order(&b.1).cmp(&op_order(&a.1))); + Ok(all) +} + +fn resolve( + doc: &AutoCommit, + parents: &[ObjId], + key: &str, + log: Option<&mut ConflictLog>, +) -> Result, StoreError> { + let mut all = candidates(doc, parents, key)?; + if all.is_empty() { + return Ok(None); + } + + let merged = matches!(all[0].0, Value::Object(ObjType::Map)); + + if all.len() > 1 { + if let Some(log) = log { + log.record(doc, key, &all, merged)?; + } + } + + if merged { + let maps = all + .into_iter() + .filter(|(value, _)| matches!(value, Value::Object(ObjType::Map))) + .map(|(_, id)| id) + .collect(); + Ok(Some(Resolved::Maps(maps))) + } else { + let (value, id) = all.swap_remove(0); + Ok(Some(Resolved::Other(value, id))) + } +} + +fn merged_keys(doc: &AutoCommit, maps: &[ObjId]) -> BTreeSet { + maps.iter().flat_map(|map| doc.keys(map)).collect() +} + +fn resolved_to_json( + doc: &AutoCommit, + resolved: &Resolved, + mut log: Option<&mut ConflictLog>, +) -> Result { + let maps = match resolved { + Resolved::Maps(maps) => maps, + Resolved::Other(Value::Scalar(scalar), _) => { + return scalar_to_json(scalar.clone().into_owned()) + } + Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log), + }; + + let mut out = serde_json::Map::new(); + + if let [map] = maps.as_slice() { + // Fast path: one pass; get_all only for keys Automerge flags as conflicted. + for item in doc.map_range(map, ..) { + let key = item.key.to_string(); + + let child = if item.conflict { + match resolve(doc, maps, &key, log.as_deref_mut())? { + Some(child) => child, + None => continue, + } + } else { + let id = item.id(); + match Value::from(item.value) { + Value::Object(ObjType::Map) => Resolved::Maps(vec![id]), + value => Resolved::Other(value.into_owned(), id), + } + }; + + let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; + out.insert(key, json); + } + } else { + // Several concurrent maps: union of their keys. + for key in merged_keys(doc, maps) { + let Some(child) = resolve(doc, maps, &key, log.as_deref_mut())? else { + continue; + }; + let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; + out.insert(key, json); + } + } + + Ok(JsonValue::Object(out)) +} + +fn object_to_json( + doc: &AutoCommit, + id: &ObjId, + mut log: Option<&mut ConflictLog>, +) -> Result { + match doc.object_type(id)? { + ObjType::Map => resolved_to_json(doc, &Resolved::Maps(vec![id.clone()]), log), + ObjType::List => { + let mut out = Vec::new(); + for (index, item) in doc.list_range(id, ..).enumerate() { + let child_id = item.id(); + out.push(match Value::from(item.value) { + Value::Object(_) => nested(&mut log, &index.to_string(), |log| { + object_to_json(doc, &child_id, log) + })?, + Value::Scalar(scalar) => scalar_to_json(scalar.into_owned())?, + }); + } + Ok(JsonValue::Array(out)) + } + ObjType::Text => Ok(JsonValue::String(doc.text(id)?)), + ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), + } +} + +fn delete_in_all(doc: &mut AutoCommit, maps: &[ObjId], key: &str) -> Result<(), StoreError> { + for map in maps { + if doc.get(map, key)?.is_some() { + doc.delete(map, key)?; + } + } + Ok(()) +} + +fn read_path_inner( + doc: &AutoCommit, + path: &[String], + mut log: Option<&mut ConflictLog>, +) -> Result, StoreError> { + let mut current = Resolved::Maps(vec![automerge::ROOT]); + + for (index, segment) in path.iter().enumerate() { + let Resolved::Maps(maps) = ¤t else { + return Ok(None); + }; + if let Some(log) = log.as_deref_mut() { + // The last segment is the requested value itself, not an ancestor. + log.in_ancestors = index + 1 < path.len(); + } + match resolve(doc, maps, segment, log.as_deref_mut())? { + Some(next) => current = next, + None => return Ok(None), + } + if let Some(log) = log.as_deref_mut() { + log.path.push(segment.clone()); + } + } + + if let Some(log) = log.as_deref_mut() { + log.in_ancestors = false; + } + Ok(Some(resolved_to_json(doc, ¤t, log)?)) +} + +/// Reads a path; the empty path is the whole store. +pub fn read_path(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { + read_path_inner(doc, path, None) +} + +pub fn read_path_with_conflicts( + doc: &AutoCommit, + path: &[String], +) -> Result)>, StoreError> { + let mut log = ConflictLog { + want_values: true, + ..Default::default() + }; + let value = read_path_inner(doc, path, Some(&mut log))?; + Ok(value.map(|value| (value, log.reports))) +} + +/// For streams: the value, plus whether the path or subtree involves any conflict. +/// Reports are collected without values, so this costs little more than read_path. +pub fn read_path_traced( + doc: &AutoCommit, + path: &[String], +) -> Result<(Option, bool), StoreError> { + let mut log = ConflictLog::default(); + let value = read_path_inner(doc, path, Some(&mut log))?; + Ok((value, !log.reports.is_empty())) +} + +/// Keys of the merged map at `path`; empty if it doesn't exist or isn't a map. +pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { + let mut maps = vec![automerge::ROOT]; + for segment in path { + maps = match resolve(doc, &maps, segment, None)? { + Some(Resolved::Maps(next)) => next, + _ => return Ok(Vec::new()), + }; + } + Ok(merged_keys(doc, &maps).into_iter().collect()) +} + +/// Writes (replaces) the value at `path`, creating intermediate maps. +pub fn write_path( + doc: &mut AutoCommit, + path: &[String], + value: &JsonValue, +) -> Result<(), StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + let maps = ensure_maps(doc, parents)?; + + // Replace semantics: clear the key in every merged parent, or an old map + // would merge into the new value. + delete_in_all(doc, &maps, leaf)?; + json_to_automerge(doc, &maps[0], leaf, value)?; + Ok(()) +} + +/// The merged maps at `path`, creating missing maps on the way. +fn ensure_maps(doc: &mut AutoCommit, path: &[String]) -> Result, StoreError> { + let mut maps = vec![automerge::ROOT]; + for segment in path { + maps = match resolve(doc, &maps, segment, None)? { + Some(Resolved::Maps(next)) => next, + Some(Resolved::Other(..)) => return Err(StoreError::Conflict), + // A concurrent creation elsewhere is merged on read. + None => vec![doc.put_object(&maps[0], segment.as_str(), ObjType::Map)?], + }; + } + Ok(maps) +} + +/// The resolved value at `path` (the empty path is the root map). +fn resolve_at(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { + let mut current = Resolved::Maps(vec![automerge::ROOT]); + for segment in path { + let Resolved::Maps(maps) = ¤t else { + return Ok(None); + }; + match resolve(doc, maps, segment, None)? { + Some(next) => current = next, + None => return Ok(None), + } + } + Ok(Some(current)) +} + +/// Makes `dst` hold at `path` what `src`'s merged view holds there: a type-preserving +/// copy (see `materialize_into`) if it exists, a delete otherwise. Replaces the whole +/// subtree at `path`; the path must not be empty. +pub(crate) fn copy_path( + src: &AutoCommit, + dst: &mut AutoCommit, + path: &[String], +) -> Result<(), StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + + let Some(resolved) = resolve_at(src, path)? else { + return match delete_path(dst, path, false) { + Ok(()) | Err(StoreError::NotFound) => Ok(()), + Err(error) => Err(error), + }; + }; + + let maps = ensure_maps(dst, parents)?; + delete_in_all(dst, &maps, leaf)?; + let mut trace = path.to_vec(); + copy_resolved( + src, + &resolved, + dst, + &maps[0], + Slot::Key(leaf), + &mut trace, + &|_| false, + ) +} + +/// Deletes the value at `path`, optionally pruning ancestors that become empty. +pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result<(), StoreError> { + let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + + // levels[d] = merged maps at path[..d] + let mut levels = vec![vec![automerge::ROOT]]; + for segment in parents { + let next = match resolve(doc, levels.last().expect("non-empty"), segment, None)? { + Some(Resolved::Maps(maps)) => maps, + _ => return Err(StoreError::NotFound), + }; + levels.push(next); + } + + let maps = levels.last().expect("non-empty"); + if resolve(doc, maps, leaf, None)?.is_none() { + return Err(StoreError::NotFound); + } + delete_in_all(doc, maps, leaf)?; + + if prune { + // Never prune the root (depth 0). + for depth in (1..levels.len()).rev() { + if !merged_keys(doc, &levels[depth]).is_empty() { + break; + } + delete_in_all(doc, &levels[depth - 1], &path[depth - 1])?; + } + } + + Ok(()) +} + +// --- materialization --------------------------------------------------------- + +/// Copies the merged view of `src` into `dst` as fresh objects. +/// +/// * Conflicts collapse exactly as reads resolve them: concurrent maps merge, everything +/// else keeps the newest candidate. +/// * Automerge types survive: text stays text, counters stay counters (normalized to +/// their current value), bytes, timestamps and floats are copied bit-exact. +/// * Map keys are visited in sorted order and list elements in order, so two documents +/// with the same view produce the same operations in the same order. +/// +/// `skip` sees the path of every map entry and list element (list indices as decimal +/// strings); returning true leaves that subtree out. +pub(crate) fn materialize_into( + src: &AutoCommit, + dst: &mut AutoCommit, + skip: &dyn Fn(&[String]) -> bool, +) -> Result<(), StoreError> { + let mut path = Vec::new(); + copy_maps( + src, + &[automerge::ROOT], + dst, + &automerge::ROOT, + &mut path, + skip, + ) +} + +fn copy_maps( + src: &AutoCommit, + maps: &[ObjId], + dst: &mut AutoCommit, + target: &ObjId, + path: &mut Vec, + skip: &dyn Fn(&[String]) -> bool, +) -> Result<(), StoreError> { + // merged_keys is a BTreeSet: sorted, so the operation order is deterministic. + for key in merged_keys(src, maps) { + path.push(key.clone()); + if !skip(path) { + if let Some(resolved) = resolve(src, maps, &key, None)? { + copy_resolved(src, &resolved, dst, target, Slot::Key(&key), path, skip)?; + } + } + path.pop(); + } + Ok(()) +} + +fn copy_resolved( + src: &AutoCommit, + resolved: &Resolved, + dst: &mut AutoCommit, + parent: &ObjId, + slot: Slot, + path: &mut Vec, + skip: &dyn Fn(&[String]) -> bool, +) -> Result<(), StoreError> { + match resolved { + Resolved::Maps(maps) => { + let id = put_object(dst, parent, slot, ObjType::Map)?; + copy_maps(src, maps, dst, &id, path, skip) + } + Resolved::Other(Value::Scalar(scalar), _) => { + let scalar = match scalar.as_ref() { + // A read counter carries its increment history; copy only its value, so + // equal counters produce equal operations. + ScalarValue::Counter(counter) => ScalarValue::counter(i64::from(counter.clone())), + other => other.clone(), + }; + put_scalar(dst, parent, slot, scalar) + } + Resolved::Other(Value::Object(obj_type), id) => { + copy_object(src, *obj_type, id, dst, parent, slot, path, skip) + } + } +} + +#[allow(clippy::too_many_arguments)] +fn copy_object( + src: &AutoCommit, + obj_type: ObjType, + id: &ObjId, + dst: &mut AutoCommit, + parent: &ObjId, + slot: Slot, + path: &mut Vec, + skip: &dyn Fn(&[String]) -> bool, +) -> Result<(), StoreError> { + match obj_type { + ObjType::Map => copy_resolved( + src, + &Resolved::Maps(vec![id.clone()]), + dst, + parent, + slot, + path, + skip, + ), + ObjType::List => { + let list = put_object(dst, parent, slot, ObjType::List)?; + // Skipped elements leave no gap in the copy. + let mut position = 0; + for (index, item) in src.list_range(id, ..).enumerate() { + path.push(index.to_string()); + if !skip(path) { + let child_id = item.id(); + let child = match Value::from(item.value).into_owned() { + Value::Object(ObjType::Map) => Resolved::Maps(vec![child_id]), + value => Resolved::Other(value, child_id), + }; + copy_resolved(src, &child, dst, &list, Slot::Index(position), path, skip)?; + position += 1; + } + path.pop(); + } + Ok(()) + } + ObjType::Text => { + let text = put_object(dst, parent, slot, ObjType::Text)?; + dst.splice_text(&text, 0, 0, &src.text(id)?)?; + Ok(()) + } + ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), + } +} + +/// Where a value is placed inside its parent object. +#[derive(Clone, Copy)] +enum Slot<'a> { + /// A key in a map. + Key(&'a str), + /// An index in a list. The value is inserted here, not overwritten. + Index(usize), +} + +fn put_object( + doc: &mut AutoCommit, + parent: &ObjId, + slot: Slot, + obj_type: ObjType, +) -> Result { + Ok(match slot { + Slot::Key(key) => doc.put_object(parent, key, obj_type)?, + Slot::Index(index) => doc.insert_object(parent, index, obj_type)?, + }) +} + +fn put_scalar( + doc: &mut AutoCommit, + parent: &ObjId, + slot: Slot, + value: impl Into, +) -> Result<(), StoreError> { + match slot { + Slot::Key(key) => doc.put(parent, key, value)?, + Slot::Index(index) => doc.insert(parent, index, value)?, + } + Ok(()) +} + +fn write_value( + doc: &mut AutoCommit, + parent: &ObjId, + slot: Slot, + value: &JsonValue, +) -> Result<(), StoreError> { + match value { + JsonValue::Object(object) => { + let id = put_object(doc, parent, slot, ObjType::Map)?; + for (child_key, child_value) in object { + write_value(doc, &id, Slot::Key(child_key), child_value)?; + } + } + JsonValue::Array(items) => { + let id = put_object(doc, parent, slot, ObjType::List)?; + for (index, item) in items.iter().enumerate() { + write_value(doc, &id, Slot::Index(index), item)?; + } + } + JsonValue::String(value) => put_scalar(doc, parent, slot, value.as_str())?, + JsonValue::Bool(value) => put_scalar(doc, parent, slot, *value)?, + JsonValue::Number(value) => { + if let Some(value) = value.as_i64() { + put_scalar(doc, parent, slot, value)?; + } else if let Some(value) = value.as_u64() { + put_scalar(doc, parent, slot, value)?; + } else if let Some(value) = value.as_f64() { + put_scalar(doc, parent, slot, value)?; + } else { + return Err(anyhow!("Unsupported JSON number").into()); + } + } + JsonValue::Null => put_scalar(doc, parent, slot, ScalarValue::Null)?, + } + + Ok(()) +} + +fn json_to_automerge( + doc: &mut AutoCommit, + parent: &ObjId, + key: &str, + value: &JsonValue, +) -> Result<(), StoreError> { + write_value(doc, parent, Slot::Key(key), value) +} + +fn scalar_to_json(scalar: ScalarValue) -> Result { + match scalar { + ScalarValue::Null => Ok(JsonValue::Null), + ScalarValue::Boolean(b) => Ok(JsonValue::Bool(b)), + ScalarValue::Int(i) => Ok(JsonValue::Number(i.into())), + ScalarValue::Uint(u) => Ok(JsonValue::Number(u.into())), + ScalarValue::F64(f) => serde_json::Number::from_f64(f) + .map(JsonValue::Number) + .ok_or(anyhow!("Non-finite float cannot be represented as JSON").into()), + ScalarValue::Str(s) => Ok(JsonValue::String(s.to_string())), + ScalarValue::Bytes(b) => Ok(JsonValue::String(BASE64.encode(b))), + ScalarValue::Counter(c) => Ok(JsonValue::Number(i64::from(c).into())), + ScalarValue::Timestamp(t) => Ok(JsonValue::Number(t.into())), + ScalarValue::Unknown { .. } => Err(anyhow!("Unsupported Automerge scalar type").into()), + } +} + +// --- conflict log ------------------------------------------------------------ + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum Resolution { + MergedMaps, + NewestWins, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CandidateReport { + pub op_id: String, + pub selected: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub value: Option, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ConflictReport { + pub path: Vec, + /// On the way to the requested path; reported without candidate values. + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub ancestor: bool, + pub resolution: Resolution, + pub candidates: Vec, +} + +#[derive(Default)] +struct ConflictLog { + path: Vec, + reports: Vec, + /// Render candidate values (only within the requested subtree). + want_values: bool, + /// Currently walking the ancestors of the requested path. + in_ancestors: bool, +} + +fn op_id_string(id: &ObjId) -> String { + match id { + ObjId::Root => "_root".to_owned(), + ObjId::Id(counter, actor, _) => format!("{counter}@{actor}"), + } +} + +fn value_to_json(doc: &AutoCommit, value: &Value<'_>, id: &ObjId) -> Result { + match value { + Value::Scalar(scalar) => Ok(scalar_to_json(scalar.clone().into_owned())?), + Value::Object(_) => object_to_json(doc, id, None), + } +} + +impl ConflictLog { + fn record( + &mut self, + doc: &AutoCommit, + key: &str, + all: &[(Value<'static>, ObjId)], + merged: bool, + ) -> Result<(), StoreError> { + let with_values = self.want_values && !self.in_ancestors; + + let candidates = all + .iter() + .enumerate() + .map(|(index, (value, id))| { + let selected = if merged { + matches!(value, Value::Object(ObjType::Map)) + } else { + index == 0 + }; + Ok(CandidateReport { + op_id: op_id_string(id), + selected, + value: if with_values { + Some(value_to_json(doc, value, id)?) + } else { + None + }, + }) + }) + .collect::, StoreError>>()?; + + let mut path = self.path.clone(); + path.push(key.to_owned()); + + self.reports.push(ConflictReport { + path, + ancestor: self.in_ancestors, + resolution: if merged { + Resolution::MergedMaps + } else { + Resolution::NewestWins + }, + candidates, + }); + Ok(()) + } +} + +/// Runs `f` with `segment` appended to the conflict path. +fn nested( + log: &mut Option<&mut ConflictLog>, + segment: &str, + f: impl FnOnce(Option<&mut ConflictLog>) -> T, +) -> T { + if let Some(log) = log.as_deref_mut() { + log.path.push(segment.to_owned()); + } + let out = f(log.as_deref_mut()); + if let Some(log) = log.as_deref_mut() { + log.path.pop(); + } + out +} diff --git a/src/firewall.rs b/src/firewall.rs deleted file mode 100644 index 917fe5f..0000000 --- a/src/firewall.rs +++ /dev/null @@ -1,105 +0,0 @@ -use std::collections::HashSet; -use std::sync::{Arc, RwLock, RwLockReadGuard}; - -use anyhow::Result; -use async_trait::async_trait; -use serde::{Deserialize, Serialize}; -use utoipa::ToSchema; - -use crate::transport::EncodedMessage; -use crate::transport::{MessageHandler, ParsedMessage, Transport}; - -#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema)] -pub struct FirewallRules { - /// Messages from these peers are dropped before the application sees them. - #[serde(default)] - pub inbound: HashSet, - /// Messages to these peers are silently dropped, like a lossy network. - #[serde(default)] - pub outbound: HashSet, -} - -/// Uses a std lock: checks are tiny and the guard is never held across an await. -#[derive(Default)] -pub struct Firewall { - rules: RwLock, -} - -impl Firewall { - fn read(&self) -> RwLockReadGuard<'_, FirewallRules> { - self.rules - .read() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - } - - pub fn rules(&self) -> FirewallRules { - self.read().clone() - } - - pub fn set_rules(&self, rules: FirewallRules) { - *self - .rules - .write() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = rules; - } - - pub fn blocks_inbound(&self, peer: &str) -> bool { - self.read().inbound.contains(peer) - } - - pub fn blocks_outbound(&self, peer: &str) -> bool { - self.read().outbound.contains(peer) - } -} - -pub struct FirewallTransport { - inner: Arc, - firewall: Arc, -} - -impl FirewallTransport { - pub fn new(inner: Arc, firewall: Arc) -> Self { - Self { inner, firewall } - } -} - -#[async_trait] -impl Transport for FirewallTransport { - fn local_id(&self) -> &str { - self.inner.local_id() - } - - async fn resolve_peer(&self, hint: &str) -> Result { - self.inner.resolve_peer(hint).await - } - - fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { - self.inner.encode(peer, message) - } - - async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - if self.firewall.blocks_outbound(&message.peer) { - tracing::trace!(peer = %message.peer, "Firewall dropped outbound message"); - return Ok(()); - } - self.inner.send_encoded(message).await - } - - async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { - let firewall = Arc::clone(&self.firewall); - - self.inner - .start_consumer(Box::new(move |sender, message| { - if firewall.blocks_inbound(&sender) { - tracing::trace!(peer = %sender, "Firewall dropped inbound message"); - return Box::pin(async { Ok(()) }); - } - handler(sender, message) - })) - .await - } - - async fn wait_for_exit(&self) -> std::io::Result { - self.inner.wait_for_exit().await - } -} diff --git a/src/main.rs b/src/main.rs index c98596b..1793523 100644 --- a/src/main.rs +++ b/src/main.rs @@ -19,23 +19,18 @@ use tracing::{debug, error, info, trace}; use url::Url; mod api; -mod auth; -mod bootstrap; mod compaction; -mod firewall; -mod mycelium; -mod peer; -mod replica; +mod document; mod state; -mod store; -mod sync; +mod swarm; mod transport; -use auth::AuthState; -use firewall::{Firewall, FirewallTransport}; -use mycelium::{MyceliumSetup, MyceliumSpawnConfig}; -use replica::{Epoch, Replica}; +use api::auth::AuthState; +use document::replica::{Epoch, Replica}; use state::{AppState, NodeLifecycle, Timing}; +use swarm::{bootstrap, peer, sync}; +use transport::firewall::{Firewall, FirewallTransport}; +use transport::mycelium::{self, MyceliumSetup, MyceliumSpawnConfig}; use transport::Transport; #[derive(Parser, Debug)] diff --git a/src/mycelium.rs b/src/mycelium.rs deleted file mode 100644 index d2deebe..0000000 --- a/src/mycelium.rs +++ /dev/null @@ -1,964 +0,0 @@ -use anyhow::{anyhow, Context, Result}; -use async_trait::async_trait; -use base64::engine::{general_purpose::STANDARD as BASE64, Engine}; -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::Keepalive; -use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; -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, - pub key_file_opt: Option, - pub network_key_file: Option, - pub network_name: String, - pub tcp_port: Option, - pub api_port: Option, - pub rpc_port: Option, - pub discovery_port: Option, - pub peers: Vec, -} - -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), - 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 { - // 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 { - match &mut self.runtime { - MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, - MyceliumRuntime::External => { - std::future::pending::>().await - } - } - } -} - -#[derive(Deserialize, Debug, Clone)] -#[serde(rename_all = "camelCase")] -pub struct MyceliumAdminInfo { - pub node_ip: String, - pub node_pubkey: String, - pub node_start_time: u64, - pub node_subnet: String, -} - -pub async fn fetch_admin_info(rest_url: &Url) -> Result { - let url = rest_url.join("/api/v1/admin")?; - let info = reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build()? - .get(url) - .send() - .await? - .error_for_status()? - .json() - .await?; - Ok(info) -} - -#[derive(Debug, Clone)] -pub struct ClusterNodeConfig { - pub tcp_port: u16, - pub api_port: u16, - pub rpc_port: u16, - pub discovery_port: Option, - pub private_network: Option, -} - -pub struct SupervisedMycelium { - child: Child, - pub config: ClusterNodeConfig, - _key_file_handle: Option, - _net_key_file_handle: Option, - key_file_path: PathBuf, - net_key_file_path: Option, - is_ephemeral_key: bool, - is_ephemeral_net_key: bool, -} - -impl SupervisedMycelium { - async fn spawn( - node_key: KeySource, - private_network: Option, - explicit_tcp: Option, - explicit_api: Option, - explicit_rpc: Option, - explicit_discovery_port: Option, - peers: &[String], - ) -> Result { - 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 = ClusterNodeConfig { - 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, is_ephemeral_key) = - Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; - - let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) = - match private_network { - Some(net_cfg) => { - let (h, p, fd, eph) = - Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; - (h, Some(p), fd, eph) - } - None => (None, None, None, false), - }; - - 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 = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); - if !fds_to_inherit.is_empty() { - unsafe { - cmd.pre_exec(move || { - for &fd in &fds_to_inherit { - libc::fcntl(fd, libc::F_SETFD, 0); - } - 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, - key_file_path, - net_key_file_path, - is_ephemeral_key, - is_ephemeral_net_key, - }; - - //let admin_info = node.poll_admin_endpoint().await?; - //Ok((node, admin_info)) - Ok(node) - } - - fn resolve_key_strategy( - desc_tag: &str, - source: KeySource, - parent_pid: u32, - ) -> Result<(Option, PathBuf, Option, bool)> { - 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), true)) - } - KeySource::Path(existing_path) => Ok((None, existing_path, None, false)), - } - } - - fn find_available_port() -> Result { - 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)) - } - - #[cfg(not(target_os = "linux"))] - { - let mut path = std::env::tempdir(); - path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4())); - let mut file = File::create(&path)?; - file.write_all(&raw_key)?; - file.flush()?; - Ok((file, -1)) - } - } - - //async fn poll_admin_endpoint(&mut self) -> Result { - // let client = reqwest::Client::new(); - // let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port); - // - // for _ in 0..30 { - // if let Some(status) = self.child.try_wait()? { - // return Err(anyhow!( - // "myceliumd process closed prematurely with status: {}", - // status - // )); - // } - // if let Ok(resp) = client.get(&url).send().await { - // if resp.status().is_success() { - // let info: MyceliumAdminInfo = resp.json().await?; - // return Ok(info); - // } - // } - // tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; - // } - // Err(anyhow!( - // "Timeout waiting for local mycelium administration socket to respond" - // )) - //} - - pub async fn wait_for_exit(&mut self) -> std::io::Result { - self.child.wait().await - } -} - -impl Drop for SupervisedMycelium { - fn drop(&mut self) { - let _ = self.child.start_kill(); - - #[cfg(not(target_os = "linux"))] - { - if self.is_ephemeral_key { - let _ = std::fs::remove_file(&self.key_file_path); - } - if self.is_ephemeral_net_key { - if let Some(ref path) = self.net_key_file_path { - let _ = std::fs::remove_file(path); - } - } - } - } -} - -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, - 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 { - 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 { - #[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_peer(&self, hint: &str) -> Result { - if is_pubkey(hint) { - return Ok(hint.to_owned()); - } - - if let Ok(ip) = hint.parse::() { - 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 { - 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::(&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::(&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::(&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 { - 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 { - let bytes = BASE64.decode(topic)?; - Ok(String::from_utf8(bytes)?) -} - -fn encode_payload(payload: &Vec) -> String { - BASE64.encode(payload) -} - -fn decode_payload(topic_encoded: &str, payload: &str) -> Result { - 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 { - 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 { - 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, - }) -} - -#[derive(Serialize, Deserialize)] -struct KeepaliveShadow { - heads_digest: ByteBuf, - #[serde(default)] - epoch: EpochShadow, -} - -fn keepalive_to_bytes(msg: &Keepalive) -> Result, rmp_serde::encode::Error> { - rmp_serde::to_vec(&KeepaliveShadow { - heads_digest: ByteBuf::from(msg.heads_digest.to_vec()), - epoch: epoch_to_shadow(&msg.epoch), - }) -} - -fn bytes_to_keepalive(bytes: &[u8]) -> Result { - 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()))?; - Ok(Keepalive { - heads_digest, - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -#[derive(Serialize, Deserialize)] -pub struct HelloRequestShadow { - #[serde(default)] - epoch: EpochShadow, -} - -#[derive(Serialize, Deserialize)] -pub struct HelloReplyShadow<'a> { - // Cow allows zero-copy references when saving, but creates an owned String when loading - pub cluster_id: Cow<'a, str>, - #[serde(default)] - pub peers: Vec, - #[serde(default)] - epoch: EpochShadow, -} - -#[derive(Serialize, Deserialize)] -pub struct SyncEnvelopeShadow<'a> { - pub cluster_id: Cow<'a, str>, - #[serde(with = "automerge_msg_codec")] - pub message: automerge::sync::Message, - #[serde(default)] - epoch: EpochShadow, -} - -// ===================================================================== -// 2. CUSTOM CODEC -// ===================================================================== -mod automerge_msg_codec { - use super::*; - use automerge::sync::Message; - - pub fn serialize(msg: &Message, serializer: S) -> Result - where - S: Serializer, - { - let bytes = msg.clone().encode(); - Bytes::new(&bytes).serialize(serializer) - } - - pub fn deserialize<'de, D>(deserializer: D) -> Result - 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, rmp_serde::encode::Error> { - rmp_serde::to_vec(&HelloRequestShadow { - epoch: epoch_to_shadow(&msg.epoch), - }) -} -pub fn bytes_to_hello_request(bytes: &[u8]) -> Result { - let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; - Ok(HelloRequest { - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -// --- HelloReply --- -pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result, rmp_serde::encode::Error> { - let shadow = HelloReplyShadow { - cluster_id: Cow::Borrowed(&msg.cluster_id), - peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), - }; - rmp_serde::to_vec(&shadow) -} -pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result { - let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?; - Ok(HelloReply { - cluster_id: shadow.cluster_id.into_owned(), - peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -// --- SyncEnvelope --- -pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result, rmp_serde::encode::Error> { - let shadow = SyncEnvelopeShadow { - cluster_id: Cow::Borrowed(&msg.cluster_id), - message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), - }; - rmp_serde::to_vec(&shadow) -} -pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result { - let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; - Ok(SyncEnvelope { - cluster_id: shadow.cluster_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, - message: shadow.message, - }) -} - -#[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/peer.rs b/src/peer.rs deleted file mode 100644 index d27cda8..0000000 --- a/src/peer.rs +++ /dev/null @@ -1,675 +0,0 @@ -//! Peers: handshake (HELLO / HELLO_REPLY), liveness and keepalives, self-healing, -//! cluster-wide forget records, and dispatch of inbound messages. - -use std::sync::Arc; -use std::time::Duration; - -use anyhow::{anyhow, bail, Context, Result}; -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::replica::{Epoch, EpochRelation}; -use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState}; -use crate::store::{self, StoreError}; -use crate::sync::{handle_sync, SyncEvent}; -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"; -const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5); - -// --- forget records ---------------------------------------------------------- - -fn forgotten_dir() -> Vec { - vec![META_ROOT.to_owned(), "forgotten".to_owned()] -} - -pub(crate) fn forgotten_path(pk: &str) -> Vec { - let mut path = forgotten_dir(); - path.push(pk.to_owned()); - path -} - -fn forgotten_peers(doc: &AutoCommit) -> Vec { - store::keys_at(doc, &forgotten_dir()).unwrap_or_default() -} - -async fn apply_forget_records(state: &AppState) { - // Before the initial sync we can't judge the records, and we have nothing to delete. - if !state.is_ready().await { - return; - } - - let forgotten = forgotten_peers(&state.doc.read().await.doc); - if forgotten.is_empty() { - return; - } - - let now = Instant::now(); - let mut cancel = Vec::new(); - - for pk in forgotten { - if pk == state.local_id() { - cancel.push(pk); // we are evidently alive - continue; - } - - let Some(peer_state_mutex) = state.get_peer_state(&pk).await else { - continue; // already forgotten here - }; - - let alive = peer_state_mutex.lock().await.is_alive(&state.timing, now); - - if alive { - cancel.push(pk); - } else if state.peer_states.write().await.remove(&pk).is_some() { - info!(peer = %pk, "Forgetting peer per cluster-wide record"); - } - } - - if cancel.is_empty() { - return; - } - - // Individual failures are logged inside; NotFound means another node already - // cancelled the record. - let _ = state - .mutate_doc(move |doc| { - for pk in &cancel { - match store::delete_path(&mut *doc, &forgotten_path(pk), false) { - Ok(()) => info!(peer = %pk, "Peer is alive; cancelling cluster-wide forget"), - Err(StoreError::NotFound) => {} - Err(error) => { - warn!(peer = %pk, error = %error, "Failed to cancel forget record") - } - } - } - Ok::<(), StoreError>(()) - }) - .await; -} - -// --- 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() -} - -pub(crate) async fn local_keepalive(state: &AppState) -> 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; - transport::Keepalive { - heads_digest: heads_digest(&mut replica.doc), - epoch: replica.epoch, - } -} - -async fn heal_if_diverged( - state: &AppState, - peer: &str, - local: &transport::Keepalive, - remote: &transport::Keepalive, -) { - // Digests of documents from different epochs are never equal and never converge. - if local.epoch != remote.epoch { - return; - } - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return; - }; - - let heal = peer_state_mutex.lock().await.needs_heal( - &local.heads_digest, - &remote.heads_digest, - &state.timing, - Instant::now(), - ); - - if heal { - warn!(peer = %peer, "Documents differ and sync is idle; re-handshaking to restart sync"); - ensure_handshake(state, peer).await; - } -} - -// --- peer discovery and HELLO ------------------------------------------------ - -pub(crate) async fn remember_peer(state: &AppState, peer: &str) { - if peer.is_empty() { - warn!("Ignoring empty peer public key"); - return; - } - - if peer == state.local_id() { - debug!("Ignoring own public key"); - return; - } - - let mut peer_states = state.peer_states.write().await; - - if !peer_states.contains_key(peer) { - peer_states.entry(peer.to_owned()).or_default(); - info!( - peer = %peer, - peer_count = peer_states.len(), - "Discovered peer" - ); - } else { - trace!(peer = %peer, "Peer already registered"); - } -} - -pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> { - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - trace!(peer = %peer, "Peer was forgotten; not sending HELLO"); - return Ok(()); - }; - - 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, - }; - true - } else { - trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending"); - false - } - }; - - if !claimed { - return Ok(()); - } - - debug!(to = %peer, "Sending HELLO"); - transmit_hello(state, peer, HELLO_RETRY_INITIAL).await -} - -async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - 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 - } - _ => return Ok(()), - } - }; - - debug!( - peer = %peer, - next_retry_ms = delay.as_millis(), - "No HELLO_REPLY yet; re-sending HELLO" - ); - transmit_hello(state, peer, delay).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<()> { - let result = state - .transport - .send_message( - peer, - &ParsedMessage::Hello(transport::HelloRequest { - epoch: state.local_epoch(), - }), - ) - .await; - - 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)), - }; - peer_state.hello_state = HelloState::AwaitingReply { - next_attempt: Instant::now() + wait, - next_delay, - }; - } - } - - result.context("Sending HELLO") -} - -async fn ensure_handshake(state: &AppState, peer: &str) { - if let Err(error) = send_hello_once(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); - } -} - -/// Sends a HELLO_REPLY if one is owed and our cluster 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.cluster_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 = 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<()> { - let cluster_id = state - .cluster_id() - .await - .ok_or_else(|| anyhow!("Cannot send HELLO_REPLY without an established cluster ID"))?; - - let candidates: Vec<(String, Arc>)> = state - .peer_states - .read() - .await - .iter() - .filter(|(known, _)| known.as_str() != peer) - .map(|(known, m)| (known.clone(), Arc::clone(m))) - .collect(); - - let mut peers = Vec::new(); - for (known, peer_state_mutex) in candidates { - let peer_state = peer_state_mutex.lock().await; - if peer_state.liveness == Liveness::Active - && peer_state.hello_state == HelloState::Established - { - peers.push(known); - } - } - - let reply = transport::HelloReply { - cluster_id, - peers, - epoch: state.local_epoch(), - }; - - let ready = state.is_ready().await; - - debug!(to = %peer, ready, "Sending HELLO_REPLY"); - - state - .transport - .send_message(peer, &ParsedMessage::HelloReply(reply)) - .await - .context("Sending HELLO_REPLY")?; - - trace!(to = %peer, "HELLO_REPLY sent"); - Ok(()) -} - -async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[String]) { - // TODO here we could add a trace t oshwo advertised_peers - for discovered_peer in advertised_peers { - if discovered_peer == from_peer { - continue; - } - - remember_peer(state, discovered_peer).await; - - if let Err(error) = send_hello_once(state, discovered_peer).await { - warn!( - peer = %discovered_peer, - error = %format!("{error:#}"), - "Failed to initiate HELLO with discovered peer" - ); - } - } -} - -// --- peer worker ------------------------------------------------------------- - -pub(crate) async fn run_peer_worker(state: AppState) { - let mut tick = tokio::time::interval(PEER_TICK); - tick.set_missed_tick_behavior(MissedTickBehavior::Delay); - - info!("Starting peer maintenance worker"); - - let mut last_forget_check: Option = None; - - loop { - tick.tick().await; - - // While joining, keep handshakes fast; dormancy would slow probes to minutes. - let ready = state.is_ready().await; - - let peers: Vec<(String, Arc>)> = state - .peer_states - .read() - .await - .iter() - .map(|(pk, m)| (pk.clone(), Arc::clone(m))) - .collect(); - - for (peer, peer_state_mutex) in &peers { - let now = Instant::now(); - - let (became_dormant, ping_due) = { - let mut peer_state = peer_state_mutex.lock().await; - let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); - let ping_due = peer_state.take_keepalive_due(&state.timing, now); - (became_dormant, ping_due) - }; - - if became_dormant { - warn!( - peer = %peer, - silent_secs = state.timing.dormant_after.as_secs(), - probe_secs = state.timing.dormant_probe.as_secs(), - "Peer went dormant" - ); - } - - if ping_due { - let local = local_keepalive(&state).await; - if let Err(error) = state - .transport - .send_message(peer, &ParsedMessage::Ping(local)) - .await - { - debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); - } - } - - 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; - last_forget_check = Some(Instant::now()); - } - } -} - -// --- peer membership --------------------------------------------------------- - -async fn establish_peer( - state: &AppState, - peer: &str, - cluster_id: String, -) -> Result> { - let expected_cluster = state.cluster_id().await; - - if let Some(expected) = &expected_cluster { - if expected != &cluster_id { - bail!( - "Incompatible cluster at {peer}: expected {}, received {}", - expected, - cluster_id - ); - } - } - - 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 { - cluster_id: existing, - } = &peer_state.membership - { - if existing != &cluster_id { - bail!( - "Peer {peer} changed cluster identity from {} to {}", - existing, - cluster_id - ); - } - } - - // Lock order: peer mutex -> lifecycle. - if expected_cluster.is_none() { - state.set_expected_cluster_id(cluster_id.clone()).await?; - } - - if matches!(peer_state.membership, PeerMembership::Unknown) { - peer_state.membership = PeerMembership::Member { - cluster_id: cluster_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 { - peer_state.hello_state = HelloState::Established; - peer_state.reset_sync(); - info!(peer = %peer, "Handshake complete; sync state reset"); - } - - let pending = std::mem::take(&mut peer_state.pending_inbound_sync); - - info!(peer = %peer, cluster_id = %cluster_id, pending_messages = pending.len(), "Peer membership established"); - Ok(pending) -} - -#[instrument(skip(state), fields(peer = %peer))] -pub(crate) async fn peer_cluster(state: &AppState, peer: &str) -> Option { - 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 { cluster_id } => Some(cluster_id.clone()), - } -} - -// --- epochs ------------------------------------------------------------------ - -/// Records the epoch a peer advertised and compares it with ours. Returns true when both -/// are on the same epoch, the only case in which the documents may be synced. -async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool { - let local = state.local_epoch(); - let relation = local.relation_to(&remote); - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return relation == EpochRelation::Same; - }; - - let previous = { - let mut peer_state = peer_state_mutex.lock().await; - let previous = peer_state.epoch.replace(remote); - if previous.is_some_and(|previous| previous != remote) { - // The peer switched documents; our sync state describes its old one. Count - // the restart as sync activity, so the keepalive heal doesn't re-handshake - // (resetting both sides again) while this fresh sync is under way. - peer_state.reset_sync(); - peer_state.last_sync_rx = Some(Instant::now()); - } - previous - }; - - // Log transitions only, not every keepalive. - if previous != Some(remote) { - match relation { - EpochRelation::Same => { - if previous.is_some() { - info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync"); - let _ = state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.to_owned())); - } - } - // Expected after a compaction: either side switches or catches up. - EpochRelation::PeerAhead | EpochRelation::PeerBehind => { - info!( - peer = %peer, - local_epoch = %local, - peer_epoch = %remote, - relation = ?relation, - "Peer is on another epoch; not syncing with it until both are on the same" - ); - } - EpochRelation::Diverged => { - warn!( - peer = %peer, - local_epoch = %local, - peer_epoch = %remote, - "Peer is on a different epoch of the same generation; not syncing with it" - ); - } - } - } - - relation == EpochRelation::Same -} - -// --- inbound dispatch -------------------------------------------------------- - -pub(crate) async fn handle_message( - state: &AppState, - peer: String, - message: transport::ParsedMessage, -) -> Result<()> { - if peer == state.local_id() { - debug!("Ignoring message from self"); - 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()); - - if reactivated { - info!(peer = %peer, "Dormant peer is active again; re-handshaking"); - } - - let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; - - //debug!(from = %peer, message = ?message, "Received message"); - - match message { - transport::ParsedMessage::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; - } - - // Sends now if we know our cluster 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"); - } - - // Restart sync with the fresh state (no-op until the peer is a validated member). - let _ = state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); - } - transport::ParsedMessage::HelloReply(reply) => { - let pending = establish_peer(state, &peer, reply.cluster_id.clone()).await?; - - discover_peers(state, &peer, &reply.peers).await; - - // Our cluster 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())) - .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; - } - transport::ParsedMessage::Sync(envelope) => { - if reactivated { - ensure_handshake(state, &peer).await; - } - if !same_epoch { - debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); - return Ok(()); - } - handle_sync(state, &peer, envelope).await?; - } - transport::ParsedMessage::Ping(remote) => { - if reactivated || peer_cluster(state, &peer).await.is_none() { - ensure_handshake(state, &peer).await; - } - let local = local_keepalive(state).await; - heal_if_diverged(state, &peer, &local, &remote).await; - if let Err(error) = state - .transport - .send_message(&peer, &ParsedMessage::Pong(local)) - .await - { - debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); - } - } - transport::ParsedMessage::Pong(remote) => { - if reactivated || peer_cluster(state, &peer).await.is_none() { - ensure_handshake(state, &peer).await; - } - let local = local_keepalive(state).await; - heal_if_diverged(state, &peer, &local, &remote).await; - } - } - - Ok(()) -} diff --git a/src/replica.rs b/src/replica.rs deleted file mode 100644 index 4579029..0000000 --- a/src/replica.rs +++ /dev/null @@ -1,130 +0,0 @@ -//! The local replica: the Automerge document together with the epoch it belongs to. -//! -//! Compaction (later steps) replaces the document with a snapshot whose history is -//! unrelated to the old one. Syncing two unrelated histories would make Automerge merge -//! them, turning every key into a conflict with its own copy, so nodes on different -//! epochs must never sync. Keeping the epoch and the document under one lock makes it -//! impossible to label a sync message from one document with the other's epoch. - -use std::fmt; -use std::ops::{Deref, DerefMut}; - -use automerge::AutoCommit; - -/// One generation of the cluster document. -/// -/// * `generation` orders epochs: every compaction increments it, so a node can tell -/// whether a peer is behind or ahead. -/// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same -/// generation apart, which only happens if two compactions committed concurrently. -#[derive(Clone, Copy, PartialEq, Eq, Hash)] -pub struct Epoch { - pub generation: u64, - pub id: [u8; 32], -} - -/// How a peer's epoch relates to ours. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum EpochRelation { - Same, - /// The peer is on an older generation and has to migrate. - PeerBehind, - /// The peer is on a newer generation; we have to migrate. - PeerAhead, - /// Same generation, different snapshot: competing compactions. Never sync. - Diverged, -} - -impl Epoch { - /// Clusters that have never been compacted. - pub const GENESIS: Epoch = Epoch { - generation: 0, - id: [0; 32], - }; - - pub fn relation_to(&self, peer: &Epoch) -> EpochRelation { - use std::cmp::Ordering; - match peer.generation.cmp(&self.generation) { - Ordering::Less => EpochRelation::PeerBehind, - Ordering::Greater => EpochRelation::PeerAhead, - Ordering::Equal if peer.id == self.id => EpochRelation::Same, - Ordering::Equal => EpochRelation::Diverged, - } - } - - pub fn id_hex(&self) -> String { - self.id.iter().map(|byte| format!("{byte:02x}")).collect() - } -} - -impl fmt::Display for Epoch { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - if *self == Epoch::GENESIS { - write!(f, "0:genesis") - } else { - write!(f, "{}:{}", self.generation, &self.id_hex()[..12]) - } - } -} - -impl fmt::Debug for Epoch { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt::Display::fmt(self, f) - } -} - -/// The document and its epoch, always locked and swapped together. -/// -/// Derefs to the document, so code that only needs the `AutoCommit` keeps working -/// with a guard of `Replica`. -pub struct Replica { - pub epoch: Epoch, - pub doc: AutoCommit, -} - -impl Replica { - pub fn new(epoch: Epoch, doc: AutoCommit) -> Self { - Self { epoch, doc } - } -} - -impl Deref for Replica { - type Target = AutoCommit; - - fn deref(&self) -> &AutoCommit { - &self.doc - } -} - -impl DerefMut for Replica { - fn deref_mut(&mut self) -> &mut AutoCommit { - &mut self.doc - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn epoch(generation: u64, byte: u8) -> Epoch { - Epoch { - generation, - id: [byte; 32], - } - } - - #[test] - fn relation() { - let local = epoch(3, 1); - assert_eq!(local.relation_to(&epoch(3, 1)), EpochRelation::Same); - assert_eq!(local.relation_to(&epoch(2, 9)), EpochRelation::PeerBehind); - assert_eq!(local.relation_to(&epoch(4, 9)), EpochRelation::PeerAhead); - assert_eq!(local.relation_to(&epoch(3, 2)), EpochRelation::Diverged); - } - - #[test] - fn display() { - assert_eq!(Epoch::GENESIS.to_string(), "0:genesis"); - assert_eq!(epoch(2, 0xab).to_string(), "2:abababababab"); - } -} diff --git a/src/state.rs b/src/state.rs index e7c1767..18bface 100644 --- a/src/state.rs +++ b/src/state.rs @@ -13,10 +13,10 @@ use tokio::time::Instant; use tracing::{debug, error, warn}; use crate::compaction::{Compaction, Host, PeerSummary}; -use crate::firewall::Firewall; -use crate::replica::{Epoch, Replica}; -use crate::sync::{ChangeSet, SyncEvent}; +use crate::document::replica::{Epoch, Replica}; +use crate::swarm::sync::{ChangeSet, SyncEvent}; use crate::transport; +use crate::transport::firewall::Firewall; #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum NodeLifecycle { @@ -539,7 +539,7 @@ impl AppState { // 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::peer::local_keepalive(self).await; + let keepalive = crate::swarm::peer::local_keepalive(self).await; for (id, _) in &peers { if let Err(error) = self .transport diff --git a/src/store.rs b/src/store.rs deleted file mode 100644 index 1815635..0000000 --- a/src/store.rs +++ /dev/null @@ -1,696 +0,0 @@ -//! JSON view over the Automerge document with conflict-aware merging. - -use anyhow::anyhow; -use automerge::{ - transaction::Transactable, ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, -}; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; -use serde::Serialize; -use serde_json::Value as JsonValue; -use std::collections::BTreeSet; - -#[derive(Debug, thiserror::Error)] -pub enum StoreError { - #[error("path crosses a non-map value")] - Conflict, - #[error("value does not exist")] - NotFound, - #[error(transparent)] - Doc(#[from] automerge::AutomergeError), - #[error(transparent)] - Other(#[from] anyhow::Error), -} - -/// Automerge's total order on operations: Lamport counter, then actor. -fn op_order(id: &ObjId) -> (u64, Option<&ActorId>) { - match id { - ObjId::Root => (0, None), - ObjId::Id(counter, actor, _) => (*counter, Some(actor)), - } -} - -#[derive(Debug, Clone)] -enum Resolved { - /// Concurrent maps presented as one. Newest first; `[0]` is canonical for writes. - Maps(Vec), - /// Any other value: the newest concurrent candidate. - Other(Value<'static>, ObjId), -} - -/// All concurrent values of `key` across a merged set of parent maps, newest first. -fn candidates( - doc: &AutoCommit, - parents: &[ObjId], - key: &str, -) -> Result, ObjId)>, StoreError> { - let mut all = Vec::new(); - for parent in parents { - for (value, id) in doc.get_all(parent, key)? { - all.push((value.into_owned(), id)); - } - } - all.sort_by(|a, b| op_order(&b.1).cmp(&op_order(&a.1))); - Ok(all) -} - -fn resolve( - doc: &AutoCommit, - parents: &[ObjId], - key: &str, - log: Option<&mut ConflictLog>, -) -> Result, StoreError> { - let mut all = candidates(doc, parents, key)?; - if all.is_empty() { - return Ok(None); - } - - let merged = matches!(all[0].0, Value::Object(ObjType::Map)); - - if all.len() > 1 { - if let Some(log) = log { - log.record(doc, key, &all, merged)?; - } - } - - if merged { - let maps = all - .into_iter() - .filter(|(value, _)| matches!(value, Value::Object(ObjType::Map))) - .map(|(_, id)| id) - .collect(); - Ok(Some(Resolved::Maps(maps))) - } else { - let (value, id) = all.swap_remove(0); - Ok(Some(Resolved::Other(value, id))) - } -} - -fn merged_keys(doc: &AutoCommit, maps: &[ObjId]) -> BTreeSet { - maps.iter().flat_map(|map| doc.keys(map)).collect() -} - -fn resolved_to_json( - doc: &AutoCommit, - resolved: &Resolved, - mut log: Option<&mut ConflictLog>, -) -> Result { - let maps = match resolved { - Resolved::Maps(maps) => maps, - Resolved::Other(Value::Scalar(scalar), _) => { - return scalar_to_json(scalar.clone().into_owned()) - } - Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log), - }; - - let mut out = serde_json::Map::new(); - - if let [map] = maps.as_slice() { - // Fast path: one pass; get_all only for keys Automerge flags as conflicted. - for item in doc.map_range(map, ..) { - let key = item.key.to_string(); - - let child = if item.conflict { - match resolve(doc, maps, &key, log.as_deref_mut())? { - Some(child) => child, - None => continue, - } - } else { - let id = item.id(); - match Value::from(item.value) { - Value::Object(ObjType::Map) => Resolved::Maps(vec![id]), - value => Resolved::Other(value.into_owned(), id), - } - }; - - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; - out.insert(key, json); - } - } else { - // Several concurrent maps: union of their keys. - for key in merged_keys(doc, maps) { - let Some(child) = resolve(doc, maps, &key, log.as_deref_mut())? else { - continue; - }; - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; - out.insert(key, json); - } - } - - Ok(JsonValue::Object(out)) -} - -fn object_to_json( - doc: &AutoCommit, - id: &ObjId, - mut log: Option<&mut ConflictLog>, -) -> Result { - match doc.object_type(id)? { - ObjType::Map => resolved_to_json(doc, &Resolved::Maps(vec![id.clone()]), log), - ObjType::List => { - let mut out = Vec::new(); - for (index, item) in doc.list_range(id, ..).enumerate() { - let child_id = item.id(); - out.push(match Value::from(item.value) { - Value::Object(_) => nested(&mut log, &index.to_string(), |log| { - object_to_json(doc, &child_id, log) - })?, - Value::Scalar(scalar) => scalar_to_json(scalar.into_owned())?, - }); - } - Ok(JsonValue::Array(out)) - } - ObjType::Text => Ok(JsonValue::String(doc.text(id)?)), - ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), - } -} - -fn delete_in_all(doc: &mut AutoCommit, maps: &[ObjId], key: &str) -> Result<(), StoreError> { - for map in maps { - if doc.get(map, key)?.is_some() { - doc.delete(map, key)?; - } - } - Ok(()) -} - -fn read_path_inner( - doc: &AutoCommit, - path: &[String], - mut log: Option<&mut ConflictLog>, -) -> Result, StoreError> { - let mut current = Resolved::Maps(vec![automerge::ROOT]); - - for (index, segment) in path.iter().enumerate() { - let Resolved::Maps(maps) = ¤t else { - return Ok(None); - }; - if let Some(log) = log.as_deref_mut() { - // The last segment is the requested value itself, not an ancestor. - log.in_ancestors = index + 1 < path.len(); - } - match resolve(doc, maps, segment, log.as_deref_mut())? { - Some(next) => current = next, - None => return Ok(None), - } - if let Some(log) = log.as_deref_mut() { - log.path.push(segment.clone()); - } - } - - if let Some(log) = log.as_deref_mut() { - log.in_ancestors = false; - } - Ok(Some(resolved_to_json(doc, ¤t, log)?)) -} - -/// Reads a path; the empty path is the whole store. -pub fn read_path(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { - read_path_inner(doc, path, None) -} - -pub fn read_path_with_conflicts( - doc: &AutoCommit, - path: &[String], -) -> Result)>, StoreError> { - let mut log = ConflictLog { - want_values: true, - ..Default::default() - }; - let value = read_path_inner(doc, path, Some(&mut log))?; - Ok(value.map(|value| (value, log.reports))) -} - -/// For streams: the value, plus whether the path or subtree involves any conflict. -/// Reports are collected without values, so this costs little more than read_path. -pub fn read_path_traced( - doc: &AutoCommit, - path: &[String], -) -> Result<(Option, bool), StoreError> { - let mut log = ConflictLog::default(); - let value = read_path_inner(doc, path, Some(&mut log))?; - Ok((value, !log.reports.is_empty())) -} - -/// Keys of the merged map at `path`; empty if it doesn't exist or isn't a map. -pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { - let mut maps = vec![automerge::ROOT]; - for segment in path { - maps = match resolve(doc, &maps, segment, None)? { - Some(Resolved::Maps(next)) => next, - _ => return Ok(Vec::new()), - }; - } - Ok(merged_keys(doc, &maps).into_iter().collect()) -} - -/// Writes (replaces) the value at `path`, creating intermediate maps. -pub fn write_path( - doc: &mut AutoCommit, - path: &[String], - value: &JsonValue, -) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - let maps = ensure_maps(doc, parents)?; - - // Replace semantics: clear the key in every merged parent, or an old map - // would merge into the new value. - delete_in_all(doc, &maps, leaf)?; - json_to_automerge(doc, &maps[0], leaf, value)?; - Ok(()) -} - -/// The merged maps at `path`, creating missing maps on the way. -fn ensure_maps(doc: &mut AutoCommit, path: &[String]) -> Result, StoreError> { - let mut maps = vec![automerge::ROOT]; - for segment in path { - maps = match resolve(doc, &maps, segment, None)? { - Some(Resolved::Maps(next)) => next, - Some(Resolved::Other(..)) => return Err(StoreError::Conflict), - // A concurrent creation elsewhere is merged on read. - None => vec![doc.put_object(&maps[0], segment.as_str(), ObjType::Map)?], - }; - } - Ok(maps) -} - -/// The resolved value at `path` (the empty path is the root map). -fn resolve_at(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { - let mut current = Resolved::Maps(vec![automerge::ROOT]); - for segment in path { - let Resolved::Maps(maps) = ¤t else { - return Ok(None); - }; - match resolve(doc, maps, segment, None)? { - Some(next) => current = next, - None => return Ok(None), - } - } - Ok(Some(current)) -} - -/// Makes `dst` hold at `path` what `src`'s merged view holds there: a type-preserving -/// copy (see `materialize_into`) if it exists, a delete otherwise. Replaces the whole -/// subtree at `path`; the path must not be empty. -pub(crate) fn copy_path( - src: &AutoCommit, - dst: &mut AutoCommit, - path: &[String], -) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - - let Some(resolved) = resolve_at(src, path)? else { - return match delete_path(dst, path, false) { - Ok(()) | Err(StoreError::NotFound) => Ok(()), - Err(error) => Err(error), - }; - }; - - let maps = ensure_maps(dst, parents)?; - delete_in_all(dst, &maps, leaf)?; - let mut trace = path.to_vec(); - copy_resolved( - src, - &resolved, - dst, - &maps[0], - Slot::Key(leaf), - &mut trace, - &|_| false, - ) -} - -/// Deletes the value at `path`, optionally pruning ancestors that become empty. -pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - - // levels[d] = merged maps at path[..d] - let mut levels = vec![vec![automerge::ROOT]]; - for segment in parents { - let next = match resolve(doc, levels.last().expect("non-empty"), segment, None)? { - Some(Resolved::Maps(maps)) => maps, - _ => return Err(StoreError::NotFound), - }; - levels.push(next); - } - - let maps = levels.last().expect("non-empty"); - if resolve(doc, maps, leaf, None)?.is_none() { - return Err(StoreError::NotFound); - } - delete_in_all(doc, maps, leaf)?; - - if prune { - // Never prune the root (depth 0). - for depth in (1..levels.len()).rev() { - if !merged_keys(doc, &levels[depth]).is_empty() { - break; - } - delete_in_all(doc, &levels[depth - 1], &path[depth - 1])?; - } - } - - Ok(()) -} - -// --- materialization --------------------------------------------------------- - -/// Copies the merged view of `src` into `dst` as fresh objects. -/// -/// * Conflicts collapse exactly as reads resolve them: concurrent maps merge, everything -/// else keeps the newest candidate. -/// * Automerge types survive: text stays text, counters stay counters (normalized to -/// their current value), bytes, timestamps and floats are copied bit-exact. -/// * Map keys are visited in sorted order and list elements in order, so two documents -/// with the same view produce the same operations in the same order. -/// -/// `skip` sees the path of every map entry and list element (list indices as decimal -/// strings); returning true leaves that subtree out. -pub(crate) fn materialize_into( - src: &AutoCommit, - dst: &mut AutoCommit, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - let mut path = Vec::new(); - copy_maps( - src, - &[automerge::ROOT], - dst, - &automerge::ROOT, - &mut path, - skip, - ) -} - -fn copy_maps( - src: &AutoCommit, - maps: &[ObjId], - dst: &mut AutoCommit, - target: &ObjId, - path: &mut Vec, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - // merged_keys is a BTreeSet: sorted, so the operation order is deterministic. - for key in merged_keys(src, maps) { - path.push(key.clone()); - if !skip(path) { - if let Some(resolved) = resolve(src, maps, &key, None)? { - copy_resolved(src, &resolved, dst, target, Slot::Key(&key), path, skip)?; - } - } - path.pop(); - } - Ok(()) -} - -fn copy_resolved( - src: &AutoCommit, - resolved: &Resolved, - dst: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - path: &mut Vec, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - match resolved { - Resolved::Maps(maps) => { - let id = put_object(dst, parent, slot, ObjType::Map)?; - copy_maps(src, maps, dst, &id, path, skip) - } - Resolved::Other(Value::Scalar(scalar), _) => { - let scalar = match scalar.as_ref() { - // A read counter carries its increment history; copy only its value, so - // equal counters produce equal operations. - ScalarValue::Counter(counter) => ScalarValue::counter(i64::from(counter.clone())), - other => other.clone(), - }; - put_scalar(dst, parent, slot, scalar) - } - Resolved::Other(Value::Object(obj_type), id) => { - copy_object(src, *obj_type, id, dst, parent, slot, path, skip) - } - } -} - -#[allow(clippy::too_many_arguments)] -fn copy_object( - src: &AutoCommit, - obj_type: ObjType, - id: &ObjId, - dst: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - path: &mut Vec, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - match obj_type { - ObjType::Map => copy_resolved( - src, - &Resolved::Maps(vec![id.clone()]), - dst, - parent, - slot, - path, - skip, - ), - ObjType::List => { - let list = put_object(dst, parent, slot, ObjType::List)?; - // Skipped elements leave no gap in the copy. - let mut position = 0; - for (index, item) in src.list_range(id, ..).enumerate() { - path.push(index.to_string()); - if !skip(path) { - let child_id = item.id(); - let child = match Value::from(item.value).into_owned() { - Value::Object(ObjType::Map) => Resolved::Maps(vec![child_id]), - value => Resolved::Other(value, child_id), - }; - copy_resolved(src, &child, dst, &list, Slot::Index(position), path, skip)?; - position += 1; - } - path.pop(); - } - Ok(()) - } - ObjType::Text => { - let text = put_object(dst, parent, slot, ObjType::Text)?; - dst.splice_text(&text, 0, 0, &src.text(id)?)?; - Ok(()) - } - ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), - } -} - -/// Where a value is placed inside its parent object. -#[derive(Clone, Copy)] -enum Slot<'a> { - /// A key in a map. - Key(&'a str), - /// An index in a list. The value is inserted here, not overwritten. - Index(usize), -} - -fn put_object( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - obj_type: ObjType, -) -> Result { - Ok(match slot { - Slot::Key(key) => doc.put_object(parent, key, obj_type)?, - Slot::Index(index) => doc.insert_object(parent, index, obj_type)?, - }) -} - -fn put_scalar( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - value: impl Into, -) -> Result<(), StoreError> { - match slot { - Slot::Key(key) => doc.put(parent, key, value)?, - Slot::Index(index) => doc.insert(parent, index, value)?, - } - Ok(()) -} - -fn write_value( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - value: &JsonValue, -) -> Result<(), StoreError> { - match value { - JsonValue::Object(object) => { - let id = put_object(doc, parent, slot, ObjType::Map)?; - for (child_key, child_value) in object { - write_value(doc, &id, Slot::Key(child_key), child_value)?; - } - } - JsonValue::Array(items) => { - let id = put_object(doc, parent, slot, ObjType::List)?; - for (index, item) in items.iter().enumerate() { - write_value(doc, &id, Slot::Index(index), item)?; - } - } - JsonValue::String(value) => put_scalar(doc, parent, slot, value.as_str())?, - JsonValue::Bool(value) => put_scalar(doc, parent, slot, *value)?, - JsonValue::Number(value) => { - if let Some(value) = value.as_i64() { - put_scalar(doc, parent, slot, value)?; - } else if let Some(value) = value.as_u64() { - put_scalar(doc, parent, slot, value)?; - } else if let Some(value) = value.as_f64() { - put_scalar(doc, parent, slot, value)?; - } else { - return Err(anyhow!("Unsupported JSON number").into()); - } - } - JsonValue::Null => put_scalar(doc, parent, slot, ScalarValue::Null)?, - } - - Ok(()) -} - -fn json_to_automerge( - doc: &mut AutoCommit, - parent: &ObjId, - key: &str, - value: &JsonValue, -) -> Result<(), StoreError> { - write_value(doc, parent, Slot::Key(key), value) -} - -fn scalar_to_json(scalar: ScalarValue) -> Result { - match scalar { - ScalarValue::Null => Ok(JsonValue::Null), - ScalarValue::Boolean(b) => Ok(JsonValue::Bool(b)), - ScalarValue::Int(i) => Ok(JsonValue::Number(i.into())), - ScalarValue::Uint(u) => Ok(JsonValue::Number(u.into())), - ScalarValue::F64(f) => serde_json::Number::from_f64(f) - .map(JsonValue::Number) - .ok_or(anyhow!("Non-finite float cannot be represented as JSON").into()), - ScalarValue::Str(s) => Ok(JsonValue::String(s.to_string())), - ScalarValue::Bytes(b) => Ok(JsonValue::String(BASE64.encode(b))), - ScalarValue::Counter(c) => Ok(JsonValue::Number(i64::from(c).into())), - ScalarValue::Timestamp(t) => Ok(JsonValue::Number(t.into())), - ScalarValue::Unknown { .. } => Err(anyhow!("Unsupported Automerge scalar type").into()), - } -} - -// --- conflict log ------------------------------------------------------------ - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub enum Resolution { - MergedMaps, - NewestWins, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct CandidateReport { - pub op_id: String, - pub selected: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub value: Option, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct ConflictReport { - pub path: Vec, - /// On the way to the requested path; reported without candidate values. - #[serde(skip_serializing_if = "std::ops::Not::not")] - pub ancestor: bool, - pub resolution: Resolution, - pub candidates: Vec, -} - -#[derive(Default)] -struct ConflictLog { - path: Vec, - reports: Vec, - /// Render candidate values (only within the requested subtree). - want_values: bool, - /// Currently walking the ancestors of the requested path. - in_ancestors: bool, -} - -fn op_id_string(id: &ObjId) -> String { - match id { - ObjId::Root => "_root".to_owned(), - ObjId::Id(counter, actor, _) => format!("{counter}@{actor}"), - } -} - -fn value_to_json(doc: &AutoCommit, value: &Value<'_>, id: &ObjId) -> Result { - match value { - Value::Scalar(scalar) => Ok(scalar_to_json(scalar.clone().into_owned())?), - Value::Object(_) => object_to_json(doc, id, None), - } -} - -impl ConflictLog { - fn record( - &mut self, - doc: &AutoCommit, - key: &str, - all: &[(Value<'static>, ObjId)], - merged: bool, - ) -> Result<(), StoreError> { - let with_values = self.want_values && !self.in_ancestors; - - let candidates = all - .iter() - .enumerate() - .map(|(index, (value, id))| { - let selected = if merged { - matches!(value, Value::Object(ObjType::Map)) - } else { - index == 0 - }; - Ok(CandidateReport { - op_id: op_id_string(id), - selected, - value: if with_values { - Some(value_to_json(doc, value, id)?) - } else { - None - }, - }) - }) - .collect::, StoreError>>()?; - - let mut path = self.path.clone(); - path.push(key.to_owned()); - - self.reports.push(ConflictReport { - path, - ancestor: self.in_ancestors, - resolution: if merged { - Resolution::MergedMaps - } else { - Resolution::NewestWins - }, - candidates, - }); - Ok(()) - } -} - -/// Runs `f` with `segment` appended to the conflict path. -fn nested( - log: &mut Option<&mut ConflictLog>, - segment: &str, - f: impl FnOnce(Option<&mut ConflictLog>) -> T, -) -> T { - if let Some(log) = log.as_deref_mut() { - log.path.push(segment.to_owned()); - } - let out = f(log.as_deref_mut()); - if let Some(log) = log.as_deref_mut() { - log.path.pop(); - } - out -} diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs new file mode 100644 index 0000000..2907e93 --- /dev/null +++ b/src/swarm/bootstrap.rs @@ -0,0 +1,104 @@ +//! Joining a cluster (or creating one as the genesis node). + +use std::sync::atomic::Ordering; +use std::time::Duration; + +use anyhow::{bail, Result}; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +use crate::state::{AppState, PeerMembership}; +use crate::swarm::peer::{remember_peer, send_hello_once}; + +const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(300); +const BOOTSTRAP_RETRY: Duration = Duration::from_millis(500); + +pub(crate) async fn bootstrap(state: &AppState, seeds: &[String]) -> Result<()> { + if seeds.is_empty() { + info!("Initializing genesis node (no seed peers configured)"); + state.initial_sync_done.store(true, Ordering::Release); + state.mark_ready(Uuid::new_v4().to_string()).await?; + return Ok(()); + } + + info!( + timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), + "Joining existing cluster" + ); + + let mut peers = Vec::new(); + + for seed in seeds { + match state.transport.resolve_peer(seed).await { + Ok(peer) => { + debug!(seed = %seed, peer = %peer, "Resolved seed to peer"); + peers.push(peer) + } + Err(e) => warn!(seed = %seed, error = %e, "Ignoring seed argument"), + } + } + + if peers.is_empty() { + bail!("None of the configured seeds could be resolved"); + } + + for peer in peers { + remember_peer(state, &peer).await; + if let Err(error) = send_hello_once(state, &peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "Initial HELLO failed; will retry"); + } + } + + let deadline = tokio::time::Instant::now() + BOOTSTRAP_TIMEOUT; + loop { + if let Some(cluster_id) = state.cluster_id().await { + // A node joining a compacted cluster rejoins the newer epoch first; it is + // ready only once that rejoin verified the epoch. + if state.initial_sync_done.load(Ordering::Acquire) && !state.compaction.is_rejoining() { + state.mark_ready(cluster_id.clone()).await?; + info!(cluster_id = %cluster_id, "Node is ready"); + return Ok(()); + } + } + + if tokio::time::Instant::now() >= deadline { + let mut peer_diagnostics: Vec = Vec::new(); + + let peer_states = state.peer_states.read().await.clone(); + + for (peer, peer_state_mutex) in &peer_states { + let peer_state = peer_state_mutex.lock().await; + let validated_cluster = match &peer_state.membership { + PeerMembership::Unknown => None, + PeerMembership::Member { cluster_id } => Some(cluster_id.as_str()), + }; + peer_diagnostics.push(format!( + "{peer}: validated_cluster={:?}, epoch={:?}, pending_inbound_sync={}", + validated_cluster, + peer_state.epoch, + peer_state.pending_inbound_sync.len() + )) + } + let local_cluster = state.cluster_id().await; + let local_epoch = state.local_epoch(); + let initial_sync_done = state.initial_sync_done.load(Ordering::Acquire); + error!( + local_cluster = ?local_cluster, + local_epoch = %local_epoch, + initial_sync_done, + peers = %peer_diagnostics.join("; "), + timeout_seconds = BOOTSTRAP_TIMEOUT.as_secs(), + "Bootstrap timed out" + ); + bail!( + "Bootstrap timed out after {:?}; local_cluster={:?}; local_epoch={}; initial_sync_done={}; peers=[{}]. Refusing to create a genesis document.", + BOOTSTRAP_TIMEOUT, + local_cluster, + local_epoch, + initial_sync_done, + peer_diagnostics.join("; ") + ); + } + tokio::time::sleep(BOOTSTRAP_RETRY).await; + } +} diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs new file mode 100644 index 0000000..7316146 --- /dev/null +++ b/src/swarm/mod.rs @@ -0,0 +1,6 @@ +//! The swarm of peers: bootstrap, handshake, keepalive and dormancy (`peer`), and +//! Automerge synchronization with every member (`sync`). + +pub(crate) mod bootstrap; +pub(crate) mod peer; +pub(crate) mod sync; diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs new file mode 100644 index 0000000..eb97a7e --- /dev/null +++ b/src/swarm/peer.rs @@ -0,0 +1,675 @@ +//! Peers: handshake (HELLO / HELLO_REPLY), liveness and keepalives, self-healing, +//! cluster-wide forget records, and dispatch of inbound messages. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, bail, Context, Result}; +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::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"; +const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5); + +// --- forget records ---------------------------------------------------------- + +fn forgotten_dir() -> Vec { + vec![META_ROOT.to_owned(), "forgotten".to_owned()] +} + +pub(crate) fn forgotten_path(pk: &str) -> Vec { + let mut path = forgotten_dir(); + path.push(pk.to_owned()); + path +} + +fn forgotten_peers(doc: &AutoCommit) -> Vec { + store::keys_at(doc, &forgotten_dir()).unwrap_or_default() +} + +async fn apply_forget_records(state: &AppState) { + // Before the initial sync we can't judge the records, and we have nothing to delete. + if !state.is_ready().await { + return; + } + + let forgotten = forgotten_peers(&state.doc.read().await.doc); + if forgotten.is_empty() { + return; + } + + let now = Instant::now(); + let mut cancel = Vec::new(); + + for pk in forgotten { + if pk == state.local_id() { + cancel.push(pk); // we are evidently alive + continue; + } + + let Some(peer_state_mutex) = state.get_peer_state(&pk).await else { + continue; // already forgotten here + }; + + let alive = peer_state_mutex.lock().await.is_alive(&state.timing, now); + + if alive { + cancel.push(pk); + } else if state.peer_states.write().await.remove(&pk).is_some() { + info!(peer = %pk, "Forgetting peer per cluster-wide record"); + } + } + + if cancel.is_empty() { + return; + } + + // Individual failures are logged inside; NotFound means another node already + // cancelled the record. + let _ = state + .mutate_doc(move |doc| { + for pk in &cancel { + match store::delete_path(&mut *doc, &forgotten_path(pk), false) { + Ok(()) => info!(peer = %pk, "Peer is alive; cancelling cluster-wide forget"), + Err(StoreError::NotFound) => {} + Err(error) => { + warn!(peer = %pk, error = %error, "Failed to cancel forget record") + } + } + } + Ok::<(), StoreError>(()) + }) + .await; +} + +// --- 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() +} + +pub(crate) async fn local_keepalive(state: &AppState) -> 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; + transport::Keepalive { + heads_digest: heads_digest(&mut replica.doc), + epoch: replica.epoch, + } +} + +async fn heal_if_diverged( + state: &AppState, + peer: &str, + local: &transport::Keepalive, + remote: &transport::Keepalive, +) { + // Digests of documents from different epochs are never equal and never converge. + if local.epoch != remote.epoch { + return; + } + + let Some(peer_state_mutex) = state.get_peer_state(peer).await else { + return; + }; + + let heal = peer_state_mutex.lock().await.needs_heal( + &local.heads_digest, + &remote.heads_digest, + &state.timing, + Instant::now(), + ); + + if heal { + warn!(peer = %peer, "Documents differ and sync is idle; re-handshaking to restart sync"); + ensure_handshake(state, peer).await; + } +} + +// --- peer discovery and HELLO ------------------------------------------------ + +pub(crate) async fn remember_peer(state: &AppState, peer: &str) { + if peer.is_empty() { + warn!("Ignoring empty peer public key"); + return; + } + + if peer == state.local_id() { + debug!("Ignoring own public key"); + return; + } + + let mut peer_states = state.peer_states.write().await; + + if !peer_states.contains_key(peer) { + peer_states.entry(peer.to_owned()).or_default(); + info!( + peer = %peer, + peer_count = peer_states.len(), + "Discovered peer" + ); + } else { + trace!(peer = %peer, "Peer already registered"); + } +} + +pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> { + let Some(peer_state_mutex) = state.get_peer_state(peer).await else { + trace!(peer = %peer, "Peer was forgotten; not sending HELLO"); + return Ok(()); + }; + + 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, + }; + true + } else { + trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending"); + false + } + }; + + if !claimed { + return Ok(()); + } + + debug!(to = %peer, "Sending HELLO"); + transmit_hello(state, peer, HELLO_RETRY_INITIAL).await +} + +async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { + let Some(peer_state_mutex) = state.get_peer_state(peer).await else { + 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 + } + _ => return Ok(()), + } + }; + + debug!( + peer = %peer, + next_retry_ms = delay.as_millis(), + "No HELLO_REPLY yet; re-sending HELLO" + ); + transmit_hello(state, peer, delay).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<()> { + let result = state + .transport + .send_message( + peer, + &ParsedMessage::Hello(transport::HelloRequest { + epoch: state.local_epoch(), + }), + ) + .await; + + 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)), + }; + peer_state.hello_state = HelloState::AwaitingReply { + next_attempt: Instant::now() + wait, + next_delay, + }; + } + } + + result.context("Sending HELLO") +} + +async fn ensure_handshake(state: &AppState, peer: &str) { + if let Err(error) = send_hello_once(state, peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); + } +} + +/// Sends a HELLO_REPLY if one is owed and our cluster 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.cluster_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 = 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<()> { + let cluster_id = state + .cluster_id() + .await + .ok_or_else(|| anyhow!("Cannot send HELLO_REPLY without an established cluster ID"))?; + + let candidates: Vec<(String, Arc>)> = state + .peer_states + .read() + .await + .iter() + .filter(|(known, _)| known.as_str() != peer) + .map(|(known, m)| (known.clone(), Arc::clone(m))) + .collect(); + + let mut peers = Vec::new(); + for (known, peer_state_mutex) in candidates { + let peer_state = peer_state_mutex.lock().await; + if peer_state.liveness == Liveness::Active + && peer_state.hello_state == HelloState::Established + { + peers.push(known); + } + } + + let reply = transport::HelloReply { + cluster_id, + peers, + epoch: state.local_epoch(), + }; + + let ready = state.is_ready().await; + + debug!(to = %peer, ready, "Sending HELLO_REPLY"); + + state + .transport + .send_message(peer, &ParsedMessage::HelloReply(reply)) + .await + .context("Sending HELLO_REPLY")?; + + trace!(to = %peer, "HELLO_REPLY sent"); + Ok(()) +} + +async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[String]) { + // TODO here we could add a trace t oshwo advertised_peers + for discovered_peer in advertised_peers { + if discovered_peer == from_peer { + continue; + } + + remember_peer(state, discovered_peer).await; + + if let Err(error) = send_hello_once(state, discovered_peer).await { + warn!( + peer = %discovered_peer, + error = %format!("{error:#}"), + "Failed to initiate HELLO with discovered peer" + ); + } + } +} + +// --- peer worker ------------------------------------------------------------- + +pub(crate) async fn run_peer_worker(state: AppState) { + let mut tick = tokio::time::interval(PEER_TICK); + tick.set_missed_tick_behavior(MissedTickBehavior::Delay); + + info!("Starting peer maintenance worker"); + + let mut last_forget_check: Option = None; + + loop { + tick.tick().await; + + // While joining, keep handshakes fast; dormancy would slow probes to minutes. + let ready = state.is_ready().await; + + let peers: Vec<(String, Arc>)> = state + .peer_states + .read() + .await + .iter() + .map(|(pk, m)| (pk.clone(), Arc::clone(m))) + .collect(); + + for (peer, peer_state_mutex) in &peers { + let now = Instant::now(); + + let (became_dormant, ping_due) = { + let mut peer_state = peer_state_mutex.lock().await; + let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); + let ping_due = peer_state.take_keepalive_due(&state.timing, now); + (became_dormant, ping_due) + }; + + if became_dormant { + warn!( + peer = %peer, + silent_secs = state.timing.dormant_after.as_secs(), + probe_secs = state.timing.dormant_probe.as_secs(), + "Peer went dormant" + ); + } + + if ping_due { + let local = local_keepalive(&state).await; + if let Err(error) = state + .transport + .send_message(peer, &ParsedMessage::Ping(local)) + .await + { + debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); + } + } + + 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; + last_forget_check = Some(Instant::now()); + } + } +} + +// --- peer membership --------------------------------------------------------- + +async fn establish_peer( + state: &AppState, + peer: &str, + cluster_id: String, +) -> Result> { + let expected_cluster = state.cluster_id().await; + + if let Some(expected) = &expected_cluster { + if expected != &cluster_id { + bail!( + "Incompatible cluster at {peer}: expected {}, received {}", + expected, + cluster_id + ); + } + } + + 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 { + cluster_id: existing, + } = &peer_state.membership + { + if existing != &cluster_id { + bail!( + "Peer {peer} changed cluster identity from {} to {}", + existing, + cluster_id + ); + } + } + + // Lock order: peer mutex -> lifecycle. + if expected_cluster.is_none() { + state.set_expected_cluster_id(cluster_id.clone()).await?; + } + + if matches!(peer_state.membership, PeerMembership::Unknown) { + peer_state.membership = PeerMembership::Member { + cluster_id: cluster_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 { + peer_state.hello_state = HelloState::Established; + peer_state.reset_sync(); + info!(peer = %peer, "Handshake complete; sync state reset"); + } + + let pending = std::mem::take(&mut peer_state.pending_inbound_sync); + + info!(peer = %peer, cluster_id = %cluster_id, pending_messages = pending.len(), "Peer membership established"); + Ok(pending) +} + +#[instrument(skip(state), fields(peer = %peer))] +pub(crate) async fn peer_cluster(state: &AppState, peer: &str) -> Option { + 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 { cluster_id } => Some(cluster_id.clone()), + } +} + +// --- epochs ------------------------------------------------------------------ + +/// Records the epoch a peer advertised and compares it with ours. Returns true when both +/// are on the same epoch, the only case in which the documents may be synced. +async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool { + let local = state.local_epoch(); + let relation = local.relation_to(&remote); + + let Some(peer_state_mutex) = state.get_peer_state(peer).await else { + return relation == EpochRelation::Same; + }; + + let previous = { + let mut peer_state = peer_state_mutex.lock().await; + let previous = peer_state.epoch.replace(remote); + if previous.is_some_and(|previous| previous != remote) { + // The peer switched documents; our sync state describes its old one. Count + // the restart as sync activity, so the keepalive heal doesn't re-handshake + // (resetting both sides again) while this fresh sync is under way. + peer_state.reset_sync(); + peer_state.last_sync_rx = Some(Instant::now()); + } + previous + }; + + // Log transitions only, not every keepalive. + if previous != Some(remote) { + match relation { + EpochRelation::Same => { + if previous.is_some() { + info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync"); + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())); + } + } + // Expected after a compaction: either side switches or catches up. + EpochRelation::PeerAhead | EpochRelation::PeerBehind => { + info!( + peer = %peer, + local_epoch = %local, + peer_epoch = %remote, + relation = ?relation, + "Peer is on another epoch; not syncing with it until both are on the same" + ); + } + EpochRelation::Diverged => { + warn!( + peer = %peer, + local_epoch = %local, + peer_epoch = %remote, + "Peer is on a different epoch of the same generation; not syncing with it" + ); + } + } + } + + relation == EpochRelation::Same +} + +// --- inbound dispatch -------------------------------------------------------- + +pub(crate) async fn handle_message( + state: &AppState, + peer: String, + message: transport::ParsedMessage, +) -> Result<()> { + if peer == state.local_id() { + debug!("Ignoring message from self"); + 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()); + + if reactivated { + info!(peer = %peer, "Dormant peer is active again; re-handshaking"); + } + + let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; + + //debug!(from = %peer, message = ?message, "Received message"); + + match message { + transport::ParsedMessage::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; + } + + // Sends now if we know our cluster 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"); + } + + // Restart sync with the fresh state (no-op until the peer is a validated member). + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.clone())); + } + transport::ParsedMessage::HelloReply(reply) => { + let pending = establish_peer(state, &peer, reply.cluster_id.clone()).await?; + + discover_peers(state, &peer, &reply.peers).await; + + // Our cluster 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())) + .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; + } + transport::ParsedMessage::Sync(envelope) => { + if reactivated { + ensure_handshake(state, &peer).await; + } + if !same_epoch { + debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); + return Ok(()); + } + handle_sync(state, &peer, envelope).await?; + } + transport::ParsedMessage::Ping(remote) => { + if reactivated || peer_cluster(state, &peer).await.is_none() { + ensure_handshake(state, &peer).await; + } + let local = local_keepalive(state).await; + heal_if_diverged(state, &peer, &local, &remote).await; + if let Err(error) = state + .transport + .send_message(&peer, &ParsedMessage::Pong(local)) + .await + { + debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); + } + } + transport::ParsedMessage::Pong(remote) => { + if reactivated || peer_cluster(state, &peer).await.is_none() { + ensure_handshake(state, &peer).await; + } + let local = local_keepalive(state).await; + heal_if_diverged(state, &peer, &local, &remote).await; + } + } + + Ok(()) +} diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs new file mode 100644 index 0000000..115d5c0 --- /dev/null +++ b/src/swarm/sync.rs @@ -0,0 +1,564 @@ +//! Automerge synchronization: the sync worker, delivery with retries, inbound sync +//! 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 anyhow::{anyhow, bail, Context, Result}; +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::swarm::peer::{peer_cluster, send_hello_once}; +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. +#[derive(Debug)] +pub(crate) struct ChangeSet { + pub(crate) paths: Vec>, +} + +async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> { + let Some(peer_state_mutex) = state.get_peer_state(peer).await else { + trace!("Peer was forgotten; skipping synchronization"); + return Ok(()); + }; + + let validated_cluster = { + let peer_state = peer_state_mutex.lock().await; + + if matches!(peer_state.liveness, Liveness::Dormant { .. }) { + // Reactivation re-handshakes, and PeerValidated then restarts sync. + trace!("Skipping synchronization with dormant peer"); + return Ok(()); + } + + if peer_state + .epoch + .is_some_and(|epoch| epoch != state.local_epoch()) + { + // Unrelated document histories must never be synced. + trace!("Skipping synchronization with peer on another epoch"); + return Ok(()); + } + + match &peer_state.membership { + PeerMembership::Unknown => return Ok(()), + PeerMembership::Member { cluster_id } => cluster_id.clone(), + } + }; + + let cluster_id = state + .cluster_id() + .await + .ok_or_else(|| anyhow!("Cannot sync without an established cluster ID"))?; + + if validated_cluster != cluster_id { + bail!("Peer {peer} belongs to a different cluster"); + } + + 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 encoded = match existing { + Some((message, next_attempt)) => { + if Instant::now() < next_attempt { + return Ok(()); + } + message + } + 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> { + // 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 sync_doc = replica.doc.sync(); + ( + epoch, + sync_doc.generate_sync_message(&mut peer_state.sync_state), + ) + }; + let Some(message) = message else { + return Ok(None); + }; + let envelope = transport::SyncEnvelope { + cluster_id, + epoch, + message, + }; + transport + .encode(&peer_owned, &ParsedMessage::Sync(envelope)) + .map(Some) + }, + ) + .await + .context("Sync generation task panicked")??; + + let Some(encoded) = generated else { + return Ok(()); + }; + + peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound { + message: encoded.clone(), + retry_delay: INITIAL_SYNC_RETRY, + next_attempt: Instant::now(), + }); + + 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) + }; + + trace!(regenerate, "Outbound sync message accepted by Mycelium"); + + if regenerate { + state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())) + .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; + } + + 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 + }; + + warn!( + retry_ms = delay.as_millis(), + error = %format!("{error:#}"), + "Sync delivery failed; retaining the same outbound envelope" + ); + + 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)); + }); + + Ok(()) + } + } +} + +pub(crate) async fn handle_sync( + state: &AppState, + peer: &str, + envelope: transport::SyncEnvelope, +) -> Result<()> { + debug!(peer = %peer, "Processing sync envelope"); + + let peer_state_mutex = state.get_or_create_peer_state(peer).await; + + let needs_hello = { + let mut 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.cluster_id == envelope.cluster_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"); + } + true + } + PeerMembership::Member { cluster_id } if cluster_id != &envelope.cluster_id => { + bail!( + "Sync cluster mismatch for peer {peer}: validated_cluster={cluster_id}, envelope_cluster={}", + envelope.cluster_id + ); + } + PeerMembership::Member { .. } => false, + } + }; + + if needs_hello { + // The peer treats us as a member but we don't know it, typically because we + // restarted. Our HELLO makes it reset its sync state and reply. + if let Err(error) = send_hello_once(state, peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to unknown sync sender failed; will retry"); + } + return Ok(()); + } + + let expected_cluster = state + .cluster_id() + .await + .ok_or_else(|| anyhow!("Missing expected cluster ID after peer membership"))?; + if envelope.cluster_id != expected_cluster { + bail!( + "Incompatible cluster from {peer}: expected {}, received {}", + expected_cluster, + envelope.cluster_id + ); + } + + let epoch_label = envelope.epoch; + + let applied = { + let doc = Arc::clone(&state.doc); + let peer_state_mutex = Arc::clone(&peer_state_mutex); + let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire); + 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> { + let mut replica = doc.blocking_write(); + + // Authoritative check, under the same lock as the apply: the document may + // have switched epochs since the message was accepted. + if replica.epoch != epoch { + return Ok(None); + } + + let mut peer_state = peer_state_mutex.blocking_lock(); + + 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) + .context("Automerge synchronization failed")?; + peer_state.last_sync_rx = Some(Instant::now()); + + 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); + + Ok(Some( + check_caught_up + && peer_state + .sync_state + .their_heads + .as_ref() + .is_some_and(|heads| { + heads + .iter() + .all(|h| replica.doc.get_change_by_hash(h).is_some()) + }), + )) + }) + .await + .context("Sync task panicked")?? + }; + + let Some(caught_up) = applied else { + debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); + return Ok(()); + }; + + if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { + info!(peer = %peer, "Initial sync complete"); + } + + state.notify_store_changed(); + + state + .sync_events + .send(SyncEvent::MessageReceived(peer.to_owned())) + .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; + + 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; + 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(); + } + _ => {} + } +} + +pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver) { + let mut dirty_peers = HashSet::::new(); + let mut retry_peers = HashSet::::new(); + let mut document_dirty = false; + + info!("Starting synchronization worker"); + + loop { + if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty { + match rx.recv().await { + Some(event) => match event { + SyncEvent::PeerValidated(peer) + | SyncEvent::MessageReceived(peer) + | SyncEvent::PeerNeedsSync(peer) => { + dirty_peers.insert(peer); + } + SyncEvent::RetryDelivery(peer) => { + retry_peers.insert(peer); + } + SyncEvent::DocumentChanged => { + document_dirty = true; + } + }, + None => { + warn!("Synchronization event channel closed"); + return; + } + } + } + + // Coalesce event bursts. + while let Ok(event) = rx.try_recv() { + match event { + SyncEvent::PeerValidated(peer) + | SyncEvent::MessageReceived(peer) + | SyncEvent::PeerNeedsSync(peer) => { + dirty_peers.insert(peer); + } + SyncEvent::RetryDelivery(peer) => { + retry_peers.insert(peer); + } + SyncEvent::DocumentChanged => { + document_dirty = true; + } + } + } + + // Let pending HTTP task locks wake up and complete their lifecycle transitions gracefully. + tokio::task::yield_now().await; + + if document_dirty { + document_dirty = false; + + // Lock and extract only the cloned string keys into a small transient vector + let peer_keys: Vec = { + let peers_guard = state.peer_states.read().await; + peers_guard.keys().cloned().collect() + }; + + for peer in peer_keys { + if peer_cluster(&state, &peer).await.is_some() { + dirty_peers.insert(peer); + } + } + } + + // Retry existing envelopes separately from generating new ones. + let retries: Vec = 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 = dirty_peers.drain().collect(); + + for peer in peers { + if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await { + warn!( + peer = %peer, + error = %format!("{error:#}"), + "Synchronization processing failed" + ); + } + } + } +} + +// --- change publisher -------------------------------------------------------- + +fn patch_store_path(patch: &automerge::Patch) -> Vec { + let mut path: Vec = patch + .path + .iter() + .map(|(_, prop)| match prop { + automerge::Prop::Map(key) => key.clone(), + automerge::Prop::Seq(index) => index.to_string(), + }) + .collect(); + + match &patch.action { + automerge::PatchAction::PutMap { key, .. } + | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()), + // List, text and counter changes: the containing object changed. + _ => {} + } + path +} + +pub(crate) fn paths_related(a: &[String], b: &[String]) -> bool { + a.starts_with(b) || b.starts_with(a) +} + +pub(crate) async fn run_change_publisher(state: AppState) { + let mut changed = state.change_tx.subscribe(); + let mut last_heads = state.write_doc(|doc| doc.get_heads()).await; + + info!("Starting change publisher"); + + while changed.changed().await.is_ok() { + let before = last_heads.clone(); + + let (heads, paths, conflicts) = state + .write_doc(move |doc| { + let heads = doc.get_heads(); + if heads == before { + return (heads, Vec::new(), Vec::new()); + } + + // After an epoch switch the previous heads belong to the old document: + // report the root as changed, which every stream is related to. + if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) { + return (heads, vec![Vec::new()], Vec::new()); + } + + let mut paths = Vec::new(); + let mut conflicts = Vec::new(); + for patch in doc.diff(&before, &heads) { + let path = patch_store_path(&patch); + if matches!( + patch.action, + automerge::PatchAction::PutMap { conflict: true, .. } + ) { + conflicts.push(path.clone()); + } + paths.push(path); + } + (heads, paths, conflicts) + }) + .await; + + if heads == last_heads { + continue; + } + last_heads = heads; + + for path in &conflicts { + warn!(path = ?path, "Concurrent writes produced a conflict"); + } + + // Fails only when nobody is subscribed. + let _ = state.change_events.send(Arc::new(ChangeSet { paths })); + } +} diff --git a/src/sync.rs b/src/sync.rs deleted file mode 100644 index 9db4b40..0000000 --- a/src/sync.rs +++ /dev/null @@ -1,564 +0,0 @@ -//! Automerge synchronization: the sync worker, delivery with retries, inbound sync -//! 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 anyhow::{anyhow, bail, Context, Result}; -use automerge::sync::SyncDoc; -use tokio::sync::mpsc; -use tokio::time::Instant; -use tracing::{debug, info, trace, warn}; - -use crate::peer::{peer_cluster, send_hello_once}; -use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound}; -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. -#[derive(Debug)] -pub(crate) struct ChangeSet { - pub(crate) paths: Vec>, -} - -async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> { - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - trace!("Peer was forgotten; skipping synchronization"); - return Ok(()); - }; - - let validated_cluster = { - let peer_state = peer_state_mutex.lock().await; - - if matches!(peer_state.liveness, Liveness::Dormant { .. }) { - // Reactivation re-handshakes, and PeerValidated then restarts sync. - trace!("Skipping synchronization with dormant peer"); - return Ok(()); - } - - if peer_state - .epoch - .is_some_and(|epoch| epoch != state.local_epoch()) - { - // Unrelated document histories must never be synced. - trace!("Skipping synchronization with peer on another epoch"); - return Ok(()); - } - - match &peer_state.membership { - PeerMembership::Unknown => return Ok(()), - PeerMembership::Member { cluster_id } => cluster_id.clone(), - } - }; - - let cluster_id = state - .cluster_id() - .await - .ok_or_else(|| anyhow!("Cannot sync without an established cluster ID"))?; - - if validated_cluster != cluster_id { - bail!("Peer {peer} belongs to a different cluster"); - } - - 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 encoded = match existing { - Some((message, next_attempt)) => { - if Instant::now() < next_attempt { - return Ok(()); - } - message - } - 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> { - // 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 sync_doc = replica.doc.sync(); - ( - epoch, - sync_doc.generate_sync_message(&mut peer_state.sync_state), - ) - }; - let Some(message) = message else { - return Ok(None); - }; - let envelope = transport::SyncEnvelope { - cluster_id, - epoch, - message, - }; - transport - .encode(&peer_owned, &ParsedMessage::Sync(envelope)) - .map(Some) - }, - ) - .await - .context("Sync generation task panicked")??; - - let Some(encoded) = generated else { - return Ok(()); - }; - - peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound { - message: encoded.clone(), - retry_delay: INITIAL_SYNC_RETRY, - next_attempt: Instant::now(), - }); - - 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) - }; - - trace!(regenerate, "Outbound sync message accepted by Mycelium"); - - if regenerate { - state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.to_owned())) - .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; - } - - 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 - }; - - warn!( - retry_ms = delay.as_millis(), - error = %format!("{error:#}"), - "Sync delivery failed; retaining the same outbound envelope" - ); - - 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)); - }); - - Ok(()) - } - } -} - -pub(crate) async fn handle_sync( - state: &AppState, - peer: &str, - envelope: transport::SyncEnvelope, -) -> Result<()> { - debug!(peer = %peer, "Processing sync envelope"); - - let peer_state_mutex = state.get_or_create_peer_state(peer).await; - - let needs_hello = { - let mut 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.cluster_id == envelope.cluster_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"); - } - true - } - PeerMembership::Member { cluster_id } if cluster_id != &envelope.cluster_id => { - bail!( - "Sync cluster mismatch for peer {peer}: validated_cluster={cluster_id}, envelope_cluster={}", - envelope.cluster_id - ); - } - PeerMembership::Member { .. } => false, - } - }; - - if needs_hello { - // The peer treats us as a member but we don't know it, typically because we - // restarted. Our HELLO makes it reset its sync state and reply. - if let Err(error) = send_hello_once(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to unknown sync sender failed; will retry"); - } - return Ok(()); - } - - let expected_cluster = state - .cluster_id() - .await - .ok_or_else(|| anyhow!("Missing expected cluster ID after peer membership"))?; - if envelope.cluster_id != expected_cluster { - bail!( - "Incompatible cluster from {peer}: expected {}, received {}", - expected_cluster, - envelope.cluster_id - ); - } - - let epoch_label = envelope.epoch; - - let applied = { - let doc = Arc::clone(&state.doc); - let peer_state_mutex = Arc::clone(&peer_state_mutex); - let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire); - 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> { - let mut replica = doc.blocking_write(); - - // Authoritative check, under the same lock as the apply: the document may - // have switched epochs since the message was accepted. - if replica.epoch != epoch { - return Ok(None); - } - - let mut peer_state = peer_state_mutex.blocking_lock(); - - 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) - .context("Automerge synchronization failed")?; - peer_state.last_sync_rx = Some(Instant::now()); - - 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); - - Ok(Some( - check_caught_up - && peer_state - .sync_state - .their_heads - .as_ref() - .is_some_and(|heads| { - heads - .iter() - .all(|h| replica.doc.get_change_by_hash(h).is_some()) - }), - )) - }) - .await - .context("Sync task panicked")?? - }; - - let Some(caught_up) = applied else { - debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch"); - return Ok(()); - }; - - if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) { - info!(peer = %peer, "Initial sync complete"); - } - - state.notify_store_changed(); - - state - .sync_events - .send(SyncEvent::MessageReceived(peer.to_owned())) - .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; - - 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; - 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(); - } - _ => {} - } -} - -pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver) { - let mut dirty_peers = HashSet::::new(); - let mut retry_peers = HashSet::::new(); - let mut document_dirty = false; - - info!("Starting synchronization worker"); - - loop { - if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty { - match rx.recv().await { - Some(event) => match event { - SyncEvent::PeerValidated(peer) - | SyncEvent::MessageReceived(peer) - | SyncEvent::PeerNeedsSync(peer) => { - dirty_peers.insert(peer); - } - SyncEvent::RetryDelivery(peer) => { - retry_peers.insert(peer); - } - SyncEvent::DocumentChanged => { - document_dirty = true; - } - }, - None => { - warn!("Synchronization event channel closed"); - return; - } - } - } - - // Coalesce event bursts. - while let Ok(event) = rx.try_recv() { - match event { - SyncEvent::PeerValidated(peer) - | SyncEvent::MessageReceived(peer) - | SyncEvent::PeerNeedsSync(peer) => { - dirty_peers.insert(peer); - } - SyncEvent::RetryDelivery(peer) => { - retry_peers.insert(peer); - } - SyncEvent::DocumentChanged => { - document_dirty = true; - } - } - } - - // Let pending HTTP task locks wake up and complete their lifecycle transitions gracefully. - tokio::task::yield_now().await; - - if document_dirty { - document_dirty = false; - - // Lock and extract only the cloned string keys into a small transient vector - let peer_keys: Vec = { - let peers_guard = state.peer_states.read().await; - peers_guard.keys().cloned().collect() - }; - - for peer in peer_keys { - if peer_cluster(&state, &peer).await.is_some() { - dirty_peers.insert(peer); - } - } - } - - // Retry existing envelopes separately from generating new ones. - let retries: Vec = 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 = dirty_peers.drain().collect(); - - for peer in peers { - if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await { - warn!( - peer = %peer, - error = %format!("{error:#}"), - "Synchronization processing failed" - ); - } - } - } -} - -// --- change publisher -------------------------------------------------------- - -fn patch_store_path(patch: &automerge::Patch) -> Vec { - let mut path: Vec = patch - .path - .iter() - .map(|(_, prop)| match prop { - automerge::Prop::Map(key) => key.clone(), - automerge::Prop::Seq(index) => index.to_string(), - }) - .collect(); - - match &patch.action { - automerge::PatchAction::PutMap { key, .. } - | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()), - // List, text and counter changes: the containing object changed. - _ => {} - } - path -} - -pub(crate) fn paths_related(a: &[String], b: &[String]) -> bool { - a.starts_with(b) || b.starts_with(a) -} - -pub(crate) async fn run_change_publisher(state: AppState) { - let mut changed = state.change_tx.subscribe(); - let mut last_heads = state.write_doc(|doc| doc.get_heads()).await; - - info!("Starting change publisher"); - - while changed.changed().await.is_ok() { - let before = last_heads.clone(); - - let (heads, paths, conflicts) = state - .write_doc(move |doc| { - let heads = doc.get_heads(); - if heads == before { - return (heads, Vec::new(), Vec::new()); - } - - // After an epoch switch the previous heads belong to the old document: - // report the root as changed, which every stream is related to. - if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) { - return (heads, vec![Vec::new()], Vec::new()); - } - - let mut paths = Vec::new(); - let mut conflicts = Vec::new(); - for patch in doc.diff(&before, &heads) { - let path = patch_store_path(&patch); - if matches!( - patch.action, - automerge::PatchAction::PutMap { conflict: true, .. } - ) { - conflicts.push(path.clone()); - } - paths.push(path); - } - (heads, paths, conflicts) - }) - .await; - - if heads == last_heads { - continue; - } - last_heads = heads; - - for path in &conflicts { - warn!(path = ?path, "Concurrent writes produced a conflict"); - } - - // Fails only when nobody is subscribed. - let _ = state.change_events.send(Arc::new(ChangeSet { paths })); - } -} diff --git a/src/transport.rs b/src/transport.rs deleted file mode 100644 index 9cfb026..0000000 --- a/src/transport.rs +++ /dev/null @@ -1,95 +0,0 @@ -use anyhow::Result; -use async_trait::async_trait; -use automerge::sync::Message; -use std::boxed::Box; -use std::future::Future; -use std::pin::Pin; - -pub use crate::replica::Epoch; - -pub type HeadsDigest = [u8; 32]; - -#[derive(Debug, Clone)] -pub struct Keepalive { - pub heads_digest: HeadsDigest, - /// The digest is only comparable between nodes on the same epoch. - pub epoch: Epoch, -} - -#[derive(Debug)] -pub enum ParsedMessage { - Hello(HelloRequest), - HelloReply(HelloReply), - Sync(SyncEnvelope), - Ping(Keepalive), - Pong(Keepalive), -} - -impl ParsedMessage { - /// Every message carries the sender's epoch. - pub fn epoch(&self) -> Epoch { - match self { - ParsedMessage::Hello(hello) => hello.epoch, - ParsedMessage::HelloReply(reply) => reply.epoch, - ParsedMessage::Sync(envelope) => envelope.epoch, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, - } - } -} - -#[derive(Debug, Clone)] -pub struct HelloRequest { - pub epoch: Epoch, -} - -#[derive(Debug, Clone)] -pub struct HelloReply { - pub cluster_id: String, - pub peers: Vec, - pub epoch: Epoch, -} - -#[derive(Debug, Clone)] -pub struct SyncEnvelope { - pub cluster_id: String, - /// Epoch of the document the message was generated from. - pub epoch: Epoch, - pub message: Message, -} - -/// A message serialized for the wire. Cloning is cheap (shared buffer). -#[derive(Clone, Debug)] -pub struct EncodedMessage { - pub peer: String, - pub body: bytes::Bytes, -} - -pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin> + Send>> - + Send - + Sync - + 'static; -pub type MessageHandler = Box; - -#[async_trait] -pub trait Transport: Send + Sync { - /// This node's identity on the transport (for Mycelium: its public key). - fn local_id(&self) -> &str; - - /// Resolves a user-supplied peer hint (e.g. a seed) to a peer identity. - async fn resolve_peer(&self, hint: &str) -> Result; - - fn encode(&self, peer: &str, message: &ParsedMessage) -> Result; - async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>; - - /// Sends an outbound message payload to a targeted peer on a specific topic. - async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> { - let encoded = self.encode(peer, message)?; - self.send_encoded(&encoded).await - } - - /// Starts the long-polling inbound event loops, driving messages to the provided handler closure. - async fn start_consumer(&self, handler: MessageHandler) -> Result<()>; - - /// Asynchronously monitors for fatal network client or sub-process failures. - async fn wait_for_exit(&self) -> std::io::Result; -} diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs new file mode 100644 index 0000000..917fe5f --- /dev/null +++ b/src/transport/firewall.rs @@ -0,0 +1,105 @@ +use std::collections::HashSet; +use std::sync::{Arc, RwLock, RwLockReadGuard}; + +use anyhow::Result; +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::transport::EncodedMessage; +use crate::transport::{MessageHandler, ParsedMessage, Transport}; + +#[derive(Debug, Clone, Default, Serialize, Deserialize, ToSchema)] +pub struct FirewallRules { + /// Messages from these peers are dropped before the application sees them. + #[serde(default)] + pub inbound: HashSet, + /// Messages to these peers are silently dropped, like a lossy network. + #[serde(default)] + pub outbound: HashSet, +} + +/// Uses a std lock: checks are tiny and the guard is never held across an await. +#[derive(Default)] +pub struct Firewall { + rules: RwLock, +} + +impl Firewall { + fn read(&self) -> RwLockReadGuard<'_, FirewallRules> { + self.rules + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + pub fn rules(&self) -> FirewallRules { + self.read().clone() + } + + pub fn set_rules(&self, rules: FirewallRules) { + *self + .rules + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = rules; + } + + pub fn blocks_inbound(&self, peer: &str) -> bool { + self.read().inbound.contains(peer) + } + + pub fn blocks_outbound(&self, peer: &str) -> bool { + self.read().outbound.contains(peer) + } +} + +pub struct FirewallTransport { + inner: Arc, + firewall: Arc, +} + +impl FirewallTransport { + pub fn new(inner: Arc, firewall: Arc) -> Self { + Self { inner, firewall } + } +} + +#[async_trait] +impl Transport for FirewallTransport { + fn local_id(&self) -> &str { + self.inner.local_id() + } + + async fn resolve_peer(&self, hint: &str) -> Result { + self.inner.resolve_peer(hint).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { + self.inner.encode(peer, message) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + if self.firewall.blocks_outbound(&message.peer) { + tracing::trace!(peer = %message.peer, "Firewall dropped outbound message"); + return Ok(()); + } + self.inner.send_encoded(message).await + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let firewall = Arc::clone(&self.firewall); + + self.inner + .start_consumer(Box::new(move |sender, message| { + if firewall.blocks_inbound(&sender) { + tracing::trace!(peer = %sender, "Firewall dropped inbound message"); + return Box::pin(async { Ok(()) }); + } + handler(sender, message) + })) + .await + } + + async fn wait_for_exit(&self) -> std::io::Result { + self.inner.wait_for_exit().await + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..6accf81 --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,98 @@ +use anyhow::Result; +use async_trait::async_trait; +use automerge::sync::Message; +use std::boxed::Box; +use std::future::Future; +use std::pin::Pin; + +pub use crate::document::replica::Epoch; + +pub mod firewall; +pub mod mycelium; + +pub type HeadsDigest = [u8; 32]; + +#[derive(Debug, Clone)] +pub struct Keepalive { + pub heads_digest: HeadsDigest, + /// The digest is only comparable between nodes on the same epoch. + pub epoch: Epoch, +} + +#[derive(Debug)] +pub enum ParsedMessage { + Hello(HelloRequest), + HelloReply(HelloReply), + Sync(SyncEnvelope), + Ping(Keepalive), + Pong(Keepalive), +} + +impl ParsedMessage { + /// Every message carries the sender's epoch. + pub fn epoch(&self) -> Epoch { + match self { + ParsedMessage::Hello(hello) => hello.epoch, + ParsedMessage::HelloReply(reply) => reply.epoch, + ParsedMessage::Sync(envelope) => envelope.epoch, + ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, + } + } +} + +#[derive(Debug, Clone)] +pub struct HelloRequest { + pub epoch: Epoch, +} + +#[derive(Debug, Clone)] +pub struct HelloReply { + pub cluster_id: String, + pub peers: Vec, + pub epoch: Epoch, +} + +#[derive(Debug, Clone)] +pub struct SyncEnvelope { + pub cluster_id: String, + /// Epoch of the document the message was generated from. + pub epoch: Epoch, + pub message: Message, +} + +/// A message serialized for the wire. Cloning is cheap (shared buffer). +#[derive(Clone, Debug)] +pub struct EncodedMessage { + pub peer: String, + pub body: bytes::Bytes, +} + +pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin> + Send>> + + Send + + Sync + + 'static; +pub type MessageHandler = Box; + +#[async_trait] +pub trait Transport: Send + Sync { + /// This node's identity on the transport (for Mycelium: its public key). + fn local_id(&self) -> &str; + + /// Resolves a user-supplied peer hint (e.g. a seed) to a peer identity. + async fn resolve_peer(&self, hint: &str) -> Result; + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result; + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>; + + /// Sends an outbound message payload to a targeted peer on a specific topic. + async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> { + let encoded = self.encode(peer, message)?; + self.send_encoded(&encoded).await + } + + /// Starts the long-polling inbound event loops, driving messages to the provided handler closure. + async fn start_consumer(&self, handler: MessageHandler) -> Result<()>; + + /// Asynchronously monitors for fatal network client or sub-process failures. + async fn wait_for_exit(&self) -> std::io::Result; +} diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs new file mode 100644 index 0000000..d2deebe --- /dev/null +++ b/src/transport/mycelium.rs @@ -0,0 +1,964 @@ +use anyhow::{anyhow, Context, Result}; +use async_trait::async_trait; +use base64::engine::{general_purpose::STANDARD as BASE64, Engine}; +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::Keepalive; +use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; +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, + pub key_file_opt: Option, + pub network_key_file: Option, + pub network_name: String, + pub tcp_port: Option, + pub api_port: Option, + pub rpc_port: Option, + pub discovery_port: Option, + pub peers: Vec, +} + +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), + 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 { + // 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 { + match &mut self.runtime { + MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, + MyceliumRuntime::External => { + std::future::pending::>().await + } + } + } +} + +#[derive(Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct MyceliumAdminInfo { + pub node_ip: String, + pub node_pubkey: String, + pub node_start_time: u64, + pub node_subnet: String, +} + +pub async fn fetch_admin_info(rest_url: &Url) -> Result { + let url = rest_url.join("/api/v1/admin")?; + let info = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()? + .get(url) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(info) +} + +#[derive(Debug, Clone)] +pub struct ClusterNodeConfig { + pub tcp_port: u16, + pub api_port: u16, + pub rpc_port: u16, + pub discovery_port: Option, + pub private_network: Option, +} + +pub struct SupervisedMycelium { + child: Child, + pub config: ClusterNodeConfig, + _key_file_handle: Option, + _net_key_file_handle: Option, + key_file_path: PathBuf, + net_key_file_path: Option, + is_ephemeral_key: bool, + is_ephemeral_net_key: bool, +} + +impl SupervisedMycelium { + async fn spawn( + node_key: KeySource, + private_network: Option, + explicit_tcp: Option, + explicit_api: Option, + explicit_rpc: Option, + explicit_discovery_port: Option, + peers: &[String], + ) -> Result { + 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 = ClusterNodeConfig { + 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, is_ephemeral_key) = + Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?; + + let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) = + match private_network { + Some(net_cfg) => { + let (h, p, fd, eph) = + Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?; + (h, Some(p), fd, eph) + } + None => (None, None, None, false), + }; + + 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 = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); + if !fds_to_inherit.is_empty() { + unsafe { + cmd.pre_exec(move || { + for &fd in &fds_to_inherit { + libc::fcntl(fd, libc::F_SETFD, 0); + } + 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, + key_file_path, + net_key_file_path, + is_ephemeral_key, + is_ephemeral_net_key, + }; + + //let admin_info = node.poll_admin_endpoint().await?; + //Ok((node, admin_info)) + Ok(node) + } + + fn resolve_key_strategy( + desc_tag: &str, + source: KeySource, + parent_pid: u32, + ) -> Result<(Option, PathBuf, Option, bool)> { + 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), true)) + } + KeySource::Path(existing_path) => Ok((None, existing_path, None, false)), + } + } + + fn find_available_port() -> Result { + 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)) + } + + #[cfg(not(target_os = "linux"))] + { + let mut path = std::env::tempdir(); + path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4())); + let mut file = File::create(&path)?; + file.write_all(&raw_key)?; + file.flush()?; + Ok((file, -1)) + } + } + + //async fn poll_admin_endpoint(&mut self) -> Result { + // let client = reqwest::Client::new(); + // let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port); + // + // for _ in 0..30 { + // if let Some(status) = self.child.try_wait()? { + // return Err(anyhow!( + // "myceliumd process closed prematurely with status: {}", + // status + // )); + // } + // if let Ok(resp) = client.get(&url).send().await { + // if resp.status().is_success() { + // let info: MyceliumAdminInfo = resp.json().await?; + // return Ok(info); + // } + // } + // tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + // } + // Err(anyhow!( + // "Timeout waiting for local mycelium administration socket to respond" + // )) + //} + + pub async fn wait_for_exit(&mut self) -> std::io::Result { + self.child.wait().await + } +} + +impl Drop for SupervisedMycelium { + fn drop(&mut self) { + let _ = self.child.start_kill(); + + #[cfg(not(target_os = "linux"))] + { + if self.is_ephemeral_key { + let _ = std::fs::remove_file(&self.key_file_path); + } + if self.is_ephemeral_net_key { + if let Some(ref path) = self.net_key_file_path { + let _ = std::fs::remove_file(path); + } + } + } + } +} + +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, + 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 { + 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 { + #[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_peer(&self, hint: &str) -> Result { + if is_pubkey(hint) { + return Ok(hint.to_owned()); + } + + if let Ok(ip) = hint.parse::() { + 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 { + 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::(&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::(&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::(&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 { + 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 { + let bytes = BASE64.decode(topic)?; + Ok(String::from_utf8(bytes)?) +} + +fn encode_payload(payload: &Vec) -> String { + BASE64.encode(payload) +} + +fn decode_payload(topic_encoded: &str, payload: &str) -> Result { + 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 { + 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 { + 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, + }) +} + +#[derive(Serialize, Deserialize)] +struct KeepaliveShadow { + heads_digest: ByteBuf, + #[serde(default)] + epoch: EpochShadow, +} + +fn keepalive_to_bytes(msg: &Keepalive) -> Result, rmp_serde::encode::Error> { + rmp_serde::to_vec(&KeepaliveShadow { + heads_digest: ByteBuf::from(msg.heads_digest.to_vec()), + epoch: epoch_to_shadow(&msg.epoch), + }) +} + +fn bytes_to_keepalive(bytes: &[u8]) -> Result { + 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()))?; + Ok(Keepalive { + heads_digest, + epoch: epoch_from_shadow(shadow.epoch)?, + }) +} + +#[derive(Serialize, Deserialize)] +pub struct HelloRequestShadow { + #[serde(default)] + epoch: EpochShadow, +} + +#[derive(Serialize, Deserialize)] +pub struct HelloReplyShadow<'a> { + // Cow allows zero-copy references when saving, but creates an owned String when loading + pub cluster_id: Cow<'a, str>, + #[serde(default)] + pub peers: Vec, + #[serde(default)] + epoch: EpochShadow, +} + +#[derive(Serialize, Deserialize)] +pub struct SyncEnvelopeShadow<'a> { + pub cluster_id: Cow<'a, str>, + #[serde(with = "automerge_msg_codec")] + pub message: automerge::sync::Message, + #[serde(default)] + epoch: EpochShadow, +} + +// ===================================================================== +// 2. CUSTOM CODEC +// ===================================================================== +mod automerge_msg_codec { + use super::*; + use automerge::sync::Message; + + pub fn serialize(msg: &Message, serializer: S) -> Result + where + S: Serializer, + { + let bytes = msg.clone().encode(); + Bytes::new(&bytes).serialize(serializer) + } + + pub fn deserialize<'de, D>(deserializer: D) -> Result + 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, rmp_serde::encode::Error> { + rmp_serde::to_vec(&HelloRequestShadow { + epoch: epoch_to_shadow(&msg.epoch), + }) +} +pub fn bytes_to_hello_request(bytes: &[u8]) -> Result { + let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; + Ok(HelloRequest { + epoch: epoch_from_shadow(shadow.epoch)?, + }) +} + +// --- HelloReply --- +pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result, rmp_serde::encode::Error> { + let shadow = HelloReplyShadow { + cluster_id: Cow::Borrowed(&msg.cluster_id), + peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching + epoch: epoch_to_shadow(&msg.epoch), + }; + rmp_serde::to_vec(&shadow) +} +pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result { + let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?; + Ok(HelloReply { + cluster_id: shadow.cluster_id.into_owned(), + peers: shadow.peers, + epoch: epoch_from_shadow(shadow.epoch)?, + }) +} + +// --- SyncEnvelope --- +pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result, rmp_serde::encode::Error> { + let shadow = SyncEnvelopeShadow { + cluster_id: Cow::Borrowed(&msg.cluster_id), + message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules + epoch: epoch_to_shadow(&msg.epoch), + }; + rmp_serde::to_vec(&shadow) +} +pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result { + let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?; + Ok(SyncEnvelope { + cluster_id: shadow.cluster_id.into_owned(), + epoch: epoch_from_shadow(shadow.epoch)?, + message: shadow.message, + }) +} + +#[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, + }, +} -- cgit v1.3.1