src: check HasCaught() in JSStream calls

`MakeCallback` can return an empty `MaybeLocal<>` even if no
exception has been generated, in particular, if
we were already terminating the current thread *before* the `TryCatch`
scope started, which meant it would not have an exception to report.

PR-URL: #26124
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: addaleax

src/js_stream.cc

@@ -46,7 +46,7 @@ bool JSStream::IsClosing() {
   TryCatchScope try_catch(env());
   Local<Value> value;
   if (!MakeCallback(env()->isclosing_string(), 0, nullptr).ToLocal(&value)) {
-    if (!try_catch.HasTerminated())
+    if (try_catch.HasCaught() && !try_catch.HasTerminated())
       FatalException(env()->isolate(), try_catch);
     return true;
   }
@@ -62,7 +62,7 @@ int JSStream::ReadStart() {
   int value_int = UV_EPROTO;
   if (!MakeCallback(env()->onreadstart_string(), 0, nullptr).ToLocal(&value) ||
       !value->Int32Value(env()->context()).To(&value_int)) {
-    if (!try_catch.HasTerminated())
+    if (try_catch.HasCaught() && !try_catch.HasTerminated())
       FatalException(env()->isolate(), try_catch);
   }
   return value_int;
@@ -77,7 +77,7 @@ int JSStream::ReadStop() {
   int value_int = UV_EPROTO;
   if (!MakeCallback(env()->onreadstop_string(), 0, nullptr).ToLocal(&value) ||
       !value->Int32Value(env()->context()).To(&value_int)) {
-    if (!try_catch.HasTerminated())
+    if (try_catch.HasCaught() && !try_catch.HasTerminated())
       FatalException(env()->isolate(), try_catch);
   }
   return value_int;
@@ -99,7 +99,7 @@ int JSStream::DoShutdown(ShutdownWrap* req_wrap) {
                     arraysize(argv),
                     argv).ToLocal(&value) ||
       !value->Int32Value(env()->context()).To(&value_int)) {
-    if (!try_catch.HasTerminated())
+    if (try_catch.HasCaught() && !try_catch.HasTerminated())
       FatalException(env()->isolate(), try_catch);
   }
   return value_int;
@@ -134,7 +134,7 @@ int JSStream::DoWrite(WriteWrap* w,
                     arraysize(argv),
                     argv).ToLocal(&value) ||
       !value->Int32Value(env()->context()).To(&value_int)) {
-    if (!try_catch.HasTerminated())
+    if (try_catch.HasCaught() && !try_catch.HasTerminated())
       FatalException(env()->isolate(), try_catch);
   }
   return value_int;

test/parallel/test-worker-http2-generic-streams-terminate.js

@@ -0,0 +1,39 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const assert = require('assert');
+const http2 = require('http2');
+const { Duplex } = require('stream');
+const { Worker, workerData } = require('worker_threads');
+
+// Tests the interaction between terminating a Worker thread and running
+// the native SetImmediate queue, which may attempt to perform multiple
+// calls into JS even though one already terminates the Worker.
+
+if (!workerData) {
+  const counter = new Int32Array(new SharedArrayBuffer(4));
+  const worker = new Worker(__filename, { workerData: { counter } });
+  worker.on('exit', common.mustCall(() => {
+    assert.strictEqual(counter[0], 1);
+  }));
+} else {
+  const { counter } = workerData;
+
+  // Start two HTTP/2 connections. This will trigger write()s call from inside
+  // the SetImmediate queue.
+  for (let i = 0; i < 2; i++) {
+    http2.connect('http://localhost', {
+      createConnection() {
+        return new Duplex({
+          write(chunk, enc, cb) {
+            Atomics.add(counter, 0, 1);
+            process.exit();
+          },
+          read() { }
+        });
+      }
+    });
+  }
+}