From 2ac2b0e2b08f14be07c46b3fd9a516a64e9933fb Mon Sep 17 00:00:00 2001
From: tv <tv@krebsco.de>
Date: Sun, 1 Nov 2026 19:29:26 +0100
Subject: LWW everywhere, mostly

---
 src/remake/switch.rs | 30 ++++++++++++++++++++----------
 1 file changed, 20 insertions(+), 10 deletions(-)

(limited to 'src/remake/switch.rs')

diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 9b03789..7d6611d 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -193,6 +193,7 @@ async fn switch<H: Host>(
         .translator()
         .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
     let late_translator = translator.clone();
+    let superseded = host.superseded();
     let previous = request.previous;
     let base = request.base_heads.clone();
     let retention = record.proposal.policy.archive_retention.clone();
@@ -235,20 +236,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 transaction, &record_path)
-            {
-                warn!(error = %error, "Could not carry the remake record into the new epoch");
+            if carry_record {
+                let mut transaction = doc.transaction();
+                if let Err(error) = store::copy_path(&copy, &mut transaction, &record_path) {
+                    warn!(error = %error, "Could not carry the remake record into the new epoch");
+                }
+                transaction.commit();
             }
 
-            let stats = delta::replay_paths(&copy, &mut transaction, &paths, &translator);
-            transaction.commit();
+            // Each write at the time it was made, so later replays judge by it.
+            let stats =
+                delta::replay_newest(&copy, &base, &heads, &mut doc, &translator, superseded);
             Ok((doc, heads, stats))
         },
     )
@@ -283,6 +284,7 @@ async fn switch<H: Host>(
         replayed = stats.replayed,
         failed = stats.failed,
         dropped = stats.dropped,
+        superseded = stats.superseded,
         quarantine = ?quarantine,
         "Switched to the new epoch"
     );
@@ -299,6 +301,7 @@ pub(super) async fn swap_in<H: Host>(
     copy_heads: Vec<ChangeHash>,
     translator: Translator,
 ) -> Result<(Document, ReplayStats), String> {
+    let superseded = host.superseded();
     // The closure runs on the blocking pool; its replay numbers come back through this.
     let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
     let late_out = std::sync::Arc::clone(&late_stats);
@@ -310,7 +313,14 @@ pub(super) async fn swap_in<H: Host>(
             }
             let mut new_doc = new_doc;
             let live_heads = live.get_heads();
-            let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc, &translator);
+            let late = delta::replay_newest(
+                live,
+                &copy_heads,
+                &live_heads,
+                &mut new_doc,
+                &translator,
+                superseded,
+            );
             if late.replayed + late.failed > 0 {
                 info!(
                     replayed = late.replayed,
-- 
cgit v1.3.1