fix compaction

- introduce DELETE /admin/compaction to abort local compaction proposal
- continue coordination if a POST /admin/compaction?wait=true terminates

2 files changed, 97 insertions, 26 deletions

diff --git a/src/compaction/api.rs b/src/compaction/api.rs
index b8c55db..a2fde13 100644
--- a/src/compaction/api.rs
+++ b/src/compaction/api.rs
@@ -9,6 +9,8 @@
 //! * `GET /admin/epochs` shows the local epoch and the archived previous ones.
 //! * `DELETE /admin/compactions/{id}` asks the coordinator to abort. Only possible while
 //!   proposed; `409` once committed (a switch can't be undone) or already over.
+//!   `DELETE /admin/compaction` does the same for the compaction of the local epoch that
+//!   is in progress, without having to know its ID.
 //!
 //! Request body for POST (all optional):
 //! `{ "replayWindow": {"epochs": 1, "seconds": 604800},
@@ -205,29 +207,21 @@ pub(crate) async fn start<H: Host>(
         peers_considered,
     };
 
-    let record_id = id.clone();
-    let written = host
-        .commit_doc(move |doc| -> Result<Record, StartError> {
-            if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
-                record.proposal.from_epoch == proposal.from_epoch
-                    && record.phase(now, grace).is_open()
-            }) {
-                return Err(StartError::Open(open));
-            }
-
-            // The base is everything before the proposal; the proposal itself is not in it.
-            let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect();
-            heads.sort();
-            proposal.base_heads = heads;
-
-            records::write_proposal(doc, &record_id, &proposal)?;
-            Ok(Record {
-                proposal,
-                decision: None,
-                abort_requested: None,
-            })
-        })
-        .await;
+    // From here on, nothing depends on the client: writing the proposal and starting the
+    // coordinator run in their own task, so a client that disconnects mid-request can't
+    // leave a proposal without a coordinator. Dropping the handle doesn't cancel it.
+    let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace));
+    let written = match proposing.await {
+        Ok(written) => written,
+        Err(error) => {
+            error!(error = %error, "Proposal task failed");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
 
     let record = match written {
         Ok(record) => record,
@@ -253,7 +247,6 @@ pub(crate) async fn start<H: Host>(
         settle_secs = settle.as_secs(),
         "Compaction proposed"
     );
-    tokio::spawn(coordinator::run(host.clone(), id.clone()));
 
     if query.wait {
         return match wait_until_final(&host, &id).await {
@@ -271,6 +264,43 @@ pub(crate) async fn start<H: Host>(
         .into_response()
 }
 
+/// Writes the proposal (unless another one is open for the epoch) and starts its
+/// coordinator. Runs as its own task, independent of the HTTP request.
+async fn propose<H: Host>(
+    host: H,
+    id: String,
+    mut proposal: Proposal,
+    now: u64,
+    grace: u64,
+) -> Result<Record, StartError> {
+    let record_id = id.clone();
+    let record = host
+        .commit_doc(move |doc| -> Result<Record, StartError> {
+            if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
+                record.proposal.from_epoch == proposal.from_epoch
+                    && record.phase(now, grace).is_open()
+            }) {
+                return Err(StartError::Open(open));
+            }
+
+            // The base is everything before the proposal; the proposal itself is not in it.
+            let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect();
+            heads.sort();
+            proposal.base_heads = heads;
+
+            records::write_proposal(doc, &record_id, &proposal)?;
+            Ok(Record {
+                proposal,
+                decision: None,
+                abort_requested: None,
+            })
+        })
+        .await?;
+
+    tokio::spawn(coordinator::run(host, id));
+    Ok(record)
+}
+
 pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
     let all = host.read_doc(records::load_all).await;
     let mut reports: Vec<Report> = all
@@ -297,11 +327,49 @@ pub(crate) async fn get<H: Host>(
     }
 }
 
+/// `DELETE /admin/compactions/{id}`.
 pub(crate) async fn abort<H: Host>(
     State(host): State<H>,
     Path(id): Path<String>,
     Query(query): Query<WaitQuery>,
 ) -> Response {
+    abort_by_id(host, id, query.wait).await
+}
+
+/// `DELETE /admin/compaction`: aborts whichever compaction of the local epoch is still
+/// proposed. There is at most one, unless two coordinators raced (then both abort anyway
+/// and this picks the oldest).
+pub(crate) async fn abort_current<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let local = EpochRecord::from(host.current_epoch());
+    let now = unix_millis();
+    let grace = grace(&host);
+    let open = host
+        .read_doc(move |doc| {
+            records::load_all(doc)
+                .into_iter()
+                .filter(|(_, record)| {
+                    record.proposal.from_epoch == local
+                        && record.phase(now, grace) == Phase::Proposed
+                })
+                .min_by_key(|(_, record)| record.proposal.proposed_at)
+                .map(|(id, _)| id)
+        })
+        .await;
+
+    match open {
+        Some(id) => abort_by_id(host, id, query.wait).await,
+        None => error_response(
+            StatusCode::NOT_FOUND,
+            "no compaction of the local epoch is in progress",
+            None,
+        ),
+    }
+}
+
+async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
     let Some(current) = load_report(&host, &id).await else {
         return error_response(StatusCode::NOT_FOUND, "no such compaction", None);
     };
@@ -346,7 +414,7 @@ pub(crate) async fn abort<H: Host>(
         info!(compaction = %id, "Abort requested");
     }
 
-    let report = if query.wait {
+    let report = if wait {
         wait_until_final(&host, &id).await
     } else {
         load_report(&host, &id).await
diff --git a/src/compaction/mod.rs b/src/compaction/mod.rs
index 6cb1a98..69c633e 100644
--- a/src/compaction/mod.rs
+++ b/src/compaction/mod.rs
@@ -102,7 +102,10 @@ pub(crate) struct PeerSummary {
 /// Admin routes. Merged into the authenticated part of the node's router.
 pub(crate) fn router<H: Host>() -> Router<H> {
     Router::new()
-        .route("/admin/compaction", post(api::start::<H>))
+        .route(
+            "/admin/compaction",
+            post(api::start::<H>).delete(api::abort_current::<H>),
+        )
         .route("/admin/compactions", get(api::list::<H>))
         .route(
             "/admin/compactions/{id}",