diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 7b4704a7..9f030a6c 100644 --- a/cmd/mesh-controller/batches.go +++ b/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 @@ -173,6 +178,12 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T fmt.Printf("%s merged into %s (%.8s); nothing the mesh holds reads it\n", repository, m.Base, m.Commit) return nil } + // **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk. + own := ownPath(touches) + waits, err := groupWait(ctx, inv, m, deliverySeatHeld(entries), own) + if err != nil { + return notNow(err) + } release, err := inv.HoldPlans(ctx, true) if err != nil { return notNow(err) @@ -198,7 +209,6 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T // **A merge on the controller's own path never shares a batch with one that waits for the delivery's // word** (decided during the build, 2026-10-10): batched together, the catalogue's deliveries would start // with the controller's and skip their turn. Each kind has a batch of its own. - own := ownPath(touches) if p, ok, err := answeredByALaterMerge(ctx, inv, heard, touches, own, deliverySeatHeld(entries)); err != nil { return notNow(err) } else if ok { @@ -218,7 +228,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T return cutBatchesHeld(ctx, f.open, now) } - batch, err := theOpenBatch(ctx, inv, heard, now, own) + batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own, waits) if err != nil { return notNow(err) } @@ -229,7 +239,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T if err := keepBatch(ctx, inv, &batch, now, ""); err != nil { return notNow(err) } - fmt.Printf("%s merged into %s (%.8s); it joins %s — %s\n", repository, m.Base, m.Commit, batch.ID, + fmt.Printf("%s merged into %s (%.8s); %s, it joins %s — %s\n", repository, m.Base, m.Commit, why, batch.ID, batchWords(batch, now)) return cutBatchesHeld(ctx, f.open, now) } @@ -320,22 +330,211 @@ func ownPath(touches []inventory.Entry) bool { return false } -// theOpenBatch is the batch a merge heard now joins: the one of its kind not yet cut, or a new one. -func theOpenBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own bool) (inventory.Plan, error) { +// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, as ADR 0306 +// decisions 2 and 3 read it). batchToJoin is the batch a merge heard now joins, and why in words: +// +// - an assembling or queued batch of its kind it is related to (related.go): it joins it and waits for the +// window, which its coming restarts. Related to two or more, the newer are folded into the oldest, which +// answers all their merges: a merge never waits in two windows, and what it shares with each walks in one. +// - none, and its delivery group has a member still to merge, or mesh-delivery could not say (groupWait, waits +// names why): a new batch, which waits for the window, beside the ones it shares nothing with; its words say +// why. +// - none, and no member of its group is still to merge: a new batch whose window is closed at once. Cut at +// once means only that: walks stay one at a time (ADR 0276 decision 4), so with a walk open the batch is +// queued behind it as any closed batch is. +func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory.BatchedMerge, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, now time.Time, own bool, waits string) (inventory.Plan, string, error) { + edges, err := inv.Dependencies(ctx) + if err != nil { + return inventory.Plan{}, "", err + } + joins, whys, _, err := relatedBatches(ctx, inv, []inventory.BatchedMerge{heard}, own, "", entries, read, edges) + if err != nil { + return inventory.Plan{}, "", err + } + if len(joins) > 0 { + into := joins[0] + why := "related to " + into.ID + " (" + whys[0] + ")" + if len(joins) > 1 { + plans, moves, err := foldInto(ctx, inv, &into, joins[1:]) + if err != nil { + return inventory.Plan{}, "", err + } + if err := inv.SavePlansMoving(ctx, plans, moves); err != nil { + return inventory.Plan{}, "", err + } + var folded []string + for i, b := range joins[1:] { + folded = append(folded, b.ID+" ("+whys[i+1]+")") + } + why += ", and to " + strings.Join(folded, ", ") + ", folded into it" + } + into.Delivery.Batch.CutAtOnce = false + return into, why, nil + } + if waits != "" { + // Its own window, beside the batches it shares nothing with: waiting is for its group's members, which + // are related to it by their head and join it (ADR 0306 decision 2). + b, err := newBatch(ctx, inv, heard, now, own, false, waits) + return b, waits + ", it waits for the window in a batch of its own", err + } + b, err := newBatch(ctx, inv, heard, now, own, true, "") + return b, "it shares nothing with a batch assembling and no member of a group is still to merge: it is cut at once", err +} + +// relatedBatches are the batches not yet cut of a kind, oldest first, parted into those related to merges (with +// the first reason of each) and the others. Each batch is judged by what its cut would move (ADR 0306 decision +// 3): its merges' moves as they were kept and as they read now — the catalogue may have moved since a merge was +// heard — widened to their dependents, and, for a catalogue batch, every walk waiting for its word, which the +// cut folds in (cutBatch's carry). A merge kept with no moves leaves its batch unknown, related to everything. +// except is a walk not to count as folded: the one being stopped. +func relatedBatches(ctx context.Context, inv *inventory.Inventory, merges []inventory.BatchedMerge, own bool, + except string, entries []inventory.Entry, read map[string][]inventory.ReadRepository, + edges []inventory.Edge) (joins []inventory.Plan, whys []string, others []inventory.Plan, err error) { batches, err := inv.Batches(ctx) if err != nil { - return inventory.Plan{}, err + return nil, nil, nil, err } + var kind []inventory.Plan for _, b := range batches { if b.OwnPath() == own { - return b, nil + kind = append(kind, b) } } + if len(kind) == 0 { + return nil, nil, nil, nil + } + carry := newFootprint() + if !own { + // An own-path batch folds no waiting walk (cutBatchesHeld); a catalogue batch may fold every one. + plans, err := inv.OpenPlans(ctx) + if err != nil { + return nil, nil, nil, err + } + for _, p := range plans { + if p.Release != nil || !p.Waiting() || p.ID == except { + continue + } + kept, err := inv.MergesOf(ctx, p.ID) + if err != nil { + return nil, nil, nil, err + } + carry = unionFootprint(carry, footprintOfWalk(p, kept)) + } + } + mine := footprintAsCut(merges, entries, read, edges) + for _, b := range kind { + held, err := inv.MergesOf(ctx, b.ID) + if err != nil { + return nil, nil, nil, err + } + if b.Delivery == nil { + b.Delivery = &inventory.PlanDelivery{} + } + if b.Delivery.Batch == nil { + b.Delivery.Batch = &inventory.PlanBatch{Own: own} + } + if ok, why := related(mine, unionFootprint(footprintAsCut(held, entries, read, edges), carry), edges); ok { + joins, whys = append(joins, b), append(whys, why) + } else { + others = append(others, b) + } + } + return joins, whys, others, nil +} + +// footprintAsCut is a batch's footprint as a cut now would read it: footprintOfBatch, widened by the modules each +// merge moves as the catalogue reads now (planBatch judges each merge again at the cut, so what it moves may +// have changed since it was heard). Both are kept: when in doubt, related. +func footprintAsCut(merges []inventory.BatchedMerge, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, edges []inventory.Edge) footprint { + f := footprintOfBatch(merges, edges) + var moved []string + for _, m := range merges { + s := announcedOf(m).SourceMoved + if s.Owner == "" { + continue + } + s.MergedAt = "" // news, whatever a cut marked seen since (as planBatch reads it) + for _, e := range wouldMove(s, entries, read) { + moved = append(moved, e.Manifest.Module) + } + } + for _, name := range reachableFrom(moved, edges) { + f.modules[name] = true + } + return f +} + +// unionFootprint is what two changes reach together. +func unionFootprint(a, b footprint) footprint { + f := newFootprint() + for _, from := range []footprint{a, b} { + for k := range from.repos { + f.repos[k] = true + } + for k := range from.modules { + f.modules[k] = true + } + for k := range from.heads { + f.heads[k] = true + } + } + f.unknown = a.unknown + if f.unknown == "" { + f.unknown = b.unknown + } + return f +} + +// foldInto folds batches into the oldest one related to a merge: the plans to write (into first, so no merge +// names a plan the store does not hold) and the merges to move, for one SavePlansMoving. The folded batches name +// the batch that took them over; a stop's closed window and its withheld modules (ADR 0299) go with their merges. +func foldInto(ctx context.Context, inv *inventory.Inventory, into *inventory.Plan, from []inventory.Plan) ([]*inventory.Plan, + []inventory.MergeMove, error) { + if into.Delivery == nil { + into.Delivery = &inventory.PlanDelivery{} + } + if into.Delivery.Batch == nil { + into.Delivery.Batch = &inventory.PlanBatch{Own: into.OwnPath()} + } + plans := []*inventory.Plan{into} + var moves []inventory.MergeMove + for i := range from { + f := from[i] + merges, err := inv.MergesOf(ctx, f.ID) + if err != nil { + return nil, nil, err + } + for _, m := range merges { + moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: into.ID, Alone: m.Alone}) + } + if f.Delivery == nil { + f.Delivery = &inventory.PlanDelivery{} + } + if w := f.Delivery.Batch; w != nil { + into.Delivery.Batch.Closed = into.Delivery.Batch.Closed || w.Closed + into.Delivery.Batch.Withheld = union(into.Delivery.Batch.Withheld, w.Withheld) + sort.Strings(into.Delivery.Batch.Withheld) + } + f.State = inventory.PlanSuperseded + f.Delivery.Batch = nil + f.Delivery.TakenOverBy = into.ID + f.Note = fmt.Sprintf("folded into %s before it was cut: a merge related to both joined it, and %s answers "+ + "its merges", into.ID, into.ID) + plans = append(plans, &f) + } + return plans, moves, nil +} + +// newBatch is a new batch of a kind, kept before its first merge names it: a merge never names a plan the store +// does not hold. cutAtOnce closes its window at once; waits says why it waits for its group, when it does. +func newBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own, + cutAtOnce bool, waits string) (inventory.Plan, error) { b := inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}, - Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}} - // Kept before its first merge names it: a merge never names a plan the store does not hold. + Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own, CutAtOnce: cutAtOnce, Waits: waits}}} if err := inv.SavePlan(ctx, &b); err != nil { return inventory.Plan{}, err } @@ -367,13 +566,16 @@ func keepBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan, } closes, latest := windowOf(merges, b.Created, window, atMost) b.Delivery.Merges = named - var closed bool + var closed, cutAtOnce bool var withheld []string + var waits string if was := b.Delivery.Batch; was != nil { closed, withheld = was.Closed, was.Withheld // a stop's, which no look takes back (novox/hq ADR 0299) + cutAtOnce = was.CutAtOnce // the merge's, until another joins it (ADR 0304 decision 1) + waits = was.Waits // its group's, as said when it was made (ADR 0306 decision 2) } b.Delivery.Batch = &inventory.PlanBatch{ClosesAt: closes, AtMost: latest, Behind: behind, Own: b.OwnPath(), - Closed: closed, Withheld: withheld} + Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits} b.State = inventory.PlanAssembling if behind != "" && windowClosed(*b, now) { b.State = inventory.PlanQueued @@ -402,13 +604,14 @@ func windowOf(merges []inventory.BatchedMerge, created time.Time, window, atMost return last.Add(window), created.Add(atMost) } -// windowClosed says a batch's window is closed: no merge for the window's length, or its maximum reached. +// windowClosed says a batch's window is closed: closed by a stop, closed at once for a merge that shares +// nothing (ADR 0304 decision 1), no merge for the window's length, or its maximum reached. func windowClosed(b inventory.Plan, now time.Time) bool { if b.Delivery == nil || b.Delivery.Batch == nil { return true } w := b.Delivery.Batch - return w.Closed || !now.Before(w.ClosesAt) || !now.Before(w.AtMost) + return w.Closed || w.CutAtOnce || !now.Before(w.ClosesAt) || !now.Before(w.AtMost) } // carriedOf is a set of merges as a walk carries them: the latest merge of each repository's branch, and @@ -617,8 +820,9 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error { return err } // The oldest batch whose window closed is cut, an own-path one before a catalogue one (it goes first, and - // the catalogue batch queues behind it rather than being cut and overtaken); one of each kind at most is - // open (theOpenBatch). + // the catalogue batch queues behind it rather than being cut and overtaken). Several of a kind may be open + // since a merge that shares nothing is cut at once (ADR 0304 decision 1): walks stay one at a time, so one + // is cut per look, and once its walk started the others are queued behind it. sort.SliceStable(batches, func(i, j int) bool { return batches[i].OwnPath() && !batches[j].OwnPath() }) for i := range batches { b := &batches[i] @@ -886,6 +1090,9 @@ func walkTimesAtCut(batch, walk inventory.Plan, entries []inventory.Entry, now t if !w.AtMost.IsZero() && w.AtMost.Before(closed) { closed = w.AtMost } + if w.CutAtOnce && !batch.Created.IsZero() { + closed = batch.Created // closed when it was made (ADR 0304 decision 1) + } if !closed.IsZero() { if closed.After(now) { closed = now @@ -1142,6 +1349,10 @@ func batchWords(b inventory.Plan, now time.Time) string { if b.State == inventory.PlanQueued { return fmt.Sprintf("queued behind %s; grouped: %s; plan not yet calculated", w.Behind, grouped) } + if w.CutAtOnce { + return fmt.Sprintf("cut at once, sharing nothing with a batch assembling; grouped: %s; plan not yet calculated", + grouped) + } left := w.ClosesAt.Sub(now) if w.AtMost.Before(w.ClosesAt) { left = w.AtMost.Sub(now) @@ -1149,8 +1360,12 @@ func batchWords(b inventory.Plan, now time.Time) string { if left < 0 { left = 0 } - return fmt.Sprintf("assembling: %s left (at the latest %s); grouped: %s; plan not yet calculated", - secondsWords(left), w.AtMost.Local().Format("15:04:05"), grouped) + waits := "" + if w.Waits != "" { + waits = " — " + w.Waits + } + return fmt.Sprintf("assembling: %s left (at the latest %s)%s; grouped: %s; plan not yet calculated", + secondsWords(left), w.AtMost.Local().Format("15:04:05"), waits, grouped) } // groupedWords is a batch's commits, each with the earlier merges it answers. diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 23557cfb..778abe73 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 { @@ -58,19 +59,70 @@ func windowed(t *testing.T) *stores { // t0 is the moment a test's clock starts: merges are dated by it, and heard after it. var t0 = time.Now().UTC().Truncate(time.Second) -// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment. +// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment, from a pull +// request's head branch of its own, as the forge announces every merge. Whether its head names a delivery group +// with a member still to merge is mesh-delivery's word, which windowed stubs as no group at all (noGroups); a test +// whose merges wait for the window as a group's says so (groupsWaiting). func catalogueMerge(commit, module string, made time.Time) link.SourceMoved { - return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit, + return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Head: "pr/" + commit, Commit: commit, MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"modules/" + module + "/module.json"}, ModuleDirs: []string{"modules/" + module}, ModuleDirsSaid: true} } -// repoMerge is a merge of a repository of one module's own. +// repoMerge is a merge of a repository of one module's own, from a head branch of its own (catalogueMerge). func repoMerge(repo, commit string, made time.Time) link.SourceMoved { - return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit, + return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Head: "pr/" + commit, Commit: commit, MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"main.go"}} } +// noGroups stubs mesh-delivery's word on groups for a test: no head names a group. +func noGroups(t *testing.T) { + t.Helper() + was := deliveryGroupOf + deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return nil, false, nil } + t.Cleanup(func() { deliveryGroupOf = was }) +} + +// groupsWaiting stubs mesh-delivery's word on groups for a test: each head named is a group with a member of another +// repository whose pull request is open (novox/elsewhere, ready), so a merge from it waits for the window (ADR 0306 +// decision 2); every other head names no group. +func groupsWaiting(t *testing.T, heads ...string) { + t.Helper() + was := deliveryGroupOf + deliveryGroupOf = func(_ context.Context, _ bool, head string) ([]groupMember, bool, error) { + if slices.Contains(heads, head) { + return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil + } + return nil, false, nil + } + t.Cleanup(func() { deliveryGroupOf = was }) +} + +// everyHeadWaits stubs mesh-delivery's word on groups for a test whose merges each wait for the window as they did +// before novox/hq ADR 0304: every head is a group with a member of another repository still to merge. +func everyHeadWaits(t *testing.T) { + t.Helper() + was := deliveryGroupOf + deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { + return []groupMember{{Repository: "novox/elsewhere", State: "ready"}}, true, nil + } + t.Cleanup(func() { deliveryGroupOf = was }) +} + +// fromHead is a merge from a head branch a test names: merges from one head are members of one delivery group, +// related by it (novox/hq ADR 0304 decision 1), so they share a batch whatever else they share. +func fromHead(m link.SourceMoved, head string) link.SourceMoved { + m.Head = head + return m +} + +// lone is a merge that names no head: known to be no delivery group's, it waits for no window unless it is +// related to a batch assembling (ADR 0304 decision 1). +func lone(m link.SourceMoved) link.SourceMoved { + m.Head = "" + return m +} + // hear is a merge heard at a moment. func hear(t *testing.T, open *stores, m link.SourceMoved, now time.Time) { t.Helper() @@ -108,6 +160,7 @@ func walks(t *testing.T, open *stores) (walks, batches []inventory.Plan) { // later commit, which names both merges — the earlier carried by the later. func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asked := asksWithPaths(t) claude := catalogueMerge("553b7191claude", "app", t0) dunst := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second)) @@ -152,9 +205,10 @@ func TestTwoMergesSecondsApartAreOneWalkAtTheLaterCommit(t *testing.T) { // Merges of three repositories in one window are one walk, one commit for each. func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) { open := windowed(t) + groupsWaiting(t, "feat/three") asksRecorded(t) for i, r := range []string{"one", "two", "three"} { - hear(t, open, repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), + hear(t, open, fromHead(repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), "feat/three"), t0.Add(time.Duration(i)*20*time.Second+time.Second)) } cutAt(t, open, t0.Add(41*time.Second+90*time.Second)) @@ -187,9 +241,10 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) { // closes at most ten minutes after its first merge however busy. func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) { open := windowed(t) + groupsWaiting(t, "feat/both", "feat/busy") asksWithPaths(t) - hear(t, open, repoMerge("one", "c1", t0), t0) - hear(t, open, repoMerge("two", "c2", t0.Add(80*time.Second)), t0.Add(80*time.Second)) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(80*time.Second)), "feat/both"), t0.Add(80*time.Second)) cutAt(t, open, t0.Add(100*time.Second)) if ws, _ := walks(t, open); len(ws) != 0 { t.Fatal("the window closed 90 seconds after its first merge, not after its last") @@ -200,12 +255,13 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) { } busy := windowed(t) + groupsWaiting(t, "feat/both", "feat/busy") asksWithPaths(t) var last time.Time for i := 0; i < 12; i++ { last = t0.Add(time.Duration(i) * time.Minute) repo := []string{"one", "two", "three"}[i%3] - hear(t, busy, repoMerge(repo, "c"+string(rune('a'+i)), last), last) + hear(t, busy, fromHead(repoMerge(repo, "c"+string(rune('a'+i)), last), "feat/busy"), last) if ws, _ := walks(t, busy); len(ws) > 0 { if i != 10 { t.Fatalf("a busy window closed at merge %d (minute %d), not at its maximum", i, i) @@ -223,6 +279,7 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) { // what is grouped, and that the plan is not yet calculated. It says so on the bus as plan-moved, too. func TestTheAssemblingBatchIsShown(t *testing.T) { open := windowed(t) + groupsWaiting(t, "pr/553b7191claude") asksWithPaths(t) var said []inventory.Plan was := inventory.PlanSaved @@ -235,7 +292,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) { dunst.Number, dunst.Title = 175, "dunst: the font the operator chose" hear(t, open, claude, now.Add(-19*time.Second)) hear(t, open, dunst, now.Add(-time.Second)) - hear(t, open, repoMerge("one", "a6bc0931one", now), now) + hear(t, open, fromHead(repoMerge("one", "a6bc0931one", now), dunst.Head), now) out := captured(t, func() error { return plansCommand(t.Context(), nil) }) lines := strings.Split(out, "\n") first := lines[0] @@ -360,6 +417,7 @@ func TestAWaitingWalkIsFoldedIntoTheNextBatch(t *testing.T) { // moves — named at once on a walk done — and joins the next batch when it does not. func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asksWithPaths(t) ctx := t.Context() later := catalogueMerge("48bda475dunst", "notes", t0.Add(17*time.Second)) @@ -391,6 +449,7 @@ func TestALateMergeIsAnsweredByTheWalkOfTheLaterOne(t *testing.T) { // first; the first delivered answers the older ones, and the search stops. func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asked := asksWithPaths(t) ctx := t.Context() // Made before the cut marks their modules seen, as merges are: walked again, they are news all the same. @@ -459,6 +518,7 @@ func TestAFailedWalkWalksItsEarlierMergesAlone(t *testing.T) { func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) { for _, grouped := range []bool{true, false} { open := windowed(t) + everyHeadWaits(t) asksWithPaths(t) ctx := t.Context() for _, m := range []string{"mesh-host"} { @@ -496,9 +556,10 @@ func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) { // store, a merge handed over again is not doubled, and the batch is cut when its window closes. func TestABatchSurvivesARestart(t *testing.T) { open := windowed(t) + groupsWaiting(t, "feat/both") asksWithPaths(t) - hear(t, open, repoMerge("one", "c1", t0), t0) - hear(t, open, repoMerge("two", "c2", t0.Add(30*time.Second)), t0.Add(30*time.Second)) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(30*time.Second)), "feat/both"), t0.Add(30*time.Second)) // A new controller: nothing of the first one's but the store. again, err := openStores(t.Context()) @@ -506,7 +567,7 @@ func TestABatchSurvivesARestart(t *testing.T) { t.Fatal(err) } t.Cleanup(again.Close) - hear(t, again, repoMerge("one", "c1", t0), t0.Add(60*time.Second)) // the bus hands it over again + hear(t, again, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0.Add(60*time.Second)) // the bus hands it over again cutAt(t, again, t0.Add(119*time.Second)) if ws, bs := walks(t, again); len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { t.Fatalf("the restarted controller's batch: %d walk(s), %+v", len(ws), bs) @@ -563,6 +624,7 @@ func TestAWaitingWalkNamesTheMergesItAnswers(t *testing.T) { // nothing, and the merge joins the next batch. func TestALateMergeOfAOneModuleRepositoryIsNamed(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asksWithPaths(t) ctx := t.Context() c2 := repoMerge("one", "c2", t0.Add(-50*time.Second)) @@ -716,6 +778,7 @@ func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) { // still waits for the word, so no catalogue delivery skips its turn. func TestAMergeOnTheControllersPathNeverSharesABatch(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asked := asksWithPaths(t) ctx := t.Context() if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1", @@ -860,6 +923,7 @@ func sameMerge(a, b inventory.PlanMerge) bool { // its delivery would skip its turn. It joins the next catalogue batch. func TestALateCatalogueMergeIsNotAnsweredByAnOwnPathWalk(t *testing.T) { open := windowed(t) + everyHeadWaits(t) asksWithPaths(t) ctx := t.Context() for _, m := range []struct { diff --git a/cmd/mesh-controller/batches_watch.go b/cmd/mesh-controller/batches_watch.go index 891d7601..9881c19e 100644 --- a/cmd/mesh-controller/batches_watch.go +++ b/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 { diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index c5f4765f..7f95d589 100644 --- a/cmd/mesh-controller/delivery.go +++ b/cmd/mesh-controller/delivery.go @@ -241,7 +241,7 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why } return p, said, inv.SavePlan(ctx, &p) } - target, err := batchToWalkAgain(ctx, inv, again, now) + target, folded, foldMoves, err := batchToWalkAgain(ctx, inv, p.ID, again, now) if err != nil { return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err) } @@ -262,8 +262,10 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why if len(said.Kept) > 0 { p.Note += "; " + keptWords(said.Kept) } - // The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7). - if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&target, &p}, moves); err != nil { + // The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7) — the + // batches folded into it too. + if err := inv.SavePlansMoving(ctx, append([]*inventory.Plan{&target, &p}, folded...), + append(moves, foldMoves...)); err != nil { return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err) } // What the batch holds and names now: written again by the cutter's next look should this fail. @@ -318,36 +320,42 @@ func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Pla return nil } -// batchToWalkAgain is the batch a stopped walk's merges join: the open batch of their kind, or a new one, not yet -// written. The kind is read from what the merges moved. -func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, again []inventory.BatchedMerge, - now time.Time) (inventory.Plan, error) { +// batchToWalkAgain is the batch a stopped walk's merges join (novox/hq ADR 0299): a batch of their kind not yet cut +// that they are related to (ADR 0304 decision 1, ADR 0306 decision 3), or a new one, not yet written. Related to +// two or more, the newer are folded into the oldest: the plans and merge moves that fold them, to be written in +// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds. +func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge, + now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) { own := false for _, n := range movesOf(again) { if _, on := onTheControllersPath[n]; on { own = true } } - batches, err := inv.Batches(ctx) + entries, read, edges, err := theGraph(ctx, inv) if err != nil { - return inventory.Plan{}, err + return inventory.Plan{}, nil, nil, err } - for _, b := range batches { - if b.OwnPath() == own { - if b.Delivery == nil { - b.Delivery = &inventory.PlanDelivery{} - } - if b.Delivery.Batch == nil { - b.Delivery.Batch = &inventory.PlanBatch{Own: own} - } - return b, nil + joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges) + if err != nil { + return inventory.Plan{}, nil, nil, err + } + if len(joins) > 0 { + into := joins[0] + if len(joins) == 1 { + return into, nil, nil, nil } + plans, moves, err := foldInto(ctx, inv, &into, joins[1:]) + if err != nil { + return inventory.Plan{}, nil, nil, err + } + return into, plans[1:], moves, nil } first := again[0] return inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}, - Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil + Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil, nil, nil } // leaveOut splits a walk's merges by the merges named to leave out, each `@` (a commit's diff --git a/cmd/mesh-controller/group_wait.go b/cmd/mesh-controller/group_wait.go new file mode 100644 index 00000000..391a9b28 --- /dev/null +++ b/cmd/mesh-controller/group_wait.go @@ -0,0 +1,158 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// **A delivery group still assembles** (novox/hq ADR 0306 decision 2): a merge whose delivery group has a member +// not yet merged waits for the merge window, whether or not a batch is assembling, so the group's members walk in +// one batch, in their order (ADR 0276 decision 5). Which groups exist and where each member stands is +// mesh-delivery's record (ADR 0239): a group is the pull requests of two repositories or more sharing a head +// branch name, which is its id. The controller asks it, through its `groups` verb, when a merge from a head is +// heard; a member whose pull request is open (proposed, checked, ready or rejected) is still to merge. When +// mesh-delivery does not answer within groupAskWithin, or answers something unreadable, the merge waits: the safe +// error is to wait, and its batch says why. + +// groupAskWithin is how long mesh-delivery is given to say a merge's group, while the merge is being heard. +var groupAskWithin = 3 * time.Second + +// openPullStates are the states of a delivery whose pull request is open and not merged (mesh-delivery's table): +// a member in one of them is still to merge. Closed unmerged is stopped, final, and waits for nothing. +var openPullStates = map[string]bool{"proposed": true, "checked": true, "ready": true, "rejected": true} + +// groupMember is one member of a delivery group as mesh-delivery's `groups` says it: `@ (state)`. +type groupMember struct { + Repository, State string +} + +// deliveryGroupOf is what mesh-delivery says of the group a head names: its members, and whether the head names a +// group at all; an error when it cannot be asked or did not answer. A seam a test replaces. +var deliveryGroupOf = func(ctx context.Context, held bool, head string) ([]groupMember, bool, error) { + if !held { + return nil, false, fmt.Errorf("no holder of %s is on record to say it", catalogue.DeliverySeat) + } + serving := servingBus.Load() + if serving == nil { + return nil, false, errors.New("this controller is not on the bus") + } + asking, cancel := context.WithTimeout(ctx, groupAskWithin) + defer cancel() + raw, err := askDeliveryOwner(asking, serving.Conn(), "groups", map[string]any{}) + if err != nil { + if errors.Is(err, link.ErrNothingServes) { + return nil, false, fmt.Errorf("nothing serves %s.groups", catalogue.DeliverySeat) + } + return nil, false, err + } + var lines []struct { + ID string `json:"id"` + Members []string `json:"members"` + } + if err := json.Unmarshal(raw, &lines); err != nil { + return nil, false, fmt.Errorf("%s.groups answered something unreadable: %w", catalogue.DeliverySeat, err) + } + for _, l := range lines { + if l.ID != head { + continue + } + var members []groupMember + for _, s := range l.Members { + members = append(members, memberOf(s)) + } + return members, true, nil + } + return nil, false, nil +} + +// memberOf reads one member as `groups` says it: `novox/mesh-sdk@d41d255f28ad (held)`. +func memberOf(s string) groupMember { + id, state := strings.TrimSpace(s), "" + if i := strings.LastIndex(id, " ("); i >= 0 && strings.HasSuffix(id, ")") { + id, state = id[:i], id[i+2:len(id)-1] + } + repository, _, _ := strings.Cut(id, "@") + return groupMember{Repository: repository, State: state} +} + +// groupWait says whether a merge heard now waits for the window for its delivery group, and why in words; nothing +// when it names no head, its head names no group, or every other member is merged or closed. own is the merge's +// kind (ownPath). +// +// A member of another repository whose merge this controller already keeps in a batch not yet cut of the same +// kind is merged, whatever mesh-delivery says yet (it marks a member merged once its walk opened): the merge joins +// that batch by their head. One kept in a batch of the other kind keeps the merge waiting, whatever mesh-delivery +// says (review of #250): kinds never share a batch, so the merge's batch is a new one, and cut at once it would +// walk before that member's batch is cut and break the group's order (ADR 0276 decision 5). Waiting, the two +// batches are cut in the order their windows close, an own-path one first at one look, as cutBatchesHeld cuts +// every pair of kinds. +// +// Asked before the plans are held: a slow answer holds up no walk. +func groupWait(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, held, own bool) (string, error) { + if m.Head == "" { + return "", nil + } + same, other, err := keptFromHead(ctx, inv, m.Head, own) + if err != nil { + return "", err + } + if len(other) > 0 { + return "group " + m.Head + " has a member waiting in a batch of the other kind: " + strings.Join(other, ", "), nil + } + members, isGroup, err := deliveryGroupOf(ctx, held, m.Head) + if err != nil { + return fmt.Sprintf("%s did not say whether group %s has a member still to merge (%v): it waits for the "+ + "window, the safe side", catalogue.DeliverySeat, m.Head, err), nil + } + if !isGroup { + return "", nil + } + var open []string + for _, g := range members { + if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || same[strings.ToLower(g.Repository)] || + !openPullStates[g.State] { + continue + } + open = append(open, g.Repository+" ("+g.State+")") + } + if len(open) == 0 { + return "", nil + } + return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), nil +} + +// keptFromHead are the merges from a head that batches not yet cut hold: the repositories (lower case) of those in +// a batch of the kind own says, and, in words, each of the other kind with its batch. +func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string, own bool) (map[string]bool, []string, error) { + batches, err := inv.Batches(ctx) + if err != nil { + return nil, nil, err + } + same := map[string]bool{} + var other []string + for _, b := range batches { + merges, err := inv.MergesOf(ctx, b.ID) + if err != nil { + return nil, nil, err + } + for _, k := range merges { + if announcedOf(k).Head != head { + continue + } + if b.OwnPath() == own { + same[strings.ToLower(k.Repository)] = true + } else { + other = append(other, fmt.Sprintf("%s@%s in %s", k.Repository, short(k.Commit), b.ID)) + } + } + } + return same, other, nil +} diff --git a/cmd/mesh-controller/issue459_test.go b/cmd/mesh-controller/issue459_test.go index 37f1782e..7fec24d4 100644 --- a/cmd/mesh-controller/issue459_test.go +++ b/cmd/mesh-controller/issue459_test.go @@ -123,11 +123,12 @@ func TestDeliveryStopTakesTheMergesToLeaveOut(t *testing.T) { // that builds the other repository's module alone. func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) { open := windowed(t) + groupsWaiting(t, "pr/cat1aaaaaaaa") endsItsLine(t) asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -181,12 +182,13 @@ func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) { // ended. A merge the walk does not answer is refused, and nothing ends. func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) { open := windowed(t) + groupsWaiting(t, "pr/cat1aaaaaaaa") endsItsLine(t) asksRecorded(t) ctx := t.Context() first := catalogueMerge("cat1aaaaaaaa", "app", t0) later := catalogueMerge("cat2cccccccc", "notes", t0.Add(10*time.Second)) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), first.Head) hear(t, open, first, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) hear(t, open, later, t0.Add(11*time.Second)) @@ -238,11 +240,12 @@ func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) { // (review of #232): its window closes, and its walk is at the newer commit, answering both. func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) { open := windowed(t) + groupsWaiting(t, "pr/cat1aaaaaaaa") endsItsLine(t) asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one1, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -290,11 +293,12 @@ func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) { // that walked its merges again is refused a retry, saying why. func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) { open := windowed(t) + groupsWaiting(t, "pr/cat1aaaaaaaa") endsItsLine(t) asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one1, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -335,11 +339,12 @@ func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) { // with the others and withholds the stopped merge's modules; with every merge left out, the batch is stopped. func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) { open := windowed(t) + everyHeadWaits(t) endsItsLine(t) asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) _, bs := walks(t, open) @@ -382,6 +387,7 @@ func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) { // else since it was read refuses all of it, and no merge moves, no plan is written. func TestSavingPlansAndMovingMergesIsOneAct(t *testing.T) { open := windowed(t) + everyHeadWaits(t) endsItsLine(t) ctx := t.Context() hear(t, open, catalogueMerge("cat1aaaaaaaa", "app", t0), t0.Add(time.Second)) diff --git a/cmd/mesh-controller/lone_merge_test.go b/cmd/mesh-controller/lone_merge_test.go new file mode 100644 index 00000000..7ab1dbc3 --- /dev/null +++ b/cmd/mesh-controller/lone_merge_test.go @@ -0,0 +1,398 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// The tests of novox/hq issue 525's window skip: a merge waits for the merge window only when a batch assembling +// holds a merge related to it, or its delivery group may have a member still to merge (ADR 0304 decision 1, ADR +// 0306 decisions 2 and 3). Walks stay one at a time. + +// A merge from no head that shares nothing, with no walk open, is cut into its walk when it is heard: no window. +func TestALoneMergeSharingNothingIsCutAtOnce(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + hear(t, open, lone(repoMerge("one", "c1", t0)), t0) + ws, bs := walks(t, open) + if len(ws) != 1 || len(bs) != 0 || !ws[0].Open() || ws[0].Commit != "c1" { + t.Fatalf("a lone merge sharing nothing: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } + if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(t0) { + t.Fatalf("its window closed at %+v; want when it was heard, %s", ws[0].Times, t0) + } +} + +// A merge related to the batch assembling joins it and waits for the window, which its coming restarts: by its +// repository, by a module in common (a dependent of what the batch moves), by an edge of the module graph between +// their modules (one that widens nothing), or by its delivery group's head. +func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) { + ctx := t.Context() + for _, c := range []string{"same repository", "a module in common", "an edge of the graph", "the same group head"} { + t.Run(c, func(t *testing.T) { + open := windowed(t) + groupsWaiting(t, "pr/c1") + asksWithPaths(t) + inv := open.inventory + first := repoMerge("one", "c1", t0) + second := lone(repoMerge("two", "c2", t0.Add(10*time.Second))) + why := map[string]string{"same repository": "both change novox/one", "a module in common": "both move two", + "an edge of the graph": "two depends on builder (built-by)", + "the same group head": "both are members of group " + first.Head}[c] + switch c { + case "same repository": + second = lone(repoMerge("one", "c2", t0.Add(10*time.Second))) + case "a module in common": + // two is built on one: a merge of one moves two too. + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "two", Version: "1", + Build: &catalogue.Build{On: []catalogue.BuildsOn{{Arg: "BASE", Module: "one"}}}}, + inventory.Source{Repository: "novox/two", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + case "an edge of the graph": + // two is built by the build machine: an edge that orders and widens nothing. + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "builder", Version: "1", + Claims: []catalogue.Claim{{Name: "mesh-build-machine", Scope: catalogue.ScopeMesh}}}, + inventory.Source{Repository: "novox/builder", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + first = repoMerge("builder", "c1", t0) + case "the same group head": + second = fromHead(repoMerge("two", "c2", t0.Add(10*time.Second)), first.Head) + } + hear(t, open, first, t0) + said := captured(t, func() error { + return (following{open: open}).hearMerge(ctx, second, t0.Add(10*time.Second)) + }) + if !strings.Contains(said, "("+why+")") { + t.Fatalf("the merge was not said related by %q: %s", why, said) + } + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { + t.Fatalf("the related merge: %d walk(s), batches %+v", len(ws), bs) + } + if bs[0].Delivery.Batch.CutAtOnce { + t.Fatal("a batch a related merge joined is cut at once") + } + cutAt(t, open, t0.Add(99*time.Second)) + if ws, _ := walks(t, open); len(ws) != 0 { + t.Fatalf("cut before the window the related merge restarted closed: %+v", ws) + } + cutAt(t, open, t0.Add(100*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || len(bs) != 0 || len(ws[0].Delivery.Merges) != 2 { + t.Fatalf("after the window: %d walk(s), %d batch(es)", len(ws), len(bs)) + } + }) + } +} + +// A merge from a head whose delivery group has a member still to merge waits for the window, even with no batch +// assembling, and says why; its sibling joins the same batch by their head (ADR 0306 decision 2). A merge from a +// head that names no group shares nothing and is cut at once, as one from no head is. +func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + groupsWaiting(t, "feat/x") + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce || windowClosed(bs[0], t0) { + t.Fatalf("a group's merge alone: %d walk(s), batches %+v", len(ws), bs) + } + if words := batchWords(bs[0], t0); !strings.HasPrefix(words, "assembling: 90 s left") || + !strings.Contains(words, "group feat/x has a member still to merge: novox/elsewhere (ready)") { + t.Fatalf("it reads %q", words) + } + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/x"), t0.Add(5*time.Second)) + if _, bs := walks(t, open); len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { + t.Fatalf("the group's second member did not join the first: %+v", bs) + } + cutAt(t, open, t0.Add(94*time.Second)) + if ws, _ := walks(t, open); len(ws) != 0 { + t.Fatalf("cut before its window closed: %+v", ws) + } + cutAt(t, open, t0.Add(95*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || len(bs) != 0 || ws[0].CommitOf("novox/one") != "c1" || ws[0].CommitOf("novox/two") != "c2" { + t.Fatalf("at its window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } +} + +// A merge from a head that names no group, sharing nothing, with no walk open, is cut at once. +func TestAHeadedMergeInNoGroupIsCutAtOnce(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "fix/alone"), t0) + if ws, bs := walks(t, open); len(ws) != 1 || len(bs) != 0 || !ws[0].Open() { + t.Fatalf("a headed merge in no group: %d walk(s), %d batch(es)", len(ws), len(bs)) + } +} + +// When mesh-delivery cannot say a head's group, the merge waits for the window, the safe side, and its batch says +// why. +func TestAGroupNobodyCanSayWaits(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + was := deliveryGroupOf + deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { + return nil, false, errors.New("nothing serves mesh-delivery.groups") + } + t.Cleanup(func() { deliveryGroupOf = was }) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce { + t.Fatalf("a merge whose group nobody could say: %d walk(s), batches %+v", len(ws), bs) + } + if words := batchWords(bs[0], t0); !strings.Contains(words, "mesh-delivery did not say whether group feat/x has a "+ + "member still to merge (nothing serves mesh-delivery.groups): it waits for the window, the safe side") { + t.Fatalf("it reads %q", words) + } +} + +// A member counts as still to merge only while its pull request is open, and not once this controller keeps its +// merge: a group whose other members are merged or closed makes nobody wait. +func TestAGroupsMergedOrClosedMembersMakeNobodyWait(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + ctx := t.Context() + members := []groupMember{{Repository: "novox/one", State: "ready"}, {Repository: "novox/two", State: "published"}, + {Repository: "novox/three", State: "stopped"}} + was := deliveryGroupOf + deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { return members, true, nil } + t.Cleanup(func() { deliveryGroupOf = was }) + m := fromHead(repoMerge("one", "c1", t0), "feat/x") + if why, err := groupWait(ctx, open.inventory, m, true, false); err != nil || why != "" { + t.Fatalf("the group's other members are merged or closed, yet: %q %v", why, err) + } + members = append(members, groupMember{Repository: "novox/four", State: "checked"}) + if why, _ := groupWait(ctx, open.inventory, m, true, false); !strings.Contains(why, "novox/four (checked)") { + t.Fatalf("an open member: %q", why) + } + // novox/four's merge is kept in a batch already: merged, whatever mesh-delivery says yet. + hear(t, open, fromHead(repoMerge("two", "c4", t0), "feat/x"), t0) + members[3].Repository = "novox/two" + if why, _ := groupWait(ctx, open.inventory, m, true, false); why != "" { + t.Fatalf("a member this controller keeps the merge of still waits: %q", why) + } + if got := memberOf("novox/mesh-sdk@d41d255f28ad (held)"); got != (groupMember{Repository: "novox/mesh-sdk", State: "held"}) { + t.Fatalf("a member reads %+v", got) + } +} + +// A merge related to two batches assembling folds the newer into the older, which answers every merge of both +// and names nothing twice; the folded one names the batch that took it over. +func TestAMergeRelatedToTwoBatchesFoldsThem(t *testing.T) { + open := windowed(t) + groupsWaiting(t, "feat/a", "feat/b") + asksWithPaths(t) + ctx := t.Context() + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/a"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/b"), t0.Add(5*time.Second)) + _, bs := walks(t, open) + if len(bs) != 2 { + t.Fatalf("two merges sharing nothing: %d batch(es)", len(bs)) + } + older, newer := bs[1], bs[0] // newest first + // Of one's repository and from two's head: related to both. + hear(t, open, fromHead(repoMerge("one", "c3", t0.Add(10*time.Second)), "feat/b"), t0.Add(10*time.Second)) + _, bs = walks(t, open) + if len(bs) != 1 || bs[0].ID != older.ID || len(bs[0].Delivery.Merges) != 3 { + t.Fatalf("after a merge related to both: batches %+v; want %s answering three merges", bs, older.ID) + } + folded, err := open.inventory.PlanByID(ctx, newer.ID) + if err != nil || folded.State != inventory.PlanSuperseded || folded.Delivery == nil || + folded.Delivery.TakenOverBy != older.ID { + t.Fatalf("the newer batch reads %s %+v (%v)", folded.State, folded.Delivery, err) + } + if left, _ := open.inventory.MergesOf(ctx, newer.ID); len(left) != 0 { + t.Fatalf("the folded batch still answers %+v", left) + } + cutAt(t, open, t0.Add(100*time.Second)) + ws, bs := walks(t, open) + var cut []inventory.Plan + for _, w := range ws { + if w.State != inventory.PlanSuperseded { + cut = append(cut, w) + } + } + if len(cut) != 1 || len(bs) != 0 || cut[0].ID != older.ID || cut[0].CommitOf("novox/one") != "c3" || + cut[0].CommitOf("novox/two") != "c2" { + t.Fatalf("the folded batch's walk: %+v, %d batch(es)", cut, len(bs)) + } +} + +// With a walk open, a batch cut at once still waits for it, said queued behind it (walks stay one at a time, ADR +// 0276 decision 4); it is cut once that walk ended, with no window to wait for. +func TestACutAtOnceBatchQueuesBehindAnOpenWalk(t *testing.T) { + open := windowed(t) + groupsWaiting(t, "pr/c1") + asksWithPaths(t) + hear(t, open, repoMerge("one", "c1", t0), t0) + cutAt(t, open, t0.Add(90*time.Second)) + ws, _ := walks(t, open) + if len(ws) != 1 || !ws[0].Open() { + t.Fatalf("no walk open: %+v", ws) + } + first := ws[0] + heard := t0.Add(100 * time.Second) + hear(t, open, lone(repoMerge("two", "c2", heard)), heard) + ws, bs := walks(t, open) + if len(ws) != 1 || len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != first.ID || + !bs[0].Delivery.Batch.CutAtOnce { + t.Fatalf("a lone merge during a walk: %d walk(s), batches %+v", len(ws), bs) + } + if !strings.HasPrefix(batchWords(bs[0], heard), "queued behind "+first.ID) { + t.Fatalf("it reads %q", batchWords(bs[0], heard)) + } + first.State = inventory.PlanDone + if err := open.inventory.SavePlan(t.Context(), &first); err != nil { + t.Fatal(err) + } + cutAt(t, open, heard.Add(time.Second)) + ws, bs = walks(t, open) + if len(ws) != 2 || len(bs) != 0 || ws[0].Commit != "c2" { + t.Fatalf("the queued batch was not cut once the walk ended: %+v %+v", ws, bs) + } + if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(heard) { + t.Fatalf("its window closed at %+v; want when its merge was heard, %s", ws[0].Times, heard) + } +} + +// A stopped walk's other merges are walked again in a batch they are related to, and a batch of their own when +// they share nothing with the one assembling (ADR 0299, ADR 0304 decision 1): never in a batch of another +// change, whose window the stop would close. +func TestAStopWalksTheOthersAgainInABatchOfTheirOwnWhenTheyShareNothing(t *testing.T) { + open := windowed(t) + groupsWaiting(t, "pr/cat1aaaaaaaa") + endsItsLine(t) + asksRecorded(t) + ctx := t.Context() + cat := catalogueMerge("cat1aaaaaaaa", "app", t0) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) + hear(t, open, cat, t0.Add(time.Second)) + hear(t, open, one, t0.Add(6*time.Second)) + cutAt(t, open, t0.Add(2*time.Minute)) + ws, _ := walks(t, open) + walk := ws[0] + hear(t, open, repoMerge("two", "two1cccccccc", t0.Add(3*time.Minute)), t0.Add(3*time.Minute)) + _, bs := walks(t, open) + if len(bs) != 1 { + t.Fatalf("the merge heard while the walk ran is in %d batches", len(bs)) + } + other := bs[0] + _, said, err := stopWalkWithout(ctx, open.inventory, walk.ID, "mesh-delivery for jochen", "its group stopped", + []string{"novox/mesh-catalog@" + cat.Commit}, t0.Add(3*time.Minute+10*time.Second)) + if err != nil { + t.Fatal(err) + } + if said.WalkedAgainBy == "" || said.WalkedAgainBy == other.ID { + t.Fatalf("walked again by %q; want a batch of its own, not %s", said.WalkedAgainBy, other.ID) + } + b, _ := open.inventory.PlanByID(ctx, other.ID) + if b.Delivery.Batch == nil || b.Delivery.Batch.Closed || len(b.Delivery.Merges) != 1 { + t.Fatalf("the other change's batch reads %+v; want it untouched by the stop", b.Delivery) + } +} + +// A catalogue batch's cut folds in every walk waiting for its delivery's word (cutBatch's carry), so what that +// walk moves is the batch's too: a lone merge related to the waiting walk joins the batch assembling and waits for +// its window, rather than being cut at once as sharing nothing with what the batch's own merges move. +func TestALoneMergeRelatedToAWalkTheCutFoldsJoinsTheBatch(t *testing.T) { + open := windowed(t) + groupsWaiting(t, "pr/c3") + asksWithPaths(t) + ctx := t.Context() + if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1", + Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}}, + inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/mesh-delivery", Ref: "main", + BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + if _, err := open.inventory.Assign(ctx, "anchor", "mesh-delivery"); err != nil { + t.Fatal(err) + } + hear(t, open, repoMerge("one", "c1", t0), t0) + cutAt(t, open, t0.Add(90*time.Second)) + if ws, _ := walks(t, open); len(ws) != 1 || !ws[0].Waiting() { + t.Fatalf("the walk does not wait for its word: %+v", ws) + } + hear(t, open, repoMerge("three", "c3", t0.Add(2*time.Minute)), t0.Add(2*time.Minute)) + hear(t, open, lone(repoMerge("one", "c2", t0.Add(2*time.Minute+5*time.Second))), t0.Add(2*time.Minute+5*time.Second)) + _, bs := walks(t, open) + if len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 || bs[0].Delivery.Batch.CutAtOnce { + t.Fatalf("a lone merge related to the waiting walk: batches %+v", bs) + } +} + +// A delivery group with a member of each kind (review of #250): a merge on the controller's own path and a +// catalogue merge never share a batch, so the later member cannot join the earlier one's. Kept in a batch of the +// other kind, the earlier member keeps the later one waiting, whatever mesh-delivery says yet — never cut at once +// to walk before it. The two batches are then cut in the order their windows close (cutBatchesHeld: the oldest +// closed batch first, an own-path one first at one look, and the other queued behind a walk that started): the +// earlier member's window closes first, so it walks first, in both orders. +func TestAGroupAcrossBothKindsKeepsItsOrder(t *testing.T) { + for _, ownFirst := range []bool{true, false} { + name := "catalogue member first" + if ownFirst { + name = "controller member first" + } + t.Run(name, func(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + ctx := t.Context() + was := deliveryGroupOf + // mesh-delivery still says both members ready: it marks one merged only once its walk opened. + deliveryGroupOf = func(context.Context, bool, string) ([]groupMember, bool, error) { + return []groupMember{{Repository: "novox/mesh-controller", State: "ready"}, + {Repository: "novox/mesh-catalog", State: "ready"}}, true, nil + } + t.Cleanup(func() { deliveryGroupOf = was }) + own := fromHead(repoMerge("mesh-controller", "k1", t0), "feat/both-kinds") + cat := fromHead(catalogueMerge("cat1aaaaaaaa", "app", t0), "feat/both-kinds") + first, second := cat, own + if ownFirst { + first, second = own, cat + } + second.MergedAt = t0.Add(10 * time.Second).Format(time.RFC3339Nano) + hear(t, open, first, t0) + hear(t, open, second, t0.Add(10*time.Second)) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 2 { + t.Fatalf("both members heard: %d walk(s), %d batch(es); want two batches, nothing cut", len(ws), len(bs)) + } + later := bs[0] // newest first + if later.Delivery.Batch.CutAtOnce || !strings.Contains(later.Delivery.Batch.Waits, + "has a member waiting in a batch of the other kind: novox/"+first.Repo+"@") { + t.Fatalf("the later member's batch reads %+v; want it waiting for the earlier member", later.Delivery.Batch) + } + // The earlier member's window closes first: its batch alone is cut. + cutAt(t, open, t0.Add(90*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || !ws[0].Open() || ws[0].CommitOf("novox/"+first.Repo) != first.Commit || len(bs) != 1 || + bs[0].ID != later.ID { + t.Fatalf("at the earlier window: walks %+v, batches %+v; want the earlier member alone cut", ws, bs) + } + earlier := ws[0] + // The later member's window closes while the earlier walk runs: it queues behind it. + cutAt(t, open, t0.Add(100*time.Second)) + if _, bs := walks(t, open); len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != earlier.ID { + t.Fatalf("the later member's batch does not queue behind the earlier walk: %+v", bs) + } + earlier.State = inventory.PlanDone + if err := open.inventory.SavePlan(ctx, &earlier); err != nil { + t.Fatal(err) + } + cutAt(t, open, t0.Add(101*time.Second)) + got, err := open.inventory.PlanByID(ctx, later.ID) + if err != nil || got.Batch() || got.CommitOf("novox/"+second.Repo) != second.Commit { + t.Fatalf("the later member was not walked after the earlier: %s %v", got.State, err) + } + }) + } +} 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 fa7f3231..b6530854 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -125,6 +125,14 @@ 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"` + // 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. diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 29b367c5..a37baf3b 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -8,7 +8,7 @@ private addresses. That is acceptable while mesh-controller is private; if it ev a second place to clean besides the catalogue itself. mesh-catalog b9de001833b1b61b297182b8e3bcdae1cbdeace9 modules/*/module.json, modules/nats/Dockerfile -mesh-host bd5cc6980419c1bd4824be6d58dfebd2381a9de3 examples/foundation-first-node-nats.lock +mesh-host a76baa4b47f07a8b2bab065603414530bd02b53a examples/foundation-first-node-nats.lock mesh-sdk 33917f91ac3cc213b879b32f3c70e2aa70395f98 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: diff --git a/testdata/beside/mesh-host/examples/foundation-first-node-nats.lock b/testdata/beside/mesh-host/examples/foundation-first-node-nats.lock index d40283a2..25d84627 100644 --- a/testdata/beside/mesh-host/examples/foundation-first-node-nats.lock +++ b/testdata/beside/mesh-host/examples/foundation-first-node-nats.lock @@ -161,7 +161,7 @@ "type": "file", "path": "/var/lib/mesh-bus-conf/accounts.conf", "mode": "0600", - "content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_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.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n" + "content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_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.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n" }, { "id": "broker", diff --git a/testdata/beside/mesh-host/internal/apply/placement_guard.go.captured b/testdata/beside/mesh-host/internal/apply/placement_guard.go.captured index 49c1d83e..01112d50 100644 --- a/testdata/beside/mesh-host/internal/apply/placement_guard.go.captured +++ b/testdata/beside/mesh-host/internal/apply/placement_guard.go.captured @@ -31,6 +31,13 @@ package apply // 5. **A mount source that is a refused directory or a refused access is refused with it**: the container // would otherwise bind the very path the engine would not place, and the runtime creates a missing one // as root. +// 6. **A file is judged as a directory is** (novox/hq issue 495): never one of the machine's own roots, nor +// in a tree nothing is placed in (rule 2), and never below a directory or an access refused here. A file +// or an archive below a refused path is refused with it, whichever module declares it and in whatever +// order. Rules 3 and 4 are a directory's owner and do not reach a file: a file below /etc is the mesh's +// ordinary configuration, and one below a home is judged by home_links.go and agent_home_guard.go. +// A refused directory also gates the rest of its module, as a step that did not complete does +// (apply.go's `gated`): what a module declares after its directory was written for that directory. // // Each is a failed resource with its reason in the node's report; nothing is touched. Paths are judged as // declared and again with every link in them resolved, so a link at /srv/x pointing at /etc places nothing. @@ -400,6 +407,7 @@ const ( placingDirectory placing = iota placingAccess placingMount + placingFile ) // refusePath says why a path is not placed or mounted; nil when it may be. module is the resource's module, @@ -466,7 +474,7 @@ func refuseOne(path string, kind placing, module, owner string) *PlacementRefuse for _, part := range strings.Split(path, "/") { ssh = ssh || part == ".ssh" } - if ssh && kind != placingDirectory { + if ssh && (kind == placingAccess || kind == placingMount) { return &PlacementRefusedError{Path: path, Why: "it is an account's .ssh, which holds its keys and who may " + "log in as it, and is never an access or a mount"} } @@ -502,8 +510,12 @@ func moduleOfID(id string) string { return module } -// refusedPlaces judges every directory and access of a declaration before anything is applied, and answers -// each refusal by path: what is refused is refused again as a container's mount source. +// refusedPlaces judges every directory, access, file and archive of a declaration before anything is applied, +// and answers each refusal by path: what is refused is refused again as a container's mount source. +// +// Directories and accesses first, so a file or an archive below one of them is refused with it whatever order +// they are declared in (novox/hq issue 495): the engine refused Docker's data root as a directory and then wrote +// a file into it, because nothing judged the file's path. func refusedPlaces(resources []declaration.Resource) map[string]error { out := map[string]error{} for _, r := range resources { @@ -520,9 +532,60 @@ func refusedPlaces(resources []declaration.Resource) map[string]error { out[filepath.Clean(r.Target())] = err } } + trees := make(map[string]error, len(out)) + for path, err := range out { + trees[path] = err + } + for _, r := range resources { + var err error + switch res := r.(type) { + case *declaration.File: + if err = belowRefused(res.Path, trees); err == nil { + err = refusePath(res.Path, placingFile, moduleOfID(res.ID), "") + } + case *declaration.Archive: + // Only below a refused path: where an archive may unpack is otherwise the archive's own judgement + // (archive.go), and the engine's own module unpacks its versions into its own tree. + err = belowRefused(res.Path, trees) + default: + continue + } + if err != nil { + out[filepath.Clean(r.Target())] = err + } + } return out } +// belowRefused says why a path below a refused directory or access is refused with it; nil when it is below +// none. Judged as declared and with its links resolved, as refusePath judges. +func belowRefused(path string, refused map[string]error) error { + clean := filepath.Clean(path) + if !filepath.IsAbs(clean) { + return nil + } + for _, p := range []string{clean, resolved(clean)} { + // The deepest refused tree it is below, so the same declaration always gives the same reason. + var tree string + for t := range refused { + if below(p, t) && len(t) > len(tree) { + tree = t + } + } + if tree == "" { + continue + } + why := refused[tree] + reason := why.Error() + var refusal *PlacementRefusedError + if errors.As(why, &refusal) { + reason = refusal.Why + } + return &PlacementRefusedError{Path: clean, Why: "it is below " + tree + ", which is refused: " + reason} + } + return nil +} + // refuseMounts says why a container's mounts are not made; nil when they may be. func refuseMounts(c *declaration.Container, refused map[string]error) error { for _, v := range c.Volumes {