diff options
Diffstat (limited to 'src/remake/delta.rs')
| -rw-r--r-- | src/remake/delta.rs | 115 |
1 files changed, 113 insertions, 2 deletions
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<ChangeHash> { @@ -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<String> { 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, |