1 files changed, 94 insertions, 92 deletions

diff --git a/src/compaction/local.rs b/src/compaction/local.rs
index adf8764..c798a4c 100644
--- a/src/compaction/local.rs
+++ b/src/compaction/local.rs
@@ -1,30 +1,25 @@
 //! Per-node, in-memory compaction state: snapshots built ahead of the switch, switch
-//! progress for reports, and archived documents of previous epochs.
+//! progress for reports, a rejoin in progress, quarantined changes, and archived
+//! documents of previous epochs.
 //!
 //! Nothing here is replicated or durable. After a restart the node simply has no
-//! archive and rebuilds snapshots when needed.
+//! archive or quarantine, and rebuilds snapshots when needed.
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex, MutexGuard};
-use std::time::Duration;
+use std::sync::{Mutex, MutexGuard};
 
 use automerge::AutoCommit;
 use serde::Serialize;
-use tokio::sync::oneshot;
 use tokio::time::Instant;
 
 use super::records::{unix_millis, Retention};
 use super::snapshot::Snapshot;
 use crate::replica::Epoch;
-use crate::transport::SnapshotReply;
 
 #[derive(Default)]
 pub(crate) struct Compaction {
     // A std mutex: every access is short and never spans an await.
     inner: Mutex<Inner>,
-    /// Set while this node saves its document for a peer; one at a time.
-    serving: AtomicBool,
 }
 
 #[derive(Default)]
@@ -35,25 +30,14 @@ struct Inner {
     first_seen: HashMap<String, Instant>,
     status: HashMap<String, LocalStatus>,
     archive: Vec<Archived>,
-    /// The chunk this node is waiting for: (peer, wanted epoch, offset).
-    awaiting: Option<(String, Epoch, u64, oneshot::Sender<SnapshotReply>)>,
+    /// Last attempt to catch up with a newer epoch.
     catch_up: Option<CatchUp>,
-    /// The saved document chunks are served from (one download source at a time).
-    served: Option<Arc<Served>>,
+    /// A rejoin in progress: the node syncs an empty document of a newer epoch.
+    rejoin: Option<Rejoin>,
+    quarantine: Vec<Quarantined>,
 }
 
-/// A saved copy of this node's document, served to downloading peers.
-pub(crate) struct Served {
-    pub(crate) epoch: Epoch,
-    pub(crate) bytes: Vec<u8>,
-    pub(crate) digest: [u8; 32],
-    pub(crate) saved_at: Instant,
-}
-
-/// A served copy is dropped this long after it was saved.
-const SERVED_KEPT_FOR: Duration = Duration::from_secs(15 * 60);
-
-/// Fetching a newer epoch from a peer.
+/// Catching up with a newer epoch: attempts and the last failure.
 #[derive(Clone, Debug)]
 pub(crate) struct CatchUp {
     pub(crate) target: Epoch,
@@ -62,11 +46,38 @@ pub(crate) struct CatchUp {
     pub(crate) error: Option<String>,
 }
 
+/// The node left `previous` and is syncing `target` from scratch.
+pub(crate) struct Rejoin {
+    pub(crate) previous: Epoch,
+    pub(crate) target: Epoch,
+    /// The document of `previous`, until it is replayed, quarantined or restored.
+    pub(crate) old: AutoCommit,
+    pub(crate) started: Instant,
+    /// The node served requests before; it does again once the rejoin is over.
+    pub(crate) was_ready: bool,
+}
+
+/// Changes of this node that were not replayed into the current epoch.
+pub(crate) struct Quarantined {
+    pub(crate) id: String,
+    /// The epoch the changes were made in, and the one the node rejoined.
+    pub(crate) epoch: Epoch,
+    pub(crate) target: Epoch,
+    pub(crate) created_at: u64,
+    pub(crate) reason: String,
+    /// Store paths the changes touched (minimal, prefix-free).
+    pub(crate) paths: Vec<Vec<String>>,
+    /// The old document; `paths` are read from its merged view.
+    pub(crate) document: AutoCommit,
+}
+
 #[derive(Clone, Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct CatchUpReport {
     pub(crate) target: String,
     pub(crate) attempts: u32,
+    /// True while the node syncs the target epoch (not serving requests).
+    pub(crate) rejoining: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) error: Option<String>,
 }
@@ -81,6 +92,9 @@ pub(crate) struct LocalStatus {
     pub(crate) replayed_paths: Option<usize>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) failed_paths: Option<usize>,
+    /// ID of the quarantine entry holding this node's changes that weren't replayed.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) quarantine: Option<String>,
     /// Why the switch did not happen (yet).
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) error: Option<String>,
@@ -88,7 +102,7 @@ pub(crate) struct LocalStatus {
 
 struct Archived {
     epoch: Epoch,
-    // Kept for later steps (serving or auditing old epochs); not read yet.
+    // Kept for auditing and recovery by hand; not read by the protocol.
     #[allow(dead_code)]
     doc: AutoCommit,
     archived_at: u64,
@@ -154,88 +168,79 @@ impl Compaction {
         prune(&mut inner.archive, current_generation);
     }
 
-    /// Drops archives past their retention and a served copy nobody needs any more.
     pub(crate) fn prune_archive(&self, current_generation: u64) {
-        let mut inner = self.lock();
-        prune(&mut inner.archive, current_generation);
-        if inner
-            .served
-            .as_ref()
-            .is_some_and(|served| served.saved_at.elapsed() > SERVED_KEPT_FOR)
-        {
-            inner.served = None;
-        }
+        prune(&mut self.lock().archive, current_generation);
     }
 
-    /// Registers interest in one chunk from `peer`, replacing any earlier request.
-    pub(crate) fn await_reply(
-        &self,
-        peer: &str,
-        wanted: Epoch,
-        offset: u64,
-    ) -> oneshot::Receiver<SnapshotReply> {
-        let (sender, receiver) = oneshot::channel();
-        self.lock().awaiting = Some((peer.to_owned(), wanted, offset, sender));
-        receiver
+    pub(crate) fn catch_up(&self) -> Option<CatchUp> {
+        self.lock().catch_up.clone()
     }
 
-    /// Hands a reply to the waiting download. False unless it answers exactly the
-    /// pending request (late replies to earlier attempts are dropped).
-    pub(crate) fn deliver_reply(&self, peer: &str, reply: SnapshotReply) -> bool {
-        let mut inner = self.lock();
-        match inner.awaiting.take() {
-            Some((expected, wanted, offset, sender))
-                if expected == peer && wanted == reply.wanted && offset == reply.offset =>
-            {
-                sender.send(reply).is_ok()
-            }
-            other => {
-                inner.awaiting = other;
-                false
-            }
-        }
+    pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) {
+        self.lock().catch_up = catch_up;
+    }
+
+    pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> {
+        let inner = self.lock();
+        let rejoining = inner.rejoin.is_some();
+        inner.catch_up.as_ref().map(|catch_up| CatchUpReport {
+            target: catch_up.target.to_string(),
+            attempts: catch_up.attempts,
+            rejoining,
+            error: catch_up.error.clone(),
+        })
     }
 
-    /// The copy served for `epoch`, if any.
-    pub(crate) fn served(&self, epoch: Epoch) -> Option<Arc<Served>> {
+    pub(crate) fn is_rejoining(&self) -> bool {
+        self.lock().rejoin.is_some()
+    }
+
+    /// Progress of the rejoin in progress: (target, started).
+    pub(crate) fn rejoin_progress(&self) -> Option<(Epoch, Instant)> {
         self.lock()
-            .served
+            .rejoin
             .as_ref()
-            .filter(|served| served.epoch == epoch)
-            .cloned()
+            .map(|rejoin| (rejoin.target, rejoin.started))
     }
 
-    pub(crate) fn set_served(&self, served: Arc<Served>) {
-        self.lock().served = Some(served);
+    pub(crate) fn start_rejoin(&self, rejoin: Rejoin) {
+        self.lock().rejoin = Some(rejoin);
     }
 
-    pub(crate) fn stop_awaiting(&self) {
-        self.lock().awaiting = None;
+    pub(crate) fn take_rejoin(&self) -> Option<Rejoin> {
+        self.lock().rejoin.take()
     }
 
-    /// Claims the right to serve one snapshot; release with [`Compaction::done_serving`].
-    pub(crate) fn try_start_serving(&self) -> bool {
-        !self.serving.swap(true, Ordering::AcqRel)
+    pub(crate) fn add_quarantine(&self, quarantined: Quarantined) {
+        self.lock().quarantine.push(quarantined);
     }
 
-    pub(crate) fn done_serving(&self) {
-        self.serving.store(false, Ordering::Release);
+    /// Runs `f` on the quarantined entry with this ID.
+    pub(crate) fn with_quarantined<T>(
+        &self,
+        id: &str,
+        f: impl FnOnce(&Quarantined) -> T,
+    ) -> Option<T> {
+        self.lock()
+            .quarantine
+            .iter()
+            .find(|entry| entry.id == id)
+            .map(f)
     }
 
-    pub(crate) fn catch_up(&self) -> Option<CatchUp> {
-        self.lock().catch_up.clone()
+    pub(crate) fn list_quarantine<T>(&self, f: impl Fn(&Quarantined) -> T) -> Vec<T> {
+        self.lock().quarantine.iter().map(f).collect()
     }
 
-    pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) {
-        self.lock().catch_up = catch_up;
+    pub(crate) fn take_quarantined(&self, id: &str) -> Option<Quarantined> {
+        let mut inner = self.lock();
+        let index = inner.quarantine.iter().position(|entry| entry.id == id)?;
+        Some(inner.quarantine.remove(index))
     }
 
-    pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> {
-        self.lock().catch_up.as_ref().map(|catch_up| CatchUpReport {
-            target: catch_up.target.to_string(),
-            attempts: catch_up.attempts,
-            error: catch_up.error.clone(),
-        })
+    /// Puts an entry back, e.g. after a failed replay.
+    pub(crate) fn restore_quarantined(&self, quarantined: Quarantined) {
+        self.lock().quarantine.push(quarantined);
     }
 
     pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> {
@@ -255,13 +260,10 @@ impl Compaction {
 fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
     let now = unix_millis();
     archive.retain(|archived| {
-        let within_epochs = archived.retention.epochs.is_some_and(|epochs| {
-            current_generation.saturating_sub(archived.epoch.generation) <= epochs
-        });
-        let within_time = archived.retention.seconds.is_some_and(|seconds| {
-            now.saturating_sub(archived.archived_at) <= seconds.saturating_mul(1000)
-        });
-        within_epochs || within_time
+        archived.retention.covers(
+            current_generation.saturating_sub(archived.epoch.generation),
+            now.saturating_sub(archived.archived_at),
+        )
     });
 }