1 files changed, 37 insertions, 47 deletions

diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index d1b3397..5885884 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -1,28 +1,28 @@
-//! The coordinator: the node that received the proposal watches the swarm during the
-//! settle period and then writes the decision.
+//! The coordinator: the node that received the proposal counts the votes (`vote`) and
+//! writes the decision.
 //!
-//! It aborts if, at any point before deciding:
+//! It commits as soon as a majority of the electorate voted for the remake. It aborts
+//! if, before that:
 //! * an operator requested an abort,
-//! * a competing proposal for the same epoch appeared (both coordinators abort),
-//! * a peer it considered went dormant (a partition may hide a competitor),
-//! * the local epoch changed, or
-//! * the decision would come later than the settle period plus the grace margin
-//!   (other nodes report the proposal as expired by then).
+//! * 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.
 //!
-//! Otherwise it builds the snapshot at the base heads and commits, recording the new
-//! epoch so every node can verify its own build against it.
+//! Committing builds the snapshot at the base heads and records the new epoch, so
+//! every node can verify its own build against it.
 //!
 //! The task lives only in memory. If the coordinator dies, nobody decides and the
-//! proposal expires.
+//! proposal expires (its votes become void).
 
-use std::collections::HashMap;
 use std::time::Duration;
 
 use tracing::{info, warn};
 
+use super::Host;
 use super::records::{self, Decision, Record};
 use super::snapshot;
-use super::Host;
+use super::vote;
 use crate::document::store::StoreError;
 
 const TICK: Duration = Duration::from_secs(1);
@@ -61,20 +61,34 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
     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).await {
+    if let Some(reason) = abort_reason(host, id, &record, &all, now, grace) {
         decide_abort(host, id, reason).await?;
         return Ok(true);
     }
 
-    if now < record.proposal.settle_until {
-        return Ok(false);
+    let tally = vote::tally(id, &record, &all, now, grace);
+    if tally.reached() {
+        info!(remake = %id, votes = %tally, "Majority reached");
+        commit(host, id, &record).await?;
+        return Ok(true);
     }
-
-    commit(host, id, &record).await?;
-    Ok(true)
+    if tally.out_of_reach() {
+        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)
 }
 
-async fn abort_reason<H: Host>(
+fn abort_reason<H: Host>(
     host: &H,
     id: &str,
     record: &Record,
@@ -89,39 +103,15 @@ async fn abort_reason<H: Host>(
     }
 
     if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch {
-        return Some("local epoch changed during settle".to_owned());
+        return Some("local epoch changed during the vote".to_owned());
     }
 
     if let Some((other, _)) = all.iter().find(|(other, candidate)| {
         other != id
             && candidate.proposal.from_epoch == proposal.from_epoch
-            && candidate.phase(now, grace).is_open()
+            && candidate.phase(now, grace) == records::Phase::Committed
     }) {
-        return Some(format!("competing remake {other} for the same epoch"));
-    }
-
-    // Peers forgotten in the meantime are ignored: an operator decided they are gone.
-    let peers: HashMap<String, bool> = host
-        .peer_summaries()
-        .await
-        .into_iter()
-        .map(|peer| (peer.id, peer.active))
-        .collect();
-    let dormant: Vec<&str> = proposal
-        .peers_considered
-        .iter()
-        .filter(|peer| peers.get(peer.as_str()) == Some(&false))
-        .map(String::as_str)
-        .collect();
-    if !dormant.is_empty() {
-        return Some(format!(
-            "peers went dormant during settle: {}",
-            dormant.join(", ")
-        ));
-    }
-
-    if now > proposal.settle_until.saturating_add(grace) {
-        return Some("decision deadline passed".to_owned());
+        return Some(format!("competing remake {other} committed"));
     }
 
     None