hydracache_server/
cluster_status.rs

1//! Honest cluster-status seam for the admin and Management Center surfaces.
2
3use std::fmt;
4use std::sync::Arc;
5use std::time::Duration;
6
7use hydracache::{ClusterMember, ClusterNodeId, ClusterRole, RaftMetadataSnapshot};
8use serde::Serialize;
9use thiserror::Error;
10
11/// Runtime state supplied by the server around the cluster-status provider.
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub struct ClusterStatusRuntime {
14    /// Whether the server runtime is ready to serve.
15    pub ready: bool,
16    /// Whether the server runtime is draining.
17    pub draining: bool,
18}
19
20impl ClusterStatusRuntime {
21    /// Build a runtime status input.
22    pub fn new(ready: bool, draining: bool) -> Self {
23        Self { ready, draining }
24    }
25}
26
27/// Where a status reading came from.
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
29#[serde(rename_all = "snake_case")]
30pub enum StatusSource {
31    /// Status came from a live grid/control-plane handle.
32    Live,
33    /// Status is the local daemon model and must not be painted as live.
34    Modeled,
35}
36
37/// Role rendered for a cluster member in management views.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
39#[serde(rename_all = "snake_case")]
40pub enum MemberRole {
41    /// Local, non-clustered runtime.
42    Local,
43    /// Client near-cache runtime.
44    Client,
45    /// Voting/cache member runtime.
46    Member,
47}
48
49impl From<ClusterRole> for MemberRole {
50    fn from(value: ClusterRole) -> Self {
51        match value {
52            ClusterRole::Local => Self::Local,
53            ClusterRole::Client => Self::Client,
54            ClusterRole::Member => Self::Member,
55        }
56    }
57}
58
59/// Reachability state reported for a known member.
60#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
61#[serde(rename_all = "snake_case")]
62pub enum Reachability {
63    /// Member is currently reachable.
64    Reachable,
65    /// Member is suspected but not yet declared unreachable.
66    Suspect,
67    /// Member is unreachable and must remain visible in the view.
68    Unreachable,
69}
70
71/// Coarse reshard lifecycle phase for read-only status surfaces.
72#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
73#[serde(rename_all = "snake_case")]
74pub enum ReshardPhase {
75    /// No reshard is active.
76    Idle,
77    /// A reshard is being planned.
78    Planning,
79    /// Partitions or replicas are moving.
80    Moving,
81    /// The reshard is finalizing.
82    Finalizing,
83}
84
85impl fmt::Display for ReshardPhase {
86    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
87        let value = match self {
88            Self::Idle => "idle",
89            Self::Planning => "planning",
90            Self::Moving => "moving",
91            Self::Finalizing => "finalizing",
92        };
93        formatter.write_str(value)
94    }
95}
96
97/// Read-only member status shown by the Management Center.
98#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
99pub struct MemberStatus {
100    /// Stable logical node id.
101    pub node_id: String,
102    /// Runtime role.
103    pub role: MemberRole,
104    /// Current reachability.
105    pub reachable: Reachability,
106    /// Process generation.
107    pub generation: u64,
108}
109
110/// Read-only cluster status snapshot.
111#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
112pub struct ClusterStatus {
113    /// Whether the snapshot is live or modeled.
114    pub source: StatusSource,
115    /// Current leader id, if a live raft source knows one.
116    pub leader: Option<String>,
117    /// Current control-plane term.
118    pub term: u64,
119    /// Current authority epoch.
120    pub epoch: u64,
121    /// Whether quorum is available while the runtime is not draining.
122    pub quorum_ok: bool,
123    /// Visible members. Unreachable members remain present.
124    pub members: Vec<MemberStatus>,
125    /// Current raft voter count, if known.
126    pub voters: u32,
127    /// Sorted current Raft voter ids, when the live runtime exposes them.
128    pub voter_ids: Vec<u64>,
129    /// Current reshard phase.
130    pub reshard_phase: ReshardPhase,
131    /// Whether the runtime is draining.
132    pub draining: bool,
133}
134
135/// Read-only state of the disk-backed Raft compaction control.
136#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
137pub struct RaftCompactionStatus {
138    /// Whether the current runtime owns a disk-backed Raft log.
139    pub available: bool,
140    /// Whether explicit compaction requests are enabled by configuration.
141    pub enabled: bool,
142    /// Last locally applied Raft index, when available.
143    pub applied_index: Option<u64>,
144    /// Durable snapshot index, when available.
145    pub snapshot_index: Option<u64>,
146    /// First retained durable log index, when available.
147    pub first_log_index: Option<u64>,
148    /// Last durable log index, when available.
149    pub last_log_index: Option<u64>,
150    /// Real HTTP snapshot send attempts since this daemon started.
151    pub snapshot_send_attempts: Option<u64>,
152    /// Successful real HTTP snapshot sends since this daemon started.
153    pub snapshot_send_successes: Option<u64>,
154    /// Failed or timed-out real HTTP snapshot sends since this daemon started.
155    pub snapshot_send_failures: Option<u64>,
156    /// Real HTTP snapshot requests currently awaiting an outcome.
157    pub snapshot_sends_in_flight: Option<u64>,
158    /// Actual per-peer sender tasks currently carrying a valid Raft snapshot.
159    pub snapshot_sender_tasks_current: Option<u64>,
160    /// Per-daemon high-water mark of concurrent snapshot sender tasks since
161    /// this process started; it resets when the daemon restarts.
162    pub snapshot_sender_tasks_high_water: Option<u64>,
163    /// Snapshots installed into the local Raft state machine this process run.
164    pub snapshot_installs: Option<u64>,
165}
166
167impl RaftCompactionStatus {
168    pub(crate) fn unavailable() -> Self {
169        Self {
170            available: false,
171            enabled: false,
172            applied_index: None,
173            snapshot_index: None,
174            first_log_index: None,
175            last_log_index: None,
176            snapshot_send_attempts: None,
177            snapshot_send_successes: None,
178            snapshot_send_failures: None,
179            snapshot_sends_in_flight: None,
180            snapshot_sender_tasks_current: None,
181            snapshot_sender_tasks_high_water: None,
182            snapshot_installs: None,
183        }
184    }
185}
186
187/// Fail-loud rejection from the narrow Raft compaction control.
188#[derive(Debug, Clone, PartialEq, Eq, Error)]
189pub enum RaftCompactionError {
190    /// The current role/runtime has no disk-backed Raft log.
191    #[error("raft compaction control is unavailable for this runtime")]
192    Unavailable,
193    /// The control exists but was not explicitly enabled.
194    #[error("raft compaction control is disabled; set HYDRACACHE_RAFT_COMPACTION=true explicitly")]
195    Disabled,
196    /// The Raft runtime or durable store rejected the operation.
197    #[error("raft compaction failed: {0}")]
198    Runtime(String),
199}
200
201/// Read-only provider of cluster status.
202pub trait ClusterStatusProvider: fmt::Debug + Send + Sync {
203    /// Notify the provider that the server has started graceful drain.
204    fn begin_drain(&self) {}
205
206    /// Wait until the control plane has durably removed the local voter.
207    ///
208    /// Modeled and non-voting providers are ready immediately. A networked
209    /// member overrides this seam after its topology leave has committed.
210    fn wait_for_drain_ready(&self, _timeout: Duration) -> bool {
211        true
212    }
213
214    /// Return a cluster status snapshot, incorporating current runtime state.
215    fn cluster_status(&self, runtime: ClusterStatusRuntime) -> ClusterStatus;
216}
217
218/// Modeled status used when the daemon has no live grid handle.
219#[derive(Debug, Clone, Default)]
220pub struct ModeledClusterStatus;
221
222impl ClusterStatusProvider for ModeledClusterStatus {
223    fn cluster_status(&self, runtime: ClusterStatusRuntime) -> ClusterStatus {
224        ClusterStatus {
225            source: StatusSource::Modeled,
226            leader: runtime.ready.then(|| "local".to_owned()),
227            term: u64::from(runtime.ready),
228            epoch: 0,
229            quorum_ok: runtime.ready && !runtime.draining,
230            members: Vec::new(),
231            voters: 0,
232            voter_ids: Vec::new(),
233            reshard_phase: ReshardPhase::Idle,
234            draining: runtime.draining,
235        }
236    }
237}
238
239/// Minimal read-only handle over a live grid/control-plane.
240pub trait GridControlPlaneHandle: fmt::Debug + Send + Sync {
241    /// Notify the live grid that graceful drain started.
242    fn begin_drain(&self);
243
244    /// Wait until any voting membership transition required by drain is
245    /// locally applied.
246    fn wait_for_drain_ready(&self, _timeout: Duration) -> bool {
247        true
248    }
249
250    /// Return a point-in-time control-plane snapshot.
251    fn snapshot(&self) -> RaftMetadataSnapshot;
252    /// Return visible members.
253    fn members(&self) -> Vec<ClusterMember>;
254    /// Return the raft soft-state leader id, if known.
255    fn raft_leader_id(&self) -> Option<String>;
256    /// Return whether the live grid currently has quorum.
257    fn has_quorum(&self) -> bool;
258    /// Return whether `observed` is still the fully applied local metadata view.
259    ///
260    /// Networked followers must fence authority while their committed index is
261    /// ahead of the locally applied metadata state. The observed-snapshot
262    /// argument also prevents a projection assembled across an apply boundary
263    /// from being published as authoritative.
264    fn metadata_authority_matches(&self, observed: &RaftMetadataSnapshot) -> bool {
265        let _ = observed;
266        true
267    }
268    /// Return current raft voter count.
269    fn voter_count(&self) -> u32;
270    /// Return sorted current Raft voter ids.
271    fn voter_ids(&self) -> Vec<u64>;
272    /// Return reachability for one known node.
273    fn reachability(&self, node: &ClusterNodeId) -> Reachability;
274    /// Return the current reshard phase.
275    fn reshard_phase(&self) -> ReshardPhase;
276    /// Return whether the grid itself is draining.
277    fn is_draining(&self) -> bool;
278
279    /// Return disk-backed Raft compaction progress and enablement.
280    fn raft_compaction_status(&self) -> Result<RaftCompactionStatus, RaftCompactionError> {
281        Ok(RaftCompactionStatus::unavailable())
282    }
283
284    /// Compact the durable Raft log exactly at current applied progress.
285    fn compact_raft_log_at_applied(&self) -> Result<RaftCompactionStatus, RaftCompactionError> {
286        Err(RaftCompactionError::Unavailable)
287    }
288}
289
290/// Live status backed by a grid/control-plane handle.
291#[derive(Debug, Clone)]
292pub struct LiveClusterStatus {
293    grid: Arc<dyn GridControlPlaneHandle>,
294}
295
296impl LiveClusterStatus {
297    /// Build a live status provider.
298    pub fn new(grid: Arc<dyn GridControlPlaneHandle>) -> Self {
299        Self { grid }
300    }
301}
302
303impl ClusterStatusProvider for LiveClusterStatus {
304    fn begin_drain(&self) {
305        self.grid.begin_drain();
306    }
307
308    fn wait_for_drain_ready(&self, timeout: Duration) -> bool {
309        self.grid.wait_for_drain_ready(timeout)
310    }
311
312    fn cluster_status(&self, runtime: ClusterStatusRuntime) -> ClusterStatus {
313        let snapshot = self.grid.snapshot();
314        let draining = runtime.draining || self.grid.is_draining();
315        let members = self
316            .grid
317            .members()
318            .into_iter()
319            .map(|member| MemberStatus {
320                node_id: member.node_id.to_string(),
321                role: MemberRole::from(member.role),
322                reachable: self.grid.reachability(&member.node_id),
323                generation: member.generation.value(),
324            })
325            .collect();
326        let metadata_authoritative = self.grid.metadata_authority_matches(&snapshot);
327
328        ClusterStatus {
329            source: StatusSource::Live,
330            leader: metadata_authoritative
331                .then(|| self.grid.raft_leader_id())
332                .flatten(),
333            term: snapshot.term,
334            epoch: snapshot.epoch.value(),
335            quorum_ok: runtime.ready
336                && metadata_authoritative
337                && self.grid.has_quorum()
338                && !draining,
339            members,
340            voters: self.grid.voter_count(),
341            voter_ids: self.grid.voter_ids(),
342            reshard_phase: self.grid.reshard_phase(),
343            draining,
344        }
345    }
346}
hydracache_server/cluster_status.rs

hydracache_server/
cluster_status.rs
