1 files changed, 30 insertions, 16 deletions

diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 6c90f44..4c72b39 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -20,7 +20,6 @@
 
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit, ChangeHash};
 use tokio::sync::broadcast;
 use tracing::{error, info, warn};
 
@@ -38,8 +37,9 @@ use super::vote;
 use crate::document::meta::{self, REMAKES};
 use crate::document::replica::Epoch;
 use crate::document::store;
+use crate::document::store::paths_related;
+use crate::document::{ActorId, ChangeHash, Document};
 use crate::locks::MutexExt;
-use crate::swarm::sync::paths_related;
 
 const TICK: Duration = Duration::from_secs(1);
 
@@ -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();
@@ -218,11 +219,11 @@ async fn switch<H: Host>(
 
     // 2. Build (or reuse), verify and replay without holding any lock.
     let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
-        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
-            let mut copy = copy;
+        move || -> Result<(Document, Vec<ChangeHash>, ReplayStats), String> {
+            let copy = copy;
             let snapshot = match prebuilt {
                 Some(snapshot) => snapshot,
-                None => snapshot::build(&mut copy, &request.params())
+                None => snapshot::build(&copy, &request.params())
                     .map_err(|error| format!("snapshot build failed: {error}"))?,
             };
             if snapshot.epoch != expected {
@@ -235,13 +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);
 
+            let heads = copy.get_heads();
+
             // Keep this remake's record visible in the new epoch, for its history.
-            if carry_record && let Err(error) = store::copy_path(&copy, &mut doc, &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 heads = copy.get_heads();
-            let stats = delta::replay(&mut copy, &base, &heads, &mut doc, &translator);
+            // 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))
         },
     )
@@ -276,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"
     );
@@ -288,25 +297,30 @@ async fn switch<H: Host>(
 pub(super) async fn swap_in<H: Host>(
     host: &H,
     target: Epoch,
-    new_doc: AutoCommit,
+    new_doc: Document,
     copy_heads: Vec<ChangeHash>,
     translator: Translator,
-) -> Result<(AutoCommit, ReplayStats), String> {
+) -> 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);
     let old = host
         .switch_epoch(target, move |live| {
             // Only one worker switches, but be sure the copy still describes the live document.
-            if !copy_heads
-                .iter()
-                .all(|head| live.get_change_by_hash(head).is_some())
-            {
+            if !crate::document::changes::missing(live, &copy_heads).is_empty() {
                 return Err("the document was replaced while preparing the switch".to_owned());
             }
             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,