diff --git a/cmd/mesh-controller/catch_up.go b/cmd/mesh-controller/catch_up.go index b960e853..f3f40bb7 100644 --- a/cmd/mesh-controller/catch_up.go +++ b/cmd/mesh-controller/catch_up.go @@ -14,6 +14,7 @@ import ( "sync" "time" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/asks" @@ -102,7 +103,10 @@ func readTrunkMerges(ctx context.Context, owner, repo, branch string) ([]trunkMe break } for _, c := range commits { - if c.SecondParent != "" { + // A merge of the trunk is a pull request's merge into it. A commit with a second parent that is none is a merge made + // inside some branch ("Merge remote-tracking branch 'origin/main' into fix/..."), which the trunk's history lists too and + // which no Approve ever covered: read as a gap it would raise a condition for nothing a person decides. + if c.SecondParent != "" && mergesIntoTrunk(c.Subject, branch) { merges = append(merges, trunkMerge{Commit: c.Commit, Head: c.SecondParent, Subject: c.Subject, At: c.At}) } } @@ -110,6 +114,23 @@ func readTrunkMerges(ctx context.Context, owner, repo, branch string) ([]trunkMe return lastMerges(merges), nil } +// sinceTheRecord are the merges that can be gaps (novox/hq ADR 0463 §2): those made since the controller's record of merges begins on +// the mesh (first, the earliest merge it holds of any repository and branch, less mergeGrace as owedLate judges). A merge before it is +// history on every trunk, and a trunk with no merge kept of its own still has the merges made since as gaps. With no merge kept +// anywhere there is no record to have a gap in. +func sinceTheRecord(merges []trunkMerge, kept bool, first time.Time) []trunkMerge { + if !kept { + return nil + } + var out []trunkMerge + for _, m := range merges { + if !m.At.Before(first.Add(-mergeGrace)) { + out = append(out, m) + } + } + return out +} + // lastMerges is the newest catchUpWindow of merges, given newest first, oldest first. func lastMerges(newestFirst []trunkMerge) []trunkMerge { if len(newestFirst) > catchUpWindow { @@ -135,6 +156,12 @@ func pullOfSubject(subject string) (title string, number int, ok bool) { return m[1], n, true } +// mergesIntoTrunk says a commit's subject is a pull request's merge into this trunk. +func mergesIntoTrunk(subject, branch string) bool { + m := mergeSubject.FindStringSubmatch(strings.TrimSpace(subject)) + return m != nil && m[4] == branch +} + // headBranchOfSubject is the branch a merge commit's subject says was merged. func headBranchOfSubject(subject string) string { if m := mergeSubject.FindStringSubmatch(strings.TrimSpace(subject)); m != nil { @@ -267,6 +294,11 @@ var routerAskIDs = func(ctx context.Context, inv catalogueReader, asker string) if err != nil { return nil, err } + return askIDsOf(ctx, kv, asker) +} + +// askIDsOf are the ids of the asks an asker has in the router's bucket. +func askIDsOf(ctx context.Context, kv jetstream.KeyValue, asker string) ([]string, error) { lister, err := kv.ListKeysFiltered(ctx, asker+".>") if err != nil { return nil, err @@ -275,7 +307,9 @@ var routerAskIDs = func(ctx context.Context, inv catalogueReader, asker string) for key := range lister.Keys() { ids = append(ids, strings.TrimPrefix(key, asker+".")) } - if err := lister.Stop(); err != nil { + // The lister stops its own watch when it has sent the last key, so a stop of ours finds it stopped: nats answers that with + // "invalid subscription", which is no failure of the read (issue 1336). Any other error stays one. + if err := lister.Stop(); err != nil && !errors.Is(err, nats.ErrBadSubscription) { return nil, err } return ids, nil @@ -437,6 +471,17 @@ func (f following) catchUp(ctx context.Context, only *caughtTrunk) (later bool) fmt.Printf("catch-up: the catalogue cannot be read, so no merge is caught up now: %v\n", err) return true } + kept, first, err := inv.EarliestKept(ctx) + if err != nil { + fmt.Printf("catch-up: what the controller kept cannot be read, so no merge is caught up now: %v\n", err) + return true + } + if kept { + fmt.Printf("catch-up: the controller's record of merges begins %s; a merge made before it, less %s, is history on every trunk\n", + first.UTC().Format(time.RFC3339), mergeGrace) + } else { + fmt.Println("catch-up: the controller holds no merge of any trunk yet, so none can have been missed") + } trunks := trunksOfCatalogue(newSourceForms(ctx, inv), entries) if only != nil { trunks = slices.DeleteFunc(trunks, func(t caughtTrunk) bool { @@ -444,14 +489,14 @@ func (f following) catchUp(ctx context.Context, only *caughtTrunk) (later bool) }) } for _, t := range trunks { - if f.catchUpTrunk(ctx, t) { + if f.catchUpTrunk(ctx, t, kept, first) { later = true } } return later } -func (f following) catchUpTrunk(ctx context.Context, t caughtTrunk) (later bool) { +func (f following) catchUpTrunk(ctx context.Context, t caughtTrunk, kept bool, first time.Time) (later bool) { inv := f.open.inventory repository := t.repository() merges, err := readTrunkMerges(ctx, t.Owner, t.Repo, t.Branch) @@ -461,6 +506,11 @@ func (f following) catchUpTrunk(ctx context.Context, t caughtTrunk) (later bool) "or the next merge heard there does it: %v\n", repository, t.Branch, err) return true } + // **A merge made before the controller's record of merges begins is history, never a gap** (novox/hq ADR 0463 §2). + read := len(merges) + merges = sinceTheRecord(merges, kept, first) + fmt.Printf("catch-up: %s %s: %d of the last %d pull request merges were made since the record began\n", repository, t.Branch, + len(merges), read) unwarranted := map[string]bool{} for _, tm := range merges { id := fmt.Sprintf("%s_%.8s", repository, tm.Commit) diff --git a/cmd/mesh-controller/catch_up_integration_test.go b/cmd/mesh-controller/catch_up_integration_test.go index ac9cf58e..c0b698fc 100644 --- a/cmd/mesh-controller/catch_up_integration_test.go +++ b/cmd/mesh-controller/catch_up_integration_test.go @@ -78,6 +78,13 @@ func aTrunkOf(t *testing.T, repo string, modules ...string) *aTrunk { return tr } +// aKeptMergeBefore is a merge the controller heard before the merges of a test: where its record begins, so what a trunk holds from +// then on can be a gap (novox/hq ADR 0463 §2). +func aKeptMergeBefore(t *testing.T, open *stores) { + t.Helper() + hear(t, open, catalogueMerge("mesh-catalog-seed", "app", t0.Add(-time.Hour)), t0.Add(-time.Hour)) +} + func mergeKnown(t *testing.T, open *stores, commit string) bool { t.Helper() _, known, err := open.inventory.MergeOf(t.Context(), "novox/mesh-catalog", commit) @@ -197,6 +204,7 @@ func TestAGapWithNoApproveRaisesOneConditionKeptOpenAndClearedWhenPlanned(t *tes open := windowed(t) everyHeadWaits(t) asksRecorded(t) + aKeptMergeBefore(t, open) keeper, _ := withConditionsInMemory(t) tr := aTrunkOf(t, "mesh-catalog", "notes") tr.approved["mesh-catalog-c01"] = false @@ -242,6 +250,7 @@ func TestAForgeThatDoesNotAnswerLeavesTheCatchUpForTheNextStart(t *testing.T) { open := windowed(t) everyHeadWaits(t) asksRecorded(t) + aKeptMergeBefore(t, open) tr := aTrunkOf(t, "mesh-catalog", "notes") tr.silent = true if later := (following{open: open}).catchUp(t.Context(), nil); !later { @@ -262,6 +271,7 @@ func TestAGapThatAddsAModuleIsHeardOnItsWarrantAndRaisedWithout(t *testing.T) { open := windowed(t) everyHeadWaits(t) asksRecorded(t) + aKeptMergeBefore(t, open) keeper, _ := withConditionsInMemory(t) tr := aTrunkOf(t, "mesh-catalog", "newmodule") tr.approved["mesh-catalog-c01"] = false @@ -289,3 +299,54 @@ func TestAGapThatAddsAModuleIsHeardOnItsWarrantAndRaisedWithout(t *testing.T) { t.Fatal("a merge adding a module was not heard once its Approve was on record") } } + +// A trunk with merges from before the controller began keeping merges, and merges since: only the later ones are gaps. +func TestAMergeFromBeforeTheRecordBeganIsHistoryAndNoCondition(t *testing.T) { + open := windowed(t) + everyHeadWaits(t) + asksRecorded(t) + keeper, _ := withConditionsInMemory(t) + tr := aTrunkOf(t, "mesh-catalog", "notes", "notes", "notes") + tr.approved["mesh-catalog-c01"], tr.approved["mesh-catalog-c02"], tr.approved["mesh-catalog-c03"] = false, false, false + // The record begins at the second merge: the first is two days older. + tr.merges[0].At = t0.Add(-48 * time.Hour) + hear(t, open, catalogueMerge("mesh-catalog-c02", "notes", t0.Add(time.Minute)), t0.Add(time.Minute)) + delete(tr.approved, "mesh-catalog-c02") + (following{open: open}).catchUp(t.Context(), nil) + // The first merge (two days before the record) was not even read from the forge: only the third, which is unknown, was. + if tr.events != 1 { + t.Fatalf("%d merge(s) were read from the forge: the one from before the record is history and is not read", tr.events) + } + all, err := keeper.Open(t.Context()) + if err != nil { + t.Fatal(err) + } + var raised int + for _, c := range all { + if c.Kind == kindMergeNotHeard { + raised++ + } + } + if raised != 1 { + t.Fatalf("%d condition(s) raised: only the merge since the record began (the third) is a gap", raised) + } +} + +// novox/hq ADR 0463 §2: the record begins at the earliest merge kept on ANY trunk. A trunk the controller kept no merge of has the +// merges made since as gaps (mesh-delivery #1 on novox), and the merges made before as history. +func TestATrunkWithNoKeptMergeHasItsLaterMergesAsGapsAndItsEarlierOnesAsHistory(t *testing.T) { + open := windowed(t) + everyHeadWaits(t) + asksRecorded(t) + // The record begins on another trunk, at t0 less an hour. + hear(t, open, repoMerge("one", "c-one", t0.Add(-time.Hour)), t0.Add(-time.Hour)) + tr := aTrunkOf(t, "mesh-catalog", "notes", "notes") + tr.merges[0].At = t0.Add(-72 * time.Hour) // before the record: history + (following{open: open}).catchUp(t.Context(), nil) + if mergeKnown(t, open, "mesh-catalog-c01") { + t.Fatal("a merge from before the record began was heard") + } + if !mergeKnown(t, open, "mesh-catalog-c02") { + t.Fatal("a merge since the record began, on a trunk with no merge kept, was not caught up") + } +} diff --git a/cmd/mesh-controller/catch_up_test.go b/cmd/mesh-controller/catch_up_test.go index 2e2ec779..b120a1cd 100644 --- a/cmd/mesh-controller/catch_up_test.go +++ b/cmd/mesh-controller/catch_up_test.go @@ -34,7 +34,8 @@ func TestOnlyTheLastFiftyMergesOfATrunkAreRead(t *testing.T) { pages := map[int][]trunkCommit{} var all []trunkCommit // newest first: 51 merges, a plain commit between each pair for i := 51; i >= 1; i-- { - all = append(all, trunkCommit{Commit: fmt.Sprintf("merge-%02d", i), SecondParent: fmt.Sprintf("head-%02d", i)}, + all = append(all, trunkCommit{Commit: fmt.Sprintf("merge-%02d", i), SecondParent: fmt.Sprintf("head-%02d", i), + Subject: fmt.Sprintf("Merge pull request 'change %d' (#%d) from fix/x into main", i, i)}, trunkCommit{Commit: fmt.Sprintf("plain-%02d", i)}) } for i := 0; i < len(all); i += 50 { @@ -264,7 +265,8 @@ func TestAShortTrunkIsReadWholeAndAnEmptyOneIsNoMerges(t *testing.T) { if page > 1 { return nil, nil } - return []trunkCommit{{Commit: "m2", SecondParent: "h2"}, {Commit: "p"}, {Commit: "m1", SecondParent: "h1"}}, nil + return []trunkCommit{{Commit: "m2", SecondParent: "h2", Subject: "Merge pull request 'two' (#2) from fix/b into main"}, + {Commit: "p"}, {Commit: "m1", SecondParent: "h1", Subject: "Merge pull request 'one' (#1) from fix/a into main"}}, nil } got, err := readTrunkMerges(t.Context(), "novox", "x", "main") if err != nil || len(got) != 2 || got[0].Commit != "m1" || got[1].Commit != "m2" { @@ -292,3 +294,53 @@ func TestTheConditionOfAMergeNotHeardIsInPlainWords(t *testing.T) { t.Fatalf("the condition's words are not plain: %s\n%q\n%q\n%q", why, c.Headline, c.Explanation, c.Needs) } } + +// The trunk's history lists the merges made inside a pull request's branch too. On novox (2026-10-20) 15 of the 15 non-pull-request +// "gaps" of mesh-catalog and mesh-controller were such merges; only a pull request's merge into this trunk is a merge to catch up. +func TestAMergeMadeInsideABranchIsNoMergeOfTheTrunk(t *testing.T) { + was := trunkCommitsPage + t.Cleanup(func() { trunkCommitsPage = was }) + trunkCommitsPage = func(_ context.Context, _, _, _ string, page int) ([]trunkCommit, error) { + if page > 1 { + return nil, nil + } + return []trunkCommit{ + {Commit: "c0880c4d", SecondParent: "b9104454", Subject: "Merge pull request 'Read the grant file' (#378) from fix/1332 into main"}, + {Commit: "6394a62a", SecondParent: "x1", Subject: "Merge remote-tracking branch 'origin/main' into fix/1081-withdraw"}, + {Commit: "896f1d28", SecondParent: "x2", Subject: "Merge feat/agents-write-git-ssh into feat/agents-write-git-ssh-f (issue 955 F)"}, + {Commit: "4d03755c", SecondParent: "x3", Subject: "Bring the branch up to main after #354 (issue 1047)"}, + {Commit: "other", SecondParent: "x4", Subject: "Merge pull request 'Elsewhere' (#5) from fix/y into nox-mesh"}, + }, nil + } + got, err := readTrunkMerges(t.Context(), "novox", "mesh-catalog", "main") + if err != nil || len(got) != 1 || got[0].Commit != "c0880c4d" { + t.Fatalf("read %+v %v: only keycloak's pull request merge into main is a merge to catch up", got, err) + } +} + +// novox/hq ADR 0463 §2: the record of merges begins at the earliest merge the controller holds on any trunk. A merge before it is +// history on every trunk; a trunk with no merge kept of its own still has the merges made since as gaps (mesh-delivery #1 on novox). +func TestOnlyTheMergesSinceTheRecordBeganAreGapsOnEveryTrunk(t *testing.T) { + first := time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC) // the first merge it heard, on some trunk of the mesh + trunk := func(prefix string) []trunkMerge { + return []trunkMerge{ + {Commit: prefix + "-old-1", At: first.Add(-30 * 24 * time.Hour)}, + {Commit: prefix + "-old-2", At: first.Add(-2 * time.Hour)}, + {Commit: prefix + "-within-grace", At: first.Add(-mergeGrace / 2)}, + {Commit: prefix + "-since", At: first.Add(24 * time.Hour)}, + } + } + // Two trunks, one of which the controller has kept no merge of: both read by the one starting point. + for _, prefix := range []string{"kept-trunk", "trunk-with-no-kept-merge"} { + var commits []string + for _, m := range sinceTheRecord(trunk(prefix), true, first) { + commits = append(commits, m.Commit) + } + if want := prefix + "-within-grace " + prefix + "-since"; strings.Join(commits, " ") != want { + t.Errorf("%s: the gaps are %v, want %q: earlier ones are history on every trunk, later ones are gaps", prefix, commits, want) + } + } + if none := sinceTheRecord(trunk("x"), false, time.Time{}); len(none) != 0 { + t.Errorf("with no merge kept anywhere there are %d gap(s), but no record to have one in", len(none)) + } +} diff --git a/cmd/mesh-controller/router_ids_integration_test.go b/cmd/mesh-controller/router_ids_integration_test.go new file mode 100644 index 00000000..2f238fe9 --- /dev/null +++ b/cmd/mesh-controller/router_ids_integration_test.go @@ -0,0 +1,52 @@ +//go:build integration + +package main + +import ( + "context" + "slices" + "testing" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// Issue 1336: the catch-up lists the ids of an asker's asks in the router's bucket. The key lister stops its own watch when the last +// key is read, and a second stop answers "nats: invalid subscription"; read as a failure, it made every warrant unreadable on novox +// (2026-10-20, 110 gaps named, none planned). +func TestTheAskIDsOfAnAskerAreReadFromTheRoutersBucketWithoutAFalseFailure(t *testing.T) { + conn, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(conn.Close) + js, err := jetstream.New(conn) + if err != nil { + t.Fatal(err) + } + ctx := t.Context() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "router_asks_test"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.DeleteKeyValue(context.Background(), "router_asks_test") }) + for _, key := range []string{"mesh-delivery.merge-378-b9104454f9a5-1", "mesh-delivery.merge-377-aaaaaaaaaaaa-1", + "mesh-delivery-teller.merge-feat_r4b_sdk_users-ba59b03220f8-989afe0a3bcf", "keycloak.something"} { + if _, err := kv.Put(ctx, key, []byte(`{}`)); err != nil { + t.Fatal(err) + } + } + got, err := askIDsOf(ctx, kv, "mesh-delivery") + if err != nil { + t.Fatalf("the ids of an asker's asks could not be read: %v", err) + } + slices.Sort(got) + if want := []string{"merge-377-aaaaaaaaaaaa-1", "merge-378-b9104454f9a5-1"}; !slices.Equal(got, want) { + t.Fatalf("read %v, want %v: the teller's and another module's asks are not mesh-delivery's", got, want) + } + if none, err := askIDsOf(ctx, kv, "nobody"); err != nil || len(none) != 0 { + t.Fatalf("an asker with no asks read as %v, %v", none, err) + } +} diff --git a/internal/inventory/batches.go b/internal/inventory/batches.go index e870a676..7ca63dce 100644 --- a/internal/inventory/batches.go +++ b/internal/inventory/batches.go @@ -75,6 +75,16 @@ func (i *Inventory) KeptOnBranch(ctx context.Context, repository, branch string) return i.chainKeptOnBranch(ctx, c, repository, branch) } +// EarliestKept says the controller keeps a merge of some repository's branch, and when it heard the first of all: where its record of +// merges begins on the whole mesh (novox/hq ADR 0463). A merge made before then is history on every trunk. +func (i *Inventory) EarliestKept(ctx context.Context) (bool, time.Time, error) { + c, err := i.theChain(ctx) + if err != nil { + return false, time.Time{}, err + } + return i.chainEarliestKept(ctx, c) +} + // AloneMerges is every merge waiting to be walked on its own commit, the newest first (ADR 0276 decision 3). func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) { c, err := i.theChain(ctx) diff --git a/internal/inventory/chain_rollouts.go b/internal/inventory/chain_rollouts.go index 6c242b6d..3988f558 100644 --- a/internal/inventory/chain_rollouts.go +++ b/internal/inventory/chain_rollouts.go @@ -437,6 +437,24 @@ func (i *Inventory) chainKeptOnBranch(ctx context.Context, c *chain.Chain, repos return true, first.UTC(), nil } +// chainEarliestKept is when the controller heard the first merge it holds a record of, on any repository and branch. +func (i *Inventory) chainEarliestKept(ctx context.Context, c *chain.Chain) (bool, time.Time, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return false, time.Time{}, err + } + var first time.Time + for _, h := range w { + if first.IsZero() || h.m.Heard.Before(first) { + first = h.m.Heard + } + } + if first.IsZero() { + return false, time.Time{}, nil + } + return true, first.UTC(), nil +} + func (i *Inventory) chainAloneMerges(ctx context.Context, c *chain.Chain) ([]BatchedMerge, error) { w, err := i.chainWinners(ctx, c) if err != nil { diff --git a/internal/inventory/chain_rollouts_test.go b/internal/inventory/chain_rollouts_test.go index fffc8576..d7e8b7f3 100644 --- a/internal/inventory/chain_rollouts_test.go +++ b/internal/inventory/chain_rollouts_test.go @@ -347,3 +347,28 @@ func TestThePlansAreReadFromAViewKeptByOneWatch(t *testing.T) { return len(open) == 3 }) } + +// Where the record of merges begins is the earliest merge kept on any repository and branch (novox/hq ADR 0463 §2). +func TestTheRecordOfMergesBeginsAtTheEarliestMergeKeptOnAnyTrunk(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + if kept, _, err := inv.chainEarliestKept(ctx, c); err != nil || kept { + t.Fatalf("an inventory with no merge kept says it keeps one: %v %v", kept, err) + } + if err := inv.chainSavePlans(ctx, c, []*Plan{aBatch("plan-1", plansAt), aBatch("plan-2", plansAt)}, nil); err != nil { + t.Fatal(err) + } + later := aMerge("m-late", "plan-1") + later.Repository, later.Branch, later.Heard = "novox/mesh-sdk", "main", plansAt.Add(48*time.Hour) + earlier := aMerge("m-early", "plan-2") + earlier.Repository, earlier.Branch, earlier.Heard = "novox/photos", "nox-mesh", plansAt.Add(time.Hour) + for _, m := range []BatchedMerge{later, earlier} { + if added, err := inv.chainAddMerge(ctx, c, m); err != nil || !added { + t.Fatalf("%s: %v %v", m.Commit, added, err) + } + } + kept, first, err := inv.chainEarliestKept(ctx, c) + if err != nil || !kept || !first.Equal(plansAt.Add(time.Hour)) { + t.Fatalf("the record begins at %v (%v, %v), want %v", first, kept, err, plansAt.Add(time.Hour)) + } +}