1use std::{
13 collections::HashSet,
14 fs,
15 path::{Path, PathBuf},
16 time::Duration,
17};
18
19use serde::Deserialize;
20
21use crate::error::ServerError;
22
23use super::{
24 AUTHORING_GLEAM_PATH_EMPTY, AUTHORING_PROJECT_ROOT_REQUIRED, AuthConfig, AuthoringConfig,
25 CORS_ALLOWED_ORIGIN_INVALID, CliOverrides, ClusterConfig, ConfigResolution,
26 DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
27 DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
28 DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
29 DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
30 DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED,
31 DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig, DrainConfig, HomeSource,
32 ListenConfig, McpConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
33 OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
34 OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
35 OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
36 OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
37 RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
38 WorkerConfig, WorkerSupervisionConfig, aion_home, config_error, env, file,
39 resolution::fill_home_defaults,
40};
41
42#[derive(Clone, Debug, Deserialize)]
44#[serde(default, deny_unknown_fields)]
45#[derive(Default)]
46pub struct ServerConfig {
47 pub server: ServerSection,
49 pub store: StoreConfig,
51 pub runtime: RuntimeSection,
53 pub drain: DrainConfig,
55 pub auth: AuthConfig,
57 pub metrics: MetricsConfig,
59 pub namespaces: NamespacesConfig,
61 pub tls: Option<TlsConfig>,
63 #[serde(alias = "dashboard")]
65 pub ops_console: OpsConsoleConfig,
66 pub namespace: NamespaceConfig,
68 pub worker: WorkerConfig,
70 pub websocket: WebSocketConfig,
72 pub workflow_packages: Vec<PathBuf>,
74 pub deploy: DeployConfig,
76 pub authoring: AuthoringConfig,
78 pub dev: DevConfig,
80 pub outbox: OutboxConfig,
82 pub observability: ObservabilityConfig,
84 pub mcp: McpConfig,
86 pub worker_supervision: WorkerSupervisionConfig,
89}
90
91pub(crate) struct LoadedConfig {
92 pub(crate) config: ServerConfig,
93 pub(crate) resolution: ConfigResolution,
94}
95
96impl ServerConfig {
97 pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
104 Ok(Self::load_resolved(cli)?.config)
105 }
106
107 pub(crate) fn load_resolved(cli: &CliOverrides) -> Result<LoadedConfig, ServerError> {
108 let home = aion_home()?;
109 let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
110 message: format!(
111 "failed to resolve the current directory for config discovery: {source}"
112 ),
113 })?;
114 Self::load_in(cli, &home.path, home.source, &working_dir, true)
115 }
116
117 fn load_in(
118 cli: &CliOverrides,
119 home: &Path,
120 home_source: HomeSource,
121 working_dir: &Path,
122 overlay_environment: bool,
123 ) -> Result<LoadedConfig, ServerError> {
124 let discovered = file::discover(cli.config_path.as_deref(), home, working_dir)?;
125 let mut config = match discovered.bytes {
126 Some(bytes) => Self::parse_unresolved(&bytes).map_err(|error| ServerError::Config {
127 message: format!("failed to parse {}: {error}", discovered.source),
128 })?,
129 None => Self::default(),
130 };
131 if overlay_environment {
132 env::overlay(&mut config)?;
133 }
134 config.apply_cli_overrides(cli);
135 #[cfg(not(unix))]
136 let home_explicit = !overlay_environment || std::env::var_os("AION_HOME").is_some();
137 #[cfg(not(unix))]
138 let data_dir_explicit = config.store.data_dir.is_some();
139 #[cfg(not(unix))]
140 let data_root_required = matches!(config.store.backend, StoreBackend::Haematite);
141 #[cfg(not(unix))]
142 let authoring_workspace_explicit = config.authoring.workspace_dir.is_some();
143 config.load_discovered_workflow_packages(cli, working_dir)?;
144 let legacy_notices = fill_home_defaults(&mut config, home, home_source, working_dir)?;
145 config.fill_operational_defaults();
146 config.validate()?;
147 let resolution = ConfigResolution {
148 home: home.to_owned(),
149 source: discovered.source,
150 data_dir: config.store.data_dir.clone(),
151 authoring_workspace: config.authoring.workspace_dir.clone(),
152 legacy_notices,
153 #[cfg(not(unix))]
154 home_explicit,
155 #[cfg(not(unix))]
156 data_dir_explicit,
157 #[cfg(not(unix))]
158 data_root_required,
159 #[cfg(not(unix))]
160 authoring_workspace_explicit,
161 };
162 Ok(LoadedConfig { config, resolution })
163 }
164
165 #[cfg(test)]
174 fn load_for_test(
175 cli: &CliOverrides,
176 home: &Path,
177 home_source: HomeSource,
178 working_dir: &Path,
179 ) -> Result<LoadedConfig, ServerError> {
180 Self::load_in(cli, home, home_source, working_dir, false)
181 }
182
183 fn fill_operational_defaults(&mut self) {
189 self.runtime
190 .query_timeout_ms
191 .get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
192 self.websocket
193 .event_broadcast_capacity
194 .get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
195 self.websocket
196 .cluster_broadcast_capacity
197 .get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
198 self.fill_outbox_defaults();
199 self.fill_deploy_defaults();
200 }
201
202 fn fill_outbox_defaults(&mut self) {
211 if !self.outbox.enabled {
212 return;
213 }
214 self.outbox
215 .poll_interval_ms
216 .get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
217 self.outbox
218 .batch_size
219 .get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
220 self.outbox
221 .max_attempts
222 .get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
223 self.outbox
224 .backoff_base_ms
225 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
226 self.outbox
227 .backoff_multiplier
228 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
229 self.outbox
230 .backoff_max_ms
231 .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
232 }
233
234 fn fill_deploy_defaults(&mut self) {
242 if !self.deploy.enabled {
243 return;
244 }
245 self.deploy
246 .max_archive_bytes
247 .get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
248 self.deploy
249 .max_inflated_bytes
250 .get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
251 }
252
253 fn load_discovered_workflow_packages(
254 &mut self,
255 cli: &CliOverrides,
256 directory: &Path,
257 ) -> Result<(), ServerError> {
258 let discovered_packages = discover_workflow_packages(directory)?;
259 merge_workflow_packages(
260 &mut self.workflow_packages,
261 discovered_packages,
262 &cli.workflow_packages,
263 );
264 Ok(())
265 }
266
267 pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
273 let home = aion_home()?;
274 Self::from_slice_with_home(bytes, &home.path)
275 }
276
277 pub fn from_slice_with_home(bytes: &[u8], home: &Path) -> Result<Self, ServerError> {
292 let working_dir = std::env::current_dir().map_err(|source| ServerError::Config {
293 message: format!(
294 "failed to resolve the current directory for config defaults: {source}"
295 ),
296 })?;
297 let mut config = Self::parse_unresolved(bytes)?;
298 fill_home_defaults(&mut config, home, HomeSource::Explicit, &working_dir)?;
302 config.fill_operational_defaults();
303 config.validate()?;
304 Ok(config)
305 }
306
307 fn parse_unresolved(bytes: &[u8]) -> Result<Self, ServerError> {
308 toml::from_slice(bytes).map_err(|source| ServerError::Config {
309 message: format!("invalid server config: {source}"),
310 })
311 }
312
313 pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
319 file::load_required(&path.into())
320 }
321
322 #[must_use]
324 pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
325 let runtime = RuntimeConfig {
326 listen: ListenConfig {
327 grpc: self.server.grpc_address,
328 http: self.server.listen_address,
329 },
330 tls: self.tls,
331 auth: self.auth,
332 ops_console: self.ops_console,
333 namespace: self.namespace,
334 worker: self.worker,
335 websocket: self.websocket,
336 workflow_packages: self.workflow_packages,
337 deploy: self.deploy,
338 authoring: self.authoring,
339 dev: self.dev,
340 outbox: self.outbox,
341 observability: self.observability,
342 mcp: self.mcp.resolved(),
343 scheduler_threads: self.runtime.scheduler_threads,
344 query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
345 default_namespace: self.namespaces.default,
346 auto_create: self.namespaces.auto_create,
347 max_in_flight_activities: self.namespaces.max_in_flight_activities,
348 drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
349 metrics: self.metrics,
350 owned_shards: self.store.owned_shards.clone(),
351 cors_allowed_origins: self.server.cors_allowed_origins.clone(),
352 };
353 (self.store, runtime)
354 }
355
356 fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
357 if let Some(address) = cli.listen_address {
358 self.server.listen_address = address;
359 }
360 if let Some(address) = cli.grpc_address {
361 self.server.grpc_address = address;
362 }
363 if let Some(url) = &cli.store_url {
364 self.store.url = Some(url.clone());
365 if matches!(
372 self.store.backend,
373 StoreBackend::Memory | StoreBackend::Haematite
374 ) {
375 self.store.backend = StoreBackend::LibSql;
376 }
377 }
378 if let Some(threads) = cli.scheduler_threads {
379 self.runtime.scheduler_threads = threads;
380 }
381 if let Some(timeout) = cli.drain_timeout_seconds {
382 self.drain.timeout_seconds = timeout;
383 }
384 if let Some(gleam_path) = &cli.gleam_path {
385 self.authoring.gleam_path = Some(gleam_path.clone());
386 }
387 if let Some(project_root) = &cli.authoring_project_root {
388 self.authoring.project_root = Some(project_root.clone());
389 }
390 }
391
392 fn validate(&self) -> Result<(), ServerError> {
393 if self.server.listen_address.port() == 0 {
394 return config_error("server.listen_address must use an explicit non-zero port");
395 }
396 if self.server.grpc_address.port() == 0 {
397 return config_error("server.grpc_address must use an explicit non-zero port");
398 }
399 validate_cors_origins(&self.server.cors_allowed_origins)?;
400 if self.runtime.scheduler_threads == 0 {
401 return config_error("runtime.scheduler_threads must be greater than zero");
402 }
403 if self.drain.timeout_seconds == 0 {
404 return config_error("drain.timeout_seconds must be greater than zero");
405 }
406 if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
407 return config_error("auth.jwks_url must not be empty when auth.enabled is true");
408 }
409 if self.auth.jwks_refresh_seconds == 0 {
410 return config_error("auth.jwks_refresh_seconds must be greater than zero");
411 }
412 if self.namespaces.default.is_empty() {
413 return config_error("namespaces.default must not be empty");
414 }
415 if matches!(self.store.backend, StoreBackend::LibSql)
416 && self.store.url.as_deref().is_none_or(str::is_empty)
417 {
418 return config_error("store.url must not be empty when store.backend is libsql");
419 }
420 if let Some(url) = &self.store.url {
421 if url.is_empty() {
422 return config_error("store.url must not be empty");
423 }
424 }
425 if matches!(self.store.backend, StoreBackend::Haematite) {
426 if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
427 return config_error(
428 "store.data_dir must not be empty when store.backend is haematite",
429 );
430 }
431 if self.store.shard_count == 0 {
432 return config_error("store.shard_count must be greater than zero");
433 }
434 if let Some(cluster) = &self.store.cluster {
435 validate_cluster(cluster)?;
436 }
437 } else if self.store.cluster.is_some() {
438 return config_error("store.cluster is only valid when store.backend is haematite");
439 }
440 if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
441 if asset_path.as_os_str().is_empty() {
442 return config_error("ops_console.source.FileSystem.asset_path must not be empty");
443 }
444 }
445 if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
446 if namespace.is_empty() {
447 return config_error("namespace.mode.SingleTenant.namespace must not be empty");
448 }
449 }
450 if self.worker.heartbeat_window.is_zero() {
451 return config_error("worker.heartbeat_window must be greater than zero");
452 }
453 self.websocket.validate()?;
454 self.observability.validate()?;
455 self.mcp.validate()?;
456 match self.runtime.query_timeout_ms {
457 None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
458 Some(_) => {}
459 }
460 if self.deploy.enabled {
461 let max_archive_bytes = match self.deploy.max_archive_bytes {
462 None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
463 Some(value) => value,
464 };
465 let max_inflated_bytes = match self.deploy.max_inflated_bytes {
466 None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
467 Some(value) => value,
468 };
469 ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
472 ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
473 if max_inflated_bytes < max_archive_bytes {
474 return config_error(format!(
475 "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"
476 ));
477 }
478 }
479 if let Some(gleam_path) = &self.authoring.gleam_path {
480 if gleam_path.as_os_str().is_empty() {
483 return config_error(AUTHORING_GLEAM_PATH_EMPTY);
484 }
485 match &self.authoring.project_root {
488 Some(root) if !root.as_os_str().is_empty() => {}
489 _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
490 }
491 }
492 self.validate_outbox()?;
493 self.worker_supervision.resolve()?;
497 Ok(())
498 }
499
500 fn validate_outbox(&self) -> Result<(), ServerError> {
508 if !self.outbox.enabled {
509 return Ok(());
510 }
511 match self.outbox.poll_interval_ms {
512 None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
513 Some(_) => {}
514 }
515 match self.outbox.batch_size {
516 None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
517 Some(_) => {}
518 }
519 match self.outbox.max_attempts {
520 None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
521 Some(_) => {}
522 }
523 let backoff_base_ms = match self.outbox.backoff_base_ms {
524 None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
525 Some(value) => value,
526 };
527 match self.outbox.backoff_multiplier {
528 None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
529 Some(_) => {}
530 }
531 match self.outbox.backoff_max_ms {
532 Some(max) if max >= backoff_base_ms => {}
533 _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
534 }
535 match (
536 self.outbox.reconcile_interval_ms,
537 self.outbox.reconcile_stale_after_ms,
538 ) {
539 (None, None) => {}
540 (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
541 (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
542 (Some(_), Some(_)) => {}
543 }
544 Ok(())
545 }
546}
547
548fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
552 if cluster.node_id.is_empty() {
553 return config_error("store.cluster.node_id must not be empty");
554 }
555 if cluster.members.iter().any(String::is_empty) {
556 return config_error("store.cluster.members entries must not be empty");
557 }
558 if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
559 return config_error("store.cluster.peers entries must name a non-empty node");
560 }
561 if matches!(cluster.failover_poll_interval_ms, Some(0)) {
562 return config_error(
563 "store.cluster.failover_poll_interval_ms must be greater than zero when set",
564 );
565 }
566 if matches!(cluster.failover_confirmations, Some(0)) {
567 return config_error("store.cluster.failover_confirmations must be at least one when set");
568 }
569 Ok(())
570}
571
572fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
574 for origin in origins {
575 validate_cors_origin(origin)?;
576 }
577 Ok(())
578}
579
580fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
585 if origin.is_empty() {
586 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
587 }
588 let scheme_split = origin.split_once("://");
592 let Some((scheme, authority)) = scheme_split else {
593 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
594 };
595 if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
596 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
597 }
598 if origin.parse::<axum::http::HeaderValue>().is_err() {
600 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
601 }
602 Ok(())
603}
604
605fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
607 if usize::try_from(value).is_err() {
608 return config_error(format!(
609 "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
610 usize::MAX
611 ));
612 }
613 Ok(())
614}
615
616fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
617 let mut packages = Vec::new();
618 let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
619 message: format!(
620 "failed to scan workflow packages in `{}`: {source}",
621 directory.display()
622 ),
623 })?;
624
625 for entry in entries {
626 let entry = entry.map_err(|source| ServerError::Config {
627 message: format!(
628 "failed to read workflow package entry in `{}`: {source}",
629 directory.display()
630 ),
631 })?;
632 let path = entry.path();
633 let has_aion_extension = path
634 .extension()
635 .is_some_and(|extension| extension == "aion");
636 if path.is_file() && has_aion_extension {
637 packages.push(path);
638 }
639 }
640
641 packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
642 Ok(packages)
643}
644
645fn merge_workflow_packages(
646 workflow_packages: &mut Vec<PathBuf>,
647 discovered_packages: Vec<PathBuf>,
648 cli_packages: &[PathBuf],
649) {
650 let mut seen: HashSet<PathBuf> = workflow_packages
651 .iter()
652 .map(|package| deduplicated_package_key(package))
653 .collect();
654 for package in discovered_packages
655 .into_iter()
656 .chain(cli_packages.iter().cloned())
657 {
658 if seen.insert(deduplicated_package_key(&package)) {
659 workflow_packages.push(package);
660 }
661 }
662}
663
664fn deduplicated_package_key(path: &Path) -> PathBuf {
665 path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
666}
667
668#[cfg(test)]
669#[path = "load_home_tests.rs"]
670mod home_tests;
671
672#[cfg(test)]
673mod tests {
674 use crate::config::{
675 AutoCreate, DEFAULT_AUTHORING_WORKSPACE_DIR, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, HomeSource,
676 OpsConsoleAssetSource,
677 };
678
679 use super::{
680 CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
681 DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
682 DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
683 DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
684 DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
685 discover_workflow_packages, merge_workflow_packages,
686 };
687
688 #[test]
689 fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
690 let config = ServerConfig::from_slice(
691 br#"
692 [server]
693 listen_address = "127.0.0.1:18080"
694 grpc_address = "127.0.0.1:15051"
695
696 [store]
697 backend = "libsql"
698 url = "aion.db"
699
700 [runtime]
701 scheduler_threads = 2
702 query_timeout_ms = 10000
703
704 [drain]
705 timeout_seconds = 45
706
707 [auth]
708 enabled = true
709 jwks_url = "https://issuer.example.com/.well-known/jwks.json"
710 jwks_refresh_seconds = 60
711
712 [metrics]
713 enabled = true
714
715 [namespaces]
716 default = "production"
717
718 [websocket]
719 outbound_buffer_bound = 16
720 event_broadcast_capacity = 1024
721 cluster_broadcast_capacity = 1024
722 "#,
723 )?;
724
725 assert_eq!(config.store.backend, StoreBackend::LibSql);
726 assert_eq!(config.store.url.as_deref(), Some("aion.db"));
727 assert_eq!(config.runtime.scheduler_threads, 2);
728 assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
729 assert_eq!(config.namespaces.default, "production");
730 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
732 assert_eq!(
735 config.namespaces.max_in_flight_activities,
736 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
737 );
738 assert_eq!(config.websocket.outbound_buffer_bound, 16);
739 assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
740 Ok(())
741 }
742
743 #[test]
744 fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
745 let config = ServerConfig::from_slice(
746 br#"
747 [namespaces]
748 default = "production"
749 auto_create = "closed"
750 "#,
751 )?;
752 assert_eq!(config.namespaces.default, "production");
753 assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
754 Ok(())
755 }
756
757 #[test]
758 fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
759 let config = ServerConfig::from_slice(
760 br#"
761 [namespaces]
762 auto_create = "open"
763 "#,
764 )?;
765 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
766 Ok(())
767 }
768
769 #[test]
770 fn namespaces_max_in_flight_activities_override_parses()
771 -> Result<(), Box<dyn std::error::Error>> {
772 let config = ServerConfig::from_slice(
773 br#"
774 [namespaces]
775 default = "production"
776 max_in_flight_activities = 32
777 "#,
778 )?;
779 assert_eq!(config.namespaces.max_in_flight_activities, 32);
780 let (_store, runtime) = config.into_parts();
782 assert_eq!(runtime.max_in_flight_activities, 32);
783 Ok(())
784 }
785
786 #[test]
787 fn namespaces_max_in_flight_activities_defaults_when_omitted()
788 -> Result<(), Box<dyn std::error::Error>> {
789 let config = ServerConfig::from_slice(
792 br#"
793 [namespaces]
794 default = "production"
795 "#,
796 )?;
797 assert_eq!(
798 config.namespaces.max_in_flight_activities,
799 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
800 );
801 Ok(())
802 }
803
804 #[test]
805 fn namespaces_auto_create_rejects_unknown_variant() {
806 let result = ServerConfig::from_slice(
807 br#"
808 [namespaces]
809 auto_create = "sometimes"
810 "#,
811 );
812 assert!(
813 result.is_err(),
814 "an unknown auto_create variant must fail to parse"
815 );
816 }
817
818 #[test]
819 fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
820 let config = ServerConfig::from_slice(
824 br"
825 [runtime]
826 query_timeout_ms = 10000
827
828 [websocket]
829 cluster_broadcast_capacity = 64
830 ",
831 )?;
832 assert_eq!(
833 config.websocket.event_broadcast_capacity,
834 Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
835 "omitted event_broadcast_capacity must resolve to the default"
836 );
837 Ok(())
838 }
839
840 #[test]
841 fn zero_event_broadcast_capacity_fails_startup_validation() {
842 let result = ServerConfig::from_slice(
843 br"
844 [websocket]
845 event_broadcast_capacity = 0
846 ",
847 );
848
849 let message = result
850 .err()
851 .map_or_else(String::new, |error| error.to_string());
852 assert!(
853 message.contains("websocket.event_broadcast_capacity"),
854 "validation message must name the zero-valued key: {message}"
855 );
856 }
857
858 #[test]
859 fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
860 let config = ServerConfig::from_slice(
864 br"
865 [runtime]
866 scheduler_threads = 1
867 query_timeout_ms = 10000
868
869 [websocket]
870 event_broadcast_capacity = 64
871 ",
872 )?;
873 assert_eq!(
874 config.websocket.cluster_broadcast_capacity,
875 Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
876 "omitted cluster_broadcast_capacity must resolve to the default"
877 );
878 Ok(())
879 }
880
881 #[test]
882 fn zero_cluster_broadcast_capacity_fails_startup_validation() {
883 let result = ServerConfig::from_slice(
884 br"
885 [runtime]
886 query_timeout_ms = 10000
887
888 [websocket]
889 event_broadcast_capacity = 64
890 cluster_broadcast_capacity = 0
891 ",
892 );
893
894 let message = result
895 .err()
896 .map_or_else(String::new, |error| error.to_string());
897 assert!(
898 message.contains("websocket.cluster_broadcast_capacity"),
899 "validation message must name the zero-valued cluster key: {message}"
900 );
901 }
902
903 #[test]
907 fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
908 let config = ServerConfig::from_slice(
909 br"
910 [runtime]
911 query_timeout_ms = 10000
912
913 [websocket]
914 event_broadcast_capacity = 64
915 cluster_broadcast_capacity = 64
916 ",
917 )?;
918 assert_eq!(
919 config.observability.max_event_bytes,
920 crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
921 );
922 assert_eq!(
923 config.observability.max_stream_events,
924 crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
925 );
926 let (_store, runtime) = config.into_parts();
928 assert_eq!(
929 runtime.observability.max_event_bytes,
930 crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
931 );
932 Ok(())
933 }
934
935 #[test]
938 fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
939 let config = ServerConfig::from_slice(
940 br"
941 [runtime]
942 query_timeout_ms = 10000
943
944 [websocket]
945 event_broadcast_capacity = 64
946 cluster_broadcast_capacity = 64
947
948 [observability]
949 max_event_bytes = 512
950 max_stream_events = 3
951 ",
952 )?;
953 assert_eq!(config.observability.max_event_bytes, 512);
954 assert_eq!(config.observability.max_stream_events, 3);
955 let (_store, runtime) = config.into_parts();
956 assert_eq!(runtime.observability.max_event_bytes, 512);
957 assert_eq!(runtime.observability.max_stream_events, 3);
958 Ok(())
959 }
960
961 #[test]
965 fn missing_mcp_section_leaves_the_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
966 let config = ServerConfig::from_slice(
967 br"
968 [runtime]
969 query_timeout_ms = 10000
970
971 [websocket]
972 event_broadcast_capacity = 64
973 cluster_broadcast_capacity = 64
974 ",
975 )?;
976 let (_store, runtime) = config.into_parts();
977 assert!(!runtime.mcp.enabled);
978 assert!(runtime.mcp.allowed_origins.is_empty());
979 assert_eq!(
980 runtime.mcp.discover_ttl_ms,
981 crate::config::DEFAULT_MCP_DISCOVER_TTL_MS
982 );
983 assert_eq!(
984 runtime.mcp.task_poll_interval_ms,
985 crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS
986 );
987 assert_eq!(
988 runtime.mcp.task_ttl_ms,
989 Some(crate::config::DEFAULT_MCP_TASK_TTL_MS)
990 );
991 Ok(())
992 }
993
994 #[test]
997 fn mcp_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
998 let config = ServerConfig::from_slice(
999 br#"
1000 [runtime]
1001 query_timeout_ms = 10000
1002
1003 [websocket]
1004 event_broadcast_capacity = 64
1005 cluster_broadcast_capacity = 64
1006
1007 [mcp]
1008 enabled = true
1009 allowed_origins = ["http://localhost:5173"]
1010 discover_ttl_ms = 1000
1011 tools_list_ttl_ms = 2000
1012 task_ttl_ms = 0
1013 task_poll_interval_ms = 250
1014 await_poll_interval_ms = 125
1015 "#,
1016 )?;
1017 let (_store, runtime) = config.into_parts();
1018 assert!(runtime.mcp.enabled);
1019 assert_eq!(runtime.mcp.allowed_origins, vec!["http://localhost:5173"]);
1020 assert_eq!(runtime.mcp.discover_ttl_ms, 1_000);
1021 assert_eq!(runtime.mcp.tools_list_ttl_ms, 2_000);
1022 assert_eq!(runtime.mcp.task_ttl_ms, None, "zero means unlimited");
1023 assert_eq!(runtime.mcp.task_poll_interval_ms, 250);
1024 assert_eq!(runtime.mcp.await_poll_interval_ms, 125);
1025 Ok(())
1026 }
1027
1028 #[test]
1031 fn zero_mcp_poll_intervals_fail_startup_validation() {
1032 for key in ["task_poll_interval_ms", "await_poll_interval_ms"] {
1033 let source = format!(
1034 "
1035 [runtime]
1036 query_timeout_ms = 10000
1037
1038 [websocket]
1039 event_broadcast_capacity = 64
1040 cluster_broadcast_capacity = 64
1041
1042 [mcp]
1043 enabled = true
1044 {key} = 0
1045 "
1046 );
1047 let result = ServerConfig::from_slice(source.as_bytes());
1048 let message = match result {
1049 Err(crate::ServerError::Config { message }) => message,
1050 _ => String::new(),
1051 };
1052 assert!(message.contains(key), "{key} must be refused: {message}");
1053 }
1054 }
1055
1056 #[test]
1059 fn a_dark_mcp_surface_does_not_validate_its_unused_knobs()
1060 -> Result<(), Box<dyn std::error::Error>> {
1061 let config = ServerConfig::from_slice(
1062 br"
1063 [runtime]
1064 query_timeout_ms = 10000
1065
1066 [websocket]
1067 event_broadcast_capacity = 64
1068 cluster_broadcast_capacity = 64
1069
1070 [mcp]
1071 task_poll_interval_ms = 0
1072 ",
1073 )?;
1074 let (_store, runtime) = config.into_parts();
1075 assert!(!runtime.mcp.enabled);
1076 assert_eq!(
1077 runtime.mcp.task_poll_interval_ms,
1078 crate::config::DEFAULT_MCP_TASK_POLL_INTERVAL_MS,
1079 "an unused zero resolves to the default rather than to a busy loop"
1080 );
1081 Ok(())
1082 }
1083
1084 #[test]
1085 fn zero_observability_max_event_bytes_fails_startup_validation() {
1086 let result = ServerConfig::from_slice(
1087 br"
1088 [runtime]
1089 query_timeout_ms = 10000
1090
1091 [websocket]
1092 event_broadcast_capacity = 64
1093 cluster_broadcast_capacity = 64
1094
1095 [observability]
1096 max_event_bytes = 0
1097 ",
1098 );
1099 let message = result
1100 .err()
1101 .map_or_else(String::new, |error| error.to_string());
1102 assert!(
1103 message.contains("observability.max_event_bytes"),
1104 "validation message must name the zero-valued key: {message}"
1105 );
1106 }
1107
1108 #[test]
1109 fn zero_observability_max_stream_events_fails_startup_validation() {
1110 let result = ServerConfig::from_slice(
1111 br"
1112 [runtime]
1113 query_timeout_ms = 10000
1114
1115 [websocket]
1116 event_broadcast_capacity = 64
1117 cluster_broadcast_capacity = 64
1118
1119 [observability]
1120 max_stream_events = 0
1121 ",
1122 );
1123 let message = result
1124 .err()
1125 .map_or_else(String::new, |error| error.to_string());
1126 assert!(
1127 message.contains("observability.max_stream_events"),
1128 "validation message must name the zero-valued key: {message}"
1129 );
1130 }
1131
1132 #[test]
1133 fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
1134 let config = ServerConfig::from_slice(
1138 br"
1139 [runtime]
1140 scheduler_threads = 1
1141
1142 [websocket]
1143 event_broadcast_capacity = 64
1144 cluster_broadcast_capacity = 64
1145 ",
1146 )?;
1147 assert_eq!(
1148 config.runtime.query_timeout_ms,
1149 Some(DEFAULT_QUERY_TIMEOUT_MS),
1150 "omitted query_timeout_ms must resolve to the default"
1151 );
1152 Ok(())
1153 }
1154
1155 #[test]
1156 fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
1157 let config = ServerConfig::from_slice(b"")?;
1162 assert_eq!(config.store.backend, StoreBackend::Haematite);
1163 assert_eq!(
1164 config.runtime.query_timeout_ms,
1165 Some(DEFAULT_QUERY_TIMEOUT_MS)
1166 );
1167 assert_eq!(
1168 config.websocket.event_broadcast_capacity,
1169 Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
1170 );
1171 assert_eq!(
1172 config.websocket.cluster_broadcast_capacity,
1173 Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
1174 );
1175 Ok(())
1176 }
1177
1178 #[test]
1179 fn zero_query_timeout_fails_startup_validation() {
1180 let result = ServerConfig::from_slice(
1181 br"
1182 [runtime]
1183 query_timeout_ms = 0
1184
1185 [websocket]
1186 event_broadcast_capacity = 64
1187 cluster_broadcast_capacity = 64
1188 ",
1189 );
1190
1191 let message = result
1192 .err()
1193 .map_or_else(String::new, |error| error.to_string());
1194 assert!(
1195 message.contains("runtime.query_timeout_ms"),
1196 "validation message must name the zero-valued key: {message}"
1197 );
1198 }
1199
1200 #[test]
1205 fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
1206 let config = ServerConfig::from_slice(
1207 br"
1208 [runtime]
1209 query_timeout_ms = 10000
1210
1211 [websocket]
1212 event_broadcast_capacity = 64
1213 cluster_broadcast_capacity = 64
1214
1215 [deploy]
1216 enabled = true
1217 ",
1218 )?;
1219
1220 assert_eq!(
1221 config.deploy.max_archive_bytes,
1222 Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
1223 "omitted max_archive_bytes must resolve to the conservative default"
1224 );
1225 assert_eq!(
1226 config.deploy.max_inflated_bytes,
1227 Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1228 "omitted max_inflated_bytes must resolve to the conservative default"
1229 );
1230 Ok(())
1231 }
1232
1233 #[test]
1234 fn deploy_zero_max_archive_bytes_fails_startup_validation() {
1235 let result = ServerConfig::from_slice(
1236 br"
1237 [runtime]
1238 query_timeout_ms = 10000
1239
1240 [websocket]
1241 event_broadcast_capacity = 64
1242 cluster_broadcast_capacity = 64
1243
1244 [deploy]
1245 enabled = true
1246 max_archive_bytes = 0
1247 ",
1248 );
1249
1250 let message = result
1251 .err()
1252 .map_or_else(String::new, |error| error.to_string());
1253 assert!(
1254 message.contains("deploy.max_archive_bytes"),
1255 "validation message must name the zero-valued key: {message}"
1256 );
1257 }
1258
1259 #[test]
1264 fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
1265 let config = ServerConfig::from_slice(
1266 br"
1267 [runtime]
1268 query_timeout_ms = 10000
1269
1270 [websocket]
1271 event_broadcast_capacity = 64
1272 cluster_broadcast_capacity = 64
1273
1274 [deploy]
1275 enabled = true
1276 max_archive_bytes = 16777216
1277 ",
1278 )?;
1279
1280 assert_eq!(
1281 config.deploy.max_archive_bytes,
1282 Some(16_777_216),
1283 "explicit max_archive_bytes must be left untouched"
1284 );
1285 assert_eq!(
1286 config.deploy.max_inflated_bytes,
1287 Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1288 "omitted max_inflated_bytes must resolve to the conservative default"
1289 );
1290 Ok(())
1291 }
1292
1293 #[test]
1294 fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
1295 let result = ServerConfig::from_slice(
1296 br"
1297 [runtime]
1298 query_timeout_ms = 10000
1299
1300 [websocket]
1301 event_broadcast_capacity = 64
1302 cluster_broadcast_capacity = 64
1303
1304 [deploy]
1305 enabled = true
1306 max_archive_bytes = 16777216
1307 max_inflated_bytes = 0
1308 ",
1309 );
1310
1311 let message = result
1312 .err()
1313 .map_or_else(String::new, |error| error.to_string());
1314 assert!(
1315 message.contains("deploy.max_inflated_bytes"),
1316 "validation message must name the zero-valued key: {message}"
1317 );
1318 }
1319
1320 #[test]
1323 fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
1324 let result = ServerConfig::from_slice(
1325 br"
1326 [runtime]
1327 query_timeout_ms = 10000
1328
1329 [websocket]
1330 event_broadcast_capacity = 64
1331 cluster_broadcast_capacity = 64
1332
1333 [deploy]
1334 enabled = true
1335 max_archive_bytes = 16777216
1336 max_inflated_bytes = 16777215
1337 ",
1338 );
1339
1340 let message = result
1341 .err()
1342 .map_or_else(String::new, |error| error.to_string());
1343 assert!(
1344 message.contains("deploy.max_inflated_bytes")
1345 && message.contains("deploy.max_archive_bytes"),
1346 "validation message must name both ceilings: {message}"
1347 );
1348 }
1349
1350 #[test]
1353 fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
1354 let config = ServerConfig::from_slice(
1355 br"
1356 [runtime]
1357 query_timeout_ms = 10000
1358
1359 [websocket]
1360 event_broadcast_capacity = 64
1361 cluster_broadcast_capacity = 64
1362 ",
1363 )?;
1364
1365 assert!(!config.deploy.enabled);
1366 assert_eq!(config.deploy.max_archive_bytes, None);
1367 assert_eq!(config.deploy.max_inflated_bytes, None);
1368 Ok(())
1369 }
1370
1371 #[test]
1372 fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
1373 let config = ServerConfig::from_slice(
1374 br"
1375 [runtime]
1376 query_timeout_ms = 10000
1377
1378 [websocket]
1379 event_broadcast_capacity = 64
1380 cluster_broadcast_capacity = 64
1381
1382 [deploy]
1383 enabled = true
1384 max_archive_bytes = 16777216
1385 max_inflated_bytes = 67108864
1386 ",
1387 )?;
1388
1389 assert!(config.deploy.enabled);
1390 assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
1391 assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
1392 Ok(())
1393 }
1394
1395 #[test]
1399 fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
1400 let config = ServerConfig::from_slice(
1401 br"
1402 [runtime]
1403 query_timeout_ms = 10000
1404
1405 [websocket]
1406 event_broadcast_capacity = 64
1407 cluster_broadcast_capacity = 64
1408 ",
1409 )?;
1410
1411 assert!(config.server.cors_allowed_origins.is_empty());
1412 let (_, runtime) = config.into_parts();
1413 assert!(runtime.cors_allowed_origins.is_empty());
1414 Ok(())
1415 }
1416
1417 #[test]
1420 fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
1421 let config = ServerConfig::from_slice(
1422 br#"
1423 [server]
1424 cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
1425
1426 [runtime]
1427 query_timeout_ms = 10000
1428
1429 [websocket]
1430 event_broadcast_capacity = 64
1431 cluster_broadcast_capacity = 64
1432 "#,
1433 )?;
1434
1435 assert_eq!(
1436 config.server.cors_allowed_origins,
1437 vec![
1438 "http://localhost:5173".to_owned(),
1439 "http://127.0.0.1:5173".to_owned()
1440 ]
1441 );
1442 let (_, runtime) = config.into_parts();
1443 assert_eq!(
1444 runtime.cors_allowed_origins,
1445 vec![
1446 "http://localhost:5173".to_owned(),
1447 "http://127.0.0.1:5173".to_owned()
1448 ]
1449 );
1450 Ok(())
1451 }
1452
1453 #[test]
1457 fn cors_allowed_origins_reject_malformed() {
1458 for bad in ["", "localhost:5173", "http://localhost:5173/"] {
1459 let toml = format!(
1460 "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
1461 );
1462 let result = ServerConfig::from_slice(toml.as_bytes());
1463 let message = result
1464 .err()
1465 .map_or_else(String::new, |error| error.to_string());
1466 assert!(
1467 message.contains("cors_allowed_origins"),
1468 "malformed origin `{bad}` must be rejected naming the key: {message}"
1469 );
1470 }
1471 }
1472
1473 #[test]
1475 fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1476 let config = ServerConfig::from_slice(
1477 br"
1478 [runtime]
1479 query_timeout_ms = 10000
1480
1481 [websocket]
1482 event_broadcast_capacity = 64
1483 cluster_broadcast_capacity = 64
1484 ",
1485 )?;
1486
1487 assert!(!config.dev.enabled);
1488 Ok(())
1489 }
1490
1491 #[test]
1494 fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
1495 let config = ServerConfig::from_slice(
1496 br"
1497 [runtime]
1498 query_timeout_ms = 10000
1499
1500 [websocket]
1501 event_broadcast_capacity = 64
1502 cluster_broadcast_capacity = 64
1503
1504 [dev]
1505 enabled = true
1506 ",
1507 )?;
1508
1509 assert!(config.dev.enabled);
1510 Ok(())
1511 }
1512
1513 #[test]
1516 fn authoring_absent_defaults_awl_workspace_but_keeps_gleam_dark()
1517 -> Result<(), Box<dyn std::error::Error>> {
1518 let home = crate::test_support::private_tempdir()?;
1519 let config = ServerConfig::from_slice_with_home(b"", home.path())?;
1520
1521 assert_eq!(config.authoring.gleam_path, None);
1522 assert_eq!(config.authoring.project_root, None);
1523 assert_eq!(
1524 config.authoring.workspace_dir.as_deref(),
1525 Some(home.path().join(DEFAULT_AUTHORING_WORKSPACE_DIR).as_path())
1526 );
1527 Ok(())
1528 }
1529
1530 #[test]
1532 fn authoring_explicit_workspace_is_honored() -> Result<(), Box<dyn std::error::Error>> {
1533 let config = ServerConfig::from_slice(
1534 br#"
1535 [authoring]
1536 workspace_dir = "/srv/aion/studio"
1537 "#,
1538 )?;
1539
1540 assert_eq!(
1541 config.authoring.workspace_dir.as_deref(),
1542 Some(std::path::Path::new("/srv/aion/studio"))
1543 );
1544 Ok(())
1545 }
1546
1547 #[test]
1550 fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1551 let config = ServerConfig::from_slice(
1552 br#"
1553 [runtime]
1554 query_timeout_ms = 10000
1555
1556 [websocket]
1557 event_broadcast_capacity = 64
1558 cluster_broadcast_capacity = 64
1559
1560 [authoring]
1561 gleam_path = "/usr/local/bin/gleam"
1562 project_root = "/srv/aion/authoring"
1563 "#,
1564 )?;
1565
1566 assert_eq!(
1567 config.authoring.gleam_path.as_deref(),
1568 Some(std::path::Path::new("/usr/local/bin/gleam"))
1569 );
1570 let (_, runtime) = config.into_parts();
1571 assert_eq!(
1572 runtime.authoring.gleam_path.as_deref(),
1573 Some(std::path::Path::new("/usr/local/bin/gleam"))
1574 );
1575 assert_eq!(
1576 runtime.authoring.project_root.as_deref(),
1577 Some(std::path::Path::new("/srv/aion/authoring"))
1578 );
1579 Ok(())
1580 }
1581
1582 #[test]
1586 fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
1587 let result = ServerConfig::from_slice(
1588 br#"
1589 [runtime]
1590 query_timeout_ms = 10000
1591
1592 [websocket]
1593 event_broadcast_capacity = 64
1594 cluster_broadcast_capacity = 64
1595
1596 [authoring]
1597 gleam_path = "/usr/local/bin/gleam"
1598 "#,
1599 );
1600
1601 let message = result
1602 .err()
1603 .map_or_else(String::new, |error| error.to_string());
1604 assert!(
1605 message.contains("authoring.project_root"),
1606 "validation message must name the missing key: {message}"
1607 );
1608 assert!(
1609 message.contains("AION_AUTHORING_PROJECT_ROOT"),
1610 "validation message must name the environment override: {message}"
1611 );
1612 }
1613
1614 #[test]
1617 fn authoring_empty_gleam_path_fails_naming_key_and_env() {
1618 let result = ServerConfig::from_slice(
1619 br#"
1620 [runtime]
1621 query_timeout_ms = 10000
1622
1623 [websocket]
1624 event_broadcast_capacity = 64
1625 cluster_broadcast_capacity = 64
1626
1627 [authoring]
1628 gleam_path = ""
1629 "#,
1630 );
1631
1632 let message = result
1633 .err()
1634 .map_or_else(String::new, |error| error.to_string());
1635 assert!(
1636 message.contains("authoring.gleam_path"),
1637 "validation message must name the empty key: {message}"
1638 );
1639 assert!(
1640 message.contains("AION_AUTHORING_GLEAM_PATH"),
1641 "validation message must name the environment override: {message}"
1642 );
1643 }
1644
1645 #[test]
1647 fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
1648 let mut config = ServerConfig::from_slice(
1649 br"
1650 [runtime]
1651 query_timeout_ms = 10000
1652
1653 [websocket]
1654 event_broadcast_capacity = 64
1655 cluster_broadcast_capacity = 64
1656 ",
1657 )?;
1658 let cli = CliOverrides {
1659 gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
1660 authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
1661 ..CliOverrides::default()
1662 };
1663
1664 config.apply_cli_overrides(&cli);
1665 config.validate()?;
1666
1667 assert_eq!(
1668 config.authoring.gleam_path.as_deref(),
1669 Some(std::path::Path::new("/opt/gleam"))
1670 );
1671 assert_eq!(
1672 config.authoring.project_root.as_deref(),
1673 Some(std::path::Path::new("/opt/project"))
1674 );
1675 Ok(())
1676 }
1677
1678 #[test]
1683 fn queue_service_settings_are_read_from_the_worker_section()
1684 -> Result<(), Box<dyn std::error::Error>> {
1685 use crate::worker::QueueServicePolicy;
1686 use std::time::Duration;
1687
1688 let bare = ServerConfig::from_slice(
1689 br"
1690 [websocket]
1691 event_broadcast_capacity = 64
1692 cluster_broadcast_capacity = 64
1693 ",
1694 )?;
1695 assert_eq!(
1696 bare.worker.queue_service.default_policy,
1697 QueueServicePolicy::Strict,
1698 "strict is the default with nothing written"
1699 );
1700 assert_eq!(
1701 bare.worker.queue_service.service_availability_deadline,
1702 None
1703 );
1704 assert_eq!(bare.worker.queue_service.schedule_to_start_timeout, None);
1705
1706 let written = ServerConfig::from_slice(
1707 br#"
1708 [websocket]
1709 event_broadcast_capacity = 64
1710 cluster_broadcast_capacity = 64
1711
1712 [worker.queue_service]
1713 service_availability_deadline = 45000
1714 schedule_to_start_timeout = 5000
1715
1716 [[worker.queue_service.overrides]]
1717 task_queue = "general"
1718 policy = "durable_pending"
1719 "#,
1720 )?;
1721 assert_eq!(
1722 written.worker.queue_service.service_availability_deadline,
1723 Some(Duration::from_secs(45))
1724 );
1725 assert_eq!(
1726 written.worker.queue_service.schedule_to_start_timeout,
1727 Some(Duration::from_secs(5))
1728 );
1729 assert_eq!(
1730 written
1731 .worker
1732 .queue_service
1733 .policy_for("default", "general"),
1734 QueueServicePolicy::DurablePending,
1735 "the written opt-in must reach the dispatch seam"
1736 );
1737 assert_eq!(
1738 written
1739 .worker
1740 .queue_service
1741 .policy_for("default", "billing"),
1742 QueueServicePolicy::Strict,
1743 "an override must not leak onto other queues"
1744 );
1745
1746 let (_store, runtime) = written.into_parts();
1748 assert_eq!(
1749 runtime
1750 .worker
1751 .queue_service
1752 .policy_for("default", "general"),
1753 QueueServicePolicy::DurablePending
1754 );
1755 Ok(())
1756 }
1757
1758 #[test]
1759 fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
1760 let config = ServerConfig::from_slice(
1761 br#"
1762 [runtime]
1763 query_timeout_ms = 10000
1764
1765 [websocket]
1766 event_broadcast_capacity = 64
1767 cluster_broadcast_capacity = 64
1768
1769 [dashboard]
1770 source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1771 "#,
1772 )?;
1773 match &config.ops_console.source {
1774 OpsConsoleAssetSource::FileSystem { asset_path } => {
1775 assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
1776 }
1777 OpsConsoleAssetSource::Embedded => {
1778 return Err("legacy [dashboard] section must map to ops_console".into());
1779 }
1780 }
1781 Ok(())
1782 }
1783
1784 #[test]
1786 fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
1787 let config = ServerConfig::from_slice(
1788 br#"
1789 [runtime]
1790 query_timeout_ms = 10000
1791
1792 [websocket]
1793 event_broadcast_capacity = 64
1794 cluster_broadcast_capacity = 64
1795
1796 [ops_console]
1797 source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1798 "#,
1799 )?;
1800 assert!(matches!(
1801 config.ops_console.source,
1802 OpsConsoleAssetSource::FileSystem { .. }
1803 ));
1804 Ok(())
1805 }
1806
1807 #[test]
1808 fn invalid_values_name_problematic_field() {
1809 let result = ServerConfig::from_slice(
1810 br"
1811 [runtime]
1812 scheduler_threads = 0
1813 ",
1814 );
1815
1816 let message = result
1817 .err()
1818 .map_or_else(String::new, |error| error.to_string());
1819 assert!(message.contains("runtime.scheduler_threads"));
1820 }
1821
1822 #[test]
1823 fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
1824 let mut config = ServerConfig::from_slice(
1825 br#"
1826 [store]
1827 backend = "libsql"
1828 url = "file.db"
1829
1830 [runtime]
1831 query_timeout_ms = 10000
1832
1833 [websocket]
1834 event_broadcast_capacity = 64
1835 cluster_broadcast_capacity = 64
1836 "#,
1837 )?;
1838 let cli = CliOverrides {
1839 store_url: Some("cli.db".to_owned()),
1840 scheduler_threads: Some(3),
1841 ..CliOverrides::default()
1842 };
1843
1844 config.apply_cli_overrides(&cli);
1845 config.validate()?;
1846
1847 assert_eq!(config.store.url.as_deref(), Some("cli.db"));
1848 assert_eq!(config.runtime.scheduler_threads, 3);
1849 Ok(())
1850 }
1851
1852 #[test]
1853 fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
1854 let mut config = ServerConfig::default();
1855
1856 assert_eq!(config.store.backend, StoreBackend::Haematite);
1859 assert_eq!(config.store.data_dir, None);
1860 assert_eq!(config.store.shard_count, 64);
1864 assert_eq!(config.store.url, None);
1865 assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
1866 assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
1867 assert_eq!(config.namespaces.default, "default");
1868 assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
1872 assert_eq!(
1875 config.namespaces.max_in_flight_activities,
1876 DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
1877 );
1878 assert_eq!(config.namespaces.max_in_flight_activities, 1024);
1879 assert!(!config.auth.enabled);
1880 assert!(config.metrics.enabled);
1881 assert_eq!(config.websocket.event_broadcast_capacity, None);
1885 assert_eq!(config.websocket.cluster_broadcast_capacity, None);
1886 assert_eq!(config.runtime.query_timeout_ms, None);
1887 config.websocket.event_broadcast_capacity = Some(64);
1888 config.websocket.cluster_broadcast_capacity = Some(64);
1889 config.runtime.query_timeout_ms = Some(10_000);
1890 let home = crate::test_support::private_tempdir()?;
1891 let working_dir = crate::test_support::private_tempdir()?;
1892 super::fill_home_defaults(
1893 &mut config,
1894 home.path(),
1895 HomeSource::Derived,
1896 working_dir.path(),
1897 )?;
1898 assert_eq!(
1899 config.store.data_dir.as_deref(),
1900 home.path().join("data").to_str()
1901 );
1902 config.validate()?;
1903 Ok(())
1904 }
1905
1906 #[test]
1907 fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
1908 {
1909 let mut config = ServerConfig::default();
1910 config.store.data_dir = Some("test-data".to_owned());
1911 config.websocket.event_broadcast_capacity = Some(64);
1912 config.websocket.cluster_broadcast_capacity = Some(64);
1913 config.runtime.query_timeout_ms = Some(10_000);
1914
1915 assert!(!config.outbox.enabled);
1919 assert_eq!(config.outbox.poll_interval_ms, None);
1920 assert_eq!(config.outbox.batch_size, None);
1921 assert_eq!(config.outbox.max_attempts, None);
1922 assert_eq!(config.outbox.backoff_base_ms, None);
1923 assert_eq!(config.outbox.backoff_multiplier, None);
1924 assert_eq!(config.outbox.backoff_max_ms, None);
1925 assert_eq!(config.outbox.reconcile_interval_ms, None);
1926 assert_eq!(config.outbox.reconcile_stale_after_ms, None);
1927 config.validate()?;
1928 Ok(())
1929 }
1930
1931 fn outbox_enabled_base() -> ServerConfig {
1932 let mut config = ServerConfig::default();
1933 config.store.data_dir = Some("test-data".to_owned());
1934 config.websocket.event_broadcast_capacity = Some(64);
1935 config.websocket.cluster_broadcast_capacity = Some(64);
1936 config.runtime.query_timeout_ms = Some(10_000);
1937 config.outbox.enabled = true;
1938 config.outbox.poll_interval_ms = Some(250);
1939 config.outbox.batch_size = Some(64);
1940 config.outbox.max_attempts = Some(5);
1941 config.outbox.backoff_base_ms = Some(100);
1942 config.outbox.backoff_multiplier = Some(2);
1943 config.outbox.backoff_max_ms = Some(30_000);
1944 config.outbox.reconcile_interval_ms = Some(1_000);
1945 config.outbox.reconcile_stale_after_ms = Some(60_000);
1946 config
1947 }
1948
1949 #[test]
1950 fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
1951 outbox_enabled_base().validate()?;
1952 Ok(())
1953 }
1954
1955 #[test]
1956 fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
1957 let config = ServerConfig::from_slice(
1961 br"
1962 [runtime]
1963 query_timeout_ms = 10000
1964
1965 [websocket]
1966 event_broadcast_capacity = 64
1967 cluster_broadcast_capacity = 64
1968
1969 [outbox]
1970 enabled = true
1971 ",
1972 )?;
1973 assert_eq!(
1974 config.outbox.poll_interval_ms,
1975 Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
1976 "omitted poll_interval_ms must resolve to the default"
1977 );
1978 Ok(())
1979 }
1980
1981 #[test]
1982 fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
1983 let config = ServerConfig::from_slice(
1986 br"
1987 [runtime]
1988 query_timeout_ms = 10000
1989
1990 [websocket]
1991 event_broadcast_capacity = 64
1992 cluster_broadcast_capacity = 64
1993
1994 [outbox]
1995 enabled = true
1996 poll_interval_ms = 250
1997 ",
1998 )?;
1999 assert_eq!(
2000 config.outbox.poll_interval_ms,
2001 Some(250),
2002 "explicit poll_interval_ms must be left untouched"
2003 );
2004 assert_eq!(
2005 config.outbox.max_attempts,
2006 Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
2007 "omitted max_attempts must resolve to the default"
2008 );
2009 Ok(())
2010 }
2011
2012 #[test]
2013 fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
2014 -> Result<(), Box<dyn std::error::Error>> {
2015 let config = ServerConfig::from_slice(
2019 br"
2020 [runtime]
2021 query_timeout_ms = 10000
2022
2023 [websocket]
2024 event_broadcast_capacity = 64
2025 cluster_broadcast_capacity = 64
2026
2027 [outbox]
2028 enabled = true
2029 ",
2030 )?;
2031 assert!(config.outbox.enabled);
2032 assert_eq!(
2033 config.outbox.poll_interval_ms,
2034 Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
2035 );
2036 assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
2037 assert_eq!(
2038 config.outbox.max_attempts,
2039 Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
2040 );
2041 assert_eq!(
2042 config.outbox.backoff_base_ms,
2043 Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
2044 );
2045 assert_eq!(
2046 config.outbox.backoff_multiplier,
2047 Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
2048 );
2049 assert_eq!(
2050 config.outbox.backoff_max_ms,
2051 Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
2052 );
2053 assert_eq!(config.outbox.reconcile_interval_ms, None);
2056 assert_eq!(config.outbox.reconcile_stale_after_ms, None);
2057 Ok(())
2058 }
2059
2060 #[test]
2061 fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2062 let mut config = outbox_enabled_base();
2065 config.outbox.poll_interval_ms = Some(0);
2066 let error = config
2067 .validate()
2068 .err()
2069 .ok_or("enabled outbox with zero poll interval must fail")?;
2070 assert!(
2071 error.to_string().contains("outbox.poll_interval_ms"),
2072 "error must name the zero-valued key: {error}"
2073 );
2074 Ok(())
2075 }
2076
2077 #[test]
2078 fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2079 let mut config = outbox_enabled_base();
2080 config.outbox.max_attempts = Some(0);
2081 let error = config
2082 .validate()
2083 .err()
2084 .ok_or("enabled outbox with zero max attempts must fail")?;
2085 assert!(
2086 error.to_string().contains("outbox.max_attempts"),
2087 "error must name the zero-valued key: {error}"
2088 );
2089 Ok(())
2090 }
2091
2092 #[test]
2093 fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
2094 let mut config = outbox_enabled_base();
2095 config.outbox.backoff_base_ms = Some(1_000);
2096 config.outbox.backoff_max_ms = Some(500);
2097 let error = config
2098 .validate()
2099 .err()
2100 .ok_or("backoff_max below backoff_base must fail")?;
2101 assert!(
2102 error.to_string().contains("outbox.backoff_max_ms"),
2103 "error must name the offending key: {error}"
2104 );
2105 Ok(())
2106 }
2107
2108 #[test]
2109 fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
2110 let mut config = outbox_enabled_base();
2111 config.outbox.reconcile_interval_ms = None;
2112 config.outbox.reconcile_stale_after_ms = None;
2113 config.validate()?;
2114 Ok(())
2115 }
2116
2117 #[test]
2118 fn outbox_reconciliation_requires_interval_when_partially_enabled()
2119 -> Result<(), Box<dyn std::error::Error>> {
2120 let mut config = outbox_enabled_base();
2121 config.outbox.reconcile_interval_ms = None;
2122 let error = config
2123 .validate()
2124 .err()
2125 .ok_or("reconciliation without interval must fail")?;
2126 assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
2127 Ok(())
2128 }
2129
2130 #[test]
2131 fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
2132 -> Result<(), Box<dyn std::error::Error>> {
2133 let mut config = outbox_enabled_base();
2134 config.outbox.reconcile_stale_after_ms = None;
2135 let error = config
2136 .validate()
2137 .err()
2138 .ok_or("reconciliation without stale threshold must fail")?;
2139 assert!(
2140 error
2141 .to_string()
2142 .contains("outbox.reconcile_stale_after_ms")
2143 );
2144 Ok(())
2145 }
2146
2147 #[test]
2148 fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
2149 let temp_dir = crate::test_support::private_tempdir()?;
2150 std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
2151 std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
2152 std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
2153 std::fs::create_dir(temp_dir.path().join("nested"))?;
2154 std::fs::write(
2155 temp_dir.path().join("nested").join("nested.aion"),
2156 b"package",
2157 )?;
2158
2159 let packages = discover_workflow_packages(temp_dir.path())?;
2160
2161 assert_eq!(
2162 packages,
2163 vec![
2164 temp_dir.path().join("alpha.aion"),
2165 temp_dir.path().join("zeta.aion"),
2166 ]
2167 );
2168 Ok(())
2169 }
2170
2171 #[test]
2172 fn workflow_package_merge_is_additive_and_deduplicated() {
2173 let mut packages = vec!["config.aion".into(), "shared.aion".into()];
2174 let discovered = vec!["auto.aion".into(), "shared.aion".into()];
2175 let cli = vec!["cli.aion".into(), "auto.aion".into()];
2176
2177 merge_workflow_packages(&mut packages, discovered, &cli);
2178
2179 assert_eq!(
2180 packages,
2181 vec![
2182 std::path::PathBuf::from("config.aion"),
2183 std::path::PathBuf::from("shared.aion"),
2184 std::path::PathBuf::from("auto.aion"),
2185 std::path::PathBuf::from("cli.aion"),
2186 ]
2187 );
2188 }
2189
2190 #[test]
2191 fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
2192 let temp_dir = crate::test_support::private_tempdir()?;
2193 let package = temp_dir.path().join("hello.aion");
2194 std::fs::write(&package, b"package")?;
2195 let mut packages = vec![package.clone()];
2196 let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
2197
2198 merge_workflow_packages(&mut packages, discovered, &[]);
2199
2200 assert_eq!(packages, vec![package]);
2201 Ok(())
2202 }
2203
2204 #[test]
2205 fn zero_config_cli_workflow_package_uses_in_memory_defaults()
2206 -> Result<(), Box<dyn std::error::Error>> {
2207 let temp_dir = crate::test_support::private_tempdir()?;
2208
2209 let cli = CliOverrides {
2210 workflow_packages: vec!["hello-world.aion".into()],
2211 ..CliOverrides::default()
2212 };
2213 let mut config = ServerConfig::default();
2214 config.store.backend = StoreBackend::Memory;
2218 config.store.data_dir = None;
2219 config.websocket.event_broadcast_capacity = Some(64);
2224 config.websocket.cluster_broadcast_capacity = Some(64);
2225 config.runtime.query_timeout_ms = Some(10_000);
2226 config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
2227
2228 config.validate()?;
2229
2230 assert_eq!(config.store.backend, StoreBackend::Memory);
2231 assert_eq!(config.store.url, None);
2232 assert_eq!(
2233 config.workflow_packages,
2234 vec![std::path::PathBuf::from("hello-world.aion")]
2235 );
2236 Ok(())
2237 }
2238
2239 #[test]
2240 fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
2241 let mut config = ServerConfig::from_slice(
2242 br#"
2243 workflow_packages = ["config.aion"]
2244
2245 [runtime]
2246 query_timeout_ms = 10000
2247
2248 [websocket]
2249 event_broadcast_capacity = 64
2250 cluster_broadcast_capacity = 64
2251 "#,
2252 )?;
2253 let cli = CliOverrides {
2254 workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
2255 ..CliOverrides::default()
2256 };
2257
2258 merge_workflow_packages(
2259 &mut config.workflow_packages,
2260 Vec::new(),
2261 &cli.workflow_packages,
2262 );
2263
2264 assert_eq!(
2265 config.workflow_packages,
2266 vec![
2267 std::path::PathBuf::from("config.aion"),
2268 std::path::PathBuf::from("cli-one.aion"),
2269 std::path::PathBuf::from("cli-two.aion"),
2270 ]
2271 );
2272 Ok(())
2273 }
2274}