[mkurz]

Summary

• Add ResponseBodyControl.execute(Runnable), which runs a task on the event loop that delivers the response body, in sequence with the callbacks AHC makes there, such as onBodyPartReceived. Calls to suspend(), resume() and cancel() from the task take effect immediately.
• This lets a consumer that buffers body parts on another thread decide whether to resume based on everything delivered so far, instead of on a decision that a part delivered in the meantime made stale.
• NettyResponseBodyControl already applied its own control calls this way; that method is now public. The default implementation runs the task on the calling thread.

Problem

A typical streaming consumer buffers body parts from onBodyPartReceived and hands them to a reader on another thread. When that reader has emptied the buffer and still wants more, it calls resume(). Called from that thread, resume() only takes effect later, on the event loop. In between, the event loop can still deliver parts from the current socket read, and those parts may already meet the reader's demand. The queued resume() then lets one more socket read through that nobody asked for.

With a highly compressible body, that one extra read can inflate to many megabytes beyond demand. For a gzipped stream with early demand from another thread, Play WS measured about 35 MB buffered beyond demand, against about 1.9 MB when the decision is re-checked on the event loop.

AHC cannot fix this by re-checking the decision itself, because only the consumer knows which state its decision depends on. Play WS currently works around it by reaching the event loop through Netty's internal ThreadExecutorMap, and then making its resume decision there.

Change

default void execute(Runnable task) {
    task.run();
}

• On the event loop, for example from within a body callback, the task runs before execute returns.
• From another thread, it is queued to the event loop and runs there after the work already pending on it. The event loop can deliver further body parts before the task runs; the task sees all of them.
• Either way, suspend(), resume() and cancel() called from the task take effect immediately.
• Not every callback runs on the event loop. onThrowable can run on the thread that cancels the request, or on a timer thread when a timeout fires. A task submitted from there is queued like one from any other thread, and can run while that callback is still running. The Javadoc says so; the implementation does not try to change it.
• The task must not block. Unlike calls to the other methods, it also runs after the response has completed (the interface Javadoc now says so). If the event loop no longer accepts tasks, it does not run.
• The default implementation runs the task on the calling thread. It suits implementations that apply control calls synchronously on the calling thread, and does not order the task with callbacks on other threads.

NettyResponseBodyControl.execute is the existing private method made public, with a null check. It works for HTTP/1.1 and for HTTP/2 stream channels.

Alternatives considered

• resumeIf(BooleanSupplier), a resume whose condition AHC evaluates on the event loop. It needs subtle ordering rules: a true condition counts as a resume at that moment and overrides an earlier suspend. A consumer whose own drain loop may run on either thread then has to fit its state machine to those rules, and we found that hard to get right. With execute, such a consumer can write control.execute(() -> { if (cond) control.resume(); }).
• Demand counting in AHC (request(n)). HTTP/1.1 decodes one socket read into several parts regardless of demand, so AHC would need its own buffer. That's a much larger change.

execute only exposes what the control already does internally. Callbacks already run on the event loop under the same rule not to block.

Compatibility

• A new default method on ResponseBodyControl. Revapi passed. Existing implementations keep compiling and behave as before, with two exceptions:
  • an implementation that declares a non-public execute(Runnable) no longer compiles;
  • one that already has a public void execute(Runnable) now implements this method, with whatever semantics it has.
• No behavior change for existing callers.

AI disclosure

Claude Code on behalf of Matthias Kurz. The commit includes Co-Authored-By per AGENTS.md.

Test plan

New tests:

• ResponseBodyControlTest:
  • execute runs inline in a callback, runs on the event loop when called from another thread, and runs after control calls queued before it;
  • a consumer thread that decides inside execute sees the part delivered in the meantime and does not resume;
  • execute after the response has completed still runs the task on the event loop;
  • the default implementation runs the task on the calling thread.
• Http2ResponseBodyControlTest: a task from another thread resumes a suspended HTTP/2 stream.
• NettyResponseBodyControlExecuteTest: once the event loop has shut down, execute neither runs the task nor throws.

Broken variants, each run against these tests and the existing control tests (24 tests in all):

• The task always runs on the calling thread: 9 failures, including "a task from another thread must run on the channel event loop".
• The task is always queued: 3 failures, including "a task from a callback must run before execute returns".
• A rejected task throws: the shutdown test fails.
• No task after completion: the completion test times out (plus failures of existing tests, since the control's own calls use the same method).

Earlier, at the first revision, deciding on the other thread without execute failed "the task must see the part buffered after it was queued".

Runs:

• ResponseBodyControlTest, Http2ResponseBodyControlTest and NettyResponseBodyControlExecuteTest on JDK 11: 24 tests passed.

• ./mvnw -B -ntp -Dgpg.skip=true clean verify on JDK 11: 1,797 tests, 0 failures, 0 errors, 22 skipped; Revapi passed. No test-skipping flags were used.

• The same 24 tests, compiled on JDK 11 and run on JDK 17, 21 and 25 (Maven Surefire's -Djvm): 0 failures on each.

• Play WS, with its resume check moved from ThreadExecutorMap to execute, against a local build of the first revision of this branch alone (Scala 2.13):

  • all Play WS tests pass, except a new cookie test that needs the separate raw-Cookie fix;
  • its gzip streaming test buffers 1,875,916 bytes beyond demand, as with the ThreadExecutorMap workaround;
  • with Play WS resuming directly instead of through execute, the same test buffers 35,589,292 bytes and fails.

  This revision changes only Javadoc and tests.

• All four open AHC changes merged together (raw Cookie headers on retries, ResponseBodyControl.execute, suspend/resume order, demand-bounded decompression): they merge without conflicts, together and in every pair, and ./mvnw -B -ntp -Dgpg.skip=true clean install on JDK 11 passes with 1,853 tests, 0 failures, 0 errors, 22 skipped; Revapi passed.

• Play WS against that merged build, Scala 2.13 and 3.3, JDK 17: all tests pass.

[hyperxpro]

This is not what happens on most of the paths that end in onThrowable. Timeouts are armed on the channel's loop in TimeoutsHolder.arm, and exceptionCaught, channelInactive and Http2Handler.streamFailed all reach abort on the loop, so inEventLoop() is true and the task runs inline before execute returns, after terminateAndExit has already deactivated the control. We should say the task can run inline from onThrowable with the control already inactive, otherwise a caller that publishes state after execute returns runs the task on stale state.

[hyperxpro]

suspend() from the task only flips the flag, it does not hold back what is already decoded: on HTTP/1.1 every HttpContent from the same read is still delivered, and on an h2 stream nothing is held back once the peer's END_STREAM is queued, doBeginRead drains it regardless of autoRead. The other direction is missing too: resume() on an h2 stream calls read() synchronously, so the remaining frames and even onCompleted can run inside the task before resume() returns. I'd say "are applied before the task returns", point at the caveat suspend() already has, and add that resume() may deliver parts and complete the response before it returns.

[hyperxpro]

What should happen when task throws? From here it escapes into HttpHandler.handleRead, which fails the request and closes the connection (resets the stream on h2), but from another thread Netty's safeExecute just logs a WARN and drops it, and the default on the interface propagates to the caller. Either catch in the queued branch and route it to the same failure path, or document that the task must not throw. Right now a caller can't write a correct task without reading this class.

[hyperxpro]

A decorator written against 3.0.14 that forwards suspend, resume and cancel to a NettyResponseBodyControl and does not override this one gets the calling-thread default and silently loses the ordering this method exists for. Worth a sentence that implementations delegating to another control must delegate execute too.

[hyperxpro]

NettyResponseBodyControl does requireNonNull(task, "task"), this one NPEs from task.run() without a message. Same check here?

[hyperxpro]

assertNotSame passes for any thread that is not the test thread, so this stays green if the task ran on a timer thread, the resume() inside gets queued to the loop anyway and the request still completes. The HTTP/1.1 tests check channel.eventLoop().inEventLoop(taskThread), can we do the same here? setHttpAdditionalChannelInitializer hands out the parent channel for h2 too and the stream channel shares its loop, or record the thread in onBodyPartReceived and assertSame against it.