fix(dio): propagate response-stream backpressure to the source

[AlexV525]

New Pull Request Checklist

• I have read the Documentation
• I have read the Agent Contribution Guidelines (required if any part of the change was produced with AI assistance)
• I have searched for a similar pull request in the project and found none — no existing issue/PR covers response-stream download backpressure. (iOS Excessive Memory Usage in MultipartFile.fromStream Compared to Direct Stream Constructor #2506 is an upload MultipartFile memory issue on a different code path.)
• I have updated this branch with the latest main branch to avoid conflicts (via merge from master or rebase)
• I have added the required tests to prove the fix/feature I am adding
• (not applicable) I have updated the documentation (if necessary) — no public API changed.
• I have run the tests without failures
• I have updated the CHANGELOG.md in the corresponding package

Motivation

Reported by a user: large streamed downloads can exhaust memory on
constrained platforms (e.g. an iOS Jetsam kill) when the consumer
reads slower than the network delivers. The consumer's pause() was
never reaching the socket, so the response kept buffering into RAM.

Root cause

handleResponseStream wraps the adapter's source stream in a
responseSink StreamController without onPause/onResume:

final responseSink = StreamController<Uint8List>(); // no backpressure hooks

So a downstream pause() only paused the responseSink internally and
buffered its data — the source subscription (the socket) was never
paused, the TCP receive window never shrank, and the sender kept pushing.
The "water bucket" filled until the OS killed the app.

This is not a regression from a recent change. The missing callbacks
already existed in IOHttpClientAdapter before #2068 extracted the
wrapping logic into handleResponseStream; the extraction preserved the
defect. Since handleResponseStream is now the single shared response
path, fixing it covers all adapters (IO / browser / native / http2).

Technical details

The change is two callback hooks plus reordering the late subscription
declaration ahead of the controller (Dart forbids a forward reference to
a late local):

late StreamSubscription<List<int>> responseSubscription;
final responseSink = StreamController<Uint8List>(
  onPause: () => responseSubscription.pause(),
  onResume: () => responseSubscription.resume(),
);

This restores the full backpressure chain:
consumer pause() → responseSink.onPause → source subscription
paused → socket stops reading → TCP window contracts.

Correctness of the late capture

• Initialization order: source.listen(...) (which assigns
  responseSubscription) runs synchronously before the stream is
  returned to the caller. A downstream pause() can only happen after
  the caller listens, so responseSubscription is always assigned when
  onPause fires.
• Cancel/timeout races: after a receiveTimeout or cancelToken
  cancellation the subscription is cancel()-ed; a later onResume
  calls resume() on an already-cancelled subscription, which is a
  documented no-op (verified by probe).

Verification

Added two regression tests to the existing
response_stream_test.dart group (no new file):

1. propagates downstream pause/resume as backpressure to the source —
   an observable upstream asserts onPause fires on a downstream pause
   and buffered data flows only after resume().
2. does not buffer an unbounded source when the consumer pauses —
   a backpressure-aware async* generator (socket-like: blocks at
   yield when paused) producing up to ~10 MB; the consumer pauses after
   the first chunk. Before the fix the generator produced all 10000
   chunks (~10 MB buffered); after it halts at < 5.

Both tests were confirmed to fail on the unpatched code (actual
values false and 10000 respectively) and pass with the fix.
dart analyze clean; the existing 8 stream tests + adapter/timeout tests
pass with no regressions.

Implementation, tests, and local review by GLM-5.2.

[github-actions]

Code Coverage Report: Only Changed Files listed

Package	Base Coverage	New Coverage	Difference
Overall Coverage	🟢 86.19%	🟢 86.2%	🟢 0.01%

Minimum allowed coverage is 0%, this run produced 86.2%

[CaiJingLong]

审查结论：可以合并

改动概述

本 PR 在 handleResponseStream 的 responseSink StreamController 上添加了 onPause/onResume 回调，将下游消费者的 pause/resume 传播到源订阅（socket），恢复了完整的背压链：consumer pause → responseSink.onPause → source subscription paused → socket stops reading → TCP window contracts。改动仅两行回调 + late 变量声明顺序调整，修复了大型流式下载在慢消费场景下内存耗尽（如 iOS Jetsam OOM）的问题。

审查明细

维度	结论	说明
正确性	通过	onPause/onResume 正确委托到 responseSubscription.pause()/resume()。responseSubscription 在 source.listen(...)（第 77 行）同步赋值，先于流返回（第 108 行），因此 onPause 触发时变量必定已初始化——无 forward reference 风险。超时/取消路径中 responseSubscription.cancel() 后再调用 resume()/pause() 是 Dart 文档定义的 no-op，无异常风险。onDone 路径先 cancel() 再 close()，close 后控制器不再触发回调。
测试	通过	新增两个回归测试到既有 response_stream_test.dart 组（无新文件）。已在未修复基线（main）上确认两测试均失败：测试 1 upstreamPaused 为 false，测试 2 produced 为 10000（~10MB 全量缓冲）；修复后均通过。第二个提交将 async* 生成器替换为基于 StreamController.isPaused 的显式生产循环，确保在 VM/dart2js/dart2wasm 三平台语义一致。本地 dart test test/response/response_stream_test.dart 全部 10 测试通过。
风格	通过	改动遵循既有代码风格，无调试残留、无无关格式化、无多余 import。
风险	通过	低回归风险：仅添加此前缺失的背压传播，不改变任何公开 API、默认行为或异常类型。既有 8 个流测试 + 超时/取消测试全部通过。handleResponseStream 是所有适配器（IO/browser/native/http2）的共享响应路径，修复一处即覆盖全部。
文档	通过	CHANGELOG.md 已在 ## Unreleased 下更新，描述面向下游用户。无公开 API 变更，README/文档无需更新（PR checklist 已正确标注 not applicable）。
仓库约束	通过	分支名 fix/response-stream-backpressure 符合 category/short-description 约定；提交消息使用 Conventional Commits 格式（fix(dio): / test(dio):）；AI 归因 Co-Authored-By: GLM-5.2 已附；无依赖变更、无公开 API 破坏、无敏感区域未声明改动。dart analyze 干净。

建议（非阻塞，可选）

• PR 正文提到用户报告的 iOS Jetsam 场景，但未关联具体 issue 编号。若有对应 issue，建议在 PR 描述中补充 Closes #NNNN 以便追踪。
• onResume 在订阅已 cancel 后被调用虽为 no-op，但可考虑在回调内加 if (!responseSink.isClosed) 守卫以更显式——非必要，当前实现已安全。

已确认关键点

• 正确性：late 变量初始化顺序安全（同步赋值先于流返回）；cancel/timeout 后对已取消订阅调用 pause/resume 为 Dart 文档 no-op；onDone/onError/onTimeout/onCancel 四条终止路径均无 double-cancel 或 use-after-close 问题。
• 测试：两个回归测试在未修复基线上确认失败（false / 10000），修复后通过；测试平台无关（VM + dart2js + dart2wasm）；dart analyze 干净；既有 10 个测试全通过无回归。
• 风险：无公开 API 变更；无依赖变更；改动仅添加缺失的背压回调，不改变既有行为语义；覆盖全部适配器。

本评论由 AI agent omp（模型: glm-5-2）生成