diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/adr548_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/adr548_test.go new file mode 100644 index 00000000..d17f9505 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/adr548_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "strings" + "testing" +) + +// One check per delivery (novox/hq ADR 0548), the gate's part: nothing merges into a repository while a check of other heads +// judges it (§2); an approved merge holds a turn on its repositories (§3); a held merge past its bound is a stalled line, read +// and handed by two tools (§4); an approved delivery whose trunk moved merges on its carried verdict with no second ask (§8). + +const otherHead = "c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3" + +// approvedLone is a lone delivery into oneRepo whose check passed, approved by the operator (the world approves at once). +func approvedLone(w *world) string { + w.t.Helper() + p := pr(oneRepo, 1, oneHead, "feat/one") + requiresCheck(w, p) + w.h.PullUpdated(p) + putCheckerVerdict(w, "pass", headOf(oneRepo, 1, oneHead)) + w.h.CommitStatus(checkStatus(oneRepo, oneHead, "success", oneTrunk)) + return IDOf(oneRepo, oneHead) +} + +func TestOneCheckPerDeliveryAtTheGate(t *testing.T) { + for name, run := range map[string]func(t *testing.T){ + // §2: a merge into a repository a running verdict names is not written to merges, and is written on that verdict's pass. + "an approved merge waits for a running check on its repository and merges on its end": func(t *testing.T) { + w := newWorld(t) + putCheckerVerdict(w, "running", headOf(oneRepo, 9, otherHead)) + id := approvedLone(w) + w.settleAll() + if w.op.count() != 1 || len(w.forge.merges) != 0 { + t.Fatalf("asked %d times, merges %v: want the Approve asked and the merge held", w.op.count(), w.forge.merges) + } + d := w.delivery(id) + if d.Ready == nil || !strings.Contains(d.Ready.Waits, "waits for the check of g") || !strings.Contains(d.Ready.Waits, oneRepo) || + d.Ready.HeldSince == nil { + t.Fatalf("the merge does not say what it waits for: %+v", d.Ready) + } + putCheckerVerdict(w, "pass", headOf(oneRepo, 9, otherHead)) + w.settleAll() + if len(w.forge.merges) != 1 || w.op.count() != 1 { + t.Fatalf("after the check's end: merges %v, asked %d times; want merged with no second ask", w.forge.merges, w.op.count()) + } + }, + "a running check on another repository holds nothing": func(t *testing.T) { + w := newWorld(t) + putCheckerVerdict(w, "running", headOf(oneRepo2, 9, otherHead)) + approvedLone(w) + w.settleAll() + if len(w.forge.merges) != 1 { + t.Fatalf("merges %v: a check of another repository held the merge (%s)", w.forge.merges, w.logs()) + } + }, + "the delivery's own running verdict does not hold it": func(t *testing.T) { + w := newWorld(t) + w.op.auto = false + id := approvedLone(w) + w.settleAll() + putCheckerVerdict(w, "running", headOf(oneRepo, 1, oneHead)) // its own heads judged again (a carry) + for _, k := range w.op.open() { + w.op.answer(k, OptionApprove) + } + w.settleAll() + if d := w.delivery(id); len(w.forge.merges) != 1 || d.Ready == nil || d.Ready.HeldSince != nil { + t.Fatalf("merges %v, run %+v: its own check held it", w.forge.merges, d.Ready) + } + }, + } { + t.Run(name, run) + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go index cfa5ffe6..71f10581 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -12,6 +12,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" ) // The holder of the mesh-delivery seat (novox/hq ADR 0239, to-be 47): the one writer of deliveries and @@ -75,8 +76,14 @@ type Holder struct { // unchecked are the open pull requests' heads seen without a merge check (unchecked.go). unchecked map[string]unchecked // checker are the checker's verdicts by the hash of the heads they judged, as its verdicts bucket holds them - // (onestatus.go, issue 1302). - checker map[string]checkerVerdict + // (onestatus.go, issue 1302); runningSince when this owner first saw each running (ADR 0548 §2, turns.go). + checker map[string]checkerVerdict + runningSince map[string]time.Time + // Turns is the gate's bucket turns (turns.go, ADR 0548 §3); nil keeps none. turnsHeld are the turns as written there, + // by repository, read once at the first look (turnsLoaded). + Turns TurnStore + turnsHeld map[string]sdk.Turn + turnsLoaded bool // tried counts the owed effects Flush tried since this owner started, by how each ended (novox/hq issue 407). tried owedTries // forgeWanted is whether the forge's open pull requests are to be read at the next React: a check may have @@ -126,6 +133,7 @@ func (h *Holder) init() { h.lostRead = map[string]time.Time{} h.unchecked = map[string]unchecked{} h.checker = map[string]checkerVerdict{} + h.runningSince, h.turnsHeld = map[string]time.Time{}, map[string]sdk.Turn{} h.started = h.Now() } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/onestatus.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/onestatus.go index fbc51d15..10151f1f 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/onestatus.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/onestatus.go @@ -72,8 +72,12 @@ func (h *Holder) CheckerVerdictChanged(key string, value []byte) { if i := strings.LastIndex(key, "."); i >= 0 { hash = key[i+1:] } + if h.runningSince == nil { + h.runningSince = map[string]time.Time{} + } if len(value) == 0 || string(value) == "null" { delete(h.checker, hash) + delete(h.runningSince, hash) h.Poke() return } @@ -85,6 +89,11 @@ func (h *Holder) CheckerVerdictChanged(key string, value []byte) { if v.Heads == "" { v.Heads = hash } + if prev, ok := h.checker[v.Heads]; v.State == checkerRunning && (!ok || prev.State != checkerRunning) { + h.runningSince[v.Heads] = h.Now() // a check starts: a merge it holds says since when (ADR 0548 §2) + } else if v.State != checkerRunning { + delete(h.runningSince, v.Heads) + } h.checker[v.Heads] = v h.Poke() } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go index 79351e47..d37f31aa 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go @@ -121,6 +121,13 @@ type ReadyRun struct { Declined *time.Time `json:"declined,omitempty"` Warrant string `json:"warrant,omitempty"` Refused string `json:"refused,omitempty"` + + // One check per delivery (novox/hq ADR 0548 §2, §4, turns.go): HeldSince is when the approved merge began to wait for the + // check of other heads HeldFor names (its verdict key), running on a repository it goes into; HandNow the operator's word + // that handed it at once past that check. + HeldSince *time.Time `json:"held_since,omitempty"` + HeldFor string `json:"held_for,omitempty"` + HandNow string `json:"hand_now,omitempty"` } // MergeBranch is the one branch this owner merges into (novox/hq ADR 0398 §4): a merge into any other is its author's. @@ -244,6 +251,9 @@ func (h *Holder) readyAsks(now time.Time) []func() { if waits == "" { waits, approvalAsk = h.approval(d, now) } + if waits == "" && gates { + waits = h.turnWaits(d, now) // nothing merges into a repository while a check of other heads judges it (ADR 0548 §2) + } if waits != r.Waits { r.Waits = waits h.keep(d) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/turns.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/turns.go new file mode 100644 index 00000000..55df4c20 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/turns.go @@ -0,0 +1,444 @@ +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 (`.`, 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{} + for _, d := range h.deliveries { + 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() { + if am.run == c { + 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 := h.wantedTurns() + 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 +}