From 0ef2749b69312e467f535508f75982c988b7239a Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Fri, 14 Aug 2026 00:10:11 +0200 Subject: [PATCH 1/2] Measure response_size via a counting body; bump runtime The metrics middleware read response_size from the response Content-Length header, but axum wraps handler responses in a streaming body with no such header at that point, so it was ~always 0. Add CountingBody, an http_body::Body wrapper that tallies data-frame bytes as they stream (never buffers) and fires a callback with the total at end-of-stream or on early drop. The metrics middleware wraps the response body and logs "request completed" (with the real response_size) from that callback. Works for JSON, downloads, and SSE alike; request_size is unchanged (Content-Length is correct for request bodies). Also bump the nvisy-engine/runtime git dep to latest main (bd96ed9 -> 3e805a7). Co-Authored-By: Claude Opus 4.8 --- Cargo.lock | 49 ++--- Cargo.toml | 1 + crates/nvisy-server/Cargo.toml | 1 + .../src/middleware/counting_body.rs | 179 ++++++++++++++++++ crates/nvisy-server/src/middleware/mod.rs | 1 + .../src/middleware/observability.rs | 48 +++-- 6 files changed, 235 insertions(+), 44 deletions(-) create mode 100644 crates/nvisy-server/src/middleware/counting_body.rs diff --git a/Cargo.lock b/Cargo.lock index 37d1bd7b..5b1bcaa3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2898,7 +2898,7 @@ checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" [[package]] name = "elide" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-codec", @@ -2919,7 +2919,7 @@ dependencies = [ [[package]] name = "elide-bento" version = "0.1.0" -source = "git+https://github.com/nvisycom/bento?branch=main#e4e2947c360cfd389cf3c4d83dcff89d49b5a4d5" +source = "git+https://github.com/nvisycom/bento?branch=main#70e1cba24bc39d01b1e35ecac4d58d05a376bd53" dependencies = [ "async-trait", "base64 0.23.1", @@ -2936,7 +2936,7 @@ dependencies = [ [[package]] name = "elide-codec" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "bytes", @@ -2959,7 +2959,7 @@ dependencies = [ [[package]] name = "elide-context" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -2970,12 +2970,14 @@ dependencies = [ [[package]] name = "elide-core" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", + "blake3", "bytes", "celes", "derive_builder", + "hex", "hipstr", "jiff", "oxilangtag", @@ -2988,7 +2990,7 @@ dependencies = [ [[package]] name = "elide-detection" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "elide-core", "futures", @@ -3000,7 +3002,7 @@ dependencies = [ [[package]] name = "elide-engine" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "bytes", "elide-codec", @@ -3015,7 +3017,7 @@ dependencies = [ [[package]] name = "elide-fake" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -3026,7 +3028,7 @@ dependencies = [ [[package]] name = "elide-lingua" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -3037,7 +3039,7 @@ dependencies = [ [[package]] name = "elide-llm" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "derive_builder", @@ -3059,7 +3061,7 @@ dependencies = [ [[package]] name = "elide-ner" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "derive_builder", @@ -3073,7 +3075,7 @@ dependencies = [ [[package]] name = "elide-ocr" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -3083,7 +3085,7 @@ dependencies = [ [[package]] name = "elide-operator" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "aes-gcm", "async-trait", @@ -3105,7 +3107,7 @@ dependencies = [ [[package]] name = "elide-pattern" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "aho-corasick", "async-trait", @@ -3125,7 +3127,7 @@ dependencies = [ [[package]] name = "elide-redaction" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -3139,7 +3141,7 @@ dependencies = [ [[package]] name = "elide-stt" version = "0.1.0" -source = "git+https://github.com/nvisycom/elide?branch=main#a743ed3a2eb8f63a01a8bb4a0e302e1f6b15d314" +source = "git+https://github.com/nvisycom/elide?branch=main#3facedc82c16584f00a69ce64d45126c3bcc0380" dependencies = [ "async-trait", "elide-core", @@ -6055,7 +6057,7 @@ dependencies = [ [[package]] name = "nvisy-engine" version = "0.1.0" -source = "git+https://github.com/nvisycom/runtime?branch=main#bd96ed9dc53d3cf42c257beb4605d0755b45958c" +source = "git+https://github.com/nvisycom/runtime?branch=main#3e805a7779fb803f9baecc6535668d224f98077f" dependencies = [ "bytes", "csv", @@ -6125,7 +6127,7 @@ dependencies = [ [[package]] name = "nvisy-policy" version = "0.1.0" -source = "git+https://github.com/nvisycom/runtime?branch=main#bd96ed9dc53d3cf42c257beb4605d0755b45958c" +source = "git+https://github.com/nvisycom/runtime?branch=main#3e805a7779fb803f9baecc6535668d224f98077f" dependencies = [ "elide-core", "elide-operator", @@ -6171,7 +6173,7 @@ dependencies = [ [[package]] name = "nvisy-schema" version = "0.1.0" -source = "git+https://github.com/nvisycom/runtime?branch=main#bd96ed9dc53d3cf42c257beb4605d0755b45958c" +source = "git+https://github.com/nvisycom/runtime?branch=main#3e805a7779fb803f9baecc6535668d224f98077f" dependencies = [ "bytes", "elide-core", @@ -6207,6 +6209,7 @@ dependencies = [ "futures", "hex", "hkdf", + "http-body", "humantime", "image", "jiff", @@ -6245,7 +6248,7 @@ dependencies = [ [[package]] name = "nvisy-template" version = "0.1.0" -source = "git+https://github.com/nvisycom/runtime?branch=main#bd96ed9dc53d3cf42c257beb4605d0755b45958c" +source = "git+https://github.com/nvisycom/runtime?branch=main#3e805a7779fb803f9baecc6535668d224f98077f" dependencies = [ "elide-core", "hipstr", @@ -6962,7 +6965,7 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "itertools 0.14.0", "log", "multimap", @@ -8549,7 +8552,7 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "proc-macro2", "quote", "syn 2.0.119", @@ -8561,7 +8564,7 @@ version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "287f59010008f0d7cf5e3b03196d666c1acc46c8d3e9cf34c28a1a7157601e72" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "proc-macro2", "quote", "syn 2.0.119", diff --git a/Cargo.toml b/Cargo.toml index 246a7946..8ff8ab5b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -136,6 +136,7 @@ derive_more = { version = "2.0", features = [] } derive_builder = { version = "0.20", features = [] } anyhow = { version = "1.0", features = [] } pin-project-lite = { version = "0.2", features = [] } +http-body = { version = "1.0", features = [] } # Primitive datatypes uuid = { version = "1.22", features = ["serde", "v4", "v7"] } diff --git a/crates/nvisy-server/Cargo.toml b/crates/nvisy-server/Cargo.toml index d4e18b9c..ca166e06 100644 --- a/crates/nvisy-server/Cargo.toml +++ b/crates/nvisy-server/Cargo.toml @@ -89,6 +89,7 @@ serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true, features = [] } toml = { workspace = true, features = ["parse"] } bytes = { workspace = true, features = [] } +http-body = { workspace = true, features = [] } validator = { workspace = true, features = [] } # Text processing diff --git a/crates/nvisy-server/src/middleware/counting_body.rs b/crates/nvisy-server/src/middleware/counting_body.rs new file mode 100644 index 00000000..b70c8d47 --- /dev/null +++ b/crates/nvisy-server/src/middleware/counting_body.rs @@ -0,0 +1,179 @@ +//! A response body that counts the bytes streamed through it. +//! +//! Axum responses are streaming bodies without a `Content-Length` header at the +//! point request middleware inspects them, so reading the header yields no size. +//! [`CountingBody`] wraps a body and tallies each data frame's length as it is +//! polled, invoking a callback with the total once the stream ends — by reaching +//! its last frame or by being dropped early (a client disconnect mid-stream). +//! +//! It never buffers: frames pass through untouched, so it is safe for large +//! downloads and long-lived event streams alike. The trade-off is timing — the +//! total is known only when the body finishes, which for a streamed response is +//! after the handler has returned. + +use std::pin::Pin; +use std::task::{Context, Poll}; + +use bytes::Buf; +use http_body::{Body, Frame, SizeHint}; +use pin_project_lite::pin_project; + +pin_project! { + /// Wraps a body, counting streamed bytes and reporting the total on end. + /// + /// The callback fires exactly once: on the last frame, or on drop if the + /// body is dropped before completing (so an interrupted response is still + /// reported, with the bytes sent before the interruption). + #[project = CountingBodyProj] + pub struct CountingBody + where + F: FnOnce(u64), + { + #[pin] + inner: B, + bytes: u64, + // `Some` until the callback fires; taken so it runs at most once, whether + // that is at end-of-stream or on drop. + on_end: Option, + } + + impl PinnedDrop for CountingBody + where + F: FnOnce(u64), + { + fn drop(this: Pin<&mut Self>) { + let this = this.project(); + if let Some(on_end) = this.on_end.take() { + on_end(*this.bytes); + } + } + } +} + +impl CountingBody +where + F: FnOnce(u64), +{ + /// Wraps `inner`, calling `on_end` with the total byte count when the body + /// finishes streaming (or is dropped). + pub fn new(inner: B, on_end: F) -> Self { + Self { + inner, + bytes: 0, + on_end: Some(on_end), + } + } +} + +impl Body for CountingBody +where + B: Body, + F: FnOnce(u64), +{ + type Data = B::Data; + type Error = B::Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.project(); + let polled = this.inner.poll_frame(cx); + + match &polled { + // A data frame: add its length to the running total. + Poll::Ready(Some(Ok(frame))) => { + if let Some(data) = frame.data_ref() { + *this.bytes += data.remaining() as u64; + } + } + // End of stream: report the total once (drop reports it otherwise). + Poll::Ready(None) => { + if let Some(on_end) = this.on_end.take() { + on_end(*this.bytes); + } + } + _ => {} + } + + polled + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +#[cfg(test)] +mod tests { + use std::future::poll_fn; + use std::sync::Arc; + use std::sync::atomic::{AtomicU64, Ordering}; + + use axum::body::Body as AxumBody; + use http_body::Body as HttpBody; + + use super::*; + + /// Drives a body to completion, returning the total bytes across its data + /// frames. + async fn drain(mut body: B) -> u64 + where + B::Error: std::fmt::Debug, + { + let mut total = 0; + while let Some(frame) = poll_fn(|cx| Pin::new(&mut body).poll_frame(cx)).await { + if let Some(data) = frame.unwrap().data_ref() { + total += data.remaining() as u64; + } + } + total + } + + /// Draining the body to completion reports the exact byte total once. + #[tokio::test] + async fn counts_streamed_bytes_on_completion() { + let reported = Arc::new(AtomicU64::new(u64::MAX)); + let sink = Arc::clone(&reported); + + let body = CountingBody::new(AxumBody::from("hello world"), move |n| { + sink.store(n, Ordering::SeqCst) + }); + + assert_eq!(drain(body).await, 11); + assert_eq!(reported.load(Ordering::SeqCst), 11); + } + + /// An empty body reports zero — distinct from the previous "header absent" + /// fallback that also produced zero for non-empty responses. + #[tokio::test] + async fn counts_zero_for_empty_body() { + let reported = Arc::new(AtomicU64::new(u64::MAX)); + let sink = Arc::clone(&reported); + + let body = CountingBody::new(AxumBody::empty(), move |n| sink.store(n, Ordering::SeqCst)); + assert_eq!(drain(body).await, 0); + assert_eq!(reported.load(Ordering::SeqCst), 0); + } + + /// Dropping the body before it is polled still reports (zero bytes seen) + /// exactly once, via the drop path rather than the end-of-stream path. + #[tokio::test] + async fn reports_on_early_drop() { + let reported = Arc::new(AtomicU64::new(u64::MAX)); + let sink = Arc::clone(&reported); + + // `CountingBody` is `Unpin` here, so dropping the owned value runs its + // `PinnedDrop` and fires the callback with the bytes seen so far (none). + let body = CountingBody::new(AxumBody::from("partial"), move |n| { + sink.store(n, Ordering::SeqCst) + }); + drop(body); + + assert_eq!(reported.load(Ordering::SeqCst), 0); + } +} diff --git a/crates/nvisy-server/src/middleware/mod.rs b/crates/nvisy-server/src/middleware/mod.rs index 675d43ea..d077ed1a 100644 --- a/crates/nvisy-server/src/middleware/mod.rs +++ b/crates/nvisy-server/src/middleware/mod.rs @@ -48,6 +48,7 @@ mod authentication; mod authorization; mod constants; +mod counting_body; mod observability; mod recovery; mod route_category; diff --git a/crates/nvisy-server/src/middleware/observability.rs b/crates/nvisy-server/src/middleware/observability.rs index 9573d6f4..4a91c915 100644 --- a/crates/nvisy-server/src/middleware/observability.rs +++ b/crates/nvisy-server/src/middleware/observability.rs @@ -7,6 +7,7 @@ use std::time::Instant; use axum::Router; +use axum::body::Body; use axum::extract::Request; use axum::http::header; use axum::middleware::{Next, from_fn}; @@ -17,6 +18,7 @@ use tower_http::sensitive_headers::SetSensitiveRequestHeadersLayer; use tower_http::trace::TraceLayer; use super::RouteCategory; +use super::counting_body::CountingBody; /// Tracing target for request metrics. const TRACING_TARGET_METRICS: &str = "nvisy_server::metrics"; @@ -105,25 +107,29 @@ pub async fn track_categorized_metrics(request: Request, next: Next) -> Response let response = next.run(request).await; let duration = start_time.elapsed(); - - let response_size = response - .headers() - .get("content-length") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - .unwrap_or(0); - - tracing::trace!( - target: TRACING_TARGET_METRICS, - method = %method, - uri = %uri, - category = category.as_str(), - status = %response.status(), - duration_ms = duration.as_millis() as u64, - request_size = request_size, - response_size = response_size, - "request completed" - ); - - response + let status = response.status(); + + // The response body streams without a `Content-Length` header, so the size + // is not known here. Wrap the body to count bytes as they flow and emit the + // "request completed" line when the stream ends (or is dropped) — which for + // a streamed response is after this function returns. + let category_str = category.as_str(); + let span = tracing::Span::current(); + let (parts, body) = response.into_parts(); + let body = CountingBody::new(body, move |response_size| { + let _guard = span.enter(); + tracing::trace!( + target: TRACING_TARGET_METRICS, + method = %method, + uri = %uri, + category = category_str, + status = %status, + duration_ms = duration.as_millis() as u64, + request_size = request_size, + response_size = response_size, + "request completed" + ); + }); + + Response::from_parts(parts, Body::new(body)) } From ef091015f61fb95e3c323d30fe7d555a9bae587e Mon Sep 17 00:00:00 2001 From: Oleh Martsokha Date: Fri, 14 Aug 2026 00:10:28 +0200 Subject: [PATCH 2/2] CI: move Machete from Security to Build Match the runtime repo's layout: the unused-dependency check (cargo-machete) belongs alongside the other build-quality gates (fmt, clippy, docs), not in the Security workflow next to cargo-deny and secret scanning. Security keeps deny and the secret scan. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/build.yml | 21 +++++++++++++++++++++ .github/workflows/security.yml | 17 ----------------- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 4383cdf2..b7821fcf 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -155,3 +155,24 @@ jobs: run: cargo doc --no-deps --all-features --workspace env: RUSTDOCFLAGS: "-D warnings" + + machete: + name: Machete + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@master + with: + toolchain: ${{ env.RUST_TOOLCHAIN }} + + - name: Install cargo-binstall + uses: cargo-bins/cargo-binstall@main + + - name: Install cargo-machete + run: cargo binstall cargo-machete --no-confirm --no-symlinks + + - name: Run machete + run: cargo machete diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index 2058def0..a0da8da1 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -53,23 +53,6 @@ jobs: - name: Run deny run: cargo deny check all - unused: - name: Unused Dependencies - runs-on: ubuntu-latest - - steps: - - name: Checkout - uses: actions/checkout@v7 - - - name: Install cargo-binstall - uses: cargo-bins/cargo-binstall@main - - - name: Install cargo-machete - run: cargo binstall cargo-machete --no-confirm --no-symlinks - - - name: Run machete - run: cargo machete - secrets: name: Secret Scan runs-on: ubuntu-latest