@@ -10,6 +10,8 @@
 //! - While one node holds the lease, only that node records runs.
 //! - Once that node is cut off from both peers, its lease lapses and it
 //!   records no further run. A peer takes the lease and records runs.
+//! - The leader balancer can move the lease between the live peers, so
+//!   either peer can fire. No two nodes ever fire the same minute.
 //!
 //! The test waits on minute boundaries, so it runs for a few minutes.
 
@@ -26,21 +28,28 @@ const CONVERGE: Duration = Duration::from_secs(30);
 const FIRE_WAIT: Duration = Duration::from_secs(150);
 const STEP: Duration = Duration::from_millis(250);
 
-/// Runs `node` recorded for the schedule.
-fn runs(node: &TestClusterNode) -> usize {
+/// The minute of each run `node` recorded for the schedule.
+fn run_minutes(node: &TestClusterNode) -> Vec<u64> {
     let Some(def) = node
         .shared
         .schedule_registry
         .list_all()
         .into_iter()
         .find(|def| def.name == SCHEDULE)
     else {
-        return 0;
+        return Vec::new();
     };
     node.shared
         .job_history
         .last_runs(def.database_id, def.tenant_id, SCHEDULE, 100)
-        .len()
+        .iter()
+        .map(|run| run.started_at / 60_000)
+        .collect()
+}
+
+/// Runs `node` recorded for the schedule.
+fn runs(node: &TestClusterNode) -> usize {
+    run_minutes(node).len()
 }
 
 /// Cut `node_id` off from every other node, both ways.
@@ -144,38 +153,25 @@ async fn a_cross_collection_schedule_fires_only_on_the_lease_holder() {
             == 1
     })
     .await;
-    let second = peers
-        .iter()
-        .find(|node| holds_vshard0_lease(node))
-        .map(|node| node.node_id)
-        .expect("the new holder");
 
-    // The new holder fires. The cut-off node, through the same minute
-    // boundaries, fires nothing more.
-    wait_for(
-        "the new lease holder fires the schedule",
-        FIRE_WAIT,
-        STEP,
-        || {
-            peers
-                .iter()
-                .any(|node| node.node_id == second && runs(node) > 0)
-        },
-    )
+    // A peer fires. The cut-off node, through the same minute boundaries,
+    // fires nothing more.
+    wait_for("a peer fires the schedule", FIRE_WAIT, STEP, || {
+        peers.iter().any(|node| runs(node) > 0)
+    })
     .await;
     assert_eq!(
         runs(cut_off),
         lapsed_runs,
         "a node without its lease fires nothing"
     );
-    for node in peers.iter().filter(|node| node.node_id != second) {
-        assert_eq!(
-            runs(node),
-            0,
-            "node {} holds no lease and fires nothing",
-            node.node_id
-        );
-    }
+
+    // One coordinator fires each minute, whichever peer holds the lease.
+    let mut minutes: Vec<u64> = cluster.nodes.iter().flat_map(run_minutes).collect();
+    let fired = minutes.len();
+    minutes.sort_unstable();
+    minutes.dedup();
+    assert_eq!(minutes.len(), fired, "two nodes fired the same minute");
 
     cluster.shutdown().await;
 }
@@ -62,6 +62,18 @@ impl StagedLogStorage {
     pub fn ticket(&self) -> Option<DurabilityTicket> {
         DurabilityTicket::for_staged(&self.disk)
     }
+
+    /// The sequence number of the last staged write. A caller reads it
+    /// before a Raft call and passes it to [`Self::reply_ticket`].
+    pub fn staged_through(&self) -> u64 {
+        self.disk.staged_through()
+    }
+
+    /// A ticket for the writes a reply built after `mark` depends on, or
+    /// `None` when they are durable. See [`DurabilityTicket::for_reply`].
+    pub fn reply_ticket(&self, mark: u64) -> Option<DurabilityTicket> {
+        DurabilityTicket::for_reply(&self.disk, mark)
+    }
 }
 
 impl Drop for StagedLogStorage {
 
@@ -39,6 +39,25 @@ impl DurabilityTicket {
         })
     }
 
+    /// A ticket for a reply staged after `mark`, or `None` when the writes it
+    /// depends on are durable. `mark` is [`GroupDisk::staged_through`] read
+    /// before the Raft call that built the reply.
+    ///
+    /// A reply depends on the latest hard state, whoever staged it: its term
+    /// and vote must survive a restart. When the call staged writes, the reply
+    /// depends on them too. Writes are durable in staging order, so the
+    /// ticket then covers every write staged so far. An `AppliedIndex` write
+    /// staged by the apply loop is no dependency of a reply that staged
+    /// nothing.
+    pub(super) fn for_reply(disk: &Arc<GroupDisk>, mark: u64) -> Option<Self> {
+        let (staged, hard_state) = disk.staged_marks();
+        let seq = if staged > mark { staged } else { hard_state };
+        (disk.progress().durable_seq < seq).then(|| Self {
+            disk: Arc::clone(disk),
+            seq,
+        })
+    }
+
     /// The group the ticket belongs to.
     pub fn group_id(&self) -> u64 {
         self.disk.group_id()
@@ -71,3 +90,63 @@ impl DurabilityTicket {
         self.disk.wait_blocking(self.seq)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use nodedb_raft::message::LogEntry;
+    use nodedb_raft::state::HardState;
+
+    use super::super::writer::StorageOp;
+    use super::*;
+
+    /// A disk with no writer thread: nothing staged becomes durable.
+    fn idle_disk() -> (tempfile::TempDir, Arc<GroupDisk>) {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join("group-3.redb");
+        let disk = Arc::new(GroupDisk::open(3, &path).expect("open"));
+        (dir, disk)
+    }
+
+    #[test]
+    fn a_reply_that_staged_nothing_waits_on_no_applied_index() {
+        let (_dir, disk) = idle_disk();
+        disk.stage(StorageOp::AppliedIndex(4));
+        let mark = disk.staged_through();
+
+        assert!(DurabilityTicket::for_reply(&disk, mark).is_none());
+        assert!(
+            DurabilityTicket::for_staged(&disk).is_some(),
+            "the applied index is not durable"
+        );
+    }
+
+    #[test]
+    fn a_reply_waits_on_the_latest_hard_state_whoever_staged_it() {
+        let (_dir, disk) = idle_disk();
+        disk.stage(StorageOp::HardState(HardState {
+            current_term: 2,
+            voted_for: 5,
+        }));
+        disk.stage(StorageOp::AppliedIndex(4));
+        let mark = disk.staged_through();
+
+        let ticket =
+            DurabilityTicket::for_reply(&disk, mark).expect("the hard state is not durable");
+        assert_eq!(ticket.seq, 1, "the later applied index is no dependency");
+    }
+
+    #[test]
+    fn a_reply_that_staged_writes_waits_on_every_staged_write() {
+        let (_dir, disk) = idle_disk();
+        disk.stage(StorageOp::AppliedIndex(4));
+        let mark = disk.staged_through();
+        disk.stage(StorageOp::Append(vec![LogEntry {
+            term: 1,
+            index: 5,
+            data: Vec::new(),
+        }]));
+
+        let ticket = DurabilityTicket::for_reply(&disk, mark).expect("the append is not durable");
+        assert_eq!(ticket.seq, 2);
+    }
+}
@@ -60,6 +60,8 @@ struct Queue {
     ops: VecDeque<(u64, StorageOp)>,
     /// The sequence number of the last staged write.
     staged_through: u64,
+    /// The sequence number of the last staged hard state, or 0.
+    hard_state_seq: u64,
     closed: bool,
 }
 
@@ -113,6 +115,9 @@ impl GroupDisk {
         let mut queue = self.queue.lock().unwrap_or_else(|p| p.into_inner());
         queue.staged_through += 1;
         let seq = queue.staged_through;
+        if matches!(op, StorageOp::HardState(_)) {
+            queue.hard_state_seq = seq;
+        }
         queue.ops.push_back((seq, op));
         self.work.notify_one();
     }
@@ -125,6 +130,13 @@ impl GroupDisk {
             .staged_through
     }
 
+    /// The sequence numbers of the last staged write and of the last staged
+    /// hard state, read together.
+    pub(super) fn staged_marks(&self) -> (u64, u64) {
+        let queue = self.queue.lock().unwrap_or_else(|p| p.into_inner());
+        (queue.staged_through, queue.hard_state_seq)
+    }
+
     /// How far the disk has come.
     pub fn progress(&self) -> DiskProgress {
         *self.progress.lock().unwrap_or_else(|p| p.into_inner())
 
@@ -36,6 +36,9 @@ pub struct MultiRaft {
     /// conf-changes applied here (via `apply_conf_change`) write THROUGH to
     /// the one table the query/data plane reads. Raft is the convergence
     /// mechanism on every applying node (leader and follower).
+    ///
+    /// Lock order: this struct's mutex first, then this routing guard.
+    /// See the lock-order section on [`RoutingTable`].
     pub(super) routing: Arc<RwLock<RoutingTable>>,
     /// Default election timeout range.
     pub(super) election_timeout_min: Duration,
 
@@ -88,14 +88,6 @@ impl MultiRaft {
         self.groups.get(&group_id)?.peer_ack_count(peer)
     }
 
-    /// Whether this node leads `group_id` by an election against other
-    /// voters that no leadership transfer started.
-    pub fn leads_by_contested_election(&self, group_id: u64) -> bool {
-        self.groups
-            .get(&group_id)
-            .is_some_and(|node| node.leads_by_contested_election())
-    }
-
     /// Whether `peer` holds every entry this leader has committed in
     /// `group_id`, and did not ask for a snapshot last. `false` when this
     /// node does not lead the group.
 
@@ -123,8 +123,8 @@ impl MultiRaft {
     /// changed since the last persist. Runs under the `MultiRaft` lock and
     /// never waits on disk. Before an RPC reply that granted a vote or bumped
     /// the term leaves this node, the caller awaits the group's
-    /// [`MultiRaft::durability_ticket`], so a restart cannot forget the vote
-    /// and let two leaders form.
+    /// [`MultiRaft::reply_ticket`], so a restart cannot forget the vote and
+    /// let two leaders form.
     ///
     /// No-op when the group is not mounted on this node.
     pub fn persist_group_hard_state(&mut self, group_id: u64) -> Result<()> {
@@ -134,6 +134,27 @@ impl MultiRaft {
         Ok(())
     }
 
+    /// The sequence number of the last write `group_id` staged, or 0 when the
+    /// group is not mounted. A reply handler reads it before its Raft call
+    /// and passes it to [`Self::reply_ticket`].
+    pub fn staged_through(&self, group_id: u64) -> u64 {
+        self.groups
+            .get(&group_id)
+            .map_or(0, |node| node.storage().staged_through())
+    }
+
+    /// A ticket for the writes a reply depends on: the group's latest hard
+    /// state, and every write staged after `mark` when there is one. `None`
+    /// when those are durable or the group is not mounted. Writes the apply
+    /// loop staged before `mark` hold no reply back.
+    pub fn reply_ticket(
+        &self,
+        group_id: u64,
+        mark: u64,
+    ) -> Option<crate::group_disk::DurabilityTicket> {
+        self.groups.get(&group_id)?.storage().reply_ticket(mark)
+    }
+
     /// Get the current term and snapshot metadata for a group (for building
     /// InstallSnapshot RPCs).
     pub fn snapshot_metadata(&self, group_id: u64) -> Result<(u64, u64, u64)> {