aion_server/config/
load.rs

1//! [`ServerConfig`]: the complete merged server configuration, its load /
2//! merge / validate pipeline, and the workflow-package discovery helpers.
3//!
4//! This is the assembly point of the config surface: [`ServerConfig`] holds one
5//! field per `[section]` (defined in [`super::sections`]), and its impls carry
6//! the load/merge/CLI-override/default-fill/validate pipeline plus
7//! [`ServerConfig::into_parts`], which splits the durable store config from the
8//! non-secret [`RuntimeConfig`] runtime view. It is re-exported from the
9//! `config` module so every existing `crate::config::X` path resolves
10//! identically.
11
12use 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, DEFAULT_CLUSTER_BROADCAST_CAPACITY,
26    DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES, DEFAULT_DEPLOY_MAX_INFLATED_BYTES,
27    DEFAULT_EVENT_BROADCAST_CAPACITY, DEFAULT_OUTBOX_BACKOFF_BASE_MS,
28    DEFAULT_OUTBOX_BACKOFF_MAX_MS, DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE,
29    DEFAULT_OUTBOX_MAX_ATTEMPTS, DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS,
30    DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED, DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig,
31    DrainConfig, ListenConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
32    OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
33    OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
34    OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
35    OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
36    RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
37    WorkerConfig, config_error, env, file,
38};
39
40/// Complete merged server configuration.
41#[derive(Clone, Debug, Deserialize)]
42#[serde(default, deny_unknown_fields)]
43#[derive(Default)]
44pub struct ServerConfig {
45    /// Public listener and transport addresses.
46    pub server: ServerSection,
47    /// Event-store backend configuration.
48    pub store: StoreConfig,
49    /// Engine runtime settings.
50    pub runtime: RuntimeSection,
51    /// Shutdown drain settings.
52    pub drain: DrainConfig,
53    /// Authentication settings defined by the operations config surface.
54    pub auth: AuthConfig,
55    /// Metrics endpoint settings.
56    pub metrics: MetricsConfig,
57    /// Namespace defaults.
58    pub namespaces: NamespacesConfig,
59    /// Optional TLS material for transports that require it.
60    pub tls: Option<TlsConfig>,
61    /// Static ops-console asset bundle location.
62    #[serde(alias = "dashboard")]
63    pub ops_console: OpsConsoleConfig,
64    /// Namespace resolver construction mode retained for existing transports.
65    pub namespace: NamespaceConfig,
66    /// Remote-worker heartbeat policy.
67    pub worker: WorkerConfig,
68    /// WebSocket event streaming policy.
69    pub websocket: WebSocketConfig,
70    /// Workflow package archives loaded into the engine at startup.
71    pub workflow_packages: Vec<PathBuf>,
72    /// Operator deploy API settings.
73    pub deploy: DeployConfig,
74    /// Server-side Gleam authoring API settings.
75    pub authoring: AuthoringConfig,
76    /// Local dev-server surface settings.
77    pub dev: DevConfig,
78    /// Durable-outbox fan-out dispatcher settings.
79    pub outbox: OutboxConfig,
80    /// Agent-observability transcript retention bounds.
81    pub observability: ObservabilityConfig,
82}
83
84impl ServerConfig {
85    /// Load and merge config from defaults, optional TOML file, environment, and CLI overrides.
86    ///
87    /// # Errors
88    ///
89    /// Returns [`ServerError::Config`] when file discovery, parsing, environment parsing, CLI
90    /// values, or validation fail.
91    pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
92        let mut config = file::load(cli.config_path.as_deref())?.unwrap_or_default();
93        env::overlay(&mut config)?;
94        config.apply_cli_overrides(cli);
95        config.load_discovered_workflow_packages(cli, Path::new("."))?;
96        config.fill_operational_defaults();
97        config.validate()?;
98        Ok(config)
99    }
100
101    /// Fill operational tuning knobs that have a sane default when omitted, so a
102    /// minimal or empty config boots without forcing the operator to hand-author
103    /// values that are pure tuning. Uses `get_or_insert`, so an explicitly set
104    /// value (including a misconfigured `0`, which [`Self::validate`] still
105    /// rejects) is left untouched; only an absent (`None`) field is defaulted.
106    fn fill_operational_defaults(&mut self) {
107        self.runtime
108            .query_timeout_ms
109            .get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
110        self.websocket
111            .event_broadcast_capacity
112            .get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
113        self.websocket
114            .cluster_broadcast_capacity
115            .get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
116        self.fill_outbox_defaults();
117        self.fill_deploy_defaults();
118    }
119
120    /// Fill the durable-outbox tuning knobs with sane defaults when the
121    /// dispatcher is enabled but a knob was omitted, so turning the feature on
122    /// does not force hand-authoring pure tuning. Inert while `outbox.enabled`
123    /// is false (the knobs are never read behind the gate). The reconciliation
124    /// pair is intentionally NOT defaulted: when both are absent reconciliation
125    /// stays dark, so forcing a default would silently commission a sweep.
126    /// `get_or_insert` leaves any explicit value (including a misconfigured `0`,
127    /// which [`Self::validate_outbox`] still rejects) untouched.
128    fn fill_outbox_defaults(&mut self) {
129        if !self.outbox.enabled {
130            return;
131        }
132        self.outbox
133            .poll_interval_ms
134            .get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
135        self.outbox
136            .batch_size
137            .get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
138        self.outbox
139            .max_attempts
140            .get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
141        self.outbox
142            .backoff_base_ms
143            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
144        self.outbox
145            .backoff_multiplier
146            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
147        self.outbox
148            .backoff_max_ms
149            .get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
150    }
151
152    /// Fill the deploy decompression-bomb ceilings with conservative defaults
153    /// when the deploy surface is enabled but a ceiling was omitted, so turning
154    /// the feature on boots rather than refusing for want of a security knob.
155    /// Inert while `deploy.enabled` is false (the ceilings are never read with
156    /// the surface dark). `get_or_insert` leaves any explicit value (including a
157    /// misconfigured `0` or an inflate ceiling below the archive ceiling, both
158    /// still rejected by [`Self::validate`]) untouched.
159    fn fill_deploy_defaults(&mut self) {
160        if !self.deploy.enabled {
161            return;
162        }
163        self.deploy
164            .max_archive_bytes
165            .get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
166        self.deploy
167            .max_inflated_bytes
168            .get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
169    }
170
171    fn load_discovered_workflow_packages(
172        &mut self,
173        cli: &CliOverrides,
174        directory: &Path,
175    ) -> Result<(), ServerError> {
176        let discovered_packages = discover_workflow_packages(directory)?;
177        merge_workflow_packages(
178            &mut self.workflow_packages,
179            discovered_packages,
180            &cli.workflow_packages,
181        );
182        Ok(())
183    }
184
185    /// Parse server configuration from TOML bytes and validate it.
186    ///
187    /// # Errors
188    ///
189    /// Returns [`ServerError::Config`] when parsing fails or values are invalid.
190    pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
191        let mut config: Self = toml::from_slice(bytes).map_err(|source| ServerError::Config {
192            message: format!("invalid server config: {source}"),
193        })?;
194        config.fill_operational_defaults();
195        config.validate()?;
196        Ok(config)
197    }
198
199    /// Load server configuration from an explicit TOML file path.
200    ///
201    /// # Errors
202    ///
203    /// Returns [`ServerError::Config`] when the file is missing, unreadable, unparsable, or invalid.
204    pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
205        file::load_required(&path.into())
206    }
207
208    /// Split store configuration from non-secret runtime settings.
209    #[must_use]
210    pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
211        let runtime = RuntimeConfig {
212            listen: ListenConfig {
213                grpc: self.server.grpc_address,
214                http: self.server.listen_address,
215            },
216            tls: self.tls,
217            auth: self.auth,
218            ops_console: self.ops_console,
219            namespace: self.namespace,
220            worker: self.worker,
221            websocket: self.websocket,
222            workflow_packages: self.workflow_packages,
223            deploy: self.deploy,
224            authoring: self.authoring,
225            dev: self.dev,
226            outbox: self.outbox,
227            observability: self.observability,
228            scheduler_threads: self.runtime.scheduler_threads,
229            query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
230            default_namespace: self.namespaces.default,
231            auto_create: self.namespaces.auto_create,
232            max_in_flight_activities: self.namespaces.max_in_flight_activities,
233            drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
234            metrics: self.metrics,
235            owned_shards: self.store.owned_shards.clone(),
236            cors_allowed_origins: self.server.cors_allowed_origins.clone(),
237        };
238        (self.store, runtime)
239    }
240
241    fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
242        if let Some(address) = cli.listen_address {
243            self.server.listen_address = address;
244        }
245        if let Some(url) = &cli.store_url {
246            self.store.url = Some(url.clone());
247            // `--store-url` names an embedded libSQL database file, so it is an
248            // explicit libSQL selection: coerce the implicit durable defaults
249            // (memory, or the new haematite default) to libsql. An operator who
250            // explicitly set `backend = "libsql"` already lands here too. The
251            // haematite backend ignores `store.url`, so the only way `--store-url`
252            // is meaningful is as a libSQL choice.
253            if matches!(
254                self.store.backend,
255                StoreBackend::Memory | StoreBackend::Haematite
256            ) {
257                self.store.backend = StoreBackend::LibSql;
258            }
259        }
260        if let Some(threads) = cli.scheduler_threads {
261            self.runtime.scheduler_threads = threads;
262        }
263        if let Some(timeout) = cli.drain_timeout_seconds {
264            self.drain.timeout_seconds = timeout;
265        }
266        if let Some(gleam_path) = &cli.gleam_path {
267            self.authoring.gleam_path = Some(gleam_path.clone());
268        }
269        if let Some(project_root) = &cli.authoring_project_root {
270            self.authoring.project_root = Some(project_root.clone());
271        }
272    }
273
274    fn validate(&self) -> Result<(), ServerError> {
275        if self.server.listen_address.port() == 0 {
276            return config_error("server.listen_address must use an explicit non-zero port");
277        }
278        if self.server.grpc_address.port() == 0 {
279            return config_error("server.grpc_address must use an explicit non-zero port");
280        }
281        validate_cors_origins(&self.server.cors_allowed_origins)?;
282        if self.runtime.scheduler_threads == 0 {
283            return config_error("runtime.scheduler_threads must be greater than zero");
284        }
285        if self.drain.timeout_seconds == 0 {
286            return config_error("drain.timeout_seconds must be greater than zero");
287        }
288        if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
289            return config_error("auth.jwks_url must not be empty when auth.enabled is true");
290        }
291        if self.auth.jwks_refresh_seconds == 0 {
292            return config_error("auth.jwks_refresh_seconds must be greater than zero");
293        }
294        if self.namespaces.default.is_empty() {
295            return config_error("namespaces.default must not be empty");
296        }
297        if matches!(self.store.backend, StoreBackend::LibSql)
298            && self.store.url.as_deref().is_none_or(str::is_empty)
299        {
300            return config_error("store.url must not be empty when store.backend is libsql");
301        }
302        if let Some(url) = &self.store.url {
303            if url.is_empty() {
304                return config_error("store.url must not be empty");
305            }
306        }
307        if matches!(self.store.backend, StoreBackend::Haematite) {
308            if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
309                return config_error(
310                    "store.data_dir must not be empty when store.backend is haematite",
311                );
312            }
313            if self.store.shard_count == 0 {
314                return config_error("store.shard_count must be greater than zero");
315            }
316            if let Some(cluster) = &self.store.cluster {
317                validate_cluster(cluster)?;
318            }
319        } else if self.store.cluster.is_some() {
320            return config_error("store.cluster is only valid when store.backend is haematite");
321        }
322        if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
323            if asset_path.as_os_str().is_empty() {
324                return config_error("ops_console.source.FileSystem.asset_path must not be empty");
325            }
326        }
327        if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
328            if namespace.is_empty() {
329                return config_error("namespace.mode.SingleTenant.namespace must not be empty");
330            }
331        }
332        if self.worker.heartbeat_window.is_zero() {
333            return config_error("worker.heartbeat_window must be greater than zero");
334        }
335        self.websocket.validate()?;
336        self.observability.validate()?;
337        match self.runtime.query_timeout_ms {
338            None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
339            Some(_) => {}
340        }
341        if self.deploy.enabled {
342            let max_archive_bytes = match self.deploy.max_archive_bytes {
343                None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
344                Some(value) => value,
345            };
346            let max_inflated_bytes = match self.deploy.max_inflated_bytes {
347                None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
348                Some(value) => value,
349            };
350            // Both ceilings size in-memory buffers, so they must be
351            // addressable on this platform (32-bit targets).
352            ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
353            ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
354            if max_inflated_bytes < max_archive_bytes {
355                return config_error(format!(
356                    "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"
357                ));
358            }
359        }
360        if let Some(gleam_path) = &self.authoring.gleam_path {
361            // The authoring surface is commissioned by a non-empty gleam_path;
362            // an empty value is a misconfiguration, not "dark".
363            if gleam_path.as_os_str().is_empty() {
364                return config_error(AUTHORING_GLEAM_PATH_EMPTY);
365            }
366            // Commissioning the loop requires a project root with no default
367            // (a Gleam project cannot be invented; the operator provisions it).
368            match &self.authoring.project_root {
369                Some(root) if !root.as_os_str().is_empty() => {}
370                _ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
371            }
372        }
373        self.validate_outbox()?;
374        Ok(())
375    }
376
377    /// Validate the durable-outbox dispatcher knobs.
378    ///
379    /// All knobs are inert while `outbox.enabled` is false (the dispatcher is
380    /// never spawned), so they are only required — and only checked — once the
381    /// operator commissions the dispatcher. This mirrors the dark-by-default
382    /// `deploy` surface: the on/off gate carries no defaults, and every
383    /// operational value behind it is an explicit operator decision.
384    fn validate_outbox(&self) -> Result<(), ServerError> {
385        if !self.outbox.enabled {
386            return Ok(());
387        }
388        match self.outbox.poll_interval_ms {
389            None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
390            Some(_) => {}
391        }
392        match self.outbox.batch_size {
393            None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
394            Some(_) => {}
395        }
396        match self.outbox.max_attempts {
397            None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
398            Some(_) => {}
399        }
400        let backoff_base_ms = match self.outbox.backoff_base_ms {
401            None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
402            Some(value) => value,
403        };
404        match self.outbox.backoff_multiplier {
405            None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
406            Some(_) => {}
407        }
408        match self.outbox.backoff_max_ms {
409            Some(max) if max >= backoff_base_ms => {}
410            _ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
411        }
412        match (
413            self.outbox.reconcile_interval_ms,
414            self.outbox.reconcile_stale_after_ms,
415        ) {
416            (None, None) => {}
417            (None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
418            (_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
419            (Some(_), Some(_)) => {}
420        }
421        Ok(())
422    }
423}
424
425/// Validate a `[store.cluster]` section: a non-empty node id, and every member /
426/// peer name non-empty. A cluster of one (no peers, members empty or `[node_id]`)
427/// is valid.
428fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
429    if cluster.node_id.is_empty() {
430        return config_error("store.cluster.node_id must not be empty");
431    }
432    if cluster.members.iter().any(String::is_empty) {
433        return config_error("store.cluster.members entries must not be empty");
434    }
435    if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
436        return config_error("store.cluster.peers entries must name a non-empty node");
437    }
438    if matches!(cluster.failover_poll_interval_ms, Some(0)) {
439        return config_error(
440            "store.cluster.failover_poll_interval_ms must be greater than zero when set",
441        );
442    }
443    if matches!(cluster.failover_confirmations, Some(0)) {
444        return config_error("store.cluster.failover_confirmations must be at least one when set");
445    }
446    Ok(())
447}
448
449/// Validate every `[server] cors_allowed_origins` entry.
450fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
451    for origin in origins {
452        validate_cors_origin(origin)?;
453    }
454    Ok(())
455}
456
457/// Validate one `[server] cors_allowed_origins` entry: it must be a non-empty,
458/// parseable HTTP origin so the `CorsLayer` can match it against the browser's
459/// `Origin` header. A malformed origin can never match a real request, so it is
460/// a misconfiguration caught at startup rather than silently never matching.
461fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
462    if origin.is_empty() {
463        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
464    }
465    // An origin is scheme + host + optional port and carries no path: reject a
466    // trailing slash or any path segment, which would never equal a browser
467    // `Origin` header value.
468    let scheme_split = origin.split_once("://");
469    let Some((scheme, authority)) = scheme_split else {
470        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
471    };
472    if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
473        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
474    }
475    // It must parse as an HTTP header value (the form the CorsLayer compares).
476    if origin.parse::<axum::http::HeaderValue>().is_err() {
477        return config_error(CORS_ALLOWED_ORIGIN_INVALID);
478    }
479    Ok(())
480}
481
482/// Refuses byte-ceiling values that cannot index memory on this platform.
483fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
484    if usize::try_from(value).is_err() {
485        return config_error(format!(
486            "{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
487            usize::MAX
488        ));
489    }
490    Ok(())
491}
492
493fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
494    let mut packages = Vec::new();
495    let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
496        message: format!(
497            "failed to scan workflow packages in `{}`: {source}",
498            directory.display()
499        ),
500    })?;
501
502    for entry in entries {
503        let entry = entry.map_err(|source| ServerError::Config {
504            message: format!(
505                "failed to read workflow package entry in `{}`: {source}",
506                directory.display()
507            ),
508        })?;
509        let path = entry.path();
510        let has_aion_extension = path
511            .extension()
512            .is_some_and(|extension| extension == "aion");
513        if path.is_file() && has_aion_extension {
514            packages.push(path);
515        }
516    }
517
518    packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
519    Ok(packages)
520}
521
522fn merge_workflow_packages(
523    workflow_packages: &mut Vec<PathBuf>,
524    discovered_packages: Vec<PathBuf>,
525    cli_packages: &[PathBuf],
526) {
527    let mut seen: HashSet<PathBuf> = workflow_packages
528        .iter()
529        .map(|package| deduplicated_package_key(package))
530        .collect();
531    for package in discovered_packages
532        .into_iter()
533        .chain(cli_packages.iter().cloned())
534    {
535        if seen.insert(deduplicated_package_key(&package)) {
536            workflow_packages.push(package);
537        }
538    }
539}
540
541fn deduplicated_package_key(path: &Path) -> PathBuf {
542    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
543}
544
545#[cfg(test)]
546mod tests {
547    use crate::config::{AutoCreate, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, OpsConsoleAssetSource};
548
549    use super::{
550        CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
551        DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
552        DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
553        DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
554        DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
555        discover_workflow_packages, merge_workflow_packages,
556    };
557
558    #[test]
559    fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
560        let config = ServerConfig::from_slice(
561            br#"
562                [server]
563                listen_address = "127.0.0.1:18080"
564                grpc_address = "127.0.0.1:15051"
565
566                [store]
567                backend = "libsql"
568                url = "aion.db"
569
570                [runtime]
571                scheduler_threads = 2
572                query_timeout_ms = 10000
573
574                [drain]
575                timeout_seconds = 45
576
577                [auth]
578                enabled = true
579                jwks_url = "https://issuer.example.com/.well-known/jwks.json"
580                jwks_refresh_seconds = 60
581
582                [metrics]
583                enabled = true
584
585                [namespaces]
586                default = "production"
587
588                [websocket]
589                outbound_buffer_bound = 16
590                event_broadcast_capacity = 1024
591                cluster_broadcast_capacity = 1024
592            "#,
593        )?;
594
595        assert_eq!(config.store.backend, StoreBackend::LibSql);
596        assert_eq!(config.store.url.as_deref(), Some("aion.db"));
597        assert_eq!(config.runtime.scheduler_threads, 2);
598        assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
599        assert_eq!(config.namespaces.default, "production");
600        // `auto_create` is omitted above, so it resolves to the Open default.
601        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
602        // `max_in_flight_activities` is omitted above, so it resolves to the
603        // generous platform default.
604        assert_eq!(
605            config.namespaces.max_in_flight_activities,
606            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
607        );
608        assert_eq!(config.websocket.outbound_buffer_bound, 16);
609        assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
610        Ok(())
611    }
612
613    #[test]
614    fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
615        let config = ServerConfig::from_slice(
616            br#"
617                [namespaces]
618                default = "production"
619                auto_create = "closed"
620            "#,
621        )?;
622        assert_eq!(config.namespaces.default, "production");
623        assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
624        Ok(())
625    }
626
627    #[test]
628    fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
629        let config = ServerConfig::from_slice(
630            br#"
631                [namespaces]
632                auto_create = "open"
633            "#,
634        )?;
635        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
636        Ok(())
637    }
638
639    #[test]
640    fn namespaces_max_in_flight_activities_override_parses()
641    -> Result<(), Box<dyn std::error::Error>> {
642        let config = ServerConfig::from_slice(
643            br#"
644                [namespaces]
645                default = "production"
646                max_in_flight_activities = 32
647            "#,
648        )?;
649        assert_eq!(config.namespaces.max_in_flight_activities, 32);
650        // The override also propagates into the runtime view.
651        let (_store, runtime) = config.into_parts();
652        assert_eq!(runtime.max_in_flight_activities, 32);
653        Ok(())
654    }
655
656    #[test]
657    fn namespaces_max_in_flight_activities_defaults_when_omitted()
658    -> Result<(), Box<dyn std::error::Error>> {
659        // An old/minimal config that predates the field omits it entirely and
660        // resolves to the generous platform default (additive, not a migration).
661        let config = ServerConfig::from_slice(
662            br#"
663                [namespaces]
664                default = "production"
665            "#,
666        )?;
667        assert_eq!(
668            config.namespaces.max_in_flight_activities,
669            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
670        );
671        Ok(())
672    }
673
674    #[test]
675    fn namespaces_auto_create_rejects_unknown_variant() {
676        let result = ServerConfig::from_slice(
677            br#"
678                [namespaces]
679                auto_create = "sometimes"
680            "#,
681        );
682        assert!(
683            result.is_err(),
684            "an unknown auto_create variant must fail to parse"
685        );
686    }
687
688    #[test]
689    fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
690        // The server unconditionally mounts /events/stream, but the channel
691        // capacity is a tuning knob: omitting it must resolve to the default and
692        // boot, not fail startup.
693        let config = ServerConfig::from_slice(
694            br"
695                [runtime]
696                query_timeout_ms = 10000
697
698                [websocket]
699                cluster_broadcast_capacity = 64
700            ",
701        )?;
702        assert_eq!(
703            config.websocket.event_broadcast_capacity,
704            Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
705            "omitted event_broadcast_capacity must resolve to the default"
706        );
707        Ok(())
708    }
709
710    #[test]
711    fn zero_event_broadcast_capacity_fails_startup_validation() {
712        let result = ServerConfig::from_slice(
713            br"
714                [websocket]
715                event_broadcast_capacity = 0
716            ",
717        );
718
719        let message = result
720            .err()
721            .map_or_else(String::new, |error| error.to_string());
722        assert!(
723            message.contains("websocket.event_broadcast_capacity"),
724            "validation message must name the zero-valued key: {message}"
725        );
726    }
727
728    #[test]
729    fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
730        // A config that sizes the workflow channel but omits the low-rate cluster
731        // channel must resolve the cluster capacity to its default and boot, not
732        // fail loudly.
733        let config = ServerConfig::from_slice(
734            br"
735                [runtime]
736                scheduler_threads = 1
737                query_timeout_ms = 10000
738
739                [websocket]
740                event_broadcast_capacity = 64
741            ",
742        )?;
743        assert_eq!(
744            config.websocket.cluster_broadcast_capacity,
745            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
746            "omitted cluster_broadcast_capacity must resolve to the default"
747        );
748        Ok(())
749    }
750
751    #[test]
752    fn zero_cluster_broadcast_capacity_fails_startup_validation() {
753        let result = ServerConfig::from_slice(
754            br"
755                [runtime]
756                query_timeout_ms = 10000
757
758                [websocket]
759                event_broadcast_capacity = 64
760                cluster_broadcast_capacity = 0
761            ",
762        );
763
764        let message = result
765            .err()
766            .map_or_else(String::new, |error| error.to_string());
767        assert!(
768            message.contains("websocket.cluster_broadcast_capacity"),
769            "validation message must name the zero-valued cluster key: {message}"
770        );
771    }
772
773    /// An omitted `[observability]` section resolves both retention bounds to
774    /// their defaults and boots — retention is on out of the box, never a
775    /// forced operator decision.
776    #[test]
777    fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
778        let config = ServerConfig::from_slice(
779            br"
780                [runtime]
781                query_timeout_ms = 10000
782
783                [websocket]
784                event_broadcast_capacity = 64
785                cluster_broadcast_capacity = 64
786            ",
787        )?;
788        assert_eq!(
789            config.observability.max_event_bytes,
790            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
791        );
792        assert_eq!(
793            config.observability.max_stream_events,
794            crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
795        );
796        // The bounds also ride into the runtime view the server state reads.
797        let (_store, runtime) = config.into_parts();
798        assert_eq!(
799            runtime.observability.max_event_bytes,
800            crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
801        );
802        Ok(())
803    }
804
805    /// Explicit `[observability]` values parse and round-trip into the runtime
806    /// view.
807    #[test]
808    fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
809        let config = ServerConfig::from_slice(
810            br"
811                [runtime]
812                query_timeout_ms = 10000
813
814                [websocket]
815                event_broadcast_capacity = 64
816                cluster_broadcast_capacity = 64
817
818                [observability]
819                max_event_bytes = 512
820                max_stream_events = 3
821            ",
822        )?;
823        assert_eq!(config.observability.max_event_bytes, 512);
824        assert_eq!(config.observability.max_stream_events, 3);
825        let (_store, runtime) = config.into_parts();
826        assert_eq!(runtime.observability.max_event_bytes, 512);
827        assert_eq!(runtime.observability.max_stream_events, 3);
828        Ok(())
829    }
830
831    #[test]
832    fn zero_observability_max_event_bytes_fails_startup_validation() {
833        let result = ServerConfig::from_slice(
834            br"
835                [runtime]
836                query_timeout_ms = 10000
837
838                [websocket]
839                event_broadcast_capacity = 64
840                cluster_broadcast_capacity = 64
841
842                [observability]
843                max_event_bytes = 0
844            ",
845        );
846        let message = result
847            .err()
848            .map_or_else(String::new, |error| error.to_string());
849        assert!(
850            message.contains("observability.max_event_bytes"),
851            "validation message must name the zero-valued key: {message}"
852        );
853    }
854
855    #[test]
856    fn zero_observability_max_stream_events_fails_startup_validation() {
857        let result = ServerConfig::from_slice(
858            br"
859                [runtime]
860                query_timeout_ms = 10000
861
862                [websocket]
863                event_broadcast_capacity = 64
864                cluster_broadcast_capacity = 64
865
866                [observability]
867                max_stream_events = 0
868            ",
869        );
870        let message = result
871            .err()
872            .map_or_else(String::new, |error| error.to_string());
873        assert!(
874            message.contains("observability.max_stream_events"),
875            "validation message must name the zero-valued key: {message}"
876        );
877    }
878
879    #[test]
880    fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
881        // The server unconditionally mounts /workflows/query, but the reply
882        // deadline is a tuning knob: omitting it must resolve to the default and
883        // boot, not fail startup.
884        let config = ServerConfig::from_slice(
885            br"
886                [runtime]
887                scheduler_threads = 1
888
889                [websocket]
890                event_broadcast_capacity = 64
891                cluster_broadcast_capacity = 64
892            ",
893        )?;
894        assert_eq!(
895            config.runtime.query_timeout_ms,
896            Some(DEFAULT_QUERY_TIMEOUT_MS),
897            "omitted query_timeout_ms must resolve to the default"
898        );
899        Ok(())
900    }
901
902    #[test]
903    fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
904        // The headline zero-config contract: an empty TOML must parse, fill every
905        // operational tuning knob with its default, and validate — so `aion
906        // server` runs with no hand-authored file. The durable default backend
907        // (haematite under its default data_dir) carries the store side.
908        let config = ServerConfig::from_slice(b"")?;
909        assert_eq!(config.store.backend, StoreBackend::Haematite);
910        assert_eq!(
911            config.runtime.query_timeout_ms,
912            Some(DEFAULT_QUERY_TIMEOUT_MS)
913        );
914        assert_eq!(
915            config.websocket.event_broadcast_capacity,
916            Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
917        );
918        assert_eq!(
919            config.websocket.cluster_broadcast_capacity,
920            Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
921        );
922        Ok(())
923    }
924
925    #[test]
926    fn zero_query_timeout_fails_startup_validation() {
927        let result = ServerConfig::from_slice(
928            br"
929                [runtime]
930                query_timeout_ms = 0
931
932                [websocket]
933                event_broadcast_capacity = 64
934                cluster_broadcast_capacity = 64
935            ",
936        );
937
938        let message = result
939            .err()
940            .map_or_else(String::new, |error| error.to_string());
941        assert!(
942            message.contains("runtime.query_timeout_ms"),
943            "validation message must name the zero-valued key: {message}"
944        );
945    }
946
947    /// The deploy surface is commissioned explicitly: enabling it without
948    /// the archive ceiling is a conservative security default, not a forced
949    /// operator decision: enabling deploy without it must resolve the ceiling
950    /// to [`DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES`] and boot, not fail startup.
951    #[test]
952    fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
953        let config = ServerConfig::from_slice(
954            br"
955                [runtime]
956                query_timeout_ms = 10000
957
958                [websocket]
959                event_broadcast_capacity = 64
960                cluster_broadcast_capacity = 64
961
962                [deploy]
963                enabled = true
964            ",
965        )?;
966
967        assert_eq!(
968            config.deploy.max_archive_bytes,
969            Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
970            "omitted max_archive_bytes must resolve to the conservative default"
971        );
972        assert_eq!(
973            config.deploy.max_inflated_bytes,
974            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
975            "omitted max_inflated_bytes must resolve to the conservative default"
976        );
977        Ok(())
978    }
979
980    #[test]
981    fn deploy_zero_max_archive_bytes_fails_startup_validation() {
982        let result = ServerConfig::from_slice(
983            br"
984                [runtime]
985                query_timeout_ms = 10000
986
987                [websocket]
988                event_broadcast_capacity = 64
989                cluster_broadcast_capacity = 64
990
991                [deploy]
992                enabled = true
993                max_archive_bytes = 0
994            ",
995        );
996
997        let message = result
998            .err()
999            .map_or_else(String::new, |error| error.to_string());
1000        assert!(
1001            message.contains("deploy.max_archive_bytes"),
1002            "validation message must name the zero-valued key: {message}"
1003        );
1004    }
1005
1006    /// The inflate ceiling defaults independently of an explicit archive
1007    /// ceiling: setting only `max_archive_bytes` must resolve the inflate
1008    /// ceiling to [`DEFAULT_DEPLOY_MAX_INFLATED_BYTES`] (which exceeds a 16 MiB
1009    /// archive, so the invariant holds) and boot.
1010    #[test]
1011    fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
1012        let config = ServerConfig::from_slice(
1013            br"
1014                [runtime]
1015                query_timeout_ms = 10000
1016
1017                [websocket]
1018                event_broadcast_capacity = 64
1019                cluster_broadcast_capacity = 64
1020
1021                [deploy]
1022                enabled = true
1023                max_archive_bytes = 16777216
1024            ",
1025        )?;
1026
1027        assert_eq!(
1028            config.deploy.max_archive_bytes,
1029            Some(16_777_216),
1030            "explicit max_archive_bytes must be left untouched"
1031        );
1032        assert_eq!(
1033            config.deploy.max_inflated_bytes,
1034            Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
1035            "omitted max_inflated_bytes must resolve to the conservative default"
1036        );
1037        Ok(())
1038    }
1039
1040    #[test]
1041    fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
1042        let result = ServerConfig::from_slice(
1043            br"
1044                [runtime]
1045                query_timeout_ms = 10000
1046
1047                [websocket]
1048                event_broadcast_capacity = 64
1049                cluster_broadcast_capacity = 64
1050
1051                [deploy]
1052                enabled = true
1053                max_archive_bytes = 16777216
1054                max_inflated_bytes = 0
1055            ",
1056        );
1057
1058        let message = result
1059            .err()
1060            .map_or_else(String::new, |error| error.to_string());
1061        assert!(
1062            message.contains("deploy.max_inflated_bytes"),
1063            "validation message must name the zero-valued key: {message}"
1064        );
1065    }
1066
1067    /// An inflate ceiling below the upload ceiling is incoherent: archives
1068    /// the upload ceiling admits would be refused even stored uncompressed.
1069    #[test]
1070    fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
1071        let result = ServerConfig::from_slice(
1072            br"
1073                [runtime]
1074                query_timeout_ms = 10000
1075
1076                [websocket]
1077                event_broadcast_capacity = 64
1078                cluster_broadcast_capacity = 64
1079
1080                [deploy]
1081                enabled = true
1082                max_archive_bytes = 16777216
1083                max_inflated_bytes = 16777215
1084            ",
1085        );
1086
1087        let message = result
1088            .err()
1089            .map_or_else(String::new, |error| error.to_string());
1090        assert!(
1091            message.contains("deploy.max_inflated_bytes")
1092                && message.contains("deploy.max_archive_bytes"),
1093            "validation message must name both ceilings: {message}"
1094        );
1095    }
1096
1097    /// An absent `[deploy]` section means the surface stays dark and the
1098    /// ceilings are not required.
1099    #[test]
1100    fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
1101        let config = ServerConfig::from_slice(
1102            br"
1103                [runtime]
1104                query_timeout_ms = 10000
1105
1106                [websocket]
1107                event_broadcast_capacity = 64
1108                cluster_broadcast_capacity = 64
1109            ",
1110        )?;
1111
1112        assert!(!config.deploy.enabled);
1113        assert_eq!(config.deploy.max_archive_bytes, None);
1114        assert_eq!(config.deploy.max_inflated_bytes, None);
1115        Ok(())
1116    }
1117
1118    #[test]
1119    fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
1120        let config = 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                [deploy]
1130                enabled = true
1131                max_archive_bytes = 16777216
1132                max_inflated_bytes = 67108864
1133            ",
1134        )?;
1135
1136        assert!(config.deploy.enabled);
1137        assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
1138        assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
1139        Ok(())
1140    }
1141
1142    /// With no `[server] cors_allowed_origins` the list is empty: the secure
1143    /// default, where no cross-origin request is permitted and no `CorsLayer`
1144    /// is installed.
1145    #[test]
1146    fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
1147        let config = ServerConfig::from_slice(
1148            br"
1149                [runtime]
1150                query_timeout_ms = 10000
1151
1152                [websocket]
1153                event_broadcast_capacity = 64
1154                cluster_broadcast_capacity = 64
1155            ",
1156        )?;
1157
1158        assert!(config.server.cors_allowed_origins.is_empty());
1159        let (_, runtime) = config.into_parts();
1160        assert!(runtime.cors_allowed_origins.is_empty());
1161        Ok(())
1162    }
1163
1164    /// A configured `[server] cors_allowed_origins` list parses and round-trips
1165    /// into `RuntimeConfig` (the value the `CorsLayer` is built from).
1166    #[test]
1167    fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
1168        let config = ServerConfig::from_slice(
1169            br#"
1170                [server]
1171                cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
1172
1173                [runtime]
1174                query_timeout_ms = 10000
1175
1176                [websocket]
1177                event_broadcast_capacity = 64
1178                cluster_broadcast_capacity = 64
1179            "#,
1180        )?;
1181
1182        assert_eq!(
1183            config.server.cors_allowed_origins,
1184            vec![
1185                "http://localhost:5173".to_owned(),
1186                "http://127.0.0.1:5173".to_owned()
1187            ]
1188        );
1189        let (_, runtime) = config.into_parts();
1190        assert_eq!(
1191            runtime.cors_allowed_origins,
1192            vec![
1193                "http://localhost:5173".to_owned(),
1194                "http://127.0.0.1:5173".to_owned()
1195            ]
1196        );
1197        Ok(())
1198    }
1199
1200    /// A malformed CORS origin (no scheme, or a trailing path) can never match a
1201    /// browser `Origin` header, so it fails startup validation rather than
1202    /// silently never matching.
1203    #[test]
1204    fn cors_allowed_origins_reject_malformed() {
1205        for bad in ["", "localhost:5173", "http://localhost:5173/"] {
1206            let toml = format!(
1207                "[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
1208            );
1209            let result = ServerConfig::from_slice(toml.as_bytes());
1210            let message = result
1211                .err()
1212                .map_or_else(String::new, |error| error.to_string());
1213            assert!(
1214                message.contains("cors_allowed_origins"),
1215                "malformed origin `{bad}` must be rejected naming the key: {message}"
1216            );
1217        }
1218    }
1219
1220    /// An absent `[dev]` section leaves the dev surface dark.
1221    #[test]
1222    fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1223        let config = ServerConfig::from_slice(
1224            br"
1225                [runtime]
1226                query_timeout_ms = 10000
1227
1228                [websocket]
1229                event_broadcast_capacity = 64
1230                cluster_broadcast_capacity = 64
1231            ",
1232        )?;
1233
1234        assert!(!config.dev.enabled);
1235        Ok(())
1236    }
1237
1238    /// `[dev] enabled = true` commissions the dev surface; it adds no other
1239    /// knobs (ADR-001: the only setting is the on/off gate).
1240    #[test]
1241    fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
1242        let config = ServerConfig::from_slice(
1243            br"
1244                [runtime]
1245                query_timeout_ms = 10000
1246
1247                [websocket]
1248                event_broadcast_capacity = 64
1249                cluster_broadcast_capacity = 64
1250
1251                [dev]
1252                enabled = true
1253            ",
1254        )?;
1255
1256        assert!(config.dev.enabled);
1257        Ok(())
1258    }
1259
1260    /// An absent `[authoring]` section leaves the surface dark: no `gleam_path`,
1261    /// no `project_root`, and validation does not require either.
1262    #[test]
1263    fn authoring_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
1264        let config = ServerConfig::from_slice(
1265            br"
1266                [runtime]
1267                query_timeout_ms = 10000
1268
1269                [websocket]
1270                event_broadcast_capacity = 64
1271                cluster_broadcast_capacity = 64
1272            ",
1273        )?;
1274
1275        assert_eq!(config.authoring.gleam_path, None);
1276        assert_eq!(config.authoring.project_root, None);
1277        Ok(())
1278    }
1279
1280    /// A configured `[authoring]` section with both `gleam_path` and
1281    /// `project_root` parses and round-trips into `RuntimeConfig`.
1282    #[test]
1283    fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
1284        let config = ServerConfig::from_slice(
1285            br#"
1286                [runtime]
1287                query_timeout_ms = 10000
1288
1289                [websocket]
1290                event_broadcast_capacity = 64
1291                cluster_broadcast_capacity = 64
1292
1293                [authoring]
1294                gleam_path = "/usr/local/bin/gleam"
1295                project_root = "/srv/aion/authoring"
1296            "#,
1297        )?;
1298
1299        assert_eq!(
1300            config.authoring.gleam_path.as_deref(),
1301            Some(std::path::Path::new("/usr/local/bin/gleam"))
1302        );
1303        let (_, runtime) = config.into_parts();
1304        assert_eq!(
1305            runtime.authoring.gleam_path.as_deref(),
1306            Some(std::path::Path::new("/usr/local/bin/gleam"))
1307        );
1308        assert_eq!(
1309            runtime.authoring.project_root.as_deref(),
1310            Some(std::path::Path::new("/srv/aion/authoring"))
1311        );
1312        Ok(())
1313    }
1314
1315    /// Commissioning the authoring loop (a `gleam_path`) without a
1316    /// `project_root` must fail startup naming the key and the environment
1317    /// override (the deploy required-config pattern).
1318    #[test]
1319    fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
1320        let result = ServerConfig::from_slice(
1321            br#"
1322                [runtime]
1323                query_timeout_ms = 10000
1324
1325                [websocket]
1326                event_broadcast_capacity = 64
1327                cluster_broadcast_capacity = 64
1328
1329                [authoring]
1330                gleam_path = "/usr/local/bin/gleam"
1331            "#,
1332        );
1333
1334        let message = result
1335            .err()
1336            .map_or_else(String::new, |error| error.to_string());
1337        assert!(
1338            message.contains("authoring.project_root"),
1339            "validation message must name the missing key: {message}"
1340        );
1341        assert!(
1342            message.contains("AION_AUTHORING_PROJECT_ROOT"),
1343            "validation message must name the environment override: {message}"
1344        );
1345    }
1346
1347    /// An empty `gleam_path` is a misconfiguration, not "dark": it must fail
1348    /// startup naming the key and the environment override.
1349    #[test]
1350    fn authoring_empty_gleam_path_fails_naming_key_and_env() {
1351        let result = ServerConfig::from_slice(
1352            br#"
1353                [runtime]
1354                query_timeout_ms = 10000
1355
1356                [websocket]
1357                event_broadcast_capacity = 64
1358                cluster_broadcast_capacity = 64
1359
1360                [authoring]
1361                gleam_path = ""
1362            "#,
1363        );
1364
1365        let message = result
1366            .err()
1367            .map_or_else(String::new, |error| error.to_string());
1368        assert!(
1369            message.contains("authoring.gleam_path"),
1370            "validation message must name the empty key: {message}"
1371        );
1372        assert!(
1373            message.contains("AION_AUTHORING_GLEAM_PATH"),
1374            "validation message must name the environment override: {message}"
1375        );
1376    }
1377
1378    /// CLI overrides commission the authoring loop after file/env merge.
1379    #[test]
1380    fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
1381        let mut config = ServerConfig::from_slice(
1382            br"
1383                [runtime]
1384                query_timeout_ms = 10000
1385
1386                [websocket]
1387                event_broadcast_capacity = 64
1388                cluster_broadcast_capacity = 64
1389            ",
1390        )?;
1391        let cli = CliOverrides {
1392            gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
1393            authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
1394            ..CliOverrides::default()
1395        };
1396
1397        config.apply_cli_overrides(&cli);
1398        config.validate()?;
1399
1400        assert_eq!(
1401            config.authoring.gleam_path.as_deref(),
1402            Some(std::path::Path::new("/opt/gleam"))
1403        );
1404        assert_eq!(
1405            config.authoring.project_root.as_deref(),
1406            Some(std::path::Path::new("/opt/project"))
1407        );
1408        Ok(())
1409    }
1410
1411    /// The config field renamed `dashboard` -> `ops_console` carries a serde
1412    /// alias so existing `[dashboard]` TOML still parses (non-breaking rename).
1413    #[test]
1414    fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
1415        let config = ServerConfig::from_slice(
1416            br#"
1417                [runtime]
1418                query_timeout_ms = 10000
1419
1420                [websocket]
1421                event_broadcast_capacity = 64
1422                cluster_broadcast_capacity = 64
1423
1424                [dashboard]
1425                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1426            "#,
1427        )?;
1428        match &config.ops_console.source {
1429            OpsConsoleAssetSource::FileSystem { asset_path } => {
1430                assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
1431            }
1432            OpsConsoleAssetSource::Embedded => {
1433                return Err("legacy [dashboard] section must map to ops_console".into());
1434            }
1435        }
1436        Ok(())
1437    }
1438
1439    /// The new `[ops_console]` section name also parses.
1440    #[test]
1441    fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
1442        let config = ServerConfig::from_slice(
1443            br#"
1444                [runtime]
1445                query_timeout_ms = 10000
1446
1447                [websocket]
1448                event_broadcast_capacity = 64
1449                cluster_broadcast_capacity = 64
1450
1451                [ops_console]
1452                source = { FileSystem = { asset_path = "/srv/aion/ui" } }
1453            "#,
1454        )?;
1455        assert!(matches!(
1456            config.ops_console.source,
1457            OpsConsoleAssetSource::FileSystem { .. }
1458        ));
1459        Ok(())
1460    }
1461
1462    #[test]
1463    fn invalid_values_name_problematic_field() {
1464        let result = ServerConfig::from_slice(
1465            br"
1466                [runtime]
1467                scheduler_threads = 0
1468            ",
1469        );
1470
1471        let message = result
1472            .err()
1473            .map_or_else(String::new, |error| error.to_string());
1474        assert!(message.contains("runtime.scheduler_threads"));
1475    }
1476
1477    #[test]
1478    fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
1479        let mut config = ServerConfig::from_slice(
1480            br#"
1481                [store]
1482                backend = "libsql"
1483                url = "file.db"
1484
1485                [runtime]
1486                query_timeout_ms = 10000
1487
1488                [websocket]
1489                event_broadcast_capacity = 64
1490                cluster_broadcast_capacity = 64
1491            "#,
1492        )?;
1493        let cli = CliOverrides {
1494            store_url: Some("cli.db".to_owned()),
1495            scheduler_threads: Some(3),
1496            ..CliOverrides::default()
1497        };
1498
1499        config.apply_cli_overrides(&cli);
1500        config.validate()?;
1501
1502        assert_eq!(config.store.url.as_deref(), Some("cli.db"));
1503        assert_eq!(config.runtime.scheduler_threads, 3);
1504        Ok(())
1505    }
1506
1507    #[test]
1508    fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
1509        let mut config = ServerConfig::default();
1510
1511        // The ablative stack is the out-of-box durable default: an empty config
1512        // selects the haematite backend rooted at the default data_dir, so a stock
1513        // server is durable without any [store] configuration. The default MUST
1514        // carry data_dir or validate() would reject it (data_dir is required for
1515        // haematite).
1516        assert_eq!(config.store.backend, StoreBackend::Haematite);
1517        assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
1518        // 64 pending #187: 4096 defeated its own lazy-materialization premise
1519        // (boot scan_prefix materializes all shards; commit then fsyncs per
1520        // shard and blows the actor timeout). See StoreConfig::default().
1521        assert_eq!(config.store.shard_count, 64);
1522        assert_eq!(config.store.url, None);
1523        assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
1524        assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
1525        assert_eq!(config.namespaces.default, "default");
1526        // Minted-on-use is OPEN by default to preserve the zero-config,
1527        // no-pre-provision model: a namespace comes into being on first
1528        // worker reference.
1529        assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
1530        // The cluster-wide in-flight ceiling defaults to the generous platform
1531        // headroom value (P2-Q1); nothing enforces it yet.
1532        assert_eq!(
1533            config.namespaces.max_in_flight_activities,
1534            DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
1535        );
1536        assert_eq!(config.namespaces.max_in_flight_activities, 1024);
1537        assert!(!config.auth.enabled);
1538        assert!(config.metrics.enabled);
1539        // event_broadcast_capacity and query_timeout_ms are the deliberately
1540        // defaultless values: defaults validate only once the operator
1541        // supplies them.
1542        assert_eq!(config.websocket.event_broadcast_capacity, None);
1543        assert_eq!(config.websocket.cluster_broadcast_capacity, None);
1544        assert_eq!(config.runtime.query_timeout_ms, None);
1545        config.websocket.event_broadcast_capacity = Some(64);
1546        config.websocket.cluster_broadcast_capacity = Some(64);
1547        config.runtime.query_timeout_ms = Some(10_000);
1548        config.validate()?;
1549        Ok(())
1550    }
1551
1552    #[test]
1553    fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
1554    {
1555        let mut config = ServerConfig::default();
1556        config.websocket.event_broadcast_capacity = Some(64);
1557        config.websocket.cluster_broadcast_capacity = Some(64);
1558        config.runtime.query_timeout_ms = Some(10_000);
1559
1560        // The dispatcher is dark by default and its operational knobs are all
1561        // absent — yet validation passes, because a disabled dispatcher never
1562        // reads them (no assumed defaults behind the gate).
1563        assert!(!config.outbox.enabled);
1564        assert_eq!(config.outbox.poll_interval_ms, None);
1565        assert_eq!(config.outbox.batch_size, None);
1566        assert_eq!(config.outbox.max_attempts, None);
1567        assert_eq!(config.outbox.backoff_base_ms, None);
1568        assert_eq!(config.outbox.backoff_multiplier, None);
1569        assert_eq!(config.outbox.backoff_max_ms, None);
1570        assert_eq!(config.outbox.reconcile_interval_ms, None);
1571        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
1572        config.validate()?;
1573        Ok(())
1574    }
1575
1576    fn outbox_enabled_base() -> ServerConfig {
1577        let mut config = ServerConfig::default();
1578        config.websocket.event_broadcast_capacity = Some(64);
1579        config.websocket.cluster_broadcast_capacity = Some(64);
1580        config.runtime.query_timeout_ms = Some(10_000);
1581        config.outbox.enabled = true;
1582        config.outbox.poll_interval_ms = Some(250);
1583        config.outbox.batch_size = Some(64);
1584        config.outbox.max_attempts = Some(5);
1585        config.outbox.backoff_base_ms = Some(100);
1586        config.outbox.backoff_multiplier = Some(2);
1587        config.outbox.backoff_max_ms = Some(30_000);
1588        config.outbox.reconcile_interval_ms = Some(1_000);
1589        config.outbox.reconcile_stale_after_ms = Some(60_000);
1590        config
1591    }
1592
1593    #[test]
1594    fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
1595        outbox_enabled_base().validate()?;
1596        Ok(())
1597    }
1598
1599    #[test]
1600    fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
1601        // Enabling the dispatcher but omitting the poll cadence must resolve it
1602        // to the default and boot, not fail startup — the cadence is pure
1603        // tuning, not a forced operator decision.
1604        let config = ServerConfig::from_slice(
1605            br"
1606                [runtime]
1607                query_timeout_ms = 10000
1608
1609                [websocket]
1610                event_broadcast_capacity = 64
1611                cluster_broadcast_capacity = 64
1612
1613                [outbox]
1614                enabled = true
1615            ",
1616        )?;
1617        assert_eq!(
1618            config.outbox.poll_interval_ms,
1619            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
1620            "omitted poll_interval_ms must resolve to the default"
1621        );
1622        Ok(())
1623    }
1624
1625    #[test]
1626    fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
1627        // Setting a tuning knob explicitly but omitting the retry budget must
1628        // leave the explicit knob untouched and default only the omitted one.
1629        let config = ServerConfig::from_slice(
1630            br"
1631                [runtime]
1632                query_timeout_ms = 10000
1633
1634                [websocket]
1635                event_broadcast_capacity = 64
1636                cluster_broadcast_capacity = 64
1637
1638                [outbox]
1639                enabled = true
1640                poll_interval_ms = 250
1641            ",
1642        )?;
1643        assert_eq!(
1644            config.outbox.poll_interval_ms,
1645            Some(250),
1646            "explicit poll_interval_ms must be left untouched"
1647        );
1648        assert_eq!(
1649            config.outbox.max_attempts,
1650            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
1651            "omitted max_attempts must resolve to the default"
1652        );
1653        Ok(())
1654    }
1655
1656    #[test]
1657    fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
1658    -> Result<(), Box<dyn std::error::Error>> {
1659        // Headline conditional-default contract: an outbox section with nothing
1660        // but `enabled = true` validates with every tuning knob resolved to its
1661        // default. The reconciliation pair stays dark (both absent), as before.
1662        let config = ServerConfig::from_slice(
1663            br"
1664                [runtime]
1665                query_timeout_ms = 10000
1666
1667                [websocket]
1668                event_broadcast_capacity = 64
1669                cluster_broadcast_capacity = 64
1670
1671                [outbox]
1672                enabled = true
1673            ",
1674        )?;
1675        assert!(config.outbox.enabled);
1676        assert_eq!(
1677            config.outbox.poll_interval_ms,
1678            Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
1679        );
1680        assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
1681        assert_eq!(
1682            config.outbox.max_attempts,
1683            Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
1684        );
1685        assert_eq!(
1686            config.outbox.backoff_base_ms,
1687            Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
1688        );
1689        assert_eq!(
1690            config.outbox.backoff_multiplier,
1691            Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
1692        );
1693        assert_eq!(
1694            config.outbox.backoff_max_ms,
1695            Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
1696        );
1697        // Reconciliation is not force-defaulted: both knobs stay absent so the
1698        // live sweep remains dark.
1699        assert_eq!(config.outbox.reconcile_interval_ms, None);
1700        assert_eq!(config.outbox.reconcile_stale_after_ms, None);
1701        Ok(())
1702    }
1703
1704    #[test]
1705    fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
1706        // An explicit zero is a misconfiguration the default never masks:
1707        // `get_or_insert` leaves `Some(0)` untouched and validate rejects it.
1708        let mut config = outbox_enabled_base();
1709        config.outbox.poll_interval_ms = Some(0);
1710        let error = config
1711            .validate()
1712            .err()
1713            .ok_or("enabled outbox with zero poll interval must fail")?;
1714        assert!(
1715            error.to_string().contains("outbox.poll_interval_ms"),
1716            "error must name the zero-valued key: {error}"
1717        );
1718        Ok(())
1719    }
1720
1721    #[test]
1722    fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
1723        let mut config = outbox_enabled_base();
1724        config.outbox.max_attempts = Some(0);
1725        let error = config
1726            .validate()
1727            .err()
1728            .ok_or("enabled outbox with zero max attempts must fail")?;
1729        assert!(
1730            error.to_string().contains("outbox.max_attempts"),
1731            "error must name the zero-valued key: {error}"
1732        );
1733        Ok(())
1734    }
1735
1736    #[test]
1737    fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
1738        let mut config = outbox_enabled_base();
1739        config.outbox.backoff_base_ms = Some(1_000);
1740        config.outbox.backoff_max_ms = Some(500);
1741        let error = config
1742            .validate()
1743            .err()
1744            .ok_or("backoff_max below backoff_base must fail")?;
1745        assert!(
1746            error.to_string().contains("outbox.backoff_max_ms"),
1747            "error must name the offending key: {error}"
1748        );
1749        Ok(())
1750    }
1751
1752    #[test]
1753    fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
1754        let mut config = outbox_enabled_base();
1755        config.outbox.reconcile_interval_ms = None;
1756        config.outbox.reconcile_stale_after_ms = None;
1757        config.validate()?;
1758        Ok(())
1759    }
1760
1761    #[test]
1762    fn outbox_reconciliation_requires_interval_when_partially_enabled()
1763    -> Result<(), Box<dyn std::error::Error>> {
1764        let mut config = outbox_enabled_base();
1765        config.outbox.reconcile_interval_ms = None;
1766        let error = config
1767            .validate()
1768            .err()
1769            .ok_or("reconciliation without interval must fail")?;
1770        assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
1771        Ok(())
1772    }
1773
1774    #[test]
1775    fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
1776    -> Result<(), Box<dyn std::error::Error>> {
1777        let mut config = outbox_enabled_base();
1778        config.outbox.reconcile_stale_after_ms = None;
1779        let error = config
1780            .validate()
1781            .err()
1782            .ok_or("reconciliation without stale threshold must fail")?;
1783        assert!(
1784            error
1785                .to_string()
1786                .contains("outbox.reconcile_stale_after_ms")
1787        );
1788        Ok(())
1789    }
1790
1791    #[test]
1792    fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
1793        let temp_dir = tempfile::tempdir()?;
1794        std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
1795        std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
1796        std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
1797        std::fs::create_dir(temp_dir.path().join("nested"))?;
1798        std::fs::write(
1799            temp_dir.path().join("nested").join("nested.aion"),
1800            b"package",
1801        )?;
1802
1803        let packages = discover_workflow_packages(temp_dir.path())?;
1804
1805        assert_eq!(
1806            packages,
1807            vec![
1808                temp_dir.path().join("alpha.aion"),
1809                temp_dir.path().join("zeta.aion"),
1810            ]
1811        );
1812        Ok(())
1813    }
1814
1815    #[test]
1816    fn workflow_package_merge_is_additive_and_deduplicated() {
1817        let mut packages = vec!["config.aion".into(), "shared.aion".into()];
1818        let discovered = vec!["auto.aion".into(), "shared.aion".into()];
1819        let cli = vec!["cli.aion".into(), "auto.aion".into()];
1820
1821        merge_workflow_packages(&mut packages, discovered, &cli);
1822
1823        assert_eq!(
1824            packages,
1825            vec![
1826                std::path::PathBuf::from("config.aion"),
1827                std::path::PathBuf::from("shared.aion"),
1828                std::path::PathBuf::from("auto.aion"),
1829                std::path::PathBuf::from("cli.aion"),
1830            ]
1831        );
1832    }
1833
1834    #[test]
1835    fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
1836        let temp_dir = tempfile::tempdir()?;
1837        let package = temp_dir.path().join("hello.aion");
1838        std::fs::write(&package, b"package")?;
1839        let mut packages = vec![package.clone()];
1840        let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
1841
1842        merge_workflow_packages(&mut packages, discovered, &[]);
1843
1844        assert_eq!(packages, vec![package]);
1845        Ok(())
1846    }
1847
1848    #[test]
1849    fn zero_config_cli_workflow_package_uses_in_memory_defaults()
1850    -> Result<(), Box<dyn std::error::Error>> {
1851        let temp_dir = tempfile::tempdir()?;
1852
1853        let cli = CliOverrides {
1854            workflow_packages: vec!["hello-world.aion".into()],
1855            ..CliOverrides::default()
1856        };
1857        let mut config = ServerConfig::default();
1858        // This test exercises CLI workflow-package discovery against the ephemeral
1859        // in-memory store, so it opts OUT of the new durable haematite default
1860        // explicitly (the default would otherwise carry a haematite data_dir).
1861        config.store.backend = StoreBackend::Memory;
1862        config.store.data_dir = None;
1863        // Even zero-config development runs must size event streaming and the
1864        // query reply deadline explicitly (config keys or the
1865        // AION_WEBSOCKET_EVENT_BROADCAST_CAPACITY /
1866        // AION_RUNTIME_QUERY_TIMEOUT_MS environment overrides).
1867        config.websocket.event_broadcast_capacity = Some(64);
1868        config.websocket.cluster_broadcast_capacity = Some(64);
1869        config.runtime.query_timeout_ms = Some(10_000);
1870        config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
1871
1872        config.validate()?;
1873
1874        assert_eq!(config.store.backend, StoreBackend::Memory);
1875        assert_eq!(config.store.url, None);
1876        assert_eq!(
1877            config.workflow_packages,
1878            vec![std::path::PathBuf::from("hello-world.aion")]
1879        );
1880        Ok(())
1881    }
1882
1883    #[test]
1884    fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
1885        let mut config = ServerConfig::from_slice(
1886            br#"
1887                workflow_packages = ["config.aion"]
1888
1889                [runtime]
1890                query_timeout_ms = 10000
1891
1892                [websocket]
1893                event_broadcast_capacity = 64
1894                cluster_broadcast_capacity = 64
1895            "#,
1896        )?;
1897        let cli = CliOverrides {
1898            workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
1899            ..CliOverrides::default()
1900        };
1901
1902        merge_workflow_packages(
1903            &mut config.workflow_packages,
1904            Vec::new(),
1905            &cli.workflow_packages,
1906        );
1907
1908        assert_eq!(
1909            config.workflow_packages,
1910            vec![
1911                std::path::PathBuf::from("config.aion"),
1912                std::path::PathBuf::from("cli-one.aion"),
1913                std::path::PathBuf::from("cli-two.aion"),
1914            ]
1915        );
1916        Ok(())
1917    }
1918}
aion_server/config/load.rs

aion_server/config/
load.rs
