diff --git a/docs/lanes/http2-contract.md b/docs/lanes/http2-contract.md new file mode 100644 index 0000000..6daea30 --- /dev/null +++ b/docs/lanes/http2-contract.md @@ -0,0 +1,425 @@ +# http2-contract — the afterlife of a terminated HTTP/2 stream + +Base `bccd379e18`, branch `http2-contract-fixes`, macOS 26.5 arm64, pinned +nightly-2026-08-20, 2026-09-16. Implementation, tests, h2spec and the local lint +ladder complete here. Linux, Windows, WASI 0.2/0.3, web, miri and loom are +**UNRUN here** (integrator CI). Nothing was benchmarked. + +Three defects, each proven by a runnable probe before anything was changed +(Perry's `docs/turnloop/http2-contract-probe.rs`, branch `turnloop/http2`), and +each now covered by a test that fails when its own fix is reverted. + +**One of the three briefs' premises was wrong and had to be measured out.** The +brief stated that Node answers a post-GOAWAY stream with +`RST_STREAM(REFUSED_STREAM)`; it sends no frame at all. See §2 — the correction, +the raw-socket measurement behind it, and why the fix is now "report and send +nothing" rather than a different hard-coded answer. + +**h2spec's strict suite — 147 tests, a required CI job — passed with all three +present, and passes unchanged now.** h2spec drives the subject from the peer +side only: every RST_STREAM in its §5.1 "Stream States" family is one *h2spec* +sends — `closed: Sends a DATA frame after sending RST_STREAM frame`, and so on — +never one the server decides to send, and the example server never closes +gracefully. So the afterlife of a stream the server itself terminated, and a +table filling with the slots those leave behind, are outside what it can reach. +§5.1.2 does fill the advertised concurrent-stream limit, with a hundred *live* +streams, and passes both before and after this change because its check accepts +a connection error as readily as a stream error. + +Everything below lives in that afterlife, which is also why §5.1's thirteen +cases pin the shape of the fix rather than obstruct it: the record for a stream +the *peer* reset is deliberately kept, so frames after it stay the error §5.1 +asks for, and a HEADERS on a numerically smaller remote id stays a connection +error for §5.1.1. + +## 1. A reset stream burned its table slot, and the connection then died + +`add_stream` recycles a closed stream's slot only when its `unreleased` credit +is zero. `reset` set `local_end`/`remote_end` and left `unreleased` alone, so a +stream reset while it still held DATA credit — which is exactly what a stream +error looks like — held its slot for the life of the connection. When the table +filled, `add_stream` returned `REFUSED_STREAM` *from inside `receive`*, which +set `failed` and emitted GOAWAY; `receive`'s error map had no case for +`REFUSED_STREAM`, so the peer was told PROTOCOL_ERROR and the session died. + +Measured on the base commit, a server table of 2 and 6 streams attempted: + +``` +-- server resets WITHOUT releasing capacity -- + server refused stream #2: CONNECTION ERROR REFUSED_STREAM + streams the server accepted: 2 +-- server releases capacity, THEN resets -- + streams the server accepted: 6 +``` + +`release_capacity`'s contract reads as "return window for data you have +consumed", and a stream you just reset is precisely the data you did *not* +consume, so the working host is the one that reasons about it wrongly. + +**Fix.** `reset` returns the stream's outstanding connection-level credit as a +connection WINDOW_UPDATE and drops the record, freeing the slot at once. RFC 9113 +§5.1 requires the credit back either way: DATA the peer sent before it saw the +RST_STREAM counts against the connection window whether or not it is delivered, +so leaving it burned the connection window as well as the slot — a second, +silent failure the probe did not reach. `h2_reset_returns_the_connection_window` +covers that half. + +Exceeding the concurrent-stream limit is now a **stream** error +(RFC 9113 §5.1.2): RST_STREAM(REFUSED_STREAM) on that stream, the connection +untouched. + +## 2. A stream opened after a graceful GOAWAY killed the connection + +After `shutdown()`, `receive`'s HEADERS arm rejected a new stream with +`protocol("invalid new stream")` — a **connection** error — because `draining` +was set. The race is unavoidable: a peer that opens a stream during a graceful +close cannot have seen the GOAWAY yet. The decision sat inside `receive_inner`, +so no host could choose otherwise. + +### The premise this lane was given was wrong, and it was measured out + +The brief for this work stated, as fact, that *"Node answers +`RST_STREAM(REFUSED_STREAM)` and keeps the session"*. This lane's first fix +implemented that, and an earlier draft of this document asserted it. **It is not +what Node does.** A sibling lane measured it, and rather than take that on +trust this lane measured it again independently, with a raw TCP peer and +hand-built frames — no HTTP/2 library anywhere in the client, so nothing under +test could shape the answer (`nodeprobe/` in this lane's scratch; Node 26.5.1, +`http2.createServer`, one stream held open so the session cannot finish +draining, `session.close()` for the graceful GOAWAY, then HEADERS on an id above +`lastStreamID`): + +``` +after /close: + <- HEADERS stream=3 flags=4 + <- DATA stream=3 flags=1 + <- GOAWAY stream=0 flags=0 last=3 code=0 +--> sending HEADERS on stream 5 (above lastStreamID 3) +Node's answer to the late stream: + (NO FRAMES AT ALL) +VERDICT: RST_STREAM for the late stream: NO +VERDICT: any frame at all naming stream 5: NO +``` + +Node sends **nothing**. The Node application never sees the stream (no `stream` +event fires). The control matters as much as the result: *no frames* could also +mean *stopped reading*, so the probe sends a PING afterwards — + +``` +--> control: PING after the late stream + <- PING stream=0 flags=1 +CONTROL: session alive and processing our frames: YES (PING ACK) +CONTROL: transport still open: YES +``` + +— so the session is alive and still servicing our frames. nghttp2 simply does +not process a HEADERS above the GOAWAY's `lastStreamID`, which is RFC 9113 +§6.8's "not processed": the peer is expected to retry it on a new connection. + +The ladder, then: a connection error (before) is clearly wrong; +`RST_STREAM(REFUSED_STREAM)` (this lane's first attempt) is closer but still not +Node; silence is Node. + +**Fix.** A stream that arrives while draining is reported as +`Event::Unprocessed { stream }` and **no frame is sent**. The header block is +still decoded and discarded, because HPACK is connection state. + +### Why the core sends nothing rather than choosing for the host + +This is the more important half, and it is the criticism that applies to this +lane's own first fix. The original defect was never "turnloop picks the wrong +answer" — it was *"the decision is inside `receive_inner`, so no host can choose +otherwise"*. Swapping one hard-coded answer for another leaves that intact. + +The two directions are not symmetrical. `output()`/`consume_output()` is a byte +stream: a host cannot retract a frame the core has already encoded into it +without parsing and splicing its own core's output, which is not a supported +operation and should never be one. Emitting is irreversible. Staying silent is +not: a host that wants to answer calls `reset(id, code)` and gets exactly one +RST_STREAM. So the permissive direction preserves every option and the emitting +direction destroys one — and it destroys precisely the option a host that must +match Node needs. + +That is also why `reset` now sends for a stream with no record, instead of +treating it as a no-op. Encoding what the host asks for is this type's job; +deciding not to is the host's. (Nothing depended on the old idempotence; a host +that resets twice now sends two frames, which is a host bug with a harmless +consequence. RFC 9113 §5.4.2's anti-looping rule is about answering a *received* +RST_STREAM, which this is not.) + +### A stream reported but not processed must not be named by GOAWAY + +Splitting the decision surfaced a second thing. `last_remote` was doing two +jobs: the id GOAWAY names, and the "have I seen this id" watermark that decides +whether a late frame is tolerated or a connection error. Those stopped being the +same quantity the moment a stream could be *reported but declined* — refused +past the limit, or arriving after our own GOAWAY. + +RFC 9113 §6.8 lets the peer retry every stream above `lastStreamID` on a new +connection. Naming a declined stream therefore tells the peer that request was +handled when it was not, and it is silently dropped rather than retried — in +exactly the graceful-close path this fix exists for. The field is now two: + +* `last_remote` — highest remote id accepted **for processing**; GOAWAY names it. +* `highest_remote` — highest remote id **seen**; the monotonicity check and + `seen()` use it, so frames the peer had in flight for a declined stream are + still tolerated. + +Both halves are pinned by `h2_goaway_names_the_last_processed_stream_not_the_last_seen`: +collapsing them one way makes GOAWAY over-claim, and the other way makes late +frames for a declined stream fatal again. + +### What is *not* copied from Node, and why + +REFUSED_STREAM's real trigger is `maxConcurrentStreams`, and this lane measured +both of Node's regimes with the same raw peer: + +| when the limit is exceeded | Node 26.5.1 sends | +|---|---| +| before the peer ACKs our SETTINGS | `RST_STREAM stream=5 code=7` (REFUSED_STREAM); the other streams live | +| after the peer ACKs our SETTINGS | `RST_STREAM code=2` then `GOAWAY code=2`, connection closed | + +turnloop implements the first and **not** the second. RFC 9113 §5.1.2 says a +HEADERS that causes the advertised limit to be exceeded *MUST* be treated as a +stream error of type STREAM_CLOSED or REFUSED_STREAM; Node's second regime is a +connection error instead. h2spec §5.1.2 exercises this case, so the RFC-shaped +answer is also the one the required gate checks. A host that wants Node's +stricter policy has `settings_awaiting_ack`-equivalent knowledge available to it +and can escalate; the core does not escalate on its behalf. + +## 3. `Step`'s zero cases are now a stated, tested contract + +`receive` returns `consumed == 0, event == None` (partial preface or partial +frame: **stop**) and `consumed > 0, event == None` (the preface itself, a +SETTINGS acknowledgement, PRIORITY, an unknown frame: **keep going**). Both are +normal and neither was written down. A host looping while "an event came back" +stalls at the *preface* — before a single frame is read, so the connection never +starts. + +**Fix.** A `# The Step contract` section in the crate root covering both +decoders in the crate, a per-decoder table on each `Step`, a doctest of the +correct loop, and `step_contract_rejects_both_wrong_loop_conditions`, which +asserts that each wrong condition actually breaks: + +* HTTP/2, "loop while an event came back": zero events, not one byte read. +* HTTP/1, "loop while input was consumed": `Event::End` reads no input + (PerryTS/turnloop#50), so the loop never sees the end of the message. + +The test also pins a third edge found while writing it: `receive` is **not +idempotent**. It advances decoder state by `consumed`, so a stalled loop that +retries the same input fails the connection (`FRAME_SIZE_ERROR`, the preface +re-read as a frame header). That is now in the contract. + +## Consequences the first two fixes forced, and which stand on their own + +Freeing a slot means the record is gone, and a refused stream never had one. +Frames the peer already had in flight then arrive for a stream with no record, +where `index()` produced a connection error. Three of those are races no host +can avoid, and RFC 9113 requires each to be tolerated: + +- **DATA for a gone stream** (§5.1): counted against the connection window and + the credit returned at once, since the payload is discarded. Without this, + fix 1 only moves the death of the connection one frame later — the ordinary + "reset a stream mid-body" case. +- **RST_STREAM for a gone stream**: crossed resets are normal; ignored. +- **WINDOW_UPDATE for a gone stream** (§6.9): ignored. +- **HEADERS for a stream *this endpoint itself opened* and has retired**: the + response a client aborted while it was already on the wire. The block is + decoded (HPACK is connection state) and discarded, no event and no second + RST_STREAM. This one was fatal before the change too — `finish_headers` + answered `STREAM_CLOSED` on a connection error — so it is a defect of the same + family rather than a consequence, and a client abort is the most ordinary + operation there is. Only a locally opened id is unambiguous here, which is why + the rule is not `seen(id)`: a *remote* id at or below `last_remote` may never + have existed, and h2spec §5.1.1 requires a HEADERS on one of those to stay a + connection error. + +A frame for a stream id this connection has *never used* is still a connection +error, which is what keeps h2spec's `5.1 idle` tests green. The distinction is +`seen(id)`: local ids below `next_id`, remote ids at or below `last_remote`. + +A stream the **peer** reset keeps its record (with its credit returned), so +frames after a peer RST_STREAM stay the error RFC 9113 §5.1 asks for and +h2spec's `closed` tests are unchanged. Only frames after *our own* RST_STREAM +are ignored, which is the asymmetry the RFC states. + +`release_capacity` for a terminated stream is now a **no-op, not an error**: +a host that buffers a body and releases when the application consumes it can +always call it, whatever happened to the stream meanwhile. Without that, fix 1 +would have turned a working host into a broken one, because slot recycling makes +the record disappear under it. + +## API + +- `Connection::unreleased(id) -> Option` — the credit the host still owes, + so the flow-control policy no longer has to mirror the core's counter by hand. +- `Connection::goaway(code, last_stream, opaque)` — the GOAWAY form `shutdown` + cannot express (`session.goaway(code, lastStreamID, opaqueData)` in Node). + `shutdown` is now `goaway(0, last_remote, &[])`; the bytes on the wire are + unchanged. Opaque data over the peer's `SETTINGS_MAX_FRAME_SIZE` is refused + here rather than sent for the peer to answer with `FRAME_SIZE_ERROR`. +- `Event::Headers` gains `kind: HeadersKind::{Head, Informational, Trailers}`. + The connection already enforced the distinction; the host had to keep its own + `received_head` to recover it. **Breaking** for an exhaustive pattern without + `..`; two in-crate sites and one test were updated. +- **Breaking, and worth calling out separately:** `Event` gained a variant, so an + exhaustive `match` on it needs one more arm. Perry's committed contract probe + stops compiling until it gets one — a one-line change, and a deliberate signal + rather than a silent behaviour shift. +- `Event::Unprocessed { stream }` — a stream the peer opened after a graceful + GOAWAY, for which **nothing has been sent**. The host decides whether to + answer. +- `Event::Reset` may now name a stream the host never saw opened — a + concurrent-stream-limit refusal is decided before any `Event::Headers` for + that stream exists. Documented on the variant. +- `Connection::reset` now sends for a stream with no record left (already reset, + or refused before one existed) rather than treating it as a no-op, so a host + can answer an `Event::Unprocessed`. An id this connection never used is still + an error. +- `send_headers`/`send_data` for a terminated stream report `STREAM_CLOSED` + rather than the idle/unknown protocol error. + +## Verification + +| check | result | +|---|---| +| `cargo test --workspace --all-features -- --test-threads=1` | 437 passed, 0 failed, 20 ignored (fixture-backed) | +| `python3 scripts/ci/h2spec.py` | **147 tests, 147 passed, 0 failed, 0 skipped** | +| sabotage: revert each fix in turn | each named test fails; see below | +| `cargo fmt --all --check`, clippy (default + all-features, all-targets) | clean | +| `RUSTDOCFLAGS=-D warnings cargo doc --all-features` | clean | +| `cargo +stable check`, `cargo +1.97.1 check` (MSRV) | clean | +| `check-paths.py`, `feature_modes.py`, `no-tokio.sh` | pass | + +A green test proves nothing if the code under it never ran, so each fix was +reverted in place and the suite re-run. Every row was re-measured against the +final tree: + +| reverted | tests that fail | +|---|---| +| `reset` returns no credit, frees no slot | `h2_reset_with_unreleased_data_keeps_its_table_slot`, `h2_reset_returns_the_connection_window`, `h2_release_after_termination_is_a_no_op`, `h2_late_frames_for_a_locally_reset_stream_are_ignored`, `h2_response_in_flight_when_the_client_aborts_is_ignored` | +| post-GOAWAY stream is a connection error | `h2_stream_after_graceful_goaway_is_unprocessed_and_unanswered` | +| post-GOAWAY stream answers RST_STREAM (the refuted premise) | `h2_stream_after_graceful_goaway_is_unprocessed_and_unanswered` | +| `reset` will not answer for a stream with no record | `h2_stream_after_graceful_goaway_is_unprocessed_and_unanswered` | +| GOAWAY names the last stream *seen* rather than processed | `h2_goaway_names_the_last_processed_stream_not_the_last_seen` | +| declined ids drop out of `seen()` | `h2_goaway_names_…`, `h2_stream_after_graceful_goaway_…` | +| stream limit is a connection error | `h2_stream_limit_refuses_one_stream_not_the_connection` | +| late DATA for a gone stream is fatal | `h2_late_frames_for_a_locally_reset_stream_are_ignored`, `h2_response_in_flight_when_the_client_aborts_is_ignored` | +| a client's in-flight response after its own reset is fatal | `h2_response_in_flight_when_the_client_aborts_is_ignored` | +| GOAWAY opaque data is not size-checked | `h2_goaway_carries_code_last_stream_and_opaque_data` | + +The first row is why `h2_reset_with_unreleased_data_keeps_its_table_slot` +asserts the *events* each stream produced and not merely that the connection +survived: with only the refusal fix in place, a burnt slot answers RST_STREAM +and the connection lives, so "6 connections survived" passes while "6 streams +were accepted" does not. + +And the artefact that specified the work — Perry's committed probe, not one line +of it changed — re-run against the fixed crate: + +``` +== gap 1: a reset stream's table slot == +-- server resets WITHOUT releasing capacity -- + streams the server accepted: 6 (was 2) +-- server releases capacity, THEN resets -- + streams the server accepted: 6 +== gap 3: a stream opened after a graceful GOAWAY == + server after late stream: ["Unprocessed s=1"] (was CONNECTION ERROR PROTOCOL_ERROR) + (and no "server emitted frame" line at all) +``` + +The probe prints its `server emitted frame kind=` line only when the server has +at least a frame header of output. Its absence is the probe independently +confirming that nothing was sent — which is what Node does. + +## One unrelated Windows defect, surfaced by CI + +`test-native (windows-2025, all-features)` failed on +`turnloop-http/interop::curl_and_node_h2_hundred_streams_against_native_server`: + +``` +thread '' panicked at protocols\turnloop-http\tests\interop.rs:374: +peer did not finish shutdown: An established connection was aborted by the +software in your host machine. (os error 10053) +``` + +Pre-existing, and not this branch's, established three ways rather than asserted: + +1. The server's wire bytes for that exact 100-stream exchange are **identical to + `main`'s** — 2955 bytes, same FNV-1a — replayed against both trees. +2. The panic is in the socket drain *after* `count == total` is asserted and the + Node client has printed `100 verified`. +3. Windows **default** and **executor** ran the same test in the same run, curl + leg included, and passed 9/9. Same code, different outcome — a race. + +The drain breaks on EOF and `ConnectionReset` but not `ConnectionAborted`, and +Windows reports the local end of a departed peer as WSAECONNABORTED. The crate +never sees this because it folds the two together everywhere — +`backend/iocp/socket.rs` maps `ERROR_CONNECTION_ABORTED` to `ConnectionReset`, +and `types.rs` and both WASI backends do the same. This test reads a raw +`std::net::TcpStream`, so it is the one place that gets the un-normalised kind. +Folded it the same way; re-running until the race went the other way would have +left the next lane to find it again. + +## Should the step contract be generalised to every sans-I/O decoder? + +Recommendation: **not as part of this**, and the reason is in the evidence. + +All three step-contract incidents — #50 (`http1::Decoder`'s zero-consume +`Event::End`, two lanes), #46 (`Event::Upgrade`'s mode asymmetry, P5) and this +one — are in `turnloop-http`, and both of its decoders are now covered by one +stated contract with tests that fail if it is violated. That is the whole +observed class. + +The workspace-wide version is a different, larger job, because the decoders do +not share a shape: + +| decoder | step type | zero cases | +|---|---|---| +| `http1::Decoder` | `Step { consumed, event }` | all four combinations occur | +| `http2::Connection` | `Step { consumed, event }` | an event always consumes | +| `compression::Decoder` | `DecodeStep { consumed, written, finished }` | documented; `decode` already errors on both-zero | +| `turnloop_websocket::Connection` | `Received { consumed, message }` | undocumented, same trap | +| postgres / mysql / redis / smtp / mongodb | `receive(&[u8]) -> Result<()>` plus a separate poll | consume the whole buffer; no step at all | + +Making that one contract means either a shared `Step` trait the five +protocol crates adopt (a breaking change across the workspace, with a +`turnloop-contract` conformance suite to make it *tested* rather than merely +stated), or a documentation convention that is not enforceable and will drift +the way this one did. Worth doing; worth scoping and reviewing as its own +change rather than riding along with a protocol fix. The one cheap piece that +does not need the design settled is `turnloop_websocket::Received`, which has +the identical two zero cases and no words about them. + +## What this cost, and the lesson that is not about HTTP/2 + +The premise that Node answers `RST_STREAM(REFUSED_STREAM)` after a graceful +GOAWAY travelled through three briefs and was implemented once before anyone put +a socket against Node. It is the same failure mode as the h2spec blind spot in +§0, seen from the other side: h2spec passed with four defects present because it +only drives from the peer side, and a stated behaviour survived three hand-offs +because nobody drove the peer at all. Both are cases of a claim nobody could +falsify from where they were standing. + +The cheap guard is the one this lane used twice by accident and once on purpose: +**put a raw socket against the reference implementation, and include a control +that distinguishes "it did nothing" from "it stopped listening".** The PING after +the late stream is what turns "no frames" from a guess into a measurement. + +## Not done + +- Server push: `turnloop_http::http2` rejects PUSH_PROMISE outright and Perry + does not implement `createPushResponse`; unchanged. +- **Trailers pipelined onto a stream the server refused.** A peer that sends + HEADERS + DATA + trailer HEADERS in one burst onto a stream refused for the + GOAWAY race or the stream limit still fails the connection on the trailer + block, because a *remote* id at or below `last_remote` cannot be distinguished + from one that was never opened, and h2spec §5.1.1 depends on that staying an + error. The DATA in the same burst is tolerated, so the behaviour is + inconsistent; making it precise needs the refused ids remembered, which is + state this change deliberately does not add. Unchanged from before. +- Rapid-reset (CVE-2023-44487) accounting. A refused stream costs an HPACK + decode and one RST_STREAM, the same as before for an accepted-then-reset + stream, so this change neither adds nor removes that exposure — but the crate + has no reset-rate bound, and now that a refusal is survivable it is worth one. +- Nothing ran on Linux, Windows or WASI here, and nothing was benchmarked. diff --git a/protocols/turnloop-http/src/asynchronous/client.rs b/protocols/turnloop-http/src/asynchronous/client.rs index 78fbf92..1ebe129 100644 --- a/protocols/turnloop-http/src/asynchronous/client.rs +++ b/protocols/turnloop-http/src/asynchronous/client.rs @@ -399,6 +399,7 @@ impl Client { stream, headers, end_stream, + .. } if stream == id => { if response.is_none() { let status = headers @@ -488,6 +489,7 @@ impl Client { stream, headers, end_stream, + .. } if stream == id => { if response.is_none() { let status = headers diff --git a/protocols/turnloop-http/src/http1.rs b/protocols/turnloop-http/src/http1.rs index ccbd5cc..737ced9 100644 --- a/protocols/turnloop-http/src/http1.rs +++ b/protocols/turnloop-http/src/http1.rs @@ -82,6 +82,20 @@ pub enum Event<'a> { End, Upgrade, } +/// One decode step: how much of `input` was consumed, and the event it +/// produced, if any. See the [step contract](crate#the-step-contract). +/// +/// **`consumed` and `event` are independent.** All four shapes occur: +/// +/// | `consumed` | `event` | meaning | +/// |---|---|---| +/// | `0` | `None` | **stop.** More input is needed - or the message is over and the decoder is done; the step does not say which. | +/// | `> 0` | `None` | **keep going.** A chunk-size line, a chunk CRLF or an empty trailer block: progress with nothing for the host. | +/// | `> 0` | `Some` | an event. | +/// | `0` | `Some` | an event that reads no input: [`Event::End`] and [`Event::Upgrade`] both arrive this way. | +/// +/// So the loop condition is `consumed > 0 || event.is_some()`, and a host that +/// stops on either zero alone is wrong in one of the two directions. #[derive(Debug)] pub struct Step<'a> { pub consumed: usize, diff --git a/protocols/turnloop-http/src/http2.rs b/protocols/turnloop-http/src/http2.rs index 873d69e..8000085 100644 --- a/protocols/turnloop-http/src/http2.rs +++ b/protocols/turnloop-http/src/http2.rs @@ -82,6 +82,18 @@ pub fn encode_frame( out.extend_from_slice(payload); Ok(()) } +/// Which of the three header blocks an [`Event::Headers`] carries. The +/// connection already enforces the distinction, so a host does not have to +/// track `received_head` itself to tell them apart. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HeadersKind { + /// The request head, or a final (>= 200) response head. + Head, + /// A 1xx response head. It never ends the stream, and a final head follows. + Informational, + /// A trailer block. It always ends the stream. + Trailers, +} #[derive(Debug)] pub enum Event<'a> { Settings, @@ -89,16 +101,38 @@ pub enum Event<'a> { stream: u32, headers: Vec
, end_stream: bool, + kind: HeadersKind, }, Data { stream: u32, bytes: &'a [u8], end_stream: bool, }, + /// The stream is over. It may name a stream the host never saw opened: a + /// stream past the concurrent-stream limit is refused with + /// RST_STREAM(REFUSED_STREAM) before any [`Event::Headers`] for it is + /// produced (RFC 9113 section 5.1.2 requires a stream error there, and + /// Node answers the same way — measured). Reset { stream: u32, code: u32, }, + /// The peer opened a stream after a graceful GOAWAY, above the last stream + /// id that GOAWAY named, and **nothing has been sent for it**. RFC 9113 + /// section 6.8 makes such a stream simply "not processed" — the peer is + /// expected to retry it on a new connection — and Node emits no frame at + /// all here (measured against Node 26.5.1 with a raw peer: no RST_STREAM, + /// no GOAWAY, the session stays alive and the request never reaches the + /// application). + /// + /// The header block was decoded and discarded, because HPACK is connection + /// state. The connection deliberately does not answer for the host: a frame + /// it emitted could not be un-emitted, so a host that must match Node would + /// have no way back. A host that wants to answer may call + /// [`Connection::reset`]. + Unprocessed { + stream: u32, + }, Goaway { last_stream: u32, code: u32, @@ -111,16 +145,71 @@ pub enum Event<'a> { stream: u32, }, } +/// One decode step: how much of `input` was consumed, and the event it +/// produced, if any. See the [step contract](crate#the-step-contract). +/// +/// **`consumed` and `event` are independent.** All three shapes are normal: +/// +/// | `consumed` | `event` | meaning | +/// |---|---|---| +/// | `0` | `None` | **stop.** A partial preface or a partial frame; read more input before calling again. | +/// | `> 0` | `None` | **keep going.** Progress with nothing for the host: the client preface, a SETTINGS acknowledgement, PRIORITY, an unknown frame type, or a frame the peer had in flight for a stream that is already gone. | +/// | `> 0` | `Some` | an event. In HTTP/2 an event always consumes; `consumed == 0` is only ever the stop case. | +/// +/// So the loop condition is `consumed > 0 || event.is_some()`: +/// +/// ``` +/// # use turnloop_http::http2::{Connection, Limits, Role, PREFACE}; +/// # let mut connection = Connection::new(Role::Server, Limits::default())?; +/// # let mut input = PREFACE.to_vec(); +/// loop { +/// let step = connection.receive(&input)?; +/// let consumed = step.consumed; +/// let progressed = consumed > 0 || step.event.is_some(); +/// if let Some(event) = step.event { +/// // DATA borrows `input`, so handle the event before draining. +/// let _ = event; +/// } +/// input.drain(..consumed); +/// if !progressed { +/// break; // read more bytes from the transport, then continue +/// } +/// } +/// # Ok::<(), turnloop_http::Error>(()) +/// ``` +/// +/// Looping while `event.is_some()` instead **stalls at the client preface** — +/// the first step of every server connection, before a single frame is read. +/// Looping while input remains spins forever on a partial frame. +#[derive(Debug)] pub struct Step<'a> { pub consumed: usize, pub event: Option>, } +/// What to do with a header block that does not belong to a live stream. Every +/// arm decodes the block first: HPACK is connection state, and a block that is +/// skipped desynchronises every block after it. +#[derive(Debug, Clone, Copy)] +enum Disposal { + /// Past the concurrent-stream limit: answer RST_STREAM and report it. + /// RFC 9113 section 5.1.2 requires a stream error here. + Refuse(u32), + /// Opened after our graceful GOAWAY: send nothing, report it, and let the + /// host decide. See [`Event::Unprocessed`]. + Unprocessed, + /// Frames the peer had in flight for a stream that is gone: discard in + /// silence. The host already knows - it is the side that ended the stream. + Discard, +} #[derive(Debug)] struct Stream { id: u32, send_window: i64, recv_window: i64, unreleased: u32, + /// The peer reset this stream and its remaining connection credit was + /// returned in bulk, so a late `release_capacity` for it is a no-op. + credited: bool, local_end: bool, remote_end: bool, received_head: bool, @@ -143,7 +232,16 @@ pub struct Connection { limits: Limits, streams: Vec, next_id: u32, + /// Highest remote stream id this connection has accepted **for processing**. + /// GOAWAY names it, so a stream that was reported but not processed - one + /// refused past the stream limit, or arriving after our own GOAWAY - must + /// not advance it: RFC 9113 section 6.8 lets the peer retry everything above + /// it on a new connection, and claiming one is processed loses that request. last_remote: u32, + /// Highest remote stream id seen on the wire, processed or not. The + /// monotonicity check and [`Connection::seen`] use this, so frames the peer + /// had in flight for a stream we declined are still tolerated. + highest_remote: u32, send_window: i64, recv_window: i64, initial_send: i64, @@ -157,6 +255,10 @@ pub struct Connection { block: Vec, scratch: Vec, continuation: Option<(u32, bool, usize)>, + /// What to do with the header block being assembled, when it does not + /// belong to a live stream. HPACK is connection state, so such a block + /// still has to advance the decoder before it is thrown away. + disposal: Option, output: Vec, output_pos: usize, draining: bool, @@ -177,6 +279,7 @@ impl Connection { streams: Vec::with_capacity(limits.streams), next_id: if role == Role::Client { 1 } else { 2 }, last_remote: 0, + highest_remote: 0, send_window: 65535, recv_window: 65535, initial_send: 65535, @@ -190,6 +293,7 @@ impl Connection { block: Vec::new(), scratch: Vec::new(), continuation: None, + disposal: None, output: Vec::new(), output_pos: 0, draining: false, @@ -281,6 +385,55 @@ impl Connection { fn active(&self) -> usize { self.streams.iter().filter(|s| !s.closed()).count() } + /// A stream id this connection has already opened, refused or finished + /// with. Its record may be gone — dropped by `reset`, never created for a + /// refused stream, or recycled — while the peer, not yet having seen the + /// RST_STREAM or GOAWAY, still sends frames for it. RFC 9113 section 5.1 + /// requires those to be tolerated; a frame for an id that was never used is + /// still a connection error. + fn seen(&self, id: u32) -> bool { + if id == 0 { + false + } else if (id % 2 == 1) == (self.role == Role::Client) { + id < self.next_id + } else { + id <= self.highest_remote + } + } + /// `STREAM_CLOSED` for a stream this connection has finished with, and the + /// idle/unknown protocol error for one it has never used. + fn gone(&self, id: u32) -> Error { + if self.seen(id) { + error("STREAM_CLOSED", "stream already terminated") + } else { + protocol("unknown or idle stream") + } + } + /// Give back the connection-level credit a terminated stream still holds. + /// RFC 9113 section 5.1: DATA the peer sent before it saw our RST_STREAM + /// counts against the connection window either way, so the credit has to go + /// back or the window shrinks for the life of the connection. No + /// stream-level WINDOW_UPDATE: that stream is over. + fn return_credit(&mut self, i: usize) -> Result<()> { + let n = std::mem::take(&mut self.streams[i].unreleased); + if n == 0 { + return Ok(()); + } + self.recv_window += n as i64; + self.frame(8, 0, 0, &n.to_be_bytes()) + } + /// Account DATA for a stream that is gone against the connection window and + /// return the credit at once: the payload is discarded, so no host release + /// will ever arrive for it. + fn discard_data(&mut self, flow: i64) -> Result<()> { + if flow > self.recv_window { + return Err(error("FLOW_CONTROL_ERROR", "receive window exceeded")); + } + if flow == 0 { + return Ok(()); + } + self.frame(8, 0, 0, &(flow as u32).to_be_bytes()) + } fn add_stream(&mut self, id: u32) -> Result { if self.active() >= self.limits.streams { return Err(error("REFUSED_STREAM", "stream limit")); @@ -290,6 +443,7 @@ impl Connection { send_window: self.initial_send, recv_window: 65535, unreleased: 0, + credited: false, local_end: false, remote_end: false, received_head: false, @@ -342,7 +496,7 @@ impl Connection { if self.failed { return Err(protocol("failed connection")); } - let i = self.index(id)?; + let i = self.index(id).map_err(|_| self.gone(id))?; let s = &self.streams[i]; if s.local_end { return Err(error("STREAM_CLOSED", "local stream closed")); @@ -403,7 +557,7 @@ impl Connection { } /// Returns accepted bytes (zero on flow-control stall). Host retains remainder. pub fn send_data(&mut self, id: u32, bytes: &[u8], end_stream: bool) -> Result { - let i = self.index(id)?; + let i = self.index(id).map_err(|_| self.gone(id))?; let s = &self.streams[i]; if self.failed || s.local_end || !s.sent_head { return Err(error("STREAM_CLOSED", "cannot send DATA")); @@ -433,9 +587,23 @@ impl Connection { s.local_end = end; Ok(n) } + /// Return `n` bytes of a stream's receive window to the peer; `n` may not + /// exceed that stream's [`unreleased`](Self::unreleased) count. + /// + /// A terminated stream has already had its remaining credit returned in + /// bulk, so a late release for one is a **no-op, not an error** — a host + /// that buffers a body and releases when the application consumes it can + /// always call this, whatever happened to the stream meanwhile. An id this + /// connection has never used is still an error. pub fn release_capacity(&mut self, id: u32, n: u32) -> Result<()> { - let i = self.index(id)?; - if n == 0 { + let Ok(i) = self.index(id) else { + return if self.seen(id) { + Ok(()) + } else { + Err(protocol("unknown or idle stream")) + }; + }; + if n == 0 || self.streams[i].credited { return Ok(()); } let s = &mut self.streams[i]; @@ -451,20 +619,67 @@ impl Connection { } Ok(()) } + /// Received DATA bytes for `id` whose flow-control credit the host has not + /// yet returned with [`release_capacity`](Self::release_capacity). `None` + /// once the stream is gone, its credit having been returned in bulk. + /// Padding never appears here: `receive` releases it as it is consumed. + pub fn unreleased(&self, id: u32) -> Option { + self.index(id).ok().map(|i| self.streams[i].unreleased) + } + /// Terminate a stream with RST_STREAM. + /// + /// The stream's outstanding connection-level credit is returned and its + /// table slot is freed at once, so resetting a stream that still holds + /// unreleased DATA — the ordinary "one stream errors while its siblings are + /// live" case — costs the connection nothing. Frames the peer already had + /// in flight for the stream are then ignored rather than failing the + /// connection (RFC 9113 section 5.1). + /// + /// The frame is sent even when the stream has no record left — it may have + /// been reset already, or refused before one existed, as after + /// [`Event::Unprocessed`]. Encoding what the host asks for is this type's + /// job; deciding not to is the host's. Only an id this connection has never + /// used is an error. pub fn reset(&mut self, id: u32, code: u32) -> Result<()> { - let i = self.index(id)?; - self.streams[i].local_end = true; - self.streams[i].remote_end = true; + let Ok(i) = self.index(id) else { + return if self.seen(id) { + self.frame(3, 0, id, &code.to_be_bytes()) + } else { + Err(protocol("unknown or idle stream")) + }; + }; + self.return_credit(i)?; + self.streams.swap_remove(i); self.frame(3, 0, id, &code.to_be_bytes()) } pub fn ping(&mut self, data: [u8; 8]) -> Result<()> { self.frame(6, 0, 0, &data) } + /// Graceful close: GOAWAY(NO_ERROR) naming the last stream this connection + /// has processed, then drain. A stream the peer opens before it sees the + /// frame does not fail the connection and is not answered: it arrives as + /// [`Event::Unprocessed`] and the host decides. pub fn shutdown(&mut self) -> Result<()> { + self.goaway(0, self.last_remote, &[]) + } + /// GOAWAY with an explicit error code, last stream id and opaque debug + /// data, the form [`shutdown`](Self::shutdown) cannot express. The + /// connection starts draining either way. + pub fn goaway(&mut self, code: u32, last_stream: u32, opaque: &[u8]) -> Result<()> { + if last_stream > 0x7fffffff { + return Err(protocol("invalid last stream")); + } + // The peer would answer a GOAWAY over its own SETTINGS_MAX_FRAME_SIZE + // with FRAME_SIZE_ERROR, so refuse the opaque data here instead. + if 8 + opaque.len() > self.peer_frame { + return Err(frame_error()); + } self.draining = true; - let mut bytes = [0; 8]; - bytes[..4].copy_from_slice(&self.last_remote.to_be_bytes()); - self.frame(7, 0, 0, &bytes) + self.scratch.clear(); + self.scratch.extend_from_slice(&last_stream.to_be_bytes()); + self.scratch.extend_from_slice(&code.to_be_bytes()); + self.scratch.extend_from_slice(opaque); + encode_frame(7, 0, 0, &self.scratch, &mut self.output) } pub fn is_drained(&self) -> bool { self.draining && self.active() == 0 && self.output().is_empty() @@ -533,7 +748,16 @@ impl Connection { if f.stream == 0 { return Err(protocol("DATA on connection")); } - let i = self.index(f.stream)?; + let Ok(i) = self.index(f.stream) else { + // In flight when the stream was refused or reset. RFC 9113 + // section 5.1 tolerates it, and the bytes still count + // against the connection window, so return the credit. + if !self.seen(f.stream) { + return Err(protocol("unknown or idle stream")); + } + self.discard_data(f.payload.len() as i64)?; + return Ok(step); + }; let s = &mut self.streams[i]; if s.remote_end || !s.received_head { return Err(protocol("DATA in invalid stream state")); @@ -572,16 +796,48 @@ impl Connection { if f.stream == 0 { return Err(protocol("HEADERS on connection")); } + self.disposal = None; if self.index(f.stream).is_err() { - if self.role != Role::Server - || f.stream % 2 == 0 - || f.stream <= self.last_remote - || self.draining - { + if self.role != Role::Server { + // A response the peer already had in flight when we + // reset the stream. Its record is gone, so decode the + // block to keep HPACK in step and ignore it (RFC 9113 + // section 5.1). Only a stream we opened ourselves is + // unambiguous here; anything else is idle. + if !self.seen(f.stream) { + return Err(protocol("invalid new stream")); + } + self.disposal = Some(Disposal::Discard); + } else if f.stream % 2 == 0 || f.stream <= self.highest_remote { return Err(protocol("invalid new stream")); + } else { + self.highest_remote = f.stream; + // A stream this connection cannot accept never fails + // the connection. The two reasons answer differently, + // and both were measured against Node 26.5.1 with a raw + // peer rather than assumed: + // + // * after a graceful GOAWAY, the stream is above the id + // that GOAWAY named and is simply not processed (RFC + // 9113 section 6.8). Node sends no frame at all, so + // neither do we - the host is told and decides. + // * past the concurrent-stream limit, RFC 9113 section + // 5.1.2 requires a stream error, and Node answers + // RST_STREAM(REFUSED_STREAM). + self.disposal = if self.draining { + Some(Disposal::Unprocessed) + } else { + match self.add_stream(f.stream) { + Ok(_) => { + // Accepted for processing, so GOAWAY may name it. + self.last_remote = f.stream; + None + } + Err(e) if e.code == "REFUSED_STREAM" => Some(Disposal::Refuse(7)), + Err(e) => return Err(e), + } + }; } - self.last_remote = f.stream; - self.add_stream(f.stream)?; } let mut payload = unpadded(f)?; if f.flags & 32 != 0 { @@ -597,7 +853,7 @@ impl Connection { self.append_block(payload)?; let end = f.flags & 1 != 0; if f.flags & 4 != 0 { - step.event = Some(self.finish_headers(f.stream, end)?); + step.event = self.finish_block(f.stream, end)?; } else { self.continuation = Some((f.stream, end, 0)); } @@ -610,7 +866,7 @@ impl Connection { self.append_block(f.payload)?; if f.flags & 4 != 0 { self.continuation = None; - step.event = Some(self.finish_headers(id, end)?); + step.event = self.finish_block(id, end)?; } else { self.continuation = Some((id, end, count + 1)); } @@ -633,9 +889,18 @@ impl Connection { if f.payload.len() != 4 { return Err(frame_error()); } - let i = self.index(f.stream)?; - self.streams[i].remote_end = true; - self.streams[i].local_end = true; + let Ok(i) = self.index(f.stream) else { + // Crossed resets, or a reset for a stream already refused. + if !self.seen(f.stream) { + return Err(protocol("unknown or idle stream")); + } + return Ok(step); + }; + self.return_credit(i)?; + let s = &mut self.streams[i]; + s.credited = true; + s.remote_end = true; + s.local_end = true; step.event = Some(Event::Reset { stream: f.stream, code: u32be(f.payload), @@ -730,6 +995,14 @@ impl Connection { if f.payload.len() != 4 { return Err(frame_error()); } + if f.stream != 0 && self.index(f.stream).is_err() { + // Credit for a stream that is already gone. RFC 9113 + // section 6.9 requires this to be tolerated. + if !self.seen(f.stream) { + return Err(protocol("unknown or idle stream")); + } + return Ok(step); + } let n = (u32be(f.payload) & 0x7fffffff) as i64; if n == 0 { if f.stream != 0 { @@ -773,6 +1046,24 @@ impl Connection { self.block.extend_from_slice(bytes); Ok(()) } + /// Complete a header block: the stream's own, or one being thrown away. + fn finish_block(&mut self, id: u32, end: bool) -> Result>> { + let Some(disposal) = self.disposal.take() else { + return self.finish_headers(id, end).map(Some); + }; + // Decode before discarding: HPACK state belongs to the connection, not + // the stream, and a skipped block desynchronises every block after it. + let mut headers = Vec::new(); + self.decoder.decode(&self.block, &mut headers)?; + match disposal { + Disposal::Refuse(code) => { + self.frame(3, 0, id, &code.to_be_bytes())?; + Ok(Some(Event::Reset { stream: id, code })) + } + Disposal::Unprocessed => Ok(Some(Event::Unprocessed { stream: id })), + Disposal::Discard => Ok(None), + } + } fn finish_headers(&mut self, id: u32, end: bool) -> Result> { let mut headers = Vec::new(); self.decoder.decode(&self.block, &mut headers)?; @@ -792,6 +1083,13 @@ impl Connection { if s.received_head && !end { return Err(protocol("trailers without END_STREAM")); } + let kind = if informational { + HeadersKind::Informational + } else if s.received_head { + HeadersKind::Trailers + } else { + HeadersKind::Head + }; if !s.received_head && !informational { s.recv_no_body = self.role == Role::Client && (s.head_request @@ -818,6 +1116,7 @@ impl Connection { stream: id, headers, end_stream: end, + kind, }) } } diff --git a/protocols/turnloop-http/src/lib.rs b/protocols/turnloop-http/src/lib.rs index c3c3116..9fec9e3 100644 --- a/protocols/turnloop-http/src/lib.rs +++ b/protocols/turnloop-http/src/lib.rs @@ -5,6 +5,50 @@ //! host owns `LocalExecutor` and calls `turn`; adapters await its streams and //! deadline futures. See the crate README and turnloop-io for ownership, streaming //! and cancellation examples. Default features retain the sans-I/O API. +//! +//! # The `Step` contract +//! +//! Both decoders here - [`http1::Decoder`] and [`http2::Connection`] - are +//! driven the same way: call `receive` with everything you have, act on the step +//! it returns, drain `consumed`, repeat. **`consumed` and `event` are +//! independent, and a host has to look at both.** The rule is one line: +//! +//! ```text +//! progressed = step.consumed > 0 || step.event.is_some() +//! ``` +//! +//! Call `receive` again while `progressed` is true. When it is false, read more +//! bytes from the transport - or declare the end of input with `eof` - before +//! calling again. Each half of that condition has cost a host a debugging cycle, +//! so neither is optional: +//! +//! * **`consumed > 0` with no event** is progress with nothing to hand up: the +//! HTTP/2 client preface, a SETTINGS acknowledgement, a PRIORITY frame, an +//! unknown frame type, or a frame the peer had in flight for a stream that is +//! already gone; an HTTP/1 chunk-size line, chunk CRLF or empty trailer +//! block. A loop that continues only while an event came back stalls here, and +//! for HTTP/2 the first such step is the preface - so the connection never +//! starts at all. +//! * **`consumed == 0` with an event** is an event that reads no input. +//! HTTP/1's [`http1::Event::End`] and [`http1::Event::Upgrade`] both arrive +//! this way. A loop that continues only while input was consumed drops the end +//! of every message. HTTP/2 has no step of this shape: there an event always +//! consumes, and `consumed == 0` is always the stop case. +//! * **`consumed == 0` with no event** is the only stop condition, and it is +//! returned whether the decoder needs more bytes or is finished for good. The +//! two are not distinguishable from the step: a host that has seen +//! [`http1::Event::End`] must remember it (or ask +//! [`http1::Decoder::reusable`]), and an HTTP/2 host closes on +//! [`http2::Connection::is_drained`]. +//! +//! `receive` is not idempotent: it advances decoder state by `consumed`, so the +//! host must drain exactly that many bytes before calling again. Feeding the +//! same input back - which is what a stalled loop does when it retries - fails +//! the connection. +//! +//! `asynchronous::Http1::event` and `asynchronous::Http2::event` are the +//! reference drivers, and `Step`'s own documentation carries the per-decoder +//! table. #![deny(unsafe_op_in_unsafe_fn)] pub mod client; pub mod compression; diff --git a/protocols/turnloop-http/tests/codecs.rs b/protocols/turnloop-http/tests/codecs.rs index a2153e5..d60f719 100644 --- a/protocols/turnloop-http/tests/codecs.rs +++ b/protocols/turnloop-http/tests/codecs.rs @@ -721,3 +721,591 @@ fn proxy_credentials_and_cross_origin_mixed_case_headers() { .is_none() ); } + +// --- HTTP/2 stream-lifetime contract ----------------------------------------- +// +// h2spec's strict suite drives the subject from the peer side only: every +// RST_STREAM in its 5.1 "Stream States" family is one h2spec sends, never one +// the server decides to send, and its example server never closes gracefully. +// The afterlife of a stream the server itself terminated is outside what it can +// reach, and that is where every test below lives. + +/// Drive `to` until it stops making progress, returning the events it produced. +/// The loop condition is the documented one: `consumed > 0 || event.is_some()`. +fn drive(to: &mut http2::Connection, input: &mut Vec) -> Result, &'static str> { + let mut seen = Vec::new(); + loop { + let step = match to.receive(input) { + Ok(step) => step, + Err(e) => return Err(e.code), + }; + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + if let Some(event) = step.event { + seen.push(match event { + http2::Event::Settings => "Settings".to_string(), + http2::Event::Headers { stream, .. } => format!("Headers s={stream}"), + http2::Event::Data { stream, bytes, .. } => { + format!("Data s={stream} n={}", bytes.len()) + } + http2::Event::Reset { stream, code } => format!("Reset s={stream} code={code}"), + http2::Event::Unprocessed { stream } => format!("Unprocessed s={stream}"), + http2::Event::Goaway { code, .. } => format!("Goaway code={code}"), + http2::Event::Ping { ack, .. } => format!("Ping ack={ack}"), + http2::Event::WindowUpdate { stream } => format!("WindowUpdate s={stream}"), + }); + } + input.drain(..consumed); + if !progressed { + return Ok(seen); + } + } +} +fn ship(from: &mut http2::Connection) -> Vec { + let wire = from.output().to_vec(); + from.consume_output(wire.len()).unwrap(); + wire +} +fn post_headers(path: &str) -> Vec
{ + vec![ + Header::new(":method", "POST"), + Header::new(":scheme", "http"), + Header::new(":path", path), + Header::new(":authority", "localhost"), + ] +} +/// A handshaked client/server pair. The server's table holds `streams` streams; +/// the client's is large, so only the server's is ever under test. +fn handshake(streams: usize) -> (http2::Connection, http2::Connection) { + let mut server = http2::Connection::new( + http2::Role::Server, + http2::Limits { + streams, + ..Default::default() + }, + ) + .unwrap(); + let mut client = http2::Connection::new(http2::Role::Client, http2::Limits::default()).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + let mut wire = ship(&mut server); + drive(&mut client, &mut wire).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + (client, server) +} + +/// A stream reset while it still holds unreleased DATA must not burn its table +/// slot. `add_stream` only recycles a closed slot whose credit has been +/// returned, so a `reset` that left `unreleased` behind leaked the slot for the +/// life of the connection — and when the table filled, the REFUSED_STREAM came +/// out of `receive` as a *connection* error and the session died. +#[test] +fn h2_reset_with_unreleased_data_keeps_its_table_slot() { + for release_first in [true, false] { + let (mut client, mut server) = handshake(2); + let mut accepted = 0; + for i in 0..6 { + let path = format!("/{i}"); + let id = client + .open(&post_headers(&path), false) + .unwrap_or_else(|e| panic!("client refused open #{i}: {}", e.code)); + assert_eq!(client.send_data(id, b"hello-body", true).unwrap(), 10); + let mut wire = ship(&mut client); + match drive(&mut server, &mut wire) { + // Accepted means the request arrived - not merely that the + // connection survived. A burnt table slot answers RST_STREAM. + Ok(events) => { + assert_eq!( + events, + vec![format!("Headers s={id}"), format!("Data s={id} n=10")], + "stream #{i} was refused (release_first={release_first})" + ); + accepted += 1; + } + Err(code) => panic!( + "server failed the connection on stream #{i} (release_first={release_first}): {code}" + ), + } + if release_first { + // The host that already knows about finding 1 and works around it. + assert_eq!(server.unreleased(id), Some(10)); + server.release_capacity(id, 10).unwrap(); + } + assert!(server.reset(id, 8).is_ok()); + // Ship RST_STREAM and any WINDOW_UPDATEs back so the client retires + // its own record: the client's peer-stream limit must not be what + // refuses the next open. + let mut back = ship(&mut server); + drive(&mut client, &mut back).unwrap(); + } + assert_eq!(accepted, 6, "release_first={release_first}"); + } +} + +/// A reset returns the stream's outstanding connection-level credit. Without +/// it the connection window shrinks by every un-released byte of every reset +/// stream until the peer can no longer send at all. +#[test] +fn h2_reset_returns_the_connection_window() { + let (mut client, mut server) = handshake(100); + let body = vec![7u8; 16384]; + let mut charged = 0; + for i in 0..4 { + let id = client.open(&post_headers(&format!("/{i}")), false).unwrap(); + assert_eq!(client.send_data(id, &body, false).unwrap(), 16384); + charged += 16384; + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + server.reset(id, 8).unwrap(); + let mut back = ship(&mut server); + drive(&mut client, &mut back).unwrap(); + } + // Every charged byte came back as connection credit, so the client can + // still fill a fresh 65535-byte connection window. + assert!(charged > 0); + let id = client.open(&post_headers("/last"), false).unwrap(); + let mut sent = 0; + while sent < 65535 { + let n = client + .send_data(id, &body[..(65535 - sent).min(16384)], false) + .unwrap(); + assert_ne!(n, 0, "connection window shrank by {charged} reset bytes"); + sent += n; + } + assert_eq!(sent, 65535); +} + +/// Frames the peer had in flight when we reset a stream arrive after the +/// RST_STREAM. RFC 9113 §5.1: the endpoint that *sent* RST_STREAM must be +/// prepared to receive them, and may ignore them. Failing the connection here +/// is a race no host can avoid. +#[test] +fn h2_late_frames_for_a_locally_reset_stream_are_ignored() { + let (mut client, mut server) = handshake(100); + let id = client.open(&post_headers("/x"), false).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + server.reset(id, 8).unwrap(); + let _ = ship(&mut server); + // The client has not seen the RST_STREAM yet. + assert_eq!(client.send_data(id, b"in-flight", false).unwrap(), 9); + client.reset(id, 8).unwrap(); + let mut late = ship(&mut client); + let events = drive(&mut server, &mut late).expect("late frames must not fail the connection"); + assert_eq!(events, Vec::::new()); + // The connection is still usable. + let next = client.open(&post_headers("/y"), true).unwrap(); + let mut wire = ship(&mut client); + assert_eq!( + drive(&mut server, &mut wire).unwrap(), + vec![format!("Headers s={next}")] + ); +} + +/// A client that aborts a request resets the stream while the server's response +/// is already on the wire. Those HEADERS and DATA arrive for a stream whose +/// record is gone, and used to end the whole session — on the most ordinary +/// client operation there is. +#[test] +fn h2_response_in_flight_when_the_client_aborts_is_ignored() { + let (mut client, mut server) = handshake(100); + let first = client.open(&post_headers("/abort"), true).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + // The server answers; the client aborts before the answer lands. + server + .send_headers(first, &[Header::new(":status", "200")], false) + .unwrap(); + server.send_data(first, b"payload", true).unwrap(); + client.reset(first, 8).unwrap(); + let _ = ship(&mut client); + let mut answer = ship(&mut server); + assert_eq!( + drive(&mut client, &mut answer).expect("an aborted request must not fail the session"), + Vec::::new() + ); + // HPACK survived the discarded block, so the next request still decodes. + let second = client.open(&post_headers("/next"), true).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + server + .send_headers(second, &[Header::new(":status", "201")], true) + .unwrap(); + let mut answer = ship(&mut server); + assert_eq!( + drive(&mut client, &mut answer).unwrap(), + vec![format!("Headers s={second}")] + ); + // A stream the client never opened is still a connection error. + let mut block = Vec::new(); + hpack::Encoder::new(4096).encode(&[Header::new(":status", "200")], &mut block); + let mut wire = Vec::new(); + http2::encode_frame(1, 4, 99, &block, &mut wire).unwrap(); + assert_eq!(drive(&mut client, &mut wire), Err("PROTOCOL_ERROR")); +} + +/// A host that buffers a body and releases capacity when the application +/// consumes it can call `release_capacity` after the stream is gone. That is a +/// no-op — the credit went back in bulk at termination — not an error. +#[test] +fn h2_release_after_termination_is_a_no_op() { + let (mut client, mut server) = handshake(100); + let id = client.open(&post_headers("/x"), false).unwrap(); + client.send_data(id, b"body", false).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + assert_eq!(server.unreleased(id), Some(4)); + server.reset(id, 8).unwrap(); + assert_eq!(server.unreleased(id), None); + server.release_capacity(id, 4).unwrap(); + // An id this connection has never seen is still an error. + assert!(server.release_capacity(99, 4).is_err()); + assert!(server.unreleased(99).is_none()); +} + +/// Exceeding the concurrent-stream limit is a stream error (RFC 9113 §5.1.2), +/// not a connection error. It was reaching `receive`'s error map as a +/// REFUSED_STREAM with no case of its own and going out as PROTOCOL_ERROR. +#[test] +fn h2_stream_limit_refuses_one_stream_not_the_connection() { + let (mut client, mut server) = handshake(2); + let mut open = Vec::new(); + for i in 0..2 { + let id = client.open(&post_headers(&format!("/{i}")), false).unwrap(); + open.push(id); + } + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + // A third concurrent stream, over the server's advertised limit of two. + // The client is told MAX_CONCURRENT_STREAMS=2, so build the frame by hand. + let third = 5; + let mut encoder = hpack::Encoder::new(4096); + let mut block = Vec::new(); + encoder.encode(&post_headers("/third"), &mut block); + let mut wire = Vec::new(); + http2::encode_frame(1, 4, third, &block, &mut wire).unwrap(); + let events = drive(&mut server, &mut wire).expect("stream limit must not fail the connection"); + assert_eq!(events, vec![format!("Reset s={third} code=7")]); + assert_eq!( + &server.output()[3..4], + &[3], + "RST_STREAM answers the refusal" + ); + // The two live streams are untouched. + for id in open { + server + .send_headers(id, &[Header::new(":status", "200")], true) + .unwrap(); + } +} + +/// A stream opened after a graceful GOAWAY is the unavoidable race: the peer +/// cannot have seen the GOAWAY when it opened. It used to be a connection +/// error, decided inside `receive` where no host could reach it. +/// +/// Node sends **no frame at all** here. Measured against Node 26.5.1 with a raw +/// TCP peer and hand-encoded frames, so no library could shape the answer: no +/// RST_STREAM, no second GOAWAY, no frame naming the stream; the request never +/// reaches the application; and the session stays alive and keeps servicing +/// frames (a PING sent afterwards is still acknowledged). RFC 9113 section 6.8 +/// makes such a stream simply "not processed", to be retried on a new +/// connection. So the connection reports it and sends nothing: a frame it +/// emitted could not be un-emitted, and a host that must match Node would then +/// have no way back. +#[test] +fn h2_stream_after_graceful_goaway_is_unprocessed_and_unanswered() { + let (mut client, mut server) = handshake(100); + server.shutdown().unwrap(); + let _ = ship(&mut server); + let late = client.open(&post_headers("/late"), true).unwrap(); + let mut wire = ship(&mut client); + let events = drive(&mut server, &mut wire).expect("the GOAWAY race must not fail the session"); + assert_eq!(events, vec![format!("Unprocessed s={late}")]); + assert!( + server.output().is_empty(), + "Node sends nothing here; so must we (got {:?})", + server.output() + ); + // HPACK state survived the discarded block, so the session still decodes. + let again = client.open(&post_headers("/after"), true).unwrap(); + let mut wire = ship(&mut client); + assert_eq!( + drive(&mut server, &mut wire).unwrap(), + vec![format!("Unprocessed s={again}")] + ); + + // The decision belongs to the host, not the connection: one that wants to + // answer still can, and gets exactly one RST_STREAM for the stream it names. + server.reset(late, 7).unwrap(); + let out = server.output().to_vec(); + assert_eq!(out.len(), 13, "one RST_STREAM frame and nothing else"); + assert_eq!(out[3], 3, "RST_STREAM"); + assert_eq!(u32::from_be_bytes(out[5..9].try_into().unwrap()), late); + assert_eq!(u32::from_be_bytes(out[9..13].try_into().unwrap()), 7); +} + +/// GOAWAY names the last stream that was actually *processed*. A stream that was +/// reported but declined — refused past the limit, or arriving after our own +/// GOAWAY — must not advance it: RFC 9113 §6.8 lets the peer retry everything +/// above `lastStreamID` on a new connection, so naming a declined stream tells +/// the peer a request was handled when it was not, and it is silently lost. +#[test] +fn h2_goaway_names_the_last_processed_stream_not_the_last_seen() { + // A stream refused by the concurrent-stream limit. + let (mut client, mut server) = handshake(2); + let mut live = Vec::new(); + for i in 0..2 { + live.push(client.open(&post_headers(&format!("/{i}")), false).unwrap()); + } + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + let processed = *live.last().unwrap(); + let refused = processed + 2; + let mut block = Vec::new(); + hpack::Encoder::new(4096).encode(&post_headers("/over"), &mut block); + let mut wire = Vec::new(); + http2::encode_frame(1, 4, refused, &block, &mut wire).unwrap(); + assert_eq!( + drive(&mut server, &mut wire).unwrap(), + vec![format!("Reset s={refused} code=7")] + ); + let _ = ship(&mut server); + server.shutdown().unwrap(); + let out = ship(&mut server); + assert_eq!(out[3], 7, "GOAWAY"); + let named = u32::from_be_bytes(out[9..13].try_into().unwrap()) & 0x7fffffff; + assert_eq!( + named, processed, + "GOAWAY named {named}; stream {refused} was refused, not processed" + ); + + // And a stream arriving after the GOAWAY does not advance it either. The + // client's own peer-stream limit is two, so build this one by hand. + let late = refused + 2; + let mut wire = Vec::new(); + http2::encode_frame(1, 4, late, &block, &mut wire).unwrap(); + assert_eq!( + drive(&mut server, &mut wire).unwrap(), + vec![format!("Unprocessed s={late}")] + ); + server.shutdown().unwrap(); + let out = ship(&mut server); + let named = u32::from_be_bytes(out[9..13].try_into().unwrap()) & 0x7fffffff; + assert_eq!(named, processed, "an unprocessed stream must not be named"); + // The declined ids are still tolerated on the wire, which is what the + // separate watermark buys: late frames for them must not fail the session. + server.release_capacity(refused, 0).unwrap(); + assert!(server.reset(late, 7).is_ok()); +} + +/// `goaway` sets the code, the last stream id and the opaque debug data that +/// `shutdown` cannot express. +#[test] +fn h2_goaway_carries_code_last_stream_and_opaque_data() { + let (mut client, mut server) = handshake(100); + server.goaway(11, 0, b"enhance").unwrap(); + let mut wire = ship(&mut server); + assert_eq!( + drive(&mut client, &mut wire).unwrap(), + vec!["Goaway code=11"] + ); + assert!(client.open(&post_headers("/after"), true).is_err()); + // Opaque data over the peer's SETTINGS_MAX_FRAME_SIZE is refused here + // rather than sent for the peer to answer with FRAME_SIZE_ERROR. + assert_eq!( + server.goaway(0, 0, &vec![0; 16384]).err().map(|e| e.code), + Some("FRAME_SIZE_ERROR") + ); +} + +/// `Step`'s two independent zero cases, both normal, neither previously +/// stated. A host looping while "an event came back" stalls on the second — +/// at the client preface, before a single frame is read. +#[test] +fn h2_step_has_two_independent_zero_cases() { + let mut server = http2::Connection::new(http2::Role::Server, http2::Limits::default()).unwrap(); + // consumed == 0, event == None: a partial preface. Wait for more input. + let step = server.receive(&http2::PREFACE[..5]).unwrap(); + assert_eq!((step.consumed, step.event.is_some()), (0, false)); + + let mut wire = http2::PREFACE.to_vec(); + http2::encode_frame(4, 0, 0, &[], &mut wire).unwrap(); // peer SETTINGS + http2::encode_frame(4, 1, 0, &[], &mut wire).unwrap(); // peer SETTINGS ack + http2::encode_frame(2, 0, 1, &[0, 0, 0, 0, 0], &mut wire).unwrap(); // PRIORITY + let mut shapes = Vec::new(); + loop { + let step = server.receive(&wire).unwrap(); + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + shapes.push((consumed, step.event.is_some())); + wire.drain(..consumed); + if !progressed { + break; + } + } + assert_eq!( + shapes, + vec![ + (24, false), // consumed > 0, event == None: the preface. KEEP GOING. + (9, true), // SETTINGS + (9, false), // consumed > 0, event == None: the SETTINGS ack. + (14, false), // consumed > 0, event == None: PRIORITY. + (0, false), // consumed == 0, event == None: exhausted. STOP. + ] + ); + // An HTTP/2 event always consumes; only the stop case has consumed == 0. + assert!(shapes.iter().all(|(n, event)| !event || *n > 0)); +} + +/// `Event::Headers` says which of the three blocks it is, so a host does not +/// have to duplicate the `received_head` state the core already keeps. +#[test] +fn h2_headers_events_name_head_informational_and_trailers() { + let (mut client, mut server) = handshake(100); + let id = client.open(&post_headers("/x"), false).unwrap(); + let mut wire = ship(&mut client); + drive(&mut server, &mut wire).unwrap(); + server + .send_headers(id, &[Header::new(":status", "103")], false) + .unwrap(); + server + .send_headers(id, &[Header::new(":status", "200")], false) + .unwrap(); + server + .send_headers(id, &[Header::new("x-trailer", "1")], true) + .unwrap(); + let mut wire = ship(&mut server); + let mut kinds = Vec::new(); + loop { + let step = client.receive(&wire).unwrap(); + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + if let Some(http2::Event::Headers { kind, .. }) = step.event { + kinds.push(kind); + } + wire.drain(..consumed); + if !progressed { + break; + } + } + assert_eq!( + kinds, + vec![ + http2::HeadersKind::Informational, + http2::HeadersKind::Head, + http2::HeadersKind::Trailers, + ] + ); +} + +/// The HTTP/1 analogue of the same contract (PerryTS/turnloop#50): `Event::End` +/// arrives with `consumed == 0`, so "consumed means progress" is wrong there in +/// the opposite direction, and `State::Done` then returns the stop shape +/// forever. +#[test] +fn http1_step_zero_cases_are_the_mirror_image() { + let mut decoder = Decoder::new(Mode::Response, Default::default()); + let mut shapes = Vec::new(); + let mut wire = + b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n4\r\nbody\r\n0\r\n\r\n".to_vec(); + for _ in 0..9 { + let step = decoder.receive(&wire).unwrap(); + let consumed = step.consumed; + shapes.push((consumed, step.event.is_some())); + wire.drain(..consumed); + } + assert_eq!( + shapes, + vec![ + (47, true), // Head + (3, false), // consumed > 0, event == None: the chunk-size line. + (4, true), // Body + (2, false), // consumed > 0, event == None: the chunk CRLF. + (3, false), // consumed > 0, event == None: the terminating size. + (2, false), // consumed > 0, event == None: the empty trailer block. + (0, true), // consumed == 0, event == Some(End): #50's shape. + (0, false), // State::Done: the stop shape, and it never changes. + (0, false), + ] + ); +} + +/// The step contract is only worth stating if the conditions it rules out +/// actually break. These two loops are the ones hosts wrote, one per decoder. +#[test] +fn step_contract_rejects_both_wrong_loop_conditions() { + // HTTP/2, "loop while an event came back": stalls at the client preface, + // before a single frame is read, so the connection never starts. + let mut server = http2::Connection::new(http2::Role::Server, http2::Limits::default()).unwrap(); + let mut wire = http2::PREFACE.to_vec(); + http2::encode_frame(4, 0, 0, &[], &mut wire).unwrap(); + let full = wire.len(); + let mut events = 0; + loop { + let step = server.receive(&wire).unwrap(); + let consumed = step.consumed; + let Some(_event) = step.event else { break }; + events += 1; + wire.drain(..consumed); + } + assert_eq!(events, 0); + assert_eq!(wire.len(), full, "not one byte was read"); + // And the stall is not recoverable in place: `receive` has already advanced + // past the preface, so a host that did not drain `consumed` now feeds it + // back and the connection fails. + assert_eq!( + server.receive(&wire).err().map(|e| e.code), + Some("FRAME_SIZE_ERROR") + ); + + // The documented condition reaches the peer's SETTINGS. + let mut server = http2::Connection::new(http2::Role::Server, http2::Limits::default()).unwrap(); + let mut events = 0; + loop { + let step = server.receive(&wire).unwrap(); + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + events += usize::from(step.event.is_some()); + wire.drain(..consumed); + if !progressed { + break; + } + } + assert_eq!(events, 1); + assert!(wire.is_empty()); + + // HTTP/1, "loop while input was consumed": drops `Event::End`, which reads + // no input at all (PerryTS/turnloop#50). + let message = b"HTTP/1.1 204 No Content\r\n\r\n"; + let mut decoder = Decoder::new(Mode::Response, Default::default()); + let mut wire = message.to_vec(); + let mut ended = false; + loop { + let step = decoder.receive(&wire).unwrap(); + if step.consumed == 0 { + break; + } + ended |= matches!(step.event, Some(Event::End)); + wire.drain(..step.consumed); + } + assert!( + !ended, + "Event::End consumes nothing, so this loop never sees it" + ); + // The documented condition does see it. + let mut decoder = Decoder::new(Mode::Response, Default::default()); + let mut wire = message.to_vec(); + let mut ended = false; + loop { + let step = decoder.receive(&wire).unwrap(); + let consumed = step.consumed; + let progressed = consumed > 0 || step.event.is_some(); + ended |= matches!(step.event, Some(Event::End)); + wire.drain(..consumed); + if !progressed { + break; + } + } + assert!(ended); +} diff --git a/protocols/turnloop-http/tests/interop.rs b/protocols/turnloop-http/tests/interop.rs index 10b2211..ea29b00 100644 --- a/protocols/turnloop-http/tests/interop.rs +++ b/protocols/turnloop-http/tests/interop.rs @@ -330,6 +330,7 @@ fn serve_h2(mut socket: TcpStream, total: usize) { stream, headers, end_stream, + .. }) = step.event { assert!(end_stream); @@ -369,7 +370,20 @@ fn serve_h2(mut socket: TcpStream, total: usize) { match socket.read(&mut tail) { Ok(0) => break, Ok(_) => {} - Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => break, + // This is a raw std socket, so it does not get the normalisation the + // crate applies everywhere else: Windows reports the local end of a + // departed peer as WSAECONNABORTED, and `backend/iocp/socket.rs` + // (`ERROR_CONNECTION_ABORTED`) and `types.rs` both fold that into + // `ConnectionReset`. Both mean the peer is gone after a verified + // exchange, which is a valid end to this drain. + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::ConnectionReset | std::io::ErrorKind::ConnectionAborted + ) => + { + break; + } Err(e) => panic!("peer did not finish shutdown: {e}"), } }