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