Something went wrong. Try again.
A local Git merge queue with content-bound gate verdicts
Something went wrong. Try again.
24 kB · 707 lines
OCaml
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708type request = { remote : string option; no_wait : bool }type context = { rules : Rules.t; rules_branch : Branch.t; scheduler : bool }
type landing = { submission : Submission.t; name : Branch.t; landed : Commit.t; since : Commit.t; pinned : bool; taken : bool;}
type facts = { branch : Branch.t; remote : string; local : Commit.t option; agreed : Commit.t option; seen : Commit.t option; tracking : Commit.t option; contains : (Commit.t * Commit.t) list; landings : landing list; foreign : int; tags : string list; checkouts : Fpath.t list; pulled : int; serial : int;}
type replay = { name : Branch.t; target : Branch.t; was : Commit.t }
type write = | Rescue of { tracking : Branch.t; serial : int; commit : Commit.t } | Requeue of { entry : Entry.t; local : Commit.t } | Reset of { branch : Branch.t; expected : Commit.t option; value : Commit.t; checkouts : Fpath.t list; } | Seen of { remote : string; branch : Branch.t; seen : Commit.t option; tracking : Commit.t option; value : Commit.t; } | Track of { branch : Branch.t; remote : string }
type event = | Local_read of Push.local list | Fetched of { remote : string; values : (Branch.t * Commit.t) list } | Unreachable of { remote : string; why : string } | Unsupported of { remote : string; why : string } | Untrusted of { remote : string; why : string } | Facts_read of facts list | Written | Requeued of Record.id | Lost of string | Overwritten of { checkout : Fpath.t; paths : string list } | Unmerged of { checkout : Fpath.t; paths : string list } | Unpinned | Signalled of Signal.t | Unanswered of action * Machine.unanswered
and action = | Read_local | Fetch of { remote : string; branches : Branch.t list } | Read_facts of { fetched : (string * (Branch.t * Commit.t) list) list; branches : (string * Branch.t) list; } | Write of write | Unpin of { name : Branch.t; expected : Record.id }
type ending = | Reconciled of { report : Report.t; replays : replay list; remotes : string; pulled : int; } | Stopped of Outcome.t
type plan = { writes : write list; replays : replay list; named : string list; pulled : int; remotes : string list;}
type phase = | Reading_local | Fetching of { all : (string * Branch.t list) list; current : string * Branch.t list; rest : (string * Branch.t list) list; fetched : (string * (Branch.t * Commit.t) list) list; } | Waiting of { fetched : (string * (Branch.t * Commit.t) list) list; branches : (string * Branch.t) list; } | Reading_facts | Writing of { plan : plan; rest : write list; pins : (Branch.t * Record.id) list; } | Unpinning of { rest : (Branch.t * Record.id) list; outcome : Outcome.t } | Done of ending
type t = { context : context; request : request; phase : phase; interrupted : bool; (** A signal arrived while the pull wrote: it finishes its writes and lands nothing. *)}
let machine = "pull"
let v context request = ( { context; request; phase = Reading_local; interrupted = false }, [ Read_local ] )
let plural n one = if n = 1 then Fmt.str "1 %s" one else Fmt.str "%d %ss" n onelet outcome ?branch ?next word = Fmt.kstr (Outcome.v ?branch ?next word)let finish t e = Ok ({ t with phase = Done e }, [])let stop t o = finish t (Stopped o)
(* {1 What is pulled from where} *)
let literals ctx = Rules.literals ctx.rules @ [ ctx.rules_branch ]let in_dag ctx b = Branch.equal b ctx.rules_branch || Rules.names ctx.rules b
(* Every branch of the DAG, the rules branch, and every local branch a pattern names, each with the remote it is pulled from: REMOTE, its own, else the one the DAG's other branches are pulled from. *)let sources ctx (request : request) (locals : Push.local list) = let local b = List.find_opt (fun (l : Push.local) -> Branch.equal l.branch b) locals in let names = List.sort_uniq Branch.compare (literals ctx @ List.filter_map (fun (l : Push.local) -> if in_dag ctx l.branch then Some l.branch else None) locals) in let fallback = List.find_map (fun b -> Option.bind (local b) (fun (l : Push.local) -> l.pull_remote)) names in List.filter_map (fun b -> let remote = match request.remote with | Some r -> Some r | None -> ( match Option.bind (local b) (fun (l : Push.local) -> l.pull_remote) with | Some r -> Some r | None -> fallback) in Option.map (fun r -> (r, b)) remote) names
let by_remote pairs = List.sort_uniq String.compare (List.map fst pairs) |> List.map (fun r -> ( r, List.filter_map (fun (r', b) -> if String.equal r r' then Some b else None) pairs ))
(* {1 One branch} *)
let tracking_of (f : facts) = Fmt.kstr Branch.v "%s/%a" f.remote Branch.pp f.branch
let contains (f : facts) a b = Commit.equal a b || List.exists (fun (x, y) -> Commit.equal x a && Commit.equal y b) f.contains
(* The value this machine last saw on the remote: its record under [refs/mq/seen/], else the remote-tracking branch, which is what Git fetched last on a machine where mq has not pulled or pushed yet. *)let last_seen (f : facts) = match f.seen with Some _ as s -> s | None -> f.tracking
let returned (l : landing) = if l.taken then Fmt.kstr Branch.v "%a-returned" Branch.pp l.name else l.name
let seen_write (f : facts) value = match f.seen with | Some s when Commit.equal s value && Option.equal Commit.equal f.tracking (Some value) -> [] | _ -> [ Seen { remote = f.remote; branch = f.branch; seen = f.seen; tracking = f.tracking; value; }; ]
let rescue (f : facts) ~agreed = match last_seen f with | Some s when not (contains f s agreed) -> ( [ Rescue { tracking = tracking_of f; serial = f.serial; commit = s } ], [ Fmt.str "%s dropped commits this machine had seen there: kept at %a" (Branch.to_string (tracking_of f)) Ref_name.pp (Ref_name.rescue (tracking_of f) f.serial); ] ) | _ -> ([], [])
let err_refused (f : facts) ?next fmt = Fmt.kstr (fun s -> Error (outcome ~branch:f.branch ?next Outcome.Refused "%s" s)) fmt
(* Each landing not pinned yet is pinned again under the name it comes back as, verified against the local branch. *)let requeues (f : facts) ~local = List.filter_map (fun (l : landing) -> if l.pinned then None else let entry = Entry.v ~submission:l.submission ~name:(returned l) ~tip:l.landed ~target:f.branch ~since:l.since ~keep_branch:false ~retry:false () in Some (Requeue { entry; local })) f.landings
let replays (f : facts) = List.map (fun (l : landing) -> { name = (if l.pinned then l.name else returned l); target = f.branch; was = l.landed; }) f.landings
let diverged (f : facts) ~agreed ~local = if f.foreign > 0 then err_refused f "%s on %a since %s/%a as this machine last saw it %s not this machine's \ landings: nothing moved" (plural f.foreign "commit") Branch.pp f.branch f.remote Branch.pp f.branch (if f.foreign = 1 then "is" else "are") else match f.tags with | tag :: _ -> err_refused f ~next:("git tag -d " ^ tag) "%s tags a commit a pull would replay on %a, and a tag is never \ replayed: nothing moved" tag Branch.pp f.branch | [] -> let rescues, named = rescue f ~agreed in let reset = Reset { branch = f.branch; expected = Some local; value = agreed; checkouts = f.checkouts; } in Ok ( rescues @ requeues f ~local @ (reset :: seen_write f agreed), replays f, named )
let reconcile (f : facts) = match (f.agreed, f.local, last_seen f) with | None, _, Some _ -> Ok ( [], [], Fmt.kstr Option.some "%a was deleted on %s: mq deletes no branch of the DAG, and this \ one stays here" Branch.pp f.branch f.remote ) | None, _, None -> Ok ([], [], None) | Some agreed, None, _ -> Ok ( Reset { branch = f.branch; expected = None; value = agreed; checkouts = []; } :: Track { branch = f.branch; remote = f.remote } :: seen_write f agreed, [], None ) | Some agreed, Some local, _ when Commit.equal agreed local -> Ok (seen_write f agreed, [], None) | Some agreed, Some local, _ when contains f local agreed -> let rescues, named = rescue f ~agreed in Ok ( rescues @ [ Reset { branch = f.branch; expected = Some local; value = agreed; checkouts = f.checkouts; }; ] @ seen_write f agreed, [], match named with n :: _ -> Some n | [] -> None ) | Some agreed, Some local, seen when contains f agreed local && Option.fold ~none:true ~some:(fun s -> contains f s agreed) seen -> Ok (seen_write f agreed, [], None) | Some _, Some _, None -> err_refused f "this machine has no record of the value it last saw on %s/%a (%a), so \ it cannot tell its own landings from the remote's: nothing moved" f.remote Branch.pp f.branch Ref_name.pp (Ref_name.seen (tracking_of f)) | Some agreed, Some local, Some _ -> ( match diverged f ~agreed ~local with | Error _ as e -> e | Ok (writes, replays, named) -> Ok (writes, replays, match named with n :: _ -> Some n | [] -> None))
(* The rules branch moves only by fast-forward, as git merge --ff-only moves a branch: a remote copy rewound behind the value this machine last saw there, or one that diverged from this machine's, is refused by name, and no landing into the rules is replayed (doc/configuration.md). *)let reconcile_rules (f : facts) = match (f.agreed, f.local) with | Some agreed, Some local when (not (contains f local agreed)) && not (contains f agreed local) -> err_refused f "%a and %s/%a have diverged: the rules move only by fast-forward; \ nothing moved" Branch.pp f.branch f.remote Branch.pp f.branch | Some agreed, Some local when (not (Commit.equal agreed local)) && contains f agreed local && Option.fold ~none:false ~some:(fun s -> not (contains f s agreed)) (last_seen f) -> err_refused f ~next:("mq push " ^ f.remote) "%s/%a was rewound behind the value this machine last saw there: %a \ keeps its rules; nothing moved" f.remote Branch.pp f.branch Branch.pp f.branch | _ -> reconcile f
(* The branches that take from a branch the pull takes back are not followed yet: their unpublished part would stand on commits the new source no longer holds. *)let unfollowed ctx facts plans = List.find_map (fun (f : facts) -> match Rules.takes ctx.rules f.branch with | None -> None | Some s -> ( let replayed = List.exists (fun ((g : facts), (_, replays, _)) -> Branch.equal g.branch s && replays <> []) plans in match (f.local, f.agreed) with | Some l, Some a when replayed && not (Commit.equal l a) -> Some f.branch | _ -> None)) facts
let plan ctx facts = let order (f : facts) = let rec depth b = match Rules.takes ctx.rules b with | Some s when not (Branch.equal s b) -> 1 + depth s | _ -> 0 in depth f.branch in let facts = List.stable_sort (fun a b -> compare (order a) (order b)) facts in let results = List.map (fun (f : facts) -> ( f, if Branch.equal f.branch ctx.rules_branch then reconcile_rules f else reconcile f )) facts in match List.find_map (function _, Error o -> Some o | _ -> None) results with | Some o -> Error (`Refused o) | None -> ( let plans = List.filter_map (function f, Ok p -> Some (f, p) | _, Error _ -> None) results in match unfollowed ctx facts plans with | Some b -> Error (`Unfollowed b) | None -> Ok { writes = List.concat_map (fun (_, (w, _, _)) -> w) plans; replays = List.concat_map (fun (_, (_, r, _)) -> r) plans; named = List.filter_map (fun (_, (_, _, n)) -> n) plans; pulled = List.fold_left (fun n ((f : facts), _) -> n + f.pulled) 0 plans; remotes = List.sort_uniq String.compare (List.map (fun ((f : facts), _) -> f.remote) plans); })
(* {1 The report} *)
let remotes (p : plan) = match p.remotes with [] -> "no remote" | rs -> String.concat ", " rs
let report (p : plan) = let remotes = remotes p in let subject = if p.pulled = 0 then Fmt.str "nothing new from %s" remotes else Fmt.str "%s from %s" (plural p.pulled "commit") remotes in let detail = match p.named with [] -> None | ns -> Some (String.concat "; " ns) in Report.v ?detail Report.Pulled subject
let summary ?named ~remote ~pulled results = let again, work = List.partition (fun (_, (o : Outcome.t), _, _) -> o.word = Outcome.Landed) results in let word = if work = [] then Outcome.Landed else Outcome.Needs_work in let landings = List.map (fun ((r : replay), (o : Outcome.t), became, _) -> { Outcome.branch = r.name; was = r.was; became = (if o.word = Outcome.Landed then became else None); }) results in let next = List.find_map (fun (_, (o : Outcome.t), _, restored) -> if o.word = Outcome.Landed then None else Option.map (fun b -> Fmt.str "mq land %a" Branch.pp b) restored) results in let needs = match work with | [] -> "" | [ _ ] -> " and 1 needs work" | w -> Fmt.str " and %d need work" (List.length w) in Fmt.kstr (Outcome.v ?next ~landed:(work = []) ~published:false ~landings word) "pulled %s from %s; of yours, %d landed again%s%s" (plural pulled "commit") remote (List.length again) needs (Option.fold ~none:"" ~some:(fun n -> "; " ^ n) named)
(* {1 Transitions} *)
let pp_phase ppf = function | Reading_local -> Fmt.string ppf "reading the local branches" | Fetching { current = r, _; _ } -> Fmt.pf ppf "fetching %s" r | Waiting _ -> Fmt.string ppf "waiting for the run" | Reading_facts -> Fmt.string ppf "reading the branches under the lock" | Writing { rest; _ } -> Fmt.pf ppf "writing, %d writes left" (List.length rest) | Unpinning _ -> Fmt.string ppf "unpinning" | Done _ -> Fmt.string ppf "done"
let pp ppf t = Fmt.pf ppf "%s: %a" machine pp_phase t.phase
let pp_write ppf = function | Rescue { tracking; serial; _ } -> Ref_name.pp ppf (Ref_name.rescue tracking serial) | Requeue { entry; _ } -> Fmt.pf ppf "requeue %a" Branch.pp entry.name | Reset { branch; _ } -> Fmt.pf ppf "reset %a" Branch.pp branch | Seen { remote; branch; _ } -> Fmt.pf ppf "seen %s/%a" remote Branch.pp branch | Track { branch; remote } -> Fmt.pf ppf "track %s/%a" remote Branch.pp branch
let pp_action ppf = function | Read_local -> Fmt.string ppf "read the local branches" | Fetch { remote; _ } -> Fmt.pf ppf "fetch %s" remote | Read_facts _ -> Fmt.string ppf "read the branches" | Write w -> pp_write ppf w | Unpin { name; _ } -> Fmt.pf ppf "unpin %a" Branch.pp name
let pp_event ppf = function | Local_read ls -> Fmt.pf ppf "%d local branches" (List.length ls) | Fetched { remote; _ } -> Fmt.pf ppf "fetched %s" remote | Unreachable { remote; why } -> Fmt.pf ppf "%s unreachable: %s" remote why | Unsupported { remote; why } -> Fmt.pf ppf "%s unsupported: %s" remote why | Untrusted { remote; why } -> Fmt.pf ppf "%s untrusted: %s" remote why | Facts_read fs -> Fmt.pf ppf "%d branches read" (List.length fs) | Written -> Fmt.string ppf "written" | Requeued _ -> Fmt.string ppf "requeued" | Lost what -> Fmt.pf ppf "lost: %s" what | Overwritten { checkout; _ } -> Fmt.pf ppf "%a overwritten" Fpath.pp checkout | Unmerged { checkout; _ } -> Fmt.pf ppf "%a in conflict" Fpath.pp checkout | Unpinned -> Fmt.string ppf "unpinned" | Signalled s -> Fmt.pf ppf "signalled %a" Signal.pp s | Unanswered (a, why) -> Fmt.pf ppf "%a: %a" pp_action a Machine.pp_unanswered why
let fetch t ~all current rest fetched = let remote, branches = current in Ok ( { t with phase = Fetching { all; current; rest; fetched } }, [ Fetch { remote; branches } ] )
let on_local t locals = if t.request.no_wait && not t.context.scheduler then stop t (outcome ~next:"mq pull" Outcome.Refused "--no-wait needs a scheduler, and none is registered on this machine") else match by_remote (sources t.context t.request locals) with | [] -> stop t (outcome Outcome.Refused "Git names no remote for any branch of the DAG, and no REMOTE was \ given: a machine with no remote has nothing to pull") | first :: rest as all -> fetch t ~all first rest []
let interrupted replays = let n = List.length replays in let next = match replays with | r :: _ -> Fmt.kstr Option.some "mq land %a" Branch.pp r.name | [] -> None in outcome ?next Outcome.Interrupted "the pull was interrupted: %s %s queued on the new tip" (plural n "landing") (if n = 1 then "stays" else "stay")
let writing t (plan : plan) rest pins = match rest with | [] when t.interrupted -> stop t (interrupted plan.replays) | [] -> finish t (Reconciled { report = report plan; replays = plan.replays; remotes = remotes plan; pulled = plan.pulled; }) | w :: rest -> Ok ({ t with phase = Writing { plan; rest; pins } }, [ Write w ])
let unpin t rest outcome = match rest with | [] -> stop t outcome | (name, expected) :: rest -> Ok ( { t with phase = Unpinning { rest; outcome } }, [ Unpin { name; expected } ] )
let on_facts t facts = match plan t.context facts with | Error (`Refused o) -> stop t o | Error (`Unfollowed b) -> Machine.not_implemented machine (Fmt.str "mq pull: taking %a back to the newest of its commits the new %s \ holds" Branch.pp b (Option.fold ~none:"source" ~some:Branch.to_string (Rules.takes t.context.rules b))) | Ok p -> writing t p p.writes []
let current_write = function | Writing { plan; rest; _ } -> let i = List.length plan.writes - List.length rest - 1 in List.nth_opt plan.writes i | _ -> None
let incoming t ~now:_ e = match (t.phase, e) with | Done _, _ -> Ok (t, []) | Writing _, Signalled _ -> Ok ({ t with interrupted = true }, []) | Unpinning _, Signalled _ -> Ok (t, []) | (Reading_local | Fetching _ | Waiting _ | Reading_facts), Signalled _ -> stop t (outcome Outcome.Interrupted "the pull was interrupted: nothing moved") | Reading_local, Local_read locals -> on_local t locals | Fetching { all; current = remote, _; rest; fetched }, Fetched { values; _ } -> ( let fetched = fetched @ [ (remote, values) ] in match rest with | next :: rest -> fetch t ~all next rest fetched | [] -> let branches = List.concat_map (fun (r, bs) -> List.map (fun b -> (r, b)) bs) all in Ok ({ t with phase = Waiting { fetched; branches } }, [])) | Fetching _, Unreachable { remote; why } -> stop t (outcome Outcome.Unavailable "%s cannot be reached (%s): nothing moved" remote why) | Fetching _, (Unsupported { remote; why } | Untrusted { remote; why }) -> stop t (outcome Outcome.Refused "%s: %s: nothing moved" remote why) | Fetching { current = remote, _; _ }, Unanswered (Fetch _, why) -> stop t (outcome Outcome.Unavailable "the fetch from %s did not finish: %a" remote Machine.pp_unanswered why) | Reading_facts, Facts_read facts -> on_facts t facts | Writing { plan; rest; pins }, Written -> writing t plan rest pins | Writing { plan; rest; pins }, Requeued id -> ( match current_write t.phase with | Some (Requeue { entry; _ }) -> writing t plan rest ((entry.name, id) :: pins) | _ -> writing t plan rest pins) | Writing { pins; _ }, Overwritten { checkout; paths } -> unpin t pins (outcome Outcome.Removed_from_queue "%a holds changes taking the branch back would overwrite: %s; \ nothing moved" Fpath.pp checkout (String.concat ", " paths)) | Writing { pins; _ }, Unmerged { checkout; paths } -> unpin t pins (outcome Outcome.Removed_from_queue "%a has an index in conflict at %s, and the branch cannot move \ under it; nothing moved" Fpath.pp checkout (String.concat ", " paths)) | Writing _, Lost what -> stop t (outcome Outcome.Removed_from_queue "%s moved while the pull wrote: what was written stays, and mq pull \ again reads it" what) | Writing _, Unanswered (Write w, why) -> stop t (outcome Outcome.Unavailable "%a could not be written: %a" pp_write w Machine.pp_unanswered why) | Unpinning { rest; outcome }, (Unpinned | Unanswered (Unpin _, _)) -> unpin t rest outcome | (Reading_local | Reading_facts), Unanswered (_, why) -> stop t (outcome Outcome.Unavailable "the branches could not be read: %a" Machine.pp_unanswered why) | ( ( Reading_local | Fetching _ | Waiting _ | Reading_facts | Writing _ | Unpinning _ ), ( Local_read _ | Fetched _ | Unreachable _ | Unsupported _ | Untrusted _ | Facts_read _ | Written | Requeued _ | Lost _ | Overwritten _ | Unmerged _ | Unpinned | Unanswered _ ) ) -> Machine.unexpected machine ~state:(Fmt.str "%a" pp t) ~event:(Fmt.str "%a" pp_event e)
let timer t ~now:_ = Ok (t, [])let next_timeout _ = Nonelet ending t = match t.phase with Done e -> Some e | _ -> Nonelet task (_ : action) = Machine.Task.Nowlet unanswered a why = Unanswered (a, why)let signal s = Signalled slet needs_lock t = match t.phase with Waiting _ -> true | _ -> falselet queued (_ : t) = Nonelet exit_code = function Reconciled _ -> 0 | Stopped o -> Outcome.exit o
let reason = function | Reconciled _ -> None | Stopped o -> Some (Outcome.first_line o)
let run_started t ~now:_ r = match (t.phase, r) with | Waiting { fetched; branches }, Ok (_ : Work.pass) -> Ok ({ t with phase = Reading_facts }, [ Read_facts { fetched; branches } ]) | _, Error (Work.Lock_held { holder; waited }) -> let by = match holder with | Some pid -> Fmt.str " by process %d" pid | None -> "" in stop t (outcome Outcome.Removed_from_queue "the run lock is held%s past bounds.wait_wall (%a): nothing moved" by Duration.pp waited) | _, Error (Work.Unavailable why) -> stop t (outcome Outcome.Unavailable "%s" why) | _, Ok _ -> Ok (t, [])