summaryrefslogtreecommitdiffstats
path: root/src/compaction/catchup.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/compaction/catchup.rs')
-rw-r--r--src/compaction/catchup.rs680
1 files changed, 362 insertions, 318 deletions
diff --git a/src/compaction/catchup.rs b/src/compaction/catchup.rs
index 4694fbc..c71bb57 100644
--- a/src/compaction/catchup.rs
+++ b/src/compaction/catchup.rs
@@ -1,59 +1,67 @@
-//! Catching up with a newer epoch from a peer.
+//! Catching up with a newer epoch: the node rejoins.
//!
-//! 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 active peer advertises a newer generation, the node downloads that peer's
-//! document (`SnapshotRequest`/`SnapshotReply`), checks that it starts with the advertised
-//! snapshot change, and switches to it:
+//! Peers on another epoch don't sync, so a node that missed a switch (new, late,
+//! returning, or one whose own build failed) can't follow through ordinary sync. Once
+//! an active peer advertises a newer generation, the node
//!
-//! * 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).
+//! 1. stops serving requests and keeps its document aside;
+//! 2. switches to an empty document labelled with the newer epoch: an empty history
+//! conflicts with nothing, so ordinary sync fills it from peers on that epoch,
+//! exactly like a new node joining;
+//! 3. once caught up with a peer, checks that the history starts with the epoch's
+//! snapshot change;
+//! 4. decides about its own changes the cluster never saw (none, for a new node):
+//! * the metadata of the epoch that followed the node's own records which changes
+//! went into it (per-actor clock); the rest are this node's missing changes;
+//! * within that epoch's replay window they are replayed (state per path, as in a
+//! switch), otherwise they are quarantined for an operator
+//! (`/admin/quarantine`), as are changes that can't be told apart;
+//! 5. archives the old document and serves requests again.
//!
-//! The document travels in chunks of `CHUNK_BYTES`, one request per chunk, so a large
-//! document neither needs one huge transport message nor one long timeout. The serving
-//! node saves its document once per download and serves every chunk from that copy; a
-//! SHA-256 digest of the whole copy is in every reply and checked at the end.
-//!
-//! Peers refuse explicitly (non-members, busy, other epoch), so the requester moves on
-//! to the next peer right away instead of timing out.
+//! If no peer completes a sync in time, or the check fails, the node restores its old
+//! document and epoch and tries again later.
-use std::sync::Arc;
+use std::collections::BTreeMap;
+use std::convert::Infallible;
use std::time::Duration;
-use automerge::{ActorId, AutoCommit};
-use sha2::{Digest, Sha256};
+use automerge::{AutoCommit, ChangeHash};
use tokio::time::Instant;
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
+use uuid::Uuid;
use super::delta::{self, ReplayStats};
-use super::local::{CatchUp, LocalStatus, Served};
+use super::local::{CatchUp, LocalStatus, Quarantined, Rejoin};
use super::records::unix_millis;
-use super::snapshot;
-use super::switch::swap_in;
+use super::snapshot::{self, EpochMetadata};
use super::Host;
use crate::replica::Epoch;
-use crate::transport::{ParsedMessage, SnapshotReply, SnapshotRequest};
-/// Pause before the next attempt (next peer) after a failed download.
-const RETRY: Duration = Duration::from_secs(10);
-/// Payload per reply.
-pub(crate) const CHUNK_BYTES: usize = 1024 * 1024;
-/// How long to wait for one chunk, and how often to ask for it.
-const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
-const CHUNK_ATTEMPTS: u32 = 3;
-/// A served copy older than this is replaced when a new download starts.
-const SERVED_FRESH_FOR: Duration = Duration::from_secs(60);
+/// Pause before trying the same target again after a failed rejoin.
+const RETRY: Duration = Duration::from_secs(30);
+/// Lower bound for how long a rejoin may take to complete a sync.
+const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
-/// Called by the switch worker when there is nothing to switch to locally.
+/// Called by the switch worker when there is nothing to switch to locally, and on every
+/// tick while a rejoin is in progress.
pub(super) async fn check<H: Host>(host: &H) {
- let local = host.current_epoch();
+ if let Some((target, started)) = host.compaction().rejoin_progress() {
+ let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT);
+ if host.initial_sync_done() {
+ finish(host).await;
+ } else if started.elapsed() > timeout {
+ if let Some(rejoin) = host.compaction().take_rejoin() {
+ let reason = format!(
+ "no peer on {target} completed a sync within {}s",
+ timeout.as_secs()
+ );
+ abandon(host, rejoin, reason).await;
+ }
+ }
+ return;
+ }
- // Any active peer will do: it proved liveness recently, and whether it accepts us
- // as a member is its decision (it refuses explicitly otherwise).
+ let local = host.current_epoch();
let mut ahead: Vec<(String, Epoch)> = host
.peer_summaries()
.await
@@ -98,14 +106,10 @@ pub(super) async fn check<H: Host>(host: &H) {
}
let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
- // Rotate through the peers on the target epoch (sorted, so the rotation is stable).
- let sources: Vec<&String> = ahead
- .iter()
- .filter(|(_, epoch)| *epoch == target)
- .map(|(peer, _)| peer)
- .collect();
- let source = sources[(attempts as usize - 1) % sources.len()].clone();
+ start(host, local, target, attempts).await;
+}
+async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
host.compaction().set_catch_up(Some(CatchUp {
target,
attempts,
@@ -113,336 +117,376 @@ pub(super) async fn check<H: Host>(host: &H) {
error: None,
}));
- info!(local = %local, target = %target, peer = %source, attempt = attempts, "Fetching newer epoch from peer");
-
- let result = match download(host, &source, local, target).await {
- Ok(document) => adopt(host, local, target, document).await,
- Err(error) => Err(error),
+ let was_ready = host.suspend().await;
+ let old = match host
+ .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new()))
+ .await
+ {
+ Ok(old) => old,
+ Err(never) => match never {},
};
+ // After the swap: messages of the old epoch are dropped from now on, so nothing
+ // can mark the new, empty document as caught up by mistake. A stale "done" from
+ // just before is caught in `finish` (the snapshot change is missing then).
+ host.reset_initial_sync();
- 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),
- }));
- }
- }
+ info!(from = %local, to = %target, attempt = attempts, "Rejoining newer epoch");
+ host.compaction().start_rejoin(Rejoin {
+ previous: local,
+ target,
+ old,
+ started: Instant::now(),
+ was_ready,
+ });
+}
+
+/// What to do with this node's own changes the new epoch lacks.
+enum Plan {
+ Nothing,
+ Replay(Vec<Vec<String>>),
+ Quarantine {
+ paths: Vec<Vec<String>>,
+ reason: String,
+ },
+}
+
+enum Prepared {
+ /// Caught up with someone, but not with the target epoch yet: keep waiting.
+ NotYet(AutoCommit),
+ Failed(AutoCommit, String),
+ Ready {
+ old: AutoCommit,
+ metadata: EpochMetadata,
+ plan: Plan,
+ own_missing: usize,
+ had_history: bool,
+ },
}
-/// Downloads `peer`'s document of `target`, chunk by chunk.
-async fn download<H: Host>(
- host: &H,
- peer: &str,
- local: Epoch,
- target: Epoch,
-) -> Result<Vec<u8>, String> {
- let mut document: Vec<u8> = Vec::new();
- let mut expected: Option<([u8; 32], u64)> = None;
- let mut last_progress = 0;
+async fn finish<H: Host>(host: &H) {
+ let Some(rejoin) = host.compaction().take_rejoin() else {
+ return;
+ };
+ let Rejoin {
+ previous,
+ target,
+ old,
+ started,
+ was_ready,
+ } = rejoin;
- loop {
- let offset = document.len() as u64;
- let reply = request_chunk(host, peer, local, target, offset).await?;
+ // Everything heavy runs on a copy, without a lock.
+ let copy = host.read_doc(|doc| doc.clone()).await;
+ let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await;
- if !reply.refused.is_empty() {
- return Err(format!("peer refused: {}", reply.refused));
+ let (old, metadata, plan, own_missing, had_history) = match prepared {
+ Ok(Prepared::Ready {
+ old,
+ metadata,
+ plan,
+ own_missing,
+ had_history,
+ }) => (old, metadata, plan, own_missing, had_history),
+ Ok(Prepared::NotYet(old)) => {
+ host.reset_initial_sync();
+ host.compaction().start_rejoin(Rejoin {
+ previous,
+ target,
+ old,
+ started,
+ was_ready,
+ });
+ return;
}
- if reply.epoch != target {
- return Err(format!("peer is on {} instead of {target}", reply.epoch));
+ Ok(Prepared::Failed(old, reason)) => {
+ let rejoin = Rejoin {
+ previous,
+ target,
+ old,
+ started,
+ was_ready,
+ };
+ return abandon(host, rejoin, reason).await;
}
- match expected {
- None => {
- info!(peer = %peer, document_bytes = reply.total, "Downloading document");
- expected = Some((reply.digest, reply.total));
- }
- Some((digest, total)) if digest != reply.digest || total != reply.total => {
- return Err("the peer's document changed during the download".to_owned());
+ Err(join_error) => {
+ // The old document is gone with the panicked task; nothing to restore.
+ error!(error = %join_error, "Rejoin task failed; staying on the new epoch");
+ if was_ready {
+ host.resume().await;
}
- Some(_) => {}
- }
- let total = reply.total;
- if reply.data.is_empty() && offset < total {
- return Err(format!("empty chunk at offset {offset} of {total}"));
- }
- document.extend_from_slice(&reply.data);
-
- let percent = (document.len() as u64 * 100)
- .checked_div(total)
- .unwrap_or(100);
- if percent >= last_progress + 10 {
- last_progress = percent;
- debug!(peer = %peer, percent, "Download progress");
+ return;
}
+ };
- if document.len() as u64 >= total {
- break;
- }
- }
+ let mut stats = ReplayStats::default();
+ let mut quarantine_id = None;
+ let mut keep_archive = true;
+ let mut old = Some(old);
- let (digest, total) = expected.expect("set by the first reply");
- if document.len() as u64 != total || sha256(&document) != digest {
- return Err("downloaded document does not match its digest".to_owned());
- }
- Ok(document)
-}
-
-async fn request_chunk<H: Host>(
- host: &H,
- peer: &str,
- local: Epoch,
- target: Epoch,
- offset: u64,
-) -> Result<SnapshotReply, String> {
- for attempt in 1..=CHUNK_ATTEMPTS {
- let reply = host.compaction().await_reply(peer, target, offset);
- let request = ParsedMessage::SnapshotRequest(SnapshotRequest {
- epoch: local,
- wanted: target,
- offset,
- });
- if let Err(error) = host.send_message(peer, request).await {
- host.compaction().stop_awaiting();
- return Err(format!("sending the request failed: {error:#}"));
+ match plan {
+ Plan::Nothing => {}
+ Plan::Replay(paths) => {
+ let document = old.take().expect("old document");
+ let (replayed, document) = host
+ .commit_doc(move |doc| {
+ let stats = delta::replay_paths(&document, doc, &paths);
+ Ok::<_, Infallible>((stats, document))
+ })
+ .await
+ .unwrap_or_else(|never| match never {});
+ stats = replayed;
+ old = Some(document);
}
-
- match tokio::time::timeout(CHUNK_TIMEOUT, reply).await {
- Ok(Ok(reply)) => return Ok(reply),
- Ok(Err(_)) => return Err("request superseded".to_owned()),
- Err(_) => {
- host.compaction().stop_awaiting();
- debug!(peer = %peer, offset, attempt, "No reply for chunk; asking again");
- }
+ Plan::Quarantine { paths, reason } => {
+ let id = Uuid::new_v4().to_string();
+ warn!(quarantine = %id, epoch = %previous, paths = paths.len(), reason = %reason, "Quarantined local changes");
+ host.compaction().add_quarantine(Quarantined {
+ id: id.clone(),
+ epoch: previous,
+ target,
+ created_at: unix_millis(),
+ reason,
+ paths,
+ document: old.take().expect("old document"),
+ });
+ quarantine_id = Some(id);
+ keep_archive = false; // the quarantine holds the document
}
}
- Err(format!(
- "no reply for the chunk at offset {offset} after {CHUNK_ATTEMPTS} attempts of {}s",
- CHUNK_TIMEOUT.as_secs()
- ))
-}
-
-/// 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 {
+ if let (true, true, Some(document)) = (had_history, keep_archive, old) {
host.compaction().archive(
- local,
- old,
+ previous,
+ document,
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),
+ quarantine: quarantine_id.clone(),
error: None,
},
);
+ host.compaction().set_catch_up(None);
+ if was_ready {
+ host.resume().await;
+ }
info!(
- from = %local,
+ from = %previous,
to = %target,
- document_bytes = bytes,
own_changes = own_missing,
replayed = stats.replayed,
failed = stats.failed,
- "Caught up with newer epoch"
+ quarantine = ?quarantine_id,
+ secs = started.elapsed().as_secs(),
+ "Rejoined newer epoch"
);
- Ok(())
}
-struct Prepared {
- doc: AutoCommit,
- copy_heads: Vec<automerge::ChangeHash>,
- stats: ReplayStats,
- own_missing: usize,
- had_history: bool,
- metadata: snapshot::EpochMetadata,
-}
+/// Verifies the synced document and decides about the old document's own changes.
+fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared {
+ if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
+ return Prepared::NotYet(old);
+ }
+ let metadata = match snapshot::verify_document(&mut copy, target) {
+ Ok(metadata) => metadata,
+ Err(reason) => return Prepared::Failed(old, reason),
+ };
-/// Answers a peer's chunk request from a saved copy of this node's document, if this
-/// node is on the wanted epoch and the peer is a member of our cluster. Refuses
-/// explicitly otherwise.
-pub(crate) async fn serve_snapshot<H: Host>(host: H, peer: String, request: SnapshotRequest) {
- let current = host.current_epoch();
- let wanted = request.wanted;
+ let old_heads = old.get_heads();
+ if old_heads.is_empty() {
+ return Prepared::Ready {
+ old,
+ metadata,
+ plan: Plan::Nothing,
+ own_missing: 0,
+ had_history: false,
+ };
+ }
- let reply = match served_copy(&host, &peer, current, wanted, request.offset).await {
- Ok(served) => {
- let start = (request.offset as usize).min(served.bytes.len());
- let end = (start + CHUNK_BYTES).min(served.bytes.len());
- if request.offset == 0 {
- info!(peer = %peer, wanted = %wanted, document_bytes = served.bytes.len(), "Serving document");
- }
- SnapshotReply {
- epoch: served.epoch,
- wanted,
- digest: served.digest,
- total: served.bytes.len() as u64,
- offset: request.offset,
- data: served.bytes[start..end].to_vec(),
- refused: String::new(),
- }
- }
- Err(reason) => {
- info!(peer = %peer, wanted = %wanted, reason = %reason, "Refusing document request");
- SnapshotReply {
- epoch: current,
- wanted,
- digest: [0; 32],
- total: 0,
- offset: request.offset,
- data: Vec::new(),
- refused: reason,
+ // The epoch that followed ours records (per actor) which of our changes it took.
+ // Snapshots carry earlier epochs' metadata forward, so this works across any gap.
+ let next = snapshot::read_epoch_metadata(&copy, previous.generation + 1)
+ .ok()
+ .filter(|next| next.previous == previous);
+
+ let own = old.get_actor().clone();
+ let clock = next
+ .as_ref()
+ .map_or_else(BTreeMap::new, |next| next.clock.clone());
+ let missing = delta::missing_from_snapshot(&mut old, &clock, &own);
+ if missing.own == 0 {
+ return Prepared::Ready {
+ old,
+ metadata,
+ plan: Plan::Nothing,
+ own_missing: 0,
+ had_history: true,
+ };
+ }
+
+ let paths = delta::changed_paths(&mut old, &missing.before, &old_heads);
+ let gap = target.generation - previous.generation;
+
+ let plan = match &next {
+ None => Plan::Quarantine {
+ paths,
+ reason: format!(
+ "the metadata of the epoch after {previous} is missing, so it is unknown which of {} local changes the cluster has",
+ missing.own
+ ),
+ },
+ Some(next) => {
+ let away_ms = unix_millis().saturating_sub(next.proposed_at);
+ if next.policy.replay_window.covers(gap, away_ms) {
+ Plan::Replay(paths)
+ } else {
+ Plan::Quarantine {
+ paths,
+ reason: format!(
+ "{} local changes are outside the replay window ({gap} epochs behind, cluster moved on {}s ago)",
+ missing.own,
+ away_ms / 1000
+ ),
+ }
}
}
};
- if let Err(error) = host
- .send_message(&peer, ParsedMessage::SnapshotReply(reply))
- .await
- {
- warn!(peer = %peer, error = %format!("{error:#}"), "Sending document chunk failed");
+ Prepared::Ready {
+ old,
+ metadata,
+ plan,
+ own_missing: missing.own,
+ had_history: true,
}
}
-/// The saved document to serve chunks from: reused while a download is in progress,
-/// saved anew when a download starts and the copy is stale.
-async fn served_copy<H: Host>(
- host: &H,
- peer: &str,
- current: Epoch,
- wanted: Epoch,
- offset: u64,
-) -> Result<Arc<Served>, String> {
- if !host.is_cluster_member(peer).await {
- return Err("not a member of this cluster (no completed handshake)".to_owned());
+/// Restores the old document and epoch, and retries later.
+async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
+ let Rejoin {
+ previous,
+ target,
+ old,
+ was_ready,
+ ..
+ } = rejoin;
+ warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch");
+
+ if let Err(never) = host
+ .switch_epoch(previous, move |_live| Ok::<_, Infallible>(old))
+ .await
+ {
+ match never {}
}
- if current != wanted {
- return Err(format!("this node is on {current}"));
+ if was_ready {
+ host.resume().await;
}
- if let Some(served) = host.compaction().served(wanted) {
- if offset > 0 || served.saved_at.elapsed() < SERVED_FRESH_FOR {
- return Ok(served);
- }
- } else if offset > 0 {
- return Err("download expired; start again".to_owned());
- }
+ let attempts = host
+ .compaction()
+ .catch_up()
+ .filter(|catch_up| catch_up.target == target)
+ .map_or(1, |catch_up| catch_up.attempts);
+ host.compaction().set_catch_up(Some(CatchUp {
+ target,
+ attempts,
+ last_attempt: Instant::now(),
+ error: Some(reason),
+ }));
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::compaction::snapshot::{build, SnapshotParams};
+ use automerge::{transaction::Transactable, ActorId, ROOT};
+ use serde_json::{json, Value as JsonValue};
- if !host.compaction().try_start_serving() {
- return Err("busy saving the document; try again".to_owned());
+ /// An old-epoch document with one own change after the base, and the cluster's
+ /// document of the next epoch (snapshot plus one foreign change).
+ fn scenario(policy: &JsonValue) -> (AutoCommit, AutoCommit, Epoch) {
+ let mut old = AutoCommit::new().with_actor(ActorId::from(&b"late-node"[..]));
+ old.put(ROOT, "base", 1_i64).unwrap();
+ old.commit();
+ let base = old.get_heads();
+
+ let snapshot = build(
+ &mut old,
+ &SnapshotParams {
+ compaction_id: "c-1",
+ previous: Epoch::GENESIS,
+ base_heads: &base,
+ policy,
+ proposed_at: 0,
+ },
+ )
+ .unwrap();
+ let target = snapshot.epoch;
+
+ let mut cluster = snapshot.doc.with_actor(ActorId::from(&b"cluster-node"[..]));
+ cluster.put(ROOT, "theirs", 1_i64).unwrap();
+ cluster.commit();
+
+ old.put(ROOT, "mine", 1_i64).unwrap();
+ old.commit();
+ (old, cluster, target)
}
- // 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;
- let saved = tokio::task::spawn_blocking(move || {
- if doc
- .get_change_by_hash(&automerge::ChangeHash(wanted.id))
- .is_some()
- {
- Some(doc.save())
- } else {
- None
+ #[test]
+ fn replays_within_the_window() {
+ let (old, cluster, target) = scenario(&json!({}));
+ match prepare(cluster, old, Epoch::GENESIS, target) {
+ Prepared::Ready {
+ plan: Plan::Replay(paths),
+ own_missing,
+ ..
+ } => {
+ assert_eq!(own_missing, 1);
+ assert_eq!(paths, vec![vec!["mine".to_owned()]]);
+ }
+ _ => panic!("expected a replay"),
}
- })
- .await;
- host.compaction().done_serving();
+ }
- let bytes = match saved {
- Ok(Some(bytes)) => bytes,
- Ok(None) => return Err("this node switched epochs".to_owned()),
- Err(error) => {
- error!(error = %error, "Saving the document for a peer failed");
- return Err("saving the document failed".to_owned());
+ #[test]
+ fn quarantines_outside_the_window() {
+ let (old, cluster, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ match prepare(cluster, old, Epoch::GENESIS, target) {
+ Prepared::Ready {
+ plan: Plan::Quarantine { paths, .. },
+ ..
+ } => assert_eq!(paths, vec![vec!["mine".to_owned()]]),
+ _ => panic!("expected a quarantine"),
}
- };
-
- let served = Arc::new(Served {
- epoch: wanted,
- digest: sha256(&bytes),
- bytes,
- saved_at: Instant::now(),
- });
- host.compaction().set_served(Arc::clone(&served));
- Ok(served)
-}
+ }
-fn sha256(bytes: &[u8]) -> [u8; 32] {
- let mut digest = [0; 32];
- digest.copy_from_slice(&Sha256::digest(bytes));
- digest
-}
+ #[test]
+ fn new_node_has_nothing_to_replay() {
+ let (_, cluster, target) = scenario(&json!({}));
+ assert!(matches!(
+ prepare(cluster, AutoCommit::new(), Epoch::GENESIS, target),
+ Prepared::Ready {
+ plan: Plan::Nothing,
+ had_history: false,
+ ..
+ }
+ ));
+ }
-/// Hands a peer's chunk to the waiting download.
-pub(crate) fn receive_snapshot_reply<H: Host>(host: &H, peer: &str, reply: SnapshotReply) {
- let offset = reply.offset;
- if !host.compaction().deliver_reply(peer, reply) {
- debug!(peer = %peer, offset, "Unexpected document chunk; dropped");
+ #[test]
+ fn waits_for_the_target_epoch() {
+ let (old, _, target) = scenario(&json!({}));
+ assert!(matches!(
+ prepare(AutoCommit::new(), old, Epoch::GENESIS, target),
+ Prepared::NotYet(_)
+ ));
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-22 16:57:24 +0000