3 files changed, 49 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);
     }
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 0d6c6c7..04ac0b0 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -53,6 +53,8 @@ struct Inner {
     contents: HashMap<[u8; 32], (Arc<Vec<u8>>, Instant)>,
     /// Content being downloaded.
     content_fetches: HashSet<[u8; 32]>,
+    /// Remakes this node coordinates now (`coordinator`).
+    coordinating: HashSet<String>,
 }
 
 /// Content no remake refers to any more is kept this long after it was stored, for
@@ -145,6 +147,19 @@ impl RemakeState {
         self.started_at
     }
 
+    /// Whether this node starts coordinating `id` now (false if it already does).
+    pub(crate) fn start_coordinating(&self, id: &str) -> bool {
+        self.lock().coordinating.insert(id.to_owned())
+    }
+
+    pub(crate) fn end_coordinating(&self, id: &str) {
+        self.lock().coordinating.remove(id);
+    }
+
+    pub(crate) fn is_coordinating(&self, id: &str) -> bool {
+        self.lock().coordinating.contains(id)
+    }
+
     fn lock(&self) -> MutexGuard<'_, Inner> {
         self.inner
             .lock()
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index c18905c..e4644fb 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -27,6 +27,7 @@ use tracing::{error, info, warn};
 use super::Host;
 use super::catchup;
 use super::content;
+use super::coordinator;
 use super::delta::{self, ReplayStats};
 use super::kind::Translator;
 use super::local::LocalStatus;
@@ -90,6 +91,9 @@ async fn check<H: Host>(host: &H) {
         host.dormant_after().as_millis() as u64,
     );
     let local_record = EpochRecord::from(local);
+    if host.is_node_ready().await {
+        coordinator::resume(host, &local_record, &all);
+    }
 
     let committed: Vec<(String, Record)> = all
         .into_iter()