diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 6ff9aed..b146faa 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -196,6 +196,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The toolchains first, so a build that starts from the first modules value finds the nodes. watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger) + // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's + // assignments are settled first, so the toolchains settle through the gate already knowing them. + gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf) + watch.put = gate.Toolchain + assignErrs, err := watchAssignments(ctx, js, gate, b.NodesChanged, logger) + if err != nil { + return err + } if err := watch.start(); err != nil { return err } @@ -268,6 +276,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err + case err := <-assignErrs: + return err case err := <-watch.errs: return err } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go @@ -0,0 +1,174 @@ +package main + +// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves +// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very +// toolchain agreement this file repairs. + +import ( + "context" + "encoding/json" + "fmt" + "log" + "maps" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provider" +) + +// A build module writes . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; it is passed on as deleted otherwise, +// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot +// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that +// refusal is the guard of novox/hq ADR 0465, and it stays. +type toolchainGate struct { + mu sync.Mutex + modules map[string]bool // the build modules whose toolchains pass the gate + forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds + logf func(format string, a ...any) + known bool // the assignments are settled + standing map[string]bool // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / said as left out, so it is said once +} + +func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate { + g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{}, + raw: map[string]map[string][]byte{}, left: map[string]bool{}} + for _, m := range modules { + g.modules[m] = true + } + return g +} + +// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed. +func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.raw[module] == nil { + g.raw[module] = map[string][]byte{} + } + if deleted { + delete(g.raw[module], key) + } else { + g.raw[module][key] = value + } + return g.pass(module, key) +} + +// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed. +func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool { + node, module, ok := strings.Cut(key, ".") + if !ok || !g.modules[module] { + return false + } + open := !deleted && openAssignment(value) + g.mu.Lock() + defer g.mu.Unlock() + if g.standing[key] == open { + return false + } + if open { + g.standing[key] = true + } else { + delete(g.standing, key) + } + if !g.known { + return false + } + changed := false + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it +// passed on changed. +func (g *toolchainGate) Settled() bool { + g.mu.Lock() + defer g.mu.Unlock() + g.known = true + changed := false + for _, module := range slices.Sorted(maps.Keys(g.raw)) { + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted +// otherwise. The caller holds mu. +func (g *toolchainGate) pass(module, key string) bool { + value, held := g.raw[module][key] + if !held { + return g.forward(module, key, nil, true) + } + node, _, err := delivery.ParseToolchainKey(key) + if err != nil || !g.known || g.standing[node+"."+module] { + delete(g.left, module+"/"+key) + return g.forward(module, key, value, false) + } + if !g.left[module+"/"+key] { + g.left[module+"/"+key] = true + g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)", + delivery.ToolchainsBucket(module), key, module, node) + } + return g.forward(module, key, nil, true) +} + +// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not +// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment). +func openAssignment(value []byte) bool { + var a struct { + Ended time.Time `json:"ended"` + State string `json:"state"` + } + if json.Unmarshal(value, &a) != nil { + return false + } + return a.Ended.IsZero() && a.State == "" +} + +// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is +// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read +// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel. +func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) { + errs := make(chan error, 1) + kv, err := js.KeyValue(ctx, provider.AssignmentsBucket) + if err != nil { + // Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery). + logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil + } + each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) } + w, err := settle(ctx, kv, each) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + } + if gate.Settled() { + changed(ctx) + } + go func() { + errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { + if gate.Assignment(e.Key(), e.Value(), !opOf(e)) { + changed(ctx) + } + }) + }() + return errs, nil +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go @@ -0,0 +1,112 @@ +package main + +// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +func gateReport(t *testing.T, node, version string) (string, []byte) { + t.Helper() + key, err := delivery.ToolchainKey(node, "typescript") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()}) + if err != nil { + t.Fatal(err) + } + return key, value +} + +func agreedOf(set *buildrun.ToolchainSet) (string, string) { + tc, none := delivery.AgreedToolchain(set.Entries("typescript")) + return tc.Version, none +} + +// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the +// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them +// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again. +func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { + set := &buildrun.ToolchainSet{} + var said []string + gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) }) + for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} { + key, value := gateReport(t, node, version) + gate.Toolchain(buildrun.NodeBuildModule, key, value, false) + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") { + t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none) + } + for _, node := range []string{"g14", "novox", "shanks"} { + gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false) + } + gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false) + gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter + if !gate.Settled() { + t.Fatal("settling the assignments left ace's report out, so what was passed on changed") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none) + } + if len(said) != 1 { + t.Fatalf("ace's report is said left out once; said %v", said) + } + // ace reports again while unassigned (an old value seen again): still left out. + key, value := gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an unassigned node's report changed what was passed on") + } + // Assigned to ace again: its report counts, and the builders really disagree: the guard stays. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) { + t.Fatal("assigning ace again passed nothing on") + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { + t.Fatalf("builders that run different versions still disagree; got %q", none) + } + // The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) { + t.Fatal("a pending assignment is no assignment on the machine") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } + gate.Assignment("ace.mesh-build-node", nil, true) + if v, _ := agreedOf(set); v != "v26.12.1" { + t.Fatalf("got %q", v) + } + // A report deleted from the bucket goes whether its node is assigned or not. + key, _ = gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) { + t.Fatal("deleting g14's report passed nothing on") + } + if n := len(set.Entries("typescript")); n != 2 { + t.Fatalf("novox and shanks are left; got %d entries", n) + } +} + +// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes. +func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { + set := &buildrun.ToolchainSet{} + gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {}) + gate.Settled() + if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) { + t.Fatal("an assignment with no report passed something on") + } + key, value := gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an assigned node's report was not passed on") + } + key, value = gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("a report of a node with no assignment key was passed on after the assignments settled") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go index 95b6fc5..69faff2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -23,9 +23,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO" // §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key // (toolchain.reported, handled by Reported); a bucket not watched yet is then watched. type toolchainWatch struct { - ctx context.Context - js bucketOpener - set *buildrun.ToolchainSet + ctx context.Context + js bucketOpener + set *buildrun.ToolchainSet + // put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940). + put func(module, key string, value []byte, deleted bool) bool changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again logger *log.Logger errs chan error @@ -34,7 +36,7 @@ type toolchainWatch struct { } func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch { - return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} + return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} } // start watches the buckets that exist now; a missing one is said and waits for its build module's event. @@ -62,14 +64,14 @@ func (w *toolchainWatch) ensure(module string) error { case err != nil: return fmt.Errorf("the bucket %s: %w", name, err) } - watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) }) + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { return err } w.watched[module] = true go func() { w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { - if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) { + if w.put(module, e.Key(), e.Value(), !opOf(e)) { w.changed(w.ctx) } }) diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 11be676..e5dd046 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -89,7 +89,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and // holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5). - if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", + if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 0b7a72f..0333c9a 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -20,6 +20,7 @@ "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-controller.assignments", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules" diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 34db5d7..297cfce 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -257,6 +257,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The toolchains first, so a check that starts from the first candidate finds the nodes. watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger) + // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's + // assignments are settled first, so the toolchains settle through the gate already knowing them. + gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf) + watch.put = gate.Toolchain + assignErrs, err := watchAssignments(ctx, js, gate, ch.NodesChanged, logger) + if err != nil { + return err + } if err := watch.start(); err != nil { return err } @@ -369,6 +377,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err + case err := <-assignErrs: + return err case err := <-watch.errs: return err } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go @@ -0,0 +1,174 @@ +package main + +// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves +// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very +// toolchain agreement this file repairs. + +import ( + "context" + "encoding/json" + "fmt" + "log" + "maps" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provider" +) + +// A build module writes . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; it is passed on as deleted otherwise, +// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot +// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that +// refusal is the guard of novox/hq ADR 0465, and it stays. +type toolchainGate struct { + mu sync.Mutex + modules map[string]bool // the build modules whose toolchains pass the gate + forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds + logf func(format string, a ...any) + known bool // the assignments are settled + standing map[string]bool // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / said as left out, so it is said once +} + +func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate { + g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{}, + raw: map[string]map[string][]byte{}, left: map[string]bool{}} + for _, m := range modules { + g.modules[m] = true + } + return g +} + +// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed. +func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.raw[module] == nil { + g.raw[module] = map[string][]byte{} + } + if deleted { + delete(g.raw[module], key) + } else { + g.raw[module][key] = value + } + return g.pass(module, key) +} + +// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed. +func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool { + node, module, ok := strings.Cut(key, ".") + if !ok || !g.modules[module] { + return false + } + open := !deleted && openAssignment(value) + g.mu.Lock() + defer g.mu.Unlock() + if g.standing[key] == open { + return false + } + if open { + g.standing[key] = true + } else { + delete(g.standing, key) + } + if !g.known { + return false + } + changed := false + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it +// passed on changed. +func (g *toolchainGate) Settled() bool { + g.mu.Lock() + defer g.mu.Unlock() + g.known = true + changed := false + for _, module := range slices.Sorted(maps.Keys(g.raw)) { + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted +// otherwise. The caller holds mu. +func (g *toolchainGate) pass(module, key string) bool { + value, held := g.raw[module][key] + if !held { + return g.forward(module, key, nil, true) + } + node, _, err := delivery.ParseToolchainKey(key) + if err != nil || !g.known || g.standing[node+"."+module] { + delete(g.left, module+"/"+key) + return g.forward(module, key, value, false) + } + if !g.left[module+"/"+key] { + g.left[module+"/"+key] = true + g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)", + delivery.ToolchainsBucket(module), key, module, node) + } + return g.forward(module, key, nil, true) +} + +// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not +// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment). +func openAssignment(value []byte) bool { + var a struct { + Ended time.Time `json:"ended"` + State string `json:"state"` + } + if json.Unmarshal(value, &a) != nil { + return false + } + return a.Ended.IsZero() && a.State == "" +} + +// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is +// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read +// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel. +func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) { + errs := make(chan error, 1) + kv, err := js.KeyValue(ctx, provider.AssignmentsBucket) + if err != nil { + // Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery). + logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil + } + each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) } + w, err := settle(ctx, kv, each) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + } + if gate.Settled() { + changed(ctx) + } + go func() { + errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { + if gate.Assignment(e.Key(), e.Value(), !opOf(e)) { + changed(ctx) + } + }) + }() + return errs, nil +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go @@ -0,0 +1,112 @@ +package main + +// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +func gateReport(t *testing.T, node, version string) (string, []byte) { + t.Helper() + key, err := delivery.ToolchainKey(node, "typescript") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()}) + if err != nil { + t.Fatal(err) + } + return key, value +} + +func agreedOf(set *buildrun.ToolchainSet) (string, string) { + tc, none := delivery.AgreedToolchain(set.Entries("typescript")) + return tc.Version, none +} + +// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the +// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them +// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again. +func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { + set := &buildrun.ToolchainSet{} + var said []string + gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) }) + for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} { + key, value := gateReport(t, node, version) + gate.Toolchain(buildrun.NodeBuildModule, key, value, false) + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") { + t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none) + } + for _, node := range []string{"g14", "novox", "shanks"} { + gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false) + } + gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false) + gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter + if !gate.Settled() { + t.Fatal("settling the assignments left ace's report out, so what was passed on changed") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none) + } + if len(said) != 1 { + t.Fatalf("ace's report is said left out once; said %v", said) + } + // ace reports again while unassigned (an old value seen again): still left out. + key, value := gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an unassigned node's report changed what was passed on") + } + // Assigned to ace again: its report counts, and the builders really disagree: the guard stays. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) { + t.Fatal("assigning ace again passed nothing on") + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { + t.Fatalf("builders that run different versions still disagree; got %q", none) + } + // The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) { + t.Fatal("a pending assignment is no assignment on the machine") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } + gate.Assignment("ace.mesh-build-node", nil, true) + if v, _ := agreedOf(set); v != "v26.12.1" { + t.Fatalf("got %q", v) + } + // A report deleted from the bucket goes whether its node is assigned or not. + key, _ = gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) { + t.Fatal("deleting g14's report passed nothing on") + } + if n := len(set.Entries("typescript")); n != 2 { + t.Fatalf("novox and shanks are left; got %d entries", n) + } +} + +// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes. +func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { + set := &buildrun.ToolchainSet{} + gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {}) + gate.Settled() + if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) { + t.Fatal("an assignment with no report passed something on") + } + key, value := gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an assigned node's report was not passed on") + } + key, value = gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("a report of a node with no assignment key was passed on after the assignments settled") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go index 7b650b0..bc848ca 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go @@ -24,9 +24,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO" // §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key // (toolchain.reported, handled by Reported); a bucket not watched yet is then watched. type toolchainWatch struct { - ctx context.Context - js bucketOpener - set *buildrun.ToolchainSet + ctx context.Context + js bucketOpener + set *buildrun.ToolchainSet + // put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940). + put func(module, key string, value []byte, deleted bool) bool changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again logger *log.Logger errs chan error @@ -35,7 +37,7 @@ type toolchainWatch struct { } func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch { - return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} + return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} } // start watches the buckets that exist now; a missing one is said and waits for its build module's event. @@ -63,14 +65,14 @@ func (w *toolchainWatch) ensure(module string) error { case err != nil: return fmt.Errorf("the bucket %s: %w", name, err) } - watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) }) + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) if err != nil { return err } w.watched[module] = true go func() { w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { - if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) { + if w.put(module, e.Key(), e.Value(), !opOf(e)) { w.changed(w.ctx) } }) diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 4a9f21b..c6af543 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", + if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 1dd6ce1..67c97f2 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -21,6 +21,7 @@ "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-controller.assignments", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 110ca94..d9142a5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -549,3 +549,88 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC t.Errorf("avahi = fp2 %q none2 %q within 5 s of the start, want a fingerprint and no reason", got.FP2, got.None2) } } + +// TestIndexerCountsOnlyTheAssignedBuildNodesToolchains (issue 1940): the toolchains bucket holds the report of n2, where +// mesh-build-go was unassigned (its assignment ended) at another version than n1's, as ace's v26.12.0 stayed beside the +// v26.12.1 of the nodes mesh-build-node runs on. The indexer gives the merged module a fingerprint from n1's toolchain +// alone; when the controller assigns mesh-build-go to n2 again, its report counts and the builders disagree (the guard +// stays); when the assignment ends again, the fingerprint comes back. No restart, no timer: the assignments' watch. +func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) { + manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") + if err != nil { + t.Fatal(err) + } + const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" + var assignments jetstream.KeyValue + js, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1}) + if err != nil { + t.Fatal(err) + } + for node, version := range map[string]string{"n1": "go1.27.3", "n2": "go1.27.2"} { + key, _ := delivery.ToolchainKey(node, "go") + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: version, Recipe: "1", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, value); err != nil { + t.Fatal(err) + } + } + // The controller's assignments mirror, in its layout (mesh-controller internal/chain Assignment). + if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_assignments", History: 1}); err != nil { + t.Fatal(err) + } + for key, value := range map[string]string{ + "n1.mesh-build-go": `{"node":"n1","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z"}`, + "n2.mesh-build-go": `{"node":"n2","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z","ended":"2026-10-23T08:00:00Z"}`, + } { + if _, err := assignments.Put(ctx, key, []byte(value)); err != nil { + t.Fatal(err) + } + } + }) + data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 302, + Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", + Trees: map[string]string{"modules/avahi": "newtree"}, + Manifests: map[string]string{"modules/avahi": string(manifest)}, + Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}}) + if err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.merged", data); err != nil { + t.Fatal(err) + } + read := func(ok func(delivery.Module) bool) delivery.Module { + var got delivery.Module + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := modules.Get(ctx, "avahi"); err == nil { + if m, err := delivery.DecodeModule(e.Value()); err == nil { + if got = m; ok(m) { + break + } + } + } + } + return got + } + agreed := func(m delivery.Module) bool { return m.FP2 != "" && m.Toolchains["go"].Version == "go1.27.3" } + if m := read(agreed); !agreed(m) { + t.Fatalf("within 5 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", m.FP2, m.None2, m.Toolchains) + } + if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","assigned":"2026-10-24T08:00:00Z"}`)); err != nil { + t.Fatal(err) + } + disagree := func(m delivery.Module) bool { + return m.FP2 == "" && strings.HasPrefix(m.None2, "builders disagree: go1.27.2 recipe 1 on n2") + } + if m := read(disagree); !disagree(m) { + t.Fatalf("within 5 s of n2's assignment avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2) + } + if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","ended":"2026-10-24T09:00:00Z"}`)); err != nil { + t.Fatal(err) + } + if m := read(agreed); !agreed(m) { + t.Fatalf("within 5 s of n2's unassignment avahi has fp2 %q, none2 %q", m.FP2, m.None2) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index d281800..fbf6ff3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -313,7 +313,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } go func() { sourceErrs <- follow(ctx, sw, each) }() } - tools := newToolchainWatch(ctx, js, toolchainSink{put: ix.PutToolchain, apply: ix.ApplyToolchain, recompute: ix.Recompute}, logger) + // Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's + // assignments are settled first, so the toolchains settle through the gate already knowing them. + gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) + assignErrs, err := watchAssignments(ctx, js, gate, func(ctx context.Context) { + if err := ix.Recompute(ctx); err != nil { + logger.Printf("writing the modules again after an assignment of a build module changed: %v", err) + } + }, logger) + if err != nil { + return err + } + tools := newToolchainWatch(ctx, js, toolchainSink{ + put: gate.Toolchain, + apply: func(ctx context.Context, module, key string, value []byte, deleted bool) error { + if !gate.Toolchain(module, key, value, deleted) { + return nil + } + return ix.Recompute(ctx) + }, + recompute: ix.Recompute}, logger) if err := tools.start(); err != nil { return err } @@ -377,6 +396,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-sourceErrs: return err + case err := <-assignErrs: + return err case err := <-tools.errs: return err case err := <-previews.errs: diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go @@ -0,0 +1,174 @@ +package main + +// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves +// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very +// toolchain agreement this file repairs. + +import ( + "context" + "encoding/json" + "fmt" + "log" + "maps" + "slices" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provider" +) + +// A build module writes . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; it is passed on as deleted otherwise, +// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot +// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that +// refusal is the guard of novox/hq ADR 0465, and it stays. +type toolchainGate struct { + mu sync.Mutex + modules map[string]bool // the build modules whose toolchains pass the gate + forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds + logf func(format string, a ...any) + known bool // the assignments are settled + standing map[string]bool // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / said as left out, so it is said once +} + +func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate { + g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{}, + raw: map[string]map[string][]byte{}, left: map[string]bool{}} + for _, m := range modules { + g.modules[m] = true + } + return g +} + +// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed. +func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.raw[module] == nil { + g.raw[module] = map[string][]byte{} + } + if deleted { + delete(g.raw[module], key) + } else { + g.raw[module][key] = value + } + return g.pass(module, key) +} + +// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed. +func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool { + node, module, ok := strings.Cut(key, ".") + if !ok || !g.modules[module] { + return false + } + open := !deleted && openAssignment(value) + g.mu.Lock() + defer g.mu.Unlock() + if g.standing[key] == open { + return false + } + if open { + g.standing[key] = true + } else { + delete(g.standing, key) + } + if !g.known { + return false + } + changed := false + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it +// passed on changed. +func (g *toolchainGate) Settled() bool { + g.mu.Lock() + defer g.mu.Unlock() + g.known = true + changed := false + for _, module := range slices.Sorted(maps.Keys(g.raw)) { + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted +// otherwise. The caller holds mu. +func (g *toolchainGate) pass(module, key string) bool { + value, held := g.raw[module][key] + if !held { + return g.forward(module, key, nil, true) + } + node, _, err := delivery.ParseToolchainKey(key) + if err != nil || !g.known || g.standing[node+"."+module] { + delete(g.left, module+"/"+key) + return g.forward(module, key, value, false) + } + if !g.left[module+"/"+key] { + g.left[module+"/"+key] = true + g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)", + delivery.ToolchainsBucket(module), key, module, node) + } + return g.forward(module, key, nil, true) +} + +// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not +// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment). +func openAssignment(value []byte) bool { + var a struct { + Ended time.Time `json:"ended"` + State string `json:"state"` + } + if json.Unmarshal(value, &a) != nil { + return false + } + return a.Ended.IsZero() && a.State == "" +} + +// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is +// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read +// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel. +func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) { + errs := make(chan error, 1) + kv, err := js.KeyValue(ctx, provider.AssignmentsBucket) + if err != nil { + // Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery). + logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) + return errs, nil + } + each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) } + w, err := settle(ctx, kv, each) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err) + } + if gate.Settled() { + changed(ctx) + } + go func() { + errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { + if gate.Assignment(e.Key(), e.Value(), !opOf(e)) { + changed(ctx) + } + }) + }() + return errs, nil +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go @@ -0,0 +1,112 @@ +package main + +// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +func gateReport(t *testing.T, node, version string) (string, []byte) { + t.Helper() + key, err := delivery.ToolchainKey(node, "typescript") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()}) + if err != nil { + t.Fatal(err) + } + return key, value +} + +func agreedOf(set *buildrun.ToolchainSet) (string, string) { + tc, none := delivery.AgreedToolchain(set.Entries("typescript")) + return tc.Version, none +} + +// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the +// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them +// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again. +func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) { + set := &buildrun.ToolchainSet{} + var said []string + gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) }) + for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} { + key, value := gateReport(t, node, version) + gate.Toolchain(buildrun.NodeBuildModule, key, value, false) + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") { + t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none) + } + for _, node := range []string{"g14", "novox", "shanks"} { + gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false) + } + gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false) + gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter + if !gate.Settled() { + t.Fatal("settling the assignments left ace's report out, so what was passed on changed") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none) + } + if len(said) != 1 { + t.Fatalf("ace's report is said left out once; said %v", said) + } + // ace reports again while unassigned (an old value seen again): still left out. + key, value := gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an unassigned node's report changed what was passed on") + } + // Assigned to ace again: its report counts, and the builders really disagree: the guard stays. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) { + t.Fatal("assigning ace again passed nothing on") + } + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { + t.Fatalf("builders that run different versions still disagree; got %q", none) + } + // The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again. + if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) { + t.Fatal("a pending assignment is no assignment on the machine") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } + gate.Assignment("ace.mesh-build-node", nil, true) + if v, _ := agreedOf(set); v != "v26.12.1" { + t.Fatalf("got %q", v) + } + // A report deleted from the bucket goes whether its node is assigned or not. + key, _ = gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) { + t.Fatal("deleting g14's report passed nothing on") + } + if n := len(set.Entries("typescript")); n != 2 { + t.Fatalf("novox and shanks are left; got %d entries", n) + } +} + +// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes. +func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { + set := &buildrun.ToolchainSet{} + gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {}) + gate.Settled() + if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) { + t.Fatal("an assignment with no report passed something on") + } + key, value := gateReport(t, "g14", "v26.12.1") + if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("an assigned node's report was not passed on") + } + key, value = gateReport(t, "ace", "v26.12.0") + if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { + t.Fatal("a report of a node with no assignment key was passed on after the assignments settled") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 0c28769..b8c3fc5 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -81,7 +81,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 755eaea..bd5a90e 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -16,6 +16,7 @@ "reads": [ "mesh-build-go.toolchains", "mesh-build-node.toolchains", + "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"