Add Axum production startup and bounded shutdown

.github/workflows/codeql.yml

@@ -15,7 +15,6 @@ on:
   push:
     branches: ["main"]
   pull_request:
-    branches: ["main"]
   schedule:
     - cron: "40 17 * * 3"
 

.github/workflows/format.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
 
 permissions:
   contents: read

.github/workflows/test.yml

@@ -4,7 +4,6 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
 
 permissions:
   contents: read

crates/edgezero-adapter-axum/Cargo.toml

@@ -31,7 +31,7 @@ cli = [
     "dep:toml",
     "dep:walkdir",
 ]
-test-utils = ["axum", "edgezero-core/test-utils"]
+test-utils = ["axum", "edgezero-core/test-utils", "dep:validator"]
 
 [dependencies]
 edgezero-adapter = { path = "../edgezero-adapter", optional = true, features = [
@@ -55,6 +55,7 @@ redb = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+validator = { workspace = true, optional = true }
 simple_logger = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, optional = true }
@@ -66,6 +67,12 @@ walkdir = { workspace = true, optional = true }
 [target.'cfg(unix)'.dependencies]
 libc = { workspace = 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

@@ -114,6 +114,23 @@ impl AxumConfigStore {
         path: &Path,
         limits: ConfigStoreLimits,
         existing: usize,
+    ) -> Result<Self, ConfigStoreError> {
+        Self::load_snapshot(path, limits, existing, false)
+    }
+
+    pub(crate) fn load_required_at_startup(
+        path: &Path,
+        limits: ConfigStoreLimits,
+        existing: usize,
+    ) -> Result<Self, ConfigStoreError> {
+        Self::load_snapshot(path, limits, existing, true)
+    }
+
+    fn load_snapshot(
+        path: &Path,
+        limits: ConfigStoreLimits,
+        existing: usize,
+        required: bool,
     ) -> Result<Self, ConfigStoreError> {
         let mut options = fs::OpenOptions::new();
         options.read(true);
@@ -137,7 +154,12 @@ impl AxumConfigStore {
                 }
                 Snapshot::load(&mut file, limits, existing)?
             }
-            Err(err) if err.kind() == ErrorKind::NotFound => Snapshot::empty(limits, existing)?,
+            Err(err) if err.kind() == ErrorKind::NotFound && !required => {
+                Snapshot::empty(limits, existing)?
+            }
+            Err(err) if err.kind() == ErrorKind::NotFound => {
+                return Err(ConfigStoreError::unavailable("config file missing"));
+            }
             Err(_io) => {
                 return Err(ConfigStoreError::unavailable("config snapshot open failed"));
             }

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

@@ -2,13 +2,15 @@ use std::error::Error as StdError;
 use std::future::{self, Future};
 use std::io;
 
+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::header_deadline::{HeaderDeadline, HeaderDeadlineExceeded, HeaderIo, HeaderService};
@@ -24,15 +26,33 @@ pub(crate) enum ConnectionExit {
     DeadlineExceeded,
 }
 
-/// Owns HTTP/1 parsing, idle/header expiry and the existing response deadline race.
+#[cfg(test)]
+pub(crate) fn serve_http1<ServiceType, ResponseBody>(
+    stream: TcpStream,
+    service: ServiceType,
+    responses: EgressConnection,
+    limits: AxumIngressConfig,
+) -> impl Future<Output = Result<ConnectionExit, hyper::Error>>
+where
+    ServiceType: Service<Request<Incoming>, Response = hyper::Response<ResponseBody>>,
+    ServiceType::Error: Into<Box<dyn StdError + Send + Sync>>,
+    ServiceType::Future: 'static,
+    ResponseBody: HttpBody + 'static,
+    ResponseBody::Error: Into<Box<dyn StdError + Send + Sync>>,
+{
+    serve_http1_with_shutdown(stream, service, responses, None, limits)
+}
+
+/// Owns bounded HTTP/1 parsing, response deadlines and graceful lifecycle drain.
 #[expect(
     clippy::integer_division_remainder_used,
     reason = "tokio::select! expands to internal randomized branch selection arithmetic"
 )]
-pub(crate) fn serve_http1<ServiceType, ResponseBody>(
+pub(crate) fn serve_http1_with_shutdown<ServiceType, ResponseBody>(
     stream: TcpStream,
     service: ServiceType,
     responses: EgressConnection,
+    mut phase: Option<watch::Receiver<LifecyclePhase>>,
     limits: AxumIngressConfig,
 ) -> impl Future<Output = Result<ConnectionExit, hyper::Error>>
 where
@@ -59,6 +79,22 @@ 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;
+                }
+            };
             if head
                 .deadline()
                 .is_some_and(|deadline| TokioInstant::now() >= deadline)
@@ -95,6 +131,7 @@ where
                 }
                 () = head.changed() => {}
                 () = responses.changed() => {}
+                () = draining => {}
             }
         }
     }