From a4116a1de742b6df30a19147172ea5de5bada5f3 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 09:18:52 +0100 Subject: [PATCH] =?UTF-8?q?Builder,=20checker=20and=20indexer=20count=20a?= =?UTF-8?q?=20build=20node's=20toolchain=20only=20where=20the=20catalogue'?= =?UTF-8?q?s=20placement=20puts=20its=20build=20module,=20through=20mesh-s?= =?UTF-8?q?dk's=20one=20gate,=20and=20read=20the=20controller's=20assignme?= =?UTF-8?q?nts=20no=20more=20(issue=20774,=20ADR=200549=20=C2=A73,=20issue?= =?UTF-8?q?=201940)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-builder/main.go | 10 +- .../cmd/mesh-delivery-builder/refused_test.go | 73 ++++++ .../cmd/mesh-delivery-builder/standing.go | 211 ---------------- .../mesh-delivery-builder/standing_test.go | 228 ------------------ .../mesh-delivery-builder/manifest_test.go | 4 +- modules/mesh-delivery-builder/module.json | 4 +- .../cmd/mesh-delivery-checker/main.go | 10 +- .../cmd/mesh-delivery-checker/refused_test.go | 62 +++++ .../cmd/mesh-delivery-checker/standing.go | 211 ---------------- .../mesh-delivery-checker/standing_test.go | 218 ----------------- .../mesh-delivery-checker/manifest_test.go | 4 +- modules/mesh-delivery-checker/module.json | 2 +- .../mesh-delivery-indexer/integration_test.go | 64 +++-- .../cmd/mesh-delivery-indexer/main.go | 15 +- .../cmd/mesh-delivery-indexer/standing.go | 211 ---------------- .../mesh-delivery-indexer/standing_test.go | 165 ------------- .../mesh-delivery-indexer/manifest_test.go | 2 +- modules/mesh-delivery-indexer/module.json | 1 - 18 files changed, 200 insertions(+), 1295 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go delete mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go delete mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/refused_test.go delete mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go delete mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go delete mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go delete mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go 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 f003aae..7fb207c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -153,11 +153,11 @@ 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) + // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the + // placement is settled first, so the toolchains settle through the gate already knowing it. + gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf) watch.put = gate.Toolchain - assignErrs, err := watchAssignments(ctx, js, gate, b.NodesChanged, logger) + placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf) if err != nil { return err } @@ -254,7 +254,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err - case err := <-assignErrs: + case err := <-placementErrs: return err case err := <-watch.errs: return err diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go new file mode 100644 index 0000000..6f17ab0 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "bytes" + "context" + "errors" + "log" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/process" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the +// watch unstarted and the part running, said; a refused bucket it cannot work without is an error naming the bucket and +// its grant. The gate over the catalogue's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate, +// issue 774). +func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) { + set := &buildrun.ToolchainSet{} + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-build-node_toolchains")} + if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { + t.Errorf("a refused toolchains bucket stopped the start: %v", err) + } + if !strings.Contains(logs.String(), "cannot be read") { + t.Errorf("the refusals were not said: %q", logs.String()) + } + if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil || + !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { + t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) + } + // The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk + // head is told why and opens it again then. + src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf} + src.start() + if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) { + t.Errorf("a trunk head with the source refused: %v, want why", err) + } + if src.opened { + t.Error("a refused source counts as opened, so it is never opened again") + } +} + +// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. +type refusingBus struct { + refuseOpen bool + err error +} + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + if r.refuseOpen { + return nil, r.err + } + return refusedWatch{err: r.err}, nil +} + +type refusedWatch struct { + jetstream.KeyValue + err error +} + +func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err +} +func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.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 deleted file mode 100644 index 9a06cec..0000000 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go +++ /dev/null @@ -1,211 +0,0 @@ -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, and only once it was written since that -// assignment; it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report -// written before the assignment is the one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node -// assigned to ace again brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement, before the module ran -// there, and the controller's take-in refused every module "builders disagree"). 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]time.Time // . with an open assignment -> since when it stands - raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it - left map[string]string // / -> why it was said left out, so each reason 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]time.Time{}, - raw: map[string]map[string][]byte{}, left: map[string]string{}} - 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 (at is when the bus wrote that revision) and answers -// whether what it passed on changed. -func (g *toolchainGate) Assignment(key string, value []byte, deleted bool, at time.Time) bool { - node, module, ok := strings.Cut(key, ".") - if !ok || !g.modules[module] { - return false - } - assigned, open := openAssignment(value) - open = open && !deleted - g.mu.Lock() - defer g.mu.Unlock() - was, stood := g.standing[key] - since := assigned - if since.IsZero() { - // A value with no time of its own (an assignment a module pulled in) stands since the revision that opened it. - since = at - if stood { - since = was - } - } - if open == stood && (!open || since.Equal(was)) { - return false - } - if open { - g.standing[key] = since - } 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 and wrote it since (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) - why := "" - if err == nil && g.known { - why = g.leftOut(module, node, value) - } - if why == "" { - delete(g.left, module+"/"+key) - return g.forward(module, key, value, false) - } - if g.left[module+"/"+key] != why { - g.left[module+"/"+key] = why - g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why) - } - return g.forward(module, key, nil, true) -} - -// leftOut is why a node's report is no builder's, or empty when it counts: its node has no open assignment of the module, -// or the report was written before that assignment. A value that is no toolchain is passed on for its reader to refuse. -// The caller holds mu. -func (g *toolchainGate) leftOut(module, node string, value []byte) string { - since, open := g.standing[node+"."+module] - if !open { - return fmt.Sprintf("%s has no assignment on %s, so it is no builder", module, node) - } - tc, err := delivery.DecodeToolchain(value) - if err == nil && tc.At.Before(since) { - return fmt.Sprintf("it was written at %s, before %s's assignment on %s at %s: it counts once %s writes it again", - tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node) - } - return "" -} - -// 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), and when -// it was assigned, zero where the value does not say. -func openAssignment(value []byte) (time.Time, bool) { - var a struct { - Assigned time.Time `json:"assigned"` - Ended time.Time `json:"ended"` - State string `json:"state"` - } - if json.Unmarshal(value, &a) != nil { - return time.Time{}, false - } - return a.Assigned, 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), e.Created()) } - w, err := settle(ctx, kv, each) - if err != nil { - // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. - logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) - return errs, nil - } - if gate.Settled() { - changed(ctx) - } - go func() { - errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { - if gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) { - 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 deleted file mode 100644 index 4a0e7ca..0000000 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go +++ /dev/null @@ -1,228 +0,0 @@ -package main - -// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). - -import ( - "bytes" - "context" - "errors" - "fmt" - "log" - "strings" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/process" - - "github.com/nats-io/nats.go/jetstream" - - "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, time.Time{}) - } - gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false, time.Time{}) - gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false, time.Time{}) // 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, time.Time{}) { - 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, time.Time{}) { - 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, time.Time{}) - 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) - } - - // 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every - // toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without - // is an error naming the bucket and its grant. - var logs bytes.Buffer - logger := log.New(&logs, "", 0) - refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")} - if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil { - t.Errorf("a refused watch of the assignments stopped the start: %v", err) - } - if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { - t.Errorf("a refused toolchains bucket stopped the start: %v", err) - } - if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") { - t.Errorf("the refusals were not said: %q", logs.String()) - } - if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil || - !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { - t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) - } - // The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk - // head is told why and opens it again then. - src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf} - src.start() - if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) { - t.Errorf("a trunk head with the source refused: %v, want why", err) - } - if src.opened { - t.Error("a refused source counts as opened, so it is never opened again") - } -} - -// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. -type refusingBus struct { - refuseOpen bool - err error -} - -func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { - if r.refuseOpen { - return nil, r.err - } - return refusedWatch{err: r.err}, nil -} - -type refusedWatch struct { - jetstream.KeyValue - err error -} - -func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - return nil, r.err -} -func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - return nil, r.err -} - -// A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not -// reported changes nothing, and the report an earlier run left on the bus (issue 1940, 2026-10-25 20:56: ace.typescript -// v26.12.0 of 2026-10-23 06:29, mesh-build-node assigned to ace again at 20:56:45) is left out, said once, until the node -// writes it again. An assignment whose value names no time stands since the revision that opened it. -func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { - left := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run - assigned := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node assigned to ace again - report := func(t *testing.T, node, version string, at time.Time) (string, []byte) { - key, _ := gateReport(t, node, version) - value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at}) - if err != nil { - t.Fatal(err) - } - return key, value - } - for _, c := range []struct { - name string - assignment string // ace's assignment value - revision time.Time // when the bus wrote it - at time.Time // when ace wrote its report - counts bool - }{ - {"no report yet, then one", `{}`, time.Time{}, time.Now(), true}, - {"a report left before the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, left, false}, - {"a report left before a value with no time", `{"node":"ace","module":"mesh-build-node"}`, assigned, left, false}, - {"a report written since the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, assigned.Add(30 * time.Second), true}, - } { - t.Run(c.name, func(t *testing.T) { - set := &buildrun.ToolchainSet{} - var said []string - gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }) - for _, node := range []string{"g14", "novox", "shanks"} { - gate.Assignment(node+".mesh-build-node", []byte(`{}`), false, time.Time{}) - key, value := gateReport(t, node, "v26.12.1") - gate.Toolchain(buildrun.NodeBuildModule, key, value, false) - } - gate.Settled() - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision) { - t.Fatal("an assignment with no report passed something on") - } - key, value := report(t, "ace", "v26.12.0", c.at) - if got := gate.Toolchain(buildrun.NodeBuildModule, key, value, false); got != c.counts { - t.Fatalf("ace's report passed on: %v, want %v", got, c.counts) - } - // A rewrite of the assignment that does not move it changes nothing, and nothing is said again. - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision.Add(time.Minute)) { - t.Fatal("the same assignment written again changed what was passed on") - } - v, none := agreedOf(set) - if c.counts { - if !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { - t.Fatalf("ace's own report counts and disagrees; got %q, %q", v, none) - } - return - } - if v != "v26.12.1" || none != "" { - t.Fatalf("ace's left report is no builder's, so g14, novox and shanks agree; got %q, %q", v, none) - } - if len(said) != 1 || !strings.Contains(said[0], "before mesh-build-node's assignment on ace at 2026-10-25T19:56:45Z") { - t.Fatalf("said %q, want once why ace's report is left out", said) - } - // ace writes its report again since it was assigned: it counts. - key, value = report(t, "ace", "v26.12.1", assigned.Add(time.Minute)) - if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { - t.Fatal("ace's new report was not passed on") - } - if n := len(set.Entries("typescript")); n != 4 { - t.Fatalf("four builders report; got %d", n) - } - }) - } -} diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 896cf0c..aac3324 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -92,8 +92,8 @@ 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", "mesh-controller.assignments", - "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index f1b273f..9e72a1b 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -32,11 +32,11 @@ "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-controller.assignments", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules" + "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.placement" ], "uses": [ "node-build-go", 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 a485138..c355fb5 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -208,11 +208,11 @@ 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) + // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the + // placement is settled first, so the toolchains settle through the gate already knowing it. + gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf) watch.put = gate.Toolchain - assignErrs, err := watchAssignments(ctx, js, gate, ch.NodesChanged, logger) + placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, ch.NodesChanged, logger.Printf) if err != nil { return err } @@ -359,7 +359,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err - case err := <-assignErrs: + case err := <-placementErrs: return err case err := <-watch.errs: return err diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/refused_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/refused_test.go new file mode 100644 index 0000000..2dcb4c6 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/refused_test.go @@ -0,0 +1,62 @@ +package main + +import ( + "bytes" + "context" + "errors" + "log" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/process" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the +// watch unstarted and the part running, said; a refused bucket it cannot work without is an error naming the bucket and +// its grant. The gate over the catalogue's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate, +// issue 774). +func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) { + set := &buildrun.ToolchainSet{} + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-build-node_toolchains")} + if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { + t.Errorf("a refused toolchains bucket stopped the start: %v", err) + } + if !strings.Contains(logs.String(), "cannot be read") { + t.Errorf("the refusals were not said: %q", logs.String()) + } + if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "checker", "mesh-delivery-indexer_modules", "work"); err == nil || + !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { + t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) + } +} + +// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. +type refusingBus struct { + refuseOpen bool + err error +} + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + if r.refuseOpen { + return nil, r.err + } + return refusedWatch{err: r.err}, nil +} + +type refusedWatch struct { + jetstream.KeyValue + err error +} + +func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err +} +func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.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 deleted file mode 100644 index 9a06cec..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go +++ /dev/null @@ -1,211 +0,0 @@ -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, and only once it was written since that -// assignment; it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report -// written before the assignment is the one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node -// assigned to ace again brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement, before the module ran -// there, and the controller's take-in refused every module "builders disagree"). 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]time.Time // . with an open assignment -> since when it stands - raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it - left map[string]string // / -> why it was said left out, so each reason 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]time.Time{}, - raw: map[string]map[string][]byte{}, left: map[string]string{}} - 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 (at is when the bus wrote that revision) and answers -// whether what it passed on changed. -func (g *toolchainGate) Assignment(key string, value []byte, deleted bool, at time.Time) bool { - node, module, ok := strings.Cut(key, ".") - if !ok || !g.modules[module] { - return false - } - assigned, open := openAssignment(value) - open = open && !deleted - g.mu.Lock() - defer g.mu.Unlock() - was, stood := g.standing[key] - since := assigned - if since.IsZero() { - // A value with no time of its own (an assignment a module pulled in) stands since the revision that opened it. - since = at - if stood { - since = was - } - } - if open == stood && (!open || since.Equal(was)) { - return false - } - if open { - g.standing[key] = since - } 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 and wrote it since (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) - why := "" - if err == nil && g.known { - why = g.leftOut(module, node, value) - } - if why == "" { - delete(g.left, module+"/"+key) - return g.forward(module, key, value, false) - } - if g.left[module+"/"+key] != why { - g.left[module+"/"+key] = why - g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why) - } - return g.forward(module, key, nil, true) -} - -// leftOut is why a node's report is no builder's, or empty when it counts: its node has no open assignment of the module, -// or the report was written before that assignment. A value that is no toolchain is passed on for its reader to refuse. -// The caller holds mu. -func (g *toolchainGate) leftOut(module, node string, value []byte) string { - since, open := g.standing[node+"."+module] - if !open { - return fmt.Sprintf("%s has no assignment on %s, so it is no builder", module, node) - } - tc, err := delivery.DecodeToolchain(value) - if err == nil && tc.At.Before(since) { - return fmt.Sprintf("it was written at %s, before %s's assignment on %s at %s: it counts once %s writes it again", - tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node) - } - return "" -} - -// 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), and when -// it was assigned, zero where the value does not say. -func openAssignment(value []byte) (time.Time, bool) { - var a struct { - Assigned time.Time `json:"assigned"` - Ended time.Time `json:"ended"` - State string `json:"state"` - } - if json.Unmarshal(value, &a) != nil { - return time.Time{}, false - } - return a.Assigned, 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), e.Created()) } - w, err := settle(ctx, kv, each) - if err != nil { - // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. - logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) - return errs, nil - } - if gate.Settled() { - changed(ctx) - } - go func() { - errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { - if gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) { - 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 deleted file mode 100644 index 0715aea..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go +++ /dev/null @@ -1,218 +0,0 @@ -package main - -// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). - -import ( - "bytes" - "context" - "errors" - "fmt" - "log" - "strings" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/process" - - "github.com/nats-io/nats.go/jetstream" - - "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, time.Time{}) - } - gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false, time.Time{}) - gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false, time.Time{}) // 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, time.Time{}) { - 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, time.Time{}) { - 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, time.Time{}) - 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) - } - - // 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every - // toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without - // is an error naming the bucket and its grant. - var logs bytes.Buffer - logger := log.New(&logs, "", 0) - refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")} - if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil { - t.Errorf("a refused watch of the assignments stopped the start: %v", err) - } - if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil { - t.Errorf("a refused toolchains bucket stopped the start: %v", err) - } - if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") { - t.Errorf("the refusals were not said: %q", logs.String()) - } - if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "checker", "mesh-delivery-indexer_modules", "work"); err == nil || - !strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") { - t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err) - } -} - -// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. -type refusingBus struct { - refuseOpen bool - err error -} - -func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { - if r.refuseOpen { - return nil, r.err - } - return refusedWatch{err: r.err}, nil -} - -type refusedWatch struct { - jetstream.KeyValue - err error -} - -func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - return nil, r.err -} -func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - return nil, r.err -} - -// A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not -// reported changes nothing, and the report an earlier run left on the bus (issue 1940, 2026-10-25 20:56: ace.typescript -// v26.12.0 of 2026-10-23 06:29, mesh-build-node assigned to ace again at 20:56:45) is left out, said once, until the node -// writes it again. An assignment whose value names no time stands since the revision that opened it. -func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { - left := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run - assigned := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node assigned to ace again - report := func(t *testing.T, node, version string, at time.Time) (string, []byte) { - key, _ := gateReport(t, node, version) - value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at}) - if err != nil { - t.Fatal(err) - } - return key, value - } - for _, c := range []struct { - name string - assignment string // ace's assignment value - revision time.Time // when the bus wrote it - at time.Time // when ace wrote its report - counts bool - }{ - {"no report yet, then one", `{}`, time.Time{}, time.Now(), true}, - {"a report left before the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, left, false}, - {"a report left before a value with no time", `{"node":"ace","module":"mesh-build-node"}`, assigned, left, false}, - {"a report written since the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, assigned.Add(30 * time.Second), true}, - } { - t.Run(c.name, func(t *testing.T) { - set := &buildrun.ToolchainSet{} - var said []string - gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }) - for _, node := range []string{"g14", "novox", "shanks"} { - gate.Assignment(node+".mesh-build-node", []byte(`{}`), false, time.Time{}) - key, value := gateReport(t, node, "v26.12.1") - gate.Toolchain(buildrun.NodeBuildModule, key, value, false) - } - gate.Settled() - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision) { - t.Fatal("an assignment with no report passed something on") - } - key, value := report(t, "ace", "v26.12.0", c.at) - if got := gate.Toolchain(buildrun.NodeBuildModule, key, value, false); got != c.counts { - t.Fatalf("ace's report passed on: %v, want %v", got, c.counts) - } - // A rewrite of the assignment that does not move it changes nothing, and nothing is said again. - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision.Add(time.Minute)) { - t.Fatal("the same assignment written again changed what was passed on") - } - v, none := agreedOf(set) - if c.counts { - if !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { - t.Fatalf("ace's own report counts and disagrees; got %q, %q", v, none) - } - return - } - if v != "v26.12.1" || none != "" { - t.Fatalf("ace's left report is no builder's, so g14, novox and shanks agree; got %q, %q", v, none) - } - if len(said) != 1 || !strings.Contains(said[0], "before mesh-build-node's assignment on ace at 2026-10-25T19:56:45Z") { - t.Fatalf("said %q, want once why ace's report is left out", said) - } - // ace writes its report again since it was assigned: it counts. - key, value = report(t, "ace", "v26.12.1", assigned.Add(time.Minute)) - if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { - t.Fatal("ace's new report was not passed on") - } - if n := len(set.Entries("typescript")); n != 4 { - t.Fatalf("four builders report; got %d", n) - } - }) - } -} diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 3bb9a7a..70803df 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -101,9 +101,9 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {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", "mesh-controller.assignments", + if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "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) { + "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 1caddc5..a4e96cb 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -34,7 +34,6 @@ "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", @@ -42,6 +41,7 @@ "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.placement", "mesh-delivery-watchman.presence", "node-engine.have" ], 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 edecd2f..87f6168 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 @@ -578,23 +578,39 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC } } -// 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) { +// TestIndexerCountsOnlyThePlacedBuildNodesToolchains (issue 1940, issue 774): the toolchains bucket holds the report of +// n2, where the catalogue no longer places mesh-build-go, 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 placement puts mesh-build-go on n2 again, its report written since counts and the builders +// disagree (the guard stays); when it is taken off again, the fingerprint comes back. No restart, no timer: the +// placement's watch (ADR 0549 §3), not the controller's assignments. +func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(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 + var placement, toolchains jetstream.KeyValue + place := func(ctx context.Context, t *testing.T, node string, modules ...string) { + p := delivery.Placement{Node: node} + for _, m := range modules { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) + } + key, _ := delivery.PlacementKey(node) + value, err := delivery.EncodePlacement(p) + if err != nil { + t.Fatal(err) + } + if _, err := placement.Put(ctx, key, value); err != nil { + t.Fatal(err) + } + } 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) } + toolchains = kv 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()}) @@ -605,18 +621,12 @@ func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) { 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 { + // The indexer's placement, as the catalogue's files put it there (ADR 0549 §2): n1 places mesh-build-go, n2 nothing. + if placement, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: delivery.PlacementHistory}); 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) - } - } + place(ctx, t, "n1", "mesh-build-go") + place(ctx, t, "n2") }) data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 302, Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0", @@ -647,16 +657,22 @@ func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) { if !agreed(held) { t.Fatalf("within 5 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", held.FP2, held.None2, held.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 { + place(ctx, t, "n2", "mesh-build-go") + key, _ := delivery.ToolchainKey("n2", "go") + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: "go1.27.2", Recipe: "1", At: time.Now().Add(time.Second)}) + if err != nil { t.Fatal(err) } - // n2's report, written since its assignment, counts: the builders disagree. A fingerprint already computed is kept + if _, err := toolchains.Put(ctx, key, value); err != nil { + t.Fatal(err) + } + // n2's report, written since its placement, counts: the builders disagree. A fingerprint already computed is kept // (issue 1940), and only a new one needs the agreement: a merge moving avahi gets none until they agree. time.Sleep(500 * time.Millisecond) if e, err := modules.Get(ctx, "avahi"); err != nil { t.Fatal(err) } else if m, err := delivery.DecodeModule(e.Value()); err != nil || m.FP2 != held.FP2 { - t.Fatalf("after n2's assignment avahi has fp2 %q, none2 %q: want its fingerprint %q kept", m.FP2, m.None2, held.FP2) + t.Fatalf("after n2's placement avahi has fp2 %q, none2 %q: want its fingerprint %q kept", m.FP2, m.None2, held.FP2) } moved := strings.Replace(string(data), `"newtree"`, `"movedtree"`, 1) if moved == string(data) { @@ -669,12 +685,10 @@ func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) { 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 a merge after 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) + t.Fatalf("within 5 s of a merge after n2's placement avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2) } + place(ctx, t, "n2") 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) + t.Fatalf("within 5 s of n2's placement taken away 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 aa732ef..e55b265 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -24,6 +24,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-indexer/internal/index" @@ -321,14 +322,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { sourceErrs <- follow(ctx, sw, each) }() } } - // 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) { + // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the + // placement is settled first, so the toolchains settle through the gate already knowing it. + gate := buildrun.NewToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf) + placementErrs, err := buildrun.FollowPlacement(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.Printf("writing the modules again after the placement of a build module changed: %v", err) } - }, logger) + }, logger.Printf) if err != nil { return err } @@ -442,7 +443,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-sourceErrs: return err - case err := <-assignErrs: + case err := <-placementErrs: return err case err := <-tools.errs: return err diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go deleted file mode 100644 index 9a06cec..0000000 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go +++ /dev/null @@ -1,211 +0,0 @@ -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, and only once it was written since that -// assignment; it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report -// written before the assignment is the one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node -// assigned to ace again brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement, before the module ran -// there, and the controller's take-in refused every module "builders disagree"). 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]time.Time // . with an open assignment -> since when it stands - raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it - left map[string]string // / -> why it was said left out, so each reason 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]time.Time{}, - raw: map[string]map[string][]byte{}, left: map[string]string{}} - 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 (at is when the bus wrote that revision) and answers -// whether what it passed on changed. -func (g *toolchainGate) Assignment(key string, value []byte, deleted bool, at time.Time) bool { - node, module, ok := strings.Cut(key, ".") - if !ok || !g.modules[module] { - return false - } - assigned, open := openAssignment(value) - open = open && !deleted - g.mu.Lock() - defer g.mu.Unlock() - was, stood := g.standing[key] - since := assigned - if since.IsZero() { - // A value with no time of its own (an assignment a module pulled in) stands since the revision that opened it. - since = at - if stood { - since = was - } - } - if open == stood && (!open || since.Equal(was)) { - return false - } - if open { - g.standing[key] = since - } 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 and wrote it since (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) - why := "" - if err == nil && g.known { - why = g.leftOut(module, node, value) - } - if why == "" { - delete(g.left, module+"/"+key) - return g.forward(module, key, value, false) - } - if g.left[module+"/"+key] != why { - g.left[module+"/"+key] = why - g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why) - } - return g.forward(module, key, nil, true) -} - -// leftOut is why a node's report is no builder's, or empty when it counts: its node has no open assignment of the module, -// or the report was written before that assignment. A value that is no toolchain is passed on for its reader to refuse. -// The caller holds mu. -func (g *toolchainGate) leftOut(module, node string, value []byte) string { - since, open := g.standing[node+"."+module] - if !open { - return fmt.Sprintf("%s has no assignment on %s, so it is no builder", module, node) - } - tc, err := delivery.DecodeToolchain(value) - if err == nil && tc.At.Before(since) { - return fmt.Sprintf("it was written at %s, before %s's assignment on %s at %s: it counts once %s writes it again", - tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node) - } - return "" -} - -// 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), and when -// it was assigned, zero where the value does not say. -func openAssignment(value []byte) (time.Time, bool) { - var a struct { - Assigned time.Time `json:"assigned"` - Ended time.Time `json:"ended"` - State string `json:"state"` - } - if json.Unmarshal(value, &a) != nil { - return time.Time{}, false - } - return a.Assigned, 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), e.Created()) } - w, err := settle(ctx, kv, each) - if err != nil { - // Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read. - logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err) - return errs, nil - } - if gate.Settled() { - changed(ctx) - } - go func() { - errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) { - if gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) { - 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 deleted file mode 100644 index 28fccc4..0000000 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go +++ /dev/null @@ -1,165 +0,0 @@ -package main - -// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940). - -import ( - "fmt" - "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, time.Time{}) - } - gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false, time.Time{}) - gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false, time.Time{}) // 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, time.Time{}) { - 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, time.Time{}) { - 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, time.Time{}) - 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 build node's report counts from the first one it writes since its assignment: a newly assigned node that has not -// reported changes nothing, and the report an earlier run left on the bus (issue 1940, 2026-10-25 20:56: ace.typescript -// v26.12.0 of 2026-10-23 06:29, mesh-build-node assigned to ace again at 20:56:45) is left out, said once, until the node -// writes it again. An assignment whose value names no time stands since the revision that opened it. -func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) { - left := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run - assigned := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node assigned to ace again - report := func(t *testing.T, node, version string, at time.Time) (string, []byte) { - key, _ := gateReport(t, node, version) - value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at}) - if err != nil { - t.Fatal(err) - } - return key, value - } - for _, c := range []struct { - name string - assignment string // ace's assignment value - revision time.Time // when the bus wrote it - at time.Time // when ace wrote its report - counts bool - }{ - {"no report yet, then one", `{}`, time.Time{}, time.Now(), true}, - {"a report left before the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, left, false}, - {"a report left before a value with no time", `{"node":"ace","module":"mesh-build-node"}`, assigned, left, false}, - {"a report written since the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, assigned.Add(30 * time.Second), true}, - } { - t.Run(c.name, func(t *testing.T) { - set := &buildrun.ToolchainSet{} - var said []string - gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }) - for _, node := range []string{"g14", "novox", "shanks"} { - gate.Assignment(node+".mesh-build-node", []byte(`{}`), false, time.Time{}) - key, value := gateReport(t, node, "v26.12.1") - gate.Toolchain(buildrun.NodeBuildModule, key, value, false) - } - gate.Settled() - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision) { - t.Fatal("an assignment with no report passed something on") - } - key, value := report(t, "ace", "v26.12.0", c.at) - if got := gate.Toolchain(buildrun.NodeBuildModule, key, value, false); got != c.counts { - t.Fatalf("ace's report passed on: %v, want %v", got, c.counts) - } - // A rewrite of the assignment that does not move it changes nothing, and nothing is said again. - if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision.Add(time.Minute)) { - t.Fatal("the same assignment written again changed what was passed on") - } - v, none := agreedOf(set) - if c.counts { - if !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { - t.Fatalf("ace's own report counts and disagrees; got %q, %q", v, none) - } - return - } - if v != "v26.12.1" || none != "" { - t.Fatalf("ace's left report is no builder's, so g14, novox and shanks agree; got %q, %q", v, none) - } - if len(said) != 1 || !strings.Contains(said[0], "before mesh-build-node's assignment on ace at 2026-10-25T19:56:45Z") { - t.Fatalf("said %q, want once why ace's report is left out", said) - } - // ace writes its report again since it was assigned: it counts. - key, value = report(t, "ace", "v26.12.1", assigned.Add(time.Minute)) - if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) { - t.Fatal("ace's new report was not passed on") - } - if n := len(set.Entries("typescript")); n != 4 { - t.Fatalf("four builders report; got %d", n) - } - }) - } -} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index e05aab4..e74b82b 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -89,7 +89,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}}; !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.assignments", "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.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 544e452..282fa60 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -22,7 +22,6 @@ "reads": [ "mesh-build-go.toolchains", "mesh-build-node.toolchains", - "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"