commit 70fe73413fdf0b2224ec54f921e1e54af797da81 · anil.recoil.org/ocaml-conpool

+51 -53

lib/conpool.ml

···

       28
        
         | Invalid_config of string

     

       29
        
         | Invalid_endpoint of string

     

       30
        
       

     

       31
       -
       exception Pool_error of error

     

       32
       -
       

     

       33
        
       let pp_error ppf = function

     

       34
        
         | Dns_resolution_failed { hostname } ->

     

       35
        
             Fmt.pf ppf "DNS resolution failed for hostname: %s" hostname

     
···

       114
        
       

     

       115
        
       let resolve_endpoint (pool : ('clock, 'net) internal) endpoint =

     

       116
        
         Log.debug (fun m -> m "Resolving %a..." Endpoint.pp endpoint);

     

       117
       -
         let addrs =

     

       118
       -
           Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint)

     

       119
       -
             ~service:(string_of_int (Endpoint.port endpoint))

     

       120
       -
         in

     

       121
       -
         Log.debug (fun m -> m "Got address list for %a" Endpoint.pp endpoint);

     

       122
       -
         match addrs with

     

       123
       -
         | addr :: _ ->

     

       124
       -
             Log.debug (fun m ->

     

       125
       -
                 m "Resolved %a to %a" Endpoint.pp endpoint Eio.Net.Sockaddr.pp addr);

     

       126
       -
             addr

     

       127
       -
         | [] ->

     

       128
       -
             Log.err (fun m ->

     

       129
       -
                 m "Failed to resolve hostname: %s" (Endpoint.host endpoint));

     

       130
       -
             raise

     

       131
       -
               (Pool_error

     

       132
       -
                  (Dns_resolution_failed { hostname = Endpoint.host endpoint }))

     

       0
        
       
     

       0
        
       
     

       133
        
       

     

       134
        
       (** {1 Connection Creation with Retry} *)

     

       135
        
       

     
···

       140
        
           Log.err (fun m ->

     

       141
        
               m "Failed to connect to %a after %d attempts" Endpoint.pp endpoint

     

       142
        
                 retry_count);

     

       143
       -
           raise

     

       144
       -
             (Pool_error

     

       145
       -
                (Connection_failed { endpoint; attempts = retry_count; last_error }))

     

       146
        
         end;

     

       147
        
       

     

       148
        
         Log.debug (fun m ->

     
···

       155
        
       

     

       156
        
           (* Connect with optional timeout *)

     

       157
        
           let socket =

     

       158
       -
             match Config.connect_timeout pool.config with

     

       159
       -
             | Some timeout ->

     

       160
       -
                 Eio.Time.with_timeout_exn pool.clock timeout (fun () ->

     

       161
       -
                     Eio.Net.connect ~sw:pool.sw pool.net addr)

     

       162
       -
             | None -> Eio.Net.connect ~sw:pool.sw pool.net addr

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       163
        
           in

     

       164
        
       

     

       165
        
           Log.debug (fun m ->

     
···

       170
        
             | None ->

     

       171
        
                 (socket :> connection)

     

       172
        
             | Some tls_config ->

     

       173
       -
                 Log.debug (fun m ->

     

       174
       -
                     m "Initiating TLS handshake with %a" Endpoint.pp endpoint);

     

       175
       -
                 let host =

     

       176
       -
                   Domain_name.(host_exn (of_string_exn (Endpoint.host endpoint)))

     

       177
       -
                 in

     

       178
       -
                 let tls_flow = Tls_eio.client_of_flow ~host tls_config socket in

     

       179
       -
                 Log.info (fun m ->

     

       180
       -
                     m "TLS connection established to %a" Endpoint.pp endpoint);

     

       181
       -
                 (tls_flow :> connection)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       182
        
           in

     

       183
        
       

     

       184
        
           let now = get_time pool in

     
···

       192
        
             mutex = Eio.Mutex.create ();

     

       193
        
           }

     

       194
        
         with

     

       195
       -
         | Eio.Time.Timeout as e ->

     

       196
        
             Log.warn (fun m ->

     

       197
        
                 m "Connection timeout to %a (attempt %d)" Endpoint.pp endpoint attempt);

     

       198
       -
             let error_msg = Printexc.to_string e in

     

       199
        
             if attempt >= Config.connect_retry_count pool.config then

     

       200
        
               (* Last attempt - convert to our error type *)

     

       201
        
               match Config.connect_timeout pool.config with

     

       202
        
               | Some timeout ->

     

       203
       -
                   raise (Pool_error (Connection_timeout { endpoint; timeout }))

     

       204
        
               | None ->

     

       205
       -
                   raise

     

       206
       -
                     (Pool_error

     

       207
       -
                        (Connection_failed

     

       208
       -
                           { endpoint; attempts = attempt; last_error = error_msg }))

     

       209
        
             else begin

     

       210
        
               (* Retry with exponential backoff *)

     

       211
        
               let delay =

     
···

       213
        
                 *. (2.0 ** float_of_int (attempt - 1))

     

       214
        
               in

     

       215
        
               Eio.Time.sleep pool.clock delay;

     

       216
       -
               create_connection_with_retry pool endpoint (attempt + 1) error_msg

     

       217
        
             end

     

       218
       -
         | e ->

     

       219
       -
             (* Other errors - retry with backoff *)

     

       220
       -
             let error_msg = Printexc.to_string e in

     

       221
        
             Log.warn (fun m ->

     

       222
        
                 m "Connection attempt %d to %a failed: %s" attempt Endpoint.pp

     

       223
        
                   endpoint error_msg);

     
···

       229
        
               Eio.Time.sleep pool.clock delay;

     

       230
        
               create_connection_with_retry pool endpoint (attempt + 1) error_msg)

     

       231
        
             else

     

       232
       -
               raise

     

       233
       -
                 (Pool_error

     

       234
       -
                    (Connection_failed

     

       235
       -
                       { endpoint; attempts = attempt; last_error = error_msg }))

     

       236
        
       

     

       237
        
       let create_connection (pool : ('clock, 'net) internal) endpoint =

     

       238
        
         create_connection_with_retry pool endpoint 1 "No attempts made"

     
···

       556
        
       

     

       557
        
       let with_connection t endpoint f =

     

       558
        
         Eio.Switch.run (fun sw -> f (connection ~sw t endpoint))

     

       559
       -
       

     

       560
       -
       let with_connection_exn t endpoint f =

     

       561
       -
         try with_connection t endpoint f with Pool_error e -> raise (err e)

     

       562
        
       

     

       563
        
       (** {1 Public API - Statistics} *)

     

       564

···

       28
        
         | Invalid_config of string

     

       29
        
         | Invalid_endpoint of string

     

       30
        
       

     

       0
        
       
     

       0
        
       
     

       31
        
       let pp_error ppf = function

     

       32
        
         | Dns_resolution_failed { hostname } ->

     

       33
        
             Fmt.pf ppf "DNS resolution failed for hostname: %s" hostname

     
···

       112
        
       

     

       113
        
       let resolve_endpoint (pool : ('clock, 'net) internal) endpoint =

     

       114
        
         Log.debug (fun m -> m "Resolving %a..." Endpoint.pp endpoint);

     

       115
       +
         try

     

       116
       +
           let addrs =

     

       117
       +
             Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint)

     

       118
       +
               ~service:(string_of_int (Endpoint.port endpoint))

     

       119
       +
           in

     

       120
       +
           Log.debug (fun m -> m "Got address list for %a" Endpoint.pp endpoint);

     

       121
       +
           match addrs with

     

       122
       +
           | addr :: _ ->

     

       123
       +
               Log.debug (fun m ->

     

       124
       +
                   m "Resolved %a to %a" Endpoint.pp endpoint Eio.Net.Sockaddr.pp addr);

     

       125
       +
               addr

     

       126
       +
           | [] ->

     

       127
       +
               Log.err (fun m ->

     

       128
       +
                   m "Failed to resolve hostname: %s" (Endpoint.host endpoint));

     

       129
       +
               raise (err (Dns_resolution_failed { hostname = Endpoint.host endpoint }))

     

       130
       +
         with Eio.Io _ as ex ->

     

       131
       +
           let bt = Printexc.get_raw_backtrace () in

     

       132
       +
           Eio.Exn.reraise_with_context ex bt "resolving %a" Endpoint.pp endpoint

     

       133
        
       

     

       134
        
       (** {1 Connection Creation with Retry} *)

     

       135
        
       

     
···

       140
        
           Log.err (fun m ->

     

       141
        
               m "Failed to connect to %a after %d attempts" Endpoint.pp endpoint

     

       142
        
                 retry_count);

     

       143
       +
           raise (err (Connection_failed { endpoint; attempts = retry_count; last_error }))

     

       0
        
       
     

       0
        
       
     

       144
        
         end;

     

       145
        
       

     

       146
        
         Log.debug (fun m ->

     
···

       153
        
       

     

       154
        
           (* Connect with optional timeout *)

     

       155
        
           let socket =

     

       156
       +
             try

     

       157
       +
               match Config.connect_timeout pool.config with

     

       158
       +
               | Some timeout ->

     

       159
       +
                   Eio.Time.with_timeout_exn pool.clock timeout (fun () ->

     

       160
       +
                       Eio.Net.connect ~sw:pool.sw pool.net addr)

     

       161
       +
               | None -> Eio.Net.connect ~sw:pool.sw pool.net addr

     

       162
       +
             with Eio.Io _ as ex ->

     

       163
       +
               let bt = Printexc.get_raw_backtrace () in

     

       164
       +
               Eio.Exn.reraise_with_context ex bt "connecting to %a" Endpoint.pp endpoint

     

       165
        
           in

     

       166
        
       

     

       167
        
           Log.debug (fun m ->

     
···

       172
        
             | None ->

     

       173
        
                 (socket :> connection)

     

       174
        
             | Some tls_config ->

     

       175
       +
                 try

     

       176
       +
                   Log.debug (fun m ->

     

       177
       +
                       m "Initiating TLS handshake with %a" Endpoint.pp endpoint);

     

       178
       +
                   let host =

     

       179
       +
                     Domain_name.(host_exn (of_string_exn (Endpoint.host endpoint)))

     

       180
       +
                   in

     

       181
       +
                   let tls_flow = Tls_eio.client_of_flow ~host tls_config socket in

     

       182
       +
                   Log.info (fun m ->

     

       183
       +
                       m "TLS connection established to %a" Endpoint.pp endpoint);

     

       184
       +
                   (tls_flow :> connection)

     

       185
       +
                 with Eio.Io _ as ex ->

     

       186
       +
                   let bt = Printexc.get_raw_backtrace () in

     

       187
       +
                   Eio.Exn.reraise_with_context ex bt "TLS handshake with %a" Endpoint.pp endpoint

     

       188
        
           in

     

       189
        
       

     

       190
        
           let now = get_time pool in

     
···

       198
        
             mutex = Eio.Mutex.create ();

     

       199
        
           }

     

       200
        
         with

     

       201
       +
         | Eio.Time.Timeout ->

     

       202
        
             Log.warn (fun m ->

     

       203
        
                 m "Connection timeout to %a (attempt %d)" Endpoint.pp endpoint attempt);

     

       0
        
       
     

       204
        
             if attempt >= Config.connect_retry_count pool.config then

     

       205
        
               (* Last attempt - convert to our error type *)

     

       206
        
               match Config.connect_timeout pool.config with

     

       207
        
               | Some timeout ->

     

       208
       +
                   raise (err (Connection_timeout { endpoint; timeout }))

     

       209
        
               | None ->

     

       210
       +
                   raise (err (Connection_failed

     

       211
       +
                           { endpoint; attempts = attempt; last_error = "Timeout" }))

     

       0
        
       
     

       0
        
       
     

       212
        
             else begin

     

       213
        
               (* Retry with exponential backoff *)

     

       214
        
               let delay =

     
···

       216
        
                 *. (2.0 ** float_of_int (attempt - 1))

     

       217
        
               in

     

       218
        
               Eio.Time.sleep pool.clock delay;

     

       219
       +
               create_connection_with_retry pool endpoint (attempt + 1) "Timeout"

     

       220
        
             end

     

       221
       +
         | Eio.Io _ as ex ->

     

       222
       +
             (* Eio IO errors - retry with backoff and add context on final failure *)

     

       223
       +
             let error_msg = Printexc.to_string ex in

     

       224
        
             Log.warn (fun m ->

     

       225
        
                 m "Connection attempt %d to %a failed: %s" attempt Endpoint.pp

     

       226
        
                   endpoint error_msg);

     
···

       232
        
               Eio.Time.sleep pool.clock delay;

     

       233
        
               create_connection_with_retry pool endpoint (attempt + 1) error_msg)

     

       234
        
             else

     

       235
       +
               let bt = Printexc.get_raw_backtrace () in

     

       236
       +
               Eio.Exn.reraise_with_context ex bt "after %d retry attempts" attempt

     

       0
        
       
     

       0
        
       
     

       237
        
       

     

       238
        
       let create_connection (pool : ('clock, 'net) internal) endpoint =

     

       239
        
         create_connection_with_retry pool endpoint 1 "No attempts made"

     
···

       557
        
       

     

       558
        
       let with_connection t endpoint f =

     

       559
        
         Eio.Switch.run (fun sw -> f (connection ~sw t endpoint))

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       560
        
       

     

       561
        
       (** {1 Public API - Statistics} *)

     

       562

+21 -16

lib/conpool.mli

···

       47
        
         | Invalid_config of string  (** Invalid configuration parameter *)

     

       48
        
         | Invalid_endpoint of string  (** Invalid endpoint specification *)

     

       49
        
       

     

       50
       -
       exception Pool_error of error

     

       51
       -
       (** Exception raised by pool operations.

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       52
        
       

     

       53
       -
           Most pool operations can raise this exception. Use {!pp_error} to get

     

       54
       -
           human-readable error messages. *)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       55
        
       

     

       56
       -
       type Eio.Exn.err += E of error

     

       57
       -
       (** Extension of Eio's error type for connection pool errors. *)

     

       58
        
       

     

       59
        
       val err : error -> exn

     

       60
        
       (** [err e] is [Eio.Exn.create (E e)].

     

       61
        
       

     

       62
        
           This converts a connection pool error to an Eio exception, allowing it to

     

       63
       -
           be handled uniformly with other Eio I/O errors. *)

     

       0
        
       
     

       64
        
       

     

       65
        
       val pp_error : error Fmt.t

     

       66
       -
       (** Pretty-printer for error values. *)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       67
        
       

     

       68
        
       (** {1 Connection Types} *)

     

       69
        
       

     
···

       142
        
                 let buf = Eio.Buf_read.of_flow conn ~max_size:4096 in

     

       143
        
                 Eio.Buf_read.take_all buf)

     

       144
        
           ]} *)

     

       145
       -
       

     

       146
       -
       val with_connection_exn : t -> Endpoint.t -> (connection -> 'a) -> 'a

     

       147
       -
       (** [with_connection_exn pool endpoint fn] is like {!with_connection} but

     

       148
       -
           converts {!Pool_error} exceptions to [Eio.Io] exceptions for better

     

       149
       -
           integration with Eio error handling.

     

       150
       -
       

     

       151
       -
           This is useful when you want pool errors to be handled uniformly with other

     

       152
       -
           Eio I/O errors. *)

     

       153
        
       

     

       154
        
       (** {1 Statistics & Monitoring} *)

     

       155

···

       47
        
         | Invalid_config of string  (** Invalid configuration parameter *)

     

       48
        
         | Invalid_endpoint of string  (** Invalid endpoint specification *)

     

       49
        
       

     

       50
       +
       type Eio.Exn.err += E of error

     

       51
       +
       (** Extension of Eio's error type for connection pool errors.

     

       52
       +
       

     

       53
       +
           Pool operations raise [Eio.Io] exceptions with context information added at

     

       54
       +
           each layer. The innermost error is often [E error], wrapped with context

     

       55
       +
           strings that describe the operation being performed.

     

       56
        
       

     

       57
       +
           Example error message:

     

       58
       +
           {[

     

       59
       +
             Eio.Io Conpool Dns_resolution_failed { hostname = "invalid.example" },

     

       60
       +
               resolving invalid.example:443,

     

       61
       +
               connecting to invalid.example:443,

     

       62
       +
               after 3 retry attempts

     

       63
       +
           ]}

     

       64
        
       

     

       65
       +
           Use {!pp_error} to format just the error code, or let Eio format the full

     

       66
       +
           exception with context. *)

     

       67
        
       

     

       68
        
       val err : error -> exn

     

       69
        
       (** [err e] is [Eio.Exn.create (E e)].

     

       70
        
       

     

       71
        
           This converts a connection pool error to an Eio exception, allowing it to

     

       72
       +
           be handled uniformly with other Eio I/O errors and enabling context to be

     

       73
       +
           added via [Eio.Exn.reraise_with_context]. *)

     

       74
        
       

     

       75
        
       val pp_error : error Fmt.t

     

       76
       +
       (** Pretty-printer for error values (without context).

     

       77
       +
       

     

       78
       +
           For full error messages including context, use [Eio.Exn.pp] or simply let

     

       79
       +
           the exception be printed naturally. *)

     

       80
        
       

     

       81
        
       (** {1 Connection Types} *)

     

       82
        
       

     
···

       155
        
                 let buf = Eio.Buf_read.of_flow conn ~max_size:4096 in

     

       156
        
                 Eio.Buf_read.take_all buf)

     

       157
        
           ]} *)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       158
        
       

     

       159
        
       (** {1 Statistics & Monitoring} *)

     

       160