diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 7b4704a7..25832623 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 @@ -218,7 +223,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T return cutBatchesHeld(ctx, f.open, now) } - batch, err := theOpenBatch(ctx, inv, heard, now, own) + batch, why, err := batchToJoin(ctx, inv, heard, entries, read, now, own) if err != nil { return notNow(err) } @@ -229,7 +234,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T if err := keepBatch(ctx, inv, &batch, now, ""); err != nil { return notNow(err) } - fmt.Printf("%s merged into %s (%.8s); it joins %s — %s\n", repository, m.Base, m.Commit, batch.ID, + fmt.Printf("%s merged into %s (%.8s); %s, it joins %s — %s\n", repository, m.Base, m.Commit, why, batch.ID, batchWords(batch, now)) return cutBatchesHeld(ctx, f.open, now) } @@ -320,22 +325,224 @@ func ownPath(touches []inventory.Entry) bool { return false } -// theOpenBatch is the batch a merge heard now joins: the one of its kind not yet cut, or a new one. -func theOpenBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own bool) (inventory.Plan, error) { +// **A merge waits only for a batch it shares something with** (novox/hq ADR 0304 decision 1, as ADR 0306 +// decisions 2 and 3 read it). batchToJoin is the batch a merge heard now joins, and why in words: +// +// - an assembling or queued batch of its kind it is related to (related.go): it joins it and waits for the +// window, which its coming restarts. Related to two or more, the newer are folded into the oldest, which +// answers all their merges: a merge never waits in two windows, and what it shares with each walks in one. +// - none, and its delivery group may have a member still to merge (stillAssembles): a new batch, which waits +// for the window as every batch did before ADR 0304, beside the ones it shares nothing with. +// - none, and no member of its group is still to merge: a new batch whose window is closed at once. Cut at +// once means only that: walks stay one at a time (ADR 0276 decision 4), so with a walk open the batch is +// queued behind it as any closed batch is. +func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory.BatchedMerge, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, now time.Time, own bool) (inventory.Plan, string, error) { + edges, err := inv.Dependencies(ctx) + if err != nil { + return inventory.Plan{}, "", err + } + joins, whys, _, err := relatedBatches(ctx, inv, []inventory.BatchedMerge{heard}, own, "", entries, read, edges) + if err != nil { + return inventory.Plan{}, "", err + } + if len(joins) > 0 { + into := joins[0] + why := "related to " + into.ID + " (" + whys[0] + ")" + if len(joins) > 1 { + plans, moves, err := foldInto(ctx, inv, &into, joins[1:]) + if err != nil { + return inventory.Plan{}, "", err + } + if err := inv.SavePlansMoving(ctx, plans, moves); err != nil { + return inventory.Plan{}, "", err + } + var folded []string + for i, b := range joins[1:] { + folded = append(folded, b.ID+" ("+whys[i+1]+")") + } + why += ", and to " + strings.Join(folded, ", ") + ", folded into it" + } + into.Delivery.Batch.CutAtOnce = false + return into, why, nil + } + if waits, why := stillAssembles(heard); waits { + // Its own window, beside the batches it shares nothing with: waiting is for its group's members, which + // are related to it by their head and join it (ADR 0306 decision 2). + b, err := newBatch(ctx, inv, heard, now, own, false) + return b, why + ", it waits for the window in a batch of its own", err + } + b, err := newBatch(ctx, inv, heard, now, own, true) + return b, "it shares nothing with a batch assembling and is cut at once", err +} + +// stillAssembles says a merge's delivery group may have a member still to merge, so the merge waits for the +// merge window whether or not a batch is assembling (novox/hq ADR 0306 decision 2), and why in words. +// +// Which members a group has and which of them are merged is mesh-delivery's record, not the controller's: the +// controller hears a merge's head branch, which is its group's name, and nothing of the pull requests still open +// on that branch in other repositories. So every merge from a head waits, as every merge did before ADR 0304: +// the safe error is to wait (issue 525). Only a merge that names no head is known to be no group's. +func stillAssembles(m inventory.BatchedMerge) (bool, string) { + if h := announcedOf(m).Head; h != "" { + return true, "group " + h + " may have a member still to merge" + } + return false, "" +} + +// relatedBatches are the batches not yet cut of a kind, oldest first, parted into those related to merges (with +// the first reason of each) and the others. Each batch is judged by what its cut would move (ADR 0306 decision +// 3): its merges' moves as they were kept and as they read now — the catalogue may have moved since a merge was +// heard — widened to their dependents, and, for a catalogue batch, every walk waiting for its word, which the +// cut folds in (cutBatch's carry). A merge kept with no moves leaves its batch unknown, related to everything. +// except is a walk not to count as folded: the one being stopped. +func relatedBatches(ctx context.Context, inv *inventory.Inventory, merges []inventory.BatchedMerge, own bool, + except string, entries []inventory.Entry, read map[string][]inventory.ReadRepository, + edges []inventory.Edge) (joins []inventory.Plan, whys []string, others []inventory.Plan, err error) { batches, err := inv.Batches(ctx) if err != nil { - return inventory.Plan{}, err + return nil, nil, nil, err } + var kind []inventory.Plan for _, b := range batches { if b.OwnPath() == own { - return b, nil + kind = append(kind, b) } } + if len(kind) == 0 { + return nil, nil, nil, nil + } + carry := newFootprint() + if !own { + // An own-path batch folds no waiting walk (cutBatchesHeld); a catalogue batch may fold every one. + plans, err := inv.OpenPlans(ctx) + if err != nil { + return nil, nil, nil, err + } + for _, p := range plans { + if p.Release != nil || !p.Waiting() || p.ID == except { + continue + } + kept, err := inv.MergesOf(ctx, p.ID) + if err != nil { + return nil, nil, nil, err + } + carry = unionFootprint(carry, footprintOfWalk(p, kept)) + } + } + mine := footprintAsCut(merges, entries, read, edges) + for _, b := range kind { + held, err := inv.MergesOf(ctx, b.ID) + if err != nil { + return nil, nil, nil, err + } + if b.Delivery == nil { + b.Delivery = &inventory.PlanDelivery{} + } + if b.Delivery.Batch == nil { + b.Delivery.Batch = &inventory.PlanBatch{Own: own} + } + if ok, why := related(mine, unionFootprint(footprintAsCut(held, entries, read, edges), carry), edges); ok { + joins, whys = append(joins, b), append(whys, why) + } else { + others = append(others, b) + } + } + return joins, whys, others, nil +} + +// footprintAsCut is a batch's footprint as a cut now would read it: footprintOfBatch, widened by the modules each +// merge moves as the catalogue reads now (planBatch judges each merge again at the cut, so what it moves may +// have changed since it was heard). Both are kept: when in doubt, related. +func footprintAsCut(merges []inventory.BatchedMerge, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, edges []inventory.Edge) footprint { + f := footprintOfBatch(merges, edges) + var moved []string + for _, m := range merges { + s := announcedOf(m).SourceMoved + if s.Owner == "" { + continue + } + s.MergedAt = "" // news, whatever a cut marked seen since (as planBatch reads it) + for _, e := range wouldMove(s, entries, read) { + moved = append(moved, e.Manifest.Module) + } + } + for _, name := range reachableFrom(moved, edges) { + f.modules[name] = true + } + return f +} + +// unionFootprint is what two changes reach together. +func unionFootprint(a, b footprint) footprint { + f := newFootprint() + for _, from := range []footprint{a, b} { + for k := range from.repos { + f.repos[k] = true + } + for k := range from.modules { + f.modules[k] = true + } + for k := range from.heads { + f.heads[k] = true + } + } + f.unknown = a.unknown + if f.unknown == "" { + f.unknown = b.unknown + } + return f +} + +// foldInto folds batches into the oldest one related to a merge: the plans to write (into first, so no merge +// names a plan the store does not hold) and the merges to move, for one SavePlansMoving. The folded batches name +// the batch that took them over; a stop's closed window and its withheld modules (ADR 0299) go with their merges. +func foldInto(ctx context.Context, inv *inventory.Inventory, into *inventory.Plan, from []inventory.Plan) ([]*inventory.Plan, + []inventory.MergeMove, error) { + if into.Delivery == nil { + into.Delivery = &inventory.PlanDelivery{} + } + if into.Delivery.Batch == nil { + into.Delivery.Batch = &inventory.PlanBatch{Own: into.OwnPath()} + } + plans := []*inventory.Plan{into} + var moves []inventory.MergeMove + for i := range from { + f := from[i] + merges, err := inv.MergesOf(ctx, f.ID) + if err != nil { + return nil, nil, err + } + for _, m := range merges { + moves = append(moves, inventory.MergeMove{Repository: m.Repository, Commit: m.Commit, Plan: into.ID, Alone: m.Alone}) + } + if f.Delivery == nil { + f.Delivery = &inventory.PlanDelivery{} + } + if w := f.Delivery.Batch; w != nil { + into.Delivery.Batch.Closed = into.Delivery.Batch.Closed || w.Closed + into.Delivery.Batch.Withheld = union(into.Delivery.Batch.Withheld, w.Withheld) + sort.Strings(into.Delivery.Batch.Withheld) + } + f.State = inventory.PlanSuperseded + f.Delivery.Batch = nil + f.Delivery.TakenOverBy = into.ID + f.Note = fmt.Sprintf("folded into %s before it was cut: a merge related to both joined it, and %s answers "+ + "its merges", into.ID, into.ID) + plans = append(plans, &f) + } + return plans, moves, nil +} + +// newBatch is a new batch of a kind, kept before its first merge names it: a merge never names a plan the store +// does not hold. cutAtOnce closes its window at once. +func newBatch(ctx context.Context, inv *inventory.Inventory, first inventory.BatchedMerge, now time.Time, own, + cutAtOnce bool) (inventory.Plan, error) { b := inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}, - Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}} - // Kept before its first merge names it: a merge never names a plan the store does not hold. + Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own, CutAtOnce: cutAtOnce}}} if err := inv.SavePlan(ctx, &b); err != nil { return inventory.Plan{}, err } @@ -367,13 +574,14 @@ func keepBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Plan, } closes, latest := windowOf(merges, b.Created, window, atMost) b.Delivery.Merges = named - var closed bool + var closed, cutAtOnce bool var withheld []string if was := b.Delivery.Batch; was != nil { closed, withheld = was.Closed, was.Withheld // a stop's, which no look takes back (novox/hq ADR 0299) + cutAtOnce = was.CutAtOnce // the merge's, until another joins it (ADR 0304 decision 1) } b.Delivery.Batch = &inventory.PlanBatch{ClosesAt: closes, AtMost: latest, Behind: behind, Own: b.OwnPath(), - Closed: closed, Withheld: withheld} + Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce} b.State = inventory.PlanAssembling if behind != "" && windowClosed(*b, now) { b.State = inventory.PlanQueued @@ -402,13 +610,14 @@ func windowOf(merges []inventory.BatchedMerge, created time.Time, window, atMost return last.Add(window), created.Add(atMost) } -// windowClosed says a batch's window is closed: no merge for the window's length, or its maximum reached. +// windowClosed says a batch's window is closed: closed by a stop, closed at once for a merge that shares +// nothing (ADR 0304 decision 1), no merge for the window's length, or its maximum reached. func windowClosed(b inventory.Plan, now time.Time) bool { if b.Delivery == nil || b.Delivery.Batch == nil { return true } w := b.Delivery.Batch - return w.Closed || !now.Before(w.ClosesAt) || !now.Before(w.AtMost) + return w.Closed || w.CutAtOnce || !now.Before(w.ClosesAt) || !now.Before(w.AtMost) } // carriedOf is a set of merges as a walk carries them: the latest merge of each repository's branch, and @@ -617,8 +826,9 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error { return err } // The oldest batch whose window closed is cut, an own-path one before a catalogue one (it goes first, and - // the catalogue batch queues behind it rather than being cut and overtaken); one of each kind at most is - // open (theOpenBatch). + // the catalogue batch queues behind it rather than being cut and overtaken). Several of a kind may be open + // since a merge that shares nothing is cut at once (ADR 0304 decision 1): walks stay one at a time, so one + // is cut per look, and once its walk started the others are queued behind it. sort.SliceStable(batches, func(i, j int) bool { return batches[i].OwnPath() && !batches[j].OwnPath() }) for i := range batches { b := &batches[i] @@ -886,6 +1096,9 @@ func walkTimesAtCut(batch, walk inventory.Plan, entries []inventory.Entry, now t if !w.AtMost.IsZero() && w.AtMost.Before(closed) { closed = w.AtMost } + if w.CutAtOnce && !batch.Created.IsZero() { + closed = batch.Created // closed when it was made (ADR 0304 decision 1) + } if !closed.IsZero() { if closed.After(now) { closed = now @@ -1142,6 +1355,10 @@ func batchWords(b inventory.Plan, now time.Time) string { if b.State == inventory.PlanQueued { return fmt.Sprintf("queued behind %s; grouped: %s; plan not yet calculated", w.Behind, grouped) } + if w.CutAtOnce { + return fmt.Sprintf("cut at once, sharing nothing with a batch assembling; grouped: %s; plan not yet calculated", + grouped) + } left := w.ClosesAt.Sub(now) if w.AtMost.Before(w.ClosesAt) { left = w.AtMost.Sub(now) diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 846804ca..45bd6c5d 100644 --- a/cmd/mesh-controller/batches_test.go +++ b/cmd/mesh-controller/batches_test.go @@ -58,19 +58,36 @@ func windowed(t *testing.T) *stores { // t0 is the moment a test's clock starts: merges are dated by it, and heard after it. var t0 = time.Now().UTC().Truncate(time.Second) -// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment. +// catalogueMerge is a merge of the catalogue changing one module's directory, made at a moment, from a pull +// request's head branch of its own, as the forge announces every merge. A merge from a head may have a group +// member still to merge, which the controller cannot tell, so it waits for the window (novox/hq ADR 0306 +// decision 2, stillAssembles): these merges keep the window's rules of ADR 0276. lone is one from no head. func catalogueMerge(commit, module string, made time.Time) link.SourceMoved { - return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit, + return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Head: "pr/" + commit, Commit: commit, MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"modules/" + module + "/module.json"}, ModuleDirs: []string{"modules/" + module}, ModuleDirsSaid: true} } -// repoMerge is a merge of a repository of one module's own. +// repoMerge is a merge of a repository of one module's own, from a head branch of its own (catalogueMerge). func repoMerge(repo, commit string, made time.Time) link.SourceMoved { - return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit, + return link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Head: "pr/" + commit, Commit: commit, MergedAt: made.Format(time.RFC3339Nano), Paths: []string{"main.go"}} } +// fromHead is a merge from a head branch a test names: merges from one head are members of one delivery group, +// related by it (novox/hq ADR 0304 decision 1), so they share a batch whatever else they share. +func fromHead(m link.SourceMoved, head string) link.SourceMoved { + m.Head = head + return m +} + +// lone is a merge that names no head: known to be no delivery group's, it waits for no window unless it is +// related to a batch assembling (ADR 0304 decision 1). +func lone(m link.SourceMoved) link.SourceMoved { + m.Head = "" + return m +} + // hear is a merge heard at a moment. func hear(t *testing.T, open *stores, m link.SourceMoved, now time.Time) { t.Helper() @@ -154,7 +171,7 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) { open := windowed(t) asksRecorded(t) for i, r := range []string{"one", "two", "three"} { - hear(t, open, repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), + hear(t, open, fromHead(repoMerge(r, "c-"+r, t0.Add(time.Duration(i)*20*time.Second)), "feat/three"), t0.Add(time.Duration(i)*20*time.Second+time.Second)) } cutAt(t, open, t0.Add(41*time.Second+90*time.Second)) @@ -188,8 +205,8 @@ func TestThreeRepositoriesInOneWindowAreOneWalk(t *testing.T) { func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) { open := windowed(t) asksWithPaths(t) - hear(t, open, repoMerge("one", "c1", t0), t0) - hear(t, open, repoMerge("two", "c2", t0.Add(80*time.Second)), t0.Add(80*time.Second)) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(80*time.Second)), "feat/both"), t0.Add(80*time.Second)) cutAt(t, open, t0.Add(100*time.Second)) if ws, _ := walks(t, open); len(ws) != 0 { t.Fatal("the window closed 90 seconds after its first merge, not after its last") @@ -205,7 +222,7 @@ func TestAMergeRestartsTheWindowAndTheMaximumClosesIt(t *testing.T) { for i := 0; i < 12; i++ { last = t0.Add(time.Duration(i) * time.Minute) repo := []string{"one", "two", "three"}[i%3] - hear(t, busy, repoMerge(repo, "c"+string(rune('a'+i)), last), last) + hear(t, busy, fromHead(repoMerge(repo, "c"+string(rune('a'+i)), last), "feat/busy"), last) if ws, _ := walks(t, busy); len(ws) > 0 { if i != 10 { t.Fatalf("a busy window closed at merge %d (minute %d), not at its maximum", i, i) @@ -235,7 +252,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) { dunst.Number, dunst.Title = 175, "dunst: the font the operator chose" hear(t, open, claude, now.Add(-19*time.Second)) hear(t, open, dunst, now.Add(-time.Second)) - hear(t, open, repoMerge("one", "a6bc0931one", now), now) + hear(t, open, fromHead(repoMerge("one", "a6bc0931one", now), dunst.Head), now) out := captured(t, func() error { return plansCommand(t.Context(), nil) }) lines := strings.Split(out, "\n") first := lines[0] @@ -497,8 +514,8 @@ func TestAGroupsOrderBecomesTiersInsideTheWalk(t *testing.T) { func TestABatchSurvivesARestart(t *testing.T) { open := windowed(t) asksWithPaths(t) - hear(t, open, repoMerge("one", "c1", t0), t0) - hear(t, open, repoMerge("two", "c2", t0.Add(30*time.Second)), t0.Add(30*time.Second)) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(30*time.Second)), "feat/both"), t0.Add(30*time.Second)) // A new controller: nothing of the first one's but the store. again, err := openStores(t.Context()) @@ -506,7 +523,7 @@ func TestABatchSurvivesARestart(t *testing.T) { t.Fatal(err) } t.Cleanup(again.Close) - hear(t, again, repoMerge("one", "c1", t0), t0.Add(60*time.Second)) // the bus hands it over again + hear(t, again, fromHead(repoMerge("one", "c1", t0), "feat/both"), t0.Add(60*time.Second)) // the bus hands it over again cutAt(t, again, t0.Add(119*time.Second)) if ws, bs := walks(t, again); len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { t.Fatalf("the restarted controller's batch: %d walk(s), %+v", len(ws), bs) 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 4055e730..716060fe 100644 --- a/cmd/mesh-controller/delivery.go +++ b/cmd/mesh-controller/delivery.go @@ -240,7 +240,7 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why } return p, said, inv.SavePlan(ctx, &p) } - target, err := batchToWalkAgain(ctx, inv, again, now) + target, folded, foldMoves, err := batchToWalkAgain(ctx, inv, p.ID, again, now) if err != nil { return p, said, fmt.Errorf("%s is not stopped: %w", p.ID, err) } @@ -261,8 +261,10 @@ func stopWalkWithout(ctx context.Context, inv *inventory.Inventory, id, by, why if len(said.Kept) > 0 { p.Note += "; " + keptWords(said.Kept) } - // The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7). - if err := inv.SavePlansMoving(ctx, []*inventory.Plan{&target, &p}, moves); err != nil { + // The batch first, so no merge names a plan the store does not hold; all of it, or nothing (decision 7) — the + // batches folded into it too. + if err := inv.SavePlansMoving(ctx, append([]*inventory.Plan{&target, &p}, folded...), + append(moves, foldMoves...)); err != nil { return p, said, fmt.Errorf("%s is not stopped, and none of its merges moved: %w", p.ID, err) } // What the batch holds and names now: written again by the cutter's next look should this fail. @@ -317,36 +319,42 @@ func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Pla return nil } -// batchToWalkAgain is the batch a stopped walk's merges join: the open batch of their kind, or a new one, not yet -// written. The kind is read from what the merges moved. -func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, again []inventory.BatchedMerge, - now time.Time) (inventory.Plan, error) { +// batchToWalkAgain is the batch a stopped walk's merges join (novox/hq ADR 0299): a batch of their kind not yet cut +// that they are related to (ADR 0304 decision 1, ADR 0306 decision 3), or a new one, not yet written. Related to +// two or more, the newer are folded into the oldest: the plans and merge moves that fold them, to be written in +// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds. +func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge, + now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) { own := false for _, n := range movesOf(again) { if _, on := onTheControllersPath[n]; on { own = true } } - batches, err := inv.Batches(ctx) + entries, read, edges, err := theGraph(ctx, inv) if err != nil { - return inventory.Plan{}, err + return inventory.Plan{}, nil, nil, err } - for _, b := range batches { - if b.OwnPath() == own { - if b.Delivery == nil { - b.Delivery = &inventory.PlanDelivery{} - } - if b.Delivery.Batch == nil { - b.Delivery.Batch = &inventory.PlanBatch{Own: own} - } - return b, nil + joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges) + if err != nil { + return inventory.Plan{}, nil, nil, err + } + if len(joins) > 0 { + into := joins[0] + if len(joins) == 1 { + return into, nil, nil, nil } + plans, moves, err := foldInto(ctx, inv, &into, joins[1:]) + if err != nil { + return inventory.Plan{}, nil, nil, err + } + return into, plans[1:], moves, nil } first := again[0] return inventory.Plan{ID: fmt.Sprintf("plan-%d", now.UnixNano()), Repository: first.Repository, Branch: first.Branch, Commit: first.Commit, Merged: first.Merged, Created: now, State: inventory.PlanAssembling, Tiers: [][]string{}, Modules: map[string]*inventory.PlanModule{}, - Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil + Delivery: &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: own}}}, nil, nil, nil } // leaveOut splits a walk's merges by the merges named to leave out, each `@` (a commit's diff --git a/cmd/mesh-controller/issue459_test.go b/cmd/mesh-controller/issue459_test.go index 37f1782e..9fef7a86 100644 --- a/cmd/mesh-controller/issue459_test.go +++ b/cmd/mesh-controller/issue459_test.go @@ -127,7 +127,7 @@ func TestAStopWithoutAMergeWalksTheOthersAgain(t *testing.T) { asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -186,7 +186,7 @@ func TestAStopWithoutAMergeEndsTheMergesThatContainIt(t *testing.T) { ctx := t.Context() first := catalogueMerge("cat1aaaaaaaa", "app", t0) later := catalogueMerge("cat2cccccccc", "notes", t0.Add(10*time.Second)) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), first.Head) hear(t, open, first, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) hear(t, open, later, t0.Add(11*time.Second)) @@ -242,7 +242,7 @@ func TestAStopWalksTheOthersAgainInTheOpenBatch(t *testing.T) { asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one1, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -294,7 +294,7 @@ func TestAStopNamesWhatStaysOnTheMachinesAndWhatStillCarriesIt(t *testing.T) { asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one1 := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one1 := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one1, t0.Add(6*time.Second)) cutAt(t, open, t0.Add(2*time.Minute)) @@ -339,7 +339,7 @@ func TestAStopWithoutAMergeLeavesItOutOfABatch(t *testing.T) { asksRecorded(t) ctx := t.Context() cat := catalogueMerge("cat1aaaaaaaa", "app", t0) - one := repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) hear(t, open, cat, t0.Add(time.Second)) hear(t, open, one, t0.Add(6*time.Second)) _, bs := walks(t, open) diff --git a/cmd/mesh-controller/lone_merge_test.go b/cmd/mesh-controller/lone_merge_test.go new file mode 100644 index 00000000..1861bda9 --- /dev/null +++ b/cmd/mesh-controller/lone_merge_test.go @@ -0,0 +1,261 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// The tests of novox/hq issue 525's window skip: a merge waits for the merge window only when a batch assembling +// holds a merge related to it, or its delivery group may have a member still to merge (ADR 0304 decision 1, ADR +// 0306 decisions 2 and 3). Walks stay one at a time. + +// A merge from no head that shares nothing, with no walk open, is cut into its walk when it is heard: no window. +func TestALoneMergeSharingNothingIsCutAtOnce(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + hear(t, open, lone(repoMerge("one", "c1", t0)), t0) + ws, bs := walks(t, open) + if len(ws) != 1 || len(bs) != 0 || !ws[0].Open() || ws[0].Commit != "c1" { + t.Fatalf("a lone merge sharing nothing: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } + if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(t0) { + t.Fatalf("its window closed at %+v; want when it was heard, %s", ws[0].Times, t0) + } +} + +// A merge related to the batch assembling joins it and waits for the window, which its coming restarts: by its +// repository, by a module in common (a dependent of what the batch moves), by an edge of the module graph between +// their modules (one that widens nothing), or by its delivery group's head. +func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) { + ctx := t.Context() + for _, c := range []string{"same repository", "a module in common", "an edge of the graph", "the same group head"} { + t.Run(c, func(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + inv := open.inventory + first := repoMerge("one", "c1", t0) + second := lone(repoMerge("two", "c2", t0.Add(10*time.Second))) + why := map[string]string{"same repository": "both change novox/one", "a module in common": "both move two", + "an edge of the graph": "two depends on builder (built-by)", + "the same group head": "both are members of group " + first.Head}[c] + switch c { + case "same repository": + second = lone(repoMerge("one", "c2", t0.Add(10*time.Second))) + case "a module in common": + // two is built on one: a merge of one moves two too. + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "two", Version: "1", + Build: &catalogue.Build{On: []catalogue.BuildsOn{{Arg: "BASE", Module: "one"}}}}, + inventory.Source{Repository: "novox/two", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + case "an edge of the graph": + // two is built by the build machine: an edge that orders and widens nothing. + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "builder", Version: "1", + Claims: []catalogue.Claim{{Name: "mesh-build-machine", Scope: catalogue.ScopeMesh}}}, + inventory.Source{Repository: "novox/builder", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + first = repoMerge("builder", "c1", t0) + case "the same group head": + second = fromHead(repoMerge("two", "c2", t0.Add(10*time.Second)), first.Head) + } + hear(t, open, first, t0) + said := captured(t, func() error { + return (following{open: open}).hearMerge(ctx, second, t0.Add(10*time.Second)) + }) + if !strings.Contains(said, "("+why+")") { + t.Fatalf("the merge was not said related by %q: %s", why, said) + } + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { + t.Fatalf("the related merge: %d walk(s), batches %+v", len(ws), bs) + } + if bs[0].Delivery.Batch.CutAtOnce { + t.Fatal("a batch a related merge joined is cut at once") + } + cutAt(t, open, t0.Add(99*time.Second)) + if ws, _ := walks(t, open); len(ws) != 0 { + t.Fatalf("cut before the window the related merge restarted closed: %+v", ws) + } + cutAt(t, open, t0.Add(100*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || len(bs) != 0 || len(ws[0].Delivery.Merges) != 2 { + t.Fatalf("after the window: %d walk(s), %d batch(es)", len(ws), len(bs)) + } + }) + } +} + +// A merge from a head names its delivery group, which may have a member still to merge: it waits for the window +// even with no batch assembling, and beside a batch it shares nothing with, in a batch of its own (ADR 0306 +// decision 2). A merge from another head that shares nothing does not join it. +func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/x"), t0) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || bs[0].Delivery.Batch.CutAtOnce || windowClosed(bs[0], t0) { + t.Fatalf("a group's merge alone: %d walk(s), batches %+v", len(ws), bs) + } + if !strings.HasPrefix(batchWords(bs[0], t0), "assembling: 90 s left") { + t.Fatalf("it reads %q", batchWords(bs[0], t0)) + } + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/y"), t0.Add(5*time.Second)) + if _, bs := walks(t, open); len(bs) != 2 { + t.Fatalf("a merge sharing nothing joined another's batch: %d batch(es)", len(bs)) + } + cutAt(t, open, t0.Add(89*time.Second)) + if ws, _ := walks(t, open); len(ws) != 0 { + t.Fatalf("cut before its window closed: %+v", ws) + } + cutAt(t, open, t0.Add(90*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || ws[0].Commit != "c1" || len(bs) != 1 { + t.Fatalf("at its window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } +} + +// A merge related to two batches assembling folds the newer into the older, which answers every merge of both +// and names nothing twice; the folded one names the batch that took it over. +func TestAMergeRelatedToTwoBatchesFoldsThem(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + ctx := t.Context() + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/a"), t0) + hear(t, open, fromHead(repoMerge("two", "c2", t0.Add(5*time.Second)), "feat/b"), t0.Add(5*time.Second)) + _, bs := walks(t, open) + if len(bs) != 2 { + t.Fatalf("two merges sharing nothing: %d batch(es)", len(bs)) + } + older, newer := bs[1], bs[0] // newest first + // Of one's repository and from two's head: related to both. + hear(t, open, fromHead(repoMerge("one", "c3", t0.Add(10*time.Second)), "feat/b"), t0.Add(10*time.Second)) + _, bs = walks(t, open) + if len(bs) != 1 || bs[0].ID != older.ID || len(bs[0].Delivery.Merges) != 3 { + t.Fatalf("after a merge related to both: batches %+v; want %s answering three merges", bs, older.ID) + } + folded, err := open.inventory.PlanByID(ctx, newer.ID) + if err != nil || folded.State != inventory.PlanSuperseded || folded.Delivery == nil || + folded.Delivery.TakenOverBy != older.ID { + t.Fatalf("the newer batch reads %s %+v (%v)", folded.State, folded.Delivery, err) + } + if left, _ := open.inventory.MergesOf(ctx, newer.ID); len(left) != 0 { + t.Fatalf("the folded batch still answers %+v", left) + } + cutAt(t, open, t0.Add(100*time.Second)) + ws, bs := walks(t, open) + var cut []inventory.Plan + for _, w := range ws { + if w.State != inventory.PlanSuperseded { + cut = append(cut, w) + } + } + if len(cut) != 1 || len(bs) != 0 || cut[0].ID != older.ID || cut[0].CommitOf("novox/one") != "c3" || + cut[0].CommitOf("novox/two") != "c2" { + t.Fatalf("the folded batch's walk: %+v, %d batch(es)", cut, len(bs)) + } +} + +// With a walk open, a batch cut at once still waits for it, said queued behind it (walks stay one at a time, ADR +// 0276 decision 4); it is cut once that walk ended, with no window to wait for. +func TestACutAtOnceBatchQueuesBehindAnOpenWalk(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + hear(t, open, repoMerge("one", "c1", t0), t0) + cutAt(t, open, t0.Add(90*time.Second)) + ws, _ := walks(t, open) + if len(ws) != 1 || !ws[0].Open() { + t.Fatalf("no walk open: %+v", ws) + } + first := ws[0] + heard := t0.Add(100 * time.Second) + hear(t, open, lone(repoMerge("two", "c2", heard)), heard) + ws, bs := walks(t, open) + if len(ws) != 1 || len(bs) != 1 || bs[0].State != inventory.PlanQueued || bs[0].Delivery.Batch.Behind != first.ID || + !bs[0].Delivery.Batch.CutAtOnce { + t.Fatalf("a lone merge during a walk: %d walk(s), batches %+v", len(ws), bs) + } + if !strings.HasPrefix(batchWords(bs[0], heard), "queued behind "+first.ID) { + t.Fatalf("it reads %q", batchWords(bs[0], heard)) + } + first.State = inventory.PlanDone + if err := open.inventory.SavePlan(t.Context(), &first); err != nil { + t.Fatal(err) + } + cutAt(t, open, heard.Add(time.Second)) + ws, bs = walks(t, open) + if len(ws) != 2 || len(bs) != 0 || ws[0].Commit != "c2" { + t.Fatalf("the queued batch was not cut once the walk ended: %+v %+v", ws, bs) + } + if ws[0].Times == nil || ws[0].Times.WindowClosed == nil || !ws[0].Times.WindowClosed.Equal(heard) { + t.Fatalf("its window closed at %+v; want when its merge was heard, %s", ws[0].Times, heard) + } +} + +// A stopped walk's other merges are walked again in a batch they are related to, and a batch of their own when +// they share nothing with the one assembling (ADR 0299, ADR 0304 decision 1): never in a batch of another +// change, whose window the stop would close. +func TestAStopWalksTheOthersAgainInABatchOfTheirOwnWhenTheyShareNothing(t *testing.T) { + open := windowed(t) + endsItsLine(t) + asksRecorded(t) + ctx := t.Context() + cat := catalogueMerge("cat1aaaaaaaa", "app", t0) + one := fromHead(repoMerge("one", "one1bbbbbbbb", t0.Add(5*time.Second)), cat.Head) + hear(t, open, cat, t0.Add(time.Second)) + hear(t, open, one, t0.Add(6*time.Second)) + cutAt(t, open, t0.Add(2*time.Minute)) + ws, _ := walks(t, open) + walk := ws[0] + hear(t, open, repoMerge("two", "two1cccccccc", t0.Add(3*time.Minute)), t0.Add(3*time.Minute)) + _, bs := walks(t, open) + if len(bs) != 1 { + t.Fatalf("the merge heard while the walk ran is in %d batches", len(bs)) + } + other := bs[0] + _, said, err := stopWalkWithout(ctx, open.inventory, walk.ID, "mesh-delivery for jochen", "its group stopped", + []string{"novox/mesh-catalog@" + cat.Commit}, t0.Add(3*time.Minute+10*time.Second)) + if err != nil { + t.Fatal(err) + } + if said.WalkedAgainBy == "" || said.WalkedAgainBy == other.ID { + t.Fatalf("walked again by %q; want a batch of its own, not %s", said.WalkedAgainBy, other.ID) + } + b, _ := open.inventory.PlanByID(ctx, other.ID) + if b.Delivery.Batch == nil || b.Delivery.Batch.Closed || len(b.Delivery.Merges) != 1 { + t.Fatalf("the other change's batch reads %+v; want it untouched by the stop", b.Delivery) + } +} + +// A catalogue batch's cut folds in every walk waiting for its delivery's word (cutBatch's carry), so what that +// walk moves is the batch's too: a lone merge related to the waiting walk joins the batch assembling and waits for +// its window, rather than being cut at once as sharing nothing with what the batch's own merges move. +func TestALoneMergeRelatedToAWalkTheCutFoldsJoinsTheBatch(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + ctx := t.Context() + if err := open.inventory.RegisterModule(ctx, catalogue.Manifest{Module: "mesh-delivery", Version: "1", + Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}}, + inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/mesh-delivery", Ref: "main", + BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + if _, err := open.inventory.Assign(ctx, "anchor", "mesh-delivery"); err != nil { + t.Fatal(err) + } + hear(t, open, repoMerge("one", "c1", t0), t0) + cutAt(t, open, t0.Add(90*time.Second)) + if ws, _ := walks(t, open); len(ws) != 1 || !ws[0].Waiting() { + t.Fatalf("the walk does not wait for its word: %+v", ws) + } + hear(t, open, repoMerge("three", "c3", t0.Add(2*time.Minute)), t0.Add(2*time.Minute)) + hear(t, open, lone(repoMerge("one", "c2", t0.Add(2*time.Minute+5*time.Second))), t0.Add(2*time.Minute+5*time.Second)) + _, bs := walks(t, open) + if len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 || bs[0].Delivery.Batch.CutAtOnce { + t.Fatalf("a lone merge related to the waiting walk: batches %+v", bs) + } +} diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index f90999a8..01fde9fa 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -117,6 +117,11 @@ type PlanBatch struct { // the batch's walk does not build them. Closed bool `json:"closed,omitempty"` Withheld []string `json:"withheld,omitempty"` + // CutAtOnce says the batch's window closed when it was made: its merge shares nothing with a batch + // assembling and has no delivery group member still to merge, so it waits for no window (novox/hq ADR 0304 + // decision 1, ADR 0306 decisions 2 and 3). A merge that joins it later opens its window again. It is not + // Closed, which is a stop's. + CutAtOnce bool `json:"cut_at_once,omitempty"` } // OwnPath says the record is a batch of merges on the controller's own path.