feat(core): add HTTP trailers and push streaming

flash/src/test/java/dev/relism/flash/ChunkedInputStreamTest.java

@@ -68,7 +68,7 @@ class ChunkedInputStreamTest {
     @Test
     void trailers_consumed() throws IOException {
         // trailing headers after 0-chunk must be consumed
-        assertEquals("hi", readAll(wrap("2\r\nhi\r\n0\r\nTrailer: value\r\n\r\n")));
+        assertEquals("hi", readAll(wrap("2\r\nhi\r\n0\r\nX-Trailer: value\r\n\r\n")));
     }
 
     // --- byte-by-byte read ---

flash/src/test/java/dev/relism/flash/Http1TrailersTest.java

@@ -0,0 +1,68 @@
+package dev.relism.flash;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.http.HttpMethod;
+import dev.relism.flash.http1.Http1ResponseWriter;
+import dev.relism.flash.models.Request;
+import dev.relism.flash.models.Response;
+import dev.relism.flash.transport.BufferedByteSource;
+import dev.relism.flash.transport.ConnectionScratch;
+import dev.relism.flash.transport.ScratchPool;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+class Http1TrailersTest {
+  @Test
+  void requestTrailersBecomeVisibleOnlyAfterBodyEof() throws Exception {
+    byte[] wire = ("POST / HTTP/1.1\r\nHost: localhost\r\nTransfer-Encoding: chunked\r\n\r\n"
+        + "3\r\nabc\r\n0\r\nGrpc-Status: 0\r\nX-Trace: done\r\n\r\n")
+        .getBytes(StandardCharsets.US_ASCII);
+    Request request = new RequestParser().parse(
+        new BufferedByteSource(new ByteArrayInputStream(wire), null));
+
+    assertThrows(IllegalStateException.class, request::trailers);
+    assertEquals("abc", new String(request.body().bytes(), StandardCharsets.US_ASCII));
+    assertEquals("0", request.trailers().first("grpc-status"));
+    assertEquals("done", request.trailers().first("x-trace"));
+  }
+
+  @Test
+  void responseTrailersUseChunkedRendering() throws Exception {
+    Response response = new Response(200, "hello", ContentType.TEXT_PLAIN)
+        .trailer("grpc-status", "0");
+    ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+    Http1ResponseWriter.writeResponse(
+        output, response, HttpMethod.GET, true, false, new ScratchPool().acquire());
+
+    String wire = output.toString(StandardCharsets.US_ASCII);
+    assertEquals(true, wire.contains("Transfer-Encoding: chunked\r\n"));
+    assertEquals(true, wire.endsWith("5\r\nhello\r\n0\r\ngrpc-status: 0\r\n\r\n"));
+  }
+
+  @Test
+  void pushStreamingAndTrailersShareTheSameHttp1Writer() throws Exception {
+    Response response = new Response(200, ContentType.BINARY).streaming(stream -> {
+      try {
+        stream.write("one".getBytes(StandardCharsets.US_ASCII), 0, 3);
+        stream.write("two".getBytes(StandardCharsets.US_ASCII), 0, 3);
+        stream.trailer("grpc-status", "0");
+      } catch (Exception failure) {
+        throw new RuntimeException(failure);
+      }
+    });
+    ByteArrayOutputStream output = new ByteArrayOutputStream();
+    Http1ResponseWriter.writeResponse(
+        output, response, HttpMethod.GET, true, false, new ScratchPool().acquire());
+
+    String wire = output.toString(StandardCharsets.US_ASCII);
+    assertEquals(true, wire.contains("one"));
+    assertEquals(true, wire.contains("two"));
+    assertEquals(true, wire.endsWith("0\r\ngrpc-status: 0\r\n\r\n"));
+  }
+}

flash/src/test/java/dev/relism/flash/http2/GrpcInteropTest.java

@@ -0,0 +1,114 @@
+package dev.relism.flash.http2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import dev.relism.flash.extension.FlashApp;
+import dev.relism.flash.extension.FlashConfiguration;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.io.TempDir;
+
+@Tag("interop")
+@EnabledIfSystemProperty(named = "grpcurl.executable", matches = ".+")
+class GrpcInteropTest {
+  private FlashApp app;
+
+  @AfterEach
+  void stop() {
+    if (app != null) app.stop().join();
+  }
+
+  @Test
+  void grpcurlCompletesUnaryStreamingAndErrorCalls(@TempDir Path directory) throws Exception {
+    int port = freePort();
+    app = FlashApp.create(FlashConfiguration.builder()
+        .host("127.0.0.1").port(port).http2Enabled(true).build());
+    app.post("/flash.test.Echo/Unary", (request, response) ->
+        response.type("application/grpc")
+            .body(request.body().bytes())
+            .trailer("grpc-status", "0"));
+    app.post("/flash.test.Echo/Stream", (request, response) -> {
+      byte[] message = request.body().bytes();
+      return response.type("application/grpc").streaming(stream -> {
+        try {
+          for (int i = 0; i < 3; i++) stream.write(message, 0, message.length);
+          stream.trailer("grpc-status", "0");
+        } catch (Exception failure) {
+          throw new RuntimeException(failure);
+        }
+      });
+    });
+    app.post("/flash.test.Echo/Fail", (request, response) ->
+        response.type("application/grpc")
+            .trailer("grpc-status", "3")
+            .trailer("grpc-message", "invalid request"));
+    app.start();
+
+    Path proto = directory.resolve("echo.proto");
+    Files.writeString(proto, """
+        syntax = "proto3";
+        package flash.test;
+        service Echo {
+          rpc Unary (Message) returns (Message);
+          rpc Stream (Message) returns (stream Message);
+          rpc Fail (Message) returns (Message);
+        }
+        message Message { string value = 1; }
+        """);
+
+    Result unary = call(directory, port, "Unary");
+    assertEquals(0, unary.exitCode);
+    assertTrue(unary.output.contains("hello"), unary.output);
+
+    Result streaming = call(directory, port, "Stream");
+    assertEquals(0, streaming.exitCode);
+    assertEquals(3, occurrences(streaming.output, "hello"), streaming.output);
+
+    Result error = call(directory, port, "Fail");
+    assertTrue(error.exitCode != 0);
+    assertTrue(error.output.contains("InvalidArgument"), error.output);
+    assertTrue(error.output.contains("invalid request"), error.output);
+  }
+
+  private static Result call(Path directory, int port, String method) throws Exception {
+    Process process = new ProcessBuilder(
+        System.getProperty("grpcurl.executable"),
+        "-plaintext",
+        "-import-path", directory.toString(),
+        "-proto", "echo.proto",
+        "-d", "{\"value\":\"hello\"}",
+        "127.0.0.1:" + port,
+        "flash.test.Echo/" + method)
+        .redirectErrorStream(true)
+        .start();
+    assertTrue(process.waitFor(10, TimeUnit.SECONDS), "grpcurl timed out");
+    return new Result(process.exitValue(),
+        new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8));
+  }
+
+  private static int occurrences(String text, String needle) {
+    int count = 0;
+    int position = 0;
+    while ((position = text.indexOf(needle, position)) >= 0) {
+      count++;
+      position += needle.length();
+    }
+    return count;
+  }
+
+  private static int freePort() throws Exception {
+    try (ServerSocket socket = new ServerSocket(0)) {
+      return socket.getLocalPort();
+    }
+  }
+
+  private record Result(int exitCode, String output) {}
+}

flash/src/test/java/dev/relism/flash/http2/Http2ConnectTest.java

@@ -0,0 +1,103 @@
+package dev.relism.flash.http2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.relism.flash.bytes.ByteWriter;
+import dev.relism.flash.extension.FlashApp;
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.http2.frame.FrameFlags;
+import dev.relism.flash.http2.frame.FrameType;
+import dev.relism.flash.http2.hpack.HpackEncoder;
+import java.io.EOFException;
+import java.io.InputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class Http2ConnectTest {
+  private FlashApp app;
+
+  @AfterEach
+  void stop() {
+    if (app != null) app.stop().join();
+  }
+
+  @Test
+  void connectTunnelCanExchangeDataBeforeEitherSideCloses() throws Exception {
+    int port = freePort();
+    app = FlashApp.create(FlashConfiguration.builder()
+        .host("127.0.0.1").port(port).http2Enabled(true).build());
+    app.connect("tunnel", (request, response) ->
+        response.type(ContentType.NONE).streaming(output -> {
+          byte[] bytes = new byte[16];
+          try {
+            int count;
+            InputStream input = request.body().stream();
+            while ((count = input.read(bytes)) >= 0) {
+              output.write(bytes, 0, count);
+              output.flush();
+            }
+          } catch (Exception failure) {
+            throw new RuntimeException(failure);
+          }
+        }));
+    app.start();
+
+    ByteWriter block = new ByteWriter(32);
+    HpackEncoder.writeLiteralWithNameIndex(block, 2, ascii("CONNECT"), false);
+    HpackEncoder.writeLiteralWithNameIndex(block, 1, ascii("tunnel"), false);
+
+    try (Socket socket = new Socket("127.0.0.1", port)) {
+      socket.setSoTimeout(5_000);
+      socket.getOutputStream().write(Http2TestFrames.concat(
+          Http2TestFrames.PREFACE, Http2TestFrames.settings(),
+          Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]),
+          Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1,
+              Arrays.copyOf(block.array(), block.length())),
+          Http2TestFrames.frame(FrameType.DATA, 0, 1, ascii("one"))));
+      socket.getOutputStream().flush();
+
+      assertEquals("one", new String(readData(socket.getInputStream()).payload(),
+          StandardCharsets.US_ASCII));
+
+      socket.getOutputStream().write(Http2TestFrames.frame(
+          FrameType.DATA, FrameFlags.END_STREAM, 1, ascii("two")));
+      socket.getOutputStream().flush();
+      assertEquals("two", new String(readData(socket.getInputStream()).payload(),
+          StandardCharsets.US_ASCII));
+    }
+  }
+
+  private static Http2TestFrames.WireFrame readData(InputStream input) throws Exception {
+    for (int i = 0; i < 12; i++) {
+      Http2TestFrames.WireFrame frame = readFrame(input);
+      if (frame.streamId() == 1 && frame.type() == FrameType.DATA.code()
+          && frame.payload().length != 0) return frame;
+    }
+    throw new AssertionError("missing tunnel DATA");
+  }
+
+  private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception {
+    byte[] header = input.readNBytes(9);
+    if (header.length != 9) throw new EOFException();
+    int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff);
+    byte[] payload = input.readNBytes(length);
+    return new Http2TestFrames.WireFrame(
+        header[3] & 0xff, header[4] & 0xff, Http2TestFrames.readInt(header, 5) & 0x7fff_ffff,
+        payload);
+  }
+
+  private static byte[] ascii(String text) {
+    return text.getBytes(StandardCharsets.US_ASCII);
+  }
+
+  private static int freePort() throws Exception {
+    try (ServerSocket socket = new ServerSocket(0)) {
+      return socket.getLocalPort();
+    }
+  }
+}

flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java

@@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.*;
 import dev.relism.flash.bytes.ByteWriter;
 import dev.relism.flash.extension.FlashApp;
 import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.http.ContentType;
 import dev.relism.flash.http2.frame.FrameFlags;
 import dev.relism.flash.http2.frame.FrameType;
 import dev.relism.flash.http2.hpack.HpackDecoder;
@@ -14,6 +15,7 @@ import dev.relism.flash.tls.TlsConfig;
 import java.io.ByteArrayInputStream;
 import java.io.EOFException;
 import java.io.InputStream;
+import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.URI;
@@ -138,6 +140,61 @@ class Http2ConnectionIntegrationTest {
     assertTrue(streamed.headers().firstValue("transfer-encoding").isEmpty());
   }
 
+  @Test
+  void pushStreamingAppliesBackpressureAcrossMultipleWindows(@TempDir Path directory)
+      throws Exception {
+    int port = freePort();
+    int length = 2 * 1024 * 1024 + 31;
+    Path keystore =
+        TestKeystores.build(
+            directory,
+            "http2-push-stream.p12",
+            "changeit",
+            TestKeystores.Entry.of("server", "localhost", "localhost"));
+    app =
+        FlashApp.create(
+            FlashConfiguration.builder()
+                .port(port)
+                .host("127.0.0.1")
+                .tls(TlsConfig.keystore(keystore, "changeit"))
+                .http2Enabled(true)
+                .build());
+    app.get(
+        "/push",
+        (request, response) ->
+            response.type(ContentType.BINARY).streaming(stream -> {
+              byte[] block = new byte[8192];
+              int written = 0;
+              try {
+                while (written < length) {
+                  int count = Math.min(block.length, length - written);
+                  for (int i = 0; i < count; i++) block[i] = (byte) ((written + i) * 31);
+                  stream.write(block, 0, count);
+                  written += count;
+                }
+                stream.trailer("grpc-status", "0");
+              } catch (IOException failure) {
+                throw new RuntimeException(failure);
+              }
+            }));
+    app.start();
+
+    HttpClient client =
+        HttpClient.newBuilder()
+            .sslContext(TestKeystores.trustAllClientContext())
+            .version(HttpClient.Version.HTTP_2)
+            .build();
+    HttpResponse<InputStream> response =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/push")).GET().build(),
+            HttpResponse.BodyHandlers.ofInputStream());
+
+    assertEquals(HttpClient.Version.HTTP_2, response.version());
+    try (InputStream body = response.body()) {
+      assertEquals(length, verifyPattern(body));
+    }
+  }
+
   @Test
   void hundredMegabyteUploadAndDownloadRemainStreaming(@TempDir Path directory) throws Exception {
     int port = freePort();

flash/src/test/java/dev/relism/flash/http2/Http2HalfCloseTest.java

@@ -0,0 +1,50 @@
+package dev.relism.flash.http2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.relism.flash.http2.stream.Http2StreamState;
+import org.junit.jupiter.api.Test;
+
+class Http2HalfCloseTest {
+  @Test
+  void remoteMayCloseBeforeLocalResponseCompletes() {
+    Http2StreamState state = Http2StreamState.IDLE;
+    state = state.transition(1, Http2StreamState.Event.RECV_HEADERS_ES);
+    assertEquals(Http2StreamState.HALF_CLOSED_REMOTE, state);
+    state = state.transition(1, Http2StreamState.Event.SEND_HEADERS);
+    state = state.transition(1, Http2StreamState.Event.SEND_DATA);
+    state = state.transition(1, Http2StreamState.Event.SEND_DATA_ES);
+    assertEquals(Http2StreamState.CLOSED, state);
+  }
+
+  @Test
+  void localMayCloseWhileRemoteBodyContinues() {
+    Http2StreamState state = Http2StreamState.IDLE;
+    state = state.transition(1, Http2StreamState.Event.RECV_HEADERS);
+    state = state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES);
+    assertEquals(Http2StreamState.HALF_CLOSED_LOCAL, state);
+    state = state.transition(1, Http2StreamState.Event.RECV_DATA);
+    state = state.transition(1, Http2StreamState.Event.RECV_DATA_ES);
+    assertEquals(Http2StreamState.CLOSED, state);
+  }
+
+  @Test
+  void bothSidesRemainOpenDuringBidirectionalData() {
+    Http2StreamState state = Http2StreamState.IDLE;
+    state = state.transition(1, Http2StreamState.Event.RECV_HEADERS);
+    state = state.transition(1, Http2StreamState.Event.SEND_HEADERS);
+    state = state.transition(1, Http2StreamState.Event.RECV_DATA);
+    state = state.transition(1, Http2StreamState.Event.SEND_DATA);
+    assertEquals(Http2StreamState.OPEN, state);
+  }
+
+  @Test
+  void trailingHeadersCanCloseRemoteAfterLocalHalfClose() {
+    Http2StreamState state = Http2StreamState.IDLE;
+    state = state.transition(1, Http2StreamState.Event.RECV_HEADERS);
+    state = state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES);
+    state = state.transition(1, Http2StreamState.Event.RECV_DATA);
+    state = state.transition(1, Http2StreamState.Event.RECV_HEADERS_ES);
+    assertEquals(Http2StreamState.CLOSED, state);
+  }
+}

flash/src/test/java/dev/relism/flash/http2/Http2TrailersTest.java

@@ -0,0 +1,150 @@
+package dev.relism.flash.http2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.relism.flash.bytes.ByteWriter;
+import dev.relism.flash.extension.FlashApp;
+import dev.relism.flash.extension.FlashConfiguration;
+import dev.relism.flash.http2.frame.FrameFlags;
+import dev.relism.flash.http2.frame.FrameType;
+import dev.relism.flash.http2.hpack.HpackEncoder;
+import java.io.EOFException;
+import java.io.InputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class Http2TrailersTest {
+  private FlashApp app;
+
+  @AfterEach
+  void stop() {
+    if (app != null) app.stop().join();
+  }
+
+  @Test
+  void requestTrailersReachHandlerAfterBodyEof() throws Exception {
+    int port = freePort();
+    app = FlashApp.create(FlashConfiguration.builder()
+        .host("127.0.0.1").port(port).http2Enabled(true).build());
+    app.post("/trailers", (request, response) -> {
+      assertEquals("abc", new String(request.body().bytes(), StandardCharsets.US_ASCII));
+      return request.trailers().first("grpc-status");
+    });
+    app.start();
+
+    byte[] initial = requestHeaders("/trailers");
+    ByteWriter trailer = new ByteWriter(32);
+    HpackEncoder.writeLiteral(
+        trailer, "grpc-status".getBytes(StandardCharsets.US_ASCII),
+        "7".getBytes(StandardCharsets.US_ASCII));
+
+    try (Socket socket = connect(port)) {
+      socket.getOutputStream().write(Http2TestFrames.concat(
+          Http2TestFrames.PREFACE,
+          Http2TestFrames.settings(),
+          Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]),
+          Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, initial),
+          Http2TestFrames.frame(FrameType.DATA, 0, 1, "abc".getBytes(StandardCharsets.US_ASCII)),
+          Http2TestFrames.frame(FrameType.HEADERS,
+              FrameFlags.END_HEADERS | FrameFlags.END_STREAM, 1,
+              Arrays.copyOf(trailer.array(), trailer.length()))));
+      socket.getOutputStream().flush();
+
+      Http2TestFrames.WireFrame data = frameOfType(socket.getInputStream(), 1, FrameType.DATA);
+      assertEquals("7", new String(data.payload(), StandardCharsets.US_ASCII));
+    }
+  }
+
+  @Test
+  void trailersWithoutEndStreamAreRejected() throws Exception {
+    int port = startBlockingRoute();
+    ByteWriter trailer = new ByteWriter(32);
+    HpackEncoder.writeLiteral(trailer, ascii("x-end"), ascii("no"));
+    try (Socket socket = connect(port)) {
+      socket.getOutputStream().write(Http2TestFrames.concat(
+          Http2TestFrames.PREFACE, Http2TestFrames.settings(),
+          Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]),
+          Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, requestHeaders("/trailers")),
+          Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1,
+              Arrays.copyOf(trailer.array(), trailer.length()))));
+      socket.getOutputStream().flush();
+      Http2TestFrames.WireFrame rst = frameOfType(socket.getInputStream(), 1, FrameType.RST_STREAM);
+      assertEquals(Http2ErrorCode.PROTOCOL_ERROR.code(), Http2TestFrames.readInt(rst.payload(), 0));
+    }
+  }
+
+  @Test
+  void pseudoHeaderInTrailersIsRejected() throws Exception {
+    int port = startBlockingRoute();
+    try (Socket socket = connect(port)) {
+      socket.getOutputStream().write(Http2TestFrames.concat(
+          Http2TestFrames.PREFACE, Http2TestFrames.settings(),
+          Http2TestFrames.frame(FrameType.SETTINGS, FrameFlags.ACK, 0, new byte[0]),
+          Http2TestFrames.frame(FrameType.HEADERS, FrameFlags.END_HEADERS, 1, requestHeaders("/trailers")),
+          Http2TestFrames.frame(FrameType.HEADERS,
+              FrameFlags.END_HEADERS | FrameFlags.END_STREAM, 1, new byte[] {(byte) 0x88})));
+      socket.getOutputStream().flush();
+      Http2TestFrames.WireFrame rst = frameOfType(socket.getInputStream(), 1, FrameType.RST_STREAM);
+      assertEquals(Http2ErrorCode.PROTOCOL_ERROR.code(), Http2TestFrames.readInt(rst.payload(), 0));
+    }
+  }
+
+  private int startBlockingRoute() throws Exception {
+    int port = freePort();
+    app = FlashApp.create(FlashConfiguration.builder()
+        .host("127.0.0.1").port(port).http2Enabled(true).build());
+    app.post("/trailers", (request, response) -> request.body().bytes());
+    app.start();
+    return port;
+  }
+
+  private static byte[] requestHeaders(String path) {
+    ByteWriter block = new ByteWriter(64);
+    HpackEncoder.writeIndexed(block, 3);
+    HpackEncoder.writeIndexed(block, 6);
+    HpackEncoder.writeLiteralWithNameIndex(block, 4, ascii(path), false);
+    HpackEncoder.writeLiteralWithNameIndex(block, 1, ascii("localhost"), false);
+    HpackEncoder.writeLiteralWithNameIndex(block, 59, ascii("trailers"), false);
+    return Arrays.copyOf(block.array(), block.length());
+  }
+
+  private static byte[] ascii(String value) {
+    return value.getBytes(StandardCharsets.US_ASCII);
+  }
+
+  private static Socket connect(int port) throws Exception {
+    Socket socket = new Socket("127.0.0.1", port);
+    socket.setSoTimeout(5_000);
+    return socket;
+  }
+
+  private static Http2TestFrames.WireFrame frameOfType(
+      InputStream input, int streamId, FrameType type) throws Exception {
+    for (int i = 0; i < 12; i++) {
+      Http2TestFrames.WireFrame frame = readFrame(input);
+      if (frame.streamId() == streamId && frame.type() == type.code()) return frame;
+    }
+    throw new AssertionError("missing " + type + " frame");
+  }
+
+  private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception {
+    byte[] header = input.readNBytes(9);
+    if (header.length != 9) throw new EOFException();
+    int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff);
+    byte[] payload = input.readNBytes(length);
+    if (payload.length != length) throw new EOFException();
+    return new Http2TestFrames.WireFrame(
+        header[3] & 0xff, header[4] & 0xff, Http2TestFrames.readInt(header, 5) & 0x7fff_ffff,
+        payload);
+  }
+
+  private static int freePort() throws Exception {
+    try (ServerSocket socket = new ServerSocket(0)) {
+      return socket.getLocalPort();
+    }
+  }
+}

flash/src/test/java/dev/relism/flash/http2/message/Http2ResponseWriterTest.java

@@ -132,6 +132,23 @@ class Http2ResponseWriterTest {
     assertEquals(List.of(FrameType.HEADERS, FrameType.DATA), parsed.types);
   }
 
+  @Test
+  void finalDataDoesNotEndStreamWhenTrailingHeadersFollow() throws Exception {
+    Response response = new Response(200, "ok", ContentType.TEXT_PLAIN)
+        .trailer("grpc-status", "0");
+    Http2ResponseWriter writer = new Http2ResponseWriter();
+
+    writer.startFlowControlled(
+        response, 1, false, false, true, false, false, 16_384, 4096, 16_384);
+    Parsed parsed = parse(writer);
+
+    assertEquals(List.of(FrameType.HEADERS, FrameType.DATA, FrameType.HEADERS), parsed.types);
+    assertEquals(0, parsed.flags.get(1) & FrameFlags.END_STREAM);
+    assertTrue((parsed.flags.get(2) & FrameFlags.END_STREAM) != 0);
+    assertTrue(decode(parsed.headerBlock).contains("grpc-status=0"));
+    assertTrue(writer.trailerHeadersInBatch());
+  }
+
   private static Parsed parse(Http2ResponseWriter writer) {
     Parsed parsed = new Parsed();
     byte[] wire = writer.buffer();

flash/src/test/java/dev/relism/flash/models/ResponseStreamTest.java

@@ -0,0 +1,50 @@
+package dev.relism.flash.models;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.relism.flash.http.ContentType;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class ResponseStreamTest {
+  @Test
+  void producerWritesBodyAndTrailersThroughBoundedBridge() throws Exception {
+    Response response = new Response(200, ContentType.BINARY);
+    response.streaming(stream -> {
+      try {
+        stream.write(new byte[] {1, 2, 3}, 0, 3);
+        stream.trailer("grpc-status", "0");
+      } catch (IOException failure) {
+        throw new RuntimeException(failure);
+      }
+    });
+
+    assertArrayEquals(new byte[] {1, 2, 3}, response.getStream().readAllBytes());
+    assertEquals(true, response.hasTrailers());
+  }
+
+  @Test
+  void writeAfterCloseFailsWithoutWritingMoreBytes() throws Exception {
+    AtomicReference<IOException> failure = new AtomicReference<>();
+    CountDownLatch attempted = new CountDownLatch(1);
+    Response response = new Response(200, ContentType.BINARY);
+    response.streaming(stream -> {
+      try {
+        stream.close();
+        stream.write(new byte[] {1}, 0, 1);
+      } catch (IOException expected) {
+        failure.set(expected);
+      } finally {
+        attempted.countDown();
+      }
+    });
+
+    assertArrayEquals(new byte[0], response.getStream().readAllBytes());
+    assertEquals(true, attempted.await(1, TimeUnit.SECONDS));
+    assertEquals("response stream is closed", failure.get().getMessage());
+  }
+}