fix(raft): fence read indexes

consensus/cloud9-raft/src/lib.rs

@@ -172,6 +172,34 @@ impl fmt::Display for ProposeError {
 
 impl std::error::Error for ProposeError {}
 
+/// Error when requesting a linearizable read index.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ReadIndexError {
+    /// This node is not the leader.
+    NotLeader { leader_hint: Option<NodeId> },
+    /// The leader has not committed an entry from its current term yet.
+    CurrentTermNotCommitted,
+    /// A previous read-index quorum round is still pending.
+    ReadInProgress { read_index: LogIndex },
+}
+
+impl fmt::Display for ReadIndexError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::NotLeader { leader_hint: Some(id) } => write!(f, "not leader, try {id}"),
+            Self::NotLeader { leader_hint: None } => write!(f, "not leader, leader unknown"),
+            Self::CurrentTermNotCommitted => {
+                write!(f, "leader has not committed an entry from its current term")
+            }
+            Self::ReadInProgress { read_index } => {
+                write!(f, "read-index quorum round pending at index {read_index}")
+            }
+        }
+    }
+}
+
+impl std::error::Error for ReadIndexError {}
+
 /// Error when transferring leadership.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TransferError {

consensus/cloud9-raft/src/raft/event.rs

@@ -58,6 +58,10 @@ pub enum Payload {
     InstallSnapshotRequest(InstallSnapshotRequest),
     /// `InstallSnapshot` response
     InstallSnapshotResponse(InstallSnapshotResponse),
+    /// Read-index heartbeat request (§6.4).
+    ReadIndexRequest(ReadIndexRequest),
+    /// Read-index heartbeat response (§6.4).
+    ReadIndexResponse(ReadIndexResponse),
     /// Leadership transfer: target should start election immediately (§3.10)
     TimeoutNow,
 }
@@ -107,6 +111,18 @@ pub struct AppendResponse {
     pub last_log_index: LogIndex,
 }
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct ReadIndexRequest {
+    pub id: u64,
+    pub read_index: LogIndex,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct ReadIndexResponse {
+    pub id: u64,
+    pub read_index: LogIndex,
+}
+
 /// `InstallSnapshot` request (§5, Figure 5.3).
 ///
 /// Sent by the leader when a follower is too far behind and needs the snapshot

consensus/cloud9-raft/src/raft/follower.rs

@@ -15,7 +15,8 @@ use super::StepResult;
 use super::core::Core;
 use super::event::{
     AppendRequest, AppendResponse, Effects, Event, InstallSnapshotRequest, InstallSnapshotResponse,
-    Message, Payload, PreVoteRequest, PreVoteResponse, VoteRequest, VoteResponse,
+    Message, Payload, PreVoteRequest, PreVoteResponse, ReadIndexRequest, ReadIndexResponse,
+    VoteRequest, VoteResponse,
 };
 use super::log::Entry;
 
@@ -76,7 +77,10 @@ impl Follower {
                 if msg.term > core.term() {
                     term_updated = core.maybe_update_term(msg.term);
                     self.reset_deadline(core);
-                    if matches!(msg.payload, Payload::AppendRequest(_)) {
+                    if matches!(
+                        msg.payload,
+                        Payload::AppendRequest(_) | Payload::ReadIndexRequest(_)
+                    ) {
                         self.leader = Some(msg.from);
                     }
                 }
@@ -89,6 +93,9 @@ impl Follower {
                 let StepResult { transition, mut effects } = match msg.payload {
                     Payload::VoteRequest(req) => self.handle_vote_request(core, msg.from, req),
                     Payload::AppendRequest(req) => self.handle_append_request(core, msg.from, &req),
+                    Payload::ReadIndexRequest(req) => {
+                        self.handle_read_index_request(core, msg.from, req)
+                    }
                     Payload::InstallSnapshotRequest(req) => {
                         self.handle_install_snapshot(core, msg.from, req)
                     }
@@ -210,6 +217,29 @@ impl Follower {
         StepResult::stay(effects.with_message(resp))
     }
 
+    /// Handle read-index heartbeat request (§6.4).
+    fn handle_read_index_request(
+        &mut self,
+        core: &mut Core,
+        from: NodeId,
+        req: ReadIndexRequest,
+    ) -> StepResult {
+        self.leader = Some(from);
+        self.record_contact(core);
+        self.reset_deadline(core);
+
+        let resp = Message {
+            from: core.id(),
+            to: from,
+            term: core.term(),
+            payload: Payload::ReadIndexResponse(ReadIndexResponse {
+                id: req.id,
+                read_index: req.read_index,
+            }),
+        };
+        StepResult::stay(Effects::none().with_message(resp))
+    }
+
     /// Handle `InstallSnapshot` RPC (§5, Figure 5.3).
     ///
     /// Per Figure 5.3 receiver implementation:
@@ -294,6 +324,10 @@ impl Follower {
                 success: false,
                 last_log_index: core.log().last_index(),
             }),
+            Payload::ReadIndexRequest(req) => Payload::ReadIndexResponse(ReadIndexResponse {
+                id: req.id,
+                read_index: req.read_index,
+            }),
             _ => return Effects::none(),
         };
         Effects::none().with_message(Message {