455 lines
16 KiB
Go
455 lines
16 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// **One check per delivery** (novox/hq ADR 0548 §2 to §4, issue 568). While a check runs, nothing merges into a repository it
|
|
// judges, and an approved merge holds a turn on its repositories so no new check starts there:
|
|
//
|
|
// - the hold (§2): an approved merge is handed to the forge bridge only when no verdict of the checker's verdicts bucket in
|
|
// the state running names, among its trunk heads, a repository the merge goes into; the verdict of the merge's own heads
|
|
// excepted. The gate watches that bucket already (onestatus.go), so the verdict leaving running gives the waiting merge its
|
|
// turn at once, with no timer. Siblings a check reads beside are not held (§2). Only while mesh/check gates the merge;
|
|
// - the turn (§3): before it reads the verdicts for a merge, the gate writes one key per repository the merge goes into in its
|
|
// bucket turns (`<owner>.<name>`, sdk.TurnsKey): the deliveries that wait there in the order of their Approve, the check
|
|
// each waits for and since when; for a group on every member's repository until its last member merged. The checker reads
|
|
// the bucket and claims no check of a repository a turn names. The key goes when the last delivery in it merged or left;
|
|
// - the bound and the tools (§4): a merge held past turnBound is a stalled line; delivery_turns reads every turn and
|
|
// delivery_hand_now hands a waiting merge at once on the operator's word, the check it waited for then carried.
|
|
|
|
// turnBound is how long an approved merge may wait for another delivery's check before it is a stalled line (ADR 0548 §4).
|
|
const turnBound = 30 * time.Minute
|
|
|
|
// TurnStore is the gate's bucket turns: one key per repository, written by the gate alone.
|
|
type TurnStore interface {
|
|
// All reads every turn, by repository (lowercase).
|
|
All() (map[string]sdk.Turn, error)
|
|
Put(t sdk.Turn) error
|
|
Delete(repository string) error
|
|
}
|
|
|
|
// kvTurns is the bucket turns over the runtime.
|
|
type kvTurns struct{}
|
|
|
|
func (kvTurns) All() (map[string]sdk.Turn, error) {
|
|
keys, err := stdio.State("turns").Keys()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := map[string]sdk.Turn{}
|
|
for _, k := range keys {
|
|
e, err := stdio.State("turns").Get(k)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if e == nil {
|
|
continue
|
|
}
|
|
repository, err := sdk.ParseTurnsKey(k)
|
|
if err != nil {
|
|
continue // not a turn's key: never the gate's, left
|
|
}
|
|
t, err := sdk.DecodeTurn(e.Value)
|
|
if err != nil {
|
|
t = sdk.Turn{Repository: repository} // unreadable: rewritten or deleted at the next look
|
|
}
|
|
out[repository] = t
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (kvTurns) Put(t sdk.Turn) error {
|
|
key, err := sdk.TurnsKey(t.Repository)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
raw, err := sdk.EncodeTurn(t)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = stdio.State("turns").Put(key, json.RawMessage(raw))
|
|
return err
|
|
}
|
|
|
|
func (kvTurns) Delete(repository string) error {
|
|
key, err := sdk.TurnsKey(repository)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return stdio.State("turns").Delete(key)
|
|
}
|
|
|
|
// approvedMerge is an approved merge not yet done: the run holding its Approve, the deliveries it covers, the hash of their
|
|
// heads as the checker keys their verdict (lowercase, onestatus.go headsOf), and the repositories it goes into (lowercase).
|
|
type approvedMerge struct {
|
|
owner *Delivery
|
|
run *ReadyRun
|
|
group string
|
|
heads string
|
|
repos []string
|
|
}
|
|
|
|
// approvedMerges are the merges the operator approved that are not done, in the order of their Approve: each ask once, its
|
|
// run on the delivery that asked it. A declined, refused, merged, rejected or final one has left. Under the lock.
|
|
func (h *Holder) approvedMerges() []approvedMerge {
|
|
var out []approvedMerge
|
|
seen := map[string]bool{}
|
|
ids := make([]string, 0, len(h.deliveries))
|
|
for id := range h.deliveries {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids) // the same owner of an ask at every look, whichever member's run holds it
|
|
for _, id := range ids {
|
|
d := h.deliveries[id]
|
|
r := d.Ready
|
|
if r == nil || r.Approved == nil || r.Declined != nil || r.AskID == "" || len(r.Heads) == 0 || seen[r.AskID] ||
|
|
d.State.Final() || d.State == Rejected || d.Base != MergeBranch || r.RefusedSince != nil || h.allMerged(r.Heads) {
|
|
continue
|
|
}
|
|
seen[r.AskID] = true
|
|
var heads []sdk.Head
|
|
repos := map[string]bool{}
|
|
for id, sha := range r.Heads {
|
|
m := h.deliveries[id]
|
|
if m == nil {
|
|
continue
|
|
}
|
|
heads = append(heads, sdk.Head{Repository: strings.ToLower(m.Repository), Number: m.Number, Head: strings.ToLower(sha)})
|
|
repos[strings.ToLower(m.Repository)] = true
|
|
}
|
|
hash, err := sdk.HeadsHash(heads)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
am := approvedMerge{owner: d, run: r, group: askGroup(d), heads: hash}
|
|
for repo := range repos {
|
|
am.repos = append(am.repos, repo)
|
|
}
|
|
sort.Strings(am.repos)
|
|
out = append(out, am)
|
|
}
|
|
sort.SliceStable(out, func(i, j int) bool {
|
|
if !out[i].run.Approved.Equal(*out[j].run.Approved) {
|
|
return out[i].run.Approved.Before(*out[j].run.Approved)
|
|
}
|
|
return out[i].heads < out[j].heads
|
|
})
|
|
return out
|
|
}
|
|
|
|
// mergeOfDelivery is the approved merge covering d, if one is approved.
|
|
func (h *Holder) mergeOfDelivery(d *Delivery) (approvedMerge, bool) {
|
|
c := h.covering(d)
|
|
if c == nil || c.Approved == nil {
|
|
return approvedMerge{}, false
|
|
}
|
|
for _, am := range h.approvedMerges() {
|
|
// By the ask, not the run: a group's members may each hold a run of the same ask, and the merge is the ask's.
|
|
if am.run.AskID == c.AskID {
|
|
return am, true
|
|
}
|
|
}
|
|
return approvedMerge{}, false
|
|
}
|
|
|
|
// runningCheck is a check of the checker in the state running that names repo among its trunk heads, its heads not own: its
|
|
// verdict key and since when the gate has seen it run. False when none runs there. Under the lock.
|
|
func (h *Holder) runningCheck(repo, own string) (key string, v checkerVerdict, since time.Time, ok bool) {
|
|
hashes := make([]string, 0, len(h.checker))
|
|
for hash := range h.checker {
|
|
hashes = append(hashes, hash)
|
|
}
|
|
sort.Strings(hashes)
|
|
for _, hash := range hashes {
|
|
v := h.checker[hash]
|
|
if v.State != checkerRunning || v.Heads == own {
|
|
continue
|
|
}
|
|
for r := range v.Trunk {
|
|
if strings.EqualFold(r, repo) {
|
|
key, err := sdk.CheckerVerdictKey(v.Group, v.Heads)
|
|
if err != nil {
|
|
key = v.Group + "." + v.Heads
|
|
}
|
|
return key, v, h.runningSince[v.Heads], true
|
|
}
|
|
}
|
|
}
|
|
return "", checkerVerdict{}, time.Time{}, false
|
|
}
|
|
|
|
// turnWaits is what an approved merge waits for before it is handed (ADR 0548 §2, §3): its turns written, then the end of
|
|
// every check of other heads running on a repository it goes into. Empty when it may be handed. It keeps on the run since
|
|
// when it is held and for which check, for the stalled line and the tools. Under the lock.
|
|
func (h *Holder) turnWaits(d *Delivery, now time.Time) string {
|
|
am, ok := h.mergeOfDelivery(d)
|
|
if !ok {
|
|
return ""
|
|
}
|
|
r := am.run
|
|
if h.Turns != nil {
|
|
if !h.turnsLoaded {
|
|
return "approved; its merge turn waits for the turns the gate wrote before to be read"
|
|
}
|
|
for _, repo := range am.repos {
|
|
if !turnNames(h.turnsHeld[repo], am.heads) {
|
|
return "approved; its merge turn on " + repo + " is being written, so no check starts there before it merges"
|
|
}
|
|
}
|
|
}
|
|
if r.HandNow != "" {
|
|
if r.HeldSince != nil {
|
|
r.HeldSince, r.HeldFor = nil, ""
|
|
h.keep(am.owner)
|
|
}
|
|
return "" // handed now on the operator's word: the check it waited for is carried after the merge
|
|
}
|
|
for _, repo := range am.repos {
|
|
key, v, since, running := h.runningCheck(repo, am.heads)
|
|
if !running {
|
|
continue
|
|
}
|
|
if r.HeldSince == nil || r.HeldFor != key {
|
|
at := now
|
|
r.HeldSince, r.HeldFor = &at, key
|
|
h.keep(am.owner)
|
|
h.Logf("[mesh-delivery] %s is approved and held: the check %s runs on %s, and nothing merges there while it does (ADR 0548 §2)", am.owner.ID, key, repo)
|
|
}
|
|
words := fmt.Sprintf("approved; waits for the check of %s (heads %s) on %s", v.Group, v.Heads, repo)
|
|
if !since.IsZero() {
|
|
words += ", running since " + since.UTC().Format(time.RFC3339)
|
|
}
|
|
return words + " (novox/hq ADR 0548 §2)"
|
|
}
|
|
if r.HeldSince != nil {
|
|
r.HeldSince, r.HeldFor = nil, ""
|
|
h.keep(am.owner)
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// turnNames says whether a turn names the heads.
|
|
func turnNames(t sdk.Turn, heads string) bool {
|
|
for _, w := range t.Waiting {
|
|
if w.Heads == heads {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// wantedTurns are the turns the approved merges hold now, by repository: each merge on every repository it goes into, in
|
|
// the order of their Approve, with the check of other heads it waits for there. Under the lock.
|
|
func (h *Holder) wantedTurns() map[string]sdk.Turn {
|
|
out := map[string]sdk.Turn{}
|
|
for _, am := range h.approvedMerges() {
|
|
for _, repo := range am.repos {
|
|
t := out[repo]
|
|
t.Repository = repo
|
|
w := sdk.TurnWait{Group: am.group, Heads: am.heads, Ask: am.run.AskID, Approved: am.run.Approved.UTC(), Since: am.run.Approved.UTC()}
|
|
if key, _, since, ok := h.runningCheck(repo, am.heads); ok {
|
|
w.WaitsFor = key
|
|
if !since.IsZero() {
|
|
w.WaitsSince = since.UTC()
|
|
}
|
|
}
|
|
t.Waiting = append(t.Waiting, w)
|
|
out[repo] = t
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// turnWrites are the writes that bring the bucket turns to what the approved merges hold: a turn written where it changed,
|
|
// a key deleted where its last delivery merged or left. The first look reads what a gate before this one wrote. Under the
|
|
// lock; the writes run outside it, and each that lands is kept and asks for a look.
|
|
func (h *Holder) turnWrites() []func() {
|
|
if h.Turns == nil {
|
|
return nil
|
|
}
|
|
if !h.turnsLoaded {
|
|
return []func(){func() {
|
|
held, err := h.Turns.All()
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] the merge turns written before cannot be read, so no merge is handed yet: %v; asked again in %s", err, askAgainAfter)
|
|
h.again("the merge turns written before, read")
|
|
return
|
|
}
|
|
h.turnsHeld, h.turnsLoaded = held, true
|
|
h.Poke()
|
|
}}
|
|
}
|
|
want := map[string]sdk.Turn{}
|
|
if h.checkGates() {
|
|
want = h.wantedTurns() // while the check gates no merge, no merge waits for one, and no turn holds a check
|
|
}
|
|
var out []func()
|
|
repos := make([]string, 0, len(want)+len(h.turnsHeld))
|
|
for repo := range want {
|
|
repos = append(repos, repo)
|
|
}
|
|
for repo := range h.turnsHeld {
|
|
if _, ok := want[repo]; !ok {
|
|
repos = append(repos, repo)
|
|
}
|
|
}
|
|
sort.Strings(repos)
|
|
for _, repo := range repos {
|
|
t, wanted := want[repo]
|
|
held, had := h.turnsHeld[repo]
|
|
switch {
|
|
case wanted && had && sameTurnValue(held, t):
|
|
case wanted:
|
|
out = append(out, func() {
|
|
err := h.Turns.Put(t)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] the merge turn on %s cannot be written: %v; asked again in %s", repo, err, askAgainAfter)
|
|
h.again("a merge turn, written")
|
|
return
|
|
}
|
|
h.turnsHeld[repo] = t
|
|
h.Poke()
|
|
})
|
|
default:
|
|
out = append(out, func() {
|
|
err := h.Turns.Delete(repo)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] the merge turn on %s cannot be deleted: %v; asked again in %s", repo, err, askAgainAfter)
|
|
h.again("a merge turn, deleted")
|
|
return
|
|
}
|
|
delete(h.turnsHeld, repo)
|
|
h.Logf("[mesh-delivery] the merge turn on %s is deleted: no approved merge waits on it, so the checker may start a check there", repo)
|
|
h.Poke()
|
|
})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// sameTurnValue says whether two turns hold the same value, the version aside.
|
|
func sameTurnValue(a, b sdk.Turn) bool {
|
|
a.V, b.V = 0, 0
|
|
return sdk.SameTurn(a, b)
|
|
}
|
|
|
|
// turnLines are the stalled lines of approved merges held past turnBound by another delivery's check (ADR 0548 §4). Under the
|
|
// lock.
|
|
func (h *Holder) turnLines(now time.Time) []StalledLine {
|
|
var out []StalledLine
|
|
for _, d := range h.deliveries {
|
|
r := d.Ready
|
|
if r == nil || r.HeldSince == nil || r.Merged != nil || d.MergedAs != "" || d.State.Final() || now.Sub(*r.HeldSince) < turnBound {
|
|
continue
|
|
}
|
|
words := r.Waits
|
|
if words == "" {
|
|
words = "approved; waits for the check " + r.HeldFor
|
|
}
|
|
out = append(out, StalledLine{ID: d.ID, Number: d.Number, State: d.State, For: now.Sub(*r.HeldSince).Round(time.Second).String(),
|
|
Bound: turnBound.String(),
|
|
H2: "none: the check it waits for ends, or the operator hands the merge now with delivery_hand_now — the operator's",
|
|
Says: fmt.Sprintf("%s#%d at %s is %s, held since %s; the step that check is at is read with the checker's checker_status",
|
|
d.Repository, d.Number, shortOf(r.Head, 12), words, r.HeldSince.UTC().Format(time.RFC3339))})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// TurnLine is one repository's turn as delivery_turns reads it: the turn as the gate holds it and, for each delivery waiting,
|
|
// the check it waits for in words.
|
|
type TurnLine struct {
|
|
Repository string `json:"repository"`
|
|
Turn sdk.Turn `json:"turn"`
|
|
Waits map[string]any `json:"waits,omitempty"`
|
|
}
|
|
|
|
// TurnsRead is the tool delivery_turns: every turn the gate holds, with the check each waiting delivery waits for, since when,
|
|
// and whether the merge was handed now on the operator's word.
|
|
func (h *Holder) TurnsRead() map[string]any {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
want := h.wantedTurns()
|
|
repos := make([]string, 0, len(want))
|
|
for repo := range want {
|
|
repos = append(repos, repo)
|
|
}
|
|
sort.Strings(repos)
|
|
lines := []TurnLine{}
|
|
for _, repo := range repos {
|
|
l := TurnLine{Repository: repo, Turn: want[repo], Waits: map[string]any{}}
|
|
for _, w := range want[repo].Waiting {
|
|
says := "nothing: no check of other heads runs on " + repo
|
|
if w.WaitsFor != "" {
|
|
says = "the check " + w.WaitsFor
|
|
if !w.WaitsSince.IsZero() {
|
|
says += ", running since " + w.WaitsSince.Format(time.RFC3339)
|
|
}
|
|
}
|
|
l.Waits[w.Group] = says
|
|
}
|
|
lines = append(lines, l)
|
|
}
|
|
out := map[string]any{"turns": lines, "check-gates-merge": h.checkGates(), "bound": turnBound.String()}
|
|
if h.Turns != nil {
|
|
out["written"] = h.turnsLoaded
|
|
var unwritten []string
|
|
for _, repo := range repos {
|
|
if held, ok := h.turnsHeld[repo]; !ok || !sameTurnValue(held, want[repo]) {
|
|
unwritten = append(unwritten, repo)
|
|
}
|
|
}
|
|
for repo := range h.turnsHeld {
|
|
if _, ok := want[repo]; !ok {
|
|
unwritten = append(unwritten, repo+" (to delete)")
|
|
}
|
|
}
|
|
sort.Strings(unwritten)
|
|
if len(unwritten) > 0 {
|
|
out["not-yet-written"] = unwritten
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// HandNow is the tool delivery_hand_now (ADR 0548 §4): on the operator's word an approved merge held for another delivery's
|
|
// check is handed at once; that check runs on and is carried to the new trunk after the merge. Refused for a delivery that is
|
|
// not an approved merge waiting, naming why.
|
|
func (h *Holder) HandNow(id, why string) (string, error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
d := h.deliveries[id]
|
|
if d == nil {
|
|
return "", fmt.Errorf("no delivery %q", id)
|
|
}
|
|
if strings.TrimSpace(why) == "" {
|
|
return "", fmt.Errorf("say why: the hand is the operator's word, kept on the delivery")
|
|
}
|
|
am, ok := h.mergeOfDelivery(d)
|
|
if !ok {
|
|
return "", fmt.Errorf("%s is not an approved merge waiting: nothing to hand (its state is %s)", id, d.State)
|
|
}
|
|
if am.run.HeldSince == nil {
|
|
return "", fmt.Errorf("%s waits for no check of another delivery: %s", id, firstOf(am.run.Waits, "it is handed at its turn"))
|
|
}
|
|
am.run.HandNow = why
|
|
h.keep(am.owner)
|
|
h.mergeAgainNow(d)
|
|
h.Logf("[mesh-delivery] %s is handed now on the operator's word, past the check %s it waited for, which is carried after the merge: %s",
|
|
id, am.run.HeldFor, why)
|
|
return fmt.Sprintf("%s is handed to the merge at the next look; the check %s it waited for runs on and is carried to the new trunk", id, am.run.HeldFor), nil
|
|
}
|