1use std::{
13 collections::HashSet,
14 fs,
15 path::{Path, PathBuf},
16 time::Duration,
17};
18
19use serde::Deserialize;
20
21use crate::error::ServerError;
22
23use super::{
24 AUTHORING_GLEAM_PATH_EMPTY, AUTHORING_PROJECT_ROOT_REQUIRED, AuthConfig, AuthoringConfig,
25 CORS_ALLOWED_ORIGIN_INVALID, CliOverrides, ClusterConfig, ConfigResolution,
26 DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
27 DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
28 DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
29 DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
30 DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED,
31 DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig, DrainConfig, HomeSource,
32 ListenConfig, McpConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
33 OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
34 OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
35 OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
36 OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
37 RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
38 WorkerConfig, WorkerSupervisionConfig, aion_home, config_error, env, file, home,
39 resolution::fill_home_defaults, sections::RetiredStoreInput,
40};
41
42#[derive(Clone, Debug, Deserialize)]
44#[serde(default, deny_unknown_fields)]
45#[derive(Default)]
46pub struct ServerConfig {
47 pub server: ServerSection,
49 pub store: StoreConfig,
51 pub runtime: RuntimeSection,
53 pub drain: DrainConfig,
55 pub auth: AuthConfig,
57 pub metrics: MetricsConfig,
59 pub namespaces: NamespacesConfig,
61 pub tls: Option<TlsConfig>,
63 #[serde(alias = "dashboard")]
65 pub ops_console: OpsConsoleConfig,
66 pub namespace: NamespaceConfig,
68 pub worker: WorkerConfig,
70 pub websocket: WebSocketConfig,
72 pub workflow_packages: Vec<PathBuf>,
74 pub deploy: DeployConfig,
76 pub authoring: AuthoringConfig,
78 pub dev: DevConfig,
80 pub outbox: OutboxConfig,
82 pub observability: ObservabilityConfig,
84 pub mcp: McpConfig,
86 pub worker_supervision: WorkerSupervisionConfig,
89}
90
91pub(crate) struct LoadedConfig {
92 pub(crate) config: ServerConfig,
93 pub(crate) resolution: ConfigResolution,
94}
95
96fn expand_operator_path_tildes(config: &mut ServerConfig) -> Result<(), ServerError> {
106 if let Some(data_dir) = config.store.data_dir.as_deref() {
107 let expanded = home::expand_tilde(Path::new(data_dir))?;
108 let expanded = expanded.to_str().ok_or_else(|| ServerError::Config {
109 message: format!(
110 "store.data_dir `{data_dir}` expands to a path that is not valid UTF-8"
111 ),
112 })?;
113 config.store.data_dir = Some(expanded.to_owned());
114 }
115 if let Some(workspace_dir) = config.authoring.workspace_dir.as_deref() {
116 config.authoring.workspace_dir = Some(home::expand_tilde(workspace_dir)?);
117 }
118 Ok(())
119}
120
121impl ServerConfig {
122 pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
129 Ok(Self::load_resolved(cli)?.config)
130 }
131
132 pub(crate) fn load_resolved(cli: &CliOverrides) -> Result<LoadedConfig, ServerError> {
133 let home = aion_home()?;
134 let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
135 message: format!(
136 "failed to resolve the current directory for config discovery: {source}"
137 ),
138 })?;
139 Self::load_in(cli, &home.path, home.source, &working_dir, true)
140 }
141
142 fn load_in(
143 cli: &CliOverrides,
144 home: &Path,
145 home_source: HomeSource,
146 working_dir: &Path,
147 overlay_environment: bool,
148 ) -> Result<LoadedConfig, ServerError> {
149 let explicit = cli
153 .config_path
154 .as_deref()
155 .map(super::home::expand_tilde)
156 .transpose()?;
157 let discovered = file::discover(explicit.as_deref(), home, working_dir)?;
158 let mut config = match discovered.bytes {
159 Some(bytes) => Self::parse_unresolved(&bytes).map_err(|error| ServerError::Config {
160 message: format!("failed to parse {}: {error}", discovered.source),
161 })?,
162 None => Self::default(),
163 };
164 if overlay_environment {
165 env::overlay(&mut config)?;
166 }
167 config.apply_cli_overrides(cli);
168 expand_operator_path_tildes(&mut config)?;
169 #[cfg(not(unix))]
170 let home_explicit = !overlay_environment || std::env::var_os("AION_HOME").is_some();
171 #[cfg(not(unix))]
172 let data_dir_explicit = config.store.data_dir.is_some();
173 #[cfg(not(unix))]
174 let data_root_required = matches!(config.store.backend, StoreBackend::Haematite);
175 #[cfg(not(unix))]
176 let authoring_workspace_explicit = config.authoring.workspace_dir.is_some();
177 config.load_discovered_workflow_packages(cli, working_dir)?;
178 let legacy_notices = fill_home_defaults(&mut config, home, home_source, working_dir)?;
179 config.fill_operational_defaults();
180 config.validate()?;
181 let resolution = ConfigResolution {
182 home: home.to_owned(),
183 source: discovered.source,
184 data_dir: config.store.data_dir.clone(),
185 authoring_workspace: config.authoring.workspace_dir.clone(),
186 legacy_notices,
187 #[cfg(not(unix))]
188 home_explicit,
189 #[cfg(not(unix))]
190 data_dir_explicit,
191 #[cfg(not(unix))]
192 data_root_required,
193 #[cfg(not(unix))]
194 authoring_workspace_explicit,
195 };
196 Ok(LoadedConfig { config, resolution })
197 }
198
199 #[cfg(test)]
208 pub(super) fn load_for_test(
209 cli: &CliOverrides,
210 home: &Path,
211 home_source: HomeSource,
212 working_dir: &Path,
213 ) -> Result<LoadedConfig, ServerError> {
214 Self::load_in(cli, home, home_source, working_dir, false)
215 }
216
217 fn fill_operational_defaults(&mut self) {
223 self.runtime
224 .query_timeout_ms
225 .get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
226 self.websocket
227 .event_broadcast_capacity
228 .get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
229 self.websocket
230 .cluster_broadcast_capacity
231 .get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
232 self.fill_outbox_defaults();
233 self.fill_deploy_defaults();
234 }
235
236 fn fill_outbox_defaults(&mut self) {
245 if !self.outbox.enabled {
246 return;
247 }
248 self.outbox
249 .poll_interval_ms
250 .get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
251 self.outbox
252 .batch_size
253 .get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
254 self.outbox
255 .max_attempts
256 .get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
257 self.outbox
258 .backoff_base_ms
259 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
260 self.outbox
261 .backoff_multiplier
262 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
263 self.outbox
264 .backoff_max_ms
265 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
266 }
267
268 fn fill_deploy_defaults(&mut self) {
276 if !self.deploy.enabled {
277 return;
278 }
279 self.deploy
280 .max_archive_bytes
281 .get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
282 self.deploy
283 .max_inflated_bytes
284 .get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
285 }
286
287 fn load_discovered_workflow_packages(
288 &mut self,
289 cli: &CliOverrides,
290 directory: &Path,
291 ) -> Result<(), ServerError> {
292 let discovered_packages = discover_workflow_packages(directory)?;
293 merge_workflow_packages(
294 &mut self.workflow_packages,
295 discovered_packages,
296 &cli.workflow_packages,
297 );
298 Ok(())
299 }
300
301 pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
307 let home = aion_home()?;
308 Self::from_slice_with_home(bytes, &home.path)
309 }
310
311 pub fn from_slice_with_home(bytes: &[u8], home: &Path) -> Result<Self, ServerError> {
326 let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
327 message: format!(
328 "failed to resolve the current directory for config defaults: {source}"
329 ),
330 })?;
331 let mut config = Self::parse_unresolved(bytes)?;
332 fill_home_defaults(&mut config, home, HomeSource::Explicit, &working_dir)?;
336 config.fill_operational_defaults();
337 config.validate()?;
338 Ok(config)
339 }
340
341 fn parse_unresolved(bytes: &[u8]) -> Result<Self, ServerError> {
342 toml::from_slice(bytes).map_err(|source| ServerError::Config {
343 message: format!("invalid server config: {source}"),
344 })
345 }
346
347 pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
353 file::load_required(&path.into())
354 }
355
356 #[must_use]
358 pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
359 let runtime = RuntimeConfig {
360 listen: ListenConfig {
361 grpc: self.server.grpc_address,
362 http: self.server.listen_address,
363 },
364 tls: self.tls,
365 auth: self.auth,
366 ops_console: self.ops_console,
367 namespace: self.namespace,
368 worker: self.worker,
369 websocket: self.websocket,
370 workflow_packages: self.workflow_packages,
371 deploy: self.deploy,
372 authoring: self.authoring,
373 dev: self.dev,
374 outbox: self.outbox,
375 observability: self.observability,
376 mcp: self.mcp.resolved(),
377 scheduler_threads: self.runtime.scheduler_threads,
378 jit_threshold: self.runtime.jit_threshold,
379 query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
380 default_namespace: self.namespaces.default,
381 auto_create: self.namespaces.auto_create,
382 max_in_flight_activities: self.namespaces.max_in_flight_activities,
383 drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
384 metrics: self.metrics,
385 owned_shards: self.store.owned_shards.clone(),
386 cors_allowed_origins: self.server.cors_allowed_origins.clone(),
387 };
388 (self.store, runtime)
389 }
390
391 fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
392 if let Some(address) = cli.listen_address {
393 self.server.listen_address = address;
394 }
395 if let Some(address) = cli.grpc_address {
396 self.server.grpc_address = address;
397 }
398 if cli.store_url.is_some() {
399 self.store.retired_input = Some(RetiredStoreInput::Flag);
400 }
401 if let Some(threads) = cli.scheduler_threads {
402 self.runtime.scheduler_threads = threads;
403 }
404 if let Some(threshold) = cli.jit_threshold {
405 self.runtime.jit_threshold = Some(threshold);
406 }
407 if let Some(timeout) = cli.drain_timeout_seconds {
408 self.drain.timeout_seconds = timeout;
409 }
410 if let Some(gleam_path) = &cli.gleam_path {
411 self.authoring.gleam_path = Some(gleam_path.clone());
412 }
413 if let Some(project_root) = &cli.authoring_project_root {
414 self.authoring.project_root = Some(project_root.clone());
415 }
416 }
417
418 fn validate(&self) -> Result<(), ServerError> {
419 if let Some(input) = self.store.retired_input {
420 let found = match input {
421 RetiredStoreInput::Backend => "backend = \"libsql\"",
422 RetiredStoreInput::BackendEnvironment => "AION_STORE_BACKEND=libsql",
423 RetiredStoreInput::Url => "store.url",
424 RetiredStoreInput::Environment => "AION_STORE_URL",
425 RetiredStoreInput::Flag => "--store-url",
426 };
427 return config_error(format!(
428 "found retired {found}; use backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build"
429 ));
430 }
431 if self.server.listen_address.port() == 0 {
432 return config_error("server.listen_address must use an explicit non-zero port");
433 }
434 if self.server.grpc_address.port() == 0 {
435 return config_error("server.grpc_address must use an explicit non-zero port");
436 }
437 validate_cors_origins(&self.server.cors_allowed_origins)?;
438 if self.runtime.scheduler_threads == 0 {
439 return config_error("runtime.scheduler_threads must be greater than zero");
440 }
441 if self.drain.timeout_seconds == 0 {
442 return config_error("drain.timeout_seconds must be greater than zero");
443 }
444 if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
445 return config_error("auth.jwks_url must not be empty when auth.enabled is true");
446 }
447 if self.auth.jwks_refresh_seconds == 0 {
448 return config_error("auth.jwks_refresh_seconds must be greater than zero");
449 }
450 if self.namespaces.default.is_empty() {
451 return config_error("namespaces.default must not be empty");
452 }
453 if matches!(self.store.backend, StoreBackend::Haematite) {
454 if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
455 return config_error(
456 "store.data_dir must not be empty when store.backend is haematite",
457 );
458 }
459 if self.store.shard_count == 0 {
460 return config_error("store.shard_count must be greater than zero");
461 }
462 if let Some(cluster) = &self.store.cluster {
463 validate_cluster(cluster)?;
464 }
465 } else if self.store.cluster.is_some() {
466 return config_error("store.cluster is only valid when store.backend is haematite");
467 }
468 if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
469 if asset_path.as_os_str().is_empty() {
470 return config_error("ops_console.source.FileSystem.asset_path must not be empty");
471 }
472 }
473 if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
474 if namespace.is_empty() {
475 return config_error("namespace.mode.SingleTenant.namespace must not be empty");
476 }
477 }
478 if self.worker.heartbeat_window.is_zero() {
479 return config_error("worker.heartbeat_window must be greater than zero");
480 }
481 self.websocket.validate()?;
482 self.observability.validate()?;
483 self.mcp.validate()?;
484 match self.runtime.query_timeout_ms {
485 None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
486 Some(_) => {}
487 }
488 if self.deploy.enabled {
489 let max_archive_bytes = match self.deploy.max_archive_bytes {
490 None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
491 Some(value) => value,
492 };
493 let max_inflated_bytes = match self.deploy.max_inflated_bytes {
494 None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
495 Some(value) => value,
496 };
497 ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
500 ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
501 if max_inflated_bytes < max_archive_bytes {
502 return config_error(format!(
503 "deploy.max_inflated_bytes ({max_inflated_bytes}) must be at least deploy.max_archive_bytes ({max_archive_bytes}): an inflate ceiling below the upload ceiling would refuse archives the upload ceiling admits, even stored uncompressed"
504 ));
505 }
506 }
507 if let Some(gleam_path) = &self.authoring.gleam_path {
508 if gleam_path.as_os_str().is_empty() {
511 return config_error(AUTHORING_GLEAM_PATH_EMPTY);
512 }
513 match &self.authoring.project_root {
516 Some(root) if !root.as_os_str().is_empty() => {}
517 _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
518 }
519 }
520 self.validate_outbox()?;
521 self.worker_supervision.resolve()?;
525 Ok(())
526 }
527
528 fn validate_outbox(&self) -> Result<(), ServerError> {
536 if !self.outbox.enabled {
537 return Ok(());
538 }
539 match self.outbox.poll_interval_ms {
540 None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
541 Some(_) => {}
542 }
543 match self.outbox.batch_size {
544 None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
545 Some(_) => {}
546 }
547 match self.outbox.max_attempts {
548 None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
549 Some(_) => {}
550 }
551 let backoff_base_ms = match self.outbox.backoff_base_ms {
552 None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
553 Some(value) => value,
554 };
555 match self.outbox.backoff_multiplier {
556 None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
557 Some(_) => {}
558 }
559 match self.outbox.backoff_max_ms {
560 Some(max) if max >= backoff_base_ms => {}
561 _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
562 }
563 match (
564 self.outbox.reconcile_interval_ms,
565 self.outbox.reconcile_stale_after_ms,
566 ) {
567 (None, None) => {}
568 (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
569 (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
570 (Some(_), Some(_)) => {}
571 }
572 Ok(())
573 }
574}
575
576fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
580 if cluster.node_id.is_empty() {
581 return config_error("store.cluster.node_id must not be empty");
582 }
583 if cluster.members.iter().any(String::is_empty) {
584 return config_error("store.cluster.members entries must not be empty");
585 }
586 if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
587 return config_error("store.cluster.peers entries must name a non-empty node");
588 }
589 if matches!(cluster.failover_poll_interval_ms, Some(0)) {
590 return config_error(
591 "store.cluster.failover_poll_interval_ms must be greater than zero when set",
592 );
593 }
594 if matches!(cluster.failover_confirmations, Some(0)) {
595 return config_error("store.cluster.failover_confirmations must be at least one when set");
596 }
597 Ok(())
598}
599
600fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
602 for origin in origins {
603 validate_cors_origin(origin)?;
604 }
605 Ok(())
606}
607
608fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
613 if origin.is_empty() {
614 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
615 }
616 let scheme_split = origin.split_once("://");
620 let Some((scheme, authority)) = scheme_split else {
621 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
622 };
623 if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
624 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
625 }
626 if origin.parse::<axum::http::HeaderValue>().is_err() {
628 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
629 }
630 Ok(())
631}
632
633fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
635 if usize::try_from(value).is_err() {
636 return config_error(format!(
637 "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
638 usize::MAX
639 ));
640 }
641 Ok(())
642}
643
644fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
645 let mut packages = Vec::new();
646 let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
647 message: format!(
648 "failed to scan workflow packages in `{}`: {source}",
649 directory.display()
650 ),
651 })?;
652
653 for entry in entries {
654 let entry = entry.map_err(|source| ServerError::Config {
655 message: format!(
656 "failed to read workflow package entry in `{}`: {source}",
657 directory.display()
658 ),
659 })?;
660 let path = entry.path();
661 let has_aion_extension = path
662 .extension()
663 .is_some_and(|extension| extension == "aion");
664 if path.is_file() && has_aion_extension {
665 packages.push(path);
666 }
667 }
668
669 packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
670 Ok(packages)
671}
672
673fn merge_workflow_packages(
674 workflow_packages: &mut Vec<PathBuf>,
675 discovered_packages: Vec<PathBuf>,
676 cli_packages: &[PathBuf],
677) {
678 let mut seen: HashSet<PathBuf> = workflow_packages
679 .iter()
680 .map(|package| deduplicated_package_key(package))
681 .collect();
682 for package in discovered_packages
683 .into_iter()
684 .chain(cli_packages.iter().cloned())
685 {
686 if seen.insert(deduplicated_package_key(&package)) {
687 workflow_packages.push(package);
688 }
689 }
690}
691
692fn deduplicated_package_key(path: &Path) -> PathBuf {
693 path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
694}
695
696#[cfg(test)]
697#[path = "load_home_tests.rs"]
698mod home_tests;
699
700#[cfg(test)]
701mod tests {
702 use crate::config::{
703 AutoCreate, DEFAULT_AUTHORING_WORKSPACE_DIR, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, HomeSource,
704 OpsConsoleAssetSource,
705 };
706
707 use super::{
708 CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
709 DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
710 DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
711 DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
712 DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
713 discover_workflow_packages, merge_workflow_packages,
714 };
715
716 #[test]
717 fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
718 let config = ServerConfig::from_slice(
719 br#"
720 [server]
721 listen_address = "127.0.0.1:18080"
722 grpc_address = "127.0.0.1:15051"
723
724 [store]
725 backend = "haematite"
726 data_dir = "aion-data"
727
728 [runtime]
729 scheduler_threads = 2
730 query_timeout_ms = 10000
731
732 [drain]
733 timeout_seconds = 45
734
735 [auth]
736 enabled = true
737 jwks_url = "https://issuer.example.com/.well-known/jwks.json"
738 jwks_refresh_seconds = 60
739
740 [metrics]
741 enabled = true
742
743 [namespaces]
744 default = "production"
745
746 [websocket]
747 outbound_buffer_bound = 16
748 event_broadcast_capacity = 1024
749 cluster_broadcast_capacity = 1024
750 "#,
751 )?;
752
753 assert_eq!(config.store.backend, StoreBackend::Haematite);
754 assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
755 assert_eq!(config.runtime.scheduler_threads, 2);
756 assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
757 assert_eq!(config.namespaces.default, "production");
758 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
760 assert_eq!(
763 config.namespaces.max_in_flight_activities,
764 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
765 );
766 assert_eq!(config.websocket.outbound_buffer_bound, 16);
767 assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
768 Ok(())
769 }
770
771 #[test]
772 fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
773 let config = ServerConfig::from_slice(
774 br#"
775 [namespaces]
776 default = "production"
777 auto_create = "closed"
778 "#,
779 )?;
780 assert_eq!(config.namespaces.default, "production");
781 assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
782 Ok(())
783 }
784
785 #[test]
786 fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
787 let config = ServerConfig::from_slice(
788 br#"
789 [namespaces]
790 auto_create = "open"
791 "#,
792 )?;
793 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
794 Ok(())
795 }
796
797 #[test]
798 fn namespaces_max_in_flight_activities_override_parses()
799 -> Result<(), Box<dyn std::error::Error>> {
800 let config = ServerConfig::from_slice(
801 br#"
802 [namespaces]
803 default = "production"
804 max_in_flight_activities = 32
805 "#,
806 )?;
807 assert_eq!(config.namespaces.max_in_flight_activities, 32);
808 let (_store, runtime) = config.into_parts();
810 assert_eq!(runtime.max_in_flight_activities, 32);
811 Ok(())
812 }
813
814 #[test]
815 fn namespaces_max_in_flight_activities_defaults_when_omitted()
816 -> Result<(), Box<dyn std::error::Error>> {
817 let config = ServerConfig::from_slice(
820 br#"
821 [namespaces]
822 default = "production"
823 "#,
824 )?;
825 assert_eq!(
826 config.namespaces.max_in_flight_activities,
827 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
828 );
829 Ok(())
830 }
831
832 #[test]
833 fn namespaces_auto_create_rejects_unknown_variant() {
834 let result = ServerConfig::from_slice(
835 br#"
836 [namespaces]
837 auto_create = "sometimes"
838 "#,
839 );
840 assert!(
841 result.is_err(),
842 "an unknown auto_create variant must fail to parse"
843 );
844 }
845
846 #[test]
847 fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
848 let config = ServerConfig::from_slice(
852 br"
853 [runtime]
854 query_timeout_ms = 10000
855
856 [websocket]
857 cluster_broadcast_capacity = 64
858 ",
859 )?;
860 assert_eq!(
861 config.websocket.event_broadcast_capacity,
862 Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
863 "omitted event_broadcast_capacity must resolve to the default"
864 );
865 Ok(())
866 }
867
868 #[test]
869 fn zero_event_broadcast_capacity_fails_startup_validation() {
870 let result = ServerConfig::from_slice(
871 br"
872 [websocket]
873 event_broadcast_capacity = 0
874 ",
875 );
876
877 let message = result
878 .err()
879 .map_or_else(String::new, |error| error.to_string());
880 assert!(
881 message.contains("websocket.event_broadcast_capacity"),
882 "validation message must name the zero-valued key: {message}"
883 );
884 }
885
886 #[test]
887 fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
888 let config = ServerConfig::from_slice(
892 br"
893 [runtime]
894 scheduler_threads = 1
895 query_timeout_ms = 10000
896
897 [websocket]
898 event_broadcast_capacity = 64
899 ",
900 )?;
901 assert_eq!(
902 config.websocket.cluster_broadcast_capacity,
903 Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
904 "omitted cluster_broadcast_capacity must resolve to the default"
905 );
906 Ok(())
907 }
908
909 #[test]
910 fn zero_cluster_broadcast_capacity_fails_startup_validation() {
911 let result = ServerConfig::from_slice(
912 br"
913 [runtime]
914 query_timeout_ms = 10000
915
916 [websocket]
917 event_broadcast_capacity = 64
918 cluster_broadcast_capacity = 0
919 ",
920 );
921
922 let message = result
923 .err()
924 .map_or_else(String::new, |error| error.to_string());
925 assert!(
926 message.contains("websocket.cluster_broadcast_capacity"),
927 "validation message must name the zero-valued cluster key: {message}"
928 );
929 }
930
931 #[test]
935 fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
936 let config = ServerConfig::from_slice(
937 br"
938 [runtime]
939 query_timeout_ms = 10000
940
941 [websocket]
942 event_broadcast_capacity = 64
943 cluster_broadcast_capacity = 64
944 ",
945 )?;
946 assert_eq!(
947 config.observability.max_event_bytes,
948 crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
949 );
950 assert_eq!(
951 config.observability.max_stream_events,
952 crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
953 );
954 let (_store, runtime) = config.into_parts();
956 assert_eq!(
957 runtime.observability.max_event_bytes,
958 crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
959 );
960 Ok(())
961 }
962
963 #[test]
966 fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
967 let config = ServerConfig::from_slice(
968 br"
969 [runtime]
970 query_timeout_ms = 10000
971
972 [websocket]
973 event_broadcast_capacity = 64
974 cluster_broadcast_capacity = 64
975
976 [observability]
977 max_event_bytes = 512
978 max_stream_events = 3
979 ",
980 )?;
981 assert_eq!(config.observability.max_event_bytes, 512);
982 assert_eq!(config.observability.max_stream_events, 3);
983 let (_store, runtime) = config.into_parts();
984 assert_eq!(runtime.observability.max_event_bytes, 512);
985 assert_eq!(runtime.observability.max_stream_events, 3);
986 Ok(())
987 }
988
989 #[test]
993 fn missing_mcp_section_leaves_the_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
994 let config = ServerConfig::from_slice(
995 br"
996 [runtime]
997 query_timeout_ms = 10000
998
999 [websocket]
1000 event_broadcast_capacity = 64
1001 cluster_broadcast_capacity = 64
1002 ",
1003 )?;
1004 let (_store, runtime) = config.into_parts();
1005 assert!(!runtime.mcp.enabled);
1006 assert!(runtime.mcp.allowed_origins.is_empty());
1007 assert_eq!(
1008 runtime.mcp.discover_ttl_ms,
1009 crate::config::DEFAULT_MCP_DISCOVER_TTL_MS
1010 );
1011 assert_eq!(
1012 runtime.mcp.task_poll_interval_ms,
1013 crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS
1014 );
1015 Ok(())
1016 }
1017
1018 #[test]
1020 fn mcp_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1021 let config = ServerConfig::from_slice(
1022 br#"
1023 [runtime]
1024 query_timeout_ms = 10000
1025
1026 [websocket]
1027 event_broadcast_capacity = 64
1028 cluster_broadcast_capacity = 64
1029
1030 [mcp]
1031 enabled = true
1032 allowed_origins = ["http://localhost:5173"]
1033 discover_ttl_ms = 1000
1034 tools_list_ttl_ms = 2000
1035 task_poll_interval_ms = 250
1036 "#,
1037 )?;
1038 let (_store, runtime) = config.into_parts();
1039 assert!(runtime.mcp.enabled);
1040 assert_eq!(runtime.mcp.allowed_origins, vec!["http://localhost:5173"]);
1041 assert_eq!(runtime.mcp.discover_ttl_ms, 1_000);
1042 assert_eq!(runtime.mcp.tools_list_ttl_ms, 2_000);
1043 assert_eq!(runtime.mcp.task_poll_interval_ms, 250);
1044 Ok(())
1045 }
1046
1047 #[test]
1056 fn the_retired_mcp_task_knobs_are_refused_by_name() {
1057 for key in ["task_ttl_ms", "await_poll_interval_ms"] {
1058 let source = format!(
1059 "
1060 [runtime]
1061 query_timeout_ms = 10000
1062
1063 [websocket]
1064 event_broadcast_capacity = 64
1065 cluster_broadcast_capacity = 64
1066
1067 [mcp]
1068 enabled = true
1069 {key} = 1000
1070 "
1071 );
1072 let message = match ServerConfig::from_slice(source.as_bytes()) {
1073 Err(crate::ServerError::Config { message }) => message,
1074 _ => String::new(),
1075 };
1076 assert!(
1077 message.contains(key),
1078 "{key} must be refused by name: {message}"
1079 );
1080 }
1081 }
1082
1083 #[test]
1086 fn a_zero_mcp_poll_interval_fails_startup_validation() {
1087 let result = ServerConfig::from_slice(
1088 br"
1089 [runtime]
1090 query_timeout_ms = 10000
1091
1092 [websocket]
1093 event_broadcast_capacity = 64
1094 cluster_broadcast_capacity = 64
1095
1096 [mcp]
1097 enabled = true
1098 task_poll_interval_ms = 0
1099 ",
1100 );
1101 let message = match result {
1102 Err(crate::ServerError::Config { message }) => message,
1103 _ => String::new(),
1104 };
1105 assert!(
1106 message.contains("task_poll_interval_ms"),
1107 "a zero poll interval must be refused: {message}"
1108 );
1109 }
1110
1111 #[test]
1114 fn a_dark_mcp_surface_does_not_validate_its_unused_knobs()
1115 -> Result<(), Box<dyn std::error::Error>> {
1116 let config = ServerConfig::from_slice(
1117 br"
1118 [runtime]
1119 query_timeout_ms = 10000
1120
1121 [websocket]
1122 event_broadcast_capacity = 64
1123 cluster_broadcast_capacity = 64
1124
1125 [mcp]
1126 task_poll_interval_ms = 0
1127 ",
1128 )?;
1129 let (_store, runtime) = config.into_parts();
1130 assert!(!runtime.mcp.enabled);
1131 assert_eq!(
1132 runtime.mcp.task_poll_interval_ms,
1133 crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS,
1134 "an unused zero resolves to the default rather than to a busy loop"
1135 );
1136 Ok(())
1137 }
1138
1139 #[test]
1144 fn observability_flush_policy_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>>
1145 {
1146 let config = ServerConfig::from_slice(
1147 br"
1148 [runtime]
1149 query_timeout_ms = 10000
1150
1151 [websocket]
1152 event_broadcast_capacity = 64
1153 cluster_broadcast_capacity = 64
1154
1155 [observability]
1156 max_batch_events = 32
1157 max_batch_hold_ms = 0
1158 ",
1159 )?;
1160 assert_eq!(config.observability.max_batch_events, Some(32));
1161 assert_eq!(config.observability.max_batch_hold_ms, Some(0));
1162 let (_store, runtime) = config.into_parts();
1163 assert_eq!(runtime.observability.max_batch_events, Some(32));
1164 assert_eq!(
1165 runtime.observability.max_batch_hold_ms,
1166 Some(0),
1167 "a stated zero hold survives as a stated zero, never collapsing to absent"
1168 );
1169 Ok(())
1170 }
1171
1172 #[test]
1176 fn node_cache_budget_parses_both_spellings() -> Result<(), Box<dyn std::error::Error>> {
1177 let bounded = ServerConfig::from_slice(
1178 br"
1179 [runtime]
1180 query_timeout_ms = 10000
1181
1182 [store]
1183 node_cache_budget = { bytes = 1073741824 }
1184 ",
1185 )?;
1186 assert_eq!(
1187 bounded.store.node_cache_budget,
1188 Some(haematite::NodeCacheBudget::bytes(1 << 30)?),
1189 "a 1 GiB ceiling arrives as a 1 GiB ceiling"
1190 );
1191
1192 let unlimited = ServerConfig::from_slice(
1193 br#"
1194 [runtime]
1195 query_timeout_ms = 10000
1196
1197 [store]
1198 node_cache_budget = "unlimited"
1199 "#,
1200 )?;
1201 assert_eq!(
1202 unlimited.store.node_cache_budget,
1203 Some(haematite::NodeCacheBudget::Unlimited),
1204 "`unlimited` is a stated choice, never collapsed to absent"
1205 );
1206 Ok(())
1207 }
1208
1209 #[test]
1216 fn retired_lock_acquisition_keys_stay_legal() -> Result<(), Box<dyn std::error::Error>> {
1217 let config = ServerConfig::from_slice(
1218 br#"
1219 [store]
1220 backend = "haematite"
1221 lock_acquisition_patience_ms = 60000
1222 lock_acquisition_retry_cadence_ms = 500
1223 "#,
1224 )?;
1225 assert_eq!(
1226 config.store.lock_acquisition_patience_ms,
1227 Some(60_000),
1228 "the retired key parses instead of refusing as unknown"
1229 );
1230 assert_eq!(
1231 config.store.lock_acquisition_retry_cadence_ms,
1232 Some(500),
1233 "both retired keys parse; nothing consumes them"
1234 );
1235 Ok(())
1236 }
1237
1238 #[test]
1243 fn an_omitted_node_cache_budget_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
1244 let config = ServerConfig::from_slice(
1245 br"
1246 [runtime]
1247 query_timeout_ms = 10000
1248
1249 [store]
1250 shard_count = 8
1251 ",
1252 )?;
1253 assert_eq!(config.store.node_cache_budget, None);
1254 Ok(())
1255 }
1256
1257 #[test]
1262 fn an_omitted_flush_policy_stays_absent() -> Result<(), Box<dyn std::error::Error>> {
1263 let config = ServerConfig::from_slice(
1264 br"
1265 [runtime]
1266 query_timeout_ms = 10000
1267
1268 [observability]
1269 max_event_bytes = 512
1270 ",
1271 )?;
1272 assert_eq!(config.observability.max_batch_events, None);
1273 assert_eq!(config.observability.max_batch_hold_ms, None);
1274 Ok(())
1275 }
1276
1277 #[test]
1278 fn zero_observability_max_event_bytes_fails_startup_validation() {
1279 let result = ServerConfig::from_slice(
1280 br"
1281 [runtime]
1282 query_timeout_ms = 10000
1283
1284 [websocket]
1285 event_broadcast_capacity = 64
1286 cluster_broadcast_capacity = 64
1287
1288 [observability]
1289 max_event_bytes = 0
1290 ",
1291 );
1292 let message = result
1293 .err()
1294 .map_or_else(String::new, |error| error.to_string());
1295 assert!(
1296 message.contains("observability.max_event_bytes"),
1297 "validation message must name the zero-valued key: {message}"
1298 );
1299 }
1300
1301 #[test]
1302 fn zero_observability_max_stream_events_fails_startup_validation() {
1303 let result = ServerConfig::from_slice(
1304 br"
1305 [runtime]
1306 query_timeout_ms = 10000
1307
1308 [websocket]
1309 event_broadcast_capacity = 64
1310 cluster_broadcast_capacity = 64
1311
1312 [observability]
1313 max_stream_events = 0
1314 ",
1315 );
1316 let message = result
1317 .err()
1318 .map_or_else(String::new, |error| error.to_string());
1319 assert!(
1320 message.contains("observability.max_stream_events"),
1321 "validation message must name the zero-valued key: {message}"
1322 );
1323 }
1324
1325 #[test]
1326 fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
1327 let config = ServerConfig::from_slice(
1331 br"
1332 [runtime]
1333 scheduler_threads = 1
1334
1335 [websocket]
1336 event_broadcast_capacity = 64
1337 cluster_broadcast_capacity = 64
1338 ",
1339 )?;
1340 assert_eq!(
1341 config.runtime.query_timeout_ms,
1342 Some(DEFAULT_QUERY_TIMEOUT_MS),
1343 "omitted query_timeout_ms must resolve to the default"
1344 );
1345 Ok(())
1346 }
1347
1348 #[test]
1349 fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
1350 let config = ServerConfig::from_slice(b"")?;
1355 assert_eq!(config.store.backend, StoreBackend::Haematite);
1356 assert_eq!(
1357 config.runtime.query_timeout_ms,
1358 Some(DEFAULT_QUERY_TIMEOUT_MS)
1359 );
1360 assert_eq!(
1361 config.websocket.event_broadcast_capacity,
1362 Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
1363 );
1364 assert_eq!(
1365 config.websocket.cluster_broadcast_capacity,
1366 Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
1367 );
1368 Ok(())
1369 }
1370
1371 #[test]
1372 fn zero_query_timeout_fails_startup_validation() {
1373 let result = ServerConfig::from_slice(
1374 br"
1375 [runtime]
1376 query_timeout_ms = 0
1377
1378 [websocket]
1379 event_broadcast_capacity = 64
1380 cluster_broadcast_capacity = 64
1381 ",
1382 );
1383
1384 let message = result
1385 .err()
1386 .map_or_else(String::new, |error| error.to_string());
1387 assert!(
1388 message.contains("runtime.query_timeout_ms"),
1389 "validation message must name the zero-valued key: {message}"
1390 );
1391 }
1392
1393 #[test]
1398 fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
1399 let config = ServerConfig::from_slice(
1400 br"
1401 [runtime]
1402 query_timeout_ms = 10000
1403
1404 [websocket]
1405 event_broadcast_capacity = 64
1406 cluster_broadcast_capacity = 64
1407
1408 [deploy]
1409 enabled = true
1410 ",
1411 )?;
1412
1413 assert_eq!(
1414 config.deploy.max_archive_bytes,
1415 Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
1416 "omitted max_archive_bytes must resolve to the conservative default"
1417 );
1418 assert_eq!(
1419 config.deploy.max_inflated_bytes,
1420 Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1421 "omitted max_inflated_bytes must resolve to the conservative default"
1422 );
1423 Ok(())
1424 }
1425
1426 #[test]
1427 fn deploy_zero_max_archive_bytes_fails_startup_validation() {
1428 let result = ServerConfig::from_slice(
1429 br"
1430 [runtime]
1431 query_timeout_ms = 10000
1432
1433 [websocket]
1434 event_broadcast_capacity = 64
1435 cluster_broadcast_capacity = 64
1436
1437 [deploy]
1438 enabled = true
1439 max_archive_bytes = 0
1440 ",
1441 );
1442
1443 let message = result
1444 .err()
1445 .map_or_else(String::new, |error| error.to_string());
1446 assert!(
1447 message.contains("deploy.max_archive_bytes"),
1448 "validation message must name the zero-valued key: {message}"
1449 );
1450 }
1451
1452 #[test]
1457 fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
1458 let config = ServerConfig::from_slice(
1459 br"
1460 [runtime]
1461 query_timeout_ms = 10000
1462
1463 [websocket]
1464 event_broadcast_capacity = 64
1465 cluster_broadcast_capacity = 64
1466
1467 [deploy]
1468 enabled = true
1469 max_archive_bytes = 16777216
1470 ",
1471 )?;
1472
1473 assert_eq!(
1474 config.deploy.max_archive_bytes,
1475 Some(16_777_216),
1476 "explicit max_archive_bytes must be left untouched"
1477 );
1478 assert_eq!(
1479 config.deploy.max_inflated_bytes,
1480 Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1481 "omitted max_inflated_bytes must resolve to the conservative default"
1482 );
1483 Ok(())
1484 }
1485
1486 #[test]
1487 fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
1488 let result = ServerConfig::from_slice(
1489 br"
1490 [runtime]
1491 query_timeout_ms = 10000
1492
1493 [websocket]
1494 event_broadcast_capacity = 64
1495 cluster_broadcast_capacity = 64
1496
1497 [deploy]
1498 enabled = true
1499 max_archive_bytes = 16777216
1500 max_inflated_bytes = 0
1501 ",
1502 );
1503
1504 let message = result
1505 .err()
1506 .map_or_else(String::new, |error| error.to_string());
1507 assert!(
1508 message.contains("deploy.max_inflated_bytes"),
1509 "validation message must name the zero-valued key: {message}"
1510 );
1511 }
1512
1513 #[test]
1516 fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
1517 let result = ServerConfig::from_slice(
1518 br"
1519 [runtime]
1520 query_timeout_ms = 10000
1521
1522 [websocket]
1523 event_broadcast_capacity = 64
1524 cluster_broadcast_capacity = 64
1525
1526 [deploy]
1527 enabled = true
1528 max_archive_bytes = 16777216
1529 max_inflated_bytes = 16777215
1530 ",
1531 );
1532
1533 let message = result
1534 .err()
1535 .map_or_else(String::new, |error| error.to_string());
1536 assert!(
1537 message.contains("deploy.max_inflated_bytes")
1538 && message.contains("deploy.max_archive_bytes"),
1539 "validation message must name both ceilings: {message}"
1540 );
1541 }
1542
1543 #[test]
1546 fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
1547 let config = ServerConfig::from_slice(
1548 br"
1549 [runtime]
1550 query_timeout_ms = 10000
1551
1552 [websocket]
1553 event_broadcast_capacity = 64
1554 cluster_broadcast_capacity = 64
1555 ",
1556 )?;
1557
1558 assert!(!config.deploy.enabled);
1559 assert_eq!(config.deploy.max_archive_bytes, None);
1560 assert_eq!(config.deploy.max_inflated_bytes, None);
1561 Ok(())
1562 }
1563
1564 #[test]
1565 fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
1566 let config = ServerConfig::from_slice(
1567 br"
1568 [runtime]
1569 query_timeout_ms = 10000
1570
1571 [websocket]
1572 event_broadcast_capacity = 64
1573 cluster_broadcast_capacity = 64
1574
1575 [deploy]
1576 enabled = true
1577 max_archive_bytes = 16777216
1578 max_inflated_bytes = 67108864
1579 ",
1580 )?;
1581
1582 assert!(config.deploy.enabled);
1583 assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
1584 assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
1585 Ok(())
1586 }
1587
1588 #[test]
1592 fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
1593 let config = ServerConfig::from_slice(
1594 br"
1595 [runtime]
1596 query_timeout_ms = 10000
1597
1598 [websocket]
1599 event_broadcast_capacity = 64
1600 cluster_broadcast_capacity = 64
1601 ",
1602 )?;
1603
1604 assert!(config.server.cors_allowed_origins.is_empty());
1605 let (_, runtime) = config.into_parts();
1606 assert!(runtime.cors_allowed_origins.is_empty());
1607 Ok(())
1608 }
1609
1610 #[test]
1613 fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
1614 let config = ServerConfig::from_slice(
1615 br#"
1616 [server]
1617 cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
1618
1619 [runtime]
1620 query_timeout_ms = 10000
1621
1622 [websocket]
1623 event_broadcast_capacity = 64
1624 cluster_broadcast_capacity = 64
1625 "#,
1626 )?;
1627
1628 assert_eq!(
1629 config.server.cors_allowed_origins,
1630 vec![
1631 "http://localhost:5173".to_owned(),
1632 "http://127.0.0.1:5173".to_owned()
1633 ]
1634 );
1635 let (_, runtime) = config.into_parts();
1636 assert_eq!(
1637 runtime.cors_allowed_origins,
1638 vec![
1639 "http://localhost:5173".to_owned(),
1640 "http://127.0.0.1:5173".to_owned()
1641 ]
1642 );
1643 Ok(())
1644 }
1645
1646 #[test]
1650 fn cors_allowed_origins_reject_malformed() {
1651 for bad in ["", "localhost:5173", "http://localhost:5173/"] {
1652 let toml = format!(
1653 "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
1654 );
1655 let result = ServerConfig::from_slice(toml.as_bytes());
1656 let message = result
1657 .err()
1658 .map_or_else(String::new, |error| error.to_string());
1659 assert!(
1660 message.contains("cors_allowed_origins"),
1661 "malformed origin `{bad}` must be rejected naming the key: {message}"
1662 );
1663 }
1664 }
1665
1666 #[test]
1668 fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1669 let config = ServerConfig::from_slice(
1670 br"
1671 [runtime]
1672 query_timeout_ms = 10000
1673
1674 [websocket]
1675 event_broadcast_capacity = 64
1676 cluster_broadcast_capacity = 64
1677 ",
1678 )?;
1679
1680 assert!(!config.dev.enabled);
1681 Ok(())
1682 }
1683
1684 #[test]
1687 fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
1688 let config = ServerConfig::from_slice(
1689 br"
1690 [runtime]
1691 query_timeout_ms = 10000
1692
1693 [websocket]
1694 event_broadcast_capacity = 64
1695 cluster_broadcast_capacity = 64
1696
1697 [dev]
1698 enabled = true
1699 ",
1700 )?;
1701
1702 assert!(config.dev.enabled);
1703 Ok(())
1704 }
1705
1706 #[test]
1709 fn authoring_absent_defaults_awl_workspace_but_keeps_gleam_dark()
1710 -> Result<(), Box<dyn std::error::Error>> {
1711 let home = crate::test_support::private_tempdir()?;
1712 let config = ServerConfig::from_slice_with_home(b"", home.path())?;
1713
1714 assert_eq!(config.authoring.gleam_path, None);
1715 assert_eq!(config.authoring.project_root, None);
1716 assert_eq!(
1717 config.authoring.workspace_dir.as_deref(),
1718 Some(home.path().join(DEFAULT_AUTHORING_WORKSPACE_DIR).as_path())
1719 );
1720 Ok(())
1721 }
1722
1723 #[test]
1725 fn authoring_explicit_workspace_is_honored() -> Result<(), Box<dyn std::error::Error>> {
1726 let config = ServerConfig::from_slice(
1727 br#"
1728 [authoring]
1729 workspace_dir = "/srv/aion/studio"
1730 "#,
1731 )?;
1732
1733 assert_eq!(
1734 config.authoring.workspace_dir.as_deref(),
1735 Some(std::path::Path::new("/srv/aion/studio"))
1736 );
1737 Ok(())
1738 }
1739
1740 #[test]
1743 fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1744 let config = ServerConfig::from_slice(
1745 br#"
1746 [runtime]
1747 query_timeout_ms = 10000
1748
1749 [websocket]
1750 event_broadcast_capacity = 64
1751 cluster_broadcast_capacity = 64
1752
1753 [authoring]
1754 gleam_path = "/usr/local/bin/gleam"
1755 project_root = "/srv/aion/authoring"
1756 "#,
1757 )?;
1758
1759 assert_eq!(
1760 config.authoring.gleam_path.as_deref(),
1761 Some(std::path::Path::new("/usr/local/bin/gleam"))
1762 );
1763 let (_, runtime) = config.into_parts();
1764 assert_eq!(
1765 runtime.authoring.gleam_path.as_deref(),
1766 Some(std::path::Path::new("/usr/local/bin/gleam"))
1767 );
1768 assert_eq!(
1769 runtime.authoring.project_root.as_deref(),
1770 Some(std::path::Path::new("/srv/aion/authoring"))
1771 );
1772 Ok(())
1773 }
1774
1775 #[test]
1779 fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
1780 let result = ServerConfig::from_slice(
1781 br#"
1782 [runtime]
1783 query_timeout_ms = 10000
1784
1785 [websocket]
1786 event_broadcast_capacity = 64
1787 cluster_broadcast_capacity = 64
1788
1789 [authoring]
1790 gleam_path = "/usr/local/bin/gleam"
1791 "#,
1792 );
1793
1794 let message = result
1795 .err()
1796 .map_or_else(String::new, |error| error.to_string());
1797 assert!(
1798 message.contains("authoring.project_root"),
1799 "validation message must name the missing key: {message}"
1800 );
1801 assert!(
1802 message.contains("AION_AUTHORING_PROJECT_ROOT"),
1803 "validation message must name the environment override: {message}"
1804 );
1805 }
1806
1807 #[test]
1810 fn authoring_empty_gleam_path_fails_naming_key_and_env() {
1811 let result = ServerConfig::from_slice(
1812 br#"
1813 [runtime]
1814 query_timeout_ms = 10000
1815
1816 [websocket]
1817 event_broadcast_capacity = 64
1818 cluster_broadcast_capacity = 64
1819
1820 [authoring]
1821 gleam_path = ""
1822 "#,
1823 );
1824
1825 let message = result
1826 .err()
1827 .map_or_else(String::new, |error| error.to_string());
1828 assert!(
1829 message.contains("authoring.gleam_path"),
1830 "validation message must name the empty key: {message}"
1831 );
1832 assert!(
1833 message.contains("AION_AUTHORING_GLEAM_PATH"),
1834 "validation message must name the environment override: {message}"
1835 );
1836 }
1837
1838 #[test]
1840 fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
1841 let mut config = ServerConfig::from_slice(
1842 br"
1843 [runtime]
1844 query_timeout_ms = 10000
1845
1846 [websocket]
1847 event_broadcast_capacity = 64
1848 cluster_broadcast_capacity = 64
1849 ",
1850 )?;
1851 let cli = CliOverrides {
1852 gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
1853 authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
1854 ..CliOverrides::default()
1855 };
1856
1857 config.apply_cli_overrides(&cli);
1858 config.validate()?;
1859
1860 assert_eq!(
1861 config.authoring.gleam_path.as_deref(),
1862 Some(std::path::Path::new("/opt/gleam"))
1863 );
1864 assert_eq!(
1865 config.authoring.project_root.as_deref(),
1866 Some(std::path::Path::new("/opt/project"))
1867 );
1868 Ok(())
1869 }
1870
1871 #[test]
1876 fn queue_service_settings_are_read_from_the_worker_section()
1877 -> Result<(), Box<dyn std::error::Error>> {
1878 use crate::worker::QueueServicePolicy;
1879 use std::time::Duration;
1880
1881 let bare = ServerConfig::from_slice(
1882 br"
1883 [websocket]
1884 event_broadcast_capacity = 64
1885 cluster_broadcast_capacity = 64
1886 ",
1887 )?;
1888 assert_eq!(
1889 bare.worker.queue_service.default_policy,
1890 QueueServicePolicy::Strict,
1891 "strict is the default with nothing written"
1892 );
1893 assert_eq!(
1894 bare.worker.queue_service.service_availability_deadline,
1895 None
1896 );
1897 assert_eq!(bare.worker.queue_service.schedule_to_start_timeout, None);
1898
1899 let written = ServerConfig::from_slice(
1900 br#"
1901 [websocket]
1902 event_broadcast_capacity = 64
1903 cluster_broadcast_capacity = 64
1904
1905 [worker.queue_service]
1906 service_availability_deadline = 45000
1907 schedule_to_start_timeout = 5000
1908
1909 [[worker.queue_service.overrides]]
1910 task_queue = "general"
1911 policy = "durable_pending"
1912 "#,
1913 )?;
1914 assert_eq!(
1915 written.worker.queue_service.service_availability_deadline,
1916 Some(Duration::from_secs(45))
1917 );
1918 assert_eq!(
1919 written.worker.queue_service.schedule_to_start_timeout,
1920 Some(Duration::from_secs(5))
1921 );
1922 assert_eq!(
1923 written
1924 .worker
1925 .queue_service
1926 .policy_for("default", "general"),
1927 QueueServicePolicy::DurablePending,
1928 "the written opt-in must reach the dispatch seam"
1929 );
1930 assert_eq!(
1931 written
1932 .worker
1933 .queue_service
1934 .policy_for("default", "billing"),
1935 QueueServicePolicy::Strict,
1936 "an override must not leak onto other queues"
1937 );
1938
1939 let (_store, runtime) = written.into_parts();
1941 assert_eq!(
1942 runtime
1943 .worker
1944 .queue_service
1945 .policy_for("default", "general"),
1946 QueueServicePolicy::DurablePending
1947 );
1948 Ok(())
1949 }
1950
1951 #[test]
1952 fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
1953 let config = ServerConfig::from_slice(
1954 br#"
1955 [runtime]
1956 query_timeout_ms = 10000
1957
1958 [websocket]
1959 event_broadcast_capacity = 64
1960 cluster_broadcast_capacity = 64
1961
1962 [dashboard]
1963 source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1964 "#,
1965 )?;
1966 match &config.ops_console.source {
1967 OpsConsoleAssetSource::FileSystem { asset_path } => {
1968 assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
1969 }
1970 OpsConsoleAssetSource::Embedded => {
1971 return Err("legacy [dashboard] section must map to ops_console".into());
1972 }
1973 }
1974 Ok(())
1975 }
1976
1977 #[test]
1979 fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
1980 let config = ServerConfig::from_slice(
1981 br#"
1982 [runtime]
1983 query_timeout_ms = 10000
1984
1985 [websocket]
1986 event_broadcast_capacity = 64
1987 cluster_broadcast_capacity = 64
1988
1989 [ops_console]
1990 source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1991 "#,
1992 )?;
1993 assert!(matches!(
1994 config.ops_console.source,
1995 OpsConsoleAssetSource::FileSystem { .. }
1996 ));
1997 Ok(())
1998 }
1999
2000 #[test]
2001 fn invalid_values_name_problematic_field() {
2002 let result = ServerConfig::from_slice(
2003 br"
2004 [runtime]
2005 scheduler_threads = 0
2006 ",
2007 );
2008
2009 let message = result
2010 .err()
2011 .map_or_else(String::new, |error| error.to_string());
2012 assert!(message.contains("runtime.scheduler_threads"));
2013 }
2014
2015 const RETIRED_STORE_REMEDY: &str = "backend = \"haematite\" with store.data_dir and store.node_cache_budget (all three are required — haematite refuses to start without the budget); existing libsql database files are not read by this build";
2016
2017 fn assert_retired_store_refusal(
2018 result: Result<ServerConfig, crate::error::ServerError>,
2019 found: &str,
2020 ) {
2021 assert!(result.is_err(), "retired libsql input must be refused");
2022 let message = result
2023 .err()
2024 .map_or_else(String::new, |error| error.to_string());
2025 assert!(
2026 message.contains(found),
2027 "refusal did not name `{found}`: {message}"
2028 );
2029 assert!(
2030 message.contains(RETIRED_STORE_REMEDY),
2031 "refusal omitted the operator remedy: {message}"
2032 );
2033 }
2034
2035 #[test]
2036 fn retired_libsql_backend_is_refused_with_remedy() {
2037 assert_retired_store_refusal(
2038 ServerConfig::from_slice(
2039 br#"
2040 [store]
2041 backend = "libsql"
2042 "#,
2043 ),
2044 "backend = \"libsql\"",
2045 );
2046 }
2047
2048 #[test]
2049 fn retired_store_url_key_is_refused_with_remedy() {
2050 assert_retired_store_refusal(
2051 ServerConfig::from_slice(
2052 br#"
2053 [store]
2054 backend = "haematite"
2055 url = "old.db"
2056 "#,
2057 ),
2058 "store.url",
2059 );
2060 }
2061
2062 #[test]
2067 fn retired_libsql_backend_environment_is_refused_with_remedy() {
2068 let mut config = ServerConfig::default();
2069 let result = super::env::overlay_vars(
2070 &mut config,
2071 [("AION_STORE_BACKEND".to_owned(), "libsql".to_owned())],
2072 )
2073 .and_then(|()| config.validate().map(|()| config));
2074 assert_retired_store_refusal(result, "AION_STORE_BACKEND=libsql");
2075 }
2076
2077 #[test]
2078 fn retired_store_url_environment_is_refused_with_remedy() {
2079 let mut config = ServerConfig::default();
2080 let result = super::env::overlay_vars(
2081 &mut config,
2082 [("AION_STORE_URL".to_owned(), "old.db".to_owned())],
2083 )
2084 .and_then(|()| config.validate().map(|()| config));
2085 assert_retired_store_refusal(result, "AION_STORE_URL");
2086 }
2087
2088 #[test]
2089 fn retired_store_url_flag_is_refused_with_remedy() {
2090 let mut config = ServerConfig::default();
2091 config.apply_cli_overrides(&CliOverrides {
2092 store_url: Some("old.db".to_owned()),
2093 ..CliOverrides::default()
2094 });
2095 let result = config.validate().map(|()| config);
2096 assert_retired_store_refusal(result, "--store-url");
2097 }
2098
2099 #[test]
2103 fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
2104 let mut config = ServerConfig::from_slice(
2105 br#"
2106 [store]
2107 backend = "haematite"
2108 data_dir = "from-the-file"
2109
2110 [runtime]
2111 scheduler_threads = 1
2112 query_timeout_ms = 10000
2113
2114 [websocket]
2115 event_broadcast_capacity = 64
2116 cluster_broadcast_capacity = 64
2117
2118 [observability]
2119 max_batch_events = 64
2120 max_batch_hold_ms = 0
2121 "#,
2122 )?;
2123 assert_eq!(
2124 config.runtime.scheduler_threads, 1,
2125 "the file's value must be what the flag then beats"
2126 );
2127 let cli = CliOverrides {
2128 scheduler_threads: Some(
2129 std::num::NonZeroUsize::new(3)
2130 .ok_or("3 is not zero")?
2131 .into(),
2132 ),
2133 ..CliOverrides::default()
2134 };
2135
2136 config.apply_cli_overrides(&cli);
2137 config.validate()?;
2138
2139 assert_eq!(config.runtime.scheduler_threads, 3);
2140 assert_eq!(
2141 config.store.data_dir.as_deref(),
2142 Some("from-the-file"),
2143 "a value the CLI did not override keeps the file's value"
2144 );
2145 Ok(())
2146 }
2147
2148 #[test]
2149 fn haematite_store_config_remains_accepted() -> Result<(), Box<dyn std::error::Error>> {
2150 let config = ServerConfig::from_slice(
2151 br#"
2152 [store]
2153 backend = "haematite"
2154 data_dir = "aion-data"
2155 "#,
2156 )?;
2157 assert_eq!(config.store.backend, StoreBackend::Haematite);
2158 assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
2159 Ok(())
2160 }
2161
2162 #[test]
2163 fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
2164 let mut config = ServerConfig::default();
2165
2166 assert_eq!(config.store.backend, StoreBackend::Haematite);
2169 assert_eq!(config.store.data_dir, None);
2170 assert_eq!(config.store.shard_count, 64);
2174 assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
2175 assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
2176 assert_eq!(config.namespaces.default, "default");
2177 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
2181 assert_eq!(
2184 config.namespaces.max_in_flight_activities,
2185 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
2186 );
2187 assert_eq!(config.namespaces.max_in_flight_activities, 1024);
2188 assert!(!config.auth.enabled);
2189 assert!(config.metrics.enabled);
2190 assert_eq!(config.websocket.event_broadcast_capacity, None);
2194 assert_eq!(config.websocket.cluster_broadcast_capacity, None);
2195 assert_eq!(config.runtime.query_timeout_ms, None);
2196 config.websocket.event_broadcast_capacity = Some(64);
2197 config.websocket.cluster_broadcast_capacity = Some(64);
2198 config.runtime.query_timeout_ms = Some(10_000);
2199 let home = crate::test_support::private_tempdir()?;
2200 let working_dir = crate::test_support::private_tempdir()?;
2201 super::fill_home_defaults(
2202 &mut config,
2203 home.path(),
2204 HomeSource::Derived,
2205 working_dir.path(),
2206 )?;
2207 assert_eq!(
2208 config.store.data_dir.as_deref(),
2209 home.path().join("data").to_str()
2210 );
2211 config.validate()?;
2212 Ok(())
2213 }
2214
2215 #[test]
2216 fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
2217 {
2218 let mut config = ServerConfig::default();
2219 config.store.data_dir = Some("test-data".to_owned());
2220 config.websocket.event_broadcast_capacity = Some(64);
2221 config.websocket.cluster_broadcast_capacity = Some(64);
2222 config.runtime.query_timeout_ms = Some(10_000);
2223
2224 assert!(!config.outbox.enabled);
2228 assert_eq!(config.outbox.poll_interval_ms, None);
2229 assert_eq!(config.outbox.batch_size, None);
2230 assert_eq!(config.outbox.max_attempts, None);
2231 assert_eq!(config.outbox.backoff_base_ms, None);
2232 assert_eq!(config.outbox.backoff_multiplier, None);
2233 assert_eq!(config.outbox.backoff_max_ms, None);
2234 assert_eq!(config.outbox.reconcile_interval_ms, None);
2235 assert_eq!(config.outbox.reconcile_stale_after_ms, None);
2236 config.validate()?;
2237 Ok(())
2238 }
2239
2240 fn outbox_enabled_base() -> ServerConfig {
2241 let mut config = ServerConfig::default();
2242 config.store.data_dir = Some("test-data".to_owned());
2243 config.websocket.event_broadcast_capacity = Some(64);
2244 config.websocket.cluster_broadcast_capacity = Some(64);
2245 config.runtime.query_timeout_ms = Some(10_000);
2246 config.outbox.enabled = true;
2247 config.outbox.poll_interval_ms = Some(250);
2248 config.outbox.batch_size = Some(64);
2249 config.outbox.max_attempts = Some(5);
2250 config.outbox.backoff_base_ms = Some(100);
2251 config.outbox.backoff_multiplier = Some(2);
2252 config.outbox.backoff_max_ms = Some(30_000);
2253 config.outbox.reconcile_interval_ms = Some(1_000);
2254 config.outbox.reconcile_stale_after_ms = Some(60_000);
2255 config
2256 }
2257
2258 #[test]
2259 fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
2260 outbox_enabled_base().validate()?;
2261 Ok(())
2262 }
2263
2264 #[test]
2265 fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
2266 let config = ServerConfig::from_slice(
2270 br"
2271 [runtime]
2272 query_timeout_ms = 10000
2273
2274 [websocket]
2275 event_broadcast_capacity = 64
2276 cluster_broadcast_capacity = 64
2277
2278 [outbox]
2279 enabled = true
2280 ",
2281 )?;
2282 assert_eq!(
2283 config.outbox.poll_interval_ms,
2284 Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
2285 "omitted poll_interval_ms must resolve to the default"
2286 );
2287 Ok(())
2288 }
2289
2290 #[test]
2291 fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
2292 let config = ServerConfig::from_slice(
2295 br"
2296 [runtime]
2297 query_timeout_ms = 10000
2298
2299 [websocket]
2300 event_broadcast_capacity = 64
2301 cluster_broadcast_capacity = 64
2302
2303 [outbox]
2304 enabled = true
2305 poll_interval_ms = 250
2306 ",
2307 )?;
2308 assert_eq!(
2309 config.outbox.poll_interval_ms,
2310 Some(250),
2311 "explicit poll_interval_ms must be left untouched"
2312 );
2313 assert_eq!(
2314 config.outbox.max_attempts,
2315 Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
2316 "omitted max_attempts must resolve to the default"
2317 );
2318 Ok(())
2319 }
2320
2321 #[test]
2322 fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
2323 -> Result<(), Box<dyn std::error::Error>> {
2324 let config = ServerConfig::from_slice(
2328 br"
2329 [runtime]
2330 query_timeout_ms = 10000
2331
2332 [websocket]
2333 event_broadcast_capacity = 64
2334 cluster_broadcast_capacity = 64
2335
2336 [outbox]
2337 enabled = true
2338 ",
2339 )?;
2340 assert!(config.outbox.enabled);
2341 assert_eq!(
2342 config.outbox.poll_interval_ms,
2343 Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
2344 );
2345 assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
2346 assert_eq!(
2347 config.outbox.max_attempts,
2348 Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
2349 );
2350 assert_eq!(
2351 config.outbox.backoff_base_ms,
2352 Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
2353 );
2354 assert_eq!(
2355 config.outbox.backoff_multiplier,
2356 Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
2357 );
2358 assert_eq!(
2359 config.outbox.backoff_max_ms,
2360 Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
2361 );
2362 assert_eq!(config.outbox.reconcile_interval_ms, None);
2365 assert_eq!(config.outbox.reconcile_stale_after_ms, None);
2366 Ok(())
2367 }
2368
2369 #[test]
2370 fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2371 let mut config = outbox_enabled_base();
2374 config.outbox.poll_interval_ms = Some(0);
2375 let error = config
2376 .validate()
2377 .err()
2378 .ok_or("enabled outbox with zero poll interval must fail")?;
2379 assert!(
2380 error.to_string().contains("outbox.poll_interval_ms"),
2381 "error must name the zero-valued key: {error}"
2382 );
2383 Ok(())
2384 }
2385
2386 #[test]
2387 fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2388 let mut config = outbox_enabled_base();
2389 config.outbox.max_attempts = Some(0);
2390 let error = config
2391 .validate()
2392 .err()
2393 .ok_or("enabled outbox with zero max attempts must fail")?;
2394 assert!(
2395 error.to_string().contains("outbox.max_attempts"),
2396 "error must name the zero-valued key: {error}"
2397 );
2398 Ok(())
2399 }
2400
2401 #[test]
2402 fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2403 let mut config = outbox_enabled_base();
2404 config.outbox.backoff_base_ms = Some(1_000);
2405 config.outbox.backoff_max_ms = Some(500);
2406 let error = config
2407 .validate()
2408 .err()
2409 .ok_or("backoff_max below backoff_base must fail")?;
2410 assert!(
2411 error.to_string().contains("outbox.backoff_max_ms"),
2412 "error must name the offending key: {error}"
2413 );
2414 Ok(())
2415 }
2416
2417 #[test]
2418 fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
2419 let mut config = outbox_enabled_base();
2420 config.outbox.reconcile_interval_ms = None;
2421 config.outbox.reconcile_stale_after_ms = None;
2422 config.validate()?;
2423 Ok(())
2424 }
2425
2426 #[test]
2427 fn outbox_reconciliation_requires_interval_when_partially_enabled()
2428 -> Result<(), Box<dyn std::error::Error>> {
2429 let mut config = outbox_enabled_base();
2430 config.outbox.reconcile_interval_ms = None;
2431 let error = config
2432 .validate()
2433 .err()
2434 .ok_or("reconciliation without interval must fail")?;
2435 assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
2436 Ok(())
2437 }
2438
2439 #[test]
2440 fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
2441 -> Result<(), Box<dyn std::error::Error>> {
2442 let mut config = outbox_enabled_base();
2443 config.outbox.reconcile_stale_after_ms = None;
2444 let error = config
2445 .validate()
2446 .err()
2447 .ok_or("reconciliation without stale threshold must fail")?;
2448 assert!(
2449 error
2450 .to_string()
2451 .contains("outbox.reconcile_stale_after_ms")
2452 );
2453 Ok(())
2454 }
2455
2456 #[test]
2457 fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
2458 let temp_dir = crate::test_support::private_tempdir()?;
2459 std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
2460 std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
2461 std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
2462 std::fs::create_dir(temp_dir.path().join("nested"))?;
2463 std::fs::write(
2464 temp_dir.path().join("nested").join("nested.aion"),
2465 b"package",
2466 )?;
2467
2468 let packages = discover_workflow_packages(temp_dir.path())?;
2469
2470 assert_eq!(
2471 packages,
2472 vec![
2473 temp_dir.path().join("alpha.aion"),
2474 temp_dir.path().join("zeta.aion"),
2475 ]
2476 );
2477 Ok(())
2478 }
2479
2480 #[test]
2481 fn workflow_package_merge_is_additive_and_deduplicated() {
2482 let mut packages = vec!["config.aion".into(), "shared.aion".into()];
2483 let discovered = vec!["auto.aion".into(), "shared.aion".into()];
2484 let cli = vec!["cli.aion".into(), "auto.aion".into()];
2485
2486 merge_workflow_packages(&mut packages, discovered, &cli);
2487
2488 assert_eq!(
2489 packages,
2490 vec![
2491 std::path::PathBuf::from("config.aion"),
2492 std::path::PathBuf::from("shared.aion"),
2493 std::path::PathBuf::from("auto.aion"),
2494 std::path::PathBuf::from("cli.aion"),
2495 ]
2496 );
2497 }
2498
2499 #[test]
2500 fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
2501 let temp_dir = crate::test_support::private_tempdir()?;
2502 let package = temp_dir.path().join("hello.aion");
2503 std::fs::write(&package, b"package")?;
2504 let mut packages = vec![package.clone()];
2505 let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
2506
2507 merge_workflow_packages(&mut packages, discovered, &[]);
2508
2509 assert_eq!(packages, vec![package]);
2510 Ok(())
2511 }
2512
2513 #[test]
2514 fn zero_config_cli_workflow_package_uses_in_memory_defaults()
2515 -> Result<(), Box<dyn std::error::Error>> {
2516 let temp_dir = crate::test_support::private_tempdir()?;
2517
2518 let cli = CliOverrides {
2519 workflow_packages: vec!["hello-world.aion".into()],
2520 ..CliOverrides::default()
2521 };
2522 let mut config = ServerConfig::default();
2523 config.store.backend = StoreBackend::Memory;
2527 config.store.data_dir = None;
2528 config.websocket.event_broadcast_capacity = Some(64);
2533 config.websocket.cluster_broadcast_capacity = Some(64);
2534 config.runtime.query_timeout_ms = Some(10_000);
2535 config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
2536
2537 config.validate()?;
2538
2539 assert_eq!(config.store.backend, StoreBackend::Memory);
2540 assert_eq!(
2541 config.workflow_packages,
2542 vec![std::path::PathBuf::from("hello-world.aion")]
2543 );
2544 Ok(())
2545 }
2546
2547 #[test]
2548 fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
2549 let mut config = ServerConfig::from_slice(
2550 br#"
2551 workflow_packages = ["config.aion"]
2552
2553 [runtime]
2554 query_timeout_ms = 10000
2555
2556 [websocket]
2557 event_broadcast_capacity = 64
2558 cluster_broadcast_capacity = 64
2559 "#,
2560 )?;
2561 let cli = CliOverrides {
2562 workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
2563 ..CliOverrides::default()
2564 };
2565
2566 merge_workflow_packages(
2567 &mut config.workflow_packages,
2568 Vec::new(),
2569 &cli.workflow_packages,
2570 );
2571
2572 assert_eq!(
2573 config.workflow_packages,
2574 vec![
2575 std::path::PathBuf::from("config.aion"),
2576 std::path::PathBuf::from("cli-one.aion"),
2577 std::path::PathBuf::from("cli-two.aion"),
2578 ]
2579 );
2580 Ok(())
2581 }
2582}