Begin the record of merges at the earliest merge kept on any trunk, not per trunk (ADR 0463 section 2, issue 1336)

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)

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")
+	}
+}

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))
+	}
+}

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)
+	}
+}

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)

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 {

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))
+	}
+}