Merge pull request #6 from socrates8300/codex/auth-crypto-ci-hygiene

v0.1-rc: graceful shutdown, Docker deploy, coverage gate; align fixes

Author: socrates8300

.claude/settings.local.json

@@ -0,0 +1,35 @@
+# Build artifacts
+target/
+
+# Version control
+.git/
+.gitignore
+
+# Editor / IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Documentation (not needed in build)
+docs/
+*.md
+LICENSE
+CODE_OF_CONDUCT.md
+
+# Dev tooling
+.devcontainer/
+.github/
+
+# Environment files
+.env
+.env.*
+
+# Temporary files
+*.tmp
+*.bak
+*.log

.dockerignore

@@ -92,4 +92,4 @@ jobs:
       - name: Install cargo-llvm-cov
         run: cargo install cargo-llvm-cov --version 0.6.16 --locked
       - name: Coverage Gate
-        run: cargo llvm-cov --workspace --summary-only --fail-under-lines 55
+        run: cargo llvm-cov --workspace --summary-only --fail-under-lines 60

.github/workflows/ci.yml

@@ -22,3 +22,15 @@ Thumbs.db
 # Temporary files
 *.tmp
 *.bak
+
+# SDD agent context database WAL sidecars (the .db itself is committed)
+docs/agent_context.db-shm
+docs/agent_context.db-wal
+
+# SDD CLI scratch/derived state — DB is the source of truth
+.sdd-iter
+.sdd-coverage-ledger.json
+docs/agent_context_schema.sql
+
+# Claude Code local-only settings (per-user permission overrides)
+.claude/settings.local.json

.gitignore

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: AGPL-3.0-only
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_trait::async_trait;
 use axum::body::Bytes;
@@ -19,7 +20,7 @@ use mx20022_channels::{
     ChannelError, ChannelHealth, DeliveryReceipt, InboundChannel, InboundMessage, OutboundChannel,
     OutboundMessage,
 };
-use tokio::sync::{mpsc, RwLock};
+use tokio::sync::{mpsc, watch, RwLock};
 use tower_http::cors::CorsLayer;
 
 #[derive(Debug, Clone)]
@@ -44,13 +45,18 @@ struct InboundState {
 pub struct HttpInboundChannel {
     config: HttpInboundConfig,
     paused: Arc<RwLock<bool>>,
+    shutdown_tx: Arc<watch::Sender<bool>>,
+    shutdown_rx: watch::Receiver<bool>,
 }
 
 impl HttpInboundChannel {
     pub fn new(config: HttpInboundConfig) -> Self {
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
         Self {
             config,
             paused: Arc::new(RwLock::new(false)),
+            shutdown_tx: Arc::new(shutdown_tx),
+            shutdown_rx,
         }
     }
 }
@@ -92,7 +98,16 @@ impl InboundChannel for HttpInboundChannel {
                         ))
                     })?;
                 tracing::info!(channel = %self.config.name, bind = %self.config.bind, "http inbound channel starting with TLS");
+                let handle = axum_server::Handle::new();
+                let shutdown_handle = handle.clone();
+                let mut shutdown_rx = self.shutdown_rx.clone();
+                tokio::spawn(async move {
+                    let _ = shutdown_rx.changed().await;
+                    tracing::info!("TLS graceful shutdown triggered");
+                    shutdown_handle.graceful_shutdown(Some(Duration::from_secs(30)));
+                });
                 axum_server::bind_rustls(socket, tls)
+                    .handle(handle)
                     .serve(app.into_make_service())
                     .await
                     .map_err(|e| ChannelError::new(format!("inbound channel serve failed: {e}")))
@@ -104,7 +119,13 @@ impl InboundChannel for HttpInboundChannel {
                         ChannelError::new(format!("failed to bind inbound channel: {e}"))
                     })?;
                 tracing::warn!(channel = %self.config.name, bind = %self.config.bind, "http inbound channel starting without TLS");
+                let mut shutdown_rx = self.shutdown_rx.clone();
+                let shutdown_signal = async move {
+                    let _ = shutdown_rx.changed().await;
+                    tracing::info!("graceful shutdown triggered");
+                };
                 axum::serve(listener, app)
+                    .with_graceful_shutdown(shutdown_signal)
                     .await
                     .map_err(|e| ChannelError::new(format!("inbound channel serve failed: {e}")))
             }
@@ -115,6 +136,8 @@ impl InboundChannel for HttpInboundChannel {
     }
 
     async fn shutdown(&self) -> Result<(), ChannelError> {
+        tracing::info!(channel = %self.config.name, "http inbound channel shutting down");
+        let _ = self.shutdown_tx.send(true);
         Ok(())
     }
 
@@ -307,16 +330,20 @@ fn now_millis() -> u128 {
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
+    use std::time::Duration;
 
     use axum::http::HeaderMap;
     use axum::http::StatusCode;
     use axum::routing::post;
     use axum::{extract::State, Router};
     use mx20022_channels::auth::InboundAuthConfig;
-    use mx20022_channels::{OutboundChannel, OutboundMessage};
+    use mx20022_channels::{InboundChannel, OutboundChannel, OutboundMessage};
     use tokio::sync::{mpsc, RwLock};
 
-    use super::{handle_post, HttpOutboundChannel, HttpOutboundConfig, InboundState};
+    use super::{
+        handle_post, HttpInboundChannel, HttpInboundConfig, HttpOutboundChannel,
+        HttpOutboundConfig, InboundState,
+    };
 
     #[tokio::test]
     async fn inbound_handler_enqueues_message() {
@@ -336,6 +363,72 @@ mod tests {
         assert_eq!(queued.content_type, "application/xml");
     }
 
+    #[tokio::test]
+    async fn shutdown_drain() {
+        let (tx, mut rx) = mpsc::channel(10);
+
+        // Bind to a free port by creating a temporary listener.
+        let temp_listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind temp listener");
+        let addr = temp_listener.local_addr().expect("resolve addr");
+        drop(temp_listener);
+
+        let channel = Arc::new(HttpInboundChannel::new(HttpInboundConfig {
+            name: "test-shutdown".to_string(),
+            bind: addr.to_string(),
+            content_type: "application/xml".to_string(),
+            auth: InboundAuthConfig::default(),
+            cors_allowed_origins: vec![],
+            tls_cert_path: None,
+            tls_key_path: None,
+        }));
+
+        let run_channel = Arc::clone(&channel);
+        let handle = tokio::spawn(async move { run_channel.run(tx).await });
+
+        // Wait for the server to be ready.
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        // Send a message before shutdown — should succeed.
+        let client = reqwest::Client::new();
+        let resp = client
+            .post(format!("http://{addr}/"))
+            .header("content-type", "application/xml")
+            .body("<Document/>")
+            .send()
+            .await
+            .expect("pre-shutdown request");
+        assert_eq!(resp.status(), StatusCode::ACCEPTED);
+
+        // Verify the message was queued.
+        let msg = rx
+            .recv()
+            .await
+            .expect("should receive pre-shutdown message");
+        assert_eq!(msg.raw, "<Document/>");
+
+        // Trigger graceful shutdown.
+        channel.shutdown().await.expect("shutdown");
+
+        // Wait for the server to drain.
+        let result = tokio::time::timeout(Duration::from_secs(5), handle).await;
+        assert!(result.is_ok(), "server should shut down within timeout");
+        assert!(result.unwrap().is_ok(), "server task should not error");
+
+        // After shutdown, new requests should be rejected.
+        let err = client
+            .post(format!("http://{addr}/"))
+            .header("content-type", "application/xml")
+            .body("<AfterShutdown/>")
+            .send()
+            .await;
+        assert!(
+            err.is_err(),
+            "post-shutdown request should fail (connection refused)"
+        );
+    }
+
     #[tokio::test]
     async fn outbound_channel_posts_payload() {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")

crates/mx20022-channels/mx20022-channel-http/src/lib.rs

@@ -105,5 +105,103 @@ fn bench_store_query(c: &mut Criterion) {
     });
 }
 
-criterion_group!(benches, bench_runtime_process, bench_store_query);
+fn build_3p_config(sqlite_path: &str) -> RuntimeConfig {
+    let raw = format!(
+        r#"
+[runtime]
+name = "bench-runtime"
+instance_id = "bench-01"
+
+[store]
+backend = "sqlite"
+url = "{sqlite_path}"
+
+[channels.http-in]
+type = "http"
+mode = "server"
+bind = "127.0.0.1:0"
+
+[[pipeline]]
+name = "bench-3p"
+channel_in = "http-in"
+participants = [
+  {{ name = "message-logger" }},
+  {{ name = "duplicate-checker" }},
+  {{ name = "circuit-breaker" }}
+]
+"#
+    );
+    RuntimeConfig::parse(&raw).expect("benchmark 3p runtime config should parse")
+}
+
+fn build_3p_runtime(rt: &tokio::runtime::Runtime) -> (TempDir, Arc<RuntimeApp>) {
+    let temp_dir = TempDir::new().expect("create temp dir");
+    let sqlite_path = temp_dir.path().join("bench-3p.db");
+    let cfg = build_3p_config(sqlite_path.to_str().expect("utf8 sqlite path"));
+    let app = rt
+        .block_on(RuntimeApp::from_config(&cfg))
+        .expect("build 3p runtime app");
+    (temp_dir, Arc::new(app))
+}
+
+fn bench_3p_process(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().expect("create tokio runtime");
+    let (_temp_dir, app) = build_3p_runtime(&rt);
+    let counter = AtomicU64::new(1);
+
+    c.bench_function("runtime_app_process_3_participants", |b| {
+        b.to_async(&rt).iter(|| async {
+            let id = counter.fetch_add(1, Ordering::Relaxed);
+            let report = app
+                .process(
+                    "bench-3p",
+                    format!("TX-BENCH-3P-{id}"),
+                    "http-in",
+                    "pacs.008.001.08",
+                    "<Document/>",
+                )
+                .await
+                .expect("process 3p benchmark transaction");
+            black_box(report);
+        });
+    });
+}
+
+fn bench_concurrent_process(c: &mut Criterion) {
+    let rt = tokio::runtime::Runtime::new().expect("create tokio runtime");
+    let (_temp_dir, app) = build_runtime(&rt);
+    let counter = AtomicU64::new(1);
+
+    c.bench_function("runtime_app_process_concurrent_10", |b| {
+        b.to_async(&rt).iter(|| async {
+            let mut tasks = Vec::with_capacity(10);
+            for _ in 0..10 {
+                let app = app.clone();
+                let id = counter.fetch_add(1, Ordering::Relaxed);
+                tasks.push(tokio::spawn(async move {
+                    app.process(
+                        "bench",
+                        format!("TX-CONC-{id}"),
+                        "http-in",
+                        "pacs.008.001.08",
+                        "<Document/>",
+                    )
+                    .await
+                }));
+            }
+            for task in tasks {
+                let result = task.await.expect("spawn should not panic");
+                black_box(result.expect("concurrent process should succeed"));
+            }
+        });
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_runtime_process,
+    bench_store_query,
+    bench_3p_process,
+    bench_concurrent_process
+);
 criterion_main!(benches);