Re-land the fix and regression test for vuln-187

src/workerd/api/streams/readable.h

@@ -255,6 +255,9 @@ class DrainingReader: public ReadableStreamController::Reader {
 
   kj::Ptr<ReadableStreamController::Reader> getPtr() { return addPtrToThis(); }
 
+  // A pointer for holders that may outlive the reader and must notice when they have.
+  kj::Weak<DrainingReader> getWeak() { return addWeakToThis(); }
+
  private:
   struct Initial {};
   using Attached = jsg::Ref<ReadableStream>;

src/workerd/api/streams/standard.c++

@@ -3479,6 +3479,25 @@ class AllReader {
   }
 };
 
+// Hands `promise` off to a holder that outlives the caller, reporting its outcome
+// through `fulfiller`. Fulfilling or rejecting a fulfiller whose promise is already
+// gone is a no-op, so the caller is free to stop listening at any point.
+template <typename T>
+kj::Promise<void> forwardToFulfiller(
+    kj::Promise<T> promise, kj::Own<kj::PromiseFulfiller<T>> fulfiller) {
+  KJ_TRY {
+    if constexpr (jsg::isVoid<T>()) {
+      co_await promise;
+      fulfiller->fulfill();
+    } else {
+      fulfiller->fulfill(co_await promise);
+    }
+  }
+  KJ_CATCH(exception) {
+    fulfiller->reject(kj::mv(exception));
+  }
+}
+
 // pumpToImpl uses a DrainingReader to efficiently pull all synchronously available
 // data from the stream in each iteration, then writes it to the sink using vectored
 // I/O. This minimizes isolate lock acquisitions by batching: each time the lock is
@@ -3487,11 +3506,23 @@ class AllReader {
 //
 // The pump loop is a kj coroutine. Dropping the returned kj::Promise drops the
 // coroutine frame, which destroys the DrainingReader (releasing the stream lock)
-// and the sink. No WeakRef/IoOwn dance is needed because ownership is clear.
+// and the sink.
 // The coroutine that implements the pump loop takes ownership of the DrainingReader
 // and sink. The jsg::Ref<ReadableStream> is not passed into the coroutine because
 // jsg::Ref is disallowed in coroutine parameters; instead, the DrainingReader holds
 // a reference to the stream internally.
+//
+// Neither isolate-lock round trip below is awaited directly. A draining read runs
+// the stream's pull() callback, and pull() can abort the request, which drops the
+// pump promise from inside the very event that is delivering the read's result.
+// Destroying an event while it is firing trips "Promise callback destroyed itself".
+// Each run() is therefore handed to the IoContext's task set, and the pump awaits an
+// unrelated fulfiller. Dropping the pump then destroys only that await; the firing
+// event stays owned by the task set, which unwinds it once it is safe to.
+//
+// The consequence is that those tasks outlive the coroutine frame, so they must not
+// name anything the frame owns. The DrainingReader is reached through a kj::Weak,
+// which reports the frame's destruction rather than dangling into it.
 kj::Promise<void> pumpToImpl(IoContext& ioContext,
     kj::Own<DrainingReader> reader,
     kj::Own<WritableStreamSink> sink,
@@ -3503,13 +3534,22 @@ kj::Promise<void> pumpToImpl(IoContext& ioContext,
     while (true) {
       // Perform a draining read to get all synchronously available data if possible
       // or fall back to a regular read if not.
-      DrainingReadResult result = co_await ioContext.run([&reader](jsg::Lock& js) mutable {
+      auto prp = kj::newPromiseAndFulfiller<DrainingReadResult>();
+      auto promise = ioContext.run([weakReader = reader->getWeak()](
+                                       jsg::Lock& js) mutable -> kj::Promise<DrainingReadResult> {
         auto& ioContext = IoContext::current();
-        // Use a 256KB limit to allow periodic yielding to the event loop,
-        // preventing a fast producer from monopolizing the thread.
-        constexpr size_t kMaxReadPerCycle = 256 * 1024;
-        return ioContext.awaitJs(js, reader->read(js, kMaxReadPerCycle));
+        KJ_IF_SOME(reader, weakReader.tryGet()) {
+          // Use a 256KB limit to allow periodic yielding to the event loop,
+          // preventing a fast producer from monopolizing the thread.
+          constexpr size_t kMaxReadPerCycle = 256 * 1024;
+          return ioContext.awaitJs(js, reader.read(js, kMaxReadPerCycle));
+        } else {
+          return KJ_EXCEPTION(DISCONNECTED, "The pump was canceled.");
+        }
       });
+      ioContext.addTask(forwardToFulfiller(kj::mv(promise), kj::mv(prp.fulfiller)));
+
+      DrainingReadResult result = co_await prp.promise;
 
       // Write all the chunks we received using vectored write for efficiency.
       if (result.chunks.size() > 0) {
@@ -3534,11 +3574,20 @@ kj::Promise<void> pumpToImpl(IoContext& ioContext,
       sink->abort(exception.clone());
     }
 
-    co_await ioContext.run([&reader, ex = exception.clone()](jsg::Lock& js) mutable {
+    auto prp = kj::newPromiseAndFulfiller<void>();
+    auto promise = ioContext.run([weakReader = reader->getWeak(), ex = exception.clone()](
+                                     jsg::Lock& js) mutable -> kj::Promise<void> {
       auto& ioContext = IoContext::current();
-      auto error = js.exceptionToJsValue(kj::mv(ex));
-      return ioContext.awaitJs(js, reader->cancel(js, error.getHandle(js)));
+      KJ_IF_SOME(reader, weakReader.tryGet()) {
+        auto error = js.exceptionToJsValue(kj::mv(ex));
+        return ioContext.awaitJs(js, reader.cancel(js, error.getHandle(js)));
+      } else {
+        return KJ_EXCEPTION(DISCONNECTED, "The pump was canceled.");
+      }
     });
+    ioContext.addTask(forwardToFulfiller(kj::mv(promise), kj::mv(prp.fulfiller)));
+
+    co_await prp.promise;
     kj::throwFatalException(kj::mv(exception));
   }
 }

src/workerd/api/tests/BUILD.bazel

@@ -804,6 +804,15 @@ wd_test(
     data = ["autovuln-319-test.js"],
 )
 
+wd_test(
+    src = "autovuln-187-test.wd-test",
+    args = ["--experimental"],
+    data = [
+        "autovuln-187-echo.js",
+        "autovuln-187-test.js",
+    ],
+)
+
 wd_test(
     src = "unsafe-test.wd-test",
     args = ["--experimental"],

src/workerd/api/tests/autovuln-187-echo.js

@@ -0,0 +1,9 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+export default {
+  fetch() {
+    return new Response('ok');
+  },
+};

src/workerd/api/tests/autovuln-187-test.js

@@ -0,0 +1,36 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+import { rejects } from 'node:assert';
+
+// Regression test for AUTOVULN-CLOUDFLARE-WORKERD-187.
+//
+// A draining read delivers its result to pumpToImpl through a kj event, and it
+// runs the stream's pull() callback along the way. A pull() that aborts the
+// request's signal makes the canceler destroy the pump's coroutine frame from
+// inside that event, so the event is destroyed while still firing and kj aborts
+// the process with "Promise callback destroyed itself".
+export default {
+  async test(_ctrl, env) {
+    const ac = new AbortController();
+    let n = 0;
+    const rs = new ReadableStream({
+      pull(c) {
+        if (++n === 2) ac.abort();
+        c.enqueue(new Uint8Array([65]));
+      },
+    });
+    await rejects(
+      env.ECHO.fetch('http://x/', {
+        method: 'POST',
+        body: rs,
+        signal: ac.signal,
+        duplex: 'half',
+      }),
+      {
+        message: 'The operation was aborted',
+      }
+    );
+  },
+};

src/workerd/api/tests/autovuln-187-test.wd-test

@@ -0,0 +1,24 @@
+using Workerd = import "/workerd/workerd.capnp";
+
+const unitTests :Workerd.Config = (
+  services = [
+    ( name = "autovuln-187-test",
+      worker = (
+        modules = [
+          (name = "worker", esModule = embed "autovuln-187-test.js")
+        ],
+        compatibilityFlags = ["nodejs_compat", "streams_enable_constructors"],
+        bindings = [
+          (name = "ECHO", service = "echo"),
+        ],
+      )
+    ),
+    ( name = "echo",
+      worker = (
+        modules = [
+          (name = "worker", esModule = embed "autovuln-187-echo.js")
+        ],
+      )
+    ),
+  ],
+);