Files
mesh-catalog/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go
T
jschoubben 373f339750
mesh/merge-gate head 373f33975077 merged into trunk 7913b56d66f7 — pass: builds mesh-delivery → novox; no bus step; every machine composes with the change …
mesh/repo-check head 373f33975077 merged into trunk 7913b56d66f7 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
Give a walk the word of a stopped delivery that waits for it (hq issue 753)
A delivery its group stopped for a member delivered since waits for a later
walk of its repository that contains its merge (issue 747). When that walk
carried no living delivery, nothing gave it its word and it waited forever.
The delivery now takes its turn on the walk that answers its merge, in its
group's order. A walk owed such a word stays under the orphans' timer, so
past the grace without it `stalled` lists it with why (held, or the
controller's refusal), and a walk no living delivery holds names the ended
deliveries it carries.
2026-10-15 00:30:26 +02:00

682 lines
20 KiB
Go

package main
import (
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"sync"
"testing"
"time"
)
// What a test stands the holder on: a store in memory that can be made to fail, a controller that records
// what it was asked and answers what it is told, a forge that records every note, view and status.
type memStore struct {
mu sync.Mutex
deliveries map[string][]byte
groups map[string][]byte
failing bool
// puts counts the deliveries written, failed or not.
puts int
}
func newMemStore() *memStore {
return &memStore{deliveries: map[string][]byte{}, groups: map[string][]byte{}}
}
func (s *memStore) PutDelivery(d *Delivery) error {
s.mu.Lock()
defer s.mu.Unlock()
s.puts++
if s.failing {
return errors.New("the bus is away")
}
s.deliveries[d.ID] = mustJSON(d)
return nil
}
func (s *memStore) DeleteDelivery(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.deliveries, id)
return nil
}
func (s *memStore) PutGroup(g *Group) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.failing {
return errors.New("the bus is away")
}
s.groups[g.ID] = mustJSON(g)
return nil
}
func (s *memStore) DeleteGroup(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.groups, id)
return nil
}
func (s *memStore) Deliveries() ([]*Delivery, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*Delivery
for _, raw := range s.deliveries {
var d Delivery
if err := json.Unmarshal(raw, &d); err != nil {
return nil, err
}
out = append(out, &d)
}
return out, nil
}
func (s *memStore) Groups() ([]*Group, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*Group
for _, raw := range s.groups {
var g Group
if err := json.Unmarshal(raw, &g); err != nil {
return nil, err
}
out = append(out, &g)
}
return out, nil
}
type fakeController struct {
mu sync.Mutex
walks map[string]Walk
held bool
order func([]Member) Order
checks []string
asked int
delivers []string
stops []string
// withouts are the stops that left merges out (novox/hq ADR 0299): "<walk> without <merge>,…".
withouts []string
// refuseWithout refuses every stop with merges to leave out, as a controller before ADR 0299 does; mergedAt are
// the merges' times, by commit, for what a later merge contains.
refuseWithout bool
mergedAt map[string]time.Time
// refuseStop refuses every whole stop too; moves are the modules each merge moves, by commit, for what a left-out
// merge shares (ADR 0299).
refuseStop bool
moves map[string][]string
plans int
down 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.
onCheck func()
// planned is what Plan answers instead of "builds app", when set.
planned *DeliveryPlan
// forgot are the pull requests whose check builds were let go of (novox/hq ADR 0350 §5), "owner/repo#n".
forgot []string
}
func (c *fakeController) ForgetChecks(owner, repo string, number int) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return errors.New("down")
}
c.forgot = append(c.forgot, fmt.Sprintf("%s/%s#%d", owner, repo, number))
return nil
}
func newFakeController() *fakeController {
return &fakeController{walks: map[string]Walk{}, held: true, mergedAt: map[string]time.Time{},
moves: map[string][]string{}}
}
func (c *fakeController) Plan(repository, base, head string, paths, dirs, removed []string) (*DeliveryPlan, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.plans++
if c.down {
return nil, errors.New("down")
}
if c.planned != nil {
p := *c.planned
return &p, nil
}
return &DeliveryPlan{Repository: repository, Base: base, Head: head, Moved: []string{"app"}, Summary: "builds app"}, nil
}
func (c *fakeController) Order(ms []Member) (Order, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return Order{}, errors.New("down")
}
if c.order != nil {
return c.order(ms), nil
}
var ids []string
for _, m := range ms {
ids = append(ids, m.ID)
}
sort.Strings(ids)
return Order{Order: ids}, nil
}
func (c *fakeController) Check(group string, ms []Member) (string, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return "", errors.New("down")
}
if c.onCheck != nil {
c.onCheck()
}
c.asked++
id := fmt.Sprintf("check-%d", c.asked)
var heads []string
for _, m := range ms {
heads = append(heads, m.ID)
}
c.checks = append(c.checks, group+":"+strings.Join(heads, ","))
return id, nil
}
func (c *fakeController) Deliver(walk, why string) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return errors.New("down")
}
if c.refuseDeliver {
return errors.New("the controller refuses the word")
}
w, known := c.walks[walk]
if !known || w.Delivery == nil || w.Delivery.Go != nil {
return fmt.Errorf("%s waits for nobody", walk)
}
now := time.Now()
w.Delivery.Go, w.Delivery.By, w.Delivery.Why = &now, byOwner, why
w.Revision++
c.walks[walk] = w
c.delivers = append(c.delivers, walk)
return nil
}
func (c *fakeController) Stop(walk, why, by string) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return errors.New("down")
}
c.stops = append(c.stops, walk)
if c.refuseStop {
return errors.New("the controller refuses")
}
if w, known := c.walks[walk]; known && !w.Open() {
// As the controller: only a walk that is building or rolling is stopped (`delivery-stop`).
return fmt.Errorf("%s is already %s", walk, w.State)
} else if known {
w.State = walkFailed
if w.Delivery == nil {
w.Delivery = &WalkDelivery{}
}
w.Delivery.Stopped, w.Delivery.StoppedWhy = by, why
w.Revision++
c.walks[walk] = w
}
return nil
}
// StopWithout ends a walk without the merges named, as the controller does (novox/hq ADR 0299): each merge named,
// and each later merge of its repository by merge time (mergedAt), is left out; the others are put in a batch,
// "<walk>-again", assembling until a test cuts it. Given a batch, the merges left out go to a stopped record,
// "<walk>-left", and the batch goes on with the others. refuseWithout refuses every such stop, as an older
// controller does.
func (c *fakeController) StopWithout(walk, why, by string, without []string) (StopAnswer, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return StopAnswer{}, errors.New("down")
}
c.withouts = append(c.withouts, walk+" without "+strings.Join(without, ","))
if c.refuseWithout {
return StopAnswer{}, errors.New(`delivery-stop takes no argument "without"`)
}
w, known := c.walks[walk]
if !known || !(w.Open() || w.IsBatch()) {
return StopAnswer{}, fmt.Errorf("%s is no open walk or batch", walk)
}
wd := WalkDelivery{}
if w.Delivery != nil {
wd = *w.Delivery
}
named := func(m WalkMerge) bool { return containsString(without, m.Repository+"@"+m.Commit) }
contains := func(m WalkMerge) string {
for _, s := range wd.Merges {
if named(s) && s.Repository == m.Repository && c.mergedAt[m.Commit].After(c.mergedAt[s.Commit]) {
return s.Commit
}
}
return ""
}
shares := func(m WalkMerge) (string, []string) {
for _, s := range wd.Merges {
if named(s) {
var both []string
for _, x := range c.moves[m.Commit] {
if containsString(c.moves[s.Commit], x) {
both = append(both, x)
}
}
if len(both) > 0 {
return s.Commit, both
}
}
}
return "", nil
}
a := StopAnswer{Stopped: walk}
var again, left []WalkMerge
for _, m := range wd.Merges {
s, both := shares(m)
switch {
case named(m):
a.LeftOut = append(a.LeftOut, WalkLeftOut{Repository: m.Repository, Commit: m.Commit})
left = append(left, m)
case len(both) > 0:
a.LeftOut = append(a.LeftOut, WalkLeftOut{Repository: m.Repository, Commit: m.Commit, Contains: s, Shares: both})
left = append(left, m)
case contains(m) != "":
a.LeftOut = append(a.LeftOut, WalkLeftOut{Repository: m.Repository, Commit: m.Commit, Contains: contains(m)})
left = append(left, m)
default:
m.Carried = ""
again = append(again, m)
a.WalkedAgain = append(a.WalkedAgain, m.Repository+"@"+m.Commit)
}
}
if len(a.LeftOut) == 0 {
return StopAnswer{}, fmt.Errorf("%s answers no merge %s", walk, strings.Join(without, ", "))
}
if w.IsBatch() && len(again) == 0 {
// Every merge left out: the batch itself is stopped, as the controller does.
wd.Stopped, wd.StoppedWhy, wd.LeftOut, wd.Batch = by, why, a.LeftOut, nil
w.Delivery, w.State = &wd, walkFailed
w.Revision++
c.walks[walk] = w
return a, nil
}
if w.IsBatch() {
rid := walk + "-left"
c.walks[rid] = Walk{ID: rid, Repository: left[0].Repository, Branch: "main", Commit: left[0].Commit,
Created: w.Created, State: walkFailed, Revision: 1, Modules: map[string]*WalkModule{},
Delivery: &WalkDelivery{Stopped: by, StoppedWhy: why, Merges: left, LeftOut: a.LeftOut, WalkedAgainBy: walk}}
wd.Merges = again
w.Delivery = &wd
w.Repository, w.Commit = again[0].Repository, again[0].Commit
w.Revision++
c.walks[walk] = w
a.Stopped, a.WalkedAgainBy = rid, walk
return a, nil
}
if len(again) > 0 {
id := walk + "-again"
closes := w.Created.Add(time.Minute)
c.walks[id] = Walk{ID: id, Repository: again[0].Repository, Branch: "main", Commit: again[0].Commit,
Created: w.Created.Add(time.Minute), State: walkAssembling, Revision: 1,
Modules: map[string]*WalkModule{}, Delivery: &WalkDelivery{Merges: again, Batch: &WalkBatch{ClosesAt: &closes}}}
wd.WalkedAgainBy, a.WalkedAgainBy = id, id
}
wd.Stopped, wd.StoppedWhy, wd.LeftOut = by, why, a.LeftOut
w.Delivery = &wd
w.State = walkFailed
w.Note = "stopped by " + by + ": " + why
w.Revision++
c.walks[walk] = w
return a, nil
}
// cut makes a batch the walk under the same id, waiting for its word, as the controller cuts it.
func (c *fakeController) cut(id string) Walk {
c.mu.Lock()
w := c.walks[id]
c.mu.Unlock()
wd := *w.Delivery
wd.Batch, wd.Awaits = nil, "mesh-delivery"
w.Delivery = &wd
w.State, w.Tiers, w.Modules = walkBuilding, [][]string{{"app"}}, map[string]*WalkModule{"app": {}}
return c.put(w)
}
func (c *fakeController) Walks(walk string) (bool, []Walk, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.down {
return false, nil, errors.New("down")
}
var out []Walk
for id, w := range c.walks {
if walk == "" || walk == id {
out = append(out, w)
}
}
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
return c.held, out, nil
}
// put is the controller keeping a walk; it answers the walk as `plan-moved` would say it.
func (c *fakeController) put(w Walk) Walk {
c.mu.Lock()
defer c.mu.Unlock()
if old, known := c.walks[w.ID]; known {
w.Revision = old.Revision + 1
} else if w.Revision == 0 {
w.Revision = 1
}
c.walks[w.ID] = w
return w
}
func (c *fakeController) walk(id string) Walk {
c.mu.Lock()
defer c.mu.Unlock()
return c.walks[id]
}
type fakeForge struct {
mu sync.Mutex
notes map[string][]string
views map[string]string
statuses map[string]string
down bool
// viewErr is what View answers instead, when set: a pull request the forge no longer holds.
viewErr error
viewCalls int
// What Statuses reads: the pull requests by owner/repo#number, the base's required statuses, whether the
// protection cannot be read, and what it was asked.
pulls map[string]ForgePull
required []string
protectionDown bool
asked []string
// contains answers gitea_contains by "<commit> in <in>"; a pair it does not hold is a forge that cannot say.
contains map[string]bool
containsDown bool
containsAsks []string
// open is what OpenPulls answers; openDown a forge that cannot say.
open []OpenPull
openDown bool
// merges, comments and unlabels are what was asked of the forge for ready marks (novox/hq issue 722).
merges, comments, unlabels []string
refuse error
}
func (f *fakeForge) OpenPulls() ([]OpenPull, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.openDown {
return nil, errors.New("the forge is away")
}
return append([]OpenPull(nil), f.open...), nil
}
func newFakeForge() *fakeForge {
return &fakeForge{notes: map[string][]string{}, views: map[string]string{}, statuses: map[string]string{},
pulls: map[string]ForgePull{}, contains: map[string]bool{}}
}
func (f *fakeForge) Note(owner, repo, commit, line string) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.down {
return errors.New("the forge is away")
}
key := owner + "/" + repo + "@" + commit
for _, l := range f.notes[key] {
if l == line {
return nil // appended once
}
}
f.notes[key] = append(f.notes[key], line)
return nil
}
func (f *fakeForge) View(owner, repo string, number int, body string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.viewCalls++
if f.viewErr != nil {
return f.viewErr
}
if f.down {
return errors.New("the forge is away")
}
f.views[fmt.Sprintf("%s/%s#%d", owner, repo, number)] = body
return nil
}
func (f *fakeForge) Status(owner, repo, commit, context, state, description, target string) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.down {
return errors.New("the forge is away")
}
f.statuses[owner+"/"+repo+"@"+commit+" "+context] = state + " " + description + " → " + target
return nil
}
// Statuses answers what the fake was told: a pull request's head from pulls, the statuses set on the commit
// (those Status set too, as the forge would keep them), and the base's required statuses from required.
func (f *fakeForge) Statuses(owner, repo string, number int, commit, base string) (*ForgeChecks, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.down {
return nil, errors.New("the forge is away")
}
f.asked = append(f.asked, fmt.Sprintf("%s/%s#%d@%s base=%s", owner, repo, number, commit, base))
out := &ForgeChecks{}
if number > 0 {
p, ok := f.pulls[fmt.Sprintf("%s/%s#%d", owner, repo, number)]
if !ok {
return nil, fmt.Errorf("no pull request %d", number)
}
out.Pull = &p
if commit == "" {
commit = p.HeadSHA
}
if base == "" {
base = p.Base
}
}
for key, said := range f.statuses {
at, context, _ := strings.Cut(key, " ")
if !strings.HasPrefix(at, owner+"/"+repo+"@") || !sameCommit(strings.TrimPrefix(at, owner+"/"+repo+"@"), commit) {
continue
}
if len(strings.TrimPrefix(at, owner+"/"+repo+"@")) > len(commit) {
commit = strings.TrimPrefix(at, owner+"/"+repo+"@")
}
state, rest, _ := strings.Cut(said, " ")
description, link, _ := strings.Cut(rest, " → ")
out.Statuses = append(out.Statuses, ForgeStatus{Context: context, State: state, Description: description,
TargetURL: link, UpdatedAt: "2026-10-06T12:00:00Z", Creator: "mesh-admin"})
}
sort.Slice(out.Statuses, func(i, j int) bool { return out.Statuses[i].Context < out.Statuses[j].Context })
out.Commit = commit
if base == "" {
base = "main"
}
out.Merge = ForgeMerge{Branch: base, Protected: f.required != nil, Required: f.required}
if f.protectionDown {
out.Merge.Says = "the protection of " + base + " could not be read"
return out, nil
}
ok := true
for _, r := range f.required {
state := "missing"
for _, st := range out.Statuses {
if st.Context == r {
state = st.State
}
}
if state != "success" {
ok = false
out.Merge.Blocking = append(out.Merge.Blocking, struct {
Context string `json:"context"`
State string `json:"state"`
}{r, state})
}
}
out.Merge.Mergeable = &ok
out.Merge.Says = fmt.Sprintf("%s requires %v", base, f.required)
return out, nil
}
// Contains answers what the fake was told of two commits.
func (f *fakeForge) Contains(owner, repo, commit, in string) (bool, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.containsAsks = append(f.containsAsks, commit+" in "+in)
if f.containsDown || f.down {
return false, errors.New("the forge is away")
}
yes, said := f.contains[commit+" in "+in]
if !said {
return false, fmt.Errorf("the forge has no commit %s or %s", commit, in)
}
return yes, nil
}
func (f *fakeForge) notesOn(commit string) []string {
f.mu.Lock()
defer f.mu.Unlock()
for key, lines := range f.notes {
if strings.HasSuffix(key, "@"+commit) {
return append([]string(nil), lines...)
}
}
return nil
}
// world is a holder over fakes, with a clock a test moves.
type world struct {
t *testing.T
h *Holder
store *memStore
ctl *fakeController
forge *fakeForge
now time.Time
said []map[string]any
mu sync.Mutex
}
func newWorld(t *testing.T) *world {
return newWorldOver(t, newMemStore(), newFakeController(), newFakeForge(), time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC))
}
func newWorldOver(t *testing.T, store *memStore, ctl *fakeController, forge *fakeForge, now time.Time) *world {
w := &world{t: t, store: store, ctl: ctl, forge: forge, now: now}
w.h = &Holder{Store: store, Controller: ctl, Forge: forge,
Now: func() time.Time { w.mu.Lock(); defer w.mu.Unlock(); return w.now },
Logf: func(format string, a ...any) { t.Logf(format, a...) },
Emit: func(event string, body any) error {
w.mu.Lock()
defer w.mu.Unlock()
var m map[string]any
_ = json.Unmarshal(mustJSON(body), &m)
m["event-kind"] = event
w.said = append(w.said, m)
return nil
}}
w.h.init()
if err := w.h.Load(); err != nil {
t.Fatal(err)
}
return w
}
func (w *world) later(d time.Duration) {
w.mu.Lock()
w.now = w.now.Add(d)
w.mu.Unlock()
}
// settleAll ticks and flushes until nothing changes, a few times.
func (w *world) settleAll() {
for range 4 {
w.h.Tick()
w.h.Flush()
}
}
func (w *world) delivery(id string) *Delivery {
w.t.Helper()
w.h.mu.Lock()
defer w.h.mu.Unlock()
d := w.h.deliveries[id]
if d == nil {
w.t.Fatalf("no delivery %s; there are %v", id, w.ids())
}
c := *d
return &c
}
func (w *world) ids() []string {
var out []string
for id := range w.h.deliveries {
out = append(out, id)
}
sort.Strings(out)
return out
}
func (w *world) state(id string) State { return w.delivery(id).State }
// pr is a pull request's head as the forge announces it.
func pr(repo string, number int, sha, branch string, paths ...string) PullEvent {
owner, name, _ := strings.Cut(repo, "/")
return PullEvent{Owner: owner, Repo: name, Number: number, Base: "main", Head: branch, HeadSHA: sha,
HTMLURL: fmt.Sprintf("https://forge.invalid/%s/pulls/%d", repo, number), Paths: paths}
}
// verdict is the controller's `checked` for a head.
func verdict(repo string, number int, sha, gate string) CheckedEvent {
owner, name, _ := strings.Cut(repo, "/")
c := CheckedEvent{Owner: owner, Repo: name, Number: number, Commit: sha, Verdict: gate, Summary: "the gate said " + gate,
ID: "check-of-" + sha, Plan: &DeliveryPlan{Repository: repo, Moved: []string{"app"}, Summary: "builds app → anchor"}}
return c
}
// merged is the forge's `pull.merged`.
func merged(repo string, number int, head, merge string) PullEvent {
p := pr(repo, number, head, "feature")
p.MergeCommit = merge
p.MergedAt = "2026-10-06T12:30:00Z"
return p
}
// aWalk is the controller's walk of a merge commit, waiting for its delivery's word.
func aWalk(id, repo, commit string, waits bool) Walk {
w := Walk{ID: id, Repository: repo, Branch: "main", Commit: commit, Created: time.Date(2026, 10, 6, 12, 31, 0, 0, time.UTC),
State: walkBuilding, Tiers: [][]string{{"app"}}, Modules: map[string]*WalkModule{"app": {}}}
if waits {
w.Delivery = &WalkDelivery{Awaits: "mesh-delivery"}
}
return w
}
// Merge, Comment and Unlabel record what was asked (novox/hq issue 722); Merge answers refuse when set.
func (f *fakeForge) Merge(owner, repo string, number int, head string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.merges = append(f.merges, fmt.Sprintf("%s/%s#%d@%s", owner, repo, number, head))
return f.refuse
}
func (f *fakeForge) Comment(owner, repo string, number int, body string) error {
f.mu.Lock()
defer f.mu.Unlock()
f.comments = append(f.comments, fmt.Sprintf("%s/%s#%d: %s", owner, repo, number, body))
return nil
}
func (f *fakeForge) Unlabel(owner, repo string, number, label int) error {
f.mu.Lock()
defer f.mu.Unlock()
f.unlabels = append(f.unlabels, fmt.Sprintf("%s/%s#%d-%d", owner, repo, number, label))
return nil
}