1use std::{
4 collections::HashSet,
5 fs,
6 net::SocketAddr,
7 path::{Path, PathBuf},
8 time::Duration,
9};
10
11use serde::Deserialize;
12
13use crate::error::ServerError;
14
15pub mod env;
17pub mod file;
19
20const DEFAULT_HTTP_ADDRESS: SocketAddr =
21 SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 8080);
22const DEFAULT_GRPC_ADDRESS: SocketAddr =
23 SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 50051);
24
25#[derive(Debug, Default)]
27pub struct CliOverrides {
28 pub config_path: Option<PathBuf>,
30 pub listen_address: Option<SocketAddr>,
32 pub store_url: Option<String>,
34 pub scheduler_threads: Option<usize>,
36 pub drain_timeout_seconds: Option<u64>,
38 pub workflow_packages: Vec<PathBuf>,
40 pub gleam_path: Option<PathBuf>,
44 pub authoring_project_root: Option<PathBuf>,
47}
48
49#[derive(Clone, Debug, Deserialize)]
51#[serde(default, deny_unknown_fields)]
52#[derive(Default)]
53pub struct ServerConfig {
54 pub server: ServerSection,
56 pub store: StoreConfig,
58 pub runtime: RuntimeSection,
60 pub drain: DrainConfig,
62 pub auth: AuthConfig,
64 pub metrics: MetricsConfig,
66 pub namespaces: NamespacesConfig,
68 pub tls: Option<TlsConfig>,
70 pub dashboard: DashboardConfig,
72 pub namespace: NamespaceConfig,
74 pub worker: WorkerConfig,
76 pub websocket: WebSocketConfig,
78 pub workflow_packages: Vec<PathBuf>,
80 pub deploy: DeployConfig,
82 pub authoring: AuthoringConfig,
84 pub dev: DevConfig,
86 pub outbox: OutboxConfig,
88}
89
90#[derive(Clone, Debug, Deserialize)]
92#[serde(default, deny_unknown_fields)]
93pub struct ServerSection {
94 pub listen_address: SocketAddr,
96 pub grpc_address: SocketAddr,
98 #[serde(default)]
109 pub cors_allowed_origins: Vec<String>,
110}
111
112#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
114#[serde(rename_all = "lowercase")]
115pub enum StoreBackend {
116 Memory,
118 LibSql,
120 Haematite,
122}
123
124#[derive(Clone, Debug, Deserialize)]
126#[serde(default, deny_unknown_fields)]
127pub struct StoreConfig {
128 pub backend: StoreBackend,
130 pub url: Option<String>,
132 pub owned_shards: Vec<usize>,
139 pub data_dir: Option<String>,
143 pub shard_count: usize,
147 pub cluster: Option<ClusterConfig>,
157}
158
159#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
167#[serde(deny_unknown_fields)]
168pub struct ClusterConfig {
169 pub node_id: String,
172 pub bind_address: SocketAddr,
175 #[serde(default)]
180 pub members: Vec<String>,
181 #[serde(default)]
185 pub peers: Vec<ClusterPeer>,
186 #[serde(default)]
190 pub failover_poll_interval_ms: Option<u64>,
191 #[serde(default)]
196 pub failover_confirmations: Option<u32>,
197}
198
199pub const DEFAULT_FAILOVER_POLL_INTERVAL_MS: u64 = 500;
202
203pub const DEFAULT_FAILOVER_CONFIRMATIONS: u32 = 3;
206
207#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
209#[serde(deny_unknown_fields)]
210pub struct ClusterPeer {
211 pub name: String,
213 pub address: SocketAddr,
215 #[serde(default)]
222 pub grpc_address: Option<SocketAddr>,
223 #[serde(default)]
231 pub owned_shards: Vec<usize>,
232}
233
234#[derive(Clone, Debug, Deserialize)]
236#[serde(default, deny_unknown_fields)]
237pub struct RuntimeSection {
238 pub scheduler_threads: usize,
240 pub query_timeout_ms: Option<u64>,
245}
246
247#[derive(Clone, Debug, Deserialize)]
249#[serde(default, deny_unknown_fields)]
250pub struct DrainConfig {
251 pub timeout_seconds: u64,
253}
254
255#[derive(Clone, Debug, Deserialize)]
257#[serde(default, deny_unknown_fields)]
258pub struct AuthConfig {
259 pub enabled: bool,
261 pub jwks_url: Option<String>,
263 pub jwks_refresh_seconds: u64,
265}
266
267#[derive(Clone, Debug, Deserialize)]
269#[serde(default, deny_unknown_fields)]
270pub struct MetricsConfig {
271 pub enabled: bool,
273}
274
275#[derive(Clone, Debug, Deserialize)]
277#[serde(default, deny_unknown_fields)]
278pub struct NamespacesConfig {
279 pub default: String,
281}
282
283#[derive(Clone, Debug, Deserialize)]
285#[serde(default, deny_unknown_fields)]
286pub struct ListenConfig {
287 pub grpc: SocketAddr,
289 pub http: SocketAddr,
291}
292
293#[derive(Clone, Debug, Deserialize)]
295#[serde(deny_unknown_fields)]
296pub struct TlsConfig {
297 pub certificate_chain_path: PathBuf,
299 pub private_key_path: PathBuf,
301}
302
303#[derive(Clone, Debug, Deserialize)]
305#[serde(default, deny_unknown_fields)]
306pub struct DashboardConfig {
307 pub source: DashboardAssetSource,
309}
310
311#[derive(Clone, Debug, Deserialize)]
313pub enum DashboardAssetSource {
314 FileSystem {
316 asset_path: PathBuf,
318 },
319 Embedded,
321}
322
323#[derive(Clone, Debug, Deserialize)]
325#[serde(default, deny_unknown_fields)]
326pub struct NamespaceConfig {
327 pub mode: NamespaceMode,
329}
330
331#[derive(Clone, Debug, Deserialize)]
333pub enum NamespaceMode {
334 SharedEngine,
336 SingleTenant {
338 namespace: String,
340 },
341}
342
343#[derive(Clone, Debug, Deserialize)]
345#[serde(default, deny_unknown_fields)]
346pub struct WorkerConfig {
347 #[serde(with = "duration_millis")]
349 pub heartbeat_window: Duration,
350}
351
352#[derive(Clone, Debug, Deserialize)]
354#[serde(default, deny_unknown_fields)]
355pub struct WebSocketConfig {
356 pub outbound_buffer_bound: usize,
358 pub event_broadcast_capacity: Option<usize>,
364}
365
366pub(crate) const EVENT_BROADCAST_CAPACITY_REQUIRED: &str = "websocket.event_broadcast_capacity is required and has no default: the server always mounts /events/stream, so live event streaming capacity must be configured explicitly; set websocket.event_broadcast_capacity (or AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY) to a positive integer sized for global event volume across all namespaces";
368
369#[derive(Clone, Debug, Default, Deserialize)]
376#[serde(default, deny_unknown_fields)]
377pub struct DeployConfig {
378 pub enabled: bool,
380 pub max_archive_bytes: Option<u64>,
384 pub max_inflated_bytes: Option<u64>,
390}
391
392pub(crate) const DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED: &str = "deploy.max_archive_bytes is required and has no default when deploy.enabled is true: the archive upload ceiling must be an explicit operator decision sized for the deployment's packages; set deploy.max_archive_bytes (or AION_DEPLOY_MAX_ARCHIVE_BYTES) to a positive number of bytes";
394
395pub(crate) const DEPLOY_MAX_INFLATED_BYTES_REQUIRED: &str = "deploy.max_inflated_bytes is required and has no default when deploy.enabled is true: the decompressed-contents ceiling for uploaded archives must be an explicit operator decision (a compressed upload under deploy.max_archive_bytes can inflate ~1000:1); set deploy.max_inflated_bytes (or AION_DEPLOY_MAX_INFLATED_BYTES) to a positive number of bytes no smaller than deploy.max_archive_bytes";
397
398pub(crate) const QUERY_TIMEOUT_REQUIRED: &str = "runtime.query_timeout_ms is required and has no default: the server always mounts /workflows/query, so the workflow query reply deadline must be configured explicitly; set runtime.query_timeout_ms (or AION_RUNTIME_QUERY_TIMEOUT_MS) to a positive number of milliseconds";
400
401#[derive(Clone, Debug, Default, Deserialize)]
411#[serde(default, deny_unknown_fields)]
412pub struct DevConfig {
413 pub enabled: bool,
415}
416
417#[derive(Clone, Debug, Default, Deserialize)]
433#[serde(default, deny_unknown_fields)]
434pub struct OutboxConfig {
435 pub enabled: bool,
438 pub poll_interval_ms: Option<u64>,
442 pub batch_size: Option<u32>,
445 pub max_attempts: Option<u32>,
448 pub backoff_base_ms: Option<u64>,
453 pub backoff_multiplier: Option<u32>,
457 pub backoff_max_ms: Option<u64>,
461 pub reconcile_interval_ms: Option<u64>,
465 pub reconcile_stale_after_ms: Option<u64>,
469 pub transport: OutboxTransport,
476 pub liminal_listen_address: Option<String>,
492}
493
494#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Deserialize)]
500#[serde(rename_all = "lowercase")]
501pub enum OutboxTransport {
502 #[default]
504 Grpc,
505 Liminal,
507}
508
509pub(crate) const OUTBOX_POLL_INTERVAL_REQUIRED: &str = "outbox.poll_interval_ms is required and has no default when outbox.enabled is true: the dispatcher claim cadence must be an explicit operator decision sized for fan-out volume and latency; set outbox.poll_interval_ms (or AION_OUTBOX_POLL_INTERVAL_MS) to a positive number of milliseconds";
511
512pub(crate) const OUTBOX_BATCH_SIZE_REQUIRED: &str = "outbox.batch_size is required and has no default when outbox.enabled is true: the per-sweep claim ceiling must be an explicit operator decision; set outbox.batch_size (or AION_OUTBOX_BATCH_SIZE) to a positive integer";
514
515pub(crate) const OUTBOX_MAX_ATTEMPTS_REQUIRED: &str = "outbox.max_attempts is required and has no default when outbox.enabled is true: the dispatch retry budget before dead-lettering must be an explicit operator decision; set outbox.max_attempts (or AION_OUTBOX_MAX_ATTEMPTS) to a positive integer";
517
518pub(crate) const OUTBOX_BACKOFF_BASE_REQUIRED: &str = "outbox.backoff_base_ms is required and has no default when outbox.enabled is true: the first-retry backoff must be an explicit operator decision; set outbox.backoff_base_ms (or AION_OUTBOX_BACKOFF_BASE_MS) to a positive number of milliseconds";
520
521pub(crate) const OUTBOX_BACKOFF_MULTIPLIER_REQUIRED: &str = "outbox.backoff_multiplier is required and has no default when outbox.enabled is true: the geometric backoff growth factor must be an explicit operator decision and must be at least one so backoff never shrinks; set outbox.backoff_multiplier (or AION_OUTBOX_BACKOFF_MULTIPLIER) to a positive integer";
523
524pub(crate) const OUTBOX_BACKOFF_MAX_REQUIRED: &str = "outbox.backoff_max_ms is required and has no default when outbox.enabled is true and must be at least outbox.backoff_base_ms: the per-retry backoff ceiling must be an explicit operator decision; set outbox.backoff_max_ms (or AION_OUTBOX_BACKOFF_MAX_MS) to a positive number of milliseconds no smaller than outbox.backoff_base_ms";
526
527pub(crate) const OUTBOX_RECONCILE_INTERVAL_REQUIRED: &str = "outbox.reconcile_interval_ms is required and has no default when live outbox reconciliation is enabled: set both outbox.reconcile_interval_ms and outbox.reconcile_stale_after_ms (or AION_OUTBOX_RECONCILE_INTERVAL_MS / AION_OUTBOX_RECONCILE_STALE_AFTER_MS) to positive millisecond values, or omit both to leave reconciliation disabled";
529
530pub(crate) const OUTBOX_RECONCILE_STALE_AFTER_REQUIRED: &str = "outbox.reconcile_stale_after_ms is required and has no default when live outbox reconciliation is enabled: set both outbox.reconcile_interval_ms and outbox.reconcile_stale_after_ms (or AION_OUTBOX_RECONCILE_INTERVAL_MS / AION_OUTBOX_RECONCILE_STALE_AFTER_MS) to positive millisecond values, or omit both to leave reconciliation disabled";
532
533#[derive(Clone, Debug, Default, Deserialize)]
542#[serde(default, deny_unknown_fields)]
543pub struct AuthoringConfig {
544 pub gleam_path: Option<PathBuf>,
549 pub project_root: Option<PathBuf>,
555}
556
557pub(crate) const AUTHORING_GLEAM_PATH_EMPTY: &str = "authoring.gleam_path must not be empty when set: it names the external gleam binary the authoring loop spawns; set authoring.gleam_path (or AION_AUTHORING_GLEAM_PATH) to the path of a runnable gleam binary, or remove it to leave the authoring surface dark";
559
560pub(crate) const AUTHORING_PROJECT_ROOT_REQUIRED: &str = "authoring.project_root is required and has no default when authoring.gleam_path is set: submitted Gleam source is written into and packaged from a built project, so the operator must provision and name the project root (a directory with gleam.toml, the aion_flow dependency, workflow.toml, and schemas/); set authoring.project_root (or AION_AUTHORING_PROJECT_ROOT)";
563
564#[derive(Clone, Debug)]
566pub struct RuntimeConfig {
567 pub listen: ListenConfig,
569 pub tls: Option<TlsConfig>,
571 pub auth: AuthConfig,
573 pub dashboard: DashboardConfig,
575 pub namespace: NamespaceConfig,
577 pub worker: WorkerConfig,
579 pub websocket: WebSocketConfig,
581 pub workflow_packages: Vec<PathBuf>,
583 pub deploy: DeployConfig,
585 pub authoring: AuthoringConfig,
587 pub dev: DevConfig,
589 pub outbox: OutboxConfig,
591 pub scheduler_threads: usize,
593 pub query_timeout: Option<Duration>,
598 pub default_namespace: String,
600 pub drain_timeout: Duration,
602 pub metrics: MetricsConfig,
604 pub owned_shards: Vec<usize>,
609 pub cors_allowed_origins: Vec<String>,
614}
615
616impl ServerConfig {
617 pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
624 let mut config = file::load(cli.config_path.as_deref())?.unwrap_or_default();
625 env::overlay(&mut config)?;
626 config.apply_cli_overrides(cli);
627 config.load_discovered_workflow_packages(cli, Path::new("."))?;
628 config.validate()?;
629 Ok(config)
630 }
631
632 fn load_discovered_workflow_packages(
633 &mut self,
634 cli: &CliOverrides,
635 directory: &Path,
636 ) -> Result<(), ServerError> {
637 let discovered_packages = discover_workflow_packages(directory)?;
638 merge_workflow_packages(
639 &mut self.workflow_packages,
640 discovered_packages,
641 &cli.workflow_packages,
642 );
643 Ok(())
644 }
645
646 pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
652 let config: Self = toml::from_slice(bytes).map_err(|source| ServerError::Config {
653 message: format!("invalid server config: {source}"),
654 })?;
655 config.validate()?;
656 Ok(config)
657 }
658
659 pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
665 file::load_required(&path.into())
666 }
667
668 #[must_use]
670 pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
671 let runtime = RuntimeConfig {
672 listen: ListenConfig {
673 grpc: self.server.grpc_address,
674 http: self.server.listen_address,
675 },
676 tls: self.tls,
677 auth: self.auth,
678 dashboard: self.dashboard,
679 namespace: self.namespace,
680 worker: self.worker,
681 websocket: self.websocket,
682 workflow_packages: self.workflow_packages,
683 deploy: self.deploy,
684 authoring: self.authoring,
685 dev: self.dev,
686 outbox: self.outbox,
687 scheduler_threads: self.runtime.scheduler_threads,
688 query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
689 default_namespace: self.namespaces.default,
690 drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
691 metrics: self.metrics,
692 owned_shards: self.store.owned_shards.clone(),
693 cors_allowed_origins: self.server.cors_allowed_origins.clone(),
694 };
695 (self.store, runtime)
696 }
697
698 fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
699 if let Some(address) = cli.listen_address {
700 self.server.listen_address = address;
701 }
702 if let Some(url) = &cli.store_url {
703 self.store.url = Some(url.clone());
704 if self.store.backend == StoreBackend::Memory {
705 self.store.backend = StoreBackend::LibSql;
706 }
707 }
708 if let Some(threads) = cli.scheduler_threads {
709 self.runtime.scheduler_threads = threads;
710 }
711 if let Some(timeout) = cli.drain_timeout_seconds {
712 self.drain.timeout_seconds = timeout;
713 }
714 if let Some(gleam_path) = &cli.gleam_path {
715 self.authoring.gleam_path = Some(gleam_path.clone());
716 }
717 if let Some(project_root) = &cli.authoring_project_root {
718 self.authoring.project_root = Some(project_root.clone());
719 }
720 }
721
722 fn validate(&self) -> Result<(), ServerError> {
723 if self.server.listen_address.port() == 0 {
724 return config_error("server.listen_address must use an explicit non-zero port");
725 }
726 if self.server.grpc_address.port() == 0 {
727 return config_error("server.grpc_address must use an explicit non-zero port");
728 }
729 validate_cors_origins(&self.server.cors_allowed_origins)?;
730 if self.runtime.scheduler_threads == 0 {
731 return config_error("runtime.scheduler_threads must be greater than zero");
732 }
733 if self.drain.timeout_seconds == 0 {
734 return config_error("drain.timeout_seconds must be greater than zero");
735 }
736 if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
737 return config_error("auth.jwks_url must not be empty when auth.enabled is true");
738 }
739 if self.auth.jwks_refresh_seconds == 0 {
740 return config_error("auth.jwks_refresh_seconds must be greater than zero");
741 }
742 if self.namespaces.default.is_empty() {
743 return config_error("namespaces.default must not be empty");
744 }
745 if matches!(self.store.backend, StoreBackend::LibSql)
746 && self.store.url.as_deref().is_none_or(str::is_empty)
747 {
748 return config_error("store.url must not be empty when store.backend is libsql");
749 }
750 if let Some(url) = &self.store.url {
751 if url.is_empty() {
752 return config_error("store.url must not be empty");
753 }
754 }
755 if matches!(self.store.backend, StoreBackend::Haematite) {
756 if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
757 return config_error(
758 "store.data_dir must not be empty when store.backend is haematite",
759 );
760 }
761 if self.store.shard_count == 0 {
762 return config_error("store.shard_count must be greater than zero");
763 }
764 if let Some(cluster) = &self.store.cluster {
765 validate_cluster(cluster)?;
766 }
767 } else if self.store.cluster.is_some() {
768 return config_error("store.cluster is only valid when store.backend is haematite");
769 }
770 if let DashboardAssetSource::FileSystem { asset_path } = &self.dashboard.source {
771 if asset_path.as_os_str().is_empty() {
772 return config_error("dashboard.source.FileSystem.asset_path must not be empty");
773 }
774 }
775 if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
776 if namespace.is_empty() {
777 return config_error("namespace.mode.SingleTenant.namespace must not be empty");
778 }
779 }
780 if self.worker.heartbeat_window.is_zero() {
781 return config_error("worker.heartbeat_window must be greater than zero");
782 }
783 if self.websocket.outbound_buffer_bound == 0 {
784 return config_error("websocket.outbound_buffer_bound must be greater than zero");
785 }
786 match self.websocket.event_broadcast_capacity {
787 None | Some(0) => return config_error(EVENT_BROADCAST_CAPACITY_REQUIRED),
788 Some(_) => {}
789 }
790 match self.runtime.query_timeout_ms {
791 None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
792 Some(_) => {}
793 }
794 if self.deploy.enabled {
795 let max_archive_bytes = match self.deploy.max_archive_bytes {
796 None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
797 Some(value) => value,
798 };
799 let max_inflated_bytes = match self.deploy.max_inflated_bytes {
800 None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
801 Some(value) => value,
802 };
803 ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
806 ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
807 if max_inflated_bytes < max_archive_bytes {
808 return config_error(format!(
809 "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"
810 ));
811 }
812 }
813 if let Some(gleam_path) = &self.authoring.gleam_path {
814 if gleam_path.as_os_str().is_empty() {
817 return config_error(AUTHORING_GLEAM_PATH_EMPTY);
818 }
819 match &self.authoring.project_root {
822 Some(root) if !root.as_os_str().is_empty() => {}
823 _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
824 }
825 }
826 self.validate_outbox()?;
827 Ok(())
828 }
829
830 fn validate_outbox(&self) -> Result<(), ServerError> {
838 if !self.outbox.enabled {
839 return Ok(());
840 }
841 match self.outbox.poll_interval_ms {
842 None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
843 Some(_) => {}
844 }
845 match self.outbox.batch_size {
846 None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
847 Some(_) => {}
848 }
849 match self.outbox.max_attempts {
850 None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
851 Some(_) => {}
852 }
853 let backoff_base_ms = match self.outbox.backoff_base_ms {
854 None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
855 Some(value) => value,
856 };
857 match self.outbox.backoff_multiplier {
858 None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
859 Some(_) => {}
860 }
861 match self.outbox.backoff_max_ms {
862 Some(max) if max >= backoff_base_ms => {}
863 _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
864 }
865 match (
866 self.outbox.reconcile_interval_ms,
867 self.outbox.reconcile_stale_after_ms,
868 ) {
869 (None, None) => {}
870 (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
871 (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
872 (Some(_), Some(_)) => {}
873 }
874 Ok(())
875 }
876}
877
878fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
882 if cluster.node_id.is_empty() {
883 return config_error("store.cluster.node_id must not be empty");
884 }
885 if cluster.members.iter().any(String::is_empty) {
886 return config_error("store.cluster.members entries must not be empty");
887 }
888 if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
889 return config_error("store.cluster.peers entries must name a non-empty node");
890 }
891 if matches!(cluster.failover_poll_interval_ms, Some(0)) {
892 return config_error(
893 "store.cluster.failover_poll_interval_ms must be greater than zero when set",
894 );
895 }
896 if matches!(cluster.failover_confirmations, Some(0)) {
897 return config_error("store.cluster.failover_confirmations must be at least one when set");
898 }
899 Ok(())
900}
901
902pub(crate) const CORS_ALLOWED_ORIGIN_INVALID: &str = "server.cors_allowed_origins entries must each be a valid HTTP origin (scheme://host[:port], e.g. http://localhost:5173) with no path or trailing slash";
905
906fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
908 for origin in origins {
909 validate_cors_origin(origin)?;
910 }
911 Ok(())
912}
913
914fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
919 if origin.is_empty() {
920 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
921 }
922 let scheme_split = origin.split_once("://");
926 let Some((scheme, authority)) = scheme_split else {
927 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
928 };
929 if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
930 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
931 }
932 if origin.parse::<axum::http::HeaderValue>().is_err() {
934 return config_error(CORS_ALLOWED_ORIGIN_INVALID);
935 }
936 Ok(())
937}
938
939fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
941 if usize::try_from(value).is_err() {
942 return config_error(format!(
943 "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
944 usize::MAX
945 ));
946 }
947 Ok(())
948}
949
950impl Default for ServerSection {
951 fn default() -> Self {
952 Self {
953 listen_address: DEFAULT_HTTP_ADDRESS,
954 grpc_address: DEFAULT_GRPC_ADDRESS,
955 cors_allowed_origins: Vec::new(),
956 }
957 }
958}
959
960impl Default for StoreConfig {
961 fn default() -> Self {
962 Self {
963 backend: StoreBackend::Memory,
964 url: None,
965 owned_shards: Vec::new(),
966 data_dir: None,
967 shard_count: 1,
968 cluster: None,
969 }
970 }
971}
972
973impl Default for RuntimeSection {
974 fn default() -> Self {
975 Self {
976 scheduler_threads: 1,
977 query_timeout_ms: None,
980 }
981 }
982}
983
984impl Default for DrainConfig {
985 fn default() -> Self {
986 Self {
987 timeout_seconds: 30,
988 }
989 }
990}
991
992impl Default for AuthConfig {
993 fn default() -> Self {
994 Self {
995 enabled: false,
996 jwks_url: None,
997 jwks_refresh_seconds: 300,
998 }
999 }
1000}
1001
1002impl Default for MetricsConfig {
1003 fn default() -> Self {
1004 Self { enabled: true }
1005 }
1006}
1007
1008impl Default for NamespacesConfig {
1009 fn default() -> Self {
1010 Self {
1011 default: "default".to_owned(),
1012 }
1013 }
1014}
1015
1016impl Default for ListenConfig {
1017 fn default() -> Self {
1018 Self {
1019 grpc: DEFAULT_GRPC_ADDRESS,
1020 http: DEFAULT_HTTP_ADDRESS,
1021 }
1022 }
1023}
1024
1025impl Default for DashboardConfig {
1026 fn default() -> Self {
1027 Self {
1028 source: DashboardAssetSource::Embedded,
1029 }
1030 }
1031}
1032
1033impl Default for NamespaceConfig {
1034 fn default() -> Self {
1035 Self {
1036 mode: NamespaceMode::SharedEngine,
1037 }
1038 }
1039}
1040
1041impl Default for WorkerConfig {
1042 fn default() -> Self {
1043 Self {
1044 heartbeat_window: Duration::from_secs(30),
1045 }
1046 }
1047}
1048
1049impl Default for WebSocketConfig {
1050 fn default() -> Self {
1051 Self {
1052 outbound_buffer_bound: 32,
1053 event_broadcast_capacity: None,
1056 }
1057 }
1058}
1059
1060pub(crate) fn config_error<T>(message: impl Into<String>) -> Result<T, ServerError> {
1061 Err(ServerError::Config {
1062 message: message.into(),
1063 })
1064}
1065
1066fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
1067 let mut packages = Vec::new();
1068 let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
1069 message: format!(
1070 "failed to scan workflow packages in `{}`: {source}",
1071 directory.display()
1072 ),
1073 })?;
1074
1075 for entry in entries {
1076 let entry = entry.map_err(|source| ServerError::Config {
1077 message: format!(
1078 "failed to read workflow package entry in `{}`: {source}",
1079 directory.display()
1080 ),
1081 })?;
1082 let path = entry.path();
1083 let has_aion_extension = path
1084 .extension()
1085 .is_some_and(|extension| extension == "aion");
1086 if path.is_file() && has_aion_extension {
1087 packages.push(path);
1088 }
1089 }
1090
1091 packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
1092 Ok(packages)
1093}
1094
1095fn merge_workflow_packages(
1096 workflow_packages: &mut Vec<PathBuf>,
1097 discovered_packages: Vec<PathBuf>,
1098 cli_packages: &[PathBuf],
1099) {
1100 let mut seen: HashSet<PathBuf> = workflow_packages
1101 .iter()
1102 .map(|package| deduplicated_package_key(package))
1103 .collect();
1104 for package in discovered_packages
1105 .into_iter()
1106 .chain(cli_packages.iter().cloned())
1107 {
1108 if seen.insert(deduplicated_package_key(&package)) {
1109 workflow_packages.push(package);
1110 }
1111 }
1112}
1113
1114fn deduplicated_package_key(path: &Path) -> PathBuf {
1115 path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
1116}
1117
1118mod duration_millis {
1119 use std::time::Duration;
1120
1121 use serde::{Deserialize, Deserializer};
1122
1123 pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<Duration, D::Error>
1124 where
1125 D: Deserializer<'de>,
1126 {
1127 let millis = u64::deserialize(deserializer)?;
1128 Ok(Duration::from_millis(millis))
1129 }
1130}
1131
1132#[cfg(test)]
1133mod tests {
1134 use super::{
1135 CliOverrides, ServerConfig, StoreBackend, discover_workflow_packages,
1136 merge_workflow_packages,
1137 };
1138
1139 #[test]
1140 fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
1141 let config = ServerConfig::from_slice(
1142 br#"
1143 [server]
1144 listen_address = "127.0.0.1:18080"
1145 grpc_address = "127.0.0.1:15051"
1146
1147 [store]
1148 backend = "libsql"
1149 url = "aion.db"
1150
1151 [runtime]
1152 scheduler_threads = 2
1153 query_timeout_ms = 10000
1154
1155 [drain]
1156 timeout_seconds = 45
1157
1158 [auth]
1159 enabled = true
1160 jwks_url = "https://issuer.example.com/.well-known/jwks.json"
1161 jwks_refresh_seconds = 60
1162
1163 [metrics]
1164 enabled = true
1165
1166 [namespaces]
1167 default = "production"
1168
1169 [websocket]
1170 outbound_buffer_bound = 16
1171 event_broadcast_capacity = 1024
1172 "#,
1173 )?;
1174
1175 assert_eq!(config.store.backend, StoreBackend::LibSql);
1176 assert_eq!(config.store.url.as_deref(), Some("aion.db"));
1177 assert_eq!(config.runtime.scheduler_threads, 2);
1178 assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
1179 assert_eq!(config.namespaces.default, "production");
1180 assert_eq!(config.websocket.outbound_buffer_bound, 16);
1181 assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
1182 Ok(())
1183 }
1184
1185 #[test]
1186 fn missing_event_broadcast_capacity_fails_startup_validation_naming_the_key() {
1187 let result = ServerConfig::default().validate();
1191
1192 let message = result
1193 .err()
1194 .map_or_else(String::new, |error| error.to_string());
1195 assert!(
1196 message.contains("websocket.event_broadcast_capacity"),
1197 "validation message must name the missing key: {message}"
1198 );
1199 assert!(
1200 message.contains("AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY"),
1201 "validation message must name the environment override: {message}"
1202 );
1203 }
1204
1205 #[test]
1206 fn zero_event_broadcast_capacity_fails_startup_validation() {
1207 let result = ServerConfig::from_slice(
1208 br"
1209 [websocket]
1210 event_broadcast_capacity = 0
1211 ",
1212 );
1213
1214 let message = result
1215 .err()
1216 .map_or_else(String::new, |error| error.to_string());
1217 assert!(
1218 message.contains("websocket.event_broadcast_capacity"),
1219 "validation message must name the zero-valued key: {message}"
1220 );
1221 }
1222
1223 #[test]
1224 fn missing_query_timeout_fails_startup_validation_naming_the_key() {
1225 let result = ServerConfig::from_slice(
1229 br"
1230 [runtime]
1231 scheduler_threads = 1
1232
1233 [websocket]
1234 event_broadcast_capacity = 64
1235 ",
1236 );
1237
1238 let message = result
1239 .err()
1240 .map_or_else(String::new, |error| error.to_string());
1241 assert!(
1242 message.contains("runtime.query_timeout_ms"),
1243 "validation message must name the missing key: {message}"
1244 );
1245 assert!(
1246 message.contains("AION_RUNTIME_QUERY_TIMEOUT_MS"),
1247 "validation message must name the environment override: {message}"
1248 );
1249 }
1250
1251 #[test]
1252 fn zero_query_timeout_fails_startup_validation() {
1253 let result = ServerConfig::from_slice(
1254 br"
1255 [runtime]
1256 query_timeout_ms = 0
1257
1258 [websocket]
1259 event_broadcast_capacity = 64
1260 ",
1261 );
1262
1263 let message = result
1264 .err()
1265 .map_or_else(String::new, |error| error.to_string());
1266 assert!(
1267 message.contains("runtime.query_timeout_ms"),
1268 "validation message must name the zero-valued key: {message}"
1269 );
1270 }
1271
1272 #[test]
1277 fn deploy_enabled_without_max_archive_bytes_fails_naming_key_and_env() {
1278 let result = ServerConfig::from_slice(
1279 br"
1280 [runtime]
1281 query_timeout_ms = 10000
1282
1283 [websocket]
1284 event_broadcast_capacity = 64
1285
1286 [deploy]
1287 enabled = true
1288 ",
1289 );
1290
1291 let message = result
1292 .err()
1293 .map_or_else(String::new, |error| error.to_string());
1294 assert!(
1295 message.contains("deploy.max_archive_bytes"),
1296 "validation message must name the missing key: {message}"
1297 );
1298 assert!(
1299 message.contains("AION_DEPLOY_MAX_ARCHIVE_BYTES"),
1300 "validation message must name the environment override: {message}"
1301 );
1302 }
1303
1304 #[test]
1305 fn deploy_zero_max_archive_bytes_fails_startup_validation() {
1306 let result = ServerConfig::from_slice(
1307 br"
1308 [runtime]
1309 query_timeout_ms = 10000
1310
1311 [websocket]
1312 event_broadcast_capacity = 64
1313
1314 [deploy]
1315 enabled = true
1316 max_archive_bytes = 0
1317 ",
1318 );
1319
1320 let message = result
1321 .err()
1322 .map_or_else(String::new, |error| error.to_string());
1323 assert!(
1324 message.contains("deploy.max_archive_bytes"),
1325 "validation message must name the zero-valued key: {message}"
1326 );
1327 }
1328
1329 #[test]
1334 fn deploy_enabled_without_max_inflated_bytes_fails_naming_key_and_env() {
1335 let result = ServerConfig::from_slice(
1336 br"
1337 [runtime]
1338 query_timeout_ms = 10000
1339
1340 [websocket]
1341 event_broadcast_capacity = 64
1342
1343 [deploy]
1344 enabled = true
1345 max_archive_bytes = 16777216
1346 ",
1347 );
1348
1349 let message = result
1350 .err()
1351 .map_or_else(String::new, |error| error.to_string());
1352 assert!(
1353 message.contains("deploy.max_inflated_bytes"),
1354 "validation message must name the missing key: {message}"
1355 );
1356 assert!(
1357 message.contains("AION_DEPLOY_MAX_INFLATED_BYTES"),
1358 "validation message must name the environment override: {message}"
1359 );
1360 }
1361
1362 #[test]
1363 fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
1364 let result = ServerConfig::from_slice(
1365 br"
1366 [runtime]
1367 query_timeout_ms = 10000
1368
1369 [websocket]
1370 event_broadcast_capacity = 64
1371
1372 [deploy]
1373 enabled = true
1374 max_archive_bytes = 16777216
1375 max_inflated_bytes = 0
1376 ",
1377 );
1378
1379 let message = result
1380 .err()
1381 .map_or_else(String::new, |error| error.to_string());
1382 assert!(
1383 message.contains("deploy.max_inflated_bytes"),
1384 "validation message must name the zero-valued key: {message}"
1385 );
1386 }
1387
1388 #[test]
1391 fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
1392 let result = ServerConfig::from_slice(
1393 br"
1394 [runtime]
1395 query_timeout_ms = 10000
1396
1397 [websocket]
1398 event_broadcast_capacity = 64
1399
1400 [deploy]
1401 enabled = true
1402 max_archive_bytes = 16777216
1403 max_inflated_bytes = 16777215
1404 ",
1405 );
1406
1407 let message = result
1408 .err()
1409 .map_or_else(String::new, |error| error.to_string());
1410 assert!(
1411 message.contains("deploy.max_inflated_bytes")
1412 && message.contains("deploy.max_archive_bytes"),
1413 "validation message must name both ceilings: {message}"
1414 );
1415 }
1416
1417 #[test]
1420 fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
1421 let config = ServerConfig::from_slice(
1422 br"
1423 [runtime]
1424 query_timeout_ms = 10000
1425
1426 [websocket]
1427 event_broadcast_capacity = 64
1428 ",
1429 )?;
1430
1431 assert!(!config.deploy.enabled);
1432 assert_eq!(config.deploy.max_archive_bytes, None);
1433 assert_eq!(config.deploy.max_inflated_bytes, None);
1434 Ok(())
1435 }
1436
1437 #[test]
1438 fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
1439 let config = ServerConfig::from_slice(
1440 br"
1441 [runtime]
1442 query_timeout_ms = 10000
1443
1444 [websocket]
1445 event_broadcast_capacity = 64
1446
1447 [deploy]
1448 enabled = true
1449 max_archive_bytes = 16777216
1450 max_inflated_bytes = 67108864
1451 ",
1452 )?;
1453
1454 assert!(config.deploy.enabled);
1455 assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
1456 assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
1457 Ok(())
1458 }
1459
1460 #[test]
1464 fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
1465 let config = ServerConfig::from_slice(
1466 br"
1467 [runtime]
1468 query_timeout_ms = 10000
1469
1470 [websocket]
1471 event_broadcast_capacity = 64
1472 ",
1473 )?;
1474
1475 assert!(config.server.cors_allowed_origins.is_empty());
1476 let (_, runtime) = config.into_parts();
1477 assert!(runtime.cors_allowed_origins.is_empty());
1478 Ok(())
1479 }
1480
1481 #[test]
1484 fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
1485 let config = ServerConfig::from_slice(
1486 br#"
1487 [server]
1488 cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
1489
1490 [runtime]
1491 query_timeout_ms = 10000
1492
1493 [websocket]
1494 event_broadcast_capacity = 64
1495 "#,
1496 )?;
1497
1498 assert_eq!(
1499 config.server.cors_allowed_origins,
1500 vec![
1501 "http://localhost:5173".to_owned(),
1502 "http://127.0.0.1:5173".to_owned()
1503 ]
1504 );
1505 let (_, runtime) = config.into_parts();
1506 assert_eq!(
1507 runtime.cors_allowed_origins,
1508 vec![
1509 "http://localhost:5173".to_owned(),
1510 "http://127.0.0.1:5173".to_owned()
1511 ]
1512 );
1513 Ok(())
1514 }
1515
1516 #[test]
1520 fn cors_allowed_origins_reject_malformed() {
1521 for bad in ["", "localhost:5173", "http://localhost:5173/"] {
1522 let toml = format!(
1523 "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
1524 );
1525 let result = ServerConfig::from_slice(toml.as_bytes());
1526 let message = result
1527 .err()
1528 .map_or_else(String::new, |error| error.to_string());
1529 assert!(
1530 message.contains("cors_allowed_origins"),
1531 "malformed origin `{bad}` must be rejected naming the key: {message}"
1532 );
1533 }
1534 }
1535
1536 #[test]
1538 fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1539 let config = ServerConfig::from_slice(
1540 br"
1541 [runtime]
1542 query_timeout_ms = 10000
1543
1544 [websocket]
1545 event_broadcast_capacity = 64
1546 ",
1547 )?;
1548
1549 assert!(!config.dev.enabled);
1550 Ok(())
1551 }
1552
1553 #[test]
1556 fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
1557 let config = ServerConfig::from_slice(
1558 br"
1559 [runtime]
1560 query_timeout_ms = 10000
1561
1562 [websocket]
1563 event_broadcast_capacity = 64
1564
1565 [dev]
1566 enabled = true
1567 ",
1568 )?;
1569
1570 assert!(config.dev.enabled);
1571 Ok(())
1572 }
1573
1574 #[test]
1577 fn authoring_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1578 let config = ServerConfig::from_slice(
1579 br"
1580 [runtime]
1581 query_timeout_ms = 10000
1582
1583 [websocket]
1584 event_broadcast_capacity = 64
1585 ",
1586 )?;
1587
1588 assert_eq!(config.authoring.gleam_path, None);
1589 assert_eq!(config.authoring.project_root, None);
1590 Ok(())
1591 }
1592
1593 #[test]
1596 fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1597 let config = ServerConfig::from_slice(
1598 br#"
1599 [runtime]
1600 query_timeout_ms = 10000
1601
1602 [websocket]
1603 event_broadcast_capacity = 64
1604
1605 [authoring]
1606 gleam_path = "/usr/local/bin/gleam"
1607 project_root = "/srv/aion/authoring"
1608 "#,
1609 )?;
1610
1611 assert_eq!(
1612 config.authoring.gleam_path.as_deref(),
1613 Some(std::path::Path::new("/usr/local/bin/gleam"))
1614 );
1615 let (_, runtime) = config.into_parts();
1616 assert_eq!(
1617 runtime.authoring.gleam_path.as_deref(),
1618 Some(std::path::Path::new("/usr/local/bin/gleam"))
1619 );
1620 assert_eq!(
1621 runtime.authoring.project_root.as_deref(),
1622 Some(std::path::Path::new("/srv/aion/authoring"))
1623 );
1624 Ok(())
1625 }
1626
1627 #[test]
1631 fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
1632 let result = ServerConfig::from_slice(
1633 br#"
1634 [runtime]
1635 query_timeout_ms = 10000
1636
1637 [websocket]
1638 event_broadcast_capacity = 64
1639
1640 [authoring]
1641 gleam_path = "/usr/local/bin/gleam"
1642 "#,
1643 );
1644
1645 let message = result
1646 .err()
1647 .map_or_else(String::new, |error| error.to_string());
1648 assert!(
1649 message.contains("authoring.project_root"),
1650 "validation message must name the missing key: {message}"
1651 );
1652 assert!(
1653 message.contains("AION_AUTHORING_PROJECT_ROOT"),
1654 "validation message must name the environment override: {message}"
1655 );
1656 }
1657
1658 #[test]
1661 fn authoring_empty_gleam_path_fails_naming_key_and_env() {
1662 let result = ServerConfig::from_slice(
1663 br#"
1664 [runtime]
1665 query_timeout_ms = 10000
1666
1667 [websocket]
1668 event_broadcast_capacity = 64
1669
1670 [authoring]
1671 gleam_path = ""
1672 "#,
1673 );
1674
1675 let message = result
1676 .err()
1677 .map_or_else(String::new, |error| error.to_string());
1678 assert!(
1679 message.contains("authoring.gleam_path"),
1680 "validation message must name the empty key: {message}"
1681 );
1682 assert!(
1683 message.contains("AION_AUTHORING_GLEAM_PATH"),
1684 "validation message must name the environment override: {message}"
1685 );
1686 }
1687
1688 #[test]
1690 fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
1691 let mut config = ServerConfig::from_slice(
1692 br"
1693 [runtime]
1694 query_timeout_ms = 10000
1695
1696 [websocket]
1697 event_broadcast_capacity = 64
1698 ",
1699 )?;
1700 let cli = CliOverrides {
1701 gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
1702 authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
1703 ..CliOverrides::default()
1704 };
1705
1706 config.apply_cli_overrides(&cli);
1707 config.validate()?;
1708
1709 assert_eq!(
1710 config.authoring.gleam_path.as_deref(),
1711 Some(std::path::Path::new("/opt/gleam"))
1712 );
1713 assert_eq!(
1714 config.authoring.project_root.as_deref(),
1715 Some(std::path::Path::new("/opt/project"))
1716 );
1717 Ok(())
1718 }
1719
1720 #[test]
1721 fn invalid_values_name_problematic_field() {
1722 let result = ServerConfig::from_slice(
1723 br"
1724 [runtime]
1725 scheduler_threads = 0
1726 ",
1727 );
1728
1729 let message = result
1730 .err()
1731 .map_or_else(String::new, |error| error.to_string());
1732 assert!(message.contains("runtime.scheduler_threads"));
1733 }
1734
1735 #[test]
1736 fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
1737 let mut config = ServerConfig::from_slice(
1738 br#"
1739 [store]
1740 backend = "libsql"
1741 url = "file.db"
1742
1743 [runtime]
1744 query_timeout_ms = 10000
1745
1746 [websocket]
1747 event_broadcast_capacity = 64
1748 "#,
1749 )?;
1750 let cli = CliOverrides {
1751 store_url: Some("cli.db".to_owned()),
1752 scheduler_threads: Some(3),
1753 ..CliOverrides::default()
1754 };
1755
1756 config.apply_cli_overrides(&cli);
1757 config.validate()?;
1758
1759 assert_eq!(config.store.url.as_deref(), Some("cli.db"));
1760 assert_eq!(config.runtime.scheduler_threads, 3);
1761 Ok(())
1762 }
1763
1764 #[test]
1765 fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
1766 let mut config = ServerConfig::default();
1767
1768 assert_eq!(config.store.backend, StoreBackend::Memory);
1769 assert_eq!(config.store.url, None);
1770 assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
1771 assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
1772 assert_eq!(config.namespaces.default, "default");
1773 assert!(!config.auth.enabled);
1774 assert!(config.metrics.enabled);
1775 assert_eq!(config.websocket.event_broadcast_capacity, None);
1779 assert_eq!(config.runtime.query_timeout_ms, None);
1780 config.websocket.event_broadcast_capacity = Some(64);
1781 config.runtime.query_timeout_ms = Some(10_000);
1782 config.validate()?;
1783 Ok(())
1784 }
1785
1786 #[test]
1787 fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
1788 {
1789 let mut config = ServerConfig::default();
1790 config.websocket.event_broadcast_capacity = Some(64);
1791 config.runtime.query_timeout_ms = Some(10_000);
1792
1793 assert!(!config.outbox.enabled);
1797 assert_eq!(config.outbox.poll_interval_ms, None);
1798 assert_eq!(config.outbox.batch_size, None);
1799 assert_eq!(config.outbox.max_attempts, None);
1800 assert_eq!(config.outbox.backoff_base_ms, None);
1801 assert_eq!(config.outbox.backoff_multiplier, None);
1802 assert_eq!(config.outbox.backoff_max_ms, None);
1803 assert_eq!(config.outbox.reconcile_interval_ms, None);
1804 assert_eq!(config.outbox.reconcile_stale_after_ms, None);
1805 config.validate()?;
1806 Ok(())
1807 }
1808
1809 fn outbox_enabled_base() -> ServerConfig {
1810 let mut config = ServerConfig::default();
1811 config.websocket.event_broadcast_capacity = Some(64);
1812 config.runtime.query_timeout_ms = Some(10_000);
1813 config.outbox.enabled = true;
1814 config.outbox.poll_interval_ms = Some(250);
1815 config.outbox.batch_size = Some(64);
1816 config.outbox.max_attempts = Some(5);
1817 config.outbox.backoff_base_ms = Some(100);
1818 config.outbox.backoff_multiplier = Some(2);
1819 config.outbox.backoff_max_ms = Some(30_000);
1820 config.outbox.reconcile_interval_ms = Some(1_000);
1821 config.outbox.reconcile_stale_after_ms = Some(60_000);
1822 config
1823 }
1824
1825 #[test]
1826 fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
1827 outbox_enabled_base().validate()?;
1828 Ok(())
1829 }
1830
1831 #[test]
1832 fn outbox_enabled_without_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>>
1833 {
1834 let mut config = outbox_enabled_base();
1835 config.outbox.poll_interval_ms = None;
1836 let error = config
1837 .validate()
1838 .err()
1839 .ok_or("enabled outbox without poll interval must fail")?;
1840 assert!(
1841 error.to_string().contains("outbox.poll_interval_ms"),
1842 "error must name the missing key: {error}"
1843 );
1844 Ok(())
1845 }
1846
1847 #[test]
1848 fn outbox_enabled_without_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
1849 let mut config = outbox_enabled_base();
1850 config.outbox.max_attempts = None;
1851 let error = config
1852 .validate()
1853 .err()
1854 .ok_or("enabled outbox without max attempts must fail")?;
1855 assert!(
1856 error.to_string().contains("outbox.max_attempts"),
1857 "error must name the missing key: {error}"
1858 );
1859 Ok(())
1860 }
1861
1862 #[test]
1863 fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
1864 let mut config = outbox_enabled_base();
1865 config.outbox.backoff_base_ms = Some(1_000);
1866 config.outbox.backoff_max_ms = Some(500);
1867 let error = config
1868 .validate()
1869 .err()
1870 .ok_or("backoff_max below backoff_base must fail")?;
1871 assert!(
1872 error.to_string().contains("outbox.backoff_max_ms"),
1873 "error must name the offending key: {error}"
1874 );
1875 Ok(())
1876 }
1877
1878 #[test]
1879 fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
1880 let mut config = outbox_enabled_base();
1881 config.outbox.reconcile_interval_ms = None;
1882 config.outbox.reconcile_stale_after_ms = None;
1883 config.validate()?;
1884 Ok(())
1885 }
1886
1887 #[test]
1888 fn outbox_reconciliation_requires_interval_when_partially_enabled()
1889 -> Result<(), Box<dyn std::error::Error>> {
1890 let mut config = outbox_enabled_base();
1891 config.outbox.reconcile_interval_ms = None;
1892 let error = config
1893 .validate()
1894 .err()
1895 .ok_or("reconciliation without interval must fail")?;
1896 assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
1897 Ok(())
1898 }
1899
1900 #[test]
1901 fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
1902 -> Result<(), Box<dyn std::error::Error>> {
1903 let mut config = outbox_enabled_base();
1904 config.outbox.reconcile_stale_after_ms = None;
1905 let error = config
1906 .validate()
1907 .err()
1908 .ok_or("reconciliation without stale threshold must fail")?;
1909 assert!(
1910 error
1911 .to_string()
1912 .contains("outbox.reconcile_stale_after_ms")
1913 );
1914 Ok(())
1915 }
1916
1917 #[test]
1918 fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
1919 let temp_dir = tempfile::tempdir()?;
1920 std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
1921 std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
1922 std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
1923 std::fs::create_dir(temp_dir.path().join("nested"))?;
1924 std::fs::write(
1925 temp_dir.path().join("nested").join("nested.aion"),
1926 b"package",
1927 )?;
1928
1929 let packages = discover_workflow_packages(temp_dir.path())?;
1930
1931 assert_eq!(
1932 packages,
1933 vec![
1934 temp_dir.path().join("alpha.aion"),
1935 temp_dir.path().join("zeta.aion"),
1936 ]
1937 );
1938 Ok(())
1939 }
1940
1941 #[test]
1942 fn workflow_package_merge_is_additive_and_deduplicated() {
1943 let mut packages = vec!["config.aion".into(), "shared.aion".into()];
1944 let discovered = vec!["auto.aion".into(), "shared.aion".into()];
1945 let cli = vec!["cli.aion".into(), "auto.aion".into()];
1946
1947 merge_workflow_packages(&mut packages, discovered, &cli);
1948
1949 assert_eq!(
1950 packages,
1951 vec![
1952 std::path::PathBuf::from("config.aion"),
1953 std::path::PathBuf::from("shared.aion"),
1954 std::path::PathBuf::from("auto.aion"),
1955 std::path::PathBuf::from("cli.aion"),
1956 ]
1957 );
1958 }
1959
1960 #[test]
1961 fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
1962 let temp_dir = tempfile::tempdir()?;
1963 let package = temp_dir.path().join("hello.aion");
1964 std::fs::write(&package, b"package")?;
1965 let mut packages = vec![package.clone()];
1966 let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
1967
1968 merge_workflow_packages(&mut packages, discovered, &[]);
1969
1970 assert_eq!(packages, vec![package]);
1971 Ok(())
1972 }
1973
1974 #[test]
1975 fn zero_config_cli_workflow_package_uses_in_memory_defaults()
1976 -> Result<(), Box<dyn std::error::Error>> {
1977 let temp_dir = tempfile::tempdir()?;
1978
1979 let cli = CliOverrides {
1980 workflow_packages: vec!["hello-world.aion".into()],
1981 ..CliOverrides::default()
1982 };
1983 let mut config = ServerConfig::default();
1984 config.websocket.event_broadcast_capacity = Some(64);
1989 config.runtime.query_timeout_ms = Some(10_000);
1990 config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
1991
1992 config.validate()?;
1993
1994 assert_eq!(config.store.backend, StoreBackend::Memory);
1995 assert_eq!(config.store.url, None);
1996 assert_eq!(
1997 config.workflow_packages,
1998 vec![std::path::PathBuf::from("hello-world.aion")]
1999 );
2000 Ok(())
2001 }
2002
2003 #[test]
2004 fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
2005 let mut config = ServerConfig::from_slice(
2006 br#"
2007 workflow_packages = ["config.aion"]
2008
2009 [runtime]
2010 query_timeout_ms = 10000
2011
2012 [websocket]
2013 event_broadcast_capacity = 64
2014 "#,
2015 )?;
2016 let cli = CliOverrides {
2017 workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
2018 ..CliOverrides::default()
2019 };
2020
2021 merge_workflow_packages(
2022 &mut config.workflow_packages,
2023 Vec::new(),
2024 &cli.workflow_packages,
2025 );
2026
2027 assert_eq!(
2028 config.workflow_packages,
2029 vec![
2030 std::path::PathBuf::from("config.aion"),
2031 std::path::PathBuf::from("cli-one.aion"),
2032 std::path::PathBuf::from("cli-two.aion"),
2033 ]
2034 );
2035 Ok(())
2036 }
2037}