fix(rustproxy-http): prevent premature idle timeouts during streamed HTTP responses and ensure TLS close_notify is sent on dropped connections

rust/crates/rustproxy-http/src/counting_body.rs

@@ -27,6 +27,10 @@ pub struct CountingBody<B> {
     connection_activity: Option<Arc<AtomicU64>>,
     /// Start instant for computing elapsed ms for connection_activity.
     activity_start: Option<std::time::Instant>,
+    /// Optional active-request counter. When set, CountingBody increments on creation
+    /// and decrements on Drop, keeping the HTTP idle watchdog aware that a response
+    /// body is still streaming (even after the request handler has returned).
+    active_requests: Option<Arc<AtomicU64>>,
 }
 
 /// Which direction the bytes flow.
@@ -55,6 +59,7 @@ impl<B> CountingBody<B> {
             direction,
             connection_activity: None,
             activity_start: None,
+            active_requests: None,
         }
     }
 
@@ -67,6 +72,15 @@ impl<B> CountingBody<B> {
         self
     }
 
+    /// Set the active-request counter for the HTTP idle watchdog.
+    /// CountingBody increments on creation and decrements on Drop, ensuring the
+    /// idle watchdog sees an "active request" while the response body streams.
+    pub fn with_active_requests(mut self, counter: Arc<AtomicU64>) -> Self {
+        counter.fetch_add(1, Ordering::Relaxed);
+        self.active_requests = Some(counter);
+        self
+    }
+
     /// Report a chunk of bytes immediately to the metrics collector.
     #[inline]
     fn report_chunk(&self, len: u64) {
@@ -122,3 +136,13 @@ where
         self.inner.size_hint()
     }
 }
+
+impl<B> Drop for CountingBody<B> {
+    fn drop(&mut self) {
+        // Decrement the active-request counter so the HTTP idle watchdog
+        // knows this response body is no longer streaming.
+        if let Some(ref counter) = self.active_requests {
+            counter.fetch_sub(1, Ordering::Relaxed);
+        }
+    }
+}