diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 25832623..15cd862b 100644 --- a/cmd/mesh-controller/batches.go +++ b/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. diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 45bd6c5d..beab9582 100644 --- a/cmd/mesh-controller/batches_test.go +++ b/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 { diff --git a/cmd/mesh-controller/group_wait.go b/cmd/mesh-controller/group_wait.go new file mode 100644 index 00000000..dc293819 --- /dev/null +++ b/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: `@ (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 +} diff --git a/cmd/mesh-controller/issue459_test.go b/cmd/mesh-controller/issue459_test.go index 9fef7a86..7fec24d4 100644 --- a/cmd/mesh-controller/issue459_test.go +++ b/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)) diff --git a/cmd/mesh-controller/lone_merge_test.go b/cmd/mesh-controller/lone_merge_test.go index 1861bda9..1a291896 100644 --- a/cmd/mesh-controller/lone_merge_test.go +++ b/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", diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 2280c9cf..ab48ea46 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -262,9 +262,12 @@ func MaySubscribe(perms Permissions, subject string) bool { // delivery held past its bound, and calls them on the operator's warrant, with its why. // // And, since novox/hq ADR 0282, `times`: the self-check reads the delivery times to say a delivery over its budget. +// +// And, since novox/hq ADR 0306, `groups`: a merge heard from a head waits for the window while its delivery group +// has a member still to merge. var VerbsTheControllerAsksTheDeliveryOwner = []SeatVerb{{Seat: "mesh-delivery", Verb: "stalled"}, {Seat: "mesh-delivery", Verb: "close"}, {Seat: "mesh-delivery", Verb: "release"}, {Seat: "mesh-delivery", Verb: "stop"}, - {Seat: "mesh-delivery", Verb: "times"}} + {Seat: "mesh-delivery", Verb: "times"}, {Seat: "mesh-delivery", Verb: "groups"}} // VerbsTheControllerActsOnAWarrant are the other seat verbs the controller calls when the operator's warrant // chooses them (novox/hq ADR 0259): a machine's service restarted, and a walk started or stopped through the diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index ea22d3c5..b3003837 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -24,7 +24,7 @@ accounts { jetstream: enabled users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index 01fde9fa..b37d69f4 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -122,6 +122,9 @@ type PlanBatch struct { // 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"` + // Waits says why a batch made for one merge waits for the window: its delivery group has a member still to + // merge, or mesh-delivery could not say whether it has (novox/hq ADR 0306 decision 2). + Waits string `json:"waits,omitempty"` } // OwnPath says the record is a batch of merges on the controller's own path.