Merge pull request 'Gate: merge turns, a merge waits for a running check of its repository, merges on a carried verdict with no second ask (ADR 0548, issue 568)' (#493) from feat/548-one-check into main

This commit was merged in pull request #493.
This commit is contained in:
2026-10-26 12:01:28 +00:00
16 changed files with 781 additions and 21 deletions
+3
View File
@@ -34,6 +34,9 @@ test("both statuses say the head and the trunk commit the verdict judged, first,
const [old] = statusesFor({ ...c, trunk: undefined, gate: { verdict: "pass", summary: "every machine composes" } });
assert.equal(old.description, "pass: every machine composes");
assert.equal(trunkNamed(old.description), undefined);
// ADR 0548 §8: a carried verdict's line names the head on the new trunk first, so the merge tool merges it on that trunk.
assert.equal(trunkNamed(`pass ${HEAD.slice(0, 12)} on ${MOVED.slice(0, 12)} | carried from ${TRUNK.slice(0, 12)}, mesh-catalog 0123456789ab | unit 0 run, 4 kept`),
MOVED.slice(0, 12));
});
test("a trunk's move puts an open head's merge check out of date, naming both commits, unless a check is on its way", () => {
@@ -0,0 +1,246 @@
package main
import (
"fmt"
"strings"
"sync"
"testing"
"time"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
// 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)
}
// memTurns is the bucket turns in memory; it logs each write with the merges the fake bridge had made by then.
type memTurns struct {
mu sync.Mutex
m map[string]sdk.Turn
log []string
merges func() int
}
func newMemTurns(w *world) *memTurns {
return &memTurns{m: map[string]sdk.Turn{}, merges: func() int { return len(w.forge.merges) }}
}
func (f *memTurns) All() (map[string]sdk.Turn, error) {
f.mu.Lock()
defer f.mu.Unlock()
out := map[string]sdk.Turn{}
for k, v := range f.m {
out[k] = v
}
return out, nil
}
func (f *memTurns) Put(t sdk.Turn) error {
if _, err := sdk.EncodeTurn(t); err != nil {
return err
}
f.mu.Lock()
defer f.mu.Unlock()
f.m[t.Repository] = t
f.log = append(f.log, fmt.Sprintf("put %s after %d merges", t.Repository, f.merges()))
return nil
}
func (f *memTurns) Delete(repository string) error {
f.mu.Lock()
defer f.mu.Unlock()
delete(f.m, repository)
f.log = append(f.log, fmt.Sprintf("delete %s after %d merges", repository, f.merges()))
return nil
}
func (f *memTurns) said() string { f.mu.Lock(); defer f.mu.Unlock(); return strings.Join(f.log, "; ") }
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)
}
// §9: deliveries says what it waits for.
for _, l := range w.h.Deliveries("", "", "", false) {
if l.ID == id && !strings.Contains(l.Waits, "approved; waits for the check of g") {
t.Errorf("deliveries says it waits for %q", l.Waits)
}
}
// §4: past 30 minutes the held merge is a stalled line naming the check it waits for and since when.
for _, l := range w.h.Stalled() {
if l.ID == id {
t.Fatalf("a stalled line within the bound: %+v", l)
}
}
w.later(turnBound + time.Minute)
w.settleAll()
var line *StalledLine
for _, l := range w.h.Stalled() {
if l.ID == id {
line = &l
}
}
if line == nil || line.Bound != "30m0s" || !strings.Contains(line.Says, "waits for the check of g") || !strings.Contains(line.Says, "held since") ||
!strings.Contains(line.Says, "checker_status") || !strings.Contains(line.H2, "delivery_hand_now") {
t.Fatalf("the stalled line of the held merge = %+v", line)
}
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)
}
},
// §3: the turn is written before the verdicts are read for the merge, and deleted after the last merge.
"a lone merge's turn is written before it is handed and deleted once it merged": func(t *testing.T) {
w := newWorld(t)
turns := newMemTurns(w)
w.h.Turns = turns
approvedLone(w)
w.settleAll()
if len(w.forge.merges) != 1 || turns.said() != "put "+oneRepo+" after 0 merges; delete "+oneRepo+" after 1 merges" {
t.Fatalf("merges %v; the turns were %q", w.forge.merges, turns.said())
}
},
"a group's turns stand on every member's repository until its last member merged": func(t *testing.T) {
w := newWorld(t)
turns := newMemTurns(w)
w.h.Turns = turns
a, b := pr(oneRepo, 1, oneHead, "feat/two"), pr(oneRepo2, 2, oneHead2, "feat/two")
requiresCheck(w, a, b)
w.h.PullUpdated(a)
w.h.PullUpdated(b)
w.h.wantForge()
putCheckerVerdict(w, "pass", headOf(oneRepo, 1, oneHead), headOf(oneRepo2, 2, oneHead2))
w.h.CommitStatus(checkStatus(oneRepo, oneHead, "success", oneTrunk))
w.h.CommitStatus(checkStatus(oneRepo2, oneHead2, "success", oneTrunk))
for range 3 {
w.settleAll()
}
want := "put " + oneRepo2 + " after 0 merges; put " + oneRepo + " after 0 merges; delete " + oneRepo2 + " after 2 merges; delete " + oneRepo + " after 2 merges"
if len(w.forge.merges) != 2 || turns.said() != want {
t.Fatalf("merges %v; the turns were %q, want %q", w.forge.merges, turns.said(), want)
}
},
// §8: an approved delivery whose trunk moved goes back to proposed, waits for its carried verdict rather than being
// handed on the old one, and merges on it with no second ask.
"an approved merge whose trunk moved merges on its carried verdict with no second ask": func(t *testing.T) {
w := newWorld(t)
putCheckerVerdict(w, "running", headOf(oneRepo, 9, otherHead)) // holds the approved merge (§2)
id := approvedLone(w)
w.settleAll()
if w.op.count() != 1 || len(w.forge.merges) != 0 {
t.Fatalf("asked %d times, merges %v; want approved and held", w.op.count(), w.forge.merges)
}
w.h.PullMerged(merged(oneRepo, 5, otherHead, oneTrunk2)) // a merge at the forge moves the trunk
putCheckerVerdict(w, "pass", headOf(oneRepo, 9, otherHead))
w.settleAll()
if len(w.forge.merges) != 0 {
t.Fatalf("merges %v: handed on a verdict judged on the old trunk", w.forge.merges)
}
d := w.delivery(id)
if d.Ready == nil || !strings.Contains(d.Ready.Waits, "being carried") {
t.Fatalf("after the trunk moved it waits for %q, want its verdict carried", d.Ready.Waits)
}
// The checker's carried line: the head on the new trunk, carried from the old one.
carried := checkStatus(oneRepo, oneHead, "success", oneTrunk2)
carried.Description = "pass " + oneHead[:12] + " on " + oneTrunk2[:12] + " | carried from " + oneTrunk[:12] + " | unit 0 run, 4 kept"
if head, trunk := checkLine(carried.Description); head != oneHead[:12] || trunk != oneTrunk2[:12] {
t.Fatalf("the carried line reads as %s on %s", head, trunk)
}
if head, trunk := checkLine("failed: a test"); head != "" || trunk != "" {
t.Fatalf("a line that names no commits gave %q %q", head, trunk)
}
putCheckerVerdict(w, "pass", headOf(oneRepo, 1, oneHead))
w.h.CommitStatus(carried)
w.settleAll()
if w.op.count() != 1 {
t.Errorf("the Approve was asked %d times, want once: it stands through the carry", w.op.count())
}
if d := w.delivery(id); d.Check == nil || d.Check.Trunk != oneTrunk2[:12] || len(w.forge.merges) != 1 {
t.Errorf("after the carry: verdict %+v, merges %v; want merged on the carried verdict", d.Check, w.forge.merges)
}
},
// §4: delivery_turns reads every turn with the check it waits for; delivery_hand_now hands a held merge at once on the
// operator's word, and refuses one no check holds.
"the turns are read, and a held merge is handed now on the operator's word": func(t *testing.T) {
w := newWorld(t)
w.h.Turns = newMemTurns(w)
putCheckerVerdict(w, "running", headOf(oneRepo, 9, otherHead))
id := approvedLone(w)
w.settleAll()
read := w.h.TurnsRead()
lines := read["turns"].([]TurnLine)
if len(lines) != 1 || lines[0].Repository != oneRepo || len(lines[0].Turn.Waiting) != 1 || lines[0].Turn.Waiting[0].WaitsFor == "" ||
!strings.Contains(lines[0].Waits["feat/one"].(string), "the check g.") || read["written"] != true {
t.Fatalf("delivery_turns = %+v", read)
}
if _, err := w.h.HandNow(id, ""); err == nil || !strings.Contains(err.Error(), "why") {
t.Errorf("a hand without the operator's why: %v", err)
}
if _, err := w.h.HandNow(IDOf(oneRepo2, oneHead2), "x"); err == nil {
t.Error("an unknown delivery was handed")
}
if _, err := w.h.HandNow(id, "the operator, on telegram: merge it now"); err != nil {
t.Fatal(err)
}
w.settleAll()
if len(w.forge.merges) != 1 || w.delivery(id).Ready.HandNow == "" {
t.Fatalf("merges %v after the hand", w.forge.merges)
}
if _, err := w.h.HandNow(id, "again"); err == nil {
t.Error("a merged delivery was handed again")
}
},
} {
t.Run(name, run)
}
}
@@ -224,6 +224,9 @@ func (h *Holder) nextStall(now time.Time) time.Time {
consider(r.RefusedSince.Add(h.answeredRefusalBound(d.Repository)))
consider(r.RefusedSince.Add(outOfDateFor))
}
if r.HeldSince != nil {
consider(r.HeldSince.Add(turnBound))
}
}
return next
}
@@ -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()
}
@@ -6,7 +6,7 @@ import "git.novox.be/novox/mesh-sdk/go/announce"
// issue 1790): the module says it itself, from its code, before it serves, and a reader is granted what it said
// (clause 3). A state is readable when another module's manifest reads it (keeps_test.go names the readers). keeps_test.go holds the list to the
// manifest's `state`, and keeps_integration_test.go that the module announces it on a bus.
var Kept = []announce.State{{Name: "asks", Readable: true}, {Name: "conditions"}, {Name: "deliveries"}, {Name: "groups"}, {Name: "merges", Readable: true}, {Name: "owed", Readable: true}, {Name: "pulls", Readable: true}, {Name: "source", Readable: true}}
var Kept = []announce.State{{Name: "asks", Readable: true}, {Name: "conditions"}, {Name: "deliveries"}, {Name: "groups"}, {Name: "merges", Readable: true}, {Name: "owed", Readable: true}, {Name: "pulls", Readable: true}, {Name: "source", Readable: true}, {Name: "turns", Readable: true}}
// announceKept says Kept through keeps, which is stdio.Keeps.
func announceKept(keeps func(...announce.State) error) error { return keeps(Kept...) }
@@ -17,6 +17,9 @@ var readers = map[string][]string{
"pulls": {"build-agent"}, // in mesh-controller
"owed": {"mesh-delivery-forge-bridge"},
"source": {"mesh-delivery-indexer", "mesh-delivery-checker"},
// ADR 0548 §3, §4: the checker reads the merge turns and claims no check of a repository one names; the gate alone writes
// them (a state is written by its owner's account alone, ADR 0495 §3: announcing it readable grants a read, never a write).
"turns": {"mesh-delivery-checker"},
}
// 1790/S: the states the module announces are the states its manifest declares, no more, no fewer, and the readable
@@ -65,6 +65,7 @@ func main() {
Operator: kvOperator{},
Merges: kvMerges{},
Source: kvSource{},
Turns: kvTurns{},
Owed: kvOwed{},
Pulls: kvPulls{},
// The gate says what stalled to the operator itself (issue 1275, ADR 0437 §9).
@@ -357,6 +358,24 @@ func tools(h *Holder, l *listening) []stdio.Tool {
}
return h.Close(strArg(a, "id"), strArg(a, "why"))
}},
{Name: "delivery_turns",
Description: "Every merge turn (novox/hq ADR 0548 §3): per repository, the approved merges that wait to merge into it in the order of " +
"their Approve, the check of another delivery each waits for and since when, whether the gate wrote the turn to its bucket " +
"turns yet (the checker claims no check of a repository a turn names), and the bound past which a held merge is a stalled " +
"line. Read-only.",
Run: func(map[string]any) (any, error) { return h.TurnsRead(), nil }},
{Name: "delivery_hand_now",
Description: "On the operator's word, hand an approved merge held for another delivery's running check to the merge at once " +
"(novox/hq ADR 0548 §4): the check it waited for runs on and is carried to the new trunk after the merge. Refused for a " +
"delivery that is not an approved merge held by a check, naming why. The why is kept on the delivery.",
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id (owner/repository@commit)"),
"why": str("the operator's word, kept on the delivery")}, "required": []string{"id", "why"}},
Run: func(a map[string]any) (any, error) {
if err := need(a, "id", "why"); err != nil {
return nil, err
}
return h.HandNow(strArg(a, "id"), strArg(a, "why"))
}},
{Name: "delivery_status",
Description: "Whether this owner works: how many deliveries in each state, the groups, what is owed to " +
"the forge and the bus and not yet done, how many owed effects were tried since it started (done, failed " +
@@ -91,9 +91,10 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) {
if !reflect.DeepEqual(m.Emits, []string{"transition", "group", "merged"}) {
t.Fatalf("emits %v", m.Emits)
}
// Eight buckets, and no `asked` (the teller asks, ADR 0437 §6): ADR 0450 §3, and the conditions it raised (issue 1275).
// Nine buckets, and no `asked` (the teller asks, ADR 0437 §6): ADR 0450 §3, the conditions it raised (issue 1275) and the
// merge turns (ADR 0548 §3).
buckets := m.buckets()
if !reflect.DeepEqual(buckets, []string{"deliveries", "groups", "pulls", "asks", "merges", "owed", "source", "conditions"}) {
if !reflect.DeepEqual(buckets, []string{"deliveries", "groups", "pulls", "asks", "merges", "owed", "source", "conditions", "turns"}) {
t.Fatalf("state %v", buckets)
}
// What it reads: the teller's answers and the checker's verdicts (issue 1302). The carry from today's module's buckets, the
@@ -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()
}
@@ -199,14 +199,3 @@ func TestAClosedPullRequestAsksNoForgetChecks(t *testing.T) {
w.h.PullUpdated(p)
w.h.PullClosed(p)
}
// 1302: the mesh/check line gives the head and the trunk it judged.
func TestTheCheckLineGivesHeadAndTrunk(t *testing.T) {
head, trunk := checkLine("pass " + oneHead[:12] + " on " + oneTrunk[:12] + " | unit 4 run, 0 kept | group g of 2 pull requests")
if head != oneHead[:12] || trunk != oneTrunk[:12] {
t.Fatalf("head %q trunk %q", head, trunk)
}
if head, trunk = checkLine("failed: a test"); head != "" || trunk != "" {
t.Fatalf("a line that names no commits gave %q %q", head, trunk)
}
}
@@ -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)
@@ -267,6 +277,7 @@ func (h *Holder) readyAsks(now time.Time) []func() {
asks = append(asks, a)
}
}
asks = append(asks, h.turnWrites()...) // the merge turns, after the holds of this look (ADR 0548 §3)
asks = append(asks, h.settleMerges()...)
asks = append(asks, h.settleForgeMerges()...)
asks = append(asks, h.withdrawnKeys()...)
@@ -376,6 +387,9 @@ func (h *Holder) readyWaits(d *Delivery) string {
case d.State == Proposed && d.outOfDate() != "" && d.AskedAgain != nil:
return "its check against the trunk's head is asked and has not answered"
case d.State == Proposed && d.outOfDate() != "" && d.Check.Passes() && h.covering(d) != nil:
// Approved, and its trunk moved: the checker carries its verdict to the trunk's head by its package digests (ADR 0548
// §5, §8). The Approve stands and is not asked again; the merge waits for the carried verdict, not a fresh check.
return "approved; its verdict is being carried to the trunk's head " + shortOf(d.TrunkNow, 12) + " (" + d.outOfDate() + ")"
case d.State == Proposed && d.outOfDate() != "":
return "its verdict is out of date: its check against the trunk's head is asked before the operator is"
case d.State == Proposed && d.AskedAgain != nil:
@@ -0,0 +1,454 @@
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
}
@@ -26,6 +26,10 @@ type Line struct {
// waitsFor is what a delivery waits for, in words.
func (h *Holder) waitsFor(d *Delivery) string {
if r := d.Ready; r != nil && r.Merged == nil && strings.HasPrefix(r.Waits, "approved;") {
// Approved and waiting for another delivery's check, or for its own verdict's carry (ADR 0548 §2, §8, §9).
return r.Waits
}
switch d.State {
case Proposed:
return "its check"
@@ -218,6 +222,7 @@ func (h *Holder) Stalled() []StalledLine {
}
out = append(out, h.withChecks(h.uncheckedLines(now))...)
out = append(out, h.readyLines(now)...)
out = append(out, h.turnLines(now)...) // an approved merge held past its bound by another delivery's check (ADR 0548 §4)
out = append(out, h.groupLines(now)...)
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
return out
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-gate
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f h1:9+TZpNq8Ifj1+qmbidE6l/7qkbua4yLdzRMrezNvf1A=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026111253-9c61918ac36f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+8 -2
View File
@@ -61,14 +61,20 @@
"name": "source",
"history": 64
},
"conditions"
"conditions",
{
"name": "turns",
"history": 5
}
],
"reads": [
"mesh-delivery-teller.answers",
"mesh-delivery-checker.verdicts"
],
"tools": [
"delivery_status"
"delivery_hand_now",
"delivery_status",
"delivery_turns"
],
"config": {
"refusal-stall-floor-minutes": {