allow superseding expired remakes

1 files changed, 188 insertions, 23 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index ffd1844..826c81e 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -12,9 +12,10 @@
 //! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
 //!   the node is catching up with a newer epoch from a peer.
 //! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while
-//!   proposed; `409` once committed (a switch can't be undone) or already over.
-//!   `DELETE /admin/remakes` does the same for the remake of the local epoch that
-//!   is in progress, without having to know its ID.
+//!   undecided; `409` once committed (a switch can't be undone) or aborted. An expired
+//!   remake is aborted right away by the node asked: its coordinator is gone, and
+//!   can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the
+//!   same for the undecided remake of the local epoch, without having to know its ID.
 //!
 //! Request body for POST (all optional):
 //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …],
@@ -22,8 +23,11 @@
 //!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
 //!    "settleSecs": 480, "switchDelaySecs": 60 }`
 //!
-//! `?force=true` accepts a `switchDelaySecs` below one keepalive interval: for tests
-//! and debugging, not for real swarms.
+//! `?force=` overrides checks: `true` all of them, or a comma-separated list of
+//! * `switchDelaySecs`: accept one below a keepalive interval (for tests and debugging,
+//!   not for real swarms);
+//! * `supersede`: abort expired remakes of this epoch and propose in their place.
+//!   Without it, an expired remake still holding votes answers `409`.
 //!
 //! `kind` is `compact` (no spec), `replace`, whose spec is
 //! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
@@ -39,6 +43,7 @@
 //! their paths:
 //! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes`
 
+use std::collections::HashSet;
 use std::time::Duration;
 
 use automerge::ChangeHash;
@@ -62,8 +67,8 @@ use uuid::Uuid;
 use super::coordinator;
 use super::local::LocalStatus;
 use super::records::{
-    self, AbortRequest, ContentRef, EpochRecord, Phase, Policy, Proposal, Record, Retention, Vote,
-    unix_millis,
+    self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record,
+    Retention, Vote, unix_millis,
 };
 use super::{Host, kind, vote};
 use crate::document::store::StoreError;
@@ -79,9 +84,8 @@ pub(crate) struct WaitQuery {
 pub(crate) struct StartQuery {
     #[serde(default)]
     wait: bool,
-    /// Allows a switch delay shorter than one keepalive interval.
-    #[serde(default)]
-    force: bool,
+    /// The checks to override (`Force`).
+    force: Option<String>,
     // Only for an import uploaded as `application/octet-stream`, which has no JSON
     // body to carry them:
     /// The rules, as JSON (as in the spec).
@@ -151,10 +155,57 @@ struct StartRequest {
     /// How long votes are collected; twice the dormancy threshold if absent.
     settle_secs: Option<u64>,
     /// How long nodes wait after seeing the commit before switching; one keepalive
-    /// interval if absent. Shorter only with `?force=true`.
+    /// interval if absent. Shorter only with `?force=switchDelaySecs`.
     switch_delay_secs: Option<u64>,
 }
 
+/// The checks `?force=` overrides: `true` all of them, or a comma-separated list.
+#[derive(Debug, Default, Clone, Copy)]
+struct Force {
+    switch_delay: bool,
+    supersede: bool,
+}
+
+impl Force {
+    fn parse(value: Option<&str>) -> Result<Self, String> {
+        let mut force = Force::default();
+        match value.map(str::trim) {
+            None | Some("" | "false") => {}
+            Some("true") => {
+                force = Force {
+                    switch_delay: true,
+                    supersede: true,
+                };
+            }
+            Some(list) => {
+                for name in list.split(',').map(str::trim) {
+                    match name {
+                        "switchDelaySecs" => force.switch_delay = true,
+                        "supersede" => force.supersede = true,
+                        other => {
+                            return Err(format!(
+                                "unknown force {other:?}: use true, or a list of \
+                                 switchDelaySecs and supersede"
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+        Ok(force)
+    }
+}
+
+/// An expired remake of the epoch a proposal would supersede.
+#[derive(Debug)]
+struct ExpiredRemake {
+    id: String,
+    coordinator: String,
+    /// When it expired (Unix milliseconds).
+    since: u64,
+    tally: vote::Tally,
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct Report {
@@ -171,6 +222,9 @@ struct Report {
 enum StartError {
     #[error("remake {0} is already open for this epoch")]
     Open(String),
+    /// Expired remakes still hold votes; superseding them takes `force=supersede`.
+    #[error("expired remakes of this epoch still hold votes")]
+    Expired(Vec<ExpiredRemake>),
     /// The proposer votes for its own remake, so it must be free to.
     #[error("this node can't vote for a new remake: {0}")]
     VoteHeld(String),
@@ -278,6 +332,10 @@ async fn start_kind<H: Host>(
     if !host.is_node_ready().await {
         return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
     }
+    let force = match Force::parse(query.force.as_deref()) {
+        Ok(force) => force,
+        Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None),
+    };
 
     let request: StartRequest = if is_octet_stream(headers) {
         match query.import_request(&body) {
@@ -368,7 +426,7 @@ async fn start_kind<H: Host>(
     let keepalive = dormant_after / 4;
     let switch_delay = request.switch_delay_secs.map(Duration::from_secs);
     match switch_delay {
-        Some(delay) if delay < keepalive && query.force => {
+        Some(delay) if delay < keepalive && force.switch_delay => {
             warn!(
                 switch_delay_secs = delay.as_secs(),
                 keepalive_secs = keepalive.as_secs(),
@@ -379,7 +437,7 @@ async fn start_kind<H: Host>(
             return error_response(
                 StatusCode::BAD_REQUEST,
                 format!(
-                    "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=true to override",
+                    "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=switchDelaySecs to override",
                     keepalive.as_secs()
                 ),
                 None,
@@ -433,6 +491,7 @@ async fn start_kind<H: Host>(
         id.clone(),
         proposal,
         base,
+        force.supersede,
         now,
         grace,
     ));
@@ -457,6 +516,35 @@ async fn start_kind<H: Host>(
                 Some(json!({ "remake": open })),
             );
         }
+        Err(StartError::Expired(expired)) => {
+            let active: HashSet<String> = host
+                .peer_summaries()
+                .await
+                .into_iter()
+                .filter(|peer| peer.active)
+                .map(|peer| peer.id)
+                .collect();
+            let me = host.node_id();
+            let expired: Vec<JsonValue> = expired
+                .iter()
+                .map(|remake| {
+                    json!({
+                        "id": remake.id,
+                        "coordinator": remake.coordinator,
+                        "coordinatorActive": remake.coordinator == me
+                            || active.contains(&remake.coordinator),
+                        "expiredAt": remake.since,
+                        "tally": remake.tally.to_string(),
+                    })
+                })
+                .collect();
+            return error_response(
+                StatusCode::CONFLICT,
+                "an expired remake of this epoch still holds votes: abort it, or add \
+                 ?force=supersede to abort it and propose in its place",
+                Some(json!({ "expired": expired, "force": "supersede" })),
+            );
+        }
         Err(error @ StartError::VoteHeld(_)) => {
             return error_response(StatusCode::CONFLICT, error.to_string(), None);
         }
@@ -502,6 +590,7 @@ async fn propose<H: Host>(
     id: String,
     mut proposal: Proposal,
     base: Option<Vec<ChangeHash>>,
+    supersede: bool,
     now: u64,
     grace: u64,
 ) -> Result<Record, StartError> {
@@ -511,14 +600,51 @@ async fn propose<H: Host>(
     let started = host.remake().started_at() + grace;
     let record = host
         .commit_doc(move |doc| -> Result<Record, StartError> {
-            let all = records::load_all(doc);
+            let mut all = records::load_all(doc);
             if let Some((open, _)) = all.iter().find(|(_, record)| {
                 record.proposal.from_epoch == proposal.from_epoch
                     && record.phase(now, grace).is_open()
             }) {
                 return Err(StartError::Open(open.clone()));
             }
-            let register = records::load_register(doc);
+            let mut register = records::load_register(doc);
+
+            // Expired remakes hold their votes; their coordinators can't commit them
+            // any more, so they may be aborted here.
+            let expired: Vec<&String> = all
+                .iter()
+                .filter(|(_, record)| {
+                    record.proposal.from_epoch == proposal.from_epoch
+                        && record.phase(now, grace) == Phase::Expired
+                })
+                .map(|(id, _)| id)
+                .collect();
+            if !expired.is_empty() {
+                if !supersede {
+                    let details = all
+                        .iter()
+                        .filter(|(id, _)| expired.contains(&id))
+                        .map(|(id, record)| ExpiredRemake {
+                            id: id.clone(),
+                            coordinator: record.proposal.coordinator.clone(),
+                            since: record.proposal.settle_until.saturating_add(grace),
+                            tally: vote::tally(id, record, &all, &register),
+                        })
+                        .collect();
+                    return Err(StartError::Expired(details));
+                }
+                let decision = Decision::Aborted {
+                    reason: format!("superseded by remake {record_id}"),
+                    decided_at: now,
+                };
+                for id in expired {
+                    records::write_decision(doc, id, &decision)?;
+                    info!(remake = %id, by = %record_id, "Superseded expired remake");
+                }
+                all = records::load_all(doc);
+                register = records::load_register(doc);
+            }
+
             match vote::freedom(&me, &proposal.from_epoch, &all, &register, started) {
                 vote::Freedom::Any => {}
                 vote::Freedom::Only(id) => {
@@ -565,6 +691,7 @@ async fn propose<H: Host>(
                 proposal,
                 decision: None,
                 abort_requested: None,
+                contested: false,
                 votes: [(me, own_vote)].into(),
             })
         })
@@ -610,22 +737,18 @@ pub(crate) async fn abort<H: Host>(
 }
 
 /// `DELETE /admin/remakes`: aborts whichever remake 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).
+/// undecided (proposed or expired), the oldest if there are several.
 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
+                    record.proposal.from_epoch == local && record.decision.is_none()
                 })
                 .min_by_key(|(_, record)| record.proposal.proposed_at)
                 .map(|(id, _)| id)
@@ -649,6 +772,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
 
     match current.phase {
         Phase::Proposed => {}
+        Phase::Expired => return abort_expired(&host, &id).await,
         Phase::Committed => {
             return error_response(
                 StatusCode::CONFLICT,
@@ -656,7 +780,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
                 Some(json!({ "remake": current })),
             );
         }
-        Phase::Aborted | Phase::Expired => {
+        Phase::Aborted => {
             return error_response(
                 StatusCode::CONFLICT,
                 "already over",
@@ -666,7 +790,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
     }
 
     // The coordinator applies the request on its next check (within about a second).
-    // If it is gone, the proposal expires anyway.
+    // If it is gone, the proposal expires, and can then be aborted right away.
     if current.record.abort_requested.is_none() {
         let request = AbortRequest {
             by: host.node_id(),
@@ -699,6 +823,47 @@ async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
     }
 }
 
+/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit
+/// it any more (`coordinator`). Answers with the final report.
+async fn abort_expired<H: Host>(host: &H, id: &str) -> Response {
+    let record_id = id.to_owned();
+    let decision = Decision::Aborted {
+        reason: format!(
+            "aborted by operator via {} after it expired",
+            host.node_id()
+        ),
+        decided_at: unix_millis(),
+    };
+    let grace = grace(host);
+    let written = host
+        .commit_doc(move |doc| -> Result<bool, StoreError> {
+            let expired = records::load(doc, &record_id)
+                .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired);
+            if expired {
+                records::write_decision(doc, &record_id, &decision)?;
+            }
+            Ok(expired)
+        })
+        .await;
+    match written {
+        Ok(true) => info!(remake = %id, "Aborted expired remake"),
+        // Decided meanwhile: the report says how.
+        Ok(false) => {}
+        Err(error) => {
+            error!(remake = %id, error = %format!("{error:#}"), "Failed to abort expired remake");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to record the abort",
+                None,
+            );
+        }
+    }
+    match load_report(host, id).await {
+        Some(report) => Json(report).into_response(),
+        None => error_response(StatusCode::NOT_FOUND, "no such remake", None),
+    }
+}
+
 /// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and
 /// an ongoing catch-up with a peer's newer epoch (`null` if none).
 pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {