Let a merge that shares nothing skip the merge window (hq issue 525)

A merge waited the whole window even when nothing assembling shared anything
with it. It now joins only a batch it is related to (ADR 0304 §1, read by ADR
0306 §3: repository, a module the cut would move, a graph edge, a group head),
folding the newer into the oldest when it relates to two. A merge related to
none starts a batch whose window is closed at once (PlanBatch.CutAtOnce),
unless its delivery group may still have a member to merge (ADR 0306 §2).

The controller cannot see a group's unmerged members (that is mesh-delivery's
record), so a merge from any head still waits, in a batch of its own: the safe
error is to wait. Walks stay one at a time; a batch cut at once queues behind
an open walk. A stop's merges walked again join a related batch, or their own.

cmd/mesh-controller/batches.go

@@ -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
@@ -218,7 +223,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)
 	if err != nil {
 		return notNow(err)
 	}
@@ -229,7 +234,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 +325,224 @@ 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 may have a member still to merge (stillAssembles): a new batch, which waits
+//     for the window as every batch did before ADR 0304, beside the ones it shares nothing with.
+//   - 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) (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, why := stillAssembles(heard); 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)
+		return b, why + ", 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 is cut at once", err
+}
+
+// stillAssembles says a merge's delivery group may have a member still to merge, so the merge waits for the
+// merge window whether or not a batch is assembling (novox/hq ADR 0306 decision 2), and why in words.
+//
+// Which members a group has and which of them are merged is mesh-delivery's record, not the controller's: the
+// controller hears a merge's head branch, which is its group's name, and nothing of the pull requests still open
+// on that branch in other repositories. So every merge from a head waits, as every merge did before ADR 0304:
+// the safe error is to wait (issue 525). Only a merge that names no head is known to be no group's.
+func stillAssembles(m inventory.BatchedMerge) (bool, string) {
+	if h := announcedOf(m).Head; h != "" {
+		return true, "group " + h + " may have a member still to merge"
+	}
+	return false, ""
+}
+
+// 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.
+func newBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own,
+	cutAtOnce bool) (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}}}
 	if err := inv.SavePlan(ctx, &b); err != nil {
 		return inventory.Plan{}, err
 	}
@@ -367,13 +574,14 @@ 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
 	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)
 	}
 	b.Delivery.Batch = &inventory.PlanBatch{ClosesAt: closes, AtMost: latest, Behind: behind, Own: b.OwnPath(),
-		Closed: closed, Withheld: withheld}
+		Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce}
 	b.State = inventory.PlanAssembling
 	if behind != "" && windowClosed(*b, now) {
 		b.State = inventory.PlanQueued
@@ -402,13 +610,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 +826,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 +1096,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 +1355,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)

cmd/mesh-controller/batches_test.go

@@ -58,19 +58,36 @@ 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. A merge from a head may have a group
+// member still to merge, which the controller cannot tell, so it waits for the window (novox/hq ADR 0306
+// decision 2, stillAssembles): these merges keep the window's rules of ADR 0276. lone is one from no head.
 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"}}
 }
 
+// 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()
@@ -154,7 +171,7 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
 	open := windowed(t)
 	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))
@@ -188,8 +205,8 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) {
 func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
 	open := windowed(t)
 	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")
@@ -205,7 +222,7 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) {
 	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)
@@ -235,7 +252,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]
@@ -497,8 +514,8 @@ func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) {
 func TestABatchSurvivesARestart(t *testing.T) {
 	open := windowed(t)
 	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 +523,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)

cmd/mesh-controller/batches_watch.go

@@ -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 {

cmd/mesh-controller/delivery.go

@@ -240,7 +240,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)
 	}
@@ -261,8 +261,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.
@@ -317,36 +319,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

cmd/mesh-controller/issue459_test.go

@@ -127,7 +127,7 @@ func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.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))
@@ -186,7 +186,7 @@ func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.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))
@@ -242,7 +242,7 @@ func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.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))
@@ -294,7 +294,7 @@ func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.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))
@@ -339,7 +339,7 @@ func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.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)

cmd/mesh-controller/lone_merge_test.go

@@ -0,0 +1,261 @@
+package main
+
+import (
+	"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)
+			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 names its delivery group, which may have a member still to merge: it waits for the window
+// even with no batch assembling, and beside a batch it shares nothing with, in a batch of its own (ADR 0306
+// decision 2). A merge from another head that shares nothing does not join it.
+func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) {
+	open := windowed(t)
+	asksWithPaths(t)
+	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 !strings.HasPrefix(batchWords(bs[0], t0), "assembling: 90 s left") {
+		t.Fatalf("it reads %q", batchWords(bs[0], t0))
+	}
+	hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/y"), t0.Add(5*time.Second))
+	if _, bs := walks(t, open); len(bs) != 2 {
+		t.Fatalf("a merge sharing nothing joined another's batch: %d batch(es)", len(bs))
+	}
+	cutAt(t, open, t0.Add(89*time.Second))
+	if ws, _ := walks(t, open); len(ws) != 0 {
+		t.Fatalf("cut before its window closed: %+v", ws)
+	}
+	cutAt(t, open, t0.Add(90*time.Second))
+	ws, bs = walks(t, open)
+	if len(ws) != 1 || ws[0].Commit != "c1" || len(bs) != 1 {
+		t.Fatalf("at its window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs))
+	}
+}
+
+// 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)
+	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)
+	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)
+	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)
+	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)
+	}
+}

internal/inventory/plans.go

@@ -117,6 +117,11 @@ type PlanBatch struct {
 	// the batch's walk does not build them.
 	Closed   bool     `json:"closed,omitempty"`
 	Withheld []string `json:"withheld,omitempty"`
+	// CutAtOnce says the batch's window closed when it was made: its merge shares nothing with a batch
+	// assembling and has no delivery group member still to merge, so it waits for no window (novox/hq ADR 0304
+	// decision 1, ADR 0306 decisions 2 and 3). A merge that joins it later opens its window again. It is not
+	// Closed, which is a stop's.
+	CutAtOnce bool `json:"cut_at_once,omitempty"`
 }
 
 // OwnPath says the record is a batch of merges on the controller's own path.