remake worker wakes up when vote is due

4 files changed, 42 insertions, 2 deletions

diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 6902515..50ed4a3 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -17,6 +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.
 //!
+//! It counts on every change to the records or votes, and once a second for the
+//! deadline: a vote that arrives just before the deadline counts, even with a short
+//! voting period.
+//!
 //! 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
@@ -24,6 +28,7 @@
 
 use std::time::Duration;
 
+use tokio::sync::broadcast;
 use tracing::{info, warn};
 
 use super::Host;
@@ -31,7 +36,9 @@ use super::content;
 use super::records::{self, Decision, EpochRecord, Phase, Record};
 use super::snapshot;
 use super::vote;
+use crate::document::meta::{self, REMAKES, VOTES};
 use crate::document::store::StoreError;
+use crate::swarm::sync::paths_related;
 
 const TICK: Duration = Duration::from_secs(1);
 
@@ -43,8 +50,29 @@ pub(crate) async fn run<H: Host>(host: H, id: String) {
     }
     info!(remake = %id, "Coordinating remake");
 
+    let watched = [meta::path(&[REMAKES]), meta::path(&[VOTES])];
+    let mut changes = host.subscribe_changes();
+    let mut changes_closed = false;
+    let mut tick = tokio::time::interval_at(tokio::time::Instant::now() + TICK, TICK);
+    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     loop {
-        tokio::time::sleep(TICK).await;
+        let due = tokio::select! {
+            _ = tick.tick() => true,
+            received = changes.recv(), if !changes_closed => match received {
+                Ok(set) => set
+                    .paths
+                    .iter()
+                    .any(|path| watched.iter().any(|dir| paths_related(path, dir))),
+                Err(broadcast::error::RecvError::Lagged(_)) => true,
+                Err(broadcast::error::RecvError::Closed) => {
+                    changes_closed = true;
+                    false
+                }
+            }
+        };
+        if !due {
+            continue;
+        }
 
         match step(&host, &id).await {
             Ok(true) => break,
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 60f7184..7810690 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -188,6 +188,14 @@ impl RemakeState {
         self.lock().coordinating.contains(id)
     }
 
+    /// How long until the vote `vote_due` scheduled is due, while it is ahead: the
+    /// worker wakes up for it (`switch`).
+    pub(crate) fn vote_wait(&self, now: UnixMillis) -> Option<Duration> {
+        let (_, due) = self.lock().vote_due.clone()?;
+        // A millisecond late rather than early: early, it would find nothing due.
+        (due > now).then(|| Duration::from_millis(due.0 - now.0 + 1))
+    }
+
     /// When this node votes for `remake` (`vote::cast`): asked first, `now` plus
     /// `delay()`; asked again, the same time.
     pub(crate) fn vote_due(
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index ea56de6..6c90f44 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -54,8 +54,12 @@ pub(crate) async fn run_worker<H: Host>(host: H) {
 
     loop {
         // React promptly to changes of the records; the timer covers the switch delay.
+        // A vote waiting for its random delay (`vote`) is cast when due, not at the next
+        // tick: with a short voting period, a second can make it too late.
+        let vote_wait = host.remake().vote_wait(host.now_millis());
         let should_check = tokio::select! {
             _ = tick.tick() => true,
+            _ = tokio::time::sleep(vote_wait.unwrap_or_default()), if vote_wait.is_some() => true,
             received = changes.recv(), if !changes_closed => match received {
                 Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)),
                 Err(broadcast::error::RecvError::Lagged(_)) => true,
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index d5b167f..e5f8dbe 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -278,7 +278,7 @@ pub(super) async fn cast<H: Host>(host: &H) {
         return;
     }
     // Spread over half a keepalive interval, so later voters see the earlier votes
-    // (the worker checks every second).
+    // (the worker wakes up when it is due, `RemakeState::vote_wait`).
     let spread = host.dormant_after() / 8;
     if now < host.remake().vote_due(&chosen, now, || random_part(spread)) {
         return;