use std::{
collections::HashSet,
fs,
path::{Path, PathBuf},
time::Duration,
};
use serde::Deserialize;
use crate::error::ServerError;
use super::{
AUTHORING_GLEAM_PATH_EMPTY, AUTHORING_PROJECT_ROOT_REQUIRED, AuthConfig, AuthoringConfig,
CORS_ALLOWED_ORIGIN_INVALID, CliOverrides, ClusterConfig, DEFAULT_CLUSTER_BROADCAST_CAPACITY,
DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES, DEFAULT_DEPLOY_MAX_INFLATED_BYTES,
DEFAULT_EVENT_BROADCAST_CAPACITY, DEFAULT_OUTBOX_BACKOFF_BASE_MS,
DEFAULT_OUTBOX_BACKOFF_MAX_MS, DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE,
DEFAULT_OUTBOX_MAX_ATTEMPTS, DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS,
DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED, DEPLOY_MAX_INFLATED_BYTES_REQUIRED, DeployConfig, DevConfig,
DrainConfig, ListenConfig, MetricsConfig, NamespaceConfig, NamespaceMode, NamespacesConfig,
OUTBOX_BACKOFF_BASE_REQUIRED, OUTBOX_BACKOFF_MAX_REQUIRED, OUTBOX_BACKOFF_MULTIPLIER_REQUIRED,
OUTBOX_BATCH_SIZE_REQUIRED, OUTBOX_MAX_ATTEMPTS_REQUIRED, OUTBOX_POLL_INTERVAL_REQUIRED,
OUTBOX_RECONCILE_INTERVAL_REQUIRED, OUTBOX_RECONCILE_STALE_AFTER_REQUIRED, ObservabilityConfig,
OpsConsoleAssetSource, OpsConsoleConfig, OutboxConfig, QUERY_TIMEOUT_REQUIRED, RuntimeConfig,
RuntimeSection, ServerSection, StoreBackend, StoreConfig, TlsConfig, WebSocketConfig,
WorkerConfig, config_error, env, file,
};
#[derive(Clone, Debug, Deserialize)]
#[serde(default, deny_unknown_fields)]
#[derive(Default)]
pub struct ServerConfig {
pub server: ServerSection,
pub store: StoreConfig,
pub runtime: RuntimeSection,
pub drain: DrainConfig,
pub auth: AuthConfig,
pub metrics: MetricsConfig,
pub namespaces: NamespacesConfig,
pub tls: Option<TlsConfig>,
#[serde(alias = "dashboard")]
pub ops_console: OpsConsoleConfig,
pub namespace: NamespaceConfig,
pub worker: WorkerConfig,
pub websocket: WebSocketConfig,
pub workflow_packages: Vec<PathBuf>,
pub deploy: DeployConfig,
pub authoring: AuthoringConfig,
pub dev: DevConfig,
pub outbox: OutboxConfig,
pub observability: ObservabilityConfig,
}
impl ServerConfig {
pub fn load(cli: &CliOverrides) -> Result<Self, ServerError> {
let mut config = file::load(cli.config_path.as_deref())?.unwrap_or_default();
env::overlay(&mut config)?;
config.apply_cli_overrides(cli);
config.load_discovered_workflow_packages(cli, Path::new("."))?;
config.fill_operational_defaults();
config.validate()?;
Ok(config)
}
fn fill_operational_defaults(&mut self) {
self.runtime
.query_timeout_ms
.get_or_insert(DEFAULT_QUERY_TIMEOUT_MS);
self.websocket
.event_broadcast_capacity
.get_or_insert(DEFAULT_EVENT_BROADCAST_CAPACITY);
self.websocket
.cluster_broadcast_capacity
.get_or_insert(DEFAULT_CLUSTER_BROADCAST_CAPACITY);
self.fill_outbox_defaults();
self.fill_deploy_defaults();
}
fn fill_outbox_defaults(&mut self) {
if !self.outbox.enabled {
return;
}
self.outbox
.poll_interval_ms
.get_or_insert(DEFAULT_OUTBOX_POLL_INTERVAL_MS);
self.outbox
.batch_size
.get_or_insert(DEFAULT_OUTBOX_BATCH_SIZE);
self.outbox
.max_attempts
.get_or_insert(DEFAULT_OUTBOX_MAX_ATTEMPTS);
self.outbox
.backoff_base_ms
.get_or_insert(DEFAULT_OUTBOX_BACKOFF_BASE_MS);
self.outbox
.backoff_multiplier
.get_or_insert(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER);
self.outbox
.backoff_max_ms
.get_or_insert(DEFAULT_OUTBOX_BACKOFF_MAX_MS);
}
fn fill_deploy_defaults(&mut self) {
if !self.deploy.enabled {
return;
}
self.deploy
.max_archive_bytes
.get_or_insert(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES);
self.deploy
.max_inflated_bytes
.get_or_insert(DEFAULT_DEPLOY_MAX_INFLATED_BYTES);
}
fn load_discovered_workflow_packages(
&mut self,
cli: &CliOverrides,
directory: &Path,
) -> Result<(), ServerError> {
let discovered_packages = discover_workflow_packages(directory)?;
merge_workflow_packages(
&mut self.workflow_packages,
discovered_packages,
&cli.workflow_packages,
);
Ok(())
}
pub fn from_slice(bytes: &[u8]) -> Result<Self, ServerError> {
let mut config: Self = toml::from_slice(bytes).map_err(|source| ServerError::Config {
message: format!("invalid server config: {source}"),
})?;
config.fill_operational_defaults();
config.validate()?;
Ok(config)
}
pub fn load_from_path(path: impl Into<PathBuf>) -> Result<Self, ServerError> {
file::load_required(&path.into())
}
#[must_use]
pub fn into_parts(self) -> (StoreConfig, RuntimeConfig) {
let runtime = RuntimeConfig {
listen: ListenConfig {
grpc: self.server.grpc_address,
http: self.server.listen_address,
},
tls: self.tls,
auth: self.auth,
ops_console: self.ops_console,
namespace: self.namespace,
worker: self.worker,
websocket: self.websocket,
workflow_packages: self.workflow_packages,
deploy: self.deploy,
authoring: self.authoring,
dev: self.dev,
outbox: self.outbox,
observability: self.observability,
scheduler_threads: self.runtime.scheduler_threads,
query_timeout: self.runtime.query_timeout_ms.map(Duration::from_millis),
default_namespace: self.namespaces.default,
auto_create: self.namespaces.auto_create,
max_in_flight_activities: self.namespaces.max_in_flight_activities,
drain_timeout: Duration::from_secs(self.drain.timeout_seconds),
metrics: self.metrics,
owned_shards: self.store.owned_shards.clone(),
cors_allowed_origins: self.server.cors_allowed_origins.clone(),
};
(self.store, runtime)
}
fn apply_cli_overrides(&mut self, cli: &CliOverrides) {
if let Some(address) = cli.listen_address {
self.server.listen_address = address;
}
if let Some(url) = &cli.store_url {
self.store.url = Some(url.clone());
if matches!(
self.store.backend,
StoreBackend::Memory | StoreBackend::Haematite
) {
self.store.backend = StoreBackend::LibSql;
}
}
if let Some(threads) = cli.scheduler_threads {
self.runtime.scheduler_threads = threads;
}
if let Some(timeout) = cli.drain_timeout_seconds {
self.drain.timeout_seconds = timeout;
}
if let Some(gleam_path) = &cli.gleam_path {
self.authoring.gleam_path = Some(gleam_path.clone());
}
if let Some(project_root) = &cli.authoring_project_root {
self.authoring.project_root = Some(project_root.clone());
}
}
fn validate(&self) -> Result<(), ServerError> {
if self.server.listen_address.port() == 0 {
return config_error("server.listen_address must use an explicit non-zero port");
}
if self.server.grpc_address.port() == 0 {
return config_error("server.grpc_address must use an explicit non-zero port");
}
validate_cors_origins(&self.server.cors_allowed_origins)?;
if self.runtime.scheduler_threads == 0 {
return config_error("runtime.scheduler_threads must be greater than zero");
}
if self.drain.timeout_seconds == 0 {
return config_error("drain.timeout_seconds must be greater than zero");
}
if self.auth.enabled && self.auth.jwks_url.as_deref().is_none_or(str::is_empty) {
return config_error("auth.jwks_url must not be empty when auth.enabled is true");
}
if self.auth.jwks_refresh_seconds == 0 {
return config_error("auth.jwks_refresh_seconds must be greater than zero");
}
if self.namespaces.default.is_empty() {
return config_error("namespaces.default must not be empty");
}
if matches!(self.store.backend, StoreBackend::LibSql)
&& self.store.url.as_deref().is_none_or(str::is_empty)
{
return config_error("store.url must not be empty when store.backend is libsql");
}
if let Some(url) = &self.store.url {
if url.is_empty() {
return config_error("store.url must not be empty");
}
}
if matches!(self.store.backend, StoreBackend::Haematite) {
if self.store.data_dir.as_deref().is_none_or(str::is_empty) {
return config_error(
"store.data_dir must not be empty when store.backend is haematite",
);
}
if self.store.shard_count == 0 {
return config_error("store.shard_count must be greater than zero");
}
if let Some(cluster) = &self.store.cluster {
validate_cluster(cluster)?;
}
} else if self.store.cluster.is_some() {
return config_error("store.cluster is only valid when store.backend is haematite");
}
if let OpsConsoleAssetSource::FileSystem { asset_path } = &self.ops_console.source {
if asset_path.as_os_str().is_empty() {
return config_error("ops_console.source.FileSystem.asset_path must not be empty");
}
}
if let NamespaceMode::SingleTenant { namespace } = &self.namespace.mode {
if namespace.is_empty() {
return config_error("namespace.mode.SingleTenant.namespace must not be empty");
}
}
if self.worker.heartbeat_window.is_zero() {
return config_error("worker.heartbeat_window must be greater than zero");
}
self.websocket.validate()?;
self.observability.validate()?;
match self.runtime.query_timeout_ms {
None | Some(0) => return config_error(QUERY_TIMEOUT_REQUIRED),
Some(_) => {}
}
if self.deploy.enabled {
let max_archive_bytes = match self.deploy.max_archive_bytes {
None | Some(0) => return config_error(DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED),
Some(value) => value,
};
let max_inflated_bytes = match self.deploy.max_inflated_bytes {
None | Some(0) => return config_error(DEPLOY_MAX_INFLATED_BYTES_REQUIRED),
Some(value) => value,
};
ensure_fits_usize("deploy.max_archive_bytes", max_archive_bytes)?;
ensure_fits_usize("deploy.max_inflated_bytes", max_inflated_bytes)?;
if max_inflated_bytes < max_archive_bytes {
return config_error(format!(
"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"
));
}
}
if let Some(gleam_path) = &self.authoring.gleam_path {
if gleam_path.as_os_str().is_empty() {
return config_error(AUTHORING_GLEAM_PATH_EMPTY);
}
match &self.authoring.project_root {
Some(root) if !root.as_os_str().is_empty() => {}
_ => return config_error(AUTHORING_PROJECT_ROOT_REQUIRED),
}
}
self.validate_outbox()?;
Ok(())
}
fn validate_outbox(&self) -> Result<(), ServerError> {
if !self.outbox.enabled {
return Ok(());
}
match self.outbox.poll_interval_ms {
None | Some(0) => return config_error(OUTBOX_POLL_INTERVAL_REQUIRED),
Some(_) => {}
}
match self.outbox.batch_size {
None | Some(0) => return config_error(OUTBOX_BATCH_SIZE_REQUIRED),
Some(_) => {}
}
match self.outbox.max_attempts {
None | Some(0) => return config_error(OUTBOX_MAX_ATTEMPTS_REQUIRED),
Some(_) => {}
}
let backoff_base_ms = match self.outbox.backoff_base_ms {
None | Some(0) => return config_error(OUTBOX_BACKOFF_BASE_REQUIRED),
Some(value) => value,
};
match self.outbox.backoff_multiplier {
None | Some(0) => return config_error(OUTBOX_BACKOFF_MULTIPLIER_REQUIRED),
Some(_) => {}
}
match self.outbox.backoff_max_ms {
Some(max) if max >= backoff_base_ms => {}
_ => return config_error(OUTBOX_BACKOFF_MAX_REQUIRED),
}
match (
self.outbox.reconcile_interval_ms,
self.outbox.reconcile_stale_after_ms,
) {
(None, None) => {}
(None | Some(0), _) => return config_error(OUTBOX_RECONCILE_INTERVAL_REQUIRED),
(_, None | Some(0)) => return config_error(OUTBOX_RECONCILE_STALE_AFTER_REQUIRED),
(Some(_), Some(_)) => {}
}
Ok(())
}
}
fn validate_cluster(cluster: &ClusterConfig) -> Result<(), ServerError> {
if cluster.node_id.is_empty() {
return config_error("store.cluster.node_id must not be empty");
}
if cluster.members.iter().any(String::is_empty) {
return config_error("store.cluster.members entries must not be empty");
}
if cluster.peers.iter().any(|peer| peer.name.is_empty()) {
return config_error("store.cluster.peers entries must name a non-empty node");
}
if matches!(cluster.failover_poll_interval_ms, Some(0)) {
return config_error(
"store.cluster.failover_poll_interval_ms must be greater than zero when set",
);
}
if matches!(cluster.failover_confirmations, Some(0)) {
return config_error("store.cluster.failover_confirmations must be at least one when set");
}
Ok(())
}
fn validate_cors_origins(origins: &[String]) -> Result<(), ServerError> {
for origin in origins {
validate_cors_origin(origin)?;
}
Ok(())
}
fn validate_cors_origin(origin: &str) -> Result<(), ServerError> {
if origin.is_empty() {
return config_error(CORS_ALLOWED_ORIGIN_INVALID);
}
let scheme_split = origin.split_once("://");
let Some((scheme, authority)) = scheme_split else {
return config_error(CORS_ALLOWED_ORIGIN_INVALID);
};
if scheme.is_empty() || authority.is_empty() || authority.contains('/') {
return config_error(CORS_ALLOWED_ORIGIN_INVALID);
}
if origin.parse::<axum::http::HeaderValue>().is_err() {
return config_error(CORS_ALLOWED_ORIGIN_INVALID);
}
Ok(())
}
fn ensure_fits_usize(key: &str, value: u64) -> Result<(), ServerError> {
if usize::try_from(value).is_err() {
return config_error(format!(
"{key} ({value}) exceeds this platform's addressable memory; set it to at most {}",
usize::MAX
));
}
Ok(())
}
fn discover_workflow_packages(directory: &Path) -> Result<Vec<PathBuf>, ServerError> {
let mut packages = Vec::new();
let entries = fs::read_dir(directory).map_err(|source| ServerError::Config {
message: format!(
"failed to scan workflow packages in `{}`: {source}",
directory.display()
),
})?;
for entry in entries {
let entry = entry.map_err(|source| ServerError::Config {
message: format!(
"failed to read workflow package entry in `{}`: {source}",
directory.display()
),
})?;
let path = entry.path();
let has_aion_extension = path
.extension()
.is_some_and(|extension| extension == "aion");
if path.is_file() && has_aion_extension {
packages.push(path);
}
}
packages.sort_by(|left, right| left.as_os_str().cmp(right.as_os_str()));
Ok(packages)
}
fn merge_workflow_packages(
workflow_packages: &mut Vec<PathBuf>,
discovered_packages: Vec<PathBuf>,
cli_packages: &[PathBuf],
) {
let mut seen: HashSet<PathBuf> = workflow_packages
.iter()
.map(|package| deduplicated_package_key(package))
.collect();
for package in discovered_packages
.into_iter()
.chain(cli_packages.iter().cloned())
{
if seen.insert(deduplicated_package_key(&package)) {
workflow_packages.push(package);
}
}
}
fn deduplicated_package_key(path: &Path) -> PathBuf {
path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}
#[cfg(test)]
mod tests {
use crate::config::{AutoCreate, DEFAULT_MAX_IN_FLIGHT_ACTIVITIES, OpsConsoleAssetSource};
use super::{
CliOverrides, DEFAULT_CLUSTER_BROADCAST_CAPACITY, DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES,
DEFAULT_DEPLOY_MAX_INFLATED_BYTES, DEFAULT_EVENT_BROADCAST_CAPACITY,
DEFAULT_OUTBOX_BACKOFF_BASE_MS, DEFAULT_OUTBOX_BACKOFF_MAX_MS,
DEFAULT_OUTBOX_BACKOFF_MULTIPLIER, DEFAULT_OUTBOX_BATCH_SIZE, DEFAULT_OUTBOX_MAX_ATTEMPTS,
DEFAULT_OUTBOX_POLL_INTERVAL_MS, DEFAULT_QUERY_TIMEOUT_MS, ServerConfig, StoreBackend,
discover_workflow_packages, merge_workflow_packages,
};
#[test]
fn valid_toml_is_parsed_into_typed_config() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[server]
listen_address = "127.0.0.1:18080"
grpc_address = "127.0.0.1:15051"
[store]
backend = "libsql"
url = "aion.db"
[runtime]
scheduler_threads = 2
query_timeout_ms = 10000
[drain]
timeout_seconds = 45
[auth]
enabled = true
jwks_url = "https://issuer.example.com/.well-known/jwks.json"
jwks_refresh_seconds = 60
[metrics]
enabled = true
[namespaces]
default = "production"
[websocket]
outbound_buffer_bound = 16
event_broadcast_capacity = 1024
cluster_broadcast_capacity = 1024
"#,
)?;
assert_eq!(config.store.backend, StoreBackend::LibSql);
assert_eq!(config.store.url.as_deref(), Some("aion.db"));
assert_eq!(config.runtime.scheduler_threads, 2);
assert_eq!(config.runtime.query_timeout_ms, Some(10_000));
assert_eq!(config.namespaces.default, "production");
assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
assert_eq!(
config.namespaces.max_in_flight_activities,
DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
);
assert_eq!(config.websocket.outbound_buffer_bound, 16);
assert_eq!(config.websocket.event_broadcast_capacity, Some(1024));
Ok(())
}
#[test]
fn namespaces_auto_create_closed_parses() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[namespaces]
default = "production"
auto_create = "closed"
"#,
)?;
assert_eq!(config.namespaces.default, "production");
assert_eq!(config.namespaces.auto_create, AutoCreate::Closed);
Ok(())
}
#[test]
fn namespaces_auto_create_open_parses() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[namespaces]
auto_create = "open"
"#,
)?;
assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
Ok(())
}
#[test]
fn namespaces_max_in_flight_activities_override_parses()
-> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[namespaces]
default = "production"
max_in_flight_activities = 32
"#,
)?;
assert_eq!(config.namespaces.max_in_flight_activities, 32);
let (_store, runtime) = config.into_parts();
assert_eq!(runtime.max_in_flight_activities, 32);
Ok(())
}
#[test]
fn namespaces_max_in_flight_activities_defaults_when_omitted()
-> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[namespaces]
default = "production"
"#,
)?;
assert_eq!(
config.namespaces.max_in_flight_activities,
DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
);
Ok(())
}
#[test]
fn namespaces_auto_create_rejects_unknown_variant() {
let result = ServerConfig::from_slice(
br#"
[namespaces]
auto_create = "sometimes"
"#,
);
assert!(
result.is_err(),
"an unknown auto_create variant must fail to parse"
);
}
#[test]
fn missing_event_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
cluster_broadcast_capacity = 64
",
)?;
assert_eq!(
config.websocket.event_broadcast_capacity,
Some(DEFAULT_EVENT_BROADCAST_CAPACITY),
"omitted event_broadcast_capacity must resolve to the default"
);
Ok(())
}
#[test]
fn zero_event_broadcast_capacity_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[websocket]
event_broadcast_capacity = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("websocket.event_broadcast_capacity"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn missing_cluster_broadcast_capacity_uses_default() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
scheduler_threads = 1
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
",
)?;
assert_eq!(
config.websocket.cluster_broadcast_capacity,
Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY),
"omitted cluster_broadcast_capacity must resolve to the default"
);
Ok(())
}
#[test]
fn zero_cluster_broadcast_capacity_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("websocket.cluster_broadcast_capacity"),
"validation message must name the zero-valued cluster key: {message}"
);
}
#[test]
fn missing_observability_section_uses_defaults() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert_eq!(
config.observability.max_event_bytes,
crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
);
assert_eq!(
config.observability.max_stream_events,
crate::config::DEFAULT_OBSERVABILITY_MAX_STREAM_EVENTS
);
let (_store, runtime) = config.into_parts();
assert_eq!(
runtime.observability.max_event_bytes,
crate::config::DEFAULT_OBSERVABILITY_MAX_EVENT_BYTES
);
Ok(())
}
#[test]
fn observability_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[observability]
max_event_bytes = 512
max_stream_events = 3
",
)?;
assert_eq!(config.observability.max_event_bytes, 512);
assert_eq!(config.observability.max_stream_events, 3);
let (_store, runtime) = config.into_parts();
assert_eq!(runtime.observability.max_event_bytes, 512);
assert_eq!(runtime.observability.max_stream_events, 3);
Ok(())
}
#[test]
fn zero_observability_max_event_bytes_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[observability]
max_event_bytes = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("observability.max_event_bytes"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn zero_observability_max_stream_events_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[observability]
max_stream_events = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("observability.max_stream_events"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn missing_query_timeout_uses_default() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
scheduler_threads = 1
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert_eq!(
config.runtime.query_timeout_ms,
Some(DEFAULT_QUERY_TIMEOUT_MS),
"omitted query_timeout_ms must resolve to the default"
);
Ok(())
}
#[test]
fn empty_config_boots_on_operational_defaults() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(b"")?;
assert_eq!(config.store.backend, StoreBackend::Haematite);
assert_eq!(
config.runtime.query_timeout_ms,
Some(DEFAULT_QUERY_TIMEOUT_MS)
);
assert_eq!(
config.websocket.event_broadcast_capacity,
Some(DEFAULT_EVENT_BROADCAST_CAPACITY)
);
assert_eq!(
config.websocket.cluster_broadcast_capacity,
Some(DEFAULT_CLUSTER_BROADCAST_CAPACITY)
);
Ok(())
}
#[test]
fn zero_query_timeout_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 0
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("runtime.query_timeout_ms"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn deploy_enabled_defaults_max_archive_bytes() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
",
)?;
assert_eq!(
config.deploy.max_archive_bytes,
Some(DEFAULT_DEPLOY_MAX_ARCHIVE_BYTES),
"omitted max_archive_bytes must resolve to the conservative default"
);
assert_eq!(
config.deploy.max_inflated_bytes,
Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
"omitted max_inflated_bytes must resolve to the conservative default"
);
Ok(())
}
#[test]
fn deploy_zero_max_archive_bytes_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
max_archive_bytes = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("deploy.max_archive_bytes"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn deploy_enabled_defaults_max_inflated_bytes() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
max_archive_bytes = 16777216
",
)?;
assert_eq!(
config.deploy.max_archive_bytes,
Some(16_777_216),
"explicit max_archive_bytes must be left untouched"
);
assert_eq!(
config.deploy.max_inflated_bytes,
Some(DEFAULT_DEPLOY_MAX_INFLATED_BYTES),
"omitted max_inflated_bytes must resolve to the conservative default"
);
Ok(())
}
#[test]
fn deploy_zero_max_inflated_bytes_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
max_archive_bytes = 16777216
max_inflated_bytes = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("deploy.max_inflated_bytes"),
"validation message must name the zero-valued key: {message}"
);
}
#[test]
fn deploy_max_inflated_below_max_archive_fails_startup_validation() {
let result = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
max_archive_bytes = 16777216
max_inflated_bytes = 16777215
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("deploy.max_inflated_bytes")
&& message.contains("deploy.max_archive_bytes"),
"validation message must name both ceilings: {message}"
);
}
#[test]
fn deploy_disabled_requires_no_archive_ceiling() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert!(!config.deploy.enabled);
assert_eq!(config.deploy.max_archive_bytes, None);
assert_eq!(config.deploy.max_inflated_bytes, None);
Ok(())
}
#[test]
fn deploy_section_parses_enabled_with_ceilings() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[deploy]
enabled = true
max_archive_bytes = 16777216
max_inflated_bytes = 67108864
",
)?;
assert!(config.deploy.enabled);
assert_eq!(config.deploy.max_archive_bytes, Some(16_777_216));
assert_eq!(config.deploy.max_inflated_bytes, Some(67_108_864));
Ok(())
}
#[test]
fn cors_allowed_origins_default_empty() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert!(config.server.cors_allowed_origins.is_empty());
let (_, runtime) = config.into_parts();
assert!(runtime.cors_allowed_origins.is_empty());
Ok(())
}
#[test]
fn cors_allowed_origins_parse_and_round_trip() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[server]
cors_allowed_origins = ["http://localhost:5173", "http://127.0.0.1:5173"]
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
"#,
)?;
assert_eq!(
config.server.cors_allowed_origins,
vec![
"http://localhost:5173".to_owned(),
"http://127.0.0.1:5173".to_owned()
]
);
let (_, runtime) = config.into_parts();
assert_eq!(
runtime.cors_allowed_origins,
vec![
"http://localhost:5173".to_owned(),
"http://127.0.0.1:5173".to_owned()
]
);
Ok(())
}
#[test]
fn cors_allowed_origins_reject_malformed() {
for bad in ["", "localhost:5173", "http://localhost:5173/"] {
let toml = format!(
"[server]\ncors_allowed_origins = [\"{bad}\"]\n\n[runtime]\nquery_timeout_ms = 10000\n\n[websocket]\nevent_broadcast_capacity = 64\n"
);
let result = ServerConfig::from_slice(toml.as_bytes());
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("cors_allowed_origins"),
"malformed origin `{bad}` must be rejected naming the key: {message}"
);
}
}
#[test]
fn dev_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert!(!config.dev.enabled);
Ok(())
}
#[test]
fn dev_section_parses_enabled() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[dev]
enabled = true
",
)?;
assert!(config.dev.enabled);
Ok(())
}
#[test]
fn authoring_absent_leaves_surface_dark() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
assert_eq!(config.authoring.gleam_path, None);
assert_eq!(config.authoring.project_root, None);
Ok(())
}
#[test]
fn authoring_section_parses_and_round_trips() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[authoring]
gleam_path = "/usr/local/bin/gleam"
project_root = "/srv/aion/authoring"
"#,
)?;
assert_eq!(
config.authoring.gleam_path.as_deref(),
Some(std::path::Path::new("/usr/local/bin/gleam"))
);
let (_, runtime) = config.into_parts();
assert_eq!(
runtime.authoring.gleam_path.as_deref(),
Some(std::path::Path::new("/usr/local/bin/gleam"))
);
assert_eq!(
runtime.authoring.project_root.as_deref(),
Some(std::path::Path::new("/srv/aion/authoring"))
);
Ok(())
}
#[test]
fn authoring_gleam_path_without_project_root_fails_naming_key_and_env() {
let result = ServerConfig::from_slice(
br#"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[authoring]
gleam_path = "/usr/local/bin/gleam"
"#,
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("authoring.project_root"),
"validation message must name the missing key: {message}"
);
assert!(
message.contains("AION_AUTHORING_PROJECT_ROOT"),
"validation message must name the environment override: {message}"
);
}
#[test]
fn authoring_empty_gleam_path_fails_naming_key_and_env() {
let result = ServerConfig::from_slice(
br#"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[authoring]
gleam_path = ""
"#,
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(
message.contains("authoring.gleam_path"),
"validation message must name the empty key: {message}"
);
assert!(
message.contains("AION_AUTHORING_GLEAM_PATH"),
"validation message must name the environment override: {message}"
);
}
#[test]
fn cli_overrides_set_authoring_paths() -> Result<(), Box<dyn std::error::Error>> {
let mut config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
",
)?;
let cli = CliOverrides {
gleam_path: Some(std::path::PathBuf::from("/opt/gleam")),
authoring_project_root: Some(std::path::PathBuf::from("/opt/project")),
..CliOverrides::default()
};
config.apply_cli_overrides(&cli);
config.validate()?;
assert_eq!(
config.authoring.gleam_path.as_deref(),
Some(std::path::Path::new("/opt/gleam"))
);
assert_eq!(
config.authoring.project_root.as_deref(),
Some(std::path::Path::new("/opt/project"))
);
Ok(())
}
#[test]
fn legacy_dashboard_section_alias_still_parses() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[dashboard]
source = { FileSystem = { asset_path = "/srv/aion/ui" } }
"#,
)?;
match &config.ops_console.source {
OpsConsoleAssetSource::FileSystem { asset_path } => {
assert_eq!(asset_path.as_os_str(), "/srv/aion/ui");
}
OpsConsoleAssetSource::Embedded => {
return Err("legacy [dashboard] section must map to ops_console".into());
}
}
Ok(())
}
#[test]
fn ops_console_section_parses() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br#"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[ops_console]
source = { FileSystem = { asset_path = "/srv/aion/ui" } }
"#,
)?;
assert!(matches!(
config.ops_console.source,
OpsConsoleAssetSource::FileSystem { .. }
));
Ok(())
}
#[test]
fn invalid_values_name_problematic_field() {
let result = ServerConfig::from_slice(
br"
[runtime]
scheduler_threads = 0
",
);
let message = result
.err()
.map_or_else(String::new, |error| error.to_string());
assert!(message.contains("runtime.scheduler_threads"));
}
#[test]
fn cli_overrides_win_over_loaded_values() -> Result<(), Box<dyn std::error::Error>> {
let mut config = ServerConfig::from_slice(
br#"
[store]
backend = "libsql"
url = "file.db"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
"#,
)?;
let cli = CliOverrides {
store_url: Some("cli.db".to_owned()),
scheduler_threads: Some(3),
..CliOverrides::default()
};
config.apply_cli_overrides(&cli);
config.validate()?;
assert_eq!(config.store.url.as_deref(), Some("cli.db"));
assert_eq!(config.runtime.scheduler_threads, 3);
Ok(())
}
#[test]
fn default_config_defaults() -> Result<(), Box<dyn std::error::Error>> {
let mut config = ServerConfig::default();
assert_eq!(config.store.backend, StoreBackend::Haematite);
assert_eq!(config.store.data_dir.as_deref(), Some("aion-data"));
assert_eq!(config.store.shard_count, 64);
assert_eq!(config.store.url, None);
assert_eq!(config.server.grpc_address.to_string(), "127.0.0.1:50051");
assert_eq!(config.server.listen_address.to_string(), "127.0.0.1:8080");
assert_eq!(config.namespaces.default, "default");
assert_eq!(config.namespaces.auto_create, AutoCreate::Open);
assert_eq!(
config.namespaces.max_in_flight_activities,
DEFAULT_MAX_IN_FLIGHT_ACTIVITIES
);
assert_eq!(config.namespaces.max_in_flight_activities, 1024);
assert!(!config.auth.enabled);
assert!(config.metrics.enabled);
assert_eq!(config.websocket.event_broadcast_capacity, None);
assert_eq!(config.websocket.cluster_broadcast_capacity, None);
assert_eq!(config.runtime.query_timeout_ms, None);
config.websocket.event_broadcast_capacity = Some(64);
config.websocket.cluster_broadcast_capacity = Some(64);
config.runtime.query_timeout_ms = Some(10_000);
config.validate()?;
Ok(())
}
#[test]
fn outbox_is_disabled_by_default_and_needs_no_knobs() -> Result<(), Box<dyn std::error::Error>>
{
let mut config = ServerConfig::default();
config.websocket.event_broadcast_capacity = Some(64);
config.websocket.cluster_broadcast_capacity = Some(64);
config.runtime.query_timeout_ms = Some(10_000);
assert!(!config.outbox.enabled);
assert_eq!(config.outbox.poll_interval_ms, None);
assert_eq!(config.outbox.batch_size, None);
assert_eq!(config.outbox.max_attempts, None);
assert_eq!(config.outbox.backoff_base_ms, None);
assert_eq!(config.outbox.backoff_multiplier, None);
assert_eq!(config.outbox.backoff_max_ms, None);
assert_eq!(config.outbox.reconcile_interval_ms, None);
assert_eq!(config.outbox.reconcile_stale_after_ms, None);
config.validate()?;
Ok(())
}
fn outbox_enabled_base() -> ServerConfig {
let mut config = ServerConfig::default();
config.websocket.event_broadcast_capacity = Some(64);
config.websocket.cluster_broadcast_capacity = Some(64);
config.runtime.query_timeout_ms = Some(10_000);
config.outbox.enabled = true;
config.outbox.poll_interval_ms = Some(250);
config.outbox.batch_size = Some(64);
config.outbox.max_attempts = Some(5);
config.outbox.backoff_base_ms = Some(100);
config.outbox.backoff_multiplier = Some(2);
config.outbox.backoff_max_ms = Some(30_000);
config.outbox.reconcile_interval_ms = Some(1_000);
config.outbox.reconcile_stale_after_ms = Some(60_000);
config
}
#[test]
fn outbox_enabled_with_all_knobs_validates() -> Result<(), Box<dyn std::error::Error>> {
outbox_enabled_base().validate()?;
Ok(())
}
#[test]
fn outbox_enabled_defaults_poll_interval() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[outbox]
enabled = true
",
)?;
assert_eq!(
config.outbox.poll_interval_ms,
Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS),
"omitted poll_interval_ms must resolve to the default"
);
Ok(())
}
#[test]
fn outbox_enabled_defaults_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[outbox]
enabled = true
poll_interval_ms = 250
",
)?;
assert_eq!(
config.outbox.poll_interval_ms,
Some(250),
"explicit poll_interval_ms must be left untouched"
);
assert_eq!(
config.outbox.max_attempts,
Some(DEFAULT_OUTBOX_MAX_ATTEMPTS),
"omitted max_attempts must resolve to the default"
);
Ok(())
}
#[test]
fn outbox_enabled_with_only_enabled_flag_uses_all_defaults()
-> Result<(), Box<dyn std::error::Error>> {
let config = ServerConfig::from_slice(
br"
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
[outbox]
enabled = true
",
)?;
assert!(config.outbox.enabled);
assert_eq!(
config.outbox.poll_interval_ms,
Some(DEFAULT_OUTBOX_POLL_INTERVAL_MS)
);
assert_eq!(config.outbox.batch_size, Some(DEFAULT_OUTBOX_BATCH_SIZE));
assert_eq!(
config.outbox.max_attempts,
Some(DEFAULT_OUTBOX_MAX_ATTEMPTS)
);
assert_eq!(
config.outbox.backoff_base_ms,
Some(DEFAULT_OUTBOX_BACKOFF_BASE_MS)
);
assert_eq!(
config.outbox.backoff_multiplier,
Some(DEFAULT_OUTBOX_BACKOFF_MULTIPLIER)
);
assert_eq!(
config.outbox.backoff_max_ms,
Some(DEFAULT_OUTBOX_BACKOFF_MAX_MS)
);
assert_eq!(config.outbox.reconcile_interval_ms, None);
assert_eq!(config.outbox.reconcile_stale_after_ms, None);
Ok(())
}
#[test]
fn outbox_enabled_zero_poll_interval_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.poll_interval_ms = Some(0);
let error = config
.validate()
.err()
.ok_or("enabled outbox with zero poll interval must fail")?;
assert!(
error.to_string().contains("outbox.poll_interval_ms"),
"error must name the zero-valued key: {error}"
);
Ok(())
}
#[test]
fn outbox_enabled_zero_max_attempts_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.max_attempts = Some(0);
let error = config
.validate()
.err()
.ok_or("enabled outbox with zero max attempts must fail")?;
assert!(
error.to_string().contains("outbox.max_attempts"),
"error must name the zero-valued key: {error}"
);
Ok(())
}
#[test]
fn outbox_backoff_max_below_base_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.backoff_base_ms = Some(1_000);
config.outbox.backoff_max_ms = Some(500);
let error = config
.validate()
.err()
.ok_or("backoff_max below backoff_base must fail")?;
assert!(
error.to_string().contains("outbox.backoff_max_ms"),
"error must name the offending key: {error}"
);
Ok(())
}
#[test]
fn outbox_enabled_can_leave_reconciliation_dark() -> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.reconcile_interval_ms = None;
config.outbox.reconcile_stale_after_ms = None;
config.validate()?;
Ok(())
}
#[test]
fn outbox_reconciliation_requires_interval_when_partially_enabled()
-> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.reconcile_interval_ms = None;
let error = config
.validate()
.err()
.ok_or("reconciliation without interval must fail")?;
assert!(error.to_string().contains("outbox.reconcile_interval_ms"));
Ok(())
}
#[test]
fn outbox_reconciliation_requires_stale_threshold_when_partially_enabled()
-> Result<(), Box<dyn std::error::Error>> {
let mut config = outbox_enabled_base();
config.outbox.reconcile_stale_after_ms = None;
let error = config
.validate()
.err()
.ok_or("reconciliation without stale threshold must fail")?;
assert!(
error
.to_string()
.contains("outbox.reconcile_stale_after_ms")
);
Ok(())
}
#[test]
fn package_discovery_is_sorted() -> Result<(), Box<dyn std::error::Error>> {
let temp_dir = tempfile::tempdir()?;
std::fs::write(temp_dir.path().join("zeta.aion"), b"package")?;
std::fs::write(temp_dir.path().join("alpha.aion"), b"package")?;
std::fs::write(temp_dir.path().join("ignored.txt"), b"package")?;
std::fs::create_dir(temp_dir.path().join("nested"))?;
std::fs::write(
temp_dir.path().join("nested").join("nested.aion"),
b"package",
)?;
let packages = discover_workflow_packages(temp_dir.path())?;
assert_eq!(
packages,
vec![
temp_dir.path().join("alpha.aion"),
temp_dir.path().join("zeta.aion"),
]
);
Ok(())
}
#[test]
fn workflow_package_merge_is_additive_and_deduplicated() {
let mut packages = vec!["config.aion".into(), "shared.aion".into()];
let discovered = vec!["auto.aion".into(), "shared.aion".into()];
let cli = vec!["cli.aion".into(), "auto.aion".into()];
merge_workflow_packages(&mut packages, discovered, &cli);
assert_eq!(
packages,
vec![
std::path::PathBuf::from("config.aion"),
std::path::PathBuf::from("shared.aion"),
std::path::PathBuf::from("auto.aion"),
std::path::PathBuf::from("cli.aion"),
]
);
}
#[test]
fn package_merge_deduplicates_canonical_files() -> Result<(), Box<dyn std::error::Error>> {
let temp_dir = tempfile::tempdir()?;
let package = temp_dir.path().join("hello.aion");
std::fs::write(&package, b"package")?;
let mut packages = vec![package.clone()];
let discovered = vec![temp_dir.path().join(".").join("hello.aion")];
merge_workflow_packages(&mut packages, discovered, &[]);
assert_eq!(packages, vec![package]);
Ok(())
}
#[test]
fn zero_config_cli_workflow_package_uses_in_memory_defaults()
-> Result<(), Box<dyn std::error::Error>> {
let temp_dir = tempfile::tempdir()?;
let cli = CliOverrides {
workflow_packages: vec!["hello-world.aion".into()],
..CliOverrides::default()
};
let mut config = ServerConfig::default();
config.store.backend = StoreBackend::Memory;
config.store.data_dir = None;
config.websocket.event_broadcast_capacity = Some(64);
config.websocket.cluster_broadcast_capacity = Some(64);
config.runtime.query_timeout_ms = Some(10_000);
config.load_discovered_workflow_packages(&cli, temp_dir.path())?;
config.validate()?;
assert_eq!(config.store.backend, StoreBackend::Memory);
assert_eq!(config.store.url, None);
assert_eq!(
config.workflow_packages,
vec![std::path::PathBuf::from("hello-world.aion")]
);
Ok(())
}
#[test]
fn cli_packages_are_additive() -> Result<(), Box<dyn std::error::Error>> {
let mut config = ServerConfig::from_slice(
br#"
workflow_packages = ["config.aion"]
[runtime]
query_timeout_ms = 10000
[websocket]
event_broadcast_capacity = 64
cluster_broadcast_capacity = 64
"#,
)?;
let cli = CliOverrides {
workflow_packages: vec!["cli-one.aion".into(), "cli-two.aion".into()],
..CliOverrides::default()
};
merge_workflow_packages(
&mut config.workflow_packages,
Vec::new(),
&cli.workflow_packages,
);
assert_eq!(
config.workflow_packages,
vec![
std::path::PathBuf::from("config.aion"),
std::path::PathBuf::from("cli-one.aion"),
std::path::PathBuf::from("cli-two.aion"),
]
);
Ok(())
}
}