v0.33.8

src/proxy/body/tests.rs

@@ -95,4 +95,134 @@ mod tests {
         assert_eq!(slots.available_permits(), 1);
         assert_eq!(metrics.errors.load(Ordering::Relaxed), 1);
     }
+
+    #[tokio::test]
+    async fn ready_and_empty_bodies_keep_accounting_without_timer_allocations() {
+        for payload in ["", "ready response"] {
+            let (exchange, slots) = observation().await;
+            let metrics = exchange.metrics.clone();
+            let logs = exchange.logs.clone();
+            let mut body = MeteredBody::new(
+                Body::from(payload),
+                exchange,
+                Direction::Response,
+                Duration::from_secs(1),
+            );
+            assert!(body.timer.is_none());
+            assert_eq!(slots.available_permits(), 0);
+            let mut received = Vec::new();
+            while let Some(frame) = body.frame().await {
+                received.extend_from_slice(frame.unwrap().data_ref().unwrap());
+                assert!(body.timer.is_none());
+            }
+            assert_eq!(received, payload.as_bytes());
+            assert_eq!(slots.available_permits(), 1);
+            assert_eq!(metrics.completed.load(Ordering::Relaxed), 1);
+            assert_eq!(
+                metrics.bytes_out.load(Ordering::Relaxed),
+                payload.len() as u64
+            );
+            assert_eq!(metrics.errors.load(Ordering::Relaxed), 0);
+            assert_eq!(logs.recent().len(), 1);
+        }
+    }
+
+    #[tokio::test]
+    async fn delayed_first_poll_does_not_extend_the_original_idle_deadline() {
+        let (exchange, slots) = observation().await;
+        let metrics = exchange.metrics.clone();
+        let pending = futures_util::stream::pending::<Result<Bytes, std::io::Error>>();
+        let mut body = MeteredBody::new(
+            Body::from_stream(pending),
+            exchange,
+            Direction::Request,
+            Duration::from_secs(1),
+        );
+        // Move the saved initial deadline into the past without wall-clock
+        // sleeps: the first Pending must create an already-expired timer.
+        body.deadline = tokio::time::Instant::now() - Duration::from_secs(1);
+        assert!(body.timer.is_none());
+        let result = tokio::time::timeout(Duration::from_millis(200), body.frame()).await;
+        assert!(result.unwrap().unwrap().is_err());
+        assert!(body.timer.is_none());
+        assert_eq!(slots.available_permits(), 1);
+        assert_eq!(metrics.errors.load(Ordering::Relaxed), 1);
+    }
+
+    #[tokio::test]
+    async fn progress_resets_a_lazy_timer_and_trailers_pass_unchanged() {
+        let (exchange, slots) = observation().await;
+        let metrics = exchange.metrics.clone();
+        let (sender, receiver) =
+            tokio::sync::mpsc::channel::<Result<Frame<Bytes>, std::io::Error>>(4);
+        let stream = futures_util::stream::unfold(receiver, |mut receiver| async {
+            receiver.recv().await.map(|frame| (frame, receiver))
+        });
+        let mut body = MeteredBody::new(
+            Body::new(http_body_util::StreamBody::new(stream)),
+            exchange,
+            Direction::Response,
+            Duration::from_secs(5),
+        );
+        assert!(futures_util::poll!(body.frame()).is_pending());
+        assert!(body.timer.is_some());
+        let original_deadline = body.deadline;
+        sender
+            .send(Ok(Frame::data(Bytes::from_static(b"part"))))
+            .await
+            .unwrap();
+        assert_eq!(
+            body.frame().await.unwrap().unwrap().data_ref().unwrap(),
+            "part"
+        );
+        assert!(body.deadline >= original_deadline);
+        assert_eq!(body.timer.as_ref().unwrap().deadline(), body.deadline);
+        let mut trailers = http::HeaderMap::new();
+        trailers.insert("x-checksum", "complete".parse().unwrap());
+        sender
+            .send(Ok(Frame::trailers(trailers.clone())))
+            .await
+            .unwrap();
+        drop(sender);
+        assert_eq!(
+            body.frame().await.unwrap().unwrap().trailers_ref(),
+            Some(&trailers)
+        );
+        assert!(body.frame().await.is_none());
+        assert!(body.timer.is_none());
+        assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 4);
+        assert_eq!(metrics.errors.load(Ordering::Relaxed), 0);
+        assert_eq!(slots.available_permits(), 1);
+    }
+
+    #[tokio::test]
+    async fn cancellation_and_read_errors_keep_their_original_outcomes() {
+        for cancel in [true, false] {
+            let (exchange, slots) = observation().await;
+            let metrics = exchange.metrics.clone();
+            let logs = exchange.logs.clone();
+            let inner = if cancel {
+                Body::from_stream(futures_util::stream::pending::<Result<Bytes, std::io::Error>>())
+            } else {
+                Body::from_stream(futures_util::stream::once(async {
+                    Err::<Bytes, _>(std::io::Error::other("test read error"))
+                }))
+            };
+            let mut body =
+                MeteredBody::new(inner, exchange, Direction::Response, Duration::from_secs(1));
+            if cancel {
+                assert!(futures_util::poll!(body.frame()).is_pending());
+                drop(body);
+            } else {
+                assert!(body.frame().await.unwrap().is_err());
+                assert!(body.timer.is_none());
+            }
+            assert_eq!(slots.available_permits(), 1);
+            assert_eq!(metrics.errors.load(Ordering::Relaxed), 1);
+            assert_eq!(
+                logs.recent()[0].outcome,
+                if cancel { "cancelled" } else { "body_error" }
+            );
+        }
+    }
 }