Add response body flow control

[mkurz]

Summary

• Add a thread-safe ResponseBodyControl callback after final response headers.
• Support suspending, resuming, and cancelling HTTP/1.1 and HTTP/2 response bodies without coupling AHC to a streaming API.
• Pause the network read timeout while reads are intentionally suspended while leaving the request timeout active.
• Keep suspended HTTP/2 streams independent and cover transport backpressure, cancellation, timeouts, and connection reuse.

Motivation

AHC 3 removed StreamedAsyncHandler and its Reactive Streams integration in pull request #1843. That removal avoids coupling AHC to a particular streaming library, but AsyncHandler by itself has no way to stop transport reads while a downstream consumer has no demand.

Play WS is the driving consumer for this change. Play WS needs transport backpressure to preserve its existing Pekko Streams and Reactive Streams response APIs while upgrading to AHC 3. A working adapter on a currently local Play WS development branch turns this control into a single-subscriber Reactive Streams publisher and has been tested against this AHC branch.

This pull request adds only the transport primitive. Streaming-library policy and dependencies remain in Play WS, so AHC does not regain a dependency on Reactive Streams or JDK Flow.

Semantics

• AsyncHandler.onResponseBodyStart runs after final headers and before body parts, including for a response with no body.
• Calls to the supplied control are thread-safe, idempotent, and ignored after response completion.
• suspend() stops requesting new transport data, although body parts already read may still be delivered.
• resume() permits transport reads again.
• Returning State.ABORT is the synchronous callback-time way to stop processing. A handler can retain the control and call cancel() when an asynchronous decision is made after the callback returns.
• Cancelling an HTTP/1.1 body closes its connection because unread bytes cannot be safely pooled. Cancelling an HTTP/2 body closes only its stream.
• Fully consumed responses retain the existing HTTP/1.1 pooling and HTTP/2 parent-connection reuse behavior.
• Suspension pauses only the network read timeout. The request timeout remains active; if the request timeout is disabled, an application that never resumes or cancels can retain the exchange and its transport resources indefinitely.

HTTP/2 flow control

Returning connection-level receive credit as DATA frames arrive prevents one suspended stream from exhausting the shared connection window and starving sibling streams. Per-stream credit is still consumption-driven, so each suspended stream remains bounded by its own receive window.

This deliberately changes the aggregate buffering bound. Before this change, the default 65,535-byte connection window capped unconsumed flow-controlled DATA across all streams on one connection. With connection-level auto-refill, that aggregate cap is removed and only the per-stream windows retain transport-level bounds. Buffering can therefore scale with the number of suspended streams; the relevant controls are http2InitialWindowSize, http2MaxConcurrentStreams, and the connection limits. The defaults do not impose a hard client-side aggregate bound: the initial per-stream window is 16 MiB and http2MaxConcurrentStreams = -1 leaves concurrency server-controlled. A rough upper-bound estimate is connection count times effective concurrent streams times the initial window, excluding network and decoder overhead. Applications requiring a finite policy must configure these values together. A true aggregate byte budget is outside this pull request.

The auto-refill mode requires a custom Http2Connection. Netty's builder treats server() and connection() as mutually exclusive, so ClientHttp2FrameCodecBuilder supplies the connection through the protected builder API and overrides isServer() to retain client mode.

The explicit gracefulShutdownTimeoutMillis(0) is not a new shutdown policy. Http2FrameCodecBuilder.forClient() selects zero through its package-private client constructor; the subclass must use the protected no-argument constructor, so it sets zero explicitly to preserve the existing client-factory behavior.

Scope and commit structure

The two commits are kept separate to make the API/lifecycle work and the HTTP/2 flow-control decision reviewable on their own, but they belong in one pull request. Without the second commit, a suspended HTTP/2 response can consume the shared connection window and block unrelated sibling streams, so the public control would not have correct multiplexed behavior.

History checked

• Issue #544 originally identified the lack of AsyncHandler backpressure, and pull request #963 addressed it by adding Reactive Streams support.
• Issues #1233 and #1721 document the interaction between downstream demand and read timeouts in the former streamed handler.
• Pull request #1843 removed StreamedAsyncHandler for AHC 3.
• Discussion #1925 asks how to migrate streamed consumers to AHC 3; the maintainer response declines restoring Reactive Streams because other libraries provide that policy and maintaining it adds overhead.
• Focused GitHub issue, pull-request, discussion, and local history searches found no existing AHC 3 proposal for a streaming-library-neutral suspend/resume/cancel response-body control.

Compatibility

• AsyncHandler.onResponseBodyStart is a new Java default method, so existing handler implementations remain source- and binary-compatible and retain their previous behavior unless they override it.
• ResponseBodyControl is a new public interface.
• NettyResponseBodyControl is public only to support AHC's cross-package transport integration and is marked @ApiStatus.Internal; consumers should depend on ResponseBodyControl instead.
• The full JDK 11 verification, including Revapi, passes.

AI disclosure

OpenAI Codex on behalf of Matthias Kurz. Both commits include Co-Authored-By: OpenAI Codex <codex@openai.com> per AGENTS.md.

Test plan

• On the first commit alone, the new HTTP/2 sibling-stream test reproduced shared connection-window starvation: a sibling response did not progress while the first response remained suspended.
• JAVA_HOME=<jdk-11> ./mvnw -pl client -Dtest=ResponseBodyControlTest,Http2ResponseBodyControlTest,TimeoutTimerTaskTest -Dio.netty.leakDetection.level=paranoid test: 16 tests passed (8 HTTP/1.1 control tests, 5 HTTP/2 control tests, and 3 timeout-task tests).
• JAVA_HOME=<jdk-11> ./mvnw clean verify -Dgpg.skip=true: BUILD SUCCESS for the full reactor, including tests, Javadocs, coverage, and Revapi.
• The Play WS feature/ahc-3-clean-upgrade branch passes its Scala 2.13 and Scala 3.3.8 test matrices, code validation, MiMa, documentation, and dependency-tree checks against the locally published AHC branch.

[hyperxpro]

This turns the connection level bound off for every H2 connection, not only the ones that use the new control. With the shipped defaults (http2InitialWindowSize 16 MiB, http2MaxConcurrentStreams -1) nothing aggregate is left, so the ceiling becomes the per stream window times whatever concurrency the server picks. Before this a slow consumer was capped at 64 KiB per connection.

Can we auto refill only while a control on that connection is actually suspended, or keep an aggregate byte budget? I would rather not change the default for users who never touch ResponseBodyControl. The README points at http2MaxConcurrentStreams as the mitigation, but that is off by default too.

[hyperxpro]

This runs before the two branches below that keep the exchange alive on purpose, the IOException filter one and recoverOnReadOrWriteException. In both we do not close the channel, but the control is already discarded, so the handler is left holding one where suspend(), resume() and cancel() are silent no ops. detach(false) also leaves autoRead off on a channel that survives. Should we move it down to the paths that really tear the channel down?

[hyperxpro]

Nit: this does an attribute lookup on every read complete for every channel, including each H2 stream child, and in the normal case readIfNeeded then no ops anyway. suspend0() always sets autoRead to false, so checking isAutoRead() first is equivalent and keeps the lookup off the hot path.

[hyperxpro]

When the HEADERS frame ends the stream we still create a control and call this, and finishUpdate runs a few lines down. A handler that suspends here gets no data and no error, and the resume() its consumer sends later hits !active and returns silently. Both new tests suspend unconditionally in onResponseBodyStart, so the path is exercised but the ignored suspend is never asserted. Either skip the callback when there is no body, or state on ResponseBodyControl that suspending in that case does nothing.

[hyperxpro]

finishUpdate calls complete() as its first statement, and if cancel() ran inline we already detached, so this never does anything. Same for the copy in HttpHandler.abortAfterStartingResponseBody. I would drop both and keep finishUpdate as the single completion point.

[hyperxpro]

Keying this on the channel is fragile. HttpHandler.handleRead returns early when applyIoExceptionFiltersAndReplayRequest replays and nothing detaches the control, so the channel enters drainChannelAndOffer still suspended with autoRead off. LastHttpContent never arrives, the drain never finishes, and the channel, its pool slot and the future are gone for good. If such a channel is reused later this throw fails a request that has nothing wrong with it. Keying the control to the exchange and replacing instead of throwing would remove both.

[hyperxpro]

ResponseBodyControl says calls made after completion have no effect, but this throws RejectedExecutionException at the caller once the event loop group is shutting down, which is exactly when a consumer calls cancel(). Catch it and treat it as the documented no op.

[hyperxpro]

With the request timeout disabled this re-arms forever and never reaches clean(), so a handler that suspends and then loses its consumer pins the future, the channel and the AsyncHandler for the life of the process. The description mentions the risk but there is no bound in the code. Can we cap the total suspended time, or at least log once when it runs long?

[hyperxpro]

cancelTimeouts() is called from finishUpdate on the event loop for every request, so this monitor now sits on the I/O path. The timer thread holds the same monitor inside startReadTimeout across nettyTimer.newTimeout(), which blocks on first use and contends on the wheel's queue. On an H2 connection that parks the loop for every other stream on it. Re-checking cancelled after scheduling, or a CAS on the readTimeout field, closes the same race without a shared lock.

[hyperxpro]

Nit: this is the same bootstrap as Http2StreamingBodyFlowControlTest and five other H2 tests. Worth pulling into a helper next to TestUtils at some point so the next pkitesting or ALPN change is one edit instead of eight.