stream: add catch handler for async _construct

lib/internal/streams/destroy.js

@@ -230,7 +230,7 @@ function constructNT(stream) {
   const s = w || r;
 
   let called = false;
-  stream._construct((err) => {
+  const result = stream._construct((err) => {
     if (r) {
       r.constructed = true;
     }
@@ -252,6 +252,25 @@ function constructNT(stream) {
       process.nextTick(emitConstructNT, stream);
     }
   });
+  if (result !== undefined && result !== null) {
+    try {
+      const then = result.then;
+      if (typeof then === 'function') {
+        then.call(result, undefined, function(err) {
+          if (r) {
+            r.constructed = true;
+          }
+          if (w) {
+            w.constructed = true;
+          }
+          called = true;
+          process.nextTick(errorOrDestroy, stream, err);
+        });
+      }
+    } catch (err) {
+      process.nextTick(emitErrorNT, stream, err);
+    }
+  }
 }
 
 function emitConstructNT(stream) {

test/parallel/test-stream-construct-async-error.js

@@ -0,0 +1,24 @@
+'use strict';
+
+const common = require('../common');
+const { Duplex } = require('stream');
+const { setTimeout } = require('timers/promises');
+const assert = require('assert');
+
+class Foo extends Duplex {
+  async _construct(cb) {
+    // eslint-disable-next-line no-restricted-syntax
+    await setTimeout(common.platformTimeout(1));
+    cb();
+    throw new Error('boom');
+  }
+}
+
+const foo = new Foo();
+foo.on('error', common.expectsError({
+  message: 'boom'
+}));
+foo.on('close', common.mustCall(() => {
+  assert(foo._writableState.constructed);
+  assert(foo._readableState.constructed);
+}));