Retry timeout errors for streams (#17875)

Rebase of https://github.com/astral-sh/uv/pull/17735 We're already retrying most timeout errors (only `retry_read_timeout_stream` didn't retry before), but not for streams after we already received some bytes which this PR adds.
2026-02-05 14:48:21 +01:00
parent e1b14cf297
commit f2957dc400
5 changed files with 170 additions and 5 deletions
@@ -5630,6 +5630,7 @@ dependencies = [
 "backon",
 "base64 0.22.1",
 "byteorder",
+ "bytes",
 "clap",
 "console 0.16.2",
 "ctrlc",
@@ -5642,6 +5643,9 @@ dependencies = [
 "fs-err",
 "futures",
 "http",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
 "ignore",
 "indexmap",
 "indicatif",
@@ -5668,6 +5672,7 @@ dependencies = [
 "textwrap",
 "thiserror 2.0.18",
 "tokio",
+ "tokio-stream",
 "tokio-util",
 "toml",
 "toml_edit 0.24.0+spec-1.1.0",
@@ -212,6 +212,7 @@ zstd = { version = "0.13.3" }
 assert_cmd = { version = "2.0.16" }
 assert_fs = { version = "1.1.2" }
 byteorder = { version = "1.5.0" }
+bytes = { version = "1.10.1" }
 filetime = { version = "0.2.25" }
 http-body-util = { version = "0.1.2" }
 hyper = { version = "1.4.1", features = ["server", "http1"] }
@@ -1246,6 +1246,8 @@ fn retryable_on_request_failure(err: &(dyn Error + 'static)) -> Option<Retryable
                io::ErrorKind::ConnectionReset,
                // https://github.com/astral-sh/uv/issues/14699
                io::ErrorKind::InvalidData,
+                // https://github.com/astral-sh/uv/issues/17697#issuecomment-3817060484
+                io::ErrorKind::TimedOut,
                // https://github.com/astral-sh/uv/issues/9246
                io::ErrorKind::UnexpectedEof,
            ];
@@ -131,8 +131,12 @@ assert_fs = { workspace = true }
 backon = { workspace = true }
 base64 = { workspace = true }
 byteorder = { workspace = true }
+bytes = { workspace = true }
 filetime = { workspace = true }
 flate2 = { workspace = true, default-features = false }
+http-body-util = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
 ignore = { workspace = true }
 indoc = { workspace = true }
 insta = { workspace = true }
@@ -143,6 +147,7 @@ sha2 = { workspace = true }
 similar = { workspace = true }
 tar = { workspace = true }
 tempfile = { workspace = true }
+tokio-stream = { workspace = true }
 tokio-util = { workspace = true }
 whoami = { workspace = true }
 wiremock = { workspace = true }
@@ -1,14 +1,22 @@
+use std::convert::Infallible;
 use std::io;
-use std::net::TcpListener;
 use std::time::{Duration, Instant};

 use assert_fs::fixture::{ChildPath, FileWriteStr, PathChild};
+use bytes::Bytes;
 use http::StatusCode;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, StreamBody};
+use hyper::body::Frame;
+use hyper::service::service_fn;
+use hyper_util::rt::TokioIo;
 use serde_json::json;
-use uv_static::EnvVars;
+use tokio_stream::wrappers::ReceiverStream;
 use wiremock::matchers::{any, method};
 use wiremock::{Mock, MockServer, Request, ResponseTemplate};

+use uv_static::EnvVars;
+
 use crate::common::{TestContext, uv_snapshot};

 /// Creates a CONNECT tunnel proxy that forwards connections to the target.
@@ -169,6 +177,63 @@ async fn mixed_error_server() -> (MockServer, String) {
    (server, mock_server_uri)
 }

+async fn time_out_response(
+    _req: hyper::Request<hyper::body::Incoming>,
+) -> Result<hyper::Response<BoxBody<Bytes, Infallible>>, Infallible> {
+    let (tx, rx) = tokio::sync::mpsc::channel(1);
+    tokio::spawn(async move {
+        let _ = tx.send(Ok(Frame::data(Bytes::new()))).await;
+        tokio::time::sleep(Duration::from_secs(60)).await;
+    });
+    let body = StreamBody::new(ReceiverStream::new(rx)).boxed();
+    Ok(hyper::Response::builder()
+        .header("Content-Type", "text/html")
+        .body(body)
+        .unwrap())
+}
+
+/// Returns the server URL and a drop guard that shuts down the server.
+///
+/// The server runs in a thread with its own tokio runtime, so it
+/// won't be starved by the subprocess blocking the test thread. Dropping the
+/// guard shuts down the runtime and all tasks running in it.
+fn read_timeout_server() -> (String, impl Drop) {
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    listener.set_nonblocking(true).unwrap();
+    let server = format!("http://{}", listener.local_addr().unwrap());
+
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+
+    std::thread::spawn(move || {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        runtime.block_on(async move {
+            let listener = tokio::net::TcpListener::from_std(listener).unwrap();
+            tokio::select! {
+                _ = async {
+                    loop {
+                        let (stream, _) = listener.accept().await.unwrap();
+                        let io = TokioIo::new(stream);
+
+                        tokio::spawn(async move {
+                           let _ = hyper_util::server::conn::auto::Builder::new(
+                                hyper_util::rt::TokioExecutor::new(),
+                            )
+                            .serve_connection(io, service_fn(time_out_response))
+                            .await;
+                        });
+                    }
+                } => {}
+                _ = shutdown_rx => {}
+            }
+        });
+    });
+
+    (server, shutdown_tx)
+}
+
 /// Check the simple index error message when the server returns HTTP status 500, a retryable error.
 #[tokio::test]
 async fn simple_http_500() {
@@ -928,11 +993,11 @@ async fn proxy_schemeless_url_in_uv_toml() {
 }

 #[test]
-fn connect_timeout() {
+fn connect_timeout_index() {
    let context = TestContext::new("3.12");

-    // Create a server that just times out.
-    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    // Create a server that never responds, causing a timeout for our requests.
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let server = listener.local_addr().unwrap().to_string();

    let start = Instant::now();
@@ -961,3 +1026,90 @@ fn connect_timeout() {
        "Test with 1s connect timeout took too long"
    );
 }
+
+#[test]
+fn connect_timeout_stream() {
+    let context = TestContext::new("3.12");
+
+    // Create a server that never responds, causing a timeout for our requests.
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+    let server = listener.local_addr().unwrap().to_string();
+
+    let start = Instant::now();
+    uv_snapshot!(context.filters(), context
+        .pip_install()
+        .arg(format!("https://{server}/tqdm-0.1-py3-none-any.whl"))
+        .env(EnvVars::UV_HTTP_CONNECT_TIMEOUT, "1")
+        .env(EnvVars::UV_HTTP_RETRIES, "0"), @"
+    success: false
+    exit_code: 1
+    ----- stdout -----
+
+    ----- stderr -----
+      × Failed to download `tqdm @ https://[LOCALHOST]/tqdm-0.1-py3-none-any.whl`
+      ├─▶ Failed to fetch: `https://[LOCALHOST]/tqdm-0.1-py3-none-any.whl`
+      ├─▶ error sending request for url (https://[LOCALHOST]/tqdm-0.1-py3-none-any.whl)
+      ├─▶ client error (Connect)
+      ╰─▶ operation timed out
+    ");
+
+    // Assumption: There's less than 2s overhead for this test and startup.
+    let elapsed = start.elapsed();
+    assert!(
+        elapsed < Duration::from_secs(3),
+        "Test with 1s connect timeout took too long"
+    );
+}
+
+#[tokio::test]
+async fn retry_read_timeout_index() {
+    let context = TestContext::new("3.12");
+
+    let (server, _guard) = read_timeout_server();
+
+    uv_snapshot!(context.filters(), context
+        .pip_install()
+        .arg("tqdm")
+        .arg("--index-url")
+        .arg(server)
+        // Speed the test up with the minimum testable values
+        .env(EnvVars::UV_HTTP_TIMEOUT, "1")
+        .env(EnvVars::UV_HTTP_RETRIES, "1"), @"
+    success: false
+    exit_code: 2
+    ----- stdout -----
+
+    ----- stderr -----
+    error: Request failed after 1 retry
+      Caused by: Failed to fetch: `http://[LOCALHOST]/tqdm/`
+      Caused by: error decoding response body
+      Caused by: request or response body error
+      Caused by: operation timed out
+    ");
+}
+
+#[tokio::test]
+async fn retry_read_timeout_stream() {
+    let context = TestContext::new("3.12");
+
+    let (server, _guard) = read_timeout_server();
+
+    uv_snapshot!(context.filters(), context
+        .pip_install()
+        .arg(format!("{server}/tqdm-0.1-py3-none-any.whl"))
+        // Speed the test up with the minimum testable values
+        .env(EnvVars::UV_HTTP_TIMEOUT, "1")
+        .env(EnvVars::UV_HTTP_RETRIES, "1"), @"
+    success: false
+    exit_code: 1
+    ----- stdout -----
+
+    ----- stderr -----
+      × Failed to download `tqdm @ http://[LOCALHOST]/tqdm-0.1-py3-none-any.whl`
+      ├─▶ Request failed after 1 retry
+      ├─▶ Failed to read metadata: `http://[LOCALHOST]/tqdm-0.1-py3-none-any.whl`
+      ├─▶ Failed to read from zip file
+      ├─▶ an upstream reader returned an error: Failed to download distribution due to network timeout. Try increasing UV_HTTP_TIMEOUT (current value: [TIME]).
+      ╰─▶ Failed to download distribution due to network timeout. Try increasing UV_HTTP_TIMEOUT (current value: [TIME]).
+    ");
+}