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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011LLwcyHUnbApCrY33gvgoa
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
787ae610d4
commit
e5bec59410
@@ -251,32 +251,38 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
|
|||||||
int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining);
|
int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining);
|
||||||
count = 0;
|
count = 0;
|
||||||
boolean eof = false;
|
boolean eof = false;
|
||||||
if (pushBody) {
|
try {
|
||||||
int read = streamBody.read(relay, 0, limit);
|
if (pushBody) {
|
||||||
if (read < 0) eof = true;
|
int read = streamBody.read(relay, 0, limit);
|
||||||
else count = read;
|
if (read < 0) eof = true;
|
||||||
} else {
|
else count = read;
|
||||||
while (count < limit) {
|
} else {
|
||||||
int read = streamBody.read(relay, count, limit - count);
|
while (count < limit) {
|
||||||
if (read < 0) {
|
int read = streamBody.read(relay, count, limit - count);
|
||||||
eof = true;
|
if (read < 0) {
|
||||||
break;
|
|
||||||
}
|
|
||||||
if (read == 0) {
|
|
||||||
int one = streamBody.read();
|
|
||||||
if (one < 0) {
|
|
||||||
eof = true;
|
eof = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
relay[count++] = (byte) one;
|
if (read == 0) {
|
||||||
} else {
|
int one = streamBody.read();
|
||||||
count += read;
|
if (one < 0) {
|
||||||
|
eof = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
relay[count++] = (byte) one;
|
||||||
|
} else {
|
||||||
|
count += read;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} catch (IOException failure) {
|
||||||
|
closeStreamBodyQuietly();
|
||||||
|
throw failure;
|
||||||
}
|
}
|
||||||
if (!unknownLength) {
|
if (!unknownLength) {
|
||||||
bodyRemaining -= count;
|
bodyRemaining -= count;
|
||||||
if (eof && bodyRemaining != 0) {
|
if (eof && bodyRemaining != 0) {
|
||||||
|
closeStreamBodyQuietly();
|
||||||
throw new IOException("streaming response ended before its declared length");
|
throw new IOException("streaming response ended before its declared length");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -297,6 +303,38 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
|
|||||||
endStreamInBatch = end;
|
endStreamInBatch = end;
|
||||||
}
|
}
|
||||||
finished = 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) {
|
private void appendTrailers(int maxFrameSize) {
|
||||||
|
|||||||
@@ -310,6 +310,11 @@ public final class Http2Stream
|
|||||||
throw new IllegalStateException("failed to restore discarded flow-control bytes", failure);
|
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() {
|
public boolean cancelled() {
|
||||||
|
|||||||
@@ -13,6 +13,8 @@ import dev.relism.flash.http2.hpack.HpackDecoder;
|
|||||||
import dev.relism.flash.models.Response;
|
import dev.relism.flash.models.Response;
|
||||||
import dev.relism.fpr.core.ByteView;
|
import dev.relism.fpr.core.ByteView;
|
||||||
import java.io.ByteArrayInputStream;
|
import java.io.ByteArrayInputStream;
|
||||||
|
import java.io.FilterInputStream;
|
||||||
|
import java.io.IOException;
|
||||||
import java.nio.charset.StandardCharsets;
|
import java.nio.charset.StandardCharsets;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
@@ -149,6 +151,78 @@ class Http2ResponseWriterTest {
|
|||||||
assertTrue(writer.trailerHeadersInBatch());
|
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) {
|
private static Parsed parse(Http2ResponseWriter writer) {
|
||||||
Parsed parsed = new Parsed();
|
Parsed parsed = new Parsed();
|
||||||
byte[] wire = writer.buffer();
|
byte[] wire = writer.buffer();
|
||||||
|
|||||||
Reference in New Issue
Block a user