summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/compaction/api.rs7
-rw-r--r--src/compaction/catchup.rs307
-rw-r--r--src/compaction/delta.rs122
-rw-r--r--src/compaction/local.rs78
-rw-r--r--src/compaction/mod.rs18
-rw-r--r--src/compaction/records.rs2
-rw-r--r--src/compaction/snapshot.rs118
-rw-r--r--src/compaction/switch.rs72
-rw-r--r--src/main.rs9
-rw-r--r--src/mycelium.rs60
-rw-r--r--src/peer.rs11
-rw-r--r--src/state.rs30
-rw-r--r--src/transport.rs31
13 files changed, 831 insertions, 34 deletions
diff --git a/src/compaction/api.rs b/src/compaction/api.rs
index 9a77e4d..f5c01f2 100644
--- a/src/compaction/api.rs
+++ b/src/compaction/api.rs
@@ -7,7 +7,8 @@
//! * `GET /admin/compactions` lists all compactions.
//! * `GET /admin/compactions/{id}` reports one; `?wait=true` long-polls until decided
//! (bounded by the proposal's own expiry).
-//! * `GET /admin/epochs` shows the local epoch and the archived previous ones.
+//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
+//! the node is catching up with a newer epoch from a peer.
//! * `DELETE /admin/compactions/{id}` asks the coordinator to abort. Only possible while
//! proposed; `409` once committed (a switch can't be undone) or already over.
//! `DELETE /admin/compaction` does the same for the compaction of the local epoch that
@@ -442,12 +443,14 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
}
}
-/// `GET /admin/epochs`: the local epoch and the previous epochs this node still keeps.
+/// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and
+/// an ongoing catch-up with a peer's newer epoch (`null` if none).
pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
let current = EpochRecord::from(host.current_epoch());
Json(json!({
"current": current,
"archived": host.compaction().archived_epochs(),
+ "catchUp": host.compaction().catch_up_report(),
}))
.into_response()
}
diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs
new file mode 100644
index 0000000..f166f56
--- /dev/null
+++ b/src/compaction/catchup.rs
@@ -0,0 +1,307 @@
+//! Catching up with a newer epoch from a peer.
+//!
+//! Peers on another epoch don't sync, so a node that missed a switch (new, late, or
+//! one whose own build failed) can't learn the new epoch through ordinary sync. Instead,
+//! once an established peer advertises a newer generation, the node asks it for its
+//! whole document (`SnapshotRequest`), checks that the document starts with the
+//! advertised snapshot change, and switches to it:
+//!
+//! * a node that wrote nothing the snapshot lacks (e.g. a brand-new node) adopts the
+//! document as is, whatever the gap;
+//! * a node exactly one epoch behind replays its own changes the snapshot lacks, found
+//! with the snapshot's per-actor clock;
+//! * a node further behind with such changes stays where it is and reports it; that is
+//! the returning-node case (replay window or quarantine, step 7).
+//!
+//! Only one fetch runs at a time, and the worker retries after `RETRY`. The reply is the
+//! peer's full document in one message; chunking comes later.
+
+use std::time::Duration;
+
+use automerge::{ActorId, AutoCommit};
+use tokio::time::Instant;
+use tracing::{debug, error, info, warn};
+
+use super::delta::{self, ReplayStats};
+use super::local::{CatchUp, LocalStatus};
+use super::records::unix_millis;
+use super::snapshot;
+use super::switch::swap_in;
+use super::Host;
+use crate::replica::Epoch;
+use crate::transport::{ParsedMessage, SnapshotReply, SnapshotRequest};
+
+const RETRY: Duration = Duration::from_secs(10);
+const FETCH_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Called by the switch worker when there is nothing to switch to locally.
+pub(super) async fn check<H: Host>(host: &H) {
+ let local = host.current_epoch();
+
+ let ahead: Vec<(String, Epoch)> = host
+ .peer_summaries()
+ .await
+ .into_iter()
+ .filter(|peer| peer.active && peer.established)
+ .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
+ .filter(|(_, epoch)| epoch.generation > local.generation)
+ .collect();
+
+ let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
+ if host.compaction().catch_up().is_some() {
+ host.compaction().set_catch_up(None);
+ }
+ return;
+ };
+
+ let mut targets: Vec<Epoch> = ahead
+ .iter()
+ .filter(|(_, epoch)| epoch.generation == generation)
+ .map(|(_, epoch)| *epoch)
+ .collect();
+ targets.sort_by_key(|epoch| epoch.id);
+ targets.dedup();
+ let [target] = targets.as_slice() else {
+ let listed: Vec<String> = targets.iter().map(ToString::to_string).collect();
+ warn!(epochs = ?listed, "Peers disagree about the newest epoch; not catching up");
+ return;
+ };
+ let target = *target;
+
+ // Back off between attempts for the same target.
+ let previous = host
+ .compaction()
+ .catch_up()
+ .filter(|catch_up| catch_up.target == target);
+ if previous
+ .as_ref()
+ .is_some_and(|catch_up| catch_up.last_attempt.elapsed() < RETRY)
+ {
+ return;
+ }
+ let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
+
+ // Rotate through the peers on the target epoch.
+ let sources: Vec<&String> = ahead
+ .iter()
+ .filter(|(_, epoch)| *epoch == target)
+ .map(|(peer, _)| peer)
+ .collect();
+ let source = sources[(attempts as usize - 1) % sources.len()].clone();
+
+ host.compaction().set_catch_up(Some(CatchUp {
+ target,
+ attempts,
+ last_attempt: Instant::now(),
+ error: None,
+ }));
+
+ info!(local = %local, target = %target, peer = %source, attempt = attempts, "Fetching newer epoch from peer");
+
+ let result = match fetch(host, &source, local, target).await {
+ Ok(document) => adopt(host, local, target, document).await,
+ Err(error) => Err(error),
+ };
+
+ match result {
+ Ok(()) => host.compaction().set_catch_up(None),
+ Err(error) => {
+ warn!(target = %target, peer = %source, error = %error, "Catching up failed; will retry");
+ host.compaction().set_catch_up(Some(CatchUp {
+ target,
+ attempts,
+ last_attempt: Instant::now(),
+ error: Some(error),
+ }));
+ }
+ }
+}
+
+async fn fetch<H: Host>(
+ host: &H,
+ peer: &str,
+ local: Epoch,
+ target: Epoch,
+) -> Result<Vec<u8>, String> {
+ let reply = host.compaction().await_reply(peer, target);
+ let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
+ epoch: local,
+ wanted: target,
+ });
+ if let Err(error) = host.send_message(peer, request).await {
+ host.compaction().stop_awaiting();
+ return Err(format!("sending the request failed: {error:#}"));
+ }
+
+ let reply = match tokio::time::timeout(FETCH_TIMEOUT, reply).await {
+ Ok(Ok(reply)) => reply,
+ Ok(Err(_)) => return Err("request superseded".to_owned()),
+ Err(_) => {
+ host.compaction().stop_awaiting();
+ return Err(format!("no reply within {}s", FETCH_TIMEOUT.as_secs()));
+ }
+ };
+
+ if reply.epoch != target || reply.document.is_empty() {
+ return Err(format!(
+ "peer is on {} and did not send {target}",
+ reply.epoch
+ ));
+ }
+ Ok(reply.document)
+}
+
+/// Verifies the fetched document, carries over this node's own missing changes, and
+/// switches to it.
+async fn adopt<H: Host>(
+ host: &H,
+ local: Epoch,
+ target: Epoch,
+ document: Vec<u8>,
+) -> Result<(), String> {
+ // Copy under the read lock only; everything else happens without a lock.
+ let copy = host.read_doc(|doc| doc.clone()).await;
+ let bytes = document.len();
+
+ let prepared = tokio::task::spawn_blocking(move || -> Result<Prepared, String> {
+ let mut fetched =
+ AutoCommit::load(&document).map_err(|error| format!("invalid document: {error}"))?;
+ let metadata = snapshot::verify_document(&mut fetched, target)?;
+
+ let mut copy = copy;
+ let own = copy.get_actor().clone();
+ let copy_heads = copy.get_heads();
+ let had_history = !copy_heads.is_empty();
+ let missing = delta::missing_from_snapshot(&mut copy, &metadata.clock, &own);
+
+ let mut doc = fetched.with_actor(ActorId::random());
+ let stats = if missing.own == 0 {
+ ReplayStats::default()
+ } else if metadata.previous == local {
+ delta::replay(&mut copy, &missing.before, &copy_heads, &mut doc)
+ } else {
+ return Err(format!(
+ "{} local changes are not in {target}, and this node is more than one epoch behind (on {local}, {target} follows {}); not replaying them automatically",
+ missing.own, metadata.previous
+ ));
+ };
+
+ Ok(Prepared {
+ doc,
+ copy_heads,
+ stats,
+ own_missing: missing.own,
+ had_history,
+ metadata,
+ })
+ })
+ .await
+ .map_err(|error| format!("catch-up task failed: {error}"))??;
+
+ let Prepared {
+ doc,
+ copy_heads,
+ stats,
+ own_missing,
+ had_history,
+ metadata,
+ } = prepared;
+
+ let old = swap_in(host, target, doc, copy_heads).await?;
+
+ if had_history {
+ host.compaction().archive(
+ local,
+ old,
+ metadata.policy.archive_retention.clone(),
+ target.generation,
+ );
+ }
+
+ // Visible in the compaction's report, if this node has its record.
+ host.compaction().set_status(
+ &metadata.compaction_id,
+ LocalStatus {
+ switched_at: Some(unix_millis()),
+ replayed_paths: Some(stats.replayed),
+ failed_paths: Some(stats.failed),
+ error: None,
+ },
+ );
+
+ info!(
+ from = %local,
+ to = %target,
+ document_bytes = bytes,
+ own_changes = own_missing,
+ replayed = stats.replayed,
+ failed = stats.failed,
+ "Caught up with newer epoch"
+ );
+ Ok(())
+}
+
+struct Prepared {
+ doc: AutoCommit,
+ copy_heads: Vec<automerge::ChangeHash>,
+ stats: ReplayStats,
+ own_missing: usize,
+ had_history: bool,
+ metadata: snapshot::EpochMetadata,
+}
+
+/// Answers a peer's snapshot request with this node's document, if it is on the wanted
+/// epoch. Only for members of our cluster; one at a time.
+pub(crate) async fn serve_snapshot<H: Host>(host: H, peer: String, request: SnapshotRequest) {
+ if !host.is_cluster_member(&peer).await {
+ debug!(peer = %peer, "Ignoring snapshot request from a non-member");
+ return;
+ }
+ if !host.compaction().try_start_serving() {
+ debug!(peer = %peer, "Already serving a snapshot; the peer will retry");
+ return;
+ }
+
+ let current = host.current_epoch();
+ let wanted = request.wanted;
+
+ let document = if current == wanted {
+ // Clone under the read lock, save without one. The history check guards against
+ // a switch between reading the epoch and cloning.
+ let mut doc = host.read_doc(|doc| doc.clone()).await;
+ tokio::task::spawn_blocking(move || {
+ if doc
+ .get_change_by_hash(&automerge::ChangeHash(wanted.id))
+ .is_some()
+ {
+ doc.save()
+ } else {
+ Vec::new()
+ }
+ })
+ .await
+ .unwrap_or_else(|error| {
+ error!(error = %error, "Saving the document for a peer failed");
+ Vec::new()
+ })
+ } else {
+ Vec::new()
+ };
+ host.compaction().done_serving();
+
+ info!(peer = %peer, wanted = %wanted, document_bytes = document.len(), "Serving snapshot");
+ let reply = ParsedMessage::SnapshotReply(SnapshotReply {
+ epoch: if document.is_empty() { current } else { wanted },
+ document,
+ });
+ if let Err(error) = host.send_message(&peer, reply).await {
+ warn!(peer = %peer, error = %format!("{error:#}"), "Sending snapshot failed");
+ }
+}
+
+/// Hands a peer's snapshot reply to the waiting fetch.
+pub(crate) fn receive_snapshot_reply<H: Host>(host: &H, peer: &str, reply: SnapshotReply) {
+ if !host.compaction().deliver_reply(peer, reply) {
+ debug!(peer = %peer, "Unexpected snapshot reply; dropped");
+ }
+}
diff --git a/src/compaction/delta.rs b/src/compaction/delta.rs
index 3bc9456..533a7be 100644
--- a/src/compaction/delta.rs
+++ b/src/compaction/delta.rs
@@ -17,7 +17,9 @@
//! The protocol's own metadata (`_continuum/compactions`, `_continuum/epochs`) is never
//! replayed: the snapshot owns it.
-use automerge::{AutoCommit, ChangeHash, PatchAction, Prop};
+use std::collections::{BTreeMap, HashSet};
+
+use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
use tracing::warn;
use super::snapshot::{COMPACTIONS, EPOCHS, META_ROOT};
@@ -92,6 +94,81 @@ fn contains_protocol_path(path: &[String]) -> bool {
matches!(path, [] | [_]) && path.first().is_none_or(|root| root == META_ROOT)
}
+/// What a late node has that a snapshot lacks.
+#[derive(Debug)]
+pub(crate) struct Missing {
+ /// Heads of everything except `own` changes the snapshot lacks (and changes built on
+ /// them). Replaying from these heads to the document's heads replays exactly those.
+ pub(crate) before: Vec<ChangeHash>,
+ /// Changes by `own` that the snapshot lacks.
+ pub(crate) own: usize,
+}
+
+/// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the
+/// snapshot iff seq <= clock[actor].
+///
+/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes'
+/// changes after the base were replayed by those nodes when they switched, and replaying
+/// them again would overwrite newer writes in the new epoch with stale ones.
+pub(crate) fn missing_from_snapshot(
+ doc: &mut AutoCommit,
+ clock: &BTreeMap<String, u64>,
+ own: &ActorId,
+) -> Missing {
+ let own_hex = own.to_string();
+ let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc
+ .get_changes(&[])
+ .iter()
+ .map(|change| {
+ let actor = change.actor_id().to_string();
+ let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq);
+ let own_missing = actor == own_hex && !in_snapshot;
+ (change.hash(), change.deps().to_vec(), own_missing)
+ })
+ .collect();
+
+ // Excluded: own missing changes and everything that (transitively) depends on them.
+ // A fixpoint, so it doesn't rely on the order `get_changes` returns.
+ let mut excluded: HashSet<ChangeHash> = changes
+ .iter()
+ .filter(|(_, _, own_missing)| *own_missing)
+ .map(|(hash, _, _)| *hash)
+ .collect();
+ let own_count = excluded.len();
+ loop {
+ let before = excluded.len();
+ for (hash, deps, _) in &changes {
+ if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) {
+ excluded.insert(*hash);
+ }
+ }
+ if excluded.len() == before {
+ break;
+ }
+ }
+
+ // Heads of the kept part: kept changes no kept change depends on.
+ let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes
+ .iter()
+ .filter(|(hash, _, _)| !excluded.contains(hash))
+ .collect();
+ let depended_on: HashSet<ChangeHash> = kept
+ .iter()
+ .flat_map(|(_, deps, _)| deps.iter().copied())
+ .collect();
+ let mut before: Vec<ChangeHash> = kept
+ .iter()
+ .map(|(hash, _, _)| *hash)
+ .filter(|hash| !depended_on.contains(hash))
+ .collect();
+ before.sort();
+
+ Missing {
+ before,
+ own: own_count,
+ }
+}
+
/// The minimal set of store paths covering every change between `before` and `after`:
/// sorted, and no path is a prefix of another.
pub(crate) fn changed_paths(
@@ -217,6 +294,49 @@ mod tests {
}
#[test]
+ fn missing_counts_only_own_changes() {
+ let own = ActorId::from(&b"own-actor"[..]);
+ let other = ActorId::from(&b"other-actor"[..]);
+
+ let mut doc = AutoCommit::new().with_actor(own.clone());
+ doc.put(ROOT, "base", 1_i64).unwrap();
+ doc.commit();
+ let base = doc.get_heads();
+
+ // The snapshot contains the base only.
+ let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect();
+
+ // A foreign change after the base (replayed by its own node) ...
+ let mut foreign = doc.fork().with_actor(other);
+ foreign.put(ROOT, "theirs", 1_i64).unwrap();
+ foreign.commit();
+ doc.merge(&mut foreign).unwrap();
+ let with_foreign = doc.get_heads();
+
+ // ... and one of ours.
+ doc.put(ROOT, "mine", 1_i64).unwrap();
+ doc.commit();
+ let after = doc.get_heads();
+
+ let missing = missing_from_snapshot(&mut doc, &clock, &own);
+ assert_eq!(missing.own, 1);
+ assert_eq!(missing.before, with_foreign);
+ assert_ne!(missing.before, base);
+ assert_eq!(
+ changed_paths(&mut doc, &missing.before, &after),
+ vec![vec!["mine".to_owned()]]
+ );
+
+ // Nothing of ours missing: nothing to replay.
+ let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect();
+ let none = missing_from_snapshot(&mut doc, &full, &own);
+ assert_eq!(none.own, 0);
+ let mut heads = after.clone();
+ heads.sort();
+ assert_eq!(none.before, heads);
+ }
+
+ #[test]
fn protocol_metadata_is_never_replayed() {
let mut old = AutoCommit::new();
old.put(ROOT, "x", 1_i64).unwrap();
diff --git a/src/compaction/local.rs b/src/compaction/local.rs
index 6290a5e..869c94e 100644
--- a/src/compaction/local.rs
+++ b/src/compaction/local.rs
@@ -5,20 +5,25 @@
//! archive and rebuilds snapshots when needed.
use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, MutexGuard};
use automerge::AutoCommit;
use serde::Serialize;
+use tokio::sync::oneshot;
use tokio::time::Instant;
use super::records::{unix_millis, Retention};
use super::snapshot::Snapshot;
use crate::replica::Epoch;
+use crate::transport::SnapshotReply;
#[derive(Default)]
pub(crate) struct Compaction {
// A std mutex: every access is short and never spans an await.
inner: Mutex<Inner>,
+ /// Set while this node saves its document for a peer; one at a time.
+ serving: AtomicBool,
}
#[derive(Default)]
@@ -29,6 +34,27 @@ struct Inner {
first_seen: HashMap<String, Instant>,
status: HashMap<String, LocalStatus>,
archive: Vec<Archived>,
+ /// The snapshot request this node is waiting for (peer, wanted epoch).
+ awaiting: Option<(String, Epoch, oneshot::Sender<SnapshotReply>)>,
+ catch_up: Option<CatchUp>,
+}
+
+/// Fetching a newer epoch from a peer.
+#[derive(Clone, Debug)]
+pub(crate) struct CatchUp {
+ pub(crate) target: Epoch,
+ pub(crate) attempts: u32,
+ pub(crate) last_attempt: Instant,
+ pub(crate) error: Option<String>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct CatchUpReport {
+ pub(crate) target: String,
+ pub(crate) attempts: u32,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub(crate) error: Option<String>,
}
/// What happened to a compaction on this node.
@@ -118,6 +144,58 @@ impl Compaction {
prune(&mut self.lock().archive, current_generation);
}
+ /// Registers interest in a snapshot reply from `peer`, replacing any earlier one.
+ pub(crate) fn await_reply(
+ &self,
+ peer: &str,
+ wanted: Epoch,
+ ) -> oneshot::Receiver<SnapshotReply> {
+ let (sender, receiver) = oneshot::channel();
+ self.lock().awaiting = Some((peer.to_owned(), wanted, sender));
+ receiver
+ }
+
+ /// Hands a reply to the waiting fetch. False if nobody asked that peer.
+ pub(crate) fn deliver_reply(&self, peer: &str, reply: SnapshotReply) -> bool {
+ let mut inner = self.lock();
+ match inner.awaiting.take() {
+ Some((expected, _, sender)) if expected == peer => sender.send(reply).is_ok(),
+ other => {
+ inner.awaiting = other;
+ false
+ }
+ }
+ }
+
+ pub(crate) fn stop_awaiting(&self) {
+ self.lock().awaiting = None;
+ }
+
+ /// Claims the right to serve one snapshot; release with [`Compaction::done_serving`].
+ pub(crate) fn try_start_serving(&self) -> bool {
+ !self.serving.swap(true, Ordering::AcqRel)
+ }
+
+ pub(crate) fn done_serving(&self) {
+ self.serving.store(false, Ordering::Release);
+ }
+
+ pub(crate) fn catch_up(&self) -> Option<CatchUp> {
+ self.lock().catch_up.clone()
+ }
+
+ pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) {
+ self.lock().catch_up = catch_up;
+ }
+
+ pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> {
+ self.lock().catch_up.as_ref().map(|catch_up| CatchUpReport {
+ target: catch_up.target.to_string(),
+ attempts: catch_up.attempts,
+ error: catch_up.error.clone(),
+ })
+ }
+
pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> {
self.lock()
.archive
diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
index 69c633e..e97d9c8 100644
--- a/src/compaction/mod.rs
+++ b/src/compaction/mod.rs
@@ -4,6 +4,7 @@
//! * `records`: the replicated compaction records under `_continuum/compactions/<id>`.
//! * `coordinator`: proposal, settle period and decision on the node that was asked.
//! * `switch`: switching this node to a committed epoch (worker).
+//! * `catchup`: fetching a newer epoch from a peer (late, new or failed nodes).
//! * `delta`: replaying this node's changes since the base into the new epoch.
//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive).
//! * `api`: `/admin/compaction(s)` handlers.
@@ -12,6 +13,7 @@
//! depends on `AppState`, peers or sync internals.
mod api;
+mod catchup;
mod coordinator;
mod delta;
mod local;
@@ -19,6 +21,7 @@ mod records;
pub(crate) mod snapshot;
mod switch;
+pub(crate) use catchup::{receive_snapshot_reply, serve_snapshot};
pub(crate) use local::Compaction;
pub(crate) use switch::run_worker;
@@ -35,6 +38,7 @@ use tokio::sync::broadcast;
use crate::replica::Epoch;
use crate::sync::ChangeSet;
+use crate::transport::ParsedMessage;
/// What compaction needs from the node. Implemented by `AppState`.
///
@@ -57,6 +61,16 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// Known (not forgotten) peers.
fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
+ /// True if `peer` completed the handshake as a member of our cluster.
+ fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send;
+
+ /// Sends a message to a peer over the transport.
+ fn send_message(
+ &self,
+ peer: &str,
+ message: ParsedMessage,
+ ) -> impl Future<Output = anyhow::Result<()>> + Send;
+
/// Store paths touched by each published document change.
fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>;
@@ -97,6 +111,10 @@ pub(crate) struct PeerSummary {
pub(crate) id: String,
/// Not dormant.
pub(crate) active: bool,
+ /// Handshake completed.
+ pub(crate) established: bool,
+ /// The epoch the peer last advertised.
+ pub(crate) epoch: Option<Epoch>,
}
/// Admin routes. Merged into the authenticated part of the node's router.
diff --git a/src/compaction/records.rs b/src/compaction/records.rs
index ce2b2a5..fbb9fb9 100644
--- a/src/compaction/records.rs
+++ b/src/compaction/records.rs
@@ -76,8 +76,10 @@ pub(crate) struct Retention {
pub(crate) struct Policy {
/// How stale a returning node's changes may be and still be replayed automatically;
/// beyond it they are quarantined.
+ #[serde(default = "Policy::default_replay_window")]
pub(crate) replay_window: Retention,
/// How long nodes keep the old epoch's document.
+ #[serde(default = "Policy::default_archive_retention")]
pub(crate) archive_retention: Retention,
}
diff --git a/src/compaction/snapshot.rs b/src/compaction/snapshot.rs
index 1f57352..86caba9 100644
--- a/src/compaction/snapshot.rs
+++ b/src/compaction/snapshot.rs
@@ -18,7 +18,8 @@
//! at the snapshot change itself, so edits made afterwards through the store API can't
//! change what the protocol sees.
-// `Snapshot::clock` is used by returning nodes (step 7).
+// `Snapshot::clock` is kept for returning nodes (step 7); late nodes read the clock from
+// the metadata instead.
#![cfg_attr(not(test), allow(dead_code))]
use std::collections::BTreeMap;
@@ -27,6 +28,7 @@ use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
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};
@@ -105,6 +107,67 @@ pub(crate) async fn build_detached(
.map_err(|error| SnapshotError::Task(error.to_string()))?
}
+/// The protocol's view of an epoch, read at its snapshot change.
+#[derive(Clone, Debug)]
+pub(crate) struct EpochMetadata {
+ pub(crate) compaction_id: String,
+ pub(crate) previous: Epoch,
+ pub(crate) clock: BTreeMap<String, u64>,
+ pub(crate) policy: Policy,
+}
+
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct MetadataRecord {
+ generation: u64,
+ compaction_id: String,
+ previous: EpochRecord,
+ clock: BTreeMap<String, u64>,
+ policy: Policy,
+}
+
+/// Checks that `doc` is a document of `epoch`: its history starts with exactly the
+/// snapshot change `epoch.id` (no dependencies), and the metadata at that change names
+/// the same generation. Returns the metadata.
+///
+/// This does not prove the snapshot's content matches the old epoch; the hash commits
+/// to it, and the peer is a cluster member trusted as much as for ordinary sync.
+pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> {
+ let root = ChangeHash(epoch.id);
+ let Some(change) = doc.get_change_by_hash(&root) else {
+ return Err(format!(
+ "document does not contain the snapshot change of {epoch}"
+ ));
+ };
+ if !change.deps().is_empty() {
+ return Err(format!("snapshot change of {epoch} has dependencies"));
+ }
+
+ let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?;
+ let value = store::read_path(&at_snapshot, &epoch_metadata_path(epoch.generation))
+ .map_err(|error| error.to_string())?
+ .ok_or_else(|| format!("no epoch metadata for generation {}", epoch.generation))?;
+ let record: MetadataRecord = serde_json::from_value(value)
+ .map_err(|error| format!("invalid epoch metadata: {error}"))?;
+ if record.generation != epoch.generation {
+ return Err(format!(
+ "epoch metadata names generation {} instead of {}",
+ record.generation, epoch.generation
+ ));
+ }
+ let previous = record
+ .previous
+ .to_epoch()
+ .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+
+ Ok(EpochMetadata {
+ compaction_id: record.compaction_id,
+ previous,
+ clock: record.clock,
+ policy: record.policy,
+ })
+}
+
/// Path of the metadata record for an epoch generation.
pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> {
vec![
@@ -493,4 +556,57 @@ mod tests {
assert!(snap.clock.is_empty());
assert_eq!(user_view(&snap.doc), json!({}));
}
+
+ #[test]
+ fn fetched_document_is_verified_and_late_changes_replayed() {
+ use crate::compaction::delta;
+
+ // Old epoch: the base, then a late write by this node.
+ let mut late = doc(b"late-node");
+ late.put(ROOT, "shared", "base").unwrap();
+ late.commit();
+ let base = late.get_heads();
+ let policy = json!({});
+ let mut snap = build(&mut late, &params(&base, &policy)).unwrap();
+
+ // Meanwhile the cluster moved on in the new epoch.
+ let mut cluster = snap
+ .doc
+ .fork()
+ .with_actor(ActorId::from(&b"cluster-node"[..]));
+ cluster.put(ROOT, "new", 1_i64).unwrap();
+ cluster.commit();
+ let bytes = cluster.save();
+
+ late.put(ROOT, "mine", "late").unwrap();
+ late.commit();
+
+ let mut fetched = AutoCommit::load(&bytes).unwrap();
+ let metadata = verify_document(&mut fetched, snap.epoch).unwrap();
+ assert_eq!(metadata.previous, Epoch::GENESIS);
+ assert_eq!(metadata.compaction_id, "c-1");
+ assert_eq!(
+ metadata.policy.replay_window,
+ Policy::default_replay_window()
+ );
+
+ // Documents of other epochs are refused.
+ let mut wrong = snap.epoch;
+ wrong.generation = 2;
+ assert!(verify_document(&mut fetched, wrong).is_err());
+ let mut wrong = snap.epoch;
+ wrong.id = [0; 32];
+ assert!(verify_document(&mut fetched, wrong).is_err());
+
+ let own = late.get_actor().clone();
+ let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own);
+ assert_eq!(missing.own, 1);
+ let heads = late.get_heads();
+ let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..]));
+ delta::replay(&mut late, &missing.before, &heads, &mut adopted);
+ assert_eq!(
+ user_view(&adopted),
+ json!({ "shared": "base", "new": 1, "mine": "late" })
+ );
+ }
}
diff --git a/src/compaction/switch.rs b/src/compaction/switch.rs
index 5eb1962..9f17df2 100644
--- a/src/compaction/switch.rs
+++ b/src/compaction/switch.rs
@@ -15,20 +15,22 @@
//! The write lock is held only for step 3, which is proportional to the changes made
//! during steps 1-2.
//!
-//! A node whose build doesn't match the committed epoch stays on the old epoch and
-//! reports the mismatch; fetching the snapshot from peers comes with step 6.
+//! A node whose build doesn't match the committed epoch, or that never saw the commit,
+//! fetches the new epoch from a peer instead (`catchup`).
use std::time::Duration;
-use automerge::{ActorId, AutoCommit};
+use automerge::{ActorId, AutoCommit, ChangeHash};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
+use super::catchup;
use super::delta::{self, ReplayStats};
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;
@@ -79,13 +81,15 @@ async fn check<H: Host>(host: &H) {
.collect();
let (id, record) = match committed.as_slice() {
- [] => return,
+ // Nothing committed here; maybe peers moved on without us.
+ [] => return catchup::check(host).await,
[(id, record)] => (id.clone(), record.clone()),
several => {
- // The settle protocol prevents this; never guess which one is right.
+ // The settle protocol prevents this; never guess which one is right. Peers
+ // that did switch can still be followed.
let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
- error!(compactions = ?ids, "Several compactions committed for this epoch; not switching");
- return;
+ error!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own");
+ return catchup::check(host).await;
}
};
@@ -94,7 +98,9 @@ async fn check<H: Host>(host: &H) {
.status(&id)
.is_some_and(|status| status.error.is_some())
{
- return; // already failed; reported in the compaction's status
+ // Our own switch failed (reported in the compaction's status): fetch the epoch
+ // from a peer that made it.
+ return catchup::check(host).await;
}
let switch_delay = host.dormant_after() / 4; // one keepalive interval
@@ -148,7 +154,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
// 2. Build (or reuse), verify and replay without holding any lock.
let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking(
- move || -> Result<(AutoCommit, Vec<automerge::ChangeHash>, ReplayStats), String> {
+ move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
let mut copy = copy;
let snapshot = match prebuilt {
Some(snapshot) => snapshot,
@@ -179,21 +185,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
.map_err(|error| format!("switch task failed: {error}"))??;
// 3. Swap, replaying what arrived since the copy.
- let old = host
- .switch_epoch(expected, move |live| {
- let mut new_doc = new_doc;
- let live_heads = live.get_heads();
- let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
- if late.replayed + late.failed > 0 {
- info!(
- replayed = late.replayed,
- failed = late.failed,
- "Replayed late changes"
- );
- }
- Ok::<_, String>(new_doc)
- })
- .await?;
+ let old = swap_in(host, expected, new_doc, copy_heads).await?;
// 4. Keep the old document as configured.
host.compaction()
@@ -208,3 +200,35 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
);
Ok(stats)
}
+
+/// Under the write lock: replays into `new_doc` whatever the live document received
+/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it
+/// in as `target`. Returns the old document.
+pub(super) async fn swap_in<H: Host>(
+ host: &H,
+ target: Epoch,
+ new_doc: AutoCommit,
+ copy_heads: Vec<ChangeHash>,
+) -> Result<AutoCommit, String> {
+ host.switch_epoch(target, move |live| {
+ // Only one worker switches, but be sure the copy still describes the live document.
+ if !copy_heads
+ .iter()
+ .all(|head| live.get_change_by_hash(head).is_some())
+ {
+ return Err("the document was replaced while preparing the switch".to_owned());
+ }
+ let mut new_doc = new_doc;
+ let live_heads = live.get_heads();
+ let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
+ if late.replayed + late.failed > 0 {
+ info!(
+ replayed = late.replayed,
+ failed = late.failed,
+ "Replayed late changes"
+ );
+ }
+ Ok(new_doc)
+ })
+ .await
+}
diff --git a/src/main.rs b/src/main.rs
index 1127cd8..f5cec55 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -187,6 +187,12 @@ async fn main() -> Result<()> {
});
info!("Peer worker task spawned");
+ // Switches to committed epochs and catches up with newer ones. Starts before
+ // bootstrapping: a node joining a compacted cluster can only finish its initial sync
+ // after fetching the current epoch from a peer.
+ tokio::spawn(compaction::run_worker(state.clone()));
+ info!("Compaction worker task spawned");
+
bootstrap::bootstrap(&state, &seeds)
.await
.context("Joining cluster")?;
@@ -196,9 +202,6 @@ async fn main() -> Result<()> {
let publisher_state = state.clone();
tokio::spawn(async move { sync::run_change_publisher(publisher_state).await });
- // Switches to committed epochs. Runs only once the node is Ready.
- tokio::spawn(compaction::run_worker(state.clone()));
-
info!(
cluster_id = ?state.cluster_id().await,
ready = state.is_ready().await,
diff --git a/src/mycelium.rs b/src/mycelium.rs
index 8d0821b..22d49c1 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -19,7 +19,9 @@ use url::Url;
use crate::transport::EncodedMessage;
use crate::transport::Keepalive;
-use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
+use crate::transport::{
+ Epoch, HelloReply, HelloRequest, SnapshotReply, SnapshotRequest, SyncEnvelope,
+};
use crate::transport::{MessageHandler, ParsedMessage, Transport};
const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
@@ -30,6 +32,8 @@ 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 TOPIC_SNAPSHOT_REQUEST: &str = "continuum.v1.snapshot-request";
+const TOPIC_SNAPSHOT_REPLY: &str = "continuum.v1.snapshot-reply";
const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
@@ -696,6 +700,12 @@ fn decode_payload(topic: &str, payload: &str) -> Result<ParsedMessage, Transport
}
TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
+ TOPIC_SNAPSHOT_REQUEST => Ok(ParsedMessage::SnapshotRequest(bytes_to_snapshot_request(
+ &bytes,
+ )?)),
+ TOPIC_SNAPSHOT_REPLY => Ok(ParsedMessage::SnapshotReply(bytes_to_snapshot_reply(
+ &bytes,
+ )?)),
t => Err(TransportError::UnknownTopic(t.to_string())),
}
}
@@ -723,6 +733,10 @@ fn format_message(
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)?),
+ ParsedMessage::SnapshotRequest(req) => {
+ (TOPIC_SNAPSHOT_REQUEST, snapshot_request_to_bytes(req)?)
+ }
+ ParsedMessage::SnapshotReply(res) => (TOPIC_SNAPSHOT_REPLY, snapshot_reply_to_bytes(res)?),
};
Ok(OutboundMessage {
dst: Destination {
@@ -900,6 +914,50 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
})
}
+// --- Snapshot transfer ---
+#[derive(Serialize, Deserialize)]
+struct SnapshotRequestShadow {
+ epoch: EpochShadow,
+ wanted: EpochShadow,
+}
+
+#[derive(Serialize, Deserialize)]
+struct SnapshotReplyShadow {
+ epoch: EpochShadow,
+ document: ByteBuf,
+}
+
+fn snapshot_request_to_bytes(msg: &SnapshotRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+ rmp_serde::to_vec(&SnapshotRequestShadow {
+ epoch: epoch_to_shadow(&msg.epoch),
+ wanted: epoch_to_shadow(&msg.wanted),
+ })
+}
+
+fn bytes_to_snapshot_request(bytes: &[u8]) -> Result<SnapshotRequest, TransportError> {
+ let shadow: SnapshotRequestShadow = rmp_serde::from_slice(bytes)?;
+ Ok(SnapshotRequest {
+ epoch: epoch_from_shadow(shadow.epoch)?,
+ wanted: epoch_from_shadow(shadow.wanted)?,
+ })
+}
+
+fn snapshot_reply_to_bytes(msg: &SnapshotReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
+ rmp_serde::to_vec(&SnapshotReplyShadow {
+ epoch: epoch_to_shadow(&msg.epoch),
+ // One copy of the document; replies are rare.
+ document: ByteBuf::from(msg.document.clone()),
+ })
+}
+
+fn bytes_to_snapshot_reply(bytes: &[u8]) -> Result<SnapshotReply, TransportError> {
+ let shadow: SnapshotReplyShadow = rmp_serde::from_slice(bytes)?;
+ Ok(SnapshotReply {
+ epoch: epoch_from_shadow(shadow.epoch)?,
+ document: shadow.document.into_vec(),
+ })
+}
+
#[derive(Error, Debug)]
pub enum TransportError {
#[error("Mycelium delivered a message without a sender public key")]
diff --git a/src/peer.rs b/src/peer.rs
index 238c341..015b5ff 100644
--- a/src/peer.rs
+++ b/src/peer.rs
@@ -660,6 +660,17 @@ pub(crate) async fn handle_message(
let local = local_keepalive(state).await;
heal_if_diverged(state, &peer, &local, &remote).await;
}
+ transport::ParsedMessage::SnapshotRequest(request) => {
+ // Saving a large document takes a while; don't stall the inbox.
+ tokio::spawn(crate::compaction::serve_snapshot(
+ state.clone(),
+ peer.clone(),
+ request,
+ ));
+ }
+ transport::ParsedMessage::SnapshotReply(reply) => {
+ crate::compaction::receive_snapshot_reply(state, &peer, reply);
+ }
}
Ok(())
diff --git a/src/state.rs b/src/state.rs
index a8b1917..b3a496c 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -566,13 +566,39 @@ impl Host for AppState {
let mut summaries = Vec::with_capacity(peers.len());
for (id, peer) in peers {
- let active = peer.lock().await.liveness == Liveness::Active;
- summaries.push(PeerSummary { id, active });
+ let peer = peer.lock().await;
+ summaries.push(PeerSummary {
+ id,
+ active: peer.liveness == Liveness::Active,
+ established: peer.hello_state == HelloState::Established,
+ epoch: peer.epoch,
+ });
}
summaries
}
}
+ fn is_cluster_member(&self, peer: &str) -> impl Future<Output = bool> + Send {
+ let state = self.clone();
+ let peer = peer.to_owned();
+ async move {
+ let Some(ours) = state.cluster_id().await else {
+ return false;
+ };
+ crate::peer::peer_cluster(&state, &peer).await.as_deref() == Some(ours.as_str())
+ }
+ }
+
+ fn send_message(
+ &self,
+ peer: &str,
+ message: transport::ParsedMessage,
+ ) -> impl Future<Output = anyhow::Result<()>> + Send {
+ let state = self.clone();
+ let peer = peer.to_owned();
+ async move { state.transport.send_message(&peer, &message).await }
+ }
+
fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>> {
self.change_events.subscribe()
}
diff --git a/src/transport.rs b/src/transport.rs
index 9cfb026..b7a4094 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -23,6 +23,9 @@ pub enum ParsedMessage {
Sync(SyncEnvelope),
Ping(Keepalive),
Pong(Keepalive),
+ /// Asks a peer for its document of a newer epoch (late or new nodes).
+ SnapshotRequest(SnapshotRequest),
+ SnapshotReply(SnapshotReply),
}
impl ParsedMessage {
@@ -33,6 +36,8 @@ impl ParsedMessage {
ParsedMessage::HelloReply(reply) => reply.epoch,
ParsedMessage::Sync(envelope) => envelope.epoch,
ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
+ ParsedMessage::SnapshotRequest(request) => request.epoch,
+ ParsedMessage::SnapshotReply(reply) => reply.epoch,
}
}
}
@@ -57,6 +62,32 @@ pub struct SyncEnvelope {
pub message: Message,
}
+#[derive(Debug, Clone)]
+pub struct SnapshotRequest {
+ /// The requester's epoch.
+ pub epoch: Epoch,
+ /// The epoch whose document is wanted.
+ pub wanted: Epoch,
+}
+
+#[derive(Clone)]
+pub struct SnapshotReply {
+ /// The sender's epoch, which is the epoch of `document`.
+ pub epoch: Epoch,
+ /// `AutoCommit::save()` of the sender's document; empty if it isn't on the wanted
+ /// epoch (any more).
+ pub document: Vec<u8>,
+}
+
+impl std::fmt::Debug for SnapshotReply {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("SnapshotReply")
+ .field("epoch", &self.epoch)
+ .field("document_bytes", &self.document.len())
+ .finish()
+ }
+}
+
/// A message serialized for the wire. Cloning is cheap (shared buffer).
#[derive(Clone, Debug)]
pub struct EncodedMessage {
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-21 19:24:12 +0000