From 23f0b629e5d80f880e5a0809f2e47b2d836ee2f9 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 8 Nov 2026 09:47:11 +0000 Subject: remake: replays judged before they go in A switching or rejoining node replays late writes as fresh changes naming each write's user, and nothing judged them before they left: a replay every node would refuse (a revoked capability, a second envelope at a signed path, readers changed) was refused everywhere, and with it everything the node wrote after it. Under a policy, replay_newest now judges each group it commits as every node will (rules::judge_held, in the new epoch); one they would refuse is taken back (forked at the heads before it, same actor) and its paths quarantined with the reason. replay_newest takes the epoch. Review finding 3 (remake replays). --- src/remake/bench.rs | 1 + src/remake/catchup.rs | 2 + src/remake/delta.rs | 115 +++++++++++++++++++++++++++++++++++++++++++++++++- src/remake/switch.rs | 2 + 4 files changed, 118 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/remake/bench.rs b/src/remake/bench.rs index 96f76d8..4c12838 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, + crate::document::replica::Epoch::GENESIS, "node", &Translator::default(), Superseded::Drop, diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 71f077a..9830c33 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -421,6 +421,7 @@ async fn finish(host: &H) { &before, &after, live, + target, &node, &translator, superseded, @@ -438,6 +439,7 @@ async fn finish(host: &H) { &began, &heads, live, + target, &node, &translator, superseded, diff --git a/src/remake/delta.rs b/src/remake/delta.rs index 06508a3..32f5ff7 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -85,12 +85,15 @@ pub(crate) enum Superseded { /// stays as `dst` has it (`superseded` says what becomes of the write). Ties go to the /// 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. +/// naming the other. Under a policy, each is judged as every node will judge it (in +/// `epoch`): one they would refuse is taken back, its paths quarantined. +#[allow(clippy::too_many_arguments)] pub(crate) fn replay_newest( src: &Document, before: &[ChangeHash], after: &[ChangeHash], dst: &mut Document, + epoch: crate::document::replica::Epoch, node: &str, translator: &Translator, superseded: Superseded, @@ -140,8 +143,9 @@ pub(crate) fn replay_newest( } crate::document::replica::authored(dst, node); for ((at, user), paths) in by_time { + let before = dst.get_heads(); let mut transaction = dst.transaction(); - stats.absorb(replay_paths(src, &mut transaction, &paths, translator)); + let group = replay_paths(src, &mut transaction, &paths, translator); let writer = crate::document::replica::Writer { node, user: user.as_deref(), @@ -151,6 +155,27 @@ pub(crate) fn replay_newest( options = options.with_message(message); } transaction.commit_with(options); + // Judged as every node will judge it (`access::rules`): one they would refuse + // goes to the quarantine instead, or everything this node writes after it + // would be refused too. + match judged(dst, epoch, &before) { + Ok(()) => stats.absorb(group), + Err(why) => { + let actor = dst.get_actor().clone(); + match dst.fork_at(&before) { + Ok(kept) => *dst = kept.with_actor(actor), + Err(error) => { + warn!(error = %error, "Could not take back a replay every node would refuse"); + stats.absorb(group); + continue; + } + } + let reason = format!("every node would refuse it: {why}"); + stats + .quarantined + .extend(paths.into_iter().map(|path| (path, reason.clone()))); + } + } } if stats.superseded > 0 { info!( @@ -166,6 +191,24 @@ pub(crate) fn replay_newest( stats } +/// The change just committed to `doc` (the one head after `before`), judged under a +/// policy as every node will judge it on receipt; why it is refused, if it is. +fn judged( + doc: &Document, + epoch: crate::document::replica::Epoch, + before: &[ChangeHash], +) -> Result<(), String> { + if !crate::access::policy::of(doc).enforced() { + return Ok(()); + } + match doc.get_heads()[..] { + [hash] if doc.get_heads() != before => { + crate::access::rules::judge_held(doc, epoch, &hash).unwrap_or(Ok(())) + } + _ => Ok(()), + } +} + /// 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 { @@ -667,6 +710,71 @@ mod tests { transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at)); } + /// Under a policy, a replayed write every node would refuse (its user may no longer + /// write there in the new epoch) goes to the quarantine, not into the document; the + /// rest goes in. + #[test] + fn replays_every_node_would_refuse_quarantined() { + let path = |text: &str| -> Vec { text.split('/').map(str::to_owned).collect() }; + let alice = crate::access::registry::entry( + "root", + 1, + &[(path("notes"), &[crate::access::Right::Write][..])], + ); + let mut old = Document::new(); + change(&mut old, |doc| { + store::write_path(doc, &path("_continuum/swarm"), &json!("s")).unwrap(); + store::write_path( + doc, + &path("_continuum/access/policy"), + &json!({"version": 1}), + ) + .unwrap(); + store::write_path(doc, &path("_continuum/members/node"), &json!({})).unwrap(); + store::write_path(doc, &path("_continuum/users/alice"), &alice).unwrap(); + store::write_path(doc, &path("_continuum/users/bob"), &alice).unwrap(); + }); + crate::document::replica::authored(&mut old, "node"); + let base = old.get_heads(); + // In the new epoch, bob may write nothing any more. + let mut new = old.clone().with_actor(automerge::ActorId::random()); + change(&mut new, |doc| { + store::write_path( + doc, + &path("_continuum/users/bob"), + &crate::access::registry::entry("root", 1, &[]), + ) + .unwrap(); + }); + let mut commit = |user: &str, write: &str| { + let mut transaction = old.transaction(); + store::write_path(&mut transaction, &path(write), &json!(1)).unwrap(); + transaction.commit_with( + automerge::transaction::CommitOptions::default() + .with_time(10) + .with_message(format!("User: {user}")), + ); + }; + commit("alice", "notes/x"); + commit("bob", "notes/y"); + let after = old.get_heads(); + let stats = replay_newest( + &old, + &base, + &after, + &mut new, + crate::document::replica::Epoch::GENESIS, + "node", + &Translator::default(), + Superseded::Drop, + ); + assert_eq!(stats.replayed, 1); + assert_eq!(stats.quarantined.len(), 1); + assert_eq!(stats.quarantined[0].0, path("notes/y")); + assert!(store::read_path(&new, &path("notes/x")).unwrap().is_some()); + assert!(store::read_path(&new, &path("notes/y")).unwrap().is_none()); + } + /// 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] @@ -699,6 +807,7 @@ mod tests { &base, &after, &mut new, + crate::document::replica::Epoch::GENESIS, "node", &Translator::default(), Superseded::Drop, @@ -762,6 +871,7 @@ mod tests { &base, &after, &mut new, + crate::document::replica::Epoch::GENESIS, "node", &Translator::default(), superseded, @@ -851,6 +961,7 @@ mod tests { &base, &after, &mut new, + crate::document::replica::Epoch::GENESIS, "node", &Translator::default(), Superseded::Drop, diff --git a/src/remake/switch.rs b/src/remake/switch.rs index dd08b38..34c492f 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -255,6 +255,7 @@ async fn switch( &base, &heads, &mut doc, + expected, &node, &translator, superseded, @@ -328,6 +329,7 @@ pub(super) async fn swap_in( ©_heads, &live_heads, &mut new_doc, + target, &node, &translator, superseded, -- cgit v1.3.1