runtime: JS-loop-driven current-thread async runtime + wait-driver with gated fast-native-drive (supersedes #5779)

crates/perry-ext-http/src/lib.rs

@@ -187,6 +187,50 @@ pub(crate) enum PendingHttpEvent {
     DeferredArmContinue { request_handle: Handle },
 }
 
+/// #5779 follow-up — count of in-flight HTTP/HTTPS CLIENT requests (the detached
+/// reqwest task spawned per `http.request`/`http.get`, from dispatch until the
+/// response fully streams or errors).
+///
+/// `EXT_BLOCKING_TASKS_INFLIGHT` (perry-stdlib's idle-kick / active-handle gate)
+/// only stays up for the SHORT outer `spawn_blocking` closure that *launches* the
+/// reqwest task and returns; it drops to 0 while the actual fetch is still in
+/// flight. So the runtime's idle-kick never fires during a fetch, and a lost
+/// tokio worker-unpark for the reqwest task (the canonical failure under a
+/// `Promise.all` burst of fetches) is never recovered — the main thread parks
+/// forever with the responses received but undelivered. This counter, exposed via
+/// [`js_ext_http_client_inflight`], lets the idle-kick + active-handle gate honor
+/// a fetch's TRUE lifetime so a stranded reqwest task gets roused.
+static CLIENT_REQUESTS_INFLIGHT: std::sync::atomic::AtomicI64 =
+    std::sync::atomic::AtomicI64::new(0);
+
+/// RAII in-flight marker. Created right before the reqwest task is spawned and
+/// MOVED INTO the task, so the count tracks the task's full lifetime — including
+/// a task scheduled-but-stranded by a lost worker-unpark (its future, holding the
+/// guard, is never dropped while stranded). Drop wakes the main loop so its
+/// active-handle gate re-evaluates promptly.
+pub(crate) struct ClientInflightGuard;
+impl ClientInflightGuard {
+    pub(crate) fn new() -> Self {
+        CLIENT_REQUESTS_INFLIGHT.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+        ClientInflightGuard
+    }
+}
+impl Drop for ClientInflightGuard {
+    fn drop(&mut self) {
+        CLIENT_REQUESTS_INFLIGHT.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+        notify_main_thread();
+    }
+}
+
+/// Exposed for perry-stdlib's idle-kick + active-handle gate (#5779 follow-up):
+/// returns nonzero while any HTTP client fetch is outstanding.
+#[no_mangle]
+pub extern "C" fn js_ext_http_client_inflight() -> i32 {
+    CLIENT_REQUESTS_INFLIGHT
+        .load(std::sync::atomic::Ordering::Acquire)
+        .clamp(0, i32::MAX as i64) as i32
+}
+
 lazy_static! {
     static ref HTTP_PENDING_EVENTS: Mutex<Vec<PendingHttpEvent>> = Mutex::new(Vec::new());
     /// Shared HTTP client — reuses connection pool, DNS cache, TLS
@@ -648,7 +692,11 @@ fn dispatch_request_over_socket(
             return;
         }
         let handle = tokio::runtime::Handle::current();
+        // #5779 follow-up: keep this fetch counted in-flight for its whole
+        // lifetime so the idle-kick recovers a lost worker-unpark.
+        let inflight_guard = ClientInflightGuard::new();
         let jh = handle.spawn(async move {
+            let _inflight = inflight_guard;
             let vtable = match perry_ffi::raw_net() {
                 Some(v) => v,
                 None => {
@@ -1642,14 +1690,36 @@ pub extern "C" fn js_http_has_pending() -> i32 {
 /// from codegen's event-loop tick. Returns count of events drained.
 #[no_mangle]
 pub unsafe extern "C" fn js_http_process_pending() -> i32 {
-    let events: Vec<PendingHttpEvent> = match HTTP_PENDING_EVENTS.lock() {
-        Ok(mut q) => q.drain(..).collect(),
-        Err(_) => return 0,
-    };
-
-    let count = events.len() as i32;
-
-    for ev in events {
+    // Process events ONE AT A TIME, re-reading the shared queue each iteration
+    // rather than draining the whole batch into a local Vec up front.
+    //
+    // Why (#5783 follow-up): a response handler may RE-ENTER the event loop —
+    // e.g. a `ResponseHead` invokes an async response callback that drives a
+    // `for await` / `.toArray()` over a `res.pipe(PassThrough())` body. That
+    // consumer's `await` block-waits, pumping the loop (which re-enters this
+    // function), and its resolution depends on the body arriving via the LATER
+    // `ResponseChunk`/`ResponseEnd` events of this same batch. If those were
+    // already drained into a local Vec, the re-entrant pump would find an empty
+    // `HTTP_PENDING_EVENTS` and the consumer would deadlock (empty body / hang).
+    // Keeping unprocessed events in the shared queue lets the re-entrant drain
+    // deliver them. FIFO `remove(0)` preserves event order; each event is taken
+    // by exactly one (outer or re-entrant) frame, so there is no double-dispatch.
+    let mut count = 0i32;
+    loop {
+        let ev = match HTTP_PENDING_EVENTS.lock() {
+            Ok(mut q) => {
+                if q.is_empty() {
+                    None
+                } else {
+                    Some(q.remove(0))
+                }
+            }
+            Err(_) => return count,
+        };
+        let Some(ev) = ev else {
+            break;
+        };
+        count += 1;
         match ev {
             PendingHttpEvent::Response {
                 request_handle,

crates/perry-ext-ws/src/lib.rs

@@ -523,7 +523,11 @@ pub unsafe extern "C" fn js_ws_wait_for_message(handle: i64, timeout_ms: f64) ->
         if start.elapsed() >= timeout {
             return std::ptr::null_mut();
         }
-        std::thread::sleep(std::time::Duration::from_millis(10));
+        // Unified single-thread model: the WS reader task only advances while the
+        // main thread drives the runtime, so drive one bounded tick here (which
+        // runs the reader and delivers messages) instead of `std::thread::sleep`,
+        // which would block this thread and never let a message arrive.
+        perry_ffi::run_pending(10);
     }
 }
 

crates/perry-ffi/src/async_runtime.rs

@@ -64,6 +64,20 @@ extern "C" {
     fn perry_ffi_spawn_blocking(ctx: *mut c_void, invoke: extern "C" fn(*mut c_void));
     fn perry_ffi_spawn_blocking_with_reactor(ctx: *mut c_void, invoke: extern "C" fn(*mut c_void));
     fn perry_ffi_spawn_async(ctx: *mut c_void);
+    fn perry_ffi_run_pending(budget_ms: u64);
+}
+
+/// Drive the shared async runtime for up to `budget_ms` (or until a producer
+/// signals work). In the unified single-thread runtime model the runtime only
+/// makes progress while the main thread drives it, so a *synchronous* native
+/// API that blocks the main thread waiting for data delivered by a spawned task
+/// (e.g. `js_ws_wait_for_message`) must call this in its poll loop instead of
+/// `std::thread::sleep` — otherwise the delivering task never runs and the wait
+/// always times out. Safe to call from the main thread outside any other
+/// `block_on`; must NOT be called from inside a spawned runtime task.
+pub fn run_pending(budget_ms: u64) {
+    // SAFETY: thin call into the perry-stdlib-provided runtime driver.
+    unsafe { perry_ffi_run_pending(budget_ms) };
 }
 
 // NaN-box tags. These values are part of perry-runtime's stable

crates/perry-ffi/src/lib.rs

@@ -46,7 +46,7 @@
 
 mod async_runtime;
 pub use async_runtime::{
-    nanbox_string_bits, spawn_async, spawn_blocking, spawn_blocking_with_reactor,
+    nanbox_string_bits, run_pending, spawn_async, spawn_blocking, spawn_blocking_with_reactor,
     JsNativeAsyncCompletion, JsPromise, PERRY_NATIVE_ASYNC_ALREADY_COMPLETED,
     PERRY_NATIVE_ASYNC_CLEANUP_ON_CANCEL, PERRY_NATIVE_ASYNC_CLEANUP_ON_REJECT,
     PERRY_NATIVE_ASYNC_CLEANUP_ON_SUCCESS, PERRY_NATIVE_ASYNC_INVALID, PERRY_NATIVE_ASYNC_OK,