Cancel the JS source when a ReadableStream pump is dropped mid-stream

[cnluzhang]

Problem

Fixes #6832.

When a client disconnects from a streaming response whose body is an async
JavaScript ReadableStream, the stream's cancel() algorithm is no longer
invoked. Because cancel() never runs, the worker cannot observe the
disconnect: the source keeps producing data (enqueuing chunks until natural
completion) long after the client is gone.

This is a regression between 1.20260617.1 and 1.20260619.1, introduced by
commit c329332 ("Remove draining read standard streams autogate"), and it is
still present in 1.20260701.1:

• 1.20260617.1: stream source receives cancel on disconnect
• 1.20260619.1 and later: stream source runs to natural completion, no cancel

Root cause

c32933263 removed the legacy PumpToReader path and left
ReadableStreamJsController::pumpTo() with only the DrainingReader +
pumpToImpl coroutine.

When the client disconnects, the HTTP layer proactively drops the
response-body pump promise. The old PumpToReader effectively preserved
cancellation on drop: tearing it down left a pending read continuation that,
on running, observed the reader was gone and canceled the controller.

The new pumpToImpl coroutine instead just tears down its frame, which
destroys the DrainingReader. ~DrainingReader calls
releaseReader(maybeJs = kj::none) — with no isolate lock available it only
clears the lock refs and never cancels the JS source. The coroutine only
canceled from its KJ_CATCH (exception) path, which a quiet disconnect never
reaches.

Fix

Cancel the source from pumpToImpl's teardown when the coroutine is dropped
before it settles. The isolate lock is unavailable during coroutine teardown,
so the cancel is scheduled onto the IoContext, with two details that matter
for lifetime correctness:

• It is scheduled as a waitUntil task, so IncomingRequest::drain() keeps
  the context open until the JS cancel() has actually run — the same
  scheduling WorkerEntrypoint uses for its disconnect abort task. A plain
  task would race context teardown (nothing awaits the tasks set after the
  request ends), so the cancel could be silently dropped in exactly the
  disconnect scenario this fixes.
• The scheduling is guarded by an IoContext::WeakRef (the pattern
  documented at IoContext::getWeakRef()). Pump promises can be owned by the
  IoContext's own task sets (e.g. ReadableStream::serialize, outbound
  fetch() with a JS-stream body), so the coroutine can be dropped during
  ~IoContext itself; the guard turns that case into a safe no-op instead of
  touching a context mid-destruction.

A pumpSettled flag suppresses the deferred cancel on the normal-completion
and error paths, which already finalize the stream themselves (the error path
cancels the reader directly). Cancelling with kj::none (undefined reason)
deliberately matches the pre-regression PumpToReader drop behavior; for
comparison, the (currently unwired) ReadableSourceKjAdapter drop path passes
a DISCONNECTED reason, so that precedent exists if a reason is preferred
here.

Testing

Two regression tests in src/workerd/api/streams-test.c++:

• ReadableStream pumpTo cancels the JS source when dropped mid-stream: a JS
  ReadableStream source suspends on its next read, the pump is dropped, and
  the test asserts the source's cancel() runs. Without the fix the cancel
  never runs and the test hangs, so it genuinely guards the fixed behavior.
• ReadableStream pumpTo does not cancel the JS source on clean completion:
  pumps a closing stream to completion, drops the settled pump, turns the
  event loop, and asserts cancel() was not invoked — pinning the
  pumpSettled suppression.

All of the following were re-run against the current head:

# Streams unit tests, default and @all-autogates variants (13/13 pass):
bazel test //src/workerd/api:streams-test@ \
  //src/workerd/api:streams/standard-test@ \
  //src/workerd/api:streams/draining-read-uaf-test@ \
  //src/workerd/api:streams/queue-test@ \
  //src/workerd/api:streams/internal-test@ \
  //src/workerd/api:system-streams-test@ \
  //src/workerd/api:streams-test@all-autogates \
  //src/workerd/api:streams/standard-test@all-autogates \
  //src/workerd/api:streams/draining-read-uaf-test@all-autogates \
  //src/workerd/api:streams/queue-test@all-autogates \
  //src/workerd/api:streams/internal-test@all-autogates \
  //src/workerd/api:system-streams-test@all-autogates

# JS-level streams test under the newest compatibility date:
bazel test //src/workerd/api:streams/streams-test@all-compat-flags

# Clean under AddressSanitizer (no use-after-free across the
# drop -> scheduled-waitUntil -> cancel path):
bazel test //src/workerd/api:streams-test@ --config=asan

End-to-end against the reporter's repro
(https://github.com/cnluzhang/workerd-stream-cancel-repro): a workerd binary
built from this branch reports stream=cancel in both the direct and proxied
topologies where stock 1.20260619.1 / 1.20260701.1 report
ended-normally.

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[cnluzhang]

I have read the CLA Document and I hereby sign the CLA

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: abd06069ce

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: fc0a147a3d

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.
To continue using code reviews, add credits to your account and enable them for code reviews in your settings.

[cnluzhang]

@danlapid — tagging you as you know this path best. This is about the PumpToReader → pumpToImpl change in c32933263: with the draining-read path, a client disconnect on an async JS ReadableStream response body tears down the pump coroutine without cancelling the JS source, so the source's cancel() doesn't fire and the stream runs to natural completion (#6832). It looks like the old PumpToReader fallback happened to cover this case.

This PR restores that behavior by cancelling the source from pumpToImpl's teardown (scheduled as a task, since the isolate lock isn't held during coroutine teardown). It's covered by a new regression test, the full streams suite (@all-autogates, @all-compat-flags), ASAN, and end-to-end against the reporter's repro.

For what it's worth on the practical side: I run WDL, a self-hosted Workers platform on stock workerd, and this is the one thing that regressed for me on 1.20260701. No rush at all — would really appreciate your eyes on the approach whenever you have a moment, since you're closest to this code.

[cnluzhang]

@danlapid heads-up before you get to this — the patch has been revised since my earlier ping (fc0a147a3d → 5591cf67d), and the PR description is updated to match. What changed:

• The deferred cancel is now scheduled via addWaitUntil rather than addTask, so IncomingRequest::drain() waits for the JS cancel() to actually run — the same scheduling the entrypoint uses for its disconnect abort task. With a plain task, nothing awaits the tasks set after the request ends, so the cancel could be silently dropped in exactly the disconnect scenario this fixes.
• The scheduling is now guarded by an IoContext::WeakRef. Pump promises can be owned by the IoContext's own task sets (ReadableStream::serialize, outbound fetch() with a JS-stream body), so the coroutine can be dropped during ~IoContext itself — the earlier unguarded version could then throw out of a destructor chain (getMetrics()'s KJ_REQUIRE(!incomingRequests.empty())). The commit message no longer cites ~WritableStreamJsRpcAdapter as the precedent: that destructor's safety depends on a revoke task in tasks that this path doesn't have; the WeakRef pattern documented at IoContext::getWeakRef() is the right anchor.
• Added a companion regression test asserting clean completion does not invoke cancel(), pinning the pumpSettled suppression.

Re-ran the full streams matrix (13/13), @all-compat-flags, ASAN, and the end-to-end repro against the new head — a binary built from this branch restores stream=cancel in both the direct and proxied repro topologies.