diff options
Diffstat (limited to 'src/remake/delta.rs')
| -rw-r--r-- | src/remake/delta.rs | 504 |
1 files changed, 379 insertions, 125 deletions
diff --git a/src/remake/delta.rs b/src/remake/delta.rs index 80e5443..638231c 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -20,15 +20,23 @@ //! //! 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 automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop}; -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}; -use crate::document::store::{self, StoreError}; +use crate::document::store::{self, InLists, StoreError}; #[derive(Debug, Default, Clone)] pub(crate) struct ReplayStats { @@ -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>, String)>, } @@ -46,17 +58,148 @@ 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<i64> = 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<ChangeHash> = dst + .get_changes(&[]) + .iter() + .filter(|change| change.timestamp() > oldest) + .map(|change| change.hash()) + .collect(); + let theirs = touched(dst, &newer); + + let mut by_time: BTreeMap<i64, Vec<Vec<String>>> = 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<ChangeHash> { + let deps: HashMap<ChangeHash, Vec<ChangeHash>> = doc + .get_changes(before) + .iter() + .map(|change| (change.hash(), change.deps().to_vec())) + .collect(); + let mut found = HashSet::new(); + let mut todo: Vec<ChangeHash> = 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, from its +/// operations (`store::changed_by`: a diff per change would cost what the maps it +/// touched hold, `remake::bench`). +fn touched(doc: &Document, changes: &[ChangeHash]) -> Vec<(i64, Vec<Vec<String>>)> { + changes + .iter() + .filter_map(|hash| { + let change = doc.get_change_by_hash(hash)?; + let paths = store::changed_by(doc, hash, InLists::Whole)?; + 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<Vec<String>>)], path: &[String]) -> Option<i64> { + 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: &mut AutoCommit, + src: &Document, before: &[ChangeHash], after: &[ChangeHash], - dst: &mut AutoCommit, + dst: &mut Document, translator: &Translator, ) -> ReplayStats { if before == after { @@ -64,14 +207,17 @@ pub(crate) fn replay( } let paths = changed_paths(src, before, after); - replay_paths(src, dst, &paths, translator) + let mut transaction = dst.transaction(); + let stats = replay_paths(src, &mut transaction, &paths, translator); + transaction.commit(); + stats } -/// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst`, as one -/// commit. Lets callers find the paths outside a lock and copy them inside it. +/// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst` (the +/// caller commits). Lets callers find the paths outside a lock and copy them inside it. pub(crate) fn replay_paths( - src: &AutoCommit, - dst: &mut AutoCommit, + src: &Document, + dst: &mut impl Transactable, paths: &[Vec<String>], translator: &Translator, ) -> ReplayStats { @@ -79,13 +225,12 @@ pub(crate) fn replay_paths( for path in paths { replay_path(src, dst, path, translator, &mut stats); } - let _ = dst.commit(); stats } fn replay_path( - src: &AutoCommit, - dst: &mut AutoCommit, + src: &Document, + dst: &mut impl Transactable, path: &[String], translator: &Translator, stats: &mut ReplayStats, @@ -164,7 +309,7 @@ pub(crate) struct Missing { /// replayed by the nodes that 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, + doc: &Document, clock: &BTreeMap<String, u64>, own: &[ActorId], ) -> Missing { @@ -225,11 +370,14 @@ pub(crate) fn missing_from_snapshot( /// 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( - doc: &mut AutoCommit, + doc: &Document, before: &[ChangeHash], after: &[ChangeHash], ) -> Vec<Vec<String>> { - let mut paths: Vec<Vec<String>> = doc.diff(before, after).iter().map(patch_path).collect(); + let mut paths: Vec<Vec<String>> = store::changed(doc, before, after, InLists::Whole) + .into_iter() + .map(|changed| changed.path) + .collect(); paths.sort(); paths.dedup(); @@ -245,49 +393,26 @@ pub(crate) fn changed_paths( minimal } -fn patch_path(patch: &automerge::Patch) -> Vec<String> { - let mut path = Vec::new(); - for (_, prop) in &patch.path { - match prop { - Prop::Map(key) => path.push(key.clone()), - // Inside a list: the list is replayed as a whole. - Prop::Seq(_) => return path, - } - } - - match &patch.action { - PatchAction::PutMap { key, .. } | PatchAction::DeleteMap { key, .. } => { - path.push(key.clone()) - } - PatchAction::Increment { - prop: Prop::Map(key), - .. - } => path.push(key.clone()), - // List and text edits: the containing object. - _ => {} - } - path -} - #[cfg(test)] mod tests { use super::*; + use crate::document::testing::{change, try_change}; use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, transaction::Transactable}; use serde_json::json; - fn view(doc: &AutoCommit) -> serde_json::Value { + fn view(doc: &Document) -> serde_json::Value { store::read_path(doc, &[]).unwrap().unwrap() } /// A random store write as the API makes it (one change): a scalar or a small map /// at a path up to three keys deep, or a delete. Writes the store refuses (e.g. into /// a scalar) are rolled back. - fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) { + fn random_write(rng: &mut crate::testrand::Rng, doc: &mut Document) { const KEYS: [&str; 3] = ["a", "b", "c"]; let depth = rng.between(1, 3); let path: Vec<String> = (0..depth).map(|_| rng.pick(&KEYS).to_string()).collect(); - let result = if rng.chance(0.2) { - store::delete_path(doc, &path, false) + let _ = if rng.chance(0.2) { + try_change(doc, |doc| store::delete_path(doc, &path, false)) } else { let value = match rng.below(4) { 0 => json!(rng.below(100)), @@ -299,16 +424,8 @@ mod tests { serde_json::Value::Object(map) } }; - store::write_path(doc, &path, &value) + try_change(doc, |doc| store::write_path(doc, &path, &value)) }; - match result { - Ok(()) => { - doc.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); - } - Err(_) => { - doc.rollback(); - } - } } /// Compaction's promise: a snapshot of the base plus everything replayed since gives @@ -317,16 +434,17 @@ mod tests { fn replay_reproduces_the_document() { crate::testrand::for_seeds(200, |rng| { let actor = |i: u8| ActorId::from(&[i; 16][..]); - let mut old = AutoCommit::new().with_actor(actor(0)); + let mut old = Document::new().with_actor(actor(0)); for _ in 0..rng.between(0, 8) { random_write(rng, &mut old); } let base = old.get_heads(); - let mut new = AutoCommit::new().with_actor(actor(100)); - store::materialize_into(&old, &mut new, &|_| false).unwrap(); - new.commit(); + let mut new = Document::new().with_actor(actor(100)); + change(&mut new, |new| { + store::materialize_into(&old, new, &|_| false).unwrap() + }); - let mut nodes: Vec<AutoCommit> = (1..=rng.between(1, 3)) + let mut nodes: Vec<Document> = (1..=rng.between(1, 3)) .map(|i| old.fork().with_actor(actor(i as u8))) .collect(); for _ in 0..rng.between(1, 20) { @@ -338,7 +456,7 @@ mod tests { } let after = old.get_heads(); - let stats = replay(&mut old, &base, &after, &mut new, &Translator::default()); + let stats = replay(&old, &base, &after, &mut new, &Translator::default()); assert_eq!(stats.failed, 0); assert_eq!(view(&new), view(&old)); }); @@ -369,12 +487,12 @@ mod tests { fn changed_paths_cover_every_change_minimally() { crate::testrand::for_seeds(300, |rng| { let actor = |i: u8| ActorId::from(&[i; 16][..]); - let mut doc = AutoCommit::new().with_actor(actor(0)); + let mut doc = Document::new().with_actor(actor(0)); for _ in 0..rng.between(0, 8) { random_write(rng, &mut doc); } let before = doc.get_heads(); - let mut forks: Vec<AutoCommit> = (1..=rng.between(1, 3)) + let mut forks: Vec<Document> = (1..=rng.between(1, 3)) .map(|i| doc.fork().with_actor(actor(i as u8))) .collect(); for _ in 0..rng.between(0, 12) { @@ -386,7 +504,7 @@ mod tests { } let after = doc.get_heads(); - let paths = changed_paths(&mut doc, &before, &after); + let paths = changed_paths(&doc, &before, &after); for pair in paths.windows(2) { assert!(pair[0] < pair[1], "not sorted: {paths:?}"); assert!(!pair[1].starts_with(&pair[0]), "not minimal: {paths:?}"); @@ -415,23 +533,25 @@ mod tests { #[test] fn minimal_paths() { - let mut doc = AutoCommit::new(); - doc.put(ROOT, "untouched", 1_i64).unwrap(); + let mut doc = Document::new(); + change(&mut doc, |doc| doc.put(ROOT, "untouched", 1_i64).unwrap()); let before = doc.get_heads(); - let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap(); - doc.put(&m, "a", 1_i64).unwrap(); - doc.put(ROOT, "x", 1_i64).unwrap(); - doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap(); - doc.commit(); + change(&mut doc, |doc| { + let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap(); + doc.put(&m, "a", 1_i64).unwrap(); + doc.put(ROOT, "x", 1_i64).unwrap(); + doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap(); + }); let middle = doc.get_heads(); - doc.increment(ROOT, "c", 2).unwrap(); - let list = doc.put_object(ROOT, "l", ObjType::List).unwrap(); - doc.insert(&list, 0, "e").unwrap(); - doc.commit(); + change(&mut doc, |doc| { + doc.increment(ROOT, "c", 2).unwrap(); + let list = doc.put_object(ROOT, "l", ObjType::List).unwrap(); + doc.insert(&list, 0, "e").unwrap(); + }); let after = doc.get_heads(); - let all = changed_paths(&mut doc, &before, &after); + let all = changed_paths(&doc, &before, &after); assert_eq!( all, vec![ @@ -442,37 +562,176 @@ mod tests { ] ); - let later = changed_paths(&mut doc, &middle, &after); + let later = changed_paths(&doc, &middle, &after); 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<i64> = 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. - let mut old = AutoCommit::new(); - let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); - old.put(&m, "keep", "base").unwrap(); - old.put(&m, "change", "base").unwrap(); - old.put(ROOT, "gone", true).unwrap(); - old.commit(); + let mut old = Document::new(); + let m = change(&mut old, |old| { + let m = old.put_object(ROOT, "m", ObjType::Map).unwrap(); + old.put(&m, "keep", "base").unwrap(); + old.put(&m, "change", "base").unwrap(); + old.put(ROOT, "gone", true).unwrap(); + m + }); let base = old.get_heads(); - let mut new = AutoCommit::new(); - store::materialize_into(&old, &mut new, &|_| false).unwrap(); - // Something the new epoch already holds that the delta must not clobber. - let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else { - panic!("m missing"); - }; - new.put(&m_new, "keep", "new epoch").unwrap(); - new.commit(); + let mut new = Document::new(); + change(&mut new, |new| { + store::materialize_into(&old, new, &|_| false).unwrap(); + // Something the new epoch already holds that the delta must not clobber. + let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else { + panic!("m missing"); + }; + new.put(&m_new, "keep", "new epoch").unwrap(); + }); // The delta: one key changed, one key deleted. - old.put(&m, "change", "delta").unwrap(); - old.delete(ROOT, "gone").unwrap(); - old.commit(); + change(&mut old, |old| { + old.put(&m, "change", "delta").unwrap(); + old.delete(ROOT, "gone").unwrap(); + }); let after = old.get_heads(); - let stats = replay(&mut old, &base, &after, &mut new, &Translator::default()); + let stats = replay(&old, &base, &after, &mut new, &Translator::default()); assert_eq!(stats.failed, 0); assert_eq!( view(&new), @@ -485,9 +744,8 @@ mod tests { 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 mut doc = Document::new().with_actor(own.clone()); + change(&mut doc, |doc| doc.put(ROOT, "base", 1_i64).unwrap()); let base = doc.get_heads(); // The snapshot contains the base only. @@ -495,28 +753,28 @@ mod tests { // 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(); + change(&mut foreign, |foreign| { + foreign.put(ROOT, "theirs", 1_i64).unwrap() + }); doc.merge(&mut foreign).unwrap(); let with_foreign = doc.get_heads(); // ... and one of ours. - doc.put(ROOT, "mine", 1_i64).unwrap(); - doc.commit(); + change(&mut doc, |doc| doc.put(ROOT, "mine", 1_i64).unwrap()); let after = doc.get_heads(); - let missing = missing_from_snapshot(&mut doc, &clock, std::slice::from_ref(&own)); + let missing = missing_from_snapshot(&doc, &clock, std::slice::from_ref(&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), + changed_paths(&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, std::slice::from_ref(&own)); + let none = missing_from_snapshot(&doc, &full, std::slice::from_ref(&own)); assert_eq!(none.own, 0); let mut heads = after.clone(); heads.sort(); @@ -525,32 +783,28 @@ mod tests { #[test] fn protocol_metadata_is_never_replayed() { - let mut old = AutoCommit::new(); - old.put(ROOT, "x", 1_i64).unwrap(); - old.commit(); + let mut old = Document::new(); + change(&mut old, |old| old.put(ROOT, "x", 1_i64).unwrap()); let base = old.get_heads(); - let mut new = AutoCommit::new(); - store::write_path( - &mut new, - &meta::path(&[EPOCHS, "1"]), - &json!({ "generation": 1 }), - ) - .unwrap(); - new.commit(); + let mut new = Document::new(); + change(&mut new, |new| { + store::write_path( + new, + &meta::path(&[EPOCHS, "1"]), + &json!({ "generation": 1 }), + ) + .unwrap() + }); // The old epoch replaces `_continuum` wholesale and writes a remake record. - store::write_path( - &mut old, - &meta::path(&[]), - &json!({ "forgotten": { "pk": 1 } }), - ) - .unwrap(); - store::write_path(&mut old, &meta::path(&[REMAKES, "c"]), &json!({ "x": 1 })).unwrap(); - old.commit(); + change(&mut old, |old| { + store::write_path(old, &meta::path(&[]), &json!({ "forgotten": { "pk": 1 } })).unwrap(); + store::write_path(old, &meta::path(&[REMAKES, "c"]), &json!({ "x": 1 })).unwrap(); + }); let after = old.get_heads(); - replay(&mut old, &base, &after, &mut new, &Translator::default()); + replay(&old, &base, &after, &mut new, &Translator::default()); let meta = &view(&new)[meta::ROOT]; assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1)); assert_eq!(meta["forgotten"]["pk"], json!(1)); |