gazagnaire.org
HTTP/2 framing, HPACK and connection state, I/O-free
1(*---------------------------------------------------------------------------
2 Copyright (c) 2019 Antonio Nuno Monteiro. Copyright (c) 2025 Anil Madhavapeddy
3 <anil@recoil.org>.
4
5 All rights reserved.
6
7 Redistribution and use in source and binary forms, with or without
8 modification, are permitted provided that the following conditions are met:
9
10 1. Redistributions of source code must retain the above copyright notice, this
11 list of conditions and the following disclaimer.
12
13 2. Redistributions in binary form must reproduce the above copyright notice,
14 this list of conditions and the following disclaimer in the documentation
15 and/or other materials provided with the distribution.
16
17 3. Neither the name of the copyright holder nor the names of its contributors
18 may be used to endorse or promote products derived from this software without
19 specific prior written permission.
20
21 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
22 AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23 IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
24 DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
25 FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
26 DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
27 SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
28 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
29 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 SPDX-License-Identifier: BSD-3-Clause
32 ---------------------------------------------------------------------------*)
33
34(** HTTP/2 Stream State Machine per RFC 9113 Section 5.1.
35
36 Implements the stream lifecycle:
37 - idle -> open -> half-closed -> closed
38 - With reserved states for server push (PUSH_PROMISE)
39
40 See
41 {{:https://datatracker.ietf.org/doc/html/rfc9113#section-5.1}RFC 9113
42 Section 5.1}. *)
43
44(* ============================================================ Stream States
45 ============================================================ *)
46
47(** Why a stream was closed. *)
48type closed_reason =
49 | Finished (** Both endpoints sent END_STREAM. *)
50 | Reset_by_us of Frame.error_code (** We sent RST_STREAM. *)
51 | Reset_by_them of Frame.error_code (** Peer sent RST_STREAM. *)
52
53(** Stream state per RFC 9113 Section 5.1. *)
54type state =
55 | Idle
56 (** Initial state. Transitions to open on send/recv HEADERS, or to
57 reserved on PUSH_PROMISE. *)
58 | Reserved_local
59 (** We sent PUSH_PROMISE. Can only send HEADERS (-> half_closed_remote) or
60 RST_STREAM (-> closed). *)
61 | Reserved_remote
62 (** Peer sent PUSH_PROMISE. Can only receive HEADERS (->
63 half_closed_local) or send/recv RST_STREAM (-> closed). *)
64 | Open
65 (** Both sides can send frames. Transitions to half_closed on END_STREAM,
66 or closed on RST_STREAM. *)
67 | Half_closed_local
68 (** We sent END_STREAM. We can only receive frames. Transitions to closed
69 on recv END_STREAM or RST_STREAM. *)
70 | Half_closed_remote
71 (** Peer sent END_STREAM. We can only send frames. Transitions to closed
72 on send END_STREAM or RST_STREAM. *)
73 | Closed of closed_reason (** Terminal state. *)
74
75(** Events that cause state transitions. *)
76type event =
77 | Send_headers of { end_stream : bool }
78 | Recv_headers of { end_stream : bool }
79 | Send_data of { end_stream : bool }
80 | Recv_data of { end_stream : bool }
81 | Send_push_promise
82 | Recv_push_promise
83 | Send_rst_stream of Frame.error_code
84 | Recv_rst_stream of Frame.error_code
85 | Send_end_stream
86 | Recv_end_stream
87
88(* ============================================================ State Transition
89 Logic ============================================================ *)
90
91(** [transition state event] applies [event] to [state] and returns the new
92 state or an error.
93
94 Implements RFC 9113 Section 5.1 state machine. *)
95let transition state event : (state, Frame.error_code * string) result =
96 match (state, event) with
97 (* === Idle state transitions === *)
98 | Idle, Send_headers { end_stream = false } -> Ok Open
99 | Idle, Send_headers { end_stream = true } -> Ok Half_closed_local
100 | Idle, Recv_headers { end_stream = false } -> Ok Open
101 | Idle, Recv_headers { end_stream = true } -> Ok Half_closed_remote
102 | Idle, Send_push_promise -> Ok Reserved_local
103 | Idle, Recv_push_promise -> Ok Reserved_remote
104 | Idle, _ -> Error (Frame.Protocol_error, "Invalid frame on idle stream")
105 (* === Reserved (local) state transitions === *)
106 | Reserved_local, Send_headers { end_stream = false } -> Ok Half_closed_remote
107 | Reserved_local, Send_headers { end_stream = true } ->
108 (* Send HEADERS with END_STREAM on reserved -> closed *)
109 Ok (Closed Finished)
110 | Reserved_local, Send_rst_stream code -> Ok (Closed (Reset_by_us code))
111 | Reserved_local, Recv_rst_stream code -> Ok (Closed (Reset_by_them code))
112 | Reserved_local, _ ->
113 Error (Frame.Protocol_error, "Invalid frame on reserved (local) stream")
114 (* === Reserved (remote) state transitions === *)
115 | Reserved_remote, Recv_headers { end_stream = false } -> Ok Half_closed_local
116 | Reserved_remote, Recv_headers { end_stream = true } ->
117 (* Recv HEADERS with END_STREAM on reserved -> closed *)
118 Ok (Closed Finished)
119 | Reserved_remote, Send_rst_stream code -> Ok (Closed (Reset_by_us code))
120 | Reserved_remote, Recv_rst_stream code -> Ok (Closed (Reset_by_them code))
121 | Reserved_remote, _ ->
122 Error (Frame.Protocol_error, "Invalid frame on reserved (remote) stream")
123 (* === Open state transitions === *)
124 | Open, Send_headers { end_stream = true } -> Ok Half_closed_local
125 | Open, Send_headers { end_stream = false } ->
126 Ok Open (* Trailers without END_STREAM, unusual but valid *)
127 | Open, Recv_headers { end_stream = true } -> Ok Half_closed_remote
128 | Open, Recv_headers { end_stream = false } ->
129 Ok Open (* Trailers without END_STREAM *)
130 | Open, Send_data { end_stream = true } -> Ok Half_closed_local
131 | Open, Send_data { end_stream = false } -> Ok Open
132 | Open, Recv_data { end_stream = true } -> Ok Half_closed_remote
133 | Open, Recv_data { end_stream = false } -> Ok Open
134 | Open, Send_end_stream -> Ok Half_closed_local
135 | Open, Recv_end_stream -> Ok Half_closed_remote
136 | Open, Send_rst_stream code -> Ok (Closed (Reset_by_us code))
137 | Open, Recv_rst_stream code -> Ok (Closed (Reset_by_them code))
138 | Open, Send_push_promise | Open, Recv_push_promise ->
139 (* PUSH_PROMISE is sent on an existing stream but creates a new reserved
140 stream *)
141 Ok Open
142 (* === Half-closed (local) state transitions === *)
143 | Half_closed_local, Recv_headers { end_stream = true } ->
144 Ok (Closed Finished)
145 | Half_closed_local, Recv_headers { end_stream = false } ->
146 Ok Half_closed_local
147 | Half_closed_local, Recv_data { end_stream = true } -> Ok (Closed Finished)
148 | Half_closed_local, Recv_data { end_stream = false } -> Ok Half_closed_local
149 | Half_closed_local, Recv_end_stream -> Ok (Closed Finished)
150 | Half_closed_local, Send_rst_stream code -> Ok (Closed (Reset_by_us code))
151 | Half_closed_local, Recv_rst_stream code -> Ok (Closed (Reset_by_them code))
152 | Half_closed_local, (Send_headers _ | Send_data _ | Send_end_stream) ->
153 Error (Frame.Stream_closed, "Cannot send on half-closed (local) stream")
154 | Half_closed_local, _ ->
155 Ok Half_closed_local (* WINDOW_UPDATE, PRIORITY allowed *)
156 (* === Half-closed (remote) state transitions === *)
157 | Half_closed_remote, Send_headers { end_stream = true } ->
158 Ok (Closed Finished)
159 | Half_closed_remote, Send_headers { end_stream = false } ->
160 Ok Half_closed_remote
161 | Half_closed_remote, Send_data { end_stream = true } -> Ok (Closed Finished)
162 | Half_closed_remote, Send_data { end_stream = false } ->
163 Ok Half_closed_remote
164 | Half_closed_remote, Send_end_stream -> Ok (Closed Finished)
165 | Half_closed_remote, Send_rst_stream code -> Ok (Closed (Reset_by_us code))
166 | Half_closed_remote, Recv_rst_stream code -> Ok (Closed (Reset_by_them code))
167 | Half_closed_remote, (Recv_headers _ | Recv_data _ | Recv_end_stream) ->
168 Error (Frame.Stream_closed, "Received data on half-closed (remote) stream")
169 | Half_closed_remote, _ ->
170 Ok Half_closed_remote (* WINDOW_UPDATE, PRIORITY allowed *)
171 (* === Closed state - terminal === *)
172 | Closed reason, Recv_rst_stream _ ->
173 (* Can receive RST_STREAM on closed stream (race condition) *)
174 Ok (Closed reason)
175 | Closed _, _ -> Error (Frame.Stream_closed, "Stream is closed")
176
177(* ============================================================ Stream Type
178 ============================================================ *)
179
180type flow_control = {
181 send_window : int; (** Bytes we're allowed to send. *)
182 recv_window : int; (** Bytes we've advertised we can receive. *)
183 initial_send_window : int; (** Initial send window from SETTINGS. *)
184 initial_recv_window : int; (** Initial receive window from SETTINGS. *)
185}
186(** Flow control state for a stream. *)
187
188type t = {
189 id : int32;
190 (** Stream identifier (odd = client-initiated, even = server-initiated).
191 *)
192 state : state; (** Current stream state. *)
193 flow : flow_control; (** Flow control windows. *)
194 request_headers : Hpack.header list option;
195 (** Request headers once received/sent. *)
196 response_headers : Hpack.header list option;
197 (** Response headers once received/sent. *)
198 trailers : Hpack.header list option; (** Trailing headers. *)
199 send_queue : Rope.t;
200 (** Outbound DATA accepted from the application but not yet emitted: held
201 here until the connection and stream send windows admit it. *)
202 send_fin : bool; (** END_STREAM has been requested for the queued data. *)
203 fin_sent : bool; (** The END_STREAM DATA frame has been emitted. *)
204}
205(** A single HTTP/2 stream: an immutable value; transitions return a new [t]. *)
206
207(** Default initial flow control window size per RFC 9113. *)
208let default_initial_window_size = 65535
209
210(** Create a new stream with the given ID. *)
211let v ?(initial_send_window = default_initial_window_size)
212 ?(initial_recv_window = default_initial_window_size) id =
213 {
214 id;
215 state = Idle;
216 flow =
217 {
218 send_window = initial_send_window;
219 recv_window = initial_recv_window;
220 initial_send_window;
221 initial_recv_window;
222 };
223 request_headers = None;
224 response_headers = None;
225 trailers = None;
226 send_queue = Rope.empty;
227 send_fin = false;
228 fin_sent = false;
229 }
230
231(** Get the stream ID. *)
232let id t = t.id
233
234(** Get the current state. *)
235let state t = t.state
236
237(** Helper for state predicate checks. *)
238let state_matches pred t = pred t.state
239
240(** Check if stream is in idle state. *)
241let is_idle = state_matches (function Idle -> true | _ -> false)
242
243(** Check if stream is open (including half-closed). *)
244let is_active =
245 state_matches (function
246 | Open | Half_closed_local | Half_closed_remote -> true
247 | _ -> false)
248
249(** Check if stream is fully open (both directions). *)
250let is_open = state_matches (function Open -> true | _ -> false)
251
252(** Check if stream is closed. *)
253let is_closed = state_matches (function Closed _ -> true | _ -> false)
254
255(** Check if we can send on this stream. *)
256let can_send =
257 state_matches (function
258 | Open | Half_closed_remote | Reserved_local -> true
259 | _ -> false)
260
261(** Check if we can receive on this stream. *)
262let can_recv =
263 state_matches (function
264 | Open | Half_closed_local | Reserved_remote -> true
265 | _ -> false)
266
267(** Apply an event to the stream. Returns [Ok t'] with the advanced stream, or
268 [Error (code, msg)] on an invalid transition. *)
269let apply_event t event =
270 match transition t.state event with
271 | Ok new_state -> Ok { t with state = new_state }
272 | Error (code, msg) -> Error (code, msg)
273
274(* ============================================================ Flow Control
275 ============================================================ *)
276
277(** Consume bytes from the send window. Returns the number of bytes actually
278 consumed (may be less if window exhausted) and the advanced stream. *)
279let consume_send_window t bytes =
280 let available = min bytes t.flow.send_window in
281 ( available,
282 {
283 t with
284 flow = { t.flow with send_window = t.flow.send_window - available };
285 } )
286
287(* RFC 9113 s6.9.1: flow-control windows must not exceed 2^31-1. Compared in
288 int64 so the bound is exact even where the native int is narrower than 32
289 bits (there the window can never reach it and the check never fires). *)
290let max_flow_window = 0x7FFF_FFFFL
291
292(** Add bytes to the send window (from WINDOW_UPDATE). *)
293let credit_send_window t increment =
294 let new_window = t.flow.send_window + increment in
295 if Int64.compare (Int64.of_int new_window) max_flow_window > 0 then
296 Error
297 ( Frame.Flow_control_error,
298 "WINDOW_UPDATE would overflow stream send window" )
299 else Ok { t with flow = { t.flow with send_window = new_window } }
300
301(** Consume bytes from the receive window. Call this when receiving DATA frames.
302*)
303let consume_recv_window t bytes =
304 { t with flow = { t.flow with recv_window = t.flow.recv_window - bytes } }
305
306(** Credit the receive window (we're sending WINDOW_UPDATE). *)
307let credit_recv_window t increment =
308 { t with flow = { t.flow with recv_window = t.flow.recv_window + increment } }
309
310(** Get available send window. *)
311let send_window t = t.flow.send_window
312
313(** Get available receive window. *)
314let recv_window t = t.flow.recv_window
315
316(** Get initial receive window size. *)
317let initial_recv_window t = t.flow.initial_recv_window
318
319(** Update initial window size (from SETTINGS). Adjusts the current window by
320 the delta. *)
321let update_initial_window_size t new_initial_size =
322 let delta = new_initial_size - t.flow.initial_send_window in
323 let new_window = t.flow.send_window + delta in
324 if
325 Int64.compare (Int64.of_int new_window) max_flow_window > 0
326 || new_window < 0
327 then
328 Error
329 ( Frame.Flow_control_error,
330 "Flow control window overflow after SETTINGS update" )
331 else Ok { t with flow = { t.flow with send_window = new_window } }
332
333(* ============================================================ Send Queue
334 ============================================================ *)
335
336(** Append [payload] to the outbound DATA queue. [end_stream] records that the
337 queued data is the last the application will send. *)
338let enqueue_send t ?(end_stream = false) payload =
339 {
340 t with
341 send_queue = Rope.concat t.send_queue payload;
342 send_fin = t.send_fin || end_stream;
343 }
344
345(** Bytes queued for sending but not yet emitted. *)
346let pending_send t = Rope.length t.send_queue
347
348(** Whether END_STREAM has been requested for the queued data. *)
349let send_fin t = t.send_fin
350
351(** Whether the END_STREAM DATA frame has already been emitted. *)
352let fin_sent t = t.fin_sent
353
354(** Record that the END_STREAM DATA frame has been emitted. *)
355let mark_fin_sent t = { t with fin_sent = true }
356
357(** Remove and return the first [n] bytes of the send queue ([n] is clamped to
358 what is queued) and the stream with that prefix dropped. *)
359let take_send t n =
360 let len = Rope.length t.send_queue in
361 let n = min n len in
362 let head = Rope.sub t.send_queue 0 n in
363 (head, { t with send_queue = Rope.sub t.send_queue n (len - n) })
364
365(* ============================================================ Stream
366 Identifier Management
367 ============================================================ *)
368
369(** Check if stream ID is client-initiated (odd). *)
370let is_client_initiated id = Int32.(logand id 1l = 1l)
371
372(** Check if stream ID is server-initiated (even, non-zero). *)
373let is_server_initiated id = Int32.(id > 0l && logand id 1l = 0l)
374
375(** Check if stream ID is valid (non-zero). *)
376let is_valid_id id = Int32.compare id 0l > 0
377
378(** Connection-level stream ID is 0. *)
379let connection_stream_id = 0l
380
381(** Check if this is the connection-level stream. *)
382let is_connection_stream id = Int32.equal id 0l
383
384(* ============================================================ Pretty Printing
385 ============================================================ *)
386
387let pp_closed_reason fmt = function
388 | Finished -> Fmt.pf fmt "Finished"
389 | Reset_by_us code -> Fmt.pf fmt "Reset_by_us(%a)" Frame.pp_error_code code
390 | Reset_by_them code ->
391 Fmt.pf fmt "Reset_by_them(%a)" Frame.pp_error_code code
392
393let pp_state fmt = function
394 | Idle -> Fmt.pf fmt "Idle"
395 | Reserved_local -> Fmt.pf fmt "Reserved(local)"
396 | Reserved_remote -> Fmt.pf fmt "Reserved(remote)"
397 | Open -> Fmt.pf fmt "Open"
398 | Half_closed_local -> Fmt.pf fmt "HalfClosed(local)"
399 | Half_closed_remote -> Fmt.pf fmt "HalfClosed(remote)"
400 | Closed reason -> Fmt.pf fmt "Closed(%a)" pp_closed_reason reason
401
402let string_of_state state = Fmt.str "%a" pp_state state
403
404let pp fmt t =
405 Fmt.pf fmt "Stream{id=%ld; state=%a; send_window=%d; recv_window=%d}" t.id
406 pp_state t.state t.flow.send_window t.flow.recv_window
407
408(* ============================================================ Headers
409 Management ============================================================ *)
410
411(** Set request headers, returning the updated stream. *)
412let set_request_headers t headers = { t with request_headers = Some headers }
413
414(** Get request headers. *)
415let request_headers t = t.request_headers
416
417(** Set response headers, returning the updated stream. *)
418let set_response_headers t headers = { t with response_headers = Some headers }
419
420(** Get response headers. *)
421let response_headers t = t.response_headers
422
423(** Set trailers, returning the updated stream. *)
424let set_trailers t headers = { t with trailers = Some headers }
425
426(** Get trailers. *)
427let trailers t = t.trailers
428
429(** Reset the stream with an error code, returning the closed stream. *)
430let reset t code =
431 match apply_event t (Send_rst_stream code) with Ok t -> t | Error _ -> t