remake worker wakes up when vote is due

1 files changed, 29 insertions, 1 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,