Fix an issue with using non-reusable body types with redirects (#19134)

Closes #19131
Closes #19039

fixes the broken auto-updater.

I had the bright idea of using streams as the most common unit of data
transfer. Unfortunately, streams are not re-usable. So HTTP redirects
that have a stream body (like our remote server and auto update
downloads), don't redirect, as they can't reuse the stream. This PR
fixes the problem and simplifies the AsyncBody implementation now that
we're not using Isahc.

Release Notes:

- N/A

crates/reqwest_client/Cargo.toml

@@ -21,7 +21,7 @@ path = "examples/client.rs"
 
 [dependencies]
 anyhow.workspace = true
-bytes = "1.0"
+bytes.workspace = true
 futures.workspace = true
 http_client.workspace = true
 serde.workspace = true

crates/reqwest_client/src/reqwest_client.rs

@@ -1,4 +1,4 @@
-use std::{any::type_name, borrow::Cow, io::Read, mem, pin::Pin, sync::OnceLock, task::Poll};
+use std::{any::type_name, mem, pin::Pin, sync::OnceLock, task::Poll};
 
 use anyhow::anyhow;
 use bytes::{BufMut, Bytes, BytesMut};
@@ -173,39 +173,6 @@ pub fn poll_read_buf(
     Poll::Ready(Ok(n))
 }
 
-struct SyncReader {
-    cursor: Option<std::io::Cursor<Cow<'static, [u8]>>>,
-}
-
-impl SyncReader {
-    fn new(cursor: std::io::Cursor<Cow<'static, [u8]>>) -> Self {
-        Self {
-            cursor: Some(cursor),
-        }
-    }
-}
-
-impl futures::stream::Stream for SyncReader {
-    type Item = Result<Bytes, std::io::Error>;
-
-    fn poll_next(
-        mut self: std::pin::Pin<&mut Self>,
-        _cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Option<Self::Item>> {
-        let Some(mut cursor) = self.cursor.take() else {
-            return Poll::Ready(None);
-        };
-
-        let mut buf = Vec::new();
-        match cursor.read_to_end(&mut buf) {
-            Ok(_) => {
-                return Poll::Ready(Some(Ok(Bytes::from(buf))));
-            }
-            Err(e) => return Poll::Ready(Some(Err(e))),
-        }
-    }
-}
-
 impl http_client::HttpClient for ReqwestClient {
     fn proxy(&self) -> Option<&http::Uri> {
         self.proxy.as_ref()
@@ -238,9 +205,7 @@ impl http_client::HttpClient for ReqwestClient {
         }
         let request = request.body(match body.0 {
             http_client::Inner::Empty => reqwest::Body::default(),
-            http_client::Inner::SyncReader(cursor) => {
-                reqwest::Body::wrap_stream(SyncReader::new(cursor))
-            }
+            http_client::Inner::Bytes(cursor) => cursor.into_inner().into(),
             http_client::Inner::AsyncReader(stream) => {
                 reqwest::Body::wrap_stream(StreamReader::new(stream))
             }