diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/config/H2Config.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/config/H2Config.java index b8a2c376d4..06dbf2303e 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/config/H2Config.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/config/H2Config.java @@ -51,10 +51,11 @@ public class H2Config { private final int maxHeaderListSize; private final boolean compressionEnabled; private final int maxContinuations; + private final int maxControlFrames; H2Config(final int headerTableSize, final boolean pushEnabled, final int maxConcurrentStreams, final int initialWindowSize, final int maxFrameSize, final int maxHeaderListSize, - final boolean compressionEnabled, final int maxContinuations) { + final boolean compressionEnabled, final int maxContinuations, final int maxControlFrames) { super(); this.headerTableSize = headerTableSize; this.pushEnabled = pushEnabled; @@ -64,6 +65,7 @@ public class H2Config { this.maxHeaderListSize = maxHeaderListSize; this.compressionEnabled = compressionEnabled; this.maxContinuations = maxContinuations; + this.maxControlFrames = maxControlFrames; } public int getHeaderTableSize() { @@ -98,6 +100,19 @@ public int getMaxContinuations() { return maxContinuations; } + /** + * Returns the maximum number of control frames (PING and SETTINGS acknowledgements, WINDOW_UPDATE + * and RST_STREAM frames generated in response to peer frames) that may be pending output before + * the connection is closed. A GOAWAY carrying {@code ENHANCE_YOUR_CALM} is sent when the output + * still permits it; a peer that stops reading forces an immediate close instead. A value of zero + * disables the limit. + * + * @since 5.5 + */ + public int getMaxControlFrames() { + return maxControlFrames; + } + @Override public String toString() { final StringBuilder builder = new StringBuilder(); @@ -109,6 +124,7 @@ public String toString() { .append(", maxHeaderListSize=").append(this.maxHeaderListSize) .append(", compressionEnabled=").append(this.compressionEnabled) .append(", maxContinuations=").append(this.maxContinuations) + .append(", maxControlFrames=").append(this.maxControlFrames) .append("]"); return builder.toString(); } @@ -122,6 +138,7 @@ public static H2Config.Builder custom() { private static final int INIT_MAX_FRAME_SIZE = FrameConsts.MIN_FRAME_SIZE; private static final int INIT_WINDOW_SIZE = 65535; private static final int INIT_CONCURRENT_STREAM = 250; + private static final int INIT_MAX_CONTROL_FRAMES = 1000; public static H2Config.Builder initial() { return new Builder() @@ -142,7 +159,9 @@ public static H2Config.Builder copy(final H2Config config) { .setInitialWindowSize(config.getInitialWindowSize()) .setMaxFrameSize(config.getMaxFrameSize()) .setMaxHeaderListSize(config.getMaxHeaderListSize()) - .setCompressionEnabled(config.isCompressionEnabled()); + .setCompressionEnabled(config.isCompressionEnabled()) + .setMaxContinuations(config.getMaxContinuations()) + .setMaxControlFrames(config.getMaxControlFrames()); } public static class Builder { @@ -155,6 +174,7 @@ public static class Builder { private int maxHeaderListSize; private boolean compressionEnabled; private int maxContinuations; + private int maxControlFrames; Builder() { this.headerTableSize = INIT_HEADER_TABLE_SIZE * 2; @@ -165,6 +185,7 @@ public static class Builder { this.maxHeaderListSize = FrameConsts.MAX_FRAME_SIZE; this.compressionEnabled = true; this.maxContinuations = 100; + this.maxControlFrames = INIT_MAX_CONTROL_FRAMES; } public Builder setHeaderTableSize(final int headerTableSize) { @@ -215,6 +236,21 @@ public Builder setMaxContinuations(final int maxContinuations) { return this; } + /** + * Sets the maximum number of control frames (PING and SETTINGS acknowledgements, WINDOW_UPDATE + * and RST_STREAM frames generated in response to peer frames) that may be pending output before + * the connection is closed. A GOAWAY carrying {@code ENHANCE_YOUR_CALM} is sent when the output + * still permits it; a peer that stops reading forces an immediate close instead. A value of zero + * disables the limit. + * + * @since 5.5 + */ + public Builder setMaxControlFrames(final int maxControlFrames) { + Args.notNegative(maxControlFrames, "Max control frames"); + this.maxControlFrames = maxControlFrames; + return this; + } + public H2Config build() { return new H2Config( headerTableSize, @@ -224,7 +260,8 @@ public H2Config build() { maxFrameSize, maxHeaderListSize, compressionEnabled, - maxContinuations); + maxContinuations, + maxControlFrames); } } diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java index 96146ce91f..f7b1024ea9 100644 --- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java +++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java @@ -143,6 +143,10 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED } private EndpointDetails endpointDetails; private boolean goAwayReceived; + // Number of control-frame responses (PING/SETTINGS acks, WINDOW_UPDATE, RST_STREAM) currently + // held in the output queue awaiting a flush. Guarded by the IO session lock. + private int queuedControlFrames; + private volatile boolean peerNoRfc7540Priorities; @@ -307,11 +311,27 @@ private void commitFrameInternal(final RawFrame frame) throws IOException { } outputBuffer.write(frame, ioSession); } else { + if (isControlFrame(frame)) { + final int maxControlFrames = localConfig.getMaxControlFrames(); + if (maxControlFrames > 0 && queuedControlFrames >= maxControlFrames) { + throw new H2ConnectionException(H2Error.ENHANCE_YOUR_CALM, + "Maximum number of outstanding control frames exceeded"); + } + queuedControlFrames++; + } outputQueue.addLast(frame); } ioSession.setEvent(SelectionKey.OP_WRITE); } + private static boolean isControlFrame(final RawFrame frame) { + final int type = frame.getType(); + return type == FrameType.SETTINGS.getValue() + || type == FrameType.PING.getValue() + || type == FrameType.WINDOW_UPDATE.getValue() + || type == FrameType.RST_STREAM.getValue(); + } + private void commitFrame(final RawFrame frame) throws IOException { Args.notNull(frame, "Frame"); ioSession.getLock().lock(); @@ -509,6 +529,9 @@ public final void onOutput() throws HttpException, IOException { while (outputBuffer.isEmpty()) { final RawFrame frame = outputQueue.poll(); if (frame != null) { + if (isControlFrame(frame)) { + queuedControlFrames--; + } if (streamListener != null) { streamListener.onFrameOutput(this, frame.getStreamId(), frame); } diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/config/TestH2Config.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/config/TestH2Config.java new file mode 100644 index 0000000000..1af6876084 --- /dev/null +++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/config/TestH2Config.java @@ -0,0 +1,62 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ + +package org.apache.hc.core5.http2.config; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestH2Config { + + @Test + void testDefaultMaxControlFrames() { + Assertions.assertEquals(1000, H2Config.DEFAULT.getMaxControlFrames()); + } + + @Test + void testMaxControlFramesIsConfigurable() { + final H2Config config = H2Config.custom().setMaxControlFrames(0).build(); + Assertions.assertEquals(0, config.getMaxControlFrames()); + } + + @Test + void testMaxControlFramesRejectsNegative() { + Assertions.assertThrows(IllegalArgumentException.class, + () -> H2Config.custom().setMaxControlFrames(-1)); + } + + @Test + void testCopyPreservesMaxControlFramesAndMaxContinuations() { + final H2Config original = H2Config.custom() + .setMaxControlFrames(42) + .setMaxContinuations(7) + .build(); + final H2Config copy = H2Config.copy(original).build(); + Assertions.assertEquals(42, copy.getMaxControlFrames()); + Assertions.assertEquals(7, copy.getMaxContinuations()); + } +} diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodIntegration.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodIntegration.java new file mode 100644 index 0000000000..e69cf00e6f --- /dev/null +++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodIntegration.java @@ -0,0 +1,222 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ + +package org.apache.hc.core5.http2.impl.nio; + +import java.io.BufferedOutputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.core5.http.impl.bootstrap.HttpAsyncServer; +import org.apache.hc.core5.http.nio.support.ImmediateResponseExchangeHandler; +import org.apache.hc.core5.http2.H2ConnectionException; +import org.apache.hc.core5.http2.H2Error; +import org.apache.hc.core5.http2.HttpVersionPolicy; +import org.apache.hc.core5.http2.config.H2Config; +import org.apache.hc.core5.http2.impl.nio.bootstrap.H2ServerBootstrap; +import org.apache.hc.core5.io.CloseMode; +import org.apache.hc.core5.reactor.IOReactorConfig; +import org.apache.hc.core5.reactor.IOSession; +import org.apache.hc.core5.reactor.IOSessionListener; +import org.apache.hc.core5.reactor.ListenerEndpoint; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** + * Drives a real HTTP/2 server with a raw-socket peer that floods PING frames without reading the + * socket. The server-side termination reason is observed through an {@link IOSessionListener} so + * the test distinguishes the control-frame bound from any other cause of connection failure. + */ +class TestH2ControlFrameFloodIntegration { + + private static final int FLOOD_SIZE = 50_000; + + private HttpAsyncServer server; + + @AfterEach + void cleanup() { + if (server != null) { + server.close(CloseMode.IMMEDIATE); + } + } + + private int startServer(final int maxControlFrames, final AtomicReference firstException, + final CountDownLatch exceptionLatch) throws Exception { + server = H2ServerBootstrap.bootstrap() + .setVersionPolicy(HttpVersionPolicy.FORCE_HTTP_2) + .setIOReactorConfig(IOReactorConfig.custom().setSndBufSize(8 * 1024).build()) + .setH2Config(H2Config.custom().setPushEnabled(false).setMaxControlFrames(maxControlFrames).build()) + .setIOSessionListener(new IOSessionListener() { + + @Override + public void connected(final IOSession session) { + } + + @Override + public void startTls(final IOSession session) { + } + + @Override + public void inputReady(final IOSession session) { + } + + @Override + public void outputReady(final IOSession session) { + } + + @Override + public void timeout(final IOSession session) { + } + + @Override + public void exception(final IOSession session, final Exception ex) { + if (firstException.compareAndSet(null, ex)) { + exceptionLatch.countDown(); + } + } + + @Override + public void disconnected(final IOSession session) { + } + + }) + .register("*", () -> new ImmediateResponseExchangeHandler(200, "OK")) + .create(); + server.start(); + final ListenerEndpoint endpoint = server.listen(new InetSocketAddress("127.0.0.1", 0)).get(); + return ((InetSocketAddress) endpoint.getAddress()).getPort(); + } + + @Test + @Timeout(30) + void testPingFloodTerminatedWithEnhanceYourCalm() throws Exception { + final AtomicReference firstException = new AtomicReference<>(); + final CountDownLatch exceptionLatch = new CountDownLatch(1); + final int port = startServer(1000, firstException, exceptionLatch); + + try (Socket socket = new Socket()) { + socket.setReceiveBufferSize(4096); + socket.connect(new InetSocketAddress("127.0.0.1", port)); + socket.setTcpNoDelay(true); + + final OutputStream out = new BufferedOutputStream(socket.getOutputStream(), 1 << 16); + out.write("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)); + writeFrame(out, 4, 0, 0, null); // empty SETTINGS + out.flush(); + + // Flood PING frames and never read the socket: the acknowledgements cannot be flushed. + final byte[] pingPayload = new byte[8]; + try { + for (int i = 0; i < FLOOD_SIZE; i++) { + writeFrame(out, 6, 0, 0, pingPayload); + if ((i & 0x3FF) == 0) { + out.flush(); + } + } + out.flush(); + } catch (final java.io.IOException ignore) { + // The server closing the connection manifests as a write failure for the flooder. + } + + // The connection is terminated for exactly one reason: the control-frame bound. + Assertions.assertTrue(exceptionLatch.await(20, TimeUnit.SECONDS), + "server did not raise a connection exception for the flood"); + final Exception ex = firstException.get(); + Assertions.assertInstanceOf(H2ConnectionException.class, ex, + "unexpected server-side failure: " + ex); + Assertions.assertEquals(H2Error.ENHANCE_YOUR_CALM.getCode(), ((H2ConnectionException) ex).getCode(), + "connection closed for a reason other than the control-frame limit: " + ex.getMessage()); + } + } + + @Test + @Timeout(30) + void testFloodNotTerminatedWhenLimitDisabled() throws Exception { + final AtomicReference firstException = new AtomicReference<>(); + final CountDownLatch exceptionLatch = new CountDownLatch(1); + final int port = startServer(0, firstException, exceptionLatch); + + try (Socket socket = new Socket()) { + socket.setReceiveBufferSize(4096); + socket.connect(new InetSocketAddress("127.0.0.1", port)); + socket.setTcpNoDelay(true); + + final OutputStream out = new BufferedOutputStream(socket.getOutputStream(), 1 << 16); + out.write("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n".getBytes(StandardCharsets.ISO_8859_1)); + writeFrame(out, 4, 0, 0, null); // empty SETTINGS + out.flush(); + + // Negative control: the same flood against a disabled limit must not trip ENHANCE_YOUR_CALM. + final byte[] pingPayload = new byte[8]; + try { + for (int i = 0; i < FLOOD_SIZE; i++) { + writeFrame(out, 6, 0, 0, pingPayload); + if ((i & 0x3FF) == 0) { + out.flush(); + } + } + out.flush(); + } catch (final java.io.IOException ignore) { + } + + // No control-frame exception is raised within a bounded observation window. + final boolean raised = exceptionLatch.await(2, TimeUnit.SECONDS); + if (raised) { + final Exception ex = firstException.get(); + final boolean enhanceYourCalm = ex instanceof H2ConnectionException + && ((H2ConnectionException) ex).getCode() == H2Error.ENHANCE_YOUR_CALM.getCode(); + Assertions.assertFalse(enhanceYourCalm, + "the flood was terminated with ENHANCE_YOUR_CALM although the limit was disabled"); + } + } + } + + private static void writeFrame(final OutputStream out, final int type, final int flags, + final int streamId, final byte[] payload) throws Exception { + final int len = payload != null ? payload.length : 0; + out.write((len >>> 16) & 0xff); + out.write((len >>> 8) & 0xff); + out.write(len & 0xff); + out.write(type & 0xff); + out.write(flags & 0xff); + out.write((streamId >>> 24) & 0xff); + out.write((streamId >>> 16) & 0xff); + out.write((streamId >>> 8) & 0xff); + out.write(streamId & 0xff); + if (payload != null) { + out.write(payload); + } + } + +} diff --git a/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodOutputQueue.java b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodOutputQueue.java new file mode 100644 index 0000000000..9cc19e0a38 --- /dev/null +++ b/httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestH2ControlFrameFloodOutputQueue.java @@ -0,0 +1,270 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ + +package org.apache.hc.core5.http2.impl.nio; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; + +import org.apache.hc.core5.function.Supplier; +import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.HttpConnection; +import org.apache.hc.core5.http.config.CharCodingConfig; +import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler; +import org.apache.hc.core5.http.nio.AsyncPushConsumer; +import org.apache.hc.core5.http.nio.AsyncPushProducer; +import org.apache.hc.core5.http.nio.HandlerFactory; +import org.apache.hc.core5.http.protocol.HttpContext; +import org.apache.hc.core5.http.protocol.HttpProcessor; +import org.apache.hc.core5.http2.H2ConnectionException; +import org.apache.hc.core5.http2.H2Error; +import org.apache.hc.core5.http2.config.H2Config; +import org.apache.hc.core5.http2.config.H2Param; +import org.apache.hc.core5.http2.config.H2Setting; +import org.apache.hc.core5.http2.frame.DefaultFrameFactory; +import org.apache.hc.core5.http2.frame.FrameFactory; +import org.apache.hc.core5.http2.frame.FrameFlag; +import org.apache.hc.core5.http2.frame.FrameType; +import org.apache.hc.core5.http2.frame.RawFrame; +import org.apache.hc.core5.http2.frame.StreamIdGenerator; +import org.apache.hc.core5.reactor.ProtocolIOSession; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Verifies that a peer flooding control frames while never reading the socket cannot drive the + * per-connection output queue without bound. The output channel is held blocked so every control + * response accumulates in the queue, reproducing the condition under which the server would + * otherwise retain one buffered frame per inbound control frame. + */ +@ExtendWith(MockitoExtension.class) +class TestH2ControlFrameFloodOutputQueue { + + private static final FrameFactory FRAME_FACTORY = DefaultFrameFactory.INSTANCE; + + @Mock(strictness = Mock.Strictness.LENIENT) + ProtocolIOSession ioSession; + @Mock + Lock lock; + @Mock + HttpProcessor httpProcessor; + @Mock + H2StreamHandler streamHandler; + + private volatile boolean writeBlocked; + private final AtomicInteger pingAckOut = new AtomicInteger(); + + @BeforeEach + void prepareMocks() throws Exception { + Mockito.when(ioSession.getLock()).thenReturn(lock); + // When blocked, every write reports zero bytes flushed; otherwise the whole buffer drains. + Mockito.lenient().when(ioSession.write(ArgumentMatchers.any(ByteBuffer.class))).thenAnswer(inv -> { + final ByteBuffer src = inv.getArgument(0, ByteBuffer.class); + if (writeBlocked) { + return 0; + } + final int n = src.remaining(); + src.position(src.limit()); + return n; + }); + } + + private final H2StreamListener frameCountingListener = new H2StreamListener() { + @Override + public void onHeaderInput(final HttpConnection c, final int id, final List h) { + } + + @Override + public void onHeaderOutput(final HttpConnection c, final int id, final List h) { + } + + @Override + public void onFrameInput(final HttpConnection c, final int id, final RawFrame f) { + } + + @Override + public void onFrameOutput(final HttpConnection c, final int id, final RawFrame f) { + if (f.getType() == FrameType.PING.getValue() && f.isFlagSet(FrameFlag.ACK)) { + pingAckOut.incrementAndGet(); + } + } + + @Override + public void onInputFlowControl(final HttpConnection c, final int id, final int d, final int s) { + } + + @Override + public void onOutputFlowControl(final HttpConnection c, final int id, final int d, final int s) { + } + }; + + // Uses the default H2Config control-frame bound so this behaviour test compiles and runs both + // with and without the production change under test. + private AbstractH2StreamMultiplexer newMux() { + return new Mux(ioSession, FRAME_FACTORY, StreamIdGenerator.ODD, httpProcessor, CharCodingConfig.DEFAULT, + H2Config.custom().setPushEnabled(false).build(), + frameCountingListener, () -> streamHandler); + } + + private static byte[] pingFrame() { + return new byte[] {0, 0, 8, 6, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; + } + + private static byte[] emptySettingsFrame() { + return new byte[] {0, 0, 0, 4, 0, 0, 0, 0, 0}; + } + + @Test + @Timeout(30) + void testControlFrameFloodIsRejectedWithEnhanceYourCalm() throws Exception { + final AbstractH2StreamMultiplexer mux = newMux(); + writeBlocked = true; + mux.onConnect(); + mux.onInput(ByteBuffer.wrap(emptySettingsFrame())); + + final byte[] ping = pingFrame(); + final H2ConnectionException ex = Assertions.assertThrows(H2ConnectionException.class, () -> { + for (int i = 0; i < 100_000; i++) { + mux.onInput(ByteBuffer.wrap(ping.clone())); + } + }); + Assertions.assertEquals(H2Error.ENHANCE_YOUR_CALM.getCode(), ex.getCode()); + } + + @Test + @Timeout(30) + void testBurstUnderLimitIsFullyAcknowledged() throws Exception { + final AbstractH2StreamMultiplexer mux = newMux(); + writeBlocked = true; + mux.onConnect(); + mux.onInput(ByteBuffer.wrap(emptySettingsFrame())); + + final int burst = 200; + final byte[] ping = pingFrame(); + Assertions.assertDoesNotThrow(() -> { + for (int i = 0; i < burst; i++) { + mux.onInput(ByteBuffer.wrap(ping.clone())); + } + }); + + // The socket becomes writable; every queued acknowledgement must be flushed. + writeBlocked = false; + mux.onOutput(); + Assertions.assertEquals(burst, pingAckOut.get()); + + // The connection remains usable: a further PING is acknowledged normally. + mux.onInput(ByteBuffer.wrap(ping.clone())); + Assertions.assertEquals(burst + 1, pingAckOut.get()); + } + + @Test + @Timeout(30) + void testReadingPeerIsNeverLimited() throws Exception { + final AbstractH2StreamMultiplexer mux = newMux(); + writeBlocked = false; + mux.onConnect(); + mux.onInput(ByteBuffer.wrap(emptySettingsFrame())); + + final int count = 2000; + final byte[] ping = pingFrame(); + Assertions.assertDoesNotThrow(() -> { + for (int i = 0; i < count; i++) { + mux.onInput(ByteBuffer.wrap(ping.clone())); + } + }); + Assertions.assertEquals(count, pingAckOut.get()); + } + + static final class Mux extends AbstractH2StreamMultiplexer { + + private final Supplier supplier; + + Mux(final ProtocolIOSession ioSession, final FrameFactory frameFactory, final StreamIdGenerator idGenerator, + final HttpProcessor httpProcessor, final CharCodingConfig charCodingConfig, final H2Config h2Config, + final H2StreamListener streamListener, final Supplier supplier) { + super(ioSession, frameFactory, idGenerator, httpProcessor, charCodingConfig, h2Config, streamListener); + this.supplier = supplier; + } + + @Override + void validateSetting(final H2Param param, final int value) { + } + + @Override + H2Setting[] generateSettings(final H2Config localConfig) { + return new H2Setting[] {new H2Setting(H2Param.ENABLE_PUSH, 0)}; + } + + @Override + void acceptHeaderFrame() { + } + + @Override + void acceptPushRequest() { + } + + @Override + void acceptPushFrame() { + } + + @Override + H2StreamHandler incomingRequest(final H2StreamChannel channel) { + return supplier.get(); + } + + @Override + H2StreamHandler outgoingRequest(final H2StreamChannel channel, final AsyncClientExchangeHandler exchangeHandler, + final HandlerFactory pushHandlerFactory, final HttpContext context) { + return null; + } + + @Override + H2StreamHandler incomingPushPromise(final H2StreamChannel channel, + final HandlerFactory pushHandlerFactory) { + return supplier.get(); + } + + @Override + H2StreamHandler outgoingPushPromise(final H2StreamChannel channel, final AsyncPushProducer pushProducer) { + return null; + } + + @Override + boolean allowGracefulAbort(final H2Stream stream) { + return false; + } + } +}