fix(streams): a throwing ReadableStream pull() errors the stream, not crash

crates/perry-runtime/src/exception.rs

@@ -144,6 +144,32 @@ pub(crate) fn current_try_depth() -> usize {
     with_exception_state(|s| unsafe { (*s).try_depth })
 }
 
+/// Invoke `f` — which may call into user JS and `js_throw` — inside a `try`
+/// trap, catching any JS exception. Returns `Ok(value)` on a normal return,
+/// or `Err(exception_bits)` if `f` threw. The `setjmp` lives in THIS frame,
+/// which stays alive while `f` runs, so the `longjmp` target is valid, and the
+/// throw unwinds only up to here — NOT past the Rust caller's frame.
+///
+/// Runtime helpers that drive user JS from a Rust-owned microtask/timer
+/// context (e.g. a Web Streams `pull` callback) use this so a throwing
+/// callback errors the relevant object per spec instead of `longjmp`-ing past
+/// their frames — skipping cleanup and corrupting state. Mirrors
+/// `combinators::combinator_catch_js`.
+pub fn js_call_catching(f: impl FnOnce() -> f64) -> Result<f64, f64> {
+    let env = js_try_push();
+    let jumped = unsafe { crate::ffi::setjmp::setjmp(env as *mut std::os::raw::c_int) };
+    if jumped == 0 {
+        let result = f();
+        js_try_end();
+        Ok(result)
+    } else {
+        js_try_end();
+        let err = js_get_exception();
+        js_clear_exception();
+        Err(err)
+    }
+}
+
 /// Throw an exception with the given value
 #[no_mangle]
 pub extern "C" fn js_throw(value: f64) -> ! {

crates/perry-stdlib/src/streams.rs

@@ -686,14 +686,28 @@ extern "C" fn readable_pull_microtask(closure: *const ClosureHeader) -> f64 {
             }
         };
         if should_pull {
-            if pull_returns_byte_chunk {
-                pull_deferred_byte_chunk(stream_id, cb);
-            } else {
-                js_closure_call1(cb as *const ClosureHeader, stream_id as f64);
-            }
+            // A Web Streams `pull()` that throws must ERROR the stream (its
+            // pending reads reject with the exception) — it must NOT `longjmp`
+            // past this Rust microtask frame. Without the local trap the throw
+            // unwinds to the microtask loop, skipping the `pulling = false`
+            // reset (and any live frames), leaving the stream wedged and, under
+            // concurrency, crashing the process. Catch it here and route to the
+            // spec error path so `reader.read()` rejects and the consumer's own
+            // handler surfaces the error, exactly as Node does.
+            let pull_outcome = perry_runtime::exception::js_call_catching(|| {
+                if pull_returns_byte_chunk {
+                    pull_deferred_byte_chunk(stream_id, cb);
+                } else {
+                    js_closure_call1(cb as *const ClosureHeader, stream_id as f64);
+                }
+                f64::from_bits(TAG_UNDEFINED)
+            });
             if let Some(s) = READABLE_STREAMS.lock().unwrap().get_mut(&stream_id) {
                 s.pulling = false;
             }
+            if let Err(exc) = pull_outcome {
+                error_readable_stream(stream_id, exc.to_bits());
+            }
         }
     }
     f64::from_bits(TAG_UNDEFINED)

test-files/test_gap_readable_stream_pull_throws.ts

@@ -0,0 +1,20 @@
+// A ReadableStream whose pull() throws must ERROR the stream: pending read()s
+// reject with the thrown value and the process stays alive (it must NOT crash
+// by unwinding out of the internal pull microtask). Byte streams take a
+// separate perry pull path, so cover both.
+async function probe(label: string, opts: any) {
+  const rs = new ReadableStream(opts);
+  const reader = rs.getReader();
+  try {
+    await reader.read();
+    console.log(label + ": UNEXPECTED resolve");
+  } catch (e) {
+    console.log(label + ": rejected: " + (e as Error).message);
+  }
+}
+async function main() {
+  await probe("default-pull-throw", { pull() { throw new Error("boom-default"); } });
+  await probe("byte-pull-throw", { type: "bytes", pull() { throw new Error("boom-byte"); } });
+  console.log("process survived");
+}
+main();