fix(retry): avoid replaying unread body data

linkerd/http/retry/src/replay.rs

@@ -189,6 +189,7 @@ where
                 tracing::trace!("Cannot replay buffered body, maximum buffer length reached");
                 return Poll::Ready(Some(Err(Capped.into())));
             }
+            this.replay_body = false;
         }
         if this.replay_trailers {
             this.replay_trailers = false;

linkerd/http/retry/src/replay/tests.rs

@@ -46,6 +46,23 @@ async fn replays_one_chunk() {
     }
 }
 
+#[tokio::test]
+async fn replays_unpolled_body() {
+    let Test {
+        mut tx,
+        initial,
+        replay,
+        _trace,
+    } = Test::new();
+    tx.send_data("hello world").await;
+    drop(tx);
+    drop(initial);
+
+    let (data, trailers) = body_to_string(replay).await;
+    assert_eq!(data, "hello world");
+    assert_eq!(trailers, None);
+}
+
 #[tokio::test]
 async fn replays_several_chunks() {
     let Test {