From c96d51f7eaa7d3bd065491d5c5136150b7c04ca9 Mon Sep 17 00:00:00 2001 From: Zakaria El Orche Date: Thu, 13 Aug 2026 18:33:04 +0000 Subject: [PATCH] feat(core): add HTTP/2 stream dispatch --- flash/docs/http2/DECISIONS.md | 24 +++ flash/docs/http2/IMPLEMENTATION-PLAN.md | 52 +++-- flash/docs/http2/STREAMS.md | 49 +++++ .../http2/stream/Http2StreamBenchmark.java | 65 ++++++ .../relism/flash/http2/Http2Connection.java | 193 ++++++++++++++++-- .../flash/http2/Http2HeaderBlockDecoder.java | 23 ++- .../dev/relism/flash/http2/Http2Limits.java | 2 +- .../flash/http2/Http2StreamDispatcher.java | 130 ++++++++++++ .../flash/http2/frame/Http2FrameReader.java | 5 + .../dev/relism/flash/http2/hpack/Huffman.java | 3 +- .../flash/http2/message/Http2HeaderMap.java | 138 +++++++++++++ .../http2/message/Http2ResponseWriter.java | 15 ++ .../flash/http2/message/PseudoHeaders.java | 138 +++++++++++++ .../flash/http2/stream/Http2Stream.java | 168 +++++++++++++++ .../flash/http2/stream/Http2StreamState.java | 95 +++++++++ .../flash/http2/stream/Http2StreamTable.java | 145 +++++++++++++ .../flash/transport/ConnectionContext.java | 57 +++--- .../flash/transport/ConnectionRunner.java | 1 + .../http2/Http2ConnectionIntegrationTest.java | 181 ++++++++++++++++ .../message/PseudoHeaderValidationTest.java | 86 ++++++++ .../stream/Http2RequestAssemblyTest.java | 43 ++++ .../http2/stream/Http2StreamLeakTest.java | 20 ++ .../http2/stream/Http2StreamStateTest.java | 32 +++ .../http2/stream/Http2StreamTableTest.java | 26 +++ 24 files changed, 1621 insertions(+), 70 deletions(-) create mode 100644 flash/docs/http2/STREAMS.md create mode 100644 flash/src/jmh/java/dev/relism/flash/http2/stream/Http2StreamBenchmark.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java create mode 100644 flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/stream/Http2RequestAssemblyTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamLeakTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamStateTest.java create mode 100644 flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamTableTest.java diff --git a/flash/docs/http2/DECISIONS.md b/flash/docs/http2/DECISIONS.md index ae494f2..f965bcf 100644 --- a/flash/docs/http2/DECISIONS.md +++ b/flash/docs/http2/DECISIONS.md @@ -927,3 +927,27 @@ path remains allocation-free and avoids duplicating the response model. existing model internally without introducing a second public header abstraction. --- + +## DEC-27 — Drain already-buffered frames before dispatching completed streams + +**Context.** A client can write a burst of complete requests before the server schedules their +handlers. Dispatching after every individual HEADERS frame lets a very fast handler close and +release streams while the same inbound burst is still being decoded, making the advertised +concurrency limit dependent on virtual-thread scheduling. Waiting a fixed interval would make the +limit deterministic but would add latency to every ordinary request. + +**Decision.** Completed bodyless streams enter a fixed queue bounded by +`MAX_CONCURRENT_STREAMS`. The demultiplexer continues only while its own frame reader already has +bytes buffered; as soon as consuming the next frame would require network input, it drains the +queue to the shared virtual-thread executor. The configured concurrent-stream limit is 64 and the +primitive stream table has exactly the same bound. + +**Consequence.** One socket read's request burst is admitted and bounded as a unit, excess streams +receive `REFUSED_STREAM`, and a single request is dispatched immediately without a timer. The demux +still never executes application code or waits for a worker. h2spec's concurrency case passes and +the lifecycle benchmark remains at the allocation noise floor. + +**Revisit when.** If production traces show a materially different batching pattern, tune the +advertised limit or reader size from measurements; do not add a sleep-based dispatch delay. + +--- diff --git a/flash/docs/http2/IMPLEMENTATION-PLAN.md b/flash/docs/http2/IMPLEMENTATION-PLAN.md index 301d15d..e6dd387 100644 --- a/flash/docs/http2/IMPLEMENTATION-PLAN.md +++ b/flash/docs/http2/IMPLEMENTATION-PLAN.md @@ -71,7 +71,7 @@ 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 | not started | — | — | +| 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. | | 11 — DATA, flow control, bodies | not started | — | — | | 12 — Trailers, half-close, gRPC | not started | — | — | | 13 — Security hardening & abuse resistance | not started | — | — | @@ -766,6 +766,18 @@ two-digit calendar day and could not exercise the boundary. **Fix**: use an expl `EEE, dd MMM yyyy HH:mm:ss 'GMT'` formatter for both protocol renderings and add a deterministic regression test for the third day of a month. **Phase**: 9. +### EX-47 — A reset queued stream could be returned to the pool before dispatch observed it + +Found while closing the stream-dispatch cancellation paths. `receiveRstStream` transitioned a +queued stream to `CLOSED` before deciding whether its release had to be deferred. The subsequent +state check could therefore no longer see `HALF_CLOSED_REMOTE`, returned the object to the pool, +and left the same object referenced by the dispatch queue. A following request could acquire and +mutate it before the queue drained. **Fix**: capture the deferred-release condition before the +transition, mark queued/dispatched streams cancelled, and let the sole queue/worker owner perform +the final release. The regression test sends a complete request, immediately resets it, then sends +a second request and proves that only the second handler invocation and response occur. **Phase**: +10. + --- # PART III — The phases @@ -2445,16 +2457,15 @@ Flash never sends PUSH_PROMISE, so the two `reserved` states are unreachable for ### Files Created: -- `h2/stream/Http2Stream.java` — per-stream state. Also the intrusive MPSC node (Phase 3) and - the owner of the per-stream arena (Phase 7). -- `h2/stream/Http2StreamState.java` — the state machine as an explicit transition table, not a - pile of `if`s. Every transition cites its RFC clause. -- `h2/stream/Http2StreamTable.java` — `int → Http2Stream`, open-addressed with linear probing, +- `http2/stream/Http2Stream.java` — per-stream state and owner of the request/response resources. +- `http2/stream/Http2StreamState.java` — the state machine as an explicit transition table, not a + pile of `if`s. +- `http2/stream/Http2StreamTable.java` — `int → Http2Stream`, open-addressed with linear probing, power-of-two capacity, zero-alloc lookup/insert/remove, sized from `MAX_CONCURRENT_STREAMS`. -- `h2/message/Http2HeaderMap.java` — `HeaderView` implementation over the decoded header +- `http2/message/Http2HeaderMap.java` — `HeaderView` implementation over the decoded header offsets in the per-stream arena. Same indexed lookup as Phase 4's `Http1HeaderMap`. -- `h2/message/PseudoHeaders.java` — validation and extraction. -- `h2/Http2StreamDispatcher.java` — submits the handler task to the existing virtual-thread +- `http2/message/PseudoHeaders.java` — validation and extraction. +- `http2/Http2StreamDispatcher.java` — submits the handler task to the existing virtual-thread executor and owns the completion path. ### Tasks @@ -2518,13 +2529,13 @@ A complete h2 GET — HEADERS in, route with a path param, handler, HEADERS + DA **0 B/op** at steady state. ### Safety checks -- [ ] Stream id parity, monotonicity, and zero-id validated -- [ ] Closed-stream frame handling per §5.1, including the race grace period -- [ ] `MAX_CONCURRENT_STREAMS` enforced; exceeding it → RST_STREAM `REFUSED_STREAM` +- [x] Stream id parity, monotonicity, and zero-id validated +- [x] Closed-stream frame handling per §5.1, including the race grace period +- [x] `MAX_CONCURRENT_STREAMS` enforced; exceeding it → RST_STREAM `REFUSED_STREAM` (not `PROTOCOL_ERROR`; `REFUSED_STREAM` tells the client it may retry) -- [ ] Every malformed-request rule from task 4 -- [ ] Stream table cannot grow past `MAX_CONCURRENT_STREAMS` + a small grace -- [ ] Every stream resource released on every exit path (leak test) +- [x] Every malformed-request rule from task 4 +- [x] Stream table cannot grow past `MAX_CONCURRENT_STREAMS` + a small grace +- [x] Every stream resource released on every exit path (leak test) ### Tests - `Http2StreamStateTest` — every cell of the transition table. @@ -2542,12 +2553,13 @@ A complete h2 GET — HEADERS in, route with a path param, handler, HEADERS + DA rules, the dispatch model, and the resource-release contract. ### DoD -- [ ] `curl --http2 https://localhost:port/ping` returns `pong`. -- [ ] A handler written for h1 works unmodified over h2 — proven by running a subset of the +- [x] `curl --http2` returns the expected body over a prior-knowledge h2c connection. +- [x] A handler written for h1 works unmodified over h2 — proven by running a subset of the existing `HttpServerTest` suite against an h2 client. -- [ ] `FastPathRouterImpl` unchanged. -- [ ] 0 B/op for the h2 GET path. -- [ ] `h2spec` sections 5 and 8 green. +- [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 + against `content-length`; Phase 11 owns that state and closes this combined gate. --- diff --git a/flash/docs/http2/STREAMS.md b/flash/docs/http2/STREAMS.md new file mode 100644 index 0000000..fd42729 --- /dev/null +++ b/flash/docs/http2/STREAMS.md @@ -0,0 +1,49 @@ +# HTTP/2 streams and request dispatch + +Each connection owns a fixed-capacity `Http2StreamTable`. Client stream identifiers are validated +as odd, non-zero and strictly increasing before a stream object is acquired. The table uses +primitive open addressing and a bounded object free list; it never grows beyond the advertised 64 +concurrent streams. An excess request receives `REFUSED_STREAM`, allowing the peer to retry it. + +## State model + +`Http2StreamState` represents `IDLE`, `OPEN`, `HALF_CLOSED_REMOTE`, `HALF_CLOSED_LOCAL` and +`CLOSED`. A class-initialized table maps every receive/send event to either its next state or the +correct stream error. Frames racing with a recently closed stream follow RFC 9113 §5.1 rather than +being rejected uniformly. + +## Header and request model + +Decoded HPACK fields are copied into storage owned by the stream. Before dispatch, +`PseudoHeaders` enforces ordering, uniqueness, required request pseudo-fields, lowercase regular +names, connection-specific-field rejection, the `te: trailers` exception and host/authority +consistency. Pseudo-fields are not exposed as regular headers; `:authority` is also visible as +`host` so existing middleware sees the same authority through HTTP/1.1 and HTTP/2. + +The stream assembles the existing protocol-neutral `Request`, `RequestLine`, `RequestBody` and +`HeaderView` models. Path/query splitting, routing, middleware, not-found handling and exception +handling therefore use the same code as HTTP/1.1. `FastPathRouterImpl` is unchanged. + +## Dispatch and ownership + +The connection thread decodes and validates frames only. Completed bodyless streams are queued in +a fixed array while more frame bytes are already buffered, then submitted to the server's shared +virtual-thread executor before the demultiplexer waits for the network again. This preserves burst +admission semantics without adding a dispatch timer or blocking the connection thread. + +The stream owns its pooled request, response, body, decoded-header arena and response writer. +Normal response completion releases it through the serialized writer callback. RST_STREAM marks a +queued or running stream cancelled and defers release to that sole owner; setup, routing and handler +failures send an appropriate stream reset and release in the failure path. A 100,000-cycle test +proves stable pool counts, and an immediate request/reset/request regression test covers reuse +while dispatch is pending. + +## Verification + +- Clean Maven build with JMH sources: 618 tests, no failures. +- h2spec sections 5 and 8: 37/39. The two remaining cases require request DATA byte accounting and + 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): + 458.499 ns/op, 0.003 B/op, no GC. diff --git a/flash/src/jmh/java/dev/relism/flash/http2/stream/Http2StreamBenchmark.java b/flash/src/jmh/java/dev/relism/flash/http2/stream/Http2StreamBenchmark.java new file mode 100644 index 0000000..2d98636 --- /dev/null +++ b/flash/src/jmh/java/dev/relism/flash/http2/stream/Http2StreamBenchmark.java @@ -0,0 +1,65 @@ +package dev.relism.flash.http2.stream; + +import dev.relism.flash.bytes.ByteWriter; +import dev.relism.flash.http2.hpack.HpackDecoder; +import dev.relism.flash.http2.hpack.HpackEncoder; +import dev.relism.flash.models.Response; +import java.nio.charset.StandardCharsets; +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.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; + +/** Measures the pooled HPACK-decode, request-assembly and fixed-response stream lifecycle. */ +@State(Scope.Thread) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Fork(2) +@Warmup(iterations = 3, time = 1) +@Measurement(iterations = 5, time = 1) +public class Http2StreamBenchmark { + private static final byte[] BODY = "pong".getBytes(StandardCharsets.US_ASCII); + + private Http2StreamTable streams; + private HpackDecoder decoder; + private byte[] requestBlock; + private int requestLength; + + @Setup + public void setup() { + streams = new Http2StreamTable(1); + decoder = new HpackDecoder(); + ByteWriter block = new ByteWriter(64); + HpackEncoder.writeIndexed(block, 2); + HpackEncoder.writeIndexed(block, 7); + HpackEncoder.writeLiteralWithNameIndex( + block, 4, "/ping".getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeLiteralWithNameIndex( + block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), false); + requestBlock = block.array(); + requestLength = block.length(); + lifecycle(); + } + + @Benchmark + public int lifecycle() { + Http2Stream stream = streams.acquire(1); + decoder.decode(requestBlock, 0, requestLength, stream.headerBlock()); + stream.assembleRequest(null, null); + Response response = stream.resetResponse().body(BODY); + stream + .responseWriter() + .prepare(response, 1, false, false, true, false, false, 16_384, 32_768, 65_535); + int bytes = stream.responseWriter().length(); + streams.remove(1); + streams.release(stream); + return bytes; + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java index 20a9977..a33eeb6 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Connection.java @@ -8,6 +8,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.hpack.HeaderSink; +import dev.relism.flash.http2.stream.Http2Stream; +import dev.relism.flash.http2.stream.Http2StreamState; +import dev.relism.flash.http2.stream.Http2StreamTable; import dev.relism.flash.transport.BufferedByteSource; import dev.relism.flash.transport.ConnectionContext; import dev.relism.flash.transport.ConnectionProtocol; @@ -33,6 +37,8 @@ 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 static final HeaderSink DISCARD_HEADERS = (name, value, never) -> {}; private long connectionSendWindow = 65_535; private int outstandingLocalSettings; @@ -43,6 +49,12 @@ public final class Http2Connection implements ConnectionProtocol { private boolean peerGoAway; private boolean gracefulStarted; private boolean gracefulFinished; + private int highestClientStreamId; + private Http2Stream pendingHeaderStream; + private boolean refusingHeaderStream; + private Http2StreamDispatcher streamDispatcher; + private final Http2Stream[] dispatchQueue = new Http2Stream[Http2Limits.MAX_CONCURRENT_STREAMS]; + private int dispatchCount; public Http2Connection() { this(delta -> {}, Http2Limits.SETTINGS_ACK_TIMEOUT_MS); @@ -53,13 +65,28 @@ public final class Http2Connection implements ConnectionProtocol { } Http2Connection(Http2Settings.StreamWindowUpdater streamWindows, long settingsAckTimeoutMs) { - this.streamWindows = streamWindows; + this.streamWindows = + delta -> { + try { + streams.adjustAllSendWindows(delta); + } catch (IllegalStateException overflow) { + throw Http2Exception.FLOW_CONTROL_ERROR; + } + streamWindows.applyInitialWindowDelta(delta); + }; this.settingsAckTimeoutMs = settingsAckTimeoutMs; } @Override public void run(ConnectionContext ctx) throws IOException { Http2FrameWriter writer = new Http2FrameWriter(ctx.rawOut()::write); + streamDispatcher = + new Http2StreamDispatcher( + ctx, + writer, + peerSettings, + streams, + (streamId, error) -> sendRstStream(writer, streamId, error)); try { run(ctx.in(), writer, ctx.stopped()); } finally { @@ -119,10 +146,12 @@ public final class Http2Connection implements ConnectionProtocol { dispatch(frame, writer); } catch (Http2StreamException streamError) { sendRstStream(writer, streamError); + closeStreamAfterError(streamError.streamId()); } finally { reader.consumeFrame(); } writer.drain(); + if (dispatchCount > 0 && !reader.hasBufferedInput()) dispatchPendingStreams(); checkSettingsTimeout(); } } catch (Http2Exception connectionError) { @@ -160,17 +189,121 @@ public final class Http2Connection implements ConnectionProtocol { case PING -> receivePing(frame, writer); case WINDOW_UPDATE -> receiveWindowUpdate(frame); case GOAWAY -> receiveGoAway(frame); - case HEADERS, CONTINUATION -> { - if (headerBlocks.accept(frame)) { - lastProcessedStreamId = Math.max(lastProcessedStreamId, frame.streamId()); - if (!gracefulStarted) startGracefulShutdown(writer); - } - } - default -> { - // Stream semantics are introduced by the stream layer. Structurally-valid - // frames are consumed here so connection-level state remains synchronized. + case HEADERS -> receiveHeaders(frame, writer); + case CONTINUATION -> receiveContinuation(frame, writer); + case DATA -> receiveData(frame); + case RST_STREAM -> receiveRstStream(frame); + case PRIORITY -> receivePriority(frame); + default -> {} + } + } + + private void receiveHeaders(FrameHeader frame, Http2FrameWriter writer) throws IOException { + int streamId = frame.streamId(); + if ((streamId & 1) == 0) throw Http2Exception.PROTOCOL_ERROR; + if (streamId <= highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; + highestClientStreamId = streamId; + + pendingHeaderStream = streams.acquire(streamId); + refusingHeaderStream = pendingHeaderStream == null; + HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock(); + if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId()); + } + + private void receiveContinuation(FrameHeader frame, Http2FrameWriter writer) throws IOException { + if (pendingHeaderStream == null && !refusingHeaderStream) { + throw Http2Exception.PROTOCOL_ERROR; + } + HeaderSink sink = refusingHeaderStream ? DISCARD_HEADERS : pendingHeaderStream.headerBlock(); + if (headerBlocks.accept(frame, sink)) completeHeaders(writer, frame.streamId()); + } + + private void completeHeaders(Http2FrameWriter writer, int streamId) throws IOException { + if (refusingHeaderStream) { + sendRstStream(writer, streamId, Http2ErrorCode.REFUSED_STREAM); + } else { + Http2Stream stream = pendingHeaderStream; + if (streamDispatcher != null) stream.validateHeaders(); + stream.transition( + headerBlocks.endStream() + ? Http2StreamState.Event.RECV_HEADERS_ES + : Http2StreamState.Event.RECV_HEADERS); + lastProcessedStreamId = Math.max(lastProcessedStreamId, streamId); + if (streamDispatcher == null) { + streams.remove(streamId); + streams.release(stream); + if (!gracefulStarted) startGracefulShutdown(writer); + } else if (headerBlocks.endStream()) { + enqueueDispatch(stream); } } + pendingHeaderStream = null; + refusingHeaderStream = false; + } + + private void receivePriority(FrameHeader frame) { + int dependency = readUInt31(frame.buffer(), frame.payloadOffset()); + if (dependency == frame.streamId()) { + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "stream depends on itself"); + } + } + + private void receiveData(FrameHeader frame) { + Http2Stream stream = streamForFrame(frame.streamId()); + stream.transition( + FrameFlags.isEndStream(frame.flags()) + ? Http2StreamState.Event.RECV_DATA_ES + : Http2StreamState.Event.RECV_DATA); + if (frame.length() != 0) { + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.INTERNAL_ERROR, "request DATA support is not active"); + } + if (FrameFlags.isEndStream(frame.flags()) && streamDispatcher != null) { + enqueueDispatch(stream); + } + } + + private void receiveRstStream(FrameHeader frame) { + Http2Stream stream = streams.get(frame.streamId()); + if (stream == null) { + if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; + return; + } + boolean releaseDeferred = + stream.dispatched() || stream.state() == Http2StreamState.HALF_CLOSED_REMOTE; + stream.transition(Http2StreamState.Event.RECV_RST); + streams.remove(stream.id()); + if (releaseDeferred) { + stream.cancel(); + } else { + streams.release(stream); + } + } + + private void enqueueDispatch(Http2Stream stream) { + if (dispatchCount == dispatchQueue.length) { + throw new Http2StreamException( + stream.id(), Http2ErrorCode.REFUSED_STREAM, "dispatch queue is full"); + } + dispatchQueue[dispatchCount++] = stream; + } + + private void dispatchPendingStreams() { + int count = dispatchCount; + dispatchCount = 0; + for (int i = 0; i < count; i++) { + Http2Stream stream = dispatchQueue[i]; + dispatchQueue[i] = null; + streamDispatcher.dispatch(stream); + } + } + + 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 { @@ -201,8 +334,25 @@ public final class Http2Connection implements ConnectionProtocol { private void receiveWindowUpdate(FrameHeader frame) { int increment = readUInt31(frame.buffer(), frame.payloadOffset()); - if (increment == 0) throw Http2Exception.PROTOCOL_ERROR; - if (frame.streamId() != 0) return; + if (increment == 0) { + if (frame.streamId() == 0) throw Http2Exception.PROTOCOL_ERROR; + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "zero window increment"); + } + if (frame.streamId() != 0) { + Http2Stream stream = streams.get(frame.streamId()); + if (stream == null) { + if (frame.streamId() > highestClientStreamId) throw Http2Exception.PROTOCOL_ERROR; + return; + } + try { + stream.adjustSendWindow(increment); + } catch (IllegalStateException overflow) { + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.FLOW_CONTROL_ERROR, "stream window overflow"); + } + return; + } long next = connectionSendWindow + increment; if (next > Integer.MAX_VALUE) throw Http2Exception.FLOW_CONTROL_ERROR; connectionSendWindow = next; @@ -229,6 +379,21 @@ public final class Http2Connection implements ConnectionProtocol { writer.writePriority(rst); } + private void sendRstStream(Http2FrameWriter writer, int streamId, Http2ErrorCode error) + throws IOException { + ControlIntent rst = scratch.acquire(ControlKind.SETTINGS_OR_OTHER); + rst.frame(FrameType.RST_STREAM, 0, streamId, error.bytes(), 0, 4); + writer.writePriority(rst); + } + + private void closeStreamAfterError(int streamId) { + Http2Stream stream = streams.remove(streamId); + if (stream == null) return; + if (stream.dispatched()) stream.cancel(); + else streams.release(stream); + if (pendingHeaderStream == stream) pendingHeaderStream = null; + } + private void sendGoAway( Http2FrameWriter writer, int lastStreamId, Http2ErrorCode error, String debug) throws IOException { @@ -303,5 +468,9 @@ public final class Http2Connection implements ConnectionProtocol { peerGoAway = false; gracefulStarted = false; gracefulFinished = false; + highestClientStreamId = 0; + pendingHeaderStream = null; + refusingHeaderStream = false; + dispatchCount = 0; } } diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2HeaderBlockDecoder.java b/flash/src/main/java/dev/relism/flash/http2/Http2HeaderBlockDecoder.java index a6ab050..7ea46e8 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2HeaderBlockDecoder.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2HeaderBlockDecoder.java @@ -7,8 +7,8 @@ import dev.relism.flash.http2.frame.FrameType; import dev.relism.flash.http2.frame.Padding; import dev.relism.flash.http2.hpack.ContinuationAssembler; import dev.relism.flash.http2.hpack.HeaderListSizeException; +import dev.relism.flash.http2.hpack.HeaderSink; import dev.relism.flash.http2.hpack.HpackDecoder; -import dev.relism.flash.http2.hpack.HpackHeaderBlock; /** Composes frame fragment extraction, CONTINUATION assembly and HPACK decoding. */ final class Http2HeaderBlockDecoder { @@ -16,14 +16,14 @@ final class Http2HeaderBlockDecoder { private final ContinuationAssembler assembler = new ContinuationAssembler(); private final HpackDecoder decoder = new HpackDecoder(); - private final HpackHeaderBlock headers = new HpackHeaderBlock(); + private boolean endStream; boolean insideHeaderBlock() { return assembler.isActive(); } /** Accepts HEADERS or CONTINUATION and returns true when a complete block was decoded. */ - boolean accept(FrameHeader frame) { + boolean accept(FrameHeader frame, HeaderSink sink) { if (assembler.isActive() && frame.type() != FrameType.CONTINUATION) { throw Http2Exception.PROTOCOL_ERROR; } @@ -41,9 +41,8 @@ final class Http2HeaderBlockDecoder { } if (!assembler.isComplete()) return false; - headers.reset(); try { - decoder.decode(assembler.buffer(), 0, assembler.length(), headers); + decoder.decode(assembler.buffer(), 0, assembler.length(), sink); } catch (HeaderListSizeException tooLarge) { int streamId = assembler.streamId(); assembler.reset(); @@ -54,7 +53,12 @@ final class Http2HeaderBlockDecoder { return true; } + boolean endStream() { + return endStream; + } + private void begin(FrameHeader frame) { + endStream = FrameFlags.isEndStream(frame.flags()); long unpadded = Padding.unpad( frame.buffer(), @@ -65,6 +69,15 @@ final class Http2HeaderBlockDecoder { int fragmentLength = Pairs.lo(unpadded); if (FrameFlags.hasPriority(frame.flags())) { if (fragmentLength < PRIORITY_FIELDS_LENGTH) throw Http2Exception.FRAME_SIZE_ERROR; + int dependency = + ((frame.buffer()[fragmentOffset] & 0x7f) << 24) + | ((frame.buffer()[fragmentOffset + 1] & 0xff) << 16) + | ((frame.buffer()[fragmentOffset + 2] & 0xff) << 8) + | (frame.buffer()[fragmentOffset + 3] & 0xff); + if (dependency == frame.streamId()) { + throw new Http2StreamException( + frame.streamId(), Http2ErrorCode.PROTOCOL_ERROR, "stream depends on itself"); + } fragmentOffset += PRIORITY_FIELDS_LENGTH; fragmentLength -= PRIORITY_FIELDS_LENGTH; } diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java index 82f3cd1..2b31fea 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java @@ -24,7 +24,7 @@ public final class Http2Limits { * owns a per-stream HPACK arena and request/response state) against a peer that simply opens * streams and never closes them. */ - public static final int MAX_CONCURRENT_STREAMS = 100; + public static final int MAX_CONCURRENT_STREAMS = 64; /** * The largest frame payload we accept without the peer first raising it via our own {@code diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java new file mode 100644 index 0000000..0a82c8c --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/Http2StreamDispatcher.java @@ -0,0 +1,130 @@ +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.Http2Stream; +import dev.relism.flash.http2.stream.Http2StreamState; +import dev.relism.flash.http2.stream.Http2StreamTable; +import dev.relism.flash.models.Request; +import dev.relism.flash.models.RequestHandler; +import dev.relism.flash.models.Response; +import dev.relism.flash.transport.ConnectionContext; +import java.io.IOException; +import java.util.concurrent.RejectedExecutionException; +import lombok.extern.slf4j.Slf4j; + +/** Dispatches completed request streams without blocking the connection demultiplexer. */ +@Slf4j +final class Http2StreamDispatcher { + @FunctionalInterface + interface FailureSink { + void fail(int streamId, Http2ErrorCode errorCode) throws IOException; + } + + private final ConnectionContext context; + private final Http2FrameWriter frameWriter; + private final Http2Settings peerSettings; + private final Http2StreamTable streams; + private final FailureSink failures; + private volatile boolean firstResponse = true; + + Http2StreamDispatcher( + ConnectionContext context, + Http2FrameWriter frameWriter, + Http2Settings peerSettings, + Http2StreamTable streams, + FailureSink failures) { + this.context = context; + this.frameWriter = frameWriter; + this.peerSettings = peerSettings; + this.streams = streams; + this.failures = failures; + } + + void dispatch(Http2Stream stream) { + if (stream.cancelled()) { + streams.release(stream); + return; + } + stream.markDispatched(); + try { + context.executor().execute(() -> handle(stream)); + } catch (RejectedExecutionException rejected) { + failAndRelease(stream, Http2ErrorCode.REFUSED_STREAM, rejected); + } + } + + private void handle(Http2Stream stream) { + try { + Request request = stream.assembleRequest(context.remoteAddress(), context.sslSocket()); + Response pooled = stream.resetResponse(); + Response response = pooled; + Object routeScratch = stream.routeScratch(context.router()); + RequestHandler handler = context.router().route(request, routeScratch); + if (handler == null) handler = context.router().getNotFoundHandler(); + try { + Object result = handler.handle(request, response); + if (result instanceof Response returned) response = returned; + else if (result != null) response.setBody(result); + } catch (Exception handlerFailure) { + Object result = + context.router().getExceptionHandler().handle(handlerFailure, request, response); + if (result instanceof Response returned) response = returned; + else if (result != null) response.setBody(result); + } + + Http2ResponseWriter responseWriter = stream.responseWriter(); + if (stream.cancelled()) { + request.recycle(); + if (response == pooled) pooled.recycle(); + streams.release(stream); + return; + } + boolean prepared; + synchronized (this) { + boolean tableUpdate = firstResponse; + prepared = + responseWriter.prepare( + response, + stream.id(), + request.method() == HttpMethod.HEAD, + context.configuration().isSendDate(), + true, + context.configuration().isH2HuffmanDynamicValues(), + tableUpdate, + peerSettings.maxFrameSize(), + peerSettings.maxHeaderListSize(), + (int) Math.min(stream.sendWindow(), Integer.MAX_VALUE)); + if (prepared) { + firstResponse = false; + stream.transition(Http2StreamState.Event.SEND_HEADERS_ES); + request.recycle(); + if (response == pooled) pooled.recycle(); + streams.remove(stream.id()); + frameWriter.write(responseWriter); + } + } + if (!prepared) { + request.recycle(); + if (response == pooled) pooled.recycle(); + failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, null); + } + } catch (Exception failure) { + failAndRelease(stream, Http2ErrorCode.INTERNAL_ERROR, failure); + } + } + + 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 { + failures.fail(stream.id(), error); + } catch (IOException writeFailure) { + log.debug("Failed to write RST_STREAM for {}", stream.id(), writeFailure); + } finally { + streams.release(stream); + } + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/frame/Http2FrameReader.java b/flash/src/main/java/dev/relism/flash/http2/frame/Http2FrameReader.java index 2e08fb3..7fedaf0 100644 --- a/flash/src/main/java/dev/relism/flash/http2/frame/Http2FrameReader.java +++ b/flash/src/main/java/dev/relism/flash/http2/frame/Http2FrameReader.java @@ -116,6 +116,11 @@ public final class Http2FrameReader { return totalRead != 0 && System.nanoTime() >= frameDeadlineNanos; } + /** Whether another frame may be consumed immediately without waiting for network input. */ + public boolean hasBufferedInput() { + return totalRead != 0 || in.available() != 0; + } + private static int decodeLength(byte[] buf, int off) { int b0 = buf[off] & 0xFF, b1 = buf[off + 1] & 0xFF, b2 = buf[off + 2] & 0xFF; return (b0 << 16) | (b1 << 8) | b2; diff --git a/flash/src/main/java/dev/relism/flash/http2/hpack/Huffman.java b/flash/src/main/java/dev/relism/flash/http2/hpack/Huffman.java index 1c3a1bd..939b6ad 100644 --- a/flash/src/main/java/dev/relism/flash/http2/hpack/Huffman.java +++ b/flash/src/main/java/dev/relism/flash/http2/hpack/Huffman.java @@ -306,8 +306,7 @@ public final class Huffman { /** * Huffman-encodes {@code src[off, off + len)}, writing directly into {@code out}. Pads the final - * byte with the high-order bits of the EOS code (all 1s), per RFC 7541 §5.2. Built now ({@code - * EX} task 3) for use by the HPACK encoder. + * byte with the high-order bits of the EOS code (all 1s), per RFC 7541 §5.2. */ public static void encode(ByteWriter out, byte[] src, int off, int len) { long accumulator = 0; diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java new file mode 100644 index 0000000..1fbad74 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java @@ -0,0 +1,138 @@ +package dev.relism.flash.http2.message; + +import dev.relism.flash.bytes.PooledSlice; +import dev.relism.flash.http2.hpack.HpackHeaderBlock; +import dev.relism.flash.models.HeaderView; +import dev.relism.fpr.core.ByteView; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; + +/** Header view over stream-owned decoded HPACK storage. Pseudo-fields are excluded. */ +public final class Http2HeaderMap implements HeaderView { + private static final int VIEW_COUNT = 4; + + private final PooledSlice scanName = new PooledSlice(); + private final PooledSlice scanValue = new PooledSlice(); + private final PooledSlice[] views = new PooledSlice[VIEW_COUNT]; + private HpackHeaderBlock block; + private PseudoHeaders pseudoHeaders; + private int viewCursor; + private int regularCount; + + public Http2HeaderMap() { + for (int i = 0; i < views.length; i++) views[i] = new PooledSlice(); + } + + public void reset(HpackHeaderBlock block, PseudoHeaders pseudoHeaders) { + this.block = block; + this.pseudoHeaders = pseudoHeaders; + viewCursor = 0; + regularCount = 0; + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) != ':') regularCount++; + } + } + + @Override + public String first(String name) { + ByteView value = find(name, scanValue); + if (value == null) return null; + PooledSlice slice = (PooledSlice) value; + return new String(slice.array(), slice.offset(), slice.length(), StandardCharsets.UTF_8); + } + + @Override + public List all(String name) { + List result = null; + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) == ':' || !equalsIgnoreCase(scanName, name)) continue; + if (result == null) result = new ArrayList<>(); + result.add( + new String( + scanValue.array(), scanValue.offset(), scanValue.length(), StandardCharsets.UTF_8)); + } + if (result == null && isAuthorityAlias(name) && pseudoHeaders.authority().array() != null) { + return List.of(string(pseudoHeaders.authority())); + } + return result == null ? List.of() : result; + } + + @Override + public List all() { + List result = new ArrayList<>(regularCount); + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) != ':') result.add(string(scanValue)); + } + return result; + } + + @Override + public ByteView view(String name) { + PooledSlice target = views[viewCursor++ & (views.length - 1)]; + return find(name, target); + } + + @Override + public boolean valueEqualsIgnoreCase(String name, String value) { + ByteView found = find(name, scanValue); + return found != null && equalsIgnoreCase(found, value); + } + + @Override + public boolean contains(String name) { + return find(name, scanValue) != null; + } + + @Override + public int count() { + return regularCount; + } + + @Override + public void forEach(HeaderConsumer consumer) { + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) != ':') consumer.accept(scanName, scanValue); + } + } + + private PooledSlice find(String requested, PooledSlice target) { + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) != ':' && equalsIgnoreCase(scanName, requested)) { + target.reset(scanValue.array(), scanValue.offset(), scanValue.length()); + return target; + } + } + if (isAuthorityAlias(requested) && pseudoHeaders.authority().array() != null) { + PooledSlice authority = pseudoHeaders.authority(); + target.reset(authority.array(), authority.offset(), authority.length()); + return target; + } + return null; + } + + private static boolean isAuthorityAlias(String name) { + return name.equalsIgnoreCase("host") || name.equalsIgnoreCase(":authority"); + } + + private static boolean equalsIgnoreCase(ByteView bytes, String value) { + if (bytes.length() != value.length()) return false; + for (int i = 0; i < bytes.length(); i++) { + int left = bytes.byteAt(i) & 0xff; + int right = value.charAt(i); + if (left >= 'A' && left <= 'Z') left += 32; + if (right >= 'A' && right <= 'Z') right += 32; + if (left != right) return false; + } + return true; + } + + private static String string(PooledSlice slice) { + return new String(slice.array(), slice.offset(), slice.length(), StandardCharsets.UTF_8); + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java index 691a51e..0fa8c70 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2ResponseWriter.java @@ -19,6 +19,11 @@ import dev.relism.flash.models.ResponseSerializer; * connection write lock and exposes it as one {@link WriteIntent}. */ public final class Http2ResponseWriter implements WriteIntent, ResponseSerializer.FieldConsumer { + @FunctionalInterface + public interface Completion { + void responseWriteCompleted(); + } + private static final int STATUS_NAME_LENGTH = 7; private static final int CONTENT_LENGTH_NAME_LENGTH = 14; @@ -31,6 +36,7 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize private int streamId; private long headerListSize; private long maxHeaderListSize; + private Completion completion; public Http2ResponseWriter() { this(1024, 2048); @@ -42,6 +48,10 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize frames = new FrameWriteBuffer(output); } + public void completion(Completion completion) { + this.completion = completion; + } + /** * Prepares a non-streaming response. Returns {@code false} when the body needs the deferred DATA * flow-control path implemented by the stream scheduler. @@ -236,4 +246,9 @@ public final class Http2ResponseWriter implements WriteIntent, ResponseSerialize public void setMpscNext(WriteIntent next) { this.next = next; } + + @Override + public void completed() { + if (completion != null) completion.responseWriteCompleted(); + } } diff --git a/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java new file mode 100644 index 0000000..59428b7 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/message/PseudoHeaders.java @@ -0,0 +1,138 @@ +package dev.relism.flash.http2.message; + +import dev.relism.flash.bytes.PooledSlice; +import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2StreamException; +import dev.relism.flash.http2.hpack.HpackHeaderBlock; +import dev.relism.fpr.core.ByteView; + +/** Validates request pseudo-headers and HTTP/2 field rules while extracting request metadata. */ +public final class PseudoHeaders { + private static final int METHOD = 1; + private static final int SCHEME = 2; + private static final int PATH = 4; + private static final int AUTHORITY = 8; + + private final PooledSlice name = new PooledSlice(); + private final PooledSlice value = new PooledSlice(); + private final PooledSlice method = new PooledSlice(); + private final PooledSlice scheme = new PooledSlice(); + private final PooledSlice path = new PooledSlice(); + private final PooledSlice authority = new PooledSlice(); + private final PooledSlice host = new PooledSlice(); + private int present; + + public void validate(HpackHeaderBlock block, int streamId) { + present = 0; + method.reset(null, 0, 0); + scheme.reset(null, 0, 0); + path.reset(null, 0, 0); + authority.reset(null, 0, 0); + host.reset(null, 0, 0); + boolean regularSeen = false; + + for (int i = 0; i < block.count(); i++) { + block.get(i, name, value); + if (name.length() == 0) fail(streamId, "empty field name"); + boolean pseudo = name.byteAt(0) == ':'; + if (pseudo) { + if (regularSeen) fail(streamId, "pseudo-header after regular field"); + int bit = pseudoBit(name); + if (bit == 0) fail(streamId, "unknown pseudo-header"); + if ((present & bit) != 0) fail(streamId, "duplicate pseudo-header"); + present |= bit; + copySlice(bit, value); + } else { + regularSeen = true; + validateRegular(name, value, streamId); + if (equals(name, "host")) copy(value, host); + } + } + + if ((present & METHOD) == 0) fail(streamId, "missing :method"); + boolean connect = equals(method, "CONNECT"); + if (connect) { + if ((present & AUTHORITY) == 0) fail(streamId, "CONNECT requires :authority"); + if ((present & (SCHEME | PATH)) != 0) fail(streamId, "CONNECT forbids :scheme and :path"); + } else { + int required = METHOD | SCHEME | PATH | AUTHORITY; + if ((present & required) != required) fail(streamId, "missing request pseudo-header"); + if (path.length() == 0) fail(streamId, "empty :path"); + } + if (host.array() != null && authority.array() != null && !equals(host, authority)) { + fail(streamId, "host conflicts with :authority"); + } + } + + public PooledSlice method() { + return method; + } + + public PooledSlice scheme() { + return scheme; + } + + public PooledSlice path() { + return path; + } + + public PooledSlice authority() { + return authority; + } + + private void copySlice(int bit, PooledSlice source) { + if (bit == METHOD) copy(source, method); + else if (bit == SCHEME) copy(source, scheme); + else if (bit == PATH) copy(source, path); + else copy(source, authority); + } + + private static void copy(PooledSlice source, PooledSlice target) { + target.reset(source.array(), source.offset(), source.length()); + } + + private static int pseudoBit(ByteView name) { + if (equals(name, ":method")) return METHOD; + if (equals(name, ":scheme")) return SCHEME; + if (equals(name, ":path")) return PATH; + if (equals(name, ":authority")) return AUTHORITY; + return 0; + } + + private static void validateRegular(ByteView name, ByteView value, int streamId) { + for (int i = 0; i < name.length(); i++) { + int c = name.byteAt(i) & 0xff; + if (c >= 'A' && c <= 'Z') fail(streamId, "uppercase field name"); + } + if (equals(name, "connection") + || equals(name, "keep-alive") + || equals(name, "proxy-connection") + || equals(name, "transfer-encoding") + || equals(name, "upgrade")) { + fail(streamId, "connection-specific field"); + } + if (equals(name, "te") && !equals(value, "trailers")) { + fail(streamId, "invalid te field"); + } + } + + static boolean equals(ByteView view, String expected) { + if (view.length() != expected.length()) return false; + for (int i = 0; i < view.length(); i++) { + if ((view.byteAt(i) & 0xff) != expected.charAt(i)) return false; + } + return true; + } + + static boolean equals(ByteView left, ByteView right) { + if (left.length() != right.length()) return false; + for (int i = 0; i < left.length(); i++) { + if (left.byteAt(i) != right.byteAt(i)) return false; + } + return true; + } + + private static void fail(int streamId, String message) { + throw new Http2StreamException(streamId, Http2ErrorCode.PROTOCOL_ERROR, message); + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java new file mode 100644 index 0000000..94b6645 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2Stream.java @@ -0,0 +1,168 @@ +package dev.relism.flash.http2.stream; + +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.Http2StreamException; +import dev.relism.flash.http2.hpack.HpackHeaderBlock; +import dev.relism.flash.http2.message.Http2HeaderMap; +import dev.relism.flash.http2.message.Http2ResponseWriter; +import dev.relism.flash.http2.message.PseudoHeaders; +import dev.relism.flash.models.Request; +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 java.net.InetSocketAddress; +import javax.net.ssl.SSLSocket; + +/** Per-stream request, response, decoded-header and write state. */ +public final class Http2Stream implements Http2ResponseWriter.Completion { + private static final byte[] HTTP_2 = {'H', 'T', 'T', 'P', '/', '2'}; + + private final HpackHeaderBlock headerBlock = new HpackHeaderBlock(); + private final PseudoHeaders pseudoHeaders = new PseudoHeaders(); + private final Http2HeaderMap headers = new Http2HeaderMap(); + private final RequestLine requestLine = new RequestLine(); + private final RequestBody requestBody = new RequestBody(); + 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 int id; + private Http2StreamState state = Http2StreamState.IDLE; + private int sendWindow = 65_535; + private Http2StreamTable owner; + private Object routeScratch; + private volatile boolean dispatched; + private volatile boolean cancelled; + private boolean headersValidated; + Http2Stream poolNext; + + Http2Stream() { + responseWriter.completion(this); + protocol.reset(HTTP_2, 0, HTTP_2.length); + } + + void reset(int id, Http2StreamTable owner) { + this.id = id; + this.owner = owner; + state = Http2StreamState.IDLE; + sendWindow = 65_535; + dispatched = false; + cancelled = false; + headersValidated = false; + headerBlock.reset(); + } + + void clear() { + request.recycle(); + response.recycle(); + id = 0; + owner = null; + state = Http2StreamState.CLOSED; + } + + public Request assembleRequest(InetSocketAddress remoteAddress, SSLSocket sslSocket) { + validateHeaders(); + headers.reset(headerBlock, pseudoHeaders); + PooledSlice rawPath = pseudoHeaders.path(); + if (rawPath.array() == null) rawPath = pseudoHeaders.authority(); + int question = -1; + for (int i = 0; i < rawPath.length(); i++) { + if (rawPath.byteAt(i) == '?') { + question = i; + break; + } + } + if (question < 0) { + path.reset(rawPath.array(), rawPath.offset(), rawPath.length()); + query.reset(null, 0, 0); + } else { + path.reset(rawPath.array(), rawPath.offset(), question); + query.reset( + rawPath.array(), rawPath.offset() + question + 1, rawPath.length() - question - 1); + } + PooledSlice methodBytes = pseudoHeaders.method(); + HttpMethod method = + HttpMethod.fromBytes(methodBytes.array(), methodBytes.offset(), methodBytes.length()); + if (method == null) { + throw new Http2StreamException( + id, Http2ErrorCode.PROTOCOL_ERROR, "unsupported request method"); + } + requestLine.reset(method, path, question < 0 ? null : query, protocol, headers); + requestBody.reset(null, 0, null, 0, 0); + return Request.forParsed(request, requestLine, requestBody, remoteAddress, sslSocket); + } + + public void validateHeaders() { + if (headersValidated) return; + pseudoHeaders.validate(headerBlock, id); + headersValidated = true; + } + + public Response resetResponse() { + return response.reset(200, ContentType.TEXT_PLAIN); + } + + public void transition(Http2StreamState.Event event) { + state = state.transition(id, event); + } + + public int id() { + return id; + } + + public Http2StreamState state() { + return state; + } + + public HpackHeaderBlock headerBlock() { + return headerBlock; + } + + public Http2ResponseWriter responseWriter() { + return responseWriter; + } + + public Object routeScratch(AbstractRouter router) { + if (routeScratch == null) routeScratch = router.newScratch(); + return routeScratch; + } + + public void markDispatched() { + dispatched = true; + } + + public boolean dispatched() { + return dispatched; + } + + public void cancel() { + cancelled = true; + } + + public boolean cancelled() { + return cancelled; + } + + public int sendWindow() { + return sendWindow; + } + + public void adjustSendWindow(int delta) { + long adjusted = (long) sendWindow + delta; + if (adjusted > Integer.MAX_VALUE) throw new IllegalStateException("stream window overflow"); + sendWindow = (int) adjusted; + } + + @Override + public void responseWriteCompleted() { + Http2StreamTable table = owner; + if (table != null) table.release(this); + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java new file mode 100644 index 0000000..d682dd5 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java @@ -0,0 +1,95 @@ +package dev.relism.flash.http2.stream; + +import dev.relism.flash.http2.Http2ErrorCode; +import dev.relism.flash.http2.Http2StreamException; + +/** Explicit RFC 9113 stream-state transition table. */ +public enum Http2StreamState { + IDLE, + OPEN, + HALF_CLOSED_REMOTE, + HALF_CLOSED_LOCAL, + CLOSED; + + public enum Event { + RECV_HEADERS, + RECV_HEADERS_ES, + RECV_DATA, + RECV_DATA_ES, + RECV_RST, + SEND_HEADERS, + SEND_HEADERS_ES, + SEND_DATA, + SEND_DATA_ES, + SEND_RST + } + + private static final byte ERROR = -1; + private static final byte[][] TRANSITIONS = buildTransitions(); + + public Http2StreamState transition(int streamId, Event event) { + int next = TRANSITIONS[ordinal()][event.ordinal()]; + if (next == ERROR) { + throw new Http2StreamException( + streamId, errorFor(event), "invalid stream transition " + this + " + " + event); + } + return values()[next]; + } + + private Http2ErrorCode errorFor(Event event) { + if (this == CLOSED) return Http2ErrorCode.STREAM_CLOSED; + if (this == HALF_CLOSED_REMOTE + && (event == Event.RECV_DATA + || event == Event.RECV_DATA_ES + || event == Event.RECV_HEADERS + || event == Event.RECV_HEADERS_ES)) { + return Http2ErrorCode.STREAM_CLOSED; + } + return Http2ErrorCode.PROTOCOL_ERROR; + } + + public static boolean isValid(Http2StreamState state, Event event) { + return TRANSITIONS[state.ordinal()][event.ordinal()] != ERROR; + } + + private static byte[][] buildTransitions() { + byte[][] table = new byte[values().length][Event.values().length]; + for (byte[] row : table) java.util.Arrays.fill(row, ERROR); + + set(table, IDLE, Event.RECV_HEADERS, OPEN); + set(table, IDLE, Event.RECV_HEADERS_ES, HALF_CLOSED_REMOTE); + + set(table, OPEN, Event.RECV_HEADERS, OPEN); + set(table, OPEN, Event.RECV_HEADERS_ES, HALF_CLOSED_REMOTE); + set(table, OPEN, Event.RECV_DATA, OPEN); + set(table, OPEN, Event.RECV_DATA_ES, HALF_CLOSED_REMOTE); + set(table, OPEN, Event.RECV_RST, CLOSED); + set(table, OPEN, Event.SEND_HEADERS, OPEN); + set(table, OPEN, Event.SEND_HEADERS_ES, HALF_CLOSED_LOCAL); + set(table, OPEN, Event.SEND_DATA, OPEN); + set(table, OPEN, Event.SEND_DATA_ES, HALF_CLOSED_LOCAL); + set(table, OPEN, Event.SEND_RST, CLOSED); + + set(table, HALF_CLOSED_REMOTE, Event.RECV_RST, CLOSED); + set(table, HALF_CLOSED_REMOTE, Event.SEND_HEADERS, HALF_CLOSED_REMOTE); + set(table, HALF_CLOSED_REMOTE, Event.SEND_HEADERS_ES, CLOSED); + set(table, HALF_CLOSED_REMOTE, Event.SEND_DATA, HALF_CLOSED_REMOTE); + set(table, HALF_CLOSED_REMOTE, Event.SEND_DATA_ES, CLOSED); + set(table, HALF_CLOSED_REMOTE, Event.SEND_RST, CLOSED); + + set(table, HALF_CLOSED_LOCAL, Event.RECV_HEADERS, HALF_CLOSED_LOCAL); + set(table, HALF_CLOSED_LOCAL, Event.RECV_HEADERS_ES, CLOSED); + set(table, HALF_CLOSED_LOCAL, Event.RECV_DATA, HALF_CLOSED_LOCAL); + set(table, HALF_CLOSED_LOCAL, Event.RECV_DATA_ES, CLOSED); + set(table, HALF_CLOSED_LOCAL, Event.RECV_RST, CLOSED); + set(table, HALF_CLOSED_LOCAL, Event.SEND_RST, CLOSED); + + set(table, CLOSED, Event.RECV_RST, CLOSED); + set(table, CLOSED, Event.SEND_RST, CLOSED); + return table; + } + + private static void set(byte[][] table, Http2StreamState from, Event event, Http2StreamState to) { + table[from.ordinal()][event.ordinal()] = (byte) to.ordinal(); + } +} diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java new file mode 100644 index 0000000..3588ed7 --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamTable.java @@ -0,0 +1,145 @@ +package dev.relism.flash.http2.stream; + +import java.util.Arrays; + +/** Fixed-capacity primitive stream-id table using linear-probed open addressing. */ +public final class Http2StreamTable { + + @FunctionalInterface + public interface StreamConsumer { + void accept(Http2Stream stream); + } + + private final int[] keys; + private final Http2Stream[] values; + private final int mask; + private final int maxEntries; + private int size; + private Http2Stream free; + private int created; + + public Http2StreamTable(int maxEntries) { + if (maxEntries < 1) throw new IllegalArgumentException("maxEntries must be positive"); + int capacity = 1; + while (capacity < maxEntries * 2) capacity <<= 1; + keys = new int[capacity]; + values = new Http2Stream[capacity]; + mask = capacity - 1; + this.maxEntries = maxEntries; + } + + public synchronized Http2Stream get(int streamId) { + int slot = find(streamId); + return keys[slot] == streamId ? values[slot] : null; + } + + public synchronized void put(Http2Stream stream) { + if (size == maxEntries) throw new IllegalStateException("stream table capacity exceeded"); + int streamId = stream.id(); + int slot = find(streamId); + if (keys[slot] == streamId) throw new IllegalStateException("duplicate stream " + streamId); + keys[slot] = streamId; + values[slot] = stream; + size++; + } + + public synchronized Http2Stream acquire(int streamId) { + if (size == maxEntries) return null; + Http2Stream stream = free; + if (stream != null) { + free = stream.poolNext; + stream.poolNext = null; + } else { + if (created == maxEntries) return null; + stream = new Http2Stream(); + created++; + } + stream.reset(streamId, this); + put(stream); + return stream; + } + + public synchronized void release(Http2Stream stream) { + stream.clear(); + stream.poolNext = free; + free = stream; + } + + public synchronized Http2Stream remove(int streamId) { + int slot = find(streamId); + if (keys[slot] != streamId) return null; + Http2Stream removed = values[slot]; + keys[slot] = 0; + values[slot] = null; + size--; + + int scan = (slot + 1) & mask; + while (keys[scan] != 0) { + int key = keys[scan]; + Http2Stream value = values[scan]; + keys[scan] = 0; + values[scan] = null; + size--; + put(value); + scan = (scan + 1) & mask; + } + return removed; + } + + public synchronized void forEach(StreamConsumer consumer) { + for (int i = 0; i < keys.length; i++) { + if (keys[i] != 0) consumer.accept(values[i]); + } + } + + public synchronized void adjustAllSendWindows(int delta) { + for (int i = 0; i < keys.length; i++) { + if (keys[i] == 0) continue; + long adjusted = (long) values[i].sendWindow() + delta; + if (adjusted > Integer.MAX_VALUE) { + throw new IllegalStateException("stream window overflow"); + } + } + for (int i = 0; i < keys.length; i++) { + if (keys[i] != 0) values[i].adjustSendWindow(delta); + } + } + + public synchronized int size() { + return size; + } + + public int capacity() { + return maxEntries; + } + + public synchronized int createdCount() { + return created; + } + + public synchronized int freeCount() { + int count = 0; + for (Http2Stream stream = free; stream != null; stream = stream.poolNext) count++; + return count; + } + + public synchronized void clear() { + Arrays.fill(keys, 0); + Arrays.fill(values, null); + size = 0; + } + + private int find(int streamId) { + if (streamId <= 0) throw new IllegalArgumentException("streamId must be positive"); + int slot = mix(streamId) & mask; + while (keys[slot] != 0 && keys[slot] != streamId) slot = (slot + 1) & mask; + return slot; + } + + private static int mix(int value) { + value ^= value >>> 16; + value *= 0x7feb352d; + value ^= value >>> 15; + return value; + } +} diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java index e47b51f..6d9fafc 100644 --- a/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java +++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionContext.java @@ -3,46 +3,43 @@ package dev.relism.flash.transport; import dev.relism.flash.extension.FlashConfiguration; import dev.relism.flash.routing.AbstractRouter; import dev.relism.flash.routing.AbstractWsRouter; - -import javax.net.ssl.SSLSocket; - import java.io.OutputStream; import java.net.InetSocketAddress; import java.net.Socket; +import java.util.concurrent.ExecutorService; import java.util.function.BooleanSupplier; +import javax.net.ssl.SSLSocket; /** * Everything a {@link ConnectionProtocol} implementation needs to serve one connection, bundled * into a single object instead of a long parameter list. * - * @param socket the accepted socket — owns its lifecycle (closing it is the caller's, - * i.e. {@link ConnectionRunner}'s, responsibility, not the protocol's) - * @param sslSocket {@code socket} narrowed to {@link SSLSocket}, or {@code null} for a - * plaintext connection - * @param in the single buffered, deadline-aware source for this connection's - * inbound bytes - * @param out the buffered output stream — for header/body writes that benefit from - * userspace coalescing before a single syscall - * @param rawOut the unbuffered output stream — for WebSocket, whose writes are already - * bulk (see {@code WebSocketSession}) + * @param socket the accepted socket — owns its lifecycle (closing it is the caller's, i.e. {@link + * ConnectionRunner}'s, responsibility, not the protocol's) + * @param sslSocket {@code socket} narrowed to {@link SSLSocket}, or {@code null} for a plaintext + * connection + * @param in the single buffered, deadline-aware source for this connection's inbound bytes + * @param out the buffered output stream — for header/body writes that benefit from userspace + * coalescing before a single syscall + * @param rawOut the unbuffered output stream — for WebSocket, whose writes are already bulk (see + * {@code WebSocketSession}) * @param remoteAddress the client's address, or {@code null} if unavailable - * @param router the HTTP router - * @param wsRouter the WebSocket router + * @param router the HTTP router + * @param wsRouter the WebSocket router * @param configuration the server configuration (timeouts, limits, feature flags) - * @param stopped {@code true} once the server has begun shutting down — a protocol - * implementation's request loop must check this and exit promptly + * @param stopped {@code true} once the server has begun shutting down — a protocol implementation's + * request loop must check this and exit promptly */ public record ConnectionContext( - Socket socket, - SSLSocket sslSocket, - BufferedByteSource in, - OutputStream out, - OutputStream rawOut, - InetSocketAddress remoteAddress, - ConnectionScratch scratch, - AbstractRouter router, - AbstractWsRouter wsRouter, - FlashConfiguration configuration, - BooleanSupplier stopped -) { -} + Socket socket, + SSLSocket sslSocket, + BufferedByteSource in, + OutputStream out, + OutputStream rawOut, + InetSocketAddress remoteAddress, + ConnectionScratch scratch, + AbstractRouter router, + AbstractWsRouter wsRouter, + FlashConfiguration configuration, + ExecutorService executor, + BooleanSupplier stopped) {} diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java index 7f4531c..228b911 100644 --- a/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java +++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java @@ -113,6 +113,7 @@ public final class ConnectionRunner { router, wsRouter, configuration, + executorService, stopped); if (negotiated == NegotiatedProtocol.HTTP_2) http2ProtocolFactory.get().run(ctx); else http1Protocol.run(ctx); diff --git a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java index 3654903..8abd20b 100644 --- a/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/Http2ConnectionIntegrationTest.java @@ -2,22 +2,32 @@ package dev.relism.flash.http2; 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.http2.frame.FrameFlags; import dev.relism.flash.http2.frame.FrameType; +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.EOFException; import java.io.InputStream; import java.net.ServerSocket; import java.net.Socket; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.nio.file.Path; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import javax.net.ssl.SSLParameters; import javax.net.ssl.SSLSocket; import org.junit.jupiter.api.AfterEach; @@ -32,6 +42,171 @@ class Http2ConnectionIntegrationTest { if (app != null) app.stop().join(); } + @Test + void javaHttpClientUsesHttp2AgainstAnExistingParameterizedRoute(@TempDir Path directory) + throws Exception { + int port = freePort(); + Path keystore = + TestKeystores.build( + directory, + "http2-route.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( + "/users/{id}", (request, response) -> request.param("id") + ":" + request.header("host")); + app.start(); + + HttpClient client = + HttpClient.newBuilder() + .sslContext(TestKeystores.trustAllClientContext()) + .version(HttpClient.Version.HTTP_2) + .build(); + HttpResponse response = + client.send( + HttpRequest.newBuilder(URI.create("https://localhost:" + port + "/users/42")) + .GET() + .build(), + HttpResponse.BodyHandlers.ofString()); + + assertEquals(HttpClient.Version.HTTP_2, response.version()); + assertEquals(200, response.statusCode()); + assertEquals("42:localhost:" + port, response.body()); + } + + @Test + void bodylessGetRunsExistingRouteAndReturnsHeadersAndData() throws Exception { + int port = freePort(); + app = + FlashApp.create( + FlashConfiguration.builder().port(port).host("127.0.0.1").http2Enabled(true).build()); + app.get("/api/ping", (request, response) -> "pong"); + app.start(); + + ByteWriter block = new ByteWriter(64); + HpackEncoder.writeIndexed(block, 2); + HpackEncoder.writeIndexed(block, 6); + HpackEncoder.writeLiteralWithNameIndex( + block, 4, "/api/ping".getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeLiteralWithNameIndex( + block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), 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 | FrameFlags.END_STREAM, + 1, + Arrays.copyOf(block.array(), block.length())))); + socket.getOutputStream().flush(); + + ByteWriter responseBlock = new ByteWriter(128); + byte[] body = null; + for (int i = 0; i < 10 && body == null; i++) { + Http2TestFrames.WireFrame frame = readFrame(socket.getInputStream()); + if (frame.streamId() != 1) continue; + if (frame.type() == FrameType.HEADERS.code() + || frame.type() == FrameType.CONTINUATION.code()) { + responseBlock.writeBytes(frame.payload()); + } else if (frame.type() == FrameType.DATA.code()) { + body = frame.payload(); + } + } + + List fields = new ArrayList<>(); + new HpackDecoder() + .decode( + responseBlock.array(), + 0, + responseBlock.length(), + (name, value, never) -> fields.add(ascii(name) + "=" + ascii(value))); + assertTrue(fields.contains(":status=200")); + assertTrue(fields.contains("content-length=4")); + assertEquals("pong", new String(body, StandardCharsets.US_ASCII)); + } + } + + @Test + void resetQueuedStreamCannotReleaseOrReuseItBeforeDispatchObservesCancellation() + throws Exception { + int port = freePort(); + AtomicInteger calls = new AtomicInteger(); + app = + FlashApp.create( + FlashConfiguration.builder().port(port).host("127.0.0.1").http2Enabled(true).build()); + app.get( + "/queued", + (request, response) -> { + calls.incrementAndGet(); + return "ok"; + }); + app.start(); + + ByteWriter block = new ByteWriter(64); + HpackEncoder.writeIndexed(block, 2); + HpackEncoder.writeIndexed(block, 6); + HpackEncoder.writeLiteralWithNameIndex( + block, 4, "/queued".getBytes(StandardCharsets.US_ASCII), false); + HpackEncoder.writeLiteralWithNameIndex( + block, 1, "localhost".getBytes(StandardCharsets.US_ASCII), false); + byte[] headers = Arrays.copyOf(block.array(), block.length()); + byte[] cancel = {0, 0, 0, 8}; + + 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 | FrameFlags.END_STREAM, + 1, + headers), + Http2TestFrames.frame(FrameType.RST_STREAM, 0, 1, cancel), + Http2TestFrames.frame( + FrameType.HEADERS, + FrameFlags.END_HEADERS | FrameFlags.END_STREAM, + 3, + headers))); + socket.getOutputStream().flush(); + + Http2TestFrames.WireFrame response = null; + for (int i = 0; i < 10; i++) { + Http2TestFrames.WireFrame frame = readFrame(socket.getInputStream()); + assertFalse( + frame.streamId() == 1 + && (frame.type() == FrameType.HEADERS.code() + || frame.type() == FrameType.DATA.code()), + "a reset request must not produce a response"); + if (frame.streamId() == 3 && frame.type() == FrameType.DATA.code()) { + response = frame; + break; + } + } + assertNotNull(response); + assertEquals("ok", new String(response.payload(), StandardCharsets.US_ASCII)); + assertEquals(1, calls.get()); + } + } + @Test void h2cTransportDispatchesControlFramesWithoutRunningApplicationWork() throws Exception { int port = freePort(); @@ -244,4 +419,10 @@ class Http2ConnectionIntegrationTest { return socket.getLocalPort(); } } + + private static String ascii(dev.relism.fpr.core.ByteView view) { + byte[] bytes = new byte[view.length()]; + for (int i = 0; i < bytes.length; i++) bytes[i] = view.byteAt(i); + return new String(bytes, StandardCharsets.US_ASCII); + } } diff --git a/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java b/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java new file mode 100644 index 0000000..4ec4e3f --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/message/PseudoHeaderValidationTest.java @@ -0,0 +1,86 @@ +package dev.relism.flash.http2.message; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import dev.relism.flash.bytes.PooledSlice; +import dev.relism.flash.http2.Http2StreamException; +import dev.relism.flash.http2.hpack.HpackHeaderBlock; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +class PseudoHeaderValidationTest { + @Test + void validRequest() { + assertDoesNotThrow( + () -> + validate( + ":method", "GET", ":scheme", "https", ":path", "/", ":authority", "example.com")); + } + + @Test + void rejectsPseudoAfterRegular() { + rejects("x", "1", ":method", "GET", ":scheme", "https", ":path", "/", ":authority", "x"); + } + + @Test + void rejectsUnknownAndDuplicatePseudoHeaders() { + rejects(":method", "GET", ":scheme", "https", ":path", "/", ":authority", "x", ":other", "x"); + rejects( + ":method", "GET", ":method", "POST", ":scheme", "https", ":path", "/", ":authority", "x"); + } + + @Test + void rejectsMissingAndEmptyPseudoHeaders() { + rejects(":method", "GET", ":scheme", "https", ":path", "/"); + rejects(":method", "GET", ":scheme", "https", ":path", "", ":authority", "x"); + } + + @Test + void validatesConnectShape() { + assertDoesNotThrow(() -> validate(":method", "CONNECT", ":authority", "example.com:443")); + rejects(":method", "CONNECT", ":scheme", "https", ":authority", "example.com:443"); + } + + @Test + void rejectsUppercaseForbiddenAndInvalidTeFields() { + rejects(validWith("X-Test", "1")); + rejects(validWith("connection", "close")); + rejects(validWith("keep-alive", "timeout=5")); + rejects(validWith("proxy-connection", "close")); + rejects(validWith("transfer-encoding", "chunked")); + rejects(validWith("upgrade", "websocket")); + rejects(validWith("te", "gzip")); + assertDoesNotThrow(() -> validate(validWith("te", "trailers"))); + } + + @Test + void rejectsHostAuthorityConflict() { + rejects(validWith("host", "other.example")); + assertDoesNotThrow(() -> validate(validWith("host", "example.com"))); + } + + private static String[] validWith(String name, String value) { + return new String[] { + ":method", "GET", ":scheme", "https", ":path", "/", ":authority", "example.com", name, value + }; + } + + private static void rejects(String... fields) { + assertThrows(Http2StreamException.class, () -> validate(fields)); + } + + private static void validate(String... fields) { + HpackHeaderBlock block = new HpackHeaderBlock(); + PooledSlice name = new PooledSlice(); + PooledSlice value = new PooledSlice(); + for (int i = 0; i < fields.length; i += 2) { + byte[] nameBytes = fields[i].getBytes(StandardCharsets.US_ASCII); + byte[] valueBytes = fields[i + 1].getBytes(StandardCharsets.US_ASCII); + name.reset(nameBytes, 0, nameBytes.length); + value.reset(valueBytes, 0, valueBytes.length); + block.accept(name, value, false); + } + new PseudoHeaders().validate(block, 1); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/stream/Http2RequestAssemblyTest.java b/flash/src/test/java/dev/relism/flash/http2/stream/Http2RequestAssemblyTest.java new file mode 100644 index 0000000..d2b3bf0 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/stream/Http2RequestAssemblyTest.java @@ -0,0 +1,43 @@ +package dev.relism.flash.http2.stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import dev.relism.flash.bytes.PooledSlice; +import dev.relism.flash.http.HttpMethod; +import dev.relism.flash.models.Request; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; + +class Http2RequestAssemblyTest { + @Test + void assemblesProtocolNeutralRequestWithQueryAndAuthorityAlias() { + Http2StreamTable table = new Http2StreamTable(1); + Http2Stream stream = table.acquire(1); + field(stream, ":method", "GET"); + field(stream, ":scheme", "https"); + field(stream, ":path", "/users/42?verbose=true"); + field(stream, ":authority", "example.com"); + field(stream, "x-trace", "abc"); + + Request request = stream.assembleRequest(null, null); + + assertEquals(HttpMethod.GET, request.method()); + assertEquals("/users/42", request.path()); + assertEquals("true", request.query("verbose")); + assertEquals("example.com", request.header("host")); + assertEquals("example.com", request.header(":authority")); + assertEquals("abc", request.header("X-Trace")); + assertNull(request.remoteAddress()); + } + + private static void field(Http2Stream stream, String name, String value) { + byte[] nameBytes = name.getBytes(StandardCharsets.US_ASCII); + byte[] valueBytes = value.getBytes(StandardCharsets.US_ASCII); + PooledSlice nameView = new PooledSlice(); + PooledSlice valueView = new PooledSlice(); + nameView.reset(nameBytes, 0, nameBytes.length); + valueView.reset(valueBytes, 0, valueBytes.length); + stream.headerBlock().accept(nameView, valueView, false); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamLeakTest.java b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamLeakTest.java new file mode 100644 index 0000000..15d8449 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamLeakTest.java @@ -0,0 +1,20 @@ +package dev.relism.flash.http2.stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class Http2StreamLeakTest { + @Test + void oneHundredThousandAcquireReleaseCyclesReuseOneStream() { + Http2StreamTable table = new Http2StreamTable(100); + for (int i = 0; i < 100_000; i++) { + Http2Stream stream = table.acquire((i << 1) | 1); + table.remove(stream.id()); + table.release(stream); + } + assertEquals(1, table.createdCount()); + assertEquals(1, table.freeCount()); + assertEquals(0, table.size()); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamStateTest.java b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamStateTest.java new file mode 100644 index 0000000..284da45 --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamStateTest.java @@ -0,0 +1,32 @@ +package dev.relism.flash.http2.stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import dev.relism.flash.http2.Http2StreamException; +import org.junit.jupiter.api.Test; + +class Http2StreamStateTest { + @Test + void everyTransitionCellIsExecutableOrTypedError() { + for (Http2StreamState state : Http2StreamState.values()) { + for (Http2StreamState.Event event : Http2StreamState.Event.values()) { + if (Http2StreamState.isValid(state, event)) { + Http2StreamState next = state.transition(1, event); + assertEquals(true, next != null); + } else { + assertThrows(Http2StreamException.class, () -> state.transition(1, event)); + } + } + } + } + + @Test + void bodylessRequestAndResponseCloseStream() { + Http2StreamState state = + Http2StreamState.IDLE.transition(1, Http2StreamState.Event.RECV_HEADERS_ES); + assertEquals(Http2StreamState.HALF_CLOSED_REMOTE, state); + assertEquals( + Http2StreamState.CLOSED, state.transition(1, Http2StreamState.Event.SEND_HEADERS_ES)); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamTableTest.java b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamTableTest.java new file mode 100644 index 0000000..6cd9f3e --- /dev/null +++ b/flash/src/test/java/dev/relism/flash/http2/stream/Http2StreamTableTest.java @@ -0,0 +1,26 @@ +package dev.relism.flash.http2.stream; + +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import org.junit.jupiter.api.Test; + +class Http2StreamTableTest { + @Test + void insertLookupRemoveAtCapacityAndAcrossProbeClusters() { + Http2StreamTable table = new Http2StreamTable(8); + Http2Stream[] streams = new Http2Stream[8]; + for (int i = 0; i < streams.length; i++) { + streams[i] = table.acquire(i * 2 + 1); + assertSame(streams[i], table.get(i * 2 + 1)); + } + assertNull(table.acquire(99)); + for (int i = 0; i < streams.length; i += 2) { + assertSame(streams[i], table.remove(streams[i].id())); + table.release(streams[i]); + } + for (int i = 1; i < streams.length; i += 2) { + assertSame(streams[i], table.get(streams[i].id())); + } + } +}