Clean stale builds from state

lib/common.ml

@@ -2,6 +2,21 @@ open Devkit
 
 module Slack_timestamp = Fresh (String) ()
 
+module Timestamp = struct
+  type t = Ptime.t
+
+  let wrap s =
+    match Ptime.of_rfc3339 s with
+    | Ok (t, _, _) -> t
+    | Error _ -> failwith "Invalid timestamp"
+  let unwrap t = Ptime.to_rfc3339 t
+
+  let wrap_with_fallback ?(fallback = Ptime_clock.now ()) s =
+    match Ptime.of_rfc3339 s with
+    | Ok (t, _, _) -> t
+    | Error _ -> fallback
+end
+
 module Slack_channel : sig
   type 'kind t
 

lib/dune

@@ -1,8 +1,27 @@
 (library
  (name lib)
- (libraries atdgen atdgen-runtime biniou cstruct curl curl.lwt base64
-   devkit devkit.core extlib hex lwt lwt.unix nocrypto omd re2 sexplib0 uri
-   yojson)
+ (libraries
+  atdgen
+  atdgen-runtime
+  base64
+  biniou
+  cstruct
+  curl
+  curl.lwt
+  devkit
+  devkit.core
+  extlib
+  hex
+  lwt
+  lwt.unix
+  nocrypto
+  omd
+  ptime
+  ptime.clock
+  re2
+  sexplib0
+  uri
+  yojson)
  (preprocess
   (pps lwt_ppx)))
 

lib/state.atd

@@ -3,7 +3,8 @@ type 'v map_as_object <ocaml from="Common"> = abstract
 type 'v int_map_as_object <ocaml from="Common"> = abstract
 type 'v table_as_object <ocaml from="Common"> = abstract
 type string_set <ocaml from="Common"> = abstract
-type timestamp = string wrap <ocaml t="Common.Slack_timestamp.t" wrap="Common.Slack_timestamp.inject" unwrap="Common.Slack_timestamp.project">
+type slack_timestamp = string wrap <ocaml t="Common.Slack_timestamp.t" wrap="Common.Slack_timestamp.inject" unwrap="Common.Slack_timestamp.project">
+type timestamp = string wrap <ocaml module="Common.Timestamp">
 type user_id = string wrap <ocaml t="Common.Slack_user_id.t" wrap="Common.Slack_user_id.inject" unwrap="Common.Slack_user_id.project">
 type channel_id = string wrap <ocaml t="Common.Slack_channel.Ident.t" wrap="Common.Slack_channel.Ident.inject" unwrap="Common.Slack_channel.Ident.project">
 type any_channel = string wrap <ocaml t="Common.Slack_channel.Any.t" wrap="Common.Slack_channel.Any.inject" unwrap="Common.Slack_channel.Any.project">
@@ -26,8 +27,8 @@ type build_status = {
   commit: ci_commit;
   is_finished: bool;
   failed_steps: failed_step list;
-  created_at: string;
-  finished_at: string nullable;
+  created_at: timestamp;
+  finished_at: timestamp nullable;
 }
 
 (* A map from builds numbers to build statuses *)
@@ -50,7 +51,7 @@ type commit_sets = {
 type pipeline_commits = commit_sets map_as_object
 
 type slack_thread = {
-  ts: timestamp;
+  ts: slack_timestamp;
   channel: any_channel;
   cid: channel_id;
 }

lib/state.ml

@@ -63,7 +63,7 @@ let set_repo_pipeline_status { state } (n : Github_t.status_notification) =
     in
     let finished_at =
       match is_finished with
-      | true -> Some n.updated_at
+      | true -> Some (Timestamp.wrap_with_fallback n.updated_at)
       | false -> None
     in
     (* Even if this is an initial build state, we can't just set an "empty" state because we don't know
@@ -84,7 +84,7 @@ let set_repo_pipeline_status { state } (n : Github_t.status_notification) =
         commit;
         is_finished;
         failed_steps;
-        created_at = n.updated_at;
+        created_at = Timestamp.wrap_with_fallback n.updated_at;
         finished_at;
       }
     in
@@ -107,11 +107,24 @@ let set_repo_pipeline_status { state } (n : Github_t.status_notification) =
     in
     let rm_successful_build =
       update_builds_in_branches ~branches:n.branches ~f:(fun builds_map ->
+          let open Ptime in
+          let threshold =
+            (* 2h as the threshold for long running or stale builds *)
+            Ptime.Span.of_int_s (60 * 60 * 2)
+          in
+          let is_past_threshold (build_status : State_t.build_status) threshold =
+            let now = Ptime_clock.now () in
+            match add_span build_status.created_at threshold with
+            | Some t_plus_threshold -> is_earlier ~than:now t_plus_threshold
+            | None -> false
+          in
           IntMap.remove build_number builds_map
           |> IntMap.filter (fun build_number' build_status ->
-                 (* Remove old builds without failed steps because they were fixed and cleaned from state *)
                  match build_status.State_t.failed_steps with
+                 (* Remove old builds without failed steps because they were fixed and cleaned from state *)
                  | [] when build_number' < build_number && build_status.is_finished -> false
+                 (* Remove builds that ran for more than the threshold or for which we didn't get an end notification *)
+                 | _ when is_past_threshold build_status threshold -> false
                  | _ -> true))
     in
     let rm_successful_step =

lib/util.ml

@@ -101,8 +101,12 @@ module Build = struct
                      String.compare s1.name s2.name)
             in
             let current_build =
-              (* We will not get an exception here because we checked that the build is failed and finished *)
-              IntMap.find current_build_number builds_maps
+              try IntMap.find current_build_number builds_maps
+              with _ ->
+                (* edge case: we got a notification for a build that ran longer than the defined threshold
+                   and was cleaned from state. This shouldn't happen, but adding an error message to make
+                   clearer what is happening if it does. *)
+                failwith "Error: failed to find current build in state, maybe it was cleaned up?"
             in
             List.filter
               (fun (step : State_t.failed_step) ->