liminal_server/config/
env.rs

1use std::ffi::{OsStr, OsString};
2use std::net::SocketAddr;
3use std::path::PathBuf;
4
5use crate::ServerError;
6
7use super::types::ServerConfig;
8
9const ENV_PREFIX: &str = "LIMINAL_";
10const LISTEN_ADDRESS: &str = "LIMINAL_LISTEN_ADDRESS";
11const HEALTH_LISTEN_ADDRESS: &str = "LIMINAL_HEALTH_LISTEN_ADDRESS";
12const DRAIN_TIMEOUT_MS: &str = "LIMINAL_DRAIN_TIMEOUT_MS";
13const PERSISTENCE_PATH: &str = "LIMINAL_PERSISTENCE_PATH";
14const CLUSTER_NODE_NAME: &str = "LIMINAL_CLUSTER_NODE_NAME";
15const CLUSTER_SEED_NODES: &str = "LIMINAL_CLUSTER_SEED_NODES";
16const CLUSTER_LISTEN_ADDRESS: &str = "LIMINAL_CLUSTER_LISTEN_ADDRESS";
17const CLUSTER_COOKIE: &str = "LIMINAL_CLUSTER_COOKIE";
18const AUTH_TOKEN: &str = "LIMINAL_AUTH_TOKEN";
19
20/// Applies supported `LIMINAL_` environment variable overrides to a config.
21///
22/// Absent variables leave the supplied configuration unchanged. Supported
23/// variables are `LIMINAL_LISTEN_ADDRESS`, `LIMINAL_HEALTH_LISTEN_ADDRESS`,
24/// `LIMINAL_DRAIN_TIMEOUT_MS`, `LIMINAL_PERSISTENCE_PATH`,
25/// `LIMINAL_CLUSTER_NODE_NAME`, `LIMINAL_CLUSTER_SEED_NODES`,
26/// `LIMINAL_CLUSTER_LISTEN_ADDRESS`, `LIMINAL_CLUSTER_COOKIE`, and
27/// `LIMINAL_AUTH_TOKEN`.
28///
29/// Unlike the cluster overrides — which refuse to fabricate a `[cluster]` section
30/// that the file did not declare (a partially-specified cluster is unsafe) —
31/// `LIMINAL_AUTH_TOKEN` MAY create the `[auth]` section when the file omits it.
32/// The auth section is a single scalar secret with no other required fields, so a
33/// deployment can inject the token purely from the environment (the idiomatic way
34/// to supply a secret) without also pinning it in a committed config file; an empty
35/// value still fails the later non-empty validation, so this cannot silently create
36/// a no-op gate.
37///
38/// # Errors
39///
40/// Returns [`ServerError`] when a present environment variable cannot be parsed
41/// or targets cluster configuration that was not declared in the file.
42pub fn apply_env_overrides(config: ServerConfig) -> Result<ServerConfig, ServerError> {
43    apply_env_overrides_from(config, std::env::vars_os())
44}
45
46pub(crate) fn apply_env_overrides_from<I>(
47    mut config: ServerConfig,
48    variables: I,
49) -> Result<ServerConfig, ServerError>
50where
51    I: IntoIterator<Item = (OsString, OsString)>,
52{
53    for (key, value) in variables {
54        let Some(key) = key.to_str() else {
55            continue;
56        };
57
58        if !key.starts_with(ENV_PREFIX) {
59            continue;
60        }
61
62        match key {
63            LISTEN_ADDRESS => {
64                config.listen_address = parse_socket_addr(LISTEN_ADDRESS, &value)?;
65            }
66            HEALTH_LISTEN_ADDRESS => {
67                config.health_listen_address = parse_socket_addr(HEALTH_LISTEN_ADDRESS, &value)?;
68            }
69            DRAIN_TIMEOUT_MS => {
70                config.drain_timeout_ms = parse_u64(DRAIN_TIMEOUT_MS, &value)?;
71            }
72            PERSISTENCE_PATH => {
73                config.persistence_path = Some(PathBuf::from(value));
74            }
75            CLUSTER_NODE_NAME => {
76                let node_name = env_string(CLUSTER_NODE_NAME, &value)?;
77                cluster_required(&mut config, CLUSTER_NODE_NAME)?.node_name = node_name;
78            }
79            CLUSTER_SEED_NODES => {
80                let seed_nodes = parse_seed_nodes(&value)?;
81                cluster_required(&mut config, CLUSTER_SEED_NODES)?.seed_nodes = seed_nodes;
82            }
83            CLUSTER_LISTEN_ADDRESS => {
84                let listen_address = parse_socket_addr(CLUSTER_LISTEN_ADDRESS, &value)?;
85                cluster_required(&mut config, CLUSTER_LISTEN_ADDRESS)?.listen_address =
86                    listen_address;
87            }
88            CLUSTER_COOKIE => {
89                let cookie = env_string(CLUSTER_COOKIE, &value)?;
90                cluster_required(&mut config, CLUSTER_COOKIE)?.cookie = cookie;
91            }
92            AUTH_TOKEN => {
93                // A single scalar secret is allowed to create the `[auth]` section
94                // even when the file omits it (see the module-level note); an empty
95                // value is left to fail the non-empty auth validation, not rejected
96                // here, so precedence and error reporting stay uniform.
97                let token = env_string(AUTH_TOKEN, &value)?;
98                config.auth = Some(super::types::AuthConfig { token });
99            }
100            _ => {}
101        }
102    }
103
104    Ok(config)
105}
106
107fn parse_socket_addr(name: &str, value: &OsStr) -> Result<SocketAddr, ServerError> {
108    let value = env_string(name, value)?;
109    value.parse::<SocketAddr>().map_err(|error| {
110        config_load(format!(
111            "environment variable {name} must be a socket address: {error}"
112        ))
113    })
114}
115
116fn parse_u64(name: &str, value: &OsStr) -> Result<u64, ServerError> {
117    let value = env_string(name, value)?;
118    value.parse::<u64>().map_err(|error| {
119        config_load(format!(
120            "environment variable {name} must be an unsigned integer: {error}"
121        ))
122    })
123}
124
125fn parse_seed_nodes(value: &OsStr) -> Result<Vec<SocketAddr>, ServerError> {
126    let value = env_string(CLUSTER_SEED_NODES, value)?;
127    if value.trim().is_empty() {
128        return Ok(Vec::new());
129    }
130
131    value
132        .split(',')
133        .enumerate()
134        .map(|(index, candidate)| parse_seed_node(index, candidate))
135        .collect()
136}
137
138fn parse_seed_node(index: usize, candidate: &str) -> Result<SocketAddr, ServerError> {
139    let candidate = candidate.trim();
140    if candidate.is_empty() {
141        return Err(config_load(format!(
142            "environment variable {CLUSTER_SEED_NODES} contains an empty seed node at position {}",
143            index + 1
144        )));
145    }
146
147    candidate.parse::<SocketAddr>().map_err(|error| {
148        config_load(format!(
149            "environment variable {CLUSTER_SEED_NODES} contains invalid seed node '{}' at position {}: {error}",
150            candidate,
151            index + 1
152        ))
153    })
154}
155
156fn env_string(name: &str, value: &OsStr) -> Result<String, ServerError> {
157    value.to_str().map(str::to_owned).ok_or_else(|| {
158        config_load(format!(
159            "environment variable {name} contains non-Unicode data"
160        ))
161    })
162}
163
164fn cluster_required<'a>(
165    config: &'a mut ServerConfig,
166    name: &str,
167) -> Result<&'a mut super::types::ClusterConfig, ServerError> {
168    config
169        .cluster
170        .as_mut()
171        .ok_or_else(|| ServerError::ConfigValidation {
172            message: format!(
173                "environment variable {name} requires a [cluster] section in the configuration file"
174            ),
175        })
176}
177
178const fn config_load(message: String) -> ServerError {
179    ServerError::ConfigLoad { message }
180}
181
182#[cfg(test)]
183mod tests {
184    use std::ffi::OsString;
185    use std::net::SocketAddr;
186    use std::path::{Path, PathBuf};
187
188    use crate::ServerError;
189
190    use super::apply_env_overrides_from;
191    use crate::config::types::{ChannelDef, ClusterConfig, RoutingRuleDef, ServerConfig};
192    use crate::config::{load_from_file, validate};
193
194    fn socket(address: &str) -> Result<SocketAddr, Box<dyn std::error::Error>> {
195        Ok(address.parse()?)
196    }
197
198    fn sample_config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
199        Ok(ServerConfig {
200            listen_address: socket("127.0.0.1:8080")?,
201            health_listen_address: socket("127.0.0.1:8081")?,
202            drain_timeout_ms: 30_000,
203            channels: vec![ChannelDef {
204                name: "orders".to_owned(),
205                schema_ref: None,
206                durable: true,
207                loaded_schema: None,
208            }],
209            routing_rules: vec![RoutingRuleDef {
210                source_channel: "orders".to_owned(),
211                target_channel: "orders".to_owned(),
212                predicate: None,
213            }],
214            persistence_path: Some(PathBuf::from("/tmp")),
215            cluster: Some(ClusterConfig {
216                node_name: "node-a".to_owned(),
217                listen_address: socket("127.0.0.1:9000")?,
218                seed_nodes: vec![socket("127.0.0.1:9001")?],
219                cookie: "test-cookie".to_owned(),
220            }),
221            auth: None,
222            services: crate::config::types::ServicesConfig::default(),
223            limits: crate::config::types::LimitsConfig::default(),
224        })
225    }
226
227    fn env_pair(name: &str, value: &str) -> (OsString, OsString) {
228        (OsString::from(name), OsString::from(value))
229    }
230
231    fn write_temp_config(contents: &str) -> Result<PathBuf, Box<dyn std::error::Error>> {
232        let path = std::env::temp_dir().join(format!(
233            "liminal-server-env-pipeline-{}.toml",
234            std::process::id()
235        ));
236        std::fs::write(&path, contents)?;
237        Ok(path)
238    }
239
240    fn remove_temp_file(path: &Path) -> Result<(), Box<dyn std::error::Error>> {
241        if path.exists() {
242            std::fs::remove_file(path)?;
243        }
244        Ok(())
245    }
246
247    #[test]
248    fn listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
249        let config = sample_config()?;
250        let config = apply_env_overrides_from(
251            config,
252            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
253        )?;
254
255        assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
256
257        Ok(())
258    }
259
260    #[test]
261    fn health_listen_address_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>>
262    {
263        let config = sample_config()?;
264        let config = apply_env_overrides_from(
265            config,
266            vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "0.0.0.0:9191")],
267        )?;
268
269        assert_eq!(config.health_listen_address, socket("0.0.0.0:9191")?);
270
271        Ok(())
272    }
273
274    #[test]
275    fn drain_timeout_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
276        let config = sample_config()?;
277        let config =
278            apply_env_overrides_from(config, vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "1250")])?;
279
280        assert_eq!(config.drain_timeout_ms, 1250);
281
282        Ok(())
283    }
284
285    #[test]
286    fn persistence_path_override_replaces_file_value() -> Result<(), Box<dyn std::error::Error>> {
287        let config = sample_config()?;
288        let config = apply_env_overrides_from(
289            config,
290            vec![env_pair("LIMINAL_PERSISTENCE_PATH", "/var/lib/liminal")],
291        )?;
292
293        assert_eq!(
294            config.persistence_path.as_deref(),
295            Some(Path::new("/var/lib/liminal"))
296        );
297
298        Ok(())
299    }
300
301    #[test]
302    fn cluster_overrides_replace_existing_cluster_values() -> Result<(), Box<dyn std::error::Error>>
303    {
304        let config = sample_config()?;
305        let config = apply_env_overrides_from(
306            config,
307            vec![
308                env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b"),
309                env_pair(
310                    "LIMINAL_CLUSTER_SEED_NODES",
311                    "127.0.0.1:9100, 127.0.0.1:9200",
312                ),
313            ],
314        )?;
315
316        let Some(cluster) = config.cluster else {
317            return Err("cluster config should remain present".into());
318        };
319        assert_eq!(cluster.node_name, "node-b");
320        assert_eq!(cluster.seed_nodes.len(), 2);
321        assert_eq!(cluster.seed_nodes[0], socket("127.0.0.1:9100")?);
322        assert_eq!(cluster.seed_nodes[1], socket("127.0.0.1:9200")?);
323
324        Ok(())
325    }
326
327    #[test]
328    fn cluster_listen_address_and_cookie_overrides_replace_values()
329    -> Result<(), Box<dyn std::error::Error>> {
330        let config = sample_config()?;
331        let config = apply_env_overrides_from(
332            config,
333            vec![
334                env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500"),
335                env_pair("LIMINAL_CLUSTER_COOKIE", "override-cookie"),
336            ],
337        )?;
338
339        let Some(cluster) = config.cluster else {
340            return Err("cluster config should remain present".into());
341        };
342        assert_eq!(cluster.listen_address, socket("127.0.0.1:9500")?);
343        assert_eq!(cluster.cookie, "override-cookie");
344
345        Ok(())
346    }
347
348    #[test]
349    fn cluster_listen_address_override_without_cluster_section_returns_validation_error()
350    -> Result<(), Box<dyn std::error::Error>> {
351        let mut config = sample_config()?;
352        config.cluster = None;
353        let result = apply_env_overrides_from(
354            config,
355            vec![env_pair("LIMINAL_CLUSTER_LISTEN_ADDRESS", "127.0.0.1:9500")],
356        );
357
358        assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
359
360        Ok(())
361    }
362
363    #[test]
364    fn auth_token_override_replaces_existing_token() -> Result<(), Box<dyn std::error::Error>> {
365        let mut config = sample_config()?;
366        config.auth = Some(crate::config::types::AuthConfig {
367            token: "file-token".to_owned(),
368        });
369
370        let config =
371            apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
372
373        let auth = config.auth.ok_or("auth section should remain present")?;
374        assert_eq!(auth.token, "env-token");
375
376        Ok(())
377    }
378
379    #[test]
380    fn auth_token_override_creates_missing_auth_section() -> Result<(), Box<dyn std::error::Error>>
381    {
382        let mut config = sample_config()?;
383        config.auth = None;
384
385        let config =
386            apply_env_overrides_from(config, vec![env_pair("LIMINAL_AUTH_TOKEN", "env-token")])?;
387
388        // Unlike cluster overrides, a scalar auth secret MAY fabricate the section.
389        let auth = config.auth.ok_or("auth section should have been created")?;
390        assert_eq!(auth.token, "env-token");
391
392        Ok(())
393    }
394
395    #[test]
396    fn absent_environment_variables_leave_config_unchanged()
397    -> Result<(), Box<dyn std::error::Error>> {
398        let config = sample_config()?;
399        let original_address = config.listen_address;
400        let original_health_address = config.health_listen_address;
401        let original_drain_timeout_ms = config.drain_timeout_ms;
402        let original_path = config.persistence_path.clone();
403        let original_cluster_name = config
404            .cluster
405            .as_ref()
406            .map(|cluster| cluster.node_name.clone());
407
408        let config = apply_env_overrides_from(config, Vec::new())?;
409
410        assert_eq!(config.listen_address, original_address);
411        assert_eq!(config.health_listen_address, original_health_address);
412        assert_eq!(config.drain_timeout_ms, original_drain_timeout_ms);
413        assert_eq!(config.persistence_path, original_path);
414        assert_eq!(
415            config
416                .cluster
417                .as_ref()
418                .map(|cluster| cluster.node_name.clone()),
419            original_cluster_name
420        );
421
422        Ok(())
423    }
424
425    #[test]
426    fn invalid_listen_address_override_returns_config_load()
427    -> Result<(), Box<dyn std::error::Error>> {
428        let config = sample_config()?;
429        let result = apply_env_overrides_from(
430            config,
431            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "not-a-socket")],
432        );
433
434        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
435
436        Ok(())
437    }
438
439    #[test]
440    fn invalid_health_listen_address_override_returns_config_load()
441    -> Result<(), Box<dyn std::error::Error>> {
442        let config = sample_config()?;
443        let result = apply_env_overrides_from(
444            config,
445            vec![env_pair("LIMINAL_HEALTH_LISTEN_ADDRESS", "not-a-socket")],
446        );
447
448        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
449
450        Ok(())
451    }
452
453    #[test]
454    fn invalid_drain_timeout_override_returns_config_load() -> Result<(), Box<dyn std::error::Error>>
455    {
456        let config = sample_config()?;
457        let result = apply_env_overrides_from(
458            config,
459            vec![env_pair("LIMINAL_DRAIN_TIMEOUT_MS", "not-a-number")],
460        );
461
462        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
463
464        Ok(())
465    }
466
467    #[test]
468    fn cluster_override_without_cluster_section_returns_validation_error()
469    -> Result<(), Box<dyn std::error::Error>> {
470        let mut config = sample_config()?;
471        config.cluster = None;
472        let result = apply_env_overrides_from(
473            config,
474            vec![env_pair("LIMINAL_CLUSTER_NODE_NAME", "node-b")],
475        );
476
477        assert!(matches!(result, Err(ServerError::ConfigValidation { .. })));
478
479        Ok(())
480    }
481
482    #[test]
483    fn file_then_env_then_validate_pipeline_gives_env_precedence()
484    -> Result<(), Box<dyn std::error::Error>> {
485        let toml = r#"
486listen_address = "127.0.0.1:8080"
487health_listen_address = "127.0.0.1:8081"
488drain_timeout_ms = 30000
489persistence_path = "/tmp"
490
491[[channels]]
492name = "orders"
493durable = true
494
495[[routing_rules]]
496source_channel = "orders"
497target_channel = "orders"
498"#;
499        let path = write_temp_config(toml)?;
500        let config = load_from_file(&path)?;
501        let mut config = apply_env_overrides_from(
502            config,
503            vec![env_pair("LIMINAL_LISTEN_ADDRESS", "0.0.0.0:9090")],
504        )?;
505        validate(&mut config, path.parent())?;
506        remove_temp_file(&path)?;
507
508        assert_eq!(config.listen_address, socket("0.0.0.0:9090")?);
509
510        Ok(())
511    }
512}
liminal_server/config/env.rs

liminal_server/config/
env.rs
