aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api/document.rs4
-rw-r--r--src/document/changes.rs25
-rw-r--r--src/document/file.rs12
-rw-r--r--src/document/lock.rs11
-rw-r--r--src/document/mod.rs14
-rw-r--r--src/document/replica.rs53
-rw-r--r--src/document/store.rs10
-rw-r--r--src/document/sync.rs55
-rw-r--r--src/document/testing.rs18
-rw-r--r--src/remake/catchup.rs32
-rw-r--r--src/remake/coordinator.rs4
-rw-r--r--src/remake/delta.rs72
-rw-r--r--src/remake/kind.rs59
-rw-r--r--src/remake/mod.rs19
-rw-r--r--src/remake/quarantine.rs8
-rw-r--r--src/remake/records.rs13
-rw-r--r--src/remake/snapshot.rs94
-rw-r--r--src/remake/switch.rs17
-rw-r--r--src/state/host.rs36
-rw-r--r--src/state/mod.rs14
-rw-r--r--src/swarm/blob.rs5
-rw-r--r--src/swarm/bootstrap.rs5
-rw-r--r--src/swarm/changes.rs7
-rw-r--r--src/swarm/document_file.rs4
-rw-r--r--src/swarm/members.rs10
-rw-r--r--src/swarm/relay.rs4
-rw-r--r--src/swarm/stats.rs4
-rw-r--r--src/swarm/sync.rs10
28 files changed, 335 insertions, 284 deletions
diff --git a/src/api/document.rs b/src/api/document.rs
index 08fd21f..bc8c47e 100644
--- a/src/api/document.rs
+++ b/src/api/document.rs
@@ -50,11 +50,11 @@ pub(crate) async fn get_document(State(state): State<AppState>) -> Response {
// Both steps take time proportional to the document: off the async threads, and
// the lock only for the copy.
let saved = tokio::task::spawn_blocking(move || {
- let (mut copy, epoch, heads) = {
+ let (copy, epoch, heads) = {
let replica = doc.blocking_read();
(replica.doc.clone(), replica.epoch, replica.heads().to_vec())
};
- (crate::document::save(&mut copy), heads, epoch)
+ (crate::document::save(&copy), heads, epoch)
})
.await;
let (bytes, heads, epoch) = match saved {
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<ChangeHash> {
+/// Those of `heads` that `doc` doesn't have.
+pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec<ChangeHash> {
heads
.iter()
.copied()
@@ -19,8 +18,8 @@ pub(crate) fn missing(doc: &mut Document, heads: &[ChangeHash]) -> Vec<ChangeHas
}
/// When the changes after `since` were committed, in Unix milliseconds by their
-/// writers' clocks (`Replica::commit`); changes without a time (snapshots) left out.
-pub(crate) fn commit_times(doc: &mut Document, since: &[ChangeHash]) -> Vec<i64> {
+/// writers' clocks (`Replica::transact`); changes without a time (snapshots) left out.
+pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> {
doc.get_changes(since)
.iter()
.map(|change| change.timestamp())
@@ -29,7 +28,7 @@ pub(crate) fn commit_times(doc: &mut Document, since: &[ChangeHash]) -> Vec<i64>
}
/// 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<Vec<u8>> {
+pub(crate) fn change_bytes(doc: &Document, hash: &ChangeHash) -> Option<Vec<u8>> {
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<u8>, hash: ChangeHash) -> Result<Chan
/// (`load_delta` reads that). The same on every node holding `to`, whatever else it
/// has. None if `doc` lacks part of `to`.
pub(crate) fn delta_bytes(
- doc: &mut Document,
+ doc: &Document,
from: &[ChangeHash],
to: &[ChangeHash],
) -> Option<Vec<u8>> {
@@ -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<ActorId>,
}
@@ -88,7 +88,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
.collect::<Result<Vec<_>>>()?;
let mut records = &rest[line_end + 1..];
- let mut doc: Option<AutoCommit> = None;
+ let mut doc: Option<Document> = 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<Self>,
epoch: Epoch,
- mut copy: AutoCommit,
+ copy: Document,
own: Vec<ActorId>,
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(&copy);
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<u8> {
+pub(crate) fn save(doc: &Document) -> Vec<u8> {
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<u8> {
+pub(crate) fn save_since(doc: &Document, since: &[ChangeHash]) -> Vec<u8> {
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<ChangeHash>,
stamp: watch::Sender<Stamp>,
}
@@ -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<T, E>(
+ &mut self,
+ at: crate::clock::UnixMillis,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+ ) -> Result<T, E> {
+ 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<SyncPayload> {
@@ -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<T>(doc: &mut Document, f: impl FnOnce(&mut Document) -> T) -> T {
- let out = f(doc);
- doc.commit();
+pub(crate) fn change<T>(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<T>(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<T, E>(
doc: &mut Document,
- f: impl FnOnce(&mut Document) -> Result<T, E>,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
) -> Result<T, E> {
- 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
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(&copy, 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(&copy, &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, &params(&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, &params(&base, &policy)).unwrap();
+ let snap = build(&a, &params(&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, &params(&heads, &policy)).unwrap();
+ let snap = build(&a, &params(&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, &params(&heads, &policy)).unwrap();
+ let snap = build(&a, &params(&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, &params(&foreign, &policy)),
+ build(&a, &params(&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, &params(&base, &policy)).unwrap();
+ let snap = build(&late, &params(&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(&copy, &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(&copy, &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(&copy, &mut doc, &record_path) {
+ if carry_record
+ && let Err(error) = store::copy_path(&copy, &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(&copy, &mut transaction, &paths, &translator);
+ transaction.commit();
Ok((doc, heads, stats))
},
)
diff --git a/src/state/host.rs b/src/state/host.rs
index a878381..24a989e 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -13,7 +13,7 @@ use tracing::debug;
use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing};
use crate::document::replica::Epoch;
use crate::document::store::ChangeSet;
-use crate::document::{ChangeHash, Document};
+use crate::document::{ChangeHash, Document, ReadDoc, Transaction};
use crate::remake::{Host, PeerSummary, RemakeState};
impl AppState {
@@ -86,7 +86,7 @@ impl Host for AppState {
crate::swarm::skew::measure(self).await
}
- fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec<String> {
+ fn electorate(&self, doc: &impl ReadDoc, now: crate::clock::UnixMillis) -> Vec<String> {
crate::swarm::members::electorate(doc, now.as_u64(), self.electorate_window)
}
@@ -218,7 +218,7 @@ impl Host for AppState {
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,
{
@@ -226,13 +226,39 @@ impl Host for AppState {
async move { state.mutate_doc(f).await }
}
+ fn apply_change(
+ &self,
+ change: crate::document::Change,
+ ) -> impl Future<Output = Result<(), String>> + Send {
+ let state = self.clone();
+ async move {
+ tokio::task::spawn_blocking(move || {
+ let result = {
+ let mut replica = state.doc.blocking_write();
+ let result = replica
+ .apply_changes([change])
+ .map_err(|error| error.to_string());
+ replica.refresh_heads();
+ result
+ }; // lock released before notifying
+
+ if result.is_ok() {
+ state.publish_change();
+ }
+ result
+ })
+ .await
+ .expect("document write task panicked")
+ }
+ }
+
fn switch_epoch<E, F>(
&self,
epoch: Epoch,
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,
{
let state = self.clone();
@@ -241,7 +267,7 @@ impl Host for AppState {
let state = state.clone();
tokio::task::spawn_blocking(move || -> Result<Document, E> {
let mut replica = state.doc.blocking_write();
- let new_doc = f(&mut replica.doc)?;
+ let new_doc = f(&replica.doc)?;
let old_doc = std::mem::replace(&mut replica.doc, new_doc);
replica.epoch = epoch;
replica.epoch_seq += 1;
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 75c3780..d8bdac5 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -36,10 +36,10 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
use tokio::time::Instant;
use tracing::{debug, error, warn};
-use crate::document::Document;
use crate::document::lock::DocumentLock;
use crate::document::replica::{Epoch, Stamp};
use crate::document::store::ChangeSet;
+use crate::document::{Document, Transaction};
use crate::locks::MutexExt;
use crate::remake::{Host, RemakeState};
use crate::swarm::sync::SyncEvent;
@@ -396,7 +396,7 @@ impl AppState {
/// cancelled request can't leave a committed write unannounced.
pub(crate) async fn mutate_doc<T, E>(
&self,
- f: impl FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
) -> Result<T, E>
where
T: Send + 'static,
@@ -407,13 +407,9 @@ impl AppState {
tokio::task::spawn_blocking(move || {
let result = {
let mut replica = state.doc.blocking_write();
- let result = f(&mut replica.doc);
- match &result {
- // One API write = one change.
- Ok(_) => replica.commit(state.clock.now_millis()),
- // Discard partial ops (e.g. old value deleted, new one not written).
- Err(_) => replica.rollback(),
- }
+ // One API write = one change; on error, its partial writes (e.g. the old
+ // value deleted, the new one not written) are dropped.
+ let result = replica.transact(state.clock.now_millis(), f);
replica.refresh_heads();
result
}; // lock released before notifying
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index a136217..3f66d50 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -217,7 +217,7 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti
let spec = delta.filter(|spec| spec.id() == delta_id)?;
let doc = Arc::clone(&state.doc);
tokio::task::spawn_blocking(move || {
- changes::delta_bytes(&mut doc.blocking_write().doc, &spec.from, &spec.to)
+ changes::delta_bytes(&doc.blocking_read().doc, &spec.from, &spec.to)
})
.await
.ok()
@@ -226,9 +226,8 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti
}
BlobId::Change(hash) => {
let doc = Arc::clone(&state.doc);
- // Looking a change up needs the document mutably.
tokio::task::spawn_blocking(move || {
- changes::change_bytes(&mut doc.blocking_write().doc, &hash)
+ changes::change_bytes(&doc.blocking_read().doc, &hash)
})
.await
.ok()
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index d2bbe24..848201e 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -281,7 +281,7 @@ pub(crate) fn starting_document(
/// The swarm and epoch `doc` is on.
fn identify(
- mut doc: Document,
+ doc: Document,
origin: Origin,
earlier_actors: Vec<ActorId>,
) -> Result<StartingDocument> {
@@ -289,8 +289,7 @@ fn identify(
let Some(swarm_id) = read_swarm_id(&doc) else {
bail!("{what} names no swarm (no _continuum/swarm); it predates swarm IDs in documents");
};
- let epoch =
- snapshot::epoch_of(&mut doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?;
+ let epoch = snapshot::epoch_of(&doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?;
Ok(StartingDocument {
doc,
epoch,
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
index 3a50769..7f3c2e9 100644
--- a/src/swarm/changes.rs
+++ b/src/swarm/changes.rs
@@ -21,9 +21,8 @@ pub(crate) async fn run_change_publisher(state: AppState) {
let before = last_heads.clone();
let doc = Arc::clone(&state.doc);
- // Looking changes up needs the document mutably.
let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
- let mut replica = doc.blocking_write();
+ let replica = doc.blocking_read();
let heads = replica.heads().to_vec();
if heads == before {
return (heads, Vec::new(), Vec::new());
@@ -31,13 +30,13 @@ pub(crate) async fn run_change_publisher(state: AppState) {
// After an epoch switch the previous heads belong to the old document:
// report the root as changed, which every stream is related to.
- if !changes::missing(&mut replica.doc, &before).is_empty() {
+ if !changes::missing(&replica.doc, &before).is_empty() {
return (heads, vec![Vec::new()], Vec::new());
}
let mut paths = Vec::new();
let mut conflicts = Vec::new();
- for changed in store::changed(&mut replica.doc, &before, &heads, InLists::Index) {
+ for changed in store::changed(&replica.doc, &before, &heads, InLists::Index) {
if changed.conflict {
conflicts.push(changed.path.clone());
}
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
index 435ff32..6a516a1 100644
--- a/src/swarm/document_file.rs
+++ b/src/swarm/document_file.rs
@@ -41,13 +41,13 @@ pub(crate) async fn flush(state: &AppState) {
}
Due::Append { epoch, since } => {
let (heads, increment) = {
- let mut replica = state.doc.write().await;
+ let replica = state.doc.read().await;
// Switched since: the next flush rewrites the file instead.
if replica.epoch != epoch {
return;
}
let heads = replica.heads().to_vec();
- (heads, crate::document::save_since(&mut replica.doc, &since))
+ (heads, crate::document::save_since(&replica.doc, &since))
};
file.append(since, heads, increment).await;
}
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 9885750..b3e2084 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -23,7 +23,7 @@ use tracing::{debug, info, warn};
use crate::document::meta;
use crate::document::store::{self, StoreError};
-use crate::document::{Document, Transactable};
+use crate::document::{Document, ReadDoc, Transactable};
use crate::state::AppState;
pub(crate) fn members_dir() -> Vec<String> {
@@ -37,7 +37,7 @@ pub(crate) fn member_path(pk: &str) -> Vec<String> {
}
/// Public keys of all registered members, sorted.
-pub(crate) fn members(doc: &Document) -> Vec<String> {
+pub(crate) fn members(doc: &impl ReadDoc) -> Vec<String> {
let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
members.sort();
members
@@ -199,7 +199,7 @@ fn alive_path(pk: &str) -> Vec<String> {
path
}
-fn millis_at(doc: &Document, path: &[String]) -> Option<u64> {
+fn millis_at(doc: &impl ReadDoc, path: &[String]) -> Option<u64> {
match store::read_path(doc, path) {
Ok(Some(JsonValue::Number(number))) => number.as_u64(),
_ => None,
@@ -207,7 +207,7 @@ fn millis_at(doc: &Document, path: &[String]) -> Option<u64> {
}
/// When `pk` last said it is alive (Unix ms): its last renewal, or its registration.
-pub(crate) fn alive_at(doc: &Document, pk: &str) -> Option<u64> {
+pub(crate) fn alive_at(doc: &impl ReadDoc, pk: &str) -> Option<u64> {
let mut since = member_path(pk);
since.push("since".to_owned());
millis_at(doc, &alive_path(pk)).max(millis_at(doc, &since))
@@ -215,7 +215,7 @@ pub(crate) fn alive_at(doc: &Document, pk: &str) -> Option<u64> {
/// The members that said they are alive within `window` before `now_ms` (sorted): the
/// electorate of a remake (`remake::coordinator`).
-pub(crate) fn electorate(doc: &Document, now_ms: u64, window: Duration) -> Vec<String> {
+pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> Vec<String> {
let oldest = now_ms.saturating_sub(window.as_millis() as u64);
members(doc)
.into_iter()
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index d11c3e0..1736cec 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -131,8 +131,8 @@ pub(crate) async fn pull_if_behind(
// Those of the listed heads we lack. With a digest only a sample is listed; if we
// have all of it, a later keepalive lists others.
let missing = {
- let mut replica = state.doc.write().await;
- crate::document::changes::missing(&mut replica.doc, remote.heads.listed())
+ let replica = state.doc.read().await;
+ crate::document::changes::missing(&replica.doc, remote.heads.listed())
};
if missing.is_empty() {
return false;
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 2398b09..04f41b2 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -278,8 +278,8 @@ pub(crate) fn measure_saved_size(state: &AppState) {
let doc = std::sync::Arc::clone(&state.doc);
// Both steps take time proportional to the document: off the async threads.
let measured = match tokio::task::spawn_blocking(move || {
- let mut copy = doc.blocking_read().doc.clone();
- crate::document::save(&mut copy).len() as u64
+ let copy = doc.blocking_read().doc.clone();
+ crate::document::save(&copy).len() as u64
})
.await
{
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 35bc4c7..46bdef6 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -136,7 +136,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
// Label the message with the epoch of the document it was generated
// from, read under the same lock.
let (session, epoch, epoch_seq, body) = {
- let mut replica = doc.blocking_write();
+ let replica = doc.blocking_read();
let mut peer_state = psm.blocking_lock();
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
// Under the same locks as the epoch label: the message belongs to
@@ -144,7 +144,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let message =
peer_state
.sync
- .generate(&mut replica.doc, Instant::now(), bytes_before);
+ .generate(&replica.doc, Instant::now(), bytes_before);
let body = message.map(|payload| {
if payload.0.len() <= delta_threshold {
return transport::SyncBody::Message(payload);
@@ -390,7 +390,7 @@ pub(crate) async fn handle_sync(
// doesn't count; the writer's still does.
let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms;
let delays: Vec<i64> =
- crate::document::changes::commit_times(&mut replica.doc, &heads_before)
+ crate::document::changes::commit_times(&replica.doc, &heads_before)
.into_iter()
.map(|time| now_ms - time)
.collect();
@@ -403,7 +403,7 @@ pub(crate) async fn handle_sync(
// catching up has them, and so does a ready peer whose document is empty
// (a misconfigured second swarm, or a genesis node in its first seconds,
// before it registered itself).
- check_caught_up && from_ready && peer_state.sync.have_all_theirs(&mut replica.doc),
+ check_caught_up && from_ready && peer_state.sync.have_all_theirs(&replica.doc),
changed,
)))
})
@@ -501,7 +501,7 @@ async fn download_and_apply_delta(
let heads_before = replica.heads().to_vec();
crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
replica.refresh_heads();
- let missing = crate::document::changes::missing(&mut replica.doc, &spec.to).len();
+ let missing = crate::document::changes::missing(&replica.doc, &spec.to).len();
if missing > 0 {
bail!("{missing} of the offered heads are still missing after loading");
}