From aa9e6dfd9442544b096e09acc559cf282ce65e40 Mon Sep 17 00:00:00 2001 From: Juergen Kunz Date: Sun, 15 Mar 2026 21:44:32 +0000 Subject: [PATCH] fix(rustproxy-http): report streamed HTTP and WebSocket bytes per chunk for real-time throughput metrics --- changelog.md | 6 +++ .../rustproxy-http/src/counting_body.rs | 44 +++++-------------- .../rustproxy-http/src/proxy_service.rs | 12 +++-- ts/00_commitinfo_data.ts | 2 +- 4 files changed, 27 insertions(+), 37 deletions(-) diff --git a/changelog.md b/changelog.md index b4705bb..5bfc9ed 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,11 @@ # Changelog +## 2026-03-15 - 25.11.4 - fix(rustproxy-http) +report streamed HTTP and WebSocket bytes per chunk for real-time throughput metrics + +- Update CountingBody to record bytes immediately on each data frame instead of aggregating until completion or drop +- Record WebSocket tunnel traffic inside both copy loops and remove the final aggregate byte report to keep throughput metrics current + ## 2026-03-15 - 25.11.3 - fix(repo) no changes to commit diff --git a/rust/crates/rustproxy-http/src/counting_body.rs b/rust/crates/rustproxy-http/src/counting_body.rs index 8882289..996b192 100644 --- a/rust/crates/rustproxy-http/src/counting_body.rs +++ b/rust/crates/rustproxy-http/src/counting_body.rs @@ -11,20 +11,17 @@ use rustproxy_metrics::MetricsCollector; /// Wraps any `http_body::Body` and counts data bytes passing through. /// -/// When the body is fully consumed or dropped, accumulated byte counts -/// are reported to the `MetricsCollector`. +/// Each chunk is reported to the `MetricsCollector` immediately so that +/// the throughput tracker (sampled at 1 Hz) reflects real-time data flow. /// /// The inner body is pinned on the heap to support `!Unpin` types like `hyper::body::Incoming`. pub struct CountingBody { inner: Pin>, - counted_bytes: AtomicU64, metrics: Arc, route_id: Option, source_ip: Option, /// Whether we count bytes as "in" (request body) or "out" (response body). direction: Direction, - /// Whether we've already reported the bytes (to avoid double-reporting on drop). - reported: bool, /// Optional connection-level activity tracker. When set, poll_frame updates this /// to keep the idle watchdog alive during active body streaming (uploads/downloads). connection_activity: Option>, @@ -52,12 +49,10 @@ impl CountingBody { ) -> Self { Self { inner: Box::pin(inner), - counted_bytes: AtomicU64::new(0), metrics, route_id, source_ip, direction, - reported: false, connection_activity: None, activity_start: None, } @@ -72,33 +67,18 @@ impl CountingBody { self } - /// Report accumulated bytes to the metrics collector. - fn report(&mut self) { - if self.reported { - return; - } - self.reported = true; - - let bytes = self.counted_bytes.load(Ordering::Relaxed); - if bytes == 0 { - return; - } - + /// Report a chunk of bytes immediately to the metrics collector. + #[inline] + fn report_chunk(&self, len: u64) { let route_id = self.route_id.as_deref(); let source_ip = self.source_ip.as_deref(); match self.direction { - Direction::In => self.metrics.record_bytes(bytes, 0, route_id, source_ip), - Direction::Out => self.metrics.record_bytes(0, bytes, route_id, source_ip), + Direction::In => self.metrics.record_bytes(len, 0, route_id, source_ip), + Direction::Out => self.metrics.record_bytes(0, len, route_id, source_ip), } } } -impl Drop for CountingBody { - fn drop(&mut self) { - self.report(); - } -} - // CountingBody is Unpin because inner is Pin> (always Unpin). impl Unpin for CountingBody {} @@ -118,7 +98,9 @@ where match this.inner.as_mut().poll_frame(cx) { Poll::Ready(Some(Ok(frame))) => { if let Some(data) = frame.data_ref() { - this.counted_bytes.fetch_add(data.len() as u64, Ordering::Relaxed); + let len = data.len() as u64; + // Report bytes immediately so the 1 Hz throughput sampler sees them + this.report_chunk(len); // Keep the connection-level idle watchdog alive during body streaming if let (Some(activity), Some(start)) = (&this.connection_activity, &this.activity_start) { activity.store(start.elapsed().as_millis() as u64, Ordering::Relaxed); @@ -127,11 +109,7 @@ where Poll::Ready(Some(Ok(frame))) } Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))), - Poll::Ready(None) => { - // Body is fully consumed — report now - this.report(); - Poll::Ready(None) - } + Poll::Ready(None) => Poll::Ready(None), Poll::Pending => Poll::Pending, } } diff --git a/rust/crates/rustproxy-http/src/proxy_service.rs b/rust/crates/rustproxy-http/src/proxy_service.rs index c5e1aa0..5ba5cbb 100644 --- a/rust/crates/rustproxy-http/src/proxy_service.rs +++ b/rust/crates/rustproxy-http/src/proxy_service.rs @@ -1919,6 +1919,9 @@ impl HttpProxyService { let conn_act_u2c = conn_activity.as_ref().map(|ca| (Arc::clone(&ca.last_activity), ca.start)); let la1 = Arc::clone(&last_activity); + let metrics_c2u = Arc::clone(&metrics); + let route_c2u = route_id_owned.clone(); + let ip_c2u = source_ip_owned.clone(); let c2u = tokio::spawn(async move { let mut buf = vec![0u8; 65536]; let mut total = 0u64; @@ -1931,6 +1934,7 @@ impl HttpProxyService { break; } total += n as u64; + metrics_c2u.record_bytes(n as u64, 0, route_c2u.as_deref(), Some(&ip_c2u)); la1.store(start.elapsed().as_millis() as u64, Ordering::Relaxed); if let Some((ref ca, ca_start)) = conn_act_c2u { ca.store(ca_start.elapsed().as_millis() as u64, Ordering::Relaxed); @@ -1941,6 +1945,9 @@ impl HttpProxyService { }); let la2 = Arc::clone(&last_activity); + let metrics_u2c = Arc::clone(&metrics); + let route_u2c = route_id_owned.clone(); + let ip_u2c = source_ip_owned.clone(); let u2c = tokio::spawn(async move { let mut buf = vec![0u8; 65536]; let mut total = 0u64; @@ -1953,6 +1960,7 @@ impl HttpProxyService { break; } total += n as u64; + metrics_u2c.record_bytes(0, n as u64, route_u2c.as_deref(), Some(&ip_u2c)); la2.store(start.elapsed().as_millis() as u64, Ordering::Relaxed); if let Some((ref ca, ca_start)) = conn_act_u2c { ca.store(ca_start.elapsed().as_millis() as u64, Ordering::Relaxed); @@ -2013,9 +2021,7 @@ impl HttpProxyService { debug!("WebSocket tunnel closed: {} bytes in, {} bytes out", bytes_in, bytes_out); upstream_selector.connection_ended(&upstream_key_owned); - if let Some(ref rid) = route_id_owned { - metrics.record_bytes(bytes_in, bytes_out, Some(rid.as_str()), Some(&source_ip_owned)); - } + // Bytes already reported per-chunk in the copy loops above }); let body: BoxBody = BoxBody::new( diff --git a/ts/00_commitinfo_data.ts b/ts/00_commitinfo_data.ts index 1f8e9ab..84d9e88 100644 --- a/ts/00_commitinfo_data.ts +++ b/ts/00_commitinfo_data.ts @@ -3,6 +3,6 @@ */ export const commitinfo = { name: '@push.rocks/smartproxy', - version: '25.11.3', + version: '25.11.4', description: 'A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.' }