diff --git a/Cargo.lock b/Cargo.lock index 0113bda27..fbb4e8fd3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -700,6 +700,7 @@ dependencies = [ "toml", "tower", "tracing", + "validator", "walkdir", ] diff --git a/crates/edgezero-adapter-axum/Cargo.toml b/crates/edgezero-adapter-axum/Cargo.toml index c1e20907b..231b7a81c 100644 --- a/crates/edgezero-adapter-axum/Cargo.toml +++ b/crates/edgezero-adapter-axum/Cargo.toml @@ -31,7 +31,7 @@ cli = [ "dep:toml", "dep:walkdir", ] -test-utils = [] +test-utils = ["dep:serde", "dep:validator"] [dependencies] edgezero-adapter = { path = "../edgezero-adapter", optional = true, features = [ @@ -53,7 +53,9 @@ hyper-util = { workspace = true, optional = true } log = { workspace = true } redb = { workspace = true, optional = true } reqwest = { workspace = true, optional = true } +serde = { workspace = true, optional = true } serde_json = { workspace = true } +validator = { workspace = true, optional = true } simple_logger = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, optional = true } @@ -62,6 +64,12 @@ tower = { workspace = true, optional = true } tracing = { workspace = true } walkdir = { workspace = true, optional = true } +[[bin]] +name = "edgezero-hosting-fixture" +path = "tests/fixtures/production_host.rs" +required-features = ["axum", "test-utils"] +test = false + [dev-dependencies] async-trait = { workspace = true } axum = { workspace = true, features = ["macros"] } diff --git a/crates/edgezero-adapter-axum/src/config_store.rs b/crates/edgezero-adapter-axum/src/config_store.rs index f58413005..c2c881238 100644 --- a/crates/edgezero-adapter-axum/src/config_store.rs +++ b/crates/edgezero-adapter-axum/src/config_store.rs @@ -15,7 +15,7 @@ use std::collections::HashMap; use std::env; use std::fs; -use std::io::ErrorKind; +use std::io::{ErrorKind, Read as _}; use std::path::{Path, PathBuf}; #[cfg(test)] use std::sync::atomic::{AtomicUsize, Ordering}; @@ -106,23 +106,54 @@ impl AxumConfigStore { /// exists but cannot be read or parsed. #[inline] pub fn from_path(path: &Path) -> Result { - let raw = match fs::read_to_string(path) { - Ok(raw) => raw, - Err(err) if err.kind() == ErrorKind::NotFound => { + Self::load_snapshot(path, false) + } + + pub(crate) fn from_required_path(path: &Path) -> Result { + Self::load_snapshot(path, true) + } + + #[expect( + clippy::verbose_file_reads, + reason = "read the same opened file whose regular-file metadata was verified" + )] + fn load_snapshot(path: &Path, required: bool) -> Result { + // Follow projected-mount symlinks. Reject non-regular targets before opening + // so a declared FIFO cannot stall startup. Roots are trusted deployment inputs. + let metadata = match fs::metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == ErrorKind::NotFound && !required => { return Ok(Self::empty()); } - Err(err) => { - return Err(ConfigStoreError::unavailable(format!( - "failed to read {}: {err}", - path.display() - ))); + Err(error) if error.kind() == ErrorKind::NotFound => { + return Err(ConfigStoreError::unavailable("config file missing")); } + Err(_error) => return Err(ConfigStoreError::unavailable("config file unreadable")), }; - let data: HashMap = serde_json::from_str(&raw).map_err(|err| { - ConfigStoreError::unavailable(format!( - "{} is not a flat string -> string JSON object: {err}", - path.display() - )) + if !metadata.is_file() { + return Err(ConfigStoreError::unavailable( + "config target is not a regular file", + )); + } + let mut file = fs::File::open(path) + .map_err(|_error| ConfigStoreError::unavailable("config file unreadable"))?; + if !file + .metadata() + .map_err(|_error| ConfigStoreError::unavailable("config file unreadable"))? + .is_file() + { + return Err(ConfigStoreError::unavailable( + "config target is not a regular file", + )); + } + let mut raw = String::new(); + file.read_to_string(&mut raw).map_err(|_error| { + ConfigStoreError::unavailable("config file unreadable or invalid UTF-8") + })?; + let data: HashMap = serde_json::from_str(&raw).map_err(|_error| { + ConfigStoreError::unavailable( + "config file malformed: expected a flat string -> string JSON object", + ) })?; Ok(Self { data, diff --git a/crates/edgezero-adapter-axum/src/connection.rs b/crates/edgezero-adapter-axum/src/connection.rs index 67037ecaf..10058cb2f 100644 --- a/crates/edgezero-adapter-axum/src/connection.rs +++ b/crates/edgezero-adapter-axum/src/connection.rs @@ -1,5 +1,7 @@ use std::error::Error as StdError; +use std::future; +use edgezero_core::probe::LifecyclePhase; use http_body::Body as HttpBody; use hyper::Request; use hyper::body::Incoming; @@ -7,6 +9,7 @@ use hyper::server::conn::http1::Builder as Http1Builder; use hyper::service::Service; use hyper_util::rt::TokioIo; use tokio::net::TcpStream; +use tokio::sync::watch; use tokio::time::{Instant as TokioInstant, sleep_until}; use crate::response::AxumBodyError; @@ -20,15 +23,31 @@ pub(crate) enum ConnectionExit { DeadlineExceeded, } -/// Owns one HTTP/1 connection until Hyper completes or an active response deadline closes it. +#[cfg(test)] +pub(crate) async fn serve_http1( + stream: TcpStream, + service: ServiceType, + responses: EgressConnection, +) -> Result +where + ServiceType: Service, Response = hyper::Response>, + ServiceType::Error: Into>, + ResponseBody: HttpBody + 'static, + ResponseBody::Error: Into>, +{ + serve_http1_with_shutdown(stream, service, responses, None).await +} + +/// Owns one HTTP/1 connection through graceful drain and existing response deadlines. #[expect( clippy::integer_division_remainder_used, reason = "tokio::select! expands to internal randomized branch selection arithmetic" )] -pub(crate) async fn serve_http1( +pub(crate) async fn serve_http1_with_shutdown( stream: TcpStream, service: ServiceType, responses: EgressConnection, + mut phase: Option>, ) -> Result where ServiceType: Service, Response = hyper::Response>, @@ -42,6 +61,23 @@ where let mut connection = Box::pin(connection_future); loop { + if phase.as_ref().is_some_and(|reader| { + matches!( + *reader.borrow(), + LifecyclePhase::Draining | LifecyclePhase::Stopped + ) + }) { + connection.as_mut().graceful_shutdown(); + phase = None; + } + let draining = async { + if let Some(reader) = phase.as_mut() { + let _changed = reader.changed().await; + } else { + future::pending::<()>().await; + } + }; + tokio::pin!(draining); let changed = responses.changed(); tokio::pin!(changed); if let Some(deadline) = responses.next_deadline() { @@ -58,6 +94,7 @@ where } } () = &mut changed => {} + () = &mut draining => {} } } else { tokio::select! { @@ -67,6 +104,7 @@ where return classify_connection_result(result); } () = &mut changed => {} + () = &mut draining => {} } } } diff --git a/crates/edgezero-adapter-axum/src/dev_server.rs b/crates/edgezero-adapter-axum/src/dev_server.rs index 81f48ada5..1fb0abc04 100644 --- a/crates/edgezero-adapter-axum/src/dev_server.rs +++ b/crates/edgezero-adapter-axum/src/dev_server.rs @@ -1,18 +1,37 @@ +#![expect( + clippy::pub_use, + reason = "validated options retain the existing dev_server API location" +)] +#![expect( + clippy::arbitrary_source_item_ordering, + reason = "compatibility entrypoints, normalization, native ownership and startup policy remain grouped by responsibility" +)] + use std::fs; use std::future; +use std::future::Future; #[cfg(test)] use std::iter; use std::net::{SocketAddr, TcpListener as StdTcpListener}; use std::path::{Path, PathBuf}; +use std::pin::Pin; use std::str::FromStr as _; use std::sync::Arc; +pub use crate::run_options::AxumRunOptions; +use crate::run_options::{failure, shutdown_grace}; use anyhow::Context as _; +use edgezero_core::probe::LifecyclePhase; +use std::time::Duration; use tokio::net::TcpListener as TokioTcpListener; use tokio::runtime::Builder as RuntimeBuilder; use tokio::signal; +#[cfg(test)] use tokio::sync::oneshot::{Receiver as ShutdownReceiver, Sender as ShutdownSender, channel}; -use tokio::task::{LocalSet, spawn_blocking, spawn_local}; +use tokio::sync::watch; +#[cfg(test)] +use tokio::task::spawn_blocking; +use tokio::task::{JoinSet, LocalSet}; use edgezero_core::addr; use edgezero_core::app::{App, Hooks, StoreMetadata, StoresMetadata}; @@ -26,15 +45,54 @@ use edgezero_core::store_registry::{ }; use log::LevelFilter; use simple_logger::SimpleLogger; + use std::collections::BTreeMap; +use std::io::ErrorKind; +use std::process; +use tokio::task::yield_now; +use tokio::time::{Instant as NativeInstant, sleep_until}; use crate::config_store::AxumConfigStore; -use crate::connection::serve_http1; +use crate::connection::serve_http1_with_shutdown; use crate::key_value_store::PersistentKvStore; +use crate::outbound::AxumOutboundClient; use crate::response::EgressConnection; use crate::secret_store::EnvSecretStore; use crate::service::AxumServiceState; +/// Local application-owned startup work, borrowing the unshared App and prepared stores. +pub type LocalStartupFuture<'startup> = + Pin> + 'startup>>; + +/// The exact registries prepared for requests. Handles may be cloned by the application; +/// retained KV clones must be released by their owners before reopening or backing up. +#[derive(Default)] +pub struct PreparedStores { + config: Option, + kv: Option, + secrets: Option, +} + +impl PreparedStores { + #[must_use] + #[inline] + pub fn config(&self) -> Option<&ConfigRegistry> { + self.config.as_ref() + } + + #[must_use] + #[inline] + pub fn kv(&self) -> Option<&KvRegistry> { + self.kv.as_ref() + } + + #[must_use] + #[inline] + pub fn secrets(&self) -> Option<&SecretRegistry> { + self.secrets.as_ref() + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum KvInitRequirement { Optional, @@ -94,32 +152,18 @@ impl AxumDevServer { /// Returns an error if the dev server fails to bind, the Tokio runtime fails to start, or the underlying request loop returns an error. #[inline] pub fn run(self) -> anyhow::Result<()> { - let runtime = RuntimeBuilder::new_multi_thread() - .enable_all() - .build() - .context("failed to build tokio runtime")?; - - runtime.block_on(async move { self.run_async().await }) - } - - async fn run_async(self) -> anyhow::Result<()> { - let AxumDevServer { + let Self { router, config, stores, } = self; - - // Allow binding to already-open listener if caller created one to surface errors early. - let std_listener = StdTcpListener::bind(config.addr) - .with_context(|| format!("failed to bind dev server to {}", config.addr))?; - std_listener - .set_nonblocking(true) - .context("failed to set listener to non-blocking")?; - - let listener = TokioTcpListener::from_std(std_listener) - .context("failed to adopt std listener into tokio")?; - - serve_with_stores(App::new(router), listener, config.enable_ctrl_c, stores).await + run_local( + ListenerSource::Bind(config.addr), + config.enable_ctrl_c, + Duration::from_secs(10), + future::pending::<()>(), + async move { prepare_transport(App::new(router), stores.into()) }, + ) } #[cfg(test)] @@ -203,18 +247,16 @@ fn kv_init_requirement(stores: StoresMetadata) -> KvInitRequirement { } } -fn kv_store_path(store_name: &str) -> PathBuf { - // Every declared id gets its own slug-based filename. The - // pre-rewrite hard-coded `.edgezero/kv.redb` shortcut for - // store_name == "EDGEZERO_KV" is gone -- the runtime no longer - // hands out a default name; if you reach here you have a real - // declared id and the slug encoding handles every shape - // uniformly. - PathBuf::from(".edgezero").join(format!( +fn kv_store_file_name(store_name: &str) -> String { + format!( "kv-{}-{:016x}.redb", store_name_slug(store_name), stable_store_name_hash(store_name) - )) + ) +} + +fn kv_store_path(store_name: &str) -> PathBuf { + PathBuf::from(".edgezero").join(kv_store_file_name(store_name)) } fn store_name_slug(store_name: &str) -> String { @@ -274,6 +316,39 @@ fn kv_handle_from_path(kv_path: &Path) -> anyhow::Result { Ok(KvHandle::new(kv_store)) } +impl From for PreparedStores { + #[inline] + fn from(stores: Stores) -> Self { + Self { + config: stores.config_registry.or_else(|| { + stores.config_store.map(|handle| { + ConfigRegistry::single_id( + "default".to_owned(), + ConfigStoreBinding { + handle, + default_key: "default".to_owned(), + }, + ) + }) + }), + kv: stores.kv_registry.or_else(|| { + stores + .kv + .map(|handle| KvRegistry::single_id("default".to_owned(), handle)) + }), + secrets: stores.secret_registry.or_else(|| { + stores.secrets.map(|handle| { + SecretRegistry::single_id( + "default".to_owned(), + BoundSecretStore::new(handle, "default".to_owned()), + ) + }) + }), + } + } +} + +#[cfg(test)] async fn serve_with_stores( app: App, listener: TokioTcpListener, @@ -309,6 +384,7 @@ async fn serve_with_stores( result } +#[cfg(test)] fn serve_local( app: App, std_listener: StdTcpListener, @@ -316,70 +392,268 @@ fn serve_local( stores: Stores, shutdown_receiver: ShutdownReceiver<()>, ) -> anyhow::Result<()> { - let mut service = AxumServiceState::from_app(app); - if let Some(registry) = stores.config_registry { - service = service.with_config_registry(registry); - } - if let Some(handle) = stores.config_store { - service = service.with_config_store_handle(handle); - } - if let Some(registry) = stores.kv_registry { - service = service.with_kv_registry(registry); - } - if let Some(handle) = stores.kv { - service = service.with_kv_handle(handle); + run_local( + ListenerSource::Prebound(std_listener), + enable_ctrl_c, + Duration::from_secs(10), + async move { + let _closed = shutdown_receiver.await; + }, + async move { prepare_transport(app, stores.into()) }, + ) +} + +type PreparedApp = (App, PreparedStores, Arc); + +enum ListenerSource { + Bind(SocketAddr), + #[cfg(test)] + Prebound(StdTcpListener), +} + +#[cfg_attr( + not(test), + expect( + clippy::needless_pass_by_value, + reason = "test embedding consumes a prebound listener through the same source" + ) +)] +fn adopt_listener(source: ListenerSource) -> anyhow::Result { + let listener = match source { + ListenerSource::Bind(addr) => StdTcpListener::bind(addr) + .map_err(|_error| failure("bind", "listener", "address unavailable"))?, + #[cfg(test)] + ListenerSource::Prebound(listener) => listener, + }; + listener + .set_nonblocking(true) + .map_err(|_error| failure("bind", "listener", "nonblocking setup failed"))?; + TokioTcpListener::from_std(listener) + .map_err(|_error| failure("bind", "listener", "runtime adoption failed")) +} + +fn prepare_transport(app: App, stores: PreparedStores) -> anyhow::Result { + let transport = AxumOutboundClient::try_transport() + .map_err(|_error| failure("transport", "HTTP client", "initialization failed"))?; + Ok((app, stores, transport)) +} + +struct Signals { + #[cfg(unix)] + interrupt: signal::unix::Signal, + #[cfg(unix)] + terminate: signal::unix::Signal, +} + +#[expect( + clippy::arbitrary_source_item_ordering, + reason = "registration precedes observation in this lifecycle owner" +)] +impl Signals { + fn register(enabled: bool) -> anyhow::Result> { + if !enabled { + return Ok(None); + } + #[cfg(unix)] + { + let interrupt = signal::unix::signal(signal::unix::SignalKind::interrupt()) + .map_err(|_error| failure("signals", "SIGINT", "registration failed"))?; + let terminate = signal::unix::signal(signal::unix::SignalKind::terminate()) + .map_err(|_error| failure("signals", "SIGTERM", "registration failed"))?; + Ok(Some(Self { + interrupt, + terminate, + })) + } + #[cfg(not(unix))] + { + Ok(Some(Self {})) + } } - if let Some(registry) = stores.secret_registry { - service = service.with_secret_registry(registry); + + #[expect( + clippy::integer_division_remainder_used, + reason = "tokio select branch arithmetic" + )] + async fn next(&mut self) -> anyhow::Result<()> { + #[cfg(unix)] + { + let received = tokio::select! { + received = self.interrupt.recv() => received, + received = self.terminate.recv() => received, + }; + received.ok_or_else(|| failure("signals", "termination", "observation closed")) + } + #[cfg(not(unix))] + { + signal::ctrl_c() + .await + .map_err(|_error| failure("signals", "SIGINT", "registration failed")) + } } - if let Some(handle) = stores.secrets { - service = service.with_secret_handle(handle); +} + +async fn next_signal(signals: &mut Option) -> anyhow::Result<()> { + match signals { + Some(source) => source.next().await, + None => future::pending().await, } +} +fn run_local( + listener: ListenerSource, + enable_signals: bool, + grace: Duration, + stop: Stop, + prepare: Prepare, +) -> anyhow::Result<()> +where + Prepare: Future>, + Stop: Future, +{ let runtime = RuntimeBuilder::new_current_thread() .enable_all() .build() - .context("failed to build axum connection runtime")?; + .map_err(|_error| failure("runtime", "local runner", "creation failed"))?; let local = LocalSet::new(); - #[expect( - clippy::integer_division_remainder_used, - reason = "tokio::select! expands to internal randomized branch selection arithmetic" - )] - let serve = async move { - let listener = TokioTcpListener::from_std(std_listener) - .context("failed to adopt listener on axum connection runtime")?; - let shutdown = async move { - let _closed = shutdown_receiver.await; - }; - let ctrl_c = async move { - if enable_ctrl_c { - let _signal = signal::ctrl_c().await; - } else { - future::pending::<()>().await; - } - }; - tokio::pin!(shutdown); - tokio::pin!(ctrl_c); + runtime.block_on(local.run_until(async move { + let mut signals = Signals::register(enable_signals)?; + let (phase, reader) = watch::channel(LifecyclePhase::Starting); + let result = run_owned(listener, grace, stop, prepare, &mut signals, &phase, reader).await; + phase.send_replace(LifecyclePhase::Stopped); + result + })) +} - loop { - tokio::select! { - accepted = listener.accept() => { - let (stream, remote_addr) = accepted.context("axum listener accept failed")?; - let responses = EgressConnection::default(); - let connection_service = service.for_connection(remote_addr, responses.clone()); - spawn_local(async move { - if let Err(error) = serve_http1(stream, connection_service, responses).await { - log::debug!("axum HTTP/1 connection ended: {error}"); - } - }); +#[expect( + clippy::integer_division_remainder_used, + reason = "tokio select branch arithmetic" +)] +async fn run_owned( + listener_source: ListenerSource, + grace: Duration, + stop: Stop, + prepare: Prepare, + signals: &mut Option, + phase: &watch::Sender, + reader: watch::Receiver, +) -> anyhow::Result<()> +where + Prepare: Future>, + Stop: Future, +{ + tokio::pin!(stop); + tokio::pin!(prepare); + let prepared = tokio::select! { + biased; + signal = next_signal(signals) => { + signal?; + phase.send_replace(LifecyclePhase::Draining); + return Ok(()); + } + () = &mut stop => { + phase.send_replace(LifecyclePhase::Draining); + return Ok(()); + } + prepared = &mut prepare => prepared?, + }; + let (app, stores, transport) = prepared; + let mut service = AxumServiceState::with_transport(app, transport).with_phase(reader.clone()); + if let Some(registry) = stores.config { + service = service.with_config_registry(registry); + } + if let Some(registry) = stores.kv { + service = service.with_kv_registry(registry); + } + if let Some(registry) = stores.secrets { + service = service.with_secret_registry(registry); + } + // Give stop observation another poll before bind/adoption and readiness. + let listener = tokio::select! { + biased; + signal = next_signal(signals) => { signal?; phase.send_replace(LifecyclePhase::Draining); return Ok(()); } + () = &mut stop => { phase.send_replace(LifecyclePhase::Draining); return Ok(()); } + result = async { yield_now().await; adopt_listener(listener_source) } => result?, + }; + phase.send_replace(LifecyclePhase::Ready); + log::info!("[edgezero] native server ready"); + let mut tasks = JoinSet::new(); + let mut outcome = Ok(()); + loop { + tokio::select! { + biased; + signal = next_signal(signals) => { outcome = signal; break; } + () = &mut stop => break, + result = tasks.join_next(), if !tasks.is_empty() => { + if result.is_some_and(|joined| joined.is_err()) { + outcome = Err(failure("runner", "connection task", "unexpected task failure")); + break; + } + } + accepted = listener.accept() => { + match accepted { + Ok((stream, remote_addr)) => { + let responses = EgressConnection::default(); + let connection_service = service.for_connection(remote_addr, responses.clone()); + let connection_phase = reader.clone(); + tasks.spawn_local(async move { + if serve_http1_with_shutdown(stream, connection_service, responses, Some(connection_phase)).await.is_err() { + log::debug!("axum HTTP/1 connection closed"); + } + }); + } + Err(_error) => { + outcome = Err(failure("runner", "listener", "accept failed")); + break; + } } - () = &mut shutdown => break, - () = &mut ctrl_c => break, } } - Ok(()) + } + phase.send_replace(LifecyclePhase::Draining); + drop(listener); + let result = drain_tasks(&mut tasks, grace, signals, outcome).await; + // Service/registry clones are dropped only after every owned connection settles. + drop(service); + result +} + +#[expect( + clippy::integer_division_remainder_used, + reason = "tokio select branch arithmetic" +)] +async fn drain_tasks( + tasks: &mut JoinSet<()>, + grace: Duration, + signals: &mut Option, + mut outcome: anyhow::Result<()>, +) -> anyhow::Result<()> { + let Some(deadline) = NativeInstant::now().checked_add(grace) else { + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + return Err(failure("shutdown", "budget", "unusable deadline")); }; - runtime.block_on(local.run_until(serve)) + while !tasks.is_empty() { + tokio::select! { + biased; + signal = next_signal(signals) => { + outcome = Err(signal.err().unwrap_or_else(|| failure("shutdown", "signal", "forced stop"))); + break; + } + () = sleep_until(deadline) => { + outcome = Err(failure("shutdown", "budget", "grace exhausted")); + break; + } + result = tasks.join_next() => { + if result.is_some_and(|joined| joined.is_err()) { + outcome = Err(failure("runner", "connection task", "unexpected task failure")); + } + } + } + } + tasks.abort_all(); + while tasks.join_next().await.is_some() {} + outcome } /// Entry point for an Axum dev-server application. @@ -392,7 +666,8 @@ fn serve_local( /// # Errors /// Returns an error if the dev server fails to bind or any required store handle cannot be initialised. fn build_app_for_dispatch() -> anyhow::Result { - App::build::(crate::AXUM_PLATFORM).context("application configuration failed") + App::build::(crate::AXUM_PLATFORM) + .map_err(|_error| anyhow::anyhow!("application configuration failed")) } /// Runs an application with the Axum development server. @@ -402,53 +677,246 @@ fn build_app_for_dispatch() -> anyhow::Result { /// binding, or connection serving fails. #[inline] pub fn run_app() -> anyhow::Result<()> { - let app = build_app_for_dispatch::()?; let env = EnvConfig::from_env(); - let stores = A::stores(); - let kv_init_requirement = kv_init_requirement(stores); - - let level = env - .logging_level() - .and_then(|raw| LevelFilter::from_str(raw).ok()) - .unwrap_or(LevelFilter::Info); + let grace = shutdown_grace(&env)?; + let resolution = resolve_addr(&env); + run_local( + ListenerSource::Bind(resolution.addr), + true, + grace, + future::pending::<()>(), + async move { + let app = build_app_for_dispatch::()?; + let metadata = A::stores(); + let level = env + .logging_level() + .and_then(|raw| LevelFilter::from_str(raw).ok()) + .unwrap_or(LevelFilter::Info); + init_logging::(level); + for _warning in &resolution.warnings { + log::warn!("invalid supplied development bind setting; using fallback"); + } + let stores = PreparedStores { + kv: build_kv_registry(metadata.kv, &env, kv_init_requirement(metadata))?, + config: build_config_registry(metadata.config, &env), + secrets: build_secret_registry(metadata.secrets, &env), + }; + prepare_transport(app, stores) + }, + ) +} +fn init_logging(level: LevelFilter) { if !A::owns_logging() { let _logger_init = SimpleLogger::new().with_level(level).init(); } +} - let resolution = resolve_addr(&env); - for warning in &resolution.warnings { - log::warn!("{warning}"); +fn no_initialize<'startup>( + _app: &'startup mut App, + _stores: &'startup PreparedStores, +) -> LocalStartupFuture<'startup> { + Box::pin(future::ready(Ok(()))) +} + +/// Runs strict framework startup with SIGINT and Unix SIGTERM. No callback is +/// required. Ready does not claim application-specific schema/bootstrap checks. +/// +/// # Errors +/// Returns classified startup/runner errors or non-success for forced shutdown. +#[inline] +pub fn run_production_app() -> anyhow::Result<()> { + run_production_selected::( + AxumRunOptions::from_env()?, + true, + future::pending::<()>(), + no_initialize, + ) +} + +/// Runs strict startup plus exactly one application-owned local initializer. +/// Pass an inline `|app, stores| Box::pin(async move { ... })` or a boxed function +/// so Rust can infer the higher-ranked borrow lifetime. Futures need not be Send. +/// +/// # Errors +/// Returns safe classified errors; callback/provider source chains are discarded. +#[inline] +pub fn run_production_app_with_initializer( + initialize: Initialize, +) -> anyhow::Result<()> +where + A: Hooks, + Initialize: for<'startup> FnOnce( + &'startup mut App, + &'startup PreparedStores, + ) -> LocalStartupFuture<'startup>, +{ + run_production_selected::( + AxumRunOptions::from_env()?, + true, + future::pending::<()>(), + initialize, + ) +} + +/// Runs production with explicit options and caller-owned stop, installing no +/// process-wide signal handlers and merging no ambient hosting/store settings. +/// +/// # Errors +/// Returns classified startup/runner errors or grace exhaustion. +#[inline] +pub fn run_app_with_options(options: AxumRunOptions, stop: Stop) -> anyhow::Result<()> +where + A: Hooks, + Stop: Future, +{ + run_app_with_options_and_initializer::(options, stop, no_initialize) +} + +/// Runs production embedding with app-selected startup checks before acceptance. +/// Retained cloned KV handles remain the caller's responsibility after return. +/// +/// # Errors +/// Returns classified startup/runner errors or grace exhaustion. +#[inline] +pub fn run_app_with_options_and_initializer( + options: AxumRunOptions, + stop: Stop, + initialize: Initialize, +) -> anyhow::Result<()> +where + A: Hooks, + Stop: Future, + Initialize: for<'startup> FnOnce( + &'startup mut App, + &'startup PreparedStores, + ) -> LocalStartupFuture<'startup>, +{ + run_production_selected::(options, false, stop, initialize) +} + +fn run_production_selected( + options: AxumRunOptions, + signals: bool, + stop: Stop, + initialize: Initialize, +) -> anyhow::Result<()> +where + A: Hooks, + Stop: Future, + Initialize: for<'startup> FnOnce( + &'startup mut App, + &'startup PreparedStores, + ) -> LocalStartupFuture<'startup>, +{ + run_local( + ListenerSource::Bind(options.addr()), + signals, + options.shutdown_grace(), + stop, + async move { + let metadata = A::stores(); + options.validate_stores(metadata)?; + let mut app = build_app_for_dispatch::()?; + init_logging::(options.logging_level()); + let stores = build_production_stores(metadata, &options)?; + let transport = AxumOutboundClient::try_transport() + .map_err(|_error| failure("transport", "HTTP client", "initialization failed"))?; + initialize(&mut app, &stores).await.map_err(|error| { + let category = error.store_extraction_reason().map_or_else( + || "application checks failed".to_owned(), + |reason| format!("{reason:?}"), + ); + failure("initializer", "application", &category) + })?; + Ok((app, stores, transport)) + }, + ) +} + +fn check_data_write_access(root: &Path) -> anyhow::Result<()> { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + for _attempt in 0..8_u8 { + let sequence = NEXT.fetch_add(1, Ordering::Relaxed); + let path = root.join(format!( + ".edgezero-write-check-{}-{sequence}", + process::id() + )); + match fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + { + Ok(file) => { + drop(file); + fs::remove_file(path).map_err(|_error| { + failure("stores", "DATA_DIR", "write-probe cleanup failed") + })?; + return Ok(()); + } + Err(error) if error.kind() == ErrorKind::AlreadyExists => {} + Err(_error) => return Err(failure("stores", "DATA_DIR", "directory is not writable")), + } } - let addr = resolution.addr; - log::info!("[edgezero] starting axum server on http://{addr}"); + Err(failure("stores", "DATA_DIR", "write-probe unavailable")) +} - let runtime = RuntimeBuilder::new_multi_thread() - .enable_all() - .build() - .context("failed to build tokio runtime")?; - - runtime.block_on(async move { - let std_listener = StdTcpListener::bind(addr) - .with_context(|| format!("failed to bind dev server to {addr}"))?; - std_listener - .set_nonblocking(true) - .context("failed to set listener to non-blocking")?; - let listener = TokioTcpListener::from_std(std_listener) - .context("failed to adopt std listener into tokio")?; - - let kv_registry = build_kv_registry(stores.kv, &env, kv_init_requirement)?; - let config_registry = build_config_registry(stores.config, &env); - let secret_registry = build_secret_registry(stores.secrets, &env); - - let request_stores = Stores { - config_registry, - kv_registry, - secret_registry, - ..Stores::default() - }; - serve_with_stores(app, listener, true, request_stores).await - }) +fn build_production_stores( + metadata: StoresMetadata, + options: &AxumRunOptions, +) -> anyhow::Result { + let mut stores = PreparedStores::default(); + if let Some(meta) = metadata.config { + let root = options + .config_dir() + .ok_or_else(|| failure("settings", "CONFIG_DIR", "required"))?; + let mut by_id = BTreeMap::new(); + for id in meta.ids { + let store = + AxumConfigStore::from_required_path(&root.join(format!("local-config-{id}.json"))) + .map_err(|error| failure("config", id, &error.to_string()))?; + by_id.insert( + (*id).to_owned(), + ConfigStoreBinding { + handle: ConfigStoreHandle::new(Arc::new(store)), + default_key: options.env_config().store_key("config", id), + }, + ); + } + stores.config = Some( + ConfigRegistry::from_parts(by_id, meta.default.to_owned()).ok_or_else(|| { + failure("config", meta.default, "invalid declared default binding") + })?, + ); + } + if let Some(meta) = metadata.kv { + let root = options + .data_dir() + .ok_or_else(|| failure("settings", "DATA_DIR", "required"))?; + check_data_write_access(root)?; + let mut by_id = BTreeMap::new(); + let mut handles: BTreeMap = BTreeMap::new(); + for id in meta.ids { + let name = options.env_config().store_name("kv", id); + let handle = if let Some(handle) = handles.get(&name) { + handle.clone() + } else { + let store = PersistentKvStore::new(root.join(kv_store_file_name(&name))) + .map_err(|_error| failure("KV", id, "database unavailable or locked"))?; + let handle = KvHandle::new(Arc::new(store)); + handles.insert(name, handle.clone()); + handle + }; + by_id.insert((*id).to_owned(), handle); + } + stores.kv = Some( + KvRegistry::from_parts(by_id, meta.default.to_owned()) + .ok_or_else(|| failure("KV", meta.default, "invalid declared default binding"))?, + ); + } + stores.secrets = build_secret_registry(metadata.secrets, options.env_config()); + Ok(stores) } /// Build the per-request KV registry from baked store metadata. @@ -466,32 +934,27 @@ fn build_kv_registry( }; let mut by_id: BTreeMap = BTreeMap::new(); + let mut handles: BTreeMap = BTreeMap::new(); for id in meta.ids { let store_name = env.store_name("kv", id); + if let Some(handle) = handles.get(&store_name) { + by_id.insert((*id).to_owned(), handle.clone()); + continue; + } let kv_path = kv_store_path(&store_name); let handle = match kv_handle_from_path(&kv_path) { Ok(handle) => handle, - Err(err) => match init { + Err(_error) => match init { KvInitRequirement::Optional => { - log::warn!( - "KV store '{}' (id `{}`) could not be initialized at {}: {}", - store_name, - id, - kv_path.display(), - err - ); + log::warn!("KV store id `{id}` unavailable; omitting optional binding"); continue; } KvInitRequirement::Required => { - return Err(err.context(format!( - "KV store '{}' (id `{}`) is explicitly configured for axum but could not be initialized at {}", - store_name, - id, - kv_path.display() - ))); + return Err(failure("KV", id, "declared database unavailable or locked")); } }, }; + handles.insert(store_name, handle.clone()); by_id.insert((*id).to_owned(), handle); } @@ -521,13 +984,9 @@ fn build_config_registry( for id in meta.ids { let store = match AxumConfigStore::from_local_file(id) { Ok(store) => store, - Err(err) => { + Err(_error) => { log::warn!( - "config store for id `{}` could not be loaded from {}: {}; \ - dropping this id from the registry", - id, - AxumConfigStore::local_path(id).display(), - err + "config store id `{id}` could not be loaded; dropping this id from the registry" ); continue; } @@ -648,6 +1107,45 @@ mod tests { assert_eq!(app.platform(), crate::AXUM_PLATFORM); } + #[test] + fn initializer_borrows_inputs_across_await_and_captures_rc() { + use futures::executor::block_on; + use std::cell::Cell; + use std::rc::Rc; + + async fn initialize( + app: &mut App, + stores: &PreparedStores, + callback: Initialize, + ) -> Result<(), EdgeError> + where + Initialize: for<'startup> FnOnce( + &'startup mut App, + &'startup PreparedStores, + ) -> LocalStartupFuture<'startup>, + { + callback(app, stores).await + } + let calls = Rc::new(Cell::new(0_u32)); + let recorded = Rc::clone(&calls); + let mut owned_app = App::new(RouterService::builder().build()); + block_on(initialize( + &mut owned_app, + &PreparedStores::default(), + move |app, stores| { + Box::pin(async move { + future::ready(()).await; + assert!(stores.config().is_none()); + app.insert_state(Arc::::from("initialized")); + recorded.set(recorded.get() + 1); + Ok(()) + }) + }, + )) + .expect("initialize"); + assert_eq!(calls.get(), 1); + } + #[test] fn default_config_uses_expected_address() { let config = AxumDevServerConfig::default(); diff --git a/crates/edgezero-adapter-axum/src/key_value_store.rs b/crates/edgezero-adapter-axum/src/key_value_store.rs index bb375cf39..c4cf72b08 100644 --- a/crates/edgezero-adapter-axum/src/key_value_store.rs +++ b/crates/edgezero-adapter-axum/src/key_value_store.rs @@ -158,11 +158,9 @@ impl PersistentKvStore { /// Returns an error if the database file cannot be opened or initialised (corrupted file, locked by another process, or insufficient permissions). #[inline] pub fn new>(path: P) -> Result { - let db_path = path.as_ref().display().to_string(); - let db = Database::create(path).map_err(|err| { + let db = Database::create(path).map_err(|_error| { KvError::Internal(anyhow::anyhow!( - "Failed to open KV database at {db_path}. If the file is corrupted or locked \ - by another process, try deleting it and restarting: {err}" + "KV database unavailable; release other owners or check storage permissions, capacity and integrity" )) })?; diff --git a/crates/edgezero-adapter-axum/src/lib.rs b/crates/edgezero-adapter-axum/src/lib.rs index d4e33f85c..1cde330e3 100644 --- a/crates/edgezero-adapter-axum/src/lib.rs +++ b/crates/edgezero-adapter-axum/src/lib.rs @@ -17,6 +17,8 @@ pub mod request; #[cfg(feature = "axum")] mod response; #[cfg(feature = "axum")] +mod run_options; +#[cfg(feature = "axum")] pub mod secret_store; #[cfg(feature = "axum")] mod service; diff --git a/crates/edgezero-adapter-axum/src/run_options.rs b/crates/edgezero-adapter-axum/src/run_options.rs new file mode 100644 index 000000000..eac43bad2 --- /dev/null +++ b/crates/edgezero-adapter-axum/src/run_options.rs @@ -0,0 +1,372 @@ +use std::env::vars_os; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::str::FromStr as _; +use std::time::{Duration, Instant}; + +use edgezero_core::addr; +use edgezero_core::app::StoresMetadata; +use edgezero_core::env_config::EnvConfig; +use log::LevelFilter; + +/// Validated explicit production hosting options. Construction never selects a +/// policy from build profile, cwd or container detection. Explicit constructors +/// do not merge environment values; only `from_env` reads hosting environment. +#[derive(Clone)] +pub struct AxumRunOptions { + addr: SocketAddr, + config_dir: Option, + data_dir: Option, + env: EnvConfig, + grace: Duration, + level: LevelFilter, + non_unicode_overrides: Vec>, +} + +impl AxumRunOptions { + pub(crate) fn addr(&self) -> SocketAddr { + self.addr + } + + pub(crate) fn config_dir(&self) -> Option<&Path> { + self.config_dir.as_deref() + } + + pub(crate) fn data_dir(&self) -> Option<&Path> { + self.data_dir.as_deref() + } + + pub(crate) fn env_config(&self) -> &EnvConfig { + &self.env + } + + /// Captures hosting settings and store overrides once. Secret values still + /// use the existing environment-backed secret backend. + /// + /// # Errors + /// Returns safe setting categories for supplied invalid hosting settings. + #[inline] + pub fn from_env() -> anyhow::Result { + let mut pairs = Vec::new(); + let mut non_unicode = Vec::new(); + for (os_key, os_value) in vars_os() { + let Some(key) = os_key.to_str() else { continue }; + let Some(rest) = key.strip_prefix("EDGEZERO__") else { + continue; + }; + let path: Vec = rest.split("__").map(str::to_ascii_lowercase).collect(); + if let Some(value) = os_value.to_str() { + pairs.push((key.to_owned(), value.to_owned())); + } else if let Some(setting) = fixed_setting(&path) { + return Err(failure("settings", setting, "non-Unicode value")); + } else { + let segments: Vec<&str> = path.iter().map(String::as_str).collect(); + if matches!( + segments.as_slice(), + ["stores", "config" | "kv" | "secrets", _, "name" | "key"] + ) { + non_unicode.push(path); + } + } + } + let mut options = Self::from_vars(pairs)?; + options.non_unicode_overrides = non_unicode; + Ok(options) + } + + /// Constructs options using only supplied string pairs and omission defaults. + /// Declared-store overrides are validated against baked metadata at startup. + /// + /// # Errors + /// Returns safe setting categories for invalid bind, logging, roots or grace. + #[inline] + pub fn from_vars(vars: I) -> anyhow::Result + where + I: IntoIterator, + K: AsRef, + V: Into, + { + let env = EnvConfig::from_vars(vars); + let host = env.adapter_host().map_or(Ok(addr::DEFAULT_HOST), |raw| { + raw.parse() + .map_err(|_error| failure("settings", "HOST", "invalid IP literal")) + })?; + let port = env.adapter_port().map_or(Ok(addr::DEFAULT_PORT), |raw| { + raw.parse::() + .ok() + .filter(|port| *port != 0) + .ok_or_else(|| failure("settings", "PORT", "invalid nonzero port")) + })?; + let mut options = Self::new(SocketAddr::new(host, port))?; + if let Some(level) = env.logging_level() { + options.level = LevelFilter::from_str(level) + .map_err(|_error| failure("settings", "LOGGING__LEVEL", "invalid level"))?; + } + options.grace = shutdown_grace(&env)?; + if let Some(root) = env.get(&["adapter", "config_dir"]) { + options = options.with_config_dir(root)?; + } + if let Some(root) = env.get(&["adapter", "data_dir"]) { + options = options.with_data_dir(root)?; + } + options.env = env; + Ok(options) + } + + pub(crate) fn logging_level(&self) -> LevelFilter { + self.level + } + + /// Explicit bind with default grace/logging, no roots, and no store overrides. + /// + /// # Errors + /// Rejects port zero. Production omission defaults apply only to omitted values. + #[inline] + pub fn new(addr: SocketAddr) -> anyhow::Result { + if addr.port() == 0 { + return Err(failure("settings", "PORT", "invalid nonzero port")); + } + Ok(Self { + addr, + config_dir: None, + data_dir: None, + env: EnvConfig::default(), + grace: Duration::from_secs(10), + level: LevelFilter::Info, + non_unicode_overrides: Vec::new(), + }) + } + + pub(crate) fn shutdown_grace(&self) -> Duration { + self.grace + } + + pub(crate) fn validate_stores(&self, metadata: StoresMetadata) -> anyhow::Result<()> { + for (kind, declaration) in [ + ("config", metadata.config), + ("kv", metadata.kv), + ("secrets", metadata.secrets), + ] { + if let Some(meta) = declaration { + for id in meta.ids { + for field in ["name", "key"] { + if self + .non_unicode_overrides + .iter() + .any(|path| match path.as_slice() { + [_, selected_kind, selected_id, selected_field] => { + selected_kind == kind + && selected_id.eq_ignore_ascii_case(id) + && selected_field == field + } + _ => false, + }) + { + return Err(failure("settings", id, "non-Unicode store override")); + } + if self.env.store_setting(kind, id, field).is_some_and(|raw| { + raw.is_empty() + || raw.chars().all(char::is_whitespace) + || raw.chars().any(char::is_control) + }) { + return Err(failure("settings", id, "invalid store NAME/KEY override")); + } + } + } + } + } + for (meta, root, label) in [ + (metadata.config, self.config_dir.as_deref(), "CONFIG_DIR"), + (metadata.kv, self.data_dir.as_deref(), "DATA_DIR"), + ] { + if meta.is_some() && root.is_none() { + return Err(failure("settings", label, "required for declared stores")); + } + if let Some(directory) = root { + validate_root(directory, label)?; + } + } + Ok(()) + } + + /// Selects an existing absolute config root; never creates it. + /// + /// # Errors + /// Rejects relative, missing or non-directory roots. + #[inline] + pub fn with_config_dir>(mut self, root: Root) -> anyhow::Result { + let directory = root.into(); + validate_root(&directory, "CONFIG_DIR")?; + self.config_dir = Some(directory); + Ok(self) + } + + /// Selects an existing absolute data root. Startup also checks actual write access. + /// + /// # Errors + /// Rejects relative, missing or non-directory roots. + #[inline] + pub fn with_data_dir>(mut self, root: Root) -> anyhow::Result { + let directory = root.into(); + validate_root(&directory, "DATA_DIR")?; + self.data_dir = Some(directory); + Ok(self) + } + + #[inline] + #[must_use] + pub fn with_logging_level(mut self, level: LevelFilter) -> Self { + self.level = level; + self + } + + /// Selects a positive, representable integer-second drain budget. + /// + /// # Errors + /// Rejects zero, fractional seconds and unusable native timer values. + #[inline] + pub fn with_shutdown_grace(mut self, grace: Duration) -> anyhow::Result { + validate_grace(grace)?; + self.grace = grace; + Ok(self) + } +} + +pub(crate) fn failure(stage: &str, subject: &str, category: &str) -> anyhow::Error { + // No source chain survives this boundary. Subject is a fixed setting name or + // baked logical id, never settings/config/secret contents or a provider string. + anyhow::anyhow!("startup {stage}: {subject}: {category}") +} + +fn fixed_setting(path: &[String]) -> Option<&'static str> { + let segments: Vec<&str> = path.iter().map(String::as_str).collect(); + match segments.as_slice() { + ["adapter", "host"] => Some("HOST"), + ["adapter", "port"] => Some("PORT"), + ["adapter", "config_dir"] => Some("CONFIG_DIR"), + ["adapter", "data_dir"] => Some("DATA_DIR"), + ["adapter", "shutdown_grace_seconds"] => Some("SHUTDOWN_GRACE_SECONDS"), + ["logging", "level"] => Some("LOGGING__LEVEL"), + _ => None, + } +} + +pub(crate) fn shutdown_grace(env: &EnvConfig) -> anyhow::Result { + let Some(raw) = env.get(&["adapter", "shutdown_grace_seconds"]) else { + return Ok(Duration::from_secs(10)); + }; + let seconds = raw + .parse::() + .ok() + .filter(|_seconds| raw.bytes().all(|byte| byte.is_ascii_digit())) + .ok_or_else(|| { + failure( + "settings", + "SHUTDOWN_GRACE_SECONDS", + "invalid positive integer seconds", + ) + })?; + let grace = Duration::from_secs(seconds); + validate_grace(grace)?; + Ok(grace) +} + +fn validate_grace(grace: Duration) -> anyhow::Result<()> { + if grace.is_zero() || grace.subsec_nanos() != 0 || Instant::now().checked_add(grace).is_none() { + return Err(failure( + "settings", + "SHUTDOWN_GRACE_SECONDS", + "invalid or unusable positive integer seconds", + )); + } + Ok(()) +} + +fn validate_root(root: &Path, setting: &str) -> anyhow::Result<()> { + if !root.is_absolute() || !root.is_dir() { + return Err(failure( + "settings", + setting, + "requires an absolute existing directory", + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use edgezero_core::app::StoreMetadata; + + #[test] + fn defaults_and_invalid_supplied_scalars_are_distinct() { + let options = AxumRunOptions::from_vars(Vec::<(String, String)>::new()).expect("defaults"); + assert_eq!( + options.addr, + SocketAddr::from((addr::DEFAULT_HOST, addr::DEFAULT_PORT)) + ); + assert_eq!(options.grace, Duration::from_secs(10)); + for (key, values) in [ + ( + "EDGEZERO__ADAPTER__HOST", + vec!["sentinel-host", "", "localhost"], + ), + ( + "EDGEZERO__ADAPTER__PORT", + vec!["0", "65536", "sentinel-port", ""], + ), + ("EDGEZERO__LOGGING__LEVEL", vec!["sentinel-level", ""]), + ( + "EDGEZERO__ADAPTER__SHUTDOWN_GRACE_SECONDS", + vec![ + "0", + "", + "-1", + "+1", + "sentinel-duration", + "18446744073709551615", + ], + ), + ] { + for value in values { + let error = AxumRunOptions::from_vars([(key, value)]) + .err() + .expect("invalid"); + assert!(error.chain().count() == 1); + assert!(!format!("{error:?}").contains("sentinel")); + } + } + } + + #[test] + fn selected_override_validation_precedes_normalization() { + let metadata = StoresMetadata { + config: Some(StoreMetadata { + ids: &["settings"], + default: "settings", + }), + ..StoresMetadata::default() + }; + let root = tempfile::tempdir().expect("root"); + for value in ["", " ", "sentinel\n"] { + let options = AxumRunOptions::from_vars([ + ( + "EDGEZERO__ADAPTER__CONFIG_DIR", + root.path().to_str().expect("path"), + ), + ("EDGEZERO__STORES__CONFIG__SETTINGS__KEY", value), + ("EDGEZERO__STORES__KV__UNDECLARED__NAME", "\n"), + ]) + .expect("native settings"); + let error = options + .validate_stores(metadata) + .expect_err("declared invalid override"); + assert!(!format!("{error:?}").contains("sentinel")); + } + let options = + AxumRunOptions::new(SocketAddr::from(([127, 0, 0, 1], 1234))).expect("explicit"); + assert!(options.env.adapter_host().is_none()); + assert_eq!(options.env.store_key("config", "settings"), "settings"); + assert!(options.validate_stores(metadata).is_err()); + } +} diff --git a/crates/edgezero-adapter-axum/src/service.rs b/crates/edgezero-adapter-axum/src/service.rs index ed1ddbfe4..76d4c6fb8 100644 --- a/crates/edgezero-adapter-axum/src/service.rs +++ b/crates/edgezero-adapter-axum/src/service.rs @@ -17,6 +17,7 @@ use edgezero_core::ingress::{ IngressHeadParts, validate_normalized_ingress_parts, }; use edgezero_core::key_value_store::KvHandle; +use edgezero_core::probe::{LifecyclePhase, LifecycleReader}; #[cfg(test)] use edgezero_core::router::RouterService; use edgezero_core::secret_store::SecretHandle; @@ -26,11 +27,13 @@ use edgezero_core::store_registry::{ use edgezero_core::time::MonotonicInstant; use hyper::body::Incoming; use hyper::service::Service as HyperService; +use tokio::sync::watch; #[cfg(test)] use tower::Service as TowerService; #[cfg(test)] use crate::connection::{ConnectionExit, serve_http1}; +#[cfg(test)] use crate::outbound::AxumOutboundClient; use crate::request::into_core_request_parts; use crate::response::{AxumEgressBody, EgressConnection, prepare_egress_response}; @@ -49,6 +52,7 @@ pub(crate) struct AxumServiceState { kv_handle: Option, kv_registry: Option, outbound_transport: Option>, + phase: Option>, secret_handle: Option, secret_registry: Option, } @@ -61,14 +65,23 @@ impl AxumServiceState { /// Creates a service that preserves all policies configured on `App`. #[must_use] #[inline] + #[cfg(test)] pub fn from_app(app: App) -> Self { + Self::with_transport( + app, + AxumOutboundClient::try_transport().expect("test transport"), + ) + } + + pub(crate) fn with_transport(app: App, transport: Arc) -> Self { Self { app: Arc::new(app), config_registry: None, config_store_handle: None, kv_handle: None, kv_registry: None, - outbound_transport: AxumOutboundClient::try_transport().ok(), + outbound_transport: Some(transport), + phase: None, secret_handle: None, secret_registry: None, } @@ -101,6 +114,7 @@ impl AxumServiceState { /// directly. #[must_use] #[inline] + #[cfg(test)] pub fn with_config_store_handle(mut self, handle: ConfigStoreHandle) -> Self { self.config_store_handle = Some(handle); self @@ -118,6 +132,7 @@ impl AxumServiceState { /// directly. #[must_use] #[inline] + #[cfg(test)] pub fn with_kv_handle(mut self, handle: KvHandle) -> Self { self.kv_handle = Some(handle); self @@ -144,6 +159,7 @@ impl AxumServiceState { /// [`Self::with_secret_registry`] directly. #[must_use] #[inline] + #[cfg(test)] pub fn with_secret_handle(mut self, handle: SecretHandle) -> Self { self.secret_handle = Some(handle); self @@ -157,6 +173,11 @@ impl AxumServiceState { self } + pub(crate) fn with_phase(mut self, phase: watch::Receiver) -> Self { + self.phase = Some(phase); + self + } + pub(crate) fn for_connection( &self, remote_addr: SocketAddr, @@ -205,12 +226,23 @@ impl AxumServiceState { (config_registry, kv_registry, secret_registry) } + fn admission_open(&self) -> bool { + self.phase + .as_ref() + .is_none_or(|phase| *phase.borrow() == LifecyclePhase::Ready) + } + async fn dispatch_request( &self, mut request: Request, connection: &EgressConnection, remote_addr: Option, ) -> Result, AxumIngressAbort> { + // This private gate defines dispatched work. Public extensions and app + // admission policy cannot admit new work after the owner starts draining. + if !self.admission_open() { + return Err(AxumIngressAbort); + } if let Some(peer_addr) = remote_addr { request.extensions_mut().insert(ConnectInfo(peer_addr)); } @@ -288,6 +320,11 @@ impl AxumServiceState { } }; + if let Some(phase) = self.phase.clone() { + core_request + .extensions_mut() + .insert(LifecycleReader::new(move || Some(*phase.borrow()))); + } if let Some(registry) = config_registry { core_request.extensions_mut().insert(registry); } diff --git a/crates/edgezero-adapter-axum/tests/fixtures/production_host.rs b/crates/edgezero-adapter-axum/tests/fixtures/production_host.rs new file mode 100644 index 000000000..c13358088 --- /dev/null +++ b/crates/edgezero-adapter-axum/tests/fixtures/production_host.rs @@ -0,0 +1,351 @@ +//! Test-only executable for real startup, signal, drain and storage acceptance. +#![expect(clippy::print_stdout, reason = "bounded subprocess IPC markers")] + +#[path = "../../../edgezero-core/tests/fixtures/production_handlers.rs"] +mod fixture; + +use std::cell::Cell; +use std::env; +use std::future; +use std::net::SocketAddr; +use std::rc::Rc; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bytes::Bytes; +use edgezero_adapter_axum::dev_server::{ + AxumRunOptions, run_app, run_app_with_options_and_initializer, run_production_app, + run_production_app_with_initializer, +}; +use edgezero_core::app::{App, Hooks, StoreMetadata, StoresMetadata}; +use edgezero_core::context::RequestContext; +use edgezero_core::extractor::{AppConfig, State}; +use edgezero_core::key_value_store::KvHandle; +use edgezero_core::probe::{self, LifecyclePhase, LifecycleReader}; +use edgezero_core::router::RouterService; +use edgezero_core::{Body, EdgeError, MonotonicClock, MonotonicInstant, action}; +use futures::stream; +use tokio::sync::oneshot; +use tokio::time::sleep; + +struct FixtureApp; + +#[expect( + clippy::missing_trait_methods, + reason = "fixture retains unused Hooks defaults" +)] +impl Hooks for FixtureApp { + fn configure(app: &mut App) -> Result<(), EdgeError> { + if mode() == "hook-error" { + return Err(EdgeError::internal(anyhow::anyhow!("HOOK_SENTINEL"))); + } + if env::var_os("FIXTURE_FREEZE_CLOCK").is_some() { + let frozen = MonotonicInstant::now(); + app.set_monotonic_clock(MonotonicClock::new(move || frozen)); + } + Ok(()) + } + + fn owns_logging() -> bool { + true + } + + fn routes() -> RouterService { + RouterService::builder() + .get("/ready", probe::readiness) + .get("/live", probe::liveness) + .get("/state", fixture::state) + .get("/typed", fixture::typed) + .get("/finite", finite) + .get("/pending", pending) + .get("/stream", endless_stream) + .get("/flood", flood) + .get("/after", after) + .get("/write", write) + .get("/read", read) + .get("/alias", alias) + .get("/distinct", distinct) + .get("/snapshot", snapshot) + .get("/stop", stop) + .build() + } + + fn stores() -> StoresMetadata { + if matches!( + mode().as_str(), + "bare" + | "bare-dev" + | "hook-error" + | "initializer-error" + | "starting" + | "embedding-bare" + ) { + return StoresMetadata::default(); + } + StoresMetadata { + config: Some(StoreMetadata { + ids: &["settings"], + default: "settings", + }), + kv: Some(StoreMetadata { + ids: &["first", "second", "third"], + default: "first", + }), + secrets: Some(StoreMetadata { + ids: &["vault"], + default: "vault", + }), + } + } +} + +struct WorkGuard(&'static str); + +impl Drop for WorkGuard { + fn drop(&mut self) { + println!("{}", self.0); + } +} + +struct StopState(Mutex>>); + +fn mode() -> String { + env::var("FIXTURE_MODE").unwrap_or_else(|_| "initialized".to_owned()) +} + +fn reader(ctx: &RequestContext) -> Result { + ctx.extensions() + .get::() + .cloned() + .ok_or_else(|| EdgeError::service_unavailable("fixture reader missing")) +} + +#[action] +async fn after() -> Result<&'static str, EdgeError> { + println!("FORBIDDEN_DISPATCH"); + Ok("forbidden dispatch") +} + +#[action] +async fn finite(ctx: RequestContext) -> Result<&'static str, EdgeError> { + let observed = reader(&ctx)?; + let _guard = WorkGuard("FINITE_DROPPED"); + println!("DISPATCH"); + while observed.phase() == Some(LifecyclePhase::Ready) { + sleep(Duration::from_millis(5)).await; + } + println!("DRAINING"); + sleep(Duration::from_millis(80)).await; + Ok("finite-complete") +} + +#[action] +async fn pending(ctx: RequestContext) -> Result<&'static str, EdgeError> { + let observed = reader(&ctx)?; + let _guard = WorkGuard("WORK_DROPPED"); + println!("DISPATCH"); + let mut reported = false; + #[expect( + clippy::infinite_loop, + reason = "fixture proves process shutdown cancels a never-ending handler" + )] + loop { + if !reported && observed.phase() == Some(LifecyclePhase::Draining) { + println!("DRAINING"); + reported = true; + } + sleep(Duration::from_millis(5)).await; + } +} + +#[action] +async fn endless_stream(ctx: RequestContext) -> Result { + let observed = reader(&ctx)?; + let guard = Rc::new(WorkGuard("STREAM_DROPPED")); + println!("STREAM_READY"); + Ok(Body::from_stream(async_stream::stream! { + let _retained = guard; + yield Ok(Bytes::from_static(b"stream-prefix")); + let mut reported = false; + loop { + if !reported && observed.phase() == Some(LifecyclePhase::Draining) { + println!("DRAINING"); + reported = true; + } + sleep(Duration::from_millis(5)).await; + } + })) +} + +#[action] +async fn flood() -> Result { + static BLOCK: [u8; 0x0001_0000] = [b'x'; 0x0001_0000]; + let guard = Rc::new(WorkGuard("STREAM_DROPPED")); + println!("STREAM_READY"); + Ok(Body::from_stream(stream::repeat_with(move || { + let _retained = &guard; + Ok(Bytes::from_static(&BLOCK)) + }))) +} + +fn kv(ctx: &RequestContext, id: &str) -> Result { + ctx.kv_store(id) + .ok_or_else(|| EdgeError::service_unavailable("fixture KV missing")) +} + +#[action] +async fn write(ctx: RequestContext) -> Result<&'static str, EdgeError> { + kv(&ctx, "first")? + .put("persisted", &"durable-value") + .await?; + Ok("written") +} + +#[action] +async fn read(ctx: RequestContext) -> Result { + Ok(kv(&ctx, "first")? + .get::("persisted") + .await? + .unwrap_or_default()) +} + +#[action] +async fn alias(ctx: RequestContext) -> Result { + Ok(kv(&ctx, "second")? + .get::("persisted") + .await? + .unwrap_or_default()) +} + +#[action] +async fn distinct(ctx: RequestContext) -> Result { + Ok(kv(&ctx, "third")? + .get::("persisted") + .await? + .unwrap_or_default()) +} + +#[action] +async fn snapshot(ctx: RequestContext) -> Result { + let binding = ctx + .config_store_default_binding() + .ok_or_else(|| EdgeError::service_unavailable("fixture config missing"))?; + Ok(binding + .handle + .get(&binding.default_key) + .await? + .unwrap_or_default()) +} + +#[action] +async fn stop(State(state): State>) -> Result<&'static str, EdgeError> { + if let Some(sender) = state + .0 + .lock() + .map_err(|_error| EdgeError::internal(anyhow::anyhow!("fixture stop lock poisoned")))? + .take() + { + let _sent = sender.send(()); + } + Ok("stopping") +} + +#[expect( + clippy::panic_in_result_fn, + reason = "test-only executable asserts initializer ownership and call counts" +)] +fn main() -> anyhow::Result<()> { + let selected = mode(); + if matches!(selected.as_str(), "development" | "bare-dev") { + return run_app::(); + } + if matches!(selected.as_str(), "framework" | "bare" | "hook-error") { + return run_production_app::(); + } + if selected == "starting" { + return run_production_app_with_initializer::(|_app, _stores| { + Box::pin(async { + let _guard = WorkGuard("STARTUP_DROPPED"); + println!("INITIALIZING"); + future::pending::>().await + }) + }); + } + if selected == "initializer-error" { + return run_production_app_with_initializer::(|_app, _stores| { + Box::pin(async { Err(EdgeError::internal(anyhow::anyhow!("CALLBACK_SENTINEL"))) }) + }); + } + let calls = Rc::new(Cell::new(0_u32)); + let initialized = Rc::clone(&calls); + if selected.starts_with("embedding") { + let address: SocketAddr = env::var("FIXTURE_BIND")?.parse()?; + let mut options = AxumRunOptions::new(address)?; + if selected == "embedding" { + options = options + .with_config_dir(env::var("FIXTURE_CONFIG_ROOT")?)? + .with_data_dir(env::var("FIXTURE_DATA_ROOT")?)?; + } + let (sender, receiver) = oneshot::channel(); + let stop_capture = Rc::clone(&calls); + return run_app_with_options_and_initializer::( + options, + async move { + let _closed = receiver.await; + assert_eq!( + stop_capture.get(), + 1, + "initializer ran once before caller stop" + ); + }, + move |app, stores| { + Box::pin(async move { + if let Some(binding) = + stores.config().and_then(|registry| registry.default_ref()) + { + let config = AppConfig::::from_binding( + binding, + None, + stores.secrets(), + app.config_extraction_limits(), + app.monotonic_clock(), + ) + .await?; + app.insert_state(Arc::new(fixture::HostState { + greeting: config.greeting, + })); + } + app.insert_state(Arc::new(StopState(Mutex::new(Some(sender))))); + initialized.set(initialized.get().saturating_add(1)); + println!("INITIALIZED"); + Ok(()) + }) + }, + ); + } + run_production_app_with_initializer::(move |app, stores| { + Box::pin(async move { + let binding = stores + .config() + .and_then(|registry| registry.default_ref()) + .ok_or_else(|| { + EdgeError::service_unavailable("required fixture binding missing") + })?; + let config = AppConfig::::from_binding( + binding, + None, + stores.secrets(), + app.config_extraction_limits(), + app.monotonic_clock(), + ) + .await?; + app.insert_state(Arc::new(fixture::HostState { + greeting: config.greeting, + })); + initialized.set(initialized.get().saturating_add(1)); + assert_eq!(initialized.get(), 1, "initializer runs exactly once"); + println!("INITIALIZED"); + Ok(()) + }) + }) +} diff --git a/crates/edgezero-adapter-axum/tests/production_hosting.rs b/crates/edgezero-adapter-axum/tests/production_hosting.rs new file mode 100644 index 000000000..97e8db9a7 --- /dev/null +++ b/crates/edgezero-adapter-axum/tests/production_hosting.rs @@ -0,0 +1,652 @@ +//! Real local processes, public entrypoints and ordinary main termination output. +#![cfg(all(feature = "axum", feature = "test-utils", unix))] + +#[cfg(test)] +#[expect( + clippy::arbitrary_source_item_ordering, + reason = "subprocess fixtures group setup, process ownership and acceptance scenarios" +)] +mod tests { + use std::fs; + use std::io::{self, BufRead as _, BufReader, Read, Write as _}; + use std::net::{SocketAddr, TcpListener, TcpStream}; + use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _, symlink}; + use std::os::unix::process::ExitStatusExt as _; + use std::process::{Child, Command, ExitStatus, Stdio}; + use std::sync::{Arc, Mutex, mpsc}; + use std::thread::{self, JoinHandle}; + use std::time::{Duration, Instant}; + + use edgezero_core::blob_envelope::BlobEnvelope; + use serde_json::{Value, json}; + use tempfile::TempDir; + + const WAIT: Duration = Duration::from_secs(5); + + struct Roots { + config: TempDir, + cwd: TempDir, + data: TempDir, + } + + impl Roots { + fn new() -> Self { + let roots = Self { + config: tempfile::tempdir().expect("config root"), + cwd: tempfile::tempdir().expect("unrelated cwd"), + data: tempfile::tempdir().expect("data root"), + }; + roots.map(&json!({"settings": blob("configured-greeting", "fixture_token")})); + roots + } + + fn map(&self, value: &Value) { + fs::write( + self.config.path().join("local-config-settings.json"), + serde_json::to_vec(value).expect("JSON"), + ) + .expect("config fixture"); + } + } + + fn blob(greeting: &str, token: &str) -> String { + serde_json::to_string(&BlobEnvelope::new( + json!({"greeting": greeting, "token": token}), + "2026-10-05T00:00:00Z".to_owned(), + )) + .expect("envelope") + } + + fn address() -> SocketAddr { + TcpListener::bind("127.0.0.1:0") + .expect("port reservation") + .local_addr() + .expect("address") + } + + fn command(mode: &str, roots: &Roots, addr: SocketAddr) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_edgezero-hosting-fixture")); + command + .env_clear() + .current_dir(roots.cwd.path()) + .env("FIXTURE_MODE", mode) + .env("FIXTURE_BIND", addr.to_string()) + .env("FIXTURE_CONFIG_ROOT", roots.config.path()) + .env("FIXTURE_DATA_ROOT", roots.data.path()) + .env("EDGEZERO__ADAPTER__HOST", "127.0.0.1") + .env("EDGEZERO__ADAPTER__PORT", addr.port().to_string()) + .env("EDGEZERO__ADAPTER__SHUTDOWN_GRACE_SECONDS", "1") + .env("EDGEZERO__ADAPTER__CONFIG_DIR", roots.config.path()) + .env("EDGEZERO__ADAPTER__DATA_DIR", roots.data.path()) + .env("EDGEZERO__STORES__KV__FIRST__NAME", "shared-fixture") + .env("EDGEZERO__STORES__KV__SECOND__NAME", "shared-fixture") + .env("fixture_token", "resolved-fixture-token"); + command + } + + struct Server { + addr: SocketAddr, + child: Child, + events: mpsc::Receiver, + output: Arc>, + readers: Vec>, + } + + impl Server { + fn spawn(mut command: Command, addr: SocketAddr) -> Self { + let mut child = command + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("fixture process"); + let output = Arc::new(Mutex::new(String::new())); + let (sender, events) = mpsc::channel(); + let stdout = child.stdout.take().expect("stdout"); + let stderr = child.stderr.take().expect("stderr"); + let mut readers = Vec::new(); + let streams: [Box; 2] = [Box::new(stdout), Box::new(stderr)]; + for stream in streams { + let captured = Arc::clone(&output); + let stream_sender = sender.clone(); + readers.push(thread::spawn(move || { + for result in BufReader::new(stream).lines() { + let Ok(line) = result else { break }; + let mut captured_output = captured.lock().expect("output"); + captured_output.push_str(&line); + captured_output.push('\n'); + drop(captured_output); + let _sent = stream_sender.send(line); + } + })); + } + Self { + addr, + child, + events, + output, + readers, + } + } + + fn log(&self) -> String { + self.output.lock().expect("output").clone() + } + + fn ready(&mut self) { + let deadline = Instant::now().checked_add(WAIT).expect("startup timer"); + loop { + assert!( + Instant::now() < deadline, + "startup deadline: {}", + self.log() + ); + assert!( + self.child.try_wait().expect("status").is_none(), + "early process failure: {}", + self.log() + ); + if let Ok(response) = self.request("/ready") + && response.starts_with("HTTP/1.1 200") + { + return; + } + thread::sleep(Duration::from_millis(5)); + } + } + + fn request(&self, path: &str) -> io::Result { + let mut socket = self.socket()?; + write!( + socket, + "GET {path} HTTP/1.1\r\nHost: fixture\r\nConnection: close\r\n\r\n" + )?; + let mut response = String::new(); + socket.read_to_string(&mut response)?; + Ok(response) + } + + fn signal(&self, name: &str) { + assert!( + Command::new("kill") + .arg(format!("-{name}")) + .arg(self.child.id().to_string()) + .status() + .expect("send signal") + .success() + ); + } + + fn socket(&self) -> io::Result { + let socket = TcpStream::connect_timeout(&self.addr, Duration::from_millis(50))?; + socket.set_read_timeout(Some(WAIT))?; + socket.set_write_timeout(Some(WAIT))?; + Ok(socket) + } + + fn wait(&mut self) -> ExitStatus { + let deadline = Instant::now().checked_add(WAIT).expect("exit timer"); + loop { + if let Some(status) = self.child.try_wait().expect("process status") { + for reader in self.readers.drain(..) { + reader.join().expect("output reader"); + } + return status; + } + assert!( + Instant::now() < deadline, + "process exit deadline: {}", + self.log() + ); + thread::sleep(Duration::from_millis(5)); + } + } + + fn wait_line(&self, expected: &str) { + let deadline = Instant::now().checked_add(WAIT).expect("event timer"); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + let line = self.events.recv_timeout(remaining).unwrap_or_else(|error| { + panic!("waiting for {expected}: {error}; {}", self.log()) + }); + if line == expected { + return; + } + } + } + } + + impl Drop for Server { + fn drop(&mut self) { + let _killed = self.child.kill(); + let _status = self.child.wait(); + for reader in self.readers.drain(..) { + let _joined = reader.join(); + } + } + } + + fn failed(command: Command, addr: SocketAddr, category: &str) -> String { + let mut server = Server::spawn(command, addr); + assert!(!server.wait().success()); + let log = server.log(); + assert!(log.contains(category), "missing category {category}: {log}"); + assert!( + !log.contains("SENTINEL"), + "startup diagnostic leaked fixture contents: {log}" + ); + assert!(!log.contains("INITIALIZED")); + TcpStream::connect_timeout(&addr, Duration::from_millis(50)) + .expect_err("failed startup never accepts connections"); + log + } + + #[test] + fn both_signals_stop_idle_processes() { + for signal in ["INT", "TERM"] { + let roots = Roots::new(); + let addr = address(); + let mut server = Server::spawn(command("bare", &roots, addr), addr); + server.ready(); + server.signal(signal); + assert!(server.wait().success(), "{}", server.log()); + } + } + + #[test] + fn sigterm_drains_dispatched_response_closes_idle_and_gates_pipeline() { + let roots = Roots::new(); + let addr = address(); + let mut server = Server::spawn(command("bare", &roots, addr), addr); + server.ready(); + let mut idle = server.socket().expect("idle socket"); + idle.write_all(b"GET /ready HTTP/1.1\r\nHost: fixture\r\n\r\n") + .expect("idle request"); + let mut idle_response = Vec::new(); + while !idle_response.ends_with(b"ready") { + let mut buffer = [0_u8; 1024]; + let read = idle.read(&mut buffer).expect("idle response"); + assert_ne!(read, 0); + idle_response.extend_from_slice(buffer.get(..read).expect("read length")); + } + let mut client = server.socket().expect("finite socket"); + client.write_all(b"GET /finite HTTP/1.1\r\nHost: fixture\r\n\r\nGET /after HTTP/1.1\r\nHost: fixture\r\nConnection: close\r\n\r\n") + .expect("pipeline"); + server.wait_line("DISPATCH"); + server.signal("TERM"); + server.wait_line("DRAINING"); + let mut response = String::new(); + client + .read_to_string(&mut response) + .expect("complete finite response"); + assert!(response.contains("finite-complete")); + assert!(!response.contains("forbidden")); + assert_eq!(idle.read(&mut [0_u8; 1]).expect("idle close"), 0); + assert!(server.wait().success(), "{}", server.log()); + assert!(!server.log().contains("FORBIDDEN_DISPATCH")); + } + + #[test] + fn process_budget_forces_pending_handler_stream_and_zero_read_peer() { + for (path, marker, dropped) in [ + ("/pending", "DISPATCH", "WORK_DROPPED"), + ("/stream", "STREAM_READY", "STREAM_DROPPED"), + ("/flood", "STREAM_READY", "STREAM_DROPPED"), + ] { + let roots = Roots::new(); + let addr = address(); + let mut cmd = command("bare", &roots, addr); + cmd.env("FIXTURE_FREEZE_CLOCK", "1"); + let mut server = Server::spawn(cmd, addr); + server.ready(); + let mut client = server.socket().expect("pending socket"); + write!(client, "GET {path} HTTP/1.1\r\nHost: fixture\r\n\r\n").expect("request"); + server.wait_line(marker); + let started = Instant::now(); + server.signal("TERM"); + assert!(!server.wait().success()); + assert!(started.elapsed() >= Duration::from_millis(700)); + assert!(started.elapsed() < Duration::from_secs(4)); + assert!(server.log().contains("grace exhausted"), "{}", server.log()); + assert_eq!(server.log().matches(dropped).count(), 1); + } + } + + #[test] + fn second_signal_forces_before_budget_exhaustion() { + let roots = Roots::new(); + let addr = address(); + let mut server = Server::spawn(command("bare", &roots, addr), addr); + server.ready(); + let mut client = server.socket().expect("pending socket"); + client + .write_all(b"GET /pending HTTP/1.1\r\nHost: fixture\r\n\r\n") + .expect("request"); + server.wait_line("DISPATCH"); + server.signal("TERM"); + server.wait_line("DRAINING"); + let started = Instant::now(); + server.signal("INT"); + assert!(!server.wait().success()); + assert!(started.elapsed() < Duration::from_millis(700)); + assert!(server.log().contains("forced stop")); + assert_eq!(server.log().matches("WORK_DROPPED").count(), 1); + } + + #[test] + fn stop_during_startup_never_binds_or_becomes_ready() { + let roots = Roots::new(); + let addr = address(); + let mut server = Server::spawn(command("starting", &roots, addr), addr); + server.wait_line("INITIALIZING"); + TcpStream::connect_timeout(&addr, Duration::from_millis(50)) + .expect_err("initialization precedes binding"); + server.signal("TERM"); + assert!(server.wait().success()); + assert!(server.log().contains("STARTUP_DROPPED")); + } + + #[test] + fn invalid_settings_and_callback_sources_are_redacted_nonzero_failures() { + let roots = Roots::new(); + for (key, value, category) in [ + ("EDGEZERO__ADAPTER__HOST", "HOST_SENTINEL", "HOST"), + ("EDGEZERO__ADAPTER__PORT", "PORT_SENTINEL", "PORT"), + ("EDGEZERO__ADAPTER__PORT", "0", "PORT"), + ("EDGEZERO__LOGGING__LEVEL", "LOGGING_SENTINEL", "LEVEL"), + ( + "EDGEZERO__ADAPTER__SHUTDOWN_GRACE_SECONDS", + "0", + "SHUTDOWN_GRACE_SECONDS", + ), + ( + "EDGEZERO__ADAPTER__CONFIG_DIR", + "ROOT_SENTINEL", + "CONFIG_DIR", + ), + ( + "EDGEZERO__STORES__CONFIG__SETTINGS__KEY", + "KEY_SENTINEL\n", + "settings", + ), + ] { + let addr = address(); + let mut cmd = command("initialized", &roots, addr); + cmd.env(key, value); + failed(cmd, addr, category); + } + for (mode, category) in [ + ("hook-error", "application configuration"), + ("initializer-error", "initializer"), + ] { + let addr = address(); + failed(command(mode, &roots, addr), addr, category); + } + let addr = address(); + let mut cmd = command("initialized", &roots, addr); + cmd.env_remove("EDGEZERO__ADAPTER__CONFIG_DIR"); + failed(cmd, addr, "CONFIG_DIR"); + let missing = roots.cwd.path().join("missing-data-SENTINEL"); + let missing_addr = address(); + let mut missing_cmd = command("initialized", &roots, missing_addr); + missing_cmd.env("EDGEZERO__ADAPTER__DATA_DIR", &missing); + failed(missing_cmd, missing_addr, "DATA_DIR"); + assert!(!missing.exists()); + } + + #[test] + fn occupied_bind_is_a_real_redacted_process_failure() { + let roots = Roots::new(); + let occupied = TcpListener::bind("127.0.0.1:0").expect("occupied port"); + let addr = occupied.local_addr().expect("address"); + let mut server = Server::spawn(command("bare", &roots, addr), addr); + assert!(!server.wait().success()); + assert!(server.log().contains("bind")); + assert!(!server.log().contains("SENTINEL")); + } + + #[test] + fn required_files_and_typed_initializer_fail_before_acceptance() { + let roots = Roots::new(); + let file = roots.config.path().join("local-config-settings.json"); + fs::remove_file(&file).expect("missing file fixture"); + let addr = address(); + failed(command("initialized", &roots, addr), addr, "missing"); + fs::create_dir_all(&file).expect("non-regular fixture"); + let directory_addr = address(); + failed( + command("initialized", &roots, directory_addr), + directory_addr, + "regular file", + ); + fs::remove_dir(&file).expect("remove fixture directory"); + roots.map(&json!({"settings": {"CONFIG_SENTINEL": true}})); + let malformed_addr = address(); + failed( + command("initialized", &roots, malformed_addr), + malformed_addr, + "malformed", + ); + for (map, category) in [ + (json!({}), "MissingBlob"), + (json!({"settings": "BLOB_SENTINEL"}), "MalformedEnvelope"), + ( + json!({"settings": blob("x", "fixture_token")}), + "Validation", + ), + ( + json!({"settings": blob("valid-greeting", "MISSING_SECRET_SENTINEL")}), + "MissingSecret", + ), + ] { + roots.map(&map); + let typed_addr = address(); + failed( + command("initialized", &roots, typed_addr), + typed_addr, + category, + ); + } + let mut envelope: Value = + serde_json::from_str(&blob("valid-greeting", "fixture_token")).expect("envelope"); + envelope["data"]["greeting"] = json!("CHANGED_CONFIG_SENTINEL"); + roots.map(&json!({"settings": serde_json::to_string(&envelope).expect("envelope")})); + let integrity_addr = address(); + failed( + command("initialized", &roots, integrity_addr), + integrity_addr, + "IntegrityMismatch", + ); + } + + #[test] + fn convenience_checks_framework_without_guessing_required_schema() { + let roots = Roots::new(); + roots.map(&json!({"settings": "INVALID_TYPED_SENTINEL"})); + let addr = address(); + let mut server = Server::spawn(command("framework", &roots, addr), addr); + server.ready(); + assert!( + !server + .request("/typed") + .expect("typed failure") + .starts_with("HTTP/1.1 200") + ); + server.signal("TERM"); + assert!(server.wait().success()); + assert!(!server.log().contains("INITIALIZED")); + } + + #[test] + fn retained_state_key_aliases_snapshot_restart_and_closed_backup_restore() { + let roots = Roots::new(); + roots.map(&json!({"settings": blob("wrong-key-greeting", "fixture_token"), "selected": blob("selected-greeting", "fixture_token")})); + let addr = address(); + let mut cmd = command("initialized", &roots, addr); + cmd.env("EDGEZERO__STORES__CONFIG__SETTINGS__KEY", "selected"); + let mut server = Server::spawn(cmd, addr); + server.ready(); + for _request in 0..3_u8 { + assert!( + server + .request("/state") + .expect("state") + .ends_with("selected-greeting") + ); + assert!( + server + .request("/typed") + .expect("typed") + .ends_with("selected-greeting") + ); + } + assert!( + server + .request("/write") + .expect("write") + .ends_with("written") + ); + assert!( + server + .request("/alias") + .expect("alias") + .ends_with("durable-value") + ); + assert!( + !server + .request("/distinct") + .expect("distinct") + .contains("durable-value") + ); + roots.map(&json!({"settings": blob("restart-greeting", "fixture_token")})); + assert!( + server + .request("/typed") + .expect("snapshot") + .ends_with("selected-greeting") + ); + let second_addr = address(); + failed( + command("initialized", &roots, second_addr), + second_addr, + "locked", + ); + assert_eq!( + fs::read_dir(roots.data.path()).expect("databases").count(), + 2 + ); + server.signal("TERM"); + assert!(server.wait().success()); + assert_eq!(server.log().matches("INITIALIZED").count(), 1); + let mut restarted = Server::spawn(command("initialized", &roots, addr), addr); + restarted.ready(); + assert!( + restarted + .request("/state") + .expect("restarted config") + .ends_with("restart-greeting") + ); + assert!( + restarted + .request("/read") + .expect("persisted") + .ends_with("durable-value") + ); + restarted.signal("TERM"); + assert!(restarted.wait().success()); + let restore = tempfile::tempdir().expect("restore root"); + for result in fs::read_dir(roots.data.path()).expect("closed databases") { + let entry = result.expect("database entry"); + fs::copy(entry.path(), restore.path().join(entry.file_name())) + .expect("stopped-writer backup"); + } + let mut restored_command = command("initialized", &roots, addr); + restored_command.env("EDGEZERO__ADAPTER__DATA_DIR", restore.path()); + let mut restored = Server::spawn(restored_command, addr); + restored.ready(); + assert!( + restored + .request("/read") + .expect("restored value") + .ends_with("durable-value") + ); + restored.signal("TERM"); + assert!(restored.wait().success()); + } + + #[test] + fn explicit_embedding_ignores_ambient_settings_and_uses_local_rc_stop() { + let roots = Roots::new(); + let addr = address(); + let mut cmd = command("embedding", &roots, addr); + cmd.env("EDGEZERO__ADAPTER__HOST", "IGNORED_HOST_SENTINEL") + .env("EDGEZERO__ADAPTER__PORT", "0") + .env("EDGEZERO__ADAPTER__CONFIG_DIR", "IGNORED_ROOT_SENTINEL") + .env( + "EDGEZERO__STORES__CONFIG__SETTINGS__KEY", + "IGNORED_KEY_SENTINEL", + ); + let mut server = Server::spawn(cmd, addr); + server.ready(); + assert!( + server + .request("/state") + .expect("state") + .ends_with("configured-greeting") + ); + assert!( + server + .request("/stop") + .expect("caller stop") + .ends_with("stopping") + ); + assert!(server.wait().success(), "{}", server.log()); + assert_eq!(server.log().matches("INITIALIZED").count(), 1); + } + + #[test] + fn embedding_does_not_install_process_signal_handlers() { + let roots = Roots::new(); + let addr = address(); + let mut server = Server::spawn(command("embedding-bare", &roots, addr), addr); + server.ready(); + server.signal("TERM"); + assert_eq!(server.wait().signal(), Some(15_i32)); + } + + #[test] + fn projected_readonly_config_and_writable_data_preserve_uid_and_permissions() { + let roots = Roots::new(); + let file = roots.config.path().join("local-config-settings.json"); + let target = roots.config.path().join("projected-value"); + fs::rename(&file, &target).expect("projected target"); + symlink(&target, &file).expect("projected symlink"); + fs::set_permissions(&target, fs::Permissions::from_mode(0o444)).expect("read-only config"); + fs::set_permissions(roots.config.path(), fs::Permissions::from_mode(0o555)) + .expect("read-only root"); + let config_before = fs::metadata(roots.config.path()).expect("config metadata"); + let data_before = fs::metadata(roots.data.path()).expect("data metadata"); + let addr = address(); + let mut server = Server::spawn(command("initialized", &roots, addr), addr); + server.ready(); + assert!( + server + .request("/write") + .expect("write") + .ends_with("written") + ); + server.signal("TERM"); + assert!(server.wait().success()); + let config_after = fs::metadata(roots.config.path()).expect("config metadata"); + let data_after = fs::metadata(roots.data.path()).expect("data metadata"); + assert_eq!( + (config_before.uid(), config_before.mode()), + (config_after.uid(), config_after.mode()) + ); + assert_eq!( + (data_before.uid(), data_before.mode()), + (data_after.uid(), data_after.mode()) + ); + fs::set_permissions(roots.config.path(), fs::Permissions::from_mode(0o700)) + .expect("fixture cleanup"); + } +} diff --git a/crates/edgezero-core/src/app.rs b/crates/edgezero-core/src/app.rs index 3adea0f37..de77c2cc1 100644 --- a/crates/edgezero-core/src/app.rs +++ b/crates/edgezero-core/src/app.rs @@ -283,6 +283,19 @@ impl App { self.ingress_head_limits } + /// Registers typed handler state during startup, before ingress admission or serving. + /// + /// Same-type insertion replaces previous state for this App only. Shared routers + /// fork on write, preserving routes and peer Apps. Previously resolved or admitted + /// dispatch tokens are invalid for this App after insertion. This is not live reload. + #[inline] + pub fn insert_state(&mut self, value: T) + where + T: Clone + Send + Sync + 'static, + { + self.router.insert_state(value); + } + /// Consume the app and return the contained router service. #[must_use] #[inline] @@ -2434,6 +2447,60 @@ mod tests { assert_eq!(complete_envelope(response).status(), StatusCode::OK); } + #[test] + fn startup_state_forks_router_and_invalidates_old_admission() { + use crate::extractor::State; + + #[crate::action] + async fn state(State(value): State>) -> Result { + Ok(value.to_string()) + } + + let router = RouterService::builder() + .get("/state/{id}", state) + .with_state(Arc::::from("builder")) + .build(); + let peer = App::new(router.clone()); + let mut app = App::new(router); + let head = || { + IngressHeadParts::new( + Method::GET, + "/state/42".parse().expect("URI"), + Version::HTTP_11, + HeaderMap::new(), + ) + }; + let IngressBeginOutcome::Admitted(old) = app + .begin_ingress(head(), MonotonicInstant::now()) + .expect("admission") + else { + panic!("expected admission") + }; + let request = || { + request_builder() + .uri("/state/42") + .body(Body::empty()) + .expect("request") + }; + app.insert_state(Arc::::from("first")); + app.insert_state(Arc::::from("startup")); + assert_eq!(app.router().routes(), peer.router().routes()); + let rejected = complete_envelope(block_on(app.dispatch_admitted(old, request()))); + assert_eq!(rejected.status(), StatusCode::INTERNAL_SERVER_ERROR); + for _ in 0..2_u8 { + let IngressBeginOutcome::Admitted(new) = app + .begin_ingress(head(), MonotonicInstant::now()) + .expect("admission") + else { + panic!("expected admission") + }; + let response = complete_envelope(block_on(app.dispatch_admitted(new, request()))); + assert_eq!(response.body().as_bytes().expect("body"), b"startup"); + } + let response = block_on(peer.router().oneshot(request())).expect("peer response"); + assert_eq!(response.body().as_bytes().expect("body"), b"builder"); + } + #[test] fn default_hooks_do_not_own_logging() { assert!(!DefaultHooks::owns_logging()); diff --git a/crates/edgezero-core/src/extractor.rs b/crates/edgezero-core/src/extractor.rs index eaf50aa70..3971e1455 100644 --- a/crates/edgezero-core/src/extractor.rs +++ b/crates/edgezero-core/src/extractor.rs @@ -844,10 +844,15 @@ where None, ) })?; - let key = binding.default_key.clone(); - extract_from_handle::(ctx, &binding.handle, &key) - .await - .map(AppConfig) + Self::from_binding( + binding, + None, + ctx.extensions().get::(), + ctx.config_extraction_limits(), + ctx.monotonic_clock(), + ) + .await + .map(AppConfig) } } @@ -855,6 +860,33 @@ impl AppConfig where C: DeserializeOwned + AppConfigMeta + Validate + Send + 'static, { + /// Loads one application-selected binding without a request or route dispatch. + /// + /// Uses the same limits, envelope checks, secret resolution, validation and + /// error categories as request extraction. `None` selects the binding's default + /// key. This does not cache config or infer required application schemas. + /// + /// # Errors + /// Returns a typed store-extraction error for backend, budget, envelope, + /// secret or schema failures. Native startup must classify errors before logging. + #[inline] + pub async fn from_binding( + binding: &ConfigStoreBinding, + key: Option<&str>, + secrets: Option<&SecretRegistry>, + limits: ConfigExtractionLimits, + clock: MonotonicClock, + ) -> Result { + extract_from_handle::( + &binding.handle, + key.unwrap_or(&binding.default_key), + secrets, + limits, + clock, + ) + .await + } + /// Read the typed config from a NON-default config store. /// `key = None` falls back to that store's `binding.default_key`. /// Returns the inner `C` directly per spec 6.2.1. @@ -874,8 +906,14 @@ where None, ) })?; - let resolved_key = key.unwrap_or(&binding.default_key).to_owned(); - extract_from_handle::(ctx, &binding.handle, &resolved_key).await + Self::from_binding( + binding, + key, + ctx.extensions().get::(), + ctx.config_extraction_limits(), + ctx.monotonic_clock(), + ) + .await } /// Read the typed config from the default store under an @@ -896,7 +934,14 @@ where None, ) })?; - extract_from_handle::(ctx, &binding.handle, key).await + Self::from_binding( + binding, + Some(key), + ctx.extensions().get::(), + ctx.config_extraction_limits(), + ctx.monotonic_clock(), + ) + .await } } @@ -1201,15 +1246,16 @@ fn future_format_reason(raw: &str) -> Option { /// Returns [`EdgeError::StoreExtraction`] with the stable reason corresponding /// to store, envelope, secret, or schema failure. async fn extract_from_handle( - ctx: &RequestContext, handle: &ConfigStoreHandle, key: &str, + secrets: Option<&SecretRegistry>, + limits: ConfigExtractionLimits, + clock: MonotonicClock, ) -> Result where C: DeserializeOwned + AppConfigMeta + Validate + Send + 'static, { - let mut budget = - ConfigExtractionBudget::start(ctx.config_extraction_limits(), ctx.monotonic_clock())?; + let mut budget = ConfigExtractionBudget::start(limits, clock)?; let read = handle .get_bounded( key, @@ -1273,7 +1319,7 @@ where budget.check_deadline()?; let mut data = parsed?; - let resolved = secret_walk::(ctx, &mut budget, &mut data).await; + let resolved = secret_walk::(secrets, &mut budget, &mut data).await; budget.check_deadline()?; resolved?; @@ -1334,7 +1380,7 @@ fn map_config_store_error(err: &ConfigStoreError) -> EdgeError { /// /// `StoreRef` fields are skipped — their value is a store id, not a key. async fn secret_walk( - ctx: &RequestContext, + secrets: Option<&SecretRegistry>, budget: &mut ConfigExtractionBudget, data: &mut serde_json::Value, ) -> Result<(), EdgeError> @@ -1348,7 +1394,7 @@ where if matches!(field.kind, SecretKind::StoreRef) { continue; } - resolve_secret_field(ctx, budget, data, &field, &field.path, String::new()).await?; + resolve_secret_field(secrets, budget, data, &field, &field.path, String::new()).await?; } Ok(()) } @@ -1357,7 +1403,7 @@ where /// secret leaf(s). `rendered` is the dotted path so far (with concrete `[n]` /// indices) for error hints. fn resolve_secret_field<'walk>( - ctx: &'walk RequestContext, + secrets: Option<&'walk SecretRegistry>, budget: &'walk mut ConfigExtractionBudget, node: &'walk mut serde_json::Value, field: &'walk SecretField, @@ -1368,7 +1414,7 @@ fn resolve_secret_field<'walk>( match remaining.split_first() { // Leaf reached: `node` is the PARENT object; the last field is the key. Some((SecretPathSegment::Field(name), [])) => { - resolve_leaf(ctx, budget, node, field, name.as_ref(), &rendered).await + resolve_leaf(secrets, budget, node, field, name.as_ref(), &rendered).await } Some((SecretPathSegment::OptionalField(name), [])) => { if node.as_object().is_some_and(|parent| { @@ -1379,7 +1425,7 @@ fn resolve_secret_field<'walk>( }) { return Ok(()); } - resolve_leaf(ctx, budget, node, field, name.as_ref(), &rendered).await + resolve_leaf(secrets, budget, node, field, name.as_ref(), &rendered).await } // Required intermediates still reject stale blobs. Optional // intermediates are represented explicitly below rather than by the @@ -1393,7 +1439,8 @@ fn resolve_secret_field<'walk>( Some(next_rendered), )), Some(child) => { - resolve_secret_field(ctx, budget, child, field, rest, next_rendered).await + resolve_secret_field(secrets, budget, child, field, rest, next_rendered) + .await } } } @@ -1409,7 +1456,8 @@ fn resolve_secret_field<'walk>( match parent.get_mut(name.as_ref()) { None | Some(serde_json::Value::Null) => Ok(()), Some(child) => { - resolve_secret_field(ctx, budget, child, field, rest, next_rendered).await + resolve_secret_field(secrets, budget, child, field, rest, next_rendered) + .await } } } @@ -1425,7 +1473,7 @@ fn resolve_secret_field<'walk>( }; for (idx, item) in items.iter_mut().enumerate() { let indexed = format!("{rendered}[{idx}]"); - resolve_secret_field(ctx, budget, item, field, rest, indexed).await?; + resolve_secret_field(secrets, budget, item, field, rest, indexed).await?; } Ok(()) } @@ -1446,7 +1494,7 @@ fn join_field(prefix: &str, name: &str) -> String { /// secret field name; `store_ref_field` (for `KeyInNamedStore`) is a sibling /// within `parent`. async fn resolve_leaf( - ctx: &RequestContext, + secrets: Option<&SecretRegistry>, budget: &mut ConfigExtractionBudget, parent: &mut serde_json::Value, field: &SecretField, @@ -1485,7 +1533,7 @@ async fn resolve_leaf( let (bound, resolved_store_id) = match field.kind { SecretKind::KeyInDefault => { - let bound = ctx.secret_store_default().ok_or_else(|| { + let bound = secrets.and_then(SecretRegistry::default).ok_or_else(|| { store_extraction_error( StoreExtractionReason::MissingRegistry, format!( @@ -1513,19 +1561,21 @@ async fn resolve_leaf( ) })? .to_owned(); - let bound = ctx.secret_store(&store_id_str).ok_or_else(|| { - // `store_id_str` is the blob's store_ref VALUE — config data that - // may be sensitive — and this message reaches the HTTP body. Name - // the field, not the stored id. - store_extraction_error( - StoreExtractionReason::UnknownStore, - format!( - "secret field `{leaf_path}` names a store_ref that is not declared in \ + let bound = secrets + .and_then(|registry| registry.named(&store_id_str)) + .ok_or_else(|| { + // `store_id_str` is the blob's store_ref VALUE — config data that + // may be sensitive — and this message reaches the HTTP body. Name + // the field, not the stored id. + store_extraction_error( + StoreExtractionReason::UnknownStore, + format!( + "secret field `{leaf_path}` names a store_ref that is not declared in \ [stores.secrets] (id redacted)" - ), - Some(leaf_path.clone()), - ) - })?; + ), + Some(leaf_path.clone()), + ) + })?; (bound, store_id_str) } }; @@ -2772,7 +2822,7 @@ mod tests { let mut budget = ConfigExtractionBudget::start(ConfigExtractionLimits::default(), ctx.monotonic_clock()) .expect("test budget"); - secret_walk::(ctx, &mut budget, data).await + secret_walk::(ctx.extensions().get::(), &mut budget, data).await } #[test] diff --git a/crates/edgezero-core/src/lib.rs b/crates/edgezero-core/src/lib.rs index cbd7dfc54..7584da69d 100644 --- a/crates/edgezero-core/src/lib.rs +++ b/crates/edgezero-core/src/lib.rs @@ -35,6 +35,7 @@ pub mod middleware; pub mod outbound; pub mod params; pub mod platform; +pub mod probe; pub mod responder; pub mod response; pub mod response_egress; diff --git a/crates/edgezero-core/src/probe.rs b/crates/edgezero-core/src/probe.rs new file mode 100644 index 000000000..7f8ec783b --- /dev/null +++ b/crates/edgezero-core/src/probe.rs @@ -0,0 +1,163 @@ +//! App-owned probe routes with read-only live lifecycle observation. +//! +//! Adapters may publish a reader in request extensions. Absence does not imply +//! health or lifecycle support. Applications choose paths and middleware; these +//! helpers register no routes. Do not override framework probe/store extensions +//! through application state or middleware. Native dispatch uses a private gate. + +use std::fmt; +use std::sync::Arc; + +use crate::action; +use crate::body::Body; +use crate::context::RequestContext; +use crate::error::EdgeError; +use crate::http::{Response, StatusCode, response_builder}; + +/// Observed native runner phase. Once draining starts, a run never becomes ready again. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum LifecyclePhase { + Draining, + Ready, + Starting, + Stopped, +} + +/// Cloneable read-only observation, sampled when a handler executes. +/// +/// Readiness means the selected entrypoint's startup contract succeeded, not that +/// undisclosed application requirements or ongoing dependencies were checked. +#[derive(Clone)] +pub struct LifecycleReader { + read: Arc Option + Send + Sync>, +} + +impl LifecycleReader { + #[inline] + #[must_use] + pub fn new(read: Read) -> Self + where + Read: Fn() -> Option + Send + Sync + 'static, + { + Self { + read: Arc::new(read), + } + } + + #[inline] + #[must_use] + pub fn phase(&self) -> Option { + (self.read)() + } +} + +impl fmt::Debug for LifecycleReader { + #[inline] + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LifecycleReader").finish_non_exhaustive() + } +} + +/// Returns 200 while starting, ready or draining, otherwise 503. +/// +/// This table applies only if the handler executes. Before acceptance and after +/// listener withdrawal, supervisors must also handle refusal, reset or EOF. +/// +/// # Errors +/// Returns an internal error only if the static response cannot be constructed. +#[action] +pub async fn liveness(ctx: RequestContext) -> Result { + match phase(&ctx) { + Some(LifecyclePhase::Starting | LifecyclePhase::Ready | LifecyclePhase::Draining) => { + response(StatusCode::OK, "live") + } + Some(LifecyclePhase::Stopped) => response(StatusCode::SERVICE_UNAVAILABLE, "stopped"), + None => response(StatusCode::SERVICE_UNAVAILABLE, "probe state unavailable"), + } +} + +fn phase(ctx: &RequestContext) -> Option { + ctx.extensions() + .get::() + .and_then(LifecycleReader::phase) +} + +/// Returns 200 only in Ready, otherwise 503. Does not poll dependencies. +/// +/// # Errors +/// Returns an internal error only if the static response cannot be constructed. +#[action] +pub async fn readiness(ctx: RequestContext) -> Result { + match phase(&ctx) { + Some(LifecyclePhase::Ready) => response(StatusCode::OK, "ready"), + Some(_) => response(StatusCode::SERVICE_UNAVAILABLE, "not ready"), + None => response(StatusCode::SERVICE_UNAVAILABLE, "probe state unavailable"), + } +} + +fn response(status: StatusCode, text: &'static str) -> Result { + response_builder() + .status(status) + .header("content-type", "text/plain; charset=utf-8") + .header("cache-control", "no-store") + .body(Body::from(text)) + .map_err(EdgeError::internal) +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use futures::executor::block_on; + + use super::*; + use crate::http::request_builder; + use crate::router::RouterService; + + #[test] + fn ordinary_routes_observe_live_phase_and_missing_state() { + #[action] + async fn health() -> Result<&'static str, EdgeError> { + Ok("app health") + } + let current = Arc::new(Mutex::new(None)); + let observed = Arc::clone(¤t); + let reader = LifecycleReader::new(move || *observed.lock().expect("phase")); + let router = RouterService::builder() + .get("/app-ready", readiness) + .get("/app-live", liveness) + .get("/health", health) + .build(); + for (value, ready, live) in [ + (None, 503, 503), + (Some(LifecyclePhase::Starting), 503, 200), + (Some(LifecyclePhase::Ready), 200, 200), + (Some(LifecyclePhase::Draining), 503, 200), + (Some(LifecyclePhase::Stopped), 503, 503), + ] { + *current.lock().expect("phase") = value; + for (path, expected) in [("/app-ready", ready), ("/app-live", live)] { + let mut request = request_builder() + .uri(path) + .body(Body::empty()) + .expect("request"); + request.extensions_mut().insert(reader.clone()); + let result = block_on(router.oneshot(request)).expect("probe"); + assert_eq!(result.status().as_u16(), expected); + assert_eq!(result.headers()["cache-control"], "no-store"); + assert_eq!( + result.headers()["content-type"], + "text/plain; charset=utf-8" + ); + } + } + for (path, expected) in [("/app-ready", 503), ("/health", 200), ("/ready", 404)] { + let request = request_builder() + .uri(path) + .body(Body::empty()) + .expect("request"); + let result = block_on(router.oneshot(request)).expect("response"); + assert_eq!(result.status().as_u16(), expected); + } + } +} diff --git a/crates/edgezero-core/src/router.rs b/crates/edgezero-core/src/router.rs index b690f95de..1914e5684 100644 --- a/crates/edgezero-core/src/router.rs +++ b/crates/edgezero-core/src/router.rs @@ -345,6 +345,7 @@ impl RouterBuilder { } } +#[derive(Clone)] struct RouterInner { manifest_json: Option>, middlewares: Vec, @@ -528,6 +529,15 @@ impl RouterService { } } + pub(crate) fn insert_state(&mut self, value: T) + where + T: Clone + Send + Sync + 'static, + { + Arc::make_mut(&mut self.inner) + .state_extensions + .insert(value); + } + fn new( routes: HashMap>, middlewares: Vec, diff --git a/crates/edgezero-core/tests/fixtures/production_handlers.rs b/crates/edgezero-core/tests/fixtures/production_handlers.rs new file mode 100644 index 000000000..5a3a1e323 --- /dev/null +++ b/crates/edgezero-core/tests/fixtures/production_handlers.rs @@ -0,0 +1,32 @@ +//! Shared native/WASM hosting fixture. No native runtime dependency belongs here. + +use std::sync::Arc; + +use edgezero_core::EdgeError; +use edgezero_core::action; +use edgezero_core::extractor::{AppConfig, State}; +use serde::Deserialize; +use validator::Validate; + +#[derive(Deserialize, Validate, edgezero_core::AppConfig)] +pub struct FixtureConfig { + #[validate(length(min = 4_u64))] + pub greeting: String, + #[secret] + #[validate(length(min = 8_u64))] + pub token: String, +} + +pub struct HostState { + pub greeting: String, +} + +#[action] +pub async fn state(State(state): State>) -> Result { + Ok(state.greeting.clone()) +} + +#[action] +pub async fn typed(AppConfig(config): AppConfig) -> Result { + Ok(config.greeting) +} diff --git a/crates/edgezero-core/tests/production_handlers.rs b/crates/edgezero-core/tests/production_handlers.rs new file mode 100644 index 000000000..afa95fe74 --- /dev/null +++ b/crates/edgezero-core/tests/production_handlers.rs @@ -0,0 +1,41 @@ +//! Compile this same handler fixture for native and every supported WASM target. + +#[path = "fixtures/production_handlers.rs"] +mod fixture; + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use edgezero_core::Body; + use edgezero_core::http::{StatusCode, request_builder}; + use edgezero_core::probe; + use edgezero_core::router::RouterService; + use futures::executor::block_on; + + use super::fixture; + + #[test] + fn shared_state_config_and_probe_actions_use_portable_routes() { + let router = RouterService::builder() + .get("/state", fixture::state) + .get("/typed", fixture::typed) + .get("/ready", probe::readiness) + .with_state(Arc::new(fixture::HostState { + greeting: "retained".to_owned(), + })) + .build(); + for (path, status) in [ + ("/state", StatusCode::OK), + ("/ready", StatusCode::SERVICE_UNAVAILABLE), + ("/typed", StatusCode::INTERNAL_SERVER_ERROR), + ] { + let request = request_builder() + .uri(path) + .body(Body::empty()) + .expect("request"); + let response = block_on(router.oneshot(request)).expect("response"); + assert_eq!(response.status(), status); + } + } +} diff --git a/docs/superpowers/specs/2026-10-05-axum-production-startup-lifecycle-design.md b/docs/superpowers/specs/2026-10-05-axum-production-startup-lifecycle-design.md new file mode 100644 index 000000000..77f43ee48 --- /dev/null +++ b/docs/superpowers/specs/2026-10-05-axum-production-startup-lifecycle-design.md @@ -0,0 +1,427 @@ +# Axum startup, storage, and bounded shutdown + +Date: 2026-10-05 +Status: initial implementation for review. The worktree implements the combined lifecycle and production-startup APIs below. Deployment guidance and the remaining acceptance checks are still open. + +Issues: [#392 lifecycle](https://github.com/stackpop/edgezero/issues/392) and [#396 configuration/storage](https://github.com/stackpop/edgezero/issues/396), under [Epic #391](https://github.com/stackpop/edgezero/issues/391). + +Base: [PR #275](https://github.com/stackpop/edgezero/pull/275), inspected at `5527b58e2a7cb6307d5c7a8926cd5dfc6f4f8b4a`. + +Worktree: `/home/pav/worktrees/issue-392-396-production-hosting-spec`. + +## Goal + +A production Axum application validates its runtime settings and initializes required stores before accepting traffic. Applications with additional configuration/bootstrap requirements validate them through the initializer variant before readiness. Its handlers can reuse startup-loaded state. On shutdown, readiness is revoked, new work stops, and dispatched work gets one finite drain budget. + +Preserve the existing permissive development entrypoint and local defaults. Strict production startup is an explicit API choice over the same runtime owner, service, and connection implementation. Do not infer production policy from release builds, containers, or the presence of path settings. + +Keep #275's fallible app assembly, owned HTTP/1 connections, local non-Send execution, and response-egress semantics. This work extends that runner rather than introducing another HTTP server or portable lifecycle framework. + +## Scope and ownership + +| Owner | Responsibility | +| ----------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| #392 | Native runtime ownership, signals/caller shutdown, lifecycle observation, dispatch gating, tracked draining, exit outcomes, and portable probe helpers | +| #396 | Explicit strict production settings/store preparation, development compatibility, app-owned initializer, shared typed-config loading, startup-only typed state, persistence/permissions/backup evidence, and production adoption | +| Application | Required schemas, config keys, extra secrets, signing/bootstrap policy, registered handler state, probe paths, and optional ongoing dependency checks | +| #395 | Residual outbound transport/wire acceptance; coordinate fallible transport startup here without building another client | +| #398/#399 | Container packaging and native architecture validation; consume this runtime contract for later integration checks | +| #397/#400 | Wider diagnostics/trusted-proxy work and deployment recipes, respectively | + +No live reload, filesystem watcher, online backup service, new backup CLI, distributed storage, automatic migrations, retry framework, tracing vendor, health configuration section, management listener, or public EdgeZero application image is included. No blanket Send bounds are added to portable futures or bodies. + +Trusted Server supplies concrete consumer requirements, but this spec does not implement its migration or certify its application image. + +## Current implementation and evidence + +All findings below are source inspection at the base above, not executed proof. Paths are repository-relative unless noted. + +| Area | Current behavior and source | +| ------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| App assembly | `crates/edgezero-core/src/app.rs`, `App::build` and `Hooks::configure`: install platform metadata and configure once. Configure receives no prepared native stores. | +| Standard entrypoint | `crates/edgezero-adapter-axum/src/dev_server.rs`, `run_app`: build app, resolve settings, bind, initialize stores, then serve. Generated main calls its no-argument form. | +| Local serving | The same file, `serve_local`: current-thread runtime/LocalSet, detached local connection tasks, Ctrl-C or internal oneshot exit. No SIGTERM or tracked graceful-drain phase. | +| Connections | `crates/edgezero-adapter-axum/src/connection.rs`, `serve_http1`: owned Hyper HTTP/1 keep-alive and response deadline enforcement. Extend this owner for draining. | +| Config files | `crates/edgezero-adapter-axum/src/config_store.rs`, `from_path` and `local_path`: flat string-map snapshots, missing-file empty stores, ancestor/cwd discovery. The runner drops malformed bindings with warnings. | +| KV identity | `dev_server.rs`, `kv_store_path` and `build_kv_registry`: name-derived slug/hash basenames under `.edgezero`. Required declared KV failures are already fatal. Each id currently attempts its own open. | +| Documented aliases | `docs/guide/kv.md:122`: ids with the same resolved platform name share a file. Per-id opens must be reconciled with this documented behavior. | +| Typed config | `crates/edgezero-core/src/extractor.rs`, `AppConfig` and `extract_from_handle`: bounded reads, envelope/integrity checks, secret resolution, schema validation, and redaction. This algorithm currently takes RequestContext. | +| Handler state | `crates/edgezero-core/src/router.rs`, `RouterBuilder::with_state`: typed state copied into requests, same-type last-write-wins. `App` currently has no post-build state setter. | +| Outbound startup | `crates/edgezero-adapter-axum/src/service.rs`, `AxumServiceState::from_app`: `try_transport().ok()` hides initialization failure; admitted requests subsequently fail. | +| Completion | `response.rs` and `crates/edgezero-core/src/response_egress.rs`: HostHandoff is not client receipt. Pre-egress cancellation can abandon callbacks while releasing captured resources. | +| Existing proof gap | `key_value_store.rs`, `data_persists_across_reopens`, tests reopen in one process, not subprocess restart, competing-writer rejection, or backup/restore. Existing server tests do not prove signal-driven draining. | + +Trusted Server at `666953a0de0dbec514b0f0bd5a8b03074f13e62b` was also inspected. Its Fastly adapter loads settings into Arc and captures that state in handlers, but its current entrypoint builds the app for each request. Native startup should retain loaded state for the native run, not assume Fastly cross-request retention. + +## Agreed runtime settings + +Use the existing `EDGEZERO__ADAPTER__*` convention. The table defines strict production options, not replacement defaults for the development runner. + +| Setting | Contract | +| ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `HOST` | Omitted defaults to `127.0.0.1`. Supplied invalid values fail, rather than warn/fallback. Preserve current IP-literal support. | +| `PORT` | Omitted defaults to `8787`. Supplied malformed/out-of-range/zero values fail. Containers supply explicit binding values. | +| `CONFIG_DIR` | Explicit absolute, existing directory when local config stores are declared in production. | +| `DATA_DIR` | Explicit absolute, existing writable directory when embedded KV stores are declared in production. | +| `SHUTDOWN_GRACE_SECONDS` | Omitted defaults to 10. Supplied values are positive integer seconds; invalid, zero, overflow, or unusable timer values fail startup. No silent clamp/fallback. | + +Directory values are not required for undeclared store kinds. Validate supplied recognized settings instead of using invalid input as omission. Preserve native defaults for genuinely omitted optional settings. Do not add a global scanner for unrelated provider variables. + +The production standalone entrypoint reads environment settings once into native options. Embedding accepts explicit options and does not silently merge ambient hosting settings. An embedding caller can deliberately use the environment-loading constructor. Store-name/key overrides belong to that selected configuration snapshot too; this does not remove the environment-backed secret backend. + +Strictness includes recognized supplied native settings that previously fell back, such as logging level or declared-store name/key overrides. Keep shared EnvConfig fallback semantics unchanged for other adapters; perform strict native validation before using normalized values. Preserve `Hooks::owns_logging` and avoid installing a competing logger. + +There is no new CLI flag or runtime configuration file for these settings. Preserve development bind warnings/fallbacks, local config discovery, missing-file empty stores, malformed-binding warnings/omission, and cwd-relative KV defaults. Declared KV initialization failure remains fatal. Explicit production options never use those fallbacks. The new shutdown duration has no legacy permissive behavior to preserve; a supplied invalid duration fails in either entrypoint. + +The Axum CLI's existing development serve path keeps its permissive behavior. A wrapper launching a production entrypoint must pass supplied hosting settings through without replacing invalid input with defaults. Keep existing development wrapper normalization out of that production path; do not add a second CLI configuration policy. + +## Startup and initialization + +Proposed production ordering for the agreed invariant. Development uses the same lifecycle owner with its existing store/default policy and no additional initializer requirement: + +```mermaid +flowchart TD + A[Parse and validate native settings] --> B[Create owned local runtime and stop observation] + B --> C[App::build installs platform and configures once] + C --> D[Prepare required registries and shared transport] + D --> E[Await app-owned initializer] + E --> F[Register validated typed handler state] + F --> G[Bind and adopt listener] + G --> H[Ready and accepting] + C -->|Failure| X[Release resources and return non-success] + D -->|Failure| X + E -->|Failure| X + G -->|Failure| X +``` + +No acceptance or Ready transition can precede successful initialization. Binding after validation avoids a misleading open port during production standalone startup. An existing embedding path with a prebound listener can preserve that listener, but still cannot accept or become Ready early. + +Register standalone signals early enough to observe stop during initialization. Embedding supplies its stop source without installing process-wide handlers. A stop before Ready cancels cooperative initialization, releases prepared resources, and never enters Ready. Completed intentional stop exits zero. Initialization failure remains non-success. Cleanup uses the same shutdown budget; expiry or a second signal is forced stop. + +Run startup and connection serving on the same owned current-thread runtime/LocalSet. The initializer and its future need not be Send. Do not capture them in the existing spawn_blocking bridge. This design is a blocking native runner; it does not add another public async hosting framework. + +A supplied production initializer runs exactly once, after `App::build`/`Hooks::configure` and strict store preparation, before App is shared with connections. It receives mutable App and borrowed access to the same registry handles/bindings that requests will receive. It cannot turn failed declarations into absent bindings or replace prepared registries. Borrowed registry access does not make the underlying KV backend read-only; any app-owned bootstrap writes remain application policy. + +The application explicitly selects every required typed blob/schema and additional required secret or startup invariant. Store metadata contains ids/defaults, not schemas or required-key lists. A successful callback is evidence that the app's selected checks passed, not proof that the framework discovered every application requirement. + +Apps without application-specific startup requirements can use a production convenience entrypoint without an initializer argument. It still validates framework settings, declared stores, transport, and fallible app assembly before Ready. An explicit initializer variant adds app-selected blob/schema, secret, and bootstrap checks. A convenience entrypoint must not claim those application-specific checks occurred. + +Preserve the existing no-argument development entrypoint. Newly generated no-store apps remain runnable without a source migration or explicit no-op callback; a generated config Rust type alone does not declare a mandatory stored blob. Ready means the selected entrypoint's startup contract completed, not that the framework discovered every application requirement. Production applications with additional requirements must select the initializer variant. + +### Proposed native API shape + +These sketches describe the boundary, not compiled APIs or final Rust lifetime ergonomics. Public names are proposed. Declarations are API outline notation, not implementation code. + +```text +pub struct AxumRunOptions { + // Validated bind, roots, shutdown duration and selected store overrides. +} + +impl AxumRunOptions { + pub fn from_env() -> anyhow::Result; + // Explicit construction validates supplied native settings as well. +} + +pub struct PreparedStores { + // Private registry fields, None only for an undeclared/unattached kind. +} + +impl PreparedStores { + pub fn config(&self) -> Option<&ConfigRegistry>; + pub fn kv(&self) -> Option<&KvRegistry>; + pub fn secrets(&self) -> Option<&SecretRegistry>; +} + +pub type LocalStartupFuture<'a> = + Pin> + 'a>>; + +// Existing development convenience API and permissive local defaults. +pub fn run_app() -> anyhow::Result<()>; + +// Production convenience: strict environment options and SIGINT/Unix SIGTERM. +// Completes framework startup checks without claiming app-specific validation. +pub fn run_production_app() -> anyhow::Result<()>; + +// Production standalone with additional app-owned initialization. +pub fn run_production_app_with_initializer( + initialize: Initialize, +) -> anyhow::Result<()> +where + A: Hooks, + Initialize: for<'a> FnOnce(&'a mut App, &'a PreparedStores) + -> LocalStartupFuture<'a>; + +// Production embedding convenience: explicit options and caller-owned stop. +// Neither embedding variant installs global signal handlers. +pub fn run_app_with_options( + options: AxumRunOptions, + stop: Stop, +) -> anyhow::Result<()> +where + A: Hooks, + Stop: Future; + +// Production embedding with additional app-owned initialization. +pub fn run_app_with_options_and_initializer( + options: AxumRunOptions, + stop: Stop, + initialize: Initialize, +) -> anyhow::Result<()> +where + A: Hooks, + Stop: Future, + Initialize: for<'a> FnOnce(&'a mut App, &'a PreparedStores) + -> LocalStartupFuture<'a>; +``` + +Caller-owned stop can use an existing oneshot/watch/notification future. No new portable shutdown-token framework is needed. The native owner handles stop idempotently; repeat caller requests do not shorten the budget. The standalone signal source separately observes the second signal for forced teardown. + +These entrypoints are thin policy/initializer selections over one native runtime owner, not separate servers. Convenience variants supply an internal successful initializer; callers need not write a no-op. Retain the existing module location and existing development signatures unless an accepted prerequisite changes them. + +Router-only AxumDevServer callers retain their configuration and handle-attachment conveniences over that same owner; do not require an unrelated initializer migration. Normalize bare handles into the existing registry representation before any selected initialization. Tests may keep a private prebound-listener entrypoint through that same owner. + +## Shared typed-config loading and state retention + +Expose a context-free entrypoint to the existing typed loader, proposed as: + +```text +AppConfig::from_binding( + binding: &ConfigStoreBinding, + key: Option<&str>, + secrets: Option<&SecretRegistry>, + limits: ConfigExtractionLimits, + clock: MonotonicClock, +) -> Result; +``` + +This is signature notation, not literal Rust syntax. Implement it as an async inherent method with the existing `C: DeserializeOwned + AppConfigMeta + Validate + Send + 'static` bounds. A missing explicit key uses the binding's resolved default key. The app chooses default or named bindings; do not apply one guessed type to all declared stores. + +Factor only the RequestContext dependencies out of `extract_from_handle` and secret traversal. Request extraction and startup then share one parser, budget, envelope/version/hash checks, secret walk, optional-secret behavior, validation, and redacted error mapping. No fake request, route dispatch, second parser, operator TOML loader, or implicit cache is involved. + +Existing extraction limits apply per selected blob and its referenced secrets. They are not a deadline for the entire initializer or arbitrary signing/bootstrap work. They also do not bound the initial allocation of the outer local-file map. Do not add a separate startup-timeout setting or new numerical file-size policy in this slice. Operator-supplied files are trusted deployment inputs, and the supervisor owns the ultimate startup/stop bound. + +Add an explicit post-build method, proposed as: + +```text +App::insert_state(&mut self, value: T) +where + T: Clone + Send + Sync + 'static; +``` + +It uses the existing router state-extension mechanism. Same-type insertion replaces earlier state, including builder-registered state, for this App. Typical usage registers Arc and handlers extract State>. + +A router shared with another App must not be mutated globally. Use copy-on-write/forked router internals for state insertion; clone existing handler/middleware references and retain route metadata and matching behavior. Do not rebuild the app/routes, require Arc uniqueness, introduce a second state overlay with competing precedence, or use a global/deferred cell. Prove shared-router isolation in a test. + +State insertion is a startup operation and must finish before ingress admission or serving begins. Existing resolved-dispatch/admission tokens retain the prior router owner; copy-on-write insertion invalidates those tokens for the modified App. Do not promise that outstanding tokens survive insertion or introduce a new route-owner identity model. Document this boundary and test that a pre-insertion token is rejected while requests admitted after insertion dispatch normally. This API is not live state replacement. + +A non-Send initializer does not relax the existing Clone/Send/Sync bounds on registered handler state. Applications own state contents and any intentional interior mutability. Retention is explicit; ordinary AppConfig request extraction does not become a cache automatically. + +## Store and filesystem contract + +The requirements below define strict production preparation. Development retains its current local config discovery, empty/malformed binding policy, and cwd-relative data paths. File identities, alias handling, ownership limitations, snapshots, and safe recovery guidance apply in both paths. + +### Config + +- Read `/local-config-.json`, retaining the existing flat string-to-string format and logical-id identity. +- Every declared file must exist, be readable, and parse successfully. No empty fallback or dropped binding. +- Permit symlinks used by projected ConfigMap/Secret mounts. The opened target must be a readable regular file. Operator-supplied roots are trusted; no generic filesystem sandbox or source-tree containment restriction is added. +- A valid outer map is not proof of a valid typed blob or required secrets. Those checks belong to the initializer. +- Config stores remain startup snapshots. Updates require process restart; no watcher/reload path. + +### KV + +- Read/create `/kv--.redb`, preserving the existing basename derived from the resolved platform NAME. Config filenames still use logical ids. +- The root must already exist and be writable. The runner may create a new database file inside it, but must not fabricate a missing mount/root directory. +- Resolve aliases before opening: two logical ids with the same resolved NAME share one database handle and keyspace in this process. Different names select different files. Do not change file identity or silently move existing data. +- The same key through either alias refers to the same value. Alias ids are not isolated namespaces. Shared handles do not change operation guarantees or make the documented read-modify-write helper atomic. +- redb remains one owning process per database. A second process fails rather than silently sharing. Shared volumes do not establish multi-writer or multi-replica safety. +- Keep runner-owned handles alive until connection tasks settle, then relinquish them on completed startup failure, orderly stop, or cooperative forced cleanup. A database lock is released only when the last handle owner drops it. Embedding callers may retain an attached handle, and initializers may clone one outside the App; those owners must release their handles before reopening or backing up the database. The runner does not revoke caller-owned handles. + +### Permissions, updates, and recovery + +Operators provision roots, ownership, and permissions for the runtime UID, including the Docker recipe's numeric UID. Config may be read-only; data must support real database writes. No startup chown, chmod 777, automatic lock repair, database deletion, or corruption recovery is added. + +Externally supplied secret updates use process restart. The existing environment-backed secret backend remains unchanged; this is not a new secret reload/rotation system. + +Initial backup support is a documented stopped-writer procedure: stop the writer, release every database handle including caller-owned clones, copy the closed database, restore into a prepared directory with correct permissions, reopen, and read stored values back. Returning from an embedded runner alone does not prove the database is closed. Online copying of an active database is not a supported backup claim. Remove ordinary recovery advice that recommends deleting production databases for lock errors or capacity maintenance; do not invent a maintenance subsystem. + +The storage guide must include these operational limits and procedures: + +- TTL expiry is lazy. Reads and paginated listing clean up encountered expired entries; unvisited expired entries can remain. An application-owned maintenance pass can traverse the existing paginated listing API through the owning process, following cursors to exhaustion even when a page is empty. There is no automatic sweeper or new maintenance CLI. +- Deletes and expiry cleanup do not automatically shrink the database file. Monitor database size and available mount capacity, provision headroom for writes and backups, and do not treat TTL as a disk quota. This slice does not expose or certify compaction; unsupported file shrinking must be stated explicitly. +- For lock conflicts, identify and stop the other owning process or release retained embedding handles before retrying. Do not remove the database or bypass its lock. Prevent overlapping writers during restart/rollout. +- For capacity exhaustion, stop the writer if needed, preserve the database, and restore usable capacity by enlarging the mount or removing unrelated disposable files. Reopen and verify stored values once capacity is available; backup/restore copying alone is not a compaction procedure. +- For suspected corruption, stop all owners, preserve the damaged file for investigation, and restore a known-good stopped-writer backup into a prepared directory. Verify stored values before resuming traffic. Automated repair and untested salvage procedures remain unsupported. + +## Lifecycle, shutdown, and results + +One native owner controls Starting, Ready, Draining, and Stopped. Acceptance and phase observation derive from that owner, not separately maintained booleans or response reports. A run never returns to Ready after draining begins. + +```mermaid +flowchart TD + A[Starting] -->|Validated service and listener| B[Ready] + A -->|Stop before ready| C[Cleanup without readiness] + A -->|Initialization failure| G[Stopped with non-success] + B -->|First stop request| D[Draining: revoke ready and block new work] + D --> E[Graceful connection completion] + E -->|All tasks settled| F[Stopped with zero exit] + C -->|Cleanup completes| F + D -->|Budget expires or second signal| H[Force cancellation and settle tasks] + C -->|Budget expires or second signal| H + H --> G +``` + +Standalone SIGINT and Unix SIGTERM feed the same first-stop transition. Registration failure is a startup error. A second termination signal of either kind requests immediate forced teardown. Embedding has no implicit global handlers. + +On first stop: + +1. Revoke readiness and stop accepting new connections. +2. Gate new request dispatch on existing keep-alive connections too. Define existing work at the service-dispatch boundary, not merely an accepted socket or queued pipelined bytes. +3. Ask owned Hyper connections to stop keep-alive/new work and finish dispatched requests. Close idle connections promptly. +4. Keep polling tracked local tasks and request/response work until completion or the single process drain deadline. +5. On expiry or second signal, cancel remaining work and settle its task ownership before returning. + +Use native monotonic runtime time for the process budget. Do not depend on an application's injected request clock for supervisor termination. Earlier accepted request/response deadlines still win. Long-lived/SSE responses share the shutdown budget; this does not impose a new ordinary stream-lifetime or handler deadline. + +Collect connection task results instead of detaching them. Preserve ordinary connection-scoped failure/disconnect behavior. Terminal listener/runtime/runner failures revoke readiness and return non-success; do not add a general restart/retry policy here. + +Completed graceful drain or completed intentional stop during startup exits zero. Startup failure, runner failure, grace exhaustion, and signal-requested forced teardown report non-success. Retain enough internal outcome distinction for focused tests and safe diagnostics, without introducing a detailed numeric exit-code taxonomy. + +The budget is cooperative. Synchronous startup code, blocking handlers, callbacks, destructors, or a stalled runtime can prevent timer/cancellation progress. The process supervisor supplies the final hard-kill boundary. Deployment examples must set a longer supervisor interval, such as 30 seconds for a 20-second drain, rather than rely on Docker's default timeout. + +Preserve #275 completion semantics. HostHandoff does not prove client receipt. Dropping unfinished egress uses its existing terminal outcome. Pre-egress cancellation may release captured resources without a completion callback/report. Do not promise a terminal report for every accepted request, SIGKILL, or process abort; connection ownership determines drain completion. + +## App-owned probe routes + +Add a small portable `edgezero_core::probe` module with: + +- A Starting/Ready/Draining/Stopped observation enum. +- A cloneable read-only live reader, following the existing MonotonicClock Arc/callback pattern. +- Ordinary `#[action]` handlers for readiness and liveness. + +The native runner publishes its reader through existing request extensions. Core contains no lifecycle writer, shutdown token, listener, task, timer, Axum/Tokio dependency, new action capability, or portable async Hooks method. Other adapters initially publish no native lifecycle observation. + +Handlers read the current phase when executed, not a copied admission-time snapshot. + +| Observed phase | Readiness | Liveness | +| --------------------- | ---------------------------- | ---------------------------- | +| Starting | 503 | 200 | +| Ready | 200 | 200 | +| Draining | 503 | 200 | +| Stopped | 503 | 503 | +| Missing/unknown state | 503, probe state unavailable | 503, probe state unavailable | + +Use short non-sensitive text, plain-text content type, and Cache-Control: no-store. Missing state must not invent healthy behavior or lifecycle support on a provider. Apps can retain an independent portable health handler when that is their desired policy. Ready reflects the selected runner's contract: development readiness is not strict production validation, and production convenience readiness does not imply application-specific initialization. Native readiness does not periodically revalidate dependencies. Applications can add ongoing checks; the framework does not poll upstreams or trigger automatic outage restarts. + +Applications bind helpers through ordinary Rust or manifest routes. No automatic paths, reserved namespace, health flag, health config section, path environment variable, management listener, or admission exemption. Existing middleware/auth remains application-owned. Do not replace framework-owned probe/store metadata through app state or middleware; the native dispatch gate must use its private owner rather than an overrideable request extension. + +There is no reachable-probe guarantee before startup acceptance or after listener withdrawal. The response table describes a helper that executes. During drain, a previously dispatched probe can observe Draining; a newly attempted request may receive refusal/reset/EOF instead of 503. No successful readiness response means not ready. Supervisor liveness policy must account for intentional listener shutdown. + +Trusted Server already owns `/health`. Its Axum handler returns 200 on a successfully built router, while startup fallback returns errors even for `/health`. Fastly answers `/health` before app construction, so that route is liveness rather than readiness. Keep those downstream routes application-owned. Generated EdgeZero root/introspection paths are editable trigger rows, not reserved paths; the old automatic `/__edgezero/routes` mechanism was removed. + +## Failure diagnostics + +Startup failures identify the stage, recognized setting/logical store id, and safe failure category. Never print config content, resolved secrets, raw invalid settings, parser input fragments, or callback/provider source strings. A malformed config must not leak through a serde error wrapped in an ordinary log or anyhow termination chain. + +Return safe classified errors at the standard runner boundary. Do not retain an unredacted source chain where ordinary main termination/debug formatting will print it. Broader permitted internal diagnostics belong to #397; this spec does not add a logging framework or debug disclosure switch. + +Use the existing public HTTP error contract for request failures. Probe bodies contain no app configuration or infrastructure details. Preserve custom application logger ownership. These guarantees apply to framework-owned error conversion/logging; arbitrary app logging and panic payloads remain application responsibilities. + +## Compatibility and entrypoint migration + +Preserve #396's permissive development requirement and the existing no-argument `run_app` signature. This review decision supersedes the earlier interview's strict-everywhere and mandatory-initializer policies. No GitHub issue amendment is needed. + +Inventory Axum main templates, app-demo's native entrypoint, CLI demo runner, AxumDevServer/test embedding callers, and downstream consumers. Do not migrate development callers merely to make them pass a no-op initializer. Add production adoption examples that select the strict convenience entrypoint or, for application-required checks, its initializer variant. Both use the existing server implementation. Document their different readiness guarantees and any prerequisite-driven source migration. + +Generated no-store apps retain their runnable development entrypoint. Show the explicit production entrypoint choice and how required stored config adds typed loading and state registration. Container recipes under #398 must select that production contract rather than certify a permissive development binary from a passing HTTP probe. Do not add schema inference, a new app! keyword, optional-store policy machinery, or a native dependency in the shared app crate. + +The native CLI's serve/push/provision recipes retain local preparation and path conventions. Production adoption documentation supplies explicit absolute roots; development does not acquire a mandatory-root requirement. `config push` retains its current project-local output convention. No automatic production push/deployment mechanism is added. + +Post-build state registration remains portable and preserves router state semantics. The context-free config helper preserves all existing request-extraction outcomes on native/WASM targets. Coordinate #379's retained-provider-app work where APIs overlap, without making it a required native hosting dependency or assuming retained provider apps imply process readiness. + +## Originally proposed PR split + +The original plan below separated the issue boundaries. This worktree delivers the shared runtime and startup/storage changes together in one PR stacked on #275, not old main. The issue ownership and acceptance boundaries still apply. + +1. **#392 lifecycle PR, first on #275.** Establish the single local runtime owner, explicit initialization barrier, caller/signal stop handling, tracked connections, dispatch gating, drain/result behavior, and core probe observation/helpers. Its initialization barrier can accept a local future without knowing application schemas. Do not claim strict config/value certification before #396 integrates. +2. **#396 startup/storage PR, stacked on #392.** Supply explicit strict production preparation and optional app initialization to that barrier; implement root/settings validation, alias handle reuse, shared typed loader, startup-only post-build state, production adoption examples, development compatibility, and persistence/permissions/backup checks. +3. **Integrated acceptance on the second PR.** Exercise failed initialization, phase transitions, signals, pending handlers/streams, durable restart, mounted permissions, and probe wiring against the combined series. Close overlapping issue acceptance only with that evidence. + +These are behavioral review boundaries, not permission for two independent agents to rewrite the same runner concurrently. Preserve existing development signatures and keep any prerequisite-required caller changes at a coherent boundary. Add production entrypoints and real required-config initialization under #396 without forcing a development migration. Do not leave a broken intermediate commit or invent a compatibility server. + +#393/#394/#395/#397 then add only residual work/evidence against the accepted foundation. #398/#399 can prepare packaging independently; final container production checks consume the integrated result. #400 documents proven behavior. + +## Likely affected files + +- `crates/edgezero-adapter-axum/src/dev_server.rs`, `connection.rs`, `service.rs`, `config_store.rs`, and `key_value_store.rs`, with colocated tests. +- `crates/edgezero-adapter-axum/src/cli.rs` and its native main template where needed to preserve development behavior and document explicit production adoption. Strict launch wrappers must not normalize invalid production inputs. +- `crates/edgezero-core/src/app.rs`, `router.rs`, and `extractor.rs` for state registration and shared typed loading; new `probe.rs` plus exports in `lib.rs`. +- CLI demo runner, root README/core-config templates where preparation guidance is stale, generated-project tests, and app-demo's Axum main. +- Axum/config/KV guides and small native/portable fixtures needed for the acceptance checks. Do not pull Trusted Server sources or operator configs into framework fixtures. + +Keep this list bounded. No general generator/dependency resolver, provider lifecycle, portable body, or distributed store rewrite is implied. + +## Acceptance and planned validation + +The table below is the full acceptance plan, not a claim that every item is complete. The initial implementation has passed workspace tests, strict workspace Clippy, formatting, the Fastly/Cloudflare/Spin feature check, and the Spin WASM compilation check. The Axum all-features suite also passed all 13 subprocess hosting tests, including signals, finite drain, forced cancellation, startup failures, embedding, persistence, aliasing, stopped-writer backup/restore, and read-only config permissions. + +Before choosing implementation changes, re-audit the accepted #275 integration revision and record its exact commit, existing contracts/tests reused, checks already covered, and residual implementation/evidence needed for each issue. Repeat the inventory if prerequisite contracts change. The draft-head source inspection above is not this acceptance gate or executed production proof. + +| Owner | Focused evidence required | +| ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| Both | Accepted-#275 revision and covered/residual inventory are recorded; add only residual changes and missing acceptance evidence. | +| #392 | Subprocess SIGINT and Unix SIGTERM each cause graceful stop; second signal forces non-success. Embedding stop installs no global handlers. | +| #392 | In a running subprocess, dispatch a delayed finite response before sending process-wide SIGTERM; observe readiness withdrawal, full response completion, and zero exit. Repeat SIGTERM with a never-ending handler and open-ended stream to prove forced non-zero exit within the configured cooperative budget. Reuse these scenarios for final-image integration; separate signal and unit drain tests do not replace them. | +| Both | Production subprocess failures cover an occupied bind port, explicit initializer failure, invalid recognized settings, missing roots/declared config files, unreadable/non-regular/malformed config, and unusable/locked KV data. Each exits non-zero without Ready or request acceptance and identifies a safe stage/setting or logical id/category; sentinel values and unredacted source chains never appear in ordinary stdout/stderr diagnostics. | +| #392 | Never Ready during startup/failure; startup stop never accepts; Ready only after successful barrier and listener setup; drain revokes it immediately. | +| #392 | Delayed dispatched work finishes; idle keep-alive closes; new keep-alive/pipelined work does not dispatch after drain. | +| #392 | Never-ending handler/stream and zero-read peer reach controlled forced teardown within the configured cooperative budget. No detached work remains when the runner returns. | +| #392 | One budget is not restarted by repeat programmatic stop. Existing earlier response deadlines/outcomes remain intact. | +| #392 | Probe helpers cover every phase and missing state through ordinary router dispatch, preserve app route ownership, and observe a live transition rather than a stale snapshot. New drain probes need not reach HTTP. | +| #396 | Production entrypoints reject invalid supplied host/port/logging/known overrides/duration without normalization into defaults. Omitted bind/duration values use documented defaults. Production launch wrappers preserve this behavior. | +| #396 | Existing development entrypoints remain runnable without new root settings or initializer arguments. Preserve missing-config empty stores, malformed-binding warning/omission, local path discovery/cwd defaults, existing bind fallbacks, and fatal declared KV failures. Diagnostics remain redacted. Selecting production explicitly enforces strict checks regardless of cwd or build profile. | +| #396 | A standalone release subprocess using the production entrypoint starts from unrelated cwd with only explicit roots; missing roots, unreadable/non-regular/malformed config, and unusable/locked data fail before acceptance. | +| #396 | Valid flat map containing malformed/missing required typed blob, integrity failure, invalid required fields, or missing referenced secret fails the initializer before Ready. Sentinel values never appear in ordinary logs/returned diagnostics. | +| #396 | A supplied initializer runs once with exactly the bindings attached to requests. Production convenience entrypoints require no callback but still enforce framework checks and never claim app-specific validation. Native production option construction does not silently merge environment settings. | +| #396 | Shared typed loader and request extraction agree on default KEY overrides, named stores/secrets, optional secrets, redaction, byte/structure/deadline limits, and existing error categories. | +| #396 | Startup-registered Arc state reaches multiple requests; last same-type insertion wins; another App sharing the original router retains its own state. Pre-insertion resolved/admission tokens are rejected for the changed App, while tokens created after insertion dispatch normally. | +| #396 | Distinct names stay distinct; same-name aliases share one handle/keyspace; fresh database in prepared root works; second process cannot own the same file. | +| #396 | Value survives actual process stop/restart; stopped-writer backup restores readable values; runtime UID works with read-only config and writable data; startup never changes root ownership or deletes locked files. An embedding owner retaining a handle keeps the file locked after runner return; dropping the last owner permits reopening. | +| #396 | Storage guidance covers lazy expiry, paginated cleanup, non-shrinking files, capacity monitoring/headroom, safe lock/capacity/corruption recovery, and unsupported compaction/repair. Backup instructions require all handle owners to release the database. | +| #396 | Config snapshot stays unchanged after file update until restart. Document restart-based secret deployment updates without unsafe process-environment mutation tests. | +| Both | Non-Send local initializer fixture compiles/runs; same shared probe/typed-state handlers compile for supported WASM targets without native dependencies. | + +Tests use disposable data, bounded waits, explicit cleanup, and no platform credentials. Ordinary unit/compile tests do not require Docker or external services. Signal/filesystem subprocess tests use local fixtures. UID/container evidence can reuse #398's opt-in runner after integration with its template PR; unavailable Docker/architecture evidence remains a reported gap. + +At implementation checkpoints, use the repository's required checks: + +```sh +cargo test -p edgezero-core --all-targets +cargo test -p edgezero-adapter-axum --all-targets --all-features +cargo test -p edgezero-cli --all-targets +cargo test --workspace --all-targets +cargo fmt --all -- --check +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo check --workspace --all-targets --features "fastly cloudflare spin" +cargo check -p edgezero-adapter-spin --target wasm32-wasip2 --features spin +cargo test -p edgezero-cli --test generated_project_builds -- --ignored +``` + +Also run docs lint/format/build and the existing example/provider gates relevant to changed callers. Installed-target or container gaps must be disclosed, not presented as passes. Existing Fastly/Cloudflare/Spin gates must stay green. Container UID, standalone release, unreadable/unusable mount, retained-handle lock, and production-guide acceptance remain open. The accepted-prerequisite re-audit must also be repeated after #275 merges. + +## Risks and review checks + +- Compile the proposed higher-ranked initializer signature with borrowed inputs and Rc-capturing non-Send futures before committing to its ergonomics. No runtime ownership workaround may add blanket Send bounds. +- Verify copy-on-write router state insertion against the locked matchit/http types and existing Clone/Send/Sync state bounds. No uniqueness assumption or cross-App mutation is acceptable. +- Production strictness is opt-in. Preserve existing development signatures, fallbacks, and data filenames. Make production adoption and readiness guarantees explicit; do not silently certify a development binary or force unrelated callers to migrate. +- App-owned initialization/state can execute arbitrary code. Cooperative cancellation does not bound blocking code/destructors or prevent unsafe app overrides of request metadata. The supervisor and trusted app-composition contract remain explicit. +- Existing typed limits do not prove startup-file or whole-process memory bounds. This spec adds no concurrency/handler-deadline policy and does not relabel BestEffort/Unsupported guarantees from #275. +- A successfully sampled readiness response does not promise the server remains ready through transmission or prove client receipt. + +Read-only audits and alternative proposals were used for lifecycle, configuration, native startup API, and probe portability. Their useful findings are incorporated above. Rejected alternatives were a second server, inferred environment modes, native dependencies in shared probe handlers, a portable async startup hook, synthetic request validation, implicit settings caches, per-id path machinery, and immediate-stop-only hosting. This review preserves permissive development and makes strict production an explicit entrypoint choice, superseding the earlier strict-everywhere and mandatory-no-op proposals. + +## Review boundary + +The interview's policy decisions and subsequent review revisions are recorded here. The initial source and test implementation covers development compatibility, optional application initialization, startup-only state insertion, explicit embedding options, restart-based config updates, and the native lifecycle owner. API and behavioral review remain necessary before merge. + +The implementation is based on #275 at `5527b58e2a7cb6307d5c7a8926cd5dfc6f4f8b4a`. Its branch has since advanced to `dfd439c613af4f6ed295e71c032881b15bf7247c`, including overlapping Axum preflight and logging changes. Integration with that newer prerequisite and its accepted-revision re-audit remain open. This work does not certify a deployment or published container image, and it does not complete all acceptance criteria for #392 or #396.