From: Julien Puydt Date: Thu, 6 Jul 2023 07:35:51 +0000 (+0200) Subject: New upstream version 5.1.0 X-Git-Tag: archive/raspbian/6.2.2-1+rpi1^2~5^2~9 X-Git-Url: https://dgit.raspbian.org/?a=commitdiff_plain;h=2099ca4572a74d4166d7253816cff7f69efd7d29;p=ocaml-cohttp.git New upstream version 5.1.0 --- diff --git a/CHANGES.md b/CHANGES.md index 1985d91..0a4aa0f 100644 --- 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 08f8398..d269641 100644 --- 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 94300ea..22c9420 100644 --- 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 b026feb..6ec5b20 100644 --- 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 fffebe0..2e71327 100644 --- 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 d89e7f9..9e6a95a 100644 --- 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}