New upstream version 5.1.0

diff --git a/CHANGES.md b/CHANGES.md
index 1985d9119eb93d637b1267df7dcea10dffe9473f..0a4aa0f99b4663b4ecca55e61a37fa4c13ff8855 100644 (file)
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,7 @@
+## v5.1.0 (2023-04-04)
+
+- cohttp,cohttp-async server: correctly close broken streams (reported by Stéphane Glondu, fix by samhot and anuragsoni)
+
 ## v5.0.0 (2021-12-15)
 
 - Cohttp.Header: new implementation (lyrm #747)

diff --git a/cohttp-async/src/server.ml b/cohttp-async/src/server.ml
index 08f83980f46c121b335fc8d2477fe36b5cfd3e60..d269641c95359b73b14d8842b862779b5cdb7229 100644 (file)
--- a/cohttp-async/src/server.ml
+++ b/cohttp-async/src/server.ml
@@ -86,6 +86,21 @@ let handle_client handle_request sock rd wr =
             io_handler rd wr >>= fun () ->
             Body.drain body >>| fun () -> Ivar.fill_if_empty finished ()
         | `Response (req, body, (res, res_body)) ->
+            (* There are scenarios if a client leaves before consuming the full response,
+            we might have a reference to an async Pipe that doesn't get drained.
+
+               Not draining or closing a pipe can lead to issues if its holding a resource like
+               a file handle as those resources will never be closed, leading to a leak.
+
+               Async writers have a promise that's fulfilled whenever they are closed,
+               so we can use it to schedule a close operation on the stream to ensure that we
+               don't leave a stream open if the underlying channels are closed. *)
+            (match res_body with
+            | `Empty | `String _ | `Strings _ -> ()
+            | `Pipe stream ->
+                Deferred.any_unit
+                  [ Writer.close_finished wr; Writer.consumer_left wr ]
+                >>> fun () -> Pipe.close_read stream);
             let keep_alive = Request.is_keep_alive req in
             let flush = Response.flush res in
             let res =

diff --git a/cohttp-lwt-unix/test/test_sanity.ml b/cohttp-lwt-unix/test/test_sanity.ml
index 94300eab034b080fbed892af45dd669da70b60ba..22c9420cb4bf559cc630434fe643b7ef85880ce4 100644 (file)
--- a/cohttp-lwt-unix/test/test_sanity.ml
+++ b/cohttp-lwt-unix/test/test_sanity.ml
@@ -15,7 +15,6 @@ let chunk_body = [ "one"; ""; " "; "bar"; "" ]
 let leak_repeat = 1024
 let () = Debug.activate_debug ()
 let () = Logs.set_level (Some Warning)
-let cond = Lwt_condition.create ()
 
 let server =
   List.map const
@@ -63,21 +62,6 @@ let server =
                 Lwt_io.write oc "8\r\nexpert 2\r\n0\r\n\r\n" >>= fun () ->
                 Lwt_io.flush oc >>= fun () -> Lwt_io.close ic )));
     ]
-  @ (* client_close *)
-  [
-    (fun _ _ ->
-      let ready = Lwt_condition.wait cond in
-      let i = ref 0 in
-      let stream =
-        Lwt_stream.from (fun () ->
-            ready >|= fun () ->
-            incr i;
-            if !i > 1000 then failwith "Connection should have failed by now!";
-            Some (String.make 4096 'X'))
-      in
-      Lwt.return
-        (`Response (Cohttp.Response.make ~status:`OK (), `Stream stream)));
-  ]
   |> response_sequence
 
 let check_logs test () =
@@ -172,22 +156,6 @@ let ts =
         Body.to_string body >|= fun body ->
         assert_equal ~printer "expert 2" body
       in
-      let client_close () =
-        Cohttp_lwt_unix.Net.connect_uri ~ctx uri >>= fun (_conn, ic, oc) ->
-        let req =
-          Cohttp.Request.make_for_client ~chunked:false `GET
-            (Uri.with_path uri "/test.html")
-        in
-        Request.write (fun _writer -> Lwt.return_unit) req oc >>= fun () ->
-        Response.read ic >>= function
-        | `Eof | `Invalid _ -> assert false
-        | `Ok rsp ->
-            assert_equal ~printer:Cohttp.Code.string_of_status `OK
-              (Cohttp.Response.status rsp);
-            Cohttp_lwt_unix.Net.close ic oc;
-            Lwt_condition.broadcast cond ();
-            Lwt.pause ()
-      in
       [
         ("sanity test", check_logs t);
         ("pipelined chunk test", check_logs pipelined_chunk);
@@ -195,7 +163,6 @@ let ts =
         ("massive chunked", check_logs massive_chunked);
         ("no leaks on requests", check_logs test_no_leak);
         ("expert response", check_logs expert_pipelined);
-        ("client_close", check_logs client_close);
       ])
 
 let _ = ts |> run_async_tests |> Lwt_main.run

diff --git a/cohttp-lwt/src/server.ml b/cohttp-lwt/src/server.ml
index b026febb3cda74bafaab3dc8c9363660d2d0110f..6ec5b2038a704ff4b39b984b5b30b5edcfb9c576 100644 (file)
--- a/cohttp-lwt/src/server.ml
+++ b/cohttp-lwt/src/server.ml
@@ -111,6 +111,20 @@ module Make (IO : S.IO) = struct
                 `Response rsp))
       (fun () -> Body.drain_body body)
 
+  let handle_response ~keep_alive oc res body handle_client =
+    IO.catch (fun () ->
+        let flush = Response.flush res in
+        Response.write ~flush
+          (fun writer -> Body.write_body (Response.write_body writer) body)
+          res oc
+        >>= fun () -> if keep_alive then handle_client oc else Lwt.return_unit)
+    >>= function
+    | Ok () -> Lwt.return_unit
+    | Error e ->
+        Log.info (fun m -> m "IO error while writing body: %a" IO.pp_error e);
+        Body.drain_body body
+
+
   let rec handle_client ic oc conn callback =
     Request.read ic >>= function
     | `Eof -> Lwt.return_unit
@@ -121,16 +135,13 @@ module Make (IO : S.IO) = struct
         let body = read_body ic req in
         handle_request callback conn req body >>= function
         | `Response (res, body) ->
-            let flush = Response.flush res in
-            Response.write ~flush
-              (fun writer -> Body.write_body (Response.write_body writer) body)
-              res oc
-            >>= fun () ->
-            if Request.is_keep_alive req then handle_client ic oc conn callback
-            else Lwt.return_unit
+            let keep_alive = Request.is_keep_alive req in
+            handle_response ~keep_alive oc res body (fun oc ->
+              handle_client ic oc conn callback)
         | `Expert (res, io_handler) ->
             Response.write_header res oc >>= fun () ->
-            io_handler ic oc >>= fun () -> handle_client ic oc conn callback)
+            io_handler ic oc >>= fun () -> 
+            handle_client ic oc conn callback)
 
   let callback spec io_id ic oc =
     let conn_id = Cohttp.Connection.create () in

diff --git a/cohttp-mirage.opam b/cohttp-mirage.opam
index fffebe0eb6b77c55b5aaa2b96c99dd2096a287c6..2e713279c4c6757475c62325ecb9a8cb8bbb9447 100644 (file)
--- a/cohttp-mirage.opam
+++ b/cohttp-mirage.opam
@@ -23,7 +23,7 @@ depends: [
   "mirage-flow" {>= "2.0.0"}
   "mirage-channel" {>= "4.0.0"}
   "conduit" {>= "2.0.2"}
-  "conduit-mirage" {>= "2.0.2"}
+  "conduit-mirage" {>= "2.3.0"}
   "mirage-kv" {>= "3.0.0"}
   "lwt" {>= "2.4.3"}
   "cohttp" {= version}

diff --git a/cohttp.opam b/cohttp.opam
index d89e7f97671429983bafd9cef260947dc686b853..9e6a95ac607dc1ccd4289567f002410d179da046 100644 (file)
--- a/cohttp.opam
+++ b/cohttp.opam
@@ -44,7 +44,7 @@ depends: [
   "fmt" {with-test}
   "jsonm" {build}
   "alcotest" {with-test}
-  "crowbar" {with-test}
+  "crowbar" {with-test & >= "0.2"}
 ]
 build: [
   ["dune" "subst"] {dev}