alopex_server/
config.rs

1use std::net::{IpAddr, SocketAddr};
2use std::path::Path;
3use std::path::PathBuf;
4use std::time::Duration;
5
6use alopex_cluster::{
7    ClusterId, ClusterIdentity, ClusterManager, ClusterManagerConfig, ClusterMode, Endpoint,
8    MembershipSource, NodeId, NodeRole, NodeState,
9};
10use serde::Deserialize;
11
12use crate::audit::AuditLogOutput;
13use crate::auth::AuthMode;
14use crate::error::{Result, ServerError};
15use crate::tls::TlsConfig;
16
17const MAX_ADMISSION_LIMIT: usize = 100_000;
18const MAX_QUERY_TIMEOUT_MS: u128 = 300_000;
19
20/// Server configuration options.
21#[derive(Clone, Debug, Deserialize)]
22#[serde(default)]
23pub struct ServerConfig {
24    /// HTTP bind address.
25    pub http_bind: SocketAddr,
26    /// gRPC bind address.
27    pub grpc_bind: SocketAddr,
28    /// Admin bind address.
29    pub admin_bind: SocketAddr,
30    /// Allowlist for admin API when non-loopback.
31    pub admin_allowlist: Vec<IpAddr>,
32    /// Data directory for storage.
33    pub data_dir: PathBuf,
34    /// API prefix for HTTP routes.
35    pub api_prefix: String,
36    /// Authentication mode.
37    pub auth_mode: AuthMode,
38    /// TLS configuration (optional).
39    pub tls: Option<TlsConfig>,
40    /// Query timeout.
41    #[serde(with = "humantime_serde")]
42    pub query_timeout: Duration,
43    /// Maximum number of concurrent requests admitted.
44    #[serde(alias = "max_connections")]
45    pub max_concurrency: usize,
46    /// Maximum number of queued requests under backpressure.
47    pub max_queue_len: usize,
48    /// Max request size in bytes.
49    pub max_request_size: usize,
50    /// Max response size in bytes.
51    pub max_response_size: usize,
52    /// Session TTL.
53    #[serde(with = "humantime_serde")]
54    pub session_ttl: Duration,
55    /// Enable Prometheus metrics.
56    pub metrics_enabled: bool,
57    /// Enable tracing.
58    pub tracing_enabled: bool,
59    /// Enable audit logging.
60    pub audit_log_enabled: bool,
61    /// Audit log output.
62    pub audit_log_output: AuditLogOutput,
63    /// Cluster-aware startup configuration.
64    pub cluster: ClusterServerConfig,
65}
66
67impl Default for ServerConfig {
68    fn default() -> Self {
69        Self {
70            http_bind: "127.0.0.1:8080".parse().unwrap(),
71            grpc_bind: "127.0.0.1:9090".parse().unwrap(),
72            admin_bind: "127.0.0.1:8081".parse().unwrap(),
73            admin_allowlist: Vec::new(),
74            data_dir: PathBuf::from("./data"),
75            api_prefix: String::new(),
76            auth_mode: AuthMode::None,
77            tls: None,
78            query_timeout: Duration::from_secs(30),
79            max_concurrency: 64,
80            max_queue_len: 256,
81            max_request_size: 100 * 1024 * 1024,
82            max_response_size: 100 * 1024 * 1024,
83            session_ttl: Duration::from_secs(300),
84            metrics_enabled: true,
85            tracing_enabled: true,
86            audit_log_enabled: true,
87            audit_log_output: AuditLogOutput::Stdout,
88            cluster: ClusterServerConfig::default(),
89        }
90    }
91}
92
93/// Cluster-aware server configuration.
94#[derive(Clone, Debug, Deserialize)]
95#[serde(default)]
96pub struct ClusterServerConfig {
97    /// Cluster operating mode.
98    pub mode: ClusterMode,
99    /// Stable local node identity for cluster-aware mode.
100    pub node_id: Option<String>,
101    /// Stable cluster identifier shared by configured nodes.
102    pub cluster_id: Option<String>,
103    /// Endpoint advertised to other cluster members.
104    pub advertised_endpoint: Option<String>,
105    /// Local node role.
106    pub role: NodeRole,
107    /// Local lifecycle state used when cluster-aware mode is enabled.
108    pub lifecycle_state: NodeState,
109    /// Membership metadata source to report and consume.
110    pub membership_source: MembershipSource,
111    /// Whether the membership source is available at startup.
112    pub membership_source_available: bool,
113}
114
115impl Default for ClusterServerConfig {
116    fn default() -> Self {
117        Self {
118            mode: ClusterMode::SingleNode,
119            node_id: None,
120            cluster_id: None,
121            advertised_endpoint: None,
122            role: NodeRole::Gateway,
123            lifecycle_state: NodeState::Active,
124            membership_source: MembershipSource::Chirps,
125            membership_source_available: true,
126        }
127    }
128}
129
130impl ServerConfig {
131    /// Load config from TOML and environment variables.
132    ///
133    /// Environment variables use `ALOPEX__` prefix with `__` separators.
134    pub fn load(path: Option<&Path>) -> Result<Self> {
135        let mut builder = config::Config::builder();
136        if let Some(path) = path {
137            builder = builder.add_source(config::File::from(path).required(false));
138        } else {
139            builder = builder.add_source(config::File::with_name("alopex").required(false));
140        }
141        builder = builder.add_source(config::Environment::with_prefix("ALOPEX").separator("__"));
142        let mut config: ServerConfig = builder
143            .build()
144            .map_err(|err| ServerError::InvalidConfig(err.to_string()))?
145            .try_deserialize()
146            .map_err(|err| ServerError::InvalidConfig(err.to_string()))?;
147        config.normalize()?;
148        Ok(config)
149    }
150
151    /// Validate config invariants.
152    pub fn validate(&self) -> Result<()> {
153        if !self.admin_bind.ip().is_loopback() && self.admin_allowlist.is_empty() {
154            return Err(ServerError::InvalidConfig(
155                "admin_allowlist is required for non-loopback admin_bind".into(),
156            ));
157        }
158        if !self.api_prefix.is_empty() && !self.api_prefix.starts_with('/') {
159            return Err(ServerError::InvalidConfig(
160                "api_prefix must start with '/' or be empty".into(),
161            ));
162        }
163        if self.max_response_size == 0 {
164            return Err(ServerError::InvalidConfig(
165                "max_response_size must be greater than 0".into(),
166            ));
167        }
168        if self.max_request_size == 0 {
169            return Err(ServerError::InvalidConfig(
170                "max_request_size must be greater than 0".into(),
171            ));
172        }
173        if self.max_concurrency == 0 {
174            return Err(ServerError::InvalidConfig(
175                "max_concurrency must be greater than 0".into(),
176            ));
177        }
178        if self.max_concurrency > MAX_ADMISSION_LIMIT {
179            return Err(ServerError::InvalidConfig(format!(
180                "max_concurrency must be <= {MAX_ADMISSION_LIMIT}"
181            )));
182        }
183        if self.max_queue_len == 0 {
184            return Err(ServerError::InvalidConfig(
185                "max_queue_len must be greater than 0".into(),
186            ));
187        }
188        if self.max_queue_len > MAX_ADMISSION_LIMIT {
189            return Err(ServerError::InvalidConfig(format!(
190                "max_queue_len must be <= {MAX_ADMISSION_LIMIT}"
191            )));
192        }
193        let query_timeout_ms = self.query_timeout.as_millis();
194        if query_timeout_ms == 0 {
195            return Err(ServerError::InvalidConfig(
196                "query_timeout must be greater than 0ms".into(),
197            ));
198        }
199        if query_timeout_ms > MAX_QUERY_TIMEOUT_MS {
200            return Err(ServerError::InvalidConfig(format!(
201                "query_timeout must be <= {MAX_QUERY_TIMEOUT_MS}ms"
202            )));
203        }
204        let cluster_config = self.cluster_manager_config()?;
205        ClusterManager::new(cluster_config)
206            .map_err(|err| ServerError::InvalidConfig(err.to_string()))?;
207        Ok(())
208    }
209
210    /// Build cluster manager configuration from server configuration.
211    pub fn cluster_manager_config(&self) -> Result<ClusterManagerConfig> {
212        self.cluster.to_manager_config()
213    }
214
215    fn normalize(&mut self) -> Result<()> {
216        if self.api_prefix == "/" {
217            self.api_prefix.clear();
218        } else if self.api_prefix.ends_with('/') {
219            while self.api_prefix.ends_with('/') {
220                self.api_prefix.pop();
221            }
222        }
223        self.validate()
224    }
225}
226
227impl ClusterServerConfig {
228    fn to_manager_config(&self) -> Result<ClusterManagerConfig> {
229        match self.mode {
230            ClusterMode::SingleNode => Ok(ClusterManagerConfig::single_node()),
231            ClusterMode::ClusterAware => {
232                let mut config =
233                    ClusterManagerConfig::cluster_aware(self.cluster_aware_identity()?);
234                config.membership_source = self.membership_source;
235                config.membership_source_available = self.membership_source_available;
236                Ok(config)
237            }
238        }
239    }
240
241    fn cluster_aware_identity(&self) -> Result<ClusterIdentity> {
242        let node_id = required_cluster_value(self.node_id.as_deref(), "cluster.node_id")?;
243        let cluster_id = required_cluster_value(self.cluster_id.as_deref(), "cluster.cluster_id")?;
244        let endpoint = required_cluster_value(
245            self.advertised_endpoint.as_deref(),
246            "cluster.advertised_endpoint",
247        )?;
248
249        Ok(ClusterIdentity {
250            node_id: NodeId::new(node_id),
251            cluster_id: Some(ClusterId::new(cluster_id)),
252            advertised_endpoint: Some(Endpoint::new(endpoint)),
253            role: self.role,
254            lifecycle_state: self.lifecycle_state,
255            metadata_schema_version: alopex_cluster::CLUSTER_METADATA_SCHEMA_VERSION,
256            update_epoch: alopex_cluster::INITIAL_UPDATE_EPOCH,
257        })
258    }
259}
260
261fn required_cluster_value(value: Option<&str>, field: &'static str) -> Result<String> {
262    let Some(value) = value else {
263        return Err(ServerError::InvalidConfig(format!(
264            "{field} is required when cluster.mode is cluster_aware"
265        )));
266    };
267    let trimmed = value.trim();
268    if trimmed.is_empty() {
269        return Err(ServerError::InvalidConfig(format!(
270            "{field} must be non-empty when cluster.mode is cluster_aware"
271        )));
272    }
273    Ok(trimmed.to_string())
274}
275
276#[cfg(test)]
277mod tests {
278    use super::*;
279
280    #[test]
281    fn default_admission_control_values_match_v06_policy() {
282        let cfg = ServerConfig::default();
283        assert_eq!(cfg.max_concurrency, 64);
284        assert_eq!(cfg.max_queue_len, 256);
285        assert_eq!(cfg.query_timeout.as_millis(), 30_000);
286    }
287
288    #[test]
289    fn validate_rejects_zero_or_excessive_admission_values() {
290        let cfg = ServerConfig {
291            max_concurrency: 0,
292            ..ServerConfig::default()
293        };
294        assert!(cfg.validate().is_err());
295
296        let cfg = ServerConfig {
297            max_queue_len: 0,
298            ..ServerConfig::default()
299        };
300        assert!(cfg.validate().is_err());
301
302        let cfg = ServerConfig {
303            max_concurrency: MAX_ADMISSION_LIMIT + 1,
304            ..ServerConfig::default()
305        };
306        assert!(cfg.validate().is_err());
307
308        let cfg = ServerConfig {
309            max_queue_len: MAX_ADMISSION_LIMIT + 1,
310            ..ServerConfig::default()
311        };
312        assert!(cfg.validate().is_err());
313
314        let cfg = ServerConfig {
315            query_timeout: Duration::from_millis(0),
316            ..ServerConfig::default()
317        };
318        assert!(cfg.validate().is_err());
319
320        let cfg = ServerConfig {
321            query_timeout: Duration::from_millis((MAX_QUERY_TIMEOUT_MS + 1) as u64),
322            ..ServerConfig::default()
323        };
324        assert!(cfg.validate().is_err());
325    }
326
327    #[test]
328    fn load_supports_legacy_max_connections_alias() {
329        let raw = r#"
330http_bind = "127.0.0.1:8080"
331grpc_bind = "127.0.0.1:9090"
332admin_bind = "127.0.0.1:8081"
333data_dir = "./data"
334api_prefix = ""
335query_timeout = "30s"
336max_connections = 77
337max_queue_len = 256
338max_request_size = 1048576
339max_response_size = 1048576
340session_ttl = "300s"
341metrics_enabled = true
342tracing_enabled = true
343audit_log_enabled = false
344
345[auth_mode]
346type = "none"
347
348[audit_log_output]
349type = "stdout"
350"#;
351
352        let built = config::Config::builder()
353            .add_source(config::File::from_str(raw, config::FileFormat::Toml))
354            .build()
355            .expect("build config");
356        let cfg: ServerConfig = built.try_deserialize().expect("deserialize");
357        assert_eq!(cfg.max_concurrency, 77);
358    }
359
360    #[test]
361    fn default_cluster_config_is_single_node_non_degraded() {
362        let cfg = ServerConfig::default();
363        let manager = ClusterManager::new(cfg.cluster_manager_config().unwrap()).unwrap();
364        let snapshot = manager.status_snapshot();
365
366        assert_eq!(snapshot.mode, ClusterMode::SingleNode);
367        assert_eq!(snapshot.identity.node_id.as_str(), "local");
368        assert_eq!(snapshot.identity.lifecycle_state, NodeState::Unconfigured);
369        assert_eq!(snapshot.membership.source, MembershipSource::LocalDefault);
370        assert!(!snapshot.degraded);
371        assert!(snapshot.diagnostics.is_empty());
372    }
373
374    #[test]
375    fn cluster_aware_config_builds_configured_identity() {
376        let cfg = ServerConfig {
377            cluster: ClusterServerConfig {
378                mode: ClusterMode::ClusterAware,
379                node_id: Some("node-a".to_string()),
380                cluster_id: Some("cluster-a".to_string()),
381                advertised_endpoint: Some("127.0.0.1:7001".to_string()),
382                role: NodeRole::Worker,
383                lifecycle_state: NodeState::Joining,
384                membership_source_available: false,
385                ..ClusterServerConfig::default()
386            },
387            ..ServerConfig::default()
388        };
389
390        let manager = ClusterManager::new(cfg.cluster_manager_config().unwrap()).unwrap();
391        let snapshot = manager.status_snapshot();
392
393        assert_eq!(snapshot.mode, ClusterMode::ClusterAware);
394        assert_eq!(snapshot.identity.node_id.as_str(), "node-a");
395        assert_eq!(snapshot.identity.cluster_id.unwrap().as_str(), "cluster-a");
396        assert_eq!(
397            snapshot.identity.advertised_endpoint.unwrap().as_str(),
398            "127.0.0.1:7001"
399        );
400        assert_eq!(snapshot.identity.role, NodeRole::Worker);
401        assert_eq!(snapshot.identity.lifecycle_state, NodeState::Joining);
402        assert_eq!(snapshot.membership.source, MembershipSource::Chirps);
403        assert!(snapshot.degraded);
404    }
405
406    #[test]
407    fn validate_rejects_invalid_cluster_identity() {
408        let cfg = ServerConfig {
409            cluster: ClusterServerConfig {
410                mode: ClusterMode::ClusterAware,
411                node_id: Some("".to_string()),
412                cluster_id: Some("cluster-a".to_string()),
413                advertised_endpoint: Some("127.0.0.1:7001".to_string()),
414                ..ClusterServerConfig::default()
415            },
416            ..ServerConfig::default()
417        };
418
419        let err = cfg.validate().unwrap_err();
420        assert!(err.to_string().contains("cluster.node_id"));
421    }
422}
alopex_server/config.rs

alopex_server/
config.rs
