fix(server): Track status for content-length terminated bodies (#580)

Streaming responses that declare a `content-length` were recorded as
client disconnects (499) in `server.requests.duration`, a regression
from #555. Hyper stops polling a response body once the declared number
of bytes has been written, so the wrapper never saw an end-of-stream
poll and its guard fell back to 499 when dropped.

`MetricsBody` now counts the bytes it forwards and completes the request
once the declared length is reached, alongside the existing
end-of-stream and trailers cases. Responses whose body hyper never polls
at all are completed up front when they are wrapped.

The tests were reworked around a helper that serves a real axum handler
through the middleware and returns the tracked status and duration,
which made room for a paused-clock test asserting the duration spans
body streaming rather than ending at the response headers. Capturing
metrics across await points needs a new
`with_capturing_test_client_async` in `objectstore-metrics`.

Author: jan-auer

objectstore-metrics/src/lib.rs

@@ -318,7 +318,7 @@ pub fn init(config: &MetricsConfig) -> Result<(), Error> {
     Ok(())
 }
 
-pub use mock::with_capturing_test_client;
+pub use mock::{with_capturing_test_client, with_capturing_test_client_async};
 
 // ---------------------------------------------------------------------------
 // Macros

objectstore-metrics/src/mock.rs

@@ -3,6 +3,7 @@
 //! Provides [`with_capturing_test_client`], which installs a thread-local
 //! recorder that captures all emitted metrics as DogStatsD-format strings.
 
+use std::future::Future;
 use std::sync::{Arc, Mutex};
 
 use metrics::{Counter, Gauge, Histogram, Key, KeyName, Metadata, Recorder, SharedString, Unit};
@@ -26,6 +27,30 @@ pub fn with_capturing_test_client(f: impl FnOnce()) -> Vec<String> {
     recorder.consume()
 }
 
+/// Awaits `future` with a thread-local mock recorder installed, then returns all captured
+/// metrics as `"name:value|type|#key:value,key:value"` strings.
+///
+/// The recorder stays installed across await points, so metrics emitted while the future is
+/// suspended are captured as well. Since it is thread-local, the future must not migrate between
+/// threads: run it on a current-thread runtime, as `#[tokio::test]` does by default.
+///
+/// # Example
+///
+/// ```ignore
+/// let captured = objectstore_metrics::with_capturing_test_client_async(async {
+///     objectstore_metrics::count!("test.counter");
+/// })
+/// .await;
+/// assert!(captured.iter().any(|m| m.starts_with("test.counter:")));
+/// ```
+pub async fn with_capturing_test_client_async(future: impl Future<Output = ()>) -> Vec<String> {
+    let recorder = MockRecorder::default();
+    let guard = metrics::set_default_local_recorder(&recorder);
+    future.await;
+    drop(guard);
+    recorder.consume()
+}
+
 /// A metrics recorder that formats and stores every operation as a string.
 #[derive(Clone, Default)]
 struct MockRecorder {

objectstore-server/Cargo.toml

@@ -64,6 +64,7 @@ objectstore-options = { workspace = true, features = ["testing"] }
 objectstore-test = { workspace = true }
 stresstest = { workspace = true }
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
 
 [[bin]]
 name = "objectstore"