bump abort margin & allow aborting active remake

3 files changed, 179 insertions, 42 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index 6b1a30c..0428a43 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -81,6 +81,14 @@ pub(crate) struct WaitQuery {
     wait: bool,
 }
 
+#[derive(Debug, Deserialize)]
+pub(crate) struct AbortQuery {
+    #[serde(default)]
+    wait: bool,
+    /// The checks to override (`force`): `coordinatorActive`.
+    force: Option<String>,
+}
+
 #[derive(Debug, Default, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct StartQuery {
@@ -162,18 +170,83 @@ struct StartRequest {
 }
 
 /// What `?force=` may override on a proposal (`force`).
-const FORCE: &[&str] = &["switchDelaySecs", "supersede"];
+const FORCE: &[&str] = &["switchDelaySecs", "supersede", "coordinatorActive"];
+/// What `?force=` may override on an abort.
+const ABORT_FORCE: &[&str] = &["coordinatorActive"];
+
+/// When a node other than the coordinator may abort an expired remake: a whole grace
+/// period after it expired. Its coordinator stops committing at expiry
+/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if
+/// their clocks are more than a grace period apart.
+fn abortable_at(record: &Record, grace: u64) -> u64 {
+    record
+        .proposal
+        .settle_until
+        .saturating_add(grace.saturating_mul(2))
+}
+
+/// Why this node may not abort expired `record` now, and the `force` name that would
+/// override it (none for one that only time cures).
+fn abort_obstacle(
+    record: &Record,
+    me: &str,
+    active_peers: &HashSet<String>,
+    force: &Force,
+    now: u64,
+    grace: u64,
+) -> Option<(String, Option<&'static str>)> {
+    let at = abortable_at(record, grace);
+    if now < at {
+        return Some((
+            format!(
+                "only its coordinator may abort it until a grace period after it expired ({}s from now)",
+                (at - now).div_ceil(1000)
+            ),
+            None,
+        ));
+    }
+    let coordinator = &record.proposal.coordinator;
+    if coordinator != me && active_peers.contains(coordinator) && !force.allows("coordinatorActive")
+    {
+        return Some((
+            format!(
+                "its coordinator {coordinator} is an active peer, so it is stuck rather than gone (look at its log); add force=coordinatorActive to abort it anyway"
+            ),
+            Some("coordinatorActive"),
+        ));
+    }
+    None
+}
+
+/// The active peers this node sees.
+async fn active_peers<H: Host>(host: &H) -> HashSet<String> {
+    host.peer_summaries()
+        .await
+        .into_iter()
+        .filter(|peer| peer.active)
+        .map(|peer| peer.id)
+        .collect()
+}
 
 /// An expired remake of the epoch a proposal would supersede.
 #[derive(Debug)]
 struct ExpiredRemake {
     id: String,
     coordinator: String,
+    coordinator_active: bool,
     /// When it expired (Unix milliseconds).
     since: u64,
+    /// When another node may abort it (`abortable_at`).
+    abortable_at: u64,
     tally: vote::Tally,
 }
 
+/// Superseding expired remakes, as the proposal's `force` asks.
+struct Supersede {
+    force: Force,
+    active_peers: HashSet<String>,
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct Report {
@@ -190,9 +263,15 @@ 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>),
+    /// Expired remakes still hold votes, and aren't superseded: not asked to
+    /// (`force=supersede`), or not possible yet (`abort_obstacle`).
+    #[error("{message}")]
+    Expired {
+        message: String,
+        remakes: Vec<ExpiredRemake>,
+        /// The `force` name that would override it, if any.
+        force: Option<&'static str>,
+    },
     /// 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),
@@ -469,7 +548,10 @@ async fn start_kind<H: Host>(
         id.clone(),
         proposal,
         base,
-        force.allows("supersede"),
+        Supersede {
+            force: force.clone(),
+            active_peers: active_peers(&host).await,
+        },
         now,
         grace,
     ));
@@ -494,33 +576,28 @@ 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
+        Err(StartError::Expired {
+            message,
+            remakes,
+            force,
+        }) => {
+            let expired: Vec<JsonValue> = remakes
                 .iter()
                 .map(|remake| {
                     json!({
                         "id": remake.id,
                         "coordinator": remake.coordinator,
-                        "coordinatorActive": remake.coordinator == me
-                            || active.contains(&remake.coordinator),
+                        "coordinatorActive": remake.coordinator_active,
                         "expiredAt": remake.since,
+                        "abortableAt": remake.abortable_at,
                         "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" })),
+                message,
+                Some(json!({ "expired": expired, "force": force })),
             );
         }
         Err(error @ StartError::VoteHeld(_)) => {
@@ -568,7 +645,7 @@ async fn propose<H: Host>(
     id: String,
     mut proposal: Proposal,
     base: Option<Vec<ChangeHash>>,
-    supersede: bool,
+    supersede: Supersede,
     now: u64,
     grace: u64,
 ) -> Result<Record, StartError> {
@@ -599,18 +676,50 @@ async fn propose<H: Host>(
                 .map(|(id, _)| id)
                 .collect();
             if !expired.is_empty() {
-                if !supersede {
-                    let details = all
-                        .iter()
+                let remakes = || {
+                    all.iter()
                         .filter(|(id, _)| expired.contains(&id))
                         .map(|(id, record)| ExpiredRemake {
                             id: id.clone(),
                             coordinator: record.proposal.coordinator.clone(),
+                            coordinator_active: supersede
+                                .active_peers
+                                .contains(&record.proposal.coordinator),
                             since: record.proposal.settle_until.saturating_add(grace),
+                            abortable_at: abortable_at(record, grace),
                             tally: vote::tally(id, record, &all, &register),
                         })
-                        .collect();
-                    return Err(StartError::Expired(details));
+                        .collect()
+                };
+                if !supersede.force.allows("supersede") {
+                    return Err(StartError::Expired {
+                        message: "an expired remake of this epoch still holds votes: abort it, \
+                                  or add ?force=supersede to abort it and propose in its place"
+                            .to_owned(),
+                        remakes: remakes(),
+                        force: Some("supersede"),
+                    });
+                }
+                let obstacle =
+                    all.iter()
+                        .filter(|(id, _)| expired.contains(&id))
+                        .find_map(|(id, record)| {
+                            let (why, force) = abort_obstacle(
+                                record,
+                                &me,
+                                &supersede.active_peers,
+                                &supersede.force,
+                                now,
+                                grace,
+                            )?;
+                            Some((format!("can't supersede remake {id}: {why}"), force))
+                        });
+                if let Some((message, force)) = obstacle {
+                    return Err(StartError::Expired {
+                        message,
+                        remakes: remakes(),
+                        force,
+                    });
                 }
                 let decision = Decision::Aborted {
                     reason: format!("superseded by remake {record_id}"),
@@ -710,16 +819,16 @@ pub(crate) async fn get<H: Host>(
 pub(crate) async fn abort<H: Host>(
     State(host): State<H>,
     Path(id): Path<String>,
-    Query(query): Query<WaitQuery>,
+    Query(query): Query<AbortQuery>,
 ) -> Response {
-    abort_by_id(host, id, query.wait).await
+    abort_by_id(host, id, query).await
 }
 
 /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still
 /// 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>,
+    Query(query): Query<AbortQuery>,
 ) -> Response {
     let local = EpochRecord::from(host.current_epoch());
     let open = host
@@ -735,7 +844,7 @@ pub(crate) async fn abort_current<H: Host>(
         .await;
 
     match open {
-        Some(id) => abort_by_id(host, id, query.wait).await,
+        Some(id) => abort_by_id(host, id, query).await,
         None => error_response(
             StatusCode::NOT_FOUND,
             "no remake of the local epoch is in progress",
@@ -744,14 +853,19 @@ pub(crate) async fn abort_current<H: Host>(
     }
 }
 
-async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
+async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Response {
+    let force = match Force::parse(query.force.as_deref(), ABORT_FORCE) {
+        Ok(force) => force,
+        Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None),
+    };
+    let wait = query.wait;
     let Some(current) = load_report(&host, &id).await else {
         return error_response(StatusCode::NOT_FOUND, "no such remake", None);
     };
 
     match current.phase {
         Phase::Proposed => {}
-        Phase::Expired => return abort_expired(&host, &id).await,
+        Phase::Expired => return abort_expired(&host, current, &force).await,
         Phase::Committed => {
             return error_response(
                 StatusCode::CONFLICT,
@@ -802,9 +916,30 @@ 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 {
+/// Aborts expired remake `current` from this node: its coordinator is gone, and can't
+/// commit it any more (`coordinator`), unless `abort_obstacle` says otherwise. Answers
+/// with the final report.
+async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Response {
+    let grace = grace(host);
+    if let Some((why, force)) = abort_obstacle(
+        &current.record,
+        &host.node_id(),
+        &active_peers(host).await,
+        force,
+        unix_millis(),
+        grace,
+    ) {
+        return error_response(
+            StatusCode::CONFLICT,
+            format!("can't abort it yet: {why}"),
+            Some(json!({
+                "remake": current,
+                "abortableAt": abortable_at(&current.record, grace),
+                "force": force,
+            })),
+        );
+    }
+    let id = current.id.as_str();
     let record_id = id.to_owned();
     let decision = Decision::Aborted {
         reason: format!(
@@ -813,11 +948,12 @@ async fn abort_expired<H: Host>(host: &H, id: &str) -> Response {
         ),
         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);
+            let now = unix_millis();
+            let expired = records::load(doc, &record_id).is_some_and(|record| {
+                record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace)
+            });
             if expired {
                 records::write_decision(doc, &record_id, &decision)?;
             }
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 3a7b071..427de08 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -12,7 +12,7 @@
 //! The decision is written in the same step that checks the records once more, so a
 //! commit never overrides an abort request or a competitor's commit that arrived while
 //! the snapshot was being built. Nor is a commit written once the remake has expired:
-//! from then on, any node may abort it (`api`), and the two mustn't race.
+//! a grace period later, any node may abort it (`api`), and the two mustn't race.
 //!
 //! Committing builds the snapshot at the base heads and records the new epoch, so
 //! every node can verify its own build against it.
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 0e40960..283f612 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -8,9 +8,10 @@
 //! * `abortRequested`: written by any node on an operator's request. It is an input to
 //!   the coordinator's decision, not a decision itself.
 //!
-//! Once a remake has expired, its coordinator no longer commits it, and any node may
-//! abort it on an operator's request (`api`). Should clocks disagree by more than the
-//! grace margin, a commit and such an abort could still be written concurrently. Such
+//! Once a remake has expired, its coordinator no longer commits it, and a grace period
+//! later any node may abort it on an operator's request (`api`). Should clocks disagree
+//! by more than that grace period, a commit and such an abort could still be written
+//! concurrently. Such
 //! a record is *contested*: it reads as committed, and nodes don't switch to it on
 //! their own.
 //!