Ensure cancellation of a standard ReadableStream pump propagates to its underlying source

[mhart]

Canceling pumpTo() destroys the pump coroutine and releases its reader without invoking the source's cancel() callback. A pending pull() could therefore remain blocked indefinitely, delaying response and request teardown.

Reproduction

const abort = new AbortController();

const stream = new ReadableStream({
  async pull(controller) {
    try {
      await scheduler.wait(60000, { signal: abort.signal });
      controller.enqueue("data");
    } catch (error) {
      if (!abort.signal.aborted) throw error;
    }
  },

  cancel(reason) {
    console.log("stream-cancel", reason);
    abort.abort();
  },
});

export default {
  fetch() {
    return new Response(stream);
  },
};

Run the worker, read only the first chunk, and close the client connection:

curl --no-buffer http://127.0.0.1:PORT/ | head -n 1
kill -TERM "$WORKERD_PID"

Before this change, the stream-cancel log was absent and the pending pull() could keep teardown active until the timer resolved.

Changes

• Added cancellation cleanup state around standard stream pumps.
• Keep the DrainingReader through IoOwn.
• Schedule reader.cancel() through the active IoContext.
• Guard cleanup against context teardown and duplicate error cleanup.
• Added a regression test with a pending pull() promise.

Testing

I only ran this on macOS:

bazel test --config=macos --nocache_test_results --test_output=all \
  //src/workerd/api:streams/standard-test@

[codspeed-hq]

Merging this PR will degrade performance by 8.78%

⚠️ Different runtime environments detected

Some benchmarks with significant performance changes were compared across different runtime environments,
which may affect the accuracy of the results.

Open the report in CodSpeed to investigate

❌ 1 regressed benchmark
✅ 71 untouched benchmarks
⏩ 129 skipped benchmarks¹

Warning

Please fix the performance issues or acknowledge them on CodSpeed.

Performance Changes

	Benchmark	BASE	HEAD	Efficiency
❌	request[RegExpBenchmark]	4.1 ms	4.5 ms	-8.78%

Tip

Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.

---

_{Comparing mhart:mhart/propagate-stream-cancel (752d9e6) with main (97a94ed)}

Footnotes

1. 129 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

[cnluzhang]

This looks like the same bug as issue #6832, which has had a fix open in PR #6833 since June 20 — same root cause, and this PR handles a couple of things better than ours does (more below). One difference seems worth raising before it lands, though, plus some tests you are welcome to take.

addTask vs addWaitUntil. ~PumpState schedules the cancel with addTask, which io-context.h documents as "It will be canceled if the request is canceled", whereas addWaitUntil is the one "drain() waits until all such promises have completed" — and IncomingRequest::drain() only awaits waitUntilTasks.onEmpty().

For a JS-backed ReadableStream body, addNoopDeferredProxy awaits the pump before fulfilling the outer promise, so the pump runs in stage 2 of WorkerEntrypoint::request(), and the ordering on disconnect is: pump dropped → ~PumpState schedules the cancel → the stage-2 KJ_DEFER fires → drain(). A task sitting in waitUntilTasks at that point gets awaited; one in tasks does not, and once the drain resolves the IncomingRequest goes away, taking the last IoContext reference with it and cancelling tasks. That leaves the cancel racing isolate-lock acquisition.

The race does not show up in a single-request repro: on an idle isolate the cancel task is queued before the drain and usually wins, and a synchronous cancel() records its effect before the context goes away. Where I would expect it to matter is an asynchronous cancel algorithm — the JS cancel() gets entered, but the promise it returns is not awaited to completion, so cleanup after the first await is quietly dropped — and a contended isolate lock, where the invocation itself can be lost. I have not measured how often that happens under real load.

To be fair about the baseline: the pre-regression PumpToReader path cancelled from a JS continuation that nothing awaited either, so waiting for the cancel algorithm to finish is a new guarantee rather than a restored one. It still seems worth having — it is the same one-word choice, the wait is bounded by limitDrain, and WorkerEntrypoint already uses addWaitUntil for its own client-disconnect abort task.

What is reproducible either way: running the five regression tests from #6833 against this branch, four pass — the drop-mid-stream cancel, clean completion, the rejecting-cancel case, and the already-errored teardown — and the drain-lifetime one fails, with the drain completing while the cancel is still pending on a gate. The existing streams tests all pass. Each of the five is verified to fail without the corresponding piece of the fix, and they run under @all-autogates and ASAN as well.

Two things this PR does better than ours: setting cleanupStarted inside the lambda, before awaiting the cancel, closes a window where our flag placement can schedule a redundant cancel; and a standalone PumpState does not lean on coroutine frame destruction order the way our KJ_DEFER does. We will likely borrow the first one either way.

Minor, take or leave: the pre-regression PumpToReader cancelled with an undefined reason (cancel(js, kj::none)) rather than an Error, if matching the old behavior matters. And hasCurrentIncomingRequest() is documented as a diagnostic hack — it does hold on the main fetch path, so this is only a readability thing.