From e5bec59410e490b513251a098fb2a56d70c51082 Mon Sep 17 00:00:00 2001 From: Zakaria El Orche Date: Fri, 14 Aug 2026 22:51:16 +0000 Subject: [PATCH] fix(http2): close a streaming response body on every exit path Http2ResponseWriter.streamBody (a streaming response's InputStream, driven across startFlowControlled/resume as flow-control windows allow) was never closed anywhere -- not on clean EOF, not on a write failure, not when the stream is abandoned (RST_STREAM from the peer, connection teardown). Same bug Http1ResponseWriter had before cf16be0, just never given the same fix: a handler stream that releases a held resource (a pooled backend connection, for a reverse proxy) from close() leaks it under any real amount of stream resets or aborted connections. - appendData closes streamBody once `end` is reached (clean completion) and on any IOException from the read itself, mirroring Http1ResponseWriter's relayAndClose/writeChunkedAndClose reasoning. - New Http2ResponseWriter#abort(), called from Http2Stream#cancel() -- symmetric with that method's existing http2Body.cancel() for the inbound leg, now covering the outbound one too. cancel() is the single hook every abandoned-stream path (RST_STREAM handling and connection teardown in Http2Connection, plus Http2StreamDispatcher) already goes through, so this covers every abort case without adding a new one. closeStreamBodyQuietly() is idempotent (nulls streamBody after closing), so the appendData and abort() close paths can't double-close or race. Four new Http2ResponseWriterTest cases: normal completion in one call, completion across a resume() (multiple flow-control windows), abort() while still streaming, and abort() as a no-op on a non-streaming response. 697/697 flash-module tests green. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_011LLwcyHUnbApCrY33gvgoa --- .../http2/message/Http2ResponseWriter.java | 72 +++++++++++++----- .../flash/http2/stream/Http2Stream.java | 5 ++ .../message/Http2ResponseWriterTest.java | 74 +++++++++++++++++++ 3 files changed, 134 insertions(+), 17 deletions(-) diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java index 137646a..8c93060 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java @@ -251,32 +251,38 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining); count = 0; boolean eof = false; - if (pushBody) { - int read = streamBody.read(relay, 0, limit); - if (read < 0) eof = true; - else count = read; - } else { - while (count < limit) { - int read = streamBody.read(relay, count, limit - count); - if (read < 0) { - eof = true; - break; - } - if (read == 0) { - int one = streamBody.read(); - if (one < 0) { + try { + if (pushBody) { + int read = streamBody.read(relay, 0, limit); + if (read < 0) eof = true; + else count = read; + } else { + while (count < limit) { + int read = streamBody.read(relay, count, limit - count); + if (read < 0) { eof = true; break; } - relay[count++] = (byte) one; - } else { - count += read; + if (read == 0) { + int one = streamBody.read(); + if (one < 0) { + eof = true; + break; + } + relay[count++] = (byte) one; + } else { + count += read; + } } } + } catch (IOException failure) { + closeStreamBodyQuietly(); + throw failure; } if (!unknownLength) { bodyRemaining -= count; if (eof && bodyRemaining != 0) { + closeStreamBodyQuietly(); throw new IOException("streaming response ended before its declared length"); } } @@ -297,6 +303,38 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize endStreamInBatch = end; } finished = end; + if (end) closeStreamBodyQuietly(); + } + + /** + * Closes any in-flight streaming response body still open when the stream is abandoned before + * finishing (RST_STREAM from the peer, connection teardown) — see {@link + * dev.relism.flash.http2.stream.Http2Stream#cancel()}, the abort-path caller. Normal completion + * already closes {@link #streamBody} itself, from {@link #appendData} once {@code end} is + * reached; this covers everything else. A no-op if the response never streamed a body, or + * already finished/closed. + */ + public void abort() { + closeStreamBodyQuietly(); + } + + /** + * Best-effort, idempotent close — mirrors {@code Http1ResponseWriter}'s identical + * {@code relayAndClose}/{@code writeChunkedAndClose} reasoning (fixed there in {@code cf16be0}): + * a handler stream that releases a held resource (a pooled backend connection, for a reverse + * proxy) from {@code close()} leaks it under any real amount of failed writes or peer resets + * unless every exit path closes it, matching {@link InputStream#close()}'s own idempotency + * contract. + */ + private void closeStreamBodyQuietly() { + if (streamBody == null) return; + try { + streamBody.close(); + } catch (IOException ignored) { + // Best-effort — nothing to do about a failure to close an already-broken stream. + } finally { + streamBody = null; + } } private void appendTrailers(int maxFrameSize) { diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java index 1c7808d..fd9e581 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java @@ -310,6 +310,11 @@ public final class Http2Stream throw new IllegalStateException("failed to restore discarded flow-control bytes", failure); } } + // Symmetric with http2Body.cancel() above, for the outbound leg: closes a still-open + // streaming response body (e.g. a reverse proxy's pooled backend connection) so an abandoned + // stream — RST_STREAM from the peer, connection teardown — doesn't leak it. No-op if the + // response never streamed a body or already finished normally. + responseWriter.abort(); } public boolean cancelled() { diff --git a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java index 181ef01..ca593be 100644 --- a/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java @@ -13,6 +13,8 @@ import dev.relism.flash.http2.hpack.HpackDecoder; import dev.relism.flash.models.Response; import dev.relism.fpr.core.ByteView; import java.io.ByteArrayInputStream; +import java.io.FilterInputStream; +import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; @@ -149,6 +151,78 @@ class Http2ResponseWriterTest { assertTrue(writer.trailerHeadersInBatch()); } + /** + * Regression for the connection leak fixed alongside {@code Http1ResponseWriter}'s identical + * bug ({@code cf16be0}): {@code streamBody} was never closed on any exit path, so a handler + * stream releasing a held resource (a pooled backend connection, for a reverse proxy) from + * {@code close()} leaked it. + */ + @Test + void startFlowControlled_closesStreamBodyOnceFullyDrainedInOneCall() throws Exception { + TrackingInputStream body = new TrackingInputStream(new byte[] {1, 2, 3, 4}); + Response response = new Response(200, ContentType.BINARY).stream(body, 4); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + writer.startFlowControlled(response, 1, false, false, true, false, false, 16_384, 4096, 16_384); + + assertTrue(writer.finished()); + assertTrue(body.closed, "streamBody must be closed once the response finished normally"); + } + + @Test + void resume_closesStreamBodyOnceFullyDrainedAcrossFlowControlWindows() throws Exception { + TrackingInputStream body = new TrackingInputStream(new byte[] {1, 2, 3, 4}); + Response response = new Response(200, ContentType.BINARY).stream(body, 4); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + // availableFlowWindow of 2 forces a second batch via resume() to drain the remaining bytes. + writer.startFlowControlled(response, 1, false, false, true, false, false, 16_384, 4096, 2); + assertFalse(writer.finished()); + assertFalse(body.closed, "must not close before the response actually finishes"); + + writer.resume(16_384, 16_384); + + assertTrue(writer.finished()); + assertTrue(body.closed); + } + + @Test + void abort_closesAStillOpenStreamBody() throws Exception { + TrackingInputStream body = new TrackingInputStream(new byte[] {1, 2, 3, 4}); + Response response = new Response(200, ContentType.BINARY).stream(body, 4); + Http2ResponseWriter writer = new Http2ResponseWriter(); + + // availableFlowWindow of 0: prepares headers only, the body is still fully unread. + writer.startFlowControlled(response, 1, false, false, true, false, false, 16_384, 4096, 0); + assertFalse(writer.finished()); + assertFalse(body.closed); + + writer.abort(); + + assertTrue(body.closed, "an abandoned stream (RST_STREAM/connection teardown) must still close streamBody"); + } + + @Test + void abort_isANoOpForANonStreamingResponse() { + Http2ResponseWriter writer = new Http2ResponseWriter(); + // No streamBody was ever set — must not throw. + writer.abort(); + } + + private static final class TrackingInputStream extends FilterInputStream { + boolean closed; + + TrackingInputStream(byte[] data) { + super(new ByteArrayInputStream(data)); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + } + private static Parsed parse(Http2ResponseWriter writer) { Parsed parsed = new Parsed(); byte[] wire = writer.buffer();