LWW everywhere, mostly

10 files changed, 385 insertions, 23 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index bb65b97..0101aa4 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -257,6 +257,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). At least 7h (members renew every seventh of their own window, hourly at most); the same on every node, or a coordinator with a shorter one counts fewer members.",
     ),
     option(
+        "superseded-writes",
+        Kind::Text,
+        "drop|quarantine",
+        Some("drop"),
+        "What becomes of a write a remake or rejoin would carry over though a newer one touched its path (last writer wins, by the writers' clocks): drop leaves it out (logged), quarantine keeps it for an operator (/admin/quarantine).",
+    ),
+    option(
         "clock-skew-warn",
         Kind::Duration,
         "DURATION",
@@ -528,6 +535,8 @@ pub(crate) struct Settings {
         with = "crate::units::duration_option"
     )]
     pub(crate) electorate_window: Duration,
+    #[serde(default)]
+    pub(crate) superseded_writes: crate::remake::Superseded,
     #[serde(
         default = "default_clock_skew_warn",
         with = "crate::units::duration_option"
diff --git a/src/main.rs b/src/main.rs
index 40f0167..f170434 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -215,6 +215,7 @@ async fn main() -> Result<()> {
         alias: args.alias.clone(),
         electorate_window: args.electorate_window,
         clock_skew_warn: args.clock_skew_warn,
+        superseded_writes: args.superseded_writes,
         // At most 4MiB (`Settings::check`).
         blob_chunk_bytes: args.blob_chunk as u32,
         swarm_key,
diff --git a/src/node.rs b/src/node.rs
index f6bdbb3..e3a832a 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -46,6 +46,8 @@ pub(crate) struct NodeConfig {
     pub(crate) alias: Option<String>,
     pub(crate) electorate_window: Duration,
     pub(crate) clock_skew_warn: Duration,
+    /// `--superseded-writes`.
+    pub(crate) superseded_writes: crate::remake::Superseded,
     pub(crate) blob_chunk_bytes: u32,
     pub(crate) swarm_key: Option<Arc<SwarmKey>>,
     pub(crate) protocol_writers: HashSet<String>,
@@ -114,6 +116,7 @@ impl Node {
             alias: config.alias,
             electorate_window: config.electorate_window,
             clock_skew_warn: config.clock_skew_warn,
+            superseded_writes: config.superseded_writes,
             clock_ahead_ms: Default::default(),
             clock_skew: Default::default(),
             blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)),
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,
                 ..
             } => {
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
index 06f2913..2c6c93d 100644
--- a/src/remake/delta.rs
+++ b/src/remake/delta.rs
@@ -20,11 +20,19 @@
 //!
 //! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never
 //! replayed: the snapshot owns it.
+//!
+//! Last writer wins, by the writers' clocks ([`replay_newest`]): a replayed value is a
+//! fresh operation, which by Automerge's order beats everything the document it goes
+//! into has, newer writes included. So a path is replayed only if no change in that
+//! document touched it (or a path above or below it) later than the write being
+//! replayed, as the changes' timestamps say; and the replayed changes carry the times
+//! of the writes they replay, not of the replay, so later replays judge by those.
 
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
-use crate::document::{ActorId, ChangeHash, Document, Transactable};
-use tracing::warn;
+use crate::document::{ActorId, ChangeHash, Document, ReadDoc, Transactable};
+use serde::{Deserialize, Serialize};
+use tracing::{info, warn};
 
 use super::kind::{Translation, Translator};
 use crate::document::meta::{self, EPOCHS, REMAKES, VOTES};
@@ -36,6 +44,10 @@ pub(crate) struct ReplayStats {
     pub(crate) failed: usize,
     /// Paths the remake declared obsolete.
     pub(crate) dropped: usize,
+    /// Paths not replayed because a newer write touched them ([`replay_newest`]), and
+    /// left out (`Superseded::Drop`; with `Superseded::Quarantine`, they are among
+    /// `quarantined`).
+    pub(crate) superseded: usize,
     /// Paths the remake had no place for, with the reason.
     pub(crate) quarantined: Vec<(Vec<String>, String)>,
 }
@@ -46,12 +58,144 @@ impl ReplayStats {
         self.replayed += other.replayed;
         self.failed += other.failed;
         self.dropped += other.dropped;
+        self.superseded += other.superseded;
         self.quarantined.extend(other.quarantined);
     }
 }
 
+/// What a replay does with a write that a newer one superseded (`--superseded-writes`):
+/// leave it out, logged (last writer wins), or put it in the quarantine, where an
+/// operator can still replay it.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum Superseded {
+    #[default]
+    Drop,
+    Quarantine,
+}
+
+/// Replays into `dst` what changed in `src` between `before` and `after` (both in
+/// `src`), last writer wins: a path whose newest write there (by its change's
+/// timestamp) is older than a change in `dst` touching it, or a path above or below it,
+/// stays as `dst` has it (`superseded` says what becomes of the write). Ties go to the
+/// replay. The replayed paths are committed grouped by the time of the writes they
+/// replay, each change stamped with it.
+pub(crate) fn replay_newest(
+    src: &Document,
+    before: &[ChangeHash],
+    after: &[ChangeHash],
+    dst: &mut Document,
+    translator: &Translator,
+    superseded: Superseded,
+) -> ReplayStats {
+    let mut stats = ReplayStats::default();
+    if before == after {
+        return stats;
+    }
+    let paths = changed_paths(src, before, after);
+    let written = touched(src, &between(src, before, after));
+    let ours: Vec<i64> = paths
+        .iter()
+        .map(|path| newest(&written, path).unwrap_or(0))
+        .collect();
+    // Only changes newer than the oldest write replayed can supersede one; the
+    // snapshot (time 0) never does.
+    let oldest = ours.iter().copied().min().unwrap_or(0);
+    let newer: Vec<ChangeHash> = dst
+        .get_changes(&[])
+        .iter()
+        .filter(|change| change.timestamp() > oldest)
+        .map(|change| change.hash())
+        .collect();
+    let theirs = touched(dst, &newer);
+
+    let mut by_time: BTreeMap<i64, Vec<Vec<String>>> = BTreeMap::new();
+    for (path, at) in paths.into_iter().zip(ours) {
+        match newest(&theirs, &path) {
+            Some(newer) if newer > at => {
+                stats.superseded += 1;
+                if superseded == Superseded::Quarantine {
+                    stats.quarantined.push((
+                        path,
+                        format!("a newer write (at {newer}, this one at {at}) superseded it"),
+                    ));
+                }
+            }
+            _ => by_time.entry(at).or_default().push(path),
+        }
+    }
+    for (at, paths) in by_time {
+        let mut transaction = dst.transaction();
+        stats.absorb(replay_paths(src, &mut transaction, &paths, translator));
+        transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
+    }
+    if stats.superseded > 0 {
+        info!(
+            superseded = stats.superseded,
+            quarantined = superseded == Superseded::Quarantine,
+            "Writes not replayed: newer ones touched their paths"
+        );
+    }
+    if superseded == Superseded::Quarantine {
+        // Counted as quarantined, not as left out.
+        stats.superseded = 0;
+    }
+    stats
+}
+
+/// The changes of `doc` after `before` up to `after`: those `after` builds on that
+/// `before` doesn't.
+fn between(doc: &Document, before: &[ChangeHash], after: &[ChangeHash]) -> Vec<ChangeHash> {
+    let deps: HashMap<ChangeHash, Vec<ChangeHash>> = doc
+        .get_changes(before)
+        .iter()
+        .map(|change| (change.hash(), change.deps().to_vec()))
+        .collect();
+    let mut found = HashSet::new();
+    let mut todo: Vec<ChangeHash> = after.to_vec();
+    while let Some(hash) = todo.pop() {
+        if let Some(parents) = deps.get(&hash)
+            && found.insert(hash)
+        {
+            todo.extend(parents.iter().copied());
+        }
+    }
+    found.into_iter().collect()
+}
+
+/// For each of `changes` (in `doc`), its time and the store paths it touched.
+fn touched(doc: &Document, changes: &[ChangeHash]) -> Vec<(i64, Vec<Vec<String>>)> {
+    changes
+        .iter()
+        .filter_map(|hash| {
+            let change = doc.get_change_by_hash(hash)?;
+            let paths = store::changed(doc, change.deps(), &[*hash], InLists::Whole)
+                .into_iter()
+                .map(|changed| changed.path)
+                .collect();
+            Some((change.timestamp(), paths))
+        })
+        .collect()
+}
+
+/// The time of the newest of `changes` that touched `path`, a path above it or one
+/// below it.
+fn newest(changes: &[(i64, Vec<Vec<String>>)], path: &[String]) -> Option<i64> {
+    changes
+        .iter()
+        .filter(|(_, paths)| {
+            paths
+                .iter()
+                .any(|other| other.starts_with(path) || path.starts_with(other))
+        })
+        .map(|(at, _)| *at)
+        .max()
+}
+
 /// Replays into `dst` everything that changed in `src` between `before` and `after`,
-/// as one commit. Both head sets must be in `src`.
+/// as one commit, whatever `dst` has (no last writer wins): what a replay copies, for
+/// tests. Both head sets must be in `src`.
+#[cfg(test)]
 pub(crate) fn replay(
     src: &Document,
     before: &[ChangeHash],
@@ -423,6 +567,141 @@ mod tests {
         assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]);
     }
 
+    /// Writes to `doc` as one change made at `at` (Unix ms).
+    fn write_at(
+        doc: &mut Document,
+        at: i64,
+        f: impl FnOnce(&mut automerge::transaction::Transaction<'_>),
+    ) {
+        let mut transaction = doc.transaction();
+        f(&mut transaction);
+        transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(at));
+    }
+
+    /// An old document with `x`, `m/a` and `m/b` written at 10 after the base, and a new
+    /// one holding the base; `newer` writes to the new one first.
+    fn superseding(
+        newer: impl FnOnce(&mut Document),
+        superseded: Superseded,
+    ) -> (Document, ReplayStats) {
+        let mut old = Document::new();
+        write_at(&mut old, 1, |old| {
+            let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
+            old.put(&m, "a", "base").unwrap();
+        });
+        let base = old.get_heads();
+        let mut new = Document::new();
+        write_at(&mut new, 0, |new| {
+            store::materialize_into(&old, new, &|_| false).unwrap();
+        });
+        newer(&mut new);
+        write_at(&mut old, 10, |old| {
+            old.put(ROOT, "x", "old").unwrap();
+            let (_, m) = old.get(ROOT, "m").unwrap().unwrap();
+            old.put(&m, "a", "old").unwrap();
+            old.put(&m, "b", "old").unwrap();
+        });
+        let after = old.get_heads();
+        let stats = replay_newest(
+            &old,
+            &base,
+            &after,
+            &mut new,
+            &Translator::default(),
+            superseded,
+        );
+        (new, stats)
+    }
+
+    #[test]
+    fn replay_newest_keeps_newer_writes() {
+        // Nothing newer: everything replayed, at the time it was written.
+        let (new, stats) = superseding(|_| {}, Superseded::Drop);
+        assert_eq!((stats.replayed, stats.superseded), (3, 0));
+        assert_eq!(
+            view(&new),
+            json!({ "x": "old", "m": { "a": "old", "b": "old" } })
+        );
+        let times: Vec<i64> = new.get_changes(&[]).iter().map(|c| c.timestamp()).collect();
+        assert_eq!(times, [0, 10]);
+
+        // A newer write at the path wins; an older one doesn't.
+        let (new, stats) = superseding(
+            |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()),
+            Superseded::Drop,
+        );
+        assert_eq!((stats.replayed, stats.superseded), (2, 1));
+        assert_eq!(view(&new)["x"], json!("newer"));
+        assert_eq!(view(&new)["m"], json!({ "a": "old", "b": "old" }));
+        let (new, _) = superseding(
+            |new| write_at(new, 5, |new| new.put(ROOT, "x", "older").unwrap()),
+            Superseded::Drop,
+        );
+        assert_eq!(view(&new)["x"], json!("old"));
+
+        // Per key: a newer `m/a` wins over the replayed `m/a`, not over `m/b`.
+        let (new, stats) = superseding(
+            |new| {
+                write_at(new, 20, |new| {
+                    let (_, m) = new.get(ROOT, "m").unwrap().unwrap();
+                    new.put(&m, "a", "newer").unwrap();
+                })
+            },
+            Superseded::Drop,
+        );
+        assert_eq!(stats.superseded, 1);
+        assert_eq!(view(&new)["m"], json!({ "a": "newer", "b": "old" }));
+        // A newer write above the paths (`m` replaced) wins over both.
+        let (new, stats) = superseding(
+            |new| write_at(new, 20, |new| new.put(ROOT, "m", "a scalar now").unwrap()),
+            Superseded::Drop,
+        );
+        assert_eq!(stats.superseded, 2);
+        assert_eq!(view(&new)["m"], json!("a scalar now"));
+
+        // Quarantined instead, with the time of each.
+        let (_, stats) = superseding(
+            |new| write_at(new, 20, |new| new.put(ROOT, "x", "newer").unwrap()),
+            Superseded::Quarantine,
+        );
+        assert_eq!(stats.superseded, 0);
+        assert_eq!(stats.quarantined.len(), 1);
+        assert_eq!(stats.quarantined[0].0, ["x"]);
+    }
+
+    /// A newer write below a replayed path wins too: the replay would replace it.
+    #[test]
+    fn replay_newest_keeps_newer_writes_below() {
+        let mut old = Document::new();
+        write_at(&mut old, 1, |old| old.put(ROOT, "base", true).unwrap());
+        let base = old.get_heads();
+        let mut new = Document::new();
+        write_at(&mut new, 0, |new| {
+            store::materialize_into(&old, new, &|_| false).unwrap();
+            new.put_object(ROOT, "m", ObjType::Map).unwrap();
+        });
+        write_at(&mut new, 20, |new| {
+            let (_, m) = new.get(ROOT, "m").unwrap().unwrap();
+            new.put(&m, "a", "newer").unwrap();
+        });
+        // The whole map, older.
+        write_at(&mut old, 10, |old| {
+            let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
+            old.put(&m, "a", "old").unwrap();
+        });
+        let after = old.get_heads();
+        let stats = replay_newest(
+            &old,
+            &base,
+            &after,
+            &mut new,
+            &Translator::default(),
+            Superseded::Drop,
+        );
+        assert_eq!(stats.superseded, 1);
+        assert_eq!(view(&new)["m"], json!({ "a": "newer" }));
+    }
+
     #[test]
     fn replays_state_per_key() {
         // Base shared by the old document and the snapshot.
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 01e1e3e..0d3d073 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -37,6 +37,7 @@ mod vote;
 
 #[cfg(test)]
 pub(crate) use api::testing;
+pub(crate) use delta::Superseded;
 pub(crate) use kind::remake_genesis;
 pub(crate) use local::RemakeState;
 pub(crate) use switch::run_worker;
@@ -193,6 +194,17 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
         T: Send + 'static,
         E: From<StoreError> + Send + 'static;
 
+    /// What a replay does with writes a newer one superseded (`--superseded-writes`).
+    fn superseded(&self) -> Superseded;
+
+    /// Changes `f` commits itself into the live document (as many as it likes, each
+    /// with its own time), under the write lock, then published and replicated.
+    /// Refused while the document isn't writable, as [`Host::commit_doc`].
+    fn commit_changes<T, F>(&self, f: F) -> impl Future<Output = Result<T, StoreError>> + Send
+    where
+        F: FnOnce(&mut Document) -> T + Send + 'static,
+        T: Send + 'static;
+
     /// Lets writes in again, after a rejoin swapped in a document that took none
     /// ([`Host::switch_epoch`] with `writable` false).
     fn accept_writes(&self) -> impl Future<Output = ()> + Send;
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,
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index aa30e7c..81bddb5 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -114,6 +114,7 @@ impl Sim {
             alias: Some(id.to_owned()),
             electorate_window: Duration::from_secs(30 * 24 * 3600),
             clock_skew_warn: Duration::from_secs(2),
+            superseded_writes: Default::default(),
             blob_chunk_bytes: 64 * 1024,
             swarm_key: None,
             protocol_writers: HashSet::new(),
diff --git a/src/state/host.rs b/src/state/host.rs
index 30ebcdb..eb7db35 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -226,6 +226,35 @@ impl Host for AppState {
         async move { state.mutate_doc(f).await }
     }
 
+    fn superseded(&self) -> crate::remake::Superseded {
+        self.superseded_writes
+    }
+
+    fn commit_changes<T, F>(&self, f: F) -> impl Future<Output = Result<T, StoreError>> + Send
+    where
+        F: FnOnce(&mut Document) -> T + Send + 'static,
+        T: Send + 'static,
+    {
+        let state = self.clone();
+        async move {
+            tokio::task::spawn_blocking(move || {
+                let result = {
+                    let mut replica = state.doc.blocking_write();
+                    if !replica.writable {
+                        return Err(StoreError::Rejoining);
+                    }
+                    let result = f(&mut replica.doc);
+                    replica.refresh_heads();
+                    result
+                };
+                state.publish_change();
+                Ok(result)
+            })
+            .await
+            .expect("document write task panicked")
+        }
+    }
+
     fn refuse_writes(&self, copy: bool) -> impl Future<Output = Option<Document>> + Send {
         let state = self.clone();
         async move {
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 5e4a4d3..6e6e48c 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -112,6 +112,8 @@ pub(crate) struct AppState {
     /// This node's clock is off when it differs from most of its peers' by more than
     /// this (`--clock-skew-warn`, `swarm::skew`); 0: never.
     pub(crate) clock_skew_warn: Duration,
+    /// What a replay does with writes a newer one superseded (`--superseded-writes`).
+    pub(crate) superseded_writes: crate::remake::Superseded,
     /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as
     /// last measured; 0 before. The propagation figures take it off.
     pub(crate) clock_ahead_ms: Arc<std::sync::atomic::AtomicI64>,