diff --git a/cmd/mesh-controller/forge_merge_bus_test.go b/cmd/mesh-controller/forge_merge_bus_test.go new file mode 100644 index 00000000..df56c4e9 --- /dev/null +++ b/cmd/mesh-controller/forge_merge_bus_test.go @@ -0,0 +1,31 @@ +//go:build integration + +package main + +import ( + "context" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/novox/mesh-controller/internal/broker" +) + +// sayForgeMerge is the forge bridge saying a merge on its stream FORGE, which it makes when missing. +func sayForgeMerge(ctx context.Context, t *testing.T, js *broker.JetStream, repo string, number int, head, merge string) { + t.Helper() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + if _, err := api.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: forgeStream, + Subjects: []string{process.ModuleStreamSubject(sdkdelivery.ForgeBridgeModule, "forge.>")}}); err != nil { + t.Fatal(err) + } + subject, body := forgeMerged(t, repo, number, head, merge) + if _, err := api.Publish(ctx, subject, body); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/mesh-controller/forge_merge_test.go b/cmd/mesh-controller/forge_merge_test.go new file mode 100644 index 00000000..0c9bf437 --- /dev/null +++ b/cmd/mesh-controller/forge_merge_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeForge is the forge bridge's stream as forgeMergeAt reads it: messages by sequence, each on a subject. +type fakeForge struct { + subjects []string + data [][]byte +} + +func (f fakeForge) last(_ context.Context, subject string) (uint64, []byte, error) { + for i := len(f.subjects) - 1; i >= 0; i-- { + if f.subjects[i] == subject { + return uint64(i + 1), f.data[i], nil + } + } + return 0, nil, jetstream.ErrMsgNotFound +} + +func (f fakeForge) next(_ context.Context, subject string, from uint64) (uint64, []byte, error) { + for i := int(from) - 1; i < len(f.subjects); i++ { + if i >= 0 && f.subjects[i] == subject { + return uint64(i + 1), f.data[i], nil + } + } + return 0, nil, jetstream.ErrMsgNotFound +} + +func forgeMerged(t *testing.T, repo string, number int, head, merge string) (string, []byte) { + t.Helper() + owner, name, _ := strings.Cut(repo, "/") + subject, err := forgeMergeSubject(owner, name) + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(sdkdelivery.Forge{V: 1, Kind: sdkdelivery.ForgeMerged, Repository: repo, Branch: "main", + Number: number, Head: head, Merge: merge, Root: "t0", Trees: map[string]string{".": "t1"}}) + if err != nil { + t.Fatal(err) + } + return subject, body +} + +// **A moved module's merge is read from the forge bridge's stream** (sweep 2026-10-30, issue 774): the controller keeps +// no copy of the forge's merges, so its registration finds the pull request and head of a merge commit on FORGE, the +// newest first and then every merge the stream keeps; a merge of another repository or commit is never taken for it. +func TestAMergeIsFoundOnTheForgeBridgesStream(t *testing.T) { + var f fakeForge + add := func(repo string, number int, head, merge string) { + s, b := forgeMerged(t, repo, number, head, merge) + f.subjects, f.data = append(f.subjects, s), append(f.data, b) + } + add("novox/mesh-host", 6, "h6", "c6c6c6c6c6") + add("novox/mesh-catalog", 9, "h9", "c7c7c7c7c7") + add("novox/mesh-host", 7, "h7", "c7c7c7c7c7") + f.subjects, f.data = append(f.subjects, f.subjects[0]), append(f.data, []byte("not a forge message")) + add("novox/mesh-host", 8, "h8", "c8c8c8c8c8") + + for name, c := range map[string]struct { + repo, commit string + number int + found bool + }{ + "the newest merge": {"novox/mesh-host", "c8c8c8c8c8", 8, true}, + "an older merge, found by scanning": {"novox/mesh-host", "c6c6c6c6c6", 6, true}, + "a short commit": {"novox/mesh-host", "c7c7c7c", 7, true}, + "a commit of another repository": {"novox/mesh-catalog", "c6c6c6c6c6", 0, false}, + "a repository with no merge": {"novox/hq", "c6c6c6c6c6", 0, false}, + } { + owner, name2, _ := strings.Cut(c.repo, "/") + k, found, err := forgeMergeAt(t.Context(), f, owner, name2, c.commit) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + if found != c.found || k.Number != c.number { + t.Errorf("%s: found %v #%d, want %v #%d", name, found, k.Number, c.found, c.number) + } + if found && (k.Owner != owner || k.Repo != name2 || k.HeadSHA == "") { + t.Errorf("%s: the merge reads as %+v", name, k.SourceMoved) + } + } +} diff --git a/cmd/mesh-controller/loop_own_consumers_test.go b/cmd/mesh-controller/loop_own_consumers_test.go index d8f5ae9f..20a7d7b4 100644 --- a/cmd/mesh-controller/loop_own_consumers_test.go +++ b/cmd/mesh-controller/loop_own_consumers_test.go @@ -21,8 +21,13 @@ func holdingOne(names ...string) func(broker.Consumer) (uint64, error) { } func deafObservations(now time.Time, held func(broker.Consumer) (uint64, error)) (string, int) { + return deafHeld(now, time.Time{}, held) +} + +// deafHeld is deafObservations with the loop's last hold due at due. +func deafHeld(now, due time.Time, held func(broker.Consumer) (uint64, error)) (string, int) { f := calm(now) - loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), loopConsumers(), held) + loop, err := readLoop(now.Add(-loopDeafAfter-time.Minute), due, loopConsumers(), held) if err != nil { panic(err) } @@ -65,3 +70,51 @@ func TestDeafLookReadsExactlyTheControllersOwnConsumers(t *testing.T) { } } } + +// Sweep 2026-10-30 (issue 774): a message the loop held back with a delay longer than the deaf bound is not a deaf +// controller until it is due and still not taken; the merges consumer's five-minute holds flapped the alarm this way. +func TestAMessageHeldBackIsNotDeafUntilItIsDue(t *testing.T) { + now := time.Now() + held := holdingOne("CONTROL/" + broker.ControllerName) + for name, c := range map[string]struct { + due time.Time + deaf int + }{ + "nothing held": {time.Time{}, 1}, + "held, due in three minutes": {now.Add(3 * time.Minute), 0}, + "held, due a moment ago": {now.Add(-time.Second), 0}, + "held, due past the deaf bound, unsent": {now.Add(-loopDeafAfter - time.Second), 1}, + } { + if _, n := deafHeld(now, c.due, held); n != c.deaf { + t.Errorf("%s: %d deaf observation(s), want %d", name, n, c.deaf) + } + } +} + +// Sweep 2026-10-30 (issue 774): the controller holds no consumer of the forge's merges any more, and removes the one +// it held at every assertion. +func TestTheMergesConsumerIsRetired(t *testing.T) { + for _, c := range broker.MeshConsumers() { + if c.Name == broker.MergesConsumerName { + t.Fatalf("the controller still holds %s on %s", c.Name, c.Stream) + } + } + r := &removed{} + if err := broker.AssertMeshConsumers(r); err != nil { + t.Fatal(err) + } + if !r.gone[broker.EventsStream+"/"+broker.MergesConsumerName] { + t.Fatalf("the retired merges consumer is not removed: removed %v", r.gone) + } +} + +type removed struct{ gone map[string]bool } + +func (*removed) EnsureConsumer(broker.Consumer) error { return nil } +func (r *removed) RemoveConsumer(stream, name string) error { + if r.gone == nil { + r.gone = map[string]bool{} + } + r.gone[stream+"/"+name] = true + return nil +} diff --git a/cmd/mesh-controller/moved.go b/cmd/mesh-controller/moved.go index d9c57391..ca5c03e9 100644 --- a/cmd/mesh-controller/moved.go +++ b/cmd/mesh-controller/moved.go @@ -16,6 +16,7 @@ import ( sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" @@ -343,64 +344,113 @@ func claimantsOf(entries []inventory.Entry, seat string) []string { return out } -// forgeMergeKey is where the forge's announcement of a merge is kept, by the merge commit: the one thing a build of a merge -// names, from which the pull request and its head are found (novox/hq ADR 0420 §7). -func forgeMergeKey(owner, repo, commit string) string { - return fmt.Sprintf("merge/%s/%s/%s", owner, repo, commit) -} - -// keepForgeMerge keeps the forge's announcement of a merge in the bucket of the merges, a week at most as that bucket -// keeps them. The controller plans nothing from a merge any more (ADR 0539 §17); this is all it keeps of one. -var keepForgeMerge = func(ctx context.Context, m link.SourceMoved) error { - serving := servingBus.Load() - if serving == nil { - return fmt.Errorf("%w: the bus is not held", link.ErrTryAgain) - } - api, err := jetstream.New(serving.Conn()) - if err != nil { - return fmt.Errorf("%w: %w", link.ErrTryAgain, err) - } - kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket) - if err != nil { - return fmt.Errorf("%w: the bucket of the merges: %w", link.ErrTryAgain, err) - } - body, err := json.Marshal(keptMerge{SourceMoved: m}) - if err != nil { - return err - } - if _, err := kv.Put(ctx, forgeMergeKey(m.Owner, m.Repo, m.Commit), body); err != nil { - return fmt.Errorf("%w: %w", link.ErrTryAgain, err) - } - return nil -} - -// keptForgeMerge is the forge's announcement of a merge, by its commit; false when none was kept. A variable so a test can -// hand it one. +// keptForgeMerge is the forge's word on the merge of a repository at a commit: the pull request and the head it merged, +// read from the forge bridge's own stream, FORGE, where mesh-delivery-forge-bridge says every merge as +// `forge...merged` and keeps it a week (novox/hq ADR 0437). False when the stream holds no merge at that +// commit. A variable so a test can hand it one. +// +// **The controller keeps no copy of the forge's merges any more** (sweep 2026-10-30, issue 774). It once kept each in +// its bucket of merges off a consumer of its own on EVENTS that never gave up a message; one announcement over the +// bucket's value size held that consumer for a day and queued every merge behind it, and the deaf watchdog read the +// queue as a deaf controller. The forge bridge's stream is the record of the merges (research 072, 03-domains: the +// delivery parts own what a merge is); the controller reads it where it needs one, at a moved module's registration. var keptForgeMerge = func(ctx context.Context, owner, repo, commit string) (keptMerge, bool, error) { serving := servingBus.Load() if serving == nil { - return keptMerge{}, false, errors.New("only the serving controller keeps what the forge announced") + return keptMerge{}, false, errors.New("only the serving controller reads the forge's merges") } api, err := jetstream.New(serving.Conn()) if err != nil { return keptMerge{}, false, err } - kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket) + stream, err := api.Stream(ctx, forgeStream) + if err != nil { + return keptMerge{}, false, fmt.Errorf("the forge bridge's stream %s cannot be read: %w", forgeStream, err) + } + return forgeMergeAt(ctx, streamMerges{stream}, owner, repo, commit) +} + +// forgeStream is the forge bridge's stream as the bus names it (its declared name, forge, in capitals). +var forgeStream = catalogue.StreamBusName(sdkdelivery.ForgeStream) + +// forgeMergeSubject is the subject of the forge bridge's merged message for a repository. +func forgeMergeSubject(owner, repo string) (string, error) { + event, err := sdkdelivery.ForgeEvent(owner+"/"+repo, sdkdelivery.ForgeMerged) + if err != nil { + return "", err + } + return process.ModuleStreamSubject(sdkdelivery.ForgeBridgeModule, event), nil +} + +// mergeMessages is what forgeMergeAt reads of a stream: the last message on a subject, and the first at or after a +// sequence. Each answers jetstream.ErrMsgNotFound when there is none. +type mergeMessages interface { + last(ctx context.Context, subject string) (uint64, []byte, error) + next(ctx context.Context, subject string, from uint64) (uint64, []byte, error) +} + +type streamMerges struct{ s jetstream.Stream } + +func (m streamMerges) last(ctx context.Context, subject string) (uint64, []byte, error) { + msg, err := m.s.GetLastMsgForSubject(ctx, subject) + if err != nil { + return 0, nil, err + } + return msg.Sequence, msg.Data, nil +} + +func (m streamMerges) next(ctx context.Context, subject string, from uint64) (uint64, []byte, error) { + msg, err := m.s.GetMsg(ctx, from, jetstream.WithGetMsgSubject(subject)) + if err != nil { + return 0, nil, err + } + return msg.Sequence, msg.Data, nil +} + +// forgeMergeAt finds the merged message of owner/repo at commit: the newest first, which is the merge a take-in +// follows nearly always, then every one the stream keeps, oldest first. A message that does not read as a forge +// message is passed over. +func forgeMergeAt(ctx context.Context, msgs mergeMessages, owner, repo, commit string) (keptMerge, bool, error) { + subject, err := forgeMergeSubject(owner, repo) if err != nil { return keptMerge{}, false, err } - e, err := kv.Get(ctx, forgeMergeKey(owner, repo, commit)) - if errors.Is(err, jetstream.ErrKeyNotFound) { + // Read for the four fields a registration needs, not checked whole (sdkdelivery.DecodeForge): a merged message a + // newer bridge writes, or one the SDK's check refuses for a part the controller never reads, still names its pull + // request, head and commit. + match := func(data []byte) (keptMerge, bool) { + var f sdkdelivery.Forge + if json.Unmarshal(data, &f) != nil || f.Kind != sdkdelivery.ForgeMerged || f.Number <= 0 || f.Head == "" || + !sameCommitID(f.Merge, commit) { + return keptMerge{}, false + } + return keptMerge{SourceMoved: link.SourceMoved{Owner: owner, Repo: repo, Base: f.Branch, Number: f.Number, + HeadSHA: f.Head, Commit: f.Merge}}, true + } + lastSeq, data, err := msgs.last(ctx, subject) + if errors.Is(err, jetstream.ErrMsgNotFound) { return keptMerge{}, false, nil } if err != nil { return keptMerge{}, false, err } - var k keptMerge - if err := json.Unmarshal(e.Value(), &k); err != nil { - return keptMerge{}, false, err + if k, ok := match(data); ok { + return k, true, nil } - return k, true, nil + for from := uint64(1); from < lastSeq; { + seq, data, err := msgs.next(ctx, subject, from) + if errors.Is(err, jetstream.ErrMsgNotFound) || (err == nil && seq >= lastSeq) { + return keptMerge{}, false, nil + } + if err != nil { + return keptMerge{}, false, err + } + if k, ok := match(data); ok { + return k, true, nil + } + from = seq + 1 + } + return keptMerge{}, false, nil } // keptMerge is a merge as the record keeps it: the forge's announcement. diff --git a/cmd/mesh-controller/moved_bus_test.go b/cmd/mesh-controller/moved_bus_test.go index 26e99fb8..819b5dcf 100644 --- a/cmd/mesh-controller/moved_bus_test.go +++ b/cmd/mesh-controller/moved_bus_test.go @@ -99,17 +99,9 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh return err == nil && kept }) - // The forge's merge of that pull request, as the controller heard it, and the build it asked for. + // The forge bridge's merge of that pull request on its stream, and the build it asked for. const commit = "c0ffee0123456789" - event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-host", "base": "main", "number": 7, - "head_sha": a.merge.HeadSHA, "merge_commit_sha": commit}) - var forgeMerge link.SourceMoved - if err := json.Unmarshal(event, &forgeMerge); err != nil { - t.Fatal(err) - } - if err := keepForgeMerge(ctx, forgeMerge); err != nil { - t.Fatal(err) - } + sayForgeMerge(ctx, t, js, "novox/mesh-host", 7, a.merge.HeadSHA, commit) id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano()) if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-host", Seat: "git", Path: "modules/sudo", Ref: "main", Commit: commit, For: "merge"}); err != nil { diff --git a/cmd/mesh-controller/new_module_bus_test.go b/cmd/mesh-controller/new_module_bus_test.go index 2300c0a3..89c24f61 100644 --- a/cmd/mesh-controller/new_module_bus_test.go +++ b/cmd/mesh-controller/new_module_bus_test.go @@ -100,17 +100,9 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB return err == nil && kept }) - // The forge's merge of that pull request, as the controller heard it, and the build it asked for. + // The forge bridge's merge of that pull request on its stream, and the build it asked for. const commit = "c0ffee0123456789" - event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "base": "main", "number": 7, - "head_sha": a.merge.HeadSHA, "merge_commit_sha": commit}) - var forgeMerge link.SourceMoved - if err := json.Unmarshal(event, &forgeMerge); err != nil { - t.Fatal(err) - } - if err := keepForgeMerge(ctx, forgeMerge); err != nil { - t.Fatal(err) - } + sayForgeMerge(ctx, t, js, "novox/mesh-delivery", 7, a.merge.HeadSHA, commit) id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano()) if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-delivery", Seat: "git", Path: "modules/mesh-delivery-view", Ref: "main", Commit: commit, For: "merge"}); err != nil { diff --git a/cmd/mesh-controller/new_module_on_merge_test.go b/cmd/mesh-controller/new_module_on_merge_test.go index 73db2ae5..7b7a46d8 100644 --- a/cmd/mesh-controller/new_module_on_merge_test.go +++ b/cmd/mesh-controller/new_module_on_merge_test.go @@ -11,7 +11,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-controller/internal/inventory" - "github.com/novox/mesh-controller/internal/link" ) // ADR 0501 §6 (issue 1851, A5): the live case of 2026-10-07 (novox/hq issue 300), on keys. A merge of the catalogue adds @@ -33,12 +32,6 @@ func TestAMergeAddingAModuleEndsRegisteredFromItsKeyAndAsksNoBuild(t *testing.T) watchNewModules(watching, open) const commit = "3da80a4b00000000" - m := link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit, - Paths: []string{"modules/systemd-resolved/module.json", "modules/systemd-resolved/cmd/main.go"}, - ModuleDirs: []string{"modules/systemd-resolved"}, ModuleDirsSaid: true} - if err := (following{open: open}).SourceMoved(ctx, m); err != nil { - t.Fatal(err) - } // The indexer writes the module's trunk value; the builder builds it, then its key is done. fp := fpOf('7') manifest := b.indexed(t, "systemd-resolved", "1", "novox/mesh-catalog", "modules/systemd-resolved", commit, fp) diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 3af94c2d..75de1c16 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -284,13 +284,17 @@ func watchLoop(f *signalFacts) []conditions.Observation { if since.IsZero() { since = f.started } + // A message the loop held back waits by the loop's own choice until it is due: deafness is counted from then. + if f.loop.due.After(since) { + since = f.loop.due + } if f.now.Sub(since) <= loopDeafAfter { return nil } return []conditions.Observation{{Scope: conditions.ScopeCore, ID: "controller", Kind: "controller-deaf", Token: "deaf", Machine: f.host, Severity: conditions.Urgent, Summary: fmt.Sprintf("the controller's event loop has taken nothing for %s while its consumers hold %d "+ - "message(s): reports, builds and merges are not being acted on", ago(f.now.Sub(since)), f.loop.pending), + "message(s): reports and events are not being acted on", ago(f.now.Sub(since)), f.loop.pending), Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}} } diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index f17c80be..73b46fcc 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -93,15 +93,6 @@ func (f following) Announceable(ctx context.Context) ([]link.Announcement, error return out, nil } -// SourceMoved is the forge announcing a merge. It is put into the open batch (novox/hq ADR 0276), which -// becomes one walk when its merge window closes: hearMerge, and cutBatchesHeld in batches.go. -func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error { - // A merge plans nothing here any more: mesh-delivery merges on the Approve, the indexer writes the fingerprints and - // the builder builds them (ADR 0539 §5, §17). The controller keeps the announcement, for the build of a merge that - // moves a module between repositories to be shown the operator's Approve (ADR 0420 §7). - return keepForgeMerge(ctx, m) -} - // actingOnMerges keeps one merge acted on at a time (novox/hq issue 266). var actingOnMerges sync.Mutex diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 062b5b9f..c2207c16 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -147,7 +147,9 @@ type machineFacts struct { func (m machineFacts) asleep() bool { return m.power.Away() } type loopFacts struct { - took time.Time + took time.Time + // due is when a message the loop held back is handed over again (link.Loop.Due): the loop is not deaf before it. + due time.Time pending uint64 where string } @@ -458,7 +460,7 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) { if w.js == nil { return loopFacts{took: link.Loop.Last()}, errors.New("this controller is not on the bus") } - return readLoop(link.Loop.Last(), loopConsumers(), func(c broker.Consumer) (uint64, error) { + return readLoop(link.Loop.Last(), link.Loop.Due(), loopConsumers(), func(c broker.Consumer) (uint64, error) { info, err := w.js.Context().ConsumerInfo(c.Stream, c.Name) if err != nil { return 0, err @@ -468,8 +470,8 @@ func (w *watchdogs) gatherLoop() (loopFacts, error) { } // readLoop sums what the given consumers hold, as held reads each. -func readLoop(took time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) { - f := loopFacts{took: took} +func readLoop(took, due time.Time, consumers []broker.Consumer, held func(broker.Consumer) (uint64, error)) (loopFacts, error) { + f := loopFacts{took: took, due: due} var where []string for _, c := range consumers { n, err := held(c) diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 052d47a8..6045431b 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -339,13 +339,36 @@ func Overlaps() []string { // ack subject is derived from (nats.go: `$JS.ACK..controller.>`). const ControllerName = "controller" -// MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the -// events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no -// max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer -// resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of -// a consumer, not of a subject. +// MergesConsumerName was the controller's durable consumer of the forge's merged event on EVENTS (novox/hq ADR 0407 +// decision 3, issue 973), with no max-deliver and one message at a time. **Retired** (sweep 2026-10-30, issue 774): +// the forge bridge's stream FORGE is the record of the merges, and the controller reads it where it needs a merge. On +// 2026-10-29 one announcement too large for the controller's bucket of merges was held on it for over a day, with +// 163 merges queued behind it, and the deaf watchdog read the queue as a deaf controller. RetiredConsumers removes it. const MergesConsumerName = "controller-merges" +// RetiredConsumers are durable consumers the controller once held and holds no more: each assertion removes them, +// so a retired consumer cannot keep a queue nobody takes from (research 072, 03-domains: the controller shrinks). +var RetiredConsumers = []Consumer{{Name: MergesConsumerName, Stream: EventsStream}} + +// Remover is the part of a JetStream connection that removes a retired consumer. +type Remover interface { + RemoveConsumer(stream, name string) error +} + +// removeRetired removes every retired consumer, when the connection can. +func removeRetired(e Ensurer) error { + r, ok := e.(Remover) + if !ok { + return nil + } + for _, c := range RetiredConsumers { + if err := r.RemoveConsumer(c.Stream, c.Name); err != nil { + return fmt.Errorf("removing the retired consumer %s on %s: %w", c.Name, c.Stream, err) + } + } + return nil +} + // ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states // under it (novox/hq ADR 0134). // @@ -513,23 +536,6 @@ func ControllerConsumers() []Consumer { Why: "the controller is the single consumer of what nodes say; explicit ack and no " + "max-deliver, because the store window's bound is the controller's own", }, - // The forge's merges (novox/hq ADR 0407 decision 3, issue 973): never dropped, so no max-deliver and no - // reset; a merge the controller cannot record yet is held with a growing delay until it can. **Asserted before - // the events consumer**, which stops filtering merges: the order of this list is the order of assertion. - { - Name: MergesConsumerName, - Stream: "EVENTS", - // Within what the controller may subscribe already (`_DELIVER.controller.>`). - DeliverSubject: "_DELIVER." + ControllerName + ".merges", - Filters: []string{ControllerFollows[3]}, - Push: true, - AckWaitSeconds: 30, - MaxAckPending: 1, - FromNow: true, - TakesOverFrom: ControllerName, - Why: "the forge's merges, one at a time; no max-deliver and never reset, because a merge is never " + - "dropped and no person releases one", - }, { Name: ControllerName, Stream: "EVENTS", @@ -581,6 +587,9 @@ func NoticesConsumer() Consumer { // one on DEAD_LETTER_NOTICES, which the keeper of dead letters asserts and retries by itself, so a fault // there never stops the controller serving (novox/hq issue 330). func AssertServingConsumers(e Ensurer) error { + if err := removeRetired(e); err != nil { + return err + } for _, c := range MeshConsumers() { if c.Stream == DeadLetterNoticesStream { continue @@ -603,6 +612,9 @@ type Ensurer interface { // After the streams, necessarily: a consumer on a stream that does not exist is refused, and the // refusal names the stream rather than the order. func AssertMeshConsumers(e Ensurer) error { + if err := removeRetired(e); err != nil { + return err + } for _, c := range MeshConsumers() { if err := e.EnsureConsumer(c); err != nil { return fmt.Errorf("asserting consumer %s on %s: %w", c.Name, c.Stream, err) diff --git a/internal/link/contracts.go b/internal/link/contracts.go index 20f3a9f2..60703291 100644 --- a/internal/link/contracts.go +++ b/internal/link/contracts.go @@ -30,9 +30,8 @@ var Contracts = map[string]Contract{ KindBuilt: {Unordered: "a build's outcome on the retired build seat's subjects: dropped, nothing takes a build in " + "from the bus any more (R4g, ADR 0501 §1); the subjects stay followed while the installer's carried user list " + "names them"}, - KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " + - "or older than the last look is history, not a second plan (issues 250, 266)", - Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}}, + KindSourceMoved: {Unordered: "the forge's merge on EVENTS: no longer followed, dropped if one arrives; the forge " + + "bridge's stream FORGE is the record of the merges, read at a moved module's registration (sweep 2026-10-30)"}, KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"}, KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " + "than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " + diff --git a/internal/link/receive_nats.go b/internal/link/receive_nats.go index 7fbecbc8..5b60dbb5 100644 --- a/internal/link/receive_nats.go +++ b/internal/link/receive_nats.go @@ -62,7 +62,7 @@ func Nats(js *broker.JetStream) Inbound { // whatever was asked for — and not at all when nothing was. func (n *natsInbound) Also(kind string) error { switch kind { - case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided, + case KindModuleMoved, KindCatchUp, KindProvisioner, KindPullUpdated, KindDecided, KindDeliveryMerged: n.follows[kind] = true return nil @@ -145,24 +145,6 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con defer func() { _ = followed.Unsubscribe() }() } - // And the forge's merges, off a consumer of their own: no max-deliver and no reset, so a merge held for the - // plans waits for them however long (novox/hq ADR 0407 decision 3). Bound to the same channel: one loop. - if n.follows[KindSourceMoved] { - if events == nil { - events = make(chan *nats.Msg, Prefetch) - } - merges, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) { - return js.ChanSubscribe(MergedSubject, events, nats.Bind("EVENTS", broker.MergesConsumerName)) - }) - if err != nil { - return fmt.Errorf("subscribing to the forge's merges: %w", err) - } - if merges == nil { - return nil - } - defer func() { _ = merges.Unsubscribe() }() - } - // A connection that dropped is said, not discovered. A controller whose bus connection is gone // is a mesh where nothing can be told anything. gone := make(chan error, 1) @@ -423,7 +405,11 @@ func (m *natsControl) Hold(after time.Duration) error { if _, already := m.on.since[m.seq]; !already { m.on.since[m.seq] = time.Now() } - return m.msg.NakWithDelay(after) + if err := m.msg.NakWithDelay(after); err != nil { + return err + } + Loop.HeldUntil(time.Now().Add(after)) + return nil } func (m *natsControl) forget() { diff --git a/internal/link/receive_nats_integration_test.go b/internal/link/receive_nats_integration_test.go index b2ee6f5a..b49bf280 100644 --- a/internal/link/receive_nats_integration_test.go +++ b/internal/link/receive_nats_integration_test.go @@ -394,35 +394,3 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) { case <-time.After(3 * time.Second): } } - -func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) { - first, most := MergeHeldFirst, MergeHeldMost - MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond - t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most }) - js := aBus(t) - up := &manyTimesBusy{busy: 8} - s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()} - if err := s.Follows(up); err != nil { - t.Fatal(err) - } - ctx, stop := context.WithCancel(context.Background()) - defer stop() - go func() { _ = s.Serve(ctx) }() - eventually(t, "the merges consumer being made", func() bool { - _, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName) - return err == nil - }) - - body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}) - if _, err := js.Context().Publish(MergedSubject, body); err != nil { - t.Fatal(err) - } - eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 }) - eventually(t, "the merge being acknowledged", func() bool { - info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName) - return err == nil && info.NumAckPending == 0 && info.NumPending == 0 - }) - if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 { - t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs) - } -} diff --git a/internal/link/receive_nats_test.go b/internal/link/receive_nats_test.go index 0f7d9ce3..fcd71630 100644 --- a/internal/link/receive_nats_test.go +++ b/internal/link/receive_nats_test.go @@ -2,7 +2,6 @@ package link import ( "context" - "errors" "io" "log" "sync" @@ -199,46 +198,3 @@ func (s slowly) Heard(context.Context, Report) (bool, error) { s.work(); return type sameAgain struct{} func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil } - -// manyTimesBusy is an upgrader whose plans are busy for its first merges, as another controller's hold is. -type manyTimesBusy struct { - toldAbout - busy int - calls int -} - -func (u *manyTimesBusy) SourceMoved(context.Context, SourceMoved) error { - u.mu.Lock() - defer u.mu.Unlock() - u.calls++ - if u.calls <= u.busy { - return errors.Join(ErrTryAgain, errors.New("another controller is working the plans")) - } - return nil -} - -func (u *manyTimesBusy) called() int { - u.mu.Lock() - defer u.mu.Unlock() - return u.calls -} - -// **A merge held more than five times is still handed over and recorded** (issue 973, novox/hq ADR 0407 decision -// 3): the merges have a durable consumer of their own with no max-deliver, so they never go to DEAD_LETTERS to wait -// for a person; the delay grows, and is capped. - -// The delay a held merge waits grows from the first to the most and stays there. -func TestAHeldMergesDelayGrowsAndIsCapped(t *testing.T) { - var got []time.Duration - for _, held := range []time.Duration{0, 4 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second, - 200 * time.Second, 20 * time.Minute} { - got = append(got, MergeRedeliverAfter(held)) - } - want := []time.Duration{5 * time.Second, 5 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second, - 200 * time.Second, 5 * time.Minute} - for i := range want { - if got[i] != want[i] { - t.Errorf("held %d: waits %v; want %v", i, got[i], want[i]) - } - } -} diff --git a/internal/link/serve.go b/internal/link/serve.go index 3e0ccf42..ea46276e 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -49,8 +49,6 @@ type Upgrader interface { // stop every upgrade behind it — except the store unreachable for the moment, which is asked // again for a bounded time (novox/hq issue 083). Upgraded(ctx context.Context, u Upgraded) error - // SourceMoved is a merge on the forge: build what that source produces, bases first. - SourceMoved(ctx context.Context, m SourceMoved) error } // Checker is what the controller does when the forge says a pull request's head moved: ask for it to be @@ -105,9 +103,6 @@ const GiveUpAfter = 2 * time.Minute // Follows says what to do about upgrades, and asks for them to be delivered. func (s *Server) Follows(u Upgrader) error { - if err := s.inbound.Also(KindSourceMoved); err != nil { - return err - } if err := s.inbound.Also(KindModuleMoved); err != nil { return err } @@ -225,7 +220,9 @@ func (s *Server) act(ctx context.Context, m Control) { case KindModuleMoved: s.moved(ctx, m) case KindSourceMoved: - s.sourceMoved(ctx, m) + // The forge's merges are no longer followed (sweep 2026-10-30): the forge bridge's stream is their record, read + // where a merge is needed (keptForgeMerge). One that arrives on an old consumer is let go of. + _ = m.Drop() case KindCatchUp: s.catchingUp(ctx, m) case KindProvisioner: @@ -521,61 +518,6 @@ func short(commit string) string { return commit } -// MergeHeldFirst is how long a merge the controller could not record yet waits before the bus hands it over -// again, and MergeHeldMost the longest it ever waits: the delay grows with how long the merge has been held, from -// the first to the most (novox/hq ADR 0407 decision 3). Variables so a test need not wait. -var ( - MergeHeldFirst = 5 * time.Second - MergeHeldMost = 5 * time.Minute -) - -// MergeRedeliverAfter is the delay for a merge held for the time given: that time, so the delay about doubles -// with each hold, between MergeHeldFirst and MergeHeldMost. -func MergeRedeliverAfter(heldFor time.Duration) time.Duration { - return min(max(heldFor, MergeHeldFirst), MergeHeldMost) -} - -// sourceMoved acts on the forge's announcement of a merge. **A merge is never dropped** (issue 921, novox/hq ADR -// 0407 decision 3): one the controller could not record for now (the plans held by another controller, the store -// away: ErrTryAgain) is held and handed over again until it is recorded; the controller giving up on it would -// lose a change for good. Anything else is an answer and is taken: a build that fails is reported by the build -// itself, and re-delivering the merge would only re-fail it. -func (s *Server) sourceMoved(ctx context.Context, m Control) { - var moved SourceMoved - if err := json.Unmarshal(m.Body(), &moved); err != nil { - s.log.Printf("a merge announcement could not be read: %v", err) - _ = m.Took() - return - } - if moved.Commit == "" || moved.Repo == "" { - s.log.Printf("a merge announcement named no repository or no commit; ignored") - _ = m.Took() - return - } - err := s.upgrader.SourceMoved(ctx, moved) - // Left unsettled when the controller is ending, whatever error that gave (the plans' hold returns a bare - // context.Canceled): the bus hands it to whatever consumes next, and taking it would lose the merge. - if err != nil && ctx.Err() != nil { - return - } - if err != nil && errors.Is(err, ErrTryAgain) { - // Also left unsettled when the bus will not hold it. - delay := MergeRedeliverAfter(m.HeldFor()) - if held := m.Hold(delay); held != nil { - s.log.Printf("%s/%s moved to %.8s and could not be held (%v): %v; left for the bus to hand over again", - moved.Owner, moved.Repo, moved.Commit, held, err) - return - } - s.log.Printf("%s/%s moved to %.8s and is held: %v; handed over again in %s", moved.Owner, moved.Repo, - moved.Commit, err, delay) - return - } - if err != nil { - s.log.Printf("%s/%s moved to %.8s and the mesh could not act on it: %v", moved.Owner, moved.Repo, moved.Commit, err) - } - _ = m.Took() -} - // pullUpdated asks for a pull request's merge check. Taken whatever happens: a check that could not be // asked is said here, and the next push to the pull request asks again. func (s *Server) pullUpdated(ctx context.Context, m Control) { diff --git a/internal/link/source_moved_held_test.go b/internal/link/source_moved_held_test.go deleted file mode 100644 index 198e6682..00000000 --- a/internal/link/source_moved_held_test.go +++ /dev/null @@ -1,60 +0,0 @@ -package link - -import ( - "context" - "encoding/json" - "errors" - "io" - "log" - "testing" - "time" -) - -// heldControl is a merge announcement that records how it was settled. -type heldControl struct { - body []byte - took, held, drops int -} - -func (c *heldControl) Kind() string { return "source-moved" } -func (c *heldControl) Body() []byte { return c.body } -func (c *heldControl) Subject() string { return "" } -func (c *heldControl) Redelivered() bool { return false } -func (c *heldControl) HeldFor() time.Duration { return 0 } -func (c *heldControl) Answer(context.Context, []byte) error { return nil } -func (c *heldControl) Took() error { c.took++; return nil } -func (c *heldControl) Hold(time.Duration) error { c.held++; return nil } -func (c *heldControl) Drop() error { c.drops++; return nil } - -type sourceMover struct { - Upgrader - answer error -} - -func (s sourceMover) SourceMoved(context.Context, SourceMoved) error { return s.answer } - -// **A merge is acknowledged only after the controller recorded it** (novox/hq ADR 0407 decision 3, issue 1092): an -// answer of "not now" — the plan it belongs to could not be saved, the bus away, the lease lost — leaves the announcement -// unacknowledged, held to be handed over again; only an answer that is not "not now" settles it. -func TestAMergeTheControllerCouldNotRecordIsHeldAndNotAcknowledged(t *testing.T) { - body, err := json.Marshal(SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: "c1"}) - if err != nil { - t.Fatal(err) - } - s := &Server{log: log.New(io.Discard, "", 0)} - for name, tc := range map[string]struct { - answer error - took, held int - }{ - "recorded": {nil, 1, 0}, - "not now": {errors.Join(ErrTryAgain, errors.New("the plan is not saved")), 0, 1}, - "an answer, not a wait": {errors.New("the mesh could not act on it"), 1, 0}, - } { - s.upgrader = sourceMover{answer: tc.answer} - m := &heldControl{body: body} - s.sourceMoved(context.Background(), m) - if m.took != tc.took || m.held != tc.held || m.drops != 0 { - t.Errorf("%s: acknowledged %d times and held %d times, want %d and %d", name, m.took, m.held, tc.took, tc.held) - } - } -} diff --git a/internal/link/store_window_test.go b/internal/link/store_window_test.go index f55ddfea..4ea2545d 100644 --- a/internal/link/store_window_test.go +++ b/internal/link/store_window_test.go @@ -17,8 +17,7 @@ func (r recordsWith) Built(context.Context, BuildResult) error { return r.err } type upgradesWith struct{ err error } -func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err } -func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return u.err } +func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err } type replaysWith struct{ err error } @@ -69,61 +68,3 @@ func TestACatchUpWaitsOutARestartingStore(t *testing.T) { } } } - -// A merge the controller could not record for now (the plans held by another controller, the store away) is held -// and handed over again, never taken: the bus keeps it until it is recorded (issue 921). Any other failure is an -// answer and is taken, as before. -func TestAMergeThatCouldNotBeRecordedNowIsHeldNotDropped(t *testing.T) { - merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"} - for _, c := range []struct { - what string - err error - want func(*settled) bool - }{ - {"the plans held by another controller", errors.Join(ErrTryAgain, errors.New("another controller is working the plans")), - func(s *settled) bool { return s.unsettled() }}, - {"the store away", errors.Join(ErrTryAgain, restarting), func(s *settled) bool { return s.unsettled() }}, - {"cannot act", errors.New("no such module"), func(s *settled) bool { return s.acked && !s.nacked }}, - {"kept", nil, func(s *settled) bool { return s.acked && !s.nacked }}, - } { - s, in := serving() - s.upgrader = upgradesWith{err: c.err} - to := &settled{} - held := in.sends(t, to, KindSourceMoved, merged).(*fakeControl) - s.act(context.Background(), held) - if !c.want(to) { - t.Errorf("%s: a merge was settled as %+v", c.what, *to) - } - if c.err != nil && errors.Is(c.err, ErrTryAgain) && len(in.held) != 1 { - t.Errorf("%s: the merge was not held to be handed over again", c.what) - } - } -} - -// A merge being held when the controller is replaced (its context ended) or when the bus will not hold it is left -// unsettled, never taken: the bus hands it to whatever consumes next (review of issue 921). -func TestAMergeIsLeftUnsettledWhenTheControllerEndsOrTheHoldFails(t *testing.T) { - merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"} - try := errors.Join(ErrTryAgain, errors.New("another controller is working the plans")) - - // The plans' hold returns a bare context.Canceled when the controller is replaced, not ErrTryAgain. - s, in := serving() - s.upgrader = upgradesWith{err: context.Canceled} - to := &settled{} - ended, end := context.WithCancel(context.Background()) - end() - s.act(ended, in.sends(t, to, KindSourceMoved, merged)) - if !to.unsettled() { - t.Errorf("a merge was settled as %+v when the controller's context ended", *to) - } - - s, in = serving() - s.upgrader = upgradesWith{err: try} - to = &settled{} - m := in.sends(t, to, KindSourceMoved, merged).(*fakeControl) - m.holdFails = errors.New("the bus will not hold it") - s.act(context.Background(), m) - if !to.unsettled() { - t.Errorf("a merge was settled as %+v when it could not be held", *to) - } -} diff --git a/internal/link/watched.go b/internal/link/watched.go index 2a7c3003..7d885efd 100644 --- a/internal/link/watched.go +++ b/internal/link/watched.go @@ -85,6 +85,26 @@ func nodeOfToolsAlive(subject string) (string, bool) { type LoopActivity struct { mu sync.Mutex took time.Time + // due is when the latest message the loop held back with a delay is handed over again: until then the loop waits + // by its own choice, which is not deafness (sweep 2026-10-30: a merge held five minutes at a time read as a deaf + // controller every time its delay outran the deaf bound). + due time.Time +} + +// HeldUntil records that the loop handed a message back to be delivered again at due. +func (l *LoopActivity) HeldUntil(due time.Time) { + l.mu.Lock() + defer l.mu.Unlock() + if due.After(l.due) { + l.due = due + } +} + +// Due is when the latest message the loop held is handed over again; zero when it held none. +func (l *LoopActivity) Due() time.Time { + l.mu.Lock() + defer l.mu.Unlock() + return l.due } // Loop is this process's event loop.