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