feat(cluster): add closed timestamp tracker and follower read gate

Introduce ClosedTimestampTracker for advancing the closed timestamp
used by bounded-staleness reads, and FollowerReadGate with ReadLevel
to decide whether a follower can serve a read at a given LSN without
forwarding to the leader.

Author: farhan-syah

nodedb-cluster/src/closed_timestamp.rs

@@ -0,0 +1,123 @@
+//! Per-group closed-timestamp tracker.
+//!
+//! Every time a Raft group applies a committed entry, the applier
+//! records the wall-clock instant as that group's "closed timestamp".
+//! A follower whose closed timestamp for a group is within the
+//! caller's staleness bound can serve reads locally — no gateway hop
+//! to the leader.
+//!
+//! The tracker is intentionally simple: one `Instant` per group,
+//! updated monotonically. There is no HLC or cross-node coordination
+//! here — the closed timestamp is local to this node. Safety comes
+//! from the fact that a follower's applied index can only advance
+//! (Raft guarantees), so a read served at a given closed timestamp
+//! sees a consistent prefix of the log.
+
+use std::collections::HashMap;
+use std::sync::RwLock;
+use std::time::{Duration, Instant};
+
+/// Tracks the most recent apply instant per Raft group.
+pub struct ClosedTimestampTracker {
+    groups: RwLock<HashMap<u64, Instant>>,
+}
+
+impl ClosedTimestampTracker {
+    pub fn new() -> Self {
+        Self {
+            groups: RwLock::new(HashMap::new()),
+        }
+    }
+
+    /// Record that `group_id` just applied one or more entries.
+    /// Called by the raft-loop applier after each apply batch.
+    pub fn mark_applied(&self, group_id: u64) {
+        let mut g = self.groups.write().unwrap_or_else(|p| p.into_inner());
+        g.insert(group_id, Instant::now());
+    }
+
+    /// Record that `group_id` just applied, using a caller-supplied
+    /// instant. Exposed for deterministic testing with paused time.
+    pub fn mark_applied_at(&self, group_id: u64, at: Instant) {
+        let mut g = self.groups.write().unwrap_or_else(|p| p.into_inner());
+        g.insert(group_id, at);
+    }
+
+    /// Check whether this node's replica of `group_id` has applied
+    /// recently enough that a read with `max_staleness` can be
+    /// served locally.
+    ///
+    /// Returns `false` if the group has never applied on this node
+    /// (no closed timestamp recorded).
+    pub fn is_fresh_enough(&self, group_id: u64, max_staleness: Duration) -> bool {
+        let g = self.groups.read().unwrap_or_else(|p| p.into_inner());
+        match g.get(&group_id) {
+            Some(last) => last.elapsed() <= max_staleness,
+            None => false,
+        }
+    }
+
+    /// Return the age of the closed timestamp for a group, or `None`
+    /// if the group has never applied on this node. Useful for
+    /// observability (metrics, SHOW commands).
+    pub fn staleness(&self, group_id: u64) -> Option<Duration> {
+        let g = self.groups.read().unwrap_or_else(|p| p.into_inner());
+        g.get(&group_id).map(|last| last.elapsed())
+    }
+}
+
+impl Default for ClosedTimestampTracker {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn unknown_group_is_not_fresh() {
+        let tracker = ClosedTimestampTracker::new();
+        assert!(!tracker.is_fresh_enough(99, Duration::from_secs(10)));
+    }
+
+    #[test]
+    fn recently_applied_is_fresh() {
+        let tracker = ClosedTimestampTracker::new();
+        tracker.mark_applied(1);
+        assert!(tracker.is_fresh_enough(1, Duration::from_secs(5)));
+    }
+
+    #[test]
+    fn stale_group_is_not_fresh() {
+        let tracker = ClosedTimestampTracker::new();
+        let old = Instant::now() - Duration::from_secs(30);
+        tracker.mark_applied_at(1, old);
+        assert!(!tracker.is_fresh_enough(1, Duration::from_secs(5)));
+    }
+
+    #[test]
+    fn staleness_returns_none_for_unknown() {
+        let tracker = ClosedTimestampTracker::new();
+        assert!(tracker.staleness(42).is_none());
+    }
+
+    #[test]
+    fn staleness_returns_age_for_known() {
+        let tracker = ClosedTimestampTracker::new();
+        tracker.mark_applied(1);
+        let s = tracker.staleness(1).unwrap();
+        assert!(s < Duration::from_millis(100));
+    }
+
+    #[test]
+    fn mark_applied_updates_monotonically() {
+        let tracker = ClosedTimestampTracker::new();
+        let old = Instant::now() - Duration::from_secs(10);
+        tracker.mark_applied_at(1, old);
+        assert!(!tracker.is_fresh_enough(1, Duration::from_secs(5)));
+        tracker.mark_applied(1);
+        assert!(tracker.is_fresh_enough(1, Duration::from_secs(5)));
+    }
+}

nodedb-cluster/src/follower_read.rs

@@ -0,0 +1,130 @@
+//! Follower-read decision gate.
+//!
+//! [`FollowerReadGate`] answers a single question: "given the
+//! session's `ReadConsistency` and the local node's role + closed
+//! timestamp for the target Raft group, can this read be served
+//! locally without forwarding to the leader?"
+//!
+//! ## Decision table
+//!
+//! | Consistency           | Local role  | Closed TS fresh? | Serve locally? |
+//! |-----------------------|-------------|------------------|----------------|
+//! | Strong                | *           | *                | Only if leader |
+//! | BoundedStaleness(d)   | Follower    | ≤ d              | Yes            |
+//! | BoundedStaleness(d)   | Follower    | > d              | No → forward   |
+//! | BoundedStaleness(d)   | Leader      | *                | Yes            |
+//! | Eventual              | *           | *                | Yes            |
+//!
+//! The gate is stateless — it reads from shared handles to the
+//! closed-timestamp tracker and the raft-status provider.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::closed_timestamp::ClosedTimestampTracker;
+
+/// Consistency level for a single read — mirrors the `ReadConsistency`
+/// enum in the `nodedb` crate without coupling `nodedb-cluster` to it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReadLevel {
+    Strong,
+    BoundedStaleness(Duration),
+    Eventual,
+}
+
+/// Answers "can this read be served locally?"
+pub struct FollowerReadGate {
+    closed_ts: Arc<ClosedTimestampTracker>,
+    /// Type-erased function that returns true if this node is the
+    /// leader for the given group. Injection seam — production wraps
+    /// `MultiRaft::group_statuses`, tests supply a closure.
+    is_leader_fn: Box<dyn Fn(u64) -> bool + Send + Sync>,
+}
+
+impl FollowerReadGate {
+    pub fn new(
+        closed_ts: Arc<ClosedTimestampTracker>,
+        is_leader_fn: Box<dyn Fn(u64) -> bool + Send + Sync>,
+    ) -> Self {
+        Self {
+            closed_ts,
+            is_leader_fn,
+        }
+    }
+
+    /// Returns `true` if the read can be served from this node's
+    /// local replica without forwarding to the leader.
+    pub fn can_serve_locally(&self, group_id: u64, level: ReadLevel) -> bool {
+        match level {
+            ReadLevel::Strong => (self.is_leader_fn)(group_id),
+            ReadLevel::Eventual => true,
+            ReadLevel::BoundedStaleness(max) => {
+                if (self.is_leader_fn)(group_id) {
+                    return true;
+                }
+                self.closed_ts.is_fresh_enough(group_id, max)
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn gate(leader_groups: &'static [u64]) -> FollowerReadGate {
+        FollowerReadGate::new(
+            Arc::new(ClosedTimestampTracker::new()),
+            Box::new(move |gid| leader_groups.contains(&gid)),
+        )
+    }
+
+    fn gate_with_tracker(
+        leader_groups: &'static [u64],
+        tracker: Arc<ClosedTimestampTracker>,
+    ) -> FollowerReadGate {
+        FollowerReadGate::new(tracker, Box::new(move |gid| leader_groups.contains(&gid)))
+    }
+
+    #[test]
+    fn strong_requires_leader() {
+        let g = gate(&[1]);
+        assert!(g.can_serve_locally(1, ReadLevel::Strong));
+        assert!(!g.can_serve_locally(2, ReadLevel::Strong));
+    }
+
+    #[test]
+    fn eventual_always_local() {
+        let g = gate(&[]);
+        assert!(g.can_serve_locally(99, ReadLevel::Eventual));
+    }
+
+    #[test]
+    fn bounded_staleness_leader_always_local() {
+        let g = gate(&[1]);
+        assert!(g.can_serve_locally(1, ReadLevel::BoundedStaleness(Duration::from_secs(5))));
+    }
+
+    #[test]
+    fn bounded_staleness_follower_fresh_enough() {
+        let tracker = Arc::new(ClosedTimestampTracker::new());
+        tracker.mark_applied(2);
+        let g = gate_with_tracker(&[], tracker);
+        assert!(g.can_serve_locally(2, ReadLevel::BoundedStaleness(Duration::from_secs(5))));
+    }
+
+    #[test]
+    fn bounded_staleness_follower_too_stale() {
+        let tracker = Arc::new(ClosedTimestampTracker::new());
+        let old = std::time::Instant::now() - Duration::from_secs(30);
+        tracker.mark_applied_at(2, old);
+        let g = gate_with_tracker(&[], tracker);
+        assert!(!g.can_serve_locally(2, ReadLevel::BoundedStaleness(Duration::from_secs(5))));
+    }
+
+    #[test]
+    fn bounded_staleness_unknown_group_not_local() {
+        let g = gate(&[]);
+        assert!(!g.can_serve_locally(99, ReadLevel::BoundedStaleness(Duration::from_secs(5))));
+    }
+}

nodedb-cluster/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod bootstrap;
 pub mod catalog;
 pub mod circuit_breaker;
+pub mod closed_timestamp;
 pub mod cluster_info;
 pub mod conf_change;
 pub mod cross_shard_txn;
@@ -12,6 +13,7 @@ pub mod distributed_spatial;
 pub mod distributed_timeseries;
 pub mod distributed_vector;
 pub mod error;
+pub mod follower_read;
 pub mod forward;
 pub mod ghost;
 pub mod ghost_sweeper;
@@ -43,6 +45,7 @@ pub mod wire;
 
 pub use bootstrap::{ClusterConfig, ClusterState, JoinRetryPolicy, start_cluster};
 pub use catalog::ClusterCatalog;
+pub use closed_timestamp::ClosedTimestampTracker;
 pub use cluster_info::{
     ClusterInfoSnapshot, ClusterObserver, GroupSnapshot, GroupStatusProvider, PeerSnapshot,
 };
@@ -52,6 +55,7 @@ pub use decommission::{
     DecommissionSafetyError, MetadataProposer, check_can_decommission, plan_full_decommission,
 };
 pub use error::{ClusterError, Result};
+pub use follower_read::{FollowerReadGate, ReadLevel};
 pub use forward::{NoopPlanExecutor, PlanExecutor};
 pub use ghost::{GhostStub, GhostTable};
 pub use health::{HealthConfig, HealthMonitor};