diff --git a/cmd/mesh-controller/apply_failing.go b/cmd/mesh-controller/apply_failing.go index 5f76ef80..4fb2eb13 100644 --- a/cmd/mesh-controller/apply_failing.go +++ b/cmd/mesh-controller/apply_failing.go @@ -30,6 +30,11 @@ const packageManagerKey = "the package manager (" // says a re-apply only when how it went changed, so the same failure is one report however long it lasts. // Names the refusal's or the failure's words; cleared by the clean apply the node-engine then says. Not while // the machine is silent, asleep or not known: S1 says silent, and asleep is not lost. +// +// **The bound gates raising, never keeping** (sweep 2026-10-30, issue 774). A report says a failure again whenever its +// words change, with a new time: on novox the vault's refusal of every secret replaced a bus refusal at 16:56, and the +// condition cleared for five minutes while every module still failed. Once raised, it stays raised while any failure +// stands on that machine, judged by the open condition the keeper holds on the bus, not by a report's time. func watchApplyFailing(f *signalFacts) []conditions.Observation { heard := map[string]bool{} for _, m := range f.machines { @@ -40,7 +45,7 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation { var out []conditions.Observation said := map[string]bool{} for _, d := range f.doing { - if !d.Wrong() || d.Since == nil || f.now.Sub(*d.Since) <= applyEvery || !heard[d.Node] { + if !d.Wrong() || d.Since == nil || !heard[d.Node] || (f.now.Sub(*d.Since) <= applyEvery && !f.applyFailingOpen[d.Node]) { continue } out = append(out, applyFailing(d, f.now)) @@ -51,7 +56,8 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation { // on a machine whose node-engine applied what it was sent. failed := map[string][]inventory.ProvReport{} for _, r := range f.provFailed { - if r.Status != inventory.ProvFailed || f.now.Sub(r.At) <= applyEvery || !heard[r.Node] || said[r.Node] { + if r.Status != inventory.ProvFailed || !heard[r.Node] || said[r.Node] || + (f.now.Sub(r.Since()) <= applyEvery && !f.applyFailingOpen[r.Node]) { continue } failed[r.Node] = append(failed[r.Node], r) @@ -62,17 +68,28 @@ func watchApplyFailing(f *signalFacts) []conditions.Observation { return out } +// applyFailingOpen are the machines whose S21 condition is open among the open conditions. +func applyFailingOpen(open []conditions.Condition) map[string]bool { + out := map[string]bool{} + for _, c := range open { + if c.Kind == kindApplyFailing && c.Subject.Machine != "" { + out[c.Subject.Machine] = true + } + } + return out +} + // provisioningFailing is one machine's S21 condition for the modules whose files its provisioner client failed to // place. func provisioningFailing(m string, failed []inventory.ProvReport, now time.Time) conditions.Observation { sort.Slice(failed, func(a, b int) bool { return failed[a].Module < failed[b].Module }) var names, words []string - since := failed[0].At + since := failed[0].Since() for _, r := range failed { names = append(names, r.Module) words = append(words, r.Module+": "+oneLine(orNoReason(r.Reason))) - if r.At.Before(since) { - since = r.At + if r.Since().Before(since) { + since = r.Since() } } o := conditions.Observation{Scope: conditions.ScopeMachine, ID: m, Kind: kindApplyFailing, Machine: m, diff --git a/cmd/mesh-controller/compose_perform.go b/cmd/mesh-controller/compose_perform.go index 6e05f2bd..b07f6181 100644 --- a/cmd/mesh-controller/compose_perform.go +++ b/cmd/mesh-controller/compose_perform.go @@ -250,7 +250,7 @@ func prepareSend(ctx context.Context, open *stores, node string) ([]string, erro if n.Stated && n.Contract >= compose.VaultBoxContract { boxes, err := coreBoxesFor(ctx, inv, node, resolved.Plan.Modules, func(m catalogue.Manifest) bool { return handing.Handing(m) - }, true) + }, true, in.BusMinted) if err != nil { return nil, err } diff --git a/cmd/mesh-controller/composed_live.go b/cmd/mesh-controller/composed_live.go index d910c03e..1c6d942d 100644 --- a/cmd/mesh-controller/composed_live.go +++ b/cmd/mesh-controller/composed_live.go @@ -124,11 +124,11 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err)) return } - if len(moved) > 0 { + if moved := builtByTheMesh(moved); len(moved) > 0 { fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n", node, strings.Join(moved, ", ")) } - if len(left) > 0 { + if left := builtByTheMesh(left); len(left) > 0 { fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+ "new form for them, and the node's want or have names their own composition (issue 1945)\n", node, strings.Join(left, ", ")) diff --git a/cmd/mesh-controller/core_box.go b/cmd/mesh-controller/core_box.go index a86d1f68..d9200f02 100644 --- a/cmd/mesh-controller/core_box.go +++ b/cmd/mesh-controller/core_box.go @@ -69,7 +69,15 @@ type coreBoxes struct { // coreModulesOf is every module of the plan on the core's list whose own secrets this machine's declaration carries, // with their plain own secret names, sorted: one handed to the provisioner has its secrets from the vault through the // client, and one issued outside the mesh is never the vault's to seal here. -func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest) bool) map[string][]string { +// +// **Nor a process's bus account nobody issued yet** (sweep 2026-10-30, issue 774). Asked with `get`, the vault holds +// none and makes a random value under the account's name: own/novox/nats/broker-nats-tools at 16:36:51. Until the +// controller issues the account and hands it to the vault, the machine has the controller's stand-in (prepareSend), as +// every module below the vault's contract does. (The controller's own broker-address and broker-management are no +// accounts: the bus's address, which broker.FromEnvironment reads, and the retired AMQP management login; no process +// declares them, so this leaves them asked as before.) +func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest) bool, node string, + minted []string) map[string][]string { out := map[string][]string{} for _, m := range modules { if !catalogue.CarriesVaultSealed(m.Module) || (handed != nil && handed(m)) { @@ -77,7 +85,7 @@ func coreModulesOf(modules []catalogue.Manifest, handed func(catalogue.Manifest) } var names []string for name, own := range m.OwnSecrets.Plain() { - if own.IssuedBy != catalogue.IssuedOutside { + if own.IssuedBy != catalogue.IssuedOutside && processAccountWarning(m, node, name, minted) == "" { names = append(names, name) } } @@ -208,8 +216,8 @@ var coreVault = func(ctx context.Context) coreAsk { // coreBoxesFor is coreBoxesOf for one machine, read from and kept in the store, and said on a send. func coreBoxesFor(ctx context.Context, inv *inventory.Inventory, node string, modules []catalogue.Manifest, - handed func(catalogue.Manifest) bool, allocating bool) (coreBoxes, error) { - core := coreModulesOf(modules, handed) + handed func(catalogue.Manifest) bool, allocating bool, minted []string) (coreBoxes, error) { + core := coreModulesOf(modules, handed, node, minted) if len(core) == 0 { if allocating { settleVaultSilent(ctx, node) diff --git a/cmd/mesh-controller/core_box_test.go b/cmd/mesh-controller/core_box_test.go index 1148d0a1..f9f4da81 100644 --- a/cmd/mesh-controller/core_box_test.go +++ b/cmd/mesh-controller/core_box_test.go @@ -124,8 +124,35 @@ func TestTheCoresModulesAreTheListsWithTheirOwnSecrets(t *testing.T) { {Module: "postgres", OwnSecrets: catalogue.OwnSecrets{"superuser": {Path: "/c"}}}, {Module: "mesh-provisioner", OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/d"}}}, } - got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" }) + got := coreModulesOf(modules, func(m catalogue.Manifest) bool { return m.Module == "mesh-provisioner" }, "anchor", nil) if want := map[string][]string{"nats": {"broker"}}; !reflect.DeepEqual(got, want) { t.Fatalf("got %v, want %v", got, want) } } + +// Sweep 2026-10-30 (issue 774): a bus process's account is asked of the vault only once it is issued; before, the vault +// would make a random value under the account's name. +func TestAnUnissuedProcessAccountIsNotAskedOfTheVault(t *testing.T) { + nats, err := catalogue.ParseManifest([]byte(`{"module":"nats","version":"1","emits":["x"], + "own-secrets":{"broker":{"path":"/a"},"broker-nats-tools":{"path":"/b","owner":"nats-tools"}}, + "resources":[{"id":"account","type":"user","name":"nats-tools"}, + {"id":"nats-tools","type":"process","run":["./nats-tools"],"user":"nats-tools","bus":{"emits":["x"]}}]}`)) + if err != nil { + t.Fatal(err) + } + if len(nats.ProcessBuses()) == 0 { + t.Fatal("the fixture declares no bus process") + } + for name, c := range map[string]struct { + minted []string + want []string + }{ + "not issued": {nil, []string{"broker"}}, + "issued": {[]string{processUser("anchor", "nats", "nats-tools")}, []string{"broker", "broker-nats-tools"}}, + } { + got := coreModulesOf([]catalogue.Manifest{nats}, nil, "anchor", c.minted) + if !reflect.DeepEqual(got["nats"], c.want) { + t.Errorf("%s: asks %v, want %v", name, got["nats"], c.want) + } + } +} 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..f3153b6a --- /dev/null +++ b/cmd/mesh-controller/forge_merge_test.go @@ -0,0 +1,106 @@ +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) + } + } +} + +// Sweep 2026-10-30 (issue 774): a module the mesh provides itself is never named as lacking a fingerprint. +func TestAModuleTheMeshProvidesIsNotSaidToLackAFingerprint(t *testing.T) { + for _, c := range []struct { + in, want []string + }{ + {[]string{"mesh-wireguard"}, nil}, + {[]string{"claude-code", "mesh-wireguard", "power"}, []string{"claude-code", "power"}}, + {nil, nil}, + } { + if got := builtByTheMesh(c.in); strings.Join(got, ",") != strings.Join(c.want, ",") { + t.Errorf("%v: %v, want %v", c.in, got, c.want) + } + } +} 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..88596958 100644 --- a/cmd/mesh-controller/moved.go +++ b/cmd/mesh-controller/moved.go @@ -16,6 +16,8 @@ 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/kvlimit" + "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" @@ -179,7 +181,12 @@ func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, emitter st if err != nil { return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err) } - if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil { + // Through kvlimit (sweep 2026-10-30, issue 774): a word too large for the bucket is refused before it is put, and + // is an answer, never "not now" — retried, it held the forge's merges for a day (10054 on de-spiegel's merge). + if _, err := kvlimit.Put(ctx, kv, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil { + if errors.Is(err, kvlimit.ErrTooLarge) { + return err + } return fmt.Errorf("%w: %w", link.ErrTryAgain, err) } if k.kept != nil { @@ -343,64 +350,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/provisioned_test.go b/cmd/mesh-controller/provisioned_test.go index 71ecac88..f86e963c 100644 --- a/cmd/mesh-controller/provisioned_test.go +++ b/cmd/mesh-controller/provisioned_test.go @@ -7,6 +7,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) @@ -72,3 +73,44 @@ func TestAFailedPlacingRaisesApplyFailing(t *testing.T) { t.Fatalf("a failed placing is not raised as anchor's apply-failing: %+v", got) } } + +// Sweep 2026-10-30 (issue 774): a failure said again in new words carries a new time, and the condition it raised stays +// raised while it stands; only a machine with no open condition waits out the bound, and a cleared failure clears it. +func TestApplyFailingStaysRaisedWhileAnyFailureStands(t *testing.T) { + now := time.Date(2026, 10, 30, 16, 57, 0, 0, time.UTC) + since := now.Add(-time.Minute) + for name, c := range map[string]struct { + open bool + prov string + doing bool + raised bool + since time.Duration // the client's failing-since, before now; zero when it says none + }{ + "a fresh placing failure, nothing open": {false, inventory.ProvFailed, false, false, 0}, + "a fresh placing failure, condition open": {true, inventory.ProvFailed, false, true, 0}, + "placing applied again, condition open": {true, inventory.ProvApplied, false, false, 0}, + "a fresh failed apply, condition open": {true, "", true, true, 0}, + "a fresh failed apply, nothing open": {false, "", true, false, 0}, + "new words of a failure standing ten minutes, nothing open": {false, inventory.ProvFailed, false, true, 10 * time.Minute}, + } { + f := calm(now) + if c.open { + f.applyFailingOpen = applyFailingOpen([]conditions.Condition{{Kind: kindApplyFailing, + Subject: conditions.Subject{Scope: conditions.ScopeMachine, ID: "anchor", Machine: "anchor"}}}) + } + if c.prov != "" { + f.provFailed = []inventory.ProvReport{{Node: "anchor", Module: "postgres", Status: c.prov, + Reason: "the vault refused it", At: now.Add(-time.Minute)}} + if c.since > 0 { + f.provFailed[0].FailingSince = now.Add(-c.since) + } + } + if c.doing { + f.doing = []inventory.Doing{{Node: "anchor", Outcome: inventory.OutcomeRefused, Refused: "no", At: since, + Since: &since}} + } + if got := watchApplyFailing(f); (len(got) == 1) != c.raised || len(got) > 1 { + t.Errorf("%s: %d observation(s), raised want %v", name, len(got), c.raised) + } + } +} diff --git a/cmd/mesh-controller/retired_sources.go b/cmd/mesh-controller/retired_sources.go new file mode 100644 index 00000000..174da63d --- /dev/null +++ b/cmd/mesh-controller/retired_sources.go @@ -0,0 +1,54 @@ +package main + +import ( + "context" + "fmt" + "regexp" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// **A condition whose check no longer exists is cleared** (sweep 2026-10-30, issue 774). A signals row or a self-check +// probe raises and clears its conditions by reconciling what it sees; one removed from the controller reconciles +// nothing again, so what it raised stayed open for good and read as current: S23's failed plans (the plans went with +// ADR 0539) and D14's stalled deliveries, from 2026-10-20 on. Each watchdogs tick clears every open condition whose +// source is a row's or a probe's name that neither the signals table nor the probe registry holds any more, saying so. +// A source of another form (an event, a module's health, a verb) is never judged here: its own word clears it. + +// checkSource is the form of a row's (S1) or a probe's (D4, DW, D-root) name. +var checkSource = regexp.MustCompile(`^(S[0-9]+|D[0-9]+|D[A-Z]+|D-[a-z]+)$`) + +// liveChecks are the sources this controller still reconciles: its signals rows and its self-check's probes. +func liveChecks() map[string]bool { + out := map[string]bool{} + for _, r := range signalsTable { + out[r.Row] = true + } + for _, p := range probeRegistry { + out[p.ID] = true + } + return out +} + +// retiredCheckConditions are the open conditions raised by a row or a probe the controller no longer has. +func retiredCheckConditions(open []conditions.Condition, live map[string]bool) []conditions.Condition { + var out []conditions.Condition + for _, c := range open { + if checkSource.MatchString(c.Source) && !live[c.Source] { + out = append(out, c) + } + } + return out +} + +// clearRetiredChecks clears each of them, saying why. +func clearRetiredChecks(ctx context.Context, k *conditions.Keeper, retired []conditions.Condition) []string { + var problems []string + for _, c := range retired { + why := fmt.Sprintf("%s, which raised it, is no longer a check of the controller", c.Source) + if _, err := k.ClearSaying(ctx, c.Key, why, "No longer checked: the check that raised this was removed"); err != nil { + problems = append(problems, c.Key+": "+err.Error()) + } + } + return problems +} diff --git a/cmd/mesh-controller/retired_sources_test.go b/cmd/mesh-controller/retired_sources_test.go new file mode 100644 index 00000000..9eb7aaed --- /dev/null +++ b/cmd/mesh-controller/retired_sources_test.go @@ -0,0 +1,31 @@ +package main + +import ( + "testing" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// Sweep 2026-10-30 (issue 774): an open condition raised by a row or a probe the controller no longer has is cleared; +// one raised by a check it still has, or by a source of another form, is left to its own word. +func TestAConditionWhoseCheckIsGoneIsCleared(t *testing.T) { + live := liveChecks() + for source, gone := range map[string]bool{ + "S23": true, + "D14": true, + "S21": false, + "S9": false, + "D-root": false, + "DM": false, + "DB": false, + "D4": false, + "health": false, + "provisioner.failing": false, + "done keys": false, + } { + got := retiredCheckConditions([]conditions.Condition{{Key: "k." + source, Source: source}}, live) + if (len(got) == 1) != gone { + t.Errorf("%s: judged retired %v, want %v", source, len(got) == 1, gone) + } + } +} 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/switch_list.go b/cmd/mesh-controller/switch_list.go index a34a277e..dcb46a15 100644 --- a/cmd/mesh-controller/switch_list.go +++ b/cmd/mesh-controller/switch_list.go @@ -160,7 +160,7 @@ func switchList(ctx context.Context, inv *inventory.Inventory, keys switchReads) var out []switchLine for _, e := range entries { m := e.Manifest.Module - if e.Provided || e.Source.Repository == "" { + if e.Provided || e.Source.Repository == "" || sdkdelivery.ProvidedByMesh(m) { continue } trunk, err := keys.ModuleAt(ctx, m, "") @@ -202,3 +202,16 @@ func heldArtifacts(ctx context.Context, inv *inventory.Inventory, module, builtF // empty-on-error: not an error; the mesh holds no build of that commit on record, which the line says return nil, nil } + +// builtByTheMesh are the modules of a list that a build of the mesh's makes: without the ones the mesh provides itself +// (sdkdelivery.ProvidedByMesh, the private network), which no fingerprint ever names, so a line saying one has none +// says nothing (sweep 2026-10-30, issue 774: "mesh-wireguard ... no fingerprint of the new form" on every composition). +func builtByTheMesh(modules []string) []string { + var out []string + for _, m := range modules { + if !sdkdelivery.ProvidedByMesh(m) { + out = append(out, m) + } + } + return out +} 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..c80f3a74 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -58,6 +58,11 @@ type signalFacts struct { // provFailed is every module whose files a machine's provisioner client last said it failed to place (S21, // novox/hq ADR 0362 §3): that machine's failed apply. provFailed []inventory.ProvReport + // applyFailingOpen are the machines whose S21 condition is open: kept raised while any failure stands, whatever + // the newest report's time (sweep 2026-10-30). + applyFailingOpen map[string]bool + // retiredChecks are the open conditions raised by a row or a probe the controller no longer has (sweep 2026-10-30). + retiredChecks []conditions.Condition loop loopFacts loopErr error @@ -147,7 +152,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 } @@ -252,6 +259,7 @@ func (w *watchdogs) see(running context.Context, f *signalFacts) { problems = append(problems, "S8: "+err.Error()) } } + problems = append(problems, clearRetiredChecks(running, w.keeper, f.retiredChecks)...) if err := w.keeper.EndSilences(running); err != nil { problems = append(problems, err.Error()) } @@ -315,6 +323,8 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { } else { f.standings = providerStandings(open) f.providerWords = providerConditions(open) + f.applyFailingOpen = applyFailingOpen(open) + f.retiredChecks = retiredCheckConditions(open, liveChecks()) } f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet)) f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories) @@ -458,7 +468,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 +478,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/go.mod b/go.mod index f29fb348..4840aa25 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-controller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 github.com/jackc/pgx/v5 v5.10.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/go.sum b/go.sum index 0b60d1bd..179d4319 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0 h1:lKU3HjNXGiVbLW2hUgFjU/TMTUYMlNnnmZxG5T7biQA= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030001116-2da0de75c3f0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24 h1:s0XXZtR7CFPBbnaDygWnE4/Q827iEFt0YfEzujk4Q9g= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030174137-a37181261c24/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/broker/announced.go b/internal/broker/announced.go index 1c8304a9..d4bc23ee 100644 --- a/internal/broker/announced.go +++ b/internal/broker/announced.go @@ -41,23 +41,6 @@ func ControllerAnnouncement(node string) announce.States { return s } -// ReadsAnnounced splits a reader's `reads` (`.`) by what each owner announced: granted when the owner -// announced the state readable, or announced nothing yet (the transition); refused when the owner announced and did -// not announce that state readable. -func ReadsAnnounced(reads []string, announced map[string]announce.States) (granted, refused []string) { - for _, r := range reads { - owner, name, ok := strings.Cut(r, ".") - s, said := announced[owner] - switch { - case !ok || !said || s.Readable(name): - granted = append(granted, r) - default: - refused = append(refused, r) - } - } - return granted, refused -} - // NotAnnounced are the owners some module reads the state of and that have announced nothing yet, sorted: read from the // readers' `reads` as before, until they announce. func NotAnnounced(r Records) []string { diff --git a/internal/broker/announced_test.go b/internal/broker/announced_test.go index ccb4fdb0..dc0a0c47 100644 --- a/internal/broker/announced_test.go +++ b/internal/broker/announced_test.go @@ -7,23 +7,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/announce" ) -// D3 (ADR 0495 clause 3, ADR 0513 §3; issue 1790): a reader's `reads` is granted where the owner announced the state -// readable, refused where it announced and did not, and granted as before where the owner announced nothing yet. -func TestAReadIsGrantedFromItsOwnersAnnouncement(t *testing.T) { - announced := map[string]announce.States{ - "owner": {Module: "owner", States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}}, - } - granted, refused := ReadsAnnounced([]string{"owner.open", "owner.closed", "owner.unsaid", "quiet.any"}, announced) - if !slices.Equal(granted, []string{"owner.open", "quiet.any"}) || !slices.Equal(refused, []string{"owner.closed", "owner.unsaid"}) { - t.Fatalf("granted %v, refused %v", granted, refused) - } - r := Records{Announced: announced, Assigned: map[string][]Declared{"one": {{Module: "reader", Reads: granted, - ReadsRefused: refused}}}} - if got := NotAnnounced(r); !slices.Equal(got, []string{"quiet"}) { - t.Fatalf("not yet announced: %v", got) - } -} - // D1 (ADR 0495 clause 2): the controller announces every bucket of its own, and readable exactly the record mirrors (the // declared packages the pacman module reads among them, ADR 0507 §5), the chain's buckets, the checks running and the // plans building. diff --git a/internal/broker/engine_guard_test.go b/internal/broker/engine_guard_test.go new file mode 100644 index 00000000..8ffdd9b4 --- /dev/null +++ b/internal/broker/engine_guard_test.go @@ -0,0 +1,39 @@ +package broker + +import "testing" + +// Sweep 2026-10-30 (issue 774), the guard of mesh-sdk #136 in the controller's copy: where a placement names the +// node-engine's manifest, it adds no user, no process and nothing the tool runner carries; a module beside it does. +func TestTheNodeEnginesManifestComposesNoPlacedUser(t *testing.T) { + engine := Declared{Module: EngineModule, Serves: []string{"health"}, State: []Bucket{{Name: "health"}}, + Processes: []ProcessPrincipal{{Process: "apply"}}} + beside := Declared{Module: "power", Serves: []string{"power_draw"}} + for name, c := range map[string]struct { + assigned []Declared + module bool + }{ + "with the runner": {[]Declared{{Module: RuntimeModule, NoAccount: true}, engine, beside}, false}, + "without the runner": {[]Declared{engine, beside}, true}, + } { + users, err := Users(Records{Nodes: []string{"anchor"}, Assigned: map[string][]Declared{"anchor": c.assigned}}) + if err != nil { + t.Fatalf("%s: %v", name, err) + } + besideSeen := false + for _, u := range users { + if u.Module == EngineModule && (u.Kind == KindModule || u.Kind == KindProcess) { + t.Errorf("%s: the node-engine's manifest composed a %v user", name, u.Kind) + } + besideSeen = besideSeen || u.Module == beside.Module + for _, carried := range u.Carries { + if carried.Module == EngineModule { + t.Errorf("%s: the tool runner carries the node-engine's manifest", name) + } + besideSeen = besideSeen || carried.Module == beside.Module + } + } + if !besideSeen { + t.Errorf("%s: the module beside it is composed nowhere", name) + } + } +} 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/broker/users.go b/internal/broker/users.go index c48b1662..d35375a0 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -181,6 +181,11 @@ func Users(r Records) ([]Principal, error) { } } for _, d := range r.Assigned[node] { + // **The node-engine's manifest is no placed module's** (novox/hq ADR 0563 §2, §3; sweep 2026-10-30, issue + // 774): the guard mesh-sdk #136 put in buscompose, kept here while the controller composes with its own copy. + if !Composed(d) { + continue + } // **A process account is its own, delivered in its own secret** (broker-, novox/hq // ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module // itself has an account, and is never carried by the machine's runtime, which runs as the @@ -227,7 +232,7 @@ func Users(r Records) ([]Principal, error) { if runtimeHere { var carried []Declared for _, d := range r.Assigned[node] { - if d.RunsAs == "" && runnerServes(d) { + if CarriedByRunner(d) { // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). d.Hears = ComposedHears(d) carried = append(carried, d) @@ -454,3 +459,15 @@ func declaredStreamsOf(r Records) []string { sort.Strings(out) return out } + +// EngineModule is the node-engine's manifest's module name (mesh-sdk buscompose.EngineModule). +const EngineModule = "mesh-host" + +// Composed says a placed module is composed as one: its own user, its processes and what the runner carries of it. The +// node-engine's manifest is not, though a placement may name it: each node's engine speaks as the node's account, which +// alone writes the engine's state (ADR 0563 §2, §3); granted to another account it is a second writer, and the writers +// table refuses the whole list (issue 774: mesh-host_acts, then mesh-host_health, 2026-10-30). mesh-sdk buscompose.Composed. +func Composed(d Declared) bool { return d.Module != EngineModule } + +// CarriedByRunner says the machine's tool runner carries this placed module (mesh-sdk buscompose.CarriedByRunner). +func CarriedByRunner(d Declared) bool { return Composed(d) && d.RunsAs == "" && runnerServes(d) } diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index 9ca930fb..e79ab5c7 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -6,7 +6,9 @@ import ( "git.novox.be/novox/mesh-sdk/go/announce" "slices" "strings" + "sync" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" "github.com/novox/mesh-controller/internal/broker" ) @@ -119,22 +121,18 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) { // announcement does not make readable is granted nothing, and said; an owner that announced nothing yet is read as // before. Announcements that cannot be read are said, and every read is granted as before: the bus's users are // composed either way. + // + // **Judged for its reader, by the SDK's judgement** (sweep 2026-10-30, issue 774): mesh-sdk buscompose's + // ReadsAnnouncedBy knows the named readers (ADR 0547 clause 9, ADR 0561 §3), so the lease is granted to the parts + // named for it and refused to every other. The controller's own reader-less copy refused every named reader: 2536 + // refused reads of mesh-controller.lease on 2026-10-30, each said on every composition. if Announcements != nil { announced, err := Announcements(ctx) if err != nil { fmt.Printf("the states the modules announced cannot be read, so every read is granted from `reads` as before: %v\n", err) } else { out.Announced = announced - for node, ds := range out.Assigned { - for k := range ds { - granted, refused := broker.ReadsAnnounced(ds[k].Reads, announced) - ds[k].Reads, ds[k].ReadsRefused = granted, refused - for _, r := range refused { - fmt.Printf("%s on %s reads %s, which its owner does not announce readable: not granted\n", - ds[k].Module, node, r) - } - } - } + judgeReads(out.Assigned, announced, sayRefusedOnce) } } @@ -615,3 +613,29 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node } return out, touches, onEvents } + +// refusalsSaid are the refused reads already said by this process, by `@ `: a refusal is said once, +// when first composed, not on every composition (about 420 times a read on 2026-10-30). `status` names them all. +var refusalsSaid sync.Map + +// sayRefusedOnce says a refused read the first time this process composes it. +func sayRefusedOnce(module, node, read string) { + if _, said := refusalsSaid.LoadOrStore(module+"@"+node+" "+read, true); said { + return + } + fmt.Printf("%s on %s reads %s, which its owner does not make readable by it: not granted\n", module, node, read) +} + +// judgeReads splits each assigned module's reads into granted and refused, for that module, by what the owners announced +// and the states' named readers (buscompose.ReadsAnnouncedBy), and says each refusal. +func judgeReads(assigned map[string][]broker.Declared, announced map[string]announce.States, say func(module, node, read string)) { + for node, ds := range assigned { + for k := range ds { + granted, refused := buscompose.ReadsAnnouncedBy(ds[k].Module, ds[k].Reads, announced) + ds[k].Reads, ds[k].ReadsRefused = granted, refused + for _, r := range refused { + say(ds[k].Module, node, r) + } + } + } +} diff --git a/internal/inventory/judge_reads_test.go b/internal/inventory/judge_reads_test.go new file mode 100644 index 00000000..9fbb5cd2 --- /dev/null +++ b/internal/inventory/judge_reads_test.go @@ -0,0 +1,44 @@ +package inventory + +import ( + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/announce" + "github.com/novox/mesh-controller/internal/broker" +) + +// Sweep 2026-10-30 (issue 774): a read is judged for its reader. The controller's lease, which its announcement marks +// readable by nobody, is granted to the parts named for it (ADR 0561 §3) and refused to every other (mesh-config named since mesh-sdk ef1ec1c9); a state its owner +// announced readable is granted; one it keeps and did not is refused; an owner that announced nothing is read as before. +// Each refusal is said, and a refusal already said is not said again. +func TestAReadIsJudgedForItsReader(t *testing.T) { + announced := map[string]announce.States{ + "mesh-controller": broker.ControllerAnnouncement("novox"), + "owner": {Module: "owner", States: []announce.State{{Name: "open", Readable: true}, {Name: "closed"}}}, + } + for name, c := range map[string]struct { + reader, read string + granted bool + }{ + "the lease, by the vault": {"mesh-vault", "mesh-controller.lease", true}, + "the lease, by the grants part": {"mesh-grants", "mesh-controller.lease", true}, + "the lease, by mesh-config": {"mesh-config", "mesh-controller.lease", true}, + "the lease, by a module not named": {"restic", "mesh-controller.lease", false}, + "a readable state": {"reader", "owner.open", true}, + "a state kept and not readable": {"reader", "owner.closed", false}, + "an owner that announced nothing": {"reader", "quiet.any", true}, + "a mirror the controller marks open": {"pacman", "mesh-controller.declared-packages", true}, + } { + assigned := map[string][]broker.Declared{"novox": {{Module: c.reader, Reads: []string{c.read}}}} + var said []string + judgeReads(assigned, announced, func(module, node, read string) { said = append(said, read) }) + d := assigned["novox"][0] + if slices.Contains(d.Reads, c.read) != c.granted || slices.Contains(d.ReadsRefused, c.read) == c.granted { + t.Errorf("%s: granted %v, refused %v", name, d.Reads, d.ReadsRefused) + } + if (len(said) == 1) == c.granted { + t.Errorf("%s: said %v", name, said) + } + } +} diff --git a/internal/inventory/provisioned.go b/internal/inventory/provisioned.go index ba2c8f0e..4423a6b2 100644 --- a/internal/inventory/provisioned.go +++ b/internal/inventory/provisioned.go @@ -26,6 +26,17 @@ type ProvReport struct { WaitsFor []string `json:"waits-for,omitempty"` Templates string `json:"templates,omitempty"` At time.Time `json:"at"` + // FailingSince is, on a failed report, when the module's standing failure began, kept by the client while only its + // words change (mesh-catalog 3aae6c08); zero from an older client. + FailingSince time.Time `json:"failing-since,omitzero"` +} + +// Since is when a failed report's failure began: FailingSince when the client says it, else At. +func (r ProvReport) Since() time.Time { + if !r.FailingSince.IsZero() { + return r.FailingSince + } + return r.At } // The words a client's report says. diff --git a/internal/link/calls.go b/internal/link/calls.go index 6332c705..c91512e6 100644 --- a/internal/link/calls.go +++ b/internal/link/calls.go @@ -127,9 +127,13 @@ type CallLog struct { // keeper keeps every call beyond this process, when Durably was given one; writes go through // one goroutine, in order, so a call's last state is the one kept. - keeper CallKeeper - holder string - writes chan Call + keeper CallKeeper + holder string + // queued are the calls' states waiting for the keeper, one per call, in the order each call was first queued, under + // qmu (keep is called with mu held); woken is told when one is queued (callQueue). + qmu sync.Mutex + queued callQueue + woken chan struct{} logger *log.Logger lost int // writes the keeper could not take, said once each keepErr error @@ -165,8 +169,8 @@ const keepTries = 5 func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, logger *log.Logger) error { l.mu.Lock() l.keeper, l.holder, l.logger = keeper, holder, logger - if l.writes == nil { - l.writes = make(chan Call, 256) + if l.woken == nil { + l.woken = make(chan struct{}, 1) go l.keepWrites() } l.mu.Unlock() @@ -203,51 +207,109 @@ func orSomebody(s string) string { return s } +// callQueueBound is how many calls may wait for the keeper at once: far beyond a burst of reads (about 200 calls in one +// second on 2026-10-30), and a bound, so a keeper that takes nothing for long is said rather than grown into. +const callQueueBound = 8192 + +// callQueue holds the latest state of each call waiting for the keeper. **One state per call** (sweep 2026-10-30, issue +// 774): a call's later state replaces its earlier one still waiting, since only the last is kept anyway; a burst of +// read calls queued begin and finish each, and a channel of 256 dropped 30 of them in one second. +type callQueue struct { + order []string + byID map[string]Call +} + +// put queues c, replacing a waiting state of the same call; false when the queue is full. +func (q *callQueue) put(c Call) bool { + if q.byID == nil { + q.byID = map[string]Call{} + } + if _, waiting := q.byID[c.ID]; waiting { + q.byID[c.ID] = c + return true + } + if len(q.order) >= callQueueBound { + return false + } + q.order = append(q.order, c.ID) + q.byID[c.ID] = c + return true +} + +// take is every waiting state, in order, and empties the queue. +func (q *callQueue) take() []Call { + out := make([]Call, 0, len(q.order)) + for _, id := range q.order { + out = append(out, q.byID[id]) + } + q.order, q.byID = nil, nil + return out +} + // keep queues one call's state for the keeper. Never blocks a call: a queue that is full is a keeper // that is not taking writes, and that is said rather than waited on. func (l *CallLog) keep(c Call) { - if l.writes == nil { + if l.woken == nil { + return + } + l.qmu.Lock() + queued := l.queued.put(c) + if !queued { + l.lost++ + } + lost := l.lost + l.qmu.Unlock() + if !queued { + if l.logger != nil { + l.logger.Printf("%s (%s.%s) is kept in memory only: %d calls wait for the bus to take their records "+ + "(%d not kept)", c.ID, c.Seat, c.Verb, callQueueBound, lost) + } return } select { - case l.writes <- c: + case l.woken <- struct{}{}: default: - l.lost++ - if l.logger != nil { - l.logger.Printf("%s (%s.%s) is kept in memory only: the bus is not taking calls' records (%d not kept)", - c.ID, c.Seat, c.Verb, l.lost) - } } } func (l *CallLog) keepWrites() { - for c := range l.writes { - l.mu.Lock() - gate := l.epoch - l.mu.Unlock() - if gate != nil { - if _, err := gate(); err != nil { - if l.logger != nil { - l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+ - "marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err) - } - continue + for range l.woken { + l.qmu.Lock() + waiting := l.queued.take() + l.qmu.Unlock() + for _, c := range waiting { + l.keepOne(c) + } + } +} + +// keepOne writes one call's state through the keeper, under the lease. +func (l *CallLog) keepOne(c Call) { + l.mu.Lock() + gate := l.epoch + l.mu.Unlock() + if gate != nil { + if _, err := gate(); err != nil { + if l.logger != nil { + l.logger.Printf("%s (%s.%s, %s) is not kept on the bus: %v — the controller holding the lease "+ + "marks it abandoned", c.ID, c.Seat, c.Verb, c.State, err) } + return } - var err error - for try := 0; try < keepTries; try++ { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - err = l.keeper.Keep(ctx, c) - cancel() - if err == nil { - break - } - time.Sleep(time.Duration(try+1) * time.Second) - } - if err != nil && l.logger != nil { - l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+ - "answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err) + } + var err error + for try := 0; try < keepTries; try++ { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err = l.keeper.Keep(ctx, c) + cancel() + if err == nil { + break } + time.Sleep(time.Duration(try+1) * time.Second) + } + if err != nil && l.logger != nil { + l.logger.Printf("%s (%s.%s, %s) could not be kept on the bus after %d tries: %v — `calls` "+ + "answers it from memory until this controller stops", c.ID, c.Seat, c.Verb, c.State, keepTries, err) } } diff --git a/internal/link/calls_test.go b/internal/link/calls_test.go index 577f7c22..2b77411e 100644 --- a/internal/link/calls_test.go +++ b/internal/link/calls_test.go @@ -254,8 +254,7 @@ func TestAHandoverRefusalIsMarkedRetryable(t *testing.T) { // who may call the seat, and a token is the one-time right to become a machine of the mesh. func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) { l, a := NewCallLog(), newAnswers(t) - writes := make(chan Call, 4) - l.writes = writes + l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below l.serveCall("mesh-controller", "token", json.RawMessage(`{"new":"laptop"}`), "_INBOX.x.1", func(context.Context, json.RawMessage) (any, error) { return "token for laptop: s3cret-join", nil }, a.respond, nil) @@ -266,8 +265,7 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) { if len(recent) != 1 || strings.Contains(string(recent[0].Answer), "s3cret-join") { t.Fatalf("the token was kept in memory: %+v", recent) } - close(writes) - for c := range writes { + for _, c := range l.queued.take() { if strings.Contains(string(c.Answer), "s3cret-join") { t.Fatalf("the token was sent to be kept on the bus: %s", c.Answer) } @@ -276,3 +274,33 @@ func TestAJoinTokenIsShownToItsCallerAndNotKept(t *testing.T) { // A verb whose handler panics answers an error, and the controller serves the next call (review of // novox/hq issue 327: read verbs are answered in the serving process now). + +// Sweep 2026-10-30 (issue 774): a burst of calls never loses a record while the keeper is slow: each call waits once, +// with its latest state, and every call of a burst far over the old queue of 256 is kept with its last state. +func TestABurstOfCallsIsKeptWithEachCallsLastState(t *testing.T) { + var q callQueue + for i := range 2000 { + id := fmt.Sprintf("call-%d", i) + if !q.put(Call{ID: id, State: CallRunning}) || !q.put(Call{ID: id, State: CallAnswered}) { + t.Fatalf("%s was refused by a queue of %d", id, callQueueBound) + } + } + got := q.take() + if len(got) != 2000 { + t.Fatalf("%d calls wait, want 2000", len(got)) + } + for i, c := range got { + if c.ID != fmt.Sprintf("call-%d", i) || c.State != CallAnswered { + t.Fatalf("call %d waits as %+v", i, c) + } + } + if len(q.take()) != 0 { + t.Fatal("taken calls wait again") + } + for i := range callQueueBound { + q.put(Call{ID: fmt.Sprint(i)}) + } + if q.put(Call{ID: "one too many"}) { + t.Fatal("the queue grew past its bound") + } +} 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/meshcli_test.go b/internal/link/meshcli_test.go index 7e5db365..c2077e10 100644 --- a/internal/link/meshcli_test.go +++ b/internal/link/meshcli_test.go @@ -125,8 +125,7 @@ func askCLI(t *testing.T, conn *nats.Conn, node string, asked CLIAsked) (CLIAnsw // bus, where `calls` answers anyone who may call the seat. func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) { l, a := NewCallLog(), newAnswers(t) - writes := make(chan Call, 4) - l.writes = writes + l.woken = make(chan struct{}, 1) // queued for the keeper, and nothing takes them: read below asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"}) l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv", func(context.Context, json.RawMessage) (any, error) { @@ -134,8 +133,7 @@ func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) { }, a.respond, nil) _ = a.only() - close(writes) - for c := range writes { + for _, c := range l.queued.take() { if carries(c.Args, "s3cret") || carries(c.Answer, "s3cret-join") { t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer) } diff --git a/internal/link/movesecret.go b/internal/link/movesecret.go index bcaa2e6a..ef498103 100644 --- a/internal/link/movesecret.go +++ b/internal/link/movesecret.go @@ -52,6 +52,9 @@ type MoveSecretAnswer struct { Secret string `json:"secret,omitempty"` Generation int `json:"generation,omitempty"` Unchanged bool `json:"unchanged,omitempty"` + // Superseded says the vault holds another value of the secret, which is the live one, and this machine's older value + // was not moved and is not asked again (mesh-host b9c24a7, sweep 2026-10-30). + Superseded bool `json:"superseded,omitempty"` } // MoveSecretWithin is how long the controller waits for one move: the engine asks the vault twice. diff --git a/internal/link/movesecret_test.go b/internal/link/movesecret_test.go index cfad665f..d72654a1 100644 --- a/internal/link/movesecret_test.go +++ b/internal/link/movesecret_test.go @@ -30,8 +30,9 @@ func TestTheMovesAskAndAnswerKeepTheirFieldNames(t *testing.T) { "provision sealed sealed_to" { t.Fatalf("the ask's fields are %q", got) } - answer := MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true} - if got := keys(answer); got != "generation refused said secret unchanged" { + answer := MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true, + Superseded: true} + if got := keys(answer); got != "generation refused said secret superseded unchanged" { t.Fatalf("the answer's fields are %q", got) } if broker.AskMoveSecretSubject("laptop") != "mesh.node.laptop.ask.move-secret" || 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. diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 9fb61a4d..1b10465e 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself. mesh-catalog f3a57dcc56fb6c198095cafd11d0decff9f720c4 modules/*/module.json, modules/*/files, modules/*/templates, predefined.json mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock -mesh-sdk 2da0de75c3f0087cff9c2f5ef789ca6dc3c59045 conformance/events/module-event.json +mesh-sdk a37181261c24a15b8092c06e11cce63e0e5f8ade conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: