Skip to main content

hydracache_server/
config.rs

1use std::env;
2use std::fs;
3use std::net::{IpAddr, SocketAddr};
4use std::path::{Path, PathBuf};
5use std::time::Duration;
6
7use hydracache_client_transport_axum::ClientSurfaceLimits;
8use hydracache_redis_compat::{
9    RedisAuthConfig, RedisKeyspaceEventConfig, RedisListenerConfig,
10    DEFAULT_REDIS_EVENT_SUBSCRIPTIONS_PER_CONNECTION,
11    DEFAULT_REDIS_EVENT_SUBSCRIPTION_BYTES_PER_CONNECTION,
12};
13use serde::{Deserialize, Serialize};
14use thiserror::Error;
15
16/// Server role selected at startup.
17#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(rename_all = "snake_case")]
19pub enum ServerRole {
20    /// Embedded-compatible single-process cache.
21    #[default]
22    Local,
23    /// Cluster member that owns partitions and durable state.
24    Member,
25    /// Client/near-cache process that connects to members.
26    Client,
27}
28
29/// Cluster startup mode for a durable member.
30///
31/// Restarting members keep their stored raft identity and log. The configured
32/// mode remains authoritative so a retained late-join volume is re-admitted by
33/// the live cluster instead of bootstrapping from a stale removed ConfState.
34#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(rename_all = "snake_case")]
36pub enum ClusterStartMode {
37    /// Bootstrap a new cluster or bootstrap cohort.
38    #[default]
39    Bootstrap,
40    /// Join an already bootstrapped cluster through seed members.
41    Join,
42}
43
44/// TLS startup policy.
45#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
46pub struct TlsConfig {
47    /// Whether TLS is enabled for externally reachable listeners.
48    pub enabled: bool,
49    /// Operator-supplied certificate path.
50    pub cert_path: Option<PathBuf>,
51    /// Operator-supplied private-key path.
52    pub key_path: Option<PathBuf>,
53    /// Operator-supplied CA bundle path.
54    pub ca_path: Option<PathBuf>,
55    /// Explicit acknowledgement for local/staging insecure deployments.
56    pub acknowledge_insecure: bool,
57}
58
59impl TlsConfig {
60    /// Return whether all configured TLS paths are present.
61    pub fn has_complete_material(&self) -> bool {
62        !self.enabled
63            || (self.cert_path.is_some() && self.key_path.is_some() && self.ca_path.is_some())
64    }
65}
66
67/// Cluster route credential policy.
68#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
69pub struct ClusterAuthConfig {
70    /// Current credential key id.
71    pub key_id: Option<String>,
72    /// File containing the current opaque token.
73    pub token_file: Option<PathBuf>,
74    /// Previous credential key id accepted during rotation.
75    pub previous_key_id: Option<String>,
76    /// File containing the previous opaque token.
77    pub previous_token_file: Option<PathBuf>,
78}
79
80impl ClusterAuthConfig {
81    /// Return whether a current credential is configured.
82    pub fn is_configured(&self) -> bool {
83        self.key_id.as_deref().is_some_and(non_empty)
84            || self.token_file.as_deref().is_some_and(non_empty_path)
85    }
86
87    fn validate(&self) -> Result<(), ServerConfigError> {
88        validate_cluster_auth_pair(
89            self.key_id.as_deref(),
90            self.token_file.as_deref(),
91            "cluster_auth",
92        )?;
93        validate_cluster_auth_pair(
94            self.previous_key_id.as_deref(),
95            self.previous_token_file.as_deref(),
96            "cluster_auth.previous",
97        )
98    }
99}
100
101/// Backup startup policy.
102#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
103pub struct BackupConfig {
104    /// Whether background backup/PITR services are enabled.
105    pub enabled: bool,
106    /// Local/object-store destination URI.
107    pub location: Option<String>,
108}
109
110/// External client API startup policy.
111#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
112pub struct ClientApiConfig {
113    /// Whether `/client/v1/*` routes are enabled.
114    pub enabled: bool,
115    /// External client request and stream limits.
116    pub limits: ClientSurfaceLimits,
117}
118
119/// Off-by-default production HC/2 gRPC client-plane policy.
120#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
121pub struct Hc2ClientPlaneConfig {
122    /// Whether the mandatory-mTLS HC/2 listener is enabled.
123    pub enabled: bool,
124    /// Dedicated HC/2 listen address.
125    pub listen_addr: SocketAddr,
126    /// Stable cluster identity returned by the HC/2 handshake.
127    pub cluster_id: String,
128}
129
130impl Default for Hc2ClientPlaneConfig {
131    fn default() -> Self {
132        Self {
133            enabled: false,
134            listen_addr: "127.0.0.1:9443"
135                .parse()
136                .expect("default HC/2 listen address is valid"),
137            cluster_id: "hydracache-local".to_owned(),
138        }
139    }
140}
141
142/// Internal operator/admin HTTP policy.
143#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
144pub struct AdminApiConfig {
145    /// Whether `/healthz`, `/readyz`, and `/admin/*` routes are enabled.
146    pub enabled: bool,
147    /// Internal admin listen address, intentionally separate from the client surface.
148    pub listen_addr: SocketAddr,
149    /// Whether the local-only destructive diagnostic reset route is enabled.
150    #[serde(default)]
151    pub diagnostic_reset_enabled: bool,
152}
153
154impl Default for AdminApiConfig {
155    fn default() -> Self {
156        Self {
157            enabled: true,
158            listen_addr: "127.0.0.1:9091"
159                .parse()
160                .expect("default admin listen address is valid"),
161            diagnostic_reset_enabled: false,
162        }
163    }
164}
165
166/// Optional Redis RESP edge facade policy.
167#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
168pub struct RedisApiConfig {
169    /// Whether the RESP listener is enabled.
170    pub enabled: bool,
171    /// RESP listen address, intentionally separate from HTTP/admin/cluster surfaces.
172    pub listen_addr: SocketAddr,
173    /// Whether Redis AUTH is required before cache/data commands.
174    pub auth_required: bool,
175    /// Optional ACL-style username for AUTH and HELLO AUTH.
176    pub auth_username: Option<String>,
177    /// File containing the Redis AUTH token/password.
178    pub auth_token_file: Option<PathBuf>,
179    /// Whether to request native rediss:// on this listener.
180    pub rediss_enabled: bool,
181    /// Whether Redis keyspace notification subscriptions are enabled.
182    pub keyspace_events_enabled: bool,
183    /// Maximum exact plus pattern subscriptions retained per RESP connection.
184    pub max_event_subscriptions_per_connection: usize,
185    /// Maximum total channel plus pattern bytes retained per RESP connection.
186    pub max_event_subscription_bytes_per_connection: usize,
187}
188
189impl Default for RedisApiConfig {
190    fn default() -> Self {
191        Self {
192            enabled: false,
193            listen_addr: "127.0.0.1:6379"
194                .parse()
195                .expect("default Redis RESP listen address is valid"),
196            auth_required: false,
197            auth_username: None,
198            auth_token_file: None,
199            rediss_enabled: false,
200            keyspace_events_enabled: false,
201            max_event_subscriptions_per_connection:
202                DEFAULT_REDIS_EVENT_SUBSCRIPTIONS_PER_CONNECTION,
203            max_event_subscription_bytes_per_connection:
204                DEFAULT_REDIS_EVENT_SUBSCRIPTION_BYTES_PER_CONNECTION,
205        }
206    }
207}
208
209/// Standalone daemon configuration.
210#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
211#[serde(default)]
212pub struct ServerConfig {
213    /// Server role.
214    pub role: ServerRole,
215    /// Public actuator/data listen address.
216    pub listen_addr: SocketAddr,
217    /// Internal cluster listen address.
218    pub cluster_addr: SocketAddr,
219    /// First-boot cluster startup mode.
220    pub cluster_start: ClusterStartMode,
221    /// Routable cluster endpoint advertised to other members.
222    pub cluster_advertise_addr: Option<String>,
223    /// Optional stable member node identity.
224    pub node_id: Option<String>,
225    /// Seed members used by member/client roles.
226    pub seeds: Vec<String>,
227    /// Durable state directory for member mode.
228    pub storage_dir: Option<PathBuf>,
229    /// Graceful shutdown drain timeout.
230    pub drain_timeout_ms: u64,
231    /// Maximum time spent waiting for explicit join admission.
232    pub join_timeout_ms: u64,
233    /// TLS policy.
234    pub tls: TlsConfig,
235    /// Cluster route authentication policy.
236    pub cluster_auth: ClusterAuthConfig,
237    /// Backup policy.
238    pub backup: BackupConfig,
239    /// External client API policy.
240    pub client_api: ClientApiConfig,
241    /// Optional production HC/2 gRPC client plane.
242    pub hc2_client_plane: Hc2ClientPlaneConfig,
243    /// Internal operator/admin HTTP policy.
244    pub admin_api: AdminApiConfig,
245    /// Optional Redis RESP edge facade policy.
246    pub redis_api: RedisApiConfig,
247    /// Whether the narrow admin Raft compaction test/ops seam is enabled.
248    ///
249    /// This remains false in normal production configuration and does not
250    /// enable automatic compaction.
251    pub raft_compaction_enabled: bool,
252}
253
254impl Default for ServerConfig {
255    fn default() -> Self {
256        Self {
257            role: ServerRole::Local,
258            listen_addr: "127.0.0.1:8080"
259                .parse()
260                .expect("default listen address is valid"),
261            cluster_addr: "127.0.0.1:7000"
262                .parse()
263                .expect("default cluster address is valid"),
264            cluster_start: ClusterStartMode::Bootstrap,
265            cluster_advertise_addr: None,
266            node_id: None,
267            seeds: Vec::new(),
268            storage_dir: None,
269            drain_timeout_ms: 30_000,
270            join_timeout_ms: 15_000,
271            tls: TlsConfig::default(),
272            cluster_auth: ClusterAuthConfig::default(),
273            backup: BackupConfig::default(),
274            client_api: ClientApiConfig::default(),
275            hc2_client_plane: Hc2ClientPlaneConfig::default(),
276            admin_api: AdminApiConfig::default(),
277            redis_api: RedisApiConfig::default(),
278            raft_compaction_enabled: false,
279        }
280    }
281}
282
283impl ServerConfig {
284    /// Load config from a TOML file.
285    pub fn from_file(path: impl AsRef<Path>) -> Result<Self, ServerConfigError> {
286        let path = path.as_ref();
287        let text = fs::read_to_string(path).map_err(|source| ServerConfigError::ConfigRead {
288            path: path.to_path_buf(),
289            source,
290        })?;
291        Self::from_toml_str(&text)
292    }
293
294    /// Load config from TOML text.
295    pub fn from_toml_str(text: &str) -> Result<Self, ServerConfigError> {
296        let config = toml::from_str::<Self>(text).map_err(ServerConfigError::ConfigParse)?;
297        config.validate()?;
298        Ok(config)
299    }
300
301    /// Load config from selected environment variables.
302    pub fn from_env() -> Result<Self, ServerConfigError> {
303        let mut config = Self::default();
304        if let Ok(role) = env::var("HYDRACACHE_ROLE") {
305            config.role = parse_role(&role)?;
306        }
307        if let Ok(listen) = env::var("HYDRACACHE_LISTEN_ADDR") {
308            config.listen_addr = listen
309                .parse()
310                .map_err(|_| ServerConfigError::InvalidAddress(listen))?;
311        }
312        if let Ok(cluster) = env::var("HYDRACACHE_CLUSTER_ADDR") {
313            config.cluster_addr = cluster
314                .parse()
315                .map_err(|_| ServerConfigError::InvalidAddress(cluster))?;
316        }
317        let cluster_start_explicit = if let Ok(cluster_start) = env::var("HYDRACACHE_CLUSTER_START")
318        {
319            config.cluster_start = parse_cluster_start(&cluster_start)?;
320            true
321        } else {
322            false
323        };
324        if let Ok(advertise_addr) = env::var("HYDRACACHE_CLUSTER_ADVERTISE_ADDR") {
325            config.cluster_advertise_addr = Some(advertise_addr);
326        }
327        if let Ok(node_id) = env::var("HYDRACACHE_NODE_ID") {
328            config.node_id = Some(node_id);
329        }
330        if let Ok(storage_dir) = env::var("HYDRACACHE_STORAGE_DIR") {
331            config.storage_dir = Some(PathBuf::from(storage_dir));
332        }
333        if let Ok(seeds) = env::var("HYDRACACHE_SEEDS") {
334            config.seeds = seeds
335                .split(',')
336                .map(str::trim)
337                .filter(|seed| !seed.is_empty())
338                .map(ToOwned::to_owned)
339                .collect();
340        }
341        apply_statefulset_env(&mut config, cluster_start_explicit)?;
342        if let Ok(join_timeout) = env::var("HYDRACACHE_JOIN_TIMEOUT_MS") {
343            config.join_timeout_ms = join_timeout
344                .parse()
345                .map_err(|_| ServerConfigError::InvalidJoinTimeoutMs(join_timeout))?;
346        }
347        if env::var("HYDRACACHE_TLS_ACK_INSECURE").as_deref() == Ok("true") {
348            config.tls.acknowledge_insecure = true;
349        }
350        if env::var("HYDRACACHE_TLS_ENABLED").as_deref() == Ok("true") {
351            config.tls.enabled = true;
352        }
353        if let Ok(path) = env::var("HYDRACACHE_TLS_CERT_PATH") {
354            config.tls.cert_path = Some(PathBuf::from(path));
355        }
356        if let Ok(path) = env::var("HYDRACACHE_TLS_KEY_PATH") {
357            config.tls.key_path = Some(PathBuf::from(path));
358        }
359        if let Ok(path) = env::var("HYDRACACHE_TLS_CA_PATH") {
360            config.tls.ca_path = Some(PathBuf::from(path));
361        }
362        if let Ok(key_id) = env::var("HYDRACACHE_CLUSTER_AUTH_KEY_ID") {
363            config.cluster_auth.key_id = Some(key_id);
364        }
365        if let Ok(path) = env::var("HYDRACACHE_CLUSTER_AUTH_TOKEN_FILE") {
366            config.cluster_auth.token_file = Some(PathBuf::from(path));
367        }
368        if let Ok(key_id) = env::var("HYDRACACHE_CLUSTER_AUTH_PREVIOUS_KEY_ID") {
369            config.cluster_auth.previous_key_id = Some(key_id);
370        }
371        if let Ok(path) = env::var("HYDRACACHE_CLUSTER_AUTH_PREVIOUS_TOKEN_FILE") {
372            config.cluster_auth.previous_token_file = Some(PathBuf::from(path));
373        }
374        if env::var("HYDRACACHE_BACKUP_ENABLED").as_deref() == Ok("true") {
375            config.backup.enabled = true;
376        }
377        if let Ok(location) = env::var("HYDRACACHE_BACKUP_LOCATION") {
378            config.backup.location = Some(location);
379        }
380        if env::var("HYDRACACHE_CLIENT_API_ENABLED").as_deref() == Ok("true") {
381            config.client_api.enabled = true;
382        }
383        if env::var("HYDRACACHE_HC2_ENABLED").as_deref() == Ok("true") {
384            config.hc2_client_plane.enabled = true;
385        }
386        if let Ok(listen) = env::var("HYDRACACHE_HC2_ADDR") {
387            config.hc2_client_plane.listen_addr = listen
388                .parse()
389                .map_err(|_| ServerConfigError::InvalidAddress(listen))?;
390        }
391        if let Ok(cluster_id) = env::var("HYDRACACHE_HC2_CLUSTER_ID") {
392            config.hc2_client_plane.cluster_id = cluster_id;
393        }
394        if let Ok(enabled) = env::var("HYDRACACHE_ADMIN_API_ENABLED") {
395            config.admin_api.enabled = enabled != "false";
396        }
397        if let Ok(listen) = env::var("HYDRACACHE_ADMIN_ADDR") {
398            config.admin_api.listen_addr = listen
399                .parse()
400                .map_err(|_| ServerConfigError::InvalidAddress(listen))?;
401        }
402        if env::var("HYDRACACHE_DIAGNOSTIC_RESET_ENABLED").as_deref() == Ok("true") {
403            config.admin_api.diagnostic_reset_enabled = true;
404        }
405        if env::var("HYDRACACHE_REDIS_API_ENABLED").as_deref() == Ok("true") {
406            config.redis_api.enabled = true;
407        }
408        if let Ok(listen) = env::var("HYDRACACHE_REDIS_ADDR") {
409            config.redis_api.listen_addr = listen
410                .parse()
411                .map_err(|_| ServerConfigError::InvalidAddress(listen))?;
412        }
413        if env::var("HYDRACACHE_REDIS_AUTH_REQUIRED").as_deref() == Ok("true") {
414            config.redis_api.auth_required = true;
415        }
416        if let Ok(username) = env::var("HYDRACACHE_REDIS_AUTH_USERNAME") {
417            config.redis_api.auth_username = Some(username);
418        }
419        if let Ok(path) = env::var("HYDRACACHE_REDIS_AUTH_TOKEN_FILE") {
420            config.redis_api.auth_token_file = Some(PathBuf::from(path));
421        }
422        if env::var("HYDRACACHE_REDIS_REDISS_ENABLED").as_deref() == Ok("true") {
423            config.redis_api.rediss_enabled = true;
424        }
425        if env::var("HYDRACACHE_REDIS_KEYSPACE_EVENTS_ENABLED").as_deref() == Ok("true") {
426            config.redis_api.keyspace_events_enabled = true;
427        }
428        if let Ok(limit) = env::var("HYDRACACHE_REDIS_MAX_EVENT_SUBSCRIPTIONS_PER_CONNECTION") {
429            config.redis_api.max_event_subscriptions_per_connection = limit
430                .parse()
431                .map_err(|_| ServerConfigError::InvalidRedisEventSubscriptionLimit)?;
432        }
433        if let Ok(limit) = env::var("HYDRACACHE_REDIS_MAX_EVENT_SUBSCRIPTION_BYTES_PER_CONNECTION")
434        {
435            config.redis_api.max_event_subscription_bytes_per_connection = limit
436                .parse()
437                .map_err(|_| ServerConfigError::InvalidRedisEventSubscriptionByteLimit)?;
438        }
439        if env::var("HYDRACACHE_RAFT_COMPACTION").as_deref() == Ok("true") {
440            config.raft_compaction_enabled = true;
441        }
442        config.validate()?;
443        Ok(config)
444    }
445
446    /// Validate startup invariants.
447    pub fn validate(&self) -> Result<(), ServerConfigError> {
448        if self.drain_timeout_ms == 0 {
449            return Err(ServerConfigError::DrainTimeoutZero);
450        }
451        if self.join_timeout_ms == 0 {
452            return Err(ServerConfigError::JoinTimeoutZero);
453        }
454        if matches!(self.cluster_start, ClusterStartMode::Join) {
455            if !matches!(self.role, ServerRole::Member) {
456                return Err(ServerConfigError::JoinRequiresMemberRole);
457            }
458            if self.seeds.is_empty() {
459                return Err(ServerConfigError::JoinRequiresSeeds);
460            }
461        }
462        if matches!(self.role, ServerRole::Member) && self.storage_dir.is_none() {
463            return Err(ServerConfigError::MissingStorageDir);
464        }
465        if self.raft_compaction_enabled && !matches!(self.role, ServerRole::Member) {
466            return Err(ServerConfigError::RaftCompactionRequiresMember);
467        }
468        if matches!(self.role, ServerRole::Member | ServerRole::Client) && self.seeds.is_empty() {
469            return Err(ServerConfigError::MissingSeeds);
470        }
471        if self
472            .node_id
473            .as_deref()
474            .is_some_and(|node_id| node_id.trim().is_empty())
475        {
476            return Err(ServerConfigError::InvalidNodeId);
477        }
478        if self
479            .cluster_advertise_addr
480            .as_deref()
481            .is_some_and(invalid_cluster_advertise_addr)
482        {
483            return Err(ServerConfigError::InvalidClusterAdvertiseAddr);
484        }
485        if self.backup.enabled
486            && self
487                .backup
488                .location
489                .as_deref()
490                .unwrap_or("")
491                .trim()
492                .is_empty()
493        {
494            return Err(ServerConfigError::MissingBackupLocation);
495        }
496        if !self.tls.has_complete_material() {
497            return Err(ServerConfigError::IncompleteTlsMaterial);
498        }
499        self.cluster_auth.validate()?;
500        if self.exposes_non_loopback() && !self.tls.enabled && !self.tls.acknowledge_insecure {
501            return Err(ServerConfigError::NonLoopbackWithoutTls);
502        }
503        if self.client_api.enabled {
504            self.client_api
505                .limits
506                .validate()
507                .map_err(|error| ServerConfigError::InvalidClientApi(error.to_string()))?;
508        }
509        if self.hc2_client_plane.enabled {
510            if !self.tls.enabled {
511                return Err(ServerConfigError::Hc2RequiresMutualTls);
512            }
513            if self.hc2_client_plane.cluster_id.trim().is_empty() {
514                return Err(ServerConfigError::InvalidHc2ClusterId);
515            }
516            for (address, surface) in [
517                (self.listen_addr, "listen_addr"),
518                (self.cluster_addr, "cluster_addr"),
519                (self.admin_api.listen_addr, "admin_api.listen_addr"),
520                (self.redis_api.listen_addr, "redis_api.listen_addr"),
521            ] {
522                if self.hc2_client_plane.listen_addr == address
523                    && (surface != "admin_api.listen_addr" || self.admin_api.enabled)
524                    && (surface != "redis_api.listen_addr" || self.redis_api.enabled)
525                {
526                    return Err(ServerConfigError::Hc2AddressConflicts { surface });
527                }
528            }
529        }
530        if self.admin_api.enabled && self.admin_api.listen_addr == self.listen_addr {
531            return Err(ServerConfigError::AdminAddressConflicts);
532        }
533        if self.admin_api.diagnostic_reset_enabled
534            && (!self.admin_api.enabled
535                || self.role != ServerRole::Local
536                || !is_loopback(self.admin_api.listen_addr.ip()))
537        {
538            return Err(ServerConfigError::UnsafeDiagnosticReset);
539        }
540        if self.redis_api.enabled {
541            if self.redis_api.max_event_subscriptions_per_connection == 0 {
542                return Err(ServerConfigError::InvalidRedisEventSubscriptionLimit);
543            }
544            if self.redis_api.max_event_subscription_bytes_per_connection == 0 {
545                return Err(ServerConfigError::InvalidRedisEventSubscriptionByteLimit);
546            }
547            if self.redis_api.rediss_enabled && !self.tls.enabled {
548                return Err(ServerConfigError::RedisRedissRequiresTls);
549            }
550            validate_redis_auth(&self.redis_api)?;
551            if self.redis_api.listen_addr == self.listen_addr {
552                return Err(ServerConfigError::RedisAddressConflicts {
553                    surface: "listen_addr",
554                });
555            }
556            if self.redis_api.listen_addr == self.cluster_addr {
557                return Err(ServerConfigError::RedisAddressConflicts {
558                    surface: "cluster_addr",
559                });
560            }
561            if self.admin_api.enabled && self.redis_api.listen_addr == self.admin_api.listen_addr {
562                return Err(ServerConfigError::RedisAddressConflicts {
563                    surface: "admin_api.listen_addr",
564                });
565            }
566        }
567        Ok(())
568    }
569
570    /// Build the Redis RESP listener config used by the optional edge server.
571    pub fn redis_listener_config(&self) -> Result<RedisListenerConfig, ServerConfigError> {
572        let mut config = RedisListenerConfig {
573            keyspace_events: RedisKeyspaceEventConfig {
574                enabled: self.redis_api.keyspace_events_enabled,
575                max_subscriptions_per_connection: self
576                    .redis_api
577                    .max_event_subscriptions_per_connection,
578                max_subscription_bytes_per_connection: self
579                    .redis_api
580                    .max_event_subscription_bytes_per_connection,
581            },
582            ..RedisListenerConfig::default()
583        };
584        if self.redis_api.auth_required {
585            let path = self
586                .redis_api
587                .auth_token_file
588                .as_deref()
589                .ok_or(ServerConfigError::IncompleteRedisAuth)?;
590            config.auth = RedisAuthConfig {
591                required: true,
592                username: self.redis_api.auth_username.clone(),
593                password: Some(read_redis_auth_token(path)?),
594            };
595        }
596        Ok(config)
597    }
598
599    /// Return the configured drain timeout.
600    pub fn drain_timeout(&self) -> Duration {
601        Duration::from_millis(self.drain_timeout_ms)
602    }
603
604    /// Return the configured join admission timeout.
605    pub fn join_timeout(&self) -> Duration {
606        Duration::from_millis(self.join_timeout_ms)
607    }
608
609    /// Return the endpoint this member should advertise to peers.
610    pub fn cluster_advertise_endpoint(&self) -> String {
611        self.cluster_advertise_addr
612            .clone()
613            .unwrap_or_else(|| self.cluster_addr.to_string())
614    }
615
616    /// Return whether any listener is externally reachable.
617    pub fn exposes_non_loopback(&self) -> bool {
618        !is_loopback(self.listen_addr.ip())
619            || !is_loopback(self.cluster_addr.ip())
620            || (self.admin_api.enabled && !is_loopback(self.admin_api.listen_addr.ip()))
621            || (self.redis_api.enabled && !is_loopback(self.redis_api.listen_addr.ip()))
622            || (self.hc2_client_plane.enabled
623                && !is_loopback(self.hc2_client_plane.listen_addr.ip()))
624    }
625}
626
627/// Fail-loud configuration errors.
628#[derive(Debug, Error)]
629pub enum ServerConfigError {
630    /// Config file could not be read.
631    #[error("failed to read config {path}: {source}")]
632    ConfigRead {
633        /// Config path.
634        path: PathBuf,
635        /// Source IO error.
636        source: std::io::Error,
637    },
638    /// Config file could not be parsed.
639    #[error("failed to parse config: {0}")]
640    ConfigParse(toml::de::Error),
641    /// Role value is unknown.
642    #[error("invalid server role: {0}")]
643    InvalidRole(String),
644    /// Cluster start mode value is unknown.
645    #[error("invalid cluster start mode: {0}")]
646    InvalidClusterStart(String),
647    /// Address value is invalid.
648    #[error("invalid listen address: {0}")]
649    InvalidAddress(String),
650    /// StatefulSet bootstrap replica count could not be parsed.
651    #[error("invalid bootstrap_replicas: {0}")]
652    InvalidBootstrapReplicas(String),
653    /// StatefulSet hostname could not be used to derive an ordinal identity.
654    #[error("invalid StatefulSet HOSTNAME: {0}")]
655    InvalidStatefulSetHostname(String),
656    /// Join timeout value could not be parsed.
657    #[error("invalid join_timeout_ms: {0}")]
658    InvalidJoinTimeoutMs(String),
659    /// Drain timeout cannot be zero.
660    #[error("drain_timeout_ms must be greater than zero")]
661    DrainTimeoutZero,
662    /// Join timeout cannot be zero.
663    #[error("join_timeout_ms must be greater than zero")]
664    JoinTimeoutZero,
665    /// Join mode is only valid for durable members.
666    #[error("cluster_start=join requires role=member")]
667    JoinRequiresMemberRole,
668    /// Join mode requires at least one seed.
669    #[error("cluster_start=join requires at least one seed")]
670    JoinRequiresSeeds,
671    /// Member mode requires durable state.
672    #[error("member role requires storage_dir")]
673    MissingStorageDir,
674    /// The disk-backed compaction seam is only valid for durable members.
675    #[error("raft compaction control requires role=member")]
676    RaftCompactionRequiresMember,
677    /// Member/client mode requires seeds.
678    #[error("member/client role requires at least one seed")]
679    MissingSeeds,
680    /// Configured node identity cannot be empty.
681    #[error("node_id must not be empty")]
682    InvalidNodeId,
683    /// Advertised cluster endpoint must be a routable non-empty endpoint.
684    #[error("cluster_advertise_addr must be non-empty and routable when set")]
685    InvalidClusterAdvertiseAddr,
686    /// Backup enabled without a destination.
687    #[error("backup.enabled requires backup.location")]
688    MissingBackupLocation,
689    /// TLS enabled without full material paths.
690    #[error("tls.enabled requires cert_path, key_path, and ca_path")]
691    IncompleteTlsMaterial,
692    /// Cluster auth material is incomplete.
693    #[error("{section} requires key_id and readable token_file")]
694    IncompleteClusterAuth {
695        /// Config section.
696        section: &'static str,
697    },
698    /// Cluster auth token file could not be read.
699    #[error("failed to read {section}.token_file {path}: {source}")]
700    ClusterAuthTokenRead {
701        /// Config section.
702        section: &'static str,
703        /// Token file path.
704        path: PathBuf,
705        /// Source IO error.
706        source: std::io::Error,
707    },
708    /// Cluster auth token file was empty.
709    #[error("{section}.token_file {path} is empty")]
710    EmptyClusterAuthToken {
711        /// Config section.
712        section: &'static str,
713        /// Token file path.
714        path: PathBuf,
715    },
716    /// External listener without TLS and without explicit acknowledgement.
717    #[error("non-loopback listeners require TLS or acknowledge_insecure=true")]
718    NonLoopbackWithoutTls,
719    /// External client API config is invalid.
720    #[error("invalid client_api config: {0}")]
721    InvalidClientApi(String),
722    /// Member grid host could not be constructed.
723    #[error("failed to start member grid host: {0}")]
724    GridHostStart(String),
725    /// Admin and client/listen surfaces must be independently bindable.
726    #[error("admin_api.listen_addr must differ from listen_addr")]
727    AdminAddressConflicts,
728    /// Destructive diagnostic reset is restricted to an enabled loopback admin surface in local mode.
729    #[error("diagnostic reset requires role=local and an enabled loopback admin API")]
730    UnsafeDiagnosticReset,
731    /// Redis RESP and existing surfaces must be independently bindable.
732    #[error("redis_api.listen_addr must differ from {surface}")]
733    RedisAddressConflicts {
734        /// Conflicting surface.
735        surface: &'static str,
736    },
737    /// Redis AUTH required without readable token material.
738    #[error("redis_api auth requires auth_token_file and a non-empty token")]
739    IncompleteRedisAuth,
740    /// Redis AUTH token file could not be read.
741    #[error("failed to read redis_api.auth_token_file {path}: {source}")]
742    RedisAuthTokenRead {
743        /// Token file path.
744        path: PathBuf,
745        /// Source IO error.
746        source: std::io::Error,
747    },
748    /// Redis AUTH token file was empty.
749    #[error("redis_api.auth_token_file {path} is empty")]
750    EmptyRedisAuthToken {
751        /// Token file path.
752        path: PathBuf,
753    },
754    /// Native rediss:// requires server TLS material.
755    #[error("redis_api.rediss_enabled requires tls.enabled with certificate/key material")]
756    RedisRedissRequiresTls,
757    /// Redis event listeners require a non-zero per-connection subscription bound.
758    #[error("redis_api.max_event_subscriptions_per_connection must be greater than zero")]
759    InvalidRedisEventSubscriptionLimit,
760    /// Redis event listeners require a non-zero retained subscription-byte bound.
761    #[error("redis_api.max_event_subscription_bytes_per_connection must be greater than zero")]
762    InvalidRedisEventSubscriptionByteLimit,
763    /// HC/2 never exposes a plaintext production constructor/listener.
764    #[error("hc2_client_plane.enabled requires tls.enabled and mutual TLS material")]
765    Hc2RequiresMutualTls,
766    /// Cluster identity must be non-empty when HC/2 is enabled.
767    #[error("hc2_client_plane.cluster_id must be non-empty")]
768    InvalidHc2ClusterId,
769    /// HC/2 must own an independently bindable port.
770    #[error("hc2_client_plane.listen_addr must differ from {surface}")]
771    Hc2AddressConflicts {
772        /// Conflicting surface.
773        surface: &'static str,
774    },
775}
776
777fn parse_role(value: &str) -> Result<ServerRole, ServerConfigError> {
778    match value.trim().to_ascii_lowercase().as_str() {
779        "local" => Ok(ServerRole::Local),
780        "member" => Ok(ServerRole::Member),
781        "client" => Ok(ServerRole::Client),
782        _ => Err(ServerConfigError::InvalidRole(value.to_owned())),
783    }
784}
785
786fn parse_cluster_start(value: &str) -> Result<ClusterStartMode, ServerConfigError> {
787    match value.trim().to_ascii_lowercase().as_str() {
788        "bootstrap" => Ok(ClusterStartMode::Bootstrap),
789        "join" => Ok(ClusterStartMode::Join),
790        _ => Err(ServerConfigError::InvalidClusterStart(value.to_owned())),
791    }
792}
793
794fn apply_statefulset_env(
795    config: &mut ServerConfig,
796    cluster_start_explicit: bool,
797) -> Result<(), ServerConfigError> {
798    let bootstrap_replicas = match env::var("HYDRACACHE_BOOTSTRAP_REPLICAS") {
799        Ok(value) => {
800            let replicas = value
801                .parse::<u32>()
802                .map_err(|_| ServerConfigError::InvalidBootstrapReplicas(value.clone()))?;
803            if replicas == 0 {
804                return Err(ServerConfigError::InvalidBootstrapReplicas(value));
805            }
806            Some(replicas)
807        }
808        Err(_) => None,
809    };
810    let headless_service = env::var("HYDRACACHE_CLUSTER_HEADLESS_SERVICE")
811        .ok()
812        .filter(|value| non_empty(value));
813    if bootstrap_replicas.is_none() && headless_service.is_none() {
814        return Ok(());
815    }
816
817    let hostname = env::var("HOSTNAME")
818        .map_err(|_| ServerConfigError::InvalidStatefulSetHostname(String::new()))
819        .and_then(|value| {
820            if non_empty(&value) {
821                Ok(value)
822            } else {
823                Err(ServerConfigError::InvalidStatefulSetHostname(value))
824            }
825        })?;
826
827    if config.node_id.is_none() {
828        config.node_id = Some(hostname.clone());
829    }
830
831    if let Some(headless) = headless_service {
832        if config.cluster_advertise_addr.is_none() {
833            config.cluster_advertise_addr = Some(format!(
834                "{hostname}.{headless}:{}",
835                config.cluster_addr.port()
836            ));
837        }
838    }
839
840    if let Some(replicas) = bootstrap_replicas {
841        if !cluster_start_explicit {
842            let ordinal = statefulset_ordinal(&hostname)
843                .ok_or_else(|| ServerConfigError::InvalidStatefulSetHostname(hostname.clone()))?;
844            config.cluster_start = if ordinal < replicas {
845                ClusterStartMode::Bootstrap
846            } else {
847                ClusterStartMode::Join
848            };
849        }
850    }
851
852    Ok(())
853}
854
855fn statefulset_ordinal(hostname: &str) -> Option<u32> {
856    let (_, ordinal) = hostname.rsplit_once('-')?;
857    ordinal.parse().ok()
858}
859
860fn validate_cluster_auth_pair(
861    key_id: Option<&str>,
862    token_file: Option<&Path>,
863    section: &'static str,
864) -> Result<(), ServerConfigError> {
865    let has_key = key_id.is_some_and(non_empty);
866    let has_file = token_file.is_some_and(non_empty_path);
867    if has_key != has_file {
868        return Err(ServerConfigError::IncompleteClusterAuth { section });
869    }
870    let Some(path) = token_file else {
871        return Ok(());
872    };
873    let token =
874        fs::read_to_string(path).map_err(|source| ServerConfigError::ClusterAuthTokenRead {
875            section,
876            path: path.to_path_buf(),
877            source,
878        })?;
879    if token.trim().is_empty() {
880        return Err(ServerConfigError::EmptyClusterAuthToken {
881            section,
882            path: path.to_path_buf(),
883        });
884    }
885    Ok(())
886}
887
888fn validate_redis_auth(config: &RedisApiConfig) -> Result<(), ServerConfigError> {
889    if config
890        .auth_username
891        .as_deref()
892        .is_some_and(|username| username.trim().is_empty())
893    {
894        return Err(ServerConfigError::IncompleteRedisAuth);
895    }
896    if !config.auth_required {
897        return Ok(());
898    }
899    let Some(path) = config
900        .auth_token_file
901        .as_deref()
902        .filter(|path| non_empty_path(path))
903    else {
904        return Err(ServerConfigError::IncompleteRedisAuth);
905    };
906    read_redis_auth_token(path).map(|_| ())
907}
908
909fn read_redis_auth_token(path: &Path) -> Result<String, ServerConfigError> {
910    let token =
911        fs::read_to_string(path).map_err(|source| ServerConfigError::RedisAuthTokenRead {
912            path: path.to_path_buf(),
913            source,
914        })?;
915    let token = token.trim().to_owned();
916    if token.is_empty() {
917        return Err(ServerConfigError::EmptyRedisAuthToken {
918            path: path.to_path_buf(),
919        });
920    }
921    Ok(token)
922}
923
924fn non_empty(value: &str) -> bool {
925    !value.trim().is_empty()
926}
927
928fn non_empty_path(path: &Path) -> bool {
929    !path.as_os_str().is_empty()
930}
931
932fn invalid_cluster_advertise_addr(value: &str) -> bool {
933    let value = value.trim();
934    if value.is_empty() {
935        return true;
936    }
937    value
938        .parse::<SocketAddr>()
939        .is_ok_and(|addr| addr.ip().is_unspecified())
940}
941
942fn is_loopback(ip: IpAddr) -> bool {
943    match ip {
944        IpAddr::V4(ip) => ip.is_loopback(),
945        IpAddr::V6(ip) => ip.is_loopback(),
946    }
947}