Add S3CachingServiceProxy

regexident · regexident · commit e1350ca65549 · 2026-03-13T13:58:09.000+01:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -80,6 +80,8 @@ tracing-subscriber = { version = "0.3", features = [
 
 [dev-dependencies]
 http = "1"
+http-body-util = "0.1"
+hyper = { version = "1", features = ["client"] }
 mock_instant = "0.6"
 testcontainers = "0.27"
 testcontainers-modules = { version = "0.15", features = ["minio"] }
diff --git a/src/lib.rs b/src/lib.rs
@@ -40,6 +40,7 @@ mod fifo_cache;
 mod metrics_writer;
 mod proxy;
 mod s3_cache;
+mod service;
 mod statistics;
 mod telemetry;
 
@@ -140,11 +141,11 @@ where
         config.cache_dry_run,
     );
 
-    // Build S3 service with auth
+    // Build S3 service with auth, wrapped in a health check layer
     let service = {
         let mut b = S3ServiceBuilder::new(caching_proxy);
         b.set_auth(auth::create_auth(&config));
-        b.build()
+        service::S3CachingServiceProxy::new(b.build())
     };
 
     // Start Prometheus metrics writer if configured
diff --git a/src/service.rs b/src/service.rs
@@ -0,0 +1,52 @@
+use bytes::Bytes;
+use hyper::service::Service;
+use hyper::{Method, Request, Response, body::Incoming};
+use s3s::{Body, HttpError};
+use std::future::Future;
+use std::pin::Pin;
+
+type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
+
+/// Wraps an S3 service and short-circuits `GET /` and `GET /health` requests,
+/// returning `200 OK` with a plain-text `"Status OK"` body without forwarding
+/// them to the S3 layer or requiring authentication.
+pub struct S3CachingServiceProxy<S> {
+    inner: S,
+}
+
+impl<S> S3CachingServiceProxy<S> {
+    pub fn new(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S: Clone> Clone for S3CachingServiceProxy<S> {
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<S> Service<Request<Incoming>> for S3CachingServiceProxy<S>
+where
+    S: Service<Request<Incoming>, Response = Response<Body>, Error = HttpError>,
+    S::Future: Send + 'static,
+{
+    type Response = Response<Body>;
+    type Error = HttpError;
+    type Future = BoxFuture<Result<Self::Response, Self::Error>>;
+
+    fn call(&self, req: Request<Incoming>) -> Self::Future {
+        if req.method() == Method::GET && (req.uri().path() == "/" || req.uri().path() == "/health")
+        {
+            let response = Response::builder()
+                .status(200)
+                .body(Body::from(Bytes::from_static(b"Status OK")))
+                .unwrap();
+            Box::pin(std::future::ready(Ok(response)))
+        } else {
+            Box::pin(self.inner.call(req))
+        }
+    }
+}
diff --git a/tests/integration_health.rs b/tests/integration_health.rs
@@ -0,0 +1,93 @@
+use bytes::Bytes;
+use http_body_util::{BodyExt, Empty};
+use hyper_util::rt::TokioIo;
+use std::net::SocketAddr;
+use tokio::net::TcpStream;
+
+async fn start_test_server() -> (SocketAddr, tokio::sync::oneshot::Sender<()>) {
+    let config = s3_cache::Config {
+        listen_addr: "127.0.0.1:0".parse().unwrap(),
+        upstream_endpoint: "http://127.0.0.1:1".to_string(),
+        upstream_access_key_id: "test".to_string(),
+        upstream_secret_access_key: "test".to_string(),
+        upstream_region: "us-east-1".to_string(),
+        client_access_key_id: "testclient".to_string(),
+        client_secret_access_key: "testsecret".to_string(),
+        cache_enabled: false,
+        cache_dry_run: false,
+        cache_shards: 4,
+        cache_max_entries: 100,
+        cache_max_size_bytes: 1024 * 1024,
+        cache_max_object_size_bytes: 1024,
+        cache_ttl_seconds: 60,
+        worker_threads: 2,
+        otel_grpc_endpoint_url: None,
+        prometheus_textfile_dir: None,
+    };
+
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+    let (addr_tx, addr_rx) = tokio::sync::oneshot::channel::<SocketAddr>();
+
+    tokio::spawn(async move {
+        let shutdown = async move {
+            let _ = shutdown_rx.await;
+        };
+        let _ = s3_cache::start_app_with_shutdown(config, shutdown, addr_tx).await;
+    });
+
+    let addr = addr_rx.await.expect("server startup failed");
+    (addr, shutdown_tx)
+}
+
+async fn http_get(addr: SocketAddr, path: &str) -> (u16, String) {
+    let stream = TcpStream::connect(addr).await.unwrap();
+    let io = TokioIo::new(stream);
+
+    let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await.unwrap();
+    tokio::spawn(conn);
+
+    let req = hyper::Request::builder()
+        .method(hyper::Method::GET)
+        .uri(path)
+        .header("Host", "localhost")
+        .body(Empty::<Bytes>::new())
+        .unwrap();
+
+    let resp = sender.send_request(req).await.unwrap();
+    let status = resp.status().as_u16();
+    let body = resp.collect().await.unwrap().to_bytes();
+
+    (status, String::from_utf8_lossy(&body).to_string())
+}
+
+// MARK: - Health
+
+#[tokio::test(flavor = "multi_thread")]
+async fn health_check_ok() {
+    let (addr, _shutdown) = start_test_server().await;
+
+    let (status, body) = http_get(addr, "/health").await;
+
+    assert_eq!(status, 200);
+    assert_eq!(body, "Status OK");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn health_check_root_ok() {
+    let (addr, _shutdown) = start_test_server().await;
+
+    let (status, body) = http_get(addr, "/").await;
+
+    assert_eq!(status, 200);
+    assert_eq!(body, "Status OK");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn health_check_does_not_require_auth() {
+    let (addr, _shutdown) = start_test_server().await;
+
+    // No Authorization header — must still succeed
+    let (status, _) = http_get(addr, "/health").await;
+
+    assert_eq!(status, 200);
+}
