refactor(loop): drive wait() promises concurrently via FuturesUnordered

Loop::wait already had the target signature and a TODO(phase-c-promise)
marker noting it drove promises serially; swap the for-loop for
FuturesUnordered so all promises are polled together instead of one at
a time, keeping the "remember only the first error" semantics.

This adds the first real use of the futures dependency (already present
in Cargo.toml/Cargo.lock from earlier prep work, now finally consumed),
so those lockfile/manifest changes land in this commit.

Real overlap still doesn't happen yet: each promise (HttpDownloader::add/
add_copy etc.) resolves through a blocking bridge (curl_runtime()/
sync_executor::block_on) that fully occupies the thread until it
settles, so this is groundwork for once a single top-level Runtime
replaces those bridges. Updated the TODO(phase-c-promise) comment to
reflect that.

4 files changed, 42 insertions, 10 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 538340cd..946f2685 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -563,6 +563,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
+name = "futures"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
 name = "futures-channel"
 version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -596,6 +611,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
 [[package]]
+name = "futures-macro"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
 name = "futures-sink"
 version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -613,8 +639,10 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -1998,6 +2026,7 @@ dependencies = [
  "async-trait",
  "base64",
  "chrono",
+ "futures",
  "indexmap",
  "jsonschema",
  "md5",
diff --git a/Cargo.toml b/Cargo.toml
index 162596d3..674fa358 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,6 +20,7 @@ async-trait = "0.1.89"
 base64 = "0.22.1"
 chrono = { version = "0.4.44", features = ["serde"] }
 fastrand = "2.4.1"
+futures = "0.3.32"
 indexmap = { version = "2.14.0", features = ["serde"] }
 jsonschema = { version = "0.46.6", default-features = false }
 md5 = "0.7.0"
diff --git a/crates/shirabe/Cargo.toml b/crates/shirabe/Cargo.toml
index 7ec9db82..00ce5211 100644
--- a/crates/shirabe/Cargo.toml
+++ b/crates/shirabe/Cargo.toml
@@ -15,6 +15,7 @@ anyhow.workspace = true
 async-trait.workspace = true
 base64.workspace = true
 chrono.workspace = true
+futures.workspace = true
 indexmap.workspace = true
 jsonschema.workspace = true
 md5.workspace = true
diff --git a/crates/shirabe/src/util/loop.rs b/crates/shirabe/src/util/loop.rs
index 4efd7707..c2fa3b08 100644
--- a/crates/shirabe/src/util/loop.rs
+++ b/crates/shirabe/src/util/loop.rs
@@ -2,6 +2,8 @@
 
 use crate::util::HttpDownloader;
 use crate::util::ProcessExecutor;
+use futures::StreamExt;
+use futures::stream::FuturesUnordered;
 use shirabe_external_packages::symfony::console::helper::ProgressBar;
 
 #[derive(Debug)]
@@ -44,25 +46,24 @@ impl Loop {
         >,
         _progress: Option<&mut ProgressBar>,
     ) -> anyhow::Result<()> {
+        let mut pending: FuturesUnordered<_> = promises.into_iter().collect();
         let mut uncaught: Option<anyhow::Error> = None;
 
-        // TODO(phase-c-promise): the asynchronous worker classes (HttpDownloader / ProcessExecutor)
-        // run single-threaded for now, so the promises are consumed serially. Once the workers run
-        // on a multi-thread runtime these futures should be driven concurrently instead of in order.
+        // TODO(phase-c-promise): promises are now polled concurrently via FuturesUnordered, but
+        // each individual future (HttpDownloader::add/add_copy etc.) still resolves through a
+        // blocking bridge (curl_runtime()/sync_executor::block_on), so real I/O overlap does not
+        // happen yet — the bridged future fully blocks the thread until it settles before the next
+        // one gets polled. That only changes once a single top-level Runtime replaces those bridges.
         // The PHP progress bar is tied to the worker active-job count and is also deferred until then.
-        for promise in promises {
-            if let Err(e) = promise.await
+        while let Some(result) = pending.next().await {
+            if let Err(e) = result
                 && uncaught.is_none()
             {
                 uncaught = Some(e);
             }
         }
 
-        if let Some(e) = uncaught {
-            return Err(e);
-        }
-
-        Ok(())
+        uncaught.map_or(Ok(()), Err)
     }
 
     pub fn abort_jobs(&self) {