Also catch exceptions from read and write in PGX.

Author: paurkedal

caqti-async/lib/caqti_async.ml

@@ -179,14 +179,19 @@ module System = struct
     type tcp_flow = Socket.t
     type tls_flow = Socket.t
 
+    let convert_io_exception exn =
+      (match Async_kernel.Monitor.extract_exn exn with
+       | Core_unix.Unix_error (err, func, arg) ->
+          Some (Msg_unix (err, func, arg))
+       | _ -> None)
+
     let intercept_exceptions f =
       Async_kernel.Monitor.try_with f >|= function
        | Ok _ as r -> r
        | Error exn ->
-          (match Async_kernel.Monitor.extract_exn exn with
-           | Core_unix.Unix_error (err, func, arg) ->
-              Error (Msg_unix (err, func, arg))
-           | _ -> raise exn)
+          (match convert_io_exception exn with
+           | Some msg -> Error msg
+           | None -> raise exn)
 
     let connect_tcp ~sw:_ ~stdenv:() addr =
       intercept_exceptions @@ fun () ->

caqti-driver-pgx/lib/caqti_driver_pgx.ml

@@ -272,7 +272,9 @@ module Connect_functor (System : Caqti_platform.System_sig.S) = struct
        | Failure msg -> (* Raised by our Pgx.Io implementation. *)
           Fiber.return (Error (h (Caqti_error.Msg msg)))
        | exn ->
-          raise exn)
+          (match Net.convert_io_exception exn with
+           | Some msg -> Fiber.return (Error (h msg))
+           | None -> raise exn))
 
   let intercept_request_failed ~uri ~query =
     intercept (Caqti_error.request_failed ~uri ~query)

caqti-eio/lib/system.ml

@@ -134,6 +134,10 @@ module Net = struct
     with Eio.Exn.Io _ as exn ->
       Error (`Msg (Format.asprintf "%a" Eio.Exn.pp exn))
 
+  let convert_io_exception = function
+   | Eio.Exn.Io (err, ctx) -> Some (Msg_io (err, ctx))
+   | _ -> None
+
   module Socket = struct
     type t = {
       flow: [Eio.Flow.two_way_ty | Eio.Resource.close_ty] Eio.Resource.t;

caqti-lwt/lib-unix/system.ml

@@ -81,6 +81,10 @@ module Net = struct
             (`Msg ("Cannot resolve host name: " ^ Unix.error_message code))
        | exn -> Lwt.fail exn)
 
+  let convert_io_exception = function
+   | Unix.Unix_error (err, fn, arg) -> Some (Msg_unix (err, fn, arg))
+   | _ -> None
+
   type socket = {
     fd: Lwt_unix.file_descr option;
     ic: Lwt_io.input_channel;

caqti-mirage/lib/caqti_mirage.ml

@@ -110,6 +110,10 @@ struct
          | false, false ->
             Lwt.return (Error (`Msg "No IP address assigned to host.")))
 
+      let convert_io_exception = function
+       | Failure msg -> Some (Caqti_error.Msg msg) (* Channel.S.error *)
+       | _ -> None
+
       module Make_stream_ops (Channel : Mirage_channel.S) = struct
         type t = Channel.t
 

caqti/lib-blocking/caqti_blocking.ml

@@ -88,6 +88,7 @@ module System = struct
   include System_core
 
   module Net = struct
+
     module Sockaddr = struct
       type t = Unix.sockaddr
       let unix s = Unix.ADDR_UNIX s
@@ -105,6 +106,10 @@ module System = struct
        | Unix.Unix_error (code, _, _) ->
           Error (`Msg ("Cannot resolve host name: " ^ Unix.error_message code))
 
+    let convert_io_exception = function
+     | Unix.Unix_error (err, fn, arg) -> Some (Msg_unix (err, fn, arg))
+     | _ -> None
+
     module Socket = struct
       type t = Tcp of in_channel * out_channel
 

caqti/lib-platform/system_sig.ml

@@ -159,6 +159,10 @@ module type NET = sig
       error return indicates that an appropriate DNS server could not be
       queried. *)
 
+  val convert_io_exception : exn -> Caqti_error.msg option
+  (** If the read and write operations in Socket raise exceptions other than
+      {!End_of_file} and {!Failure}, this function is used to intercept them. *)
+
   (** A socket with input and output channels and dedicated IO functions.  This
       bundling is done to support the various APIs involved for networking and
       StartTLS. *)