[H2.Serialize]: Fix buffer sharing bug in header blocks (#182)

* [H2.Serialize]: Demonstrate buffer sharing bug

* Allocate a new buffer for every header block

CHANGES.md

@@ -23,6 +23,8 @@ Unreleased
   ([#172](https://github.com/anmonteiro/ocaml-h2/pull/172))
 - h2-async: Add an OCaml-TLS client to `h2-async`
   ([#174](https://github.com/anmonteiro/ocaml-h2/pull/174))
+- h2: Fix a bug that caused different requests to share the same headers buffer
+  under concurrency ([#182](https://github.com/anmonteiro/ocaml-h2/pull/182))
 
 0.8.0 2021-04-11
 ---------------

lib/serialize.ml

@@ -311,19 +311,13 @@ module Writer = struct
           (* The number of bytes that were not written due to the output stream
            * being closed before all buffered output could be written. Useful
            * for detecting error cases. *)
-    ; headers_block_buffer : Bigstringaf.t
     ; mutable wakeup : Optional_thunk.t
     }
 
   let create buffer_size =
     let buffer = Bigstringaf.create buffer_size in
     let encoder = Faraday.of_bigstring buffer in
-    { buffer
-    ; encoder
-    ; drained_bytes = 0
-    ; headers_block_buffer = Bigstringaf.create 0x1000
-    ; wakeup = Optional_thunk.none
-    }
+    { buffer; encoder; drained_bytes = 0; wakeup = Optional_thunk.none }
 
   let faraday t = t.encoder
 
@@ -470,7 +464,7 @@ module Writer = struct
 
   let write_request_like_frame t hpack_encoder ~write_frame frame_info request =
     let { Request.meth; target; scheme; headers } = request in
-    let faraday = Faraday.of_bigstring t.headers_block_buffer in
+    let faraday = Faraday.create 0x1000 in
     Hpack.Encoder.encode_header
       hpack_encoder
       faraday
@@ -509,7 +503,7 @@ module Writer = struct
     if not (is_closed t.encoder)
     then (
       let { Response.status; headers; _ } = response in
-      let faraday = Faraday.of_bigstring t.headers_block_buffer in
+      let faraday = Faraday.create 0x1000 in
       (* From RFC7540§8.1.2.4:
        *   For HTTP/2 responses, a single :status pseudo-header field is defined
        *   that carries the HTTP status code field (see [RFC7231], Section 6).
@@ -533,7 +527,7 @@ module Writer = struct
   let write_response_trailers t hpack_encoder frame_info trailers =
     if not (is_closed t.encoder)
     then (
-      let faraday = Faraday.of_bigstring t.headers_block_buffer in
+      let faraday = Faraday.create 0x1000 in
       (* From RFC7540§8.1:
        *  optionally, one HEADERS frame, followed by zero or more
        *  CONTINUATION frames containing the trailer-part, if present (see

lib_test/test_common.ml

@@ -71,6 +71,14 @@ let encode_headers hpack_encoder headers =
   Serialize.Writer.encode_headers hpack_encoder f headers;
   Faraday.serialize_to_bigstring f
 
+let decode_headers decoder bigstring =
+  let parser = Angstrom.Buffered.parse (Hpack.Decoder.decode_headers decoder) in
+  let state = Angstrom.Buffered.feed parser (`Bigstring bigstring) in
+  let state' = Angstrom.Buffered.feed state `Eof in
+  match Angstrom.Buffered.state_to_option state' with
+  | Some (Ok headers) -> headers
+  | Some _ | None -> assert false
+
 let preface =
   let writer = Serialize.Writer.create 0x400 in
   Serialize.Writer.write_connection_preface writer [];

lib_test/test_h2_client.ml

@@ -1214,6 +1214,94 @@ module Client_connection_tests = struct
            Frame.FrameType.serialize frame_type)
          frames)
 
+  let test_header_buffer_sharing () =
+    let t = create_and_handle_preface () in
+    let request1 =
+      Request.create
+        ~scheme:"http"
+        `GET
+        "/"
+        ~headers:
+          Headers.(
+            of_list
+              [ "headerA", "valueA"; "headerB", "valueB"; "headerC", "valueC" ])
+    in
+    let request2 =
+      Request.create
+        ~scheme:"http"
+        `GET
+        "/"
+        ~headers:
+          Headers.(
+            of_list
+              [ "headerD", "valueD"; "headerE", "valueE"; "headerF", "valueF" ])
+    in
+    let response_handler _response _response_body = assert false in
+    let do_request req =
+      Client_connection.request
+        ~flush_headers_immediately:false
+        t
+        req
+        ~error_handler:default_error_handler
+        ~response_handler
+    in
+    let req1_body = do_request request1 in
+    let req2_body = do_request request2 in
+    (* Writer yields when `~flush_headers_immediately` is false *)
+    writer_yielded t;
+    (* Write to the body *)
+    let frames, lenv = flush_pending_writes t in
+    Alcotest.(check (list int))
+      "Batches both header frames together"
+      (List.map Frame.FrameType.serialize Frame.FrameType.[ Headers; Headers ])
+      (List.map
+         (fun Frame.{ frame_header = { frame_type; _ }; _ } ->
+           Frame.FrameType.serialize frame_type)
+         frames);
+
+    let[@ocaml.warning "-8"] [ headers1; headers2 ] = frames in
+    (match headers1.frame_payload, headers2.frame_payload with
+    | Headers (_, block1), Headers (_, block2) ->
+      let headers1 = decode_headers t.hpack_decoder block1 in
+      let headers2 = decode_headers t.hpack_decoder block2 in
+      let headers_testable = Alcotest.of_pp Headers.pp_hum in
+      Alcotest.(check headers_testable)
+        "Headers block 1 matches"
+        (Headers.of_list
+           [ ":method", "GET"
+           ; ":path", "/"
+           ; ":scheme", "http"
+           ; "headerA", "valueA"
+           ; "headerB", "valueB"
+           ; "headerC", "valueC"
+           ])
+        headers1;
+
+      Alcotest.(check headers_testable)
+        "Headers block 2 matches"
+        (Headers.of_list
+           [ ":method", "GET"
+           ; ":path", "/"
+           ; ":scheme", "http"
+           ; "headerD", "valueD"
+           ; "headerE", "valueE"
+           ; "headerF", "valueF"
+           ])
+        headers2
+    | _ -> Alcotest.fail "expected both frame payloads to be header blocks");
+
+    Body.Writer.close req1_body;
+    Body.Writer.close req2_body;
+    report_write_result t (`Ok lenv);
+    let frames, _lenv = flush_pending_writes t in
+    Alcotest.(check (list int))
+      "Writes empty DATA frame"
+      (List.map Frame.FrameType.serialize Frame.FrameType.[ Data ])
+      (List.map
+         (fun Frame.{ frame_header = { frame_type; _ }; _ } ->
+           Frame.FrameType.serialize frame_type)
+         frames)
+
   let suite =
     [ "initial reader state", `Quick, test_initial_reader_state
     ; "set up client connection", `Quick, test_set_up_connection
@@ -1259,6 +1347,7 @@ module Client_connection_tests = struct
     ; ( "don't flush headers immediately"
       , `Quick
       , test_dont_flush_headers_immediately )
+    ; "headers blocks don't share buffers", `Quick, test_header_buffer_sharing
     ]
 end
 

nix/ci/test.nix

@@ -5,7 +5,6 @@ let
   src = fetchGit {
     url = with lock.nodes.nixpkgs.locked; "https://github.com/${owner}/${repo}";
     inherit (lock.nodes.nixpkgs.locked) rev;
-    # inherit (lock.nodes.nixpkgs.original) ref;
     allRefs = true;
   };
 

nix/default.nix

@@ -12,8 +12,7 @@ let
     with nix-filter; filter {
       root = ./..;
       include = [ "dune-project" ] ++ files ++ (builtins.map inDirectory dirs);
-    }
-  ;
+    };
   buildH2 = args: buildDunePackage ({
     version = "0.6.0-dev";
     useDune2 = true;