LWW everywhere, mostly

1 files changed, 25 insertions, 9 deletions

diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 2da8540..319037d 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -49,7 +49,6 @@ use super::quarantine;
 use super::snapshot::{self, EpochMetadata};
 use crate::clock::UnixMillis;
 use crate::document::replica::Epoch;
-use crate::document::store::StoreError;
 
 /// Pause before trying the same target again after a failed rejoin.
 const RETRY: Duration = Duration::from_secs(30);
@@ -308,8 +307,14 @@ async fn fetch_snapshot<H: Host>(host: H, target: Epoch) {
 /// What to do with this node's own changes the new epoch lacks.
 enum Plan {
     Nothing,
-    /// Replay these paths, carried through every remake since the node's epoch.
-    Replay(Vec<Vec<String>>, Translator),
+    /// Replay these paths (what changed from `before` to `after` in the old
+    /// document), carried through every remake since the node's epoch.
+    Replay {
+        paths: Vec<Vec<String>>,
+        before: Vec<ChangeHash>,
+        after: Vec<ChangeHash>,
+        translator: Translator,
+    },
     Quarantine {
         paths: Vec<Vec<String>>,
         reason: String,
@@ -402,13 +407,18 @@ async fn finish<H: Host>(host: &H) {
 
     match plan {
         Plan::Nothing => {}
-        Plan::Replay(paths, translator) => {
+        Plan::Replay {
+            paths,
+            before,
+            after,
+            translator,
+        } => {
             let document = Arc::new(old.take().expect("old document"));
+            let superseded = host.superseded();
             let replayed = {
                 let document = Arc::clone(&document);
-                let paths = paths.clone();
-                host.commit_doc(move |doc| {
-                    Ok::<_, StoreError>(delta::replay_paths(&document, doc, &paths, &translator))
+                host.commit_changes(move |doc| {
+                    delta::replay_newest(&document, &before, &after, doc, &translator, superseded)
                 })
                 .await
             };
@@ -478,6 +488,7 @@ async fn finish<H: Host>(host: &H) {
         own_changes = own_missing,
         replayed = stats.replayed,
         failed = stats.failed,
+        superseded = stats.superseded,
         quarantine = ?quarantine_id,
         secs = started.elapsed().as_secs(),
         "Rejoined newer epoch"
@@ -578,7 +589,12 @@ fn prepare(
         (Some(next), Ok(translator)) => {
             let away_ms = now.ms_since(next.proposed_at);
             if next.policy.replay_window.covers(gap, away_ms) {
-                Plan::Replay(paths, translator)
+                Plan::Replay {
+                    paths,
+                    before: missing.before.clone(),
+                    after: old_heads.clone(),
+                    translator,
+                }
             } else {
                 Plan::Quarantine {
                     paths,
@@ -733,7 +749,7 @@ mod tests {
             UnixMillis(0),
         ) {
             Prepared::Ready {
-                plan: Plan::Replay(paths, _),
+                plan: Plan::Replay { paths, .. },
                 own_missing,
                 ..
             } => {