diff options
Diffstat (limited to 'src/remake')
| -rw-r--r-- | src/remake/catchup.rs | 32 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 4 | ||||
| -rw-r--r-- | src/remake/delta.rs | 72 | ||||
| -rw-r--r-- | src/remake/kind.rs | 59 | ||||
| -rw-r--r-- | src/remake/mod.rs | 19 | ||||
| -rw-r--r-- | src/remake/quarantine.rs | 8 | ||||
| -rw-r--r-- | src/remake/records.rs | 13 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 94 | ||||
| -rw-r--r-- | src/remake/switch.rs | 17 |
9 files changed, 174 insertions, 144 deletions
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index e265d0b..695e4d7 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -31,7 +31,7 @@ use std::collections::{BTreeMap, HashMap}; use std::convert::Infallible; use std::time::Duration; -use automerge::{ActorId, AutoCommit, ChangeHash}; +use crate::document::{ActorId, ChangeHash, Document, ReadDoc}; use tokio::time::Instant; use tracing::{debug, error, info, warn}; @@ -181,7 +181,7 @@ async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) { host.remake().add_own_actor(actor.clone()); let old = match host .switch_epoch(target, move |_live| { - Ok::<_, Infallible>(AutoCommit::new().with_actor(actor)) + Ok::<_, Infallible>(Document::new().with_actor(actor)) }) .await { @@ -218,9 +218,7 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) { if still_rejoining { match outcome { Ok(change) => { - let applied = host - .commit_doc(move |doc| doc.apply_changes([change])) - .await; + let applied = host.apply_change(change).await; match applied { Ok(()) => { info!(epoch = %target, "Applied the epoch's snapshot; syncing the changes made since") @@ -258,10 +256,10 @@ enum Plan { #[allow(clippy::large_enum_variant)] enum Prepared { /// Caught up with someone, but not with the target epoch yet: keep waiting. - NotYet(AutoCommit), - Failed(AutoCommit, String), + NotYet(Document), + Failed(Document, String), Ready { - old: AutoCommit, + old: Document, metadata: EpochMetadata, plan: Plan, own_missing: usize, @@ -411,8 +409,8 @@ async fn finish<H: Host>(host: &H) { /// `own`: this node's actors (`RemakeState::own_actors`); the old document's own actor /// counts in any case. `now`: Unix milliseconds, by the node's clock. fn prepare( - mut copy: AutoCommit, - mut old: AutoCommit, + copy: Document, + old: Document, previous: Epoch, target: Epoch, mut own: Vec<ActorId>, @@ -421,7 +419,7 @@ fn prepare( if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() { return Prepared::NotYet(old); } - let metadata = match snapshot::verify_document(&mut copy, target) { + let metadata = match snapshot::verify_document(©, target) { Ok(metadata) => metadata, Err(reason) => return Prepared::Failed(old, reason), }; @@ -447,7 +445,7 @@ fn prepare( let clock = next .as_ref() .map_or_else(BTreeMap::new, |next| next.clock.clone()); - let missing = delta::missing_from_snapshot(&mut old, &clock, &own); + let missing = delta::missing_from_snapshot(&old, &clock, &own); if missing.own == 0 { return Prepared::Ready { old, @@ -458,7 +456,7 @@ fn prepare( }; } - let paths = delta::changed_paths(&mut old, &missing.before, &old_heads); + let paths = delta::changed_paths(&old, &missing.before, &old_heads); let gap = target.generation - previous.generation; // Writes made in our epoch go through every remake since, in order. @@ -562,8 +560,8 @@ mod tests { /// An old-epoch document with one own change after the base, and the swarm's /// document of the next epoch (snapshot plus one foreign change). - fn scenario(policy: &JsonValue) -> (AutoCommit, AutoCommit, Epoch) { - let mut old = AutoCommit::new().with_actor(ActorId::from(&b"late-node"[..])); + fn scenario(policy: &JsonValue) -> (Document, Document, Epoch) { + let mut old = Document::new().with_actor(ActorId::from(&b"late-node"[..])); change(&mut old, |old| old.put(ROOT, "base", 1_i64).unwrap()); let base = old.get_heads(); @@ -640,7 +638,7 @@ mod tests { assert!(matches!( prepare( swarm, - AutoCommit::new(), + Document::new(), Epoch::GENESIS, target, Vec::new(), @@ -659,7 +657,7 @@ mod tests { let (old, _, target) = scenario(&json!({})); assert!(matches!( prepare( - AutoCommit::new(), + Document::new(), old, Epoch::GENESIS, target, diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs index 01937c7..314cd87 100644 --- a/src/remake/coordinator.rs +++ b/src/remake/coordinator.rs @@ -178,7 +178,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor }; // Clone under the read lock, build without any lock. - let mut copy = host.read_doc(|doc| doc.clone()).await; + let copy = host.read_doc(|doc| doc.clone()).await; // A base left to the commit is what this node has now. Every node that sees the // decision has it too: the decision change depends on it. let (base, fixed_now) = match &record.proposal.base_heads { @@ -195,7 +195,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor }; let base_heads = fixed_now.then(|| records::encode_heads(&request.base_heads)); request.content = content; - let mut snapshot = match snapshot::build_detached(copy, request).await { + let snapshot = match snapshot::build_detached(copy, request).await { Ok((snapshot, _copy)) => snapshot, Err(error) => { return decide_abort(host, id, format!("snapshot build failed: {error}")).await; diff --git a/src/remake/delta.rs b/src/remake/delta.rs index 21c07b7..06f2913 100644 --- a/src/remake/delta.rs +++ b/src/remake/delta.rs @@ -23,7 +23,7 @@ use std::collections::{BTreeMap, HashSet}; -use automerge::{ActorId, AutoCommit, ChangeHash}; +use crate::document::{ActorId, ChangeHash, Document, Transactable}; use tracing::warn; use super::kind::{Translation, Translator}; @@ -53,10 +53,10 @@ impl ReplayStats { /// Replays into `dst` everything that changed in `src` between `before` and `after`, /// as one commit. Both head sets must be in `src`. 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 +64,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 +82,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 +166,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,7 +227,7 @@ 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>> { @@ -255,14 +257,14 @@ mod tests { 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(); @@ -289,17 +291,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)); + 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) { @@ -311,7 +313,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)); }); @@ -342,12 +344,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) { @@ -359,7 +361,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:?}"); @@ -388,7 +390,7 @@ mod tests { #[test] fn minimal_paths() { - let mut doc = AutoCommit::new(); + let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "untouched", 1_i64).unwrap()); let before = doc.get_heads(); @@ -406,7 +408,7 @@ mod tests { }); let after = doc.get_heads(); - let all = changed_paths(&mut doc, &before, &after); + let all = changed_paths(&doc, &before, &after); assert_eq!( all, vec![ @@ -417,14 +419,14 @@ 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()]]); } #[test] fn replays_state_per_key() { // Base shared by the old document and the snapshot. - let mut old = AutoCommit::new(); + 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(); @@ -434,7 +436,7 @@ mod tests { }); let base = old.get_heads(); - let mut new = AutoCommit::new(); + 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. @@ -451,7 +453,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), @@ -464,7 +466,7 @@ 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()); + 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(); @@ -483,18 +485,18 @@ mod tests { 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(); @@ -503,11 +505,11 @@ mod tests { #[test] fn protocol_metadata_is_never_replayed() { - let mut old = AutoCommit::new(); + 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(); + let mut new = Document::new(); change(&mut new, |new| { store::write_path( new, @@ -524,7 +526,7 @@ mod tests { }); 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)); diff --git a/src/remake/kind.rs b/src/remake/kind.rs index 1df03cb..03f2ef7 100644 --- a/src/remake/kind.rs +++ b/src/remake/kind.rs @@ -19,8 +19,8 @@ use std::path::Path; +use crate::document::{Document, Transaction}; use anyhow::{Context, anyhow, bail}; -use automerge::AutoCommit; use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; @@ -69,9 +69,9 @@ pub(crate) trait RemakeKind: Send + Sync + 'static { /// must not write under `_continuum` (anything there is replaced). fn build( &self, - old_at_base: &AutoCommit, + old_at_base: &Document, content: Option<&[u8]>, - target: &mut AutoCommit, + target: &mut Transaction<'_>, ) -> Result<(), StoreError>; /// Every node: where a user path written after the base goes. @@ -105,7 +105,7 @@ pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> { pub(crate) fn remake_genesis( spec: &str, files: &Files, - target: &mut AutoCommit, + target: &mut Document, ) -> anyhow::Result<()> { let (name, path) = match spec.split_once(':') { Some((name, path)) => (name, Some(Path::new(path))), @@ -135,8 +135,13 @@ pub(crate) fn remake_genesis( let prepared = kind .propose(&request) .map_err(|reason| anyhow!("{name}: {reason}"))?; - kind.build(&AutoCommit::new(), prepared.content.as_deref(), target)?; - target.commit(); + let mut transaction = target.transaction(); + kind.build( + &Document::new(), + prepared.content.as_deref(), + &mut transaction, + )?; + transaction.commit(); Ok(()) } @@ -163,9 +168,9 @@ impl RemakeKind for Compact { fn build( &self, - old_at_base: &AutoCommit, + old_at_base: &Document, _content: Option<&[u8]>, - target: &mut AutoCommit, + target: &mut Transaction<'_>, ) -> Result<(), StoreError> { store::materialize_into(old_at_base, target, &meta::is_protocol) } @@ -267,9 +272,9 @@ impl RemakeKind for Replace { fn build( &self, - _old_at_base: &AutoCommit, + _old_at_base: &Document, content: Option<&[u8]>, - target: &mut AutoCommit, + target: &mut Transaction<'_>, ) -> Result<(), StoreError> { let content = content.ok_or_else(|| anyhow::anyhow!("replace without content"))?; let document: JsonValue = serde_json::from_slice(content) @@ -365,11 +370,11 @@ struct ImportSpec { } /// The document in a request (base64 of a saved Automerge document). -fn load_encoded(encoded: &str) -> Result<AutoCommit, String> { +fn load_encoded(encoded: &str) -> Result<Document, String> { let bytes = BASE64 .decode(encoded.trim()) .map_err(|error| format!("the document is not valid base64: {error}"))?; - AutoCommit::load(&bytes) + Document::load(&bytes) .map_err(|error| format!("the document is not an Automerge document: {error}")) } @@ -386,9 +391,11 @@ impl RemakeKind for Import { // Keep only what goes into the new epoch: the merged view without protocol data // and history. Usually much smaller than what was uploaded, and it is what every // voter downloads. - let mut content = AutoCommit::new(); - store::materialize_into(&imported, &mut content, &meta::is_protocol) + let mut content = Document::new(); + let mut transaction = content.transaction(); + store::materialize_into(&imported, &mut transaction, &meta::is_protocol) .map_err(|error| format!("the document can't be read: {error}"))?; + transaction.commit(); Ok(Prepared { content: Some(content.save()), @@ -398,12 +405,12 @@ impl RemakeKind for Import { fn build( &self, - _old_at_base: &AutoCommit, + _old_at_base: &Document, content: Option<&[u8]>, - target: &mut AutoCommit, + target: &mut Transaction<'_>, ) -> Result<(), StoreError> { let content = content.ok_or_else(|| anyhow::anyhow!("import without content"))?; - let imported = AutoCommit::load(content)?; + let imported = Document::load(content)?; // Deterministic: the same bytes give the same view, copied in a fixed order. store::materialize_into(&imported, target, &meta::is_protocol) } @@ -473,9 +480,9 @@ mod tests { } fn build( &self, - _old: &AutoCommit, + _old: &Document, _content: Option<&[u8]>, - _target: &mut AutoCommit, + _target: &mut Transaction<'_>, ) -> Result<(), StoreError> { Ok(()) } @@ -595,7 +602,7 @@ mod tests { use automerge::{ObjType, ROOT, ReadDoc, transaction::Transactable}; use serde_json::json; - let mut source = AutoCommit::new(); + let mut source = Document::new(); change(&mut source, |source| { let text = source.put_object(ROOT, "notes", ObjType::Text).unwrap(); source.splice_text(&text, 0, 0, "hello").unwrap(); @@ -622,15 +629,15 @@ mod tests { .propose(&json!({ "document": encoded, "rules": [{ "from": "", "to": "" }] })) .unwrap(); let build = || { - let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); + let mut target = Document::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); try_change(&mut target, |target| { - Import.build(&AutoCommit::new(), prepared.content.as_deref(), target) + Import.build(&Document::new(), prepared.content.as_deref(), target) }) .unwrap(); target }; - let mut a = build(); - let mut b = build(); + let a = build(); + let b = build(); assert_eq!(a.get_heads(), b.get_heads()); assert!(a.get(ROOT, meta::ROOT).unwrap().is_none()); @@ -655,9 +662,9 @@ mod tests { use serde_json::json; let build = |document: serde_json::Value| { let prepared = replace(json!({ "document": document })).unwrap(); - let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); + let mut target = Document::new().with_actor(automerge::ActorId::from(&b"fixed"[..])); try_change(&mut target, |target| { - Replace.build(&AutoCommit::new(), prepared.content.as_deref(), target) + Replace.build(&Document::new(), prepared.content.as_deref(), target) }) .unwrap(); ( diff --git a/src/remake/mod.rs b/src/remake/mod.rs index 9206bbf..e9d3021 100644 --- a/src/remake/mod.rs +++ b/src/remake/mod.rs @@ -52,9 +52,9 @@ use axum::{ }; use tokio::sync::broadcast; -use crate::document::Document; use crate::document::replica::Epoch; use crate::document::store::ChangeSet; +use crate::document::{Document, Transaction}; /// Remakes and their aftermath on this node, in numbers (`/admin/stats`). #[derive(Debug, Clone, Copy, Default)] @@ -129,7 +129,11 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// The electorate of a remake proposed now (`vote`): the members that said they are /// alive recently enough, read from `doc`. - fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec<String>; + fn electorate( + &self, + doc: &impl crate::document::ReadDoc, + now: crate::clock::UnixMillis, + ) -> Vec<String>; /// Downloads a change from peers on the local epoch, in chunks from several at once, /// checked against its hash. @@ -183,10 +187,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { /// One committed change (rolled back on error), published and replicated. fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send where - F: FnOnce(&mut Document) -> Result<T, E> + Send + 'static, + F: FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static, T: Send + 'static, E: Send + 'static; + /// Applies a change downloaded from peers (`fetch_change`), then publishes it like + /// a change received by sync. + fn apply_change( + &self, + change: crate::document::Change, + ) -> impl Future<Output = Result<(), String>> + Send; + /// Replaces the document and its epoch atomically. Under the write lock, `f` gets /// the live document and returns the new one. Afterwards the node resets every /// peer's sync state and publishes the change. Returns the old document. @@ -196,7 +207,7 @@ pub(crate) trait Host: Clone + Send + Sync + 'static { f: F, ) -> impl Future<Output = Result<Document, E>> + Send where - F: FnOnce(&mut Document) -> Result<Document, E> + Send + 'static, + F: FnOnce(&Document) -> Result<Document, E> + Send + 'static, E: Send + 'static; } diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs index 5300f54..3ad5005 100644 --- a/src/remake/quarantine.rs +++ b/src/remake/quarantine.rs @@ -5,7 +5,7 @@ //! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value //! the old document's merged view holds there (`deleted` if none). //! * `GET /admin/quarantine/{id}/document` downloads the old document -//! (`AutoCommit::save()`), for inspection with other tools. +//! (`Document::save()`), for inspection with other tools. //! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as //! fresh writes, so they win over what is there) and removes the entry. //! * `DELETE /admin/quarantine/{id}` discards the entry. @@ -14,7 +14,7 @@ use std::convert::Infallible; -use automerge::AutoCommit; +use crate::document::Document; use axum::{ Json, extract::{Path, State}, @@ -69,7 +69,7 @@ pub(super) fn add<H: Host>( host: &H, epoch: Epoch, target: Epoch, - document: &AutoCommit, + document: &Document, quarantined: Vec<(Vec<String>, String)>, ) -> Option<String> { if quarantined.is_empty() { @@ -132,7 +132,7 @@ pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) } pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response { - let Some(mut document) = host + let Some(document) = host .remake() .with_quarantined(&id, |entry| entry.document.clone()) else { diff --git a/src/remake/records.rs b/src/remake/records.rs index fecae4c..55fa8a0 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -38,7 +38,7 @@ use crate::clock::UnixMillis; use crate::document::meta::{self, REMAKES, VOTES}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; -use crate::document::{ChangeHash, Document, Transactable}; +use crate::document::{ChangeHash, ReadDoc, Transactable}; /// One day, for retention defaults. const DAY_MS: u64 = 24 * 60 * 60 * 1000; @@ -337,13 +337,13 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> { } /// The record with this ID, if it exists and is readable, with its votes. -pub(crate) fn load(doc: &Document, id: &str) -> Option<Record> { +pub(crate) fn load(doc: &impl ReadDoc, id: &str) -> Option<Record> { let record = load_record(doc, id)?; Some(with_votes(id, record, &load_register(doc))) } /// All readable records, with their votes. -pub(crate) fn load_all(doc: &Document) -> Vec<(String, Record)> { +pub(crate) fn load_all(doc: &impl ReadDoc) -> Vec<(String, Record)> { let register = load_register(doc); store::keys_at(doc, &records_dir()) .unwrap_or_default() @@ -366,7 +366,7 @@ fn with_votes(id: &str, mut record: Record, register: &Register) -> Record { record } -fn load_record(doc: &Document, id: &str) -> Option<Record> { +fn load_record(doc: &impl ReadDoc, id: &str) -> Option<Record> { let value = match store::read_path(doc, &record_path(id, "")) { Ok(Some(value)) => value, Ok(None) => return None, @@ -390,7 +390,7 @@ fn load_record(doc: &Document, id: &str) -> Option<Record> { } /// A commit among concurrent decisions of remake `id` that disagree, if they do. -fn contested_decision(doc: &Document, id: &str) -> Option<Decision> { +fn contested_decision(doc: &impl ReadDoc, id: &str) -> Option<Decision> { let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision")) .ok()? .into_iter() @@ -418,7 +418,7 @@ fn register_path(voter: &str) -> Vec<String> { } /// The votes register; unreadable entries are skipped with a warning. -pub(crate) fn load_register(doc: &Document) -> Register { +pub(crate) fn load_register(doc: &impl ReadDoc) -> Register { store::keys_at(doc, &meta::path(&[VOTES])) .unwrap_or_default() .into_iter() @@ -503,6 +503,7 @@ pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> { #[cfg(test)] mod tests { use super::*; + use crate::document::Document; use crate::document::testing::change; #[test] diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index c818af6..157a28d 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -20,7 +20,8 @@ use std::collections::BTreeMap; -use automerge::{ActorId, AutoCommit, ChangeHash, transaction::CommitOptions}; +use crate::document::{ActorId, ChangeHash, Document, ReadDoc}; +use automerge::transaction::CommitOptions; use serde_json::{Map as JsonMap, Value as JsonValue, json}; use sha2::{Digest, Sha256}; @@ -55,7 +56,7 @@ pub(crate) struct Snapshot { /// The new epoch: next generation, ID = hash of the snapshot change. pub(crate) epoch: Epoch, /// A document consisting of exactly that one change. - pub(crate) doc: AutoCommit, + pub(crate) doc: Document, /// Highest sequence number per actor (hex) contained in the base. A change /// (actor, seq) is part of the snapshot iff seq <= clock[actor]. Read by tests only: /// nodes read the clock from the epoch metadata (`EpochMetadata::clock`). @@ -116,11 +117,11 @@ impl SnapshotRequest { /// No lock is held while building; the caller only needs a lock for the clone. /// Returns the copy as well, for callers that continue working with it. pub(crate) async fn build_detached( - mut copy: AutoCommit, + copy: Document, request: SnapshotRequest, -) -> Result<(Snapshot, AutoCommit), SnapshotError> { +) -> Result<(Snapshot, Document), SnapshotError> { tokio::task::spawn_blocking(move || { - let snapshot = build(&mut copy, &request.params())?; + let snapshot = build(©, &request.params())?; Ok((snapshot, copy)) }) .await @@ -173,7 +174,7 @@ struct MetadataRecord { /// carry the metadata of all earlier epochs forward (only remake records stay /// behind), so a node many epochs behind can still find the epoch that followed its own. pub(crate) fn read_epoch_metadata( - doc: &AutoCommit, + doc: &Document, generation: u64, ) -> Result<EpochMetadata, String> { let value = store::read_path(doc, &epoch_metadata_path(generation)) @@ -209,7 +210,7 @@ pub(crate) fn read_epoch_metadata( /// /// This does not prove the snapshot's content matches the old epoch; the hash commits /// to it, and the peer is a swarm member trusted as much as for ordinary sync. -pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> { +pub(crate) fn verify_document(doc: &Document, epoch: Epoch) -> Result<EpochMetadata, String> { let root = ChangeHash(epoch.id); let Some(change) = doc.get_change_by_hash(&root) else { return Err(format!( @@ -227,7 +228,7 @@ pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<Epoc /// The epoch `doc` is on: genesis if it holds no epoch metadata; otherwise the newest /// generation it has metadata for, whose snapshot change is the document's only change /// without dependencies (verified). For a document loaded from a backup. -pub(crate) fn epoch_of(doc: &mut AutoCommit) -> Result<Epoch, String> { +pub(crate) fn epoch_of(doc: &Document) -> Result<Epoch, String> { let generation = store::keys_at(doc, &meta::path(&[EPOCHS])) .map_err(|error| error.to_string())? .iter() @@ -263,7 +264,7 @@ pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> { /// Builds the snapshot of `source` at `params.base_heads`. pub(crate) fn build( - source: &mut AutoCommit, + source: &Document, params: &SnapshotParams<'_>, ) -> Result<Snapshot, SnapshotError> { // fork_at with unknown heads would quietly build something else. @@ -273,21 +274,24 @@ pub(crate) fn build( } } - let mut base = source.fork_at(params.base_heads)?; - let clock = clock_of(&mut base); + let base = source.fork_at(params.base_heads)?; + let clock = clock_of(&base); let generation = params.previous.generation + 1; let kind = by_name(params.kind).ok_or_else(|| SnapshotError::UnknownKind(params.kind.to_owned()))?; - let mut snapshot = AutoCommit::new().with_actor(snapshot_actor(params.remake_id, generation)); + let mut snapshot = Document::new().with_actor(snapshot_actor(params.remake_id, generation)); + + // One transaction, so the snapshot is a single change. + let mut target = snapshot.transaction(); // The user content is the kind's. - kind.build(&base, params.content, &mut snapshot)?; + kind.build(&base, params.content, &mut target)?; // Protocol data carries forward, whatever the kind: forget records, earlier epochs' // metadata (rejoining nodes need it). Remake records and the votes register describe // the old epoch's protocol and stay behind. - match store::delete_path(&mut snapshot, &meta::path(&[]), false) { + match store::delete_path(&mut target, &meta::path(&[]), false) { Ok(()) | Err(StoreError::NotFound) => {} Err(error) => return Err(error.into()), } @@ -297,7 +301,7 @@ pub(crate) fn build( .into_iter() .filter(|key| key != REMAKES && key != VOTES) { - store::copy_path(&base, &mut snapshot, &meta::path(&[key.as_str()]))?; + store::copy_path(&base, &mut target, &meta::path(&[key.as_str()]))?; } let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect(); @@ -317,9 +321,9 @@ pub(crate) fn build( "kind": kind.name(), "translation": params.translation, })); - store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?; + store::write_path(&mut target, &epoch_metadata_path(generation), &metadata)?; - let _ = snapshot.commit_with(CommitOptions::default().with_time(0)); + target.commit_with(CommitOptions::default().with_time(0)); let heads = snapshot.get_heads(); let [id] = heads.as_slice() else { @@ -337,7 +341,7 @@ pub(crate) fn build( } /// Highest sequence number per actor in `doc`'s history. -fn clock_of(doc: &mut AutoCommit) -> BTreeMap<String, u64> { +fn clock_of(doc: &Document) -> BTreeMap<String, u64> { let mut clock = BTreeMap::new(); for change in doc.get_changes(&[]) { let seq = clock.entry(change.actor_id().to_string()).or_insert(0); @@ -379,8 +383,8 @@ mod tests { use crate::document::testing::change; use automerge::{ObjType, ROOT, ReadDoc, ScalarValue, Value, transaction::Transactable}; - fn doc(actor: &[u8]) -> AutoCommit { - AutoCommit::new().with_actor(ActorId::from(actor)) + fn doc(actor: &[u8]) -> Document { + Document::new().with_actor(ActorId::from(actor)) } fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> { @@ -398,14 +402,14 @@ mod tests { static NULL: JsonValue = JsonValue::Null; - fn snapshot_of(source: &mut AutoCommit) -> Snapshot { + fn snapshot_of(source: &Document) -> Snapshot { let heads = source.get_heads(); let policy = json!({}); build(source, ¶ms(&heads, &policy)).expect("snapshot") } /// The user-visible view, without our own metadata. - fn user_view(doc: &AutoCommit) -> JsonValue { + fn user_view(doc: &Document) -> JsonValue { let mut view = store::read_path(doc, &[]).unwrap().unwrap(); view.as_object_mut().unwrap().remove(meta::ROOT); view @@ -443,8 +447,8 @@ mod tests { assert_eq!(user_view(&a), user_view(&b)); - let snap_a = snapshot_of(&mut a); - let snap_b = snapshot_of(&mut b); + let snap_a = snapshot_of(&a); + let snap_b = snapshot_of(&b); // Different histories mean different clocks, and the clock is part of the // metadata. Equal views alone therefore give equal content but not equal IDs; @@ -468,9 +472,9 @@ mod tests { let mut b = doc(b"actor-b"); b.merge(&mut a.clone()).unwrap(); - let first = snapshot_of(&mut a); - let again = snapshot_of(&mut a); - let elsewhere = snapshot_of(&mut b); + let first = snapshot_of(&a); + let again = snapshot_of(&a); + let elsewhere = snapshot_of(&b); assert_eq!(first.epoch, again.epoch); assert_eq!(first.epoch, elsewhere.epoch); @@ -498,7 +502,7 @@ mod tests { "test setup should produce a conflict" ); - let snap = snapshot_of(&mut a); + let snap = snapshot_of(&a); assert_eq!(user_view(&snap.doc), json!({ "cfg": { "x": 1, "y": 2 } })); let (_, conflicts) = store::read_path_with_conflicts(&snap.doc, &[]) @@ -529,7 +533,7 @@ mod tests { change(&mut a, |a| a.put(ROOT, "late", true).unwrap()); let policy = json!({}); - let snap = build(&mut a, ¶ms(&base, &policy)).unwrap(); + let snap = build(&a, ¶ms(&base, &policy)).unwrap(); let view = user_view(&snap.doc); assert!(view.get("late").is_none()); @@ -548,11 +552,11 @@ mod tests { change(&mut a, |a| { store::write_path(a, &["k".into()], &json!(1)).unwrap() }); - assert_eq!(epoch_of(&mut a), Ok(Epoch::GENESIS)); + assert_eq!(epoch_of(&a), Ok(Epoch::GENESIS)); let heads = a.get_heads(); let policy = json!({}); - let mut snap = build(&mut a, ¶ms(&heads, &policy)).unwrap(); + let snap = build(&a, ¶ms(&heads, &policy)).unwrap(); // Saved and loaded, with a write since, as a backup would be. let mut later = snap .doc @@ -561,8 +565,8 @@ mod tests { change(&mut later, |later| { store::write_path(later, &["k".into()], &json!(2)).unwrap() }); - let mut backup = AutoCommit::load(&later.save()).unwrap(); - assert_eq!(epoch_of(&mut backup), Ok(snap.epoch)); + let backup = Document::load(&later.save()).unwrap(); + assert_eq!(epoch_of(&backup), Ok(snap.epoch)); } #[test] @@ -591,7 +595,7 @@ mod tests { let heads = a.get_heads(); let policy = json!({ "replayWindow": { "epochs": 1 } }); - let snap = build(&mut a, ¶ms(&heads, &policy)).unwrap(); + let snap = build(&a, ¶ms(&heads, &policy)).unwrap(); let meta = store::read_path(&snap.doc, &meta::path(&[])) .unwrap() @@ -624,7 +628,7 @@ mod tests { a.put(ROOT, "neg_zero", -0.0_f64).unwrap(); }); - let snap = snapshot_of(&mut a); + let snap = snapshot_of(&a); let d = &snap.doc; match d.get(ROOT, "c").unwrap() { @@ -665,7 +669,7 @@ mod tests { let mut a = doc(b"actor-a"); change(&mut a, |a| a.put(ROOT, "x", 1_i64).unwrap()); - let mut snap = snapshot_of(&mut a); + let snap = snapshot_of(&a); let heads = snap.doc.get_heads(); assert_eq!(heads.len(), 1); assert_eq!(heads[0].0, snap.epoch.id); @@ -683,15 +687,15 @@ mod tests { let policy = json!({}); assert!(matches!( - build(&mut a, ¶ms(&foreign, &policy)), + build(&a, ¶ms(&foreign, &policy)), Err(SnapshotError::MissingBase(_)) )); } #[test] fn empty_document() { - let mut a = doc(b"actor-a"); - let snap = snapshot_of(&mut a); + let a = doc(b"actor-a"); + let snap = snapshot_of(&a); assert_eq!(snap.epoch.generation, 1); assert!(snap.clock.is_empty()); assert_eq!(user_view(&snap.doc), json!({})); @@ -706,7 +710,7 @@ mod tests { change(&mut late, |late| late.put(ROOT, "shared", "base").unwrap()); let base = late.get_heads(); let policy = json!({}); - let mut snap = build(&mut late, ¶ms(&base, &policy)).unwrap(); + let snap = build(&late, ¶ms(&base, &policy)).unwrap(); // Meanwhile the swarm moved on in the new epoch. let mut swarm = snap @@ -718,8 +722,8 @@ mod tests { change(&mut late, |late| late.put(ROOT, "mine", "late").unwrap()); - let mut fetched = AutoCommit::load(&bytes).unwrap(); - let metadata = verify_document(&mut fetched, snap.epoch).unwrap(); + let fetched = Document::load(&bytes).unwrap(); + let metadata = verify_document(&fetched, snap.epoch).unwrap(); assert_eq!(metadata.previous, Epoch::GENESIS); assert_eq!(metadata.remake_id, "c-1"); assert_eq!( @@ -730,14 +734,14 @@ mod tests { // Documents of other epochs are refused. let mut wrong = snap.epoch; wrong.generation = 2; - assert!(verify_document(&mut fetched, wrong).is_err()); + assert!(verify_document(&fetched, wrong).is_err()); let mut wrong = snap.epoch; wrong.id = [0; 32]; - assert!(verify_document(&mut fetched, wrong).is_err()); + assert!(verify_document(&fetched, wrong).is_err()); let own = late.get_actor().clone(); let missing = - delta::missing_from_snapshot(&mut late, &metadata.clock, std::slice::from_ref(&own)); + delta::missing_from_snapshot(&late, &metadata.clock, std::slice::from_ref(&own)); assert_eq!(missing.own, 1); let heads = late.get_heads(); let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..])); diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 9aae748..e60c942 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -219,10 +219,10 @@ async fn switch<H: Host>( // 2. Build (or reuse), verify and replay without holding any lock. let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking( move || -> Result<(Document, Vec<ChangeHash>, ReplayStats), String> { - let mut copy = copy; + let copy = copy; let snapshot = match prebuilt { Some(snapshot) => snapshot, - None => snapshot::build(&mut copy, &request.params()) + None => snapshot::build(©, &request.params()) .map_err(|error| format!("snapshot build failed: {error}"))?, }; if snapshot.epoch != expected { @@ -235,13 +235,20 @@ async fn switch<H: Host>( // A fresh actor: this node's writes in the new epoch. let mut doc = snapshot.doc.with_actor(actor); + // The carried record and the replayed writes go into one change. + let heads = copy.get_heads(); + let paths = delta::changed_paths(©, &base, &heads); + let mut transaction = doc.transaction(); + // Keep this remake's record visible in the new epoch, for its history. - if carry_record && let Err(error) = store::copy_path(©, &mut doc, &record_path) { + if carry_record + && let Err(error) = store::copy_path(©, &mut transaction, &record_path) + { warn!(error = %error, "Could not carry the remake record into the new epoch"); } - let heads = copy.get_heads(); - let stats = delta::replay(&mut copy, &base, &heads, &mut doc, &translator); + let stats = delta::replay_paths(©, &mut transaction, &paths, &translator); + transaction.commit(); Ok((doc, heads, stats)) }, ) |