document: introduce patch log

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 {