turn compaction into a kind of remake

1 files changed, 67 insertions, 30 deletions

diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 94a0df7..4eb72b0 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -27,7 +27,9 @@ use tracing::{error, info, warn};
 use super::Host;
 use super::catchup;
 use super::delta::{self, ReplayStats};
+use super::kind::Translator;
 use super::local::LocalStatus;
+use super::quarantine;
 use super::records::{self, EpochRecord, Record, unix_millis};
 use super::snapshot::{self, META_ROOT, REMAKES};
 use crate::document::replica::Epoch;
@@ -115,14 +117,14 @@ async fn check<H: Host>(host: &H) {
     }
 
     match switch(host, &id, &record).await {
-        Ok(stats) => {
+        Ok((stats, quarantine)) => {
             host.remake().set_status(
                 &id,
                 LocalStatus {
                     switched_at: Some(unix_millis()),
                     replayed_paths: Some(stats.replayed),
                     failed_paths: Some(stats.failed),
-                    quarantine: None,
+                    quarantine,
                     error: None,
                 },
             );
@@ -140,11 +142,21 @@ async fn check<H: Host>(host: &H) {
     }
 }
 
-async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> {
+/// Returns the replay numbers and the quarantine entry for writes the remake had no
+/// place for, if any.
+async fn switch<H: Host>(
+    host: &H,
+    id: &str,
+    record: &Record,
+) -> Result<(ReplayStats, Option<String>), String> {
     let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
     else {
         return Err("unreadable remake record".to_owned());
     };
+    let translator = request
+        .translator()
+        .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
+    let late_translator = translator.clone();
     let previous = request.previous;
     let base = request.base_heads.clone();
     let retention = record.proposal.policy.archive_retention.clone();
@@ -165,7 +177,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
         .filter(|snapshot| snapshot.epoch == expected);
 
     // 2. Build (or reuse), verify and replay without holding any lock.
-    let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking(
+    let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
         move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
             let mut copy = copy;
             let snapshot = match prebuilt {
@@ -191,7 +203,7 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
             }
 
             let heads = copy.get_heads();
-            let stats = delta::replay(&mut copy, &base, &heads, &mut doc);
+            let stats = delta::replay(&mut copy, &base, &heads, &mut doc, &translator);
             Ok((doc, heads, stats))
         },
     )
@@ -199,9 +211,18 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
     .map_err(|error| format!("switch task failed: {error}"))??;
 
     // 3. Swap, replaying what arrived since the copy.
-    let old = swap_in(host, expected, new_doc, copy_heads).await?;
+    let (old, late) = swap_in(host, expected, new_doc, copy_heads, late_translator).await?;
+    stats.absorb(late);
 
-    // 4. Keep the old document as configured.
+    // 4. Writes the remake had no place for wait for an operator (the old document
+    //    holds them); the old document is kept as configured either way.
+    let quarantine = quarantine::add(
+        host,
+        previous,
+        expected,
+        &old,
+        std::mem::take(&mut stats.quarantined),
+    );
     host.remake()
         .archive(previous, old, retention, expected.generation);
 
@@ -210,9 +231,11 @@ async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplaySt
         epoch = %expected,
         replayed = stats.replayed,
         failed = stats.failed,
+        dropped = stats.dropped,
+        quarantine = ?quarantine,
         "Switched to the new epoch"
     );
-    Ok(stats)
+    Ok((stats, quarantine))
 }
 
 /// Under the write lock: replays into `new_doc` whatever the live document received
@@ -223,26 +246,40 @@ pub(super) async fn swap_in<H: Host>(
     target: Epoch,
     new_doc: AutoCommit,
     copy_heads: Vec<ChangeHash>,
-) -> Result<AutoCommit, String> {
-    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())
-        {
-            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);
-        if late.replayed + late.failed > 0 {
-            info!(
-                replayed = late.replayed,
-                failed = late.failed,
-                "Replayed late changes"
-            );
-        }
-        Ok(new_doc)
-    })
-    .await
+    translator: Translator,
+) -> Result<(AutoCommit, ReplayStats), String> {
+    // 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())
+            {
+                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);
+            if late.replayed + late.failed > 0 {
+                info!(
+                    replayed = late.replayed,
+                    failed = late.failed,
+                    "Replayed late changes"
+                );
+            }
+            *late_out
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner()) = late;
+            Ok(new_doc)
+        })
+        .await?;
+    let late = std::mem::take(
+        &mut *late_stats
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()),
+    );
+    Ok((old, late))
 }