Add Axum production startup and bounded shutdown

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"] }

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<Self, ConfigStoreError> {
-        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, ConfigStoreError> {
+        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<Self, ConfigStoreError> {
+        // 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<String, String> = 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<String, String> = serde_json::from_str(&raw).map_err(|_error| {
+            ConfigStoreError::unavailable(
+                "config file malformed: expected a flat string -> string JSON object",
+            )
         })?;
         Ok(Self {
             data,

crates/edgezero-adapter-axum/src/connection.rs

@@ -1,12 +1,15 @@
 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;
 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<ServiceType, ResponseBody>(
+    stream: TcpStream,
+    service: ServiceType,
+    responses: EgressConnection,
+) -> Result<ConnectionExit, hyper::Error>
+where
+    ServiceType: Service<Request<Incoming>, Response = hyper::Response<ResponseBody>>,
+    ServiceType::Error: Into<Box<dyn StdError + Send + Sync>>,
+    ResponseBody: HttpBody + 'static,
+    ResponseBody::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    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<ServiceType, ResponseBody>(
+pub(crate) async fn serve_http1_with_shutdown<ServiceType, ResponseBody>(
     stream: TcpStream,
     service: ServiceType,
     responses: EgressConnection,
+    mut phase: Option<watch::Receiver<LifecyclePhase>>,
 ) -> Result<ConnectionExit, hyper::Error>
 where
     ServiceType: Service<Request<Incoming>, Response = hyper::Response<ResponseBody>>,
@@ -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 => {}
             }
         }
     }