aboutsummaryrefslogtreecommitdiffstats
path: root/src/remake/catchup.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/catchup.rs')
-rw-r--r--src/remake/catchup.rs563
1 files changed, 370 insertions, 193 deletions
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 0be7dfd..1ee9508 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -26,12 +26,16 @@
//!
//! If no peer completes a sync in time, or the check fails, the node restores its old
//! document and epoch and tries again later.
+//!
+//! Two remakes committed for one epoch make two epochs of the next generation, which
+//! never sync. A node behind follows the one more peers advertise (`most_followed`);
+//! a ready node on the other rejoins *sideways* once more ready peers are on it than on
+//! its own (`sideways`), its writes on the epoch it leaves going to the quarantine.
use std::collections::{BTreeMap, HashMap};
-use std::convert::Infallible;
use std::time::Duration;
-use automerge::{ActorId, AutoCommit, ChangeHash};
+use crate::document::{ActorId, ChangeHash, Document, ReadDoc};
use tokio::time::Instant;
use tracing::{debug, error, info, warn};
@@ -87,13 +91,17 @@ pub(super) async fn check<H: Host>(host: &H) {
.count();
// Only ready members count: a rejoining peer advertises an epoch it doesn't hold
// yet, and a node of another swarm (or a stranger) advertises one that isn't ours.
- let mut ahead: Vec<(String, Epoch)> = peers
+ let mut serving: Vec<(String, Epoch)> = peers
.into_iter()
.filter(|peer| peer.serves())
.filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
+ .collect();
+ serving.sort_by(|a, b| a.0.cmp(&b.0));
+ let mut ahead: Vec<(String, Epoch)> = serving
+ .iter()
.filter(|(_, epoch)| epoch.generation > local.generation)
+ .cloned()
.collect();
- ahead.sort_by(|a, b| a.0.cmp(&b.0));
ahead.retain(|(peer, epoch)| {
let plausible = epoch.generation - local.generation <= MAX_GENERATION_GAP;
if !plausible {
@@ -105,7 +113,8 @@ pub(super) async fn check<H: Host>(host: &H) {
// One peer's word is not enough for a ready node to give up its document: a buggy
// or hostile member could otherwise keep the swarm rejoining an epoch nobody
// serves. A newer epoch is followed once enough distinct peers advertise it.
- let required = if host.is_node_ready().await {
+ let ready = host.is_node_ready().await;
+ let required = if ready {
active_members.clamp(1, CORROBORATION)
} else {
1
@@ -114,38 +123,47 @@ pub(super) async fn check<H: Host>(host: &H) {
for (_, epoch) in &ahead {
*advertised.entry(*epoch).or_default() += 1;
}
- let corroborated: Vec<Epoch> = advertised
+ let corroborated: Vec<(Epoch, usize)> = advertised
.into_iter()
.filter(|(_, count)| *count >= required)
- .map(|(epoch, _)| epoch)
.collect();
- let Some(generation) = corroborated.iter().map(|epoch| epoch.generation).max() else {
- if host.remake().catch_up().is_some() {
- host.remake().set_catch_up(None);
+ let mut aside = false;
+ let target = match corroborated.iter().map(|(epoch, _)| epoch.generation).max() {
+ Some(generation) => {
+ let targets: Vec<(Epoch, usize)> = corroborated
+ .into_iter()
+ .filter(|(epoch, _)| epoch.generation == generation)
+ .collect();
+ // Several epochs of the newest generation (two remakes committed for one
+ // epoch): the one most peers are on, as `sideways` would pick.
+ let target = most_followed(&targets).expect("corroborated has this generation");
+ if targets.len() > 1 {
+ let listed: Vec<String> =
+ targets.iter().map(|(epoch, _)| epoch.to_string()).collect();
+ warn!(epochs = ?listed, chosen = %target, "Peers disagree about the newest epoch; following the one most of them are on");
+ }
+ target
}
- if !ahead.is_empty() {
- debug!(
- peers = ahead.len(),
- required, "A newer epoch is advertised, but not by enough peers yet"
- );
+ None => {
+ if !ahead.is_empty() {
+ debug!(
+ peers = ahead.len(),
+ required, "A newer epoch is advertised, but not by enough peers yet"
+ );
+ }
+ let sideways = ready.then(|| sideways(local, &serving, required)).flatten();
+ let Some(target) = sideways else {
+ if host.remake().catch_up().is_some() {
+ host.remake().set_catch_up(None);
+ }
+ return;
+ };
+ aside = true;
+ target
}
- return;
};
- let mut targets: Vec<Epoch> = corroborated
- .into_iter()
- .filter(|epoch| epoch.generation == generation)
- .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
.remake()
@@ -159,9 +177,50 @@ pub(super) async fn check<H: Host>(host: &H) {
}
let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
+ if aside {
+ warn!(
+ local = %local,
+ theirs = %target,
+ "More peers are on another epoch of this generation than on ours; rejoining theirs (writes made on ours since go to the quarantine)"
+ );
+ }
start(host, local, target, attempts).await;
}
+/// For a ready node, another epoch of its own generation to rejoin, if any. Two remakes
+/// committed for one epoch (`remakes.md`, Limits: a coordinator cut off after it
+/// committed, electorates of an old partition, a vote lost with a restart) leave two
+/// epochs of a generation that never sync. The one more ready members are on wins:
+/// a node on the other rejoins it, with `required` of them at least; on a tie, the
+/// epoch with the lower ID. Each node judges by the peers it sees, so the side that
+/// moves is the one that sees itself outnumbered.
+fn sideways(local: Epoch, serving: &[(String, Epoch)], required: usize) -> Option<Epoch> {
+ let mut counts: HashMap<Epoch, usize> = HashMap::new();
+ for (_, epoch) in serving {
+ if epoch.generation == local.generation {
+ *counts.entry(*epoch).or_default() += 1;
+ }
+ }
+ // This node is on its own epoch too.
+ let ours = counts.remove(&local).unwrap_or(0) + 1;
+ let others: Vec<(Epoch, usize)> = counts
+ .into_iter()
+ .filter(|(epoch, count)| {
+ *count >= required && (*count > ours || (*count == ours && epoch.id < local.id))
+ })
+ .collect();
+ most_followed(&others)
+}
+
+/// Of epochs with how many peers are on each, the one most are on; of those, the one
+/// with the lowest ID, so every node picks the same.
+fn most_followed(epochs: &[(Epoch, usize)]) -> Option<Epoch> {
+ epochs
+ .iter()
+ .max_by(|a, b| a.1.cmp(&b.1).then(b.0.id.cmp(&a.0.id)))
+ .map(|(epoch, _)| *epoch)
+}
+
async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
host.remake().set_catch_up(Some(CatchUp {
target,
@@ -170,24 +229,12 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
error: None,
}));
- // Keep answering reads, from a copy of the current document marked stale, before
- // the node stops being ready: no window in which reads get 503.
- if host.is_node_ready().await {
- let copy = host.read_doc(|doc| doc.clone()).await;
- host.set_stale_view(Some((local, copy)));
- }
- let was_ready = host.suspend().await;
let actor = ActorId::random();
host.remake().add_own_actor(actor.clone());
- let old = match host
- .switch_epoch(target, move |_live| {
- Ok::<_, Infallible>(AutoCommit::new().with_actor(actor))
- })
- .await
- {
- Ok(old) => old,
- Err(never) => match never {},
- };
+ // At once: the fresh document takes no writes until the rejoin is done; a ready
+ // node's document so far answers reads, marked stale, and takes its writes
+ // meanwhile (`Host::begin_rejoin`).
+ let begun = host.begin_rejoin(target, actor).await;
// 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).
@@ -197,9 +244,10 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
host.remake().start_rejoin(Rejoin {
previous: local,
target,
- old,
+ old: begun.old,
+ heads: begun.heads,
started: Instant::now(),
- was_ready,
+ was_ready: begun.was_ready,
});
// Sync waits until the snapshot is here (or its download failed).
@@ -218,9 +266,7 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
if still_rejoining {
match outcome {
Ok(change) => {
- let applied = host
- .commit_doc(move |doc| doc.apply_changes([change]))
- .await;
+ let applied = host.apply_change(change).await;
match applied {
Ok(()) => {
info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since")
@@ -242,30 +288,35 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
host.request_sync();
}
-/// What to do with this node's own changes the new epoch lacks.
+/// What to do with this node's own changes the new epoch lacks, those made before the
+/// rejoin.
enum Plan {
Nothing,
- /// Replay these paths, carried through every remake since the node's epoch.
- Replay(Vec<Vec<String>>, Translator),
+ /// Replay what changed from `before` to `after` in the old document, carried
+ /// through every remake since the node's epoch.
+ Replay {
+ before: Vec<ChangeHash>,
+ after: Vec<ChangeHash>,
+ translator: Translator,
+ },
Quarantine {
paths: Vec<Vec<String>>,
reason: String,
},
}
-// Every variant holds a whole document; one value lives for one step of a rejoin, so
-// boxing would only add an indirection.
-#[allow(clippy::large_enum_variant)]
enum Prepared {
/// Caught up with someone, but not with the target epoch yet: keep waiting.
- NotYet(AutoCommit),
- Failed(AutoCommit, String),
+ NotYet,
+ Failed(String),
Ready {
- old: AutoCommit,
metadata: EpochMetadata,
plan: Plan,
+ /// Where the writes the old document took during the rejoin go: fresh, they
+ /// are replayed whatever the replay window says, unless there is no telling
+ /// where (the reason).
+ during: Result<Translator, String>,
own_missing: usize,
- had_history: bool,
},
}
@@ -277,99 +328,139 @@ async fn finish<H: Host>(host: &H) {
previous,
target,
old,
+ heads: began,
started,
was_ready,
} = rejoin;
- // Everything heavy runs on a copy, without a lock.
+ // Everything heavy runs on copies, without a lock; the old document takes writes
+ // meanwhile, and those after `began` are replayed under the lock in the end.
let copy = host.read_doc(|doc| doc.clone()).await;
+ let kept = host.rejoin_document().await;
let own = host.remake().own_actors();
let now = host.now_millis();
- let prepared =
- tokio::task::spawn_blocking(move || prepare(copy, old, previous, target, own, now)).await;
+ let prepared = {
+ let began = began.clone();
+ tokio::task::spawn_blocking(move || {
+ let empty = Document::new();
+ let reference = kept.as_ref().or(old.as_ref()).unwrap_or(&empty);
+ let prepared = prepare(&copy, reference, &began, previous, target, own, now);
+ (prepared, old)
+ })
+ .await
+ };
- let (old, metadata, plan, own_missing, had_history) = match prepared {
- Ok(Prepared::Ready {
+ let (metadata, plan, during, own_missing, old) = match prepared {
+ Ok((
+ Prepared::Ready {
+ metadata,
+ plan,
+ during,
+ own_missing,
+ },
old,
- metadata,
- plan,
- own_missing,
- had_history,
- }) => (old, metadata, plan, own_missing, had_history),
- Ok(Prepared::NotYet(old)) => {
+ )) => (metadata, plan, during, own_missing, old),
+ Ok((Prepared::NotYet, old)) => {
host.reset_initial_sync();
host.remake().start_rejoin(Rejoin {
previous,
target,
old,
+ heads: began,
started,
was_ready,
});
return;
}
- Ok(Prepared::Failed(old, reason)) => {
+ Ok((Prepared::Failed(reason), old)) => {
let rejoin = Rejoin {
previous,
target,
old,
+ heads: began,
started,
was_ready,
};
return abandon(host, rejoin, reason).await;
}
Err(join_error) => {
- // The old document is gone with the panicked task; nothing to restore.
+ // A document of a node that wasn't ready is gone with the panicked task; a
+ // ready node's writes during the rejoin wait in the quarantine.
error!(error = %join_error, "Rejoin task failed; staying on the new epoch");
- if was_ready {
- host.resume().await;
- }
- host.set_stale_view(None);
+ let (unplaced, old) = host
+ .finish_rejoin(None, move |_live, old| {
+ let reason = "written during a rejoin whose check failed".to_owned();
+ delta::changed_paths(old, &began, &old.get_heads())
+ .into_iter()
+ .map(|path| (path, reason.clone()))
+ .collect::<Vec<_>>()
+ })
+ .await;
+ quarantine::add(host, previous, target, &old, unplaced);
return;
}
};
- let mut stats = ReplayStats::default();
- let mut quarantine_id = None;
- let mut keep_archive = true;
- let mut old = Some(old);
-
- match plan {
- Plan::Nothing => {}
- Plan::Replay(paths, translator) => {
- let document = old.take().expect("old document");
- let (replayed, document) = host
- .commit_doc(move |doc| {
- let stats = delta::replay_paths(&document, doc, &paths, &translator);
- Ok::<_, Infallible>((stats, document))
- })
- .await
- .unwrap_or_else(|never| match never {});
- stats = replayed;
+ // The writes, under the lock, with the switch back to serving requests.
+ let quarantined_whole = matches!(plan, Plan::Quarantine { .. });
+ let superseded = host.superseded();
+ let ((stats, unplaced), old) = host
+ .finish_rejoin(old, move |live, old| {
+ let mut stats = ReplayStats::default();
+ let mut unplaced = Vec::new();
+ match plan {
+ Plan::Nothing => {}
+ Plan::Replay {
+ before,
+ after,
+ translator,
+ } => {
+ stats.absorb(delta::replay_newest(
+ old,
+ &before,
+ &after,
+ live,
+ &translator,
+ superseded,
+ ));
+ }
+ Plan::Quarantine { paths, reason } => {
+ unplaced.extend(paths.into_iter().map(|path| (path, reason.clone())));
+ }
+ }
+ let heads = old.get_heads();
+ match during {
+ Ok(translator) => {
+ stats.absorb(delta::replay_newest(
+ old,
+ &began,
+ &heads,
+ live,
+ &translator,
+ superseded,
+ ));
+ }
+ Err(reason) => {
+ let reason = format!("written during the rejoin, but {reason}");
+ unplaced.extend(
+ delta::changed_paths(old, &began, &heads)
+ .into_iter()
+ .map(|path| (path, reason.clone())),
+ );
+ }
+ }
// Writes the remakes since had no place for.
- quarantine_id = quarantine::add(
- host,
- previous,
- target,
- &document,
- std::mem::take(&mut stats.quarantined),
- );
- old = Some(document);
- }
- Plan::Quarantine { paths, reason } => {
- let document = old.take().expect("old document");
- let entries = paths
- .into_iter()
- .map(|path| (path, reason.clone()))
- .collect();
- quarantine_id = quarantine::add(host, previous, target, &document, entries);
- keep_archive = false; // the quarantine holds the document
- }
- }
+ unplaced.append(&mut stats.quarantined);
+ (stats, unplaced)
+ })
+ .await;
+ let quarantine_id = quarantine::add(host, previous, target, &old, unplaced);
- if let (true, true, Some(document)) = (had_history, keep_archive, old) {
+ // Unless the quarantine holds the document.
+ if !old.get_heads().is_empty() && !quarantined_whole {
host.remake().archive(
previous,
- document,
+ old,
metadata.policy.archive_retention.clone(),
target.generation,
host.now_millis(),
@@ -389,11 +480,6 @@ async fn finish<H: Host>(host: &H) {
host.remake().set_catch_up(None);
let actor = host.read_doc(|doc| doc.get_actor().clone()).await;
host.remake().reset_own_actors(actor);
- if was_ready {
- host.resume().await;
- }
- // After resuming, so reads move from the stale view to the live document directly.
- host.set_stale_view(None);
info!(
from = %previous,
@@ -401,71 +487,47 @@ async fn finish<H: Host>(host: &H) {
own_changes = own_missing,
replayed = stats.replayed,
failed = stats.failed,
+ superseded = stats.superseded,
quarantine = ?quarantine_id,
secs = started.elapsed().as_secs(),
"Rejoined newer epoch"
);
}
-/// Verifies the synced document and decides about the old document's own changes.
-/// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor
-/// counts in any case. `now`: Unix milliseconds, by the node's clock.
+/// Verifies the synced document `copy` and decides about the old document's own
+/// changes: those up to `began` (when the rejoin began) by the replay window, those
+/// after it (made during the rejoin) whatever it says. `own`: this node's actors
+/// (`RemakeState::own_actors`); the old document's own actor counts in any case.
+/// `now`: Unix milliseconds, by the node's clock.
fn prepare(
- mut copy: AutoCommit,
- mut old: AutoCommit,
+ copy: &Document,
+ old: &Document,
+ began: &[ChangeHash],
previous: Epoch,
target: Epoch,
mut own: Vec<ActorId>,
now: UnixMillis,
) -> Prepared {
if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
- return Prepared::NotYet(old);
+ return Prepared::NotYet;
}
- let metadata = match snapshot::verify_document(&mut copy, target) {
+ let metadata = match snapshot::verify_document(copy, target) {
Ok(metadata) => metadata,
- Err(reason) => return Prepared::Failed(old, reason),
+ Err(reason) => return Prepared::Failed(reason),
};
- 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,
- };
- }
-
// 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)
+ let next = snapshot::read_epoch_metadata(copy, previous.generation + 1)
.ok()
.filter(|next| next.previous == previous);
-
- own.push(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;
// Writes made in our epoch go through every remake since, in order.
let translator = (previous.generation + 1..=target.generation).try_fold(
Translator::default(),
|mut translator, generation| {
- let epoch = snapshot::read_epoch_metadata(&copy, generation)?;
+ let epoch = snapshot::read_epoch_metadata(copy, generation)?;
let kind = epoch.kind().ok_or_else(|| {
format!(
"epoch generation {generation} was made by unknown remake kind {:?}",
@@ -476,8 +538,51 @@ fn prepare(
Ok::<_, String>(translator)
},
);
+ let during = match (&next, &translator) {
+ // Sideways, to another epoch of our generation: the same paths.
+ _ if gap == 0 => Ok(Translator::default()),
+ (Some(_), Ok(translator)) => Ok(translator.clone()),
+ (Some(_), Err(reason)) => Err(format!("cannot be translated: {reason}")),
+ (None, _) => Err(format!(
+ "the metadata of the epoch after {previous} is missing"
+ )),
+ };
+
+ if began.is_empty() {
+ return Prepared::Ready {
+ metadata,
+ plan: Plan::Nothing,
+ during,
+ own_missing: 0,
+ };
+ }
+
+ own.push(old.get_actor().clone());
+ let clock = next
+ .as_ref()
+ .map_or_else(BTreeMap::new, |next| next.clock.clone());
+ let missing = delta::missing_from_snapshot(old, &clock, &own);
+ if missing.own == 0 {
+ return Prepared::Ready {
+ metadata,
+ plan: Plan::Nothing,
+ during,
+ own_missing: 0,
+ };
+ }
+ // Up to when the rejoin began; what came after is `during`'s.
+ let paths = delta::changed_paths(old, &missing.before, began);
let plan = match (&next, translator) {
+ // Rejoined sideways (`sideways`): whatever was written on our epoch, the swarm
+ // went another way; nothing says where it would go.
+ _ if gap == 0 => Plan::Quarantine {
+ paths,
+ reason: format!(
+ "{} local changes were made on {previous}, which the swarm left for {target}, another epoch of the same generation",
+ missing.own
+ ),
+ },
(Some(_), Err(reason)) => Plan::Quarantine {
paths,
reason: format!("cannot translate {} local changes: {reason}", missing.own),
@@ -492,7 +597,11 @@ fn prepare(
(Some(next), Ok(translator)) => {
let away_ms = now.ms_since(next.proposed_at);
if next.policy.replay_window.covers(gap, away_ms) {
- Plan::Replay(paths, translator)
+ Plan::Replay {
+ before: missing.before.clone(),
+ after: began.to_vec(),
+ translator,
+ }
} else {
Plan::Quarantine {
paths,
@@ -507,37 +616,27 @@ fn prepare(
};
Prepared::Ready {
- old,
metadata,
plan,
+ during,
own_missing: missing.own,
- had_history: true,
}
}
-/// Restores the old document and epoch, and retries later.
+/// Restores the old document (with the writes it took meanwhile) 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 {}
- }
+ host.abandon_rejoin(previous, old).await;
// A node that was never ready must sync again before bootstrap may finish.
host.reset_initial_sync();
- if was_ready {
- host.resume().await;
- }
- host.set_stale_view(None);
let attempts = host
.remake()
@@ -555,16 +654,59 @@ async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::document::testing::change;
use crate::remake::snapshot::{SnapshotParams, build};
use automerge::{ActorId, ROOT, transaction::Transactable};
use serde_json::{Value as JsonValue, json};
+ #[test]
+ fn sideways_follows_the_epoch_more_peers_are_on() {
+ let epoch = |id: u8| Epoch {
+ generation: 3,
+ id: [id; 32],
+ };
+ let on = |peers: &[(&str, Epoch)]| -> Vec<(String, Epoch)> {
+ peers
+ .iter()
+ .map(|(peer, epoch)| ((*peer).to_owned(), *epoch))
+ .collect()
+ };
+ let (ours, theirs) = (epoch(5), epoch(9));
+ // Alone against two: theirs.
+ let serving = on(&[("b", theirs), ("c", theirs)]);
+ assert_eq!(sideways(ours, &serving, 2), Some(theirs));
+ // Two (with us) against two: the lower ID, ours; they come to us.
+ let serving = on(&[("b", theirs), ("c", theirs), ("d", ours)]);
+ assert_eq!(sideways(ours, &serving, 2), None);
+ assert_eq!(
+ sideways(theirs, &on(&[("a", ours), ("d", ours), ("b", theirs)]), 2),
+ Some(ours)
+ );
+ // Fewer than required: nothing, even outnumbered.
+ assert_eq!(sideways(ours, &on(&[("b", theirs)]), 2), None);
+ // Another generation isn't sideways.
+ let newer = Epoch {
+ generation: 4,
+ id: [1; 32],
+ };
+ assert_eq!(sideways(ours, &on(&[("b", newer), ("c", newer)]), 2), None);
+ // Several: the one most are on.
+ let other = epoch(2);
+ let serving = on(&[
+ ("b", theirs),
+ ("c", theirs),
+ ("d", theirs),
+ ("e", other),
+ ("f", other),
+ ]);
+ assert_eq!(sideways(ours, &serving, 2), Some(theirs));
+ }
+
/// An old-epoch document with one own change after the base, and the swarm's
/// document of the next epoch (snapshot plus one foreign change).
- 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();
+ fn scenario(policy: &JsonValue) -> (Document, Document, Epoch) {
+ let mut old = Document::new().with_actor(ActorId::from(&b"late-node"[..]));
+ change(&mut old, |old| old.put(ROOT, "base", 1_i64).unwrap());
let base = old.get_heads();
let snapshot = build(
@@ -584,43 +726,76 @@ mod tests {
let target = snapshot.epoch;
let mut swarm = snapshot.doc.with_actor(ActorId::from(&b"swarm-node"[..]));
- swarm.put(ROOT, "theirs", 1_i64).unwrap();
- swarm.commit();
+ change(&mut swarm, |swarm| {
+ swarm.put(ROOT, "theirs", 1_i64).unwrap()
+ });
- old.put(ROOT, "mine", 1_i64).unwrap();
- old.commit();
+ change(&mut old, |old| old.put(ROOT, "mine", 1_i64).unwrap());
(old, swarm, target)
}
#[test]
fn replays_within_the_window() {
let (old, swarm, target) = scenario(&json!({}));
+ let began = old.get_heads();
match prepare(
- swarm,
- old,
+ &swarm,
+ &old,
+ &began,
Epoch::GENESIS,
target,
Vec::new(),
UnixMillis(0),
) {
Prepared::Ready {
- plan: Plan::Replay(paths, _),
+ plan: Plan::Replay { before, after, .. },
own_missing,
..
} => {
assert_eq!(own_missing, 1);
- assert_eq!(paths, vec![vec!["mine".to_owned()]]);
+ assert_eq!(
+ delta::changed_paths(&old, &before, &after),
+ vec![vec!["mine".to_owned()]]
+ );
}
_ => panic!("expected a replay"),
}
}
+ /// Writes made during the rejoin aren't the window's: those after `began` go
+ /// wherever the remakes since put them, even outside it.
+ #[test]
+ fn writes_during_the_rejoin_are_replayed_outside_the_window() {
+ let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ let began = old.get_heads();
+ let mut during = old.clone();
+ change(&mut during, |doc| doc.put(ROOT, "later", 1_i64).unwrap());
+ match prepare(
+ &swarm,
+ &during,
+ &began,
+ Epoch::GENESIS,
+ target,
+ Vec::new(),
+ UnixMillis(0),
+ ) {
+ Prepared::Ready {
+ plan: Plan::Quarantine { paths, .. },
+ during: Ok(_),
+ ..
+ } => assert_eq!(paths, vec![vec!["mine".to_owned()]]),
+ _ => panic!("expected a quarantine before, and a replay during the rejoin"),
+ }
+ }
+
#[test]
fn quarantines_outside_the_window() {
let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+ let began = old.get_heads();
match prepare(
- swarm,
- old,
+ &swarm,
+ &old,
+ &began,
Epoch::GENESIS,
target,
Vec::new(),
@@ -639,8 +814,9 @@ mod tests {
let (_, swarm, target) = scenario(&json!({}));
assert!(matches!(
prepare(
- swarm,
- AutoCommit::new(),
+ &swarm,
+ &Document::new(),
+ &[],
Epoch::GENESIS,
target,
Vec::new(),
@@ -648,7 +824,7 @@ mod tests {
),
Prepared::Ready {
plan: Plan::Nothing,
- had_history: false,
+ own_missing: 0,
..
}
));
@@ -659,14 +835,15 @@ mod tests {
let (old, _, target) = scenario(&json!({}));
assert!(matches!(
prepare(
- AutoCommit::new(),
- old,
+ &Document::new(),
+ &old,
+ &old.get_heads(),
Epoch::GENESIS,
target,
Vec::new(),
UnixMillis(0)
),
- Prepared::NotYet(_)
+ Prepared::NotYet
));
}
}