Merge remote-tracking branch 'origin/main' into fix/451-a-recheck-is-the-step
mesh/merge-gate pass: builds mesh-controller, route-proxy → ace, novox; no bus step; every machine composes with the change as it did without (4 of 4 compo…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
mesh/delivery-group group fix/451-a-recheck-is-the-step delivered: every member is delivered

This commit is contained in:
2026-10-12 13:56:52 +02:00
13 changed files with 987 additions and 60 deletions
+232 -17
View File
@@ -37,6 +37,11 @@ import (
// own commit, the newest first, one walk each, until one is delivered; the older ones are then answered
// by that one.
//
// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, ADR 0306 decisions
// 2 and 3, issue 525): it joins a batch it is related to, and one that is related to none and whose delivery
// group has no member still to merge is a batch whose window is closed at once (batchToJoin). Walks stay one at
// a time: such a batch is cut when no walk is open, and queued behind the one that is otherwise.
//
// A merge on the controller's own path (a module whose walk waits for nobody's word) never shares a batch with
// one that waits for mesh-delivery's: each kind has a batch of its own, an own-path batch is cut first and folds
// no waiting catalogue walk, and a catalogue walk let go while another walk runs starts once it ended — so no
@@ -173,6 +178,12 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
fmt.Printf("%s merged into %s (%.8s); nothing the mesh holds reads it\n", repository, m.Base, m.Commit)
return nil
}
// **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk.
own := ownPath(touches)
waits, err := groupWait(ctx, inv, m, deliverySeatHeld(entries), own)
if err != nil {
return notNow(err)
}
release, err := inv.HoldPlans(ctx, true)
if err != nil {
return notNow(err)
@@ -198,7 +209,6 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
// **A merge on the controller's own path never shares a batch with one that waits for the delivery's
// word** (decided during the build, 2026-10-10): batched together, the catalogue's deliveries would start
// with the controller's and skip their turn. Each kind has a batch of its own.
own := ownPath(touches)
if p, ok, err := answeredByALaterMerge(ctx, inv, heard, touches, own, deliverySeatHeld(entries)); err != nil {
return notNow(err)
} else if ok {
@@ -218,7 +228,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
return cutBatchesHeld(ctx, f.open, now)
}
batch, err := theOpenBatch(ctx, inv, heard, now, own)
batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own, waits)
if err != nil {
return notNow(err)
}
@@ -229,7 +239,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
if err := keepBatch(ctx, inv, &batch, now, ""); err != nil {
return notNow(err)
}
fmt.Printf("%s merged into %s (%.8s); it joins %s — %s\n", repository, m.Base, m.Commit, batch.ID,
fmt.Printf("%s merged into %s (%.8s); %s, it joins %s — %s\n", repository, m.Base, m.Commit, why, batch.ID,
batchWords(batch, now))
return cutBatchesHeld(ctx, f.open, now)
}
@@ -320,22 +330,211 @@ func ownPath(touches []inventory.Entry) bool {
return false
}
// theOpenBatch is the batch a merge heard now joins: the one of its kind not yet cut, or a new one.
func theOpenBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own bool) (inventory.Plan, error) {
// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, as ADR 0306
// decisions 2 and 3 read it). batchToJoin is the batch a merge heard now joins, and why in words:
//
// - an assembling or queued batch of its kind it is related to (related.go): it joins it and waits for the
// window, which its coming restarts. Related to two or more, the newer are folded into the oldest, which
// answers all their merges: a merge never waits in two windows, and what it shares with each walks in one.
// - none, and its delivery group has a member still to merge, or mesh-delivery could not say (groupWait, waits
// names why): a new batch, which waits for the window, beside the ones it shares nothing with; its words say
// why.
// - none, and no member of its group is still to merge: a new batch whose window is closed at once. Cut at
// once means only that: walks stay one at a time (ADR 0276 decision 4), so with a walk open the batch is
// queued behind it as any closed batch is.
func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory.BatchedMerge, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, now time.Time, own bool, waits string) (inventory.Plan, string, error) {
edges, err := inv.Dependencies(ctx)
if err != nil {
return inventory.Plan{}, "", err
}
joins, whys, _, err := relatedBatches(ctx, inv, []inventory.BatchedMerge{heard}, own, "", entries, read, edges)
if err != nil {
return inventory.Plan{}, "", err
}
if len(joins) > 0 {
into := joins[0]
why := "related to " + into.ID + " (" + whys[0] + ")"
if len(joins) > 1 {
plans, moves, err := foldInto(ctx, inv, &into, joins[1:])
if err != nil {
return inventory.Plan{}, "", err
}
if err := inv.SavePlansMoving(ctx, plans, moves); err != nil {
return inventory.Plan{}, "", err
}
var folded []string
for i, b := range joins[1:] {
folded = append(folded, b.ID+" ("+whys[i+1]+")")
}
why += ", and to " + strings.Join(folded, ", ") + ", folded into it"
}
into.Delivery.Batch.CutAtOnce = false
return into, why, nil
}
if waits != "" {
// Its own window, beside the batches it shares nothing with: waiting is for its group's members, which
// are related to it by their head and join it (ADR 0306 decision 2).
b, err := newBatch(ctx, inv, heard, now, own, false, waits)
return b, waits + ", it waits for the window in a batch of its own", err
}
b, err := newBatch(ctx, inv, heard, now, own, true, "")
return b, "it shares nothing with a batch assembling and no member of a group is still to merge: it is cut at once", err
}
// relatedBatches are the batches not yet cut of a kind, oldest first, parted into those related to merges (with
// the first reason of each) and the others. Each batch is judged by what its cut would move (ADR 0306 decision
// 3): its merges' moves as they were kept and as they read now — the catalogue may have moved since a merge was
// heard — widened to their dependents, and, for a catalogue batch, every walk waiting for its word, which the
// cut folds in (cutBatch's carry). A merge kept with no moves leaves its batch unknown, related to everything.
// except is a walk not to count as folded: the one being stopped.
func relatedBatches(ctx context.Context, inv *inventory.Inventory, merges []inventory.BatchedMerge, own bool,
except string, entries []inventory.Entry, read map[string][]inventory.ReadRepository,
edges []inventory.Edge) (joins []inventory.Plan, whys []string, others []inventory.Plan, err error) {
batches, err := inv.Batches(ctx)
if err != nil {
return inventory.Plan{}, err
return nil, nil, nil, err
}
var kind []inventory.Plan
for _, b := range batches {
if b.OwnPath() == own {
return b, nil
kind = append(kind, b)
}
}
if len(kind) == 0 {
return nil, nil, nil, nil
}
carry := newFootprint()
if !own {
// An own-path batch folds no waiting walk (cutBatchesHeld); a catalogue batch may fold every one.
plans, err := inv.OpenPlans(ctx)
if err != nil {
return nil, nil, nil, err
}
for _, p := range plans {
if p.Release != nil || !p.Waiting() || p.ID == except {
continue
}
kept, err := inv.MergesOf(ctx, p.ID)
if err != nil {
return nil, nil, nil, err
}
carry = unionFootprint(carry, footprintOfWalk(p, kept))
}
}
mine := footprintAsCut(merges, entries, read, edges)
for _, b := range kind {
held, err := inv.MergesOf(ctx, b.ID)
if err != nil {
return nil, nil, nil, err
}
if b.Delivery == nil {
b.Delivery = &inventory.PlanDelivery{}
}
if b.Delivery.Batch == nil {
b.Delivery.Batch = &inventory.PlanBatch{Own: own}
}
if ok, why := related(mine, unionFootprint(footprintAsCut(held, entries, read, edges), carry), edges); ok {
joins, whys = append(joins, b), append(whys, why)
} else {
others = append(others, b)
}
}
return joins, whys, others, nil
}
// footprintAsCut is a batch's footprint as a cut now would read it: footprintOfBatch, widened by the modules each
// merge moves as the catalogue reads now (planBatch judges each merge again at the cut, so what it moves may
// have changed since it was heard). Both are kept: when in doubt, related.
func footprintAsCut(merges []inventory.BatchedMerge, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, edges []inventory.Edge) footprint {
f := footprintOfBatch(merges, edges)
var moved []string
for _, m := range merges {
s := announcedOf(m).SourceMoved
if s.Owner == "" {
continue
}
s.MergedAt = "" // news, whatever a cut marked seen since (as planBatch reads it)
for _, e := range wouldMove(s, entries, read) {
moved = append(moved, e.Manifest.Module)
}
}
for _, name := range reachableFrom(moved, edges) {
f.modules[name] = true
}
return f
}
// unionFootprint is what two changes reach together.
func unionFootprint(a, b footprint) footprint {
f := newFootprint()
for _, from := range []footprint{a, b} {
for k := range from.repos {
f.repos[k] = true
}
for k := range from.modules {
f.modules[k] = true
}
for k := range from.heads {
f.heads[k] = true
}
}
f.unknown = a.unknown
if f.unknown == "" {
f.unknown = b.unknown
}
return f
}
// foldInto folds batches into the oldest one related to a merge: the plans to write (into first, so no merge
// names a plan the store does not hold) and the merges to move, for one SavePlansMoving. The folded batches name
// the batch that took them over; a stop's closed window and its withheld modules (ADR 0299) go with their merges.
func foldInto(ctx context.Context, inv *inventory.Inventory, into *inventory.Plan, from []inventory.Plan) ([]*inventory.Plan,
[]inventory.MergeMove, error) {
if into.Delivery == nil {
into.Delivery = &inventory.PlanDelivery{}
}
if into.Delivery.Batch == nil {
into.Delivery.Batch = &inventory.PlanBatch{Own: into.OwnPath()}
}
plans := []*inventory.Plan{into}
var moves []inventory.MergeMove
for i := range from {
f := from[i]
merges, err := inv.MergesOf(ctx, f.ID)
if err != nil {
return nil, nil, err
}
for _, m := range merges {
moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: into.ID, Alone: m.Alone})
}
if f.Delivery == nil {
f.Delivery = &inventory.PlanDelivery{}
}
if w := f.Delivery.Batch; w != nil {
into.Delivery.Batch.Closed = into.Delivery.Batch.Closed || w.Closed
into.Delivery.Batch.Withheld = union(into.Delivery.Batch.Withheld, w.Withheld)
sort.Strings(into.Delivery.Batch.Withheld)
}
f.State = inventory.PlanSuperseded
f.Delivery.Batch = nil
f.Delivery.TakenOverBy = into.ID
f.Note = fmt.Sprintf("folded into %s before it was cut: a merge related to both joined it, and %s answers "+
"its merges", into.ID, into.ID)
plans = append(plans, &f)
}
return plans, moves, nil
}
// newBatch is a new batch of a kind, kept before its first merge names it: a merge never names a plan the store
// does not hold. cutAtOnce closes its window at once; waits says why it waits for its group, when it does.
func newBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own,
cutAtOnce bool, waits string) (inventory.Plan, error) {
b := inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch,
Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling,
Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{},
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}
// Kept before its first merge names it: a merge never names a plan the store does not hold.
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own, CutAtOnce: cutAtOnce, Waits: waits}}}
if err := inv.SavePlan(ctx, &b); err != nil {
return inventory.Plan{}, err
}
@@ -367,13 +566,16 @@ func keepBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan,
}
closes, latest := windowOf(merges, b.Created, window, atMost)
b.Delivery.Merges = named
var closed bool
var closed, cutAtOnce bool
var withheld []string
var waits string
if was := b.Delivery.Batch; was != nil {
closed, withheld = was.Closed, was.Withheld // a stop's, which no look takes back (novox/hq ADR 0299)
cutAtOnce = was.CutAtOnce // the merge's, until another joins it (ADR 0304 decision 1)
waits = was.Waits // its group's, as said when it was made (ADR 0306 decision 2)
}
b.Delivery.Batch = &inventory.PlanBatch{ClosesAt: closes, AtMost: latest, Behind: behind, Own: b.OwnPath(),
Closed: closed, Withheld: withheld}
Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits}
b.State = inventory.PlanAssembling
if behind != "" && windowClosed(*b, now) {
b.State = inventory.PlanQueued
@@ -402,13 +604,14 @@ func windowOf(merges []inventory.BatchedMerge, created time.Time, window, atMost
return last.Add(window), created.Add(atMost)
}
// windowClosed says a batch's window is closed: no merge for the window's length, or its maximum reached.
// windowClosed says a batch's window is closed: closed by a stop, closed at once for a merge that shares
// nothing (ADR 0304 decision 1), no merge for the window's length, or its maximum reached.
func windowClosed(b inventory.Plan, now time.Time) bool {
if b.Delivery == nil || b.Delivery.Batch == nil {
return true
}
w := b.Delivery.Batch
return w.Closed || !now.Before(w.ClosesAt) || !now.Before(w.AtMost)
return w.Closed || w.CutAtOnce || !now.Before(w.ClosesAt) || !now.Before(w.AtMost)
}
// carriedOf is a set of merges as a walk carries them: the latest merge of each repository's branch, and
@@ -617,8 +820,9 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error {
return err
}
// The oldest batch whose window closed is cut, an own-path one before a catalogue one (it goes first, and
// the catalogue batch queues behind it rather than being cut and overtaken); one of each kind at most is
// open (theOpenBatch).
// the catalogue batch queues behind it rather than being cut and overtaken). Several of a kind may be open
// since a merge that shares nothing is cut at once (ADR 0304 decision 1): walks stay one at a time, so one
// is cut per look, and once its walk started the others are queued behind it.
sort.SliceStable(batches, func(i, j int) bool { return batches[i].OwnPath() && !batches[j].OwnPath() })
for i := range batches {
b := &batches[i]
@@ -886,6 +1090,9 @@ func walkTimesAtCut(batch, walk inventory.Plan, entries []inventory.Entry, now t
if !w.AtMost.IsZero() && w.AtMost.Before(closed) {
closed = w.AtMost
}
if w.CutAtOnce && !batch.Created.IsZero() {
closed = batch.Created // closed when it was made (ADR 0304 decision 1)
}
if !closed.IsZero() {
if closed.After(now) {
closed = now
@@ -1142,6 +1349,10 @@ func batchWords(b inventory.Plan, now time.Time) string {
if b.State == inventory.PlanQueued {
return fmt.Sprintf("queued behind %s; grouped: %s; plan not yet calculated", w.Behind, grouped)
}
if w.CutAtOnce {
return fmt.Sprintf("cut at once, sharing nothing with a batch assembling; grouped: %s; plan not yet calculated",
grouped)
}
left := w.ClosesAt.Sub(now)
if w.AtMost.Before(w.ClosesAt) {
left = w.AtMost.Sub(now)
@@ -1149,8 +1360,12 @@ func batchWords(b inventory.Plan, now time.Time) string {
if left < 0 {
left = 0
}
return fmt.Sprintf("assembling: %s left (at the latest %s); grouped: %s; plan not yet calculated",
secondsWords(left), w.AtMost.Local().Format("15:04:05"), grouped)
waits := ""
if w.Waits != "" {
waits = " — " + w.Waits
}
return fmt.Sprintf("assembling: %s left (at the latest %s)%s; grouped: %s; plan not yet calculated",
secondsWords(left), w.AtMost.Local().Format("15:04:05"), waits, grouped)
}
// groupedWords is a batch's commits, each with the earlier merges it answers.
+76 -12
View File
@@ -34,6 +34,7 @@ func windowed(t *testing.T) *stores {
t.Fatal(err)
}
t.Setenv(mergeWindowFileVar, settings)
noGroups(t)
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-controller", Version: "1"},
inventory.Source{Repository: "novox/mesh-controller", Seat: "git", Ref: "main", BuiltFrom: "c0",
Head: "c0"}); err != nil {
@@ -58,19 +59,70 @@ func windowed(t *testing.T) *stores {
// t0 is the moment a test's clock starts: merges are dated by it, and heard after it.
var t0 = time.Now().UTC().Truncate(time.Second)
// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment.
// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment, from a pull
// request's head branch of its own, as the forge announces every merge. Whether its head names a delivery group
// with a member still to merge is mesh-delivery's word, which windowed stubs as no group at all (noGroups); a test
// whose merges wait for the window as a group's says so (groupsWaiting).
func catalogueMerge(commit, module string, made time.Time) link.SourceMoved {
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit,
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Head: "pr/" + commit, Commit: commit,
MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"modules/" + module + "/module.json"},
ModuleDirs: []string{"modules/" + module}, ModuleDirsSaid: true}
}
// repoMerge is a merge of a repository of one module's own.
// repoMerge is a merge of a repository of one module's own, from a head branch of its own (catalogueMerge).
func repoMerge(repo, commit string, made time.Time) link.SourceMoved {
return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit,
return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Head: "pr/" + commit, Commit: commit,
MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"main.go"}}
}
// noGroups stubs mesh-delivery's word on groups for a test: no head names a group.
func noGroups(t *testing.T) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return nil, false, nil }
t.Cleanup(func() { deliveryGroupOf = was })
}
// groupsWaiting stubs mesh-delivery's word on groups for a test: each head named is a group with a member of another
// repository whose pull request is open (novox/elsewhere, ready), so a merge from it waits for the window (ADR 0306
// decision 2); every other head names no group.
func groupsWaiting(t *testing.T, heads ...string) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(_ context.Context, _ bool, head string) ([]groupMember, bool, error) {
if slices.Contains(heads, head) {
return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil
}
return nil, false, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
}
// everyHeadWaits stubs mesh-delivery's word on groups for a test whose merges each wait for the window as they did
// before novox/hq ADR 0304: every head is a group with a member of another repository still to merge.
func everyHeadWaits(t *testing.T) {
t.Helper()
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
}
// fromHead is a merge from a head branch a test names: merges from one head are members of one delivery group,
// related by it (novox/hq ADR 0304 decision 1), so they share a batch whatever else they share.
func fromHead(m link.SourceMoved, head string) link.SourceMoved {
m.Head = head
return m
}
// lone is a merge that names no head: known to be no delivery group's, it waits for no window unless it is
// related to a batch assembling (ADR 0304 decision 1).
func lone(m link.SourceMoved) link.SourceMoved {
m.Head = ""
return m
}
// hear is a merge heard at a moment.
func hear(t *testing.T, open *stores, m link.SourceMoved, now time.Time) {
t.Helper()
@@ -108,6 +160,7 @@ func walks(t *testing.T, open *stores) (walks, batches []inventory.Plan) {
// later commit, which names both merges — the earlier carried by the later.
func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
claude := catalogueMerge("553b7191claude", "app", t0)
dunst := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second))
@@ -152,9 +205,10 @@ func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) {
// Merges of three repositories in one window are one walk, one commit for each.
func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/three")
asksRecorded(t)
for i, r := range []string{"one", "two", "three"} {
hear(t, open, repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)),
hear(t, open, fromHead(repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), "feat/three"),
t0.Add(time.Duration(i)*20*time.Second+time.Second))
}
cutAt(t, open, t0.Add(41*time.Second+90*time.Second))
@@ -187,9 +241,10 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
// closes at most ten minutes after its first merge however busy.
func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/both", "feat/busy")
asksWithPaths(t)
hear(t, open, repoMerge("one", "c1", t0), t0)
hear(t, open, repoMerge("two", "c2", t0.Add(80*time.Second)), t0.Add(80*time.Second))
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(80*time.Second)), "feat/both"), t0.Add(80*time.Second))
cutAt(t, open, t0.Add(100*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatal("the window closed 90 seconds after its first merge, not after its last")
@@ -200,12 +255,13 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
}
busy := windowed(t)
groupsWaiting(t, "feat/both", "feat/busy")
asksWithPaths(t)
var last time.Time
for i := 0; i < 12; i++ {
last = t0.Add(time.Duration(i) * time.Minute)
repo := []string{"one", "two", "three"}[i%3]
hear(t, busy, repoMerge(repo, "c"+string(rune('a'+i)), last), last)
hear(t, busy, fromHead(repoMerge(repo, "c"+string(rune('a'+i)), last), "feat/busy"), last)
if ws, _ := walks(t, busy); len(ws) > 0 {
if i != 10 {
t.Fatalf("a busy window closed at merge %d (minute %d), not at its maximum", i, i)
@@ -223,6 +279,7 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
// what is grouped, and that the plan is not yet calculated. It says so on the bus as plan-moved, too.
func TestTheAssemblingBatchIsShown(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/553b7191claude")
asksWithPaths(t)
var said []inventory.Plan
was := inventory.PlanSaved
@@ -235,7 +292,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) {
dunst.Number, dunst.Title = 175, "dunst: the font the operator chose"
hear(t, open, claude, now.Add(-19*time.Second))
hear(t, open, dunst, now.Add(-time.Second))
hear(t, open, repoMerge("one", "a6bc0931one", now), now)
hear(t, open, fromHead(repoMerge("one", "a6bc0931one", now), dunst.Head), now)
out := captured(t, func() error { return plansCommand(t.Context(), nil) })
lines := strings.Split(out, "\n")
first := lines[0]
@@ -360,6 +417,7 @@ func TestAWaitingWalkIsFoldedIntoTheNextBatch(t *testing.T) {
// moves — named at once on a walk done — and joins the next batch when it does not.
func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
later := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second))
@@ -391,6 +449,7 @@ func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) {
// first; the first delivered answers the older ones, and the search stops.
func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
ctx := t.Context()
// Made before the cut marks their modules seen, as merges are: walked again, they are news all the same.
@@ -459,6 +518,7 @@ func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) {
func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) {
for _, grouped := range []bool{true, false} {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
for _, m := range []string{"mesh-host"} {
@@ -496,9 +556,10 @@ func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) {
// store, a merge handed over again is not doubled, and the batch is cut when its window closes.
func TestABatchSurvivesARestart(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/both")
asksWithPaths(t)
hear(t, open, repoMerge("one", "c1", t0), t0)
hear(t, open, repoMerge("two", "c2", t0.Add(30*time.Second)), t0.Add(30*time.Second))
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(30*time.Second)), "feat/both"), t0.Add(30*time.Second))
// A new controller: nothing of the first one's but the store.
again, err := openStores(t.Context())
@@ -506,7 +567,7 @@ func TestABatchSurvivesARestart(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(again.Close)
hear(t, again, repoMerge("one", "c1", t0), t0.Add(60*time.Second)) // the bus hands it over again
hear(t, again, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0.Add(60*time.Second)) // the bus hands it over again
cutAt(t, again, t0.Add(119*time.Second))
if ws, bs := walks(t, again); len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the restarted controller's batch: %d walk(s), %+v", len(ws), bs)
@@ -563,6 +624,7 @@ func TestAWaitingWalkNamesTheMergesItAnswers(t *testing.T) {
// nothing, and the merge joins the next batch.
func TestALateMergeOfAOneModuleRepositoryIsNamed(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
c2 := repoMerge("one", "c2", t0.Add(-50*time.Second))
@@ -716,6 +778,7 @@ func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) {
// still waits for the word, so no catalogue delivery skips its turn.
func TestAMergeOnTheControllersPathNeverSharesABatch(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asked := asksWithPaths(t)
ctx := t.Context()
if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1",
@@ -860,6 +923,7 @@ func sameMerge(a, b inventory.PlanMerge) bool {
// its delivery would skip its turn. It joins the next catalogue batch.
func TestALateCatalogueMergeIsNotAnsweredByAnOwnPathWalk(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
asksWithPaths(t)
ctx := t.Context()
for _, m := range []struct {
+4
View File
@@ -48,6 +48,10 @@ func gatherBatches(ctx context.Context, inv *inventory.Inventory, now time.Time)
if f.atMost.Before(f.closed) {
f.closed = f.atMost
}
if w.Batch.CutAtOnce && !b.Created.IsZero() {
// Its window closed when it was made, which is its latest too (ADR 0304 decision 1).
f.atMost, f.closed = b.Created, b.Created
}
}
if f.behind != "" {
if walk, err := inv.PlanByID(ctx, f.behind); err == nil {
+27 -19
View File
@@ -241,7 +241,7 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why
}
return p, said, inv.SavePlan(ctx, &p)
}
target, err := batchToWalkAgain(ctx, inv, again, now)
target, folded, foldMoves, err := batchToWalkAgain(ctx, inv, p.ID, again, now)
if err != nil {
return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err)
}
@@ -262,8 +262,10 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why
if len(said.Kept) > 0 {
p.Note += "; " + keptWords(said.Kept)
}
// The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7).
if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&target, &p}, moves); err != nil {
// The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7) — the
// batches folded into it too.
if err := inv.SavePlansMoving(ctx, append([]*inventory.Plan{&target, &p}, folded...),
append(moves, foldMoves...)); err != nil {
return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err)
}
// What the batch holds and names now: written again by the cutter's next look should this fail.
@@ -318,36 +320,42 @@ func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Pla
return nil
}
// batchToWalkAgain is the batch a stopped walk's merges join: the open batch of their kind, or a new one, not yet
// written. The kind is read from what the merges moved.
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, again []inventory.BatchedMerge,
now time.Time) (inventory.Plan, error) {
// batchToWalkAgain is the batch a stopped walk's merges join (novox/hq ADR 0299): a batch of their kind not yet cut
// that they are related to (ADR 0304 decision 1, ADR 0306 decision 3), or a new one, not yet written. Related to
// two or more, the newer are folded into the oldest: the plans and merge moves that fold them, to be written in
// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds.
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge,
now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) {
own := false
for _, n := range movesOf(again) {
if _, on := onTheControllersPath[n]; on {
own = true
}
}
batches, err := inv.Batches(ctx)
entries, read, edges, err := theGraph(ctx, inv)
if err != nil {
return inventory.Plan{}, err
return inventory.Plan{}, nil, nil, err
}
for _, b := range batches {
if b.OwnPath() == own {
if b.Delivery == nil {
b.Delivery = &inventory.PlanDelivery{}
}
if b.Delivery.Batch == nil {
b.Delivery.Batch = &inventory.PlanBatch{Own: own}
}
return b, nil
joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges)
if err != nil {
return inventory.Plan{}, nil, nil, err
}
if len(joins) > 0 {
into := joins[0]
if len(joins) == 1 {
return into, nil, nil, nil
}
plans, moves, err := foldInto(ctx, inv, &into, joins[1:])
if err != nil {
return inventory.Plan{}, nil, nil, err
}
return into, plans[1:], moves, nil
}
first := again[0]
return inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch,
Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling,
Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{},
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil
Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil, nil, nil
}
// leaveOut splits a walk's merges by the merges named to leave out, each `<repository>@<commit>` (a commit's
+158
View File
@@ -0,0 +1,158 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// **A delivery group still assembles** (novox/hq ADR 0306 decision 2): a merge whose delivery group has a member
// not yet merged waits for the merge window, whether or not a batch is assembling, so the group's members walk in
// one batch, in their order (ADR 0276 decision 5). Which groups exist and where each member stands is
// mesh-delivery's record (ADR 0239): a group is the pull requests of two repositories or more sharing a head
// branch name, which is its id. The controller asks it, through its `groups` verb, when a merge from a head is
// heard; a member whose pull request is open (proposed, checked, ready or rejected) is still to merge. When
// mesh-delivery does not answer within groupAskWithin, or answers something unreadable, the merge waits: the safe
// error is to wait, and its batch says why.
// groupAskWithin is how long mesh-delivery is given to say a merge's group, while the merge is being heard.
var groupAskWithin = 3 * time.Second
// openPullStates are the states of a delivery whose pull request is open and not merged (mesh-delivery's table):
// a member in one of them is still to merge. Closed unmerged is stopped, final, and waits for nothing.
var openPullStates = map[string]bool{"proposed": true, "checked": true, "ready": true, "rejected": true}
// groupMember is one member of a delivery group as mesh-delivery's `groups` says it: `<owner/repo>@<commit> (state)`.
type groupMember struct {
Repository, State string
}
// deliveryGroupOf is what mesh-delivery says of the group a head names: its members, and whether the head names a
// group at all; an error when it cannot be asked or did not answer. A seam a test replaces.
var deliveryGroupOf = func(ctx context.Context, held bool, head string) ([]groupMember, bool, error) {
if !held {
return nil, false, fmt.Errorf("no holder of %s is on record to say it", catalogue.DeliverySeat)
}
serving := servingBus.Load()
if serving == nil {
return nil, false, errors.New("this controller is not on the bus")
}
asking, cancel := context.WithTimeout(ctx, groupAskWithin)
defer cancel()
raw, err := askDeliveryOwner(asking, serving.Conn(), "groups", map[string]any{})
if err != nil {
if errors.Is(err, link.ErrNothingServes) {
return nil, false, fmt.Errorf("nothing serves %s.groups", catalogue.DeliverySeat)
}
return nil, false, err
}
var lines []struct {
ID string `json:"id"`
Members []string `json:"members"`
}
if err := json.Unmarshal(raw, &lines); err != nil {
return nil, false, fmt.Errorf("%s.groups answered something unreadable: %w", catalogue.DeliverySeat, err)
}
for _, l := range lines {
if l.ID != head {
continue
}
var members []groupMember
for _, s := range l.Members {
members = append(members, memberOf(s))
}
return members, true, nil
}
return nil, false, nil
}
// memberOf reads one member as `groups` says it: `novox/mesh-sdk@d41d255f28ad (held)`.
func memberOf(s string) groupMember {
id, state := strings.TrimSpace(s), ""
if i := strings.LastIndex(id, " ("); i >= 0 && strings.HasSuffix(id, ")") {
id, state = id[:i], id[i+2:len(id)-1]
}
repository, _, _ := strings.Cut(id, "@")
return groupMember{Repository: repository, State: state}
}
// groupWait says whether a merge heard now waits for the window for its delivery group, and why in words; nothing
// when it names no head, its head names no group, or every other member is merged or closed. own is the merge's
// kind (ownPath).
//
// A member of another repository whose merge this controller already keeps in a batch not yet cut of the same
// kind is merged, whatever mesh-delivery says yet (it marks a member merged once its walk opened): the merge joins
// that batch by their head. One kept in a batch of the other kind keeps the merge waiting, whatever mesh-delivery
// says (review of #250): kinds never share a batch, so the merge's batch is a new one, and cut at once it would
// walk before that member's batch is cut and break the group's order (ADR 0276 decision 5). Waiting, the two
// batches are cut in the order their windows close, an own-path one first at one look, as cutBatchesHeld cuts
// every pair of kinds.
//
// Asked before the plans are held: a slow answer holds up no walk.
func groupWait(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, held, own bool) (string, error) {
if m.Head == "" {
return "", nil
}
same, other, err := keptFromHead(ctx, inv, m.Head, own)
if err != nil {
return "", err
}
if len(other) > 0 {
return "group " + m.Head + " has a member waiting in a batch of the other kind: " + strings.Join(other, ", "), nil
}
members, isGroup, err := deliveryGroupOf(ctx, held, m.Head)
if err != nil {
return fmt.Sprintf("%s did not say whether group %s has a member still to merge (%v): it waits for the "+
"window, the safe side", catalogue.DeliverySeat, m.Head, err), nil
}
if !isGroup {
return "", nil
}
var open []string
for _, g := range members {
if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || same[strings.ToLower(g.Repository)] ||
!openPullStates[g.State] {
continue
}
open = append(open, g.Repository+" ("+g.State+")")
}
if len(open) == 0 {
return "", nil
}
return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), nil
}
// keptFromHead are the merges from a head that batches not yet cut hold: the repositories (lower case) of those in
// a batch of the kind own says, and, in words, each of the other kind with its batch.
func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string, own bool) (map[string]bool, []string, error) {
batches, err := inv.Batches(ctx)
if err != nil {
return nil, nil, err
}
same := map[string]bool{}
var other []string
for _, b := range batches {
merges, err := inv.MergesOf(ctx, b.ID)
if err != nil {
return nil, nil, err
}
for _, k := range merges {
if announcedOf(k).Head != head {
continue
}
if b.OwnPath() == own {
same[strings.ToLower(k.Repository)] = true
} else {
other = append(other, fmt.Sprintf("%s@%s in %s", k.Repository, short(k.Commit), b.ID))
}
}
}
return same, other, nil
}
+11 -5
View File
@@ -123,11 +123,12 @@ func TestDeliveryStopTakesTheMergesToLeaveOut(t *testing.T) {
// that builds the other repository's module alone.
func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -181,12 +182,13 @@ func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) {
// ended. A merge the walk does not answer is refused, and nothing ends.
func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
first := catalogueMerge("cat1aaaaaaaa", "app", t0)
later := catalogueMerge("cat2cccccccc", "notes", t0.Add(10*time.Second))
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), first.Head)
hear(t, open, first, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
hear(t, open, later, t0.Add(11*time.Second))
@@ -238,11 +240,12 @@ func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) {
// (review of #232): its window closes, and its walk is at the newer commit, answering both.
func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one1, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -290,11 +293,12 @@ func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) {
// that walked its merges again is refused a retry, saying why.
func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one1, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
@@ -335,11 +339,12 @@ func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) {
// with the others and withholds the stopped merge's modules; with every merge left out, the batch is stopped.
func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second))
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
_, bs := walks(t, open)
@@ -382,6 +387,7 @@ func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) {
// else since it was read refuses all of it, and no merge moves, no plan is written.
func TestSavingPlansAndMovingMergesIsOneAct(t *testing.T) {
open := windowed(t)
everyHeadWaits(t)
endsItsLine(t)
ctx := t.Context()
hear(t, open, catalogueMerge("cat1aaaaaaaa", "app", t0), t0.Add(time.Second))
+398
View File
@@ -0,0 +1,398 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// The tests of novox/hq issue 525's window skip: a merge waits for the merge window only when a batch assembling
// holds a merge related to it, or its delivery group may have a member still to merge (ADR 0304 decision 1, ADR
// 0306 decisions 2 and 3). Walks stay one at a time.
// A merge from no head that shares nothing, with no walk open, is cut into its walk when it is heard: no window.
func TestALoneMergeSharingNothingIsCutAtOnce(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
hear(t, open, lone(repoMerge("one", "c1", t0)), t0)
ws, bs := walks(t, open)
if len(ws) != 1 || len(bs) != 0 || !ws[0].Open() || ws[0].Commit != "c1" {
t.Fatalf("a lone merge sharing nothing: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs))
}
if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(t0) {
t.Fatalf("its window closed at %+v; want when it was heard, %s", ws[0].Times, t0)
}
}
// A merge related to the batch assembling joins it and waits for the window, which its coming restarts: by its
// repository, by a module in common (a dependent of what the batch moves), by an edge of the module graph between
// their modules (one that widens nothing), or by its delivery group's head.
func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) {
ctx := t.Context()
for _, c := range []string{"same repository", "a module in common", "an edge of the graph", "the same group head"} {
t.Run(c, func(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c1")
asksWithPaths(t)
inv := open.inventory
first := repoMerge("one", "c1", t0)
second := lone(repoMerge("two", "c2", t0.Add(10*time.Second)))
why := map[string]string{"same repository": "both change novox/one", "a module in common": "both move two",
"an edge of the graph": "two depends on builder (built-by)",
"the same group head": "both are members of group " + first.Head}[c]
switch c {
case "same repository":
second = lone(repoMerge("one", "c2", t0.Add(10*time.Second)))
case "a module in common":
// two is built on one: a merge of one moves two too.
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "two", Version: "1",
Build: &catalogue.Build{On: []catalogue.BuildsOn{{Arg: "BASE", Module: "one"}}}},
inventory.Source{Repository: "novox/two", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
case "an edge of the graph":
// two is built by the build machine: an edge that orders and widens nothing.
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "builder", Version: "1",
Claims: []catalogue.Claim{{Name: "mesh-build-machine", Scope: catalogue.ScopeMesh}}},
inventory.Source{Repository: "novox/builder", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
first = repoMerge("builder", "c1", t0)
case "the same group head":
second = fromHead(repoMerge("two", "c2", t0.Add(10*time.Second)), first.Head)
}
hear(t, open, first, t0)
said := captured(t, func() error {
return (following{open: open}).hearMerge(ctx, second, t0.Add(10*time.Second))
})
if !strings.Contains(said, "("+why+")") {
t.Fatalf("the merge was not said related by %q: %s", why, said)
}
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the related merge: %d walk(s), batches %+v", len(ws), bs)
}
if bs[0].Delivery.Batch.CutAtOnce {
t.Fatal("a batch a related merge joined is cut at once")
}
cutAt(t, open, t0.Add(99*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatalf("cut before the window the related merge restarted closed: %+v", ws)
}
cutAt(t, open, t0.Add(100*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || len(bs) != 0 || len(ws[0].Delivery.Merges) != 2 {
t.Fatalf("after the window: %d walk(s), %d batch(es)", len(ws), len(bs))
}
})
}
}
// A merge from a head whose delivery group has a member still to merge waits for the window, even with no batch
// assembling, and says why; its sibling joins the same batch by their head (ADR 0306 decision 2). A merge from a
// head that names no group shares nothing and is cut at once, as one from no head is.
func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
groupsWaiting(t, "feat/x")
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0)
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce || windowClosed(bs[0], t0) {
t.Fatalf("a group's merge alone: %d walk(s), batches %+v", len(ws), bs)
}
if words := batchWords(bs[0], t0); !strings.HasPrefix(words, "assembling: 90 s left") ||
!strings.Contains(words, "group feat/x has a member still to merge: novox/elsewhere (ready)") {
t.Fatalf("it reads %q", words)
}
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/x"), t0.Add(5*time.Second))
if _, bs := walks(t, open); len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 {
t.Fatalf("the group's second member did not join the first: %+v", bs)
}
cutAt(t, open, t0.Add(94*time.Second))
if ws, _ := walks(t, open); len(ws) != 0 {
t.Fatalf("cut before its window closed: %+v", ws)
}
cutAt(t, open, t0.Add(95*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || len(bs) != 0 || ws[0].CommitOf("novox/one") != "c1" || ws[0].CommitOf("novox/two") != "c2" {
t.Fatalf("at its window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs))
}
}
// A merge from a head that names no group, sharing nothing, with no walk open, is cut at once.
func TestAHeadedMergeInNoGroupIsCutAtOnce(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
hear(t, open, fromHead(repoMerge("one", "c1", t0), "fix/alone"), t0)
if ws, bs := walks(t, open); len(ws) != 1 || len(bs) != 0 || !ws[0].Open() {
t.Fatalf("a headed merge in no group: %d walk(s), %d batch(es)", len(ws), len(bs))
}
}
// When mesh-delivery cannot say a head's group, the merge waits for the window, the safe side, and its batch says
// why.
func TestAGroupNobodyCanSayWaits(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return nil, false, errors.New("nothing serves mesh-delivery.groups")
}
t.Cleanup(func() { deliveryGroupOf = was })
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0)
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a merge whose group nobody could say: %d walk(s), batches %+v", len(ws), bs)
}
if words := batchWords(bs[0], t0); !strings.Contains(words, "mesh-delivery did not say whether group feat/x has a "+
"member still to merge (nothing serves mesh-delivery.groups): it waits for the window, the safe side") {
t.Fatalf("it reads %q", words)
}
}
// A member counts as still to merge only while its pull request is open, and not once this controller keeps its
// merge: a group whose other members are merged or closed makes nobody wait.
func TestAGroupsMergedOrClosedMembersMakeNobodyWait(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
ctx := t.Context()
members := []groupMember{{Repository: "novox/one", State: "ready"}, {Repository: "novox/two", State: "published"},
{Repository: "novox/three", State: "stopped"}}
was := deliveryGroupOf
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return members, true, nil }
t.Cleanup(func() { deliveryGroupOf = was })
m := fromHead(repoMerge("one", "c1", t0), "feat/x")
if why, err := groupWait(ctx, open.inventory, m, true, false); err != nil || why != "" {
t.Fatalf("the group's other members are merged or closed, yet: %q %v", why, err)
}
members = append(members, groupMember{Repository: "novox/four", State: "checked"})
if why, _ := groupWait(ctx, open.inventory, m, true, false); !strings.Contains(why, "novox/four (checked)") {
t.Fatalf("an open member: %q", why)
}
// novox/four's merge is kept in a batch already: merged, whatever mesh-delivery says yet.
hear(t, open, fromHead(repoMerge("two", "c4", t0), "feat/x"), t0)
members[3].Repository = "novox/two"
if why, _ := groupWait(ctx, open.inventory, m, true, false); why != "" {
t.Fatalf("a member this controller keeps the merge of still waits: %q", why)
}
if got := memberOf("novox/mesh-sdk@d41d255f28ad (held)"); got != (groupMember{Repository: "novox/mesh-sdk", State: "held"}) {
t.Fatalf("a member reads %+v", got)
}
}
// A merge related to two batches assembling folds the newer into the older, which answers every merge of both
// and names nothing twice; the folded one names the batch that took it over.
func TestAMergeRelatedToTwoBatchesFoldsThem(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "feat/a", "feat/b")
asksWithPaths(t)
ctx := t.Context()
hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/a"), t0)
hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/b"), t0.Add(5*time.Second))
_, bs := walks(t, open)
if len(bs) != 2 {
t.Fatalf("two merges sharing nothing: %d batch(es)", len(bs))
}
older, newer := bs[1], bs[0] // newest first
// Of one's repository and from two's head: related to both.
hear(t, open, fromHead(repoMerge("one", "c3", t0.Add(10*time.Second)), "feat/b"), t0.Add(10*time.Second))
_, bs = walks(t, open)
if len(bs) != 1 || bs[0].ID != older.ID || len(bs[0].Delivery.Merges) != 3 {
t.Fatalf("after a merge related to both: batches %+v; want %s answering three merges", bs, older.ID)
}
folded, err := open.inventory.PlanByID(ctx, newer.ID)
if err != nil || folded.State != inventory.PlanSuperseded || folded.Delivery == nil ||
folded.Delivery.TakenOverBy != older.ID {
t.Fatalf("the newer batch reads %s %+v (%v)", folded.State, folded.Delivery, err)
}
if left, _ := open.inventory.MergesOf(ctx, newer.ID); len(left) != 0 {
t.Fatalf("the folded batch still answers %+v", left)
}
cutAt(t, open, t0.Add(100*time.Second))
ws, bs := walks(t, open)
var cut []inventory.Plan
for _, w := range ws {
if w.State != inventory.PlanSuperseded {
cut = append(cut, w)
}
}
if len(cut) != 1 || len(bs) != 0 || cut[0].ID != older.ID || cut[0].CommitOf("novox/one") != "c3" ||
cut[0].CommitOf("novox/two") != "c2" {
t.Fatalf("the folded batch's walk: %+v, %d batch(es)", cut, len(bs))
}
}
// With a walk open, a batch cut at once still waits for it, said queued behind it (walks stay one at a time, ADR
// 0276 decision 4); it is cut once that walk ended, with no window to wait for.
func TestACutAtOnceBatchQueuesBehindAnOpenWalk(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c1")
asksWithPaths(t)
hear(t, open, repoMerge("one", "c1", t0), t0)
cutAt(t, open, t0.Add(90*time.Second))
ws, _ := walks(t, open)
if len(ws) != 1 || !ws[0].Open() {
t.Fatalf("no walk open: %+v", ws)
}
first := ws[0]
heard := t0.Add(100 * time.Second)
hear(t, open, lone(repoMerge("two", "c2", heard)), heard)
ws, bs := walks(t, open)
if len(ws) != 1 || len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != first.ID ||
!bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a lone merge during a walk: %d walk(s), batches %+v", len(ws), bs)
}
if !strings.HasPrefix(batchWords(bs[0], heard), "queued behind "+first.ID) {
t.Fatalf("it reads %q", batchWords(bs[0], heard))
}
first.State = inventory.PlanDone
if err := open.inventory.SavePlan(t.Context(), &first); err != nil {
t.Fatal(err)
}
cutAt(t, open, heard.Add(time.Second))
ws, bs = walks(t, open)
if len(ws) != 2 || len(bs) != 0 || ws[0].Commit != "c2" {
t.Fatalf("the queued batch was not cut once the walk ended: %+v %+v", ws, bs)
}
if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(heard) {
t.Fatalf("its window closed at %+v; want when its merge was heard, %s", ws[0].Times, heard)
}
}
// A stopped walk's other merges are walked again in a batch they are related to, and a batch of their own when
// they share nothing with the one assembling (ADR 0299, ADR 0304 decision 1): never in a batch of another
// change, whose window the stop would close.
func TestAStopWalksTheOthersAgainInABatchOfTheirOwnWhenTheyShareNothing(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/cat1aaaaaaaa")
endsItsLine(t)
asksRecorded(t)
ctx := t.Context()
cat := catalogueMerge("cat1aaaaaaaa", "app", t0)
one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head)
hear(t, open, cat, t0.Add(time.Second))
hear(t, open, one, t0.Add(6*time.Second))
cutAt(t, open, t0.Add(2*time.Minute))
ws, _ := walks(t, open)
walk := ws[0]
hear(t, open, repoMerge("two", "two1cccccccc", t0.Add(3*time.Minute)), t0.Add(3*time.Minute))
_, bs := walks(t, open)
if len(bs) != 1 {
t.Fatalf("the merge heard while the walk ran is in %d batches", len(bs))
}
other := bs[0]
_, said, err := stopWalkWithout(ctx, open.inventory, walk.ID, "mesh-delivery for jochen", "its group stopped",
[]string{"novox/mesh-catalog@" + cat.Commit}, t0.Add(3*time.Minute+10*time.Second))
if err != nil {
t.Fatal(err)
}
if said.WalkedAgainBy == "" || said.WalkedAgainBy == other.ID {
t.Fatalf("walked again by %q; want a batch of its own, not %s", said.WalkedAgainBy, other.ID)
}
b, _ := open.inventory.PlanByID(ctx, other.ID)
if b.Delivery.Batch == nil || b.Delivery.Batch.Closed || len(b.Delivery.Merges) != 1 {
t.Fatalf("the other change's batch reads %+v; want it untouched by the stop", b.Delivery)
}
}
// A catalogue batch's cut folds in every walk waiting for its delivery's word (cutBatch's carry), so what that
// walk moves is the batch's too: a lone merge related to the waiting walk joins the batch assembling and waits for
// its window, rather than being cut at once as sharing nothing with what the batch's own merges move.
func TestALoneMergeRelatedToAWalkTheCutFoldsJoinsTheBatch(t *testing.T) {
open := windowed(t)
groupsWaiting(t, "pr/c3")
asksWithPaths(t)
ctx := t.Context()
if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1",
Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}},
inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/mesh-delivery", Ref: "main",
BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
if _, err := open.inventory.Assign(ctx, "anchor", "mesh-delivery"); err != nil {
t.Fatal(err)
}
hear(t, open, repoMerge("one", "c1", t0), t0)
cutAt(t, open, t0.Add(90*time.Second))
if ws, _ := walks(t, open); len(ws) != 1 || !ws[0].Waiting() {
t.Fatalf("the walk does not wait for its word: %+v", ws)
}
hear(t, open, repoMerge("three", "c3", t0.Add(2*time.Minute)), t0.Add(2*time.Minute))
hear(t, open, lone(repoMerge("one", "c2", t0.Add(2*time.Minute+5*time.Second))), t0.Add(2*time.Minute+5*time.Second))
_, bs := walks(t, open)
if len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 || bs[0].Delivery.Batch.CutAtOnce {
t.Fatalf("a lone merge related to the waiting walk: batches %+v", bs)
}
}
// A delivery group with a member of each kind (review of #250): a merge on the controller's own path and a
// catalogue merge never share a batch, so the later member cannot join the earlier one's. Kept in a batch of the
// other kind, the earlier member keeps the later one waiting, whatever mesh-delivery says yet — never cut at once
// to walk before it. The two batches are then cut in the order their windows close (cutBatchesHeld: the oldest
// closed batch first, an own-path one first at one look, and the other queued behind a walk that started): the
// earlier member's window closes first, so it walks first, in both orders.
func TestAGroupAcrossBothKindsKeepsItsOrder(t *testing.T) {
for _, ownFirst := range []bool{true, false} {
name := "catalogue member first"
if ownFirst {
name = "controller member first"
}
t.Run(name, func(t *testing.T) {
open := windowed(t)
asksWithPaths(t)
ctx := t.Context()
was := deliveryGroupOf
// mesh-delivery still says both members ready: it marks one merged only once its walk opened.
deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) {
return []groupMember{{Repository: "novox/mesh-controller", State: "ready"},
{Repository: "novox/mesh-catalog", State: "ready"}}, true, nil
}
t.Cleanup(func() { deliveryGroupOf = was })
own := fromHead(repoMerge("mesh-controller", "k1", t0), "feat/both-kinds")
cat := fromHead(catalogueMerge("cat1aaaaaaaa", "app", t0), "feat/both-kinds")
first, second := cat, own
if ownFirst {
first, second = own, cat
}
second.MergedAt = t0.Add(10 * time.Second).Format(time.RFC3339Nano)
hear(t, open, first, t0)
hear(t, open, second, t0.Add(10*time.Second))
ws, bs := walks(t, open)
if len(ws) != 0 || len(bs) != 2 {
t.Fatalf("both members heard: %d walk(s), %d batch(es); want two batches, nothing cut", len(ws), len(bs))
}
later := bs[0] // newest first
if later.Delivery.Batch.CutAtOnce || !strings.Contains(later.Delivery.Batch.Waits,
"has a member waiting in a batch of the other kind: novox/"+first.Repo+"@") {
t.Fatalf("the later member's batch reads %+v; want it waiting for the earlier member", later.Delivery.Batch)
}
// The earlier member's window closes first: its batch alone is cut.
cutAt(t, open, t0.Add(90*time.Second))
ws, bs = walks(t, open)
if len(ws) != 1 || !ws[0].Open() || ws[0].CommitOf("novox/"+first.Repo) != first.Commit || len(bs) != 1 ||
bs[0].ID != later.ID {
t.Fatalf("at the earlier window: walks %+v, batches %+v; want the earlier member alone cut", ws, bs)
}
earlier := ws[0]
// The later member's window closes while the earlier walk runs: it queues behind it.
cutAt(t, open, t0.Add(100*time.Second))
if _, bs := walks(t, open); len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != earlier.ID {
t.Fatalf("the later member's batch does not queue behind the earlier walk: %+v", bs)
}
earlier.State = inventory.PlanDone
if err := open.inventory.SavePlan(ctx, &earlier); err != nil {
t.Fatal(err)
}
cutAt(t, open, t0.Add(101*time.Second))
got, err := open.inventory.PlanByID(ctx, later.ID)
if err != nil || got.Batch() || got.CommitOf("novox/"+second.Repo) != second.Commit {
t.Fatalf("the later member was not walked after the earlier: %s %v", got.State, err)
}
})
}
}