1 files changed, 30 insertions, 5 deletions

diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index d90cca0..9bcf8ba 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -17,8 +17,10 @@
 //! 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; its votes stay cast (`vote`).
+//! 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::time::Duration;
 
@@ -26,27 +28,50 @@ use tracing::{info, warn};
 
 use super::Host;
 use super::content;
-use super::records::{self, Decision, Phase, Record};
+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.
@@ -70,7 +95,7 @@ async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
         return Ok(true);
     }
 
-    if records::EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch {
+    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);
     }