Better S3 error context (#8829)

backend/windmill-common/src/client.rs

@@ -2,7 +2,7 @@ use anyhow::Context;
 use reqwest::{Body, Response};
 use serde::de::DeserializeOwned;
 
-use crate::utils::HTTP_CLIENT;
+use crate::utils::{HTTP_CLIENT, HTTP_CLIENT_STREAMING};
 
 #[derive(Clone)]
 pub struct AuthedClient {
@@ -166,14 +166,17 @@ impl AuthedClient {
         if let Some(storage) = storage {
             query.push(("storage", storage));
         }
+        let url = format!(
+            "{}/api/w/{}/job_helpers/upload_s3_file",
+            self.base_internal_url, workspace_id
+        );
+        // Use the streaming HTTP client (no total request timeout) because the
+        // upload body is streamed and total time depends on data size.
         let response = self
             .force_client
             .as_ref()
-            .unwrap_or(&HTTP_CLIENT)
-            .post(format!(
-                "{}/api/w/{}/job_helpers/upload_s3_file",
-                self.base_internal_url, workspace_id
-            ))
+            .unwrap_or(&HTTP_CLIENT_STREAMING)
+            .post(&url)
             .query(&query)
             .header(
                 reqwest::header::ACCEPT,
@@ -187,12 +190,17 @@ impl AuthedClient {
             .body(Body::wrap_stream(body))
             .send()
             .await
-            .context(format!("Sent upload_s3_file request",))
-            .map_err(|e| anyhow::anyhow!(e.to_string()))?;
+            .context(format!("Failed to send upload_s3_file request to {url}"))?;
 
         match response.status().as_u16() {
             200u16 => Ok(()),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default()))?,
+            _ => {
+                let status = response.status();
+                let body = response.text().await.unwrap_or_default();
+                Err(anyhow::anyhow!(
+                    "upload_s3_file request to {url} failed with status {status}: {body}"
+                ))
+            }
         }
     }
 

backend/windmill-common/src/utils.rs

@@ -69,6 +69,19 @@ lazy_static::lazy_static! {
 
         builder.build().unwrap()
     };
+    /// HTTP client for streaming uploads (no total request timeout, only connect timeout).
+    /// Used for S3 file uploads where the body is streamed and total time depends on data size.
+    pub static ref HTTP_CLIENT_STREAMING: Client = {
+        let mut builder = reqwest::ClientBuilder::new()
+            .user_agent("windmill/beta")
+            .connect_timeout(std::time::Duration::from_secs(10));
+
+        if *FORCE_IPV4 {
+            builder = builder.local_address(std::net::IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)));
+        }
+
+        builder.build().unwrap()
+    };
     pub static ref HTTP_CLIENT_PERMISSIVE: Client = configure_client(reqwest::ClientBuilder::new()
         .user_agent("windmill/beta")
         .connect_timeout(std::time::Duration::from_secs(10))

backend/windmill-object-store/src/lib.rs

@@ -1083,14 +1083,29 @@ pub async fn convert_json_line_stream<E: Into<anyhow::Error>>(
                 Err(e) => tracing::error!("Error in blocking task: {:?}", &e),
             };
         }
-        task::spawn_blocking(move || {
+        let close_result = task::spawn_blocking(move || {
             writer.lock().unwrap().take().unwrap().close()?;
             drop(writer);
             Ok::<_, anyhow::Error>(())
         })
-        .await??;
+        .await;
+        match close_result {
+            Ok(Ok(())) => {}
+            Ok(Err(e)) => {
+                tracing::error!("Error closing S3 stream writer: {:?}", e);
+                let _ = tx.send(Err(e)).await;
+            }
+            Err(e) => {
+                tracing::error!("S3 stream writer close task panicked: {:?}", e);
+                let _ = tx
+                    .send(Err(anyhow::anyhow!("writer close task panicked: {e}")))
+                    .await;
+            }
+        }
         drop(ctx);
-        tokio::fs::remove_file(&path).await?;
+        if let Err(e) = tokio::fs::remove_file(&path).await {
+            tracing::error!("Error removing temp file {}: {:?}", path.display(), e);
+        }
         Ok::<_, anyhow::Error>(())
     });