aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/document/replica.rs138
-rw-r--r--src/document/store.rs7
-rw-r--r--src/document/sync.rs15
-rw-r--r--src/swarm/changes.rs4
-rw-r--r--src/swarm/sync.rs6
5 files changed, 148 insertions, 22 deletions
diff --git a/src/document/replica.rs b/src/document/replica.rs
index a6fc504..0d1998a 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -121,6 +121,37 @@ pub(crate) struct Replica {
pub(crate) writable: bool,
heads: Vec<ChangeHash>,
stamp: watch::Sender<Stamp>,
+ /// What local writes and sync changed since the change publisher last looked
+ /// (`changed_since`), so it needn't diff the document for it.
+ log: std::sync::Mutex<Log>,
+}
+
+/// The patches of the changes from `from` to `to` (sorted heads) of the document of
+/// `epoch_seq`, as they were made. Changes made otherwise (applied from a delta, a
+/// merge) aren't in it: the heads then aren't `to` any more, and whoever looks diffs
+/// instead. Another document (an epoch switch) starts a log of its own.
+struct Log {
+ epoch_seq: u32,
+ from: Vec<ChangeHash>,
+ to: Vec<ChangeHash>,
+ patches: automerge::PatchLog,
+}
+
+impl Log {
+ fn at(epoch_seq: u32, heads: Vec<ChangeHash>) -> Self {
+ Self {
+ epoch_seq,
+ from: heads.clone(),
+ to: heads,
+ patches: automerge::PatchLog::active(),
+ }
+ }
+}
+
+fn sorted_heads(doc: &Document) -> Vec<ChangeHash> {
+ let mut heads = doc.get_heads();
+ heads.sort_unstable();
+ heads
}
/// The epoch label and the heads of the replica, as of its last refresh, published
@@ -140,6 +171,7 @@ impl Replica {
epoch_seq: 0,
heads: Arc::from([]),
});
+ let log = std::sync::Mutex::new(Log::at(0, sorted_heads(&doc)));
let mut replica = Self {
epoch,
epoch_seq: 0,
@@ -147,6 +179,7 @@ impl Replica {
writable: true,
heads: Vec::new(),
stamp,
+ log,
};
replica.refresh_heads();
replica
@@ -184,7 +217,53 @@ impl Replica {
writer: &Writer,
f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
) -> Result<T, E> {
- transact(&mut self.doc, at, writer, f)
+ self.logged(|doc, log| transact_logged(doc, at, writer, log, f))
+ }
+
+ /// Runs `f`, which changes the document and records what it changed in the log it
+ /// is given; it returns whether the log still holds everything (a rolled-back
+ /// transaction takes the log with it).
+ pub(crate) fn logged<T>(
+ &mut self,
+ f: impl FnOnce(&mut Document, &mut automerge::PatchLog) -> (T, bool),
+ ) -> T {
+ use crate::locks::MutexExt;
+ let before = sorted_heads(&self.doc);
+ let mut log = self.log.lock_or_recover();
+ if log.epoch_seq != self.epoch_seq || log.to != before {
+ // Another document, or changed otherwise since: what the log holds doesn't
+ // lead here.
+ *log = Log::at(self.epoch_seq, before);
+ }
+ let (out, whole) = f(&mut self.doc, &mut log.patches);
+ let after = sorted_heads(&self.doc);
+ if whole {
+ log.to = after;
+ } else {
+ *log = Log::at(self.epoch_seq, after);
+ }
+ out
+ }
+
+ /// The store paths changed since `before` (sorted heads, of this document) to
+ /// now (`store::changed`): from the log where it covers exactly that, else by a
+ /// diff. Starts the log again from now.
+ pub(crate) fn changed_since(
+ &self,
+ before: &[ChangeHash],
+ lists: super::store::InLists,
+ ) -> Vec<super::store::Changed> {
+ use crate::locks::MutexExt;
+ let now = sorted_heads(&self.doc);
+ let mut log = self.log.lock_or_recover();
+ let changed = if log.epoch_seq == self.epoch_seq && log.from == before && log.to == now {
+ let patches = self.doc.make_patches(&mut log.patches);
+ super::store::changed_in(&patches, lists)
+ } else {
+ super::store::changed(&self.doc, before, &now, lists)
+ };
+ *log = Log::at(self.epoch_seq, now);
+ changed
}
}
@@ -283,23 +362,52 @@ pub(crate) fn transact<T, E>(
writer: &Writer,
f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
) -> Result<T, E> {
+ transact_logged(doc, at, writer, &mut automerge::PatchLog::inactive(), f).0
+}
+
+/// [`transact`], what it changes recorded in `log`; with whether `log` still holds
+/// everything (`Replica::logged`).
+fn transact_logged<T, E>(
+ doc: &mut Document,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ log: &mut automerge::PatchLog,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+) -> (Result<T, E>, bool) {
authored(doc, writer.node);
- let mut transaction = doc.transaction();
- let out = f(&mut transaction);
- match &out {
- Ok(_) => {
- let mut options =
- automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64);
- if let Some(user) = writer.user {
- options.set_message(user);
- }
- transaction.commit_with(options);
- }
- Err(_) => {
- transaction.rollback();
+ let taken = std::mem::replace(log, automerge::PatchLog::inactive());
+ if let Ok(transaction) = doc.transaction_log_patches(taken) {
+ let (out, logged) = run(transaction, at, writer, f);
+ let whole = logged.is_some();
+ if let Some(logged) = logged {
+ *log = logged;
}
+ return (out, whole);
+ }
+ // Another document's log (the epoch switched): this change goes unlogged.
+ (run(doc.transaction(), at, writer, f).0, false)
+}
+
+/// Runs `f` in `transaction`, committing what it writes if it succeeds (stamped `at`,
+/// by `writer`) with the transaction's log, rolling it back if not.
+fn run<T, E>(
+ mut transaction: Transaction<'_>,
+ at: crate::clock::UnixMillis,
+ writer: &Writer,
+ f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E>,
+) -> (Result<T, E>, Option<automerge::PatchLog>) {
+ let out = f(&mut transaction);
+ if out.is_err() {
+ transaction.rollback();
+ return (out, None);
+ }
+ let mut options =
+ automerge::transaction::CommitOptions::default().with_time(at.as_u64() as i64);
+ if let Some(user) = writer.user {
+ options.set_message(user);
}
- out
+ let (_, logged) = transaction.commit_with(options);
+ (out, Some(logged))
}
/// Makes `node` the author of `doc`'s changes from now on. Setting an author gives the
diff --git a/src/document/store.rs b/src/document/store.rs
index 13824b9..4d9bc74 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -42,7 +42,12 @@ pub(crate) fn changed(
after: &[automerge::ChangeHash],
lists: InLists,
) -> Vec<Changed> {
- doc.diff(before, after)
+ changed_in(&doc.diff(before, after), lists)
+}
+
+/// The store paths `patches` touched, as [`changed`] has them.
+pub(crate) fn changed_in(patches: &[automerge::Patch], lists: InLists) -> Vec<Changed> {
+ patches
.iter()
.map(|patch| Changed {
path: patch_path(patch, lists),
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 2119eda..44ad833 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -165,13 +165,26 @@ impl SyncSession {
/// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
/// answer: the caller measures the exchange first (`reply`).
+ #[cfg(test)]
pub(crate) fn receive(
&mut self,
doc: &mut Document,
message: Received,
) -> Result<(), AutomergeError> {
+ self.receive_logged(doc, message, &mut automerge::PatchLog::inactive())
+ }
+
+ /// Applies a sync message from the peer to `doc`, what it changes recorded in `log`
+ /// (`Replica::logged`). Doesn't settle the awaited answer: the caller measures the
+ /// exchange first (`reply`).
+ pub(crate) fn receive_logged(
+ &mut self,
+ doc: &mut Document,
+ message: Received,
+ log: &mut automerge::PatchLog,
+ ) -> Result<(), AutomergeError> {
self.delta_offered = false;
- doc.receive_sync_message(&mut self.state, message.0)
+ doc.receive_sync_message_log_patches(&mut self.state, message.0, log)
}
/// The message just generated is too large to send in one piece: take back what it
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
index 8565cb6..0f22f52 100644
--- a/src/swarm/changes.rs
+++ b/src/swarm/changes.rs
@@ -8,7 +8,7 @@ use std::sync::Arc;
use tracing::{info, warn};
use crate::document::changes;
-use crate::document::store::{self, ChangeSet, InLists};
+use crate::document::store::{ChangeSet, InLists};
use crate::state::AppState;
pub(crate) async fn run_change_publisher(state: AppState) {
@@ -37,7 +37,7 @@ pub(crate) async fn run_change_publisher(state: AppState) {
let mut paths = Vec::new();
let mut conflicts = Vec::new();
- for changed in store::changed(&replica.doc, &before, &heads, InLists::Index) {
+ for changed in replica.changed_since(&before, InLists::Index) {
if changed.conflict {
conflicts.push(changed.path.clone());
}
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b881609..3bfbf7d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -389,9 +389,9 @@ pub(crate) async fn handle_sync(
peer_state.reports_ready = Some(from_ready);
let heads_before = replica.heads().to_vec();
- peer_state
- .sync
- .receive(&mut replica.doc, message)
+ let sync = &mut peer_state.sync;
+ replica
+ .logged(|doc, log| (sync.receive_logged(doc, message, log), true))
.context("Automerge sync failed")?;
peer_state.last_sync_rx = Some(Instant::now());
if let Some(wait) = &peer_state.sync.reply {