1 files changed, 160 insertions, 0 deletions
diff --git a/src/compaction/quarantine.rs b/src/compaction/quarantine.rs
new file mode 100644
index 0000000..da39ec0
--- /dev/null
+++ b/src/compaction/quarantine.rs
@@ -0,0 +1,160 @@
+//! `/admin/quarantine`: this node's own changes that were not replayed when it rejoined
+//! a newer epoch (outside the replay window, or impossible to tell apart).
+//!
+//! * `GET /admin/quarantine` lists the entries.
+//! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value
+//!   the old document's merged view holds there (`deleted` if none).
+//! * `GET /admin/quarantine/{id}/document` downloads the old document
+//!   (`AutoCommit::save()`), for inspection with other tools.
+//! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as
+//!   fresh writes, so they win over what is there) and removes the entry.
+//! * `DELETE /admin/quarantine/{id}` discards the entry.
+//!
+//! Quarantine lives in memory on the node that holds it.
+
+use std::convert::Infallible;
+
+use axum::{
+    extract::{Path, State},
+    http::{header, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::Serialize;
+use serde_json::{json, Value as JsonValue};
+use tracing::info;
+
+use super::api::error_response;
+use super::delta;
+use super::local::Quarantined;
+use super::Host;
+use crate::store;
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct Summary {
+    id: String,
+    /// The epoch the changes were made in.
+    epoch: String,
+    /// The epoch the node rejoined.
+    target: String,
+    created_at: u64,
+    reason: String,
+    paths: usize,
+}
+
+fn summary(entry: &Quarantined) -> Summary {
+    Summary {
+        id: entry.id.clone(),
+        epoch: entry.epoch.to_string(),
+        target: entry.target.to_string(),
+        created_at: entry.created_at,
+        reason: entry.reason.clone(),
+        paths: entry.paths.len(),
+    }
+}
+
+fn not_found() -> Response {
+    error_response(StatusCode::NOT_FOUND, "no such quarantine entry", None)
+}
+
+pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
+    Json(host.compaction().list_quarantine(summary)).into_response()
+}
+
+pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    let Some((summary, document, paths)) = host.compaction().with_quarantined(&id, |entry| {
+        (summary(entry), entry.document.clone(), entry.paths.clone())
+    }) else {
+        return not_found();
+    };
+
+    let writes = tokio::task::spawn_blocking(move || {
+        paths
+            .into_iter()
+            .map(|path| match store::read_path(&document, &path) {
+                Ok(Some(value)) => json!({ "path": path, "value": value }),
+                Ok(None) => json!({ "path": path, "deleted": true }),
+                Err(error) => json!({ "path": path, "error": error.to_string() }),
+            })
+            .collect::<Vec<JsonValue>>()
+    })
+    .await;
+
+    match writes {
+        Ok(writes) => Json(json!({ "entry": summary, "writes": writes })).into_response(),
+        Err(_) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "reading the writes failed",
+            None,
+        ),
+    }
+}
+
+pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    let Some(mut document) = host
+        .compaction()
+        .with_quarantined(&id, |entry| entry.document.clone())
+    else {
+        return not_found();
+    };
+
+    match tokio::task::spawn_blocking(move || document.save()).await {
+        Ok(bytes) => (
+            [
+                (header::CONTENT_TYPE, "application/octet-stream".to_owned()),
+                (
+                    header::CONTENT_DISPOSITION,
+                    format!("attachment; filename=\"quarantine-{id}.automerge\""),
+                ),
+            ],
+            bytes,
+        )
+            .into_response(),
+        Err(_) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "saving the document failed",
+            None,
+        ),
+    }
+}
+
+pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    if !host.is_node_ready().await {
+        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
+    }
+    let Some(entry) = host.compaction().take_quarantined(&id) else {
+        return not_found();
+    };
+
+    let (stats, entry) = host
+        .commit_doc(move |doc| {
+            let stats = delta::replay_paths(&entry.document, doc, &entry.paths);
+            Ok::<_, Infallible>((stats, entry))
+        })
+        .await
+        .unwrap_or_else(|never| match never {});
+
+    if stats.failed > 0 && stats.replayed == 0 {
+        // Nothing went in; keep the entry for another look.
+        host.compaction().restore_quarantined(entry);
+        return error_response(
+            StatusCode::CONFLICT,
+            "none of the writes could be replayed",
+            Some(json!({ "failed": stats.failed })),
+        );
+    }
+
+    info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, "Replayed quarantined changes");
+    Json(json!({ "replayed": stats.replayed, "failed": stats.failed })).into_response()
+}
+
+pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    match host.compaction().take_quarantined(&id) {
+        Some(_) => {
+            info!(quarantine = %id, "Discarded quarantined changes");
+            StatusCode::NO_CONTENT.into_response()
+        }
+        None => not_found(),
+    }
+}