Bug/2281 reconnect in worker

pool/app/invitation/dune

@@ -2,6 +2,6 @@
  (name invitation)
  (libraries contact experiment filter mailing pool_common sihl sihl-email)
  (preprocess
-  (pps lwt_ppx ppx_deriving.eq ppx_deriving.show)))
+  (pps lwt_ppx ppx_deriving.eq ppx_deriving.show ppx_string)))
 
 (include_subdirs unqualified)

pool/app/invitation/repo/repo.ml

@@ -222,7 +222,7 @@ let bulk_insert ?mailing_id pool contacts experiment_id =
     |sql}
   in
   let invitations = CCList.map CCFun.(uncurry (fun id -> Entity.create ~id)) contacts in
-  let values, value_insert =
+  let Dynparam.Pack (pt, pv), value_insert =
     CCList.fold_left
       (fun (dyn, sql) entity ->
          let sql_line =
@@ -244,10 +244,10 @@ let bulk_insert ?mailing_id pool contacts experiment_id =
       (Dynparam.empty, [])
       invitations
   in
-  let (Dynparam.Pack (pt, pv)) = values in
   let prepare_request =
     let open Caqti_request.Infix in
-    Format.asprintf "%s \n %s" insert_sql (CCString.concat ",\n" value_insert)
+    let values = CCString.concat ",\n" value_insert in
+    [%string "%{insert_sql} \n %{values} \n ON DUPLICATE KEY UPDATE updated_at = now()"]
     |> (pt ->. Caqti_type.unit) ~oneshot:true
   in
   let%lwt () = Database.exec pool prepare_request pv in

pool/database/database.mli

@@ -270,7 +270,7 @@ module Pool : sig
     -> (t, Pool_message.Error.t) Lwt_result.t
 
   val connect : Label.t -> (unit, Pool_message.Error.t) Lwt_result.t
-  val disconnect : ?error:Caqti_error.t -> Entity.Label.t -> unit Lwt.t
+  val disconnect : ?error:Caqti_error.t -> Label.t -> unit Lwt.t
   val all : ?allowed_status:Status.t list -> ?exclude:Label.t list -> unit -> Label.t list
   val is_root : Label.t -> bool
 

pool/database/pools.ml

@@ -206,7 +206,6 @@ module Make (Config : Pools_sig.ConfigSig) = struct
       | Ok resp -> Lwt.return resp
       | Error `Unsupported -> raise Pool_message.Error.(Exn (Unsupported "Caqti error"))
       | Error (#load_or_connect as err) ->
-        Logs.info (fun m -> m "ERROR load or connect: %s" (Caqti_error.show err));
         let%lwt () = disconnect ~error:err label in
         let () = Cache.log_pools () in
         raise (Exn err)

pool/pool_queue/pool_queue.ml

@@ -58,10 +58,7 @@ let dev_dispatch
     m ?tags "Environment is not 'production' and/or var `QUEUE_FORCE_ASYNC` not set");
   match%lwt decode input |> Lwt_result.lift >>= handle database_label with
   | Ok () -> callback instance
-  | Error msg ->
-    let job = Instance.(instance |> add_error msg |> failed) in
-    Logs.err (fun m -> m ?tags "Job failed: %s" ([%show: Instance.t] job));
-    Lwt.return_unit
+  | Error msg -> Instance.default_error_handler database_label msg instance
 ;;
 
 let dispatch
@@ -234,12 +231,6 @@ let create_schedule (database_label, (job : AnyJob.t)) : Schedule.t =
 
 let start () =
   let tags = Database.Logger.Tags.create Database.Pool.Root.label in
-  let database_labels = Database.(Pool.Tenant.all ~status:[ Status.Active ] ()) in
-  Logs.info (fun m ->
-    m
-      ~tags
-      "Start queue for databases: %s"
-      ([%show: Database.Label.t list] database_labels));
   let archive_and_reset database_label =
     let%lwt () = Repo.archive_all_processed database_label in
     Repo.reset_pending_jobs database_label
@@ -254,12 +245,18 @@ let start () =
       Lwt.return_unit
     | err -> raise err
   in
-  let%lwt () = Lwt_list.iter_s (handle archive_and_reset) database_labels in
   match !registered_jobs with
   | [] ->
     Logs.warn (fun m -> m ~tags "No jobs registered");
     Lwt.return_unit
   | jobs ->
+    let database_labels = Database.Pool.Tenant.all () in
+    Logs.info (fun m ->
+      m
+        ~tags
+        "Start queue for databases: %s"
+        ([%show: Database.Label.t list] database_labels));
+    let%lwt () = Lwt_list.iter_s (handle archive_and_reset) database_labels in
     let jobs_per_database =
       let open CCList in
       fold_left

pool/pool_queue/repo.ml

@@ -41,6 +41,7 @@ let sql_select_columns ?(decode = true) table =
   ; go "last_error_at"
   ; go "database_label"
   ; go_binary "clone_of"
+  ; go "created_at"
   ]
 ;;
 
@@ -157,7 +158,11 @@ let count_all_workable_request =
 ;;
 
 let count_all_workable label =
-  Database.find_opt label count_all_workable_request ()
+  Lwt.catch
+    (fun () -> Database.find_opt label count_all_workable_request ())
+    (function
+      | Caqti_error.(Exn #load_or_connect) -> Lwt.return_none
+      | exn -> Lwt.reraise exn)
   ||> CCOption.get_or ~default:0
   ||> CCResult.return
 ;;

pool/schedule/schedule.ml

@@ -86,8 +86,9 @@ module Registered = struct
   ;;
 end
 
-let run ({ label; scheduled_time; status; _ } as schedule : t) =
+let run ({ database_label; label; scheduled_time; status; _ } as schedule : t) =
   let open Utils.Lwt_result.Infix in
+  let tags = CCOption.map_or ~default:tags Database.Logger.Tags.create database_label in
   let delay = run_in scheduled_time in
   let notify status =
     Logs.debug ~src (fun m -> m ~tags "%s: Run is %s" label (Status.show status));
@@ -110,26 +111,28 @@ let run ({ label; scheduled_time; status; _ } as schedule : t) =
     let rerun () = Lwt_unix.sleep delay >|> loop in
     let database_ok () =
       let open Database in
+      let open Status in
       let database_status =
-        match schedule.database_label with
-        | None -> Status.Active
-        | Some label ->
-          Pool.Tenant.find_status_by_label label |> CCOption.value ~default:Status.Active
+        CCOption.map_or ~default:Active (fun label ->
+          match Pool.Tenant.find label with
+          | Ok database -> status database
+          | Error _ -> Disabled)
       in
-      let open Status in
       let retry_connection label =
-        Database.Pool.connect label
+        (* Reset database connection to "Close" before retry *)
+        let%lwt () = Pool.disconnect label in
+        Pool.connect label
         >|> function
         | Ok () ->
           let%lwt () =
             let open Pool_database in
-            StatusUpdated (label, Status.Active) |> handle_event Database.Pool.Root.label
+            StatusUpdated (label, Active) |> handle_event Pool.Root.label
           in
           Lwt.return_true
         | Error _ -> Lwt.return_false
       in
-      match database_status with
-      | Active -> Lwt.return true
+      match database_status database_label with
+      | Active -> Lwt.return_true
       | ConnectionIssue ->
         CCOption.map_or ~default:Lwt.return_false retry_connection schedule.database_label
       | Disabled