1use std::net::SocketAddr;
4
5use crate::{
6 config::{ServerConfig, StoreBackend, config_error, sections::RetiredStoreInput},
7 error::ServerError,
8};
9
10#[derive(Clone, Debug, Eq, PartialEq)]
20pub(crate) struct RetiredVarNotice {
21 pub(crate) variable: String,
23}
24
25impl RetiredVarNotice {
26 fn warn(&self) {
29 tracing::warn!(
30 variable = %self.variable,
31 "retired and ignored: the server waits indefinitely for the \
32 data-directory writer lock and reports while it waits. \
33 Unset this variable"
34 );
35 }
36}
37
38pub fn overlay(config: &mut ServerConfig) -> Result<(), ServerError> {
48 for notice in overlay_vars(config, std::env::vars())? {
49 notice.warn();
50 }
51 Ok(())
52}
53
54pub(crate) fn overlay_vars(
63 config: &mut ServerConfig,
64 vars: impl IntoIterator<Item = (String, String)>,
65) -> Result<Vec<RetiredVarNotice>, ServerError> {
66 let mut notices = Vec::new();
67 for (name, value) in vars {
68 match name.as_str() {
69 "AION_SERVER_LISTEN_ADDRESS" => {
70 config.server.listen_address = parse_socket_addr(&name, &value)?;
71 }
72 "AION_SERVER_GRPC_ADDRESS" => {
73 config.server.grpc_address = parse_socket_addr(&name, &value)?;
74 }
75 "AION_SERVER_CORS_ALLOWED_ORIGINS" => {
76 config.server.cors_allowed_origins = parse_csv_origins(&value);
77 }
78 "AION_STORE_BACKEND" => {
79 if value.eq_ignore_ascii_case("libsql") {
86 config.store.retired_input = Some(RetiredStoreInput::BackendEnvironment);
87 } else {
88 config.store.backend = parse_store_backend(&name, &value)?;
89 }
90 }
91 "AION_STORE_URL" => {
92 config.store.retired_input = Some(RetiredStoreInput::Environment);
93 }
94 "AION_STORE_DATA_DIR" => {
95 if value.is_empty() {
96 return config_error("AION_STORE_DATA_DIR must not be empty");
97 }
98 config.store.data_dir = Some(value);
99 }
100 "AION_STORE_SHARD_COUNT" => {
101 config.store.shard_count = parse_positive_usize(&name, &value)?;
102 }
103 "AION_STORE_NODE_CACHE_BUDGET" => {
104 config.store.node_cache_budget = Some(parse_node_cache_budget(&name, &value)?);
105 }
106 "AION_STORE_LOCK_ACQUISITION_PATIENCE_MS"
112 | "AION_STORE_LOCK_ACQUISITION_RETRY_CADENCE_MS" => {
113 notices.push(RetiredVarNotice {
114 variable: name.clone(),
115 });
116 }
117 "AION_RUNTIME_SCHEDULER_THREADS" => {
118 config.runtime.scheduler_threads = parse_positive_usize(&name, &value)?;
119 }
120 "AION_RUNTIME_JIT_THRESHOLD" => {
121 config.runtime.jit_threshold = Some(parse_positive_u32(&name, &value)?);
122 }
123 "AION_RUNTIME_QUERY_TIMEOUT_MS" => {
124 config.runtime.query_timeout_ms = Some(parse_positive_u64(&name, &value)?);
125 }
126 "AION_RUNTIME_WORKLOOP_SWEEP_INTERVAL_MS" => {
127 config.runtime.workloop_sweep_interval_ms =
128 Some(parse_positive_u64(&name, &value)?);
129 }
130 "AION_RUNTIME_STOP_DRAIN_TIMEOUT_MS" => {
131 config.runtime.stop_drain_timeout_ms = Some(parse_positive_u64(&name, &value)?);
132 }
133 "AION_DRAIN_TIMEOUT_SECONDS" => {
134 config.drain.timeout_seconds = parse_positive_u64(&name, &value)?;
135 }
136 "AION_AUTH_ENABLED" => {
137 config.auth.enabled = parse_bool(&name, &value)?;
138 }
139 "AION_AUTH_JWKS_URL" => {
140 if value.is_empty() {
141 return config_error("AION_AUTH_JWKS_URL must not be empty");
142 }
143 config.auth.jwks_url = Some(value);
144 }
145 "AION_AUTH_JWKS_REFRESH_SECONDS" => {
146 config.auth.jwks_refresh_seconds = parse_positive_u64(&name, &value)?;
147 }
148 "AION_METRICS_ENABLED" => {
149 config.metrics.enabled = parse_bool(&name, &value)?;
150 }
151 "AION_WEBSOCKET_OUTBOUND_BUFFER_BOUND" => {
152 config.websocket.outbound_buffer_bound = parse_positive_usize(&name, &value)?;
153 }
154 "AION_DEPLOY_ENABLED" => {
155 config.deploy.enabled = parse_bool(&name, &value)?;
156 }
157 "AION_DEPLOY_MAX_ARCHIVE_BYTES" => {
158 config.deploy.max_archive_bytes = Some(parse_positive_u64(&name, &value)?);
159 }
160 "AION_DEPLOY_MAX_INFLATED_BYTES" => {
161 config.deploy.max_inflated_bytes = Some(parse_positive_u64(&name, &value)?);
162 }
163 "AION_DEV_ENABLED" => {
164 config.dev.enabled = parse_bool(&name, &value)?;
165 }
166 other => overlay_authoring(config, other, &value)?,
167 }
168 }
169 Ok(notices)
170}
171
172fn overlay_authoring(
178 config: &mut ServerConfig,
179 name: &str,
180 value: &str,
181) -> Result<(), ServerError> {
182 match name {
183 "AION_AUTHORING_GLEAM_PATH" => {
184 if value.is_empty() {
185 return config_error("AION_AUTHORING_GLEAM_PATH must not be empty");
186 }
187 config.authoring.gleam_path = Some(std::path::PathBuf::from(value));
188 }
189 "AION_AUTHORING_PROJECT_ROOT" => {
190 if value.is_empty() {
191 return config_error("AION_AUTHORING_PROJECT_ROOT must not be empty");
192 }
193 config.authoring.project_root = Some(std::path::PathBuf::from(value));
194 }
195 "AION_AUTHORING_WORKSPACE_DIR" => {
196 if value.is_empty() {
197 return config_error("AION_AUTHORING_WORKSPACE_DIR must not be empty");
198 }
199 config.authoring.workspace_dir = Some(std::path::PathBuf::from(value));
200 }
201 "AION_NAMESPACES_DEFAULT" => {
202 if value.is_empty() {
203 return config_error("AION_NAMESPACES_DEFAULT must not be empty");
204 }
205 value.clone_into(&mut config.namespaces.default);
206 }
207 other => overlay_websocket(config, other, value)?,
208 }
209 Ok(())
210}
211
212fn overlay_websocket(
218 config: &mut ServerConfig,
219 name: &str,
220 value: &str,
221) -> Result<(), ServerError> {
222 match name {
223 "AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY" => {
224 config.websocket.event_broadcast_capacity = Some(parse_positive_usize(name, value)?);
225 }
226 "AION_WEBSOCKET_CLUSTER_BROADCAST_CAPACITY" => {
227 config.websocket.cluster_broadcast_capacity = Some(parse_positive_usize(name, value)?);
228 }
229 other => overlay_observability(config, other, value)?,
230 }
231 Ok(())
232}
233
234fn parse_node_cache_budget(
254 name: &str,
255 value: &str,
256) -> Result<haematite::NodeCacheBudget, ServerError> {
257 #[derive(serde::Deserialize)]
259 struct Document {
260 node_cache_budget: haematite::NodeCacheBudget,
261 }
262
263 let document: Document =
264 toml::from_str(&format!("node_cache_budget = {value}")).map_err(|error| {
265 ServerError::Config {
266 message: format!(
267 "{name} must be a node cache budget written exactly as it would be in \
268 config.toml — `{{ bytes = <positive integer> }}` or `\"unlimited\"` — got \
269 `{value}`: {error}"
270 ),
271 }
272 })?;
273 Ok(document.node_cache_budget)
274}
275
276fn overlay_observability(
277 config: &mut ServerConfig,
278 name: &str,
279 value: &str,
280) -> Result<(), ServerError> {
281 match name {
282 "AION_OBSERVABILITY_MAX_EVENT_BYTES" => {
283 config.observability.max_event_bytes = parse_positive_usize(name, value)?;
284 }
285 "AION_OBSERVABILITY_MAX_STREAM_EVENTS" => {
286 config.observability.max_stream_events = parse_positive_u64(name, value)?;
287 }
288 "AION_OBSERVABILITY_MAX_BATCH_EVENTS" => {
289 config.observability.max_batch_events = Some(parse_positive_usize(name, value)?);
290 }
291 "AION_OBSERVABILITY_MAX_BATCH_HOLD_MS" => {
292 config.observability.max_batch_hold_ms = Some(parse_u64(name, value)?);
295 }
296 other => overlay_outbox(config, other, value)?,
297 }
298 Ok(())
299}
300
301fn overlay_outbox(config: &mut ServerConfig, name: &str, value: &str) -> Result<(), ServerError> {
308 match name {
309 "AION_OUTBOX_ENABLED" => {
310 config.outbox.enabled = parse_bool(name, value)?;
311 }
312 "AION_OUTBOX_POLL_INTERVAL_MS" => {
313 config.outbox.poll_interval_ms = Some(parse_positive_u64(name, value)?);
314 }
315 "AION_OUTBOX_BATCH_SIZE" => {
316 config.outbox.batch_size = Some(parse_positive_u32(name, value)?);
317 }
318 "AION_OUTBOX_MAX_ATTEMPTS" => {
319 config.outbox.max_attempts = Some(parse_positive_u32(name, value)?);
320 }
321 "AION_OUTBOX_BACKOFF_BASE_MS" => {
322 config.outbox.backoff_base_ms = Some(parse_positive_u64(name, value)?);
323 }
324 "AION_OUTBOX_BACKOFF_MULTIPLIER" => {
325 config.outbox.backoff_multiplier = Some(parse_positive_u32(name, value)?);
326 }
327 "AION_OUTBOX_BACKOFF_MAX_MS" => {
328 config.outbox.backoff_max_ms = Some(parse_positive_u64(name, value)?);
329 }
330 "AION_OUTBOX_RECONCILE_INTERVAL_MS" => {
331 config.outbox.reconcile_interval_ms = Some(parse_positive_u64(name, value)?);
332 }
333 "AION_OUTBOX_RECONCILE_STALE_AFTER_MS" => {
334 config.outbox.reconcile_stale_after_ms = Some(parse_positive_u64(name, value)?);
335 }
336 "AION_OUTBOX_LIMINAL_LISTEN_ADDRESS" => {
337 config.outbox.liminal_listen_address = Some(value.to_owned());
338 }
339 "AION_OUTBOX_LIMINAL_MAX_CONNECTION_OUTBOUND_BYTES" => {
340 config.outbox.liminal_max_connection_outbound_bytes =
341 Some(parse_positive_u64(name, value)?);
342 }
343 _ => {}
344 }
345 Ok(())
346}
347
348fn parse_csv_origins(value: &str) -> Vec<String> {
353 value
354 .split(',')
355 .map(str::trim)
356 .filter(|origin| !origin.is_empty())
357 .map(str::to_owned)
358 .collect()
359}
360
361fn parse_socket_addr(name: &str, value: &str) -> Result<SocketAddr, ServerError> {
362 value.parse().map_err(|source| ServerError::Config {
363 message: format!("{name} must be a socket address: {source}"),
364 })
365}
366
367fn parse_store_backend(name: &str, value: &str) -> Result<StoreBackend, ServerError> {
368 match value.to_ascii_lowercase().as_str() {
369 "memory" => Ok(StoreBackend::Memory),
370 "haematite" => Ok(StoreBackend::Haematite),
371 _ => config_error(format!("{name} must be one of: memory, haematite")),
372 }
373}
374
375fn parse_positive_usize(name: &str, value: &str) -> Result<usize, ServerError> {
376 let parsed = value
377 .parse::<usize>()
378 .map_err(|source| ServerError::Config {
379 message: format!("{name} must be a positive integer: {source}"),
380 })?;
381 if parsed == 0 {
382 return config_error(format!("{name} must be a positive integer"));
383 }
384 Ok(parsed)
385}
386
387fn parse_positive_u32(name: &str, value: &str) -> Result<u32, ServerError> {
388 let parsed = value.parse::<u32>().map_err(|source| ServerError::Config {
389 message: format!("{name} must be a positive integer: {source}"),
390 })?;
391 if parsed == 0 {
392 return config_error(format!("{name} must be a positive integer"));
393 }
394 Ok(parsed)
395}
396
397fn parse_positive_u64(name: &str, value: &str) -> Result<u64, ServerError> {
398 let parsed = value.parse::<u64>().map_err(|source| ServerError::Config {
399 message: format!("{name} must be a positive integer: {source}"),
400 })?;
401 if parsed == 0 {
402 return config_error(format!("{name} must be a positive integer"));
403 }
404 Ok(parsed)
405}
406
407fn parse_u64(name: &str, value: &str) -> Result<u64, ServerError> {
411 value.parse::<u64>().map_err(|source| ServerError::Config {
412 message: format!("{name} must be a non-negative integer: {source}"),
413 })
414}
415
416fn parse_bool(name: &str, value: &str) -> Result<bool, ServerError> {
417 match value.to_ascii_lowercase().as_str() {
418 "true" | "1" | "yes" | "on" => Ok(true),
419 "false" | "0" | "no" | "off" => Ok(false),
420 _ => config_error(format!("{name} must be a boolean")),
421 }
422}
423
424#[cfg(test)]
425mod tests {
426 use super::{RetiredVarNotice, overlay_vars, parse_node_cache_budget};
427
428 #[test]
432 fn the_outbound_bound_env_override_threads_and_refuses_zero()
433 -> Result<(), Box<dyn std::error::Error>> {
434 let var = "AION_OUTBOX_LIMINAL_MAX_CONNECTION_OUTBOUND_BYTES";
435 let mut config = crate::config::ServerConfig::default();
436 overlay_vars(&mut config, [(var.to_owned(), "8388608".to_owned())])?;
437 assert_eq!(
438 config.outbox.liminal_max_connection_outbound_bytes,
439 Some(8_388_608)
440 );
441 for bad in ["0", "four-megabytes"] {
442 let mut config = crate::config::ServerConfig::default();
443 let refusal = overlay_vars(&mut config, [(var.to_owned(), bad.to_owned())])
444 .err()
445 .ok_or_else(|| format!("{bad:?} must be refused"))?;
446 assert!(refusal.to_string().contains(var), "{refusal}");
447 assert_eq!(config.outbox.liminal_max_connection_outbound_bytes, None);
448 }
449 Ok(())
450 }
451
452 #[test]
458 fn a_retired_variable_is_returned_as_data_not_logged() -> Result<(), Box<dyn std::error::Error>>
459 {
460 let mut config = crate::config::ServerConfig::default();
461 let (captured, notices) = crate::test_support::CapturedLogs::capture(|| {
462 overlay_vars(
463 &mut config,
464 [(
465 "AION_STORE_LOCK_ACQUISITION_PATIENCE_MS".to_owned(),
466 "60000".to_owned(),
467 )],
468 )
469 });
470 let notices = notices?;
471 assert_eq!(
472 notices,
473 vec![RetiredVarNotice {
474 variable: "AION_STORE_LOCK_ACQUISITION_PATIENCE_MS".to_owned(),
475 }],
476 "the retired variable must surface as exactly one notice"
477 );
478 let logged = captured.text()?;
479 assert!(
480 logged.is_empty(),
481 "the overlay parse must emit nothing itself: {logged}"
482 );
483 Ok(())
484 }
485
486 #[test]
490 fn the_emit_site_warns_once_naming_the_variable() -> Result<(), Box<dyn std::error::Error>> {
491 let notice = RetiredVarNotice {
492 variable: "AION_STORE_LOCK_ACQUISITION_RETRY_CADENCE_MS".to_owned(),
493 };
494 let (captured, ()) = crate::test_support::CapturedLogs::capture(|| notice.warn());
495 let logged = captured.text()?;
496 assert_eq!(
497 logged.matches("retired and ignored").count(),
498 1,
499 "one notice, one warning: {logged}"
500 );
501 assert!(
502 logged.contains("AION_STORE_LOCK_ACQUISITION_RETRY_CADENCE_MS"),
503 "the warning must name the variable: {logged}"
504 );
505 Ok(())
506 }
507
508 #[test]
512 fn the_env_override_accepts_both_haematite_spellings() -> Result<(), Box<dyn std::error::Error>>
513 {
514 let name = "AION_STORE_NODE_CACHE_BUDGET";
515 assert_eq!(
516 parse_node_cache_budget(name, "{ bytes = 1073741824 }")?,
517 haematite::NodeCacheBudget::bytes(1 << 30)?,
518 "a 1 GiB ceiling written as it would be in config.toml"
519 );
520 assert_eq!(
521 parse_node_cache_budget(name, "\"unlimited\"")?,
522 haematite::NodeCacheBudget::Unlimited,
523 "the pre-budget behaviour, spelled out loud"
524 );
525 Ok(())
526 }
527
528 #[test]
533 fn the_env_override_refuses_a_zero_ceiling() -> Result<(), Box<dyn std::error::Error>> {
534 let error = parse_node_cache_budget("AION_STORE_NODE_CACHE_BUDGET", "{ bytes = 0 }")
535 .err()
536 .ok_or("a zero byte ceiling must be refused, not accepted as 'no cache'")?;
537 let crate::error::ServerError::Config { message } = error else {
538 return Err("a bad env value must be a config refusal".into());
539 };
540 assert!(
541 message.contains("AION_STORE_NODE_CACHE_BUDGET"),
542 "the refusal must name the variable, got: {message}"
543 );
544 assert!(
545 message.contains("greater than zero"),
546 "the refusal must carry haematite's own reason, got: {message}"
547 );
548 Ok(())
549 }
550
551 #[test]
554 fn the_env_override_refuses_junk() -> Result<(), Box<dyn std::error::Error>> {
555 let error = parse_node_cache_budget("AION_STORE_NODE_CACHE_BUDGET", "1GiB")
556 .err()
557 .ok_or("`1GiB` is not a spelling haematite accepts and must be refused")?;
558 let crate::error::ServerError::Config { message } = error else {
559 return Err("a bad env value must be a config refusal".into());
560 };
561 assert!(
562 message.contains("unlimited") && message.contains("bytes"),
563 "the refusal must show the accepted spellings, got: {message}"
564 );
565 Ok(())
566 }
567}