perf(process-executor): make execute_async genuinely concurrent

execute_async was a serial pump: it queued a job, then drove it to
completion itself via wait_id()'s blocking usleep loop before
returning, so concurrent callers never overlapped even when polled
together through FuturesUnordered.

Rewrite it as a single &self async fn mirroring the CurlDownloader/
HttpDownloader rework: a tokio Semaphore sized by max_jobs
(COMPOSER_MAX_PARALLEL_PROCESSES, PHP parity) gates admission, the
child is started non-blocking, and an async 1ms sleep loop pumps
is_running()/check_timeout() while yielding to the reactor so sibling
jobs genuinely run in parallel. The Job table, STATUS_* lifecycle,
start_job/mark_job_done/count_active_jobs/wait/wait_id all had no
remaining callers and are removed.

&self also lets callers hold only a shared borrow across their awaits
(zip_downloader, version_guesser, filesystem via the new
get_process_handle), which would otherwise panic with 'already
mutably borrowed' once two async jobs overlap on the same
Rc<RefCell<ProcessExecutor>>.

The async mock branch no longer consumes the expectation before
hitting its todo!(): the panic made that bookkeeping unobservable.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

1 files changed, 12 insertions, 3 deletions

diff --git a/crates/shirabe/src/util/filesystem.rs b/crates/shirabe/src/util/filesystem.rs
index b92bee94..e12d7fee 100644
--- a/crates/shirabe/src/util/filesystem.rs
+++ b/crates/shirabe/src/util/filesystem.rs
@@ -187,8 +187,9 @@ impl Filesystem {
             vec!["rm".to_string(), "-rf".to_string(), directory.to_string()]
         };
 
-        let mut process = self
-            .get_process()
+        let process_executor = self.get_process_handle();
+        let mut process = process_executor
+            .borrow()
             .execute_async(
                 PhpMixed::List(cmd.iter().map(|s| PhpMixed::String(s.clone())).collect()),
                 None,
@@ -887,13 +888,21 @@ impl Filesystem {
     }
 
     pub(crate) fn get_process(&mut self) -> std::cell::RefMut<'_, ProcessExecutor> {
+        self.get_process_handle();
+
+        self.process_executor.as_ref().unwrap().borrow_mut()
+    }
+
+    /// Hands out the executor handle itself so async callers can hold only a shared borrow
+    /// across their awaits (a RefMut held across an await panics once calls overlap).
+    pub(crate) fn get_process_handle(&mut self) -> std::rc::Rc<std::cell::RefCell<ProcessExecutor>> {
         if self.process_executor.is_none() {
             self.process_executor = Some(std::rc::Rc::new(std::cell::RefCell::new(
                 ProcessExecutor::new(None),
             )));
         }
 
-        self.process_executor.as_ref().unwrap().borrow_mut()
+        self.process_executor.as_ref().unwrap().clone()
     }
 
     /// delete symbolic link implementation (commonly known as "unlink()")