diff options
| -rw-r--r-- | docs/remakes.md | 4 | ||||
| -rw-r--r-- | src/document/replica.rs | 4 | ||||
| -rw-r--r-- | src/remake/bench.rs | 1 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 3 | ||||
| -rw-r--r-- | src/remake/delta.rs | 202 | ||||
| -rw-r--r-- | src/remake/switch.rs | 17 |
6 files changed, 203 insertions, 28 deletions
diff --git a/docs/remakes.md b/docs/remakes.md index 7b21b9e..cad429f 100644 --- a/docs/remakes.md +++ b/docs/remakes.md @@ -305,7 +305,9 @@ The swarm stays writable throughout. * **Last writer wins**: a write is carried over only if nothing newer touched its path, a path above it or one below it, in the new epoch; newer by the times the writers' clocks gave their changes. A carried-over write keeps the time it was made, so a - later carry-over judges by that. A write that loses is left out, logged (`Writes not + later carry-over judges by that, and the user it was made for (`GET /admin/changes`); + the carrying node commits it. Another member's own entry (its alive stamp, its + links) is carried over by that member only. A write that loses is left out, logged (`Writes not replayed: newer ones touched their paths`), or with `--superseded-writes quarantine` kept in the quarantine. Clocks that disagree (beyond `--clock-skew-warn`) can make the older write win. diff --git a/src/document/replica.rs b/src/document/replica.rs index ead42f6..a409969 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -385,7 +385,7 @@ pub(crate) const USER_TRAILER: &str = "User"; impl Writer<'_> { /// The change's message: its trailers (`trailers`), if it has any. - fn message(&self) -> Option<String> { + pub(crate) fn message(&self) -> Option<String> { self.user.map(|user| format!("{USER_TRAILER}: {user}")) } } @@ -481,7 +481,7 @@ fn run<T, E>( /// Makes `node` the author of `doc`'s changes from now on. Setting an author gives the /// document a new actor; this node keeps its own (`with_actor`), which it tracks /// (`RemakeState::own_actors`, the document file), as only it writes with it. -fn authored(doc: &mut Document, node: &str) { +pub(crate) fn authored(doc: &mut Document, node: &str) { if doc .get_author() .is_some_and(|author| author.as_bytes() == node.as_bytes()) diff --git a/src/remake/bench.rs b/src/remake/bench.rs index fa9f358..96f76d8 100644 --- a/src/remake/bench.rs +++ b/src/remake/bench.rs @@ -33,6 +33,7 @@ fn replay(keys: usize, late: usize, newer: usize) { &base, &after, &mut target, + "node", &Translator::default(), Superseded::Drop, )); diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 1ee9508..71f077a 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -404,6 +404,7 @@ async fn finish<H: Host>(host: &H) { // The writes, under the lock, with the switch back to serving requests. let quarantined_whole = matches!(plan, Plan::Quarantine { .. }); let superseded = host.superseded(); + let node = host.node_id(); let ((stats, unplaced), old) = host .finish_rejoin(old, move |live, old| { let mut stats = ReplayStats::default(); @@ -420,6 +421,7 @@ async fn finish<H: Host>(host: &H) { &before, &after, live, + &node, &translator, superseded, )); @@ -436,6 +438,7 @@ async fn finish<H: Host>(host: &H) { &began, &heads, live, + &node, &translator, superseded, )); diff --git a/src/remake/delta.rs b/src/remake/delta.rs index 6864c4d..06508a3 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -21,6 +21,11 @@ //! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never //! replayed: the snapshot owns it. //! +//! Replayed changes are the replaying node's, and name the user each write was made +//! for (`replica::Writer`), as rules by path judge them (`access::rules`): they are +//! grouped by user as by time. Another member's entry written without a user (its +//! alive stamp, its links) is left to that member to replay: only it may write 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 @@ -78,13 +83,15 @@ pub(crate) enum Superseded { /// `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. +/// replay. The replayed paths are committed by `node`, grouped by the time of the writes +/// they replay and the user each was made for, each change stamped with the one and +/// naming the other. pub(crate) fn replay_newest( src: &Document, before: &[ChangeHash], after: &[ChangeHash], dst: &mut Document, + node: &str, translator: &Translator, superseded: Superseded, ) -> ReplayStats { @@ -92,15 +99,19 @@ pub(crate) fn replay_newest( 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 + // Split where several users wrote below a path: each part replayed for its own. + let paths = by_user(src, dst, &written, changed_paths(src, before, after)); + let ours: Vec<(i64, Option<String>)> = paths .iter() - .map(|path| newest(&written, path).unwrap_or(0)) + .map(|path| { + let at = newest(&written, path).map_or(0, |(at, _)| at); + (at, writer(&written, path)) + }) .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 oldest = ours.iter().map(|(at, _)| *at).min().unwrap_or(0); let newer: Vec<ChangeHash> = dst .get_changes(&[]) .iter() @@ -109,10 +120,13 @@ pub(crate) fn replay_newest( .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) { + let mut by_time: BTreeMap<(i64, Option<String>), Vec<Vec<String>>> = BTreeMap::new(); + for (path, (at, user)) in paths.into_iter().zip(ours) { + if user.is_none() && others_entry(&path, node) { + continue; + } match newest(&theirs, &path) { - Some(newer) if newer > at => { + Some((newer, _)) if newer > at => { stats.superseded += 1; if superseded == Superseded::Quarantine { stats.quarantined.push(( @@ -121,13 +135,22 @@ pub(crate) fn replay_newest( )); } } - _ => by_time.entry(at).or_default().push(path), + _ => by_time.entry((at, user)).or_default().push(path), } } - for (at, paths) in by_time { + crate::document::replica::authored(dst, node); + for ((at, user), 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)); + let writer = crate::document::replica::Writer { + node, + user: user.as_deref(), + }; + let mut options = automerge::transaction::CommitOptions::default().with_time(at); + if let Some(message) = writer.message() { + options = options.with_message(message); + } + transaction.commit_with(options); } if stats.superseded > 0 { info!( @@ -163,32 +186,99 @@ fn between(doc: &Document, before: &[ChangeHash], after: &[ChangeHash]) -> Vec<C 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>>)> { +/// A change's time, the user it names (`replica::user_of`), and the store paths it +/// touched. +type Touched = (i64, Option<String>, Vec<Vec<String>>); + +/// For each of `changes` (in `doc`), its time, its user 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<Touched> { 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)) + let user = change + .message() + .and_then(|message| crate::document::replica::user_of(message)) + .map(str::to_owned); + Some((change.timestamp(), user, 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> { +/// The time and user of the newest of `changes` that touched `path`, a path above it or +/// one below it (of two at once, the user named last in order). +fn newest(changes: &[Touched], path: &[String]) -> Option<(i64, Option<String>)> { changes .iter() - .filter(|(_, paths)| { + .filter(|(_, _, paths)| { paths .iter() .any(|other| other.starts_with(path) || path.starts_with(other)) }) - .map(|(at, _)| *at) + .map(|(at, user, _)| (*at, user.clone())) + .max() +} + +/// The user of the newest of `changes` that wrote at `path` or below it: whose write it +/// is (of two at once, the one named last in order). +fn writer(changes: &[Touched], path: &[String]) -> Option<String> { + changes + .iter() + .filter(|(_, _, paths)| paths.iter().any(|other| other.starts_with(path))) + .map(|(at, user, _)| (*at, user.clone())) .max() + .and_then(|(_, user)| user) +} + +/// `paths`, each a map that writes for more than one user changed below split into its +/// entries (in `src` or `dst`), until each part is one user's: a replayed change names +/// one user, and rules by path judge it by that user (`access::rules`). +fn by_user( + src: &Document, + dst: &Document, + changes: &[Touched], + paths: Vec<Vec<String>>, +) -> Vec<Vec<String>> { + let mut out = Vec::new(); + let mut todo = paths; + while let Some(path) = todo.pop() { + let users: HashSet<&Option<String>> = changes + .iter() + .filter(|(_, _, paths)| paths.iter().any(|other| other.starts_with(&path))) + .map(|(_, user, _)| user) + .collect(); + let keys = store::keys_at(src, &path).ok().flatten(); + match keys { + Some(mut keys) if users.len() > 1 => { + keys.extend( + store::keys_at(dst, &path) + .ok() + .flatten() + .unwrap_or_default(), + ); + keys.sort(); + keys.dedup(); + for key in keys { + let mut child = path.clone(); + child.push(key); + todo.push(child); + } + } + _ => out.push(path), + } + } + out.sort(); + out +} + +/// Whether `path` is in another member's entry than `node`'s (`_continuum/members/<pk>`): +/// only that member writes it of its own (`access::rules`). +fn others_entry(path: &[String], node: &str) -> bool { + matches!(path, [root, members, owner, ..] + if root == meta::ROOT && members == meta::MEMBERS && owner != node) } /// Replays into `dst` everything that changed in `src` between `before` and `after`, @@ -577,6 +667,72 @@ mod tests { transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at)); } + /// Replayed changes are the replaying node's and name each write's user; another + /// member's entry, written without a user, is left to that member. + #[test] + fn replays_keep_their_writers() { + let path = |text: &str| -> Vec<String> { text.split('/').map(str::to_owned).collect() }; + let mut old = Document::new(); + change(&mut old, |doc| { + store::write_path(doc, &path("_continuum/members/node"), &json!({})).unwrap(); + store::write_path(doc, &path("_continuum/members/other"), &json!({})).unwrap(); + }); + let base = old.get_heads(); + let mut new = old.clone().with_actor(automerge::ActorId::random()); + let mut commit = |at: i64, user: Option<&str>, write: &str, value: serde_json::Value| { + let mut transaction = old.transaction(); + store::write_path(&mut transaction, &path(write), &value).unwrap(); + let mut options = automerge::transaction::CommitOptions::default().with_time(at); + if let Some(user) = user { + options = options.with_message(format!("User: {user}")); + } + transaction.commit_with(options); + }; + commit(10, Some("alice"), "notes/x", json!(1)); + commit(10, Some("bob"), "notes/y", json!(2)); + commit(11, None, "_continuum/members/other/alive", json!(11)); + commit(11, None, "_continuum/members/node/alive", json!(11)); + let after = old.get_heads(); + let new_heads = new.get_heads(); + let stats = replay_newest( + &old, + &base, + &after, + &mut new, + "node", + &Translator::default(), + Superseded::Drop, + ); + assert_eq!(stats.replayed, 3); + assert!( + store::read_path(&new, &path("_continuum/members/other/alive")) + .unwrap() + .is_none() + ); + assert_eq!( + store::read_path(&new, &path("_continuum/members/node/alive")).unwrap(), + Some(json!(11)) + ); + // One change per user and time, each by the node, naming its user. + let replayed: Vec<(Option<String>, i64)> = new + .get_changes(&new_heads) + .iter() + .map(|change| (change.message().map(str::to_owned), change.timestamp())) + .collect(); + assert_eq!( + replayed, + vec![ + (Some("User: alice".to_owned()), 10), + (Some("User: bob".to_owned()), 10), + (None, 11), + ] + ); + assert_eq!( + new.get_author().map(|author| author.as_bytes().to_vec()), + Some(b"node".to_vec()) + ); + } + /// 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( @@ -606,6 +762,7 @@ mod tests { &base, &after, &mut new, + "node", &Translator::default(), superseded, ); @@ -694,6 +851,7 @@ mod tests { &base, &after, &mut new, + "node", &Translator::default(), Superseded::Drop, ); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 4c72b39..dd08b38 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -216,6 +216,7 @@ async fn switch<H: Host>( let actor = ActorId::random(); host.remake().add_own_actor(actor.clone()); let new_actor = actor.clone(); + let node = host.node_id(); // 2. Build (or reuse), verify and replay without holding any lock. let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking( @@ -233,8 +234,9 @@ async fn switch<H: Host>( )); } - // A fresh actor: this node's writes in the new epoch. + // A fresh actor: this node's writes in the new epoch, made by it. let mut doc = snapshot.doc.with_actor(actor); + crate::document::replica::authored(&mut doc, &node); let heads = copy.get_heads(); @@ -248,8 +250,15 @@ async fn switch<H: Host>( } // 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); + let stats = delta::replay_newest( + ©, + &base, + &heads, + &mut doc, + &node, + &translator, + superseded, + ); Ok((doc, heads, stats)) }, ) @@ -302,6 +311,7 @@ pub(super) async fn swap_in<H: Host>( translator: Translator, ) -> Result<(Document, ReplayStats), String> { let superseded = host.superseded(); + let node = host.node_id(); // 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); @@ -318,6 +328,7 @@ pub(super) async fn swap_in<H: Host>( ©_heads, &live_heads, &mut new_doc, + &node, &translator, superseded, ); |