hydracache_server/
bootstrap.rs

1use hydracache::{ClusterGridCounters, HydraCache};
2use hydracache_client_transport_axum::{ClientSurfaceDrain, ClientSurfaceRuntime};
3use hydracache_observability::{
4    ClusterMemberView, ClusterOverview, ClusterTopologyOverview, ConsistencyView,
5    HydraCacheRegistry, LeaderView, LifecycleView, PartitionSummary, TopologyReshardPhase,
6    TopologyStatusSource,
7};
8use hydracache_redis_compat::{RedisListenerConfig, RedisRespServer, RedisServeError};
9use serde::Serialize;
10use std::net::SocketAddr;
11use std::sync::Arc;
12use thiserror::Error;
13
14use crate::cluster_status::{
15    ClusterStatus, ClusterStatusProvider, ClusterStatusRuntime, LiveClusterStatus, MemberRole,
16    ModeledClusterStatus, Reachability, ReshardPhase, StatusSource,
17};
18use crate::config::{ServerConfig, ServerConfigError, ServerRole};
19use crate::redis_tcp::{RedisTlsAcceptor, RedisTlsError};
20use crate::services::{DrainOutcome, GracefulShutdown, ServiceSet};
21
22/// Runtime state exposed by health/readiness checks.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
24#[serde(rename_all = "snake_case")]
25pub enum ServerState {
26    /// Runtime has not started.
27    Created,
28    /// Runtime accepts requests.
29    Running,
30    /// Runtime is draining in-flight work.
31    Draining,
32    /// Runtime stopped cleanly.
33    Stopped,
34}
35
36/// Liveness response.
37#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
38pub struct ServerHealth {
39    /// Stable status field for probes.
40    pub status: &'static str,
41    /// Current runtime state.
42    pub state: ServerState,
43}
44
45/// Readiness response.
46#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
47pub struct ServerReadiness {
48    /// Whether the daemon can serve traffic.
49    pub ready: bool,
50    /// Whether durable storage has opened.
51    pub storage_open: bool,
52    /// Whether the configured cluster role is ready.
53    pub cluster_ready: bool,
54    /// Whether listeners are accepting new work.
55    pub accepting: bool,
56    /// Whether the external client surface is accepting work.
57    pub client_surface_ready: bool,
58    /// Whether the optional Redis RESP surface is accepting work.
59    pub redis_surface_ready: bool,
60}
61
62/// Admin status consumed by the Kubernetes operator.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
64pub struct ServerAdminStatus {
65    /// Whether this status is live or modeled.
66    pub source: StatusSource,
67    /// Current leader id if known to the runtime.
68    pub leader: Option<String>,
69    /// Current control-plane term if known.
70    pub term: u64,
71    /// Whether the runtime believes quorum is available.
72    pub quorum_ok: bool,
73    /// Observed member count.
74    pub members: u32,
75    /// Observed raft voter count.
76    pub voters: u32,
77    /// Current reshard phase.
78    pub reshard_phase: String,
79    /// Whether the runtime is draining.
80    pub draining: bool,
81}
82
83/// Accepted admin action response.
84#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
85pub struct ServerAdminAction {
86    /// Stable action name.
87    pub action: &'static str,
88    /// Stable outcome string.
89    pub outcome: &'static str,
90    /// Human-readable detail, safe for operator Conditions.
91    pub detail: String,
92}
93
94/// Modeled Redis RESP listener runtime.
95#[derive(Debug, Clone, PartialEq, Eq)]
96pub struct RedisSurfaceRuntime {
97    accepting: bool,
98    active_connections: u64,
99}
100
101impl RedisSurfaceRuntime {
102    fn new() -> Self {
103        Self {
104            accepting: false,
105            active_connections: 0,
106        }
107    }
108
109    fn start(&mut self) {
110        self.accepting = true;
111    }
112
113    fn accepting(&self) -> bool {
114        self.accepting
115    }
116
117    fn begin_connection(&mut self) -> bool {
118        if !self.accepting {
119            return false;
120        }
121        self.active_connections = self.active_connections.saturating_add(1);
122        true
123    }
124
125    fn finish_connection(&mut self) {
126        self.active_connections = self.active_connections.saturating_sub(1);
127    }
128
129    fn active_connections(&self) -> u64 {
130        self.active_connections
131    }
132
133    fn shutdown(&mut self) -> RedisSurfaceDrain {
134        self.accepting = false;
135        let started_with = self.active_connections;
136        self.active_connections = 0;
137        RedisSurfaceDrain {
138            started_with,
139            remaining: self.active_connections,
140        }
141    }
142}
143
144/// Result of draining modeled Redis RESP connections.
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
146pub struct RedisSurfaceDrain {
147    /// Active RESP connections observed when drain started.
148    pub started_with: u64,
149    /// Active RESP connections after drain.
150    pub remaining: u64,
151}
152
153/// Additional read-only observability signals supplied by the daemon host.
154#[derive(Debug, Clone, PartialEq, Eq)]
155pub struct ServerObservabilityModel {
156    cluster_grid: ClusterGridCounters,
157    partition_count: u64,
158    configured_default_consistency: Option<String>,
159    backup_age_seconds: Option<u64>,
160    upgrade_phase: String,
161}
162
163impl ServerObservabilityModel {
164    /// Attach aggregate grid counters supplied by the hosting runtime.
165    pub fn with_cluster_grid_counters(mut self, counters: ClusterGridCounters) -> Self {
166        self.cluster_grid = counters;
167        self
168    }
169
170    /// Attach the effective partition count.
171    pub fn with_partition_count(mut self, count: u64) -> Self {
172        self.partition_count = count;
173        self
174    }
175
176    /// Attach the configured default consistency label.
177    pub fn with_configured_default_consistency(mut self, level: impl Into<String>) -> Self {
178        self.configured_default_consistency = Some(level.into());
179        self
180    }
181
182    /// Attach the worst known backup age.
183    pub fn with_backup_age_seconds(mut self, seconds: u64) -> Self {
184        self.backup_age_seconds = Some(seconds);
185        self
186    }
187
188    /// Attach backup ages for namespaces, keeping the oldest/worst age.
189    pub fn with_backup_age_seconds_from_namespaces(
190        mut self,
191        ages: impl IntoIterator<Item = u64>,
192    ) -> Self {
193        self.backup_age_seconds = ages.into_iter().max();
194        self
195    }
196
197    /// Attach the current graceful-upgrade phase.
198    pub fn with_upgrade_phase(mut self, phase: impl Into<String>) -> Self {
199        self.upgrade_phase = phase.into();
200        self
201    }
202}
203
204impl Default for ServerObservabilityModel {
205    fn default() -> Self {
206        Self {
207            cluster_grid: ClusterGridCounters::default(),
208            partition_count: 0,
209            configured_default_consistency: None,
210            backup_age_seconds: None,
211            upgrade_phase: "idle".to_owned(),
212        }
213    }
214}
215
216/// Fail-loud admin action errors.
217#[derive(Debug, Clone, PartialEq, Eq, Error)]
218pub enum ServerAdminActionError {
219    /// The runtime is not ready to accept the requested action.
220    #[error("server is not ready for admin action: {0}")]
221    NotReady(&'static str),
222    /// The action requires member mode in the current server model.
223    #[error("{0} requires member mode")]
224    RequiresMember(&'static str),
225    /// Backup cannot run without configured backup support.
226    #[error("backup admin action requires backup.enabled and backup.location")]
227    BackupDisabled,
228}
229
230/// Standalone server runtime.
231#[derive(Debug, Clone)]
232pub struct ServerRuntime {
233    config: ServerConfig,
234    cache: HydraCache,
235    services: ServiceSet,
236    state: ServerState,
237    storage_open: bool,
238    cluster_ready: bool,
239    accepting: bool,
240    flushed: bool,
241    client_surface: Option<ClientSurfaceRuntime>,
242    redis_client_state: Option<Arc<hydracache_client_transport_axum::ClientSurfaceState>>,
243    redis_listener_config: Option<RedisListenerConfig>,
244    redis_surface: Option<RedisSurfaceRuntime>,
245    cluster_status: Arc<dyn ClusterStatusProvider>,
246    observability: ServerObservabilityModel,
247    last_client_surface_drain: Option<ClientSurfaceDrain>,
248    last_redis_surface_drain: Option<RedisSurfaceDrain>,
249    last_drain: Option<DrainOutcome>,
250}
251
252impl ServerRuntime {
253    /// Validate config and construct a runtime.
254    pub fn new(config: ServerConfig) -> Result<Self, ServerConfigError> {
255        config.validate()?;
256        let (cache, cluster_status): (HydraCache, Arc<dyn ClusterStatusProvider>) =
257            match config.role {
258                ServerRole::Member => {
259                    let (cache, grid) = crate::grid_host::build_member(&config)?;
260                    (cache, Arc::new(LiveClusterStatus::new(grid)))
261                }
262                ServerRole::Local | ServerRole::Client => {
263                    (HydraCache::local().build(), Arc::new(ModeledClusterStatus))
264                }
265            };
266        let client_surface = if config.client_api.enabled {
267            Some(
268                ClientSurfaceRuntime::new(config.client_api.limits)
269                    .map_err(|error| ServerConfigError::InvalidClientApi(error.to_string()))?,
270            )
271        } else {
272            None
273        };
274        let redis_client_state = if config.redis_api.enabled {
275            Some(match &client_surface {
276                Some(surface) => surface.state(),
277                None => Arc::new(
278                    hydracache_client_transport_axum::ClientSurfaceState::new(
279                        config.client_api.limits,
280                    )
281                    .map_err(|error| ServerConfigError::InvalidClientApi(error.to_string()))?,
282                ),
283            })
284        } else {
285            None
286        };
287        let redis_surface = if config.redis_api.enabled {
288            Some(RedisSurfaceRuntime::new())
289        } else {
290            None
291        };
292        let redis_listener_config = if config.redis_api.enabled {
293            Some(config.redis_listener_config()?)
294        } else {
295            None
296        };
297        Ok(Self {
298            config,
299            cache,
300            services: ServiceSet::default(),
301            state: ServerState::Created,
302            storage_open: false,
303            cluster_ready: false,
304            accepting: false,
305            flushed: false,
306            client_surface,
307            redis_client_state,
308            redis_listener_config,
309            redis_surface,
310            cluster_status,
311            observability: ServerObservabilityModel::default(),
312            last_client_surface_drain: None,
313            last_redis_surface_drain: None,
314            last_drain: None,
315        })
316    }
317
318    /// Override the cluster-status provider.
319    ///
320    /// This is the W0 seam used by tests and later by the member-role grid host.
321    pub fn with_cluster_status_provider(
322        mut self,
323        cluster_status: Arc<dyn ClusterStatusProvider>,
324    ) -> Self {
325        self.cluster_status = cluster_status;
326        self
327    }
328
329    /// Override additional read-only observability signals.
330    pub fn with_observability_model(mut self, observability: ServerObservabilityModel) -> Self {
331        self.observability = observability;
332        self
333    }
334
335    /// Start storage, cluster membership, listeners, and background services.
336    pub fn start(mut self) -> Self {
337        self.storage_open = true;
338        self.cluster_ready = matches!(
339            self.config.role,
340            ServerRole::Local | ServerRole::Member | ServerRole::Client
341        );
342        self.accepting = true;
343        if let Some(surface) = self.client_surface.as_mut() {
344            surface.start();
345        }
346        if let Some(surface) = self.redis_surface.as_mut() {
347            surface.start();
348        }
349        self.services.start();
350        self.state = ServerState::Running;
351        self
352    }
353
354    /// Return liveness.
355    pub fn health(&self) -> ServerHealth {
356        ServerHealth {
357            status: if self.state == ServerState::Stopped {
358                "stopped"
359            } else {
360                "ok"
361            },
362            state: self.state,
363        }
364    }
365
366    /// Return readiness.
367    pub fn ready(&self) -> ServerReadiness {
368        ServerReadiness {
369            ready: self.can_serve(),
370            storage_open: self.storage_open,
371            cluster_ready: self.cluster_ready,
372            accepting: self.accepting,
373            client_surface_ready: self.client_surface_ready(),
374            redis_surface_ready: self.redis_surface_ready(),
375        }
376    }
377
378    /// Return whether the runtime can serve traffic.
379    pub fn can_serve(&self) -> bool {
380        self.state == ServerState::Running
381            && self.storage_open
382            && self.cluster_ready
383            && self.accepting
384    }
385
386    /// Return whether the runtime is currently draining.
387    pub fn is_draining(&self) -> bool {
388        self.state == ServerState::Draining
389    }
390
391    /// Begin one in-flight request.
392    pub fn begin_request(&mut self) -> bool {
393        if !self.accepting {
394            return false;
395        }
396        self.services.begin_request();
397        true
398    }
399
400    /// Complete one in-flight request.
401    pub fn finish_request(&mut self) {
402        self.services.finish_request();
403    }
404
405    /// Return whether the external client surface is accepting work.
406    pub fn client_surface_ready(&self) -> bool {
407        self.client_surface
408            .as_ref()
409            .is_some_and(ClientSurfaceRuntime::accepting)
410    }
411
412    /// Begin a modeled client subscription stream.
413    pub fn begin_client_subscription(&self) -> bool {
414        self.client_surface
415            .as_ref()
416            .is_some_and(|surface| surface.begin_subscription().is_ok())
417    }
418
419    /// Return active modeled client subscription streams.
420    pub fn client_active_subscriptions(&self) -> u64 {
421        self.client_surface
422            .as_ref()
423            .map_or(0, |surface| surface.state().active_subscriptions())
424    }
425
426    /// Return the last client-surface drain result, if the surface is enabled.
427    pub fn client_surface_drain(&self) -> Option<ClientSurfaceDrain> {
428        self.last_client_surface_drain
429    }
430
431    /// Return whether the optional Redis RESP surface is accepting work.
432    pub fn redis_surface_ready(&self) -> bool {
433        self.redis_surface
434            .as_ref()
435            .is_some_and(RedisSurfaceRuntime::accepting)
436    }
437
438    /// Begin a modeled RESP connection.
439    pub fn begin_redis_connection(&mut self) -> bool {
440        self.redis_surface
441            .as_mut()
442            .is_some_and(RedisSurfaceRuntime::begin_connection)
443    }
444
445    /// Finish a modeled RESP connection.
446    pub fn finish_redis_connection(&mut self) {
447        if let Some(surface) = self.redis_surface.as_mut() {
448            surface.finish_connection();
449        }
450    }
451
452    /// Return active modeled RESP connections.
453    pub fn redis_active_connections(&self) -> u64 {
454        self.redis_surface
455            .as_ref()
456            .map_or(0, RedisSurfaceRuntime::active_connections)
457    }
458
459    /// Return the last Redis RESP surface drain result, if the surface is enabled.
460    pub fn redis_surface_drain(&self) -> Option<RedisSurfaceDrain> {
461        self.last_redis_surface_drain
462    }
463
464    /// Return the Redis RESP bind address when that optional listener is enabled.
465    pub fn redis_listener_addr(&self) -> Option<SocketAddr> {
466        self.redis_surface
467            .as_ref()
468            .map(|_| self.config.redis_api.listen_addr)
469    }
470
471    /// Build a Redis RESP executor using this runtime's shared client-surface state.
472    pub fn redis_resp_server(&self) -> Result<Option<RedisRespServer>, RedisServeError> {
473        let Some(state) = &self.redis_client_state else {
474            return Ok(None);
475        };
476        let Some(config) = &self.redis_listener_config else {
477            return Ok(None);
478        };
479        RedisRespServer::new(Arc::clone(state), config.clone()).map(Some)
480    }
481
482    /// Build the optional Redis TLS acceptor when rediss:// is enabled.
483    pub fn redis_tls_acceptor(&self) -> Result<Option<RedisTlsAcceptor>, RedisTlsError> {
484        if !self.config.redis_api.enabled || !self.config.redis_api.rediss_enabled {
485            return Ok(None);
486        }
487        RedisTlsAcceptor::from_tls_config(&self.config.tls).map(Some)
488    }
489
490    /// Stop accepting new work and enter the draining state.
491    pub fn begin_drain(&mut self) {
492        self.begin_local_drain();
493        self.cluster_status.begin_drain();
494    }
495
496    /// Accept an operator/admin drain request without stopping the daemon process.
497    pub fn request_admin_drain(&mut self) -> DrainOutcome {
498        if self.state == ServerState::Stopped {
499            return self.last_drain.unwrap_or(DrainOutcome {
500                started_with: 0,
501                remaining: 0,
502                timed_out: false,
503            });
504        }
505        self.begin_local_drain();
506        self.leave_cluster_for_shutdown();
507        self.cluster_status.begin_drain();
508        let outcome = GracefulShutdown::new(self.config.drain_timeout()).drain(&mut self.services);
509        self.last_drain = Some(outcome);
510        outcome
511    }
512
513    fn begin_local_drain(&mut self) {
514        if matches!(self.state, ServerState::Stopped) {
515            return;
516        }
517        self.accepting = false;
518        self.state = ServerState::Draining;
519        if let Some(surface) = self.client_surface.as_mut() {
520            if self
521                .last_client_surface_drain
522                .is_none_or(|drain| drain.remaining > 0)
523            {
524                self.last_client_surface_drain = Some(surface.shutdown());
525            }
526        }
527        if let Some(surface) = self.redis_surface.as_mut() {
528            if self
529                .last_redis_surface_drain
530                .is_none_or(|drain| drain.remaining > 0)
531            {
532                self.last_redis_surface_drain = Some(surface.shutdown());
533            }
534        }
535    }
536
537    /// Gracefully stop accepting, drain, flush, and stop services.
538    pub fn graceful_shutdown(&mut self) -> DrainOutcome {
539        if self.state == ServerState::Stopped {
540            return self.last_drain.unwrap_or(DrainOutcome {
541                started_with: 0,
542                remaining: 0,
543                timed_out: false,
544            });
545        }
546        self.begin_local_drain();
547        self.leave_cluster_for_shutdown();
548        self.cluster_status.begin_drain();
549        let outcome = GracefulShutdown::new(self.config.drain_timeout()).drain(&mut self.services);
550        self.flushed = true;
551        self.storage_open = false;
552        self.cluster_ready = false;
553        self.services.stop();
554        self.state = ServerState::Stopped;
555        self.last_drain = Some(outcome);
556        outcome
557    }
558
559    /// Backward-compatible alias for graceful shutdown.
560    pub fn shutdown(&mut self) -> DrainOutcome {
561        self.graceful_shutdown()
562    }
563
564    fn leave_cluster_for_shutdown(&self) {
565        if matches!(self.config.role, ServerRole::Member | ServerRole::Client) {
566            let _ = block_on_cluster_leave(&self.cache);
567        }
568    }
569
570    /// Return admin/operator status derived from the runtime model.
571    pub fn admin_status(&self) -> ServerAdminStatus {
572        let status = self.cluster_status_snapshot();
573        ServerAdminStatus {
574            source: status.source,
575            leader: status.leader,
576            term: status.term,
577            quorum_ok: status.quorum_ok,
578            members: status.members.len() as u32,
579            voters: status.voters,
580            reshard_phase: status.reshard_phase.to_string(),
581            draining: status.draining,
582        }
583    }
584
585    /// Build a metrics registry snapshot for the admin surface.
586    pub fn metrics_registry(&self) -> HydraCacheRegistry {
587        let status = self.cluster_status_snapshot();
588        let registry = HydraCacheRegistry::new()
589            .with_cache("server", self.cache.clone())
590            .with_cluster_grid_counters(self.observability.cluster_grid)
591            .with_topology(ClusterTopologyOverview::new(
592                topology_status_source(status.source),
593                status.members.len() as u64,
594                status.leader,
595                status.epoch,
596                topology_reshard_phase(status.reshard_phase),
597            ));
598        if let Some(seconds) = self.observability.backup_age_seconds {
599            registry.with_backup_age_seconds(seconds)
600        } else {
601            registry
602        }
603    }
604
605    /// Build a read-only Management Center cluster overview.
606    pub fn cluster_overview(&self) -> ClusterOverview {
607        let status = self.cluster_status_snapshot();
608        let counters = overview_cluster_grid_counters(
609            self.cache.cluster_grid_counters(),
610            self.observability.cluster_grid,
611        );
612        ClusterOverview::new(
613            topology_status_source(status.source),
614            status
615                .members
616                .iter()
617                .map(|member| {
618                    ClusterMemberView::new(
619                        member.node_id.clone(),
620                        member_role_label(member.role),
621                        member.reachable == Reachability::Reachable,
622                        reachability_label(member.reachable),
623                        member.generation,
624                    )
625                })
626                .collect(),
627            cluster_overview_leader(&status),
628            PartitionSummary::from_grid_counters(counters, self.observability.partition_count),
629            ConsistencyView::from_grid_counters(
630                self.observability.configured_default_consistency.clone(),
631                counters,
632            ),
633            self.observability.backup_age_seconds,
634            LifecycleView::new(
635                status.reshard_phase.to_string(),
636                self.observability.upgrade_phase.clone(),
637            ),
638        )
639    }
640
641    fn cluster_status_snapshot(&self) -> ClusterStatus {
642        let cluster_ready = self.cluster_ready && self.state != ServerState::Stopped;
643        self.cluster_status
644            .cluster_status(ClusterStatusRuntime::new(cluster_ready, self.is_draining()))
645    }
646
647    /// Request an online reshard through the current runtime model.
648    pub fn request_reshard(&self) -> Result<ServerAdminAction, ServerAdminActionError> {
649        if !self.can_serve() {
650            return Err(ServerAdminActionError::NotReady("reshard"));
651        }
652        if !matches!(self.config.role, ServerRole::Member) {
653            return Err(ServerAdminActionError::RequiresMember("reshard"));
654        }
655        Ok(ServerAdminAction {
656            action: "reshard",
657            outcome: "accepted",
658            detail: "reshard request accepted by member runtime".to_owned(),
659        })
660    }
661
662    /// Request a backup through the current runtime model.
663    pub fn request_backup(&self) -> Result<ServerAdminAction, ServerAdminActionError> {
664        if !self.can_serve() {
665            return Err(ServerAdminActionError::NotReady("backup"));
666        }
667        if !self.config.backup.enabled
668            || self
669                .config
670                .backup
671                .location
672                .as_deref()
673                .unwrap_or("")
674                .trim()
675                .is_empty()
676        {
677            return Err(ServerAdminActionError::BackupDisabled);
678        }
679        Ok(ServerAdminAction {
680            action: "backup",
681            outcome: "accepted",
682            detail: "backup request accepted by configured runtime".to_owned(),
683        })
684    }
685
686    /// Return whether shutdown flushed durable state.
687    pub fn flushed(&self) -> bool {
688        self.flushed
689    }
690
691    /// Return cache handle used by embedded tests/adapters.
692    pub fn cache(&self) -> &HydraCache {
693        &self.cache
694    }
695
696    /// Return the runtime config.
697    pub fn config(&self) -> &ServerConfig {
698        &self.config
699    }
700}
701
702fn topology_status_source(source: StatusSource) -> TopologyStatusSource {
703    match source {
704        StatusSource::Live => TopologyStatusSource::Live,
705        StatusSource::Modeled => TopologyStatusSource::Modeled,
706    }
707}
708
709fn block_on_cluster_leave(cache: &HydraCache) -> hydracache::CacheResult<()> {
710    let cache = cache.clone();
711    if tokio::runtime::Handle::try_current().is_ok() {
712        return std::thread::spawn(move || block_on_cluster_leave_without_current(cache))
713            .join()
714            .map_err(|_| {
715                hydracache::CacheError::Backend("cluster leave helper thread panicked".to_owned())
716            })?;
717    }
718
719    block_on_cluster_leave_without_current(cache)
720}
721
722fn block_on_cluster_leave_without_current(cache: HydraCache) -> hydracache::CacheResult<()> {
723    let runtime = tokio::runtime::Builder::new_current_thread()
724        .enable_all()
725        .build()
726        .map_err(|error| {
727            hydracache::CacheError::Backend(format!(
728                "failed to build cluster leave runtime: {error}"
729            ))
730        })?;
731    let left = runtime.block_on(cache.leave_cluster())?;
732    let _ = left;
733    Ok(())
734}
735
736fn topology_reshard_phase(phase: ReshardPhase) -> TopologyReshardPhase {
737    match phase {
738        ReshardPhase::Idle => TopologyReshardPhase::Idle,
739        ReshardPhase::Planning => TopologyReshardPhase::Planning,
740        ReshardPhase::Moving => TopologyReshardPhase::Moving,
741        ReshardPhase::Finalizing => TopologyReshardPhase::Finalizing,
742    }
743}
744
745fn cluster_overview_leader(status: &ClusterStatus) -> Option<LeaderView> {
746    if status.source != StatusSource::Live {
747        return None;
748    }
749    status
750        .leader
751        .as_ref()
752        .map(|node_id| LeaderView::new(node_id.clone(), status.term, status.epoch))
753}
754
755fn member_role_label(role: MemberRole) -> &'static str {
756    match role {
757        MemberRole::Local => "local",
758        MemberRole::Client => "client",
759        MemberRole::Member => "member",
760    }
761}
762
763fn reachability_label(reachability: Reachability) -> &'static str {
764    match reachability {
765        Reachability::Reachable => "reachable",
766        Reachability::Suspect => "suspect",
767        Reachability::Unreachable => "unreachable",
768    }
769}
770
771fn overview_cluster_grid_counters(
772    mut left: ClusterGridCounters,
773    right: ClusterGridCounters,
774) -> ClusterGridCounters {
775    left.under_replicated_keys = left
776        .under_replicated_keys
777        .saturating_add(right.under_replicated_keys);
778    left.consistency_level_operations_total = left
779        .consistency_level_operations_total
780        .saturating_add(right.consistency_level_operations_total);
781    left
782}
hydracache_server/bootstrap.rs

hydracache_server/
bootstrap.rs
