v25.11.4

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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartproxy",
-  "version": "25.11.3",
+  "version": "25.11.4",
   "private": false,
   "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.",
   "main": "dist_ts/index.js",

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
+/// Each chunk is reported to the `MetricsCollector` immediately so that
-/// are reported to the `MetricsCollector`.
+/// 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<B> {
     inner: Pin<Box<B>>,
-    counted_bytes: AtomicU64,
     metrics: Arc<MetricsCollector>,
     route_id: Option<String>,
     source_ip: Option<String>,
     /// 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<Arc<AtomicU64>>,
@@ -52,12 +49,10 @@ impl<B> CountingBody<B> {
     ) -> 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<B> CountingBody<B> {
         self
     }
 
-    /// Report accumulated bytes to the metrics collector.
+    /// Report a chunk of bytes immediately to the metrics collector.
-    fn report(&mut self) {
+    #[inline]
-        if self.reported {
+    fn report_chunk(&self, len: u64) {
-            return;
-        }
-        self.reported = true;
-
-        let bytes = self.counted_bytes.load(Ordering::Relaxed);
-        if bytes == 0 {
-            return;
-        }
-
         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::In => self.metrics.record_bytes(len, 0, route_id, source_ip),
-            Direction::Out => self.metrics.record_bytes(0, bytes, route_id, source_ip),
+            Direction::Out => self.metrics.record_bytes(0, len, route_id, source_ip),
         }
     }
 }
 
-impl<B> Drop for CountingBody<B> {
-    fn drop(&mut self) {
-        self.report();
-    }
-}
-
 // CountingBody is Unpin because inner is Pin<Box<B>> (always Unpin).
 impl<B> Unpin for CountingBody<B> {}
 
@@ -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) => {
+            Poll::Ready(None) => Poll::Ready(None),
-                // Body is fully consumed — report now
-                this.report();
-                Poll::Ready(None)
-            }
             Poll::Pending => Poll::Pending,
         }
     }

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 {
+            // Bytes already reported per-chunk in the copy loops above
-                metrics.record_bytes(bytes_in, bytes_out, Some(rid.as_str()), Some(&source_ip_owned));
-            }
         });
 
         let body: BoxBody<Bytes, hyper::Error> = BoxBody::new(

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.'
 }