diff --git a/crates/tools/src/error.rs b/crates/tools/src/error.rs index 49ec6d00..70b95310 100644 --- a/crates/tools/src/error.rs +++ b/crates/tools/src/error.rs @@ -41,7 +41,8 @@ impl From for ToolError { ToolsError::WorkdirSession( workdir::WorkdirError::NotFound(_) | workdir::WorkdirError::Io { .. } - | workdir::WorkdirError::Unavailable(_), + | workdir::WorkdirError::Unavailable(_) + | workdir::WorkdirError::Transport(_), ) => ToolError::ExecutionFailed(err.to_string()), ToolsError::FileSystem(_) | ToolsError::WorkdirSession(_) diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs index 63c34662..8dd73c17 100644 --- a/crates/workdir/src/http.rs +++ b/crates/workdir/src/http.rs @@ -121,6 +121,7 @@ impl WorkdirTransportError { (Code::UnknownCommand, "Workdir command was not found") } WorkdirError::Unavailable(_) => (Code::Unavailable, "Workdir session is unavailable"), + WorkdirError::Transport(_) => (Code::Internal, "Workdir transport failed"), WorkdirError::InvalidPath(_) | WorkdirError::RelativePath(_) | WorkdirError::InvalidGlob(_) @@ -151,7 +152,8 @@ impl WorkdirTransportError { Code::Unsupported => WorkdirError::Unavailable(self.message), Code::UnknownCommand => WorkdirError::UnknownCommand("".to_string()), Code::InvalidRequest => WorkdirError::InvalidArgument(self.message), - Code::Unavailable | Code::Internal => WorkdirError::Unavailable(self.message), + Code::Unavailable => WorkdirError::Unavailable(self.message), + Code::Internal => WorkdirError::Transport(self.message), } } } @@ -485,6 +487,19 @@ pub use client::{ClientSession as RemoteWorkdirSession, WorkdirHttpAuthorization mod tests { use super::*; + #[test] + fn transport_failure_remains_distinct_from_session_unavailable() { + let transport = WorkdirTransportError::from_workdir_error(&WorkdirError::Transport( + "Workspace API request timed out".to_string(), + )); + assert_eq!(transport.code, WorkdirTransportErrorCode::Internal); + assert_eq!(transport.message, "Workdir transport failed"); + assert!(matches!( + transport.into_workdir_error(), + WorkdirError::Transport(_) + )); + } + #[test] fn transport_error_does_not_expose_host_path() { let error = WorkdirError::Io { diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 9be023a6..957c7bbd 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -270,6 +270,9 @@ pub enum WorkdirError { #[error("Workdir session is unavailable: {0}")] Unavailable(String), + #[error("Workdir transport failed: {0}")] + Transport(String), + #[error("Workdir content was modified externally before the operation could be applied: {0}")] Conflict(String), diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index dc392f1b..fd8daa44 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use worker::{ WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, @@ -291,6 +291,7 @@ pub struct RuntimeOwnedWorkspaceClient { base_url: String, runtime_id: String, worker_id: String, + request_timeout: Option, worker_remove: Option, } @@ -306,6 +307,7 @@ impl RuntimeOwnedWorkspaceClient { base_url: base_url.into().trim_end_matches('/').to_string(), runtime_id: runtime_id.into(), worker_id: worker_id.into(), + request_timeout: None, worker_remove: None, } } @@ -314,6 +316,12 @@ impl RuntimeOwnedWorkspaceClient { self.worker_remove = Some(worker_remove); self } + + #[cfg(test)] + fn with_request_timeout(mut self, request_timeout: Option) -> Self { + self.request_timeout = request_timeout; + self + } } impl std::fmt::Debug for RuntimeOwnedWorkspaceClient { @@ -351,9 +359,16 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { let base_url = self.base_url.clone(); let runtime_id = self.runtime_id.clone(); let worker_id = self.worker_id.clone(); + let request_timeout = self.request_timeout; if tokio::runtime::Handle::try_current().is_ok() { std::thread::spawn(move || { - execute_runtime_owned_workspace_http(&base_url, &runtime_id, &worker_id, request) + execute_runtime_owned_workspace_http( + &base_url, + &runtime_id, + &worker_id, + request_timeout, + request, + ) }) .join() .map_err(|_| { @@ -364,6 +379,7 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { &self.base_url, &self.runtime_id, &self.worker_id, + self.request_timeout, request, ) } @@ -397,6 +413,7 @@ fn execute_runtime_owned_workspace_http( base_url: &str, runtime_id: &str, worker_id: &str, + request_timeout: Option, request: WorkspaceRequest, ) -> Result { if !request.path.starts_with('/') || request.path.starts_with("//") { @@ -410,7 +427,16 @@ fn execute_runtime_owned_workspace_http( WorkspaceRequestMethod::Patch => reqwest::Method::PATCH, WorkspaceRequestMethod::Delete => reqwest::Method::DELETE, }; - let client = reqwest::blocking::Client::new(); + let client = reqwest::blocking::Client::builder() + .timeout(request_timeout) + .build() + .map_err(|error| { + WorkspaceClientError::Unavailable(format!( + "failed to build Workspace API HTTP client: {}", + reqwest_error_chain(&error) + )) + })?; + let request_label = format!("{method} {}", request.path); let mut request_builder = client .request(method, url) .header("x-yoi-runtime-id", runtime_id) @@ -422,14 +448,52 @@ fn execute_runtime_owned_workspace_http( } let response = request_builder .send() - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + .map_err(|error| workspace_http_error(&request_label, "waiting for response", error))?; let status = response.status().as_u16(); let body = response .text() - .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + .map_err(|error| workspace_http_error(&request_label, "reading response body", error))?; Ok(WorkspaceResponse { status, body }) } +fn workspace_http_error( + request_label: &str, + stage: &str, + error: reqwest::Error, +) -> WorkspaceClientError { + let details = reqwest_error_chain(&error); + if error.is_timeout() { + WorkspaceClientError::Request(format!( + "Workspace API {request_label} timed out while {stage}: {details}" + )) + } else if error.is_connect() { + WorkspaceClientError::Unavailable(format!( + "Workspace API {request_label} could not connect while {stage}: {details}" + )) + } else { + WorkspaceClientError::Request(format!( + "Workspace API {request_label} transport failed while {stage}: {details}" + )) + } +} + +fn reqwest_error_chain(error: &reqwest::Error) -> String { + let mut details = error.to_string(); + let mut source = std::error::Error::source(error); + for _ in 0..4 { + let Some(current) = source else { + break; + }; + let current_text = current.to_string(); + if !current_text.is_empty() && !details.ends_with(¤t_text) { + details.push_str(": "); + details.push_str(¤t_text); + } + source = std::error::Error::source(current); + } + details +} + #[derive(Debug, thiserror::Error)] pub enum RuntimeWorkerMutationForwardError { #[error(transparent)] @@ -665,6 +729,64 @@ mod tests { assert_eq!(reason, "retire obsolete Worker"); } + #[test] + fn runtime_owned_workspace_client_has_no_fixed_request_timeout() { + let (base_url, server) = delayed_workspace_response(Duration::from_millis(75)); + let client = + RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a"); + assert_eq!(client.request_timeout, None); + + let response = client + .execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + "/api/test", + "{}", + )) + .unwrap(); + assert_eq!(response.status, 200); + assert_eq!(response.body, r#"{"ok":true}"#); + server.join().unwrap(); + } + + #[test] + fn runtime_owned_workspace_client_reports_request_timeouts() { + let (base_url, server) = delayed_workspace_response(Duration::from_millis(75)); + let client = + RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a") + .with_request_timeout(Some(Duration::from_millis(20))); + + let error = client + .execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + "/api/test", + "{}", + )) + .unwrap_err(); + assert!(matches!(error, WorkspaceClientError::Request(_))); + let message = error.to_string(); + assert!(message.contains("POST /api/test"), "{message}"); + assert!(message.contains("timed out"), "{message}"); + server.join().unwrap(); + } + + fn delayed_workspace_response(delay: Duration) -> (String, std::thread::JoinHandle<()>) { + use std::io::{Read, Write}; + use std::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut bytes = [0_u8; 8192]; + let _ = stream.read(&mut bytes); + std::thread::sleep(delay); + let _ = stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}", + ); + }); + (format!("http://{address}"), server) + } + #[test] fn embedded_authority_uses_the_same_claim_contract_without_a_credential() { let authority = diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index 2f4a5392..1f592124 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -29,7 +29,10 @@ use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, ToolDeclaration, }; -use crate::worker::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse}; +use crate::worker::{ + WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, + WorkspaceResponse, +}; const FEATURE_ID: &str = "manage-workdir"; const FEATURE_NAME: &str = "Manage Workdir"; @@ -174,7 +177,7 @@ impl WorkspaceAttachedWorkdirSession { encode_path_segment(workspace_id) ), serde_json::to_string(&operation).map_err(|error| { - WorkdirError::Unavailable(format!( + WorkdirError::Transport(format!( "failed to encode Workspace Workdir operation: {error}" )) })?, @@ -182,28 +185,35 @@ impl WorkspaceAttachedWorkdirSession { let response = self .client .execute(request) - .map_err(|error| WorkdirError::Unavailable(error.to_string()))?; + .map_err(workspace_workdir_error)?; if !response.is_success() { - return Err(WorkdirError::Unavailable(format!( + return Err(WorkdirError::Transport(format!( "Workspace Workdir API returned HTTP {}: {}", response.status, bounded_error_body(&response.body) ))); } serde_json::from_str(&response.body).map_err(|error| { - WorkdirError::Unavailable(format!( + WorkdirError::Transport(format!( "failed to decode Workspace Workdir operation result: {error}" )) }) } fn mismatch(expected: &str) -> WorkdirError { - WorkdirError::Unavailable(format!( + WorkdirError::Transport(format!( "Workspace Backend returned a mismatched Workdir operation result; expected {expected}" )) } } +fn workspace_workdir_error(error: WorkspaceClientError) -> WorkdirError { + match error { + WorkspaceClientError::Unavailable(message) => WorkdirError::Unavailable(message), + other => WorkdirError::Transport(other.to_string()), + } +} + #[async_trait] impl WorkdirSession for WorkspaceAttachedWorkdirSession { fn workdir(&self) -> &Workdir { @@ -732,6 +742,25 @@ mod tests { }) } + #[test] + fn workspace_request_failure_is_not_reported_as_session_unavailable() { + let error = workspace_workdir_error(WorkspaceClientError::Request( + "Workspace API POST /workdir timed out".to_string(), + )); + assert!(matches!(error, WorkdirError::Transport(_))); + let message = error.to_string(); + assert!(message.contains("timed out"), "{message}"); + assert!(!message.contains("session is unavailable"), "{message}"); + } + + #[test] + fn workspace_unavailable_remains_a_session_error() { + let error = workspace_workdir_error(WorkspaceClientError::Unavailable( + "Workspace API could not connect".to_string(), + )); + assert!(matches!(error, WorkdirError::Unavailable(_))); + } + #[test] fn descriptor_declares_only_workdir_lifecycle_tools() { let feature = manage_workdir_feature(Arc::new(RecordingWorkspaceClient::new(Vec::new())));