diff options
| author | nsfisis <nsfisis@gmail.com> | 2026-07-17 16:37:58 +0900 |
|---|---|---|
| committer | nsfisis <nsfisis@gmail.com> | 2026-07-17 16:37:58 +0900 |
| commit | c5c527d87425ff93a4f983d36e12bd1c4b565e52 (patch) | |
| tree | 47696883b4f7bf132f6dc6ebc1db777ab57addbc | |
| parent | 2dc55eec52ee2c6993e64d8bb11c96a4f2ec5c6d (diff) | |
| download | php-shirabe-c5c527d87425ff93a4f983d36e12bd1c4b565e52.tar.gz php-shirabe-c5c527d87425ff93a4f983d36e12bd1c4b565e52.tar.zst php-shirabe-c5c527d87425ff93a4f983d36e12bd1c4b565e52.zip | |
refactor(curl-downloader): rewrite as a single async fn, drop Job/tick
Replaces the Job-table + tick()-driven polling loop with one async
download() that sends, decides (retry/redirect/fail/succeed via a new
decide() extracted from the former run_job), and loops until it
resolves — no more resolve/reject callbacks. The client switches from
reqwest::blocking::Client to the non-blocking reqwest::Client, with
body streaming now via tokio::fs.
Because real async I/O needs a live tokio reactor and none runs yet at
the process level (sync_executor::block_on is a no-reactor busy-spin
executor that only works when awaited futures resolve synchronously),
HttpDownloader::start_job drives CurlDownloader::download() through a
dedicated temporary current_thread Runtime (curl_runtime(), marked
TODO(phase-e)) instead. This keeps concurrency characteristics
unchanged for now — start_job still resolves one job at a time — real
parallel I/O lands once HttpDownloader/Loop are rearchitected on top of
FuturesUnordered.
count_active_jobs' curl.tick() polling and the Job.settled/curl_id
plumbing are removed as dead weight now that start_job settles curl
jobs synchronously, same as the rfs path already did.
abort_request is dropped: it had no caller (the PHP Promise-cancellation
flow it backs was never ported), and the job table it operated on no
longer exists.
Verified manually against real network I/O (sandbox disabled): `shirabe
show -a` (JSON metadata, in-memory body) and `shirabe create-project`
(actual dist zip download + extraction) both complete correctly with no
hang. Two unrelated pre-existing bugs surfaced during manual testing
(an event-dispatcher subscriber wiring gap during `require`, and a
RefCell reentrancy panic in `diagnose`) reproduce identically on the
pre-change code and are out of scope here.
| -rw-r--r-- | crates/shirabe/src/util/http/curl_downloader.rs | 760 | ||||
| -rw-r--r-- | crates/shirabe/src/util/http_downloader.rs | 106 | ||||
| -rw-r--r-- | crates/shirabe/src/util/sync_executor.rs | 14 |
3 files changed, 363 insertions, 517 deletions
diff --git a/crates/shirabe/src/util/http/curl_downloader.rs b/crates/shirabe/src/util/http/curl_downloader.rs index a91455c1..93661d86 100644 --- a/crates/shirabe/src/util/http/curl_downloader.rs +++ b/crates/shirabe/src/util/http/curl_downloader.rs @@ -1,15 +1,22 @@ //! ref: composer/src/Composer/Util/Http/CurlDownloader.php //! -//! reqwest-based re-implementation. The libcurl multi-handle event loop of the PHP original is -//! replaced by a blocking reqwest client: `download()` queues a job, and `tick()` drives one job to -//! completion with a synchronous HTTP request. A blocking client is used (rather than the async -//! client + a per-downloader tokio runtime) so the request never starts a tokio runtime from within -//! another one when invoked through the codebase's sync bridges (see util/sync_executor.rs). +//! reqwest-based re-implementation. Unlike the libcurl multi-handle event loop of the PHP +//! original, `download()` is a single `async fn`: it sends the request, hands the result to +//! `decide()` (retry / redirect / fail / succeed), and loops until it resolves. There is no job +//! table or `tick()` driver anymore; the caller `.await`s `download()` directly and gets the +//! final `Response` (or `TransportException`) back. //! //! The PHP control flow (insecure-URL check, redirect following, transport/status retries, //! authenticated-retry detection, max_file_size enforcement, atomic rename of the `~` temp file) //! is preserved. Per-request TLS/proxy/IP-resolve settings that reqwest only exposes per-Client //! are simplified to a single default Client; see the TODOs below. +//! +//! `abortRequest()` (PHP `CurlDownloader::abortRequest`, called from +//! `HttpDownloader.php:275` when a React\Promise consumer cancels a download) has no equivalent +//! here: shirabe has never ported the Promise/canceler machinery (`HttpDownloader::STATUS_ABORTED` +//! is likewise unused), and there is no job table left to cancel now that `download()` runs to +//! completion in one `.await`. Re-adding cancellation support belongs to whichever future task +//! ports that Promise-cancellation flow. use crate::config::Config; use crate::downloader::MaxFileSizeExceededException; @@ -30,46 +37,12 @@ use shirabe_php_shim::{ }; use std::sync::atomic::{AtomicBool, Ordering}; -/// resolve callback supplied by `HttpDownloader`. Receives the final `Response` on success. -pub type ResolveCallback = Box<dyn Fn(Response) + Send + Sync>; -/// reject callback supplied by `HttpDownloader`. Receives the recoverable error on failure. -pub type RejectCallback = Box<dyn Fn(anyhow::Error) + Send + Sync>; - -/// One in-flight download. PHP stored this as a loosely-typed `array` and additionally kept the -/// header/body stream resources and the resolve/reject callables out-of-band. Here a typed struct -/// holds everything, which is what the PHP `Job` array modelled. -struct CurlJob { - url: String, - origin: String, - attributes: IndexMap<String, PhpMixed>, - /// `options` after defaults/auth/stream-context have been merged in. - options: IndexMap<String, PhpMixed>, - /// Destination path when copying to a file (PHP `filename`), `None` for in-memory downloads. - filename: Option<String>, - resolve: ResolveCallback, - reject: RejectCallback, -} - -impl std::fmt::Debug for CurlJob { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("CurlJob") - .field("url", &self.url) - .field("origin", &self.origin) - .field("attributes", &self.attributes) - .field("options", &self.options) - .field("filename", &self.filename) - .finish() - } -} - #[derive(Debug)] pub struct CurlDownloader { /// Connection pool / cookie / TLS-session sharing — reqwest::Client handles this internally, /// replacing the PHP multiHandle + shareHandle. Redirects are disabled because we follow them /// manually (to control auth-header re-attachment), matching CURLOPT_FOLLOWLOCATION = false. - client: reqwest::blocking::Client, - jobs: IndexMap<i64, CurlJob>, - next_id: i64, + client: reqwest::Client, io: std::rc::Rc<std::cell::RefCell<dyn IOInterface>>, config: std::rc::Rc<std::cell::RefCell<Config>>, auth_helper: std::rc::Rc<std::cell::RefCell<AuthHelper>>, @@ -80,6 +53,15 @@ pub struct CurlDownloader { /// Function-static `$timeoutWarning` from `tick()`. static TIMEOUT_WARNING: AtomicBool = AtomicBool::new(false); +/// The outcome of one send attempt, decided by `decide()`. Replaces the PHP promise +/// resolve/reject callbacks (and this crate's former restart_job/resolve_job/reject_job side +/// effects): `download()`'s loop matches on this instead. +enum Decision { + Retry { url: String, delay_ms: Option<u64> }, + Done(Response), + Failed(TransportException), +} + impl CurlDownloader { /// @param mixed[] $options pub fn new( @@ -98,7 +80,7 @@ impl CurlDownloader { // (one HttpDownloader owns one CurlDownloader) but not pooled across them. // TODO: cookie sharing (CURL_LOCK_DATA_COOKIE) would need reqwest's `cookies` feature // (.cookie_store(true)); omitted as it is not required for package downloads. - let client = reqwest::blocking::Client::builder() + let client = reqwest::Client::builder() .pool_max_idle_per_host(8) .redirect(reqwest::redirect::Policy::none()) .build() @@ -113,8 +95,6 @@ impl CurlDownloader { Self { client, - jobs: IndexMap::new(), - next_id: 1, io, config, auth_helper, @@ -126,17 +106,24 @@ impl CurlDownloader { /// @param mixed[] $options /// @param non-empty-string $url /// - /// @return int internal job id - pub fn download( - &mut self, - resolve: ResolveCallback, - reject: RejectCallback, + /// Runs the request through the redirect/retry/status state machine until it resolves, + /// mirroring what the PHP promise resolver + `tick()` loop used to do together. + pub async fn download( + &self, origin: &str, url: &str, mut options: IndexMap<String, PhpMixed>, copy_to: Option<&str>, - ) -> anyhow::Result<Result<i64, TransportException>> { - let mut attributes: IndexMap<String, PhpMixed> = IndexMap::new(); + ) -> anyhow::Result<Result<Response, TransportException>> { + let mut attributes: IndexMap<String, PhpMixed> = { + let mut m = IndexMap::new(); + m.insert("retryAuthFailure".to_string(), PhpMixed::Bool(true)); + m.insert("redirects".to_string(), PhpMixed::Int(0)); + m.insert("retries".to_string(), PhpMixed::Int(0)); + m.insert("storeAuth".to_string(), PhpMixed::Bool(false)); + m.insert("ipResolve".to_string(), PhpMixed::Null); + m + }; if options.contains_key("retry-auth-failure") { attributes.insert( "retryAuthFailure".to_string(), @@ -148,37 +135,6 @@ impl CurlDownloader { options.shift_remove("retry-auth-failure"); } - self.init_download(resolve, reject, origin, url, options, copy_to, attributes) - } - - #[allow(clippy::too_many_arguments, reason = "to keep PHP signature")] - fn init_download( - &mut self, - resolve: ResolveCallback, - reject: RejectCallback, - origin: &str, - url: &str, - options: IndexMap<String, PhpMixed>, - copy_to: Option<&str>, - attributes: IndexMap<String, PhpMixed>, - ) -> anyhow::Result<Result<i64, TransportException>> { - let defaults: IndexMap<String, PhpMixed> = { - let mut m = IndexMap::new(); - m.insert("retryAuthFailure".to_string(), PhpMixed::Bool(true)); - m.insert("redirects".to_string(), PhpMixed::Int(0)); - m.insert("retries".to_string(), PhpMixed::Int(0)); - m.insert("storeAuth".to_string(), PhpMixed::Bool(false)); - m.insert("ipResolve".to_string(), PhpMixed::Null); - m - }; - let mut attributes: IndexMap<String, PhpMixed> = { - let mut m = defaults; - for (k, v) in attributes { - m.insert(k, v); - } - m - }; - if attributes .get("ipResolve") .map(|v| v.is_null()) @@ -208,8 +164,8 @@ impl CurlDownloader { } // PHP added the auth options and ran StreamContextFactory::initOptions here, and would fail - // up-front if the body temp file could not be opened. reqwest opens no file until tick(), - // so the auth/stream-context merge happens once at send time (see send_once). + // up-front if the body temp file could not be opened. reqwest opens no file until + // send_once(), so the auth/stream-context merge happens once per attempt inside the loop. let header_strings = Self::header_list(&options); let if_modified = if shirabe_php_shim::stripos( @@ -231,307 +187,297 @@ impl CurlDownloader { .map_err(|e| anyhow::anyhow!(e.message))? .and_then(|p| p.get_status(Some(" using proxy (%s)")).ok()) .unwrap_or_default(); - if attributes.get("redirects").and_then(|v| v.as_int()) == Some(0) - && attributes.get("retries").and_then(|v| v.as_int()) == Some(0) - { - self.io.write_error3( - &format!( - "Downloading {}{}{}", - Url::sanitize(url.to_string()), - using_proxy, - if_modified - ), - true, - crate::io::DEBUG, - ); - } - - let id = self.next_id; - self.next_id += 1; - self.jobs.insert( - id, - CurlJob { - url: url.to_string(), - origin: origin.to_string(), - attributes, - options, - filename: copy_to.map(|s| s.to_string()), - resolve, - reject, - }, + // `attributes.redirects == 0 && attributes.retries == 0` in PHP is always true here since + // this runs exactly once, before the retry loop below has a chance to advance either. + self.io.write_error3( + &format!( + "Downloading {}{}{}", + Url::sanitize(url.to_string()), + using_proxy, + if_modified + ), + true, + crate::io::DEBUG, ); - Ok(Ok(id)) - } - - pub fn abort_request(&mut self, id: i64) { - if let Some(job) = self.jobs.shift_remove(&id) - && let Some(filename) = &job.filename - { - unlink_silent(&format!("{}~", filename)); - } - } - - pub fn tick(&mut self) -> anyhow::Result<()> { - if self.jobs.is_empty() { - return Ok(()); - } - - // Drive every queued job to completion. The PHP multi-handle progressed all easy handles - // a little per tick(); here each tick() fully resolves one job (one blocking request), - // which is observationally equivalent for the sync wait_id() loop that calls tick(). - let ids: Vec<i64> = self.jobs.keys().copied().collect(); - for id in ids { - self.run_job(id)?; - } - Ok(()) - } - - /// Runs a single job through the redirect/retry/status state machine until it resolves or - /// rejects, invoking the stored resolve/reject callback. Mirrors the body of PHP `tick()`. - fn run_job(&mut self, id: i64) -> anyhow::Result<()> { + let mut current_url = url.to_string(); + let mut current_origin = origin.to_string(); loop { - let (url, origin, filename, options, attributes) = { - let job = match self.jobs.get(&id) { - Some(j) => j, - None => return Ok(()), - }; - ( - job.url.clone(), - job.origin.clone(), - job.filename.clone(), - job.options.clone(), - job.attributes.clone(), - ) - }; - // PHP merges auth options + stream-context options at curl_setopt time. We need the // resulting header/method/content/timeout/ssl/max_file_size, so do it here per send. let send_options = self.auth_helper.borrow_mut().add_authentication_options( options.clone(), - &origin, - &url, + ¤t_origin, + ¤t_url, )?; let send_options = - crate::util::StreamContextFactory::init_options(&url, send_options, true) + crate::util::StreamContextFactory::init_options(¤t_url, send_options, true) .map_err(|e| anyhow::anyhow!(e.message))?; - let send_result = self.send_once(&url, &send_options, filename.as_deref(), &attributes); - - let response = match send_result { - Ok(resp) => resp, - Err(transport_err) => { - // CURLE_OPERATION_TIMEDOUT one-time warning (errno 28). reqwest cannot report - // the curl errno, so this fires on the is_timeout() branch instead. - if transport_err.was_timeout && !TIMEOUT_WARNING.load(Ordering::Relaxed) { - TIMEOUT_WARNING.store(true, Ordering::Relaxed); - self.io.write_error3( - "<warning>A connection timeout was encountered. If you intend to run Composer without connecting to the internet, run the command again prefixed with COMPOSER_DISABLE_NETWORK=1 to make Composer run in offline mode.</warning>", - true, - crate::io::NORMAL, - ); - } + let send_result = self + .send_once(¤t_url, &send_options, copy_to, &attributes) + .await; - // PHP retried on a set of curl errnos (7/16/92/6/28 and 56/35 with "Connection - // reset by peer"); reqwest does not expose those errnos, so approximate with - // is_connect/is_timeout/is_request on GET requests. - let retries = attributes - .get("retries") - .and_then(|v| v.as_int()) - .unwrap_or(0); - if transport_err.retryable - && Self::method_is_get(&options) - && retries < self.max_retries - { - let mut new_attrs = attributes.clone(); - new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1)); - // CURLE_COULDNT_CONNECT analogue: force IPv4 if no IP stack chosen. - if transport_err.is_connect && !attributes.contains_key("ipResolve") { - new_attrs.insert("ipResolve".to_string(), PhpMixed::Int(4)); - } - self.io.write_error3( - &format!( - "Retrying ({}) {} due to connection error", - retries + 1, - Url::sanitize(url.clone()) - ), - true, - crate::io::DEBUG, - ); - self.restart_job_with_delay(id, &url, new_attrs); - continue; - } - // PHP throws a MaxFileSizeExceededException (a TransportException subclass) with - // the raw "Maximum allowed download size reached..." message; preserve it - // verbatim rather than wrapping it in the generic curl-error text. - let message = if transport_err.is_max_file_size { - MaxFileSizeExceededException(TransportException::new( - transport_err.message.clone(), - 0, - )) - .0 - .message - } else { - TransportException::new( - format!( - "curl error while downloading {}: {}", - Url::sanitize(url.clone()), - transport_err.message - ), - 0, - ) - .message - }; - self.reject_job(id, anyhow::anyhow!(message)); - return Ok(()); - } - }; - - let status_code = response.status; - - let curl_response = response.into_curl_response(&url); - - self.io.write_error3( - &format!("[{}] {}", status_code, Url::sanitize(url.clone())), - true, - crate::io::DEBUG, - ); - - // Output JSON warnings (PHP HttpDownloader::outputWarnings) for >= 300 JSON bodies. - if curl_response.inner.get_status_code() >= 300 - && curl_response.inner.get_header("content-type").as_deref() - == Some("application/json") - && let Some(body) = curl_response.inner.get_body() + match self + .decide( + send_result, + ¤t_url, + ¤t_origin, + copy_to, + &options, + &mut attributes, + ) + .await? { - let decoded = shirabe_php_shim::json_decode(body, true)?; - if let PhpMixed::Array(a) = decoded { - HttpDownloader::output_warnings(self.io.clone(), &origin, &a)?; - } - } - - // Authenticated-retry detection (401/403, Bitbucket login page, GitLab archive 404). - let auth_result = self.is_authenticated_retry_needed( - &url, - &origin, - filename.as_deref(), - &attributes, - &curl_response, - )?; - match auth_result { - Ok(prompt) if prompt.retry => { - let mut new_attrs = attributes.clone(); - new_attrs.insert( - "storeAuth".to_string(), - match prompt.store_auth { - StoreAuth::Bool(b) => PhpMixed::Bool(b), - StoreAuth::Prompt => PhpMixed::String("prompt".to_string()), - }, - ); - let retries = attributes - .get("retries") - .and_then(|v| v.as_int()) - .unwrap_or(0); - new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1)); - self.restart_job(id, &url, new_attrs); + Decision::Retry { + url: new_url, + delay_ms, + } => { + if let Some(ms) = delay_ms { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + } + current_url = new_url; + current_origin = Url::get_origin(&self.config.borrow(), ¤t_url); continue; } - Ok(_) => {} - Err(e) => { - self.reject_job(id, anyhow::anyhow!(e.message)); - return Ok(()); - } + Decision::Done(response) => return Ok(Ok(response)), + Decision::Failed(e) => return Ok(Err(e)), } + } + } - // Handle 3xx redirects, 304 Not Modified excluded. - let redirects = attributes - .get("redirects") - .and_then(|v| v.as_int()) - .unwrap_or(0); - if (300..=399).contains(&status_code) - && status_code != 304 - && redirects < self.max_redirects - { - match self.handle_redirect(&url, &attributes, &curl_response)? { - Ok(location) if !location.is_empty() => { - let mut new_attrs = attributes.clone(); - new_attrs.insert("redirects".to_string(), PhpMixed::Int(redirects + 1)); - // The redirect target becomes the new url; origin is recomputed in restart. - self.restart_job(id, &location, new_attrs); - continue; - } - Ok(_) => {} - Err(e) => { - self.reject_job(id, anyhow::anyhow!(e.message)); - return Ok(()); - } + /// Decides what to do with one send attempt's result: retry (transport error, auth-retry, + /// redirect, retryable status code), fail, or succeed. Mirrors the body of the former + /// `run_job` loop, minus the resolve/reject/restart side effects, which are now expressed as + /// the returned `Decision`. + async fn decide( + &self, + send_result: Result<RawResponse, TransportError>, + url: &str, + origin: &str, + filename: Option<&str>, + options: &IndexMap<String, PhpMixed>, + attributes: &mut IndexMap<String, PhpMixed>, + ) -> anyhow::Result<Decision> { + let response = match send_result { + Ok(resp) => resp, + Err(transport_err) => { + // CURLE_OPERATION_TIMEDOUT one-time warning (errno 28). reqwest cannot report + // the curl errno, so this fires on the is_timeout() branch instead. + if transport_err.was_timeout && !TIMEOUT_WARNING.load(Ordering::Relaxed) { + TIMEOUT_WARNING.store(true, Ordering::Relaxed); + self.io.write_error3( + "<warning>A connection timeout was encountered. If you intend to run Composer without connecting to the internet, run the command again prefixed with COMPOSER_DISABLE_NETWORK=1 to make Composer run in offline mode.</warning>", + true, + crate::io::NORMAL, + ); } - } - // Fail 4xx and 5xx responses (some are retried on GET). - if (400..=599).contains(&status_code) { + // PHP retried on a set of curl errnos (7/16/92/6/28 and 56/35 with "Connection + // reset by peer"); reqwest does not expose those errnos, so approximate with + // is_connect/is_timeout/is_request on GET requests. let retries = attributes .get("retries") .and_then(|v| v.as_int()) .unwrap_or(0); - if Self::method_is_get(&options) - && in_array( - PhpMixed::Int(status_code), - &PhpMixed::List( - [423, 425, 500, 502, 503, 504, 507, 510] - .iter() - .map(|c| PhpMixed::Int(*c)) - .collect(), - ), - true, - ) + if transport_err.retryable + && Self::method_is_get(options) && retries < self.max_retries { + attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1)); + // CURLE_COULDNT_CONNECT analogue: force IPv4 if no IP stack chosen. + if transport_err.is_connect && !attributes.contains_key("ipResolve") { + attributes.insert("ipResolve".to_string(), PhpMixed::Int(4)); + } self.io.write_error3( &format!( - "Retrying ({}) {} due to status code {}", + "Retrying ({}) {} due to connection error", retries + 1, - Url::sanitize(url.clone()), - status_code + Url::sanitize(url.to_string()) ), true, crate::io::DEBUG, ); - let mut new_attrs = attributes.clone(); - new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1)); - self.restart_job_with_delay(id, &url, new_attrs); - continue; + if let Some(filename) = filename { + unlink_silent(&format!("{}~", filename)); + } + return Ok(Decision::Retry { + url: url.to_string(), + delay_ms: Self::retry_delay_ms(retries + 1), + }); } - let status_msg = curl_response.inner.get_status_message().unwrap_or_default(); - let e = self.fail_response(&url, filename.as_deref(), &curl_response, &status_msg); - self.reject_job_with_response(id, e, &curl_response); - return Ok(()); + if let Some(filename) = filename { + unlink_silent(&format!("{}~", filename)); + } + // PHP throws a MaxFileSizeExceededException (a TransportException subclass) with + // the raw "Maximum allowed download size reached..." message verbatim rather than + // wrapping it in the generic curl-error text. + if transport_err.is_max_file_size { + return Ok(Decision::Failed( + MaxFileSizeExceededException(TransportException::new( + transport_err.message, + 0, + )) + .0, + )); + } + return Ok(Decision::Failed(TransportException::new( + format!( + "curl error while downloading {}: {}", + Url::sanitize(url.to_string()), + transport_err.message + ), + 0, + ))); } + }; + + let status_code = response.status; + let curl_response = response.into_curl_response(url); - // storeAuth on success. - let store_auth = attributes.get("storeAuth").cloned(); - if !matches!(store_auth, Some(PhpMixed::Bool(false))) { - let store_auth = match store_auth { - Some(PhpMixed::String(ref s)) if s == "prompt" => StoreAuth::Prompt, - Some(PhpMixed::Bool(b)) => StoreAuth::Bool(b), - _ => StoreAuth::Bool(false), - }; - self.auth_helper.borrow().store_auth(&origin, store_auth)?; + self.io.write_error3( + &format!("[{}] {}", status_code, Url::sanitize(url.to_string())), + true, + crate::io::DEBUG, + ); + + // Output JSON warnings (PHP HttpDownloader::outputWarnings) for >= 300 JSON bodies. + if curl_response.inner.get_status_code() >= 300 + && curl_response.inner.get_header("content-type").as_deref() == Some("application/json") + && let Some(body) = curl_response.inner.get_body() + { + let decoded = shirabe_php_shim::json_decode(body, true)?; + if let PhpMixed::Array(a) = decoded { + HttpDownloader::output_warnings(self.io.clone(), origin, &a)?; } + } - // Atomic rename of the `~` temp file to its final name (file mode). - if let Some(filename) = &filename { - rename(format!("{}~", filename), filename); + // Authenticated-retry detection (401/403, Bitbucket login page, GitLab archive 404). + let auth_result = + self.is_authenticated_retry_needed(url, origin, filename, attributes, &curl_response)?; + match auth_result { + Ok(prompt) if prompt.retry => { + attributes.insert( + "storeAuth".to_string(), + match prompt.store_auth { + StoreAuth::Bool(b) => PhpMixed::Bool(b), + StoreAuth::Prompt => PhpMixed::String("prompt".to_string()), + }, + ); + let retries = attributes + .get("retries") + .and_then(|v| v.as_int()) + .unwrap_or(0); + attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1)); + if let Some(filename) = filename { + unlink_silent(&format!("{}~", filename)); + } + return Ok(Decision::Retry { + url: url.to_string(), + delay_ms: None, + }); } + Ok(_) => {} + Err(e) => return Ok(Decision::Failed(e)), + } - self.resolve_job(id, curl_response.inner); - return Ok(()); + // Handle 3xx redirects, 304 Not Modified excluded. + let redirects = attributes + .get("redirects") + .and_then(|v| v.as_int()) + .unwrap_or(0); + if (300..=399).contains(&status_code) + && status_code != 304 + && redirects < self.max_redirects + { + match self.handle_redirect(url, attributes, &curl_response)? { + Ok(location) if !location.is_empty() => { + attributes.insert("redirects".to_string(), PhpMixed::Int(redirects + 1)); + if let Some(filename) = filename { + unlink_silent(&format!("{}~", filename)); + } + return Ok(Decision::Retry { + url: location, + delay_ms: None, + }); + } + Ok(_) => {} + Err(e) => return Ok(Decision::Failed(e)), + } } + + // Fail 4xx and 5xx responses (some are retried on GET). + if (400..=599).contains(&status_code) { + let retries = attributes + .get("retries") + .and_then(|v| v.as_int()) + .unwrap_or(0); + if Self::method_is_get(options) + && in_array( + PhpMixed::Int(status_code), + &PhpMixed::List( + [423, 425, 500, 502, 503, 504, 507, 510] + .iter() + .map(|c| PhpMixed::Int(*c)) + .collect(), + ), + true, + ) + && retries < self.max_retries + { + self.io.write_error3( + &format!( + "Retrying ({}) {} due to status code {}", + retries + 1, + Url::sanitize(url.to_string()), + status_code + ), + true, + crate::io::DEBUG, + ); + attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1)); + if let Some(filename) = filename { + unlink_silent(&format!("{}~", filename)); + } + return Ok(Decision::Retry { + url: url.to_string(), + delay_ms: Self::retry_delay_ms(retries + 1), + }); + } + + let status_msg = curl_response.inner.get_status_message().unwrap_or_default(); + // Enrich with the response headers/status/body, mirroring PHP's catch block that + // calls setHeaders/setStatusCode/setResponse before reject(). + let mut e = self.fail_response(url, filename, &curl_response, &status_msg); + e.set_headers(curl_response.inner.get_headers().clone()); + e.set_status_code(Some(curl_response.inner.get_status_code())); + e.set_response(curl_response.inner.get_body().map(|s| s.to_string())); + return Ok(Decision::Failed(e)); + } + + // storeAuth on success. + let store_auth = attributes.get("storeAuth").cloned(); + if !matches!(store_auth, Some(PhpMixed::Bool(false))) { + let store_auth = match store_auth { + Some(PhpMixed::String(ref s)) if s == "prompt" => StoreAuth::Prompt, + Some(PhpMixed::Bool(b)) => StoreAuth::Bool(b), + _ => StoreAuth::Bool(false), + }; + self.auth_helper.borrow().store_auth(origin, store_auth)?; + } + + // Atomic rename of the `~` temp file to its final name (file mode). + if let Some(filename) = filename { + rename(format!("{}~", filename), filename); + } + + Ok(Decision::Done(curl_response.inner)) } - /// Performs one blocking HTTP request via reqwest, enforcing max_file_size and streaming the - /// body to the `~` temp file when in file mode. Replaces PHP's curl_setopt block + curl I/O. - fn send_once( + /// Performs one non-blocking HTTP request via reqwest, enforcing max_file_size and streaming + /// the body to the `~` temp file when in file mode. Replaces PHP's curl_setopt block + curl I/O. + async fn send_once( &self, url: &str, options: &IndexMap<String, PhpMixed>, @@ -584,7 +530,7 @@ impl CurlDownloader { builder = builder.body(body); } - let resp = builder.send().map_err(|e| TransportError { + let resp = builder.send().await.map_err(|e| TransportError { message: e.to_string(), retryable: e.is_timeout() || e.is_connect() || e.is_request(), is_connect: e.is_connect(), @@ -611,15 +557,15 @@ impl CurlDownloader { headers_out.push(format!("{}: {}", k, v.to_str().unwrap_or(""))); } - let body = Self::read_body_with_limit(resp, max_file_size, filename).map_err( - |(message, is_max_file_size)| TransportError { + let body = Self::read_body_with_limit(resp, max_file_size, filename) + .await + .map_err(|(message, is_max_file_size)| TransportError { message, retryable: false, is_connect: false, was_timeout: false, is_max_file_size, - }, - )?; + })?; Ok(RawResponse { status, @@ -630,35 +576,34 @@ impl CurlDownloader { /// Reads the body, enforcing max_file_size, writing to the `~` temp file when in file mode. /// The `bool` in the error is `true` when the failure is a max_file_size violation. - fn read_body_with_limit( - resp: reqwest::blocking::Response, + async fn read_body_with_limit( + mut resp: reqwest::Response, max_file_size: Option<u64>, filename: Option<&str>, ) -> Result<Body, (String, bool)> { - use std::io::{Read as _, Write as _}; + use tokio::io::AsyncWriteExt as _; - let mut stream = resp; let mut written: u64 = 0; enum Sink { - File(std::fs::File), + File(tokio::fs::File), Memory(Vec<u8>), } let mut sink = match filename { Some(f) => Sink::File( - std::fs::File::create(format!("{}~", f)).map_err(|e| (e.to_string(), false))?, + tokio::fs::File::create(format!("{}~", f)) + .await + .map_err(|e| (e.to_string(), false))?, ), None => Sink::Memory(Vec::new()), }; - let mut buffer = [0u8; 16 * 1024]; loop { - let n = match stream.read(&mut buffer) { - Ok(0) => break, - Ok(n) => n, + let chunk = match resp.chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => break, Err(e) => return Err((e.to_string(), false)), }; - let chunk = &buffer[..n]; - written += n as u64; + written += chunk.len() as u64; if let Some(max) = max_file_size && written > max { @@ -671,8 +616,11 @@ impl CurlDownloader { )); } match &mut sink { - Sink::File(f) => f.write_all(chunk).map_err(|e| (e.to_string(), false))?, - Sink::Memory(buf) => buf.extend_from_slice(chunk), + Sink::File(f) => f + .write_all(&chunk) + .await + .map_err(|e| (e.to_string(), false))?, + Sink::Memory(buf) => buf.extend_from_slice(&chunk), } } @@ -755,7 +703,7 @@ impl CurlDownloader { } fn is_authenticated_retry_needed( - &mut self, + &self, url: &str, origin: &str, filename: Option<&str>, @@ -856,55 +804,17 @@ impl CurlDownloader { })) } - fn restart_job(&mut self, id: i64, url: &str, attributes: IndexMap<String, PhpMixed>) { - let filename = match self.jobs.get(&id) { - Some(job) => job.filename.clone(), - None => return, - }; - if let Some(filename) = &filename { - unlink_silent(&format!("{}~", filename)); - } - - // Merge the new attributes over the job's existing ones. - let merged = { - let mut m = match self.jobs.get(&id) { - Some(job) => job.attributes.clone(), - None => return, - }; - for (k, v) in attributes { - m.insert(k, v); - } - m - }; - let origin = Url::get_origin(&self.config.borrow(), url); - - // options/filename/resolve/reject are preserved across the restart, mirroring PHP forwarding - // the original job's resolve/reject into the restarted download; only url/origin/attributes - // change. - if let Some(job) = self.jobs.get_mut(&id) { - job.url = url.to_string(); - job.origin = origin; - job.attributes = merged; - } - } - - fn restart_job_with_delay( - &mut self, - id: i64, - url: &str, - attributes: IndexMap<String, PhpMixed>, - ) { - let retries = attributes - .get("retries") - .and_then(|v| v.as_int()) - .unwrap_or(0); + /// The delay `restart_job_with_delay` used to sleep before restarting a retried job: half a + /// second from the 3rd retry onward, 100ms for the 2nd, none for the 1st. `retries` is the + /// post-increment retry count. + fn retry_delay_ms(retries: i64) -> Option<u64> { if retries >= 3 { - shirabe_php_shim::usleep(500000); // half a second delay for 3rd retry and beyond + Some(500) } else if retries >= 2 { - shirabe_php_shim::usleep(100000); // 100ms delay for 2nd retry - } // no sleep for the first retry - - self.restart_job(id, url, attributes); + Some(100) + } else { + None + } } fn fail_response( @@ -955,40 +865,6 @@ impl CurlDownloader { ) } - /// Invokes the stored resolve callback and removes the job. - fn resolve_job(&mut self, id: i64, response: Response) { - if let Some(job) = self.jobs.shift_remove(&id) { - (job.resolve)(response); - } - } - - /// Invokes the stored reject callback and removes the job, deleting the temp file. - fn reject_job(&mut self, id: i64, e: anyhow::Error) { - if let Some(job) = self.jobs.shift_remove(&id) { - if let Some(filename) = &job.filename { - unlink_silent(&format!("{}~", filename)); - } - (job.reject)(e); - } - } - - /// Reject after enriching the TransportException with the response headers/status/body, mirroring - /// PHP's catch block that calls setHeaders/setStatusCode/setResponse before reject(). - fn reject_job_with_response( - &mut self, - id: i64, - mut e: TransportException, - response: &CurlResponse, - ) { - e.set_headers(response.inner.get_headers().clone()); - e.set_status_code(Some(response.inner.get_status_code())); - e.set_response(response.inner.get_body().map(|s| s.to_string())); - let msg = e.message.clone(); - // Carry the enriched exception through anyhow; the typed payload is reconstructed from the - // message on the HttpDownloader side. TransportException is the recoverable error here. - self.reject_job(id, anyhow::Error::new(e).context(msg)); - } - fn method_is_get(options: &IndexMap<String, PhpMixed>) -> bool { let method = options .get("http") diff --git a/crates/shirabe/src/util/http_downloader.rs b/crates/shirabe/src/util/http_downloader.rs index 448c7019..c2c83b85 100644 --- a/crates/shirabe/src/util/http_downloader.rs +++ b/crates/shirabe/src/util/http_downloader.rs @@ -97,34 +97,15 @@ impl Default for HttpDownloaderMockHandler { } } +#[derive(Debug)] struct Job { id: i64, status: i64, request: Request, sync: bool, origin: String, - curl_id: Option<i64>, response: Option<Response>, exception: Option<anyhow::Error>, - /// Completion slot written by the curl resolve/reject closures (driven by `curl.tick()`) - /// and read by `count_active_jobs`. Uses `Arc<Mutex>` because `CurlDownloader::download` - /// requires `Send + Sync` callbacks. - settled: std::sync::Arc<std::sync::Mutex<Option<anyhow::Result<Response>>>>, -} - -impl std::fmt::Debug for Job { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Job") - .field("id", &self.id) - .field("status", &self.status) - .field("request", &self.request) - .field("sync", &self.sync) - .field("origin", &self.origin) - .field("curl_id", &self.curl_id) - .field("response", &self.response) - .field("exception", &self.exception) - .finish() - } } #[derive(Debug, Clone)] @@ -134,6 +115,24 @@ struct Request { copy_to: Option<String>, } +/// A single-threaded tokio Runtime used only to drive `CurlDownloader::download()` (which needs a +/// real reactor now that it uses the non-blocking `reqwest::Client`, unlike `sync_executor::block_on` +/// which assumes every awaited future resolves synchronously). `current_thread` is used because +/// `block_on` (unlike `spawn`) has no `Send` bound, and `download()`'s future closes over +/// `Rc<RefCell<...>>` handles that are not `Send`. +/// +/// TODO(phase-e): remove this once `HttpDownloader::add`/`get` are driven by `Loop::wait`'s +/// `FuturesUnordered` under a single top-level Runtime (see the async re-architecture design). +fn curl_runtime() -> &'static tokio::runtime::Runtime { + static RT: std::sync::LazyLock<tokio::runtime::Runtime> = std::sync::LazyLock::new(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("failed to build the temporary CurlDownloader bridge runtime") + }); + &RT +} + impl HttpDownloader { const STATUS_QUEUED: i64 = 1; const STATUS_STARTED: i64 = 2; @@ -397,10 +396,8 @@ impl HttpDownloader { request: request.clone(), sync, origin, - curl_id: None, response: None, exception: None, - settled: std::sync::Arc::new(std::sync::Mutex::new(None)), }; let can_use_curl = self.can_use_curl(&job); self.jobs.insert(id, job); @@ -560,38 +557,20 @@ impl HttpDownloader { return; } - // curl branch: register the request with the curl multi handle. Completion is delivered - // by curl.tick() into the job's `settled` slot (read by count_active_jobs). The resolve - // callback stores the Response, reject stores the error; this mirrors PHP's promise - // resolve/reject firing during tick(). PHP catches any exception from download() and - // rejects the job. - let settled = self.jobs.get(&id).unwrap().settled.clone(); - let settled_for_reject = settled.clone(); - let resolve: Box<dyn Fn(Response) + Send + Sync> = Box::new(move |response: Response| { - *settled.lock().unwrap() = Some(Ok(response)); - }); - let reject: Box<dyn Fn(anyhow::Error) + Send + Sync> = - Box::new(move |error: anyhow::Error| { - *settled_for_reject.lock().unwrap() = Some(Err(error)); - }); - - let download_result = { - let curl = self.curl.as_mut().unwrap(); - curl.download(resolve, reject, &origin, &url, options, copy_to.as_deref()) - }; - match download_result { - Ok(Ok(curl_id)) => { - if let Some(job) = self.jobs.get_mut(&id) { - job.curl_id = Some(curl_id); - } - } - Ok(Err(e)) => { - self.settle_job(id, Err(e.into())); - } - Err(e) => { - self.settle_job(id, Err(e)); + // curl branch: `CurlDownloader::download` now runs the whole redirect/retry/auth state + // machine to completion itself and returns the settled result directly, so this drives + // it through the temporary bridge runtime instead of PHP's promise resolve/reject firing + // during tick(). PHP catches any exception from download() and rejects the job. + let result: anyhow::Result<Response> = { + let curl = self.curl.as_ref().unwrap(); + match curl_runtime().block_on(curl.download(&origin, &url, options, copy_to.as_deref())) + { + Ok(Ok(response)) => Ok(response), + Ok(Err(transport_exception)) => Err(transport_exception.into()), + Err(e) => Err(e), } - } + }; + self.settle_job(id, result); } fn mark_job_done(&mut self) { @@ -637,24 +616,9 @@ impl HttpDownloader { } } - if let Some(curl) = self.curl.as_mut() { - curl.tick()?; - } - - // Apply completions delivered by curl.tick() into each started job's `settled` slot. - // This reproduces the effect of PHP's resolve/reject callbacks firing during tick(). - let started_ids: Vec<i64> = self - .jobs - .values() - .filter(|j| j.status == Self::STATUS_STARTED) - .map(|j| j.id) - .collect(); - for id in started_ids { - let settled = self.jobs.get(&id).unwrap().settled.lock().unwrap().take(); - if let Some(result) = settled { - self.settle_job(id, result); - } - } + // Unlike the old tick()-driven curl path, `start_job` now settles curl jobs synchronously + // (via the temporary bridge runtime), so no job is ever left lingering in STATUS_STARTED + // by the time we get here — nothing left to poll or collect. if let Some(index) = index { return Ok( diff --git a/crates/shirabe/src/util/sync_executor.rs b/crates/shirabe/src/util/sync_executor.rs index b7512c2d..ee0995fc 100644 --- a/crates/shirabe/src/util/sync_executor.rs +++ b/crates/shirabe/src/util/sync_executor.rs @@ -1,10 +1,16 @@ //! Minimal synchronous future executor used as a drop-in for the `tokio::runtime::Runtime::new()` //! + `block_on` sync bridges scattered across the codebase (repository / installer / downloader). //! -//! Those bridges drive `async fn`s whose `.await` points all resolve synchronously: the only async -//! I/O (reqwest in `CurlDownloader`) is performed through a blocking client, so no reactor is -//! required. Nesting `tokio` runtimes is forbidden ("Cannot start a runtime from within a runtime"), -//! which is why this no-reactor executor exists; it can be nested freely. +//! Those bridges drive `async fn`s whose `.await` points all resolve synchronously — this relies +//! on the invariant below. `CurlDownloader` no longer qualifies: it now performs real async I/O +//! via a non-blocking `reqwest::Client`, so `HttpDownloader` drives it through its own dedicated +//! `tokio::runtime::Runtime` instead (see `http_downloader::curl_runtime`), not through this +//! module. The other call sites (`file_downloader.rs`, `version_guesser.rs`, +//! `installation_manager.rs`, `composer_repository.rs`, `sync_helper.rs`) still rely on this +//! module because none of their awaited futures actually park on a reactor. +//! +//! Nesting `tokio` runtimes is forbidden ("Cannot start a runtime from within a runtime"), which is +//! why this no-reactor executor exists for those remaining call sites; it can be nested freely. //! //! TODO(phase-e): remove this module once the async bridges are either made genuinely synchronous or //! consolidated onto a single shared runtime driven from `main`. |
