From 7dfc0324a8db5dc554fc8700d31609dd0d3f9b84 Mon Sep 17 00:00:00 2001 From: xlx1212 Date: Sat, 8 Aug 2026 19:01:32 +0800 Subject: [PATCH 1/2] fix(miniapp): allow concurrent RPCs per worker by narrowing lock scope The pool held each app's Arc> for the whole RPC, so one long call queued every other call to that app for minutes, even though JsWorker already routes responses concurrently via its id-keyed pending map. Drop the outer mutex (Arc): the only serialized section is now the stdin write inside JsWorker::call. The old mutex also doubled as the busy signal (try_lock failure) for the idle reaper, LRU eviction, and admission. Replace that with an explicit in-flight counter maintained by an RAII guard that also clears the pending entry and refreshes last_activity on every exit path, so a long quiet call can no longer be reaped mid-flight. On stderr EOF the worker now fails all in-flight calls with "Worker exited" instead of letting each caller run out its own timeout. Co-Authored-By: Claude Fable 5 --- .../core/src/miniapp/js_worker_pool.rs | 5 +- .../src/miniapp/worker.rs | 206 +++++++++++++----- .../src/miniapp/worker_pool.rs | 46 ++-- 3 files changed, 172 insertions(+), 85 deletions(-) diff --git a/src/crates/assembly/core/src/miniapp/js_worker_pool.rs b/src/crates/assembly/core/src/miniapp/js_worker_pool.rs index c41e0b9abc..206556ea02 100644 --- a/src/crates/assembly/core/src/miniapp/js_worker_pool.rs +++ b/src/crates/assembly/core/src/miniapp/js_worker_pool.rs @@ -19,7 +19,6 @@ use bitfun_services_integrations::miniapp::worker_pool::{ use serde_json::Value; use std::path::{Path, PathBuf}; use std::sync::Arc; -use tokio::sync::Mutex; pub struct JsWorkerPool { inner: ServiceJsWorkerPool, @@ -50,7 +49,7 @@ impl JsWorkerPool { worker_revision: &str, policy_json: &str, node_perms: Option<&NodePermissions>, - ) -> BitFunResult>> { + ) -> BitFunResult> { self.inner .get_or_spawn(app_id, worker_revision, policy_json, node_perms) .await @@ -65,7 +64,7 @@ impl JsWorkerPool { worker_revision: &str, policy_json: &str, node_perms: Option<&NodePermissions>, - ) -> BitFunResult>> { + ) -> BitFunResult> { self.inner .get_or_spawn_with_app_dir( worker_key, diff --git a/src/crates/services/services-integrations/src/miniapp/worker.rs b/src/crates/services/services-integrations/src/miniapp/worker.rs index 1c8ededb03..12d5c24971 100644 --- a/src/crates/services/services-integrations/src/miniapp/worker.rs +++ b/src/crates/services/services-integrations/src/miniapp/worker.rs @@ -1,4 +1,11 @@ //! JS Worker — single child process (Bun/Node) with stdin/stderr JSON-RPC. +//! +//! Concurrency model: `call` takes `&self` and locks stdin only while writing +//! the request line; responses are routed by id through the `pending` map, so +//! multiple RPCs on the same worker proceed independently. An in-flight +//! counter marks the worker busy — the pool must never idle-reap or LRU-evict +//! a worker with pending RPCs, since `last_activity` alone goes stale during +//! long calls that emit no stderr output. use bitfun_product_domains::miniapp::runtime::DetectedRuntime; use serde_json::Value; @@ -6,8 +13,8 @@ use std::collections::HashMap; use std::future::Future; use std::path::Path; use std::pin::Pin; -use std::sync::atomic::{AtomicI64, Ordering}; -use std::sync::Arc; +use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex as StdMutex, MutexGuard}; use std::time::Duration; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::{Child, ChildStdin}; @@ -31,12 +38,41 @@ pub trait MiniAppWorkerEventSink: Send + Sync { pub type SharedMiniAppWorkerEventSink = Arc; +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +fn lock_pending(pending: &StdMutex) -> MutexGuard<'_, PendingResponseMap> { + pending.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// Removes the pending entry and drops the busy mark on every exit path of +/// `call`, including timeout and caller cancellation. +struct InflightGuard { + id: String, + pending: Arc>, + inflight: Arc, + last_activity: Arc, +} + +impl Drop for InflightGuard { + fn drop(&mut self) { + lock_pending(&self.pending).remove(&self.id); + self.inflight.fetch_sub(1, Ordering::SeqCst); + self.last_activity.store(now_ms(), Ordering::SeqCst); + } +} + /// Single JS Worker process: stdin for requests, stderr for RPC responses, stdout for user logs. pub struct JsWorker { - _child: Child, + child: Mutex, stdin: Mutex>, - pending: Arc>, + pending: Arc>, last_activity: Arc, + inflight: Arc, } impl JsWorker { @@ -67,13 +103,8 @@ impl JsWorker { let stderr = child.stderr.take().ok_or("No stderr")?; let _stdout = child.stdout.take(); - let pending = Arc::new(Mutex::new(PendingResponseMap::new())); - let last_activity = Arc::new(AtomicI64::new( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64, - )); + let pending = Arc::new(StdMutex::new(PendingResponseMap::new())); + let last_activity = Arc::new(AtomicI64::new(now_ms())); let pending_clone = pending.clone(); let last_activity_clone = last_activity.clone(); @@ -84,15 +115,7 @@ impl JsWorker { if line.is_empty() { continue; } - let _ = - last_activity_clone.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |_| { - Some( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64, - ) - }); + last_activity_clone.store(now_ms(), Ordering::SeqCst); let msg: Value = match serde_json::from_str(&line) { Ok(v) => v, Err(_) => continue, @@ -112,8 +135,8 @@ impl JsWorker { .cloned() .ok_or_else(|| "Missing result".to_string()) }; - let mut guard = pending_clone.lock().await; - if let Some(tx) = guard.remove(&id) { + let tx = lock_pending(&pending_clone).remove(&id); + if let Some(tx) = tx { let _ = tx.send(result); } continue; @@ -133,17 +156,29 @@ impl JsWorker { .await; } } + + // stderr EOF: the process died — fail all in-flight calls now + // instead of letting each caller run out its own timeout. + let waiters: Vec = { + let mut guard = lock_pending(&pending_clone); + guard.drain().map(|(_, tx)| tx).collect() + }; + for tx in waiters { + let _ = tx.send(Err("Worker exited".to_string())); + } }); Ok(Self { - _child: child, + child: Mutex::new(child), stdin: Mutex::new(Some(stdin_handle)), pending, last_activity, + inflight: Arc::new(AtomicUsize::new(0)), }) } /// Send a JSON-RPC request and wait for the response (with timeout). + /// Concurrent calls interleave: the stdin lock covers only the write. pub async fn call( &self, method: &str, @@ -160,51 +195,112 @@ impl JsWorker { let line = serde_json::to_string(&request).map_err(|e| e.to_string())? + "\n"; let (tx, rx) = oneshot::channel(); + self.inflight.fetch_add(1, Ordering::SeqCst); + let _inflight = InflightGuard { + id: id.clone(), + pending: Arc::clone(&self.pending), + inflight: Arc::clone(&self.inflight), + last_activity: Arc::clone(&self.last_activity), + }; + lock_pending(&self.pending).insert(id, tx); + self.last_activity.store(now_ms(), Ordering::SeqCst); + { - let mut guard = self.pending.lock().await; - guard.insert(id.clone(), tx); + let mut stdin_guard = self.stdin.lock().await; + let stdin = stdin_guard.as_mut().ok_or("Worker stdin closed")?; + use tokio::io::AsyncWriteExt; + stdin + .write_all(line.as_bytes()) + .await + .map_err(|e| e.to_string())?; + stdin.flush().await.map_err(|e| e.to_string())?; } - self.last_activity.store( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64, - Ordering::SeqCst, - ); - - let mut stdin_guard = self.stdin.lock().await; - let stdin = stdin_guard.as_mut().ok_or("Worker stdin closed")?; - use tokio::io::AsyncWriteExt; - stdin - .write_all(line.as_bytes()) - .await - .map_err(|e| e.to_string())?; - stdin.flush().await.map_err(|e| e.to_string())?; - drop(stdin_guard); let timeout = Duration::from_millis(timeout_ms); match tokio::time::timeout(timeout, rx).await { Ok(Ok(Ok(v))) => Ok(v), Ok(Ok(Err(e))) => Err(e), - Ok(Err(_)) => { - let _ = self.pending.lock().await.remove(&id); - Err("Worker dropped response".to_string()) - } - Err(_) => { - let _ = self.pending.lock().await.remove(&id); - Err(format!("Worker call timeout ({}ms)", timeout_ms)) - } + Ok(Err(_)) => Err("Worker dropped response".to_string()), + Err(_) => Err(format!("Worker call timeout ({}ms)", timeout_ms)), } } - /// Last activity timestamp (millis since epoch). + /// Last activity timestamp (millis since epoch). Refreshed on request + /// send, on every stderr line, and when an in-flight call settles. pub fn last_activity_ms(&self) -> i64 { self.last_activity.load(Ordering::SeqCst) } - /// Kill the worker process. - pub async fn kill(&mut self) { - let _ = self._child.start_kill(); - let _ = tokio::time::timeout(Duration::from_secs(2), self._child.wait()).await; + /// Number of RPCs currently awaiting a response. + pub fn inflight_count(&self) -> usize { + self.inflight.load(Ordering::SeqCst) + } + + /// True while any RPC is in flight. Busy workers must not be reaped or + /// evicted: `last_activity` alone can look stale during a long quiet call. + pub fn is_busy(&self) -> bool { + self.inflight_count() > 0 + } + + /// Kill the worker process. Safe to call while RPCs are in flight — the + /// stderr reader fails them with "Worker exited" once the process dies. + pub async fn kill(&self) { + let mut child = self.child.lock().await; + let _ = child.start_kill(); + let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inflight_guard_settles_call_state_on_drop() { + let pending: Arc> = + Arc::new(StdMutex::new(PendingResponseMap::new())); + let inflight = Arc::new(AtomicUsize::new(1)); + let last_activity = Arc::new(AtomicI64::new(0)); + + let (tx, mut rx) = oneshot::channel(); + lock_pending(&pending).insert("rpc-1".to_string(), tx); + + drop(InflightGuard { + id: "rpc-1".to_string(), + pending: Arc::clone(&pending), + inflight: Arc::clone(&inflight), + last_activity: Arc::clone(&last_activity), + }); + + assert!(lock_pending(&pending).is_empty()); + assert_eq!(inflight.load(Ordering::SeqCst), 0); + assert!(last_activity.load(Ordering::SeqCst) > 0); + // The sender side is gone, so a late response would be dropped. + assert!(rx.try_recv().is_err()); + } + + #[test] + fn inflight_guard_leaves_other_pending_calls_untouched() { + let pending: Arc> = + Arc::new(StdMutex::new(PendingResponseMap::new())); + let inflight = Arc::new(AtomicUsize::new(2)); + let last_activity = Arc::new(AtomicI64::new(0)); + + let (tx1, _rx1) = oneshot::channel(); + let (tx2, _rx2) = oneshot::channel(); + lock_pending(&pending).insert("rpc-1".to_string(), tx1); + lock_pending(&pending).insert("rpc-2".to_string(), tx2); + + drop(InflightGuard { + id: "rpc-1".to_string(), + pending: Arc::clone(&pending), + inflight: Arc::clone(&inflight), + last_activity: Arc::clone(&last_activity), + }); + + let guard = lock_pending(&pending); + assert_eq!(guard.len(), 1); + assert!(guard.contains_key("rpc-2")); + assert_eq!(inflight.load(Ordering::SeqCst), 1); } } diff --git a/src/crates/services/services-integrations/src/miniapp/worker_pool.rs b/src/crates/services/services-integrations/src/miniapp/worker_pool.rs index 2a6d91409a..dd7e049110 100644 --- a/src/crates/services/services-integrations/src/miniapp/worker_pool.rs +++ b/src/crates/services/services-integrations/src/miniapp/worker_pool.rs @@ -75,7 +75,13 @@ pub type MiniAppWorkerPoolResult = Result; struct WorkerEntry { revision: String, - worker: Arc>, + worker: Arc, +} + +/// A worker is reclaimable only when nothing is in flight: `last_activity` +/// goes stale during long RPCs that emit no stderr output. +fn worker_is_reclaimable(now: i64, worker: &JsWorker) -> bool { + !worker.is_busy() && worker_is_idle(now, worker.last_activity_ms()) } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -127,13 +133,7 @@ fn spawn_worker_reaper() -> Arc>> { fn drain_idle_workers(guard: &mut HashMap, now: i64) -> Vec { let to_remove = guard .iter() - .filter(|(_, entry)| { - entry - .worker - .try_lock() - .map(|worker| worker_is_idle(now, worker.last_activity_ms())) - .unwrap_or(false) - }) + .filter(|(_, entry)| worker_is_reclaimable(now, &entry.worker)) .map(|(key, _)| key.clone()) .collect::>(); @@ -144,12 +144,9 @@ fn drain_idle_workers(guard: &mut HashMap, now: i64) -> Vec } fn drain_lru_worker(guard: &mut HashMap) -> Option { + // Workers with in-flight RPCs are not eviction candidates. let oldest_id = select_lru_worker(guard.iter().filter_map(|(id, entry)| { - entry - .worker - .try_lock() - .map(|worker| (id.as_str(), worker.last_activity_ms())) - .ok() + (!entry.worker.is_busy()).then(|| (id.as_str(), entry.worker.last_activity_ms())) })) .unwrap_or_default(); @@ -162,8 +159,7 @@ fn drain_lru_worker(guard: &mut HashMap) -> Option) { for entry in entries { - let mut worker = entry.worker.lock().await; - worker.kill().await; + entry.worker.kill().await; } } @@ -228,7 +224,7 @@ impl JsWorkerPool { worker_revision: &str, policy_json: &str, node_perms: Option<&NodePermissions>, - ) -> MiniAppWorkerPoolResult>> { + ) -> MiniAppWorkerPoolResult> { let app_dir = self.miniapp_dir(app_id); self.get_or_spawn_with_app_dir( app_id, @@ -249,7 +245,7 @@ impl JsWorkerPool { worker_revision: &str, policy_json: &str, node_perms: Option<&NodePermissions>, - ) -> MiniAppWorkerPoolResult>> { + ) -> MiniAppWorkerPoolResult> { let _spawn_guard = self.spawn_lock.lock().await; let mut workers_to_kill = Vec::new(); @@ -260,11 +256,7 @@ impl JsWorkerPool { .unwrap_or_default() .as_millis() as i64; if let Some(entry) = guard.remove(worker_key) { - let is_idle = entry - .worker - .try_lock() - .map(|worker| worker_is_idle(now, worker.last_activity_ms())) - .unwrap_or(false); + let is_idle = worker_is_reclaimable(now, &entry.worker); match plan_existing_worker_admission(&entry.revision, worker_revision, is_idle) { ExistingWorkerAdmission::Reuse => { let worker = Arc::clone(&entry.worker); @@ -317,7 +309,7 @@ impl JsWorkerPool { .map_err(MiniAppWorkerPoolError::validation)?; let _timeout_ms = node_perms.and_then(|n| n.timeout_ms).unwrap_or(30_000); - let worker = Arc::new(Mutex::new(worker)); + let worker = Arc::new(worker); { let mut guard = self.workers.lock().await; guard.insert( @@ -345,8 +337,9 @@ impl JsWorkerPool { .get_or_spawn(app_id, worker_revision, policy_json, permissions) .await?; let timeout_ms = permissions.and_then(|n| n.timeout_ms).unwrap_or(30_000); - let guard = worker.lock().await; - guard + // No pool-level lock here: JsWorker::call serializes only the stdin + // write, so concurrent RPCs to the same app interleave. + worker .call(method, params, timeout_ms) .await .map_err(MiniAppWorkerPoolError::validation) @@ -374,8 +367,7 @@ impl JsWorkerPool { ) .await?; let timeout_ms = permissions.and_then(|n| n.timeout_ms).unwrap_or(30_000); - let guard = worker.lock().await; - guard + worker .call(method, params, timeout_ms) .await .map_err(MiniAppWorkerPoolError::validation) From c2031ea01f11451721d1e69851ed9a69e22fbf83 Mon Sep 17 00:00:00 2001 From: xlx1212 Date: Sat, 8 Aug 2026 19:01:39 +0800 Subject: [PATCH 2/2] test(core): fix stale collapsed_tool_names field in catalog test Renamed to deferred_tool_names in c56456f7f; this one test was missed and has blocked compiling the bitfun-core test target since. Co-Authored-By: Claude Fable 5 --- .../assembly/core/src/agentic/tools/product_runtime/catalog.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/crates/assembly/core/src/agentic/tools/product_runtime/catalog.rs b/src/crates/assembly/core/src/agentic/tools/product_runtime/catalog.rs index 042f320e07..bc2fbb0cf6 100644 --- a/src/crates/assembly/core/src/agentic/tools/product_runtime/catalog.rs +++ b/src/crates/assembly/core/src/agentic/tools/product_runtime/catalog.rs @@ -782,7 +782,7 @@ mod tests { .allowed_tool_names .contains(&"ReviewPlatform".to_string())); assert!(manifest - .collapsed_tool_names + .deferred_tool_names .contains(&"ReviewPlatform".to_string())); assert!(manifest .tool_definitions