From 2c8c1543baf26001648c2c3acfb4000c4cc031df Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 13:48:54 +0200 Subject: [PATCH] issue 975: the pull request's key written by compare-and-set, the ready flag on a ready mark's check, build-agent reads the key --- modules/build-agent/module.json | 3 + .../cmd/mesh-delivery/bus_test.go | 35 +++++ .../cmd/mesh-delivery/fakes_test.go | 5 + .../mesh-delivery/cmd/mesh-delivery/holder.go | 4 +- .../mesh-delivery/cmd/mesh-delivery/main.go | 3 + .../cmd/mesh-delivery/manifest_test.go | 2 +- .../mesh-delivery/cmd/mesh-delivery/ports.go | 4 +- .../cmd/mesh-delivery/pullkey.go | 115 ++++++++++++++ .../cmd/mesh-delivery/pullkey_test.go | 140 ++++++++++++++++++ .../mesh-delivery/cmd/mesh-delivery/ready.go | 1 + modules/mesh-delivery/module.json | 3 +- 11 files changed, 311 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-delivery/cmd/mesh-delivery/pullkey.go create mode 100644 modules/mesh-delivery/cmd/mesh-delivery/pullkey_test.go diff --git a/modules/build-agent/module.json b/modules/build-agent/module.json index b960eacf..f9b789af 100644 --- a/modules/build-agent/module.json +++ b/modules/build-agent/module.json @@ -12,6 +12,9 @@ "serves": ["current", "kill", "pause", "resume"] } ], + "reads": [ + "mesh-delivery.pulls" + ], "requires": [ "artifact-store", { diff --git a/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go b/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go index 9d2f681f..ded238e3 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/bus_test.go @@ -286,6 +286,8 @@ func (r *runtime) answer(m rpcMessage) { case "seat:" + ControllerSeat + ".delivery-plan": return toolAnswer(map[string]any{"plan": DeliveryPlan{Repository: "novox/app", Moved: []string{"app"}, Summary: "builds app, sent to the node"}}), nil + case "seat:" + ControllerSeat + ".delivery-check": + return toolAnswer(map[string]any{"asked": "check-1"}), nil case "gitea.gitea_open_pulls": return toolAnswer(map[string]any{"pulls": pulls}), nil case "gitea.gitea_note_append", "gitea.gitea_delivery_view", "gitea.gitea_commit_status", "gitea.gitea_merge_pull_request", @@ -404,6 +406,7 @@ func TestAReadyMarkAndItsApproveAreMergedWithinSecondsOfTheirEventsOverARealBus( // A pull request into main is announced, and the controller's verdict on its head passes. const head = "1111111111111111111111111111111111111111" announced := pr("novox/app", 7, head, "fix/x", "src/a.go") + announced.BaseSHA = "t1t1t1t1t1t1" announced.Title = "Fix the apostrophe (issue 975)" b.publish(t, "mesh.mod.gitea.event.pull.updated", announced) b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/app", 7, head, "pass", "t1t1t1t1t1t1")) @@ -421,6 +424,12 @@ func TestAReadyMarkAndItsApproveAreMergedWithinSecondsOfTheirEventsOverARealBus( } return false }) + // The pull request's key holds its head, the trunk commit it was asked against and the verdict. + rt.until("the pull request's key to hold head, trunk and verdict", 5*time.Second, func() bool { + e, err := b.kvs["pulls"].Get(context.Background(), "novox.app.7") + var k PullKey + return err == nil && json.Unmarshal(e.Value(), &k) == nil && k == PullKey{Head: head, Trunk: "t1t1t1t1t1t1", Verdict: "pass"} + }) reads := len(rt.asksOf("gitea.gitea_open_pulls")) // The reviewer's label and comment: the forge shows the label alone at first, and the mark whole once the comment is @@ -479,3 +488,29 @@ func TestAReadyMarkAndItsApproveAreMergedWithinSecondsOfTheirEventsOverARealBus( t.Fatalf("%d asks of the mesh in a quiet second and a half after the merge: something polls", more) } } + +// A ready mark on a pull request whose check nobody asked yet asks it at once, on the ready stream (the check's `ready`). +func TestAReadyMarkAsksTheCheckOfAPullRequestWithNoVerdictAsReady(t *testing.T) { + b := newTestBus(t) + rt := b.start(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rt.deliverEvents(ctx) + rt.until("the module to subscribe", 20*time.Second, func() bool { + rt.mu.Lock() + defer rt.mu.Unlock() + return len(rt.subs) >= len(followed)+1 + }) + const head = "2222222222222222222222222222222222222222" + b.publish(t, "mesh.mod.gitea.event.pull.updated", pr("novox/app", 8, head, "fix/y", "src/a.go")) + rt.mu.Lock() + rt.openPulls = []OpenPull{markedPull("novox/app", 8, head, head)} + rt.mu.Unlock() + b.publish(t, "mesh.mod.gitea.event.forge.pull-request.commented", commentedReady(8, head)) + rt.until("the check to be asked", 5*time.Second, func() bool { return len(rt.asksOf("seat:"+ControllerSeat+".delivery-check")) >= 1 }) + body := rt.asksOf("seat:" + ControllerSeat + ".delivery-check")[0].Body + var members []Member + if err := json.Unmarshal([]byte(fmt.Sprint(body["members"])), &members); err != nil || len(members) != 1 || !members[0].Ready { + t.Fatalf("the check was asked for %v (%v); wanted one member with ready", body["members"], err) + } +} diff --git a/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go b/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go index cdb962e6..1c2a9a94 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go @@ -108,6 +108,8 @@ type fakeController struct { down bool // walkReads counts the reads of the controller's walks (Walks). walkReads int + // checkReady is the ready flag of each member of each check asked. + checkReady []bool // refuseDeliver refuses every word, as a controller that will not take it does (review of novox/hq issue 753). refuseDeliver bool // onCheck runs as a check is asked, before it is answered: what the forge held at that moment. @@ -178,6 +180,9 @@ func (c *fakeController) Check(group string, ms []Member) (string, error) { heads = append(heads, m.ID) } c.checks = append(c.checks, group+":"+strings.Join(heads, ",")) + for _, m := range ms { + c.checkReady = append(c.checkReady, m.Ready) + } return id, nil } func (c *fakeController) Deliver(walk, why string) error { diff --git a/modules/mesh-delivery/cmd/mesh-delivery/holder.go b/modules/mesh-delivery/cmd/mesh-delivery/holder.go index e5e529af..7b448e2b 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/holder.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/holder.go @@ -29,6 +29,8 @@ type Holder struct { Logf func(format string, a ...any) // Operator is the operator channel, asked for the Approve of a merge into main (approval.go). Operator Operator + // Pulls keeps the pull requests' keys on the bus (pullkey.go); nil keeps none. + Pulls PullKeys // cancels are asks to take back, gathered under the lock and run with the other asks of the look. cancels []func() @@ -1705,7 +1707,7 @@ func (h *Holder) askCheck(d *Delivery, now time.Time) func() { if d.State != Proposed || d.Check != nil || d.CheckAsk != nil || d.MergedAs != "" || d.Number == 0 { return nil } - if now.Sub(d.Since) < checkGrace { + if now.Sub(d.Since) < checkGrace && d.Ready == nil { // a ready mark makes the check worth asking at once h.lookAt(d.Since.Add(checkGrace), "a proposed delivery's check, asked once the grace has passed with no verdict") return nil } diff --git a/modules/mesh-delivery/cmd/mesh-delivery/main.go b/modules/mesh-delivery/cmd/mesh-delivery/main.go index 6bfaffcb..a2966be8 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/main.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/main.go @@ -46,6 +46,7 @@ func main() { Emit: func(event string, body any) error { return stdio.Emit(event, body) }, Logf: logf, Operator: operator, + Pulls: kvPulls{}, } h.init() l := &listening{now: "not yet: starting"} @@ -143,6 +144,7 @@ func take(h *Holder, e stdio.Envelope) error { if h.PullUpdated(p) { h.wantForge() // a head not seen before: the forge shows whether anyone checks it } + h.KeepPull(p) case "gitea.pull.merged": h.PullMerged(p) default: @@ -176,6 +178,7 @@ func take(h *Holder, e stdio.Envelope) error { return nil } h.Checked(c) + h.KeepVerdict(c) case ControllerSeat + ".plan-moved": var w Walk if err := json.Unmarshal(e.Body, &w); err != nil { diff --git a/modules/mesh-delivery/cmd/mesh-delivery/manifest_test.go b/modules/mesh-delivery/cmd/mesh-delivery/manifest_test.go index 1915e3b2..b724779a 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/manifest_test.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/manifest_test.go @@ -57,7 +57,7 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { if !reflect.DeepEqual(m.Emits, []string{"transition", "group"}) { t.Fatalf("emits %v", m.Emits) } - if !reflect.DeepEqual(m.State, []string{"deliveries", "groups", "asked"}) { + if !reflect.DeepEqual(m.State, []string{"deliveries", "groups", "asked", "pulls"}) { t.Fatalf("state %v", m.State) } // It calls exactly what its ports ask: the controller's six verbs and the forge's four tools. diff --git a/modules/mesh-delivery/cmd/mesh-delivery/ports.go b/modules/mesh-delivery/cmd/mesh-delivery/ports.go index 06fa071c..b884cf0b 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/ports.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/ports.go @@ -32,6 +32,8 @@ type Member struct { ModuleDirsSaid bool `json:"module_dirs_said,omitempty"` CloneURL string `json:"clone_url,omitempty"` After []string `json:"after,omitempty"` + // Ready is whether the pull request carries a ready mark: its check is published on the ready stream (ADR 0406 §3). + Ready bool `json:"ready,omitempty"` // BaseSHA is the base branch's head the forge last announced the head with (novox/hq ADR 0323): the trunk commit // the controller's verdict names when no build judges the head. BaseSHA string `json:"base_sha,omitempty"` @@ -39,7 +41,7 @@ type Member struct { // MemberOf is a delivery as a member. func MemberOf(d *Delivery) Member { - return Member{ID: d.ID, Repository: d.Repository, Base: d.Base, Head: d.Commit, Number: d.Number, Paths: d.Paths, + return Member{Ready: d.Ready != nil && d.Ready.Merged == nil, ID: d.ID, Repository: d.Repository, Base: d.Base, Head: d.Commit, Number: d.Number, Paths: d.Paths, PathsTruncated: d.PathsTruncated, Removed: d.Removed, ModuleDirs: d.ModuleDirs, ModuleDirsSaid: d.ModuleDirsSaid, CloneURL: d.CloneURL, After: d.After, BaseSHA: d.BaseSHA} } diff --git a/modules/mesh-delivery/cmd/mesh-delivery/pullkey.go b/modules/mesh-delivery/cmd/mesh-delivery/pullkey.go new file mode 100644 index 00000000..2447b91b --- /dev/null +++ b/modules/mesh-delivery/cmd/mesh-delivery/pullkey.go @@ -0,0 +1,115 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// The pull request's key (novox/hq ADR 0406 §4, issue 975): mesh-delivery keeps one key per pull request on the bus, +// holding its head, the trunk commit its current check is asked against and its verdict, written by compare-and-set. +// It writes a new head when the forge says one (`pull.updated`, which carries the head and the trunk commit it was +// announced against), and a new trunk commit only when the check is asked again against one — the merge tool's ask is +// that same announcement. A merge in a group's order, and a trunk move that leaves an inactive pull request unasked, +// are announced as nothing (the gitea module announces `pull.updated` only for a head and for the merge tool's ask), so +// they write nothing. A verdict for that head and trunk commit writes the verdict. The build seat reads the key. + +// PullKey is what the bus keeps of one pull request. +type PullKey struct { + Head string `json:"head"` + Trunk string `json:"trunk,omitempty"` + Verdict string `json:"verdict,omitempty"` +} + +// PullKeys keeps the pull requests' keys: Change reads the key, applies the change to what it read and writes the result +// by compare-and-set when the change says to write; a write another writer won reads the key again and applies again. +type PullKeys interface { + Change(key string, apply func(PullKey) (PullKey, bool)) error +} + +// pullKeyName is a pull request's key: owner.repo.number, with any dot in a name made an underscore. +func pullKeyName(owner, repo string, number int) string { + clean := func(s string) string { return strings.NewReplacer(".", "_", "/", "_").Replace(s) } + return fmt.Sprintf("%s.%s.%d", clean(owner), clean(repo), number) +} + +// KeepPull writes the key of a pull request whose head was announced, when the head or the trunk commit it was +// announced against is not what the key holds; a new head or trunk clears the verdict. +func (h *Holder) KeepPull(p PullEvent) { + if h.Pulls == nil || p.Number == 0 || p.HeadSHA == "" { + return + } + err := h.Pulls.Change(pullKeyName(p.Owner, p.Repo, p.Number), func(k PullKey) (PullKey, bool) { + if sameCommit(k.Head, p.HeadSHA) && k.Head != "" && (p.BaseSHA == "" || sameCommit(k.Trunk, p.BaseSHA)) { + return k, false + } + return PullKey{Head: p.HeadSHA, Trunk: p.BaseSHA}, true + }) + if err != nil { + h.Logf("[mesh-delivery] the key of %s/%s#%d could not be written: %v", p.Owner, p.Repo, p.Number, err) + h.again("a pull request's key that could not be written") + } +} + +// KeepVerdict writes a verdict into the key of its pull request, when the key holds that head and trunk commit. +func (h *Holder) KeepVerdict(c CheckedEvent) { + if h.Pulls == nil || c.Group != "" || c.Number == 0 || c.Commit == "" { + return + } + verdict := c.Verdict + if verdict == "" { + return + } + err := h.Pulls.Change(pullKeyName(c.Owner, c.Repo, c.Number), func(k PullKey) (PullKey, bool) { + if !sameCommit(k.Head, c.Commit) || (c.Trunk != "" && !sameCommit(k.Trunk, c.Trunk)) || k.Verdict == verdict { + return k, false + } + k.Verdict = verdict + return k, true + }) + if err != nil { + h.Logf("[mesh-delivery] the verdict in the key of %s/%s#%d could not be written: %v", c.Owner, c.Repo, c.Number, err) + h.again("a pull request's key that could not be written") + } +} + +// kvPulls keeps the keys in the module's own state `pulls`. +type kvPulls struct{} + +// changeAttempts is how often a lost compare-and-set is read and applied again before it is said. +const changeAttempts = 10 + +func (kvPulls) Change(key string, apply func(PullKey) (PullKey, bool)) error { + state := stdio.State("pulls") + for range changeAttempts { + e, err := state.Get(key) + if err != nil { + return err + } + var k PullKey + if e != nil { + if err := json.Unmarshal(e.Value, &k); err != nil { + return err + } + } + next, write := apply(k) + if !write { + return nil + } + if e == nil { + _, err = state.Create(key, next) + } else { + _, err = state.Update(key, next, e.Revision) + } + if err == nil { + return nil + } + if !errors.Is(err, stdio.ErrChanged) { + return err + } + } + return fmt.Errorf("%s was written by another writer first %d times: %w", key, changeAttempts, stdio.ErrChanged) +} diff --git a/modules/mesh-delivery/cmd/mesh-delivery/pullkey_test.go b/modules/mesh-delivery/cmd/mesh-delivery/pullkey_test.go new file mode 100644 index 00000000..936c4e4f --- /dev/null +++ b/modules/mesh-delivery/cmd/mesh-delivery/pullkey_test.go @@ -0,0 +1,140 @@ +package main + +import ( + "errors" + "sync" + "testing" +) + +// **The pull request's key** (novox/hq ADR 0406 §4, issue 975): one key per pull request on the bus, holding its head, +// the trunk commit its current check is asked against and its verdict, written by compare-and-set. A head the forge +// announces writes head and trunk; a verdict for that head and trunk writes the verdict; nothing else writes. + +type fakePulls struct { + mu sync.Mutex + keys map[string]PullKey + writes int + // lose makes the first n writes lose to another writer, which sets rival. + lose int + rival PullKey +} + +func newFakePulls() *fakePulls { return &fakePulls{keys: map[string]PullKey{}} } + +func (f *fakePulls) Change(key string, apply func(PullKey) (PullKey, bool)) error { + f.mu.Lock() + defer f.mu.Unlock() + for tries := 0; tries < 5; tries++ { + next, write := apply(f.keys[key]) + if !write { + return nil + } + if f.lose > 0 { // another writer was first: read again and apply again + f.lose-- + f.keys[key] = f.rival + continue + } + f.keys[key] = next + f.writes++ + return nil + } + return errors.New("lost the race too often") +} + +func (f *fakePulls) get(key string) PullKey { + f.mu.Lock() + defer f.mu.Unlock() + return f.keys[key] +} + +const appKey = "novox.app.7" + +func announcedAt(head, trunk string) PullEvent { + p := pr("novox/app", 7, head, "feat/x", "src/a.go") + p.BaseSHA = trunk + return p +} + +func TestAnAnnouncedHeadWritesTheKeyAndTheSameAnnouncementAgainWritesNothing(t *testing.T) { + w := newWorld(t) + pulls := newFakePulls() + w.h.Pulls = pulls + p := announcedAt(readyHead, "t1t1t1t1t1t1") + w.h.PullUpdated(p) + w.h.KeepPull(p) + if got := pulls.get(appKey); got.Head != readyHead || got.Trunk != "t1t1t1t1t1t1" || got.Verdict != "" { + t.Fatalf("the key holds %+v", got) + } + w.h.KeepPull(p) + if pulls.writes != 1 { + t.Fatalf("the same announcement wrote %d times", pulls.writes) + } +} + +func TestAVerdictForTheKeysHeadAndTrunkIsWrittenAndOneForAnotherIsNot(t *testing.T) { + w := newWorld(t) + pulls := newFakePulls() + w.h.Pulls = pulls + w.h.KeepPull(announcedAt(readyHead, "t1t1t1t1t1t1")) + old := judgedAt("novox/app", 7, pushed, "pass", "t1t1t1t1t1t1") + w.h.KeepVerdict(old) + if pulls.get(appKey).Verdict != "" { + t.Fatal("a verdict for another head was written") + } + other := judgedAt("novox/app", 7, readyHead, "pass", "t9t9t9t9t9t9") + w.h.KeepVerdict(other) + if pulls.get(appKey).Verdict != "" { + t.Fatal("a verdict for another trunk commit was written") + } + w.h.KeepVerdict(judgedAt("novox/app", 7, readyHead, "pass", "t1t1t1t1t1t1")) + if got := pulls.get(appKey); got.Verdict != "pass" { + t.Fatalf("the key holds %+v", got) + } +} + +func TestANewTrunkCommitClearsTheVerdictAndANewHeadReplacesTheKey(t *testing.T) { + w := newWorld(t) + pulls := newFakePulls() + w.h.Pulls = pulls + w.h.KeepPull(announcedAt(readyHead, "t1t1t1t1t1t1")) + w.h.KeepVerdict(judgedAt("novox/app", 7, readyHead, "pass", "t1t1t1t1t1t1")) + w.h.KeepPull(announcedAt(readyHead, "t2t2t2t2t2t2")) + if got := pulls.get(appKey); got.Trunk != "t2t2t2t2t2t2" || got.Verdict != "" { + t.Fatalf("the check asked again against a new trunk left %+v", got) + } + w.h.KeepPull(announcedAt(pushed, "t2t2t2t2t2t2")) + if got := pulls.get(appKey); got.Head != pushed { + t.Fatalf("a new head left %+v", got) + } +} + +func TestALostRaceReadsTheKeyAgainAndAppliesTheChangeToWhatItFound(t *testing.T) { + w := newWorld(t) + pulls := newFakePulls() + pulls.lose, pulls.rival = 1, PullKey{Head: readyHead, Trunk: "t1t1t1t1t1t1", Verdict: "pass"} + w.h.Pulls = pulls + // The rival writer already holds this very head and trunk with a verdict: the change finds nothing to write. + w.h.KeepPull(announcedAt(readyHead, "t1t1t1t1t1t1")) + if got := pulls.get(appKey); got.Verdict != "pass" || pulls.writes != 0 { + t.Fatalf("after the lost race the key holds %+v after %d writes", got, pulls.writes) + } +} + +func TestAPullRequestMarkedReadyHasItsCheckAskedOnTheReadyStream(t *testing.T) { + w := newWorld(t) + w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go")) + w.forge.open = []OpenPull{markedPull("novox/app", 7, readyHead, readyHead)} + w.h.wantForge() + w.settleAll() + if len(w.ctl.checkReady) == 0 || !w.ctl.checkReady[0] { + t.Fatalf("the check of a marked pull request was asked with ready %v", w.ctl.checkReady) + } + // An unmarked one is asked as before. + w2 := newWorld(t) + w2.h.PullUpdated(pr("novox/app", 8, pushed, "feat/y", "src/a.go")) + w2.later(checkGrace + 1) + w2.settleAll() + if len(w2.ctl.checkReady) == 0 || w2.ctl.checkReady[0] { + t.Fatalf("the check of an unmarked pull request was asked with ready %v", w2.ctl.checkReady) + } +} diff --git a/modules/mesh-delivery/cmd/mesh-delivery/ready.go b/modules/mesh-delivery/cmd/mesh-delivery/ready.go index faa89a5e..32c23bd0 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/ready.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/ready.go @@ -199,6 +199,7 @@ func (h *Holder) readyAsks(now time.Time) []func() { } if d.Ready == nil || d.Ready.Head != mark.Head || (d.Ready.Declined != nil && mark.At != d.Ready.MarkedAt) { d.Ready = &ReadyRun{Head: mark.Head, By: mark.By, MarkedAt: mark.At, Seen: now} + h.Poke() // its check, if none is asked, is worth asking now (askCheck) h.Logf("[mesh-delivery] %s is marked ready at %s by %s: merged once its check passes against its trunk's head", d.ID, shortOf(mark.Head, 12), orNone(mark.By)) h.keep(d) diff --git a/modules/mesh-delivery/module.json b/modules/mesh-delivery/module.json index 617e5615..8cad5f5e 100644 --- a/modules/mesh-delivery/module.json +++ b/modules/mesh-delivery/module.json @@ -64,7 +64,8 @@ "state": [ "deliveries", "groups", - "asked" + "asked", + "pulls" ], "tools": [ "delivery_status"