1 files changed, 157 insertions, 94 deletions

diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index d1b3397..9bcf8ba 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -1,53 +1,90 @@
-//! 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,
+//! * the voting period (`settleSecs`) passed without a majority, or
+//! * a majority is out of reach (too many members voted for a competing remake).
 //!
-//! 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.
+//! 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.
 //!
-//! The task lives only in memory. If the coordinator dies, nobody decides and the
-//! proposal expires.
+//! 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. A coordinator that restarts takes its undecided
+//! remakes up again once ready (`resume`); past the voting period, that means aborting
+//! them. If it is gone for good, nobody decides and the proposal expires; its votes
+//! stay cast (`vote`).
 
-use std::collections::HashMap;
 use std::time::Duration;
 
 use tracing::{info, warn};
 
-use super::records::{self, Decision, Record};
-use super::snapshot;
 use super::Host;
+use super::content;
+use super::records::{self, Decision, EpochRecord, Phase, Record};
+use super::snapshot;
+use super::vote;
 use crate::document::store::StoreError;
 
 const TICK: Duration = Duration::from_secs(1);
 
+/// Coordinates remake `id` until it is decided; returns at once if this node already
+/// does.
 pub(crate) async fn run<H: Host>(host: H, id: String) {
+    if !host.remake().start_coordinating(&id) {
+        return;
+    }
     info!(remake = %id, "Coordinating remake");
 
     loop {
         tokio::time::sleep(TICK).await;
 
         match step(&host, &id).await {
-            Ok(true) => return,
+            Ok(true) => break,
             Ok(false) => {}
             Err(error) => {
                 warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
             }
         }
     }
+    host.remake().end_coordinating(&id);
+}
+
+/// Takes up coordinating this node's undecided remakes of `epoch` among `all` that it
+/// doesn't coordinate (it restarted since proposing them). Called by the remake worker
+/// once the node is ready, so its records are complete.
+pub(super) fn resume<H: Host>(host: &H, epoch: &EpochRecord, all: &[(String, Record)]) {
+    let me = host.node_id();
+    for (id, record) in all {
+        if record.decision.is_none()
+            && record.proposal.coordinator == me
+            && &record.proposal.from_epoch == epoch
+            && !host.remake().is_coordinating(id)
+        {
+            info!(remake = %id, "Resuming coordination of own remake");
+            tokio::spawn(run(host.clone(), id.clone()));
+        }
+    }
 }
 
 /// One evaluation. Returns true once the coordinator is done.
 async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
     let wanted = id.to_owned();
-    let (record, all) = host
-        .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc)))
+    let (record, all, register) = host
+        .read_doc(move |doc| {
+            (
+                records::load(doc, &wanted),
+                records::load_all(doc),
+                records::load_register(doc),
+            )
+        })
         .await;
 
     let Some(record) = record else {
@@ -58,77 +95,53 @@ 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).await {
+    if 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);
     }
 
-    if now < record.proposal.settle_until {
-        return Ok(false);
+    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);
     }
-
-    commit(host, id, &record).await?;
-    Ok(true)
+    if tally.reached() {
+        info!(remake = %id, votes = %tally, "Majority reached");
+        commit(host, id, &record).await?;
+        return Ok(true);
+    }
+    if tally.out_of_reach() {
+        decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?;
+        return Ok(true);
+    }
+    Ok(false)
 }
 
-async 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 settle".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).is_open()
-    }) {
-        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());
-    }
-
-    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> {
-    let Some(request) = record.snapshot_request(id) else {
+    let Some(mut request) = record.snapshot_request(id) else {
         return decide_abort(
             host,
             id,
@@ -136,6 +149,12 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
         )
         .await;
     };
+    if let Some(reference) = &record.proposal.content {
+        match content::ensure(host, reference).await {
+            Ok(bytes) => request.content = Some(bytes),
+            Err(reason) => return decide_abort(host, id, reason).await,
+        }
+    }
 
     // Clone under the read lock, build without any lock.
     let copy = host.read_doc(|doc| doc.clone()).await;
@@ -154,30 +173,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
 }