From 6105507200231e67ba544c12aa10b988e1c23826 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 19:53:27 +0100 Subject: use Automerge as Document instead of AutoCommit This allows a lot of operations to drop mut. --- src/document/changes.rs | 25 +++++++++++----------- src/document/file.rs | 12 +++++------ src/document/lock.rs | 11 ++-------- src/document/mod.rs | 14 +++++++------ src/document/replica.rs | 53 +++++++++++++++++++++++++++-------------------- src/document/store.rs | 10 ++++----- src/document/sync.rs | 55 +++++++++++++++++++++++-------------------------- src/document/testing.rs | 18 +++++++++------- 8 files changed, 100 insertions(+), 98 deletions(-) (limited to 'src/document') diff --git a/src/document/changes.rs b/src/document/changes.rs index 33029fc..92b7744 100644 --- a/src/document/changes.rs +++ b/src/document/changes.rs @@ -6,11 +6,10 @@ use std::collections::{HashMap, HashSet}; use anyhow::{Context, Result, anyhow, bail}; -use super::{Change, ChangeHash, Document}; +use super::{Change, ChangeHash, Document, ReadDoc}; -/// Those of `heads` that `doc` doesn't have. (Looking a change up needs the document -/// mutably.) -pub(crate) fn missing(doc: &mut Document, heads: &[ChangeHash]) -> Vec { +/// Those of `heads` that `doc` doesn't have. +pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec { heads .iter() .copied() @@ -19,8 +18,8 @@ pub(crate) fn missing(doc: &mut Document, heads: &[ChangeHash]) -> Vec Vec { +/// writers' clocks (`Replica::transact`); changes without a time (snapshots) left out. +pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec { doc.get_changes(since) .iter() .map(|change| change.timestamp()) @@ -29,7 +28,7 @@ pub(crate) fn commit_times(doc: &mut Document, since: &[ChangeHash]) -> Vec } /// The change `hash` as the document encodes it; None if `doc` doesn't have it. -pub(crate) fn change_bytes(doc: &mut Document, hash: &ChangeHash) -> Option> { +pub(crate) fn change_bytes(doc: &Document, hash: &ChangeHash) -> Option> { doc.get_change_by_hash(hash) .map(|change| change.raw_bytes().to_vec()) } @@ -49,7 +48,7 @@ pub(crate) fn change_from_bytes(bytes: Vec, hash: ChangeHash) -> Result Option> { @@ -147,8 +146,8 @@ mod tests { // Beyond `to`: must stay out of the delta. random_write(rng, &mut two); - let bytes = delta_bytes(&mut one, &from, &to).unwrap(); - assert_eq!(delta_bytes(&mut two, &from, &to).unwrap(), bytes); + let bytes = delta_bytes(&one, &from, &to).unwrap(); + assert_eq!(delta_bytes(&two, &from, &to).unwrap(), bytes); let mut behind = base.fork(); load_delta(&mut behind, &bytes).unwrap(); @@ -160,7 +159,7 @@ mod tests { fn no_delta_without_its_heads() { let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap()); - assert!(delta_bytes(&mut doc, &[], &[ChangeHash([7; 32])]).is_none()); + assert!(delta_bytes(&doc, &[], &[ChangeHash([7; 32])]).is_none()); } /// A change loads from its bytes only as the change it was asked for. @@ -169,9 +168,9 @@ mod tests { let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "k", 1_i64).unwrap()); let hash = doc.get_heads()[0]; - let bytes = change_bytes(&mut doc, &hash).unwrap(); + let bytes = change_bytes(&doc, &hash).unwrap(); assert_eq!(change_from_bytes(bytes.clone(), hash).unwrap().hash(), hash); assert!(change_from_bytes(bytes, ChangeHash([7; 32])).is_err()); - assert!(change_bytes(&mut doc, &ChangeHash([7; 32])).is_none()); + assert!(change_bytes(&doc, &ChangeHash([7; 32])).is_none()); } } diff --git a/src/document/file.rs b/src/document/file.rs index 6439e27..9efae42 100644 --- a/src/document/file.rs +++ b/src/document/file.rs @@ -20,8 +20,8 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; +use super::{ActorId, ChangeHash, Document}; use anyhow::{Context, Result, anyhow, bail}; -use automerge::{ActorId, AutoCommit, ChangeHash}; use serde::{Deserialize, Serialize}; use tokio::time::Instant; use tracing::{debug, error, warn}; @@ -42,7 +42,7 @@ struct Header { /// A document file as read at start. pub(crate) struct Loaded { /// Under a fresh actor. - pub(crate) doc: AutoCommit, + pub(crate) doc: Document, /// The actors this node wrote with before. pub(crate) actors: Vec, } @@ -88,7 +88,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> { .collect::>>()?; let mut records = &rest[line_end + 1..]; - let mut doc: Option = None; + let mut doc: Option = None; while !records.is_empty() { let Some((length, after)) = records.split_first_chunk::<4>() else { break; @@ -226,7 +226,7 @@ impl DocumentFile { pub(crate) async fn rewrite( self: &Arc, epoch: Epoch, - mut copy: AutoCommit, + copy: Document, own: Vec, reason: &str, ) { @@ -247,7 +247,7 @@ impl DocumentFile { .into_iter() .filter(|actor| *actor == current || present.contains(actor)) .collect(); - let save = super::save(&mut copy); + let save = super::save(©); let bytes = whole(&actors, &save)?; writer .files @@ -327,7 +327,7 @@ mod tests { #[test] fn whole_then_increments() { - let mut doc = AutoCommit::new(); + let mut doc = Document::new(); change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap()); let actor = doc.get_actor().clone(); let heads = doc.get_heads(); diff --git a/src/document/lock.rs b/src/document/lock.rs index a394b52..f05c5e3 100644 --- a/src/document/lock.rs +++ b/src/document/lock.rs @@ -27,8 +27,8 @@ pub(crate) struct DocumentLock { pub(crate) struct LockWaits { /// Readers: wait for a writer only. pub(crate) read: Waits, - /// Writers (applying sync messages, generating them, writes, saves): wait for - /// everyone. + /// Writers (applying sync messages and fetched changes, writes, epoch switches): + /// wait for everyone. pub(crate) write: Waits, } @@ -125,13 +125,6 @@ impl DocumentLock { guard } - pub(crate) async fn write(&self) -> RwLockWriteGuard<'_, Replica> { - let since = Instant::now(); - let guard = self.replica.write().await; - self.waited(Mode::Write, since); - guard - } - /// Outside the runtime only (`spawn_blocking`), like `RwLock::blocking_read`. pub(crate) fn blocking_read(&self) -> RwLockReadGuard<'_, Replica> { let since = Instant::now(); diff --git a/src/document/mod.rs b/src/document/mod.rs index 9633241..fa1fa77 100644 --- a/src/document/mod.rs +++ b/src/document/mod.rs @@ -2,21 +2,23 @@ //! (merged-view paths over JSON) the HTTP API and remakes work with. /// The engine's types, under the names the rest of the crate uses: the document -/// (Automerge's `AutoCommit`), its changes and their hashes, and its writers (actors); -/// and what can be written like a document (`Transactable`). +/// (Automerge's `Automerge`), its changes and their hashes, and its writers (actors); +/// a write to it (`Transaction`: one change, committed or dropped whole), and what can +/// be read like a document (`ReadDoc`: the document, or a write in progress) and written +/// like one (`Transactable`). /// Only `document` and `remake` use Automerge itself (`architecture.md`, The document). -pub(crate) use automerge::transaction::Transactable; -pub(crate) use automerge::{ActorId, AutoCommit as Document, Change, ChangeHash}; +pub(crate) use automerge::transaction::{Transactable, Transaction}; +pub(crate) use automerge::{ActorId, Automerge as Document, Change, ChangeHash, ReadDoc}; /// The whole document as bytes, as the engine saves it: a backup, the document /// file's full save, the size a joining node downloads. -pub(crate) fn save(doc: &mut Document) -> Vec { +pub(crate) fn save(doc: &Document) -> Vec { doc.save() } /// The changes after `since` as bytes that, appended to a save, load with it (the /// document file's appends). -pub(crate) fn save_since(doc: &mut Document, since: &[ChangeHash]) -> Vec { +pub(crate) fn save_since(doc: &Document, since: &[ChangeHash]) -> Vec { doc.save_after(since) } diff --git a/src/document/replica.rs b/src/document/replica.rs index 5a87016..20c02e5 100644 --- a/src/document/replica.rs +++ b/src/document/replica.rs @@ -10,7 +10,7 @@ use std::fmt; use std::ops::{Deref, DerefMut}; use std::sync::Arc; -use automerge::{AutoCommit, ChangeHash}; +use super::{ChangeHash, Document, Transaction}; use tokio::sync::watch; /// One generation of the swarm document. @@ -95,12 +95,11 @@ impl fmt::Debug for Epoch { /// The document and its epoch, always locked and swapped together. /// -/// Derefs to the document, so code that only needs the `AutoCommit` keeps working -/// with a guard of `Replica`. +/// Derefs to the document, so code that only needs the document keeps working with a +/// guard of `Replica`. /// -/// The heads are cached: `AutoCommit::get_heads` takes `&mut self` (it closes a pending -/// transaction), so without the cache every keepalive and status read would need the -/// write lock. Every path that changes the document under the write lock calls +/// The heads are cached, sorted, so keepalives and status reads (and the stamp below) +/// don't recompute them. Every path that changes the document under the write lock calls /// [`Replica::refresh_heads`] before releasing it (`AppState::mutate_doc`, /// `switch_epoch`, the sync apply); readers take the read lock and use /// [`Replica::heads`]. @@ -113,7 +112,7 @@ pub(crate) struct Replica { /// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this /// run: messages carry it, so a late one can't flip a peer's view of our epoch back. pub(crate) epoch_seq: u32, - pub(crate) doc: AutoCommit, + pub(crate) doc: Document, heads: Vec, stamp: watch::Sender, } @@ -129,7 +128,7 @@ pub(crate) struct Stamp { } impl Replica { - pub(crate) fn new(epoch: Epoch, doc: AutoCommit) -> Self { + pub(crate) fn new(epoch: Epoch, doc: Document) -> Self { let (stamp, _) = watch::channel(Stamp { epoch, epoch_seq: 0, @@ -168,30 +167,40 @@ impl Replica { self.stamp.subscribe() } - /// The writes since the last commit, as one change stamped `at`: receivers tell by - /// it how long the change took to reach them (`swarm::stats`). - pub(crate) fn commit(&mut self, at: crate::clock::UnixMillis) { - self.doc.commit_with( - automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64), - ); - } - - /// The writes since the last commit, discarded. - pub(crate) fn rollback(&mut self) { - self.doc.rollback(); + /// The writes `f` makes, as one change stamped `at` if it succeeds (receivers tell + /// by the stamp how long the change took to reach them, `swarm::stats`), dropped if + /// not. Doesn't refresh the heads: the caller does, under the same lock. + pub(crate) fn transact( + &mut self, + at: crate::clock::UnixMillis, + f: impl FnOnce(&mut Transaction<'_>) -> Result, + ) -> Result { + let mut transaction = self.doc.transaction(); + let out = f(&mut transaction); + match &out { + Ok(_) => { + transaction.commit_with( + automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64), + ); + } + Err(_) => { + transaction.rollback(); + } + } + out } } impl Deref for Replica { - type Target = AutoCommit; + type Target = Document; - fn deref(&self) -> &AutoCommit { + fn deref(&self) -> &Document { &self.doc } } impl DerefMut for Replica { - fn deref_mut(&mut self) -> &mut AutoCommit { + fn deref_mut(&mut self) -> &mut Document { &mut self.doc } } diff --git a/src/document/store.rs b/src/document/store.rs index 7d54626..0a3c8ad 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -2,7 +2,7 @@ use anyhow::anyhow; use automerge::{ - ActorId, AutoCommit, ObjId, ObjType, Prop, ReadDoc, ScalarValue, Value, + ActorId, Automerge, ObjId, ObjType, Prop, ReadDoc, ScalarValue, Value, transaction::Transactable, }; use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; @@ -37,7 +37,7 @@ pub(crate) struct Changed { /// incremented in a map by its key, and list and text edits by the containing object; /// inside a list as `lists` says. pub(crate) fn changed( - doc: &mut AutoCommit, + doc: &Automerge, before: &[automerge::ChangeHash], after: &[automerge::ChangeHash], lists: InLists, @@ -1088,7 +1088,7 @@ mod tests { #[test] fn paths_go_through_lists() { - let mut doc = AutoCommit::new(); + let mut doc = Automerge::new(); try_change(&mut doc, |doc| { write_path(doc, &path("list"), &json!(["b", {"k": 1}, "d"])) }) @@ -1133,7 +1133,7 @@ mod tests { #[test] fn negative_indexes_count_from_the_end() { - let mut doc = AutoCommit::new(); + let mut doc = Automerge::new(); try_change(&mut doc, |doc| { write_path(doc, &path("list"), &json!(["a", "b"])) }) @@ -1161,7 +1161,7 @@ mod tests { #[test] fn inserts_into_lists() { - let mut doc = AutoCommit::new(); + let mut doc = Automerge::new(); try_change(&mut doc, |doc| { write_path(doc, &path("list"), &json!(["b", "d"])) }) diff --git a/src/document/sync.rs b/src/document/sync.rs index 625d805..2a24007 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -14,7 +14,7 @@ use automerge::sync::{self, SyncDoc}; use tokio::time::Instant; use tracing::debug; -use super::Document; +use super::{Document, ReadDoc}; use crate::backoff::Backoff; use crate::swarm::throughput::SyncBytes; use crate::transport::SyncPayload; @@ -126,7 +126,7 @@ impl SyncSession { /// peer so far (`swarm::throughput`). pub(crate) fn generate( &mut self, - doc: &mut Document, + doc: &Document, now: Instant, bytes: SyncBytes, ) -> Option { @@ -135,7 +135,7 @@ impl SyncSession { self.state.last_sent_heads.clone(), self.state.have_responded, ); - let message = doc.sync().generate_sync_message(&mut self.state); + let message = doc.generate_sync_message(&mut self.state); let in_step = self.state.their_heads.as_ref() == Some(&doc.get_heads()); let sent: Vec<_> = self .state @@ -167,7 +167,7 @@ impl SyncSession { message: Received, ) -> Result<(), AutomergeError> { self.delta_offered = false; - doc.sync().receive_sync_message(&mut self.state, message.0) + doc.receive_sync_message(&mut self.state, message.0) } /// The message just generated is too large to send in one piece: take back what it @@ -194,7 +194,7 @@ impl SyncSession { /// Whether `doc` holds everything the peer had when it last told us its heads: they /// are known, not empty (empty heads are "contained" in every document, so they /// prove nothing), and all in `doc`. - pub(crate) fn have_all_theirs(&self, doc: &mut Document) -> bool { + pub(crate) fn have_all_theirs(&self, doc: &Document) -> bool { self.state.their_heads.as_ref().is_some_and(|heads| { !heads.is_empty() && heads @@ -392,16 +392,16 @@ mod tests { change(&mut a, |a| a.put(automerge::ROOT, "early", 1).unwrap()); // Until in step. for _ in 0..10 { - let to_b = a.sync().generate_sync_message(&mut at_a); - let to_a = b.sync().generate_sync_message(&mut at_b); + let to_b = a.generate_sync_message(&mut at_a); + let to_a = b.generate_sync_message(&mut at_b); if to_b.is_none() && to_a.is_none() { break; } if let Some(m) = to_b { - b.sync().receive_sync_message(&mut at_b, m).unwrap(); + b.receive_sync_message(&mut at_b, m).unwrap(); } if let Some(m) = to_a { - a.sync().receive_sync_message(&mut at_a, m).unwrap(); + a.receive_sync_message(&mut at_a, m).unwrap(); } } assert_eq!(a.get_heads(), b.get_heads()); @@ -411,21 +411,21 @@ mod tests { a.put(automerge::ROOT, format!("lost{i}"), i).unwrap() }); } - let _lost = a.sync().generate_sync_message(&mut at_a).unwrap(); + let _lost = a.generate_sync_message(&mut at_a).unwrap(); change(&mut a, |a| a.put(automerge::ROOT, "last", 6).unwrap()); at_a.in_flight = false; // As after our reply timeout. - let last = a.sync().generate_sync_message(&mut at_a).unwrap(); + let last = a.generate_sync_message(&mut at_a).unwrap(); assert_eq!(last.changes.len(), 1); - b.sync().receive_sync_message(&mut at_b, last).unwrap(); + b.receive_sync_message(&mut at_b, last).unwrap(); // b asks for the fifth, saying nothing of what it has. - let ask = b.sync().generate_sync_message(&mut at_b).unwrap(); + let ask = b.generate_sync_message(&mut at_b).unwrap(); assert!(ask.have.is_empty() && ask.need.len() == 1); - a.sync().receive_sync_message(&mut at_a, ask).unwrap(); + a.receive_sync_message(&mut at_a, ask).unwrap(); assert!(answer_for_everything_since_shared(&mut at_a)); // All since the shared heads, in one message (here the whole document: more // than a third of it is missing). - let answer = a.sync().generate_sync_message(&mut at_a).unwrap(); - b.sync().receive_sync_message(&mut at_b, answer).unwrap(); + let answer = a.generate_sync_message(&mut at_a).unwrap(); + b.receive_sync_message(&mut at_b, answer).unwrap(); assert_eq!(b.get_heads(), a.get_heads()); // A message with what it has is left alone. assert!(!answer_for_everything_since_shared(&mut at_b)); @@ -485,13 +485,13 @@ mod tests { let bytes = SyncBytes::default(); // a opens, b answers with its (empty) heads, then a sends the changes. - let opening = a.generate(&mut ours, now, bytes).unwrap(); + let opening = a.generate(&ours, now, bytes).unwrap(); a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&opening).unwrap()).unwrap(); - let answer = b.generate(&mut theirs, now, bytes).unwrap(); + let answer = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&answer).unwrap()).unwrap(); - let carrying = a.generate(&mut ours, now, bytes).unwrap(); + let carrying = a.generate(&ours, now, bytes).unwrap(); let carried = carrying.0.len(); assert!(carried > 4_000, "the changes took {carried} bytes"); @@ -506,7 +506,7 @@ mod tests { .is_some() ); assert!(!a.delta_offered); - let again = a.generate(&mut ours, now, bytes).unwrap(); + let again = a.generate(&ours, now, bytes).unwrap(); let carried_again = again.0.len(); assert!(carried_again > 4_000, "sent again in {carried_again} bytes"); } @@ -531,27 +531,24 @@ mod tests { let bytes = SyncBytes::default(); // b (joining) opens; a's answer would carry everything: withdrawn for a delta. - let opening = b.generate(&mut theirs, now, bytes).unwrap(); + let opening = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&opening).unwrap()).unwrap(); - assert!(a.generate(&mut ours, now, bytes).is_some()); + assert!(a.generate(&ours, now, bytes).is_some()); a.withdraw_for_delta(); a.send_succeeded(a.id()); // b loads the delta and tells a its heads, which are a's now. theirs.merge(&mut ours.clone()).unwrap(); - assert!( - !b.have_all_theirs(&mut theirs), - "b hasn't heard a's heads yet" - ); - let after_delta = b.generate(&mut theirs, now, bytes).unwrap(); + assert!(!b.have_all_theirs(&theirs), "b hasn't heard a's heads yet"); + let after_delta = b.generate(&theirs, now, bytes).unwrap(); b.send_succeeded(b.id()); a.receive(&mut ours, decode(&after_delta).unwrap()).unwrap(); let answer = a - .generate(&mut ours, now, bytes) + .generate(&ours, now, bytes) .expect("a answers with its heads"); b.receive(&mut theirs, decode(&answer).unwrap()).unwrap(); - assert!(b.have_all_theirs(&mut theirs)); + assert!(b.have_all_theirs(&theirs)); } } diff --git a/src/document/testing.rs b/src/document/testing.rs index b34d85f..7b75a96 100644 --- a/src/document/testing.rs +++ b/src/document/testing.rs @@ -1,12 +1,13 @@ //! For tests: writing to documents the way the node does, one change at a time, so //! that tests don't depend on how the engine groups writes into changes. -use super::Document; +use super::{Document, Transaction}; /// The writes `f` makes to `doc`, as one change; what `f` returns. -pub(crate) fn change(doc: &mut Document, f: impl FnOnce(&mut Document) -> T) -> T { - let out = f(doc); - doc.commit(); +pub(crate) fn change(doc: &mut Document, f: impl FnOnce(&mut Transaction<'_>) -> T) -> T { + let mut transaction = doc.transaction(); + let out = f(&mut transaction); + transaction.commit(); out } @@ -14,15 +15,16 @@ pub(crate) fn change(doc: &mut Document, f: impl FnOnce(&mut Document) -> T) /// `f` returns. Committed at time 0: the same writes make the same change. pub(crate) fn try_change( doc: &mut Document, - f: impl FnOnce(&mut Document) -> Result, + f: impl FnOnce(&mut Transaction<'_>) -> Result, ) -> Result { - let out = f(doc); + let mut transaction = doc.transaction(); + let out = f(&mut transaction); match &out { Ok(_) => { - doc.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); + transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(0)); } Err(_) => { - doc.rollback(); + transaction.rollback(); } } out -- cgit v1.3.1