diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java index f305fb3f3..7bd5977d2 100755 --- a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java @@ -1027,10 +1027,21 @@ protected void initChannel(Channel channel) throws Exception { } /** - * Checks whether the given channel is an HTTP/2 connection (i.e. has the HTTP/2 multiplex handler installed). + * Checks whether the given channel is an HTTP/2 connection: the parent that multiplexes streams, not one of + * its stream children, which carry neither the multiplex handler nor connection state of their own. + *

+ * Answered from the {@link Http2ConnectionState} attached to the connection rather than by looking + * {@link #HTTP2_MULTIPLEX} up in the pipeline. The two are attached together, in + * {@link #upgradePipelineToHttp2}, and agree for as long as the connection is live; a pipeline lookup, + * though, compares handler names down the chain, and an HTTP/1.1 connection, which has no such handler, is + * walked to the end to say no. + *

+ * They part on close, where Netty's own teardown strips the pipeline and leaves the attribute: a closed + * connection answers yes here and would have answered no to a lookup. Callers reach this while deciding + * what to do with a channel they have just taken from the pool, having checked it is active. */ public static boolean isHttp2(Channel channel) { - return channel.pipeline().get(HTTP2_MULTIPLEX) != null; + return channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get() != null; } /** @@ -1068,6 +1079,21 @@ public void upgradePipelineToHttp2(ChannelPipeline pipeline) { pipeline.remove(AHC_HTTP_HANDLER); } + // Attach HTTP/2 connection state for MAX_CONCURRENT_STREAMS tracking and GOAWAY draining. Its + // presence is also what marks the connection as HTTP/2; see isHttp2. Attached before the handlers + // rather than after them, so that there is no instant at which the pipeline speaks HTTP/2 and the + // connection does not yet say so -- a write landing there would take the HTTP/1.1 branch onto an + // HTTP/2 pipeline. Nothing can reach this channel that early today; the ordering is what keeps that + // from being something each new caller has to know. + Http2ConnectionState state = new Http2ConnectionState(); + int configMaxStreams = config.getHttp2MaxConcurrentStreams(); + if (configMaxStreams > 0) { + // Client's own cap; the server-advertised value (applied by the http2-settings-listener below) + // can only lower the effective limit, never raise it above this. + state.setClientMaxConcurrentStreams(configMaxStreams); + } + pipeline.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state); + // Add HTTP/2 frame codec (handles connection preface, SETTINGS, PING, flow control, etc.) Http2Settings settings = new Http2Settings() .initialWindowSize(config.getHttp2InitialWindowSize()) @@ -1097,16 +1123,6 @@ protected void initChannel(Channel ch) { pipeline.addLast(HTTP2_FRAME_CODEC, frameCodec); pipeline.addLast(HTTP2_MULTIPLEX, multiplexHandler); - // Attach HTTP/2 connection state for MAX_CONCURRENT_STREAMS tracking and GOAWAY draining - Http2ConnectionState state = new Http2ConnectionState(); - int configMaxStreams = config.getHttp2MaxConcurrentStreams(); - if (configMaxStreams > 0) { - // Client's own cap; the server-advertised value (applied by the http2-settings-listener below) - // can only lower the effective limit, never raise it above this. - state.setClientMaxConcurrentStreams(configMaxStreams); - } - pipeline.channel().attr(Http2ConnectionState.HTTP2_STATE_KEY).set(state); - // Install SETTINGS listener to update MAX_CONCURRENT_STREAMS from server pipeline.addLast("http2-settings-listener", new ChannelInboundHandlerAdapter() { @Override diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java index a17f74e31..33987acb8 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Continue100Interceptor.java @@ -41,9 +41,9 @@ public boolean exitAfterHandling100(final Channel channel, final NettyResponseFu if (channel instanceof Http2StreamChannel) { // HTTP/2: the HEADERS frame was already sent with endStream=false; now send the deferred body - // as DATA frame(s). writeRequest() can't be reused here — its isHttp2() check looks for the - // parent connection's multiplex handler, which a stream child channel doesn't have, so it would - // mis-route to the HTTP/1.1 writer (UnsupportedMessageTypeException + use-after-free). + // as DATA frame(s). writeRequest() can't be reused here — it routes on the parent connection's + // HTTP/2 state, which a stream child channel doesn't carry, so it would mis-route to the + // HTTP/1.1 writer (UnsupportedMessageTypeException + use-after-free). // // Only resume when the body was genuinely deferred. For a request WITHOUT Expect: 100-continue // the body was already written with endStream=true, so writing DATA now would be a frame after diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index c142bc62a..25bbad763 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -703,9 +703,12 @@ public void writeRequest(NettyResponseFuture future, Channel channel) { return; } - // Route to HTTP/2 path if the parent channel has the HTTP/2 multiplex handler installed - if (ChannelManager.isHttp2(channel)) { - writeHttp2Request(future, channel); + // Route to HTTP/2 when the connection carries HTTP/2 state, which is attached where the multiplex + // handler is. Read here rather than asked of ChannelManager.isHttp2, because the HTTP/2 path needs the + // state itself and would otherwise look up what this line has already found. + Http2ConnectionState http2State = channel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); + if (http2State != null) { + writeHttp2Request(future, channel, http2State); return; } @@ -771,11 +774,12 @@ public void writeRequest(NettyResponseFuture future, Channel channel) { * as HTTP/2 frames ({@link DefaultHttp2HeadersFrame} + optional {@link DefaultHttp2DataFrame}). * The stream child channel has the {@link org.asynchttpclient.netty.handler.Http2Handler} installed * and the {@link NettyResponseFuture} attached to it, mirroring the HTTP/1.1 channel model. + * + * @param state the connection's HTTP/2 state, which is what identified it as an HTTP/2 connection in the + * first place, so the caller has it in hand */ - private void writeHttp2Request(NettyResponseFuture future, Channel parentChannel) { - Http2ConnectionState state = parentChannel.attr(Http2ConnectionState.HTTP2_STATE_KEY).get(); - - if (state != null && !state.tryAcquireStream()) { + private void writeHttp2Request(NettyResponseFuture future, Channel parentChannel, Http2ConnectionState state) { + if (!state.tryAcquireStream()) { if (state.isDraining()) { // Connection is draining from GOAWAY — fail the future so it retries on a new connection. // Don't close the parent channel since it may still have active streams. sendHttp2Frames @@ -850,14 +854,12 @@ protected void initChannel(Http2StreamChannel streamCh) { if (openedRequest != null) { openedRequest.release(); } - if (state != null) { - state.releaseStream(); - // Close the parent once it has no active streams AND it is either draining - // (GOAWAY) or a redundant duplicate (#10 thundering-herd loser) — neither - // will serve further requests, so it must not linger open. - if ((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) { - channelManager.closeChannel(parentChannel); - } + state.releaseStream(); + // Close the parent once it has no active streams AND it is either draining + // (GOAWAY) or a redundant duplicate (#10 thundering-herd loser) — neither + // will serve further requests, so it must not linger open. + if ((state.isDraining() || state.isRedundant()) && state.getActiveStreams() <= 0) { + channelManager.closeChannel(parentChannel); } }); @@ -896,9 +898,7 @@ protected void initChannel(Http2StreamChannel streamCh) { } else { // Stream channel was never opened — no closeFuture will fire, so release the // acquired slot and the unsent request body inline. - if (state != null) { - state.releaseStream(); - } + state.releaseStream(); releaseHttp2Request(future); // Fail ONLY this future (future.abort, not abort(parentChannel, ...)): opening one stream // can fail for a stream-local reason (e.g. Netty rejecting it as the outbound max-streams diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2MarkerTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2MarkerTest.java new file mode 100644 index 000000000..16e8fcb28 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/ChannelManagerHttp2MarkerTest.java @@ -0,0 +1,113 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.channel; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http2.Http2StreamChannelBootstrap; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * {@link ChannelManager#isHttp2(Channel)} answers from the {@link Http2ConnectionState} attached to a + * connection, while the thing it stands for is the multiplex handler in the pipeline. These pin the two + * together: either both say HTTP/2 or neither does, whichever way a later change to the upgrade attaches them. + */ +class ChannelManagerHttp2MarkerTest { + + // One per class: the upgrade is what is under test and it needs a ChannelManager only to be called. Building + // one per test costs an SslContext and an event loop group each time, for state that lives on the channel. + private static ChannelManager channelManager; + private static Timer timer; + + private EmbeddedChannel channel; + + @BeforeAll + static void startManager() { + timer = new HashedWheelTimer(); + channelManager = new ChannelManager(config().build(), timer); + } + + @AfterAll + static void stopManager() { + if (channelManager != null) { + channelManager.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @BeforeEach + void setUp() { + channel = new EmbeddedChannel(); + } + + @AfterEach + void tearDown() { + if (channel != null) { + channel.finishAndReleaseAll(); + } + } + + @Test + void aConnectionThatWasNeverUpgradedIsNotHttp2() { + assertNull(channel.pipeline().get(ChannelManager.HTTP2_MULTIPLEX), + "an untouched pipeline should not carry the multiplex handler"); + assertFalse(ChannelManager.isHttp2(channel), "and should not be reported as HTTP/2"); + } + + @Test + void upgradingAConnectionBothInstallsTheHandlerAndReportsHttp2() { + channelManager.upgradePipelineToHttp2(channel.pipeline()); + + assertNotNull(channel.pipeline().get(ChannelManager.HTTP2_MULTIPLEX), + "the upgrade should install the multiplex handler"); + assertTrue(ChannelManager.isHttp2(channel), "and should report the connection as HTTP/2"); + } + + @Test + void aStreamOfAnHttp2ConnectionIsNotTheConnection() { + // A real stream child rather than a bare channel: what is worth pinning is that a stream does not + // inherit the connection state its parent carries, since that is now what identifies an HTTP/2 + // connection. The stream is where a request is written, so mistaking it for its parent would loop. + channelManager.upgradePipelineToHttp2(channel.pipeline()); + channel.runPendingTasks(); + + Channel stream = new Http2StreamChannelBootstrap(channel) + .handler(new ChannelInboundHandlerAdapter()) + .open().syncUninterruptibly().getNow(); + try { + assertNull(stream.pipeline().get(ChannelManager.HTTP2_MULTIPLEX), + "a stream child carries no multiplex handler of its own"); + assertFalse(ChannelManager.isHttp2(stream), "and is not the connection that multiplexes it"); + } finally { + stream.close().syncUninterruptibly(); + } + } +}