Count only the toolchain reports of nodes a build module is assigned to, so a node it left does not make the builders disagree (issue 1940)
mesh/delivery delivered
mesh/delivery delivered
mesh-build-node was unassigned from ace on 2026-10-23 and its ace.typescript key stayed in mesh-build-node_toolchains at v26.12.0: nothing deletes it, so the indexer, the builder and the checker took it for a builder and every Node fingerprint said 'builders disagree'. The three now read mesh-controller_assignments and pass a toolchain report on only while its build module has an open assignment on its node; builders that run different versions still disagree.
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 <node>.<language> 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 <node>.<module>) 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 // <node>.<module> with an open assignment
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]bool // <module>/<key> 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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user