Something went wrong. Try again.
A local Git merge queue with content-bound gate verdicts
Something went wrong. Try again.
17 kB · 480 lines
OCaml
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481exception Defect = Loop.Defectexception Not_implemented = Loop.Not_implemented
module M = Merge_queue
type context = { env : Eio_unix.Stdenv.base; cwd : Fpath.t; directory : (Fpath.t, string) result; lookup : string -> string option; environ : string list; environment : (Repo.environment, string) result; caller : (Merge_queue.Caller.t, string) result; uid : int; say : Merge_queue.Progress.t -> unit;}
let version = match Build_info.V1.version () with | Some v -> Build_info.V1.Version.to_string v | None -> "dev"
let caller env ~lookup = let clock = Clock.v (Eio.Stdenv.clock env) (Eio.Stdenv.mono_clock env) in let user = match lookup "USER" with | Some u when u <> "" -> u | _ -> ( try Unix.getlogin () with Unix.Unix_error _ -> "unknown") in match M.Host.of_string (Unix.gethostname ()) with | Error (`Msg m) -> Error m | Ok host -> Ok { Merge_queue.Caller.user; host; terminal = Console_eio.is_tty Unix.stdin; tz_offset_s = Clock.tz_offset_s clock; }
(* A variable of the environment, the last one when it is set twice, as getenv(3) reads it on the platforms mq runs on. *)let lookup_of environ = let vars = Array.fold_left (fun acc binding -> match String.index_opt binding '=' with | Some i -> ( String.sub binding 0 i, String.sub binding (i + 1) (String.length binding - i - 1) ) :: acc | None -> acc) [] environ in fun name -> List.assoc_opt name vars
(* Why [dir] cannot be the directory a command runs in, when it cannot. *)let not_a_directory fs dir = match Eio.Path.kind ~follow:true Eio.Path.(fs / Fpath.to_string dir) with | `Directory -> None | `Not_found -> Fmt.kstr Option.some "%a does not exist" Fpath.pp dir | _ -> Fmt.kstr Option.some "%a is not a directory" Fpath.pp dir
(* The current directory, then each -C in turn as git -C takes it: a relative one from the one before, each a directory to change to. *)let directory fs ~cwd dirs = List.fold_left (fun acc p -> Result.bind acc (fun dir -> let dir = Fpath.append dir p in match not_a_directory fs dir with | None -> Ok dir | Some why -> Error why)) (Ok cwd) dirs
let context env ~environ ~uid ~cwd dirs = let fs = (Eio.Stdenv.fs env :> Eio.Fs.dir_ty Eio.Path.t) in let lookup = lookup_of environ in { env; cwd; directory = directory fs ~cwd dirs; lookup; environ = Array.to_list environ; environment = Repo.environment ~fs ~lookup ~uid; caller = caller env ~lookup; uid; say = (fun p -> Fmt.epr "%a@." M.Progress.pp p); }
let todo what = raise (Not_implemented what)
let run_name (caller : M.Caller.t) ~now = Fmt.str "%a:%d:%s" M.Host.pp caller.host (Unix.getpid ()) (Timestamp.Rfc3339.format now)
(* {1 The queue a path names} *)
type queue = { perform : Perform.t; location : Repo.location; rules : (M.Rules.t * M.Commit.t) option; unloadable : (M.Commit.t * string) option; registration : M.Registration.t option; process : [ `Generic | `Unix ] Eio.Process.mgr_ty Eio.Resource.t;}
let refused ?branch ?next detail = M.Outcome.v ?branch ?next M.Outcome.Refused detail
let unavailable ?branch detail = M.Outcome.v ?branch M.Outcome.Unavailable detail
(* The repository's build slots: [mq.slots], else as many as the cores hold builds of the width a check runs at, which is every core while a check is given no width of its own. *)let slots ~fs ~mq_dir (local : M.Local_config.t) = let slots = match local.slots with | Some n -> n | None -> Build_slot.default_slots ~width:(Domain.recommended_domain_count ()) in Build_slot.v ~slots Eio.Path.(fs / Fpath.to_string (M.Slot.file ~mq_dir))
(* The shell's handles on one repository, once its local configuration is read. Nothing is written here: each store is made by its first write. *)(* Each check's median run over the last day of the journal, the window [mq stats] reads by default: the estimates the build slots are taken by (Order). A record the journal cannot read is left out. *)let estimates ~now journal = let day = Ptime.Span.of_int_s 86_400 in let since = Option.value ~default:Ptime.epoch (Ptime.sub_span now day) in let reading = Journal.since journal since in let flow = M.Flow.v { M.Flow.now; since; landings = []; moves = []; events = List.map (fun (r : Journal.record) -> r.event) reading.records; ranges = []; skipped = List.length reading.skipped; } in M.Order.estimates (List.filter_map (fun (c, (t : M.Flow.timing)) -> Option.map (fun m -> (c, m)) t.median_ms) flow.per_check)
let perform_of ctx ~sw ~fs ~repo ~root ~caller local = let clock = Clock.v (Eio.Stdenv.clock ctx.env) (Eio.Stdenv.mono_clock ctx.env) in let mq_dir = Repo.mq_dir repo in let logs = Fpath.(mq_dir / "logs") in let pool = M.Local_config.pool local ~repository:root in let locks = match Run_lock.group [ Repo.common_dir repo ] with | Ok group -> group | Error reason -> failwith reason in let journal = Journal.v ~fs mq_dir ~source:("file://" ^ Fpath.to_string (Repo.common_dir repo)) in let checks = { Checks.fs; repo; pool; logs; logs_bound = Checks.default_bound; clock; confine = Confine.v ~say:(fun line -> ctx.say (M.Progress.Line line)) ~process:(Eio.Stdenv.process_mgr ctx.env) ~fs ~lookup:ctx.lookup ~sleep:(Eio.Time.sleep (Eio.Stdenv.clock ctx.env)) ~repo ~pool local.M.Local_config.confine; path = Option.value ~default:"/usr/bin:/bin" (ctx.lookup "PATH"); home = Option.value ~default:"/" (ctx.lookup "HOME"); locks; slots = slots ~fs ~mq_dir local; owner = "mq"; estimates = estimates ~now:(Clock.now clock) journal; lease = Checks.lease (); } in { Perform.repo; journal; clock; caller; local; pool; checks; sw; fs; locks; run = None; repair = None; crontab = (fun () -> Scheduler.crontab ~fs ~path:(Option.value ~default:"" (ctx.lookup "PATH")) ~process: (Eio.Stdenv.process_mgr ctx.env :> [ `Generic | `Unix ] Eio.Process.mgr_ty Eio.Resource.t) ~sleep:(Eio.Time.sleep (Eio.Stdenv.clock ctx.env))); say = ctx.say; }
(* Why no repository serves [path]. *)let discovery_failure path : Repo.discovery_error -> string = function | No_such_directory -> Fmt.str "%a does not exist" Fpath.pp path | Not_a_directory -> Fmt.str "%a is not a directory" Fpath.pp path | No_repository -> Fmt.str "%a is not in a Git repository" Fpath.pp path | No_checkout common -> Fmt.str "%a is in the bare repository %a, which has no main checkout" Fpath.pp path Fpath.pp common | Refused why -> why
(* How a call failed to open its queue: refused, the path naming none or a key of the machine's holding a value the caller sets, with the command that fixes it; or unavailable, the queue unreadable. *)type failure = | Refuse of { why : string; next : string option } | Unavailable of string
let refuse why = Refuse { why; next = None }let ( let* ) = Result.bind
let config_of repo = Result.map_error (fun m -> Unavailable m) (Repo.configuration repo)
let keys ~environment c = Result.map_error (fun (r : M.Local_config.refusal) -> Refuse { why = r.reason; next = r.next }) (M.Local_config.v ~env:(Repo.config_env environment) c)
let network ctx = Repo.network ~net:(Eio.Stdenv.net ctx.env) ~mono:(Eio.Stdenv.mono_clock ctx.env)
(* The repository a directory names, opened at its main checkout, with its keys (doc/commands.md, -C PATH). *)let open_repo ~sw ~network ~environment ~hooks (location : Repo.location) = let repo = Repo.v ~network ~sw ~environment ~hooks location.checkout in let* local = Result.bind (config_of repo) (keys ~environment) in Ok (repo, local)
(* What mq schedule start recorded, and the keys a run reads under it: the rules branch the registration recorded, in place of mq.rules. *)let registered ~fs ~recorded repo local = match Scheduler.registered ~fs (Repo.mq_dir repo) with | Error m -> Error (Unavailable m) | Ok (Some r) when recorded -> Ok (Some r, M.Local_config.with_rules local r.M.Registration.rules) | Ok r -> Ok (r, local)
(* Whether [cwd] is outside the repository's main checkout and every worktree of it: a command sent elsewhere by -C or GIT_DIR, whose caller is told which repository it opened (doc/commands.md, -C PATH). *)let outside cwd (location : Repo.location) repo = let real p = match Unix.realpath (Fpath.to_string p) with | r -> Fpath.to_dir_path (Fpath.v r) | exception Unix.Unix_error _ -> Fpath.to_dir_path p in let cwd = real cwd in let within dir = Fpath.is_prefix (real dir) cwd in Result.map (fun worktrees -> not (within location.checkout || List.exists (fun (w : Repo.worktree) -> within w.path) worktrees)) (Repo.readable repo)
(* The rules at the tip of [local]'s rules branch. Rules that do not load there open the repository with no rules, and with the tip and why: each command answers them itself ({!initialised}, {!repairing}). A rules branch with no tip is unavailable. *)let rules_of repo (local : M.Local_config.t) = match Perform.read_rules repo local.rules with | Ok rules -> Ok (rules, None) | Error m -> ( match Repo.read_ref repo (M.Ref_name.branch local.rules) with | Some c -> Ok (None, Some (c, m)) | None -> Error (Unavailable m))
(* [repo] under the walls and the remote bound its rules set. *)let bounded repo = function | Some (r, _) -> let b = r.M.Rules.bounds in Repo.with_remote_bytes (Repo.with_wait_wall (Repo.with_hook_wall repo (M.Duration.of_seconds b.hook_wall)) (M.Duration.of_seconds b.wait_wall)) b.remote_bytes | None -> repo
(* mq's own reference transactions start no hook mq init wrote: the hook's committed run is {!hook_committed}, in the work tree it would run in. *)let rec open_queue ?(recorded = false) ctx ~sw ~writes = let fs = (Eio.Stdenv.fs ctx.env :> Eio.Fs.dir_ty Eio.Path.t) in let process = (Eio.Stdenv.process_mgr ctx.env :> [ `Generic | `Unix ] Eio.Process.mgr_ty Eio.Resource.t) in let hooks = Repo.hooks ~committed:(fun work_tree lines -> hook_committed { ctx with directory = Ok work_tree } lines) ~process ~clock: (Eio.Stdenv.clock ctx.env :> float Eio.Time.clock_ty Eio.Resource.t) ~wall:(M.Duration.of_seconds M.Rules.default_bounds.hook_wall) () in let* caller = Result.map_error (fun m -> Unavailable m) ctx.caller in let* environment = Result.map_error (fun m -> Unavailable m) ctx.environment in let* dir = Result.map_error refuse ctx.directory in let* location = Result.map_error (fun e -> refuse (discovery_failure dir e)) (Repo.discover environment dir) in (* A repository whose configuration or rules mq cannot read is one it cannot serve: unavailable, in the words of what failed. A key of the machine's whose value is not one is the caller's to fix: refused, with the command that removes it (doc/configuration.md). *) let* repo, local, (rules, unloadable), registration = Loop.catching (fun () -> let* repo, local = open_repo ~sw ~network:(network ctx) ~environment ~hooks location in let* registration, local = registered ~fs ~recorded repo local in let* rules = rules_of repo local in Ok (repo, local, rules, registration)) ~on_failure:(fun m -> Error (Unavailable m)) in let* announce = if not writes then Ok false else Result.map_error (fun u -> Fmt.kstr (fun s -> Unavailable s) "%a" Repo.pp_unreadable u) (outside ctx.cwd location repo) in if announce then Fmt.kstr (fun s -> ctx.say (M.Progress.Line s)) "repository: %a" Fpath.pp location.checkout; let repo = bounded repo rules in let perform = perform_of ctx ~sw ~fs ~repo ~root:location.checkout ~caller local in Ok { perform; location; rules; unloadable; registration; process }
(* Git aborts a transaction whose hook fails at preparing or prepared, and a move made outside mq is ordinary Git (doc/invariants.md, 1): the hook acts only once the transaction is committed, and whatever it meets, it writes nothing but the path the scheduler watches. *)and hook_committed ctx lines = match M.Registration.moved lines with | [] -> () | _ :: _ -> ( try Eio.Switch.run @@ fun sw -> match open_queue ctx ~sw ~writes:false with | Ok { rules = Some (rules, _); registration = Some r; perform; _ } when M.Registration.woken rules lines ~current:(fun name -> Option.map Git.Hash.to_hex (Repo.read_value perform.repo name)) -> Scheduler.wake ~fs:perform.fs r | Ok _ | Error _ -> () with Failure _ | Eio.Io _ | Unix.Unix_error _ | Sys_error _ -> ())
(* doc/runs.md: a target with branches queued and no run started within bounds.run_start is one event, and so is an edge left unserved for twice its interval. Nobody waits on a branch queued with --no-wait nor on an edge, so each command that writes journals the crossings its reading of the queue finds and the journal does not record yet. A queue it cannot read is left to the command's own reading, which names it. *)let journal_unserved q = match (q.rules, q.registration) with | None, _ | _, None -> Ok () | Some (rules, _), Some _ -> let p = q.perform in let run_start = M.Duration.of_seconds rules.M.Rules.bounds.run_start in let now = Perform.now p in let pins = match Result.bind (Repo.snapshot p.repo [ M.Ref_name.Prefix.queue ]) (Queue_read.collect (Queue_read.pin_of p.repo)) with | Ok pins -> pins | Error _ -> [] in let behind (e : M.Schedule.t) = e.behind > 0 in if M.Status.unserved ~run_start ~now pins [] = [] && not (List.exists behind (Queue_read.edges p rules [])) then Ok () else Loop.catching (fun () -> Ok (Journal.append_unrecorded p.journal ~segments:Queue_read.tail_segments (fun events -> List.map (fun (pin : M.Status.pin) -> Perform.event p ~branch:pin.entry.name (M.Event.Overdue { target = pin.entry.target; since = pin.queued })) (M.Status.unreported ~run_start ~now pins events) @ List.filter_map (fun (e : M.Schedule.t) -> Option.map (fun since -> Perform.event p ~branch:e.source (M.Event.Overdue { target = e.target; since })) e.served) (M.Schedule.unreported ~now (Queue_read.edges p rules events) events)))) ~on_failure:(fun m -> Error m)
let with_queue ?branch ?recorded ~writes ctx f = Eio.Switch.run @@ fun sw -> match open_queue ?recorded ctx ~sw ~writes with | Error (Refuse { why; next }) -> Error (refused ?branch ?next why) | Error (Unavailable m) -> Error (unavailable ?branch m) | Ok q when writes -> ( match journal_unserved q with | Ok () -> f q | Error m -> Error (unavailable ?branch m)) | Ok q -> f q
let initialised ?branch q k = let rules_branch = q.perform.local.rules in match (q.rules, q.unloadable) with | Some rules, _ -> k rules | None, Some (_, why) -> Error (M.Outcome.diagnosed ?branch why) | None, None when M.Branch.equal rules_branch M.Branch.rules -> ( match Perform.past q.perform with | Never -> Error (refused ?branch ~next:"mq init" "mq init has not run in this repository") | Ran last -> Error (M.Init.gone ?branch ~rules_branch last)) | None, None -> Error (Fmt.kstr (refused ?branch) "mq.rules names %a, which is not a branch here" M.Branch.pp rules_branch)
let repairing ?branch q ~target k = match (q.unloadable, target) with | Some (c, _), Some t when M.Branch.equal t q.perform.local.rules -> k { q with perform = { q.perform with repair = Some c } } (M.Rules.repairing, c) | _ -> initialised ?branch q (k q)
let recorded_base q name = let p = q.perform in M.Description.recorded_base name ~upstream:(Perform.upstream p name) (Perform.description p name)
let admitted ctx = with_queue ~writes:false ctx (fun q -> initialised q (fun _ -> Ok ()))