document: introduce patch log

1 files changed, 123 insertions, 15 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