feat(core): add HTTP/2 flow-controlled bodies

flash/docs/http2/DECISIONS.md

@@ -951,3 +951,33 @@ the lifecycle benchmark remains at the allocation noise floor.
 advertised limit or reader size from measurements; do not add a sleep-based dispatch delay.
 
 ---
+
+## DEC-28 — Align the receive window with a coalescing bounded DATA pool
+
+**Context.** The demultiplexer cannot block waiting for an application handler, but delaying
+WINDOW_UPDATE only provides backpressure after the peer has spent the window it already owns. A
+pool smaller than that outstanding credit can be exhausted legitimately. Allocating one buffer per
+DATA frame is also unsafe because many tiny or heavily padded frames can consume little payload
+storage while exhausting an object-per-frame pool.
+
+**Decision.** Advertise 1 MiB at both the connection and stream receive levels and back the
+connection with exactly 64 reusable 16 KiB buffers (also 1 MiB). Adjacent DATA payloads coalesce
+into available tail space; padding contributes to flow credit but not storage. WINDOW_UPDATE is
+sent at half-window consumption, never merely on receipt. Bodies at or below 64 KiB with a known
+length use one reusable contiguous stream buffer and dispatch at END_STREAM; all other bodies
+dispatch immediately onto the same protocol-neutral `RequestBody` over a blocking pooled source.
+
+Response DATA uses the same serialized writer with a progress cursor. A stream object itself is
+the executor task for initial handling and resumptions, so no per-resume closure is created.
+WINDOW_UPDATE only schedules work; it never reads an application `InputStream` on the demux thread.
+
+**Consequence.** Outstanding peer credit and worst-case pooled payload storage match exactly,
+small frames do not multiply objects, slow consumers withhold credit naturally, and streaming
+responses resume without recursive writer completion or demux blocking. The 100 MiB bidirectional
+integration test remains bounded, while JMH measures the streaming request and response paths at
+the allocation noise floor.
+
+**Revisit when.** Production memory/throughput measurements justify a different window. Change the
+window and pool byte capacity together; never raise credit independently of bounded storage.
+
+---

flash/docs/http2/FLOW-CONTROL.md

@@ -0,0 +1,48 @@
+# HTTP/2 bodies and flow control
+
+HTTP/2 applies flow control independently to the connection and to every stream. Flash advertises
+a 1 MiB receive window at both levels and sends WINDOW_UPDATE only after the application has
+consumed at least half a window. A DATA frame decrements both windows by its complete payload
+length, including the pad-length byte and padding; only its unpadded data reaches the handler.
+
+## Request bodies
+
+Known bodies up to 64 KiB remain in one reusable contiguous stream buffer. Their handler is
+dispatched at END_STREAM, and `RequestBody.bytes()` performs the only allocation: the byte array
+returned to application code. For a 1,024-byte body JMH reports exactly 1,040 B/op, the array plus
+its object header, with no framework allocation around it.
+
+Larger or unknown-length bodies dispatch after request headers. DATA is copied out of the frame
+reader into a connection-owned pool of 64 reusable 16 KiB buffers. Small adjacent frames coalesce
+inside a buffer, so the pool is bounded by bytes rather than frame count. The existing
+`RequestBody.stream()` blocks only the handler's virtual thread when data is absent. Buffers return
+to the pool as reads consume them, and that consumption reopens both receive windows. If a handler
+does not read its body, the normal post-handler drain performs the same bounded consumption.
+
+The connection window and pool both cover exactly 1 MiB, so the peer can never hold more credit
+than the server can store before backpressure takes effect. Per-stream accepted body bytes remain
+bounded by `MAX_REQUEST_BODY_SIZE`. Declared content length is parsed without a String and checked
+against the unpadded DATA total at END_STREAM.
+
+## Responses
+
+Fixed byte arrays, known-length streams and unknown-length streams all use one resumable
+`Http2ResponseWriter`. It emits DATA frames no larger than the peer's frame limit, the available
+connection window, the available stream window and the reusable 16 KiB relay buffer. A
+WINDOW_UPDATE schedules the stream on the shared virtual-thread executor; application streams are
+never read by the demultiplexer.
+
+`Response.chunked(InputStream)` means unknown-length streaming at the application API. HTTP/2 has
+no chunked transfer coding, so Flash emits ordinary DATA followed by END_STREAM and never sends a
+`transfer-encoding` field. `Response.stream(InputStream, length)` emits `content-length` and fails
+the stream if the source ends before that length.
+
+## Verification
+
+- A real Java HTTP/2 client uploads and downloads 100 MiB over TLS; both directions are validated
+  byte-for-byte without materializing the test payload.
+- A synthetic 100 MiB response proves serialized scratch storage stays below 64 KiB.
+- h2spec sections 5, 6.1, 6.9 and 8: 50 passed, one h2spec-skipped case, zero failures.
+- Clean Maven build with JMH sources: 633 tests, no failures.
+- JMH request streaming: 159.408 ns/op, 0.001 B/op, no GC.
+- JMH response streaming frame: 219.090 ns/op, 0.002 B/op, no GC.

flash/docs/http2/IMPLEMENTATION-PLAN.md

@@ -71,8 +71,8 @@ Status values: `not started` / `in progress` / `blocked` / `done`.
 | 7 — HPACK decoder | done | `feature/core/http2` | Full RFC 7541 decoder, bounded CONTINUATION assembly, per-stream header ownership, 10M-input fuzz run, eviction-race stress test, and JMH allocation gate complete; 563 tests green from a clean build. |
 | 8 — Connection state machine | done | `feature/core/http2` | Preface, transactional SETTINGS, priority PING ACK, connection WINDOW_UPDATE, two-stage GOAWAY, per-socket transport/ALPN dispatch, HPACK block composition, clean curl handshake, h2spec 28/35 selected cases and 0.008 B/op JMH gate complete. Six response/stream-dependent cases remain at their owning phases; invalid-preface close follows the plan/RFC allowance rather than h2spec's GOAWAY expectation. |
 | 9 — HPACK encoder + h2 response path | done | `feature/core/http2` | Stateless static-table HPACK encoder; precompiled status/content-type/date fields; reusable response writer with header filtering, bounds, CONTINUATION splitting and fixed DATA happy path; HTTP/1/2 serializer parity test. EX-46 fixed the one-digit Date day-of-month bug. JMH: 174.309 ns/op, 0.001 B/op (noise floor), no GC. 603/603 tests green from a clean `-Pjmh` build. |
-| 10 — Stream state machine + dispatch | done | `feature/core/http2` | Explicit stream transition table, bounded primitive stream table and pool, pseudo-header/message validation, protocol-neutral `Request` assembly, virtual-thread dispatch and exception path, cancellation-safe release, raw h2c + Java HTTP/2 integration. h2spec sections 5/8: 37/39; the two content-length/DATA accounting cases are owned by Phase 11. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green from a clean `-Pjmh` build. |
+| 10 — Stream state machine + dispatch | done | `feature/core/http2` | Explicit stream transition table, bounded primitive stream table and pool, pseudo-header/message validation, protocol-neutral `Request` assembly, virtual-thread dispatch and exception path, cancellation-safe release, raw h2c + Java HTTP/2 integration. Phase 11 closed the two deferred content-length/DATA cases; h2spec sections 5/8 are now 39/39. JMH pooled lifecycle: 458.499 ns/op, 0.003 B/op, no GC. 618/618 tests green at phase closure. |
-| 11 — DATA, flow control, bodies | not started | — | — |
+| 11 — DATA, flow control, bodies | done | `feature/core/http2` | Two-level receive/send flow control, consumption-driven WINDOW_UPDATE hysteresis, bounded/coalescing DATA pool, inline and blocking streaming request bodies through the existing `RequestBody`, resumable fixed/known/unknown response streams, content-length and empty-DATA validation. Real TLS HTTP/2 transfer: 100 MiB upload + 100 MiB download verified byte-for-byte. h2spec combined sections 5, 6.1, 6.9 and 8: 50 passed, 1 tool-skipped, 0 failed. JMH: inline materialization exactly one 1,040-byte array; request streaming 0.001 B/op; response streaming 0.002 B/op; full pooled lifecycle 0.003 B/op. 633/633 tests green from a clean `-Pjmh` build. |
 | 12 — Trailers, half-close, gRPC | not started | — | — |
 | 13 — Security hardening & abuse resistance | not started | — | — |
 | 14 — h2c prior knowledge + proxy support | not started | — | — |
@@ -2558,8 +2558,7 @@ rules, the dispatch model, and the resource-release contract.
       existing `HttpServerTest` suite against an h2 client.
 - [x] `FastPathRouterImpl` unchanged.
 - [x] 0 B/op for the pooled protocol-side h2 GET lifecycle (0.003 B/op JMH noise floor).
-- [~] `h2spec` sections 5 and 8: 37/39 green. Both remaining cases validate DATA-byte totals
+- [x] `h2spec` sections 5 and 8: 39/39 green after DATA-byte accounting landed.
-      against `content-length`; Phase 11 owns that state and closes this combined gate.
 
 ---
 
@@ -2571,13 +2570,13 @@ backpressure.
 ### Files
 
 Created:
-- `h2/stream/Http2FlowController.java` — connection and stream windows, both directions.
+- `http2/stream/Http2FlowController.java` — connection and stream windows, both directions.
-- `h2/message/Http2RequestBody.java` — DATA frames → the `RequestBody` contract.
+- `http2/message/Http2RequestBody.java` — DATA frames → the `RequestBody` contract.
-- `h2/message/DataBufferPool.java` — the fixed-size buffer free list.
+- `http2/message/DataBufferPool.java` — the fixed-size buffer free list.
 
 Modified:
-- `h2/Http2Connection.java` — DATA dispatch.
+- `http2/Http2Connection.java` — DATA dispatch.
-- `h2/message/Http2ResponseWriter.java` — multi-frame and streaming bodies.
+- `http2/message/Http2ResponseWriter.java` — multi-frame and streaming bodies.
 - `models/RequestBody.java` — accept an h2 backing (the Phase 6 refactor made this possible).
 
 ### Tasks
@@ -2635,16 +2634,16 @@ Modified:
 - Streaming path: 0 B/op at steady state; all buffers come from `DataBufferPool`.
 
 ### Safety checks
-- [ ] Connection-level **and** stream-level WINDOW_UPDATE both sent
+- [x] Connection-level **and** stream-level WINDOW_UPDATE both sent
-- [ ] Window overflow (> 2^31-1) rejected
+- [x] Window overflow (> 2^31-1) rejected
-- [ ] Window underflow (peer exceeds its window) rejected with the correct scope
+- [x] Window underflow (peer exceeds its window) rejected with the correct scope
-- [ ] Padding counted toward flow control
+- [x] Padding counted toward flow control
-- [ ] Flow control accounted for RST streams until settled
+- [x] Flow control accounted for RST streams until settled
-- [ ] `content-length` verified against actual DATA
+- [x] `content-length` verified against actual DATA
-- [ ] Empty DATA frame flood bounded
+- [x] Empty DATA frame flood bounded
-- [ ] `DataBufferPool` bounded; exhaustion applies backpressure rather than allocating without
+- [x] `DataBufferPool` bounded; exhaustion applies backpressure rather than allocating without
       limit
-- [ ] Body size bounded by `Http2Limits.MAX_REQUEST_BODY_SIZE` when no handler consumes it
+- [x] Body size bounded by `Http2Limits.MAX_REQUEST_BODY_SIZE` when no handler consumes it
 
 ### Tests
 - `Http2FlowControlTest` — the classic scenarios: a 10 MB upload with a 64 KB window; a
@@ -2663,9 +2662,9 @@ Modified:
 and the dispatch-on-END_STREAM optimization with its rationale.
 
 ### DoD
-- [ ] 100 MB upload and 100 MB download both correct, both bounded memory.
+- [x] 100 MB upload and 100 MB download both correct, both bounded memory.
-- [ ] `h2spec` DATA and WINDOW_UPDATE sections green.
+- [x] `h2spec` DATA and WINDOW_UPDATE sections green (13 passed, one tool-skipped, zero failed).
-- [ ] Small-body path allocates exactly one `byte[]` (the user's body).
+- [x] Small-body path allocates exactly one `byte[]` (1,040 B/op for a 1,024-byte body).
 
 ---
 

flash/docs/http2/STREAMS.md

@@ -40,10 +40,9 @@ while dispatch is pending.
 
 ## Verification
 
-- Clean Maven build with JMH sources: 618 tests, no failures.
+- Phase 11 clean Maven build with JMH sources: 633 tests, no failures.
-- h2spec sections 5 and 8: 37/39. The two remaining cases require request DATA byte accounting and
+- h2spec sections 5 and 8: 39/39 after request DATA byte accounting landed.
-  are completed with body flow control.
 - Java `HttpClient` negotiates HTTP/2 over TLS and runs an existing parameterized route unchanged.
 - curl prior-knowledge h2c receives a valid `200` response and body.
-- JMH pooled lifecycle (HPACK decode, request assembly, response write and release):
+- Current JMH pooled lifecycle (HPACK decode, request assembly, response write and release):
-  458.499 ns/op, 0.003 B/op, no GC.
+  483.571 ns/op, 0.003 B/op, no GC.

flash/src/jmh/java/dev/relism/flash/http2/message/Http2BodyBenchmark.java

@@ -0,0 +1,105 @@
+package dev.relism.flash.http2.message;
+
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.models.RequestBody;
+import dev.relism.flash.models.Response;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@Fork(2)
+@State(Scope.Thread)
+public class Http2BodyBenchmark {
+  private static final Http2RequestBody.ConsumptionListener NOOP = bytes -> {};
+
+  private final byte[] payload = new byte[1024];
+  private final byte[] target = new byte[1024];
+  private DataBufferPool pool;
+  private Http2RequestBody source;
+  private RequestBody body;
+  private Response response;
+  private Http2ResponseWriter responseWriter;
+  private ResettableInputStream responseSource;
+
+  @Setup(Level.Trial)
+  public void setup() throws IOException {
+    pool = new DataBufferPool(16_384, 1);
+    source = new Http2RequestBody(pool);
+    body = new RequestBody();
+    response = new Response(200, ContentType.BINARY);
+    responseWriter = new Http2ResponseWriter();
+    responseSource = new ResettableInputStream(payload);
+    source.begin(-1, false, NOOP);
+    source.offer(1, payload, 0, payload.length, payload.length);
+    source.finish(1);
+    source.read(target);
+  }
+
+  @Benchmark
+  public byte[] inlineBytes() {
+    source.begin(payload.length, true, NOOP);
+    source.offer(1, payload, 0, payload.length, payload.length);
+    source.finish(1);
+    body.reset(source, payload.length, null, 0, 0);
+    return body.bytes();
+  }
+
+  @Benchmark
+  public int streamingRead() throws IOException {
+    source.begin(-1, false, NOOP);
+    source.offer(1, payload, 0, payload.length, payload.length);
+    source.finish(1);
+    return source.read(target, 0, target.length);
+  }
+
+  @Benchmark
+  public int streamingResponseFrame() throws IOException {
+    responseSource.rewind();
+    response.reset(200, ContentType.BINARY).stream(responseSource, payload.length);
+    responseWriter.startFlowControlled(
+        response, 1, false, false, true, false, false, 16_384, 32_768, 16_384);
+    return responseWriter.length();
+  }
+
+  private static final class ResettableInputStream extends InputStream {
+    private final byte[] source;
+    private int position;
+
+    ResettableInputStream(byte[] source) {
+      this.source = source;
+    }
+
+    void rewind() {
+      position = 0;
+    }
+
+    @Override
+    public int read() {
+      return position == source.length ? -1 : source[position++] & 0xff;
+    }
+
+    @Override
+    public int read(byte[] target, int offset, int length) {
+      if (position == source.length) return -1;
+      int count = Math.min(length, source.length - position);
+      System.arraycopy(source, position, target, offset, count);
+      position += count;
+      return count;
+    }
+  }
+}

flash/src/main/java/dev/relism/flash/http2/Http2Connection.java

@@ -1,5 +1,6 @@
 package dev.relism.flash.http2;
 
+import dev.relism.flash.bytes.Pairs;
 import dev.relism.flash.http2.Http2ConnectionScratch.ControlIntent;
 import dev.relism.flash.http2.Http2ConnectionScratch.ControlKind;
 import dev.relism.flash.http2.frame.FrameFlags;
@@ -8,7 +9,10 @@ import dev.relism.flash.http2.frame.FrameType;
 import dev.relism.flash.http2.frame.FrameValidator;
 import dev.relism.flash.http2.frame.Http2FrameReader;
 import dev.relism.flash.http2.frame.Http2FrameWriter;
+import dev.relism.flash.http2.frame.Padding;
 import dev.relism.flash.http2.hpack.HeaderSink;
+import dev.relism.flash.http2.message.DataBufferPool;
+import dev.relism.flash.http2.stream.Http2FlowController;
 import dev.relism.flash.http2.stream.Http2Stream;
 import dev.relism.flash.http2.stream.Http2StreamState;
 import dev.relism.flash.http2.stream.Http2StreamTable;
@@ -37,10 +41,13 @@ public final class Http2Connection implements ConnectionProtocol {
   private final Http2Settings.StreamWindowUpdater streamWindows;
   private final long settingsAckTimeoutMs;
   private final Http2HeaderBlockDecoder headerBlocks = new Http2HeaderBlockDecoder();
-  private final Http2StreamTable streams = new Http2StreamTable(Http2Limits.MAX_CONCURRENT_STREAMS);
+  private final DataBufferPool dataBuffers =
+      new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, Http2Limits.DATA_BUFFER_POOL_SIZE);
+  private final Http2StreamTable streams =
+      new Http2StreamTable(Http2Limits.MAX_CONCURRENT_STREAMS, dataBuffers);
   private static final HeaderSink DISCARD_HEADERS = (name, value, never) -> {};
 
-  private long connectionSendWindow = 65_535;
+  private Http2FlowController flowController;
   private int outstandingLocalSettings;
   private long oldestSettingsSentNanos;
   private int lastProcessedStreamId;
@@ -68,7 +75,8 @@ public final class Http2Connection implements ConnectionProtocol {
     this.streamWindows =
         delta -> {
           try {
-            streams.adjustAllSendWindows(delta);
+            if (flowController == null) streams.adjustAllSendWindows(delta);
+            else flowController.applyInitialWindowDelta(streams, delta);
           } catch (IllegalStateException overflow) {
             throw Http2Exception.FLOW_CONTROL_ERROR;
           }
@@ -80,12 +88,16 @@ public final class Http2Connection implements ConnectionProtocol {
   @Override
   public void run(ConnectionContext ctx) throws IOException {
     Http2FrameWriter writer = new Http2FrameWriter(ctx.rawOut()::write);
+    flowController =
+        new Http2FlowController(
+            (streamId, increment) -> sendWindowUpdate(writer, streamId, increment));
     streamDispatcher =
         new Http2StreamDispatcher(
             ctx,
             writer,
             peerSettings,
             streams,
+            flowController,
             (streamId, error) -> sendRstStream(writer, streamId, error));
     try {
       run(ctx.in(), writer, ctx.stopped());
@@ -96,6 +108,11 @@ public final class Http2Connection implements ConnectionProtocol {
 
   void run(BufferedByteSource input, Http2FrameWriter writer, BooleanSupplier stopped)
       throws IOException {
+    if (flowController == null) {
+      flowController =
+          new Http2FlowController(
+              (streamId, increment) -> sendWindowUpdate(writer, streamId, increment));
+    }
     Http2FrameReader reader = new Http2FrameReader(input);
     runPrepared(input, reader, writer, stopped);
   }
@@ -206,6 +223,10 @@ public final class Http2Connection implements ConnectionProtocol {
 
     pendingHeaderStream = streams.acquire(streamId);
     refusingHeaderStream = pendingHeaderStream == null;
+    if (pendingHeaderStream != null) {
+      flowController.initializeStreamSendWindow(
+          pendingHeaderStream, peerSettings.initialWindowSize());
+    }
     HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock();
     if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId());
   }
@@ -224,6 +245,8 @@ public final class Http2Connection implements ConnectionProtocol {
     } else {
       Http2Stream stream = pendingHeaderStream;
       if (streamDispatcher != null) stream.validateHeaders();
+      boolean dispatch =
+          stream.prepareRequestBody(flowController, headerBlocks.endStream());
       stream.transition(
           headerBlocks.endStream()
               ? Http2StreamState.Event.RECV_HEADERS_ES
@@ -233,7 +256,7 @@ public final class Http2Connection implements ConnectionProtocol {
         streams.remove(streamId);
         streams.release(stream);
         if (!gracefulStarted) startGracefulShutdown(writer);
-      } else if (headerBlocks.endStream()) {
+      } else if (dispatch) {
         enqueueDispatch(stream);
       }
     }
@@ -250,17 +273,47 @@ public final class Http2Connection implements ConnectionProtocol {
   }
 
   private void receiveData(FrameHeader frame) {
-    Http2Stream stream = streamForFrame(frame.streamId());
+    flowController.receiveConnectionBytes(frame.length());
-    stream.transition(
+    Http2Stream stream = streams.get(frame.streamId());
-        FrameFlags.isEndStream(frame.flags())
+    if (stream == null) {
-            ? Http2StreamState.Event.RECV_DATA_ES
+      discardConnectionBytes(frame.length());
-            : Http2StreamState.Event.RECV_DATA);
+      if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
-    if (frame.length() != 0) {
       throw new Http2StreamException(
-          frame.streamId(), Http2ErrorCode.INTERNAL_ERROR, "request DATA support is not active");
+          frame.streamId(), Http2ErrorCode.STREAM_CLOSED, "stream is closed");
     }
-    if (FrameFlags.isEndStream(frame.flags()) && streamDispatcher != null) {
+    boolean bodyAccepted = false;
-      enqueueDispatch(stream);
+    try {
+      stream.transition(
+          FrameFlags.isEndStream(frame.flags())
+              ? Http2StreamState.Event.RECV_DATA_ES
+              : Http2StreamState.Event.RECV_DATA);
+      flowController.receiveStreamBytes(stream, frame.length());
+      long unpadded =
+          Padding.unpad(
+              frame.buffer(),
+              frame.payloadOffset(),
+              frame.length(),
+              FrameFlags.isPadded(frame.flags()));
+      int dataOffset = Pairs.hi(unpadded);
+      int dataLength = Pairs.lo(unpadded);
+      if (frame.length() == 0) {
+        if (stream.incrementEmptyDataFrames()
+            > Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM) {
+          throw new Http2StreamException(
+              frame.streamId(), Http2ErrorCode.ENHANCE_YOUR_CALM, "empty DATA frame limit exceeded");
+        }
+      } else {
+        stream.resetEmptyDataFrames();
+      }
+      stream.receiveData(frame.buffer(), dataOffset, dataLength, frame.length());
+      bodyAccepted = true;
+      if (FrameFlags.isEndStream(frame.flags())) {
+        stream.finishRequestBody();
+        if (streamDispatcher != null && !stream.dispatched()) enqueueDispatch(stream);
+      }
+    } catch (RuntimeException failure) {
+      if (!bodyAccepted) discardConnectionBytes(frame.length());
+      throw failure;
     }
   }
 
@@ -276,6 +329,7 @@ public final class Http2Connection implements ConnectionProtocol {
     streams.remove(stream.id());
     if (releaseDeferred) {
       stream.cancel();
+      if (stream.responseStarted() && !stream.responseInFlight()) streams.release(stream);
     } else {
       streams.release(stream);
     }
@@ -286,6 +340,7 @@ public final class Http2Connection implements ConnectionProtocol {
       throw new Http2StreamException(
           stream.id(), Http2ErrorCode.REFUSED_STREAM, "dispatch queue is full");
     }
+    stream.markDispatched();
     dispatchQueue[dispatchCount++] = stream;
   }
 
@@ -299,13 +354,6 @@ public final class Http2Connection implements ConnectionProtocol {
     }
   }
 
-  private Http2Stream streamForFrame(int streamId) {
-    Http2Stream stream = streams.get(streamId);
-    if (stream != null) return stream;
-    if (streamId > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR;
-    throw new Http2StreamException(streamId, Http2ErrorCode.STREAM_CLOSED, "stream is closed");
-  }
-
   private void receiveSettings(FrameHeader frame, Http2FrameWriter writer) throws IOException {
     boolean ack = FrameFlags.isAck(frame.flags());
     if (ack) {
@@ -346,16 +394,16 @@ public final class Http2Connection implements ConnectionProtocol {
         return;
       }
       try {
-        stream.adjustSendWindow(increment);
+        flowController.increaseStreamSendWindow(stream, increment);
       } catch (IllegalStateException overflow) {
         throw new Http2StreamException(
             frame.streamId(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream window overflow");
       }
+      if (streamDispatcher != null) streamDispatcher.streamWindowUpdated(stream);
       return;
     }
-    long next = connectionSendWindow + increment;
+    flowController.increaseConnectionSendWindow(increment);
-    if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR;
+    if (streamDispatcher != null) streamDispatcher.connectionWindowUpdated();
-    connectionSendWindow = next;
   }
 
   private void receiveGoAway(FrameHeader frame) {
@@ -386,6 +434,21 @@ public final class Http2Connection implements ConnectionProtocol {
     writer.writePriority(rst);
   }
 
+  private void sendWindowUpdate(Http2FrameWriter writer, int streamId, int increment)
+      throws IOException {
+    ControlIntent update = scratch.acquire(ControlKind.SETTINGS_OR_OTHER);
+    update.windowUpdate(streamId, increment);
+    writer.writePriority(update);
+  }
+
+  private void discardConnectionBytes(int bytes) {
+    try {
+      flowController.discarded(bytes);
+    } catch (IOException failure) {
+      throw new IllegalStateException("failed to restore connection flow-control window", failure);
+    }
+  }
+
   private void closeStreamAfterError(int streamId) {
     Http2Stream stream = streams.remove(streamId);
     if (stream == null) return;
@@ -446,7 +509,7 @@ public final class Http2Connection implements ConnectionProtocol {
   }
 
   public long connectionSendWindow() {
-    return connectionSendWindow;
+    return flowController == null ? 65_535 : flowController.connectionSendWindow();
   }
 
   public int peerLastStreamId() {
@@ -459,7 +522,7 @@ public final class Http2Connection implements ConnectionProtocol {
 
   void reset() {
     peerSettings.reset();
-    connectionSendWindow = 65_535;
+    flowController = null;
     outstandingLocalSettings = 0;
     oldestSettingsSentNanos = 0;
     lastProcessedStreamId = 0;

flash/src/main/java/dev/relism/flash/http2/Http2ConnectionScratch.java

@@ -119,6 +119,17 @@ final class Http2ConnectionScratch {
       length = 17 + debugLength;
     }
 
+    void windowUpdate(int streamId, int increment) {
+      buffer[0] = 0;
+      buffer[1] = 0;
+      buffer[2] = 4;
+      buffer[3] = (byte) FrameType.WINDOW_UPDATE.code();
+      buffer[4] = 0;
+      writeUInt31(buffer, 5, streamId);
+      writeUInt31(buffer, 9, increment);
+      length = 13;
+    }
+
     private static void writeUInt31(byte[] target, int off, int value) {
       writeUInt32(target, off, value & 0x7FFF_FFFF);
     }

flash/src/main/java/dev/relism/flash/http2/Http2Limits.java

@@ -111,6 +111,15 @@ public final class Http2Limits {
    */
   public static final int MAX_EMPTY_DATA_FRAMES_PER_STREAM = 1_000;
 
+  /** Largest request body retained contiguously before dispatching its handler. */
+  public static final int INLINE_BODY_THRESHOLD = 64 * 1024;
+
+  /** Hard limit for request body bytes accepted on one stream. */
+  public static final int MAX_REQUEST_BODY_SIZE = 100 * 1024 * 1024;
+
+  /** Number of frame-sized buffers available to streaming request bodies on one connection. */
+  public static final int DATA_BUFFER_POOL_SIZE = 64;
+
   /**
    * The value of {@code SETTINGS_INITIAL_WINDOW_SIZE} Flash advertises for every new stream:
    * deliberately large (1 MiB, versus the RFC default of 65535) so that a normal-sized
@@ -125,7 +134,7 @@ public final class Http2Limits {
    * windows, and sizing for that worst case would commit 100 MiB of receive window to every
    * connection regardless of load.
    */
-  public static final int CONNECTION_WINDOW_SIZE_LOCAL = 16 * 1_048_576;
+  public static final int CONNECTION_WINDOW_SIZE_LOCAL = 1_048_576;
 
   /**
    * The HPACK dynamic table size Flash's decoder honours, in bytes of RFC 7541 §4.1 accounting. RFC

flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java

@@ -3,6 +3,7 @@ package dev.relism.flash.http2;
 import dev.relism.flash.http.HttpMethod;
 import dev.relism.flash.http2.frame.Http2FrameWriter;
 import dev.relism.flash.http2.message.Http2ResponseWriter;
+import dev.relism.flash.http2.stream.Http2FlowController;
 import dev.relism.flash.http2.stream.Http2Stream;
 import dev.relism.flash.http2.stream.Http2StreamState;
 import dev.relism.flash.http2.stream.Http2StreamTable;
@@ -16,7 +17,7 @@ import lombok.extern.slf4j.Slf4j;
 
 /** Dispatches completed request streams without blocking the connection demultiplexer. */
 @Slf4j
-final class Http2StreamDispatcher {
+final class Http2StreamDispatcher implements Http2Stream.ResponseSink {
   @FunctionalInterface
   interface FailureSink {
     void fail(int streamId, Http2ErrorCode errorCode) throws IOException;
@@ -26,7 +27,10 @@ final class Http2StreamDispatcher {
   private final Http2FrameWriter frameWriter;
   private final Http2Settings peerSettings;
   private final Http2StreamTable streams;
+  private final Http2FlowController flowController;
   private final FailureSink failures;
+  private final Http2Stream[] resumeScratch =
+      new Http2Stream[Http2Limits.MAX_CONCURRENT_STREAMS];
   private volatile boolean firstResponse = true;
 
   Http2StreamDispatcher(
@@ -34,28 +38,65 @@ final class Http2StreamDispatcher {
       Http2FrameWriter frameWriter,
       Http2Settings peerSettings,
       Http2StreamTable streams,
+      Http2FlowController flowController,
       FailureSink failures) {
     this.context = context;
     this.frameWriter = frameWriter;
     this.peerSettings = peerSettings;
     this.streams = streams;
+    this.flowController = flowController;
     this.failures = failures;
   }
 
+  void streamWindowUpdated(Http2Stream stream) {
+    scheduleResume(stream);
+  }
+
+  void connectionWindowUpdated() {
+    int count = streams.copyValues(resumeScratch);
+    for (int i = 0; i < count; i++) {
+      Http2Stream stream = resumeScratch[i];
+      resumeScratch[i] = null;
+      scheduleResume(stream);
+    }
+  }
+
+  private void scheduleResume(Http2Stream stream) {
+    if (!stream.responseStarted() || stream.cancelled()) return;
+    if (!stream.beginResponseBatch()) return;
+    stream.markResumeTask();
+    try {
+      context.executor().execute(stream);
+    } catch (RejectedExecutionException rejected) {
+      stream.endResponseBatch();
+      failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected);
+    }
+  }
+
   void dispatch(Http2Stream stream) {
     if (stream.cancelled()) {
       streams.release(stream);
       return;
     }
     stream.markDispatched();
+    stream.responseSink(this);
     try {
-      context.executor().execute(() -> handle(stream));
+      context.executor().execute(stream);
     } catch (RejectedExecutionException rejected) {
       failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected);
     }
   }
 
+  @Override
+  public void handleRequest(Http2Stream stream) {
+    handle(stream);
+  }
+
   private void handle(Http2Stream stream) {
+    if (stream.cancelled()) {
+      streams.release(stream);
+      return;
+    }
     try {
       Request request = stream.assembleRequest(context.remoteAddress(), context.sslSocket());
       Response pooled = stream.resetResponse();
@@ -74,6 +115,7 @@ final class Http2StreamDispatcher {
         else if (result != null) response.setBody(result);
       }
 
+      request.drain();
       Http2ResponseWriter responseWriter = stream.responseWriter();
       if (stream.cancelled()) {
         request.recycle();
@@ -81,44 +123,120 @@ final class Http2StreamDispatcher {
         streams.release(stream);
         return;
       }
-      boolean prepared;
+      boolean headRequest = request.method() == HttpMethod.HEAD;
+      int reserved;
+      int used;
       synchronized (this) {
         boolean tableUpdate = firstResponse;
-        prepared =
+        reserved = flowController.reserveSend(stream, peerSettings.maxFrameSize());
-            responseWriter.prepare(
+        used = 0;
-                response,
+        try {
-                stream.id(),
+          used =
-                request.method() == HttpMethod.HEAD,
+              responseWriter.startFlowControlled(
-                context.configuration().isSendDate(),
+                  response,
-                true,
+                  stream.id(),
-                context.configuration().isH2HuffmanDynamicValues(),
+                  headRequest,
-                tableUpdate,
+                  context.configuration().isSendDate(),
-                peerSettings.maxFrameSize(),
+                  true,
-                peerSettings.maxHeaderListSize(),
+                  context.configuration().isH2HuffmanDynamicValues(),
-                (int) Math.min(stream.sendWindow(), Integer.MAX_VALUE));
+                  tableUpdate,
-        if (prepared) {
+                  peerSettings.maxFrameSize(),
-          firstResponse = false;
+                  peerSettings.maxHeaderListSize(),
-          stream.transition(Http2StreamState.Event.SEND_HEADERS_ES);
+                  reserved);
-          request.recycle();
+        } finally {
-          if (response == pooled) pooled.recycle();
+          flowController.refundSend(stream, reserved - used);
-          streams.remove(stream.id());
-          frameWriter.write(responseWriter);
         }
+        firstResponse = false;
       }
-      if (!prepared) {
+      request.recycle();
-        request.recycle();
+      if (response == pooled) pooled.recycle();
-        if (response == pooled) pooled.recycle();
+      stream.markResponseStarted();
-        failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, null);
+      applyBatchTransition(stream, responseWriter);
+      if (!stream.beginResponseBatch()) {
+        throw new IllegalStateException("response batch already in flight");
       }
+      frameWriter.write(responseWriter);
     } catch (Exception failure) {
       failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure);
     }
   }
 
+  private void tryResumeResponse(Http2Stream stream) {
+    if (stream.cancelled()) {
+      stream.endResponseBatch();
+      streams.release(stream);
+      return;
+    }
+    Http2ResponseWriter responseWriter = stream.responseWriter();
+    if (responseWriter.finished()) {
+      stream.endResponseBatch();
+      streams.remove(stream.id());
+      streams.release(stream);
+      return;
+    }
+    int reserved = flowController.reserveSend(stream, peerSettings.maxFrameSize());
+    if (reserved == 0) {
+      stream.endResponseBatch();
+      return;
+    }
+    try {
+      int used = 0;
+      try {
+        used = responseWriter.resume(peerSettings.maxFrameSize(), reserved);
+      } finally {
+        flowController.refundSend(stream, reserved - used);
+      }
+      applyBatchTransition(stream, responseWriter);
+      frameWriter.write(responseWriter);
+    } catch (Exception failure) {
+      stream.endResponseBatch();
+      failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure);
+    }
+  }
+
+  @Override
+  public void resumeResponse(Http2Stream stream) {
+    tryResumeResponse(stream);
+  }
+
+  private static void applyBatchTransition(
+      Http2Stream stream, Http2ResponseWriter responseWriter) {
+    if (responseWriter.headersInBatch()) {
+      if (responseWriter.finished() && responseWriter.dataBytesInBatch() == 0) {
+        stream.transition(Http2StreamState.Event.SEND_HEADERS_ES);
+        return;
+      }
+      stream.transition(Http2StreamState.Event.SEND_HEADERS);
+    }
+    if (responseWriter.dataBytesInBatch() != 0 || responseWriter.endStreamInBatch()) {
+      stream.transition(
+          responseWriter.endStreamInBatch()
+              ? Http2StreamState.Event.SEND_DATA_ES
+              : Http2StreamState.Event.SEND_DATA);
+    }
+  }
+
+  @Override
+  public void responseBatchCompleted(Http2Stream stream) {
+    stream.endResponseBatch();
+    if (stream.id() == 0) return;
+    if (stream.cancelled() || stream.responseWriter().finished()) {
+      streams.remove(stream.id());
+      streams.release(stream);
+    } else {
+      scheduleResume(stream);
+    }
+  }
+
   private void failAndRelease(Http2Stream stream, Http2ErrorCode error, Exception cause) {
     if (stream.id() == 0) return;
     if (cause != null) log.error("HTTP/2 stream {} failed", stream.id(), cause);
     streams.remove(stream.id());
+    try {
+      stream.cancel();
+    } catch (RuntimeException cancellationFailure) {
+      log.debug("Failed to cancel HTTP/2 stream {} cleanly", stream.id(), cancellationFailure);
+    }
     try {
       failures.fail(stream.id(), error);
     } catch (IOException writeFailure) {

flash/src/main/java/dev/relism/flash/http2/message/DataBufferPool.java

@@ -0,0 +1,69 @@
+package dev.relism.flash.http2.message;
+
+/** Bounded connection-owned free list of frame-sized request-body buffers. */
+public final class DataBufferPool {
+  static final class DataBuffer {
+    final byte[] bytes;
+    DataBuffer next;
+    int position;
+    int length;
+    int flowControlledBytes;
+
+    DataBuffer(int size) {
+      bytes = new byte[size];
+    }
+
+    void reset() {
+      next = null;
+      position = 0;
+      length = 0;
+      flowControlledBytes = 0;
+    }
+  }
+
+  private final int bufferSize;
+  private final int maxBuffers;
+  private DataBuffer free;
+  private int created;
+  private int available;
+
+  public DataBufferPool(int bufferSize, int maxBuffers) {
+    if (bufferSize < 1 || maxBuffers < 1) {
+      throw new IllegalArgumentException("bufferSize and maxBuffers must be positive");
+    }
+    this.bufferSize = bufferSize;
+    this.maxBuffers = maxBuffers;
+  }
+
+  synchronized DataBuffer acquire() {
+    DataBuffer buffer = free;
+    if (buffer != null) {
+      free = buffer.next;
+      available--;
+      buffer.reset();
+      return buffer;
+    }
+    if (created == maxBuffers) return null;
+    created++;
+    return new DataBuffer(bufferSize);
+  }
+
+  synchronized void release(DataBuffer buffer) {
+    buffer.reset();
+    buffer.next = free;
+    free = buffer;
+    available++;
+  }
+
+  public synchronized int createdCount() {
+    return created;
+  }
+
+  public synchronized int availableCount() {
+    return available;
+  }
+
+  public int capacity() {
+    return maxBuffers;
+  }
+}

flash/src/main/java/dev/relism/flash/http2/message/Http2RequestBody.java

@@ -0,0 +1,232 @@
+package dev.relism.flash.http2.message;
+
+import dev.relism.flash.http2.Http2ErrorCode;
+import dev.relism.flash.http2.Http2Limits;
+import dev.relism.flash.http2.Http2StreamException;
+import dev.relism.flash.http2.message.DataBufferPool.DataBuffer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+/** Reusable request-body source fed by the connection demultiplexer. */
+public final class Http2RequestBody extends InputStream {
+  @FunctionalInterface
+  public interface ConsumptionListener {
+    void consumed(int flowControlledBytes) throws IOException;
+  }
+
+  private final DataBufferPool pool;
+  private final ReentrantLock lock = new ReentrantLock();
+  private final Condition dataAvailable = lock.newCondition();
+  private final byte[] oneByte = new byte[1];
+  private byte[] inline;
+  private DataBuffer head;
+  private DataBuffer tail;
+  private ConsumptionListener listener;
+  private long declaredLength;
+  private long received;
+  private int inlinePosition;
+  private int inlineFlowControlledBytes;
+  private boolean inlineMode;
+  private boolean finished;
+
+  public Http2RequestBody(DataBufferPool pool) {
+    this.pool = pool;
+  }
+
+  public void begin(long declaredLength, boolean inlineMode, ConsumptionListener listener) {
+    releaseQueued();
+    this.declaredLength = declaredLength;
+    this.inlineMode = inlineMode;
+    this.listener = listener;
+    received = 0;
+    inlinePosition = 0;
+    inlineFlowControlledBytes = 0;
+    finished = false;
+    if (inlineMode && inline == null) inline = new byte[Http2Limits.INLINE_BODY_THRESHOLD];
+  }
+
+  public void offer(
+      int streamId, byte[] source, int offset, int length, int flowControlledBytes) {
+    long next = received + length;
+    if (next > Http2Limits.MAX_REQUEST_BODY_SIZE) {
+      throw new Http2StreamException(
+          streamId, Http2ErrorCode.PROTOCOL_ERROR, "request body exceeds configured limit");
+    }
+    if (declaredLength >= 0 && next > declaredLength) {
+      throw new Http2StreamException(
+          streamId, Http2ErrorCode.PROTOCOL_ERROR, "request body exceeds content-length");
+    }
+    if (length == 0) {
+      notifyConsumed(flowControlledBytes);
+      return;
+    }
+    if (inlineMode) {
+      if (next > inline.length) {
+        throw new Http2StreamException(
+            streamId, Http2ErrorCode.PROTOCOL_ERROR, "inline request body exceeded its bound");
+      }
+      System.arraycopy(source, offset, inline, (int) received, length);
+      received = next;
+      inlineFlowControlledBytes += flowControlledBytes;
+      return;
+    }
+
+    lock.lock();
+    try {
+      int remaining = length;
+      int sourcePosition = offset;
+      while (remaining > 0) {
+        if (tail == null || tail.length == tail.bytes.length) {
+          DataBuffer buffer = pool.acquire();
+          if (buffer == null) {
+            throw new Http2StreamException(
+                streamId,
+                Http2ErrorCode.ENHANCE_YOUR_CALM,
+                "request body buffer pool exhausted");
+          }
+          if (tail == null) head = buffer;
+          else tail.next = buffer;
+          tail = buffer;
+        }
+        int copied = Math.min(remaining, tail.bytes.length - tail.length);
+        System.arraycopy(source, sourcePosition, tail.bytes, tail.length, copied);
+        tail.length += copied;
+        sourcePosition += copied;
+        remaining -= copied;
+      }
+      tail.flowControlledBytes += flowControlledBytes;
+      received = next;
+      dataAvailable.signal();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public void finish(int streamId) {
+    if (declaredLength >= 0 && received != declaredLength) {
+      throw new Http2StreamException(
+          streamId,
+          Http2ErrorCode.PROTOCOL_ERROR,
+          "content-length does not match received DATA bytes");
+    }
+    lock.lock();
+    try {
+      finished = true;
+      dataAvailable.signalAll();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  public int cancel() {
+    int discarded;
+    lock.lock();
+    try {
+      finished = true;
+      discarded = inlineFlowControlledBytes + releaseQueuedLocked();
+      inlineFlowControlledBytes = 0;
+      dataAvailable.signalAll();
+    } finally {
+      lock.unlock();
+    }
+    return discarded;
+  }
+
+  public long declaredLength() {
+    return declaredLength;
+  }
+
+  @Override
+  public int read() throws IOException {
+    int count = read(oneByte, 0, 1);
+    return count < 0 ? -1 : oneByte[0] & 0xff;
+  }
+
+  @Override
+  public int read(byte[] target, int offset, int length) throws IOException {
+    if (length == 0) return 0;
+    if (inlineMode) return readInline(target, offset, length);
+
+    DataBuffer consumed = null;
+    int copied;
+    int flowControlled = 0;
+    lock.lock();
+    try {
+      while (head == null && !finished) {
+        try {
+          dataAvailable.await();
+        } catch (InterruptedException interrupted) {
+          Thread.currentThread().interrupt();
+          throw new IOException("interrupted while waiting for request DATA", interrupted);
+        }
+      }
+      if (head == null) return -1;
+      DataBuffer buffer = head;
+      copied = Math.min(length, buffer.length - buffer.position);
+      System.arraycopy(buffer.bytes, buffer.position, target, offset, copied);
+      buffer.position += copied;
+      if (buffer.position == buffer.length) {
+        head = buffer.next;
+        if (head == null) tail = null;
+        flowControlled = buffer.flowControlledBytes;
+        consumed = buffer;
+      }
+    } finally {
+      lock.unlock();
+    }
+    if (consumed != null) {
+      pool.release(consumed);
+      notifyConsumed(flowControlled);
+    }
+    return copied;
+  }
+
+  private int readInline(byte[] target, int offset, int length) throws IOException {
+    if (!finished) {
+      throw new IOException("inline request body is not complete");
+    }
+    if (inlinePosition == received) return -1;
+    int copied = (int) Math.min(length, received - inlinePosition);
+    System.arraycopy(inline, inlinePosition, target, offset, copied);
+    inlinePosition += copied;
+    if (inlinePosition == received && inlineFlowControlledBytes != 0) {
+      int flowControlled = inlineFlowControlledBytes;
+      inlineFlowControlledBytes = 0;
+      notifyConsumed(flowControlled);
+    }
+    return copied;
+  }
+
+  private void notifyConsumed(int bytes) {
+    if (bytes == 0 || listener == null) return;
+    try {
+      listener.consumed(bytes);
+    } catch (IOException failure) {
+      cancel();
+      throw new IllegalStateException("failed to update request flow-control window", failure);
+    }
+  }
+
+  private void releaseQueued() {
+    lock.lock();
+    try {
+      releaseQueuedLocked();
+    } finally {
+      lock.unlock();
+    }
+  }
+
+  private int releaseQueuedLocked() {
+    int flowControlled = 0;
+    while (head != null) {
+      DataBuffer released = head;
+      head = released.next;
+      flowControlled += released.flowControlledBytes;
+      pool.release(released);
+    }
+    tail = null;
+    return flowControlled;
+  }
+}

flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java

@@ -5,6 +5,7 @@ import dev.relism.flash.http.ContentType;
 import dev.relism.flash.http.DateHeader;
 import dev.relism.flash.http.HttpStatus;
 import dev.relism.flash.http2.Http2ErrorCode;
+import dev.relism.flash.http2.Http2Limits;
 import dev.relism.flash.http2.Http2StreamException;
 import dev.relism.flash.http2.frame.FrameFlags;
 import dev.relism.flash.http2.frame.FrameType;
@@ -13,6 +14,8 @@ import dev.relism.flash.http2.frame.WriteIntent;
 import dev.relism.flash.http2.hpack.HpackEncoder;
 import dev.relism.flash.models.Response;
 import dev.relism.flash.models.ResponseSerializer;
+import java.io.IOException;
+import java.io.InputStream;
 
 /**
  * Reusable per-stream HTTP/2 response serializer. It prepares a complete small response outside the
@@ -30,13 +33,23 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
   private final ByteWriter headerBlock;
   private final ByteWriter output;
   private final FrameWriteBuffer frames;
-  private final byte[] decimalScratch = new byte[10];
+  private final byte[] decimalScratch = new byte[20];
+  private final byte[] relay = new byte[Http2Limits.MAX_FRAME_SIZE_LOCAL];
   private WriteIntent next;
   private boolean huffmanDynamicValues;
   private int streamId;
   private long headerListSize;
   private long maxHeaderListSize;
   private Completion completion;
+  private byte[] fixedBody;
+  private InputStream streamBody;
+  private long bodyRemaining;
+  private int fixedPosition;
+  private boolean unknownLength;
+  private boolean finished;
+  private boolean headersInBatch;
+  private boolean endStreamInBatch;
+  private int dataBytesInBatch;
 
   public Http2ResponseWriter() {
     this(1024, 2048);
@@ -117,6 +130,157 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
     return true;
   }
 
+  /** Starts a response whose DATA may span multiple flow-control windows. */
+  public int startFlowControlled(
+      Response response,
+      int streamId,
+      boolean headRequest,
+      boolean sendDate,
+      boolean sendContentLength,
+      boolean huffmanDynamicValues,
+      boolean emitTableSizeUpdate,
+      int maxFrameSize,
+      long maxHeaderListSize,
+      int availableFlowWindow)
+      throws IOException {
+    if (streamId <= 0) throw new IllegalArgumentException("streamId must be positive");
+    if (maxFrameSize <= 0 || availableFlowWindow < 0) {
+      throw new IllegalArgumentException("frame size must be positive and flow window non-negative");
+    }
+    headerBlock.reset();
+    output.reset();
+    this.streamId = streamId;
+    this.huffmanDynamicValues = huffmanDynamicValues;
+    this.maxHeaderListSize = maxHeaderListSize;
+    headerListSize = 0;
+    next = null;
+    headersInBatch = true;
+    endStreamInBatch = false;
+    dataBytesInBatch = 0;
+    fixedPosition = 0;
+    fixedBody = response.isStreaming() ? null : response.getBody();
+    streamBody = response.isStreaming() ? response.getStream() : null;
+    unknownLength = response.isStreaming() && response.isChunked();
+    if (response.isStreaming() && !unknownLength && response.getStreamLength() < 0) {
+      throw new IllegalArgumentException("known response stream length must not be negative");
+    }
+    bodyRemaining =
+        response.isStreaming()
+            ? (unknownLength ? -1 : response.getStreamLength())
+            : (fixedBody == null ? 0 : fixedBody.length);
+    long representationLength = bodyRemaining;
+    boolean representationUnknownLength = unknownLength;
+
+    int statusCode = response.getStatusCode();
+    boolean bodyForbidden =
+        statusCode == 204 || statusCode == 304 || (statusCode >= 100 && statusCode < 200);
+    if (headRequest || bodyForbidden) {
+      fixedBody = null;
+      streamBody = null;
+      unknownLength = false;
+      bodyRemaining = 0;
+    }
+
+    if (emitTableSizeUpdate) HpackEncoder.writeDynamicTableSizeUpdateZero(headerBlock);
+    writeStatus(statusCode);
+    writeContentType(response.getContentType());
+    if (sendDate) {
+      addHeaderListSize(4, 29);
+      headerBlock.writeBytes(DateHeader.hpackBytes());
+    }
+    if (sendContentLength && !bodyForbidden && !representationUnknownLength) {
+      addHeaderListSize(CONTENT_LENGTH_NAME_LENGTH, decimalLength(representationLength));
+      writeDecimalLiteral(28, representationLength);
+    }
+    ResponseSerializer.forEachCustomField(response, this);
+
+    boolean hasBody = unknownLength || bodyRemaining > 0;
+    writeHeaderFrames(maxFrameSize, !hasBody);
+    finished = !hasBody;
+    if (hasBody && availableFlowWindow > 0) {
+      appendData(maxFrameSize, availableFlowWindow);
+    }
+    return dataBytesInBatch;
+  }
+
+  /** Serializes the next DATA batch after a WINDOW_UPDATE or previous write completion. */
+  public int resume(int maxFrameSize, int availableFlowWindow) throws IOException {
+    if (finished || availableFlowWindow <= 0) return 0;
+    output.reset();
+    next = null;
+    headersInBatch = false;
+    endStreamInBatch = false;
+    dataBytesInBatch = 0;
+    appendData(maxFrameSize, availableFlowWindow);
+    return dataBytesInBatch;
+  }
+
+  private void appendData(int maxFrameSize, int availableFlowWindow) throws IOException {
+    int target = Math.min(relay.length, Math.min(maxFrameSize, availableFlowWindow));
+    int count;
+    boolean end;
+    if (fixedBody != null) {
+      count = (int) Math.min(target, bodyRemaining);
+      frames.beginFrame(
+          FrameType.DATA, count == bodyRemaining ? FrameFlags.END_STREAM : 0, streamId);
+      output.writeBytes(fixedBody, fixedPosition, count);
+      frames.endFrame();
+      fixedPosition += count;
+      bodyRemaining -= count;
+      end = bodyRemaining == 0;
+    } else {
+      int limit = unknownLength ? target : (int) Math.min(target, bodyRemaining);
+      count = 0;
+      boolean eof = false;
+      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) {
+            eof = true;
+            break;
+          }
+          relay[count++] = (byte) one;
+        } else {
+          count += read;
+        }
+      }
+      if (!unknownLength) {
+        bodyRemaining -= count;
+        if (eof && bodyRemaining != 0) {
+          throw new IOException("streaming response ended before its declared length");
+        }
+      }
+      end = unknownLength ? eof : bodyRemaining == 0;
+      frames.beginFrame(FrameType.DATA, end ? FrameFlags.END_STREAM : 0, streamId);
+      output.writeBytes(relay, 0, count);
+      frames.endFrame();
+    }
+    dataBytesInBatch = count;
+    endStreamInBatch = end;
+    finished = end;
+  }
+
+  public boolean finished() {
+    return finished;
+  }
+
+  public boolean headersInBatch() {
+    return headersInBatch;
+  }
+
+  public boolean endStreamInBatch() {
+    return endStreamInBatch;
+  }
+
+  public int dataBytesInBatch() {
+    return dataBytesInBatch;
+  }
+
   @Override
   public void accept(
       byte[] name, int nameOff, int nameLen, byte[] value, int valueOff, int valueLen) {
@@ -151,10 +315,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
     }
   }
 
-  private void writeDecimalLiteral(int nameIndex, int value) {
+  private void writeDecimalLiteral(int nameIndex, long value) {
     int length = decimalLength(value);
     int offset = decimalScratch.length - length;
-    int current = value;
+    long current = value;
     for (int i = decimalScratch.length - 1; i >= offset; i--) {
       decimalScratch[i] = (byte) ('0' + current % 10);
       current /= 10;
@@ -209,17 +373,13 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize
     return true;
   }
 
-  private static int decimalLength(int value) {
+  private static int decimalLength(long value) {
-    if (value < 10) return 1;
+    int length = 1;
-    if (value < 100) return 2;
+    while (value >= 10) {
-    if (value < 1000) return 3;
+      value /= 10;
-    if (value < 10000) return 4;
+      length++;
-    if (value < 100000) return 5;
+    }
-    if (value < 1000000) return 6;
+    return length;
-    if (value < 10000000) return 7;
-    if (value < 100000000) return 8;
-    if (value < 1000000000) return 9;
-    return 10;
   }
 
   @Override

flash/src/main/java/dev/relism/flash/http2/stream/Http2FlowController.java

@@ -0,0 +1,108 @@
+package dev.relism.flash.http2.stream;
+
+import dev.relism.flash.http2.Http2ErrorCode;
+import dev.relism.flash.http2.Http2Exception;
+import dev.relism.flash.http2.Http2Limits;
+import dev.relism.flash.http2.Http2StreamException;
+import java.io.IOException;
+
+/** Connection-level half of HTTP/2's two-level flow-control accounting. */
+public final class Http2FlowController {
+  @FunctionalInterface
+  public interface WindowUpdateSink {
+    void update(int streamId, int increment) throws IOException;
+  }
+
+  private final WindowUpdateSink updates;
+  private int receiveWindow = Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL;
+  private int consumedSinceUpdate;
+  private long sendWindow = 65_535;
+
+  public Http2FlowController(WindowUpdateSink updates) {
+    this.updates = updates;
+  }
+
+  public synchronized void receiveConnectionBytes(int bytes) {
+    if (bytes < 0) throw new IllegalArgumentException("bytes must not be negative");
+    if (bytes > receiveWindow) throw Http2Exception.FLOW_CONTROL_ERROR;
+    receiveWindow -= bytes;
+  }
+
+  public void consumed(Http2Stream stream, int bytes) throws IOException {
+    int connectionIncrement = 0;
+    synchronized (this) {
+      consumedSinceUpdate += bytes;
+      if (consumedSinceUpdate >= Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL / 2) {
+        connectionIncrement = consumedSinceUpdate;
+        receiveWindow += connectionIncrement;
+        consumedSinceUpdate = 0;
+      }
+    }
+    int streamIncrement = stream.consumedReceiveBytes(bytes);
+    if (streamIncrement != 0) updates.update(stream.id(), streamIncrement);
+    if (connectionIncrement != 0) updates.update(0, connectionIncrement);
+  }
+
+  public void discarded(int bytes) throws IOException {
+    int increment = 0;
+    synchronized (this) {
+      consumedSinceUpdate += bytes;
+      if (consumedSinceUpdate >= Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL / 2) {
+        increment = consumedSinceUpdate;
+        receiveWindow += increment;
+        consumedSinceUpdate = 0;
+      }
+    }
+    if (increment != 0) updates.update(0, increment);
+  }
+
+  public synchronized int reserveSend(Http2Stream stream, int requested) {
+    int streamWindow = stream.sendWindow();
+    if (requested <= 0 || sendWindow <= 0 || streamWindow <= 0) return 0;
+    int granted =
+        (int)
+            Math.min(requested, Math.min(sendWindow, Math.min(streamWindow, Integer.MAX_VALUE)));
+    sendWindow -= granted;
+    stream.adjustSendWindow(-granted);
+    return granted;
+  }
+
+  public synchronized void refundSend(Http2Stream stream, int bytes) {
+    if (bytes == 0) return;
+    sendWindow += bytes;
+    stream.adjustSendWindow(bytes);
+  }
+
+  public synchronized void increaseConnectionSendWindow(int increment) {
+    long next = sendWindow + increment;
+    if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR;
+    sendWindow = next;
+  }
+
+  public synchronized void increaseStreamSendWindow(Http2Stream stream, int increment) {
+    stream.adjustSendWindow(increment);
+  }
+
+  public synchronized void initializeStreamSendWindow(Http2Stream stream, int initialWindow) {
+    stream.adjustSendWindow(initialWindow - 65_535);
+  }
+
+  public synchronized void applyInitialWindowDelta(Http2StreamTable streams, int delta) {
+    streams.adjustAllSendWindows(delta);
+  }
+
+  public void receiveStreamBytes(Http2Stream stream, int bytes) {
+    if (!stream.receiveBytes(bytes)) {
+      throw new Http2StreamException(
+          stream.id(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream receive window exceeded");
+    }
+  }
+
+  public synchronized int connectionReceiveWindow() {
+    return receiveWindow;
+  }
+
+  public synchronized long connectionSendWindow() {
+    return sendWindow;
+  }
+}

flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java

@@ -4,9 +4,12 @@ import dev.relism.flash.bytes.PooledSlice;
 import dev.relism.flash.http.ContentType;
 import dev.relism.flash.http.HttpMethod;
 import dev.relism.flash.http2.Http2ErrorCode;
+import dev.relism.flash.http2.Http2Limits;
 import dev.relism.flash.http2.Http2StreamException;
 import dev.relism.flash.http2.hpack.HpackHeaderBlock;
+import dev.relism.flash.http2.message.DataBufferPool;
 import dev.relism.flash.http2.message.Http2HeaderMap;
+import dev.relism.flash.http2.message.Http2RequestBody;
 import dev.relism.flash.http2.message.Http2ResponseWriter;
 import dev.relism.flash.http2.message.PseudoHeaders;
 import dev.relism.flash.models.Request;
@@ -14,11 +17,22 @@ import dev.relism.flash.models.RequestBody;
 import dev.relism.flash.models.RequestLine;
 import dev.relism.flash.models.Response;
 import dev.relism.flash.routing.AbstractRouter;
+import dev.relism.fpr.core.ByteView;
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import javax.net.ssl.SSLSocket;
 
 /** Per-stream request, response, decoded-header and write state. */
-public final class Http2Stream implements Http2ResponseWriter.Completion {
+public final class Http2Stream
+    implements Http2ResponseWriter.Completion, Http2RequestBody.ConsumptionListener, Runnable {
+  public interface ResponseSink {
+    void handleRequest(Http2Stream stream);
+
+    void responseBatchCompleted(Http2Stream stream);
+
+    void resumeResponse(Http2Stream stream);
+  }
+
   private static final byte[] HTTP_2 = {'H', 'T', 'T', 'P', '/', '2'};
 
   private final HpackHeaderBlock headerBlock = new HpackHeaderBlock();
@@ -26,24 +40,37 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
   private final Http2HeaderMap headers = new Http2HeaderMap();
   private final RequestLine requestLine = new RequestLine();
   private final RequestBody requestBody = new RequestBody();
+  private final Http2RequestBody http2Body;
   private final Request request = new Request();
   private final Response response = new Response(200, ContentType.TEXT_PLAIN);
   private final Http2ResponseWriter responseWriter = new Http2ResponseWriter();
   private final PooledSlice path = new PooledSlice();
   private final PooledSlice query = new PooledSlice();
   private final PooledSlice protocol = new PooledSlice();
+  private final PooledSlice scanName = new PooledSlice();
+  private final PooledSlice scanValue = new PooledSlice();
 
   private int id;
   private Http2StreamState state = Http2StreamState.IDLE;
   private int sendWindow = 65_535;
+  private int receiveWindow = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL;
+  private int consumedReceiveBytes;
+  private int emptyDataFrames;
   private Http2StreamTable owner;
   private Object routeScratch;
   private volatile boolean dispatched;
   private volatile boolean cancelled;
   private boolean headersValidated;
+  private Http2FlowController flowController;
+  private ResponseSink responseSink;
+  private volatile boolean responseInFlight;
+  private volatile boolean responseStarted;
+  private boolean releaseClaimed;
+  private volatile boolean resumeTask;
   Http2Stream poolNext;
 
-  Http2Stream() {
+  Http2Stream(DataBufferPool dataBuffers) {
+    http2Body = new Http2RequestBody(dataBuffers);
     responseWriter.completion(this);
     protocol.reset(HTTP_2, 0, HTTP_2.length);
   }
@@ -53,9 +80,17 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
     this.owner = owner;
     state = Http2StreamState.IDLE;
     sendWindow = 65_535;
+    receiveWindow = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL;
+    consumedReceiveBytes = 0;
+    emptyDataFrames = 0;
     dispatched = false;
     cancelled = false;
     headersValidated = false;
+    responseInFlight = false;
+    responseStarted = false;
+    releaseClaimed = false;
+    resumeTask = false;
+    responseSink = null;
     headerBlock.reset();
   }
 
@@ -95,7 +130,7 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
           id, Http2ErrorCode.PROTOCOL_ERROR, "unsupported request method");
     }
     requestLine.reset(method, path, question < 0 ? null : query, protocol, headers);
-    requestBody.reset(null, 0, null, 0, 0);
+    requestBody.reset(http2Body, http2Body.declaredLength(), null, 0, 0);
     return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket);
   }
 
@@ -105,6 +140,63 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
     headersValidated = true;
   }
 
+  public boolean prepareRequestBody(Http2FlowController flowController, boolean endStream) {
+    this.flowController = flowController;
+    long contentLength = parseContentLength();
+    if (contentLength < 0 && endStream) contentLength = 0;
+    boolean inline = contentLength >= 0 && contentLength <= Http2Limits.INLINE_BODY_THRESHOLD;
+    http2Body.begin(contentLength, inline, this);
+    if (endStream) http2Body.finish(id);
+    return endStream || !inline;
+  }
+
+  public void receiveData(byte[] source, int offset, int length, int flowControlledBytes) {
+    http2Body.offer(id, source, offset, length, flowControlledBytes);
+  }
+
+  public void finishRequestBody() {
+    http2Body.finish(id);
+  }
+
+  private long parseContentLength() {
+    long parsed = -1;
+    for (int i = 0; i < headerBlock.count(); i++) {
+      headerBlock.get(i, scanName, scanValue);
+      if (!equals(scanName, "content-length")) continue;
+      long value = parseDecimal(scanValue);
+      if (parsed >= 0 && parsed != value) {
+        throw new Http2StreamException(
+            id, Http2ErrorCode.PROTOCOL_ERROR, "conflicting content-length fields");
+      }
+      parsed = value;
+    }
+    return parsed;
+  }
+
+  private long parseDecimal(ByteView value) {
+    if (value.length() == 0) {
+      throw new Http2StreamException(id, Http2ErrorCode.PROTOCOL_ERROR, "empty content-length");
+    }
+    long parsed = 0;
+    for (int i = 0; i < value.length(); i++) {
+      int digit = (value.byteAt(i) & 0xff) - '0';
+      if (digit < 0 || digit > 9 || parsed > (Http2Limits.MAX_REQUEST_BODY_SIZE - digit) / 10L) {
+        throw new Http2StreamException(
+            id, Http2ErrorCode.PROTOCOL_ERROR, "invalid or oversized content-length");
+      }
+      parsed = parsed * 10 + digit;
+    }
+    return parsed;
+  }
+
+  private static boolean equals(ByteView value, String expected) {
+    if (value.length() != expected.length()) return false;
+    for (int i = 0; i < value.length(); i++) {
+      if ((value.byteAt(i) & 0xff) != expected.charAt(i)) return false;
+    }
+    return true;
+  }
+
   public Response resetResponse() {
     return response.reset(200, ContentType.TEXT_PLAIN);
   }
@@ -129,6 +221,42 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
     return responseWriter;
   }
 
+  public void responseSink(ResponseSink responseSink) {
+    this.responseSink = responseSink;
+  }
+
+  public void markResponseStarted() {
+    responseStarted = true;
+  }
+
+  public boolean responseStarted() {
+    return responseStarted;
+  }
+
+  public void markResumeTask() {
+    resumeTask = true;
+  }
+
+  public synchronized boolean beginResponseBatch() {
+    if (responseInFlight) return false;
+    responseInFlight = true;
+    return true;
+  }
+
+  public synchronized void endResponseBatch() {
+    responseInFlight = false;
+  }
+
+  public synchronized boolean responseInFlight() {
+    return responseInFlight;
+  }
+
+  synchronized boolean claimRelease() {
+    if (releaseClaimed) return false;
+    releaseClaimed = true;
+    return true;
+  }
+
   public Object routeScratch(AbstractRouter router) {
     if (routeScratch == null) routeScratch = router.newScratch();
     return routeScratch;
@@ -144,25 +272,71 @@ public final class Http2Stream implements Http2ResponseWriter.Completion {
 
   public void cancel() {
     cancelled = true;
+    int discarded = http2Body.cancel();
+    if (discarded != 0 && flowController != null) {
+      try {
+        flowController.discarded(discarded);
+      } catch (IOException failure) {
+        throw new IllegalStateException("failed to restore discarded flow-control bytes", failure);
+      }
+    }
   }
 
   public boolean cancelled() {
     return cancelled;
   }
 
-  public int sendWindow() {
+  public synchronized int sendWindow() {
     return sendWindow;
   }
 
-  public void adjustSendWindow(int delta) {
+  public synchronized void adjustSendWindow(int delta) {
     long adjusted = (long) sendWindow + delta;
     if (adjusted > Integer.MAX_VALUE) throw new IllegalStateException("stream window overflow");
     sendWindow = (int) adjusted;
   }
 
+  public synchronized boolean receiveBytes(int bytes) {
+    if (bytes > receiveWindow) return false;
+    receiveWindow -= bytes;
+    return true;
+  }
+
+  public synchronized int consumedReceiveBytes(int bytes) {
+    consumedReceiveBytes += bytes;
+    if (consumedReceiveBytes < Http2Limits.INITIAL_WINDOW_SIZE_LOCAL / 2) {
+      return 0;
+    }
+    int increment = consumedReceiveBytes;
+    receiveWindow += increment;
+    consumedReceiveBytes = 0;
+    return increment;
+  }
+
+  public int incrementEmptyDataFrames() {
+    return ++emptyDataFrames;
+  }
+
+  public void resetEmptyDataFrames() {
+    emptyDataFrames = 0;
+  }
+
+  @Override
+  public void consumed(int flowControlledBytes) throws IOException {
+    flowController.consumed(this, flowControlledBytes);
+  }
+
   @Override
   public void responseWriteCompleted() {
-    Http2StreamTable table = owner;
+    ResponseSink sink = responseSink;
-    if (table != null) table.release(this);
+    if (sink != null) sink.responseBatchCompleted(this);
+  }
+
+  @Override
+  public void run() {
+    ResponseSink sink = responseSink;
+    if (sink == null) return;
+    if (resumeTask) sink.resumeResponse(this);
+    else sink.handleRequest(this);
   }
 }

flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java

@@ -1,5 +1,7 @@
 package dev.relism.flash.http2.stream;
 
+import dev.relism.flash.http2.Http2Limits;
+import dev.relism.flash.http2.message.DataBufferPool;
 import java.util.Arrays;
 
 /** Fixed-capacity primitive stream-id table using linear-probed open addressing. */
@@ -17,8 +19,15 @@ public final class Http2StreamTable {
   private int size;
   private Http2Stream free;
   private int created;
+  private final DataBufferPool dataBuffers;
 
   public Http2StreamTable(int maxEntries) {
+    this(
+        maxEntries,
+        new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, Http2Limits.DATA_BUFFER_POOL_SIZE));
+  }
+
+  public Http2StreamTable(int maxEntries, DataBufferPool dataBuffers) {
     if (maxEntries < 1) throw new IllegalArgumentException("maxEntries must be positive");
     int capacity = 1;
     while (capacity < maxEntries * 2) capacity <<= 1;
@@ -26,6 +35,7 @@ public final class Http2StreamTable {
     values = new Http2Stream[capacity];
     mask = capacity - 1;
     this.maxEntries = maxEntries;
+    this.dataBuffers = dataBuffers;
   }
 
   public synchronized Http2Stream get(int streamId) {
@@ -51,7 +61,7 @@ public final class Http2StreamTable {
       stream.poolNext = null;
     } else {
       if (created == maxEntries) return null;
-      stream = new Http2Stream();
+      stream = new Http2Stream(dataBuffers);
       created++;
     }
     stream.reset(streamId, this);
@@ -60,6 +70,7 @@ public final class Http2StreamTable {
   }
 
   public synchronized void release(Http2Stream stream) {
+    if (!stream.claimRelease()) return;
     stream.clear();
     stream.poolNext = free;
     free = stream;
@@ -92,6 +103,14 @@ public final class Http2StreamTable {
     }
   }
 
+  public synchronized int copyValues(Http2Stream[] target) {
+    int count = 0;
+    for (int i = 0; i < keys.length && count < target.length; i++) {
+      if (keys[i] != 0) target[count++] = values[i];
+    }
+    return count;
+  }
+
   public synchronized void adjustAllSendWindows(int delta) {
     for (int i = 0; i < keys.length; i++) {
       if (keys[i] == 0) continue;

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

@@ -0,0 +1,37 @@
+package dev.relism.flash.http2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.relism.flash.http2.message.DataBufferPool;
+import dev.relism.flash.http2.message.Http2RequestBody;
+import dev.relism.flash.http2.stream.Http2FlowController;
+import dev.relism.flash.http2.stream.Http2Stream;
+import dev.relism.flash.http2.stream.Http2StreamTable;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+class Http2BackpressureTest {
+  @Test
+  void windowUpdatesAreWithheldUntilTheHandlerConsumesQueuedData() throws Exception {
+    AtomicInteger updates = new AtomicInteger();
+    Http2FlowController flow =
+        new Http2FlowController((streamId, increment) -> updates.addAndGet(increment));
+    Http2Stream stream = new Http2StreamTable(1).acquire(1);
+    DataBufferPool pool = new DataBufferPool(Http2Limits.MAX_FRAME_SIZE_LOCAL, 32);
+    Http2RequestBody body = new Http2RequestBody(pool);
+    body.begin(-1, false, bytes -> flow.consumed(stream, bytes));
+    byte[] frame = new byte[Http2Limits.MAX_FRAME_SIZE_LOCAL];
+
+    for (int i = 0; i < 32; i++) {
+      flow.receiveConnectionBytes(frame.length);
+      flow.receiveStreamBytes(stream, frame.length);
+      body.offer(1, frame, 0, frame.length, frame.length);
+    }
+    assertEquals(0, updates.get(), "receiving alone must not reopen either window");
+
+    body.finish(1);
+    assertEquals(32L * frame.length, body.readAllBytes().length);
+    assertEquals(2 * 32 * frame.length, updates.get());
+    assertEquals(32, pool.availableCount());
+  }
+}

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

@@ -11,6 +11,7 @@ import dev.relism.flash.http2.hpack.HpackDecoder;
 import dev.relism.flash.http2.hpack.HpackEncoder;
 import dev.relism.flash.tls.TestKeystores;
 import dev.relism.flash.tls.TlsConfig;
+import java.io.ByteArrayInputStream;
 import java.io.EOFException;
 import java.io.InputStream;
 import java.net.ServerSocket;
@@ -81,6 +82,115 @@ class Http2ConnectionIntegrationTest {
     assertEquals("42:localhost:" + port, response.body());
   }
 
+  @Test
+  void javaHttpClientUploadsAndDownloadsFlowControlledBodies(@TempDir Path directory)
+      throws Exception {
+    int port = freePort();
+    Path keystore =
+        TestKeystores.build(
+            directory,
+            "http2-bodies.p12",
+            "changeit",
+            TestKeystores.Entry.of("server", "localhost", "localhost"));
+    byte[] upload = new byte[2 * 1024 * 1024];
+    for (int i = 0; i < upload.length; i++) upload[i] = (byte) (i * 31);
+    byte[] download = new byte[2 * 1024 * 1024 + 17];
+    for (int i = 0; i < download.length; i++) download[i] = (byte) (i * 17);
+
+    app =
+        FlashApp.create(
+            FlashConfiguration.builder()
+                .port(port)
+                .host("127.0.0.1")
+                .tls(TlsConfig.keystore(keystore, "changeit"))
+                .http2Enabled(true)
+                .build());
+    app.post("/echo", (request, response) -> request.body().bytes());
+    app.get("/fixed", (request, response) -> response.body(download));
+    app.get(
+        "/stream",
+        (request, response) -> response.chunked(new ByteArrayInputStream(download)));
+    app.start();
+
+    HttpClient client =
+        HttpClient.newBuilder()
+            .sslContext(TestKeystores.trustAllClientContext())
+            .version(HttpClient.Version.HTTP_2)
+            .build();
+    HttpResponse<byte[]> echoed =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/echo"))
+                .POST(HttpRequest.BodyPublishers.ofByteArray(upload))
+                .build(),
+            HttpResponse.BodyHandlers.ofByteArray());
+    HttpResponse<byte[]> fixed =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/fixed")).GET().build(),
+            HttpResponse.BodyHandlers.ofByteArray());
+    HttpResponse<byte[]> streamed =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/stream")).GET().build(),
+            HttpResponse.BodyHandlers.ofByteArray());
+
+    assertArrayEquals(upload, echoed.body());
+    assertArrayEquals(download, fixed.body());
+    assertArrayEquals(download, streamed.body());
+    assertTrue(streamed.headers().firstValue("transfer-encoding").isEmpty());
+  }
+
+  @Test
+  void hundredMegabyteUploadAndDownloadRemainStreaming(@TempDir Path directory) throws Exception {
+    int port = freePort();
+    long length = 100L * 1024 * 1024;
+    Path keystore =
+        TestKeystores.build(
+            directory,
+            "http2-large-bodies.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.post(
+        "/upload",
+        (request, response) -> {
+          long count = verifyPattern(request.body().stream());
+          return Long.toString(count);
+        });
+    app.get(
+        "/download",
+        (request, response) -> response.stream(new PatternInputStream(length), length));
+    app.start();
+
+    HttpClient client =
+        HttpClient.newBuilder()
+            .sslContext(TestKeystores.trustAllClientContext())
+            .version(HttpClient.Version.HTTP_2)
+            .build();
+    HttpResponse<String> upload =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/upload"))
+                .POST(HttpRequest.BodyPublishers.ofInputStream(() -> new PatternInputStream(length)))
+                .build(),
+            HttpResponse.BodyHandlers.ofString());
+    HttpResponse<InputStream> download =
+        client.send(
+            HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/download"))
+                .GET()
+                .build(),
+            HttpResponse.BodyHandlers.ofInputStream());
+
+    assertEquals(Long.toString(length), upload.body());
+    try (InputStream body = download.body()) {
+      assertEquals(length, verifyPattern(body));
+    }
+  }
+
   @Test
   void bodylessGetRunsExistingRouteAndReturnsHeadersAndData() throws Exception {
     int port = freePort();
@@ -401,6 +511,39 @@ class Http2ConnectionIntegrationTest {
     throw new AssertionError();
   }
 
+  private static long verifyPattern(InputStream input) throws Exception {
+    byte[] buffer = new byte[64 * 1024];
+    long position = 0;
+    int count;
+    while ((count = input.read(buffer)) >= 0) {
+      for (int i = 0; i < count; i++) assertEquals((byte) (position++ * 31), buffer[i]);
+    }
+    return position;
+  }
+
+  private static final class PatternInputStream extends InputStream {
+    private final long length;
+    private long position;
+
+    PatternInputStream(long length) {
+      this.length = length;
+    }
+
+    @Override
+    public int read() {
+      if (position == length) return -1;
+      return (byte) (position++ * 31) & 0xff;
+    }
+
+    @Override
+    public int read(byte[] target, int offset, int requested) {
+      if (position == length) return -1;
+      int count = (int) Math.min(requested, length - position);
+      for (int i = 0; i < count; i++) target[offset + i] = (byte) (position++ * 31);
+      return count;
+    }
+  }
+
   private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception {
     byte[] header = input.readNBytes(9);
     if (header.length != 9) throw new EOFException("truncated frame header");

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

@@ -0,0 +1,54 @@
+package dev.relism.flash.http2.message;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import dev.relism.flash.http.ContentType;
+import dev.relism.flash.models.Response;
+import java.io.InputStream;
+import org.junit.jupiter.api.Test;
+
+class Http2LargeResponseTest {
+  @Test
+  void hundredMegabyteStreamUsesOneBoundedReusableFrameBuffer() throws Exception {
+    long length = 100L * 1024 * 1024;
+    Response response =
+        new Response(200, ContentType.BINARY).stream(new RepeatingInputStream(length), length);
+    Http2ResponseWriter writer = new Http2ResponseWriter();
+    long written =
+        writer.startFlowControlled(
+            response, 1, false, false, true, false, false, 16_384, 32_768, 16_384);
+    int largestBuffer = writer.buffer().length;
+    while (!writer.finished()) {
+      written += writer.resume(16_384, 16_384);
+      largestBuffer = Math.max(largestBuffer, writer.buffer().length);
+    }
+
+    assertEquals(length, written);
+    assertTrue(largestBuffer <= 65_536, "serialized storage must not scale with body length");
+  }
+
+  private static final class RepeatingInputStream extends InputStream {
+    private long remaining;
+
+    RepeatingInputStream(long remaining) {
+      this.remaining = remaining;
+    }
+
+    @Override
+    public int read() {
+      if (remaining == 0) return -1;
+      remaining--;
+      return 0x5a;
+    }
+
+    @Override
+    public int read(byte[] target, int offset, int length) {
+      if (remaining == 0) return -1;
+      int count = (int) Math.min(length, remaining);
+      java.util.Arrays.fill(target, offset, offset + count, (byte) 0x5a);
+      remaining -= count;
+      return count;
+    }
+  }
+}

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

@@ -0,0 +1,86 @@
+package dev.relism.flash.http2.message;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import dev.relism.flash.http2.Http2ErrorCode;
+import dev.relism.flash.http2.Http2Limits;
+import dev.relism.flash.http2.Http2StreamException;
+import dev.relism.flash.models.RequestBody;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+class Http2RequestBodyTest {
+  @Test
+  void inlineBodyFeedsTheProtocolNeutralRequestBodyWithOneMaterialization() {
+    AtomicInteger consumed = new AtomicInteger();
+    Http2RequestBody source = new Http2RequestBody(new DataBufferPool(16, 1));
+    source.begin(3, true, consumed::addAndGet);
+    source.offer(1, "abc".getBytes(StandardCharsets.US_ASCII), 0, 3, 5);
+    source.finish(1);
+    RequestBody body = new RequestBody();
+    body.reset(source, 3, null, 0, 0);
+
+    assertArrayEquals("abc".getBytes(StandardCharsets.US_ASCII), body.bytes());
+    assertEquals(5, consumed.get());
+    assertEquals(3, body.contentLength());
+  }
+
+  @Test
+  void streamingBodyReusesAndReturnsPooledBuffers() throws Exception {
+    DataBufferPool pool = new DataBufferPool(8, 2);
+    AtomicInteger consumed = new AtomicInteger();
+    Http2RequestBody source = new Http2RequestBody(pool);
+    source.begin(-1, false, consumed::addAndGet);
+    source.offer(1, new byte[] {1, 2, 3, 4, 5, 6, 7, 8}, 0, 8, 8);
+    source.offer(1, new byte[] {9, 10}, 0, 2, 2);
+    source.finish(1);
+
+    assertArrayEquals(new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, source.readAllBytes());
+    assertEquals(10, consumed.get());
+    assertEquals(2, pool.createdCount());
+    assertEquals(2, pool.availableCount());
+  }
+
+  @Test
+  void contentLengthMismatchIsAProtocolStreamError() {
+    Http2RequestBody source = new Http2RequestBody(new DataBufferPool(8, 1));
+    source.begin(4, true, bytes -> {});
+    source.offer(3, new byte[] {1, 2, 3}, 0, 3, 3);
+
+    Http2StreamException failure =
+        assertThrows(Http2StreamException.class, () -> source.finish(3));
+    assertEquals(Http2ErrorCode.PROTOCOL_ERROR, failure.errorCode());
+  }
+
+  @Test
+  void boundedPoolNeverAllocatesPastItsCapacity() {
+    DataBufferPool pool = new DataBufferPool(4, 1);
+    Http2RequestBody source = new Http2RequestBody(pool);
+    source.begin(-1, false, bytes -> {});
+    source.offer(1, new byte[] {1, 2, 3, 4}, 0, 4, 4);
+
+    Http2StreamException failure =
+        assertThrows(
+            Http2StreamException.class,
+            () -> source.offer(1, new byte[] {2}, 0, 1, 1));
+    assertEquals(Http2ErrorCode.ENHANCE_YOUR_CALM, failure.errorCode());
+    assertEquals(1, pool.createdCount());
+  }
+
+  @Test
+  void unknownLengthBodyCannotExceedTheConfiguredMaximum() {
+    Http2RequestBody source = new Http2RequestBody(new DataBufferPool(8, 1));
+    source.begin(-1, false, bytes -> {});
+
+    Http2StreamException failure =
+        assertThrows(
+            Http2StreamException.class,
+            () ->
+                source.offer(
+                    1, new byte[1], 0, Http2Limits.MAX_REQUEST_BODY_SIZE + 1, 1));
+    assertEquals(Http2ErrorCode.PROTOCOL_ERROR, failure.errorCode());
+  }
+}

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

@@ -12,6 +12,7 @@ import dev.relism.flash.http2.frame.FrameType;
 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.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -99,6 +100,38 @@ class Http2ResponseWriterTest {
         () -> writer.prepare(response, 5, false, false, true, false, false, 16_384, 41, 65_535));
   }
 
+  @Test
+  void flowControlledHeadPreservesKnownRepresentationLength() throws Exception {
+    Response response =
+        new Response(200, ContentType.BINARY)
+            .stream(new ByteArrayInputStream(new byte[] {1, 2, 3, 4}), 4);
+    Http2ResponseWriter writer = new Http2ResponseWriter();
+
+    writer.startFlowControlled(
+        response, 1, true, false, true, false, false, 16_384, 4096, 16_384);
+    Parsed parsed = parse(writer);
+
+    assertEquals(List.of(FrameType.HEADERS), parsed.types);
+    assertTrue(decode(parsed.headerBlock).contains("content-length=4"));
+  }
+
+  @Test
+  void unknownLengthStreamUsesNativeDataWithoutTransferEncoding() throws Exception {
+    Response response =
+        new Response(200, ContentType.BINARY)
+            .chunked(new ByteArrayInputStream(new byte[] {1, 2, 3, 4}));
+    Http2ResponseWriter writer = new Http2ResponseWriter();
+
+    writer.startFlowControlled(
+        response, 1, false, false, true, false, false, 16_384, 4096, 16_384);
+    Parsed parsed = parse(writer);
+    List<String> fields = decode(parsed.headerBlock);
+
+    assertFalse(fields.stream().anyMatch(field -> field.startsWith("content-length=")));
+    assertFalse(fields.stream().anyMatch(field -> field.startsWith("transfer-encoding=")));
+    assertEquals(List.of(FrameType.HEADERS, FrameType.DATA), parsed.types);
+  }
+
   private static Parsed parse(Http2ResponseWriter writer) {
     Parsed parsed = new Parsed();
     byte[] wire = writer.buffer();

flash/src/test/java/dev/relism/flash/http2/stream/Http2FlowControlTest.java

@@ -0,0 +1,81 @@
+package dev.relism.flash.http2.stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import dev.relism.flash.http2.Http2Exception;
+import dev.relism.flash.http2.Http2Limits;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+class Http2FlowControlTest {
+  @Test
+  void receiveWindowsReopenAtHalfWindowAtBothLevels() throws Exception {
+    AtomicInteger connectionUpdates = new AtomicInteger();
+    AtomicInteger streamUpdates = new AtomicInteger();
+    Http2FlowController controller =
+        new Http2FlowController(
+            (streamId, increment) -> {
+              if (streamId == 0) connectionUpdates.addAndGet(increment);
+              else streamUpdates.addAndGet(increment);
+            });
+    Http2Stream stream = new Http2StreamTable(1).acquire(1);
+    int half = Http2Limits.INITIAL_WINDOW_SIZE_LOCAL / 2;
+
+    controller.receiveConnectionBytes(half);
+    controller.receiveStreamBytes(stream, half);
+    controller.consumed(stream, half);
+
+    assertEquals(half, connectionUpdates.get());
+    assertEquals(half, streamUpdates.get());
+    assertEquals(Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL, controller.connectionReceiveWindow());
+  }
+
+  @Test
+  void connectionAndStreamUnderflowUseTheirCorrectErrorScope() {
+    Http2FlowController controller = new Http2FlowController((streamId, increment) -> {});
+    Http2Stream stream = new Http2StreamTable(1).acquire(1);
+
+    assertSame(
+        Http2Exception.FLOW_CONTROL_ERROR,
+        assertThrows(
+            Http2Exception.class,
+            () ->
+                controller.receiveConnectionBytes(
+                    Http2Limits.CONNECTION_WINDOW_SIZE_LOCAL + 1)));
+    assertEquals(
+        dev.relism.flash.http2.Http2ErrorCode.FLOW_CONTROL_ERROR,
+        assertThrows(
+                dev.relism.flash.http2.Http2StreamException.class,
+                () ->
+                    controller.receiveStreamBytes(
+                        stream, Http2Limits.INITIAL_WINDOW_SIZE_LOCAL + 1))
+            .errorCode());
+  }
+
+  @Test
+  void sendReservationHonoursBothWindowsAndRejectsOverflow() {
+    Http2FlowController controller = new Http2FlowController((streamId, increment) -> {});
+    Http2Stream stream = new Http2StreamTable(1).acquire(1);
+
+    assertEquals(65_535, controller.reserveSend(stream, 100_000));
+    assertEquals(0, controller.reserveSend(stream, 1));
+    controller.increaseConnectionSendWindow(Integer.MAX_VALUE);
+    assertSame(
+        Http2Exception.FLOW_CONTROL_ERROR,
+        assertThrows(Http2Exception.class, () -> controller.increaseConnectionSendWindow(1)));
+  }
+
+  @Test
+  void emptyDataFrameCounterCrossesTheConfiguredLimitDeterministically() {
+    Http2Stream stream = new Http2StreamTable(1).acquire(1);
+    for (int i = 1; i <= Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM; i++) {
+      assertEquals(i, stream.incrementEmptyDataFrames());
+    }
+    assertEquals(
+        Http2Limits.MAX_EMPTY_DATA_FRAMES_PER_STREAM + 1, stream.incrementEmptyDataFrames());
+    stream.resetEmptyDataFrames();
+    assertEquals(1, stream.incrementEmptyDataFrames());
+  }
+}