From 8b3a9163d0ba5b75d14f05a8642be1a59f74b5ef Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 21:06:33 +0200 Subject: [PATCH] Hold a group's batch open for its last member and keep both kinds of member in one batch and one plan (issue 977, part C, ADR 0418) --- cmd/mesh-controller/batches.go | 111 +++++++++++++++- cmd/mesh-controller/batches_test.go | 14 +- cmd/mesh-controller/batches_watch.go | 50 ++++++- cmd/mesh-controller/group_one_batch_test.go | 140 ++++++++++++++++++++ cmd/mesh-controller/group_wait.go | 110 +++++++++------ cmd/mesh-controller/lone_merge_test.go | 76 +---------- cmd/mesh-controller/merge_window_bus.go | 4 +- cmd/mesh-controller/signals.go | 9 ++ cmd/mesh-controller/signals_test.go | 13 ++ internal/inventory/plans.go | 6 + 10 files changed, 406 insertions(+), 127 deletions(-) create mode 100644 cmd/mesh-controller/group_one_batch_test.go diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 5004b55b..beee8b63 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -181,7 +181,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } // **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) + waits, holding, err := groupWaitHolding(ctx, inv, m, deliverySeatHeld(entries)) if err != nil { return notNow(err) } @@ -237,6 +237,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T if added, err := inv.AddMerge(ctx, heard); err != nil || !added { return notNow(err) } + holdFor(&batch, m.Head, waits, holding, own, now) if err := keepBatch(ctx, inv, &batch, now, ""); err != nil { return notNow(err) } @@ -245,6 +246,43 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T return cutBatchesHeld(ctx, f.open, now) } +// holdFor keeps in a batch that a merge of a group joined whether that group still has a member to merge (ADR 0418 +// clause 3): held while it has, the window running again from the moment the last hold is cleared. A group with a +// merge of the controller's own path makes its batch an own-path one: such a plan waits for nobody's word, and is +// started by the controller when the group's batch closes (clause 6). +func holdFor(batch *inventory.Plan, head, why string, holding, own bool, now time.Time) { + if batch.Delivery == nil { + batch.Delivery = &inventory.PlanDelivery{} + } + if batch.Delivery.Batch == nil { + batch.Delivery.Batch = &inventory.PlanBatch{Own: own} + } + w := batch.Delivery.Batch + w.Own = w.Own || own + if head == "" { + return + } + if holding { + if w.Holds == nil { + w.Holds = map[string]string{} + } + w.Holds[head] = why + return + } + if _, held := w.Holds[head]; held { + holds := map[string]string{} + for h, y := range w.Holds { + if h != head { + holds[h] = y + } + } + w.Holds = holds + if len(holds) == 0 { + w.Holds, w.HeldAt = nil, now + } + } +} + // owedLate says a merge that reads as history is still the record's to answer (review of this change): a cut // marks every module it moved seen at that moment, so a merge of that branch // handed over late — made before the cut, heard after it — reads as history and was dropped. One that touches @@ -349,6 +387,15 @@ func batchToJoin(ctx context.Context, inv *inventory.Inventory, heard inventory. if err != nil { return inventory.Plan{}, "", err } + // **A group is one batch, whatever kinds its members are** (novox/hq ADR 0418 clause 6): a merge joins the batch + // of the other kind that holds a merge from its head. + if group, ok, err := batchOfGroup(ctx, inv, announcedOf(heard).Head, heard.Repository); err != nil { + return inventory.Plan{}, "", err + } else if ok && group.OwnPath() != own && group.Delivery != nil && group.Delivery.Batch != nil { + group.Delivery.Batch.CutAtOnce = false + return group, "its group " + announcedOf(heard).Head + " has a merge in " + group.ID + ", a batch of the other kind: " + + "a group is one batch", nil + } joins, whys, _, err := relatedBatches(ctx, inv, []inventory.BatchedMerge{heard}, own, "", entries, read, edges) if err != nil { return inventory.Plan{}, "", err @@ -572,18 +619,25 @@ func keepBatchBecause(ctx context.Context, inv *inventory.Inventory, b *inventor if b.Delivery == nil { b.Delivery = &inventory.PlanDelivery{} } - closes, latest := windowOf(merges, b.Created, window, atMost) + anchor := b.Created + if was := b.Delivery.Batch; was != nil && was.HeldAt.After(anchor) { + anchor = was.HeldAt // the maximum counts from the last hold's end (ADR 0418 clause 3) + } + closes, latest := windowOf(merges, anchor, window, atMost) b.Delivery.Merges = named var closed, cutAtOnce bool var withheld []string var waits string + var holds map[string]string + var heldAt time.Time if was := b.Delivery.Batch; was != nil { + holds, heldAt = was.Holds, was.HeldAt 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, BehindWhy: why, Own: b.OwnPath(), - Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits} + Closed: closed, Withheld: withheld, CutAtOnce: cutAtOnce, Waits: waits, Holds: holds, HeldAt: heldAt} b.State = inventory.PlanAssembling if behind != "" && windowClosed(*b, now) { b.State = inventory.PlanQueued @@ -619,13 +673,20 @@ func windowOf(merges []inventory.BatchedMerge, created time.Time, window, atMost } // 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. +// nothing (ADR 0304 decision 1), no merge for the window's length, or its maximum reached. **A batch holding a group +// with a member still to merge is closed by none of the last three** (novox/hq ADR 0418 clause 3). 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 || w.CutAtOnce || !now.Before(w.ClosesAt) || !now.Before(w.AtMost) + if w.Closed { + return true + } + if len(w.Holds) > 0 { + return false // a group's last member is still to merge: neither the window nor its maximum closes the batch + } + return 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 @@ -789,6 +850,11 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error { if err != nil { return err } + // **A hold is looked at again at every cut** (ADR 0418 clause 3): a group whose last member merged or closed + // since is released, and its window runs again from now. + if batches, err = releaseHolds(ctx, inv, batches, now); err != nil { + return err + } keepAll := func(behind string) error { for i := range batches { if err := keepBatch(ctx, inv, &batches[i], now, behind); err != nil { @@ -812,6 +878,41 @@ func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error { return cutUnrelated(ctx, open, batches, started, waiting, now) } +// releaseHolds asks mesh-delivery again about each group a batch holds, and clears the holds of those with no member +// left to merge. A group mesh-delivery cannot be asked about stays held: the safe error is to wait. It answers the +// batches as kept. +func releaseHolds(ctx context.Context, inv *inventory.Inventory, batches []inventory.Plan, now time.Time) ([]inventory.Plan, error) { + var entries []inventory.Entry + for i := range batches { + b := &batches[i] + if b.Delivery == nil || b.Delivery.Batch == nil || len(b.Delivery.Batch.Holds) == 0 { + continue + } + if entries == nil { + var err error + if entries, err = inv.Catalogued(ctx); err != nil { + return nil, err + } + } + changed := false + for head := range b.Delivery.Batch.Holds { + open, isGroup, err := openMembersOf(ctx, inv, deliverySeatHeld(entries), head, "") + if err != nil || (isGroup && len(open) > 0) { + continue + } + holdFor(b, head, "", false, false, now) + changed = true + fmt.Printf("%s: group %s has no member left to merge; its window runs again from now\n", b.ID, head) + } + if changed { + if err := keepBatch(ctx, inv, b, now, ""); err != nil { + return nil, err + } + } + } + return batches, nil +} + // cutUnrelated cuts the closed batches that share nothing with an open walk, each into a walk beside the open // ones, and says the others queued behind the first walk they are related to (novox/hq ADR 0407 decision 4, // issue 973). A batch cut in this look counts as open for the batches after it. diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 0c02f436..a5010630 100644 --- a/cmd/mesh-controller/batches_test.go +++ b/cmd/mesh-controller/batches_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "io" "os" "slices" @@ -91,15 +92,16 @@ func noGroups(t *testing.T) { 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. +// groupsWaiting stubs mesh-delivery's word on groups for a test of the merge window's own timing: each head named +// is one whose group mesh-delivery cannot say, so a merge from it waits for the window (ADR 0306 decision 2) and +// nothing holds the batch open; every other head names no group. A group with a member still to merge, which holds +// its batch open however long it takes (ADR 0418 clause 3), is groupStates'. 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, errors.New("the test's mesh-delivery does not say") } return nil, false, nil } @@ -107,12 +109,12 @@ func groupsWaiting(t *testing.T, heads ...string) { } // 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. +// before novox/hq ADR 0304: mesh-delivery cannot say any head's group. 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 + return nil, false, errors.New("the test's mesh-delivery does not say") } t.Cleanup(func() { deliveryGroupOf = was }) } diff --git a/cmd/mesh-controller/batches_watch.go b/cmd/mesh-controller/batches_watch.go index 9881c19e..ae933763 100644 --- a/cmd/mesh-controller/batches_watch.go +++ b/cmd/mesh-controller/batches_watch.go @@ -3,6 +3,8 @@ package main import ( "context" "fmt" + "sort" + "strings" "time" "github.com/novox/mesh-controller/internal/conditions" @@ -17,6 +19,9 @@ import ( const ( kindBatchNotCut = "batch-not-cut" kindBatchBehindWalk = "batch-behind-walk" + // kindGroupNotMerged is raised for a group whose last member has not merged by the merge window's maximum after + // its first (novox/hq ADR 0418 clause 3). + kindGroupNotMerged = "group-not-merged" ) // batchFacts is one batch not yet cut, as the watchdogs read it. @@ -28,6 +33,11 @@ type batchFacts struct { // tiers. behind string walkBound time.Duration + // holds are the groups the batch is held for, by head, with why; created is when the batch was made, the first + // merge of the group heard; atMostWindow is the maximum a window may take (the controller's setting). + holds map[string]string + created time.Time + atMostWindow time.Duration } // gatherBatches is every batch not yet cut, with the bound of the walk it waits behind. @@ -42,8 +52,10 @@ func gatherBatches(ctx context.Context, inv *inventory.Inventory, now time.Time) } var out []batchFacts for _, b := range batches { - f := batchFacts{id: b.ID, state: b.State, grouped: groupedWords(b)} + _, atMostWindow := mergeWindowOf() + f := batchFacts{id: b.ID, state: b.State, grouped: groupedWords(b), created: b.Created, atMostWindow: atMostWindow} if w := b.Delivery; w != nil && w.Batch != nil { + f.holds = w.Batch.Holds f.atMost, f.closed, f.behind = w.Batch.AtMost, w.Batch.ClosesAt, w.Batch.Behind if f.atMost.Before(f.closed) { f.closed = f.atMost @@ -67,7 +79,8 @@ func gatherBatches(ctx context.Context, inv *inventory.Inventory, now time.Time) func watchBatchesNotCut(f *signalFacts) []conditions.Observation { var out []conditions.Observation for _, b := range f.batches { - if b.state != inventory.PlanAssembling || b.atMost.IsZero() || f.now.Sub(b.atMost) <= batchLateAfter { + if b.state != inventory.PlanAssembling || b.atMost.IsZero() || f.now.Sub(b.atMost) <= batchLateAfter || + len(b.holds) > 0 { // a batch held for a group waits on a member, which has its own condition continue } late := f.now.Sub(b.atMost) @@ -105,3 +118,36 @@ func watchBatchesBehind(f *signalFacts) []conditions.Observation { } return out } + +// watchGroupsNotMerged is the condition of a batch held for a group whose last member has not merged by the merge +// window's maximum after the first member merged, naming the member and why (novox/hq ADR 0418 clause 3): nothing of +// the group is delivered while it waits. +func watchGroupsNotMerged(f *signalFacts) []conditions.Observation { + var out []conditions.Observation + for _, b := range f.batches { + if len(b.holds) == 0 || b.created.IsZero() || f.now.Sub(b.created) <= b.atMostWindow { + continue + } + heads := make([]string, 0, len(b.holds)) + for h := range b.holds { + heads = append(heads, h) + } + sort.Strings(heads) + var why []string + for _, h := range heads { + why = append(why, b.holds[h]) + } + waited := f.now.Sub(b.created) + out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: b.id, + Kind: kindGroupNotMerged, Severity: conditions.Warning, + Summary: fmt.Sprintf("%s; its first member merged %s ago, past the merge window's maximum (%s), and nothing "+ + "of the group is delivered while it waits (batch %s)", strings.Join(why, "; "), ago(waited), ago(b.atMostWindow), b.id), + Said: "a group waits for a member that has not merged", + Headline: "A change waits for the rest of its group to merge", + Explanation: fmt.Sprintf("The changes of a group are delivered together, once the last one is merged. "+ + "One has not merged in %s. Nothing is lost; the merged ones are kept and delivered with it.", + humanDuration(waited)), + Resolved: "The group's last change merged"}) + } + return out +} diff --git a/cmd/mesh-controller/group_one_batch_test.go b/cmd/mesh-controller/group_one_batch_test.go new file mode 100644 index 00000000..e792e40e --- /dev/null +++ b/cmd/mesh-controller/group_one_batch_test.go @@ -0,0 +1,140 @@ +package main + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// A group is one batch and one plan (novox/hq ADR 0418 clauses 3 and 6): the batch holding a merge of a group with a +// member still to merge does not close by the merge window, however long that member takes, and a group with a +// member of each kind is one batch whatever order its members are heard in. + +// groupStates stubs mesh-delivery's word on one group: its members and their states, which a test changes as the +// members merge. +func groupStates(t *testing.T, head string, states map[string]string) { + t.Helper() + was := deliveryGroupOf + deliveryGroupOf = func(_ context.Context, _ bool, asked string) ([]groupMember, bool, error) { + if asked != head { + return nil, false, nil + } + var members []groupMember + for repository, state := range states { + members = append(members, groupMember{Repository: repository, State: state}) + } + return members, true, nil + } + t.Cleanup(func() { deliveryGroupOf = was }) +} + +func TestAGroupsBatchWaitsForItsLastMemberHoweverLongItTakes(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + states := map[string]string{"novox/one": "ready", "novox/two": "ready"} + groupStates(t, "feat/g", states) + first := fromHead(repoMerge("one", "c1", t0), "feat/g") + hear(t, open, first, t0) + // Past the window and past its maximum, with the second member still to merge: one batch, no walk. + for _, after := range []time.Duration{90 * time.Second, 11 * time.Minute, 30 * time.Minute} { + cutAt(t, open, t0.Add(after)) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Batch.Holds) != 1 { + t.Fatalf("%s after the first member: %d walk(s), %d batch(es) %+v", after, len(ws), len(bs), bs) + } + } + // The last member merges: the window runs again from that merge. + at := t0.Add(30 * time.Minute) + states["novox/two"] = "merged" + hear(t, open, fromHead(repoMerge("two", "c2", at), "feat/g"), at.Add(time.Second)) + cutAt(t, open, at.Add(time.Second+89*time.Second)) + if ws, _ := walks(t, open); len(ws) != 0 { + t.Fatalf("cut before the window ran again: %+v", ws) + } + cutAt(t, open, at.Add(time.Second+90*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("after the window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } +} + +// A group with a member of each kind is one batch and one walk, in both orders of hearing, started without +// mesh-delivery's word. +func TestAGroupWithAMemberOfEachKindIsOneBatchAndOneWalk(t *testing.T) { + for name, ownFirst := range map[string]bool{"controller member first": true, "catalogue member first": false} { + t.Run(strings.ReplaceAll(name, " ", "_"), func(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + states := map[string]string{"novox/mesh-controller": "ready", "novox/mesh-catalog": "ready"} + groupStates(t, "feat/both-kinds", states) + 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) + states["novox/"+second.Repo] = "merged" + hear(t, open, second, t0.Add(10*time.Second)) + ws, bs := walks(t, open) + if len(ws) != 0 || len(bs) != 1 || len(bs[0].Delivery.Merges) != 2 { + t.Fatalf("both members heard: %d walk(s), %d batch(es) %+v; want one batch of both", len(ws), len(bs), bs) + } + cutAt(t, open, t0.Add(10*time.Second+90*time.Second)) + ws, bs = walks(t, open) + if len(ws) != 1 || len(bs) != 0 || ws[0].CommitOf("novox/mesh-controller") != "k1" || + ws[0].CommitOf("novox/mesh-catalog") != cat.Commit { + t.Fatalf("after the window: %d walk(s) %+v, %d batch(es)", len(ws), ws, len(bs)) + } + if ws[0].Delivery != nil && ws[0].Delivery.Awaits != "" { + t.Fatalf("a plan carrying the controller waits for %s's word", ws[0].Delivery.Awaits) + } + }) + } +} + +// A group whose last member has not merged ten minutes after its first raises a condition naming it, and the held +// batch is not the batch S18 reads as a cut the controller failed to make. +func TestAHeldBatchRaisesTheGroupConditionAndNotTheBatchNotCutOne(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + groupStates(t, "feat/g", map[string]string{"novox/one": "ready", "novox/two": "ready"}) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/g"), t0) + now := t0.Add(12 * time.Minute) + batches, err := gatherBatches(t.Context(), open.inventory, now) + if err != nil || len(batches) != 1 { + t.Fatalf("batches %+v: %v", batches, err) + } + f := &signalFacts{now: now, batches: batches} + if got := watchBatchesNotCut(f); len(got) != 0 { + t.Fatalf("a held batch is read as not cut: %+v", got) + } + got := watchGroupsNotMerged(f) + if len(got) != 1 || got[0].Kind != kindGroupNotMerged || !strings.Contains(got[0].Summary, "novox/two") || + !strings.Contains(got[0].Summary, "feat/g") { + t.Fatalf("the group's condition: %+v", got) + } + if early := watchGroupsNotMerged(&signalFacts{now: t0.Add(5 * time.Minute), batches: batches}); len(early) != 0 { + t.Fatalf("raised before the maximum: %+v", early) + } +} + +// The batch held for a group survives a restart: the hold is in the record. +func TestAHeldBatchSurvivesARestart(t *testing.T) { + open := windowed(t) + asksWithPaths(t) + groupStates(t, "feat/g", map[string]string{"novox/one": "ready", "novox/two": "ready"}) + hear(t, open, fromHead(repoMerge("one", "c1", t0), "feat/g"), t0) + reopened, err := open.inventory.Batches(t.Context()) + if err != nil || len(reopened) != 1 || len(reopened[0].Delivery.Batch.Holds) != 1 { + t.Fatalf("the record of the held batch: %+v %v", reopened, err) + } + cutAt(t, open, t0.Add(time.Hour)) + if ws, bs := walks(t, open); len(ws) != 0 || len(bs) != 1 || bs[0].State != inventory.PlanAssembling { + t.Fatalf("after an hour: %d walk(s), batches %+v", len(ws), bs) + } +} diff --git a/cmd/mesh-controller/group_wait.go b/cmd/mesh-controller/group_wait.go index 1be26410..b2384cf8 100644 --- a/cmd/mesh-controller/group_wait.go +++ b/cmd/mesh-controller/group_wait.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "strings" "time" @@ -116,75 +117,100 @@ func memberOf(s string) groupMember { } // 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). +// when it names no head, its head names no group, or every other member is merged or closed. // -// A member of another repository whose merge this controller already keeps in a batch not yet cut 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. +// A member of another repository whose merge this controller already keeps in a batch not yet cut, of either +// 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. **Kinds do not part a group** (novox/hq ADR 0418 clause 6): a controller member and a +// catalogue member share one batch. // // 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) { + why, _, err := groupWaitHolding(ctx, inv, m, held) + return why, err +} + +// groupWaitHolding is groupWait, and whether the wait is for a member still to merge: such a batch is held open +// however long it takes (ADR 0418 clause 3). A wait only because mesh-delivery did not say is no hold. +func groupWaitHolding(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, held bool) (string, bool, error) { if m.Head == "" { - return "", nil + return "", false, 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) + open, isGroup, err := openMembersOf(ctx, inv, held, m.Head, m.Owner+"/"+m.Repo) 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 + "window, the safe side", catalogue.DeliverySeat, m.Head, err), false, nil } - if !isGroup { - return "", nil + if !isGroup || len(open) == 0 { + return "", false, nil + } + return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), true, nil +} + +// openMembersOf are the members of the group a head names that are still to merge, in words: those whose pull +// request is open and whose merge no batch not yet cut keeps. self is a repository to leave out, the merge being heard. +func openMembersOf(ctx context.Context, inv *inventory.Inventory, held bool, head, self string) ([]string, bool, error) { + kept, err := keptFromHead(ctx, inv, head) + if err != nil { + return nil, false, err + } + members, isGroup, err := deliveryGroupOf(ctx, held, head) + if err != nil || !isGroup { + return nil, isGroup, err } var open []string for _, g := range members { - if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || same[strings.ToLower(g.Repository)] || - !openPullStates[g.State] { + if strings.EqualFold(g.Repository, self) || kept[strings.ToLower(g.Repository)] || !openPullStates[g.State] { continue } open = append(open, g.Repository+" ("+g.State+")") } - if len(open) == 0 { - return "", nil - } - return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), nil + sort.Strings(open) + return open, true, 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) { +// keptFromHead are the repositories (lower case) of the merges from a head that batches not yet cut hold, of +// either kind. +func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string) (map[string]bool, error) { batches, err := inv.Batches(ctx) if err != nil { - return nil, nil, err + return nil, err } - same := map[string]bool{} - var other []string + kept := map[string]bool{} for _, b := range batches { merges, err := inv.MergesOf(ctx, b.ID) if err != nil { - return nil, nil, err + return 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)) + if announcedOf(k).Head == head { + kept[strings.ToLower(k.Repository)] = true } } } - return same, other, nil + return kept, nil +} + +// batchOfGroup is the batch not yet cut that holds a merge of another repository from the same head: the group's +// batch, whatever kind it is. +func batchOfGroup(ctx context.Context, inv *inventory.Inventory, head, repository string) (inventory.Plan, bool, error) { + if head == "" { + return inventory.Plan{}, false, nil + } + batches, err := inv.Batches(ctx) + if err != nil { + return inventory.Plan{}, false, err + } + for _, b := range batches { + merges, err := inv.MergesOf(ctx, b.ID) + if err != nil { + return inventory.Plan{}, false, err + } + for _, k := range merges { + if announcedOf(k).Head == head && !strings.EqualFold(k.Repository, repository) { + return b, true, nil + } + } + } + return inventory.Plan{}, false, nil } diff --git a/cmd/mesh-controller/lone_merge_test.go b/cmd/mesh-controller/lone_merge_test.go index 344c12a5..c2bb3df6 100644 --- a/cmd/mesh-controller/lone_merge_test.go +++ b/cmd/mesh-controller/lone_merge_test.go @@ -99,16 +99,19 @@ func TestARelatedMergeJoinsTheBatchAndWaitsForTheWindow(t *testing.T) { func TestAMergeOfAGroupWaitsEvenAlone(t *testing.T) { open := windowed(t) asksWithPaths(t) - groupsWaiting(t, "feat/x") + states := map[string]string{"novox/one": "ready", "novox/two": "ready", "novox/elsewhere": "ready"} + groupStates(t, "feat/x", states) 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)") { + if words := batchWords(bs[0], t0); !strings.Contains(words, "group feat/x has a member still to merge: novox/elsewhere (ready), novox/two (ready)") { t.Fatalf("it reads %q", words) } + // Its members merge: the batch is held for the last, and the window runs again from that merge. + delete(states, "novox/elsewhere") + states["novox/two"] = "merged" 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) @@ -331,70 +334,3 @@ func TestALoneMergeRelatedToAWalkTheCutFoldsJoinsTheBatch(t *testing.T) { 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/cmd/mesh-controller/merge_window_bus.go b/cmd/mesh-controller/merge_window_bus.go index 84ff718e..5b4cb0e3 100644 --- a/cmd/mesh-controller/merge_window_bus.go +++ b/cmd/mesh-controller/merge_window_bus.go @@ -54,8 +54,8 @@ func windowClockOf() windowClock { // most, each at least a second (the least a key can live). Not armed for a batch whose window is closed already // (by a stop, by its time, or because it is cut at once) or that is no batch. func windowDeadlines(b inventory.Plan, now time.Time) (quiet, most time.Duration, armed bool) { - if b.Delivery == nil || b.Delivery.Batch == nil || windowClosed(b, now) { - return 0, 0, false + if b.Delivery == nil || b.Delivery.Batch == nil || windowClosed(b, now) || len(b.Delivery.Batch.Holds) > 0 { + return 0, 0, false // a held batch has no clock: its group's last member closes it (ADR 0418 clause 3) } // Rounded up to a whole second, at least one: the server rounds a key's time to live down to whole seconds, so // a fraction would close the key before the record's time and the look would cut nothing. diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index a0a5b560..e086e5d9 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -225,6 +225,15 @@ var signalsTable = []signalRow{ newest: func(f *signalFacts) time.Time { return newestOf(f.batches, func(b batchFacts) time.Time { return b.closed }) }}, + {Row: "S22", Signal: "the last member of a delivery group merges", Emitter: "mesh-delivery's group, held by the controller's merge window", + Trigger: "each batch held for a group with a member still to merge (novox/hq ADR 0418)", + Bound: "merge-window-at-most from the group's first merge: naming the member and why it did not merge; " + + "nothing of the group is delivered while it waits", + Kind: kindGroupNotMerged, Severity: conditions.Warning, Phase: 3, + needs: func(f *signalFacts) error { return f.plansErr }, watch: watchGroupsNotMerged, + newest: func(f *signalFacts) time.Time { + return newestOf(f.batches, func(b batchFacts) time.Time { return b.created }) + }}, {Row: "S20", Signal: "a send refused for the assignment generation it was composed from", Emitter: "node-engine", Trigger: "each refusal (novox/hq issue 234)", Bound: "none: raised at the first refusal, naming its sender, the generation it came from and the one the " + diff --git a/cmd/mesh-controller/signals_test.go b/cmd/mesh-controller/signals_test.go index cde92547..efa2ae7e 100644 --- a/cmd/mesh-controller/signals_test.go +++ b/cmd/mesh-controller/signals_test.go @@ -147,6 +147,19 @@ var suppressions = map[string]suppression{ atMost: f.now.Add(-61 * time.Second), closed: f.now.Add(-61 * time.Second)}} }, }, + // A batch held for a group whose last member has not merged by the window's maximum after the first (ADR 0418). + "S22": { + inside: func(f *signalFacts) { + f.batches = []batchFacts{{id: "plan-3", state: inventory.PlanAssembling, grouped: "novox/app@c0ffee11", + holds: map[string]string{"feat/g": "group feat/g has a member still to merge: novox/two (ready)"}, + created: f.now.Add(-9 * time.Minute), atMostWindow: 10 * time.Minute}} + }, + past: func(f *signalFacts) { + f.batches = []batchFacts{{id: "plan-3", state: inventory.PlanAssembling, grouped: "novox/app@c0ffee11", + holds: map[string]string{"feat/g": "group feat/g has a member still to merge: novox/two (ready)"}, + created: f.now.Add(-11 * time.Minute), atMostWindow: 10 * time.Minute}} + }, + }, // A batch queued behind an open walk past that walk's bound, named. "S19": { inside: func(f *signalFacts) { diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index 4f983b37..ae7e9fb9 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -139,6 +139,12 @@ type PlanBatch struct { // 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"` + // Holds are the groups whose last member is still to merge, by head branch name, each with why in words + // (novox/hq ADR 0418 clause 3): the batch does not close by the merge window while one is held, however long the + // member takes. Cleared when every member is merged or closed; the window then runs again from that moment. + Holds map[string]string `json:"holds,omitempty"` + // HeldAt is when the last hold was cleared: the maximum is counted from it, not from the batch's making. + HeldAt time.Time `json:"held_at,omitempty"` } // OwnPath says the record is a batch of merges on the controller's own path.