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