From 389b11b597b6bf60f120b190002bbb76fff28dbf Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 7 Nov 2026 22:35:21 +0100 Subject: access: screening received changes access::screen: a batch of received changes judged change by change, those that change what rules read applied before the next is judged; a refused change stays out with whatever builds on it, kept aside for operators; changes missing dependencies wait. Auth plan M6. --- src/access/mod.rs | 3 + src/access/rules.rs | 9 +- src/access/screen.rs | 378 ++++++++++++++++++++++++++++++++++++++++++++++++ src/document/changes.rs | 27 ++++ src/document/mod.rs | 3 +- 5 files changed, 418 insertions(+), 2 deletions(-) create mode 100644 src/access/screen.rs (limited to 'src') diff --git a/src/access/mod.rs b/src/access/mod.rs index 99c8ecd..1109fa6 100644 --- a/src/access/mod.rs +++ b/src/access/mod.rs @@ -17,6 +17,9 @@ pub(crate) mod registry; // Judging changes on receipt is what uses it; until then, the tests. #[cfg_attr(not(test), allow(dead_code))] pub(crate) mod rules; +// Judging changes on receipt is what uses it; until then, the tests. +#[cfg_attr(not(test), allow(dead_code))] +pub(crate) mod screen; pub(crate) mod token; /// What a capability lets its holder do, at its path and below. diff --git a/src/access/rules.rs b/src/access/rules.rs index 9a9e5e2..e151cbb 100644 --- a/src/access/rules.rs +++ b/src/access/rules.rs @@ -35,6 +35,10 @@ use crate::document::{ChangeHash, Document, store}; pub(crate) struct Change<'a> { pub(crate) hash: ChangeHash, pub(crate) deps: &'a [ChangeHash], + /// The heads to judge it at: its dependencies, or, for those not applied yet, the + /// changes of the document they build on, where nothing between changed what rules + /// read (`access::screen`). + pub(crate) seen: &'a [ChangeHash], /// The node that made it (its node ID), as its signature proves. pub(crate) committer: Option<&'a str>, /// The user it names (`User:`), if any. @@ -77,7 +81,7 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result )) }; } - let at = At::new(doc, change.deps).ok_or(Refusal::Unready)?; + let at = At::new(doc, change.seen).ok_or(Refusal::Unready)?; let refuse = |why: String| Err(Refusal::Rule(why)); let Some(node) = change.committer else { return refuse("it names no node".to_owned()); @@ -239,6 +243,7 @@ mod tests { &Change { hash: ChangeHash([9; 32]), deps: &deps, + seen: &deps, committer: Some(committer), user, paths: &touched, @@ -363,6 +368,7 @@ mod tests { &Change { hash: ChangeHash([9; 32]), deps, + seen: deps, committer: Some("a"), user: Some("alice"), paths: &touched, @@ -386,6 +392,7 @@ mod tests { &Change { hash: ChangeHash(hash), deps: &[], + seen: &[], committer: None, user: None, paths: &touched, diff --git a/src/access/screen.rs b/src/access/screen.rs new file mode 100644 index 0000000..dd9ffcd --- /dev/null +++ b/src/access/screen.rs @@ -0,0 +1,378 @@ +//! Received changes judged by rules by path before they are applied (`rules::judge`), +//! and applied if allowed (`apply`): a refused change stays out, and so does whatever +//! builds on it; it is kept aside for an operator to look at (`Refused`). +//! +//! A change is judged at the heads its writer saw. Those of a batch aren't in the +//! document yet, so one is judged at the changes of the document its batch ancestors +//! build on: what rules read there (the members, the user registry, remake proposals) +//! is what its writer saw, unless one of those ancestors changed it. So a change that +//! does (`relevant`) is applied before the next is judged: the rest go in at once. +//! +//! A change whose dependencies are neither in the document nor among those received +//! waits (`Waiting`) until they come, a bounded number; Automerge never sees it before. + +use std::collections::{BTreeSet, HashMap, HashSet}; + +use anyhow::Result; + +use super::rules::{self, Refusal}; +use crate::document::inspect::{Inspector, Path}; +use crate::document::meta; +use crate::document::replica::{Epoch, user_of}; +use crate::document::{Change, ChangeHash, Document, ReadDoc}; + +/// Changes waiting for their dependencies, kept at most. +const MAX_WAITING: usize = 10_000; +/// Refused changes kept for operators, at most: the oldest go. +pub(crate) const MAX_REFUSED: usize = 1_000; + +/// A refused change, and why. +#[derive(Debug, Clone)] +pub(crate) struct Refused { + pub(crate) change: Change, + pub(crate) committer: Option, + pub(crate) user: Option, + pub(crate) paths: BTreeSet, + pub(crate) why: String, +} + +/// Received changes waiting for their dependencies. +#[derive(Debug, Default)] +pub(crate) struct Waiting(Vec); + +impl Waiting { + pub(crate) fn len(&self) -> usize { + self.0.len() + } +} + +/// What `apply` did: how many changes it applied, and those it refused. +#[derive(Debug, Default)] +pub(crate) struct Screened { + pub(crate) applied: usize, + pub(crate) refused: Vec, +} + +/// Judges `received` (and those `waiting` before) and applies to `doc` (of `epoch`) +/// those rules allow, what they change recorded in `log`. Those already in `doc` are +/// left out; those whose dependencies are missing wait. +pub(crate) fn apply( + doc: &mut Document, + epoch: Epoch, + received: Vec, + waiting: &mut Waiting, + log: &mut crate::document::PatchLog, +) -> Result { + let mut screened = Screened::default(); + // Each once, not in the document yet. + let mut seen = HashSet::new(); + let candidates: Vec = std::mem::take(&mut waiting.0) + .into_iter() + .chain(received) + .filter(|change| seen.insert(change.hash())) + .filter(|change| doc.get_change_by_hash(&change.hash()).is_none()) + .collect(); + let (ready, unready) = in_order(doc, candidates); + waiting.0 = unready; + waiting.0.truncate(MAX_WAITING); + if ready.is_empty() { + return Ok(screened); + } + + let decoded: Vec = + ready.iter().map(|change| change.decode()).collect(); + let paths = Inspector::new(doc).paths(&decoded); + // Who made each: its author, or that of its actor's first change (in the document, + // or among these). + let mut authors: HashMap = HashMap::new(); + let committers: Vec> = ready + .iter() + .map(|change| { + let actor = change.actor_id(); + let named = change + .author() + .or_else(|| doc.get_author_for_actor(actor)) + .and_then(|author| { + std::str::from_utf8(author.as_bytes()) + .ok() + .map(str::to_owned) + }) + .or_else(|| authors.get(actor).cloned()); + if let Some(named) = &named { + authors.insert(actor.clone(), named.clone()); + } + named + }) + .collect(); + + // The changes of the document each accepted one, not applied yet, builds on. + let mut frontier: HashMap> = HashMap::new(); + let mut refused: HashSet = HashSet::new(); + let mut accepted: Vec = Vec::new(); + for (index, change) in ready.iter().enumerate() { + let hash = change.hash(); + let committer = committers[index].as_deref(); + let user = change.message().and_then(user_of).map(str::to_owned); + let refuse = |why: String| Refused { + change: change.clone(), + committer: committer.map(str::to_owned), + user: user.clone(), + paths: paths[index].clone(), + why, + }; + if let Some(dep) = change.deps().iter().find(|dep| refused.contains(dep)) { + refused.insert(hash); + screened + .refused + .push(refuse(format!("it builds on refused change {dep}"))); + continue; + } + let mut seen_at: BTreeSet = BTreeSet::new(); + for dep in change.deps() { + match frontier.get(dep) { + Some(heads) => seen_at.extend(heads.iter().copied()), + None => { + seen_at.insert(*dep); + } + } + } + let seen_at: Vec = seen_at.into_iter().collect(); + let judged = rules::judge( + doc, + epoch, + &rules::Change { + hash, + deps: change.deps(), + seen: &seen_at, + committer, + user: user.as_deref(), + paths: &paths[index], + }, + ); + match judged { + Ok(()) => { + accepted.push(change.clone()); + if relevant(&paths[index]) { + // What rules read changed: in before the next is judged. + screened.applied += accepted.len(); + doc.apply_changes_log_patches(std::mem::take(&mut accepted), log)?; + frontier.clear(); + } else { + frontier.insert(hash, seen_at); + } + } + Err(Refusal::Unready) => { + // Can't be: its dependencies are in the document or among these. + refused.insert(hash); + screened.refused.push(refuse(Refusal::Unready.to_string())); + } + Err(Refusal::Rule(why)) => { + refused.insert(hash); + screened.refused.push(refuse(why)); + } + } + } + screened.applied += accepted.len(); + if !accepted.is_empty() { + doc.apply_changes_log_patches(accepted, log)?; + } + Ok(screened) +} + +/// Whether a change touching `paths` changes what rules read: the members, the user +/// registry, remake records, the protocol data as a whole. +fn relevant(paths: &BTreeSet) -> bool { + paths.iter().any(|path| match path.as_slice() { + [] => true, + [root, ..] if root != meta::ROOT => false, + [_] => true, + [_, part, ..] => [meta::MEMBERS, meta::USERS, meta::REMAKES].contains(&part.as_str()), + }) +} + +/// `candidates` in an order where every change comes after its dependencies among them +/// (ready), and those with a dependency neither in `doc` nor among them, nor ready +/// (unready). +fn in_order(doc: &Document, candidates: Vec) -> (Vec, Vec) { + let mut by_hash: HashMap = candidates + .into_iter() + .map(|change| (change.hash(), change)) + .collect(); + let mut ready = Vec::new(); + let mut placed: HashSet = HashSet::new(); + loop { + let mut progress = false; + let mut hashes: Vec = by_hash.keys().copied().collect(); + // The same order on every node, whatever order they came in. + hashes.sort_unstable(); + for hash in hashes { + let change = &by_hash[&hash]; + if change + .deps() + .iter() + .all(|dep| placed.contains(dep) || doc.get_change_by_hash(dep).is_some()) + { + placed.insert(hash); + ready.push(by_hash.remove(&hash).expect("there")); + progress = true; + } + } + if !progress { + break; + } + } + (ready, by_hash.into_values().collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::access::Right; + use crate::access::registry; + use crate::clock::UnixMillis; + use crate::document::replica::{Writer, transact}; + use crate::document::store; + use serde_json::json; + + fn path(text: &str) -> Path { + text.split('/').map(str::to_owned).collect() + } + + fn write( + doc: &mut Document, + node: &str, + user: Option<&str>, + at: &str, + value: serde_json::Value, + ) { + let writer = Writer { node, user }; + let at = path(at); + transact(doc, UnixMillis(1), &writer, |doc| { + store::write_path(doc, &at, &value) + }) + .unwrap(); + } + + /// The changes of `doc` after `since`, as received. + fn received(doc: &Document, since: &[ChangeHash]) -> Vec { + doc.get_changes(since).into_iter().collect() + } + + fn screen(doc: &mut Document, changes: Vec, waiting: &mut Waiting) -> Screened { + apply( + doc, + Epoch::GENESIS, + changes, + waiting, + &mut crate::document::PatchLog::inactive(), + ) + .unwrap() + } + + /// A swarm whose genesis registers member `a` and user `alice`, who may write + /// `notes`; written on `theirs`, received by an empty document in one batch. + fn genesis() -> Document { + let mut theirs = Document::new(); + let alice = registry::entry("a", 1, &[(path("notes"), &[Right::Write][..])]); + let writer = Writer { + node: "a", + user: None, + }; + transact(&mut theirs, UnixMillis(1), &writer, |doc| { + store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?; + store::write_path(doc, &path("_continuum/users/alice"), &alice) + }) + .unwrap(); + write(&mut theirs, "a", None, "_continuum/members/a", json!({})); + theirs + } + + /// A batch from the genesis on: honest changes applied, a refused one left out with + /// what builds on it, the rest in. + #[test] + fn a_batch_judged_change_by_change() { + let mut theirs = genesis(); + write(&mut theirs, "a", Some("alice"), "notes/x", json!(1)); + let before_bad = theirs.get_heads(); + write(&mut theirs, "a", Some("alice"), "config/x", json!(2)); + write(&mut theirs, "a", Some("alice"), "notes/y", json!(3)); + // Concurrent with the refused one: in. + let mut side = theirs.fork_at(&before_bad).unwrap(); + write(&mut side, "a", Some("alice"), "notes/z", json!(4)); + theirs.merge(&mut side).unwrap(); + + let mut ours = Document::new(); + let mut waiting = Waiting::default(); + let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting); + assert_eq!(screened.applied, 4); + let why: Vec<&str> = screened.refused.iter().map(|r| r.why.as_str()).collect(); + assert_eq!(why.len(), 2, "{why:?}"); + assert!(why[0].contains("may not write /config"), "{why:?}"); + assert!(why[1].contains("builds on refused change"), "{why:?}"); + assert_eq!( + store::read_path(&ours, &path("notes/x")).unwrap(), + Some(json!(1)) + ); + assert_eq!( + store::read_path(&ours, &path("notes/z")).unwrap(), + Some(json!(4)) + ); + assert_eq!(store::read_path(&ours, &path("config")).unwrap(), None); + assert!(waiting.len() == 0); + } + + /// Judged as its writer saw the registry: a grant earlier in the same batch counts. + #[test] + fn a_grant_in_the_same_batch_counts() { + let mut theirs = genesis(); + let bob = registry::entry("a", 1, &[(path("config"), &[Right::Write][..])]); + write(&mut theirs, "a", Some("alice"), "_continuum/users/bob", bob); + write(&mut theirs, "a", Some("bob"), "config/x", json!(1)); + let mut ours = Document::new(); + let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default()); + // Alice may not grant (no write on the registry): both refused. + assert_eq!(screened.refused.len(), 2); + + let mut theirs = genesis(); + let root = registry::entry("a", 1, ®istry::all_caps()); + transact( + &mut theirs, + UnixMillis(1), + &Writer { + node: "a", + user: None, + }, + |doc| store::write_path(doc, &path("_continuum/users/root"), &root), + ) + .unwrap(); + // Written as a member without a user: refused, and so is what builds on it. + let mut ours = Document::new(); + let screened = screen(&mut ours, received(&theirs, &[]), &mut Waiting::default()); + assert_eq!(screened.refused.len(), 1); + } + + /// Changes whose dependencies haven't come wait, and are judged once they do. + #[test] + fn changes_wait_for_their_dependencies() { + let mut theirs = genesis(); + write(&mut theirs, "a", Some("alice"), "notes/x", json!(1)); + let middle = theirs.get_heads(); + write(&mut theirs, "a", Some("alice"), "notes/y", json!(2)); + let mut ours = Document::new(); + let mut waiting = Waiting::default(); + // The last first: it waits. + let last = received(&theirs, &middle); + let screened = screen(&mut ours, last, &mut waiting); + assert_eq!(screened.applied, 0); + assert_eq!(waiting.len(), 1); + // The rest: all in, the waiting one too. + let screened = screen(&mut ours, received(&theirs, &[]), &mut waiting); + assert_eq!(screened.applied, 4); + assert!(screened.refused.is_empty()); + assert_eq!(waiting.len(), 0); + assert_eq!( + store::read_path(&ours, &path("notes/y")).unwrap(), + Some(json!(2)) + ); + } +} diff --git a/src/document/changes.rs b/src/document/changes.rs index 57e2bf7..eef7dbc 100644 --- a/src/document/changes.rs +++ b/src/document/changes.rs @@ -240,6 +240,33 @@ pub(crate) fn misplaced_author(mut data: &[u8]) -> Option { None } +/// The changes in `data`, chunks as sync messages and downloads carry them: changes, +/// compressed or not, bundles, documents. None if a chunk doesn't load. +pub(crate) fn changes_in(mut data: &[u8]) -> Option> { + const MAGIC: [u8; 4] = [0x85, 0x6f, 0x4a, 0x83]; + const HEADER: usize = 9; + let mut changes = Vec::new(); + while !data.is_empty() { + if data.len() <= HEADER || data[..4] != MAGIC { + return None; + } + let kind = data[8]; + let (length, read) = uleb128(&data[HEADER..])?; + let end = HEADER + .checked_add(read)? + .checked_add(usize::try_from(length).ok()?)?; + let chunk = data.get(..end)?; + data = &data[end..]; + match kind { + 1 | 2 => changes.push(Change::try_from(chunk).ok()?), + 3 => changes.extend(automerge::Bundle::try_from(chunk).ok()?.to_changes().ok()?), + 0 => changes.extend(Document::load(chunk).ok()?.get_changes(&[])), + _ => return None, + } + } + Some(changes) +} + /// `change`, if it names an author but isn't its actor's first (`misplaced_author`). fn misplaced_in(change: &Change) -> Option { let author = change.author()?; diff --git a/src/document/mod.rs b/src/document/mod.rs index 7f59942..c183537 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -9,7 +9,8 @@ /// Only `document` and `remake` use Automerge itself (`architecture.md`, The document). pub(crate) use automerge::transaction::{Transactable, Transaction}; pub(crate) use automerge::{ - ActorId, Automerge as Document, Change, ChangeHash, ReadDoc, ScalarValue, + ActorId, Automerge as Document, Change, ChangeHash, ExpandedChange, PatchLog, ReadDoc, + ScalarValue, }; /// The whole document as bytes, as the engine saves it: a backup, the document -- cgit v1.3.1