use std::any::{Any, TypeId};
use std::collections::{BTreeSet, HashMap, HashSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use futures::FutureExt as _;
use tracing::Instrument as _;
use crate::config::{AutumnConfig, ConfigLoader};
#[cfg(feature = "maud")]
use crate::error_pages::{ErrorPageRenderer, SharedRenderer};
use crate::middleware::exception_filter::ExceptionFilter;
#[cfg(feature = "db")]
use crate::migrate;
use crate::route::Route;
use crate::state::AppState;
#[must_use]
pub fn app() -> AppBuilder {
AppBuilder {
routes: Vec::new(),
api_versions: Vec::new(),
route_sources: Vec::new(),
current_plugin: None,
tasks: Vec::new(),
one_off_tasks: Vec::new(),
jobs: Vec::new(),
listeners: Vec::new(),
static_metas: Vec::new(),
exception_filters: Vec::new(),
scoped_groups: Vec::new(),
merge_routers: Vec::new(),
nest_routers: Vec::new(),
custom_layers: Vec::new(),
static_gate_layers: Vec::new(),
startup_hooks: Vec::new(),
state_initializers: Vec::new(),
shutdown_hooks: Vec::new(),
extensions: HashMap::new(),
registered_plugins: HashSet::new(),
plugin_config_roots: BTreeSet::new(),
#[cfg(feature = "maud")]
error_page_renderer: None,
#[cfg(feature = "db")]
migrations: Vec::new(),
config_loader_factory: None,
#[cfg(feature = "db")]
pool_provider_factory: None,
#[cfg(feature = "db")]
shard_provider_factory: None,
#[cfg(feature = "db")]
shard_router: None,
#[cfg(feature = "db")]
directory_shard_router: false,
telemetry_provider: None,
session_store: None,
#[cfg(feature = "ws")]
channels_backend: None,
#[cfg(feature = "storage")]
blob_store: None,
cache_backend: None,
#[cfg(feature = "reporting")]
error_reporters: Vec::new(),
alert_channels: Vec::new(),
#[cfg(feature = "openapi")]
openapi: None,
#[cfg(feature = "mcp")]
mcp: None,
audit_logger: None,
#[cfg(feature = "i18n")]
i18n_bundle: None,
#[cfg(feature = "i18n")]
i18n_auto_load: false,
#[cfg(feature = "embed-assets")]
embedded_static: None,
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
embedded_locales: None,
policy_registrations: Vec::new(),
#[cfg(feature = "mail")]
mail_delivery_queue_factory: None,
#[cfg(feature = "mail")]
suppression_store: None,
#[cfg(feature = "mail")]
mail_suppression_store: None,
#[cfg(feature = "mail")]
mount_unsubscribe_endpoint: false,
#[cfg(feature = "mail")]
mail_previews: Vec::new(),
#[cfg(feature = "maud")]
story_gallery: None,
declared_routes: Vec::new(),
idempotency_enabled: false,
#[cfg(feature = "mail")]
mail_interceptor: None,
job_interceptor: None,
#[cfg(feature = "db")]
db_interceptor: None,
#[cfg(feature = "ws")]
channels_interceptor: None,
#[cfg(feature = "oauth2")]
http_interceptor: None,
seo_sources: Vec::new(),
metrics_sources: Vec::new(),
health_indicators: Vec::new(),
#[cfg(feature = "inbound-mail")]
inbound_mail_router: None,
}
}
fn omitted_router_count<'a>(
merge_routers: usize,
nest_prefixes: impl IntoIterator<Item = &'a str>,
declared_routes: &[crate::route_listing::RouteInfo],
) -> usize {
let uncovered_nests = nest_prefixes
.into_iter()
.filter(|prefix| !nest_prefix_is_covered(prefix, declared_routes))
.count();
merge_routers + uncovered_nests
}
fn nest_prefix_is_covered(
prefix: &str,
declared_routes: &[crate::route_listing::RouteInfo],
) -> bool {
declared_routes
.iter()
.any(|route| path_is_under_prefix(&route.path, prefix))
}
fn path_is_under_prefix(path: &str, prefix: &str) -> bool {
let prefix = prefix.trim_end_matches('/');
if prefix.is_empty() {
return true;
}
path == prefix
|| path
.strip_prefix(prefix)
.is_some_and(|rest| rest.starts_with('/'))
}
type StartupHookFuture = Pin<Box<dyn Future<Output = crate::AutumnResult<()>> + Send>>;
type StartupHook = Box<dyn Fn(AppState) -> StartupHookFuture + Send + Sync>;
type StateInitializer = Box<dyn FnOnce(&AppState) + Send>;
type ShutdownHookFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
type ShutdownHook = Box<dyn Fn() -> ShutdownHookFuture + Send + Sync>;
type ConfigLoaderFactory = Box<
dyn FnOnce() -> Pin<
Box<dyn Future<Output = Result<AutumnConfig, crate::config::ConfigError>> + Send>,
> + Send,
>;
#[cfg(feature = "db")]
type PoolProviderFactory = Box<
dyn FnOnce(
crate::config::DatabaseConfig,
) -> Pin<
Box<
dyn Future<
Output = Result<Option<crate::db::DatabaseTopology>, crate::db::PoolError>,
> + Send,
>,
> + Send,
>;
#[cfg(feature = "db")]
type ShardProviderFactory = Box<
dyn FnOnce(
crate::config::DatabaseConfig,
) -> Pin<
Box<
dyn Future<Output = Result<Vec<crate::db::DatabaseTopology>, crate::db::PoolError>>
+ Send,
>,
> + Send,
>;
type PolicyRegistration = Box<dyn FnOnce(&crate::authorization::PolicyRegistry) + Send>;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
pub struct ApiVersion {
pub version: String,
pub deprecated_at: Option<chrono::DateTime<chrono::Utc>>,
pub sunset_at: Option<chrono::DateTime<chrono::Utc>>,
}
#[derive(Clone, Debug)]
pub struct RegisteredApiVersions(pub Vec<ApiVersion>);
#[allow(clippy::struct_excessive_bools)]
pub struct AppBuilder {
pub(crate) routes: Vec<Route>,
pub api_versions: Vec<ApiVersion>,
route_sources: Vec<crate::route_listing::RouteSource>,
current_plugin: Option<String>,
tasks: Vec<crate::task::TaskInfo>,
one_off_tasks: Vec<crate::task::OneOffTaskInfo>,
pub(crate) jobs: Vec<crate::job::JobInfo>,
pub(crate) listeners: Vec<crate::events::ListenerInfo>,
pub(crate) static_metas: Vec<crate::static_gen::StaticRouteMeta>,
pub(crate) exception_filters: Vec<Arc<dyn ExceptionFilter>>,
pub(crate) scoped_groups: Vec<ScopedGroup>,
pub(crate) merge_routers: Vec<axum::Router<AppState>>,
pub(crate) nest_routers: Vec<(String, axum::Router<AppState>)>,
pub(crate) custom_layers: Vec<CustomLayerRegistration>,
pub(crate) static_gate_layers: Vec<CustomLayerRegistration>,
pub(crate) startup_hooks: Vec<StartupHook>,
pub(crate) state_initializers: Vec<StateInitializer>,
pub(crate) shutdown_hooks: Vec<ShutdownHook>,
pub(crate) extensions: HashMap<TypeId, Box<dyn Any + Send>>,
pub(crate) registered_plugins: HashSet<String>,
pub(crate) plugin_config_roots: BTreeSet<String>,
#[cfg(feature = "maud")]
error_page_renderer: Option<SharedRenderer>,
#[cfg(feature = "db")]
migrations: Vec<migrate::EmbeddedMigrations>,
config_loader_factory: Option<ConfigLoaderFactory>,
#[cfg(feature = "db")]
pool_provider_factory: Option<PoolProviderFactory>,
#[cfg(feature = "db")]
shard_provider_factory: Option<ShardProviderFactory>,
#[cfg(feature = "db")]
shard_router: Option<Arc<dyn crate::sharding::ShardRouter>>,
#[cfg(feature = "db")]
directory_shard_router: bool,
telemetry_provider: Option<Box<dyn crate::telemetry::TelemetryProvider>>,
session_store: Option<Arc<dyn crate::session::BoxedSessionStore>>,
#[cfg(feature = "ws")]
channels_backend: Option<Arc<dyn crate::channels::ChannelsBackend>>,
#[cfg(feature = "storage")]
blob_store: Option<crate::storage::SharedBlobStore>,
cache_backend: Option<Arc<dyn crate::cache::Cache>>,
#[cfg(feature = "reporting")]
pub(crate) error_reporters: Vec<Arc<dyn crate::reporting::ErrorReporter>>,
pub(crate) alert_channels: Vec<Arc<dyn crate::alerts::AlertChannel>>,
#[cfg(feature = "openapi")]
openapi: Option<crate::openapi::OpenApiConfig>,
#[cfg(feature = "mcp")]
mcp: Option<crate::mcp::McpRuntime>,
audit_logger: Option<Arc<crate::audit::AuditLogger>>,
#[cfg(feature = "i18n")]
i18n_bundle: Option<Arc<crate::i18n::Bundle>>,
#[cfg(feature = "i18n")]
i18n_auto_load: bool,
#[cfg(feature = "embed-assets")]
embedded_static: Option<crate::assets::EmbeddedStaticDir>,
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
embedded_locales: Option<&'static include_dir::Dir<'static>>,
policy_registrations: Vec<PolicyRegistration>,
#[cfg(feature = "mail")]
mail_delivery_queue_factory: Option<MailDeliveryQueueFactory>,
#[cfg(feature = "mail")]
pub(crate) suppression_store: Option<crate::mail::SuppressionStoreHandle>,
#[cfg(feature = "mail")]
pub(crate) mail_suppression_store: Option<crate::mail::suppression::SuppressionStoreHandle>,
#[cfg(feature = "mail")]
pub(crate) mount_unsubscribe_endpoint: bool,
#[cfg(feature = "mail")]
mail_previews: Vec<crate::mail::MailPreview>,
#[cfg(feature = "maud")]
story_gallery: Option<crate::stories::StoryGallery>,
declared_routes: Vec<crate::route_listing::RouteInfo>,
idempotency_enabled: bool,
#[cfg(feature = "mail")]
mail_interceptor: Option<Arc<dyn crate::interceptor::MailInterceptor>>,
job_interceptor: Option<Arc<dyn crate::interceptor::JobInterceptor>>,
#[cfg(feature = "db")]
db_interceptor: Option<Arc<dyn crate::interceptor::DbConnectionInterceptor>>,
#[cfg(feature = "ws")]
channels_interceptor: Option<Arc<dyn crate::interceptor::ChannelsInterceptor>>,
#[cfg(feature = "oauth2")]
http_interceptor: Option<Arc<dyn crate::interceptor::HttpInterceptor>>,
seo_sources: Vec<Arc<dyn crate::seo::SitemapSource>>,
pub(crate) metrics_sources: Vec<(String, Arc<dyn crate::actuator::MetricsSource>)>,
pub(crate) health_indicators: Vec<(
String,
crate::actuator::IndicatorGroup,
Arc<dyn crate::actuator::HealthIndicator>,
)>,
#[cfg(feature = "inbound-mail")]
pub(crate) inbound_mail_router: Option<Arc<crate::inbound_mail::InboundMailRouter>>,
}
#[cfg(feature = "mail")]
pub(crate) type MailDeliveryQueueFactory = Box<
dyn FnOnce(&AppState) -> crate::AutumnResult<Arc<dyn crate::mail::MailDeliveryQueue>> + Send,
>;
pub struct ScopedGroup {
pub prefix: String,
pub routes: Vec<Route>,
pub source: crate::route_listing::RouteSource,
pub apply_layer: Box<dyn FnOnce(axum::Router<AppState>) -> axum::Router<AppState> + Send>,
}
pub(crate) type CustomLayerApplier =
Box<dyn FnOnce(axum::Router<AppState>) -> axum::Router<AppState> + Send>;
pub(crate) struct CustomLayerRegistration {
pub(crate) type_id: TypeId,
pub(crate) type_name: &'static str,
pub(crate) apply: CustomLayerApplier,
}
mod sealed {
pub trait Sealed {}
}
#[diagnostic::on_unimplemented(
message = "`{Self}` is not a usable Autumn app-wide Tower layer",
label = "this type does not implement `tower::Layer<axum::routing::Route>` with the required service bounds",
note = "`AppBuilder::layer(..)` requires:\n L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,\n L::Service: Service<axum::extract::Request, Response = axum::response::Response, Error = Infallible> + Clone + Send + Sync + 'static,\n <L::Service as Service<axum::extract::Request>>::Future: Send + 'static\nSee docs/guide/middleware.md for common patterns and how to wrap raw-error layers (e.g. TimeoutLayer) with HandleErrorLayer."
)]
pub trait IntoAppLayer: sealed::Sealed + Send + Sync + 'static {
#[doc(hidden)]
fn apply_to(self, router: axum::Router<AppState>) -> axum::Router<AppState>;
}
impl<L> sealed::Sealed for L
where
L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,
L::Service: tower::Service<
axum::extract::Request,
Response = axum::response::Response,
Error = std::convert::Infallible,
> + Clone
+ Send
+ Sync
+ 'static,
<L::Service as tower::Service<axum::extract::Request>>::Future: Send + 'static,
{
}
impl<L> IntoAppLayer for L
where
L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,
L::Service: tower::Service<
axum::extract::Request,
Response = axum::response::Response,
Error = std::convert::Infallible,
> + Clone
+ Send
+ Sync
+ 'static,
<L::Service as tower::Service<axum::extract::Request>>::Future: Send + 'static,
{
fn apply_to(self, router: axum::Router<AppState>) -> axum::Router<AppState> {
router.layer(self)
}
}
impl AppBuilder {
#[must_use]
pub fn routes(mut self, routes: Vec<Route>) -> Self {
let source = self
.current_plugin
.as_ref()
.map_or(crate::route_listing::RouteSource::User, |name| {
crate::route_listing::RouteSource::Plugin(name.clone())
});
for _ in &routes {
self.route_sources.push(source.clone());
}
self.routes.extend(routes);
self
}
#[must_use]
pub fn tasks(mut self, tasks: Vec<crate::task::TaskInfo>) -> Self {
self.tasks.extend(tasks);
self
}
#[must_use]
pub fn one_off_tasks(mut self, tasks: Vec<crate::task::OneOffTaskInfo>) -> Self {
self.one_off_tasks.extend(tasks);
self
}
#[must_use]
pub fn jobs(mut self, jobs: Vec<crate::job::JobInfo>) -> Self {
self.jobs.extend(jobs);
self
}
#[must_use]
pub fn listeners(mut self, listeners: Vec<crate::events::ListenerInfo>) -> Self {
self.listeners.extend(listeners);
self
}
#[must_use]
pub fn static_routes(mut self, metas: Vec<crate::static_gen::StaticRouteMeta>) -> Self {
self.static_metas.extend(metas);
self
}
#[must_use]
pub fn seo_source<S: crate::seo::SitemapSource + 'static>(mut self, source: S) -> Self {
self.seo_sources.push(Arc::new(source));
self
}
#[cfg(feature = "openapi")]
#[must_use]
pub fn openapi(mut self, config: crate::openapi::OpenApiConfig) -> Self {
self.openapi = Some(config);
self
}
#[cfg(feature = "mcp")]
#[must_use]
pub fn mount_mcp(mut self, path: impl Into<String>) -> Self {
let path = path.into();
if let Some(rt) = self.mcp.as_mut() {
rt.mount_path = path;
} else {
self.mcp = Some(crate::mcp::McpRuntime::new(path));
}
self
}
#[cfg(feature = "mcp")]
#[must_use]
pub fn expose_all_as_mcp(mut self) -> Self {
if let Some(rt) = self.mcp.as_mut() {
rt.expose_all = true;
} else {
let mut rt = crate::mcp::McpRuntime::new("/mcp");
rt.expose_all = true;
self.mcp = Some(rt);
}
self
}
#[cfg(feature = "mcp")]
#[must_use]
pub fn secure_mcp<L>(mut self, layer: L) -> Self
where
L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,
L::Service: tower::Service<
axum::http::Request<axum::body::Body>,
Response = axum::http::Response<axum::body::Body>,
Error = std::convert::Infallible,
> + Clone
+ Send
+ Sync
+ 'static,
<L::Service as tower::Service<axum::http::Request<axum::body::Body>>>::Future:
Send + 'static,
{
let applier: crate::mcp::McpEndpointLayer = Box::new(move |router| router.layer(layer));
if let Some(rt) = self.mcp.as_mut() {
rt.endpoint_layer = Some(applier);
} else {
let mut rt = crate::mcp::McpRuntime::new("/mcp");
rt.endpoint_layer = Some(applier);
self.mcp = Some(rt);
}
self
}
#[must_use]
pub fn exception_filter(mut self, filter: impl ExceptionFilter) -> Self {
self.exception_filters.push(Arc::new(filter));
self
}
#[must_use]
#[cfg(feature = "maud")]
pub fn error_pages(mut self, renderer: impl ErrorPageRenderer) -> Self {
self.error_page_renderer = Some(Arc::new(renderer));
self
}
#[must_use]
pub fn scoped<L>(mut self, prefix: &str, layer: L, routes: Vec<Route>) -> Self
where
L: tower::Layer<axum::routing::Route> + Clone + Send + Sync + 'static,
L::Service: tower::Service<
axum::http::Request<axum::body::Body>,
Response = axum::http::Response<axum::body::Body>,
Error = std::convert::Infallible,
> + Clone
+ Send
+ Sync
+ 'static,
<L::Service as tower::Service<axum::http::Request<axum::body::Body>>>::Future:
Send + 'static,
{
let source = self
.current_plugin
.as_ref()
.map_or(crate::route_listing::RouteSource::User, |name| {
crate::route_listing::RouteSource::Plugin(name.clone())
});
self.scoped_groups.push(ScopedGroup {
prefix: prefix.to_owned(),
routes,
source,
apply_layer: Box::new(move |router| router.layer(layer)),
});
self
}
#[must_use]
pub fn layer<L: IntoAppLayer>(mut self, layer: L) -> Self {
self.custom_layers.push(CustomLayerRegistration {
type_id: TypeId::of::<L>(),
type_name: std::any::type_name::<L>(),
apply: Box::new(move |router| layer.apply_to(router)),
});
self
}
#[must_use]
pub fn has_layer<L: 'static>(&self) -> bool {
let layer_type = TypeId::of::<L>();
self.custom_layers
.iter()
.any(|registered| registered.type_id == layer_type)
}
#[must_use]
pub const fn idempotent(mut self) -> Self {
self.idempotency_enabled = true;
self
}
#[must_use]
pub fn get_layer_types(&self) -> Vec<TypeId> {
self.custom_layers
.iter()
.map(|registered| registered.type_id)
.collect()
}
#[must_use]
pub fn static_gate<L: IntoAppLayer>(mut self, layer: L) -> Self {
self.static_gate_layers.push(CustomLayerRegistration {
type_id: TypeId::of::<L>(),
type_name: std::any::type_name::<L>(),
apply: Box::new(move |router| layer.apply_to(router)),
});
self
}
#[must_use]
pub fn has_static_gate<L: 'static>(&self) -> bool {
let layer_type = TypeId::of::<L>();
self.static_gate_layers
.iter()
.any(|registered| registered.type_id == layer_type)
}
#[must_use]
pub fn get_static_gate_types(&self) -> Vec<TypeId> {
self.static_gate_layers
.iter()
.map(|registered| registered.type_id)
.collect()
}
#[must_use]
pub fn merge(mut self, router: axum::Router<AppState>) -> Self {
self.merge_routers.push(router);
self
}
#[must_use]
pub fn nest(mut self, path: &str, router: axum::Router<AppState>) -> Self {
self.nest_routers.push((path.to_owned(), router));
self
}
#[must_use]
pub fn declare_plugin_routes(
mut self,
routes: impl IntoIterator<Item = crate::route_listing::RouteInfo>,
) -> Self {
let source = self
.current_plugin
.as_deref()
.map_or(crate::route_listing::RouteSource::User, |name| {
crate::route_listing::RouteSource::Plugin(name.to_owned())
});
for mut route in routes {
route.source = source.clone();
self.declared_routes.push(route);
}
self
}
#[must_use]
pub fn on_startup<F, Fut>(mut self, hook: F) -> Self
where
F: Fn(AppState) -> Fut + Send + Sync + 'static,
Fut: Future<Output = crate::AutumnResult<()>> + Send + 'static,
{
self.startup_hooks
.push(Box::new(move |state| Box::pin(hook(state))));
self
}
#[must_use]
pub fn state_initializer<F>(mut self, initializer: F) -> Self
where
F: FnOnce(&AppState) + Send + 'static,
{
self.state_initializers.push(Box::new(initializer));
self
}
#[must_use]
pub fn on_shutdown<F, Fut>(mut self, hook: F) -> Self
where
F: Fn() -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
self.shutdown_hooks.push(Box::new(move || Box::pin(hook())));
self
}
#[must_use]
pub fn api_version(mut self, version: ApiVersion) -> Self {
if let Some(pos) = self
.api_versions
.iter()
.position(|v| v.version == version.version)
{
self.api_versions[pos] = version;
} else {
self.api_versions.push(version);
}
self
}
#[must_use]
pub fn api_versions(mut self, versions: impl IntoIterator<Item = ApiVersion>) -> Self {
for version in versions {
if let Some(pos) = self
.api_versions
.iter()
.position(|v| v.version == version.version)
{
self.api_versions[pos] = version;
} else {
self.api_versions.push(version);
}
}
self
}
#[must_use]
pub fn with_extension<T>(mut self, value: T) -> Self
where
T: Any + Send + 'static,
{
self.extensions.insert(TypeId::of::<T>(), Box::new(value));
self
}
#[must_use]
pub fn update_extension<T, Init, Update>(mut self, init: Init, update: Update) -> Self
where
T: Any + Send + 'static,
Init: FnOnce() -> T,
Update: FnOnce(&mut T),
{
let type_id = TypeId::of::<T>();
let entry = self
.extensions
.entry(type_id)
.or_insert_with(|| Box::new(init()));
let typed = entry
.downcast_mut::<T>()
.expect("extension type map corrupted");
update(typed);
self
}
#[must_use]
pub fn extension<T>(&self) -> Option<&T>
where
T: Any + Send + 'static,
{
self.extensions.get(&TypeId::of::<T>())?.downcast_ref::<T>()
}
#[cfg(feature = "mail")]
#[must_use]
pub fn with_mail_interceptor(
mut self,
interceptor: impl crate::interceptor::MailInterceptor,
) -> Self {
self.mail_interceptor = Some(Arc::new(interceptor));
self
}
#[must_use]
pub fn with_job_interceptor(
mut self,
interceptor: impl crate::interceptor::JobInterceptor,
) -> Self {
self.job_interceptor = Some(Arc::new(interceptor));
self
}
#[cfg(feature = "db")]
#[must_use]
pub fn with_db_interceptor(
mut self,
interceptor: impl crate::interceptor::DbConnectionInterceptor,
) -> Self {
self.db_interceptor = Some(Arc::new(interceptor));
self
}
#[cfg(feature = "ws")]
#[must_use]
pub fn with_channels_interceptor(
mut self,
interceptor: impl crate::interceptor::ChannelsInterceptor,
) -> Self {
self.channels_interceptor = Some(Arc::new(interceptor));
self
}
#[cfg(feature = "oauth2")]
#[must_use]
pub fn with_http_interceptor(
mut self,
interceptor: impl crate::interceptor::HttpInterceptor,
) -> Self {
self.http_interceptor = Some(Arc::new(interceptor));
self
}
#[cfg(feature = "i18n")]
#[must_use]
pub fn i18n(mut self, bundle: crate::i18n::Bundle) -> Self {
self.i18n_bundle = Some(Arc::new(bundle));
self.i18n_auto_load = false;
self
}
#[cfg(feature = "i18n")]
#[must_use]
pub fn i18n_auto(mut self) -> Self {
self.i18n_bundle = None;
self.i18n_auto_load = true;
self
}
#[must_use]
pub fn with_config_loader<L>(mut self, loader: L) -> Self
where
L: crate::config::ConfigLoader,
{
if self.config_loader_factory.is_some() {
tracing::warn!(
"config loader replaced; the previously-installed loader was overwritten"
);
}
self.config_loader_factory = Some(Box::new(move || {
Box::pin(async move { loader.load().await })
}));
self
}
#[cfg(feature = "db")]
#[must_use]
pub fn with_pool_provider<P>(mut self, provider: P) -> Self
where
P: crate::db::DatabasePoolProvider,
{
if self.pool_provider_factory.is_some() {
tracing::warn!(
"database pool provider replaced; the previously-installed provider was overwritten"
);
}
let provider = Arc::new(provider);
let shard_provider = Arc::clone(&provider);
self.pool_provider_factory =
Some(Box::new(move |config: crate::config::DatabaseConfig| {
Box::pin(async move { provider.create_topology(&config).await })
}));
self.shard_provider_factory =
Some(Box::new(move |config: crate::config::DatabaseConfig| {
Box::pin(async move {
let mut topologies = Vec::with_capacity(config.shards.len());
for shard in &config.shards {
topologies
.push(shard_provider.create_shard_topology(shard, &config).await?);
}
Ok(topologies)
})
}));
self
}
#[cfg(feature = "db")]
#[must_use]
pub fn with_shard_router<R>(mut self, router: R) -> Self
where
R: crate::sharding::ShardRouter,
{
if self.shard_router.is_some() {
tracing::warn!(
"shard router replaced; the previously-installed router was overwritten"
);
}
self.shard_router = Some(Arc::new(router));
self
}
#[cfg(feature = "db")]
#[must_use]
pub const fn with_directory_shard_router(mut self) -> Self {
self.directory_shard_router = true;
self
}
#[must_use]
pub fn with_telemetry_provider<T>(mut self, provider: T) -> Self
where
T: crate::telemetry::TelemetryProvider,
{
if self.telemetry_provider.is_some() {
tracing::warn!(
"telemetry provider replaced; the previously-installed provider was overwritten"
);
}
self.telemetry_provider = Some(Box::new(provider));
self
}
#[must_use]
pub fn with_session_store<S>(mut self, store: S) -> Self
where
S: crate::session::SessionStore,
{
if self.session_store.is_some() {
tracing::warn!(
"session store replaced; the previously-installed store was overwritten"
);
}
self.session_store = Some(Arc::new(store));
self
}
#[cfg(feature = "ws")]
#[must_use]
pub fn with_channels_backend<B>(mut self, backend: B) -> Self
where
B: crate::channels::ChannelsBackend,
{
if self.channels_backend.is_some() {
tracing::warn!(
"channels backend replaced; the previously-installed backend was overwritten"
);
}
self.channels_backend = Some(Arc::new(backend));
self
}
#[cfg(feature = "storage")]
#[must_use]
pub fn with_blob_store<B>(mut self, store: B) -> Self
where
B: crate::storage::BlobStore,
{
if self.blob_store.is_some() {
tracing::warn!("blob store replaced; the previously-installed store was overwritten");
}
self.blob_store = Some(std::sync::Arc::new(store));
self
}
#[must_use]
pub fn with_cache_backend<C: crate::cache::Cache>(mut self, cache: C) -> Self {
if self.cache_backend.is_some() {
tracing::warn!(
"cache backend replaced; the previously-installed backend was overwritten"
);
}
self.cache_backend = Some(Arc::new(cache) as Arc<dyn crate::cache::Cache>);
self
}
#[cfg(feature = "reporting")]
#[must_use]
pub fn with_error_reporter<R: crate::reporting::ErrorReporter>(mut self, reporter: R) -> Self {
self.error_reporters
.push(Arc::new(reporter) as Arc<dyn crate::reporting::ErrorReporter>);
self
}
#[must_use]
pub fn with_alert_channel<C: crate::alerts::AlertChannel>(mut self, channel: C) -> Self {
self.alert_channels
.push(Arc::new(channel) as Arc<dyn crate::alerts::AlertChannel>);
self
}
#[must_use]
pub fn with_flag_store<S>(self, store: S) -> Self
where
S: crate::feature_flags::FlagStore,
{
let service = crate::feature_flags::FeatureFlagService::new(Arc::new(store) as Arc<_>);
self.state_initializer(move |state| {
state.insert_extension(service);
})
}
#[must_use]
pub fn with_flag_store_and_resolver<S>(
self,
store: S,
resolver: crate::feature_flags::GroupResolver,
) -> Self
where
S: crate::feature_flags::FlagStore,
{
let service = crate::feature_flags::FeatureFlagService::new(Arc::new(store) as Arc<_>)
.with_group_resolver(resolver);
self.state_initializer(move |state| {
state.insert_extension(service);
})
}
#[must_use]
pub fn with_experiment_store<S>(self, store: S) -> Self
where
S: crate::experiments::ExperimentStore,
{
let service = crate::experiments::ExperimentService::new(Arc::new(store) as Arc<_>);
self.state_initializer(move |state| {
state.insert_extension(service);
})
}
#[must_use]
pub fn with_experiment_store_and_sink<S>(
self,
store: S,
sink: Arc<dyn crate::experiments::ExposureSink>,
) -> Self
where
S: crate::experiments::ExperimentStore,
{
let service = crate::experiments::ExperimentService::new(Arc::new(store) as Arc<_>)
.with_exposure_sink(sink);
self.state_initializer(move |state| {
state.insert_extension(service);
})
}
#[cfg(feature = "mail")]
#[must_use]
pub fn with_mail_delivery_queue(
mut self,
queue: impl crate::mail::MailDeliveryQueue + 'static,
) -> Self {
let arc: Arc<dyn crate::mail::MailDeliveryQueue> = Arc::new(queue);
self.mail_delivery_queue_factory = Some(Box::new(move |_state| Ok(arc)));
self
}
#[cfg(feature = "mail")]
#[must_use]
pub fn with_mail_delivery_queue_factory<F, Q>(mut self, factory: F) -> Self
where
F: FnOnce(&AppState) -> crate::AutumnResult<Q> + Send + 'static,
Q: crate::mail::MailDeliveryQueue + 'static,
{
self.mail_delivery_queue_factory = Some(Box::new(move |state| {
factory(state).map(|q| Arc::new(q) as Arc<dyn crate::mail::MailDeliveryQueue>)
}));
self
}
#[cfg(feature = "mail")]
#[must_use]
pub fn with_suppression_store(
mut self,
store: impl crate::mail::SuppressionStore + 'static,
) -> Self {
self.suppression_store = Some(crate::mail::SuppressionStoreHandle::new(store));
self
}
#[cfg(feature = "mail")]
#[must_use]
pub fn with_mail_suppression_store(
mut self,
store: impl crate::mail::suppression::SuppressionStore + 'static,
) -> Self {
self.mail_suppression_store =
Some(crate::mail::suppression::SuppressionStoreHandle::new(store));
self
}
#[cfg(feature = "mail")]
#[must_use]
pub const fn mount_unsubscribe_endpoint(mut self) -> Self {
self.mount_unsubscribe_endpoint = true;
self
}
#[cfg(feature = "inbound-mail")]
#[must_use]
pub fn inbound_mail_router(mut self, router: crate::inbound_mail::InboundMailRouter) -> Self {
self.inbound_mail_router = Some(Arc::new(router));
self
}
#[cfg(feature = "mail")]
#[must_use]
pub fn mail_previews(
mut self,
previews: impl IntoIterator<Item = crate::mail::MailPreview>,
) -> Self {
self.mail_previews.extend(previews);
self
}
#[cfg(feature = "maud")]
#[must_use]
pub fn with_story_gallery(mut self, gallery: crate::stories::StoryGallery) -> Self {
self.story_gallery = Some(gallery);
self
}
#[must_use]
pub fn with_audit_sink<S>(mut self, sink: S) -> Self
where
S: crate::audit::AuditSink,
{
let logger = self
.audit_logger
.take()
.map_or_else(crate::audit::AuditLogger::new, |logger| (*logger).clone())
.with_sink(Arc::new(sink));
self.audit_logger = Some(Arc::new(logger));
self
}
#[must_use]
pub fn policy<R, P>(mut self, policy: P) -> Self
where
R: Send + Sync + 'static,
P: crate::authorization::Policy<R>,
{
self.policy_registrations.push(Box::new(move |registry| {
registry.register_policy::<R, _>(policy);
}));
self
}
#[must_use]
pub fn scope<R, S>(mut self, scope: S) -> Self
where
R: Send + Sync + 'static,
S: crate::authorization::Scope<R>,
{
self.policy_registrations.push(Box::new(move |registry| {
registry.register_scope::<R, _>(scope);
}));
self
}
#[must_use]
#[track_caller]
pub fn plugin<P>(mut self, plugin: P) -> Self
where
P: crate::plugin::Plugin,
{
let name = plugin.name();
if self.registered_plugins.contains(name.as_ref()) {
tracing::warn!(
plugin = name.as_ref(),
"plugin already registered; skipping duplicate"
);
return self;
}
let name_str = name.into_owned();
self.registered_plugins.insert(name_str.clone());
let outer_plugin = self.current_plugin.replace(name_str);
let mut result = plugin.build(self);
result.current_plugin = outer_plugin;
result
}
#[must_use]
pub fn plugins<P>(self, plugins: P) -> Self
where
P: crate::plugin::Plugins,
{
plugins.apply(self)
}
#[must_use]
pub fn has_plugin(&self, name: &str) -> bool {
self.registered_plugins.contains(name)
}
#[must_use]
pub fn config_section(mut self, root: impl Into<String>) -> Self {
self.plugin_config_roots.insert(root.into());
self
}
#[must_use]
pub fn has_config_section(&self, root: &str) -> bool {
self.plugin_config_roots.contains(root)
}
#[must_use]
pub fn metrics_source(
mut self,
name: impl Into<String>,
source: Arc<dyn crate::actuator::MetricsSource>,
) -> Self {
let name = name.into();
if self.metrics_sources.iter().any(|(n, _)| n == &name) {
tracing::warn!(
source_name = %name,
"MetricsSource '{}' is already registered; skipping duplicate",
name
);
return self;
}
self.metrics_sources.push((name, source));
self
}
#[must_use]
pub fn health_indicator(
mut self,
name: impl Into<String>,
indicator: Arc<dyn crate::actuator::HealthIndicator>,
) -> Self {
let name = name.into();
#[cfg(feature = "db")]
if name == "db" || name.starts_with("db:shard:") {
tracing::warn!(
indicator_name = %name,
"\"db\" and \"db:shard:*\" are reserved built-in health indicator names; \
registration skipped. Use a different name for your custom indicator."
);
return self;
}
if self.health_indicators.iter().any(|(n, _, _)| n == &name) {
tracing::warn!(
indicator_name = %name,
"HealthIndicator '{}' is already registered; skipping duplicate",
name
);
return self;
}
let group = indicator.group();
self.health_indicators.push((name, group, indicator));
self
}
#[cfg(feature = "db")]
#[must_use]
pub fn migrations(mut self, migrations: migrate::EmbeddedMigrations) -> Self {
self.migrations.push(migrations);
self
}
#[cfg(feature = "embed-assets")]
#[must_use]
pub const fn embedded_static(mut self, dir: &'static include_dir::Dir<'static>) -> Self {
self.embedded_static = Some(crate::assets::EmbeddedStaticDir(dir));
self
}
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
#[must_use]
pub const fn embedded_locales(mut self, dir: &'static include_dir::Dir<'static>) -> Self {
self.embedded_locales = Some(dir);
self
}
#[allow(clippy::too_many_lines)]
#[allow(clippy::cognitive_complexity)]
pub async fn run(self) {
if is_static_build_mode() {
self.run_build_mode().await;
return;
}
if is_dump_routes_mode() {
self.run_dump_routes_mode().await;
return;
}
if is_dump_jobs_mode() {
self.run_dump_jobs_mode().await;
return;
}
if is_list_one_off_tasks_mode() {
self.run_list_one_off_tasks_mode();
return;
}
if let Some(task_name) = one_off_task_name_from_env() {
self.run_one_off_task_mode(task_name).await;
return;
}
if is_migrate_only_mode() {
self.run_migrate_only_mode().await;
return;
}
let Self {
routes,
api_versions,
route_sources: _,
current_plugin: _,
tasks,
one_off_tasks: _,
mut jobs,
listeners,
static_metas,
exception_filters,
scoped_groups,
merge_routers,
nest_routers,
custom_layers,
static_gate_layers,
startup_hooks,
state_initializers,
shutdown_hooks,
extensions: _,
registered_plugins: _,
plugin_config_roots,
#[cfg(feature = "maud")]
error_page_renderer,
#[cfg(feature = "db")]
migrations,
config_loader_factory,
#[cfg(feature = "db")]
pool_provider_factory,
#[cfg(feature = "db")]
shard_provider_factory,
#[cfg(feature = "db")]
shard_router,
#[cfg(feature = "db")]
directory_shard_router,
telemetry_provider,
session_store,
#[cfg(feature = "ws")]
channels_backend,
#[cfg(feature = "storage")]
blob_store,
cache_backend,
#[cfg(feature = "reporting")]
error_reporters,
alert_channels,
#[cfg(feature = "openapi")]
openapi,
#[cfg(feature = "mcp")]
mcp,
audit_logger,
#[cfg(feature = "i18n")]
i18n_bundle,
#[cfg(feature = "i18n")]
i18n_auto_load,
#[cfg(feature = "embed-assets")]
embedded_static,
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
embedded_locales,
policy_registrations,
#[cfg(feature = "mail")]
mail_delivery_queue_factory,
#[cfg(feature = "mail")]
suppression_store,
#[cfg(feature = "mail")]
mail_suppression_store,
#[cfg(feature = "mail")]
mount_unsubscribe_endpoint,
#[cfg(feature = "mail")]
mail_previews,
#[cfg(feature = "maud")]
story_gallery,
declared_routes: _,
idempotency_enabled,
#[cfg(feature = "mail")]
mail_interceptor,
job_interceptor,
#[cfg(feature = "db")]
db_interceptor,
#[cfg(feature = "ws")]
channels_interceptor,
#[cfg(feature = "oauth2")]
http_interceptor,
seo_sources,
metrics_sources,
health_indicators,
#[cfg(feature = "inbound-mail")]
inbound_mail_router,
} = self;
let all_routes = routes;
let (mut config, telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
let role = config.role;
if crate::config::split_role_requires_durable_backend(role, &config.jobs.backend) {
tracing::error!(
role = role.as_str(),
jobs_backend = %config.jobs.backend,
"process role '{}' requires a durable jobs backend: backend '{}' is not \
a recognized durable backend and falls through to the in-process 'local' \
runtime, which cannot be shared across a split web/worker topology. \
Set jobs.backend = \"postgres\" or \"redis\", or run the combined role.",
role.as_str(),
config.jobs.backend,
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(feature = "mail")]
if mount_unsubscribe_endpoint {
config.mail.mount_unsubscribe_endpoint = true;
}
if idempotency_enabled {
let env_disabled = std::env::var("AUTUMN_IDEMPOTENCY__ENABLED")
.is_ok_and(|v| matches!(v.to_lowercase().as_str(), "false" | "0" | "no" | "off"));
if !env_disabled && config.idempotency.enabled != Some(false) {
config.idempotency.enabled = Some(true);
}
}
#[cfg(feature = "embed-assets")]
register_embedded_static_dir(embedded_static);
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
let i18n_bundle = embedded_i18n_bundle(i18n_bundle, embedded_locales, &config);
#[cfg(feature = "i18n")]
let i18n_bundle =
resolve_i18n_bundle(i18n_bundle, i18n_auto_load, &config, &crate::config::OsEnv);
assert!(
!all_routes.is_empty(),
"No routes registered. Did you forget to call .routes()?"
);
let profile_display = config.profile.as_deref().unwrap_or("none");
tracing::info!(
version = env!("CARGO_PKG_VERSION"),
profile = profile_display,
"Autumn starting"
);
let show_config = std::env::var("AUTUMN_SHOW_CONFIG").as_deref() == Ok("1");
if show_config {
log_startup_transparency(&all_routes, &tasks, &scoped_groups, &config);
}
fail_fast_on_invalid_session_config(&config, session_store.is_some());
fail_fast_on_invalid_signing_secret(&config);
fail_fast_on_missing_encryption_keys(&config);
fail_fast_on_invalid_trusted_hosts(&config);
fail_fast_on_invalid_webhook_config(&config);
fail_fast_on_invalid_idempotency_config(&config);
#[cfg(feature = "storage")]
let storage_bootstrap = blob_store.map_or_else(
|| preflight_storage(&config),
|store| {
Some(StorageBootstrap {
store,
serving: None,
})
},
);
#[cfg(feature = "db")]
let database = setup_database(
&config,
migrations,
pool_provider_factory,
shard_provider_factory,
shard_router,
directory_shard_router,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.unwrap_or_else(|e| {
tracing::error!("{e}");
std::process::exit(1);
});
#[cfg(feature = "db")]
let pool = database.topology;
#[cfg(feature = "db")]
let shards = database.shards;
#[cfg(feature = "db")]
let replica_readiness = database.replica_readiness;
#[cfg(feature = "db")]
let replica_migration_check = database.replica_migration_check;
#[cfg(feature = "db")]
if pool.is_some() || shards.is_some() {
let shard_max_connections = shards
.as_ref()
.map_or(0, crate::sharding::ShardSet::total_max_connections);
let control_max_connections = pool.as_ref().map_or(0, |topology| {
topology.primary().status().max_size
+ topology.replica().map_or(0, |p| p.status().max_size)
});
let total_max_connections = control_max_connections + shard_max_connections;
tracing::info!(
primary_max_connections = config.database.effective_primary_pool_size(),
replica_configured = config.database.replica_url.is_some(),
replica_max_connections = config.database.effective_replica_pool_size(),
shard_count = shards.as_ref().map_or(0, crate::sharding::ShardSet::len),
total_max_connections,
"Database topology configured"
);
let warn_threshold = config.database.max_connections_warn_threshold;
if crate::config::should_warn_total_connections(total_max_connections, warn_threshold) {
tracing::warn!(
total_max_connections,
warn_threshold,
"Aggregate database connection count is high: the control \
topology and all shard pools together may open \
{total_max_connections} connections (warn threshold \
{warn_threshold}). Ensure each Postgres server's \
max_connections (plus headroom for migrations and \
psql) exceeds the pools that target it, or lower \
database.pool_size. Set \
database.max_connections_warn_threshold = 0 to silence."
);
}
} else {
tracing::info!("Database not configured");
}
validate_repository_api_policies(&all_routes, &scoped_groups, &config);
let mut state = build_state(
&config,
#[cfg(feature = "db")]
pool.as_ref(),
#[cfg(feature = "db")]
shards,
#[cfg(feature = "ws")]
channels_backend,
);
if let Some(buf) = telemetry_guard.log_buffer.clone() {
state.insert_extension(buf);
}
if let Some(handle) = telemetry_guard.filter_reload.clone() {
state.log_levels().attach_reload_handle(handle);
}
let maintenance_state = crate::maintenance::MaintenanceState::new();
let flag_path = std::path::Path::new(crate::maintenance::MAINTENANCE_FLAG_FILE);
if let Ok(Some(cfg)) = crate::maintenance::MaintenanceState::load_from_file(flag_path) {
maintenance_state.enable(cfg);
}
state.insert_extension(maintenance_state.clone());
let poller_state = maintenance_state.clone();
tokio::spawn(async move {
let path = std::path::Path::new(crate::maintenance::MAINTENANCE_FLAG_FILE);
let interval = std::time::Duration::from_millis(500);
loop {
let load_res = tokio::task::spawn_blocking(move || {
crate::maintenance::MaintenanceState::load_from_file(path)
})
.await;
match load_res {
Ok(Ok(Some(cfg))) => {
if poller_state.get() != Some(cfg.clone()) {
poller_state.enable(cfg);
}
}
Ok(Ok(None)) => {
if poller_state.is_active() {
poller_state.disable();
}
}
Ok(Err(e)) => {
tracing::error!(error = %e, "failed to load maintenance flag file");
}
Err(e) => {
tracing::error!(error = %e, "maintenance poller task panicked");
}
}
tokio::time::sleep(interval).await;
}
});
let canary_state = crate::canary::CanaryState::from_env();
if canary_state.is_canary() {
tracing::info!(
version = canary_state.version(),
"canary: replica labelled as canary cohort"
);
}
state.insert_extension(canary_state);
if crate::canary::CanaryState::rollback_flag_present(std::path::Path::new(
crate::canary::CANARY_ROLLBACK_FLAG_FILE,
)) {
tracing::warn!(
"canary: rollback flag present at startup; /ready will report draining until \
the flag is cleared (`autumn canary promote`)"
);
state.begin_shutdown();
}
#[cfg(feature = "mail")]
if let Some(interceptor) = mail_interceptor {
state.insert_extension(interceptor);
}
if let Some(interceptor) = job_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "db")]
if let Some(interceptor) = db_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "ws")]
if let Some(interceptor) = channels_interceptor {
state.insert_extension(interceptor.clone());
state.channels = crate::channels::Channels::with_shared_backend(std::sync::Arc::new(
crate::channels::InterceptedChannelsBackend::new(
state.channels.backend().clone(),
vec![interceptor],
),
));
#[cfg(feature = "presence")]
{
state.presence = crate::presence::Presence::new(state.channels.clone());
}
}
#[cfg(feature = "oauth2")]
if let Some(interceptor) = http_interceptor {
state.insert_extension(interceptor);
}
for (name, source) in metrics_sources {
if let Err(e) = state.metrics_source_registry.register(name, source) {
tracing::warn!("{e}");
}
}
for (name, group, indicator) in health_indicators {
if let Err(e) = state
.health_indicator_registry
.register(name, group, indicator)
{
tracing::warn!("{e}");
}
}
#[cfg(feature = "acme")]
let acme_status: Option<crate::acme::renewal::AcmeStatus> = if let Some(acme_cfg) =
config.server.tls.as_ref().and_then(|t| t.acme.as_ref())
{
let status = crate::acme::renewal::AcmeStatus::new();
let indicator = std::sync::Arc::new(crate::acme::renewal::AcmeHealthIndicator::new(
status.clone(),
acme_cfg.renew_before_days,
));
if let Err(e) = state.health_indicator_registry.register(
"acme",
crate::actuator::IndicatorGroup::HealthOnly,
indicator,
) {
tracing::warn!("{e}");
}
Some(status)
} else {
None
};
#[cfg(feature = "db")]
configure_replica_migration_check(&state, replica_migration_check);
#[cfg(feature = "db")]
apply_replica_migration_readiness(&state, replica_readiness);
if let Some(cache) = cache_backend {
crate::cache::set_global_cache(cache.clone());
state.shared_cache = Some(cache);
} else {
crate::cache::clear_global_cache();
}
state.insert_extension(RegisteredApiVersions(api_versions));
#[cfg(all(feature = "acme", feature = "reporting"))]
let acme_reporters = error_reporters.clone();
#[cfg(feature = "reporting")]
if !error_reporters.is_empty() {
state.insert_extension(crate::reporting::RegisteredReporters(error_reporters));
}
for register in policy_registrations {
register(state.policy_registry());
}
validate_repository_policies_registered(&all_routes, &scoped_groups, &state, &config);
#[cfg(feature = "mail")]
if let Some(handle) = suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
if let Some(handle) = mail_suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
crate::mail::install_mailer_with_factory(
&state,
&config.mail,
mail_delivery_queue_factory,
true,
)
.unwrap_or_else(|error| {
tracing::error!(error = %error, "Failed to configure mailer");
exit_stop_managed_pg();
std::process::exit(1);
});
#[cfg(feature = "mail")]
state.insert_extension(crate::mail::MailPreviewRegistry::new(mail_previews));
#[cfg(feature = "maud")]
install_story_registry(&state, story_gallery);
crate::alerts::install_from_config(&state, &config.alerts, alert_channels);
if let Some(logger) = audit_logger {
state.insert_extension::<crate::audit::AuditLogger>((*logger).clone());
}
#[cfg(feature = "i18n")]
let custom_layers = install_i18n_bundle_layer(custom_layers, &state, i18n_bundle);
#[cfg(feature = "storage")]
let storage_router = storage_bootstrap.and_then(|b| b.install(&state));
install_webhook_registry(&state, &config);
run_state_initializers(state_initializers, &state);
finalize_event_bus(listeners, &mut jobs, &state);
let env = crate::config::OsEnv;
let dist_dir = project_dir("dist", &env);
let dist_ref = if dist_dir.exists() {
Some(dist_dir.as_path())
} else {
None
};
#[cfg_attr(
not(any(feature = "storage", feature = "inbound-mail")),
allow(unused_mut)
)]
let mut merge_routers = merge_routers;
#[cfg(feature = "storage")]
if let Some(router) = storage_router {
merge_routers.push(router);
}
if !seo_sources.is_empty() || crate::seo::has_seo_config(&config.seo) {
let seo_cfg = &config.seo;
let raw_profile = config.profile.as_deref().unwrap_or("dev");
let profile = crate::seo::effective_seo_profile(raw_profile, seo_cfg.robots.allow_all);
let static_paths: Vec<&str> = static_metas.iter().map(|m| m.path).collect();
let (robots_body, sitemap_body) = crate::seo::assemble_seo_bodies(
profile,
seo_cfg.base_url.as_deref(),
seo_cfg.robots.sitemap_url.as_deref(),
&seo_cfg.robots.additional_rules,
&seo_sources,
&static_paths,
)
.await;
let seo_router = crate::seo::build_seo_router_from_bodies(robots_body, sitemap_body);
let is_seo_path = |p: &str| p == "/robots.txt" || p == "/sitemap.xml";
let seo_collision = all_routes.iter().any(|r| is_seo_path(r.path))
|| static_metas.iter().any(|m| is_seo_path(m.path))
|| scoped_groups.iter().any(|g| {
let prefix = g.prefix.trim_end_matches('/');
g.routes
.iter()
.any(|r| is_seo_path(&format!("{prefix}{}", r.path)))
});
if seo_collision {
tracing::warn!(
"seo: /robots.txt or /sitemap.xml is already registered by the application; \
skipping automatic SEO routes to prevent a startup panic"
);
} else {
merge_routers.push(seo_router);
}
}
#[cfg(feature = "inbound-mail")]
if let Some(ref im_router) = inbound_mail_router {
let mut registered_inbound: std::collections::HashSet<String> =
std::collections::HashSet::new();
for (path, axum_router) in crate::inbound_mail::build_routes(im_router) {
if all_routes
.iter()
.any(|r| r.method == http::Method::POST && r.path == path)
|| scoped_groups.iter().any(|g| {
g.routes.iter().any(|r| {
r.method == http::Method::POST
&& crate::router::join_nested_path(&g.prefix, r.path)
== path.as_str()
})
})
|| nest_routers.iter().any(|(nest_path, _)| {
let p = nest_path.as_str();
path.as_str() == p
|| path.starts_with(p)
&& (p.ends_with('/') || path.as_bytes().get(p.len()) == Some(&b'/'))
})
{
tracing::warn!(
path = %path,
"inbound_mail: skipping webhook route — a POST handler is \
already registered at this path by the application"
);
continue;
}
if !registered_inbound.insert(path.clone()) {
tracing::warn!(
path = %path,
"inbound_mail: skipping duplicate inbound webhook path"
);
continue;
}
config.security.csrf.exempt_paths.push(path.clone());
config.security.captcha_exempt_paths.push(path);
merge_routers.push(axum_router);
}
}
let router_build = if role.serves_http() {
crate::router::try_build_router_with_static_inner(
all_routes,
&config,
state.clone(),
dist_ref,
crate::router::RouterContext {
exception_filters,
scoped_groups,
merge_routers,
nest_routers,
custom_layers,
static_gate_layers,
#[cfg(feature = "maud")]
error_page_renderer,
session_store,
#[cfg(feature = "openapi")]
openapi: if config.openapi_runtime.enabled {
openapi
} else {
None
},
#[cfg(feature = "mcp")]
mcp,
},
)
} else {
crate::router::try_build_probe_only_router(&config, state.clone())
};
let router = router_build.unwrap_or_else(|error| {
tracing::error!(error = %error, "Failed to build router");
exit_stop_managed_pg();
std::process::exit(1);
});
if let Some(tls_cfg) = config.server.tls.as_ref() {
if let Err(msg) = tls_cfg.validate() {
tracing::error!("Invalid [server.tls] configuration: {msg}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(not(feature = "tls"))]
{
tracing::error!(
"[server.tls] is configured but this binary was built without the `tls` \
feature; rebuild with `--features tls`, or remove [server.tls] to serve \
plain HTTP"
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(all(feature = "tls", not(feature = "acme")))]
if tls_cfg.acme.is_some() {
tracing::error!(
"[server.tls.acme] is configured but this binary was built without the \
`acme` feature; rebuild with `--features acme`, or configure a static \
cert_path/key_path instead"
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(feature = "tls")]
if config.server.unix_socket.is_some() {
tracing::error!(
"[server.tls] cannot be combined with server.unix_socket; direct TLS \
terminates on host:port. Unset one of them"
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
let server_shutdown = tokio_util::sync::CancellationToken::new();
#[cfg(feature = "tls")]
let mut tls_reload_state: Option<TlsReloadState> = None;
#[cfg(feature = "acme")]
let mut acme_bind_state: Option<AcmeBindState> = None;
let (bound_listener, bound_desc, unix_socket_cleanup): (
BoundListener,
String,
Option<(std::path::PathBuf, u64, u64)>,
) = if let Some(socket_path) = config.server.unix_socket.as_deref() {
let _ = socket_path;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let path = std::path::Path::new(socket_path);
if let Err(e) = prepare_unix_socket_path(path) {
tracing::error!(socket = %socket_path, "Failed to prepare unix socket: {e}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
let bind_result = {
static UMASK_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
let _umask_guard = UMASK_LOCK
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let prev_umask =
nix::sys::stat::umask(nix::sys::stat::Mode::from_bits_truncate(0o177));
let result = tokio::net::UnixListener::bind(path);
nix::sys::stat::umask(prev_umask);
result
};
let listener = match bind_result {
Ok(listener) => listener,
Err(e) => {
tracing::error!(socket = %socket_path, "Failed to bind unix socket: {e}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
};
if let Err(e) =
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
{
tracing::error!(socket = %socket_path, "Failed to enforce owner-only permissions on unix socket: {e}");
let _ = std::fs::remove_file(path);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
let (dev, ino) = {
use std::os::unix::fs::MetadataExt;
std::fs::metadata(path).map_or((0, 0), |m| (m.dev(), m.ino()))
};
(
BoundListener::Unix(listener),
format!("unix:{socket_path}"),
Some((path.to_path_buf(), dev, ino)),
)
}
#[cfg(not(unix))]
{
tracing::error!(
"server.unix_socket is only supported on Unix platforms; \
unset it or use server.host/server.port"
);
std::process::exit(1);
}
} else {
let addr = format!("{}:{}", config.server.host, config.server.port);
let listener = match tokio::net::TcpListener::bind(&addr).await {
Ok(listener) => listener,
Err(e) => {
tracing::error!(addr = %addr, "Failed to bind: {e}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
};
#[cfg(feature = "tls")]
{
if let Some(tls_cfg) = config.server.tls.as_ref() {
#[cfg(feature = "acme")]
if let Some(acme_cfg) = tls_cfg.acme.as_ref() {
let https_port = config.server.port;
match build_acme_tls_listener(
listener,
tls_cfg,
acme_cfg,
https_port,
acme_status.clone(),
server_shutdown.child_token(),
)
.await
{
Ok((tls_listener, bind_state)) => {
acme_bind_state = Some(bind_state);
(
BoundListener::Tls(tls_listener),
format!("https://{addr} (ACME)"),
None,
)
}
Err(e) => {
tracing::error!(error = %e, "Failed to configure [server.tls.acme]");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
} else {
match build_tls_listener(listener, tls_cfg, server_shutdown.child_token()) {
Ok((tls_listener, reload)) => {
tls_reload_state = Some(reload);
(
BoundListener::Tls(tls_listener),
format!("https://{addr}"),
None,
)
}
Err(e) => {
tracing::error!(error = %e, "Failed to configure [server.tls]");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
}
#[cfg(not(feature = "acme"))]
match build_tls_listener(listener, tls_cfg, server_shutdown.child_token()) {
Ok((tls_listener, reload)) => {
tls_reload_state = Some(reload);
(
BoundListener::Tls(tls_listener),
format!("https://{addr}"),
None,
)
}
Err(e) => {
tracing::error!(error = %e, "Failed to configure [server.tls]");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
} else {
(BoundListener::Tcp(listener), addr, None)
}
}
#[cfg(not(feature = "tls"))]
{
(BoundListener::Tcp(listener), addr, None)
}
};
let shutdown_timeout = config.server.shutdown_timeout_secs;
let prestop_grace = config.server.prestop_grace_secs;
if let Err(error) = initialize_job_runtime(
jobs,
&state,
&server_shutdown,
&config.jobs,
role.runs_workers(),
) {
tracing::error!(error = %error, "job runtime initialization failed");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(feature = "db")]
{
#[cfg(feature = "ws")]
crate::repository_commit_hooks::set_global_channels(state.channels().clone());
}
#[cfg(all(feature = "db", not(feature = "sqlite")))]
if role.runs_workers()
&& let Some(pool) = state.pool().cloned()
{
#[cfg(feature = "ws")]
{
let channels = state.channels().clone();
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
Some(channels),
server_shutdown.child_token(),
);
}
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
server_shutdown.child_token(),
);
}
#[cfg(all(feature = "db", not(feature = "sqlite")))]
if role.runs_workers()
&& let Some(shards) = state.shards()
{
for shard in shards.iter() {
#[cfg(feature = "ws")]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
shard.primary_pool().clone(),
Some(state.channels().clone()),
server_shutdown.child_token(),
);
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
shard.primary_pool().clone(),
server_shutdown.child_token(),
);
}
}
#[cfg(all(feature = "db", feature = "sqlite"))]
if role.runs_workers()
&& let Some(pool) = state.pool().cloned()
{
#[cfg(feature = "ws")]
{
let channels = state.channels().clone();
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
Some(channels),
server_shutdown.child_token(),
);
}
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
server_shutdown.child_token(),
);
}
#[cfg(feature = "presence")]
{
let presence = state.presence().clone();
let sweep_shutdown = server_shutdown.child_token();
tokio::spawn(async move {
let interval = std::time::Duration::from_secs(15);
loop {
tokio::select! {
() = tokio::time::sleep(interval) => {
presence.sweep_expired();
}
() = sweep_shutdown.cancelled() => break,
}
}
});
}
#[cfg(feature = "tls")]
if let Some(reload) = tls_reload_state.take() {
let reload_shutdown = server_shutdown.child_token();
tokio::spawn(async move {
run_tls_cert_reload(reload, reload_shutdown).await;
});
}
#[cfg(feature = "acme")]
if let Some(bind_state) = acme_bind_state.take() {
let AcmeBindState {
mut renewal_task,
tokens,
http_challenge_port,
https_port,
} = bind_state;
let challenge_listeners =
match crate::acme::challenge::bind_challenge_listeners(http_challenge_port).await {
Ok(listeners) => listeners,
Err(e) => {
tracing::error!(
port = http_challenge_port,
"Failed to bind the ACME HTTP-01 challenge listener: {e}. Port \
{http_challenge_port} typically needs privilege (grant \
CAP_NET_BIND_SERVICE), or set [server.tls.acme] http_challenge_port \
to a port a front-end forwards :80 to"
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
};
let challenge_router = crate::acme::challenge::challenge_router(tokens, https_port);
for challenge_listener in challenge_listeners {
let router = challenge_router.clone();
let challenge_shutdown = server_shutdown.child_token();
tokio::spawn(async move {
if let Err(e) = axum::serve(challenge_listener, router)
.with_graceful_shutdown(async move {
challenge_shutdown.cancelled().await;
})
.await
{
tracing::error!(
error = %e,
"ACME challenge listener stopped with an error"
);
}
});
}
let mut leadership_degraded = false;
let coordinator =
match crate::scheduler::coordinator_from_config(&config.scheduler, &state) {
Ok(c) => c,
Err(e) => {
tracing::warn!(
error = %e,
"ACME renewal: falling back to an in-process coordinator"
);
leadership_degraded = !matches!(
config.scheduler.backend,
crate::config::SchedulerBackend::InProcess
);
std::sync::Arc::new(crate::scheduler::InProcessSchedulerCoordinator::new(
config.scheduler.resolved_replica_id(),
))
}
};
renewal_task.leadership_degraded = leadership_degraded;
if !matches!(
config.scheduler.backend,
crate::config::SchedulerBackend::InProcess
) {
tracing::warn!(
scheduler_backend = coordinator.backend(),
"ACME HTTP-01 validation is not fleet-safe with the local on-disk token \
store: behind a load balancer the CA's :80 challenge may reach a replica \
without the token (404), and non-leader replicas cannot adopt issued \
certificates from a non-shared store. Run ACME on a single host, or use a \
shared token store / DNS-01 (#1620)"
);
}
#[cfg(feature = "reporting")]
let reporter = make_acme_reporter(acme_reporters);
#[cfg(not(feature = "reporting"))]
let reporter = make_acme_reporter();
let renewal_shutdown = server_shutdown.child_token();
tokio::spawn(async move {
renewal_task
.run(coordinator, reporter, renewal_shutdown)
.await;
});
}
tracing::info!(bound = %bound_desc, "Listening");
let server_shutdown_wait = server_shutdown.clone();
let after_method = tower::Layer::layer(
&crate::middleware::MethodOverrideLayer::new()
.with_max_scan_bytes(config.security.upload.max_request_size_bytes),
router,
);
let service = tower::Layer::layer(
&crate::security::TrustedProxiesLayer::from_config(&config.security.trusted_proxies),
after_method,
);
let server_task = match bound_listener {
BoundListener::Tcp(listener) => {
let make_service =
axum::ServiceExt::<axum::extract::Request>::into_make_service_with_connect_info::<
std::net::SocketAddr,
>(service);
tokio::spawn(async move {
axum::serve(listener, make_service)
.with_graceful_shutdown(async move {
server_shutdown_wait.cancelled().await;
})
.await
})
}
#[cfg(unix)]
BoundListener::Unix(listener) => {
let service = tower::Layer::layer(
&axum::middleware::from_fn(stamp_loopback_connect_info),
service,
);
let make_service =
axum::ServiceExt::<axum::extract::Request>::into_make_service_with_connect_info::<
UdsConnectInfo,
>(service);
tokio::spawn(async move {
axum::serve(listener, make_service)
.with_graceful_shutdown(async move {
server_shutdown_wait.cancelled().await;
})
.await
})
}
#[cfg(feature = "tls")]
BoundListener::Tls(listener) => {
use axum::serve::ListenerExt as _;
let listener = listener.tap_io(|_io| {});
let make_service =
axum::ServiceExt::<axum::extract::Request>::into_make_service_with_connect_info::<
std::net::SocketAddr,
>(service);
tokio::spawn(async move {
axum::serve(listener, make_service)
.with_graceful_shutdown(async move {
server_shutdown_wait.cancelled().await;
})
.await
})
}
};
let shutdown_state = state.clone();
let shutdown_signal_token = server_shutdown.clone();
#[cfg(feature = "ws")]
let websocket_shutdown = state.shutdown.clone();
let shutdown_metrics = state.metrics.clone();
let drain_started_at: std::sync::Arc<std::sync::OnceLock<std::time::Instant>> =
std::sync::Arc::new(std::sync::OnceLock::new());
let drain_started_clone = std::sync::Arc::clone(&drain_started_at);
let drain_phase_notify = std::sync::Arc::new(tokio::sync::Notify::new());
let drain_phase_notify_for_watchdog = std::sync::Arc::clone(&drain_phase_notify);
let server_entered_drain = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let server_entered_drain_for_watchdog = std::sync::Arc::clone(&server_entered_drain);
let shutdown_task = tokio::spawn(async move {
shutdown_signal().await;
tracing::info!(
phase = "signal_received",
prestop_grace_secs = prestop_grace,
shutdown_timeout_secs = shutdown_timeout,
"shutdown: graceful shutdown initiated"
);
shutdown_state.begin_shutdown();
tracing::info!(phase = "ready_draining", "shutdown: /ready now 503");
if prestop_grace > 0 {
tokio::time::sleep(std::time::Duration::from_secs(prestop_grace)).await;
}
tracing::info!(phase = "listener_stopping", "shutdown: stopping listener");
#[cfg(feature = "ws")]
websocket_shutdown.cancel();
let _ = drain_started_clone.set(std::time::Instant::now());
shutdown_signal_token.cancel();
if !server_entered_drain_for_watchdog.load(std::sync::atomic::Ordering::Acquire) {
tracing::warn!(
phase = "signal_during_startup",
"shutdown: SIGTERM during startup hooks; waiting for drain phase \
to begin before enforcing the drain deadline"
);
drain_phase_notify_for_watchdog.notified().await;
}
tokio::time::sleep(std::time::Duration::from_secs(shutdown_timeout)).await;
if shutdown_metrics.snapshot().http.requests_active == 0 {
return;
}
let aborted = shutdown_metrics.snapshot().http.requests_active;
shutdown_metrics.record_shutdown_aborted(aborted);
tracing::error!(
phase = "in_flight_drain",
timeout_secs = shutdown_timeout,
autumn_shutdown_aborted_requests_total = aborted,
exit_code = 1,
"shutdown: in_flight_drain phase exceeded deadline; terminating"
);
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
});
if let Err(error) = run_startup_hooks(&startup_hooks, state.clone()).await {
tracing::error!(error = %error, "startup hook failed");
server_shutdown.cancel();
server_task.abort();
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
if !state.probes().is_shutting_down() {
if role.runs_workers() && !tasks.is_empty() {
let res = start_task_scheduler_with_config(
tasks,
&state,
&server_shutdown,
&config.scheduler,
);
if let Err(err) = res {
tracing::error!(error = %err, "scheduled task runtime initialization failed");
server_shutdown.cancel();
server_task.abort();
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
state.probes().mark_startup_complete();
signal_serve_ready(
config
.server
.prestop_grace_secs
.saturating_add(config.server.shutdown_timeout_secs),
);
}
server_entered_drain.store(true, std::sync::atomic::Ordering::Release);
drain_phase_notify.notify_one();
let server_result = server_task.await.unwrap_or_else(|e| {
tracing::error!("Server task join error: {e}");
exit_stop_managed_pg();
std::process::exit(1);
});
shutdown_task.abort();
server_result.unwrap_or_else(|e| {
tracing::error!("Server error: {e}");
exit_stop_managed_pg();
std::process::exit(1);
});
let drain_elapsed = drain_started_at
.get()
.map_or(std::time::Duration::ZERO, std::time::Instant::elapsed);
let hook_budget =
std::time::Duration::from_secs(shutdown_timeout).saturating_sub(drain_elapsed);
run_shutdown_hooks_with_timeout(&shutdown_hooks, hook_budget, hook_budget).await;
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
#[cfg(unix)]
if let Some((path, dev, ino)) = &unix_socket_cleanup {
use std::os::unix::fs::MetadataExt;
let still_ours =
std::fs::metadata(path).is_ok_and(|m| m.dev() == *dev && m.ino() == *ino);
if still_ours {
let _ = std::fs::remove_file(path);
}
}
#[cfg(not(unix))]
let _ = &unix_socket_cleanup;
tracing::info!(exit_code = 0, "shutdown: all phases completed cleanly");
}
#[allow(clippy::too_many_lines)]
async fn run_build_mode(self) {
let Self {
routes,
api_versions,
route_sources: _,
current_plugin: _,
tasks: _,
one_off_tasks: _,
jobs: _,
listeners,
static_metas,
exception_filters: _,
scoped_groups,
merge_routers: _,
nest_routers: _,
custom_layers,
static_gate_layers: _,
startup_hooks: _,
state_initializers,
shutdown_hooks: _,
extensions: _,
registered_plugins: _,
plugin_config_roots,
#[cfg(feature = "maud")]
error_page_renderer: _,
#[cfg(feature = "db")]
migrations: _,
config_loader_factory,
#[cfg(feature = "db")]
pool_provider_factory,
#[cfg(feature = "db")]
shard_provider_factory,
#[cfg(feature = "db")]
shard_router,
#[cfg(feature = "db")]
directory_shard_router,
telemetry_provider,
session_store,
#[cfg(feature = "ws")]
channels_backend,
#[cfg(feature = "storage")]
blob_store,
cache_backend,
#[cfg(feature = "reporting")]
error_reporters,
alert_channels: _,
#[cfg(feature = "openapi")]
openapi,
#[cfg(feature = "mcp")]
mcp: _,
audit_logger: _,
#[cfg(feature = "i18n")]
i18n_bundle,
#[cfg(feature = "i18n")]
i18n_auto_load,
#[cfg(feature = "embed-assets")]
embedded_static,
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
embedded_locales,
policy_registrations,
#[cfg(feature = "mail")]
mail_delivery_queue_factory,
#[cfg(feature = "mail")]
suppression_store,
#[cfg(feature = "mail")]
mail_suppression_store,
#[cfg(feature = "mail")]
mount_unsubscribe_endpoint,
#[cfg(feature = "mail")]
mail_previews,
#[cfg(feature = "maud")]
story_gallery,
declared_routes: _,
idempotency_enabled,
#[cfg(feature = "mail")]
mail_interceptor,
job_interceptor,
#[cfg(feature = "db")]
db_interceptor,
#[cfg(feature = "ws")]
channels_interceptor,
#[cfg(feature = "oauth2")]
http_interceptor,
seo_sources,
metrics_sources,
health_indicators,
#[cfg(feature = "inbound-mail")]
inbound_mail_router: _,
} = self;
let _ = &api_versions;
let _ = &metrics_sources;
let _ = &health_indicators;
let all_routes = routes;
let (mut config, telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
#[cfg(feature = "mail")]
if mount_unsubscribe_endpoint {
config.mail.mount_unsubscribe_endpoint = true;
}
if idempotency_enabled {
let env_disabled = std::env::var("AUTUMN_IDEMPOTENCY__ENABLED")
.is_ok_and(|v| matches!(v.to_lowercase().as_str(), "false" | "0" | "no" | "off"));
if !env_disabled && config.idempotency.enabled != Some(false) {
config.idempotency.enabled = Some(true);
}
}
#[cfg(feature = "embed-assets")]
register_embedded_static_dir(embedded_static);
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
let i18n_bundle = embedded_i18n_bundle(i18n_bundle, embedded_locales, &config);
#[cfg(feature = "i18n")]
let i18n_bundle =
resolve_i18n_bundle(i18n_bundle, i18n_auto_load, &config, &crate::config::OsEnv);
#[cfg(feature = "openapi")]
let api_docs_snapshot: Vec<crate::openapi::ApiDoc> = {
let mut docs: Vec<crate::openapi::ApiDoc> = all_routes
.iter()
.map(|r| {
let mut doc = r.api_doc.clone();
doc.api_version = r.api_version;
doc.sunset_opt_out = r.sunset_opt_out;
doc
})
.collect();
for group in &scoped_groups {
let prefix_params = crate::router::extract_path_params(&group.prefix);
for route in &group.routes {
let mut doc = route.api_doc.clone();
doc.api_version = route.api_version;
doc.sunset_opt_out = route.sunset_opt_out;
let full = crate::router::join_nested_path(&group.prefix, route.api_doc.path);
doc.path = Box::leak(full.into_boxed_str());
if !prefix_params.is_empty() {
let mut merged: Vec<&'static str> = prefix_params
.iter()
.map(|p| &*Box::leak(p.clone().into_boxed_str()))
.collect();
merged.extend_from_slice(doc.path_params);
doc.path_params = Box::leak(merged.into_boxed_slice());
}
docs.push(doc);
}
}
docs
};
if static_metas.is_empty() {
eprintln!("No static routes registered. Nothing to build.");
eprintln!("Hint: use .static_routes(static_routes![...]) on your AppBuilder.");
std::process::exit(1);
}
fail_fast_on_invalid_session_config(&config, session_store.is_some());
fail_fast_on_invalid_signing_secret(&config);
fail_fast_on_missing_encryption_keys(&config);
fail_fast_on_invalid_trusted_hosts(&config);
#[cfg(feature = "storage")]
let storage_bootstrap = blob_store.map_or_else(
|| preflight_storage(&config),
|store| {
Some(StorageBootstrap {
store,
serving: None,
})
},
);
#[cfg(feature = "db")]
let database = setup_database(
&config,
vec![],
pool_provider_factory,
shard_provider_factory,
shard_router,
directory_shard_router,
RepositoryCommitHookQueueMigrationMode::StaticBuild,
)
.await
.unwrap_or_else(|e| {
eprintln!("{e}");
std::process::exit(1);
});
#[cfg(feature = "db")]
let pool = database.topology;
#[cfg(feature = "db")]
let shards = database.shards;
#[cfg(feature = "db")]
let replica_readiness = database.replica_readiness;
#[cfg(feature = "db")]
let replica_migration_check = database.replica_migration_check;
let mut state = build_state(
&config,
#[cfg(feature = "db")]
pool.as_ref(),
#[cfg(feature = "db")]
shards,
#[cfg(feature = "ws")]
channels_backend,
);
if let Some(buf) = telemetry_guard.log_buffer.clone() {
state.insert_extension(buf);
}
if let Some(handle) = telemetry_guard.filter_reload.clone() {
state.log_levels().attach_reload_handle(handle);
}
state.insert_extension(RegisteredApiVersions(api_versions.clone()));
#[cfg(feature = "mail")]
if let Some(interceptor) = mail_interceptor {
state.insert_extension(interceptor);
}
if let Some(interceptor) = job_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "db")]
if let Some(interceptor) = db_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "ws")]
if let Some(interceptor) = channels_interceptor {
state.insert_extension(interceptor.clone());
state.channels = crate::channels::Channels::with_shared_backend(std::sync::Arc::new(
crate::channels::InterceptedChannelsBackend::new(
state.channels.backend().clone(),
vec![interceptor],
),
));
#[cfg(feature = "presence")]
{
state.presence = crate::presence::Presence::new(state.channels.clone());
}
}
#[cfg(feature = "oauth2")]
if let Some(interceptor) = http_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "db")]
configure_replica_migration_check(&state, replica_migration_check);
#[cfg(feature = "db")]
apply_replica_migration_readiness(&state, replica_readiness);
if let Some(cache) = cache_backend {
crate::cache::set_global_cache(cache.clone());
state.shared_cache = Some(cache);
} else {
crate::cache::clear_global_cache();
}
#[cfg(feature = "reporting")]
if !error_reporters.is_empty() {
state.insert_extension(crate::reporting::RegisteredReporters(error_reporters));
}
#[cfg(feature = "mail")]
if let Some(handle) = suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
if let Some(handle) = mail_suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
crate::mail::install_mailer_with_factory(
&state,
&config.mail,
mail_delivery_queue_factory,
false,
)
.unwrap_or_else(|error| {
eprintln!("Failed to configure mailer: {error}");
exit_stop_managed_pg();
std::process::exit(1);
});
#[cfg(feature = "mail")]
state.insert_extension(crate::mail::MailPreviewRegistry::new(mail_previews));
#[cfg(feature = "maud")]
install_story_registry(&state, story_gallery);
state.probes = crate::probe::ProbeState::default();
for register in policy_registrations {
register(state.policy_registry());
}
#[cfg(feature = "i18n")]
let custom_layers = install_i18n_bundle_layer(custom_layers, &state, i18n_bundle);
#[cfg(feature = "storage")]
let storage_router = storage_bootstrap.and_then(|b| b.install(&state));
install_webhook_registry(&state, &config);
run_state_initializers(state_initializers, &state);
let sync_listeners: Vec<_> = listeners
.into_iter()
.filter(|listener| listener.mode == crate::events::DispatchMode::Sync)
.collect();
finalize_event_bus(sync_listeners, &mut Vec::new(), &state);
#[cfg_attr(not(feature = "storage"), allow(unused_mut))]
let mut merge_routers: Vec<axum::Router<AppState>> = Vec::new();
#[cfg(feature = "storage")]
if let Some(router) = storage_router {
merge_routers.push(router);
}
let router = crate::router::try_build_router_inner(
all_routes,
&config,
state,
crate::router::RouterContext {
exception_filters: Vec::new(),
scoped_groups,
merge_routers,
nest_routers: Vec::new(),
custom_layers,
static_gate_layers: Vec::new(),
#[cfg(feature = "maud")]
error_page_renderer: None,
session_store,
#[cfg(feature = "openapi")]
openapi: None,
#[cfg(feature = "mcp")]
mcp: None,
},
)
.unwrap_or_else(|error| {
eprintln!("Failed to build router: {error}");
exit_stop_managed_pg();
std::process::exit(1);
});
let env = crate::config::OsEnv;
let dist_dir = project_dir("dist", &env);
eprintln!("Building {} static route(s)...", static_metas.len());
match crate::static_gen::render_static_routes(router, &static_metas, &dist_dir).await {
Ok(()) => {
eprintln!(
"\n \u{2713} Static build complete \u{2192} {}",
dist_dir.display()
);
}
Err(e) => {
eprintln!("\n \u{2717} Static build failed: {e}");
exit_stop_managed_pg();
std::process::exit(1);
}
}
#[cfg(feature = "openapi")]
if let Some(mut openapi_config) = openapi {
openapi_config.api_versions = api_versions;
let openapi_config =
openapi_config.session_cookie_name(config.session.cookie_name.clone());
let docs: Vec<&crate::openapi::ApiDoc> = api_docs_snapshot.iter().collect();
let spec = crate::openapi::generate_spec(&openapi_config, &docs);
match crate::openapi::write_openapi_spec_to_dist(&spec, &dist_dir) {
Ok(()) => {
eprintln!(
" \u{2713} OpenAPI spec written \u{2192} {}/openapi.json",
dist_dir.display()
);
}
Err(e) => {
eprintln!(" \u{26A0} Failed to write OpenAPI spec: {e}");
}
}
}
if !seo_sources.is_empty() || crate::seo::has_seo_config(&config.seo) {
let seo_cfg = &config.seo;
let raw_profile = config.profile.as_deref().unwrap_or("dev");
let profile = crate::seo::effective_seo_profile(raw_profile, seo_cfg.robots.allow_all);
let static_paths: Vec<&str> = static_metas.iter().map(|m| m.path).collect();
let (robots_body, sitemap_body) = crate::seo::assemble_seo_bodies(
profile,
seo_cfg.base_url.as_deref(),
seo_cfg.robots.sitemap_url.as_deref(),
&seo_cfg.robots.additional_rules,
&seo_sources,
&static_paths,
)
.await;
let robots_path = dist_dir.join("robots.txt");
let sitemap_path = dist_dir.join("sitemap.xml");
if robots_path.exists() {
eprintln!(
" \u{2713} SEO: robots.txt already present (custom static route), skipping"
);
} else {
match tokio::fs::write(&robots_path, robots_body).await {
Ok(()) => eprintln!(
" \u{2713} SEO: robots.txt written \u{2192} {}",
robots_path.display()
),
Err(e) => eprintln!(" \u{26A0} Failed to write robots.txt: {e}"),
}
}
if sitemap_path.exists() {
eprintln!(
" \u{2713} SEO: sitemap.xml already present (custom static route), skipping"
);
} else {
match tokio::fs::write(&sitemap_path, sitemap_body).await {
Ok(()) => eprintln!(
" \u{2713} SEO: sitemap.xml written \u{2192} {}",
sitemap_path.display()
),
Err(e) => eprintln!(" \u{26A0} Failed to write sitemap.xml: {e}"),
}
}
}
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
}
#[allow(clippy::too_many_lines)]
async fn run_dump_routes_mode(self) {
let Self {
routes,
api_versions,
route_sources,
scoped_groups,
merge_routers,
nest_routers,
declared_routes,
config_loader_factory,
telemetry_provider,
#[cfg(feature = "openapi")]
openapi,
plugin_config_roots,
..
} = self;
let registered_versions: std::collections::HashSet<&str> =
api_versions.iter().map(|av| av.version.as_str()).collect();
for route in &routes {
if let Some(ver) = route
.api_version
.filter(|ver| !registered_versions.contains(*ver))
{
eprintln!(
"Failed to build router: route '{}' uses unregistered API version '{}'",
route.name, ver
);
std::process::exit(1);
}
}
for group in &scoped_groups {
for route in &group.routes {
if let Some(ver) = route
.api_version
.filter(|ver| !registered_versions.contains(*ver))
{
eprintln!(
"Failed to build router: route '{}' uses unregistered API version '{}'",
route.name, ver
);
std::process::exit(1);
}
}
}
let hidden = omitted_router_count(
merge_routers.len(),
nest_routers.iter().map(|(prefix, _)| prefix.as_str()),
&declared_routes,
);
if hidden > 0 {
eprintln!(
"[autumn routes] warning: {hidden} raw router(s) added via \
.merge()/.nest() are not enumerable and are omitted from this listing"
);
eprintln!(
"{marker}{hidden}",
marker = crate::route_listing::OMITTED_ROUTES_MARKER
);
}
let (config, _telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
if is_dump_security_mode() {
let security = crate::route_listing::SecurityDump::from_config(&config);
match serde_json::to_string(&security) {
Ok(json) => eprintln!(
"{marker}{json}",
marker = crate::route_listing::SECURITY_CONFIG_MARKER
),
Err(e) => eprintln!("Failed to serialize security config: {e}"),
}
}
let mut infos = match crate::route_listing::collect_route_infos(
&routes,
&route_sources,
&scoped_groups,
&api_versions,
) {
Ok(infos) => infos,
Err(e) => {
eprintln!("Failed to build router: {e}");
std::process::exit(1);
}
};
infos.extend(declared_routes);
crate::route_listing::append_framework_routes(&mut infos, &config);
#[cfg(feature = "openapi")]
if let Some(ref oa) = openapi {
crate::route_listing::append_openapi_routes(&mut infos, oa);
}
crate::route_listing::append_dev_reload_routes(&mut infos);
crate::route_listing::sort_route_infos(&mut infos);
let json = serde_json::to_string_pretty(&infos).unwrap_or_else(|e| {
eprintln!("Failed to serialize route listing: {e}");
std::process::exit(1);
});
println!("{json}");
std::process::exit(0);
}
async fn run_dump_jobs_mode(self) {
let Self {
jobs,
listeners,
config_loader_factory,
telemetry_provider,
plugin_config_roots,
..
} = self;
let (config, _telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
let manifest = dump_jobs_manifest(&config.jobs.queues, jobs, listeners);
print!("{manifest}");
std::process::exit(0);
}
fn run_list_one_off_tasks_mode(self) {
let Self { one_off_tasks, .. } = self;
if let Err(error) = crate::task::validate_unique_one_off_task_names(&one_off_tasks) {
eprintln!("Invalid task registration: {error}");
std::process::exit(1);
}
let listing = crate::task::list_one_off_tasks(&one_off_tasks);
let json = serde_json::to_string_pretty(&listing).unwrap_or_else(|error| {
eprintln!("Failed to serialize task listing: {error}");
std::process::exit(1);
});
println!("{json}");
std::process::exit(0);
}
#[cfg(feature = "db")]
async fn run_migrate_only_mode(self) {
let Self {
migrations,
config_loader_factory,
telemetry_provider,
plugin_config_roots,
..
} = self;
let (config, _telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
let migrations = migrations_with_repository_framework_migrations(
migrations,
crate::repository_commit_hooks::has_repository_commit_hook_descriptors(),
crate::version_history::has_versioned_repository_descriptors(),
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let control_url = config.database.effective_primary_url().map(str::to_owned);
let shard_targets: Vec<(String, String)> = config
.database
.shards
.iter()
.map(|shard| (format!("shard:{}", shard.name), shard.primary_url.clone()))
.collect();
if migrations.is_empty() || (control_url.is_none() && shard_targets.is_empty()) {
eprintln!(
"autumn migrate: no database configured or no migrations registered — nothing to apply"
);
std::process::exit(0);
}
#[cfg(feature = "sqlite")]
{
let sqlite_guard_shard_urls: Vec<&str> =
shard_targets.iter().map(|(_, url)| url.as_str()).collect();
if let Err(e) = sqlite_sharding_unsupported_guard(
control_url.as_deref(),
!shard_targets.is_empty(),
&sqlite_guard_shard_urls,
) {
eprintln!("autumn migrate: {e}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop();
std::process::exit(1);
}
}
let applied_total = tokio::task::spawn_blocking(move || {
let mut total = 0_usize;
if let Some(url) = &control_url {
#[cfg(feature = "sqlite")]
let is_sqlite_control = crate::config::DatabaseBackend::detect(url)
== Some(crate::config::DatabaseBackend::Sqlite);
#[cfg(not(feature = "sqlite"))]
let is_sqlite_control = false;
if is_sqlite_control {
#[cfg(feature = "sqlite")]
for mig in &migrations {
total += apply_pending_sqlite_or_exit(url, mig, "control");
}
} else {
for mig in &migrations {
total += apply_pending_or_exit(url, mig, "control");
}
}
}
for (label, url) in &shard_targets {
for mig in migrations
.iter()
.filter(|mig| !migration_set_is_control_framework(mig))
{
total += apply_pending_or_exit(url, mig, label);
}
}
total
})
.await
.unwrap_or_else(|error| {
eprintln!("autumn migrate: migration task panicked: {error}");
std::process::exit(1);
});
eprintln!(
"autumn migrate: applied {applied_total} pending migration(s); database is up to date"
);
std::process::exit(0);
}
#[cfg(not(feature = "db"))]
#[allow(clippy::unused_async)]
async fn run_migrate_only_mode(self) {
eprintln!("autumn migrate: this build has no database support — nothing to migrate");
std::process::exit(0);
}
#[allow(clippy::too_many_lines)]
#[allow(clippy::cognitive_complexity)]
async fn run_one_off_task_mode(self, requested_name: String) {
let Self {
one_off_tasks,
mut jobs,
listeners,
#[cfg(feature = "i18n")]
custom_layers,
#[cfg(not(feature = "i18n"))]
custom_layers: _,
startup_hooks,
state_initializers,
shutdown_hooks,
config_loader_factory,
#[cfg(feature = "db")]
migrations,
#[cfg(feature = "db")]
pool_provider_factory,
#[cfg(feature = "db")]
shard_provider_factory,
#[cfg(feature = "db")]
shard_router,
#[cfg(feature = "db")]
directory_shard_router,
telemetry_provider,
session_store,
#[cfg(feature = "ws")]
channels_backend,
#[cfg(feature = "storage")]
blob_store,
audit_logger,
#[cfg(feature = "i18n")]
i18n_bundle,
#[cfg(feature = "i18n")]
i18n_auto_load,
#[cfg(feature = "embed-assets")]
embedded_static,
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
embedded_locales,
policy_registrations,
cache_backend,
#[cfg(feature = "mail")]
mail_delivery_queue_factory,
#[cfg(feature = "mail")]
suppression_store,
#[cfg(feature = "mail")]
mail_suppression_store,
#[cfg(feature = "mail")]
mount_unsubscribe_endpoint: _,
#[cfg(feature = "mail")]
mail_interceptor,
job_interceptor,
#[cfg(feature = "db")]
db_interceptor,
#[cfg(feature = "ws")]
channels_interceptor,
#[cfg(feature = "oauth2")]
http_interceptor,
plugin_config_roots,
..
} = self;
if let Err(error) = crate::task::validate_unique_one_off_task_names(&one_off_tasks) {
eprintln!("Invalid task registration: {error}");
std::process::exit(1);
}
let Some((task_name, task_handler)) = one_off_tasks
.iter()
.find(|task| task.name == requested_name)
.map(|task| (task.name.clone(), task.handler))
else {
eprintln!("No one-off task named '{requested_name}' is registered.");
print_available_one_off_tasks(&one_off_tasks);
std::process::exit(1);
};
let args = one_off_task_args_from_env().unwrap_or_else(|error| {
eprintln!("Invalid task args: {error}");
std::process::exit(1);
});
let (config, telemetry_guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
#[cfg(feature = "embed-assets")]
register_embedded_static_dir(embedded_static);
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
let i18n_bundle = embedded_i18n_bundle(i18n_bundle, embedded_locales, &config);
#[cfg(feature = "i18n")]
let i18n_bundle =
resolve_i18n_bundle(i18n_bundle, i18n_auto_load, &config, &crate::config::OsEnv);
fail_fast_on_invalid_session_config(&config, session_store.is_some());
fail_fast_on_invalid_signing_secret(&config);
fail_fast_on_missing_encryption_keys(&config);
fail_fast_on_invalid_trusted_hosts(&config);
#[cfg(feature = "storage")]
let storage_bootstrap = blob_store.map_or_else(
|| preflight_storage(&config),
|store| {
Some(StorageBootstrap {
store,
serving: None,
})
},
);
#[cfg(feature = "db")]
let database = setup_database(
&config,
migrations,
pool_provider_factory,
shard_provider_factory,
shard_router,
directory_shard_router,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.unwrap_or_else(|error| {
eprintln!("{error}");
std::process::exit(1);
});
#[cfg(feature = "db")]
let pool = database.topology;
#[cfg(feature = "db")]
let shards = database.shards;
#[cfg(feature = "db")]
let replica_readiness = database.replica_readiness;
#[cfg(feature = "db")]
let replica_migration_check = database.replica_migration_check;
let mut state = build_state(
&config,
#[cfg(feature = "db")]
pool.as_ref(),
#[cfg(feature = "db")]
shards,
#[cfg(feature = "ws")]
channels_backend,
);
if let Some(buf) = telemetry_guard.log_buffer.clone() {
state.insert_extension(buf);
}
if let Some(handle) = telemetry_guard.filter_reload.clone() {
state.log_levels().attach_reload_handle(handle);
}
#[cfg(feature = "mail")]
if let Some(interceptor) = mail_interceptor {
state.insert_extension(interceptor);
}
if let Some(interceptor) = job_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "db")]
if let Some(interceptor) = db_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "ws")]
if let Some(interceptor) = channels_interceptor {
state.insert_extension(interceptor.clone());
state.channels = crate::channels::Channels::with_shared_backend(std::sync::Arc::new(
crate::channels::InterceptedChannelsBackend::new(
state.channels.backend().clone(),
vec![interceptor],
),
));
#[cfg(feature = "presence")]
{
state.presence = crate::presence::Presence::new(state.channels.clone());
}
}
#[cfg(feature = "oauth2")]
if let Some(interceptor) = http_interceptor {
state.insert_extension(interceptor);
}
#[cfg(feature = "db")]
configure_replica_migration_check(&state, replica_migration_check);
#[cfg(feature = "db")]
apply_replica_migration_readiness(&state, replica_readiness);
if let Some(cache) = cache_backend {
crate::cache::set_global_cache(cache.clone());
state.shared_cache = Some(cache);
} else {
crate::cache::clear_global_cache();
}
for register in policy_registrations {
register(state.policy_registry());
}
#[cfg(feature = "mail")]
if let Some(handle) = suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
if let Some(handle) = mail_suppression_store {
state.insert_extension(handle);
}
#[cfg(feature = "mail")]
crate::mail::install_mailer_with_factory(
&state,
&config.mail,
mail_delivery_queue_factory,
true,
)
.unwrap_or_else(|error| {
eprintln!("Failed to configure mailer: {error}");
exit_stop_managed_pg();
std::process::exit(1);
});
if let Some(logger) = audit_logger {
state.insert_extension::<crate::audit::AuditLogger>((*logger).clone());
}
#[cfg(feature = "i18n")]
let _custom_layers = install_i18n_bundle_layer(custom_layers, &state, i18n_bundle);
#[cfg(feature = "storage")]
let _storage_router = storage_bootstrap.and_then(|bootstrap| bootstrap.install(&state));
run_state_initializers(state_initializers, &state);
finalize_event_bus(listeners, &mut jobs, &state);
let task_shutdown = tokio_util::sync::CancellationToken::new();
if let Err(error) = initialize_job_runtime(jobs, &state, &task_shutdown, &config.jobs, true)
{
eprintln!("job runtime initialization failed: {error}");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
#[cfg(feature = "db")]
{
#[cfg(feature = "ws")]
crate::repository_commit_hooks::set_global_channels(state.channels().clone());
}
#[cfg(all(feature = "db", not(feature = "sqlite")))]
if let Some(pool) = state.pool().cloned() {
#[cfg(feature = "ws")]
{
let channels = state.channels().clone();
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
Some(channels),
task_shutdown.child_token(),
);
}
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
task_shutdown.child_token(),
);
}
#[cfg(all(feature = "db", not(feature = "sqlite")))]
if let Some(shards) = state.shards() {
for shard in shards.iter() {
#[cfg(feature = "ws")]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
shard.primary_pool().clone(),
Some(state.channels().clone()),
task_shutdown.child_token(),
);
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
shard.primary_pool().clone(),
task_shutdown.child_token(),
);
}
}
#[cfg(all(feature = "db", feature = "sqlite"))]
if let Some(pool) = state.pool().cloned() {
#[cfg(feature = "ws")]
{
let channels = state.channels().clone();
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
Some(channels),
task_shutdown.child_token(),
);
}
#[cfg(not(feature = "ws"))]
crate::repository_commit_hooks::start_repository_commit_hook_worker(
pool,
task_shutdown.child_token(),
);
}
if let Err(error) = run_startup_hooks(&startup_hooks, state.clone()).await {
eprintln!("startup hook failed: {error}");
task_shutdown.cancel();
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
state.probes().mark_startup_complete();
tracing::info!(task = %task_name, "Running one-off task");
let span = tracing::info_span!("one_off_task", task = %task_name);
#[cfg(feature = "oauth2")]
let result = {
use crate::interceptor::{ACTIVE_HTTP_INTERCEPTORS, HttpInterceptor};
let interceptors: Vec<std::sync::Arc<dyn HttpInterceptor>> = state
.extension::<std::sync::Arc<dyn HttpInterceptor>>()
.map(|interceptor_arc| vec![(*interceptor_arc).clone()])
.unwrap_or_default();
ACTIVE_HTTP_INTERCEPTORS
.scope(
interceptors,
(task_handler)(state.clone(), args).instrument(span),
)
.await
};
#[cfg(not(feature = "oauth2"))]
let result = (task_handler)(state.clone(), args).instrument(span).await;
task_shutdown.cancel();
run_shutdown_hooks(&shutdown_hooks).await;
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
match result {
Ok(()) => {
tracing::info!(task = %task_name, "One-off task completed");
}
Err(error) => {
tracing::error!(task = %task_name, error = %error, "One-off task failed");
eprintln!("Task '{task_name}' failed: {error}");
for cause in error.source_chain() {
eprintln!("Caused by: {cause}");
}
std::process::exit(1);
}
}
}
}
pub(crate) fn is_static_build_mode() -> bool {
std::env::var("AUTUMN_BUILD_STATIC").as_deref() == Ok("1")
}
#[allow(clippy::missing_const_for_fn)]
fn exit_stop_managed_pg() {
#[cfg(feature = "managed-pg")]
{
let _ = std::thread::spawn(crate::managed_pg::emergency_stop).join();
}
}
pub(crate) fn is_dump_routes_mode() -> bool {
std::env::var("AUTUMN_DUMP_ROUTES").as_deref() == Ok("1")
}
pub(crate) fn is_dump_security_mode() -> bool {
std::env::var("AUTUMN_DUMP_SECURITY").as_deref() == Ok("1")
}
pub(crate) fn is_dump_jobs_mode() -> bool {
std::env::var("AUTUMN_DUMP_JOBS").as_deref() == Ok("1")
}
pub(crate) fn is_list_one_off_tasks_mode() -> bool {
std::env::var("AUTUMN_LIST_TASKS").as_deref() == Ok("1")
}
pub(crate) fn is_migrate_only_mode() -> bool {
std::env::var("AUTUMN_MIGRATE").as_deref() == Ok("1")
}
fn one_off_task_name_from_env() -> Option<String> {
std::env::var("AUTUMN_RUN_TASK")
.ok()
.map(|value| value.trim().to_owned())
.filter(|value| !value.is_empty())
}
fn one_off_task_args_from_env() -> Result<Vec<String>, String> {
match std::env::var("AUTUMN_TASK_ARGS_JSON") {
Ok(raw) if !raw.trim().is_empty() => serde_json::from_str(&raw)
.map_err(|error| format!("AUTUMN_TASK_ARGS_JSON must be a JSON string array: {error}")),
_ => Ok(Vec::new()),
}
}
fn print_available_one_off_tasks(tasks: &[crate::task::OneOffTaskInfo]) {
let listing = crate::task::list_one_off_tasks(tasks);
if listing.is_empty() {
eprintln!("No one-off tasks are registered. Add .one_off_tasks(one_off_tasks![...]).");
return;
}
eprintln!("Available tasks:");
for task in listing {
if task.description.is_empty() {
eprintln!(" {}", task.name);
} else {
eprintln!(" {:<24} {}", task.name, task.description);
}
}
}
#[allow(clippy::cast_possible_truncation)]
#[allow(clippy::cognitive_complexity)]
#[allow(dead_code)]
fn start_task_scheduler(
tasks: Vec<crate::task::TaskInfo>,
state: &AppState,
shutdown: &tokio_util::sync::CancellationToken,
) {
if let Err(error) = start_task_scheduler_with_config(
tasks,
state,
shutdown,
&crate::config::SchedulerConfig::default(),
) {
tracing::error!(error = %error, "scheduled task runtime initialization failed");
}
}
#[allow(clippy::cast_possible_truncation)]
#[allow(clippy::cognitive_complexity)]
fn start_task_scheduler_with_config(
tasks: Vec<crate::task::TaskInfo>,
state: &AppState,
shutdown: &tokio_util::sync::CancellationToken,
scheduler_config: &crate::config::SchedulerConfig,
) -> crate::AutumnResult<()> {
tracing::info!(count = tasks.len(), "Starting scheduled tasks");
let coordinator = crate::scheduler::coordinator_from_config(scheduler_config, state)?;
let lease_ttl = std::time::Duration::from_secs(scheduler_config.lease_ttl_secs);
for task_info in &tasks {
let schedule_desc = task_info.schedule.to_string();
tracing::info!(
name = %task_info.name,
schedule = %schedule_desc,
coordination = %task_info.coordination,
scheduler_backend = coordinator.backend(),
replica_id = coordinator.replica_id(),
lease_ttl_secs = scheduler_config.lease_ttl_secs,
"Registered task"
);
}
let mut cron_tasks: Vec<CronTaskSpec> = Vec::new();
for task_info in tasks {
let state = state.clone();
let name = task_info.name.clone();
let handler = task_info.handler;
let coordination = task_info.coordination;
let schedule_desc = task_info.schedule.to_string();
state.task_registry.register_scheduled(
&name,
&schedule_desc,
coordination,
coordinator.backend(),
coordinator.replica_id(),
);
match task_info.schedule {
crate::task::Schedule::FixedDelay(delay) => {
let coordinator = Arc::clone(&coordinator);
let shutdown = shutdown.child_token();
tokio::spawn(async move {
loop {
state
.task_registry
.record_next_run_at(&name, &format_next_task_run_after(delay));
tokio::select! {
() = shutdown.cancelled() => break,
() = tokio::time::sleep(delay) => {
execute_fixed_delay_task(
name.clone(),
state.clone(),
handler,
delay,
coordination,
Arc::clone(&coordinator),
lease_ttl,
)
.await;
}
}
}
});
}
crate::task::Schedule::Cron {
expression,
timezone,
} => {
cron_tasks.push(CronTaskSpec {
name,
expression,
timezone,
coordination,
handler,
});
}
}
}
run_cron_scheduler(cron_tasks, state, shutdown, &coordinator, lease_ttl);
Ok(())
}
#[allow(unused_variables, clippy::needless_pass_by_value)]
fn send_ws_sys_task_msg(
state: &AppState,
event: &str,
name: &str,
extra: Vec<(&str, serde_json::Value)>,
) {
#[cfg(feature = "ws")]
{
let mut msg = serde_json::json!({
"event": event,
"task": name,
"timestamp": chrono::Utc::now().to_rfc3339(),
});
if let Some(map) = msg.as_object_mut() {
for (k, v) in extra {
map.insert(k.to_string(), v);
}
}
let _ = state.channels().sender("sys:tasks").send(msg.to_string());
}
}
async fn execute_task_result(
state: &AppState,
handler: crate::task::TaskHandler,
start: std::time::Instant,
name: &str,
schedule: &'static str,
) -> Result<u64, (u64, String)> {
let task_span = tracing::info_span!(
parent: None,
"scheduled_task",
otel.kind = "internal",
task = %name,
schedule = schedule,
);
let future = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
(handler)(state.clone()).instrument(task_span)
})) {
Ok(future) => future,
Err(panic) => {
let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
return Err((duration_ms, format_scheduled_task_panic(panic.as_ref())));
}
};
let result = std::panic::AssertUnwindSafe(future).catch_unwind().await;
let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
match result {
Ok(Ok(())) => Ok(duration_ms),
Ok(Err(e)) => Err((duration_ms, e.to_string())),
Err(panic) => Err((duration_ms, format_scheduled_task_panic(panic.as_ref()))),
}
}
fn format_scheduled_task_panic(panic: &(dyn Any + Send)) -> String {
let detail = panic
.downcast_ref::<String>()
.map(String::as_str)
.or_else(|| panic.downcast_ref::<&'static str>().copied())
.unwrap_or("non-string panic payload");
format!("scheduled task handler panicked: {detail}")
}
async fn execute_task_result_with_optional_lease_ttl(
state: &AppState,
handler: crate::task::TaskHandler,
start: std::time::Instant,
name: &str,
schedule: &'static str,
lease_ttl: Option<std::time::Duration>,
) -> Result<u64, (u64, String)> {
let Some(lease_ttl) = lease_ttl else {
return execute_task_result(state, handler, start, name, schedule).await;
};
tokio::time::timeout(
lease_ttl,
execute_task_result(state, handler, start, name, schedule),
)
.await
.unwrap_or_else(|_| {
let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
Err((
duration_ms,
format!(
"scheduled task exceeded lease TTL of {}s",
lease_ttl.as_secs()
),
))
})
}
#[allow(clippy::cognitive_complexity)]
async fn execute_fixed_delay_task(
name: String,
state: AppState,
handler: crate::task::TaskHandler,
delay: std::time::Duration,
coordination: crate::task::TaskCoordination,
coordinator: Arc<dyn crate::scheduler::SchedulerCoordinator>,
lease_ttl: std::time::Duration,
) {
let tick_key = crate::scheduler::fixed_delay_tick_key(
&name,
delay,
crate::time::clock_unix_duration(state.clock()),
);
let lease = match coordinator
.try_acquire(&name, &tick_key, coordination)
.await
{
Ok(Some(lease)) => lease,
Ok(None) => {
tracing::debug!(task = %name, tick = %tick_key, "Scheduled task tick already claimed");
return;
}
Err(error) => {
tracing::warn!(task = %name, tick = %tick_key, error = %error, "Failed to acquire scheduled task lease");
return;
}
};
state
.task_registry
.record_leader(&name, lease.leader_id(), &tick_key);
tracing::debug!(task = %name, "Running scheduled task");
state.task_registry.record_start(&name);
send_ws_sys_task_msg(&state, "started", &name, vec![]);
let start = std::time::Instant::now();
let lease_ttl = lease_ttl_for_run(&lease, coordination, lease_ttl);
match execute_task_result_with_optional_lease_ttl(
&state,
handler,
start,
&name,
"fixed_delay",
lease_ttl,
)
.await
{
Ok(duration_ms) => {
state.task_registry.record_success(&name, duration_ms);
crate::alerts::notify_scheduled_task_recovered(&state, &name);
tracing::debug!(task = %name, "Task completed");
send_ws_sys_task_msg(
&state,
"success",
&name,
vec![("duration_ms", serde_json::json!(duration_ms))],
);
}
Err((duration_ms, error_str)) => {
state
.task_registry
.record_failure(&name, duration_ms, &error_str);
crate::alerts::notify_scheduled_task_failure(&state, &name, &error_str);
tracing::warn!(task = %name, error = %error_str, "Task failed");
send_ws_sys_task_msg(
&state,
"failure",
&name,
vec![
("duration_ms", serde_json::json!(duration_ms)),
("error", serde_json::json!(error_str)),
],
);
}
}
if let Err(error) = lease.release().await {
tracing::warn!(task = %name, tick = %tick_key, error = %error, "Failed to release scheduled task lease");
}
}
#[allow(clippy::cognitive_complexity)]
async fn execute_cron_task(
name: String,
state: AppState,
handler: crate::task::TaskHandler,
coordination: crate::task::TaskCoordination,
coordinator: Arc<dyn crate::scheduler::SchedulerCoordinator>,
lease_ttl: std::time::Duration,
scheduled_unix_secs: u64,
) {
let tick_key = crate::scheduler::cron_tick_key(&name, scheduled_unix_secs);
let lease = match coordinator
.try_acquire(&name, &tick_key, coordination)
.await
{
Ok(Some(lease)) => lease,
Ok(None) => {
tracing::debug!(task = %name, tick = %tick_key, "Cron task tick already claimed");
return;
}
Err(error) => {
tracing::warn!(task = %name, tick = %tick_key, error = %error, "Failed to acquire cron task lease");
return;
}
};
state
.task_registry
.record_leader(&name, lease.leader_id(), &tick_key);
tracing::debug!(task = %name, "Running cron task");
state.task_registry.record_start(&name);
send_ws_sys_task_msg(&state, "started", &name, vec![]);
let start = std::time::Instant::now();
let lease_ttl = lease_ttl_for_run(&lease, coordination, lease_ttl);
match execute_task_result_with_optional_lease_ttl(
&state, handler, start, &name, "cron", lease_ttl,
)
.await
{
Ok(duration_ms) => {
state.task_registry.record_success(&name, duration_ms);
crate::alerts::notify_scheduled_task_recovered(&state, &name);
tracing::debug!(task = %name, "Cron task completed");
send_ws_sys_task_msg(
&state,
"success",
&name,
vec![("duration_ms", serde_json::json!(duration_ms))],
);
}
Err((duration_ms, error_str)) => {
state
.task_registry
.record_failure(&name, duration_ms, &error_str);
crate::alerts::notify_scheduled_task_failure(&state, &name, &error_str);
tracing::warn!(task = %name, error = %error_str, "Cron task failed");
send_ws_sys_task_msg(
&state,
"failure",
&name,
vec![
("duration_ms", serde_json::json!(duration_ms)),
("error", serde_json::json!(error_str)),
],
);
}
}
if let Err(error) = lease.release().await {
tracing::warn!(task = %name, tick = %tick_key, error = %error, "Failed to release cron task lease");
}
}
struct CronTaskSpec {
name: String,
expression: String,
timezone: Option<String>,
coordination: crate::task::TaskCoordination,
handler: crate::task::TaskHandler,
}
fn lease_ttl_for_run(
lease: &crate::scheduler::SchedulerLease,
coordination: crate::task::TaskCoordination,
lease_ttl: std::time::Duration,
) -> Option<std::time::Duration> {
(coordination == crate::task::TaskCoordination::Fleet && lease.backend() == "postgres")
.then_some(lease_ttl)
}
fn run_cron_scheduler(
tasks: Vec<CronTaskSpec>,
state: &AppState,
shutdown: &tokio_util::sync::CancellationToken,
coordinator: &Arc<dyn crate::scheduler::SchedulerCoordinator>,
lease_ttl: std::time::Duration,
) {
if tasks.is_empty() {
return;
}
tracing::info!(count = tasks.len(), "Cron scheduler started");
for task in tasks {
let state = state.clone();
let coordinator = Arc::clone(coordinator);
let shutdown = shutdown.child_token();
tokio::spawn(async move {
run_cron_task_loop(task, state, shutdown, coordinator, lease_ttl).await;
});
}
}
#[allow(clippy::cognitive_complexity)]
async fn run_cron_task_loop(
task: CronTaskSpec,
state: AppState,
shutdown: tokio_util::sync::CancellationToken,
coordinator: Arc<dyn crate::scheduler::SchedulerCoordinator>,
lease_ttl: std::time::Duration,
) {
let CronTaskSpec {
name,
expression,
timezone,
coordination,
handler,
} = task;
let cron = match expression.parse::<croner::Cron>() {
Ok(cron) => cron,
Err(error) => {
tracing::error!(task = %name, expression = %expression, error = %error, "Failed to create cron job");
return;
}
};
let timezone = timezone
.as_deref()
.and_then(|timezone| {
timezone.parse::<chrono_tz::Tz>().map_or_else(
|_| {
tracing::warn!(task = %name, timezone = %timezone, "Unrecognized timezone; falling back to UTC");
None
},
Some,
)
})
.unwrap_or(chrono_tz::UTC);
let mut cursor = chrono::Utc::now().with_timezone(&timezone);
loop {
let now = chrono::Utc::now().with_timezone(&timezone);
let scheduled_at = match next_cron_occurrence_after(&cron, &cursor, &now) {
Ok(scheduled_at) => scheduled_at,
Err(error) => {
tracing::error!(task = %name, expression = %expression, error = %error, "Failed to compute next cron tick");
return;
}
};
state.task_registry.record_next_run_at(
&name,
&scheduled_at.with_timezone(&chrono::Utc).to_rfc3339(),
);
let sleep_for = cron_sleep_duration_until(&scheduled_at);
tokio::select! {
() = shutdown.cancelled() => break,
() = tokio::time::sleep(sleep_for) => {
let woke_at = chrono::Utc::now().with_timezone(&timezone);
match cron_occurrence_is_overdue(&cron, &scheduled_at, &woke_at) {
Ok(true) => {
tracing::warn!(
task = %name,
scheduled_at = %scheduled_at,
woke_at = %woke_at,
"Skipping overdue cron task tick"
);
cursor = woke_at;
continue;
}
Ok(false) => {}
Err(error) => {
tracing::error!(task = %name, expression = %expression, error = %error, "Failed to evaluate cron tick lateness");
return;
}
}
let scheduled_unix_secs = u64::try_from(scheduled_at.timestamp()).unwrap_or_default();
tokio::spawn(execute_cron_task(
name.clone(),
state.clone(),
handler,
coordination,
Arc::clone(&coordinator),
lease_ttl,
scheduled_unix_secs,
));
cursor = scheduled_at;
}
}
}
}
fn format_next_task_run_after(delay: std::time::Duration) -> String {
let now = chrono::Utc::now();
let Ok(delay) = chrono::TimeDelta::from_std(delay) else {
return now.to_rfc3339();
};
(now + delay).to_rfc3339()
}
fn next_cron_occurrence_after<Tz: chrono::TimeZone>(
cron: &croner::Cron,
cursor: &chrono::DateTime<Tz>,
now: &chrono::DateTime<Tz>,
) -> Result<chrono::DateTime<Tz>, croner::errors::CronError> {
let anchor = if cursor < now { now } else { cursor };
cron.find_next_occurrence(anchor, false)
}
fn cron_occurrence_is_overdue<Tz: chrono::TimeZone>(
cron: &croner::Cron,
scheduled_at: &chrono::DateTime<Tz>,
now: &chrono::DateTime<Tz>,
) -> Result<bool, croner::errors::CronError> {
let next_after_scheduled = cron.find_next_occurrence(scheduled_at, false)?;
Ok(&next_after_scheduled <= now)
}
fn cron_sleep_duration_until<Tz: chrono::TimeZone>(
scheduled_at: &chrono::DateTime<Tz>,
) -> std::time::Duration {
scheduled_at
.with_timezone(&chrono::Utc)
.signed_duration_since(chrono::Utc::now())
.to_std()
.unwrap_or_default()
}
async fn run_startup_hooks(hooks: &[StartupHook], state: AppState) -> crate::AutumnResult<()> {
for hook in hooks {
hook(state.clone()).await?;
}
Ok(())
}
fn run_state_initializers(initializers: Vec<StateInitializer>, state: &AppState) {
for initializer in initializers {
initializer(state);
}
}
fn synthesize_durable_listener_jobs(
listeners: Vec<crate::events::ListenerInfo>,
jobs: &mut Vec<crate::job::JobInfo>,
) -> crate::events::EventRegistry {
let registry = crate::events::EventRegistry::from_listeners(listeners);
jobs.extend(registry.durable_job_infos());
registry
}
fn finalize_event_bus(
listeners: Vec<crate::events::ListenerInfo>,
jobs: &mut Vec<crate::job::JobInfo>,
state: &AppState,
) {
let registry = synthesize_durable_listener_jobs(listeners, jobs);
state.insert_extension(registry.clone());
crate::events::init_global_event_bus(®istry, state, None);
}
fn dump_jobs_manifest(
cfg: &crate::config::JobQueuesConfig,
mut jobs: Vec<crate::job::JobInfo>,
listeners: Vec<crate::events::ListenerInfo>,
) -> String {
synthesize_durable_listener_jobs(listeners, &mut jobs);
crate::job::render_jobs_manifest(cfg, &jobs)
}
fn initialize_job_runtime(
jobs: Vec<crate::job::JobInfo>,
state: &AppState,
shutdown: &tokio_util::sync::CancellationToken,
config: &crate::config::JobConfig,
run_workers: bool,
) -> crate::AutumnResult<()> {
crate::job::clear_global_job_client();
if jobs.is_empty() {
Ok(())
} else {
crate::job::start_runtime(jobs, state, shutdown, config, run_workers)
}
}
enum BoundListener {
Tcp(tokio::net::TcpListener),
#[cfg(unix)]
Unix(tokio::net::UnixListener),
#[cfg(feature = "tls")]
Tls(crate::tls::TlsListener),
}
#[cfg(feature = "tls")]
fn now_unix() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
}
#[cfg(feature = "tls")]
struct TlsReloadState {
resolver: std::sync::Arc<crate::tls::ReloadableCertResolver>,
provider: std::sync::Arc<rustls::crypto::CryptoProvider>,
cert_path: std::path::PathBuf,
key_path: std::path::PathBuf,
interval: std::time::Duration,
}
#[cfg(feature = "tls")]
fn build_tls_listener(
tcp: tokio::net::TcpListener,
cfg: &crate::config::TlsConfig,
shutdown: tokio_util::sync::CancellationToken,
) -> Result<(crate::tls::TlsListener, TlsReloadState), crate::tls::TlsError> {
let provider = crate::tls::crypto_provider();
let cert_path = cfg
.cert_path
.as_deref()
.expect("validated: static [server.tls] sets cert_path");
let key_path = cfg
.key_path
.as_deref()
.expect("validated: static [server.tls] sets key_path");
let certified = crate::tls::load_certified_key(cert_path, key_path, &provider, now_unix())?;
let resolver = std::sync::Arc::new(crate::tls::ReloadableCertResolver::new(certified));
let server_config = crate::tls::build_server_config(
std::sync::Arc::clone(&provider),
std::sync::Arc::clone(&resolver),
)?;
let handshake_timeout = std::time::Duration::from_secs(cfg.handshake_timeout_secs.max(1));
let listener = crate::tls::TlsListener::new(tcp, server_config, handshake_timeout, shutdown);
let reload = TlsReloadState {
resolver,
provider,
cert_path: cert_path.to_path_buf(),
key_path: key_path.to_path_buf(),
interval: std::time::Duration::from_secs(cfg.reload_interval_secs.max(1)),
};
Ok((listener, reload))
}
#[cfg(feature = "acme")]
struct AcmeBindState {
renewal_task: crate::acme::renewal::AcmeRenewalTask,
tokens: crate::acme::challenge::Http01Tokens,
http_challenge_port: u16,
https_port: u16,
}
#[cfg(feature = "acme")]
async fn build_acme_tls_listener(
tcp: tokio::net::TcpListener,
tls_cfg: &crate::config::TlsConfig,
acme_cfg: &crate::config::AcmeConfig,
https_port: u16,
status: Option<crate::acme::renewal::AcmeStatus>,
shutdown: tokio_util::sync::CancellationToken,
) -> Result<(crate::tls::TlsListener, AcmeBindState), String> {
use crate::acme::store::{AcmeStore, CertId, FsAcmeStore};
let provider = crate::tls::crypto_provider();
let cert_id = CertId::from_domains(&acme_cfg.domains);
let directory_label = crate::acme::directory_label(&acme_cfg.directory);
let store: std::sync::Arc<dyn AcmeStore> = std::sync::Arc::new(FsAcmeStore::new(
acme_cfg.cache_dir.clone(),
directory_label,
));
let status = status.unwrap_or_default();
let (initial, serving_stored_cert) = match store.load_cert(&cert_id).await {
Ok(Some(stored)) => match crate::tls::certified_key_from_pem(
stored.chain_pem.as_bytes(),
stored.key_pem.as_bytes(),
&provider,
) {
Ok(ck) => {
if let Ok(not_after) =
crate::tls::leaf_not_after_from_pem(stored.chain_pem.as_bytes())
{
status.set_cert_not_after(not_after);
}
(ck, true)
}
Err(e) => {
tracing::warn!(
"stored ACME certificate is unusable ({e}); serving a self-signed \
placeholder until the renewal task issues a real one"
);
(acme_placeholder_key(&acme_cfg.domains, &provider)?, false)
}
},
Ok(None) => (acme_placeholder_key(&acme_cfg.domains, &provider)?, false),
Err(e) => {
tracing::warn!(
"failed to read the stored ACME certificate ({e}); serving a self-signed \
placeholder"
);
(acme_placeholder_key(&acme_cfg.domains, &provider)?, false)
}
};
let resolver = std::sync::Arc::new(crate::tls::ReloadableCertResolver::new(initial));
let server_config = crate::tls::build_server_config(
std::sync::Arc::clone(&provider),
std::sync::Arc::clone(&resolver),
)
.map_err(|e| e.to_string())?;
let handshake_timeout = std::time::Duration::from_secs(tls_cfg.handshake_timeout_secs.max(1));
let listener = crate::tls::TlsListener::new(tcp, server_config, handshake_timeout, shutdown);
let tokens = crate::acme::challenge::Http01Tokens::new();
let renewal_task = crate::acme::renewal::AcmeRenewalTask {
resolver,
provider,
store,
cert_id,
tokens: tokens.clone(),
status,
config: acme_cfg.clone(),
serving_stored_cert,
leadership_degraded: false,
renew_window_misconfigured: std::sync::atomic::AtomicBool::new(false),
};
Ok((
listener,
AcmeBindState {
renewal_task,
tokens,
http_challenge_port: acme_cfg.http_challenge_port,
https_port,
},
))
}
#[cfg(feature = "acme")]
fn acme_placeholder_key(
domains: &[String],
provider: &rustls::crypto::CryptoProvider,
) -> Result<std::sync::Arc<rustls::sign::CertifiedKey>, String> {
let placeholder = crate::acme::renewal::self_signed_placeholder(domains)?;
crate::tls::certified_key_from_pem(
placeholder.chain_pem.as_bytes(),
placeholder.key_pem.as_bytes(),
provider,
)
}
#[cfg(all(feature = "acme", feature = "reporting"))]
fn make_acme_reporter(
reporters: Vec<std::sync::Arc<dyn crate::reporting::ErrorReporter>>,
) -> crate::acme::renewal::ReporterFn {
std::sync::Arc::new(move |message: String| {
if reporters.is_empty() {
return;
}
let reporters = reporters.clone();
tokio::spawn(async move {
let event = crate::reporting::ErrorEvent {
status: axum::http::StatusCode::INTERNAL_SERVER_ERROR,
message,
problem_type: None,
request_id: None,
route: Some("acme-renewal".to_owned()),
method: None,
panic: None,
};
for reporter in &reporters {
reporter.report(&event).await;
}
});
})
}
#[cfg(all(feature = "acme", not(feature = "reporting")))]
fn make_acme_reporter() -> crate::acme::renewal::ReporterFn {
std::sync::Arc::new(|_message: String| {})
}
#[cfg(feature = "tls")]
fn tls_file_mtimes(
cert: &std::path::Path,
key: &std::path::Path,
) -> (Option<std::time::SystemTime>, Option<std::time::SystemTime>) {
let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok();
(mtime(cert), mtime(key))
}
#[cfg(feature = "tls")]
async fn run_tls_cert_reload(state: TlsReloadState, shutdown: tokio_util::sync::CancellationToken) {
let stat_mtimes = |cert: std::path::PathBuf, key: std::path::PathBuf| {
tokio::task::spawn_blocking(move || tls_file_mtimes(&cert, &key))
};
let mut last = match stat_mtimes(state.cert_path.clone(), state.key_path.clone()).await {
Ok(mtimes) => mtimes,
Err(e) => {
tracing::warn!(error = %e, "TLS reload: initial mtime read failed; assuming unknown");
(None, None)
}
};
loop {
tokio::select! {
() = tokio::time::sleep(state.interval) => {}
() = shutdown.cancelled() => break,
}
let current = match stat_mtimes(state.cert_path.clone(), state.key_path.clone()).await {
Ok(mtimes) => mtimes,
Err(e) => {
tracing::warn!(error = %e, "TLS reload: mtime read task failed; skipping tick");
continue;
}
};
if current == last {
continue;
}
let cert_path = state.cert_path.clone();
let key_path = state.key_path.clone();
let provider = std::sync::Arc::clone(&state.provider);
let loaded = tokio::task::spawn_blocking(move || {
crate::tls::load_certified_key(&cert_path, &key_path, &provider, now_unix())
})
.await;
let loaded = match loaded {
Ok(result) => result,
Err(e) => {
tracing::warn!(error = %e, "TLS reload: load task failed; skipping tick");
continue;
}
};
match loaded {
Ok(next) => {
state.resolver.store(next);
last = current;
tracing::info!(
cert = %state.cert_path.display(),
"Reloaded TLS certificate after detecting a change on disk"
);
}
Err(e) => {
tracing::error!(
error = %e,
cert = %state.cert_path.display(),
"TLS certificate reload failed; keeping the previously loaded certificate"
);
}
}
}
}
#[cfg(unix)]
#[derive(Clone, Debug)]
struct UdsConnectInfo;
#[cfg(unix)]
impl
axum::extract::connect_info::Connected<
axum::serve::IncomingStream<'_, tokio::net::UnixListener>,
> for UdsConnectInfo
{
fn connect_info(_stream: axum::serve::IncomingStream<'_, tokio::net::UnixListener>) -> Self {
Self
}
}
#[cfg(unix)]
async fn stamp_loopback_connect_info(
mut req: axum::extract::Request,
next: axum::middleware::Next,
) -> axum::response::Response {
if req
.extensions()
.get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
.is_none()
{
let loopback =
std::net::SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0);
req.extensions_mut()
.insert(axum::extract::ConnectInfo(loopback));
}
next.run(req).await
}
fn signal_serve_ready(drain_budget_secs: u64) {
let Some(path) = std::env::var_os("AUTUMN_SERVE_READY_FILE") else {
return;
};
if path.is_empty() {
return;
}
let path = std::path::PathBuf::from(path);
let mut tmp = path.clone();
tmp.as_mut_os_string().push(".tmp");
if let Err(e) = std::fs::write(&tmp, drain_budget_secs.to_string())
.and_then(|()| std::fs::rename(&tmp, &path))
{
let _ = std::fs::remove_file(&tmp);
tracing::warn!(error = %e, path = %path.display(),
"could not write serve readiness file");
}
}
#[cfg(unix)]
fn prepare_unix_socket_path(path: &std::path::Path) -> std::io::Result<()> {
use std::os::unix::fs::FileTypeExt;
match std::fs::symlink_metadata(path) {
Ok(meta) if meta.file_type().is_socket() => {
match std::os::unix::net::UnixStream::connect(path) {
Ok(_) => Err(std::io::Error::new(
std::io::ErrorKind::AddrInUse,
format!(
"refusing to bind unix socket: {} is already in use by a \
live listener",
path.display()
),
)),
Err(e)
if matches!(
e.kind(),
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
) =>
{
std::fs::remove_file(path)
}
Err(e) => Err(std::io::Error::new(
std::io::ErrorKind::AddrInUse,
format!(
"refusing to bind unix socket: cannot determine whether {} \
is live ({e}); not removing it",
path.display()
),
)),
}
}
Ok(_) => Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
format!(
"refusing to bind unix socket: {} exists and is not a socket",
path.display()
),
)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
}
}
async fn run_shutdown_hooks(hooks: &[ShutdownHook]) {
for hook in hooks.iter().rev() {
hook().await;
}
}
async fn run_shutdown_hooks_with_timeout(
hooks: &[ShutdownHook],
per_hook_budget: std::time::Duration,
total_budget: std::time::Duration,
) {
let deadline = tokio::time::Instant::now() + total_budget;
for hook in hooks.iter().rev() {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
tracing::warn!("shutdown: total hook budget exhausted; skipping remaining hooks");
break;
}
let timeout = remaining.min(per_hook_budget);
if tokio::time::timeout(timeout, hook()).await.is_err() {
tracing::warn!(
per_hook_budget_ms = timeout.as_millis(),
"shutdown: hook overran per-hook timeout; continuing with remaining budget"
);
}
}
}
#[allow(clippy::cognitive_complexity)]
fn log_startup_transparency(
routes: &[Route],
tasks: &[crate::task::TaskInfo],
scoped_groups: &[ScopedGroup],
config: &AutumnConfig,
) {
tracing::info!(
"Registered routes:{}",
format_route_lines(routes, scoped_groups, config)
);
if let Some(task_lines) = format_task_lines(tasks) {
tracing::info!("Scheduled tasks:{task_lines}");
}
tracing::info!("Active middleware: {}", format_middleware_list(config));
tracing::info!("Configuration:{}", format_config_summary(config));
}
fn fail_fast_on_invalid_session_config(config: &AutumnConfig, has_custom_session_store: bool) {
if has_custom_session_store {
return;
}
if let Err(error) = config.session.backend_plan(config.profile.as_deref()) {
eprintln!("Invalid session backend config: {error}");
std::process::exit(1);
}
}
fn fail_fast_on_missing_encryption_keys(config: &AutumnConfig) {
if let Err(diagnostic) = crate::encryption::init_attribute_encryption(config.credentials()) {
let is_production = matches!(config.profile.as_deref(), Some("prod" | "production"));
if is_production {
eprintln!("Attribute encryption misconfiguration: {diagnostic}");
std::process::exit(1);
}
eprintln!(
"warning: attribute encryption is not fully configured (dev): {diagnostic}\n \
note: encrypted-column reads/writes will fail until keys are set; \
this is a hard error in production."
);
}
}
fn fail_fast_on_invalid_signing_secret(config: &AutumnConfig) {
use crate::security::config::validate_signing_secret;
let is_production = matches!(config.profile.as_deref(), Some("prod" | "production"));
let secret = config.security.signing_secret.secret.as_deref();
if let Err(error) = validate_signing_secret(secret, is_production) {
eprintln!("Invalid signing secret configuration: {error}");
eprintln!(
" hint: generate a secret with `openssl rand -hex 32` and set \
AUTUMN_SECURITY__SIGNING_SECRET"
);
std::process::exit(1);
}
if is_production {
for (i, prev) in config
.security
.signing_secret
.previous_secrets
.iter()
.enumerate()
{
if let Err(error) = validate_signing_secret(Some(prev.as_str()), true) {
eprintln!("Invalid signing secret configuration: previous_secrets[{i}]: {error}");
eprintln!(
" hint: every previous secret must meet the same entropy requirement \
as the current secret"
);
std::process::exit(1);
}
}
}
}
fn fail_fast_on_invalid_webhook_config(config: &AutumnConfig) {
let is_production = matches!(config.profile.as_deref(), Some("prod" | "production"));
if let Err(error) = config.security.webhooks.validate(is_production) {
eprintln!("Invalid signed webhook configuration: {error}");
std::process::exit(1);
}
}
fn fail_fast_on_invalid_trusted_hosts(config: &AutumnConfig) {
let is_production = matches!(config.profile.as_deref(), Some("prod" | "production"));
if !is_production {
return;
}
let hosts: Vec<String> = config
.security
.trusted_hosts
.hosts
.iter()
.map(|h| h.trim().to_owned())
.filter(|h| !h.is_empty())
.collect();
if hosts.is_empty() {
eprintln!(
"[security.trusted_hosts] is required in production; set hosts = [\"example.com\"] or explicit entries"
);
std::process::exit(1);
}
if hosts.iter().any(|h| h == "*") {
tracing::warn!("trusted host validation disabled via wildcard '*' in production");
}
}
fn fail_fast_on_invalid_idempotency_config(config: &AutumnConfig) {
if !config.idempotency.enabled.unwrap_or(false) {
return;
}
let is_production = matches!(config.profile.as_deref(), Some("prod" | "production"));
if is_production
&& config.idempotency.backend == crate::config::IdempotencyBackend::Memory
&& !config.idempotency.allow_memory_in_production
{
eprintln!(
"The in-memory idempotency backend is not safe for multi-replica production use.\n\
Set `[idempotency] backend = \"redis\"` in autumn.toml, or set \
`allow_memory_in_production = true` to suppress this check."
);
std::process::exit(1);
}
#[cfg(feature = "redis")]
if config.idempotency.backend == crate::config::IdempotencyBackend::Redis {
let url_missing = config
.idempotency
.redis
.url
.as_deref()
.is_none_or(|u| u.trim().is_empty());
if url_missing {
eprintln!(
"Redis idempotency backend requires a connection URL.\n\
Set AUTUMN_IDEMPOTENCY__REDIS__URL or `[idempotency.redis] url` in autumn.toml."
);
std::process::exit(1);
}
}
}
pub(crate) fn install_webhook_registry(state: &AppState, config: &AutumnConfig) {
if let Err(error) =
crate::webhook::install_registry_from_config(state, &config.security.webhooks)
{
eprintln!("Invalid signed webhook configuration: {error}");
std::process::exit(1);
}
}
#[cfg(feature = "storage")]
struct StorageBootstrap {
store: crate::storage::SharedBlobStore,
serving: Option<axum::Router<AppState>>,
}
#[cfg(feature = "storage")]
impl StorageBootstrap {
fn install(self, state: &AppState) -> Option<axum::Router<AppState>> {
state.insert_extension::<crate::storage::BlobStoreState>(
crate::storage::BlobStoreState::new(self.store),
);
self.serving
}
}
#[cfg(feature = "storage")]
#[allow(clippy::too_many_lines)] fn preflight_storage(config: &AutumnConfig) -> Option<StorageBootstrap> {
use crate::storage::StorageBackendPlan;
let plan = config
.storage
.backend_plan(config.profile.as_deref())
.unwrap_or_else(|error| {
tracing::error!(%error, "invalid storage backend config; aborting startup");
std::process::exit(1);
});
match plan {
StorageBackendPlan::Disabled => None,
StorageBackendPlan::Local {
provider_id,
root,
mount_path,
default_url_expiry_secs,
warn_in_production,
} => Some(bootstrap_local_storage(
config,
&provider_id,
&root,
&mount_path,
default_url_expiry_secs,
warn_in_production,
)),
StorageBackendPlan::S3 { .. } => {
tracing::error!(
"storage.backend=s3 requires the `autumn-storage-s3` plugin. \
Add it to your Cargo.toml, build an S3BlobStore from your config, \
and call `.with_blob_store(store)` on your AppBuilder. \
Aborting startup."
);
std::process::exit(1);
}
}
}
#[cfg(feature = "storage")]
fn bootstrap_local_storage(
config: &AutumnConfig,
provider_id: &str,
root: &std::path::Path,
mount_path: &str,
default_url_expiry_secs: u64,
warn_in_production: bool,
) -> StorageBootstrap {
use crate::storage::{LocalBlobStore, SharedBlobStore, local::SigningKey};
if warn_in_production {
tracing::warn!(
"prod profile is using the local-disk blob store; \
bytes won't survive replica turnover. Set \
storage.backend=s3 or storage.allow_local_in_production=true \
to acknowledge"
);
}
let (signing_key, previous_signing_keys) = config
.security
.signing_secret
.secret
.as_deref()
.filter(|s| !s.is_empty())
.map_or_else(
|| {
config
.storage
.local
.signing_key
.as_deref()
.filter(|s| !s.is_empty())
.map_or_else(
|| {
if matches!(config.profile.as_deref(), Some("prod" | "production")) {
tracing::warn!(
"no signing secret configured in prod; blob URL signatures \
won't survive a process restart. Set \
AUTUMN_SECURITY__SIGNING_SECRET."
);
}
(SigningKey::random(), vec![])
},
|legacy| (SigningKey::new(legacy.as_bytes().to_vec()), vec![]),
)
},
|secret| {
let current = SigningKey::new(secret.as_bytes().to_vec());
let previous = config
.security
.signing_secret
.previous_secrets
.iter()
.map(|s| SigningKey::new(s.as_bytes().to_vec()))
.collect::<Vec<_>>();
(current, previous)
},
);
let store = match LocalBlobStore::new(
provider_id.to_string(),
root.to_path_buf(),
mount_path.to_string(),
std::time::Duration::from_secs(default_url_expiry_secs),
signing_key,
previous_signing_keys,
) {
Ok(store) => store,
Err(err) => {
tracing::error!(
error = %err,
root = %root.display(),
"failed to initialize local blob store; aborting startup"
);
std::process::exit(1);
}
};
let serving = crate::storage::local::serve_router(&store);
let arc: SharedBlobStore = std::sync::Arc::new(store);
tracing::info!(
provider = %provider_id,
root = %root.display(),
mount = %mount_path,
"Local blob store mounted"
);
StorageBootstrap {
store: arc,
serving: Some(serving),
}
}
async fn load_config_and_telemetry(
config_loader: Option<ConfigLoaderFactory>,
telemetry_provider: Option<Box<dyn crate::telemetry::TelemetryProvider>>,
plugin_config_roots: BTreeSet<String>,
) -> (AutumnConfig, crate::telemetry::TelemetryGuard) {
let mut config = match config_loader {
Some(factory) => factory().await,
None => {
crate::config::TomlEnvConfigLoader::new()
.with_plugin_config_roots(plugin_config_roots)
.load()
.await
}
}
.unwrap_or_else(|e| {
eprintln!("Failed to load configuration: {e}");
std::process::exit(1);
});
if let Ok(forced) = std::env::var("AUTUMN_SERVE_FORCE_UNIX_SOCKET")
&& !forced.is_empty()
{
config.server.unix_socket = Some(forced);
}
let provider: Box<dyn crate::telemetry::TelemetryProvider> = telemetry_provider
.unwrap_or_else(|| Box::new(crate::telemetry::TracingOtlpTelemetryProvider::new()));
let telemetry_guard = provider
.init(&config.log, &config.telemetry, config.profile.as_deref())
.unwrap_or_else(|error| {
eprintln!("Failed to initialize telemetry: {error}");
std::process::exit(1);
});
(config, telemetry_guard)
}
#[cfg(feature = "embed-assets")]
fn register_embedded_static_dir(embedded_static: Option<crate::assets::EmbeddedStaticDir>) {
if let Some(dir) = embedded_static {
crate::assets::register_embedded_static(dir);
}
}
#[cfg(all(feature = "embed-assets", feature = "i18n"))]
fn embedded_i18n_bundle(
explicit: Option<Arc<crate::i18n::Bundle>>,
embedded_locales: Option<&'static include_dir::Dir<'static>>,
config: &AutumnConfig,
) -> Option<Arc<crate::i18n::Bundle>> {
explicit.or_else(|| {
embedded_locales.map(|dir| {
Arc::new(
crate::i18n::Bundle::load_from_embedded(dir, &config.i18n)
.unwrap_or_else(|e| panic!("embedded_locales: {e}")),
)
})
})
}
#[cfg(feature = "i18n")]
fn resolve_i18n_bundle(
explicit_bundle: Option<Arc<crate::i18n::Bundle>>,
auto_load: bool,
config: &AutumnConfig,
env: &dyn crate::config::Env,
) -> Option<Arc<crate::i18n::Bundle>> {
if explicit_bundle.is_some() {
return explicit_bundle;
}
if !auto_load {
return None;
}
let dir = project_dir(&config.i18n.dir, env);
Some(Arc::new(
crate::i18n::Bundle::load_from_dir(&dir, &config.i18n)
.unwrap_or_else(|e| panic!("i18n_auto: {e}")),
))
}
#[cfg(feature = "i18n")]
fn install_i18n_bundle_layer(
mut custom_layers: Vec<CustomLayerRegistration>,
state: &AppState,
bundle: Option<Arc<crate::i18n::Bundle>>,
) -> Vec<CustomLayerRegistration> {
let Some(bundle) = bundle else {
return custom_layers;
};
tracing::info!(
locales = ?bundle.locales(),
default = bundle.default_locale(),
"i18n bundle loaded"
);
state.insert_extension::<Arc<crate::i18n::Bundle>>(bundle.clone());
let ext_layer = axum::Extension(bundle);
custom_layers.push(CustomLayerRegistration {
type_id: TypeId::of::<axum::Extension<Arc<crate::i18n::Bundle>>>(),
type_name: std::any::type_name::<axum::Extension<Arc<crate::i18n::Bundle>>>(),
apply: Box::new(move |router| router.layer(ext_layer)),
});
custom_layers
}
#[cfg(feature = "db")]
struct DatabaseBootstrap {
topology: Option<crate::db::DatabaseTopology>,
shards: Option<crate::sharding::ShardSet>,
replica_readiness: Option<crate::migrate::ReplicaMigrationReadiness>,
replica_migration_check: Option<(String, String)>,
}
#[cfg(feature = "db")]
async fn resolve_shard_set(
config: &AutumnConfig,
shard_router: Option<Arc<dyn crate::sharding::ShardRouter>>,
shard_provider: Option<ShardProviderFactory>,
directory_routing_enabled: bool,
spawn_directory_listener: bool,
topology: Option<&crate::db::DatabaseTopology>,
) -> Result<Option<crate::sharding::ShardSet>, String> {
if !config.database.has_shards() {
return Ok(None);
}
let router: Arc<dyn crate::sharding::ShardRouter> = match shard_router {
Some(explicit) => explicit,
None if directory_routing_enabled => {
let control_primary = topology
.map(crate::db::DatabaseTopology::primary)
.ok_or_else(|| {
"directory_shard_router is enabled but no control database is configured. \
The directory router needs a control `database.primary_url`/`url` to read \
the tenant→shard directory. Set one, or disable directory routing to use \
the hash router."
.to_owned()
})?;
let control_max = control_primary.status().max_size;
if control_max < 2 {
return Err(format!(
"directory_shard_router requires a control database pool of at least 2 \
connections, but the configured maximum is {control_max}. Directory \
routing checks out a second control connection during extraction to \
resolve the tenant→shard key, which deadlocks a pool sized to 1 when a \
handler already holds a control connection (e.g. `Db` + `ShardedDb`). \
Increase the control pool size (database.pool.max_size), or disable \
directory routing to use the hash router."
));
}
let timeout_ms = config.database.statement_timeout.map_or(0, |d| {
u64::try_from(d.as_millis())
.unwrap_or(i32::MAX as u64)
.min(i32::MAX as u64)
});
let dir_router = Arc::new(
crate::sharding::DirectoryShardRouter::new(control_primary.clone())
.with_statement_timeout_ms(timeout_ms),
);
if spawn_directory_listener {
if let Some(control_url) = topology
.and_then(crate::db::DatabaseTopology::migration_url)
.or_else(|| config.database.effective_primary_url())
{
drop(
crate::sharding::DirectoryShardRouter::spawn_invalidation_listener(
Arc::clone(&dir_router),
control_url.to_owned(),
crate::sharding::DEFAULT_DIRECTORY_INVALIDATION_SWEEP_INTERVAL,
),
);
} else {
tracing::warn!(
"directory shard routing is enabled but no control database URL is \
configured (database.primary_url/url is unset, e.g. a custom \
DatabasePoolProvider supplied the control pool); the cache-\
invalidation LISTEN/NOTIFY task cannot be started, so directory \
re-pins will only take effect after the cache TTL expires rather \
than fleet-wide on commit"
);
}
}
dir_router
}
None => Arc::new(crate::sharding::HashShardRouter),
};
let set = match shard_provider {
Some(factory) => {
let topologies = factory(config.database.clone())
.await
.map_err(|e| format!("Failed to create shard pools: {e}"))?;
#[cfg(feature = "sqlite")]
crate::db::reject_sqlite_statement_timeout(config.database.statement_timeout)
.map_err(|e| format!("Failed to create shard pools: {e}"))?;
crate::sharding::build_shard_set(&config.database, topologies, router)
}
None => crate::sharding::create_shard_set(&config.database, router)
.map(|set| set.expect("has_shards() checked above")),
}
.map_err(|e| format!("Failed to configure shards: {e}"))?;
Ok(Some(set))
}
#[cfg(feature = "db")]
#[allow(clippy::too_many_lines)]
async fn setup_database(
config: &AutumnConfig,
migrations: Vec<crate::migrate::EmbeddedMigrations>,
pool_provider: Option<PoolProviderFactory>,
shard_provider: Option<ShardProviderFactory>,
shard_router: Option<Arc<dyn crate::sharding::ShardRouter>>,
directory_shard_router: bool,
hook_queue_migration_mode: RepositoryCommitHookQueueMigrationMode,
) -> Result<DatabaseBootstrap, String> {
let migrations = migrations_with_repository_framework_migrations(
migrations,
crate::repository_commit_hooks::has_repository_commit_hook_descriptors(),
crate::version_history::has_versioned_repository_descriptors(),
hook_queue_migration_mode,
);
let use_directory_router = shard_router.is_none()
&& (directory_shard_router || config.database.directory_shard_router);
let directory_migration_required = directory_migration_is_required(
use_directory_router,
config.database.has_shards(),
hook_queue_migration_mode,
);
let shard_map_migration_required =
shard_map_migration_is_required(config.database.has_shards(), hook_queue_migration_mode);
let check_replica_migrations = !migrations.is_empty();
let topology = match pool_provider {
Some(factory) => factory(config.database.clone()).await,
None => crate::db::create_topology(&config.database),
}
.map_err(|e| format!("Failed to create database pool: {e}"))?;
#[cfg(feature = "sqlite")]
if topology.is_some() {
crate::db::reject_sqlite_statement_timeout(config.database.statement_timeout)
.map_err(|e| format!("Failed to create database pool: {e}"))?;
}
let runtime_boot = hook_queue_migration_mode == RepositoryCommitHookQueueMigrationMode::Runtime;
let shards = match resolve_shard_set(
config,
shard_router,
shard_provider,
use_directory_router,
runtime_boot,
topology.as_ref(),
)
.await
{
Ok(shards) => shards,
Err(e) => {
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
return Err(e);
}
};
let provider_migration_url = topology
.as_ref()
.and_then(|t| t.migration_url())
.map(str::to_owned);
#[cfg(feature = "sqlite")]
let sqlite_guard_shard_urls: Vec<&str> = if shards.is_some() {
config
.database
.shards
.iter()
.map(|shard| shard.primary_url.as_str())
.collect()
} else {
Vec::new()
};
#[cfg(feature = "sqlite")]
#[allow(clippy::question_mark)] if let Err(e) = sqlite_sharding_unsupported_guard(
if topology.is_some() {
provider_migration_url
.as_deref()
.or_else(|| config.database.effective_primary_url())
} else {
None
},
directory_migration_required
|| shard_map_migration_required
|| config.database.has_shards(),
&sqlite_guard_shard_urls,
) {
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
return Err(e);
}
run_startup_migrations(
config,
topology.is_some(),
shards.is_some(),
provider_migration_url,
migrations,
directory_migration_required,
shard_map_migration_required,
)
.await;
let (replica_readiness, replica_migration_check) = if topology
.as_ref()
.is_some_and(|topology| check_replica_migrations && topology.replica().is_some())
{
match (
config.database.effective_primary_url(),
config.database.replica_url.as_deref(),
) {
(Some(primary_url), Some(replica_url)) => {
let primary_url = primary_url.to_owned();
let replica_url = replica_url.to_owned();
let readiness = crate::migrate::check_replica_migration_readiness_blocking(
primary_url.clone(),
replica_url.clone(),
)
.await;
(Some(readiness), Some((primary_url, replica_url)))
}
_ => (None, None),
}
} else {
(None, None)
};
if check_replica_migrations && let Some(set) = &shards {
check_shard_replica_migration_parity(config, set).await;
}
#[allow(clippy::question_mark)]
if let Err(e) = Box::pin(enforce_shard_map_guard(
config,
topology.as_ref(),
runtime_boot,
))
.await
{
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
return Err(e);
}
Ok(DatabaseBootstrap {
topology,
shards,
replica_readiness,
replica_migration_check,
})
}
#[cfg(feature = "db")]
fn apply_pending_or_exit(
database_url: &str,
migrations: &crate::migrate::EmbeddedMigrations,
target: &str,
) -> usize {
match crate::migrate::run_pending_locked(
database_url,
crate::migrate::EmbeddedMigrationsRef(migrations),
None,
) {
Ok(result) => result.applied.len(),
Err(error) => {
let reason = match error {
crate::migrate::MigrationError::Connection(_) => {
"could not connect to the database"
}
crate::migrate::MigrationError::Migration(_) => "a migration failed to apply",
_ => "migration error",
};
eprintln!("autumn migrate: {reason} (target {target})");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop();
std::process::exit(1);
}
}
}
#[cfg(feature = "sqlite")]
fn apply_pending_sqlite_or_exit(
database_url: &str,
migrations: &crate::migrate::EmbeddedMigrations,
target: &str,
) -> usize {
if let Some(err) = crate::migrate::reject_in_memory_migrations(
database_url,
&crate::migrate::EmbeddedMigrationsRef(migrations),
) {
eprintln!("autumn migrate: {err} (target {target})");
std::process::exit(1);
}
match crate::migrate::run_pending_sqlite(
database_url,
crate::migrate::EmbeddedMigrationsRef(migrations),
) {
Ok(result) => result.applied.len(),
Err(error) => {
let reason = match error {
crate::migrate::MigrationError::Connection(_) => {
"could not connect to the database"
}
crate::migrate::MigrationError::Migration(_) => "a migration failed to apply",
_ => "migration error",
};
eprintln!("autumn migrate: {reason} (target {target})");
std::process::exit(1);
}
}
}
#[cfg(feature = "sqlite")]
fn sqlite_sharding_unsupported_guard(
control_url: Option<&str>,
control_sharding_required: bool,
shard_urls: &[&str],
) -> Result<(), String> {
fn is_sqlite(url: &str) -> bool {
crate::config::DatabaseBackend::detect(url) == Some(crate::config::DatabaseBackend::Sqlite)
}
if control_url.is_some_and(is_sqlite) && control_sharding_required {
return Err(
"SQLite deployments do not support sharding. The configured sqlite:// control target \
has sharding enabled (shards and/or the directory/shard-map control migrations), \
which is a Postgres-only capability \u{2014} remove the shard configuration to run \
on SQLite, or use a Postgres control database. Tracking: #1614."
.to_owned(),
);
}
if shard_urls.iter().copied().any(is_sqlite) {
return Err(
"SQLite deployments do not support sharding. A configured shard targets a SQLite \
database, and per-shard migration/fan-out is a Postgres-only capability \u{2014} \
remove the SQLite shard configuration to run on SQLite, or use Postgres shard \
targets. Tracking: #1614."
.to_owned(),
);
}
Ok(())
}
#[cfg(all(test, feature = "sqlite"))]
mod sqlite_sharding_unsupported_guard_tests {
use super::sqlite_sharding_unsupported_guard;
#[test]
fn sqlite_control_target_with_sharding_fails_fast() {
for url in [
"sqlite:///var/lib/app.db",
"sqlite://./relative.db",
"sqlite::memory:",
] {
let err = sqlite_sharding_unsupported_guard(Some(url), true, &[])
.expect_err("sqlite control target + sharding must be rejected");
assert!(
err.contains("do not support sharding"),
"message must name the sharding situation clearly: {err}"
);
assert!(
err.contains("#1614"),
"message must point at the tracking issue: {err}"
);
}
}
#[test]
fn sqlite_control_target_without_sharding_boots() {
for url in [
"sqlite:///var/lib/app.db",
"sqlite://./relative.db",
"sqlite::memory:",
] {
assert!(
sqlite_sharding_unsupported_guard(Some(url), false, &[]).is_ok(),
"sqlite target without sharding must boot (migrations now applied): {url}"
);
}
}
#[test]
fn postgres_target_is_unchanged() {
for url in [
"postgres://u@h/db",
"postgresql://user:pass@db:5432/app",
"host=db user=app sslmode=require",
] {
assert!(
sqlite_sharding_unsupported_guard(Some(url), true, &[]).is_ok(),
"postgres target must never be gated: {url}"
);
assert!(
sqlite_sharding_unsupported_guard(Some(url), false, &[]).is_ok(),
"postgres target must never be gated: {url}"
);
}
}
#[test]
fn absent_control_url_boots() {
assert!(sqlite_sharding_unsupported_guard(None, true, &[]).is_ok());
assert!(sqlite_sharding_unsupported_guard(None, false, &[]).is_ok());
}
#[test]
fn sqlite_shard_target_fails_fast() {
for shards in [
&["sqlite:///var/lib/shard0.db"][..],
&["sqlite:///var/lib/shard0.db", "postgres://u@h/shard1"][..],
] {
let err = sqlite_sharding_unsupported_guard(None, false, shards)
.expect_err("a sqlite shard target must be rejected");
assert!(
err.contains("do not support sharding") && err.contains("shard"),
"message must name the SQLite shard situation clearly: {err}"
);
assert!(
err.contains("#1614"),
"message must point at the tracking issue: {err}"
);
}
}
#[test]
fn postgres_shard_targets_are_unchanged() {
assert!(
sqlite_sharding_unsupported_guard(
Some("postgres://u@h/control"),
true,
&["postgres://u@h/shard0", "postgres://u@h/shard1"],
)
.is_ok(),
"all-postgres shard targets must never be gated"
);
}
#[test]
fn migrate_only_mode_reuses_the_boot_guard_for_sqlite_targets() {
let err = sqlite_sharding_unsupported_guard(Some("sqlite:///var/lib/app.db"), true, &[])
.expect_err("sqlite migrate control target with sharding must be rejected");
assert!(
err.contains("do not support sharding") && err.contains("#1614"),
"migrate-only sqlite control error must be the actionable sharding message: {err}"
);
let shard_err = sqlite_sharding_unsupported_guard(
Some("postgres://u@h/control"),
true,
&["sqlite:///var/lib/shard0.db"],
)
.expect_err("sqlite migrate shard target must be rejected");
assert!(
shard_err.contains("do not support sharding") && shard_err.contains("shard"),
"migrate-only sqlite shard error must be the actionable sharding message: {shard_err}"
);
assert!(
sqlite_sharding_unsupported_guard(
Some("postgres://u@h/control"),
true,
&["postgres://u@h/shard0"],
)
.is_ok(),
"an all-postgres migrate configuration must proceed unchanged"
);
assert!(
sqlite_sharding_unsupported_guard(Some("sqlite:///var/lib/app.db"), false, &[]).is_ok(),
"sqlite control target without sharding must never be gated"
);
}
}
#[cfg(feature = "db")]
#[allow(clippy::too_many_arguments, clippy::fn_params_excessive_bools)]
async fn run_startup_migrations(
config: &AutumnConfig,
control_configured: bool,
shards_configured: bool,
provider_migration_url: Option<String>,
migrations: Vec<crate::migrate::EmbeddedMigrations>,
directory_migration_required: bool,
shard_map_migration_required: bool,
) {
let control_url = if control_configured {
provider_migration_url
.or_else(|| config.database.effective_primary_url().map(str::to_owned))
} else {
None
};
let shard_targets: Vec<(String, String)> = if shards_configured {
config
.database
.shards
.iter()
.map(|shard| (format!("shard:{}", shard.name), shard.primary_url.clone()))
.collect()
} else {
Vec::new()
};
let profile = config.profile.clone();
let auto_in_prod = config.database.auto_migrate_in_production;
let migration_result = tokio::task::spawn_blocking(move || {
#[cfg(feature = "sqlite")]
if let Some(url) = control_url.as_deref()
&& crate::config::DatabaseBackend::detect(url)
== Some(crate::config::DatabaseBackend::Sqlite)
{
for mig in &migrations {
crate::migrate::auto_migrate_sqlite(
url,
profile.as_deref(),
auto_in_prod,
mig,
"control",
);
}
return;
}
if let Some(url) = control_url {
for mig in &migrations {
crate::migrate::auto_migrate(
&url,
profile.as_deref(),
auto_in_prod,
mig,
"control",
);
}
if directory_migration_required {
crate::migrate::auto_migrate(
&url,
profile.as_deref(),
auto_in_prod,
&crate::sharding::SHARD_DIRECTORY_MIGRATIONS,
"control",
);
}
if shard_map_migration_required {
crate::migrate::auto_migrate(
&url,
profile.as_deref(),
true,
&crate::sharding::SHARD_MAP_MIGRATIONS,
"control",
);
}
}
for (target, url) in &shard_targets {
for mig in migrations
.iter()
.filter(|mig| !migration_set_is_control_framework(mig))
{
crate::migrate::auto_migrate(url, profile.as_deref(), auto_in_prod, mig, target);
}
}
})
.await;
if let Err(e) = migration_result {
tracing::error!(error = %e, "Migration task panicked");
#[cfg(feature = "managed-pg")]
crate::managed_pg::emergency_stop_async().await;
std::process::exit(1);
}
}
#[cfg(feature = "db")]
async fn check_shard_replica_migration_parity(
config: &AutumnConfig,
set: &crate::sharding::ShardSet,
) {
for (shard_config, shard) in config.database.shards.iter().zip(set.iter()) {
let Some(replica_url) = shard_config.replica_url.as_deref() else {
continue;
};
shard
.runtime()
.configure_migration_check(shard_config.primary_url.clone(), replica_url.to_owned());
let _ = shard.runtime().parity_check_due();
let readiness = crate::migrate::check_replica_migration_readiness_blocking(
shard_config.primary_url.clone(),
replica_url.to_owned(),
)
.await;
if readiness.is_ready() {
shard.runtime().mark_replica_migrations_ready();
} else if let Some(detail) = readiness.detail() {
tracing::warn!(
shard = %shard.name(),
detail = %detail,
"shard replica migrations are not ready"
);
shard.runtime().mark_replica_migrations_unready(detail);
}
}
}
#[cfg(feature = "db")]
const REPOSITORY_COMMIT_HOOK_QUEUE_MIGRATION: &str =
"20260515000000_create_repository_commit_hook_queue";
#[cfg(feature = "db")]
const VERSION_HISTORY_MIGRATION: &str = "20260526000000_create_version_history";
#[cfg(feature = "db")]
const fn directory_migration_is_required(
directory_routing_enabled: bool,
has_shards: bool,
mode: RepositoryCommitHookQueueMigrationMode,
) -> bool {
directory_routing_enabled
&& has_shards
&& matches!(mode, RepositoryCommitHookQueueMigrationMode::Runtime)
}
#[cfg(feature = "db")]
const fn shard_map_migration_is_required(
has_shards: bool,
mode: RepositoryCommitHookQueueMigrationMode,
) -> bool {
has_shards && matches!(mode, RepositoryCommitHookQueueMigrationMode::Runtime)
}
#[cfg(feature = "db")]
#[derive(diesel::QueryableByName)]
struct ShardMapRow {
#[diesel(sql_type = diesel::sql_types::Text)]
shard_name: String,
#[diesel(sql_type = diesel::sql_types::Text)]
slots: String,
}
#[cfg(feature = "db")]
pub async fn run_shard_map_guard(
control_pool: &deadpool::managed::Pool<
diesel_async::pooled_connection::AsyncDieselConnectionManager<
diesel_async::AsyncPgConnection,
>,
>,
computed: &[crate::config::ShardSlotAssignment],
auto_split: bool,
) -> Result<(), String> {
use diesel_async::RunQueryDsl as _;
if !auto_split {
return Ok(());
}
let mut conn = match control_pool.get().await {
Ok(conn) => conn,
Err(e) => {
return Err(format!(
"shard-map guard could not acquire a control connection: {e} — \
ensure the control database is reachable to enforce topology \
change detection"
));
}
};
let rows: Vec<ShardMapRow> = match diesel::sql_query(
"SELECT shard_name, slots FROM _autumn_shard_map ORDER BY shard_name",
)
.load::<ShardMapRow>(&mut conn)
.await
{
Ok(rows) => rows,
Err(e) => {
return Err(format!(
"shard-map guard could not read _autumn_shard_map: {e} — \
run `autumn migrate` to create the control schema before \
starting with auto-split shards"
));
}
};
let stored: Vec<crate::config::ShardSlotAssignment> = rows
.into_iter()
.map(|r| crate::config::ShardSlotAssignment {
name: r.shard_name,
ranges: r.slots,
})
.collect();
let stored_opt = if stored.is_empty() {
None
} else {
Some(stored.as_slice())
};
crate::config::check_stored_slot_map(auto_split, computed, stored_opt)?;
if stored.is_empty() {
use diesel_async::AsyncConnection as _;
let assignments: Vec<_> = computed.to_vec();
conn.transaction::<(), diesel::result::Error, _>(async move |conn| {
for assignment in &assignments {
diesel::sql_query(
"INSERT INTO _autumn_shard_map (shard_name, slots) VALUES ($1, $2) \
ON CONFLICT (shard_name) DO UPDATE \
SET slots = EXCLUDED.slots, updated_at = NOW()",
)
.bind::<diesel::sql_types::Text, _>(&assignment.name)
.bind::<diesel::sql_types::Text, _>(&assignment.ranges)
.execute(conn)
.await?;
}
Ok(())
})
.await
.map_err(|e| format!("shard-map guard could not persist map: {e}"))?;
}
Ok(())
}
#[cfg(all(feature = "db", feature = "sqlite"))]
#[allow(clippy::unused_async)]
async fn enforce_shard_map_guard(
config: &AutumnConfig,
topology: Option<&crate::db::DatabaseTopology>,
runtime_boot: bool,
) -> Result<(), String> {
let _ = (config, topology, runtime_boot);
Ok(())
}
#[cfg(all(feature = "db", not(feature = "sqlite")))]
async fn enforce_shard_map_guard(
config: &AutumnConfig,
topology: Option<&crate::db::DatabaseTopology>,
runtime_boot: bool,
) -> Result<(), String> {
if !runtime_boot || !config.database.has_shards() {
return Ok(());
}
let Some(topology) = topology else {
return Ok(());
};
if !config.database.shards_auto_split() {
return Ok(());
}
let computed = config
.database
.resolved_shard_assignments()
.map_err(|e| format!("shard-map guard: {e}"))?;
run_shard_map_guard(topology.primary(), &computed, true).await
}
#[cfg(feature = "db")]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RepositoryCommitHookQueueMigrationMode {
Runtime,
StaticBuild,
}
#[cfg(feature = "db")]
fn migrations_with_repository_framework_migrations(
mut migrations: Vec<crate::migrate::EmbeddedMigrations>,
hook_queue_required: bool,
version_history_required: bool,
mode: RepositoryCommitHookQueueMigrationMode,
) -> Vec<crate::migrate::EmbeddedMigrations> {
if hook_queue_required
&& mode == RepositoryCommitHookQueueMigrationMode::Runtime
&& !shard_applied_sets_include(&migrations, REPOSITORY_COMMIT_HOOK_QUEUE_MIGRATION)
{
migrations.push(crate::repository_commit_hooks::REPOSITORY_COMMIT_HOOK_MIGRATIONS);
}
if version_history_required
&& mode == RepositoryCommitHookQueueMigrationMode::Runtime
&& !shard_applied_sets_include(&migrations, VERSION_HISTORY_MIGRATION)
{
migrations.push(crate::version_history::VERSION_HISTORY_MIGRATIONS);
}
migrations
}
#[cfg(feature = "db")]
fn shard_applied_sets_include(
migrations: &[crate::migrate::EmbeddedMigrations],
migration_name: &str,
) -> bool {
use diesel::migration::{Migration, MigrationSource as _};
use diesel::pg::Pg;
migrations
.iter()
.filter(|set| !migration_set_is_control_framework(set))
.any(|source| {
let Ok(source_migrations): Result<Vec<Box<dyn Migration<Pg>>>, _> = source.migrations()
else {
return false;
};
source_migrations
.iter()
.any(|migration| migration.name().to_string() == migration_name)
})
}
#[cfg(feature = "db")]
fn migration_set_is_control_framework(set: &crate::migrate::EmbeddedMigrations) -> bool {
use diesel::migration::{Migration, MigrationSource as _};
use diesel::pg::Pg;
fn names(set: &crate::migrate::EmbeddedMigrations) -> std::collections::HashSet<String> {
let migrations: Vec<Box<dyn Migration<Pg>>> = set.migrations().unwrap_or_default();
migrations.iter().map(|m| m.name().to_string()).collect()
}
let mut control_only = names(&crate::migrate::FRAMEWORK_MIGRATIONS);
for shard_required in [
&crate::version_history::VERSION_HISTORY_MIGRATIONS,
&crate::repository_commit_hooks::REPOSITORY_COMMIT_HOOK_MIGRATIONS,
] {
for name in names(shard_required) {
control_only.remove(&name);
}
}
names(set).iter().any(|name| control_only.contains(name))
}
#[cfg(feature = "db")]
fn apply_replica_migration_readiness(
state: &AppState,
readiness: Option<crate::migrate::ReplicaMigrationReadiness>,
) {
let Some(readiness) = readiness else {
return;
};
if readiness.is_ready() {
state.probes().mark_replica_migrations_ready();
} else if let Some(detail) = readiness.detail() {
state.probes().mark_replica_migrations_unready(detail);
}
}
#[cfg(feature = "db")]
fn configure_replica_migration_check(state: &AppState, check: Option<(String, String)>) {
let Some((primary_url, replica_url)) = check else {
return;
};
state
.probes()
.configure_replica_migration_check(primary_url, replica_url);
}
fn collect_unguarded_repository_writes(
routes: &[Route],
scoped_groups: &[ScopedGroup],
) -> Vec<(String, String)> {
let mut offenders: Vec<(String, String)> = Vec::new();
let mut seen: std::collections::HashSet<(&'static str, &'static str)> =
std::collections::HashSet::new();
let mut record_route = |route: &Route| {
if let Some(meta) = route.repository
&& !meta.has_policy
&& is_mutating_method(&route.method)
&& seen.insert((meta.resource_type_name, meta.api_path))
{
offenders.push((meta.resource_type_name.to_owned(), meta.api_path.to_owned()));
}
};
for route in routes {
record_route(route);
}
for group in scoped_groups {
for route in &group.routes {
record_route(route);
}
}
offenders
}
fn format_unguarded_repository_listing(offenders: &[(String, String)]) -> String {
use std::fmt::Write;
let mut s = String::new();
let mut first = true;
for (name, path) in offenders {
if !first {
s.push('\n');
}
first = false;
write!(s, " - #[repository({name}, api = \"{path}\")]").unwrap();
}
s
}
fn validate_repository_api_policies(
routes: &[Route],
scoped_groups: &[ScopedGroup],
config: &AutumnConfig,
) {
let profile = config.profile.as_deref().unwrap_or("default");
let strict =
is_production_profile(profile) && !config.security.allow_unauthorized_repository_api;
let offenders = collect_unguarded_repository_writes(routes, scoped_groups);
if offenders.is_empty() {
return;
}
let listing = format_unguarded_repository_listing(&offenders);
if strict {
tracing::error!(
"refusing to start: the following #[repository(api = ...)] mutating endpoints have no paired `policy = ...` argument:\n{listing}\n\
Add `policy = SomePolicy` to each, or set `[security] allow_unauthorized_repository_api = true` to opt out explicitly."
);
std::process::exit(1);
} else {
tracing::warn!(
"the following #[repository(api = ...)] mutating endpoints have no paired `policy = ...` argument; \
auto-generated POST/PUT/PATCH/DELETE handlers will accept writes from any authenticated user:\n{listing}\n\
This will become a startup-time error in `prod` profile builds."
);
}
}
type MissingRepositoryRegistration = (String, String);
fn collect_unregistered_repository_handlers(
routes: &[Route],
scoped_groups: &[ScopedGroup],
registry: &crate::authorization::PolicyRegistry,
) -> (
Vec<MissingRepositoryRegistration>,
Vec<MissingRepositoryRegistration>,
) {
let mut missing_policies: Vec<(String, String)> = Vec::new();
let mut missing_scopes: Vec<(String, String)> = Vec::new();
let mut seen_policies: std::collections::HashSet<(&'static str, &'static str)> =
std::collections::HashSet::new();
let mut seen_scopes: std::collections::HashSet<(&'static str, &'static str)> =
std::collections::HashSet::new();
let mut record_route = |route: &Route| {
if let Some(meta) = route.repository {
if let Some(check) = meta.policy_check
&& !check(registry)
&& seen_policies.insert((meta.resource_type_name, meta.api_path))
{
missing_policies
.push((meta.resource_type_name.to_owned(), meta.api_path.to_owned()));
}
if let Some(check) = meta.scope_check
&& !check(registry)
&& seen_scopes.insert((meta.resource_type_name, meta.api_path))
{
missing_scopes.push((meta.resource_type_name.to_owned(), meta.api_path.to_owned()));
}
}
};
for route in routes {
record_route(route);
}
for group in scoped_groups {
for route in &group.routes {
record_route(route);
}
}
(missing_policies, missing_scopes)
}
fn format_missing_policy_listing(missing: &[(String, String)]) -> String {
use std::fmt::Write;
let mut s = String::new();
let mut first = true;
for (name, path) in missing {
if !first {
s.push('\n');
}
first = false;
write!(s, " - #[repository({name}, api = \"{path}\", policy = ...)]: call `.policy::<{name}, _>(...)` on the app builder").unwrap();
}
s
}
fn format_missing_scope_listing(missing: &[(String, String)]) -> String {
use std::fmt::Write;
let mut s = String::new();
let mut first = true;
for (name, path) in missing {
if !first {
s.push('\n');
}
first = false;
write!(s, " - #[repository({name}, api = \"{path}\", scope = ...)]: call `.scope::<{name}, _>(...)` on the app builder").unwrap();
}
s
}
#[allow(clippy::cognitive_complexity)]
fn validate_repository_policies_registered(
routes: &[Route],
scoped_groups: &[ScopedGroup],
state: &AppState,
config: &AutumnConfig,
) {
let profile = config.profile.as_deref().unwrap_or("default");
let strict = is_production_profile(profile);
let (missing_policies, missing_scopes) =
collect_unregistered_repository_handlers(routes, scoped_groups, state.policy_registry());
if missing_policies.is_empty() && missing_scopes.is_empty() {
return;
}
if !missing_policies.is_empty() {
let listing = format_missing_policy_listing(&missing_policies);
if strict {
tracing::error!(
"refusing to start: the following #[repository] routes declare a `policy = ...` argument, but no policy is registered for the resource type. Without registration, every protected request would fail at runtime with `500 no policy registered`:\n{listing}"
);
} else {
tracing::warn!(
"the following #[repository] routes declare `policy = ...` but no matching `.policy::<R, _>(...)` registration is on the app builder. Protected requests will 500 at runtime:\n{listing}\n\
This will become a startup-time error in `prod` profile builds."
);
}
}
if !missing_scopes.is_empty() {
let listing = format_missing_scope_listing(&missing_scopes);
if strict {
tracing::error!(
"refusing to start: the following #[repository] routes declare a `scope = ...` argument, but no scope is registered for the resource type. Without registration, every list request would fail at runtime with `500 missing scope registration`:\n{listing}"
);
} else {
tracing::warn!(
"the following #[repository] routes declare `scope = ...` but no matching `.scope::<R, _>(...)` registration is on the app builder. List requests will 500 at runtime:\n{listing}\n\
This will become a startup-time error in `prod` profile builds."
);
}
}
if strict {
std::process::exit(1);
}
}
const fn is_mutating_method(method: &http::Method) -> bool {
matches!(
*method,
http::Method::POST | http::Method::PUT | http::Method::PATCH | http::Method::DELETE
)
}
fn is_production_profile(profile: &str) -> bool {
matches!(profile, "prod" | "production")
}
#[cfg(test)]
mod validate_repository_api_policies_tests {
use super::*;
use crate::RepositoryApiMeta;
fn build_route(
method: http::Method,
path: &'static str,
meta: Option<RepositoryApiMeta>,
) -> Route {
Route {
method,
path,
handler: axum::routing::any(|| async { "" }),
name: "test_route",
api_doc: crate::openapi::ApiDoc::default(),
repository: meta,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}
}
fn unguarded(path: &'static str, type_name: &'static str) -> RepositoryApiMeta {
RepositoryApiMeta {
resource_type_name: type_name,
api_path: path,
has_policy: false,
policy_check: None,
scope_check: None,
}
}
fn collect_offenders(routes: &[Route]) -> Vec<(String, String)> {
collect_unguarded_repository_writes(routes, &[])
}
#[test]
fn read_only_mount_without_policy_is_not_an_offender() {
let routes = vec![
build_route(
http::Method::GET,
"/api/posts",
Some(unguarded("/api/posts", "Post")),
),
build_route(
http::Method::GET,
"/api/posts/{id}",
Some(unguarded("/api/posts", "Post")),
),
];
let offenders = collect_offenders(&routes);
assert!(
offenders.is_empty(),
"read-only mounts should not trigger the unauthorized-repo guard"
);
}
#[test]
fn write_mount_without_policy_is_an_offender() {
let routes = vec![build_route(
http::Method::POST,
"/api/posts",
Some(unguarded("/api/posts", "Post")),
)];
let offenders = collect_offenders(&routes);
assert_eq!(offenders.len(), 1);
assert_eq!(offenders[0].0, "Post");
assert_eq!(offenders[0].1, "/api/posts");
}
#[test]
fn mixed_mount_only_dedups_one_offender_per_repository() {
let routes = vec![
build_route(
http::Method::GET,
"/api/posts",
Some(unguarded("/api/posts", "Post")),
),
build_route(
http::Method::POST,
"/api/posts",
Some(unguarded("/api/posts", "Post")),
),
build_route(
http::Method::PUT,
"/api/posts/{id}",
Some(unguarded("/api/posts", "Post")),
),
build_route(
http::Method::DELETE,
"/api/posts/{id}",
Some(unguarded("/api/posts", "Post")),
),
];
let offenders = collect_offenders(&routes);
assert_eq!(offenders.len(), 1);
}
#[test]
fn is_mutating_method_classifies_methods() {
assert!(is_mutating_method(&http::Method::POST));
assert!(is_mutating_method(&http::Method::PUT));
assert!(is_mutating_method(&http::Method::PATCH));
assert!(is_mutating_method(&http::Method::DELETE));
assert!(!is_mutating_method(&http::Method::GET));
assert!(!is_mutating_method(&http::Method::HEAD));
assert!(!is_mutating_method(&http::Method::OPTIONS));
}
use crate::authorization::{Policy, PolicyRegistry};
#[derive(Debug, Clone, PartialEq)]
struct TestPost;
#[derive(Default)]
struct TestPostPolicy;
impl Policy<TestPost> for TestPostPolicy {}
fn guarded_with_check(path: &'static str, type_name: &'static str) -> RepositoryApiMeta {
RepositoryApiMeta {
resource_type_name: type_name,
api_path: path,
has_policy: true,
policy_check: Some(|registry: &PolicyRegistry| registry.has_policy::<TestPost>()),
scope_check: None,
}
}
fn collect_missing(routes: &[Route], registry: &PolicyRegistry) -> Vec<(String, String)> {
let (missing_policies, _) = collect_unregistered_repository_handlers(routes, &[], registry);
missing_policies
}
#[test]
fn registry_check_flags_routes_missing_their_policy_registration() {
let registry = PolicyRegistry::default();
let routes = vec![build_route(
http::Method::POST,
"/api/posts",
Some(guarded_with_check("/api/posts", "TestPost")),
)];
let missing = collect_missing(&routes, ®istry);
assert_eq!(missing.len(), 1);
assert_eq!(missing[0].0, "TestPost");
assert_eq!(missing[0].1, "/api/posts");
}
#[test]
fn registry_check_passes_when_policy_is_registered() {
let registry = PolicyRegistry::default();
registry.register_policy::<TestPost, _>(TestPostPolicy);
let routes = vec![build_route(
http::Method::POST,
"/api/posts",
Some(guarded_with_check("/api/posts", "TestPost")),
)];
let missing = collect_missing(&routes, ®istry);
assert!(missing.is_empty(), "policy is registered, no offenders");
}
#[test]
fn registry_check_skips_routes_without_policy_check_fn() {
let registry = PolicyRegistry::default();
let routes = vec![build_route(
http::Method::POST,
"/api/posts",
Some(unguarded("/api/posts", "TestPost")),
)];
let missing = collect_missing(&routes, ®istry);
assert!(missing.is_empty());
}
#[test]
fn registry_check_dedups_one_offender_per_repository() {
let registry = PolicyRegistry::default();
let routes = vec![
build_route(
http::Method::GET,
"/api/posts",
Some(guarded_with_check("/api/posts", "TestPost")),
),
build_route(
http::Method::POST,
"/api/posts",
Some(guarded_with_check("/api/posts", "TestPost")),
),
build_route(
http::Method::DELETE,
"/api/posts/{id}",
Some(guarded_with_check("/api/posts", "TestPost")),
),
];
let missing = collect_missing(&routes, ®istry);
assert_eq!(missing.len(), 1);
}
use crate::authorization::{BoxFuture, PolicyContext, Scope};
#[derive(Default)]
struct TestPostScope;
impl Scope<TestPost> for TestPostScope {
fn list<'a>(
&'a self,
_ctx: &'a PolicyContext,
_conn: &'a mut crate::db::RuntimeConnection,
) -> BoxFuture<'a, crate::AutumnResult<Vec<TestPost>>> {
Box::pin(async { Ok(Vec::new()) })
}
}
fn scope_only_meta(path: &'static str, type_name: &'static str) -> RepositoryApiMeta {
RepositoryApiMeta {
resource_type_name: type_name,
api_path: path,
has_policy: false,
policy_check: None,
scope_check: Some(|registry: &PolicyRegistry| registry.scope::<TestPost>().is_some()),
}
}
fn collect_missing_scopes(
routes: &[Route],
registry: &PolicyRegistry,
) -> Vec<(String, String)> {
let (_, missing_scopes) = collect_unregistered_repository_handlers(routes, &[], registry);
missing_scopes
}
#[test]
fn scope_check_flags_unregistered_scope() {
let registry = PolicyRegistry::default();
let routes = vec![build_route(
http::Method::GET,
"/api/posts",
Some(scope_only_meta("/api/posts", "TestPost")),
)];
let missing = collect_missing_scopes(&routes, ®istry);
assert_eq!(missing.len(), 1);
assert_eq!(missing[0].0, "TestPost");
}
#[test]
fn scope_check_passes_when_scope_is_registered() {
let registry = PolicyRegistry::default();
registry.register_scope::<TestPost, _>(TestPostScope);
let routes = vec![build_route(
http::Method::GET,
"/api/posts",
Some(scope_only_meta("/api/posts", "TestPost")),
)];
let missing = collect_missing_scopes(&routes, ®istry);
assert!(missing.is_empty());
}
#[test]
fn scope_check_skips_routes_without_scope_check_fn() {
let registry = PolicyRegistry::default();
let routes = vec![build_route(
http::Method::POST,
"/api/posts",
Some(unguarded("/api/posts", "TestPost")),
)];
let missing = collect_missing_scopes(&routes, ®istry);
assert!(missing.is_empty());
}
#[test]
fn is_production_profile_matches_both_aliases() {
assert!(is_production_profile("prod"));
assert!(is_production_profile("production"));
assert!(!is_production_profile("dev"));
assert!(!is_production_profile("staging"));
assert!(!is_production_profile("test"));
assert!(!is_production_profile("default"));
assert!(!is_production_profile("Prod"));
assert!(!is_production_profile("Production"));
}
#[test]
fn format_unguarded_listing_renders_one_bullet_per_offender() {
let offenders = vec![
("Post".to_owned(), "/api/posts".to_owned()),
("Comment".to_owned(), "/api/comments".to_owned()),
];
let listing = format_unguarded_repository_listing(&offenders);
assert!(listing.contains("Post"));
assert!(listing.contains("/api/posts"));
assert!(listing.contains("Comment"));
assert!(listing.contains("/api/comments"));
assert_eq!(listing.matches("\n - ").count() + 1, 2);
}
#[test]
fn format_unguarded_listing_empty_input_yields_empty_string() {
let listing = format_unguarded_repository_listing(&[]);
assert!(listing.is_empty());
}
#[test]
fn format_missing_policy_listing_includes_policy_call_hint() {
let missing = vec![("Post".to_owned(), "/api/posts".to_owned())];
let listing = format_missing_policy_listing(&missing);
assert!(listing.contains("Post"));
assert!(listing.contains("/api/posts"));
assert!(listing.contains(".policy::<Post, _>"));
assert!(listing.contains("policy = ..."));
}
#[test]
fn format_missing_scope_listing_includes_scope_call_hint() {
let missing = vec![("Post".to_owned(), "/api/posts".to_owned())];
let listing = format_missing_scope_listing(&missing);
assert!(listing.contains("Post"));
assert!(listing.contains("/api/posts"));
assert!(listing.contains(".scope::<Post, _>"));
assert!(listing.contains("scope = ..."));
}
#[test]
fn collect_unguarded_walks_scoped_groups() {
let group_route = build_route(
http::Method::POST,
"/api/posts",
Some(unguarded("/api/posts", "Post")),
);
let group = ScopedGroup {
prefix: "/scoped".to_owned(),
routes: vec![group_route],
source: crate::route_listing::RouteSource::User,
apply_layer: Box::new(|r| r),
};
let offenders = collect_unguarded_repository_writes(&[], std::slice::from_ref(&group));
assert_eq!(offenders.len(), 1);
assert_eq!(offenders[0].0, "Post");
}
#[test]
fn collect_unregistered_walks_scoped_groups() {
let group_route = build_route(
http::Method::POST,
"/api/posts",
Some(guarded_with_check("/api/posts", "TestPost")),
);
let group = ScopedGroup {
prefix: "/scoped".to_owned(),
routes: vec![group_route],
source: crate::route_listing::RouteSource::User,
apply_layer: Box::new(|r| r),
};
let registry = PolicyRegistry::default();
let (missing, _) =
collect_unregistered_repository_handlers(&[], std::slice::from_ref(&group), ®istry);
assert_eq!(missing.len(), 1);
assert_eq!(missing[0].0, "TestPost");
}
}
#[cfg(feature = "maud")]
fn install_story_registry(state: &AppState, story_gallery: Option<crate::stories::StoryGallery>) {
if let Some(gallery) = story_gallery {
state.insert_extension(gallery.into_registry());
}
}
fn build_state(
config: &AutumnConfig,
#[cfg(feature = "db")] database_topology: Option<&crate::db::DatabaseTopology>,
#[cfg(feature = "db")] shards: Option<crate::sharding::ShardSet>,
#[cfg(feature = "ws")] channels_backend: Option<Arc<dyn crate::channels::ChannelsBackend>>,
) -> AppState {
#[cfg(feature = "ws")]
let shutdown = tokio_util::sync::CancellationToken::new();
#[cfg(feature = "ws")]
let channels = channels_backend.map_or_else(
|| {
crate::channels::Channels::from_config(&config.channels, shutdown.child_token())
.unwrap_or_else(|error| {
tracing::error!(error = %error, "Failed to configure channels backend");
std::process::exit(1);
})
},
crate::channels::Channels::with_shared_backend,
);
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())),
#[cfg(feature = "db")]
pool: database_topology.map(|topology| topology.primary().clone()),
#[cfg(feature = "db")]
replica_pool: database_topology.and_then(|topology| topology.replica().cloned()),
#[cfg(feature = "db")]
shards,
profile: config.profile.clone(),
role: config.role,
started_at: std::time::Instant::now(),
health_detailed: config.health.detailed,
probes: crate::probe::ProbeState::pending_startup(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new(&config.log.level),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::from_config(config),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(channels.clone()),
#[cfg(feature = "ws")]
channels,
#[cfg(feature = "ws")]
shutdown,
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: config.security.forbidden_response,
auth_session_key: config.auth.session_key.clone(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
#[cfg(feature = "db")]
if state.replica_pool.is_some() {
state
.probes()
.configure_replica_dependency(config.database.replica_fallback);
}
#[cfg(feature = "db")]
if let Some(set) = state.shards() {
crate::sharding::register_shard_health_indicators(set, &state.health_indicator_registry);
}
state.insert_extension(config.clone());
state.insert_extension(crate::step_up::StepUpGlobalConfig {
default_max_age_secs: config.auth.step_up.default_max_age_secs,
});
#[cfg(feature = "http-client")]
state.insert_extension(crate::http_client::SharedReqwestClient {
client: crate::http_client::Client::build_inner(&config.http.client),
timeout_secs: config.http.client.timeout_secs,
});
state
}
fn format_route_lines(
routes: &[Route],
scoped_groups: &[ScopedGroup],
config: &AutumnConfig,
) -> String {
use std::fmt::Write as _;
let mut out = String::new();
for route in routes {
let _ = write!(
out,
"\n {} {:<8} -> {}",
route.path, route.method, route.name
);
}
for group in scoped_groups {
for route in &group.routes {
let _ = write!(
out,
"\n {}{} {:<8} -> {} (scoped)",
group.prefix, route.path, route.method, route.name
);
}
}
let mut probe_paths = std::collections::HashSet::new();
for (path, name) in [
(config.health.live_path.as_str(), "live"),
(config.health.ready_path.as_str(), "ready"),
(config.health.startup_path.as_str(), "startup"),
(config.health.path.as_str(), "health"),
] {
if probe_paths.insert(path) {
let _ = write!(out, "\n {} {:<8} -> {}", path, "GET", name);
}
}
let _ = write!(
out,
"\n {} {:<8} -> actuator",
crate::actuator::actuator_route_glob(&config.actuator.prefix),
"GET"
);
#[cfg(feature = "htmx")]
{
out.push_str("\n /static/js/htmx.min.js GET -> htmx");
out.push_str("\n /static/js/autumn-htmx-csrf.js GET -> htmx csrf");
}
out
}
fn format_task_lines(tasks: &[crate::task::TaskInfo]) -> Option<String> {
use std::fmt::Write as _;
if tasks.is_empty() {
return None;
}
let mut out = String::new();
for task in tasks {
let schedule = task.schedule.to_string();
let _ = write!(out, "\n {} ({schedule})", task.name);
}
Some(out)
}
fn format_middleware_list(config: &AutumnConfig) -> String {
let mut items = vec![
"RequestId",
"SecurityHeaders",
"Session (in-memory)",
"ErrorPages",
];
if !config.cors.allowed_origins.is_empty() {
items.push("CORS");
}
if config.security.csrf.enabled {
items.push("CSRF");
}
items.push("Metrics");
items.join(", ")
}
fn mask_database_url(url: &str, pool_size: usize) -> String {
if let Ok(mut parsed_url) = url::Url::parse(url) {
if parsed_url.password().is_some() {
let _ = parsed_url.set_password(Some("****"));
return format!("{parsed_url} (pool_size={pool_size})");
}
format!("{parsed_url} (pool_size={pool_size})")
} else {
format!("**** (pool_size={pool_size})")
}
}
fn format_config_summary(config: &AutumnConfig) -> String {
let profile = config.profile.as_deref().unwrap_or("none");
let db_status = config.database.effective_primary_url().map_or_else(
|| "not configured".to_owned(),
|url| {
let primary = mask_database_url(url, config.database.effective_primary_pool_size());
if config.database.replica_url.is_some() {
format!(
"primary={primary}, replica=configured (pool_size={})",
config.database.effective_replica_pool_size()
)
} else {
primary
}
},
);
let telemetry_status = if config.telemetry.enabled {
let endpoint = config
.telemetry
.otlp_endpoint
.as_deref()
.unwrap_or("<missing endpoint>");
format!("{:?} -> {endpoint}", config.telemetry.protocol)
} else {
"disabled".to_owned()
};
format!(
"\
\n profile: {profile}\
\n server: {}:{}\
\n database: {db_status}\
\n log_level: {}\
\n log_format: {:?}\
\n telemetry: {telemetry_status}\
\n health: {} (detailed={})\
\n actuator: sensitive={}\
\n shutdown: prestop={}s drain={}s",
config.server.host,
config.server.port,
config.log.level,
config.log.format,
config.health.path,
config.health.detailed,
config.actuator.sensitive,
config.server.prestop_grace_secs,
config.server.shutdown_timeout_secs,
)
}
pub(crate) fn project_dir(subdir: &str, env: &dyn crate::config::Env) -> std::path::PathBuf {
env.var("AUTUMN_MANIFEST_DIR").map_or_else(
|_| std::path::PathBuf::from(subdir),
|d| std::path::PathBuf::from(d).join(subdir),
)
}
async fn shutdown_signal() {
let ctrl_c = async {
tokio::signal::ctrl_c()
.await
.expect("Failed to install Ctrl+C handler");
tracing::info!("Received Ctrl+C, starting graceful shutdown");
};
#[cfg(unix)]
let terminate = async {
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("Failed to install SIGTERM handler")
.recv()
.await;
tracing::info!("Received SIGTERM, starting graceful shutdown");
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
let canary_rollback = async {
canary_rollback_signal(std::path::Path::new(
crate::canary::CANARY_ROLLBACK_FLAG_FILE,
))
.await;
tracing::info!("Canary rollback signalled, starting graceful shutdown");
};
tokio::select! {
() = ctrl_c => {},
() = terminate => {},
() = canary_rollback => {},
}
}
async fn canary_rollback_signal(path: &std::path::Path) {
let interval = std::time::Duration::from_millis(500);
loop {
if tokio::fs::metadata(path).await.is_ok() {
return;
}
tokio::time::sleep(interval).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use std::sync::atomic::{AtomicUsize, Ordering};
use tower::ServiceExt;
#[test]
fn plugin_declares_config_section_via_build() {
struct DummyMediaPlugin;
impl crate::plugin::Plugin for DummyMediaPlugin {
fn build(self, app: AppBuilder) -> AppBuilder {
app.config_section("media")
}
}
let builder = app().plugin(DummyMediaPlugin);
assert!(
builder.has_config_section("media"),
"a plugin's build() must declare its [media] config section"
);
assert!(
!builder.has_config_section("definitely_not_a_root"),
"only explicitly-declared roots are registered — the seam is fail-closed"
);
}
fn omitted_for(builder: &AppBuilder) -> usize {
omitted_router_count(
builder.merge_routers.len(),
builder
.nest_routers
.iter()
.map(|(prefix, _)| prefix.as_str()),
&builder.declared_routes,
)
}
#[test]
fn documented_nest_then_declare_is_not_counted_as_omitted() {
let raw =
axum::Router::<AppState>::new().route("/ping", axum::routing::get(|| async { "pong" }));
let declared = vec![crate::route_listing::RouteInfo {
method: "GET".to_owned(),
path: "/admin/ping".to_owned(),
handler: "admin::ping".to_owned(),
..Default::default()
}];
let builder = app().nest("/admin", raw).declare_plugin_routes(declared);
assert_eq!(builder.nest_routers.len(), 1);
assert_eq!(builder.declared_routes.len(), 1);
assert_eq!(
omitted_for(&builder),
0,
"a nest whose endpoints are declared is enumerable and must not count as omitted",
);
}
#[test]
fn undeclared_nest_and_merge_still_count_as_omitted() {
let raw_nest =
axum::Router::<AppState>::new().route("/x", axum::routing::get(|| async { "x" }));
let raw_merge =
axum::Router::<AppState>::new().route("/y", axum::routing::get(|| async { "y" }));
let builder = app().nest("/v2", raw_nest).merge(raw_merge);
assert_eq!(builder.nest_routers.len(), 1);
assert_eq!(builder.merge_routers.len(), 1);
assert!(builder.declared_routes.is_empty());
assert_eq!(
omitted_for(&builder),
2,
"an undeclared nest and a merge are both opaque and must be reported",
);
}
#[test]
fn declared_routes_do_not_cover_a_rootless_merge() {
let raw_merge =
axum::Router::<AppState>::new().route("/y", axum::routing::get(|| async { "y" }));
let builder =
app()
.merge(raw_merge)
.declare_plugin_routes(vec![crate::route_listing::RouteInfo {
method: "GET".to_owned(),
path: "/admin/ok".to_owned(),
handler: "admin::ok".to_owned(),
..Default::default()
}]);
assert_eq!(
omitted_for(&builder),
1,
"a merge has no prefix to match declarations against and must always count",
);
}
#[test]
fn mixed_declared_and_undeclared_nests_count_only_the_undeclared() {
let declared_raw =
axum::Router::<AppState>::new().route("/ok", axum::routing::get(|| async { "ok" }));
let undeclared_raw = axum::Router::<AppState>::new()
.route("/opaque", axum::routing::get(|| async { "opaque" }));
let builder = app()
.nest("/admin", declared_raw)
.declare_plugin_routes(vec![crate::route_listing::RouteInfo {
method: "GET".to_owned(),
path: "/admin/ok".to_owned(),
handler: "admin::ok".to_owned(),
..Default::default()
}])
.nest("/raw", undeclared_raw);
assert_eq!(builder.nest_routers.len(), 2);
assert_eq!(builder.declared_routes.len(), 1);
assert_eq!(
omitted_for(&builder),
1,
"only the bare nest() is omitted; the declared mount is covered",
);
}
#[test]
fn prefix_match_respects_path_segment_boundaries() {
let raw = axum::Router::<AppState>::new().route("/x", axum::routing::get(|| async { "x" }));
let builder = app().nest("/admin", raw).declare_plugin_routes(vec![
crate::route_listing::RouteInfo {
method: "GET".to_owned(),
path: "/administrators".to_owned(),
handler: "other::index".to_owned(),
..Default::default()
},
]);
assert_eq!(
omitted_for(&builder),
1,
"`/administrators` is not under the `/admin` nest prefix; the nest stays omitted",
);
}
#[test]
fn is_dump_jobs_mode_only_true_for_exactly_one() {
temp_env::with_var("AUTUMN_DUMP_JOBS", Some("1"), || {
assert!(is_dump_jobs_mode(), "`1` must select the jobs-dump path");
});
temp_env::with_var("AUTUMN_DUMP_JOBS", Some("0"), || {
assert!(!is_dump_jobs_mode(), "`0` must not select the dump path");
});
temp_env::with_var("AUTUMN_DUMP_JOBS", Some("true"), || {
assert!(
!is_dump_jobs_mode(),
"only the literal `1` enables the mode"
);
});
temp_env::with_var("AUTUMN_DUMP_JOBS", None::<&str>, || {
assert!(!is_dump_jobs_mode(), "unset must not select the dump path");
});
}
#[test]
fn dump_jobs_manifest_includes_synthesized_durable_listener_default_queue() {
fn listener_handler(
_state: AppState,
_payload: serde_json::Value,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = crate::AutumnResult<()>> + Send + 'static>,
> {
Box::pin(async move { Ok(()) })
}
let durable = crate::events::ListenerInfo {
event_name: "UserSignedUp",
listener_name: "app::send_welcome_email".to_string(),
mode: crate::events::DispatchMode::Durable,
job_name: Some("__event_listener::send_welcome_email".to_string()),
max_attempts: 4,
initial_backoff_ms: 250,
handler: listener_handler,
};
let cfg = crate::config::JobQueuesConfig::strict_list(["critical"]);
let manifest = dump_jobs_manifest(&cfg, Vec::new(), vec![durable]);
assert_eq!(manifest, "queues = [\"critical\", \"default\"]\n");
}
#[cfg(feature = "db")]
const APP_TEST_MIGRATIONS: crate::migrate::EmbeddedMigrations =
diesel_migrations::embed_migrations!("test_migrations");
#[cfg(feature = "mail")]
struct MailTestNoopQueue;
#[cfg(feature = "mail")]
impl crate::mail::MailDeliveryQueue for MailTestNoopQueue {
fn enqueue<'a>(
&'a self,
_mail: crate::mail::Mail,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<(), crate::mail::MailError>> + Send + 'a>,
> {
Box::pin(async { Ok(()) })
}
}
#[cfg(feature = "mail")]
fn test_mail() -> crate::mail::Mail {
crate::mail::Mail::builder()
.to("test@example.com")
.subject("hi")
.text("hello")
.build()
.expect("test mail should build")
}
pub fn test_router(routes: Vec<Route>) -> axum::Router {
let config = AutumnConfig::default();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
crate::router::build_router(routes, &config, state)
}
#[tokio::test]
async fn canary_rollback_signal_resolves_when_flag_newly_written() {
let tmp = tempfile::TempDir::new().unwrap();
let path = tmp.path().join("canary-rollback.json");
let writer_path = path.clone();
let writer = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
crate::canary::CanaryState::write_rollback_flag(
&writer_path,
&crate::canary::RollbackSignal::default(),
)
.unwrap();
});
let signalled = tokio::time::timeout(
std::time::Duration::from_secs(5),
canary_rollback_signal(&path),
)
.await;
assert!(signalled.is_ok(), "rollback signal should resolve");
writer.await.unwrap();
}
#[tokio::test]
async fn canary_rollback_signal_resolves_immediately_when_flag_present_at_boot() {
let tmp = tempfile::TempDir::new().unwrap();
let path = tmp.path().join("canary-rollback.json");
crate::canary::CanaryState::write_rollback_flag(
&path,
&crate::canary::RollbackSignal::default(),
)
.unwrap();
let signalled = tokio::time::timeout(
std::time::Duration::from_secs(5),
canary_rollback_signal(&path),
)
.await;
assert!(
signalled.is_ok(),
"a flag present at boot must trigger rollback (sticky across restarts)"
);
}
#[cfg(feature = "db")]
#[test]
fn build_state_applies_replica_fallback_policy_to_read_routing() {
let mut config = AutumnConfig::default();
config.database.primary_url = Some("postgres://localhost/primary".to_owned());
config.database.primary_pool_size = Some(5);
config.database.replica_url = Some("postgres://localhost/replica".to_owned());
config.database.replica_pool_size = Some(2);
config.database.replica_fallback = crate::config::ReplicaFallback::Primary;
let topology = crate::db::create_topology(&config.database)
.expect("topology should build")
.expect("database should be configured");
let state = build_state(
&config,
Some(&topology),
None,
#[cfg(feature = "ws")]
None,
);
state
.probes()
.mark_replica_unready("replica migrations lag primary");
assert_eq!(state.read_pool().expect("read pool").status().max_size, 5);
}
#[test]
fn build_state_exposes_resolved_process_role() {
use crate::config::ProcessRole;
let mut config = AutumnConfig::default();
let state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
None,
);
assert_eq!(state.role(), ProcessRole::Combined);
assert!(state.role().serves_http());
assert!(state.role().runs_workers());
config.role = ProcessRole::Worker;
let state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
None,
);
assert_eq!(state.role(), ProcessRole::Worker);
assert!(state.role().runs_workers());
assert!(!state.role().serves_http());
}
#[cfg(feature = "db")]
#[tokio::test]
async fn custom_pool_provider_preserves_configured_replica_topology() {
struct PassthroughPoolProvider;
impl crate::db::DatabasePoolProvider for PassthroughPoolProvider {
async fn create_pool(
&self,
config: &crate::config::DatabaseConfig,
) -> Result<
Option<
diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
>,
crate::db::PoolError,
> {
crate::db::create_pool(config)
}
}
let mut config = AutumnConfig::default();
config.database.primary_url = Some("postgres://localhost/primary".to_owned());
config.database.primary_pool_size = Some(5);
config.database.replica_url = Some("postgres://localhost/replica".to_owned());
config.database.replica_pool_size = Some(2);
config.database.replica_fallback = crate::config::ReplicaFallback::FailReadiness;
let AppBuilder {
pool_provider_factory,
..
} = app().with_pool_provider(PassthroughPoolProvider);
let database = setup_database(
&config,
Vec::new(),
pool_provider_factory,
None,
None,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.expect("custom provider should build database topology");
let topology = database.topology.expect("database should be configured");
assert_eq!(topology.primary().status().max_size, 5);
assert_eq!(
topology
.replica()
.expect("custom provider should create replica pool")
.status()
.max_size,
2
);
let state = build_state(
&config,
Some(&topology),
None,
#[cfg(feature = "ws")]
None,
);
state
.probes()
.mark_replica_connection_unready("replica connection failed");
assert!(state.read_pool().is_none());
let (status, _) = crate::probe::readiness_response(&state).await;
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn custom_pool_provider_with_established_sqlite_pool_fails_closed_on_statement_timeout() {
struct RealSqlitePoolProvider;
impl crate::db::DatabasePoolProvider for RealSqlitePoolProvider {
async fn create_pool(
&self,
config: &crate::config::DatabaseConfig,
) -> Result<
Option<
diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
>,
crate::db::PoolError,
> {
let mut relaxed = config.clone();
relaxed.statement_timeout = None;
crate::db::create_pool(&relaxed)
}
}
let mut config = AutumnConfig::default();
config.database.primary_url = Some("sqlite::memory:".to_owned());
config.database.statement_timeout = Some(std::time::Duration::from_secs(30));
let AppBuilder {
pool_provider_factory,
shard_provider_factory,
..
} = app().with_pool_provider(RealSqlitePoolProvider);
let Err(err) = setup_database(
&config,
Vec::new(),
pool_provider_factory,
shard_provider_factory,
None,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
else {
panic!(
"sqlite + statement_timeout must fail closed once the provider establishes a pool"
);
};
assert!(
err.contains("database.statement_timeout") && err.contains("SQLite"),
"dispatch guard error must name the config key and SQLite, got: {err}"
);
}
#[cfg(feature = "sqlite")]
#[tokio::test]
async fn custom_pool_provider_no_database_mode_boots_with_statement_timeout() {
struct NoDatabaseProvider;
impl crate::db::DatabasePoolProvider for NoDatabaseProvider {
async fn create_pool(
&self,
_config: &crate::config::DatabaseConfig,
) -> Result<
Option<
diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
>,
crate::db::PoolError,
> {
Ok(None)
}
}
let mut config = AutumnConfig::default();
config.database.primary_url = Some("sqlite::memory:".to_owned());
config.database.statement_timeout = Some(std::time::Duration::from_secs(30));
let AppBuilder {
pool_provider_factory,
shard_provider_factory,
..
} = app().with_pool_provider(NoDatabaseProvider);
let bootstrap = setup_database(
&config,
Vec::new(),
pool_provider_factory,
shard_provider_factory,
None,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.expect("no-database provider must boot even with a nonzero statement_timeout");
assert!(
bootstrap.topology.is_none(),
"no-database mode must yield no control topology"
);
assert!(
bootstrap.shards.is_none(),
"no-database mode must yield no shard set"
);
}
#[cfg(feature = "db")]
fn sharded_test_config() -> AutumnConfig {
let mut config = AutumnConfig::default();
config.database.primary_url = Some("postgres://localhost/control".to_owned());
config.database.shards = vec![
crate::config::ShardConfig {
name: "shard0".to_owned(),
primary_url: "postgres://localhost/shard0".to_owned(),
slots: Some(vec![crate::config::SlotSpec::Range("0-8191".to_owned())]),
replica_url: None,
primary_pool_size: Some(3),
replica_pool_size: None,
replica_fallback: None,
},
crate::config::ShardConfig {
name: "shard1".to_owned(),
primary_url: "postgres://localhost/shard1".to_owned(),
slots: Some(vec![crate::config::SlotSpec::Range(
"8192-16383".to_owned(),
)]),
replica_url: Some("postgres://localhost/shard1_ro".to_owned()),
primary_pool_size: None,
replica_pool_size: Some(2),
replica_fallback: None,
},
];
config
}
#[cfg(feature = "db")]
#[tokio::test]
async fn setup_database_builds_shard_set_from_config() {
let config = sharded_test_config();
let database = setup_database(
&config,
Vec::new(),
None,
None,
None,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.expect("sharded config should bootstrap");
assert!(database.topology.is_some(), "control role configured");
let shards = database.shards.expect("shards configured");
assert_eq!(shards.len(), 2);
assert_eq!(
shards
.by_name("shard0")
.expect("shard0")
.primary_pool()
.status()
.max_size,
3
);
assert_eq!(
shards
.by_name("shard1")
.expect("shard1")
.replica_pool()
.expect("shard1 replica")
.status()
.max_size,
2
);
let state = build_state(
&config,
database.topology.as_ref(),
Some(shards),
#[cfg(feature = "ws")]
None,
);
let state_shards = state.shards().expect("state should expose shards");
assert_eq!(state_shards.len(), 2);
let routed = state_shards.route("tenant-1").await.expect("route");
assert!(["shard0", "shard1"].contains(&routed.name()));
}
#[cfg(feature = "db")]
#[tokio::test]
async fn custom_pool_provider_builds_shard_topologies() {
struct CountingProvider(std::sync::Arc<std::sync::atomic::AtomicUsize>);
impl crate::db::DatabasePoolProvider for CountingProvider {
async fn create_pool(
&self,
config: &crate::config::DatabaseConfig,
) -> Result<
Option<
diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
>,
crate::db::PoolError,
> {
crate::db::create_pool(config)
}
async fn create_shard_topology(
&self,
shard: &crate::config::ShardConfig,
defaults: &crate::config::DatabaseConfig,
) -> Result<crate::db::DatabaseTopology, crate::db::PoolError> {
self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
crate::db::create_shard_topology(shard, defaults)
}
}
let config = sharded_test_config();
let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let AppBuilder {
pool_provider_factory,
shard_provider_factory,
..
} = app().with_pool_provider(CountingProvider(calls.clone()));
let database = setup_database(
&config,
Vec::new(),
pool_provider_factory,
shard_provider_factory,
None,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
)
.await
.expect("provider should build shard topologies");
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
assert_eq!(database.shards.expect("shards").len(), 2);
}
#[cfg(feature = "db")]
#[test]
fn repository_commit_hook_worker_starts_after_job_runtime_initialization() {
let source = include_str!("app.rs").replace("\r\n", "\n");
let server_init = "initialize_job_runtime(
jobs,
&state,
&server_shutdown,";
let server_worker = "start_repository_commit_hook_worker(\n pool,\n server_shutdown.child_token(),\n );";
let task_init = "initialize_job_runtime(jobs, &state, &task_shutdown, &config.jobs, true)";
let task_worker = "start_repository_commit_hook_worker(\n pool,\n task_shutdown.child_token(),\n );";
assert!(
source
.find(server_init)
.expect("normal server path should initialize jobs")
< source
.find(server_worker)
.expect("normal server path should start repository hook worker"),
"normal server startup must initialize jobs before repository commit hooks can enqueue them"
);
assert!(
source
.find(task_init)
.expect("task runner path should initialize jobs")
< source
.find(task_worker)
.expect("task runner path should start repository hook worker"),
"task runner startup must initialize jobs before repository commit hooks can enqueue them"
);
}
#[cfg(feature = "db")]
#[test]
fn repository_commit_hook_workers_are_gated_on_worker_role() {
let source = include_str!("app.rs").replace("\r\n", "\n");
let primary_gate =
"if role.runs_workers()\n && let Some(pool) = state.pool().cloned()";
let shard_gate = "if role.runs_workers()\n && let Some(shards) = state.shards()";
assert!(
source.contains(primary_gate),
"primary-pool commit-hook worker must be gated on role.runs_workers()"
);
assert!(
source.contains(shard_gate),
"shard commit-hook workers must be gated on role.runs_workers()"
);
}
#[test]
fn state_initializers_run_before_job_runtime_initialization() {
let source = include_str!("app.rs").replace("\r\n", "\n");
let server_start = source
.find("pub async fn run(self)")
.expect("normal server path should exist");
let build_mode_start = source
.find("async fn run_build_mode(self)")
.expect("static build path should follow server path");
let task_start = source
.find("async fn run_one_off_task_mode(self, requested_name: String)")
.expect("task runner path should exist");
let server_source = &source[server_start..build_mode_start];
let task_source = &source[task_start..];
let server_init = "initialize_job_runtime(
jobs,
&state,
&server_shutdown,";
let task_init = "initialize_job_runtime(jobs, &state, &task_shutdown, &config.jobs, true)";
let server_initializer = server_source
.find("run_state_initializers(state_initializers, &state);")
.expect("normal server path should run state initializers");
let task_initializer = task_source
.find("run_state_initializers(state_initializers, &state);")
.expect("task runner path should run state initializers");
let server_job = server_source
.find(server_init)
.expect("normal server path should initialize jobs");
let task_job = task_source
.find(task_init)
.expect("task runner path should initialize jobs");
assert!(
server_initializer < server_job,
"normal server startup must install state-initialized resources before job workers start"
);
assert!(
task_initializer < task_job,
"task runner startup must install state-initialized resources before job workers start"
);
}
#[test]
fn static_builds_run_state_initializers_before_router_build() {
let source = include_str!("app.rs").replace("\r\n", "\n");
let build_mode_start = source
.find("async fn run_build_mode(self)")
.expect("static build path should exist");
let dump_mode_start = source
.find("async fn run_dump_routes_mode(self)")
.expect("route dump path should follow static build path");
let build_mode_source = &source[build_mode_start..dump_mode_start];
let state_initializer = build_mode_source
.find("run_state_initializers(state_initializers, &state);")
.expect("static build path should run state initializers");
let router_build = build_mode_source
.find("let router = crate::router::try_build_router_inner(")
.expect("static build path should build a router");
assert!(
state_initializer < router_build,
"static builds must install state-initialized resources before rendering routes"
);
}
#[test]
fn migrate_only_one_shot_applies_and_exits_without_serving() {
let source = include_str!("app.rs").replace("\r\n", "\n");
let run_start = source.find("pub async fn run(self)").expect("run() exists");
let run_end = source
.find("async fn run_build_mode(self)")
.expect("build mode follows run()");
let run_body = &source[run_start..run_end];
let dispatch = run_body
.find("if is_migrate_only_mode() {")
.expect("run() dispatches the migrate one-shot");
let server_start = run_body
.find("let Self {")
.expect("run() destructures self to start the server");
assert!(
dispatch < server_start,
"AUTUMN_MIGRATE must be handled before the server-start path"
);
let migrate_branch = &run_body[dispatch..server_start];
assert!(
migrate_branch.contains("self.run_migrate_only_mode().await;")
&& migrate_branch.contains("return;"),
"the migrate one-shot must run then return before server start"
);
let handler_start = source
.find("async fn run_migrate_only_mode(self)")
.expect("migrate handler exists");
let handler_end = source
.find("async fn run_one_off_task_mode(self, requested_name: String)")
.expect("one-off task handler follows the migrate handler");
let handler = &source[handler_start..handler_end];
assert!(
handler.contains("apply_pending_or_exit"),
"the migrate handler applies pending migrations per target"
);
assert!(
handler.contains("std::process::exit(0)"),
"the migrate handler exits after applying"
);
let guard_call = handler
.find("sqlite_sharding_unsupported_guard(")
.expect("migrate handler applies the SQLite sharding guard");
let first_apply = handler
.find("apply_pending_or_exit")
.expect("migrate handler applies per target");
assert!(
guard_call < first_apply,
"the SQLite guard must run BEFORE the migration loop / apply_pending_or_exit"
);
assert!(
!handler.contains("initialize_job_runtime")
&& !handler.contains("try_build_router_inner"),
"the migrate one-shot must not start the server"
);
let helper_start = source
.find("fn apply_pending_or_exit(")
.expect("apply_pending_or_exit exists");
let helper = &source[helper_start..helper_start + 1200];
assert!(
helper.contains("crate::migrate::run_pending_locked("),
"must reuse the shared locked applier, not duplicate migration logic"
);
assert!(
helper.contains("std::process::exit(1)"),
"a failed migration must exit non-zero (abort before cutover)"
);
}
#[cfg(feature = "db")]
#[test]
fn hooked_repository_apps_include_hook_queue_framework_migration() {
let migrations = migrations_with_repository_framework_migrations(
vec![APP_TEST_MIGRATIONS],
true,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let names = migration_names(&migrations);
assert!(
names
.iter()
.any(|name| name == REPOSITORY_COMMIT_HOOK_QUEUE_MIGRATION),
"hooked repository apps must auto-register the durable hook queue migration"
);
assert!(
names.iter().all(|name| !name.contains("api_tokens")),
"hooked repository apps must not auto-register unrelated framework migrations: {names:?}"
);
}
#[cfg(feature = "db")]
#[test]
fn runtime_hooked_apps_include_hook_queue_framework_migration_without_app_migrations() {
let migrations = migrations_with_repository_framework_migrations(
Vec::new(),
true,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let names = migration_names(&migrations);
assert!(
names
.iter()
.any(|name| name == REPOSITORY_COMMIT_HOOK_QUEUE_MIGRATION),
"runtime hooked repository apps must install the durable hook queue even when app migrations are managed elsewhere"
);
}
#[cfg(feature = "db")]
#[test]
fn versioned_repository_apps_include_version_history_framework_migration() {
let migrations = migrations_with_repository_framework_migrations(
vec![APP_TEST_MIGRATIONS],
false,
true,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let names = migration_names(&migrations);
assert!(
names.iter().any(|name| name == VERSION_HISTORY_MIGRATION),
"versioned repository apps must auto-register the version-history migration"
);
assert!(
names
.iter()
.all(|name| !name.contains("repository_commit_hook_queue")),
"versioned-only repository apps must not auto-register the durable hook queue: {names:?}"
);
}
#[cfg(feature = "db")]
#[test]
fn runtime_versioned_apps_include_version_history_framework_migration_without_app_migrations() {
let migrations = migrations_with_repository_framework_migrations(
Vec::new(),
false,
true,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let names = migration_names(&migrations);
assert!(
names.iter().any(|name| name == VERSION_HISTORY_MIGRATION),
"runtime versioned repository apps must install version history even when app migrations are managed elsewhere"
);
}
#[cfg(feature = "db")]
#[test]
fn static_builds_do_not_auto_add_hook_queue_when_no_migrations_registered() {
let migrations = migrations_with_repository_framework_migrations(
Vec::new(),
true,
true,
RepositoryCommitHookQueueMigrationMode::StaticBuild,
);
assert!(
migrations.is_empty(),
"static/export builds that pass no migrations must not mutate the database"
);
}
#[cfg(feature = "db")]
#[test]
fn directory_migration_required_only_at_runtime_with_shards_and_routing() {
use RepositoryCommitHookQueueMigrationMode::{Runtime, StaticBuild};
assert!(directory_migration_is_required(true, true, Runtime));
assert!(!directory_migration_is_required(true, true, StaticBuild));
assert!(!directory_migration_is_required(false, true, Runtime));
assert!(!directory_migration_is_required(true, false, Runtime));
}
#[test]
fn shard_map_migration_required_only_at_runtime_with_shards() {
use RepositoryCommitHookQueueMigrationMode::{Runtime, StaticBuild};
assert!(shard_map_migration_is_required(true, Runtime));
assert!(!shard_map_migration_is_required(true, StaticBuild));
assert!(!shard_map_migration_is_required(false, Runtime));
}
#[cfg(feature = "db")]
#[test]
fn unhooked_apps_do_not_auto_add_hook_queue_framework_migration() {
let migrations = migrations_with_repository_framework_migrations(
Vec::new(),
false,
false,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
assert!(
migrations.is_empty(),
"unhooked apps should not get durable hook queue migrations for free"
);
}
#[cfg(feature = "db")]
fn migration_names(migrations: &[crate::migrate::EmbeddedMigrations]) -> Vec<String> {
use diesel::migration::{Migration, MigrationSource as _};
use diesel::pg::Pg;
migrations
.iter()
.flat_map(|source| {
let migrations: Vec<Box<dyn Migration<Pg>>> = source.migrations().unwrap();
migrations
})
.map(|migration| migration.name().to_string())
.collect()
}
#[cfg(feature = "db")]
#[test]
fn control_framework_filter_skips_control_but_keeps_shard_required_sets() {
assert!(migration_set_is_control_framework(
&crate::migrate::FRAMEWORK_MIGRATIONS
));
assert!(!migration_set_is_control_framework(
&crate::version_history::VERSION_HISTORY_MIGRATIONS
));
assert!(!migration_set_is_control_framework(
&crate::repository_commit_hooks::REPOSITORY_COMMIT_HOOK_MIGRATIONS
));
}
#[cfg(feature = "db")]
#[test]
fn sharded_app_with_full_framework_still_gets_shard_required_sets() {
use diesel::migration::{Migration, MigrationSource as _};
use diesel::pg::Pg;
let migrations = migrations_with_repository_framework_migrations(
vec![crate::migrate::FRAMEWORK_MIGRATIONS],
true,
true,
RepositoryCommitHookQueueMigrationMode::Runtime,
);
let shard_names: Vec<String> = migrations
.iter()
.filter(|set| !migration_set_is_control_framework(set))
.flat_map(|set| {
let ms: Vec<Box<dyn Migration<Pg>>> = set.migrations().unwrap_or_default();
ms.into_iter()
.map(|m| m.name().to_string())
.collect::<Vec<_>>()
})
.collect();
assert!(
shard_names
.iter()
.any(|name| name == REPOSITORY_COMMIT_HOOK_QUEUE_MIGRATION),
"shards must receive the commit-hook queue migration even when the full \
control framework set is also registered: {shard_names:?}"
);
assert!(
shard_names
.iter()
.any(|name| name == VERSION_HISTORY_MIGRATION),
"shards must receive the version-history migration even when the full \
control framework set is also registered: {shard_names:?}"
);
}
#[cfg(feature = "db")]
#[test]
fn configure_replica_migration_check_stores_recheck_urls() {
let mut config = AutumnConfig::default();
config.database.primary_url = Some("postgres://localhost/primary".to_owned());
config.database.replica_url = Some("postgres://localhost/replica".to_owned());
let topology = crate::db::create_topology(&config.database)
.expect("topology should build")
.expect("database should be configured");
let state = build_state(
&config,
Some(&topology),
None,
#[cfg(feature = "ws")]
None,
);
assert!(
state.probes().replica_migration_check().is_none(),
"build_state should not enable migration checks without registered migrations"
);
configure_replica_migration_check(
&state,
Some((
"postgres://localhost/primary".to_owned(),
"postgres://localhost/replica".to_owned(),
)),
);
let check = state
.probes()
.replica_migration_check()
.expect("replica migration check should be configured");
assert_eq!(check.primary_url, "postgres://localhost/primary");
assert_eq!(check.replica_url, "postgres://localhost/replica");
}
#[cfg(feature = "db")]
#[tokio::test]
async fn replica_migration_readiness_marks_ready_endpoint_degraded() {
let mut config = AutumnConfig::default();
config.database.primary_url = Some("postgres://localhost/primary".to_owned());
config.database.primary_pool_size = Some(5);
config.database.replica_url = Some("postgres://localhost/replica".to_owned());
config.database.replica_pool_size = Some(2);
config.database.replica_fallback = crate::config::ReplicaFallback::FailReadiness;
let topology = crate::db::create_topology(&config.database)
.expect("topology should build")
.expect("database should be configured");
let state = build_state(
&config,
Some(&topology),
None,
#[cfg(feature = "ws")]
None,
);
apply_replica_migration_readiness(
&state,
Some(crate::migrate::ReplicaMigrationReadiness::Stale {
primary_latest: Some("00000000000002".to_owned()),
replica_latest: Some("00000000000001".to_owned()),
}),
);
let (status, _) = crate::probe::readiness_response(&state).await;
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
}
#[cfg(feature = "db")]
#[tokio::test]
async fn blocking_replica_migration_readiness_reports_unknown_connection_errors() {
let readiness = crate::migrate::check_replica_migration_readiness_blocking(
"not-a-primary-url".to_owned(),
"not-a-replica-url".to_owned(),
)
.await;
assert!(matches!(
readiness,
crate::migrate::ReplicaMigrationReadiness::Unknown(_)
));
}
#[cfg(feature = "ws")]
#[test]
fn with_channels_backend_overrides_config_driven_backend_selection() {
let builder = app().with_channels_backend(crate::channels::LocalChannelsBackend::new(4));
let AppBuilder {
channels_backend, ..
} = builder;
assert!(channels_backend.is_some());
let mut config = AutumnConfig::default();
config.channels.backend = crate::config::ChannelBackend::Redis;
config.channels.redis.url = None;
let state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
channels_backend,
);
let mut rx = state.channels().subscribe("override");
state
.broadcast()
.publish("override", "ok")
.expect("custom local backend should publish");
assert_eq!(rx.try_recv().expect("message should arrive").as_str(), "ok");
}
pub fn test_get_route(path: &'static str, name: &'static str) -> Route {
Route {
method: http::Method::GET,
path,
handler: axum::routing::get(|| async { "ok" }),
name,
api_doc: crate::openapi::ApiDoc {
method: "GET",
path,
operation_id: name,
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}
}
#[cfg(feature = "i18n")]
fn test_i18n_bundle(key: &str, value: &str) -> Arc<crate::i18n::Bundle> {
let mut messages = std::collections::HashMap::new();
let mut en = std::collections::HashMap::new();
en.insert(key.to_owned(), value.to_owned());
messages.insert("en".to_owned(), en);
Arc::new(crate::i18n::Bundle::from_messages(
messages,
&crate::i18n::I18nConfig::default(),
))
}
#[cfg(feature = "i18n")]
#[test]
fn i18n_auto_defers_loading_until_runtime_config_is_available() {
let builder = app().i18n_auto();
assert!(builder.i18n_bundle.is_none());
assert!(builder.i18n_auto_load);
}
#[cfg(feature = "i18n")]
#[derive(Clone)]
struct StaticConfigLoader {
config: AutumnConfig,
}
#[cfg(feature = "i18n")]
impl crate::config::ConfigLoader for StaticConfigLoader {
async fn load(&self) -> Result<AutumnConfig, crate::config::ConfigError> {
Ok(self.config.clone())
}
}
#[cfg(feature = "i18n")]
struct NoopTelemetryProvider;
#[cfg(feature = "i18n")]
impl crate::telemetry::TelemetryProvider for NoopTelemetryProvider {
fn init(
&self,
_log: &crate::config::LogConfig,
_telemetry: &crate::config::TelemetryConfig,
_profile: Option<&str>,
) -> Result<crate::telemetry::TelemetryGuard, crate::telemetry::TelemetryInitError>
{
Ok(crate::telemetry::TelemetryGuard::disabled())
}
}
#[cfg(feature = "i18n")]
#[tokio::test]
async fn i18n_auto_uses_config_loader_output_for_bundle_dir() {
let project = tempfile::tempdir().expect("project dir");
let i18n_dir = project.path().join("custom-i18n");
std::fs::create_dir_all(&i18n_dir).expect("i18n dir");
std::fs::write(i18n_dir.join("en.ftl"), "nav.home = Loader Home\n").expect("bundle");
let mut config = AutumnConfig::default();
config.i18n.dir = "custom-i18n".to_owned();
let builder = app()
.with_config_loader(StaticConfigLoader { config })
.with_telemetry_provider(NoopTelemetryProvider)
.i18n_auto();
let AppBuilder {
config_loader_factory,
telemetry_provider,
i18n_bundle,
i18n_auto_load,
plugin_config_roots,
..
} = builder;
let (loaded_config, _guard) = load_config_and_telemetry(
config_loader_factory,
telemetry_provider,
plugin_config_roots,
)
.await;
let env = crate::config::MockEnv::new().with(
"AUTUMN_MANIFEST_DIR",
project.path().to_str().expect("utf-8 path"),
);
let bundle = resolve_i18n_bundle(i18n_bundle, i18n_auto_load, &loaded_config, &env)
.expect("bundle loaded from configured dir");
assert_eq!(bundle.translate("en", "nav.home", &[]), "Loader Home");
}
#[cfg(feature = "i18n")]
#[tokio::test]
async fn i18n_bundle_layer_is_applied_to_static_route_rendering() {
async fn localized(locale: crate::i18n::Locale) -> String {
locale.t("nav.home")
}
let config = AutumnConfig::default();
let state = AppState::for_test();
let custom_layers = install_i18n_bundle_layer(
Vec::new(),
&state,
Some(test_i18n_bundle("nav.home", "Home")),
);
let router = crate::router::try_build_router_inner(
vec![Route {
method: http::Method::GET,
path: "/about",
handler: axum::routing::get(localized),
name: "localized",
api_doc: crate::openapi::ApiDoc {
method: "GET",
path: "/about",
operation_id: "localized",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}],
&config,
state,
crate::router::RouterContext {
exception_filters: Vec::new(),
scoped_groups: Vec::new(),
merge_routers: Vec::new(),
nest_routers: Vec::new(),
custom_layers,
static_gate_layers: Vec::new(),
#[cfg(feature = "maud")]
error_page_renderer: None,
session_store: None,
#[cfg(feature = "openapi")]
openapi: None,
#[cfg(feature = "mcp")]
mcp: None,
},
)
.expect("router builds");
let tmp = tempfile::tempdir().expect("dist parent");
let dist = tmp.path().join("dist");
crate::static_gen::render_static_routes(
router,
&[crate::static_gen::StaticRouteMeta {
path: "/about",
name: "localized",
revalidate: None,
params_fn: None,
}],
&dist,
)
.await
.expect("static render succeeds");
let html = std::fs::read_to_string(dist.join("about/index.html")).expect("rendered html");
assert_eq!(html, "Home");
}
#[test]
fn app_builder_routes_adds_routes() {
let builder = app();
assert_eq!(builder.routes.len(), 0);
let builder = builder.routes(vec![test_get_route("/1", "route1")]);
assert_eq!(builder.routes.len(), 1);
let builder = builder.routes(vec![
test_get_route("/2", "route2"),
test_get_route("/3", "route3"),
]);
assert_eq!(builder.routes.len(), 3);
assert_eq!(builder.routes[0].path, "/1");
assert_eq!(builder.routes[1].path, "/2");
assert_eq!(builder.routes[2].path, "/3");
}
#[test]
fn app_builder_extensions_store_and_update_typed_values() {
let builder = app()
.with_extension::<String>("haunted".into())
.update_extension::<String, _, _>(String::new, |value| value.push_str(" harvest"));
let value = builder
.extension::<String>()
.expect("string extension should be present");
assert_eq!(value, "haunted harvest");
}
#[cfg(feature = "mail")]
#[tokio::test]
async fn app_builder_with_mail_delivery_queue_stores_queue_for_install() {
let builder = app().with_mail_delivery_queue(MailTestNoopQueue);
let factory = builder
.mail_delivery_queue_factory
.expect("with_mail_delivery_queue should store a factory on the builder");
let state = AppState::for_test();
let queue = factory(&state).expect("trivial factory should produce the queue");
assert!(Arc::strong_count(&queue) >= 1);
queue
.enqueue(test_mail())
.await
.expect("noop queue should always succeed");
}
#[cfg(feature = "mail")]
#[test]
fn app_builder_with_mail_delivery_queue_factory_runs_with_app_state() {
let observed_profile: Arc<std::sync::Mutex<Option<String>>> =
Arc::new(std::sync::Mutex::new(None));
let captured = Arc::clone(&observed_profile);
let builder = app().with_mail_delivery_queue_factory(move |state| {
*captured.lock().expect("lock") = Some(state.profile().to_owned());
Ok::<_, crate::AutumnError>(MailTestNoopQueue)
});
let factory = builder
.mail_delivery_queue_factory
.expect("factory should be stored on the builder");
let state = AppState::for_test().with_profile("dev");
let _queue = factory(&state).expect("factory should succeed");
assert_eq!(
observed_profile.lock().expect("lock").as_deref(),
Some("dev"),
"factory must run with the live AppState"
);
}
#[cfg(feature = "mail")]
#[test]
fn app_builder_with_mail_delivery_queue_factory_propagates_errors() {
let builder = app().with_mail_delivery_queue_factory(|_state| {
Err::<MailTestNoopQueue, _>(crate::AutumnError::service_unavailable_msg("factory boom"))
});
let factory = builder
.mail_delivery_queue_factory
.expect("factory present");
let state = AppState::for_test();
match factory(&state) {
Ok(_) => panic!("factory should have errored"),
Err(err) => assert!(err.to_string().contains("factory boom")),
}
}
#[tokio::test]
async fn startup_and_shutdown_hooks_run_in_expected_order() {
let events = Arc::new(std::sync::Mutex::new(Vec::<&'static str>::new()));
let startup_events = Arc::clone(&events);
let shutdown_a = Arc::clone(&events);
let shutdown_b = Arc::clone(&events);
let builder = app()
.on_startup(move |_state| {
let startup_events = Arc::clone(&startup_events);
async move {
startup_events
.lock()
.expect("events lock poisoned")
.push("start");
Ok(())
}
})
.on_shutdown(move || {
let shutdown_a = Arc::clone(&shutdown_a);
async move {
shutdown_a
.lock()
.expect("events lock poisoned")
.push("stop-a");
}
})
.on_shutdown(move || {
let shutdown_b = Arc::clone(&shutdown_b);
async move {
shutdown_b
.lock()
.expect("events lock poisoned")
.push("stop-b");
}
});
run_startup_hooks(&builder.startup_hooks, AppState::for_test())
.await
.expect("startup hooks should succeed");
run_shutdown_hooks(&builder.shutdown_hooks).await;
let recorded_events = events.lock().expect("events lock poisoned").clone();
assert_eq!(recorded_events, vec!["start", "stop-b", "stop-a"]);
}
fn startup_noop_job_handler(
_state: AppState,
_payload: serde_json::Value,
) -> Pin<Box<dyn Future<Output = crate::AutumnResult<()>> + Send + 'static>> {
Box::pin(async move { Ok(()) })
}
#[tokio::test]
async fn startup_hooks_can_enqueue_jobs_after_runtime_init() {
let _guard = crate::job::global_job_runtime_test_lock().lock().await;
crate::job::clear_global_job_client();
let builder = app()
.jobs(vec![crate::job::JobInfo {
version: 1,
name: "startup-seed".to_string(),
max_attempts: 1,
initial_backoff_ms: 1,
queue: "default".to_string(),
uniqueness: None,
concurrency: None,
handler: startup_noop_job_handler,
}])
.on_startup(|_state| async {
crate::job::enqueue("startup-seed", serde_json::json!({ "kind": "warmup" })).await
});
let state = AppState::for_test().with_profile("dev");
let shutdown = tokio_util::sync::CancellationToken::new();
initialize_job_runtime(
builder.jobs.clone(),
&state,
&shutdown,
&crate::config::JobConfig::default(),
true,
)
.expect("job runtime should initialize before startup hooks");
run_startup_hooks(&builder.startup_hooks, state.clone())
.await
.expect("startup hook should be able to enqueue jobs");
tokio::time::timeout(std::time::Duration::from_secs(1), async {
loop {
let snapshot = state.job_registry().snapshot();
let status = snapshot
.get("startup-seed")
.expect("job should be registered before startup hooks run");
if status.total_successes == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.expect("startup-enqueued job should complete");
shutdown.cancel();
crate::job::clear_global_job_client();
}
#[tokio::test]
async fn initialize_job_runtime_propagates_redis_init_errors() {
let _guard = crate::job::global_job_runtime_test_lock().lock().await;
crate::job::clear_global_job_client();
let state = AppState::for_test().with_profile("dev");
let shutdown = tokio_util::sync::CancellationToken::new();
let config = crate::config::JobConfig {
backend: "redis".to_string(),
..Default::default()
};
let error = initialize_job_runtime(
vec![crate::job::JobInfo {
version: 1,
name: "startup-seed".to_string(),
max_attempts: 1,
initial_backoff_ms: 1,
queue: "default".to_string(),
uniqueness: None,
concurrency: None,
handler: startup_noop_job_handler,
}],
&state,
&shutdown,
&config,
true,
)
.expect_err("redis init errors should abort startup");
#[cfg(feature = "redis")]
assert!(
error
.to_string()
.contains("jobs.backend=redis requires jobs.redis.url"),
"unexpected error: {error}"
);
#[cfg(not(feature = "redis"))]
assert!(
error
.to_string()
.contains("jobs.backend=redis requested but redis feature is disabled"),
"unexpected error: {error}"
);
}
#[tokio::test]
async fn startup_hook_errors_propagate() {
let builder = app().on_startup(|_state| async {
Err(crate::AutumnError::service_unavailable_msg(
"startup ritual failed",
))
});
let error = run_startup_hooks(&builder.startup_hooks, AppState::for_test())
.await
.expect_err("startup hook should fail");
assert!(error.to_string().contains("startup ritual failed"));
}
#[tokio::test]
async fn build_router_mounts_user_routes() {
let router = test_router(vec![test_get_route("/test", "test_handler")]);
let response = router
.oneshot(Request::builder().uri("/test").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"ok");
}
#[tokio::test]
async fn build_router_mounts_health_check_at_default_path() {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri("/health")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["status"], "ok");
}
#[tokio::test]
async fn build_router_mounts_health_check_at_custom_path() {
let mut config = AutumnConfig::default();
config.health.path = "/healthz".to_owned();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
let router =
crate::router::build_router(vec![test_get_route("/dummy", "dummy")], &config, state);
let response = router
.oneshot(
Request::builder()
.uri("/healthz")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn build_router_adds_request_id_header() {
let router = test_router(vec![test_get_route("/test", "test")]);
let response = router
.oneshot(Request::builder().uri("/test").body(Body::empty()).unwrap())
.await
.unwrap();
assert!(response.headers().contains_key("x-request-id"));
}
#[tokio::test]
async fn build_router_unknown_route_returns_404() {
let router = test_router(vec![test_get_route("/exists", "exists")]);
let response = router
.oneshot(Request::builder().uri("/nope").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn build_router_multiple_routes() {
let router = test_router(vec![test_get_route("/a", "a"), test_get_route("/b", "b")]);
let resp_a = router
.clone()
.oneshot(Request::builder().uri("/a").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp_a.status(), StatusCode::OK);
let resp_b = router
.oneshot(Request::builder().uri("/b").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp_b.status(), StatusCode::OK);
}
#[tokio::test]
async fn build_router_post_route() {
let post_routes = vec![Route {
method: http::Method::POST,
path: "/submit",
handler: axum::routing::post(|| async { "posted" }),
name: "submit",
api_doc: crate::openapi::ApiDoc {
method: "POST",
path: "/submit",
operation_id: "submit",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}];
let config = AutumnConfig::default();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
let router = crate::router::build_router(post_routes, &config, state);
let response = router
.oneshot(
Request::builder()
.method("POST")
.uri("/submit")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn build_router_merges_methods_on_same_path() {
let route_list = vec![
Route {
method: http::Method::GET,
path: "/admin",
handler: axum::routing::get(|| async { "list" }),
name: "admin_list",
api_doc: crate::openapi::ApiDoc {
method: "GET",
path: "/admin",
operation_id: "admin_list",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
},
Route {
method: http::Method::POST,
path: "/admin",
handler: axum::routing::post(|| async { "created" }),
name: "create",
api_doc: crate::openapi::ApiDoc {
method: "POST",
path: "/admin",
operation_id: "create",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
},
];
let config = AutumnConfig::default();
let router = crate::router::build_router(route_list, &config, AppState::for_test());
let resp = router
.clone()
.oneshot(
Request::builder()
.uri("/admin")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"list");
let resp = router
.oneshot(
Request::builder()
.method("POST")
.uri("/admin")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"created");
}
#[cfg(feature = "htmx")]
#[tokio::test]
async fn htmx_handler_returns_javascript_with_correct_headers() {
let app = axum::Router::new().route(
crate::htmx::HTMX_JS_PATH,
axum::routing::get(crate::router::htmx_handler),
);
let response = app
.oneshot(
Request::builder()
.uri(crate::htmx::HTMX_JS_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let content_type = response
.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap();
assert!(
content_type.contains("application/javascript"),
"Expected application/javascript, got {content_type}"
);
let cache_control = response
.headers()
.get("cache-control")
.unwrap()
.to_str()
.unwrap();
assert!(
cache_control.contains("immutable"),
"Expected immutable cache, got {cache_control}"
);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(body.len(), crate::htmx::HTMX_JS.len());
let start = std::str::from_utf8(&body[..50]).expect("htmx should be valid UTF-8");
assert!(
start.contains("htmx") || start.contains("function"),
"Response doesn't look like htmx JavaScript: {start}"
);
}
#[cfg(feature = "htmx")]
#[tokio::test]
async fn htmx_csrf_handler_returns_csp_compatible_javascript() {
let app = axum::Router::new().route(
crate::htmx::HTMX_CSRF_JS_PATH,
axum::routing::get(crate::router::htmx_csrf_handler),
);
let response = app
.oneshot(
Request::builder()
.uri(crate::htmx::HTMX_CSRF_JS_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok()),
Some("application/javascript")
);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let js = std::str::from_utf8(&body).expect("csrf helper should be valid utf-8");
assert!(js.contains("htmx:configRequest"));
assert!(js.contains("X-CSRF-Token"));
assert!(!js.contains("<script"));
}
#[cfg(feature = "htmx")]
#[tokio::test]
async fn build_router_serves_htmx_js() {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri(crate::htmx::HTMX_JS_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let ct = response
.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap();
assert!(ct.contains("javascript"));
}
#[cfg(feature = "htmx")]
#[tokio::test]
async fn build_router_serves_htmx_csrf_js() {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri(crate::htmx::HTMX_CSRF_JS_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let csp = response
.headers()
.get("content-security-policy")
.expect("framework JS should still receive security headers")
.to_str()
.unwrap();
assert!(csp.contains("script-src 'self'"), "csp = {csp}");
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let js = std::str::from_utf8(&body).expect("csrf helper should be valid utf-8");
assert!(js.contains("htmx:configRequest"));
assert!(js.contains("X-CSRF-Token"));
}
#[tokio::test]
async fn build_router_serves_default_favicon_without_404() {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri(crate::router::DEFAULT_FAVICON_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NO_CONTENT);
assert!(
response.headers().contains_key("content-security-policy"),
"framework fallback responses should still receive security headers"
);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert!(body.is_empty());
}
#[tokio::test]
async fn build_router_does_not_override_user_favicon_route() {
let router = test_router(vec![test_get_route(
crate::router::DEFAULT_FAVICON_PATH,
"favicon",
)]);
let response = router
.oneshot(
Request::builder()
.uri(crate::router::DEFAULT_FAVICON_PATH)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"ok");
}
#[tokio::test]
async fn build_router_serves_static_files_for_unmatched_paths() {
use std::collections::HashMap;
let tmp = tempfile::tempdir().expect("tempdir");
let dist = tmp.path().join("dist");
std::fs::create_dir_all(dist.join("docs")).expect("mkdir");
std::fs::write(dist.join("docs/index.html"), "<h1>Static Docs</h1>").expect("write");
let manifest = crate::static_gen::StaticManifest {
generated_at: "2026-03-27T00:00:00Z".to_owned(),
autumn_version: "0.2.0".to_owned(),
routes: HashMap::from([(
"/docs".to_owned(),
crate::static_gen::ManifestEntry {
file: "docs/index.html".to_owned(),
revalidate: None,
},
)]),
};
let json = serde_json::to_string(&manifest).expect("serialize");
std::fs::write(dist.join("manifest.json"), json).expect("write manifest");
let config = AutumnConfig::default();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
let router = crate::router::build_router_with_static(
vec![test_get_route("/other", "other_page")],
&config,
state,
Some(dist.as_path()),
);
let response = router
.oneshot(
Request::builder()
.uri("/docs/")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let csp = response
.headers()
.get("content-security-policy")
.expect("static-first HTML should still receive security headers")
.to_str()
.unwrap();
assert!(csp.contains("script-src 'self'"), "csp = {csp}");
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(std::str::from_utf8(&body).unwrap(), "<h1>Static Docs</h1>");
}
#[tokio::test]
async fn build_mode_static_rendering_bypasses_startup_barrier() {
temp_env::async_with_vars([("AUTUMN_BUILD_STATIC", Some("1"))], async {
let config = AutumnConfig::default();
let state = AppState::for_test().with_startup_complete(false);
let router = crate::router::build_router(
vec![Route {
method: http::Method::GET,
path: "/about",
handler: axum::routing::get(|| async { "About Page Content" }),
name: "about",
api_doc: crate::openapi::ApiDoc {
method: "GET",
path: "/about",
operation_id: "about",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}],
&config,
state,
);
let tmp = tempfile::tempdir().unwrap();
let dist = tmp.path().join("dist");
let result = crate::static_gen::render_static_routes(
router,
&[crate::static_gen::StaticRouteMeta {
path: "/about",
name: "about",
revalidate: None,
params_fn: None,
}],
&dist,
)
.await;
assert!(result.is_ok(), "build failed: {:?}", result.err());
let html = std::fs::read_to_string(dist.join("about/index.html")).unwrap();
assert_eq!(html, "About Page Content");
})
.await;
}
#[tokio::test]
async fn build_router_injects_live_reload_script_when_enabled() {
let reload_file = tempfile::NamedTempFile::new().expect("reload state file");
std::fs::write(reload_file.path(), r#"{"version":0,"kind":"full"}"#).expect("write");
temp_env::async_with_vars(
[
("AUTUMN_DEV_RELOAD", Some("1")),
(
"AUTUMN_DEV_RELOAD_STATE",
Some(reload_file.path().to_str().expect("utf-8 path")),
),
],
async {
let router = test_router(vec![Route {
method: http::Method::GET,
path: "/page",
handler: axum::routing::get(|| async {
axum::response::Html("<html><body><main>ok</main></body></html>")
}),
name: "page",
api_doc: crate::openapi::ApiDoc {
method: "GET",
path: "/page",
operation_id: "page",
success_status: 200,
..Default::default()
},
repository: None,
idempotency: crate::route::RouteIdempotency::Direct,
timeout: crate::route::RouteTimeout::Inherit,
api_version: None,
sunset_opt_out: false,
}]);
let response = router
.oneshot(Request::builder().uri("/page").body(Body::empty()).unwrap())
.await
.unwrap();
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let html = std::str::from_utf8(&body).expect("utf-8");
assert!(html.contains("/__autumn/live-reload"));
},
)
.await;
}
#[tokio::test]
async fn build_router_mounts_dev_reload_script_endpoint_when_enabled() {
let reload_file = tempfile::NamedTempFile::new().expect("reload state file");
std::fs::write(reload_file.path(), r#"{"version":0,"kind":"full"}"#).expect("write");
temp_env::async_with_vars(
[
("AUTUMN_DEV_RELOAD", Some("1")),
(
"AUTUMN_DEV_RELOAD_STATE",
Some(reload_file.path().to_str().expect("utf-8 path")),
),
],
async {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri("/__autumn/live-reload.js")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("content-type")
.and_then(|v| v.to_str().ok()),
Some("application/javascript; charset=utf-8")
);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let js = std::str::from_utf8(&body).expect("utf-8");
assert!(js.contains("fetch("), "js body: {js}");
},
)
.await;
}
#[tokio::test]
async fn build_router_mounts_dev_reload_endpoint_when_enabled() {
let reload_file = tempfile::NamedTempFile::new().expect("reload state file");
std::fs::write(reload_file.path(), r#"{"version":7,"kind":"css"}"#).expect("write");
temp_env::async_with_vars(
[
("AUTUMN_DEV_RELOAD", Some("1")),
(
"AUTUMN_DEV_RELOAD_STATE",
Some(reload_file.path().to_str().expect("utf-8 path")),
),
],
async {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri("/__autumn/live-reload")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get("cache-control").unwrap(),
"no-store, no-cache, must-revalidate"
);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], br#"{"version":7,"kind":"css"}"#);
},
)
.await;
}
#[tokio::test]
async fn build_router_disables_cache_for_static_assets_in_dev_reload_mode() {
let project = tempfile::tempdir().expect("project dir");
let static_dir = project.path().join("static");
std::fs::create_dir_all(&static_dir).expect("mkdir");
std::fs::write(static_dir.join("demo.txt"), "hello").expect("write static file");
let reload_file = tempfile::NamedTempFile::new().expect("reload state file");
std::fs::write(reload_file.path(), r#"{"version":0,"kind":"full"}"#).expect("write");
temp_env::async_with_vars(
[
(
"AUTUMN_MANIFEST_DIR",
Some(project.path().to_str().expect("utf-8 path")),
),
("AUTUMN_DEV_RELOAD", Some("1")),
(
"AUTUMN_DEV_RELOAD_STATE",
Some(reload_file.path().to_str().expect("utf-8 path")),
),
],
async {
let router = test_router(vec![test_get_route("/dummy", "dummy")]);
let response = router
.oneshot(
Request::builder()
.uri("/static/demo.txt")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get("cache-control").unwrap(),
"no-store, no-cache, must-revalidate"
);
},
)
.await;
}
#[test]
fn app_builder_accepts_static_routes() {
use crate::static_gen::StaticRouteMeta;
let metas = vec![StaticRouteMeta {
path: "/about",
name: "about",
revalidate: None,
params_fn: None,
}];
let builder = app().static_routes(metas);
assert_eq!(builder.static_metas.len(), 1);
}
#[test]
fn project_dir_defaults_to_subdir() {
let env = crate::config::MockEnv::new();
let dir = super::project_dir("dist", &env);
assert_eq!(dir, std::path::PathBuf::from("dist"));
}
pub fn test_router_with_config(routes: Vec<Route>, config: &AutumnConfig) -> axum::Router {
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
crate::router::build_router(routes, config, state)
}
#[tokio::test]
async fn cors_wildcard_allows_any_origin() {
let mut config = AutumnConfig::default();
config.cors.allowed_origins = vec!["*".to_owned()];
let router = test_router_with_config(vec![test_get_route("/test", "test")], &config);
let response = router
.oneshot(
Request::builder()
.uri("/test")
.header("Origin", "https://example.com")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("access-control-allow-origin")
.unwrap(),
"*"
);
}
#[tokio::test]
async fn cors_specific_origin_reflected() {
let mut config = AutumnConfig::default();
config.cors.allowed_origins = vec!["https://example.com".to_owned()];
let router = test_router_with_config(vec![test_get_route("/test", "test")], &config);
let response = router
.oneshot(
Request::builder()
.uri("/test")
.header("Origin", "https://example.com")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("access-control-allow-origin")
.unwrap(),
"https://example.com"
);
}
#[tokio::test]
async fn cors_disabled_when_no_origins() {
let config = AutumnConfig::default();
assert!(config.cors.allowed_origins.is_empty());
let router = test_router_with_config(vec![test_get_route("/test", "test")], &config);
let response = router
.oneshot(
Request::builder()
.uri("/test")
.header("Origin", "https://example.com")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert!(
response
.headers()
.get("access-control-allow-origin")
.is_none()
);
}
#[tokio::test]
async fn cors_preflight_returns_204() {
let mut config = AutumnConfig::default();
config.cors.allowed_origins = vec!["https://example.com".to_owned()];
let router = test_router_with_config(vec![test_get_route("/test", "test")], &config);
let response = router
.oneshot(
Request::builder()
.method("OPTIONS")
.uri("/test")
.header("Origin", "https://example.com")
.header("Access-Control-Request-Method", "GET")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert!(
response
.headers()
.contains_key("access-control-allow-methods")
);
}
#[tokio::test]
async fn build_router_with_static_skips_without_manifest() {
let tmp = tempfile::tempdir().expect("tempdir");
let dist = tmp.path().join("dist");
std::fs::create_dir_all(&dist).expect("mkdir");
let config = AutumnConfig::default();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
let router = crate::router::build_router_with_static(
vec![test_get_route("/test", "test")],
&config,
state,
Some(dist.as_path()),
);
let response = router
.oneshot(Request::builder().uri("/test").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn build_router_with_static_none_dist() {
let config = AutumnConfig::default();
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
#[cfg(feature = "ws")]
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
#[cfg(feature = "ws")]
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
};
let router = crate::router::build_router_with_static(
vec![test_get_route("/test", "test")],
&config,
state,
None,
);
let response = router
.oneshot(Request::builder().uri("/test").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[test]
fn format_route_lines_lists_user_routes() {
let routes = vec![
test_get_route("/", "index"),
test_get_route("/users/{id}", "get_user"),
];
let config = AutumnConfig::default();
let output = format_route_lines(&routes, &[], &config);
assert!(output.contains("-> index"));
assert!(output.contains("/ GET"));
assert!(output.contains("/users/{id}"));
assert!(output.contains("-> get_user"));
}
#[test]
fn config_runtime_drift_format_route_lines_uses_actuator_prefix() {
let mut config = AutumnConfig::default();
config.actuator.prefix = "/ops".to_owned();
let output = format_route_lines(&[], &[], &config);
assert!(output.contains("-> health"));
assert!(output.contains("/ops/*"));
}
#[test]
fn format_task_lines_none_when_empty() {
assert!(format_task_lines(&[]).is_none());
}
#[test]
fn format_task_lines_fixed_delay() {
let tasks = vec![crate::task::TaskInfo {
name: "cleanup".into(),
schedule: crate::task::Schedule::FixedDelay(std::time::Duration::from_secs(300)),
coordination: crate::task::TaskCoordination::Fleet,
handler: |_| Box::pin(async { Ok(()) }),
}];
let output = format_task_lines(&tasks).unwrap();
assert!(output.contains("cleanup (every 300s)"));
}
#[test]
fn format_task_lines_cron() {
let tasks = vec![crate::task::TaskInfo {
name: "nightly".into(),
schedule: crate::task::Schedule::Cron {
expression: "0 0 * * *".into(),
timezone: None,
},
coordination: crate::task::TaskCoordination::Fleet,
handler: |_| Box::pin(async { Ok(()) }),
}];
let output = format_task_lines(&tasks).unwrap();
assert!(output.contains("nightly (cron 0 0 * * *)"));
}
#[test]
fn format_middleware_list_default() {
let config = AutumnConfig::default();
let output = format_middleware_list(&config);
assert!(output.contains("RequestId"));
assert!(output.contains("SecurityHeaders"));
assert!(output.contains("Session (in-memory)"));
assert!(output.contains("Metrics"));
assert!(!output.contains("CORS"));
assert!(!output.contains("CSRF"));
}
#[test]
fn format_middleware_list_with_cors_and_csrf() {
let config = AutumnConfig {
cors: crate::config::CorsConfig {
allowed_origins: vec!["https://example.com".into()],
..crate::config::CorsConfig::default()
},
security: crate::security::config::SecurityConfig {
csrf: crate::security::config::CsrfConfig {
enabled: true,
..crate::security::config::CsrfConfig::default()
},
..crate::security::config::SecurityConfig::default()
},
..AutumnConfig::default()
};
let output = format_middleware_list(&config);
assert!(output.contains("CORS"));
assert!(output.contains("CSRF"));
}
#[test]
fn mask_database_url_with_password() {
let masked = mask_database_url("postgres://user:secret@localhost:5432/mydb", 10);
assert!(masked.contains("****"));
assert!(!masked.contains("secret"));
assert!(masked.contains("postgres://user:****@localhost:5432/mydb"));
assert!(masked.contains("pool_size=10"));
}
#[test]
fn mask_database_url_without_password() {
let masked = mask_database_url("postgres://localhost/mydb", 5);
assert!(!masked.contains("****"));
assert!(masked.contains("postgres://localhost/mydb"));
assert!(masked.contains("pool_size=5"));
}
#[test]
fn mask_database_url_edge_cases() {
let masked2 = mask_database_url("postgres://user:p%40ssw%3Ard%21@localhost:5432/mydb", 10);
assert!(masked2.contains("****"));
assert!(!masked2.contains("p%40ssw%3Ard%21"));
assert!(masked2.contains("postgres://user:****@localhost:5432/mydb"));
let masked3 = mask_database_url("postgres://:secret@localhost:5432/mydb", 10);
assert!(masked3.contains("****"));
assert!(!masked3.contains("secret"));
assert!(masked3.contains("postgres://:****@localhost:5432/mydb"));
}
#[test]
fn mask_database_url_invalid_url_fallback() {
let masked = mask_database_url("this is completely invalid as a URL with supersecret", 10);
assert!(masked.contains("****"));
assert!(!masked.contains("supersecret"));
assert!(masked.contains("pool_size=10"));
}
#[test]
fn format_config_summary_defaults() {
let config = AutumnConfig::default();
let output = format_config_summary(&config);
assert!(output.contains("profile: none"));
assert!(output.contains("server: 127.0.0.1:3000"));
assert!(output.contains("database: not configured"));
assert!(output.contains("log_level:"));
assert!(output.contains("telemetry: disabled"));
assert!(output.contains("health: /health"));
}
#[test]
fn format_config_summary_with_db() {
let config = AutumnConfig {
database: crate::config::DatabaseConfig {
url: Some("postgres://user:pass@host/db".into()),
pool_size: 20,
..crate::config::DatabaseConfig::default()
},
..AutumnConfig::default()
};
let output = format_config_summary(&config);
assert!(output.contains("user:****@host/db"));
assert!(output.contains("pool_size=20"));
assert!(!output.contains("pass"));
}
#[test]
fn format_config_summary_with_profile() {
let config = AutumnConfig {
profile: Some("prod".into()),
..AutumnConfig::default()
};
let output = format_config_summary(&config);
assert!(output.contains("profile: prod"));
}
#[test]
fn format_config_summary_with_telemetry() {
let config = AutumnConfig {
telemetry: crate::config::TelemetryConfig {
enabled: true,
service_name: "orders-api".into(),
otlp_endpoint: Some("http://otel-collector:4317".into()),
..crate::config::TelemetryConfig::default()
},
..AutumnConfig::default()
};
let output = format_config_summary(&config);
assert!(output.contains("telemetry: Grpc -> http://otel-collector:4317"));
}
#[test]
fn log_startup_transparency_runs_without_panic() {
let routes = vec![test_get_route("/", "index")];
let tasks = vec![crate::task::TaskInfo {
name: "cleanup".into(),
schedule: crate::task::Schedule::FixedDelay(std::time::Duration::from_secs(60)),
coordination: crate::task::TaskCoordination::Fleet,
handler: |_| Box::pin(async { Ok(()) }),
}];
let config = AutumnConfig::default();
log_startup_transparency(&routes, &tasks, &[], &config);
}
#[test]
fn log_startup_transparency_no_tasks() {
let routes = vec![test_get_route("/health", "check")];
let config = AutumnConfig::default();
log_startup_transparency(&routes, &[], &[], &config);
}
#[cfg(feature = "ws")]
#[tokio::test]
async fn start_task_scheduler_broadcasts_events() {
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
};
let mut rx = state.channels().subscribe("sys:tasks");
let task = crate::task::TaskInfo {
name: "test_broadcaster".into(),
schedule: crate::task::Schedule::FixedDelay(std::time::Duration::from_millis(1)),
coordination: crate::task::TaskCoordination::Fleet,
handler: |_| Box::pin(async { Ok(()) }),
};
let state_clone = state.clone();
tokio::spawn(async move {
super::start_task_scheduler(
vec![task],
&state_clone,
&tokio_util::sync::CancellationToken::new(),
);
});
let msg1 = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
.await
.expect("timeout waiting for start event")
.expect("channel closed");
let json1: serde_json::Value = serde_json::from_str(msg1.as_str()).unwrap();
assert_eq!(json1["event"], "started");
assert_eq!(json1["task"], "test_broadcaster");
let msg2 = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
.await
.expect("timeout waiting for success event")
.expect("channel closed");
let json2: serde_json::Value = serde_json::from_str(msg2.as_str()).unwrap();
assert_eq!(json2["event"], "success");
assert_eq!(json2["task"], "test_broadcaster");
assert!(json2.get("duration_ms").is_some());
}
#[cfg(feature = "ws")]
#[tokio::test]
async fn start_task_scheduler_broadcasts_failure_events() {
let state = AppState {
extensions: std::sync::Arc::new(std::sync::RwLock::new(
std::collections::HashMap::new(),
)),
#[cfg(feature = "db")]
pool: None,
#[cfg(feature = "db")]
replica_pool: None,
#[cfg(feature = "db")]
shards: None,
profile: None,
role: crate::config::ProcessRole::Combined,
started_at: std::time::Instant::now(),
health_detailed: true,
probes: crate::probe::ProbeState::ready_for_test(),
metrics: crate::middleware::MetricsCollector::new(),
log_levels: crate::actuator::LogLevels::new("info"),
task_registry: crate::actuator::TaskRegistry::new(),
job_registry: crate::actuator::JobRegistry::new(),
config_props: crate::actuator::ConfigProperties::default(),
channels: crate::channels::Channels::new(32),
#[cfg(feature = "presence")]
presence: crate::presence::Presence::new(crate::channels::Channels::new(32)),
shutdown: tokio_util::sync::CancellationToken::new(),
policy_registry: crate::authorization::PolicyRegistry::default(),
forbidden_response: crate::authorization::ForbiddenResponse::default(),
auth_session_key: "user_id".to_owned(),
shared_cache: None,
clock: std::sync::Arc::new(crate::time::SystemClock),
app_id: AppState::next_app_id(),
metrics_source_registry: crate::actuator::MetricsSourceRegistry::new(),
health_indicator_registry: crate::actuator::HealthIndicatorRegistry::new(),
};
let mut rx = state.channels().subscribe("sys:tasks");
let task = crate::task::TaskInfo {
name: "test_failing_task".into(),
schedule: crate::task::Schedule::FixedDelay(std::time::Duration::from_millis(1)),
coordination: crate::task::TaskCoordination::Fleet,
handler: |_| {
Box::pin(async { Err(crate::AutumnError::bad_request_msg("forced error")) })
},
};
let state_clone = state.clone();
tokio::spawn(async move {
super::start_task_scheduler(
vec![task],
&state_clone,
&tokio_util::sync::CancellationToken::new(),
);
});
let _ = rx.recv().await.unwrap();
let msg2 = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
.await
.expect("timeout waiting for failure event")
.expect("channel closed");
let json2: serde_json::Value = serde_json::from_str(msg2.as_str()).unwrap();
assert_eq!(json2["event"], "failure");
assert_eq!(json2["task"], "test_failing_task");
assert_eq!(json2["error"], "forced error");
}
#[tokio::test]
async fn execute_task_result_ok_returns_duration() {
let state = AppState::for_test();
let handler: crate::task::TaskHandler = |_| Box::pin(async { Ok(()) });
let start = std::time::Instant::now();
let result =
super::execute_task_result(&state, handler, start, "test_task", "fixed_delay").await;
assert!(result.is_ok(), "expected Ok from successful handler");
assert!(result.unwrap() < u64::MAX);
}
#[tokio::test]
async fn execute_task_result_err_returns_duration_and_message() {
let state = AppState::for_test();
let handler: crate::task::TaskHandler =
|_| Box::pin(async { Err(crate::AutumnError::bad_request_msg("test error")) });
let start = std::time::Instant::now();
let result =
super::execute_task_result(&state, handler, start, "test_task", "fixed_delay").await;
assert!(result.is_err(), "expected Err from failing handler");
let (duration_ms, msg) = result.unwrap_err();
assert!(duration_ms < u64::MAX);
assert!(msg.contains("test error"));
}
fn instantly_panicking_scheduled_handler(
_state: AppState,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = crate::AutumnResult<()>> + Send>> {
panic!("panic before scheduled future")
}
#[tokio::test]
async fn execute_task_result_reports_immediate_handler_panics() {
let state = AppState::for_test();
let start = std::time::Instant::now();
let result = super::execute_task_result(
&state,
instantly_panicking_scheduled_handler,
start,
"test_task",
"fixed_delay",
)
.await;
let (duration_ms, msg) = result.expect_err("expected Err from panicking handler");
assert!(duration_ms < u64::MAX);
assert!(msg.contains("scheduled task handler panicked: panic before scheduled future"));
}
#[tokio::test]
async fn execute_fixed_delay_task_does_not_timeout_in_process_runs() {
let state = AppState::for_test();
state.task_registry.register_scheduled(
"slow_task",
"every 1s",
crate::task::TaskCoordination::Fleet,
"in_process",
"replica-a",
);
let handler: crate::task::TaskHandler = |_| {
Box::pin(async {
tokio::time::sleep(std::time::Duration::from_millis(30)).await;
Ok(())
})
};
let coordinator = std::sync::Arc::new(
crate::scheduler::InProcessSchedulerCoordinator::new("replica-a"),
);
super::execute_fixed_delay_task(
"slow_task".to_owned(),
state.clone(),
handler,
std::time::Duration::from_secs(1),
crate::task::TaskCoordination::Fleet,
coordinator,
std::time::Duration::from_millis(10),
)
.await;
let snapshot = state.task_registry.snapshot();
let status = &snapshot["slow_task"];
assert_eq!(status.status, "idle");
assert_eq!(status.last_result.as_deref(), Some("ok"));
assert_eq!(status.total_runs, 1);
assert_eq!(status.total_failures, 0);
assert!(status.last_error.is_none());
}
static SKIPPED_LEASE_HANDLER_CALLS: AtomicUsize = AtomicUsize::new(0);
struct DenyingSchedulerCoordinator;
impl crate::scheduler::SchedulerCoordinator for DenyingSchedulerCoordinator {
fn backend(&self) -> &'static str {
"postgres"
}
fn replica_id(&self) -> &'static str {
"replica-a"
}
fn try_acquire<'a>(
&'a self,
_task_name: &'a str,
_tick_key: &'a str,
_coordination: crate::task::TaskCoordination,
) -> crate::scheduler::SchedulerFuture<
'a,
crate::AutumnResult<Option<crate::scheduler::SchedulerLease>>,
> {
Box::pin(async { Ok(None) })
}
}
struct GrantingSchedulerCoordinator {
backend: &'static str,
tick_keys: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
release_count: Option<std::sync::Arc<AtomicUsize>>,
}
impl crate::scheduler::SchedulerCoordinator for GrantingSchedulerCoordinator {
fn backend(&self) -> &'static str {
self.backend
}
fn replica_id(&self) -> &'static str {
"replica-a"
}
fn try_acquire<'a>(
&'a self,
_task_name: &'a str,
tick_key: &'a str,
_coordination: crate::task::TaskCoordination,
) -> crate::scheduler::SchedulerFuture<
'a,
crate::AutumnResult<Option<crate::scheduler::SchedulerLease>>,
> {
Box::pin(async move {
self.tick_keys.lock().unwrap().push(tick_key.to_owned());
let lease = self.release_count.as_ref().map_or_else(
|| crate::scheduler::SchedulerLease::local(self.backend, "replica-a"),
|release_count| {
crate::scheduler::SchedulerLease::tracked(
self.backend,
"replica-a",
std::sync::Arc::clone(release_count),
)
},
);
Ok(Some(lease))
})
}
}
fn counted_scheduled_handler(
_state: AppState,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = crate::AutumnResult<()>> + Send>> {
Box::pin(async {
SKIPPED_LEASE_HANDLER_CALLS.fetch_add(1, Ordering::SeqCst);
Ok(())
})
}
#[tokio::test]
async fn execute_fixed_delay_task_skips_handler_when_lease_is_not_acquired() {
SKIPPED_LEASE_HANDLER_CALLS.store(0, Ordering::SeqCst);
let state = AppState::for_test();
state.task_registry.register_scheduled(
"claimed_elsewhere",
"every 1s",
crate::task::TaskCoordination::Fleet,
"postgres",
"replica-a",
);
let coordinator = std::sync::Arc::new(DenyingSchedulerCoordinator);
super::execute_fixed_delay_task(
"claimed_elsewhere".to_owned(),
state.clone(),
counted_scheduled_handler,
std::time::Duration::from_secs(1),
crate::task::TaskCoordination::Fleet,
coordinator,
std::time::Duration::from_secs(1),
)
.await;
let snapshot = state.task_registry.snapshot();
let status = &snapshot["claimed_elsewhere"];
assert_eq!(SKIPPED_LEASE_HANDLER_CALLS.load(Ordering::SeqCst), 0);
assert_eq!(status.total_runs, 0);
assert!(status.current_leader.is_none());
assert!(status.last_tick.is_none());
}
#[tokio::test]
async fn execute_fixed_delay_task_records_distributed_lease_ttl_timeout() {
let state = AppState::for_test();
state.task_registry.register_scheduled(
"slow_distributed_task",
"every 1s",
crate::task::TaskCoordination::Fleet,
"postgres",
"replica-a",
);
let handler: crate::task::TaskHandler = |_| {
Box::pin(async {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
Ok(())
})
};
let coordinator = std::sync::Arc::new(GrantingSchedulerCoordinator {
backend: "postgres",
tick_keys: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
release_count: None,
});
super::execute_fixed_delay_task(
"slow_distributed_task".to_owned(),
state.clone(),
handler,
std::time::Duration::from_secs(1),
crate::task::TaskCoordination::Fleet,
coordinator,
std::time::Duration::from_millis(10),
)
.await;
let snapshot = state.task_registry.snapshot();
let status = &snapshot["slow_distributed_task"];
assert_eq!(status.status, "idle");
assert_eq!(status.last_result.as_deref(), Some("failed"));
assert_eq!(status.total_runs, 1);
assert_eq!(status.total_failures, 1);
assert!(
status
.last_error
.as_deref()
.is_some_and(|error| error.contains("lease TTL"))
);
}
#[tokio::test]
async fn execute_cron_task_uses_scheduled_occurrence_for_tick_key() {
let state = AppState::for_test();
state.task_registry.register_scheduled(
"cron_review_task",
"cron */10 * * * * *",
crate::task::TaskCoordination::Fleet,
"postgres",
"replica-a",
);
let tick_keys = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let coordinator = std::sync::Arc::new(GrantingSchedulerCoordinator {
backend: "postgres",
tick_keys: std::sync::Arc::clone(&tick_keys),
release_count: None,
});
let handler: crate::task::TaskHandler = |_| Box::pin(async { Ok(()) });
let scheduled_unix_secs = 1_700_000_000;
super::execute_cron_task(
"cron_review_task".to_owned(),
state.clone(),
handler,
crate::task::TaskCoordination::Fleet,
coordinator,
std::time::Duration::from_secs(30),
scheduled_unix_secs,
)
.await;
assert_eq!(
tick_keys.lock().unwrap().as_slice(),
["cron_review_task:1700000000"]
);
}
#[tokio::test]
async fn execute_fixed_delay_task_releases_lease_when_handler_panics() {
let state = AppState::for_test();
state.task_registry.register_scheduled(
"panic_task",
"every 1s",
crate::task::TaskCoordination::Fleet,
"postgres",
"replica-a",
);
let release_count = std::sync::Arc::new(AtomicUsize::new(0));
let coordinator = std::sync::Arc::new(GrantingSchedulerCoordinator {
backend: "postgres",
tick_keys: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
release_count: Some(std::sync::Arc::clone(&release_count)),
});
let handler: crate::task::TaskHandler = |_| {
Box::pin(async {
panic!("forced scheduled panic");
#[allow(unreachable_code)]
Ok(())
})
};
super::execute_fixed_delay_task(
"panic_task".to_owned(),
state.clone(),
handler,
std::time::Duration::from_secs(1),
crate::task::TaskCoordination::Fleet,
coordinator,
std::time::Duration::from_secs(30),
)
.await;
let snapshot = state.task_registry.snapshot();
let status = &snapshot["panic_task"];
assert_eq!(release_count.load(Ordering::SeqCst), 1);
assert_eq!(status.status, "idle");
assert_eq!(status.last_result.as_deref(), Some("failed"));
assert_eq!(status.total_runs, 1);
assert_eq!(status.total_failures, 1);
assert!(
status
.last_error
.as_deref()
.is_some_and(|error| error.contains("scheduled task handler panicked"))
);
}
#[test]
fn next_cron_occurrence_skips_overdue_slots() {
use chrono::TimeZone as _;
let cron = "0 * * * * *"
.parse::<croner::Cron>()
.expect("cron expression should parse");
let stale_cursor = chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 0, 0)
.unwrap();
let now = chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 30, 5)
.unwrap();
let next = super::next_cron_occurrence_after(&cron, &stale_cursor, &now)
.expect("next cron occurrence should resolve");
assert_eq!(
next,
chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 31, 0)
.unwrap()
);
}
#[test]
fn cron_occurrence_is_overdue_after_later_slot_passed() {
use chrono::TimeZone as _;
let cron = "0 * * * * *"
.parse::<croner::Cron>()
.expect("cron expression should parse");
let scheduled_at = chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 1, 0)
.unwrap();
let slightly_late = chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 1, 5)
.unwrap();
let after_later_slot = chrono_tz::UTC
.with_ymd_and_hms(2026, 5, 5, 12, 30, 5)
.unwrap();
assert!(
!super::cron_occurrence_is_overdue(&cron, &scheduled_at, &slightly_late)
.expect("overdue check should resolve")
);
assert!(
super::cron_occurrence_is_overdue(&cron, &scheduled_at, &after_later_slot)
.expect("overdue check should resolve")
);
}
#[cfg(feature = "storage")]
mod storage_preflight {
use super::super::{StorageBootstrap, preflight_storage};
use crate::AppState;
use crate::config::AutumnConfig;
use crate::storage::{BlobStoreState, StorageBackend, StorageConfig, StorageLocalConfig};
fn config_with_storage(storage: StorageConfig) -> AutumnConfig {
AutumnConfig {
profile: Some("dev".into()),
storage,
..AutumnConfig::default()
}
}
#[test]
fn preflight_returns_none_when_disabled() {
let cfg = config_with_storage(StorageConfig {
backend: StorageBackend::Disabled,
..StorageConfig::default()
});
assert!(preflight_storage(&cfg).is_none());
}
#[test]
fn preflight_provisions_local_backend_against_tempdir() {
let dir = tempfile::tempdir().unwrap();
let cfg = config_with_storage(StorageConfig {
backend: StorageBackend::Local,
local: StorageLocalConfig {
root: dir.path().to_path_buf(),
..StorageLocalConfig::default()
},
..StorageConfig::default()
});
let bootstrap = preflight_storage(&cfg).expect("local backend should provision");
assert_eq!(bootstrap.store.provider_id(), "default");
assert!(bootstrap.serving.is_some(), "local backend mounts a route");
}
#[tokio::test]
async fn install_registers_blob_store_on_state() {
let dir = tempfile::tempdir().unwrap();
let cfg = config_with_storage(StorageConfig {
backend: StorageBackend::Local,
local: StorageLocalConfig {
root: dir.path().to_path_buf(),
..StorageLocalConfig::default()
},
..StorageConfig::default()
});
let bootstrap: StorageBootstrap = preflight_storage(&cfg).unwrap();
let state = AppState::for_test();
assert!(state.extension::<BlobStoreState>().is_none());
let serving = bootstrap.install(&state);
assert!(serving.is_some());
assert!(state.extension::<BlobStoreState>().is_some());
}
#[test]
fn with_blob_store_stores_custom_store() {
use crate::storage::{
Blob, BlobFuture, BlobMeta, BlobStore, BlobStoreError, ByteStream,
};
use bytes::Bytes;
use std::time::Duration;
struct FakeStore;
impl BlobStore for FakeStore {
fn provider_id(&self) -> &'static str {
"fake"
}
fn put<'a>(&'a self, _k: &'a str, _ct: &'a str, _b: Bytes) -> BlobFuture<'a, Blob> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn put_stream<'a>(
&'a self,
_k: &'a str,
_ct: &'a str,
_d: ByteStream<'a>,
) -> BlobFuture<'a, Blob> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn get<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, Bytes> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn delete<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, ()> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn head<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, Option<BlobMeta>> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn presigned_url<'a>(
&'a self,
_k: &'a str,
_e: Duration,
) -> BlobFuture<'a, String> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
}
let builder = crate::app().with_blob_store(FakeStore);
assert!(builder.blob_store.is_some());
}
#[tokio::test]
async fn with_blob_store_is_installed_on_state() {
use crate::storage::{
Blob, BlobFuture, BlobMeta, BlobStore, BlobStoreError, ByteStream,
};
use bytes::Bytes;
use std::time::Duration;
struct FakeStore;
impl BlobStore for FakeStore {
fn provider_id(&self) -> &'static str {
"fake-installed"
}
fn put<'a>(&'a self, _k: &'a str, _ct: &'a str, _b: Bytes) -> BlobFuture<'a, Blob> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn put_stream<'a>(
&'a self,
_k: &'a str,
_ct: &'a str,
_d: ByteStream<'a>,
) -> BlobFuture<'a, Blob> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn get<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, Bytes> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn delete<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, ()> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn head<'a>(&'a self, _k: &'a str) -> BlobFuture<'a, Option<BlobMeta>> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
fn presigned_url<'a>(
&'a self,
_k: &'a str,
_e: Duration,
) -> BlobFuture<'a, String> {
Box::pin(async { Err(BlobStoreError::Unsupported("fake".into())) })
}
}
let builder = crate::app().with_blob_store(FakeStore);
let bootstrap = builder.blob_store.map(|store| StorageBootstrap {
store,
serving: None,
});
let state = AppState::for_test();
assert!(state.extension::<BlobStoreState>().is_none());
if let Some(b) = bootstrap {
b.install(&state);
}
let installed = state
.extension::<BlobStoreState>()
.expect("store should be installed");
assert_eq!(installed.store().provider_id(), "fake-installed");
}
}
struct TestPlugin {
name: &'static str,
route: Route,
}
impl crate::plugin::Plugin for TestPlugin {
fn name(&self) -> std::borrow::Cow<'static, str> {
std::borrow::Cow::Borrowed(self.name)
}
fn build(self, app: AppBuilder) -> AppBuilder {
app.routes(vec![self.route])
}
}
#[test]
fn routes_registered_before_plugin_are_user_sourced() {
let user_route = test_get_route("/home", "home");
let builder = app().routes(vec![user_route]);
assert_eq!(builder.route_sources.len(), 1);
assert_eq!(
builder.route_sources[0],
crate::route_listing::RouteSource::User
);
}
#[test]
fn routes_registered_inside_plugin_are_plugin_sourced() {
let plugin_route = test_get_route("/plugin-page", "plugin_page");
let plugin = TestPlugin {
name: "my-plugin",
route: plugin_route,
};
let builder = app().plugin(plugin);
assert_eq!(builder.route_sources.len(), 1);
assert_eq!(
builder.route_sources[0],
crate::route_listing::RouteSource::Plugin("my-plugin".to_owned())
);
}
#[test]
fn routes_registered_after_plugin_revert_to_user_sourced() {
let plugin_route = test_get_route("/plugin-page", "plugin_page");
let user_route = test_get_route("/home", "home");
let plugin = TestPlugin {
name: "my-plugin",
route: plugin_route,
};
let builder = app().plugin(plugin).routes(vec![user_route]);
assert_eq!(builder.route_sources.len(), 2);
assert_eq!(
builder.route_sources[0],
crate::route_listing::RouteSource::Plugin("my-plugin".to_owned())
);
assert_eq!(
builder.route_sources[1],
crate::route_listing::RouteSource::User
);
}
struct OuterPlugin;
impl crate::plugin::Plugin for OuterPlugin {
fn name(&self) -> std::borrow::Cow<'static, str> {
"outer".into()
}
fn build(self, app: AppBuilder) -> AppBuilder {
let inner = TestPlugin {
name: "inner",
route: test_get_route("/inner", "inner"),
};
app.plugin(inner)
.routes(vec![test_get_route("/outer-after", "outer_after")])
}
}
#[test]
fn outer_plugin_source_restored_after_nested_plugin() {
let builder = app().plugin(OuterPlugin);
assert_eq!(builder.route_sources.len(), 2);
assert_eq!(
builder.route_sources[0],
crate::route_listing::RouteSource::Plugin("inner".to_owned()),
"first route should be attributed to inner plugin"
);
assert_eq!(
builder.route_sources[1],
crate::route_listing::RouteSource::Plugin("outer".to_owned()),
"second route should be re-attributed to outer plugin after nested build"
);
}
#[tokio::test]
async fn shutdown_hooks_with_timeout_runs_all_fast_hooks() {
use std::sync::atomic::{AtomicUsize, Ordering};
let counter = Arc::new(AtomicUsize::new(0));
let c1 = Arc::clone(&counter);
let c2 = Arc::clone(&counter);
let hooks: Vec<ShutdownHook> = vec![
Box::new(move || {
let c = Arc::clone(&c1);
Box::pin(async move {
c.fetch_add(1, Ordering::SeqCst);
})
}),
Box::new(move || {
let c = Arc::clone(&c2);
Box::pin(async move {
c.fetch_add(1, Ordering::SeqCst);
})
}),
];
run_shutdown_hooks_with_timeout(
&hooks,
std::time::Duration::from_secs(2),
std::time::Duration::from_secs(10),
)
.await;
assert_eq!(counter.load(Ordering::SeqCst), 2, "both hooks must run");
}
#[tokio::test]
async fn shutdown_hooks_with_timeout_tolerates_slow_hook_overrun() {
use std::sync::atomic::{AtomicBool, Ordering};
let fast_ran = Arc::new(AtomicBool::new(false));
let fr = Arc::clone(&fast_ran);
let hooks: Vec<ShutdownHook> = vec![
Box::new(move || {
let fr = Arc::clone(&fr);
Box::pin(async move {
fr.store(true, Ordering::SeqCst);
})
}),
Box::new(|| {
Box::pin(async move {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
})
}),
];
run_shutdown_hooks_with_timeout(
&hooks,
std::time::Duration::from_millis(50),
std::time::Duration::from_secs(1),
)
.await;
assert!(
fast_ran.load(Ordering::SeqCst),
"fast hook must still run even after slow hook overruns its per-hook budget"
);
}
#[cfg(feature = "http-client")]
#[test]
fn build_state_registers_shared_reqwest_client() {
let config = AutumnConfig::default();
let state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
None,
);
assert!(
state
.extension::<crate::http_client::SharedReqwestClient>()
.is_some(),
"build_state must register a SharedReqwestClient for connection-pool sharing"
);
}
#[cfg(feature = "maud")]
#[test]
fn with_story_gallery_installs_story_registry_extension() {
let builder = crate::app().with_story_gallery(crate::stories::StoryGallery::builtin());
let gallery = builder
.story_gallery
.expect("with_story_gallery must store the gallery on the builder");
let expected_count = gallery.stories().len();
assert!(expected_count > 0, "builtin gallery must not be empty");
let config = AutumnConfig::default();
let state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
None,
);
install_story_registry(&state, Some(gallery));
let registry = state
.extension::<crate::stories::StoryRegistry>()
.expect("install_story_registry must publish the StoryRegistry extension");
assert_eq!(
registry.stories().len(),
expected_count,
"every registered story must reach the state extension"
);
let bare_state = build_state(
&config,
#[cfg(feature = "db")]
None,
#[cfg(feature = "db")]
None,
#[cfg(feature = "ws")]
None,
);
install_story_registry(&bare_state, None);
assert!(
bare_state
.extension::<crate::stories::StoryRegistry>()
.is_none(),
"no gallery registered must mean no StoryRegistry extension"
);
}
}
#[cfg(all(test, unix))]
mod unix_socket_tests {
use super::prepare_unix_socket_path;
#[test]
fn prepare_unix_socket_path_noop_when_absent() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("missing.sock");
prepare_unix_socket_path(&path).expect("absent path is fine");
assert!(!path.exists());
}
#[test]
fn prepare_unix_socket_path_removes_stale_socket() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("stale.sock");
let listener = std::os::unix::net::UnixListener::bind(&path).expect("bind socket");
drop(listener);
assert!(path.exists(), "socket file should exist before prepare");
prepare_unix_socket_path(&path).expect("stale socket should be removed");
assert!(!path.exists(), "stale socket should be unlinked");
}
#[test]
fn prepare_unix_socket_path_refuses_live_socket() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("live.sock");
let _listener = std::os::unix::net::UnixListener::bind(&path).expect("bind socket");
let err = prepare_unix_socket_path(&path).expect_err("must refuse a live socket");
assert_eq!(err.kind(), std::io::ErrorKind::AddrInUse);
assert!(path.exists(), "live socket must not be removed");
}
#[test]
fn prepare_unix_socket_path_errors_on_regular_file() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("not-a-socket");
std::fs::write(&path, b"i am a regular file").expect("write file");
let err = prepare_unix_socket_path(&path).expect_err("must refuse a non-socket file");
assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists);
assert!(path.exists(), "regular file must not be removed");
}
}