Ask mesh-delivery whether a merge's group still assembles (hq issue 525)

A merge from any head waited for the window because the controller could not
see a group's members, so the window skip changed nothing in production and
unrelated merges walked one after another. The controller now asks
mesh-delivery's groups, within 3 s and before the plans are held: a merge
waits only when its head's group has another member whose pull request is
open and whose merge it does not keep yet (ADR 0306 §2). No answer means it
waits, and its batch says why. A headed merge in no waiting group and related
to no batch is cut at once, as one from no head is (ADR 0304 §1, ADR 0306 §3).

The controller's grant names mesh-delivery.groups. The installer's first user
list in mesh-host must carry it too before the genesis test can pass.

cmd/mesh-controller/batches.go

@@ -178,6 +178,11 @@ 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.
+	waits, err := groupWait(ctx, inv, m, deliverySeatHeld(entries))
+	if err != nil {
+		return notNow(err)
+	}
 	release, err := inv.HoldPlans(ctx, true)
 	if err != nil {
 		return notNow(err)
@@ -223,7 +228,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
 		return cutBatchesHeld(ctx, f.open, now)
 	}
 
-	batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own)
+	batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own, waits)
 	if err != nil {
 		return notNow(err)
 	}
@@ -331,13 +336,14 @@ func ownPath(touches []inventory.Entry) bool {
 //   - 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 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) (inventory.Plan, string, error) {
+	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
@@ -366,28 +372,14 @@ func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory.
 		into.Delivery.Batch.CutAtOnce = false
 		return into, why, nil
 	}
-	if waits, why := stillAssembles(heard); waits {
+	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)
-		return b, why + ", it waits for the window in a batch of its own", err
+		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 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, ""
+	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
@@ -536,13 +528,13 @@ func foldInto(ctx context.Context, inv *inventory.Inventory, into *inventory.Pla
 }
 
 // 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.
+// 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) (inventory.Plan, error) {
+	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, CutAtOnce: cutAtOnce}}}
+		Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own, CutAtOnce: cutAtOnce, Waits: waits}}}
 	if err := inv.SavePlan(ctx, &b); err != nil {
 		return inventory.Plan{}, err
 	}
@@ -576,12 +568,14 @@ func keepBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan,
 	b.Delivery.Merges = named
 	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, CutAtOnce: cutAtOnce}
+		Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits}
 	b.State = inventory.PlanAssembling
 	if behind != "" && windowClosed(*b, now) {
 		b.State = inventory.PlanQueued
@@ -1366,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.

cmd/mesh-controller/batches_test.go

@@ -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 {
@@ -59,9 +60,9 @@ func windowed(t *testing.T) *stores {
 var t0 = time.Now().UTC().Truncate(time.Second)
 
 // 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.
+// 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", Head: "pr/" + commit, Commit: commit,
 		MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"modules/" + module + "/module.json"},
@@ -74,6 +75,40 @@ func repoMerge(repo, commit string, made time.Time) link.SourceMoved {
 		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 {
@@ -125,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))
@@ -169,6 +205,7 @@ 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, fromHead(repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), "feat/three"),
@@ -204,6 +241,7 @@ 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, 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))
@@ -217,6 +255,7 @@ 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++ {
@@ -240,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
@@ -377,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))
@@ -408,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.
@@ -476,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"} {
@@ -513,6 +556,7 @@ 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, 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))
@@ -580,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))
@@ -733,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",
@@ -877,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 {

cmd/mesh-controller/group_wait.go

@@ -0,0 +1,140 @@
+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. A member of another
+// repository whose merge this controller already keeps in a batch not yet cut is merged, whatever mesh-delivery
+// says yet (it marks a member merged once its walk opened). 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 bool) (string, error) {
+	if m.Head == "" {
+		return "", 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
+	}
+	kept, err := keptFromHead(ctx, inv, m.Head)
+	if err != nil {
+		return "", err
+	}
+	var open []string
+	for _, g := range members {
+		if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || kept[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 repositories (lower case) of the merges from a head that batches not yet cut hold.
+func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string) (map[string]bool, error) {
+	batches, err := inv.Batches(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := map[string]bool{}
+	for _, b := range batches {
+		merges, err := inv.MergesOf(ctx, b.ID)
+		if err != nil {
+			return nil, err
+		}
+		for _, k := range merges {
+			if announcedOf(k).Head == head {
+				out[strings.ToLower(k.Repository)] = true
+			}
+		}
+	}
+	return out, nil
+}

cmd/mesh-controller/issue459_test.go

@@ -123,6 +123,7 @@ 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()
@@ -181,6 +182,7 @@ 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()
@@ -238,6 +240,7 @@ 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()
@@ -290,6 +293,7 @@ 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()
@@ -335,6 +339,7 @@ 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()
@@ -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))

cmd/mesh-controller/lone_merge_test.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"context"
+	"errors"
 	"strings"
 	"testing"
 	"time"
@@ -35,6 +37,7 @@ func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) {
 	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)
@@ -90,39 +93,103 @@ func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) {
 	}
 }
 
-// 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.
+// 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 !strings.HasPrefix(batchWords(bs[0], t0), "assembling: 90 s left") {
-		t.Fatalf("it reads %q", batchWords(bs[0], t0))
+	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/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))
+	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(89*time.Second))
+	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(90*time.Second))
+	cutAt(t, open, t0.Add(95*time.Second))
 	ws, bs = walks(t, open)
-	if len(ws) != 1 || ws[0].Commit != "c1" || len(bs) != 1 {
+	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); 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); !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); 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)
@@ -164,6 +231,7 @@ func TestAMergeRelatedToTwoBatchesFoldsThem(t *testing.T) {
 // 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))
@@ -201,6 +269,7 @@ func TestACutAtOnceBatchQueuesBehindAnOpenWalk(t *testing.T) {
 // 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()
@@ -236,6 +305,7 @@ func TestAStopWalksTheOthersAgainInABatchOfTheirOwnWhenTheyShareNothing(t *testi
 // 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",