don't commit expired remakes

1 files changed, 93 insertions, 58 deletions

diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 96edb91..d90cca0 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -6,8 +6,13 @@
 //! * an operator requested an abort,
 //! * the local epoch changed,
 //! * a competing remake of the same epoch committed,
-//! * a majority is out of reach (too many members voted for a competing remake), or
-//! * the voting period (`settleSecs`) passed without a majority.
+//! * the voting period (`settleSecs`) passed without a majority, or
+//! * a majority is out of reach (too many members voted for a competing remake).
+//!
+//! 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.
 //!
 //! Committing builds the snapshot at the base heads and records the new epoch, so
 //! every node can verify its own build against it.
@@ -21,7 +26,7 @@ use tracing::{info, warn};
 
 use super::Host;
 use super::content;
-use super::records::{self, Decision, Record};
+use super::records::{self, Decision, Phase, Record};
 use super::snapshot;
 use super::vote;
 use crate::document::store::StoreError;
@@ -65,15 +70,26 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
         return Ok(true);
     }
 
-    let now = records::unix_millis();
-    let grace = host.dormant_after().as_millis() as u64;
-
-    if let Some(reason) = abort_reason(host, id, &record, &all, now, grace) {
+    if records::EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch {
+        decide_abort(host, id, "local epoch changed during the vote".to_owned()).await?;
+        return Ok(true);
+    }
+    if let Some(reason) = obstacle(id, &record, &all) {
         decide_abort(host, id, reason).await?;
         return Ok(true);
     }
 
     let tally = vote::tally(id, &record, &all, &register);
+    // The deadline first: a majority counted after it is too late.
+    if records::unix_millis() >= record.proposal.settle_until {
+        decide_abort(
+            host,
+            id,
+            format!("no majority in the voting period: {tally}"),
+        )
+        .await?;
+        return Ok(true);
+    }
     if tally.reached() {
         info!(remake = %id, votes = %tally, "Majority reached");
         commit(host, id, &record).await?;
@@ -83,45 +99,20 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
         decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?;
         return Ok(true);
     }
-    if now >= record.proposal.settle_until {
-        decide_abort(
-            host,
-            id,
-            format!("no majority in the voting period: {tally}"),
-        )
-        .await?;
-        return Ok(true);
-    }
     Ok(false)
 }
 
-fn abort_reason<H: Host>(
-    host: &H,
-    id: &str,
-    record: &Record,
-    all: &[(String, Record)],
-    now: u64,
-    grace: u64,
-) -> Option<String> {
-    let proposal = &record.proposal;
-
+/// Why remake `id` must be aborted whatever its votes, if it must.
+fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<String> {
     if let Some(request) = &record.abort_requested {
         return Some(format!("aborted by operator via {}", request.by));
     }
-
-    if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch {
-        return Some("local epoch changed during the vote".to_owned());
-    }
-
-    if let Some((other, _)) = all.iter().find(|(other, candidate)| {
+    let (other, _) = all.iter().find(|(other, candidate)| {
         other != id
-            && candidate.proposal.from_epoch == proposal.from_epoch
-            && candidate.phase(now, grace) == records::Phase::Committed
-    }) {
-        return Some(format!("competing remake {other} committed"));
-    }
-
-    None
+            && candidate.proposal.from_epoch == record.proposal.from_epoch
+            && candidate.committed_epoch().is_some()
+    })?;
+    Some(format!("competing remake {other} committed"))
 }
 
 async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
@@ -157,30 +148,74 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
         snapshot_bytes,
         decided_at: records::unix_millis(),
     };
-    write_decision(host, id, decision).await?;
-
-    // The local switch reuses it instead of building again.
-    host.remake().store_prebuilt(id, snapshot);
-
-    info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed");
+    match decide(host, id, decision).await? {
+        Some(Decision::Committed { .. }) => {
+            // The local switch reuses it instead of building again.
+            host.remake().store_prebuilt(id, snapshot);
+            info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed");
+        }
+        Some(Decision::Aborted { reason, .. }) => {
+            warn!(remake = %id, reason = %reason, "Remake aborted while building its snapshot");
+        }
+        None => {}
+    }
     Ok(())
 }
 
 async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
-    warn!(remake = %id, reason = %reason, "Remake aborted");
-    write_decision(
-        host,
-        id,
-        Decision::Aborted {
-            reason,
-            decided_at: records::unix_millis(),
-        },
-    )
-    .await
+    let decision = Decision::Aborted {
+        reason: reason.clone(),
+        decided_at: records::unix_millis(),
+    };
+    if decide(host, id, decision).await?.is_some() {
+        warn!(remake = %id, reason = %reason, "Remake aborted");
+    }
+    Ok(())
 }
 
-async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> {
+/// Writes `decision`, checked again under the write lock against the records as they
+/// are now (building the snapshot takes a while): a commit becomes an abort if the
+/// operator asked for one, a competing remake committed, the remake expired, or the
+/// majority is gone.
+/// Returns what was written; nothing if the remake is decided already or gone.
+async fn decide<H: Host>(
+    host: &H,
+    id: &str,
+    decision: Decision,
+) -> Result<Option<Decision>, StoreError> {
     let id = id.to_owned();
-    host.commit_doc(move |doc| records::write_decision(doc, &id, &decision))
-        .await
+    let grace = host.dormant_after().as_millis() as u64;
+    host.commit_doc(move |doc| -> Result<Option<Decision>, StoreError> {
+        let all = records::load_all(doc);
+        let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else {
+            return Ok(None);
+        };
+        if record.decision.is_some() {
+            return Ok(None);
+        }
+        let decision = match decision {
+            Decision::Committed { decided_at, .. } => {
+                let register = records::load_register(doc);
+                let tally = vote::tally(&id, record, &all, &register);
+                let expired = record.phase(records::unix_millis(), grace) == Phase::Expired;
+                match obstacle(&id, record, &all) {
+                    Some(reason) => Decision::Aborted { reason, decided_at },
+                    // Others may abort it by now (`api`): too late to commit.
+                    None if expired => Decision::Aborted {
+                        reason: format!("expired while building the snapshot: {tally}"),
+                        decided_at,
+                    },
+                    None if !tally.reached() => Decision::Aborted {
+                        reason: format!("the majority was lost: {tally}"),
+                        decided_at,
+                    },
+                    None => decision,
+                }
+            }
+            aborted @ Decision::Aborted { .. } => aborted,
+        };
+        records::write_decision(doc, &id, &decision)?;
+        Ok(Some(decision))
+    })
+    .await
 }