From a9838a756fa7730e887598592a2f1b5b294e68a7 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 19:29:26 +0100 Subject: LWW everywhere, mostly --- src/remake/catchup.rs | 34 ++++-- src/remake/delta.rs | 287 +++++++++++++++++++++++++++++++++++++++++++++++++- src/remake/mod.rs | 12 +++ src/remake/switch.rs | 30 ++++-- 4 files changed, 340 insertions(+), 23 deletions(-) (limited to 'src/remake') diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 2da8540..319037d 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -49,7 +49,6 @@ use super::quarantine; use super::snapshot::{self, EpochMetadata}; use crate::clock::UnixMillis; use crate::document::replica::Epoch; -use crate::document::store::StoreError; /// Pause before trying the same target again after a failed rejoin. const RETRY: Duration = Duration::from_secs(30); @@ -308,8 +307,14 @@ async fn fetch_snapshot(host: H, target: Epoch) { /// What to do with this node's own changes the new epoch lacks. enum Plan { Nothing, - /// Replay these paths, carried through every remake since the node's epoch. - Replay(Vec>, Translator), + /// Replay these paths (what changed from `before` to `after` in the old + /// document), carried through every remake since the node's epoch. + Replay { + paths: Vec>, + before: Vec, + after: Vec, + translator: Translator, + }, Quarantine { paths: Vec>, reason: String, @@ -402,13 +407,18 @@ async fn finish(host: &H) { match plan { Plan::Nothing => {} - Plan::Replay(paths, translator) => { + Plan::Replay { + paths, + before, + after, + translator, + } => { let document = Arc::new(old.take().expect("old document")); + let superseded = host.superseded(); let replayed = { let document = Arc::clone(&document); - let paths = paths.clone(); - host.commit_doc(move |doc| { - Ok::<_, StoreError>(delta::replay_paths(&document, doc, &paths, &translator)) + host.commit_changes(move |doc| { + delta::replay_newest(&document, &before, &after, doc, &translator, superseded) }) .await }; @@ -478,6 +488,7 @@ async fn finish(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" @@ -578,7 +589,12 @@ 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 { + paths, + before: missing.before.clone(), + after: old_heads.clone(), + translator, + } } else { Plan::Quarantine { paths, @@ -733,7 +749,7 @@ mod tests { UnixMillis(0), ) { Prepared::Ready { - plan: Plan::Replay(paths, _), + plan: Plan::Replay { paths, .. }, own_missing, .. } => { diff --git a/src/remake/delta.rs b/src/remake/delta.rs index 06f2913..2c6c93d 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -20,11 +20,19 @@ //! //! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never //! replayed: the snapshot owns it. +//! +//! Last writer wins, by the writers' clocks ([`replay_newest`]): a replayed value is a +//! fresh operation, which by Automerge's order beats everything the document it goes +//! into has, newer writes included. So a path is replayed only if no change in that +//! document touched it (or a path above or below it) later than the write being +//! replayed, as the changes' timestamps say; and the replayed changes carry the times +//! of the writes they replay, not of the replay, so later replays judge by those. -use std::collections::{BTreeMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; -use crate::document::{ActorId, ChangeHash, Document, Transactable}; -use tracing::warn; +use crate::document::{ActorId, ChangeHash, Document, ReadDoc, Transactable}; +use serde::{Deserialize, Serialize}; +use tracing::{info, warn}; use super::kind::{Translation, Translator}; use crate::document::meta::{self, EPOCHS, REMAKES, VOTES}; @@ -36,6 +44,10 @@ pub(crate) struct ReplayStats { pub(crate) failed: usize, /// Paths the remake declared obsolete. pub(crate) dropped: usize, + /// Paths not replayed because a newer write touched them ([`replay_newest`]), and + /// left out (`Superseded::Drop`; with `Superseded::Quarantine`, they are among + /// `quarantined`). + pub(crate) superseded: usize, /// Paths the remake had no place for, with the reason. pub(crate) quarantined: Vec<(Vec, String)>, } @@ -46,12 +58,144 @@ impl ReplayStats { self.replayed += other.replayed; self.failed += other.failed; self.dropped += other.dropped; + self.superseded += other.superseded; self.quarantined.extend(other.quarantined); } } +/// What a replay does with a write that a newer one superseded (`--superseded-writes`): +/// leave it out, logged (last writer wins), or put it in the quarantine, where an +/// operator can still replay it. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Superseded { + #[default] + Drop, + Quarantine, +} + +/// Replays into `dst` what changed in `src` between `before` and `after` (both in +/// `src`), last writer wins: a path whose newest write there (by its change's +/// timestamp) is older than a change in `dst` touching it, or a path above or below it, +/// stays as `dst` has it (`superseded` says what becomes of the write). Ties go to the +/// replay. The replayed paths are committed grouped by the time of the writes they +/// replay, each change stamped with it. +pub(crate) fn replay_newest( + src: &Document, + before: &[ChangeHash], + after: &[ChangeHash], + dst: &mut Document, + translator: &Translator, + superseded: Superseded, +) -> ReplayStats { + let mut stats = ReplayStats::default(); + if before == after { + return stats; + } + let paths = changed_paths(src, before, after); + let written = touched(src, &between(src, before, after)); + let ours: Vec = paths + .iter() + .map(|path| newest(&written, path).unwrap_or(0)) + .collect(); + // Only changes newer than the oldest write replayed can supersede one; the + // snapshot (time 0) never does. + let oldest = ours.iter().copied().min().unwrap_or(0); + let newer: Vec = dst + .get_changes(&[]) + .iter() + .filter(|change| change.timestamp() > oldest) + .map(|change| change.hash()) + .collect(); + let theirs = touched(dst, &newer); + + let mut by_time: BTreeMap>> = BTreeMap::new(); + for (path, at) in paths.into_iter().zip(ours) { + match newest(&theirs, &path) { + Some(newer) if newer > at => { + stats.superseded += 1; + if superseded == Superseded::Quarantine { + stats.quarantined.push(( + path, + format!("a newer write (at {newer}, this one at {at}) superseded it"), + )); + } + } + _ => by_time.entry(at).or_default().push(path), + } + } + for (at, paths) in by_time { + let mut transaction = dst.transaction(); + stats.absorb(replay_paths(src, &mut transaction, &paths, translator)); + transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at)); + } + if stats.superseded > 0 { + info!( + superseded = stats.superseded, + quarantined = superseded == Superseded::Quarantine, + "Writes not replayed: newer ones touched their paths" + ); + } + if superseded == Superseded::Quarantine { + // Counted as quarantined, not as left out. + stats.superseded = 0; + } + stats +} + +/// The changes of `doc` after `before` up to `after`: those `after` builds on that +/// `before` doesn't. +fn between(doc: &Document, before: &[ChangeHash], after: &[ChangeHash]) -> Vec { + let deps: HashMap> = doc + .get_changes(before) + .iter() + .map(|change| (change.hash(), change.deps().to_vec())) + .collect(); + let mut found = HashSet::new(); + let mut todo: Vec = after.to_vec(); + while let Some(hash) = todo.pop() { + if let Some(parents) = deps.get(&hash) + && found.insert(hash) + { + todo.extend(parents.iter().copied()); + } + } + found.into_iter().collect() +} + +/// For each of `changes` (in `doc`), its time and the store paths it touched. +fn touched(doc: &Document, changes: &[ChangeHash]) -> Vec<(i64, Vec>)> { + changes + .iter() + .filter_map(|hash| { + let change = doc.get_change_by_hash(hash)?; + let paths = store::changed(doc, change.deps(), &[*hash], InLists::Whole) + .into_iter() + .map(|changed| changed.path) + .collect(); + Some((change.timestamp(), paths)) + }) + .collect() +} + +/// The time of the newest of `changes` that touched `path`, a path above it or one +/// below it. +fn newest(changes: &[(i64, Vec>)], path: &[String]) -> Option { + changes + .iter() + .filter(|(_, paths)| { + paths + .iter() + .any(|other| other.starts_with(path) || path.starts_with(other)) + }) + .map(|(at, _)| *at) + .max() +} + /// Replays into `dst` everything that changed in `src` between `before` and `after`, -/// as one commit. Both head sets must be in `src`. +/// as one commit, whatever `dst` has (no last writer wins): what a replay copies, for +/// tests. Both head sets must be in `src`. +#[cfg(test)] pub(crate) fn replay( src: &Document, before: &[ChangeHash], @@ -423,6 +567,141 @@ mod tests { assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]); } + /// Writes to `doc` as one change made at `at` (Unix ms). + fn write_at( + doc: &mut Document, + at: i64, + f: impl FnOnce(&mut automerge::transaction::Transaction<'_>), + ) { + let mut transaction = doc.transaction(); + f(&mut transaction); + transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at)); + } + + /// An old document with `x`, `m/a` and `m/b` written at 10 after the base, and a new + /// one holding the base; `newer` writes to the new one first. + fn superseding( + newer: impl FnOnce(&mut Document), + superseded: Superseded, + ) -> (Document, ReplayStats) { + let mut old = Document::new(); + write_at(&mut old, 1, |old| { + let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); + old.put(&m, "a", "base").unwrap(); + }); + let base = old.get_heads(); + let mut new = Document::new(); + write_at(&mut new, 0, |new| { + store::materialize_into(&old, new, &|_| false).unwrap(); + }); + newer(&mut new); + write_at(&mut old, 10, |old| { + old.put(ROOT, "x", "old").unwrap(); + let (_, m) = old.get(ROOT, "m").unwrap().unwrap(); + old.put(&m, "a", "old").unwrap(); + old.put(&m, "b", "old").unwrap(); + }); + let after = old.get_heads(); + let stats = replay_newest( + &old, + &base, + &after, + &mut new, + &Translator::default(), + superseded, + ); + (new, stats) + } + + #[test] + fn replay_newest_keeps_newer_writes() { + // Nothing newer: everything replayed, at the time it was written. + let (new, stats) = superseding(|_| {}, Superseded::Drop); + assert_eq!((stats.replayed, stats.superseded), (3, 0)); + assert_eq!( + view(&new), + json!({ "x": "old", "m": { "a": "old", "b": "old" } }) + ); + let times: Vec = new.get_changes(&[]).iter().map(|c| c.timestamp()).collect(); + assert_eq!(times, [0, 10]); + + // A newer write at the path wins; an older one doesn't. + let (new, stats) = superseding( + |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()), + Superseded::Drop, + ); + assert_eq!((stats.replayed, stats.superseded), (2, 1)); + assert_eq!(view(&new)["x"], json!("newer")); + assert_eq!(view(&new)["m"], json!({ "a": "old", "b": "old" })); + let (new, _) = superseding( + |new| write_at(new, 5, |new| new.put(ROOT, "x", "older").unwrap()), + Superseded::Drop, + ); + assert_eq!(view(&new)["x"], json!("old")); + + // Per key: a newer `m/a` wins over the replayed `m/a`, not over `m/b`. + let (new, stats) = superseding( + |new| { + write_at(new, 20, |new| { + let (_, m) = new.get(ROOT, "m").unwrap().unwrap(); + new.put(&m, "a", "newer").unwrap(); + }) + }, + Superseded::Drop, + ); + assert_eq!(stats.superseded, 1); + assert_eq!(view(&new)["m"], json!({ "a": "newer", "b": "old" })); + // A newer write above the paths (`m` replaced) wins over both. + let (new, stats) = superseding( + |new| write_at(new, 20, |new| new.put(ROOT, "m", "a scalar now").unwrap()), + Superseded::Drop, + ); + assert_eq!(stats.superseded, 2); + assert_eq!(view(&new)["m"], json!("a scalar now")); + + // Quarantined instead, with the time of each. + let (_, stats) = superseding( + |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()), + Superseded::Quarantine, + ); + assert_eq!(stats.superseded, 0); + assert_eq!(stats.quarantined.len(), 1); + assert_eq!(stats.quarantined[0].0, ["x"]); + } + + /// A newer write below a replayed path wins too: the replay would replace it. + #[test] + fn replay_newest_keeps_newer_writes_below() { + let mut old = Document::new(); + write_at(&mut old, 1, |old| old.put(ROOT, "base", true).unwrap()); + let base = old.get_heads(); + let mut new = Document::new(); + write_at(&mut new, 0, |new| { + store::materialize_into(&old, new, &|_| false).unwrap(); + new.put_object(ROOT, "m", ObjType::Map).unwrap(); + }); + write_at(&mut new, 20, |new| { + let (_, m) = new.get(ROOT, "m").unwrap().unwrap(); + new.put(&m, "a", "newer").unwrap(); + }); + // The whole map, older. + write_at(&mut old, 10, |old| { + let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); + old.put(&m, "a", "old").unwrap(); + }); + let after = old.get_heads(); + let stats = replay_newest( + &old, + &base, + &after, + &mut new, + &Translator::default(), + Superseded::Drop, + ); + assert_eq!(stats.superseded, 1); + assert_eq!(view(&new)["m"], json!({ "a": "newer" })); + } + #[test] fn replays_state_per_key() { // Base shared by the old document and the snapshot. diff --git a/src/remake/mod.rs b/src/remake/mod.rs index 01e1e3e..0d3d073 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -37,6 +37,7 @@ mod vote; #[cfg(test)] pub(crate) use api::testing; +pub(crate) use delta::Superseded; pub(crate) use kind::remake_genesis; pub(crate) use local::RemakeState; pub(crate) use switch::run_worker; @@ -193,6 +194,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { T: Send + 'static, E: From + Send + 'static; + /// What a replay does with writes a newer one superseded (`--superseded-writes`). + fn superseded(&self) -> Superseded; + + /// Changes `f` commits itself into the live document (as many as it likes, each + /// with its own time), under the write lock, then published and replicated. + /// Refused while the document isn't writable, as [`Host::commit_doc`]. + fn commit_changes(&self, f: F) -> impl Future> + Send + where + F: FnOnce(&mut Document) -> T + Send + 'static, + T: Send + 'static; + /// Lets writes in again, after a rejoin swapped in a document that took none /// ([`Host::switch_epoch`] with `writable` false). fn accept_writes(&self) -> impl Future + Send; diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 9b03789..7d6611d 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -193,6 +193,7 @@ async fn switch( .translator() .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?; let late_translator = translator.clone(); + let superseded = host.superseded(); let previous = request.previous; let base = request.base_heads.clone(); let retention = record.proposal.policy.archive_retention.clone(); @@ -235,20 +236,20 @@ async fn switch( // A fresh actor: this node's writes in the new epoch. let mut doc = snapshot.doc.with_actor(actor); - // The carried record and the replayed writes go into one change. let heads = copy.get_heads(); - let paths = delta::changed_paths(©, &base, &heads); - let mut transaction = doc.transaction(); // Keep this remake's record visible in the new epoch, for its history. - if carry_record - && let Err(error) = store::copy_path(©, &mut transaction, &record_path) - { - warn!(error = %error, "Could not carry the remake record into the new epoch"); + if carry_record { + let mut transaction = doc.transaction(); + if let Err(error) = store::copy_path(©, &mut transaction, &record_path) { + warn!(error = %error, "Could not carry the remake record into the new epoch"); + } + transaction.commit(); } - let stats = delta::replay_paths(©, &mut transaction, &paths, &translator); - transaction.commit(); + // Each write at the time it was made, so later replays judge by it. + let stats = + delta::replay_newest(©, &base, &heads, &mut doc, &translator, superseded); Ok((doc, heads, stats)) }, ) @@ -283,6 +284,7 @@ async fn switch( replayed = stats.replayed, failed = stats.failed, dropped = stats.dropped, + superseded = stats.superseded, quarantine = ?quarantine, "Switched to the new epoch" ); @@ -299,6 +301,7 @@ pub(super) async fn swap_in( copy_heads: Vec, translator: Translator, ) -> Result<(Document, ReplayStats), String> { + let superseded = host.superseded(); // The closure runs on the blocking pool; its replay numbers come back through this. let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default())); let late_out = std::sync::Arc::clone(&late_stats); @@ -310,7 +313,14 @@ pub(super) async fn swap_in( } let mut new_doc = new_doc; let live_heads = live.get_heads(); - let late = delta::replay(live, ©_heads, &live_heads, &mut new_doc, &translator); + let late = delta::replay_newest( + live, + ©_heads, + &live_heads, + &mut new_doc, + &translator, + superseded, + ); if late.replayed + late.failed > 0 { info!( replayed = late.replayed, -- cgit v1.3.1