stream: keep async context of pipeTo watchers

lib/internal/webstreams/readablestream.js

@@ -47,6 +47,8 @@ const {
   DOMException,
 } = internalBinding('messaging');
 
+const AsyncContextFrame = require('internal/async_context_frame');
+
 const {
   isArrayBufferView,
   isDataView,
@@ -1541,21 +1543,30 @@ function readableStreamFromIterable(iterable) {
 // record enqueues the watcher at the microtask position a reaction on the
 // promise would have had, without materializing the promise; the shared
 // pending promise satisfies the probes on the erroring/release paths.
+// The reaction is registered in the async context captured at install
+// time, where the spec registers it, rather than in the context of
+// whoever settles the record.
 class ClosedPromiseHook {
   constructor(onResolved, onRejected) {
     this.promise = kPendingPromise;
     this.onResolved = onResolved;
     this.onRejected = onRejected;
+    this.contextFrame = AsyncContextFrame.current();
   }
 
   resolve() {
-    if (this.onResolved !== undefined)
-      PromisePrototypeThen(kResolvedPromise, this.onResolved);
+    if (this.onResolved === undefined)
+      return;
+    const prior = AsyncContextFrame.exchange(this.contextFrame);
+    PromisePrototypeThen(kResolvedPromise, this.onResolved);
+    AsyncContextFrame.set(prior);
   }
 
   reject(error) {
     const onRejected = this.onRejected;
+    const prior = AsyncContextFrame.exchange(this.contextFrame);
     PromisePrototypeThen(kResolvedPromise, () => onRejected(error));
+    AsyncContextFrame.set(prior);
   }
 }