fix leak on head client requests

lib_test/test_sanity.ml

@@ -10,6 +10,8 @@ let message = "Hello sanity!"
 
 let chunk_body = ["one"; ""; " "; "bar"; ""]
 
+let leak_repeat = 1024
+
 let server =
   [ (* t *)
     Server.respond_string ~status:`OK ~body:message ();
@@ -41,7 +43,10 @@ let server =
     tests @ [
       (fun _ body -> (* Returns 500 on bad file *)
          Cohttp_lwt_body.to_string body >>= fun fname ->
-         Server.respond_file ~fname ())])
+         Server.respond_file ~fname ())] @
+    (Array.init (leak_repeat * 2) (fun _ _ _ ->
+         (* no leaks *)
+         Server.respond_string ~status:`OK ~body:"no leak" ()) |> Array.to_list))
   |> response_sequence
 
 let ts =
@@ -129,13 +134,23 @@ let ts =
           body "Error: Internal Server Error"
       ) (fun () -> Lwt_unix.unlink fname)
     in
+    let test_no_leak () =
+      let stream = Array.init leak_repeat (fun _ -> uri) |> Lwt_stream.of_array in
+      Lwt_stream.fold_s (fun uri () ->
+          Client.head uri >>= fun resp_head ->
+          assert_equal (Response.status resp_head) `OK;
+          Client.get uri >>= fun (resp_get, body) ->
+          assert_equal (Response.status resp_get) `OK;
+          Cohttp_lwt_body.drain_body body) stream ()
+    in
     [ "sanity test", t
     ; "empty chunk test", empty_chunk
     ; "pipelined chunk test", pipelined_chunk
     ; "no body when response is not modified", not_modified_has_no_body
     ; "pipelined with interleaving requests", pipelined_interleave
     ; "massive chunked", massive_chunked
     ; "unreadable file returns 500", unreadable_file_500
+    ; "no leaks on requests", test_no_leak
     ]
   end
 

lwt-core/cohttp_lwt.ml

@@ -71,7 +71,7 @@ module Make_client
           Gc.finalise gcfn stream;
           let body = Body.of_stream stream in
           return (res, body)
-        | `No -> return (res, `Empty)
+        | `No -> closefn (); return (res, `Empty)
       end
     end
     |> fun t ->

lwt/cohttp_lwt_unix_test.ml

@@ -20,12 +20,14 @@ let temp_server ?port spec callback =
     | Some p -> p in
   let server = Server.make ~callback:(fun _ req body -> spec req body) () in
   let uri = Uri.of_string ("http://0.0.0.0:" ^ (string_of_int port)) in
+  let server_failed, server_failed_wake = Lwt.task () in
   let server = Lwt.catch
                  (fun () -> Server.create ~mode:(`TCP (`Port port)) server)
                  (function
                    | Lwt.Canceled -> Lwt.return_unit
-                   | x -> Lwt.fail x) in
-  callback uri >|= fun res ->
+                   | x -> Lwt.wakeup_exn server_failed_wake x; Lwt.fail x)
+  in
+  Lwt.pick [ callback uri; server_failed ] >|= fun res ->
   Lwt.cancel server;
   res