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

hydracache_server/
config.rs
