worker: fix stream racing with terminate

`OnStreamAfterReqFinished` uses `v8::Object::Has` to check if it needs
to call `oncomplete`. `v8::Object::Has` needs to execute Javascript.
However when worker threads are involved, `OnStreamAfterReqFinished` may
be called after the worker thread termination has begun via
`worker.terminate()`. This makes `v8::Object::Has` return `Nothing`,
which triggers an assert.

This diff fixes the issue by simply defaulting us to `false` in the case
where `Nothing` is returned. This is sound because we can't execute
`oncomplete` anyway as the isolate is terminating.

Fixes: #38418

PR-URL: #42874
Reviewed-By: Darshan Sen <raisinten@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>

Author: airtable-keyhanvakil

src/stream_base.cc

@@ -601,6 +601,7 @@ void ReportWritesToJSStreamListener::OnStreamAfterReqFinished(
     StreamReq* req_wrap, int status) {
   StreamBase* stream = static_cast<StreamBase*>(stream_);
   Environment* env = stream->stream_env();
+  if (env->is_stopping()) return;
   AsyncWrap* async_wrap = req_wrap->GetAsyncWrap();
   HandleScope handle_scope(env->isolate());
   Context::Scope context_scope(env->context());

test/parallel/test-worker-http2-stream-terminate.js

@@ -0,0 +1,63 @@
+'use strict';
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const makeDuplexPair = require('../common/duplexpair');
+const { Worker, parentPort } = require('worker_threads');
+
+// This test ensures that workers can be terminated without error while
+// stream activity is ongoing, in particular the C++ function
+// ReportWritesToJSStreamListener::OnStreamAfterReqFinished.
+
+const MAX_ITERATIONS = 20;
+const MAX_THREADS = 10;
+
+// Do not use isMainThread so that this test itself can be run inside a Worker.
+if (!process.env.HAS_STARTED_WORKER) {
+  process.env.HAS_STARTED_WORKER = 1;
+
+  function spinWorker(iter) {
+    const w = new Worker(__filename);
+    w.on('message', common.mustCall((msg) => {
+      assert.strictEqual(msg, 'terminate');
+      w.terminate();
+    }));
+
+    w.on('exit', common.mustCall(() => {
+      if (iter < MAX_ITERATIONS)
+        spinWorker(++iter);
+    }));
+  }
+
+  for (let i = 0; i < MAX_THREADS; i++) {
+    spinWorker(0);
+  }
+} else {
+  const server = http2.createServer();
+  let i = 0;
+  server.on('stream', (stream, headers) => {
+    if (i === 1) {
+      parentPort.postMessage('terminate');
+    }
+    i++;
+
+    stream.end('');
+  });
+
+  const { clientSide, serverSide } = makeDuplexPair();
+  server.emit('connection', serverSide);
+
+  const client = http2.connect('http://localhost:80', {
+    createConnection: () => clientSide,
+  });
+
+  function makeRequests() {
+    for (let i = 0; i < 3; i++) {
+      client.request().end();
+    }
+    setImmediate(makeRequests);
+  }
+  makeRequests();
+}