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)

This commit is contained in:
2026-10-17 21:06:33 +02:00
parent e43d52c1ea
commit 8b3a9163d0
10 changed files with 406 additions and 127 deletions
+106 -5
View File
@@ -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.
+8 -6
View File
@@ -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 })
}
+48 -2
View File
@@ -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
}
+140
View File
@@ -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)
}
}
+68 -42
View File
@@ -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
}
+6 -70
View File
@@ -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)
}
})
}
}
+2 -2
View File
@@ -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.
+9
View File
@@ -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 " +
+13
View File
@@ -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) {
+6
View File
@@ -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.