From ac2fcfc6af9794a2d076571808b2ee50d136d42e Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Tue, 14 Jul 2026 20:54:39 +0200
Subject: [PATCH 1/7] fix(server): Measure request duration end-to-end

The server.requests.duration metric previously stopped as soon as the handler
produced the response headers, excluding the time spent streaming the response
body. This under-measured GET, batch, and multipart:complete responses, whose
bodies are streamed after the middleware returns.

Move the timing guard into a MetricsBody wrapper around the response body so it
is dropped only once the body has finished streaming. When the body does not
stream to completion (client disconnect or a server-side stream error, folded
together), the metric is tagged status:499 instead of the header status.
---
 objectstore-server/docs/architecture.md    |  11 ++
 objectstore-server/src/web/metrics_body.rs |  79 ++++++++
 objectstore-server/src/web/middleware.rs   | 201 +++++++++++++++++++--
 objectstore-server/src/web/mod.rs          |   1 +
 4 files changed, 277 insertions(+), 15 deletions(-)
 create mode 100644 objectstore-server/src/web/metrics_body.rs

diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md
index 18058444..43dbf2e5 100644
--- a/objectstore-server/docs/architecture.md
+++ b/objectstore-server/docs/architecture.md
@@ -76,6 +76,17 @@ A request flows through several layers before reaching the storage service:
    [`objectstore-types` docs](objectstore_types) for the header mapping) and
    the payload is streamed back.
 
+The `server.requests.duration` metric is measured **end-to-end**: the timing guard is
+moved into the response body, so it records the full request lifetime including the time
+spent streaming the response body back to the client (e.g. GET payloads, batch and
+multipart responses), not just the time to produce the response headers.
+
+The metric's `status` tag reflects the response status sent in the headers, **except** when
+the response body does not stream to completion, in which case it is tagged `499` (nginx'
+non-standard "client closed request"). This covers two intentionally conflated cases: the
+client disconnecting mid-stream, and a server-side error while streaming the body. Both are
+reported as `499`.
+
 ## Authentication & Authorization
 
 Objectstore uses **JWT tokens with EdDSA signatures** (Ed25519) for
diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
new file mode 100644
index 00000000..d6682174
--- /dev/null
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -0,0 +1,79 @@
+//! Response body wrapper that keeps a metrics guard alive until the body ends.
+//!
+//! Axum handlers produce response *headers* while the middleware stack unwinds,
+//! but the response *body* is streamed by hyper afterwards. To measure the true
+//! end-to-end request duration, the timing guard must live until the body has
+//! been fully streamed. [`MetricsBody`] owns an [`EmitMetricsGuard`] and drops
+//! it when the inner body reaches end-of-stream (or is dropped on client
+//! disconnect), at which point the guard emits `server.requests.duration`.
+//!
+//! When the body streams to completion, [`MetricsBody`] calls
+//! [`EmitMetricsGuard::mark_completed`] so the metric is tagged with the real
+//! response status. If the body is dropped before end-of-stream — a client
+//! disconnect or a server-side stream error, which are intentionally conflated —
+//! the guard reports a `499` status instead.
+//!
+//! Because this delegates [`size_hint`](http_body::Body::size_hint) and
+//! [`is_end_stream`](http_body::Body::is_end_stream) to the inner body,
+//! buffered responses keep their exact size hint and hyper still emits a
+//! `Content-Length` header instead of chunked encoding.
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use axum::body::Body;
+use bytes::Bytes;
+use http_body::{Frame, SizeHint};
+use pin_project_lite::pin_project;
+
+use crate::web::middleware::EmitMetricsGuard;
+
+pin_project! {
+    /// Wraps an axum [`Body`] and holds an [`EmitMetricsGuard`] until the body ends.
+    ///
+    /// The guard emits the request-duration metric on drop, so keeping it inside
+    /// the body defers that emission until streaming completes.
+    pub struct MetricsBody {
+        // Dropped together with the body once streaming finishes; its `Drop`
+        // emits the request-duration metric.
+        guard: EmitMetricsGuard,
+        #[pin]
+        inner: Body,
+    }
+}
+
+impl MetricsBody {
+    /// Creates a new [`MetricsBody`] that keeps `guard` alive while polling `inner`.
+    pub fn new(guard: EmitMetricsGuard, inner: Body) -> Self {
+        Self { guard, inner }
+    }
+}
+
+impl http_body::Body for MetricsBody {
+    type Data = Bytes;
+    type Error = axum::Error;
+
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        let this = self.project();
+        let poll = this.inner.poll_frame(cx);
+        // `Ready(None)` is the only end-of-stream signal that is uniform across buffered and
+        // streamed bodies (`StreamBody` does not override `is_end_stream`). Reaching it means
+        // the body streamed to completion; anything else (client disconnect, stream error)
+        // leaves the guard marked incomplete, so it reports `499` on drop.
+        if let Poll::Ready(None) = poll {
+            this.guard.mark_completed();
+        }
+        poll
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        self.inner.size_hint()
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+}
diff --git a/objectstore-server/src/web/middleware.rs b/objectstore-server/src/web/middleware.rs
index 6574b835..d84d076e 100644
--- a/objectstore-server/src/web/middleware.rs
+++ b/objectstore-server/src/web/middleware.rs
@@ -14,6 +14,7 @@ use tower_http::set_header::SetResponseHeaderLayer;
 use crate::endpoints::is_internal_route;
 use crate::extractors::downstream_service::DownstreamService;
 use crate::web::RequestCounter;
+use crate::web::metrics_body::MetricsBody;
 use crate::web::sentry_body::SentryBody;
 
 /// The value for the `Server` HTTP header.
@@ -100,6 +101,10 @@ pub async fn bind_sentry_body(request: Request, next: Next) -> Response {
 ///
 /// Use this with [`from_fn`](axum::middleware::from_fn).
 ///
+/// The request-duration metric is measured **end-to-end**: the timing guard is moved into
+/// the response body (see [`MetricsBody`]) so it is dropped only once the body has finished
+/// streaming, not when the handler produces the response headers.
+///
 /// Internal routes (see [`is_internal_route`]) are excluded from metrics.
 pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response {
     let matched_path = request.extract_parts::<MatchedPath>().await;
@@ -111,26 +116,44 @@ pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response
 
     let response = next.run(request).await;
 
-    if let Some(guard) = guard {
-        guard.finish(response.status());
+    // Record the actual response status, then move the guard into the response body so the
+    // duration metric is emitted only when the body has finished streaming.
+    match guard {
+        Some(mut guard) => {
+            guard.finish(response.status());
+            let (parts, body) = response.into_parts();
+            let body = Body::new(MetricsBody::new(guard, body));
+            Response::from_parts(parts, body)
+        }
+        None => response,
     }
-    response
 }
 
 /// Helper for [`emit_request_metrics`].
 ///
-/// This tracks relevant generic request parameters and emits metrics. If the guard is dropped
-/// without calling [`Self::finish`], it will emit a `499`` status code (derived from nginx'
-/// non-standard "client closed request").
-struct EmitMetricsGuard<'a> {
-    route: &'a str,
+/// This tracks relevant generic request parameters and emits the `server.requests.duration`
+/// metric on drop. It is moved into the response body ([`MetricsBody`]) so timing spans the
+/// full response, including body streaming.
+///
+/// The metric is tagged with a `499` status (derived from nginx' non-standard "client closed
+/// request") whenever the response body did not stream to completion. This conflates two
+/// distinct cases: the client disconnecting mid-stream, and the server-side body stream
+/// erroring out. Both are reported as `499` regardless of the response status set via
+/// [`Self::finish`]. A `499` is also emitted when no response was produced at all (the guard
+/// dropped without [`Self::finish`] being called).
+pub struct EmitMetricsGuard {
+    route: String,
     method: Method,
     start: Instant,
     status: Option<StatusCode>,
+    /// Whether the response body streamed to completion (reached end-of-stream). Set via
+    /// [`Self::mark_completed`] from [`MetricsBody`]. When `false` at drop time, the metric is
+    /// tagged `499` instead of the response status.
+    completed: bool,
 }
 
-impl<'a> EmitMetricsGuard<'a> {
-    fn new(route: &'a str, method: &Method, service: DownstreamService) -> Self {
+impl EmitMetricsGuard {
+    fn new(route: &str, method: &Method, service: DownstreamService) -> Self {
         objectstore_metrics::count!(
             "server.requests",
             route = route.to_owned(),
@@ -139,24 +162,43 @@ impl<'a> EmitMetricsGuard<'a> {
         );
 
         Self {
-            route,
+            route: route.to_owned(),
             method: method.clone(),
             start: Instant::now(),
             status: None,
+            completed: false,
         }
     }
 
-    fn finish(mut self, status: StatusCode) {
+    /// Records the response status to tag the emitted metric with.
+    fn finish(&mut self, status: StatusCode) {
         self.status = Some(status);
     }
+
+    /// Marks the response body as having streamed to completion.
+    ///
+    /// Called by [`MetricsBody`] when the inner body reaches end-of-stream. If this is never
+    /// called before the guard drops (client disconnect or a server-side stream error), the
+    /// metric is tagged `499`.
+    pub(crate) fn mark_completed(&mut self) {
+        self.completed = true;
+    }
 }
 
-impl Drop for EmitMetricsGuard<'_> {
+impl Drop for EmitMetricsGuard {
     fn drop(&mut self) {
-        let status = self.status.map(|s| s.as_u16()).unwrap_or(499).to_string();
+        // A body that never reached end-of-stream is reported as `499` regardless of the
+        // status sent in the headers: the client disconnected mid-stream, or the body stream
+        // errored server-side. These two cases are intentionally conflated.
+        let status = if self.completed {
+            self.status.map(|s| s.as_u16()).unwrap_or(499)
+        } else {
+            499
+        }
+        .to_string();
         objectstore_metrics::record!(
             "server.requests.duration" = self.start.elapsed(),
-            route = self.route.to_owned(),
+            route = self.route.clone(),
             method = self.method.as_str().to_owned(),
             status = status,
             // service omitted to limit cardinality
@@ -253,4 +295,133 @@ mod tests {
         resume.notify_one();
         blocking.await.unwrap().unwrap();
     }
+
+    /// The request-duration metric must be emitted only once the response body has finished
+    /// streaming, not when the handler produces the response headers.
+    ///
+    /// This runs on a current-thread runtime inside [`with_capturing_test_client`] so all
+    /// metric emissions happen on the capturing thread. A sentinel `test.marker` metric is
+    /// emitted after the response headers are received but before the body is consumed; the
+    /// duration metric must appear *after* that sentinel.
+    #[test]
+    fn duration_measured_end_to_end() {
+        use axum::body::{self, Bytes};
+        use axum::middleware::from_fn;
+
+        let captured = objectstore_metrics::with_capturing_test_client(|| {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+
+            rt.block_on(async {
+                let app = Router::new()
+                    .route(
+                        "/v1/test/_/key",
+                        get(|| async {
+                            let stream = async_stream::stream! {
+                                yield Ok::<_, std::io::Error>(Bytes::from_static(b"hello"));
+                            };
+                            Body::from_stream(stream).into_response()
+                        }),
+                    )
+                    .layer(from_fn(emit_request_metrics));
+
+                let resp = app.oneshot(make_request("/v1/test/_/key")).await.unwrap();
+                assert_eq!(resp.status(), StatusCode::OK);
+
+                // Headers received. The duration metric must not have been emitted yet.
+                objectstore_metrics::count!("test.marker");
+
+                // Consuming the body drops the wrapping `MetricsBody` and its guard, which
+                // emits the duration metric.
+                let bytes = body::to_bytes(resp.into_body(), usize::MAX).await.unwrap();
+                assert_eq!(&bytes[..], b"hello");
+            });
+        });
+
+        let marker = captured
+            .iter()
+            .position(|m| m.starts_with("test.marker:"))
+            .expect("sentinel marker not captured");
+        let duration = captured
+            .iter()
+            .position(|m| m.starts_with("server.requests.duration:"))
+            .expect("duration metric not captured");
+
+        assert!(
+            duration > marker,
+            "duration metric emitted before body was consumed: {captured:?}"
+        );
+    }
+
+    /// Runs a request whose handler returns `body`, driving it to completion on a
+    /// current-thread runtime, and returns the captured `server.requests.duration` metric
+    /// string. If `consume_body` is false, the response body is dropped without being read,
+    /// simulating a client that disconnects mid-stream.
+    fn capture_duration_metric(body: Body, consume_body: bool) -> String {
+        use axum::body as axum_body;
+        use axum::middleware::from_fn;
+
+        let captured = objectstore_metrics::with_capturing_test_client(|| {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+
+            rt.block_on(async {
+                let shared = Arc::new(std::sync::Mutex::new(Some(body)));
+                let app = Router::new()
+                    .route(
+                        "/v1/test/_/key",
+                        get(move || {
+                            let shared = shared.clone();
+                            async move { shared.lock().unwrap().take().unwrap() }
+                        }),
+                    )
+                    .layer(from_fn(emit_request_metrics));
+
+                let resp = app.oneshot(make_request("/v1/test/_/key")).await.unwrap();
+                assert_eq!(resp.status(), StatusCode::OK);
+
+                if consume_body {
+                    axum_body::to_bytes(resp.into_body(), usize::MAX)
+                        .await
+                        .unwrap();
+                } else {
+                    // Drop the response (and its body) without reading it: the wrapping
+                    // `MetricsBody` never reaches end-of-stream, mirroring a client disconnect.
+                    drop(resp);
+                }
+            });
+        });
+
+        captured
+            .into_iter()
+            .find(|m| m.starts_with("server.requests.duration:"))
+            .expect("duration metric not captured")
+    }
+
+    /// A body streamed to completion reports the real response status.
+    #[test]
+    fn completed_stream_reports_real_status() {
+        let stream = async_stream::stream! {
+            yield Ok::<_, std::io::Error>(axum::body::Bytes::from_static(b"hello"));
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// A body dropped before end-of-stream (client disconnect) reports `499`, overriding the
+    /// `200` status that was sent in the headers.
+    #[test]
+    fn interrupted_stream_reports_499() {
+        // A stream that yields one chunk then pends forever, so it never reaches end-of-stream.
+        let stream = async_stream::stream! {
+            yield Ok::<_, std::io::Error>(axum::body::Bytes::from_static(b"hello"));
+            std::future::pending::<()>().await;
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), false);
+        assert!(metric.contains("status:499"), "unexpected metric: {metric}");
+    }
 }
diff --git a/objectstore-server/src/web/mod.rs b/objectstore-server/src/web/mod.rs
index 64dd787b..9f12dc81 100644
--- a/objectstore-server/src/web/mod.rs
+++ b/objectstore-server/src/web/mod.rs
@@ -13,6 +13,7 @@
 //! to start a test server and interact with it using the client SDK.
 
 mod app;
+mod metrics_body;
 mod middleware;
 mod request_counter;
 mod sentry_body;

From 28fd02bc05c2323ce102acaf4a36b98f4b98b50f Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Tue, 14 Jul 2026 20:58:13 +0200
Subject: [PATCH 2/7] fix(server): Map server-side stream errors to 500

Distinguish a server-side body stream error from a client disconnect: an error
frame while streaming the response body is now tagged status:500, while a client
disconnecting mid-stream remains status:499.
---
 objectstore-server/docs/architecture.md    | 11 +--
 objectstore-server/src/web/metrics_body.rs | 18 +++--
 objectstore-server/src/web/middleware.rs   | 80 +++++++++++++++-------
 3 files changed, 73 insertions(+), 36 deletions(-)

diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md
index 43dbf2e5..952365ab 100644
--- a/objectstore-server/docs/architecture.md
+++ b/objectstore-server/docs/architecture.md
@@ -82,10 +82,13 @@ spent streaming the response body back to the client (e.g. GET payloads, batch a
 multipart responses), not just the time to produce the response headers.
 
 The metric's `status` tag reflects the response status sent in the headers, **except** when
-the response body does not stream to completion, in which case it is tagged `499` (nginx'
-non-standard "client closed request"). This covers two intentionally conflated cases: the
-client disconnecting mid-stream, and a server-side error while streaming the body. Both are
-reported as `499`.
+the response body does not stream to completion:
+
+- **Client disconnect** mid-stream (the body is dropped before end-of-stream) is tagged
+  `499` (nginx' non-standard "client closed request").
+- **Server-side stream error** (the body yields an error while streaming) is tagged `500`.
+
+Both override the status sent in the headers.
 
 ## Authentication & Authorization
 
diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index d6682174..f593ab2f 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -9,9 +9,10 @@
 //!
 //! When the body streams to completion, [`MetricsBody`] calls
 //! [`EmitMetricsGuard::mark_completed`] so the metric is tagged with the real
-//! response status. If the body is dropped before end-of-stream — a client
-//! disconnect or a server-side stream error, which are intentionally conflated —
-//! the guard reports a `499` status instead.
+//! response status. A server-side stream error instead calls
+//! [`EmitMetricsGuard::mark_errored`] and is reported as `500`. If the body is
+//! dropped before either happens — a client disconnect — the guard reports a
+//! `499` status.
 //!
 //! Because this delegates [`size_hint`](http_body::Body::size_hint) and
 //! [`is_end_stream`](http_body::Body::is_end_stream) to the inner body,
@@ -61,10 +62,13 @@ impl http_body::Body for MetricsBody {
         let poll = this.inner.poll_frame(cx);
         // `Ready(None)` is the only end-of-stream signal that is uniform across buffered and
         // streamed bodies (`StreamBody` does not override `is_end_stream`). Reaching it means
-        // the body streamed to completion; anything else (client disconnect, stream error)
-        // leaves the guard marked incomplete, so it reports `499` on drop.
-        if let Poll::Ready(None) = poll {
-            this.guard.mark_completed();
+        // the body streamed to completion. A `Ready(Some(Err))` frame is a server-side stream
+        // error. If neither is observed before the body is dropped (client disconnect), the
+        // guard stays pending and reports `499`.
+        match &poll {
+            Poll::Ready(None) => this.guard.mark_completed(),
+            Poll::Ready(Some(Err(_))) => this.guard.mark_errored(),
+            _ => {}
         }
         poll
     }
diff --git a/objectstore-server/src/web/middleware.rs b/objectstore-server/src/web/middleware.rs
index d84d076e..d142f677 100644
--- a/objectstore-server/src/web/middleware.rs
+++ b/objectstore-server/src/web/middleware.rs
@@ -129,27 +129,39 @@ pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response
     }
 }
 
+/// How the response body streaming ended, which determines the `status` tag on the emitted
+/// metric when it differs from the status sent in the headers.
+#[derive(Clone, Copy)]
+enum BodyOutcome {
+    /// The body has not finished streaming. At drop time this means the client disconnected
+    /// mid-stream (or no response was produced at all), reported as `499`.
+    Pending,
+    /// The body streamed to completion; the real response status is reported.
+    Completed,
+    /// The body stream errored server-side, reported as `500`.
+    Errored,
+}
+
 /// Helper for [`emit_request_metrics`].
 ///
 /// This tracks relevant generic request parameters and emits the `server.requests.duration`
 /// metric on drop. It is moved into the response body ([`MetricsBody`]) so timing spans the
 /// full response, including body streaming.
 ///
-/// The metric is tagged with a `499` status (derived from nginx' non-standard "client closed
-/// request") whenever the response body did not stream to completion. This conflates two
-/// distinct cases: the client disconnecting mid-stream, and the server-side body stream
-/// erroring out. Both are reported as `499` regardless of the response status set via
-/// [`Self::finish`]. A `499` is also emitted when no response was produced at all (the guard
-/// dropped without [`Self::finish`] being called).
+/// The `status` tag reflects how streaming ended (see [`BodyOutcome`]):
+/// - streamed to completion: the response status set via [`Self::finish`];
+/// - client disconnect mid-stream, or no response produced at all: `499` (nginx' non-standard
+///   "client closed request");
+/// - server-side stream error: `500`.
 pub struct EmitMetricsGuard {
     route: String,
     method: Method,
     start: Instant,
     status: Option<StatusCode>,
-    /// Whether the response body streamed to completion (reached end-of-stream). Set via
-    /// [`Self::mark_completed`] from [`MetricsBody`]. When `false` at drop time, the metric is
-    /// tagged `499` instead of the response status.
-    completed: bool,
+    /// How body streaming ended. Updated via [`Self::mark_completed`] / [`Self::mark_errored`]
+    /// from [`MetricsBody`]; stays [`BodyOutcome::Pending`] if the body is dropped before it
+    /// finishes (client disconnect).
+    outcome: BodyOutcome,
 }
 
 impl EmitMetricsGuard {
@@ -166,7 +178,7 @@ impl EmitMetricsGuard {
             method: method.clone(),
             start: Instant::now(),
             status: None,
-            completed: false,
+            outcome: BodyOutcome::Pending,
         }
     }
 
@@ -177,23 +189,29 @@ impl EmitMetricsGuard {
 
     /// Marks the response body as having streamed to completion.
     ///
-    /// Called by [`MetricsBody`] when the inner body reaches end-of-stream. If this is never
-    /// called before the guard drops (client disconnect or a server-side stream error), the
-    /// metric is tagged `499`.
+    /// Called by [`MetricsBody`] when the inner body reaches end-of-stream.
     pub(crate) fn mark_completed(&mut self) {
-        self.completed = true;
+        self.outcome = BodyOutcome::Completed;
+    }
+
+    /// Marks the response body stream as having errored server-side.
+    ///
+    /// Called by [`MetricsBody`] when the inner body yields an error frame.
+    pub(crate) fn mark_errored(&mut self) {
+        self.outcome = BodyOutcome::Errored;
     }
 }
 
 impl Drop for EmitMetricsGuard {
     fn drop(&mut self) {
-        // A body that never reached end-of-stream is reported as `499` regardless of the
-        // status sent in the headers: the client disconnected mid-stream, or the body stream
-        // errored server-side. These two cases are intentionally conflated.
-        let status = if self.completed {
-            self.status.map(|s| s.as_u16()).unwrap_or(499)
-        } else {
-            499
+        // The status tag depends on how streaming ended. A body still `Pending` at drop time
+        // never reached end-of-stream, meaning the client disconnected mid-stream (or no
+        // response was produced), reported as `499`. A server-side stream error is reported
+        // as `500`. Both override the status sent in the headers.
+        let status = match self.outcome {
+            BodyOutcome::Completed => self.status.map(|s| s.as_u16()).unwrap_or(499),
+            BodyOutcome::Errored => 500,
+            BodyOutcome::Pending => 499,
         }
         .to_string();
         objectstore_metrics::record!(
@@ -385,9 +403,9 @@ mod tests {
                 assert_eq!(resp.status(), StatusCode::OK);
 
                 if consume_body {
-                    axum_body::to_bytes(resp.into_body(), usize::MAX)
-                        .await
-                        .unwrap();
+                    // Drain the body to end-of-stream (or until it errors); the result is
+                    // ignored so a server-side stream error does not panic the test.
+                    let _ = axum_body::to_bytes(resp.into_body(), usize::MAX).await;
                 } else {
                     // Drop the response (and its body) without reading it: the wrapping
                     // `MetricsBody` never reaches end-of-stream, mirroring a client disconnect.
@@ -424,4 +442,16 @@ mod tests {
         let metric = capture_duration_metric(Body::from_stream(stream), false);
         assert!(metric.contains("status:499"), "unexpected metric: {metric}");
     }
+
+    /// A body stream that errors server-side reports `500`, overriding the `200` status that
+    /// was sent in the headers.
+    #[test]
+    fn errored_stream_reports_500() {
+        let stream = async_stream::stream! {
+            yield Ok(axum::body::Bytes::from_static(b"hello"));
+            yield Err(std::io::Error::other("boom"));
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), true);
+        assert!(metric.contains("status:500"), "unexpected metric: {metric}");
+    }
 }

From c1d2a368b8376f8ced833b6265b6916a816808e5 Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Thu, 16 Jul 2026 13:32:37 +0200
Subject: [PATCH 3/7] fix(server): Handle all response body completion paths

Classify empty, buffered, streamed, and trailers-only responses as completed so request duration metrics retain the response status. Keep stream errors and interrupted bodies distinguishable, and simplify body wrapping with Response::map.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 objectstore-server/docs/architecture.md    |  14 --
 objectstore-server/src/web/metrics_body.rs | 125 ++++++++++++++--
 objectstore-server/src/web/middleware.rs   | 162 +++++++--------------
 3 files changed, 166 insertions(+), 135 deletions(-)

diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md
index 952365ab..18058444 100644
--- a/objectstore-server/docs/architecture.md
+++ b/objectstore-server/docs/architecture.md
@@ -76,20 +76,6 @@ A request flows through several layers before reaching the storage service:
    [`objectstore-types` docs](objectstore_types) for the header mapping) and
    the payload is streamed back.
 
-The `server.requests.duration` metric is measured **end-to-end**: the timing guard is
-moved into the response body, so it records the full request lifetime including the time
-spent streaming the response body back to the client (e.g. GET payloads, batch and
-multipart responses), not just the time to produce the response headers.
-
-The metric's `status` tag reflects the response status sent in the headers, **except** when
-the response body does not stream to completion:
-
-- **Client disconnect** mid-stream (the body is dropped before end-of-stream) is tagged
-  `499` (nginx' non-standard "client closed request").
-- **Server-side stream error** (the body yields an error while streaming) is tagged `500`.
-
-Both override the status sent in the headers.
-
 ## Authentication & Authorization
 
 Objectstore uses **JWT tokens with EdDSA signatures** (Ed25519) for
diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index f593ab2f..5da6d8b4 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -7,12 +7,18 @@
 //! it when the inner body reaches end-of-stream (or is dropped on client
 //! disconnect), at which point the guard emits `server.requests.duration`.
 //!
-//! When the body streams to completion, [`MetricsBody`] calls
+//! When the body finishes, [`MetricsBody`] calls
 //! [`EmitMetricsGuard::mark_completed`] so the metric is tagged with the real
-//! response status. A server-side stream error instead calls
+//! response status. Detecting completion is not as simple as watching for
+//! `Poll::Ready(None)`: hyper only polls a body while
+//! [`is_end_stream`](http_body::Body::is_end_stream) is false, so empty bodies
+//! are never polled and buffered (`Full`) bodies stop after their single data
+//! frame — neither ever yields `Ready(None)`.
+//! [`MetricsBody`] therefore also checks `is_end_stream` at construction and
+//! after each data frame. A server-side stream error instead calls
 //! [`EmitMetricsGuard::mark_errored`] and is reported as `500`. If the body is
-//! dropped before either happens — a client disconnect — the guard reports a
-//! `499` status.
+//! dropped before any of these happen — a client disconnect — the guard reports
+//! a `499` status.
 //!
 //! Because this delegates [`size_hint`](http_body::Body::size_hint) and
 //! [`is_end_stream`](http_body::Body::is_end_stream) to the inner body,
@@ -23,11 +29,88 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use axum::body::Body;
+use axum::http::{Method, StatusCode};
 use bytes::Bytes;
-use http_body::{Frame, SizeHint};
+use http_body::{Body as HttpBody, Frame, SizeHint};
 use pin_project_lite::pin_project;
+use tokio::time::Instant;
 
-use crate::web::middleware::EmitMetricsGuard;
+use crate::extractors::downstream_service::DownstreamService;
+
+/// How response-body streaming ended, which determines the `status` tag when it differs from
+/// the status sent in the headers.
+#[derive(Clone, Copy)]
+enum BodyOutcome {
+    /// The body was dropped before streaming completed, reported as `499`.
+    Pending,
+    /// The body streamed to completion, reported with the response status.
+    Completed,
+    /// The body yielded a server-side error, reported as `500`.
+    Errored,
+}
+
+/// Tracks request timing and emits `server.requests.duration` when dropped.
+///
+/// [`MetricsBody`] owns this guard so request duration spans full response-body streaming.
+pub(crate) struct EmitMetricsGuard {
+    route: String,
+    method: Method,
+    start: Instant,
+    status: Option<StatusCode>,
+    outcome: BodyOutcome,
+}
+
+impl EmitMetricsGuard {
+    pub(crate) fn new(route: &str, method: &Method, service: DownstreamService) -> Self {
+        objectstore_metrics::count!(
+            "server.requests",
+            route = route.to_owned(),
+            method = method.as_str().to_owned(),
+            service = service.to_string(),
+        );
+
+        Self {
+            route: route.to_owned(),
+            method: method.clone(),
+            start: Instant::now(),
+            status: None,
+            outcome: BodyOutcome::Pending,
+        }
+    }
+
+    /// Records the response status to use after successful body completion.
+    pub(crate) fn finish(&mut self, status: StatusCode) {
+        self.status = Some(status);
+    }
+
+    /// Marks the response body as having streamed to completion.
+    fn mark_completed(&mut self) {
+        self.outcome = BodyOutcome::Completed;
+    }
+
+    /// Marks the response body stream as having errored server-side.
+    fn mark_errored(&mut self) {
+        self.outcome = BodyOutcome::Errored;
+    }
+}
+
+impl Drop for EmitMetricsGuard {
+    fn drop(&mut self) {
+        let status = match self.outcome {
+            BodyOutcome::Completed => self.status.map(|status| status.as_u16()).unwrap_or(499),
+            BodyOutcome::Errored => 500,
+            BodyOutcome::Pending => 499,
+        }
+        .to_string();
+        objectstore_metrics::record!(
+            "server.requests.duration" = self.start.elapsed(),
+            route = self.route.clone(),
+            method = self.method.as_str().to_owned(),
+            status = status,
+            // service omitted to limit cardinality
+        );
+    }
+}
 
 pin_project! {
     /// Wraps an axum [`Body`] and holds an [`EmitMetricsGuard`] until the body ends.
@@ -45,7 +128,14 @@ pin_project! {
 
 impl MetricsBody {
     /// Creates a new [`MetricsBody`] that keeps `guard` alive while polling `inner`.
-    pub fn new(guard: EmitMetricsGuard, inner: Body) -> Self {
+    pub fn new(mut guard: EmitMetricsGuard, inner: Body) -> Self {
+        // A body that already reports end-of-stream (e.g. an empty body) may never be polled
+        // by hyper at all — it sets `body_rx = None` up front and writes a `Content-Length: 0`
+        // response without ever driving the body. Detect that here so the guard is not left
+        // `Pending` (which would misreport `499`).
+        if inner.is_end_stream() {
+            guard.mark_completed();
+        }
         Self { guard, inner }
     }
 }
@@ -58,15 +148,22 @@ impl http_body::Body for MetricsBody {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
-        let this = self.project();
-        let poll = this.inner.poll_frame(cx);
-        // `Ready(None)` is the only end-of-stream signal that is uniform across buffered and
-        // streamed bodies (`StreamBody` does not override `is_end_stream`). Reaching it means
-        // the body streamed to completion. A `Ready(Some(Err))` frame is a server-side stream
-        // error. If neither is observed before the body is dropped (client disconnect), the
-        // guard stays pending and reports `499`.
+        let mut this = self.project();
+        let poll = this.inner.as_mut().poll_frame(cx);
+        // Detect completion so the guard reports the real status instead of `499`:
+        // - `Ready(None)` is the end-of-stream signal for ordinary streamed bodies.
+        // - A buffered body (`Full`) yields its single data frame and then reports
+        //   `is_end_stream()`; hyper stops polling after that frame without ever returning
+        //   `Ready(None)`, so we must check `is_end_stream()` after each frame too.
+        // - Hyper treats trailers as terminal and drops the body without polling again.
+        // A `Ready(Some(Err))` frame is a server-side stream error, reported as `500`. If none
+        // of these is observed before the body is dropped (client disconnect), the guard stays
+        // pending and reports `499`.
         match &poll {
             Poll::Ready(None) => this.guard.mark_completed(),
+            Poll::Ready(Some(Ok(frame))) if frame.is_trailers() || this.inner.is_end_stream() => {
+                this.guard.mark_completed()
+            }
             Poll::Ready(Some(Err(_))) => this.guard.mark_errored(),
             _ => {}
         }
diff --git a/objectstore-server/src/web/middleware.rs b/objectstore-server/src/web/middleware.rs
index d142f677..88938d21 100644
--- a/objectstore-server/src/web/middleware.rs
+++ b/objectstore-server/src/web/middleware.rs
@@ -4,17 +4,16 @@ use std::net::SocketAddr;
 use axum::RequestExt;
 use axum::body::Body;
 use axum::extract::{ConnectInfo, MatchedPath, Request, State};
-use axum::http::{HeaderValue, Method, StatusCode, header};
+use axum::http::{HeaderValue, StatusCode, header};
 use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 use objectstore_log::tracing;
-use tokio::time::Instant;
 use tower_http::set_header::SetResponseHeaderLayer;
 
 use crate::endpoints::is_internal_route;
 use crate::extractors::downstream_service::DownstreamService;
 use crate::web::RequestCounter;
-use crate::web::metrics_body::MetricsBody;
+use crate::web::metrics_body::{EmitMetricsGuard, MetricsBody};
 use crate::web::sentry_body::SentryBody;
 
 /// The value for the `Server` HTTP header.
@@ -91,10 +90,9 @@ pub fn handle_panic(err: Box<dyn Any + Send + 'static>) -> Response {
 /// tower layers so that `Hub::current()` returns the request-scoped hub.
 pub async fn bind_sentry_body(request: Request, next: Next) -> Response {
     let hub = sentry::Hub::current();
-    let response = next.run(request).await;
-    let (parts, body) = response.into_parts();
-    let sentry_body = Body::new(SentryBody::new(hub, body));
-    Response::from_parts(parts, sentry_body)
+    next.run(request)
+        .await
+        .map(|body| Body::new(SentryBody::new(hub, body)))
 }
 
 /// A middleware that logs web request timings as metrics.
@@ -121,119 +119,26 @@ pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response
     match guard {
         Some(mut guard) => {
             guard.finish(response.status());
-            let (parts, body) = response.into_parts();
-            let body = Body::new(MetricsBody::new(guard, body));
-            Response::from_parts(parts, body)
+            response.map(|body| Body::new(MetricsBody::new(guard, body)))
         }
         None => response,
     }
 }
 
-/// How the response body streaming ended, which determines the `status` tag on the emitted
-/// metric when it differs from the status sent in the headers.
-#[derive(Clone, Copy)]
-enum BodyOutcome {
-    /// The body has not finished streaming. At drop time this means the client disconnected
-    /// mid-stream (or no response was produced at all), reported as `499`.
-    Pending,
-    /// The body streamed to completion; the real response status is reported.
-    Completed,
-    /// The body stream errored server-side, reported as `500`.
-    Errored,
-}
-
-/// Helper for [`emit_request_metrics`].
-///
-/// This tracks relevant generic request parameters and emits the `server.requests.duration`
-/// metric on drop. It is moved into the response body ([`MetricsBody`]) so timing spans the
-/// full response, including body streaming.
-///
-/// The `status` tag reflects how streaming ended (see [`BodyOutcome`]):
-/// - streamed to completion: the response status set via [`Self::finish`];
-/// - client disconnect mid-stream, or no response produced at all: `499` (nginx' non-standard
-///   "client closed request");
-/// - server-side stream error: `500`.
-pub struct EmitMetricsGuard {
-    route: String,
-    method: Method,
-    start: Instant,
-    status: Option<StatusCode>,
-    /// How body streaming ended. Updated via [`Self::mark_completed`] / [`Self::mark_errored`]
-    /// from [`MetricsBody`]; stays [`BodyOutcome::Pending`] if the body is dropped before it
-    /// finishes (client disconnect).
-    outcome: BodyOutcome,
-}
-
-impl EmitMetricsGuard {
-    fn new(route: &str, method: &Method, service: DownstreamService) -> Self {
-        objectstore_metrics::count!(
-            "server.requests",
-            route = route.to_owned(),
-            method = method.as_str().to_owned(),
-            service = service.to_string(),
-        );
-
-        Self {
-            route: route.to_owned(),
-            method: method.clone(),
-            start: Instant::now(),
-            status: None,
-            outcome: BodyOutcome::Pending,
-        }
-    }
-
-    /// Records the response status to tag the emitted metric with.
-    fn finish(&mut self, status: StatusCode) {
-        self.status = Some(status);
-    }
-
-    /// Marks the response body as having streamed to completion.
-    ///
-    /// Called by [`MetricsBody`] when the inner body reaches end-of-stream.
-    pub(crate) fn mark_completed(&mut self) {
-        self.outcome = BodyOutcome::Completed;
-    }
-
-    /// Marks the response body stream as having errored server-side.
-    ///
-    /// Called by [`MetricsBody`] when the inner body yields an error frame.
-    pub(crate) fn mark_errored(&mut self) {
-        self.outcome = BodyOutcome::Errored;
-    }
-}
-
-impl Drop for EmitMetricsGuard {
-    fn drop(&mut self) {
-        // The status tag depends on how streaming ended. A body still `Pending` at drop time
-        // never reached end-of-stream, meaning the client disconnected mid-stream (or no
-        // response was produced), reported as `499`. A server-side stream error is reported
-        // as `500`. Both override the status sent in the headers.
-        let status = match self.outcome {
-            BodyOutcome::Completed => self.status.map(|s| s.as_u16()).unwrap_or(499),
-            BodyOutcome::Errored => 500,
-            BodyOutcome::Pending => 499,
-        }
-        .to_string();
-        objectstore_metrics::record!(
-            "server.requests.duration" = self.start.elapsed(),
-            route = self.route.clone(),
-            method = self.method.as_str().to_owned(),
-            status = status,
-            // service omitted to limit cardinality
-        );
-    }
-}
-
 #[cfg(test)]
 mod tests {
+    use std::convert::Infallible;
+    use std::pin::Pin;
     use std::sync::Arc;
+    use std::task::{Context, Poll};
 
     use axum::Router;
-    use axum::body::Body;
-    use axum::http::{Request, StatusCode};
+    use axum::body::{Body, Bytes};
+    use axum::http::{HeaderMap, Request, StatusCode};
     use axum::middleware::from_fn_with_state;
     use axum::response::IntoResponse;
     use axum::routing::get;
+    use http_body::Frame;
     use tokio::sync::Notify;
     use tower::ServiceExt;
 
@@ -430,6 +335,49 @@ mod tests {
         assert!(metric.contains("status:200"), "unexpected metric: {metric}");
     }
 
+    /// A body that needs no polling is completed when it is wrapped, because hyper may not poll
+    /// it before dropping it.
+    #[test]
+    fn empty_body_reports_real_status() {
+        let metric = capture_duration_metric(Body::empty(), false);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// A buffered body completes after its final data frame, without requiring a subsequent poll
+    /// that returns `None`.
+    #[test]
+    fn buffered_body_reports_real_status() {
+        let metric = capture_duration_metric(Body::from("hello"), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// Hyper treats trailers as terminal, so completion must be recorded when their frame is
+    /// yielded rather than waiting for an unobserved following `None`.
+    #[test]
+    fn trailers_report_real_status() {
+        struct TrailersBody(bool);
+
+        impl http_body::Body for TrailersBody {
+            type Data = Bytes;
+            type Error = Infallible;
+
+            fn poll_frame(
+                mut self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+            ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+                Poll::Ready(if self.0 {
+                    None
+                } else {
+                    self.0 = true;
+                    Some(Ok(Frame::trailers(HeaderMap::new())))
+                })
+            }
+        }
+
+        let metric = capture_duration_metric(Body::new(TrailersBody(false)), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
     /// A body dropped before end-of-stream (client disconnect) reports `499`, overriding the
     /// `200` status that was sent in the headers.
     #[test]

From 8c8a7ee4d85a1c8fea7ef6c1071e5c1cafbfde97 Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Thu, 16 Jul 2026 15:01:09 +0200
Subject: [PATCH 4/7] test(server): Co-locate metrics body lifecycle tests

Keep response-body outcome coverage next to the MetricsBody state machine while retaining middleware timing coverage with its middleware tests.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 objectstore-server/src/web/metrics_body.rs | 144 +++++++++++++++++++++
 objectstore-server/src/web/middleware.rs   | 133 +------------------
 2 files changed, 146 insertions(+), 131 deletions(-)

diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index 5da6d8b4..1c50767e 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -178,3 +178,147 @@ impl http_body::Body for MetricsBody {
         self.inner.is_end_stream()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::convert::Infallible;
+    use std::pin::Pin;
+    use std::sync::Arc;
+    use std::task::{Context, Poll};
+
+    use axum::Router;
+    use axum::body::{self, Body, Bytes};
+    use axum::http::{HeaderMap, Request, StatusCode};
+    use axum::middleware::from_fn;
+    use axum::routing::get;
+    use http_body::Frame;
+    use tower::ServiceExt;
+
+    use crate::web::middleware::emit_request_metrics;
+
+    fn make_request(uri: &str) -> Request<Body> {
+        Request::builder().uri(uri).body(Body::empty()).unwrap()
+    }
+
+    /// Runs a request whose handler returns `body`, driving it to completion on a
+    /// current-thread runtime, and returns the captured `server.requests.duration` metric
+    /// string. If `consume_body` is false, the response body is dropped without being read,
+    /// simulating a client that disconnects mid-stream.
+    fn capture_duration_metric(body: Body, consume_body: bool) -> String {
+        let captured = objectstore_metrics::with_capturing_test_client(|| {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+
+            rt.block_on(async {
+                let shared = Arc::new(std::sync::Mutex::new(Some(body)));
+                let app = Router::new()
+                    .route(
+                        "/v1/test/_/key",
+                        get(move || {
+                            let shared = shared.clone();
+                            async move { shared.lock().unwrap().take().unwrap() }
+                        }),
+                    )
+                    .layer(from_fn(emit_request_metrics));
+
+                let resp = app.oneshot(make_request("/v1/test/_/key")).await.unwrap();
+                assert_eq!(resp.status(), StatusCode::OK);
+
+                if consume_body {
+                    // Drain the body to end-of-stream (or until it errors); the result is
+                    // ignored so a server-side stream error does not panic the test.
+                    let _ = body::to_bytes(resp.into_body(), usize::MAX).await;
+                } else {
+                    // Drop the response (and its body) without reading it: the wrapping
+                    // `MetricsBody` never reaches end-of-stream, mirroring a client disconnect.
+                    drop(resp);
+                }
+            });
+        });
+
+        captured
+            .into_iter()
+            .find(|m| m.starts_with("server.requests.duration:"))
+            .expect("duration metric not captured")
+    }
+
+    /// A body streamed to completion reports the real response status.
+    #[test]
+    fn completed_stream_reports_real_status() {
+        let stream = async_stream::stream! {
+            yield Ok::<_, std::io::Error>(Bytes::from_static(b"hello"));
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// A body that needs no polling is completed when it is wrapped, because hyper may not poll
+    /// it before dropping it.
+    #[test]
+    fn empty_body_reports_real_status() {
+        let metric = capture_duration_metric(Body::empty(), false);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// A buffered body completes after its final data frame, without requiring a subsequent poll
+    /// that returns `None`.
+    #[test]
+    fn buffered_body_reports_real_status() {
+        let metric = capture_duration_metric(Body::from("hello"), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// Hyper treats trailers as terminal, so completion must be recorded when their frame is
+    /// yielded rather than waiting for an unobserved following `None`.
+    #[test]
+    fn trailers_report_real_status() {
+        struct TrailersBody(bool);
+
+        impl http_body::Body for TrailersBody {
+            type Data = Bytes;
+            type Error = Infallible;
+
+            fn poll_frame(
+                mut self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+            ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+                Poll::Ready(if self.0 {
+                    None
+                } else {
+                    self.0 = true;
+                    Some(Ok(Frame::trailers(HeaderMap::new())))
+                })
+            }
+        }
+
+        let metric = capture_duration_metric(Body::new(TrailersBody(false)), true);
+        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
+    }
+
+    /// A body dropped before end-of-stream (client disconnect) reports `499`, overriding the
+    /// `200` status that was sent in the headers.
+    #[test]
+    fn interrupted_stream_reports_499() {
+        // A stream that yields one chunk then pends forever, so it never reaches end-of-stream.
+        let stream = async_stream::stream! {
+            yield Ok::<_, std::io::Error>(Bytes::from_static(b"hello"));
+            std::future::pending::<()>().await;
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), false);
+        assert!(metric.contains("status:499"), "unexpected metric: {metric}");
+    }
+
+    /// A body stream that errors server-side reports `500`, overriding the `200` status that
+    /// was sent in the headers.
+    #[test]
+    fn errored_stream_reports_500() {
+        let stream = async_stream::stream! {
+            yield Ok(Bytes::from_static(b"hello"));
+            yield Err(std::io::Error::other("boom"));
+        };
+        let metric = capture_duration_metric(Body::from_stream(stream), true);
+        assert!(metric.contains("status:500"), "unexpected metric: {metric}");
+    }
+}
diff --git a/objectstore-server/src/web/middleware.rs b/objectstore-server/src/web/middleware.rs
index 88938d21..f37b6348 100644
--- a/objectstore-server/src/web/middleware.rs
+++ b/objectstore-server/src/web/middleware.rs
@@ -127,18 +127,14 @@ pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response
 
 #[cfg(test)]
 mod tests {
-    use std::convert::Infallible;
-    use std::pin::Pin;
     use std::sync::Arc;
-    use std::task::{Context, Poll};
 
     use axum::Router;
-    use axum::body::{Body, Bytes};
-    use axum::http::{HeaderMap, Request, StatusCode};
+    use axum::body::Body;
+    use axum::http::{Request, StatusCode};
     use axum::middleware::from_fn_with_state;
     use axum::response::IntoResponse;
     use axum::routing::get;
-    use http_body::Frame;
     use tokio::sync::Notify;
     use tower::ServiceExt;
 
@@ -277,129 +273,4 @@ mod tests {
             "duration metric emitted before body was consumed: {captured:?}"
         );
     }
-
-    /// Runs a request whose handler returns `body`, driving it to completion on a
-    /// current-thread runtime, and returns the captured `server.requests.duration` metric
-    /// string. If `consume_body` is false, the response body is dropped without being read,
-    /// simulating a client that disconnects mid-stream.
-    fn capture_duration_metric(body: Body, consume_body: bool) -> String {
-        use axum::body as axum_body;
-        use axum::middleware::from_fn;
-
-        let captured = objectstore_metrics::with_capturing_test_client(|| {
-            let rt = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .unwrap();
-
-            rt.block_on(async {
-                let shared = Arc::new(std::sync::Mutex::new(Some(body)));
-                let app = Router::new()
-                    .route(
-                        "/v1/test/_/key",
-                        get(move || {
-                            let shared = shared.clone();
-                            async move { shared.lock().unwrap().take().unwrap() }
-                        }),
-                    )
-                    .layer(from_fn(emit_request_metrics));
-
-                let resp = app.oneshot(make_request("/v1/test/_/key")).await.unwrap();
-                assert_eq!(resp.status(), StatusCode::OK);
-
-                if consume_body {
-                    // Drain the body to end-of-stream (or until it errors); the result is
-                    // ignored so a server-side stream error does not panic the test.
-                    let _ = axum_body::to_bytes(resp.into_body(), usize::MAX).await;
-                } else {
-                    // Drop the response (and its body) without reading it: the wrapping
-                    // `MetricsBody` never reaches end-of-stream, mirroring a client disconnect.
-                    drop(resp);
-                }
-            });
-        });
-
-        captured
-            .into_iter()
-            .find(|m| m.starts_with("server.requests.duration:"))
-            .expect("duration metric not captured")
-    }
-
-    /// A body streamed to completion reports the real response status.
-    #[test]
-    fn completed_stream_reports_real_status() {
-        let stream = async_stream::stream! {
-            yield Ok::<_, std::io::Error>(axum::body::Bytes::from_static(b"hello"));
-        };
-        let metric = capture_duration_metric(Body::from_stream(stream), true);
-        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
-    }
-
-    /// A body that needs no polling is completed when it is wrapped, because hyper may not poll
-    /// it before dropping it.
-    #[test]
-    fn empty_body_reports_real_status() {
-        let metric = capture_duration_metric(Body::empty(), false);
-        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
-    }
-
-    /// A buffered body completes after its final data frame, without requiring a subsequent poll
-    /// that returns `None`.
-    #[test]
-    fn buffered_body_reports_real_status() {
-        let metric = capture_duration_metric(Body::from("hello"), true);
-        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
-    }
-
-    /// Hyper treats trailers as terminal, so completion must be recorded when their frame is
-    /// yielded rather than waiting for an unobserved following `None`.
-    #[test]
-    fn trailers_report_real_status() {
-        struct TrailersBody(bool);
-
-        impl http_body::Body for TrailersBody {
-            type Data = Bytes;
-            type Error = Infallible;
-
-            fn poll_frame(
-                mut self: Pin<&mut Self>,
-                _cx: &mut Context<'_>,
-            ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
-                Poll::Ready(if self.0 {
-                    None
-                } else {
-                    self.0 = true;
-                    Some(Ok(Frame::trailers(HeaderMap::new())))
-                })
-            }
-        }
-
-        let metric = capture_duration_metric(Body::new(TrailersBody(false)), true);
-        assert!(metric.contains("status:200"), "unexpected metric: {metric}");
-    }
-
-    /// A body dropped before end-of-stream (client disconnect) reports `499`, overriding the
-    /// `200` status that was sent in the headers.
-    #[test]
-    fn interrupted_stream_reports_499() {
-        // A stream that yields one chunk then pends forever, so it never reaches end-of-stream.
-        let stream = async_stream::stream! {
-            yield Ok::<_, std::io::Error>(axum::body::Bytes::from_static(b"hello"));
-            std::future::pending::<()>().await;
-        };
-        let metric = capture_duration_metric(Body::from_stream(stream), false);
-        assert!(metric.contains("status:499"), "unexpected metric: {metric}");
-    }
-
-    /// A body stream that errors server-side reports `500`, overriding the `200` status that
-    /// was sent in the headers.
-    #[test]
-    fn errored_stream_reports_500() {
-        let stream = async_stream::stream! {
-            yield Ok(axum::body::Bytes::from_static(b"hello"));
-            yield Err(std::io::Error::other("boom"));
-        };
-        let metric = capture_duration_metric(Body::from_stream(stream), true);
-        assert!(metric.contains("status:500"), "unexpected metric: {metric}");
-    }
 }

From f8d073b6bb60443dfbc40794660c40b95a2e08a5 Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Fri, 24 Jul 2026 14:04:16 +0200
Subject: [PATCH 5/7] wip

---
 objectstore-server/src/web/metrics_body.rs | 115 +++++++--------------
 objectstore-server/src/web/middleware.rs   |  10 +-
 2 files changed, 44 insertions(+), 81 deletions(-)

diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index 1c50767e..a8a6f81f 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -1,29 +1,4 @@
-//! Response body wrapper that keeps a metrics guard alive until the body ends.
-//!
-//! Axum handlers produce response *headers* while the middleware stack unwinds,
-//! but the response *body* is streamed by hyper afterwards. To measure the true
-//! end-to-end request duration, the timing guard must live until the body has
-//! been fully streamed. [`MetricsBody`] owns an [`EmitMetricsGuard`] and drops
-//! it when the inner body reaches end-of-stream (or is dropped on client
-//! disconnect), at which point the guard emits `server.requests.duration`.
-//!
-//! When the body finishes, [`MetricsBody`] calls
-//! [`EmitMetricsGuard::mark_completed`] so the metric is tagged with the real
-//! response status. Detecting completion is not as simple as watching for
-//! `Poll::Ready(None)`: hyper only polls a body while
-//! [`is_end_stream`](http_body::Body::is_end_stream) is false, so empty bodies
-//! are never polled and buffered (`Full`) bodies stop after their single data
-//! frame — neither ever yields `Ready(None)`.
-//! [`MetricsBody`] therefore also checks `is_end_stream` at construction and
-//! after each data frame. A server-side stream error instead calls
-//! [`EmitMetricsGuard::mark_errored`] and is reported as `500`. If the body is
-//! dropped before any of these happen — a client disconnect — the guard reports
-//! a `499` status.
-//!
-//! Because this delegates [`size_hint`](http_body::Body::size_hint) and
-//! [`is_end_stream`](http_body::Body::is_end_stream) to the inner body,
-//! buffered responses keep their exact size hint and hyper still emits a
-//! `Content-Length` header instead of chunked encoding.
+//! Response body wrapper that emits request-duration metrics after the body finishes.
 
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -37,15 +12,14 @@ use tokio::time::Instant;
 
 use crate::extractors::downstream_service::DownstreamService;
 
-/// How response-body streaming ended, which determines the `status` tag when it differs from
-/// the status sent in the headers.
+/// How response-body streaming ended, which determines the metric `status` tag.
 #[derive(Clone, Copy)]
-enum BodyOutcome {
-    /// The body was dropped before streaming completed, reported as `499`.
+enum BodyStatus {
+    /// The body was still streaming.
     Pending,
-    /// The body streamed to completion, reported with the response status.
-    Completed,
-    /// The body yielded a server-side error, reported as `500`.
+    /// The body streamed to completion.
+    Completed(StatusCode),
+    /// The body yielded a server-side error.
     Errored,
 }
 
@@ -56,8 +30,7 @@ pub(crate) struct EmitMetricsGuard {
     route: String,
     method: Method,
     start: Instant,
-    status: Option<StatusCode>,
-    outcome: BodyOutcome,
+    status: BodyStatus,
 }
 
 impl EmitMetricsGuard {
@@ -73,40 +46,28 @@ impl EmitMetricsGuard {
             route: route.to_owned(),
             method: method.clone(),
             start: Instant::now(),
-            status: None,
-            outcome: BodyOutcome::Pending,
+            status: BodyStatus::Pending,
         }
     }
 
-    /// Records the response status to use after successful body completion.
-    pub(crate) fn finish(&mut self, status: StatusCode) {
-        self.status = Some(status);
-    }
-
-    /// Marks the response body as having streamed to completion.
-    fn mark_completed(&mut self) {
-        self.outcome = BodyOutcome::Completed;
-    }
-
-    /// Marks the response body stream as having errored server-side.
-    fn mark_errored(&mut self) {
-        self.outcome = BodyOutcome::Errored;
+    fn set_status(&mut self, status: BodyStatus) {
+        self.status = status;
     }
 }
 
 impl Drop for EmitMetricsGuard {
     fn drop(&mut self) {
-        let status = match self.outcome {
-            BodyOutcome::Completed => self.status.map(|status| status.as_u16()).unwrap_or(499),
-            BodyOutcome::Errored => 500,
-            BodyOutcome::Pending => 499,
-        }
-        .to_string();
+        let status = match self.status {
+            BodyStatus::Completed(status) => status.as_u16(),
+            BodyStatus::Errored => 500,
+            // `Pending` at drop time means the body was cancelled mid-stream (client disconnect).
+            BodyStatus::Pending => 499,
+        };
         objectstore_metrics::record!(
             "server.requests.duration" = self.start.elapsed(),
             route = self.route.clone(),
             method = self.method.as_str().to_owned(),
-            status = status,
+            status = status.to_string(),
             // service omitted to limit cardinality
         );
     }
@@ -121,6 +82,8 @@ pin_project! {
         // Dropped together with the body once streaming finishes; its `Drop`
         // emits the request-duration metric.
         guard: EmitMetricsGuard,
+        // Response status from headers, applied when the body completes successfully.
+        status: StatusCode,
         #[pin]
         inner: Body,
     }
@@ -128,15 +91,19 @@ pin_project! {
 
 impl MetricsBody {
     /// Creates a new [`MetricsBody`] that keeps `guard` alive while polling `inner`.
-    pub fn new(mut guard: EmitMetricsGuard, inner: Body) -> Self {
-        // A body that already reports end-of-stream (e.g. an empty body) may never be polled
-        // by hyper at all — it sets `body_rx = None` up front and writes a `Content-Length: 0`
-        // response without ever driving the body. Detect that here so the guard is not left
-        // `Pending` (which would misreport `499`).
+    ///
+    /// `status` is the response status from headers.
+    pub fn new(mut guard: EmitMetricsGuard, status: StatusCode, inner: Body) -> Self {
+        // An empty response body reports end-of-stream immediately and is never polled by hyper,
+        // so we must check and mark it as completed immediately.
         if inner.is_end_stream() {
-            guard.mark_completed();
+            guard.set_status(BodyStatus::Completed(status));
+        }
+        Self {
+            guard,
+            status,
+            inner,
         }
-        Self { guard, inner }
     }
 }
 
@@ -150,27 +117,23 @@ impl http_body::Body for MetricsBody {
     ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
         let mut this = self.project();
         let poll = this.inner.as_mut().poll_frame(cx);
-        // Detect completion so the guard reports the real status instead of `499`:
-        // - `Ready(None)` is the end-of-stream signal for ordinary streamed bodies.
-        // - A buffered body (`Full`) yields its single data frame and then reports
-        //   `is_end_stream()`; hyper stops polling after that frame without ever returning
-        //   `Ready(None)`, so we must check `is_end_stream()` after each frame too.
-        // - Hyper treats trailers as terminal and drops the body without polling again.
-        // A `Ready(Some(Err))` frame is a server-side stream error, reported as `500`. If none
-        // of these is observed before the body is dropped (client disconnect), the guard stays
-        // pending and reports `499`.
         match &poll {
-            Poll::Ready(None) => this.guard.mark_completed(),
+            // End-of-stream for a streamed body.
+            Poll::Ready(None) => this.guard.set_status(BodyStatus::Completed(*this.status)),
+            // End-of-stream for a buffered body (yields a single frame and reports end-of-stream
+            // immediatley, so hyper doesn't attempt to poll it again).
             Poll::Ready(Some(Ok(frame))) if frame.is_trailers() || this.inner.is_end_stream() => {
-                this.guard.mark_completed()
+                this.guard.set_status(BodyStatus::Completed(*this.status))
             }
-            Poll::Ready(Some(Err(_))) => this.guard.mark_errored(),
+            Poll::Ready(Some(Err(_))) => this.guard.set_status(BodyStatus::Errored),
             _ => {}
         }
         poll
     }
 
     fn size_hint(&self) -> SizeHint {
+        // Delegating [`size_hint`](http_body::Body::size_hint) preserves `Content-Length` on buffered
+        // responses instead of forcing chunked encoding.
         self.inner.size_hint()
     }
 
diff --git a/objectstore-server/src/web/middleware.rs b/objectstore-server/src/web/middleware.rs
index f37b6348..1d037612 100644
--- a/objectstore-server/src/web/middleware.rs
+++ b/objectstore-server/src/web/middleware.rs
@@ -114,12 +114,12 @@ pub async fn emit_request_metrics(mut request: Request, next: Next) -> Response
 
     let response = next.run(request).await;
 
-    // Record the actual response status, then move the guard into the response body so the
-    // duration metric is emitted only when the body has finished streaming.
+    // Move the guard into the response body so the duration metric is emitted only when the
+    // body has finished streaming. The header status is applied on successful completion.
     match guard {
-        Some(mut guard) => {
-            guard.finish(response.status());
-            response.map(|body| Body::new(MetricsBody::new(guard, body)))
+        Some(guard) => {
+            let status = response.status();
+            response.map(|body| Body::new(MetricsBody::new(guard, status, body)))
         }
         None => response,
     }

From cc190f95bfb1248f2af8e3c372680520eae0bd82 Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Fri, 24 Jul 2026 14:21:29 +0200
Subject: [PATCH 6/7] wip

---
 objectstore-server/src/web/metrics_body.rs | 61 +++++++++++-----------
 1 file changed, 30 insertions(+), 31 deletions(-)

diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index a8a6f81f..6a351805 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -12,14 +12,14 @@ use tokio::time::Instant;
 
 use crate::extractors::downstream_service::DownstreamService;
 
-/// How response-body streaming ended, which determines the metric `status` tag.
+/// State of a response body.
 #[derive(Clone, Copy)]
-enum BodyStatus {
-    /// The body was still streaming.
-    Pending,
-    /// The body streamed to completion.
+enum BodyState {
+    /// The body is still streaming.
+    Streaming(StatusCode),
+    /// The body was streamed to completion.
     Completed(StatusCode),
-    /// The body yielded a server-side error.
+    /// An error was encountered while streaming the body.
     Errored,
 }
 
@@ -30,7 +30,7 @@ pub(crate) struct EmitMetricsGuard {
     route: String,
     method: Method,
     start: Instant,
-    status: BodyStatus,
+    body_state: Option<BodyState>,
 }
 
 impl EmitMetricsGuard {
@@ -46,28 +46,27 @@ impl EmitMetricsGuard {
             route: route.to_owned(),
             method: method.clone(),
             start: Instant::now(),
-            status: BodyStatus::Pending,
+            body_state: None,
         }
     }
 
-    fn set_status(&mut self, status: BodyStatus) {
-        self.status = status;
+    fn set_body_state(&mut self, state: BodyState) {
+        self.body_state = Some(state);
     }
 }
 
 impl Drop for EmitMetricsGuard {
     fn drop(&mut self) {
-        let status = match self.status {
-            BodyStatus::Completed(status) => status.as_u16(),
-            BodyStatus::Errored => 500,
-            // `Pending` at drop time means the body was cancelled mid-stream (client disconnect).
-            BodyStatus::Pending => 499,
+        let state = match self.body_state {
+            Some(BodyState::Completed(status)) => status.as_u16(),
+            Some(BodyState::Streaming(_)) | None => 499,
+            Some(BodyState::Errored) => 500,
         };
         objectstore_metrics::record!(
             "server.requests.duration" = self.start.elapsed(),
             route = self.route.clone(),
             method = self.method.as_str().to_owned(),
-            status = status.to_string(),
+            status = state.to_string(),
             // service omitted to limit cardinality
         );
     }
@@ -79,11 +78,7 @@ pin_project! {
     /// The guard emits the request-duration metric on drop, so keeping it inside
     /// the body defers that emission until streaming completes.
     pub struct MetricsBody {
-        // Dropped together with the body once streaming finishes; its `Drop`
-        // emits the request-duration metric.
         guard: EmitMetricsGuard,
-        // Response status from headers, applied when the body completes successfully.
-        status: StatusCode,
         #[pin]
         inner: Body,
     }
@@ -95,15 +90,13 @@ impl MetricsBody {
     /// `status` is the response status from headers.
     pub fn new(mut guard: EmitMetricsGuard, status: StatusCode, inner: Body) -> Self {
         // An empty response body reports end-of-stream immediately and is never polled by hyper,
-        // so we must check and mark it as completed immediately.
+        // so mark it completed up front.
         if inner.is_end_stream() {
-            guard.set_status(BodyStatus::Completed(status));
-        }
-        Self {
-            guard,
-            status,
-            inner,
+            guard.set_body_state(BodyState::Completed(status));
+        } else {
+            guard.set_body_state(BodyState::Streaming(status));
         }
+        Self { guard, inner }
     }
 }
 
@@ -119,13 +112,19 @@ impl http_body::Body for MetricsBody {
         let poll = this.inner.as_mut().poll_frame(cx);
         match &poll {
             // End-of-stream for a streamed body.
-            Poll::Ready(None) => this.guard.set_status(BodyStatus::Completed(*this.status)),
+            Poll::Ready(None) => {
+                if let Some(BodyState::Streaming(status)) = this.guard.body_state {
+                    this.guard.set_body_state(BodyState::Completed(status));
+                }
+            }
             // End-of-stream for a buffered body (yields a single frame and reports end-of-stream
-            // immediatley, so hyper doesn't attempt to poll it again).
+            // immediately, so hyper doesn't attempt to poll it again).
             Poll::Ready(Some(Ok(frame))) if frame.is_trailers() || this.inner.is_end_stream() => {
-                this.guard.set_status(BodyStatus::Completed(*this.status))
+                if let Some(BodyState::Streaming(status)) = this.guard.body_state {
+                    this.guard.set_body_state(BodyState::Completed(status));
+                }
             }
-            Poll::Ready(Some(Err(_))) => this.guard.set_status(BodyStatus::Errored),
+            Poll::Ready(Some(Err(_))) => this.guard.set_body_state(BodyState::Errored),
             _ => {}
         }
         poll

From 88bf634f95c649fb87bdef556a397dc50e84d777 Mon Sep 17 00:00:00 2001
From: lcian <17258265+lcian@users.noreply.github.com>
Date: Fri, 24 Jul 2026 14:23:06 +0200
Subject: [PATCH 7/7] wip

---
 objectstore-server/src/web/metrics_body.rs | 31 ++++++++++++----------
 1 file changed, 17 insertions(+), 14 deletions(-)

diff --git a/objectstore-server/src/web/metrics_body.rs b/objectstore-server/src/web/metrics_body.rs
index 6a351805..29f39554 100644
--- a/objectstore-server/src/web/metrics_body.rs
+++ b/objectstore-server/src/web/metrics_body.rs
@@ -50,8 +50,18 @@ impl EmitMetricsGuard {
         }
     }
 
-    fn set_body_state(&mut self, state: BodyState) {
-        self.body_state = Some(state);
+    fn mark_streaming(&mut self, status: StatusCode) {
+        self.body_state = Some(BodyState::Streaming(status));
+    }
+
+    fn mark_completed(&mut self) {
+        if let Some(BodyState::Streaming(status)) = self.body_state {
+            self.body_state = Some(BodyState::Completed(status));
+        }
+    }
+
+    fn mark_errored(&mut self) {
+        self.body_state = Some(BodyState::Errored);
     }
 }
 
@@ -89,12 +99,11 @@ impl MetricsBody {
     ///
     /// `status` is the response status from headers.
     pub fn new(mut guard: EmitMetricsGuard, status: StatusCode, inner: Body) -> Self {
+        guard.mark_streaming(status);
         // An empty response body reports end-of-stream immediately and is never polled by hyper,
         // so mark it completed up front.
         if inner.is_end_stream() {
-            guard.set_body_state(BodyState::Completed(status));
-        } else {
-            guard.set_body_state(BodyState::Streaming(status));
+            guard.mark_completed();
         }
         Self { guard, inner }
     }
@@ -112,19 +121,13 @@ impl http_body::Body for MetricsBody {
         let poll = this.inner.as_mut().poll_frame(cx);
         match &poll {
             // End-of-stream for a streamed body.
-            Poll::Ready(None) => {
-                if let Some(BodyState::Streaming(status)) = this.guard.body_state {
-                    this.guard.set_body_state(BodyState::Completed(status));
-                }
-            }
+            Poll::Ready(None) => this.guard.mark_completed(),
             // End-of-stream for a buffered body (yields a single frame and reports end-of-stream
             // immediately, so hyper doesn't attempt to poll it again).
             Poll::Ready(Some(Ok(frame))) if frame.is_trailers() || this.inner.is_end_stream() => {
-                if let Some(BodyState::Streaming(status)) = this.guard.body_state {
-                    this.guard.set_body_state(BodyState::Completed(status));
-                }
+                this.guard.mark_completed()
             }
-            Poll::Ready(Some(Err(_))) => this.guard.set_body_state(BodyState::Errored),
+            Poll::Ready(Some(Err(_))) => this.guard.mark_errored(),
             _ => {}
         }
         poll