state: add lock

lib/action.ml

@@ -100,7 +100,7 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
     let rules = cfg.status_rules.rules in
     let action_on_match (branches : branch list) =
       let default = Option.to_list cfg.prefix_rules.default_channel in
-      State.set_repo_pipeline_status ctx.state repo.url ~pipeline ~branches ~status:current_status;
+      let%lwt () = State.set_repo_pipeline_status ctx.state repo.url ~pipeline ~branches ~status:current_status in
       match List.is_empty branches with
       | true -> Lwt.return []
       | false ->
@@ -118,7 +118,7 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
         )
     in
     if Context.is_pipeline_allowed ctx repo.url ~pipeline then begin
-      let repo_state = State.find_or_add_repo ctx.state repo.url in
+      let%lwt repo_state = State.find_or_add_repo ctx.state repo.url in
       match Rule.Status.match_rules ~rules n with
       | Some Ignore | None -> Lwt.return []
       | Some Allow -> action_on_match n.branches
@@ -261,7 +261,7 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
     let fetch_bot_user_id () =
       match%lwt Slack_api.send_auth_test ~ctx () with
       | Ok { user_id; _ } ->
-        ctx.state.bot_user_id <- Some user_id;
+        State.set_bot_user_id ctx.state user_id;
         let%lwt () =
           Option.value_map ctx.state_filepath ~default:Lwt.return_unit ~f:(fun path ->
             match%lwt State.save ctx.state path with
@@ -298,7 +298,7 @@ module Action (Github_api : Api.Github) (Slack_api : Api.Slack) = struct
         )
     in
     let%lwt bot_user_id =
-      match ctx.state.bot_user_id with
+      match State.get_bot_user_id ctx.state with
       | Some id -> Lwt.return_some id
       | None -> fetch_bot_user_id ()
     in

lib/context.ml

@@ -12,7 +12,7 @@ type t = {
   state_filepath : string option;
   mutable secrets : Config_t.secrets option;
   config : Config_t.config Stringtbl.t;
-  state : State_t.state;
+  state : State.t;
 }
 
 let default_config_filename = "monorobot.json"
@@ -95,7 +95,7 @@ let refresh_state ctx =
       match get_local_file path with
       | Error e -> fmt_error "error while getting local file: %s\nfailed to get state from file %s" e path
       | Ok file ->
-        let state = State_j.state_of_string file in
+        let state = { ctx.state with state = State_j.state_of_string file } in
         Ok { ctx with state }
     end
     else Ok ctx

lib/state.ml

@@ -2,25 +2,45 @@ open Base
 open Common
 open Devkit
 
+type t = {
+  state : State_t.state;
+  lock : Lwt_mutex.t;
+}
+
 let empty_repo_state () : State_t.repo_state = { pipeline_statuses = StringMap.empty }
 
-let empty () : State_t.state = { repos = Stringtbl.empty (); bot_user_id = None }
+let empty () : t =
+  let state = State_t.{ repos = Stringtbl.empty (); bot_user_id = None } in
+  { state; lock = Lwt_mutex.create () }
+
+let find_or_add_repo' state repo_url = Stringtbl.find_or_add state.State_t.repos repo_url ~default:empty_repo_state
+
+let set_repo_state { state; lock } repo_url repo_state =
+  Lwt_mutex.with_lock lock @@ fun () ->
+  Stringtbl.set state.repos ~key:repo_url ~data:repo_state;
+  Lwt.return_unit
 
-let find_or_add_repo (state : State_t.state) repo_url =
-  Stringtbl.find_or_add state.repos repo_url ~default:empty_repo_state
+let find_or_add_repo { state; lock } repo_url =
+  Lwt_mutex.with_lock lock @@ fun () -> find_or_add_repo' state repo_url |> Lwt.return
 
-let set_repo_pipeline_status (state : State_t.state) repo_url ~pipeline ~(branches : Github_t.branch list) ~status =
+let set_repo_pipeline_status { state; lock } repo_url ~pipeline ~(branches : Github_t.branch list) ~status =
   let set_branch_status branch_statuses =
     let new_statuses = List.map branches ~f:(fun b -> b.name, status) in
     let init = Option.value branch_statuses ~default:(Map.empty (module String)) in
     List.fold_left new_statuses ~init ~f:(fun m (key, data) -> Map.set m ~key ~data)
   in
-  let repo_state = find_or_add_repo state repo_url in
-  repo_state.pipeline_statuses <- Map.update repo_state.pipeline_statuses pipeline ~f:set_branch_status
+  Lwt_mutex.with_lock lock @@ fun () ->
+  let repo_state = find_or_add_repo' state repo_url in
+  repo_state.pipeline_statuses <- Map.update repo_state.pipeline_statuses pipeline ~f:set_branch_status;
+  Lwt.return_unit
+
+let set_bot_user_id { state; _ } user_id = state.State_t.bot_user_id <- Some user_id
+
+let get_bot_user_id { state; _ } = state.State_t.bot_user_id
 
 let log = Log.from "state"
 
-let save state path =
+let save { state; _ } path =
   let data = State_j.string_of_state state |> Yojson.Basic.from_string |> Yojson.Basic.pretty_to_string in
   match write_to_local_file ~data path with
   | Ok () -> Lwt.return @@ Ok ()

test/test.ml

@@ -26,7 +26,7 @@ let process ~(secrets : Config_t.secrets) ~config (kind, path, state_path) =
     let repo = Github.repo_of_notification @@ Github.parse_exn headers event in
     let ctx = Context.make () in
     ctx.secrets <- Some secrets;
-    ignore (State.find_or_add_repo ctx.state repo.url);
+    let%lwt _ = State.find_or_add_repo ctx.state repo.url in
     match state_path with
     | None ->
       Context.set_repo_config ctx repo.url config;
@@ -38,7 +38,7 @@ let process ~(secrets : Config_t.secrets) ~config (kind, path, state_path) =
       Lwt.return ctx
     | Ok file ->
       let repo_state = State_j.repo_state_of_string file in
-      Common.Stringtbl.set ctx.state.repos ~key:repo.url ~data:repo_state;
+      let%lwt () = State.set_repo_state ctx.state repo.url repo_state in
       Context.set_repo_config ctx repo.url config;
       Lwt.return ctx
   in