diff --git a/modules/mesh-delivery-builder/buildrun/names.go b/modules/mesh-delivery-builder/buildrun/names.go index 50ab553c..f42177cf 100644 --- a/modules/mesh-delivery-builder/buildrun/names.go +++ b/modules/mesh-delivery-builder/buildrun/names.go @@ -11,6 +11,3 @@ const ( // StepsBucket is the bucket of a build module's state steps. func StepsBucket(module string) string { return module + "_steps" } - -// ToolchainsBucket is the bucket of a build module's state toolchains. -func ToolchainsBucket(module string) string { return module + "_toolchains" } diff --git a/modules/mesh-delivery-builder/buildrun/standing.go b/modules/mesh-delivery-builder/buildrun/standing.go index f97ae7c0..52a490b1 100644 --- a/modules/mesh-delivery-builder/buildrun/standing.go +++ b/modules/mesh-delivery-builder/buildrun/standing.go @@ -3,9 +3,6 @@ package buildrun import ( "context" "errors" - "fmt" - "maps" - "slices" "sort" "sync" "time" @@ -15,120 +12,60 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" - "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-sdk/go/process" ) -// A build module writes . of its toolchains state on every node it runs on, and nothing deletes the key -// when the module leaves a node: no part writes another module's state, and the module is gone. A report of a node the -// build module no longer runs on is therefore no builder's (issue 1940: mesh-build-node taken off 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 -// ToolchainSet): an entry is passed on only while the assignments (the seat mesh-module-assigner's state `assignments`, held by -// the assigner, AssignmentsBucket, read by meshmoduleassignerclient.Placed; ADR 0567, ADR 0549 §3) places its build module on -// its node, and only once it was written since that assignments; it is passed on as deleted otherwise, said once, and -// passed on when its node writes it again. A report written before the assignments is one an earlier run of the module -// left (issue 1940, 2026-10-25 20:56: mesh-build-node put on ace again brought ace.typescript v26.12.0 of 2026-10-23 -// back into the agreement before the module ran there). Until the assignments is settled, and while no node's assignments -// is held (before the assigner's adopt writes the first one, or where the state cannot be read), every entry is passed -// on. Builders that run and report different versions still disagree: that refusal is the guard of novox/hq ADR 0465, -// and it stays. -// -// This replaces the copies of the gate the indexer, the builder and the checker each kept over the controller's -// assignments (issue 1940's subtask, issue 774: those go with the controller's lease, ADR 0549 §6). -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) - placed placements - known bool // the assignments is settled - 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 - // OnPlaced is told, outside the gate's lock, once the assignments's history is read and after each revision of it +// PlacedNodes holds which modules the catalogue places on which node: the seat mesh-module-assigner's state `assignments`, +// held by the assigner (AssignmentsBucket, read by meshmoduleassignerclient.Placed; ADR 0567), or the assignment half +// (meshwantsclient.Placed, fed by Revision). The builder counts the nodes it names (Nodes; ADR 0624 §1). It holds no +// toolchain: the builder keeps the toolchains it heard in its state toolchains-heard and forgets one when the node-engine +// seat says the build module stopped on the node, never on a placement (design 47 G §546). +type PlacedNodes struct { + mu sync.Mutex + logf func(format string, a ...any) + placed placements + known bool // the assignments is settled + // OnPlaced is told, outside the lock, once the assignments's history is read and after each revision of it // FollowAssignments hears from then on (the counted nodes, Nodes, may have changed). Nil tells nobody. OnPlaced func() } -// placements is where the gate reads which modules stand on which node, and since when: the assigner's state -// (meshmoduleassignerclient.Placed, fed by Assignments) or the assignment half (meshwantsclient.Placed, fed by Revision). +// placements is where the part reads which modules stand on which node. type placements interface { Seeded() bool - Since(node, module string) (time.Time, bool) ByNode() map[string][]string } -// NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what -// it leaves out. -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, placed: meshmoduleassignerclient.NewPlaced(), - raw: map[string]map[string][]byte{}, left: map[string]string{}} - for _, m := range modules { - g.modules[m] = true - } - return g +// NewPlacedNodes is the placements of the builder, saying with logf what it cannot read. +func NewPlacedNodes(logf func(string, ...any)) *PlacedNodes { + return &PlacedNodes{logf: logf, placed: meshmoduleassignerclient.NewPlaced()} } -// UseModules replaces the build modules whose reports the gate passes on again when the assignments moves (a part that -// finds them from the seat list and the manifests, as the indexer does, gives them at each change), and passes every -// report it holds on again. It answers whether what it passed on changed. -func (g *ToolchainGate) UseModules(modules []string) bool { - g.mu.Lock() - defer g.mu.Unlock() - g.modules = map[string]bool{} - for _, m := range modules { - g.modules[m] = true - } - return g.all() -} - -// 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) -} - -// Assignments takes one revision of the seat mesh-module-assigner's assignments in (at is when the bus wrote it) and answers whether what it -// passed on changed. A value that does not read is said, and places nothing on its node. -func (g *ToolchainGate) Assignments(key string, value []byte, deleted bool, at time.Time) bool { +// Assignments takes one revision of the seat mesh-module-assigner's assignments in (at is when the bus wrote it) and +// answers whether the nodes the modules stand on changed. A value that does not read is said, and places nothing. +func (g *PlacedNodes) Assignments(key string, value []byte, deleted bool, at time.Time) bool { g.mu.Lock() defer g.mu.Unlock() placed, ok := g.placed.(*meshmoduleassignerclient.Placed) if !ok { - g.logf("%s: %s is not read, the gate takes the assignment half (design 47)", AssignmentsBucket, key) + g.logf("%s: %s is not read, the placements take the assignment half (design 47)", AssignmentsBucket, key) return false } was := placed.Seeded() - node, changed, err := placed.Put(key, value, deleted, at) + _, changed, err := placed.Put(key, value, deleted, at) if err != nil { g.logf("%s: %v (issue 1940)", AssignmentsBucket, err) } if !g.known { return false } - if was != g.placed.Seeded() { - // The first assignments came, or the last went: every report's standing changed. - return g.all() - } - return g.again(node, changed) + return was != g.placed.Seeded() || len(changed) > 0 } // Revision takes one assignment graph revision the part's view holds in (design 47 *Each part acts on its own entries*: -// the placements come from the seat mesh-wants' assignment half, never the assigner's state; assignmentview's Held hands -// each) and answers whether what it passed on changed. The first revision taken settles the gate, as the assignments' -// history read does; a later one passes on again the reports of each node whose build modules' standing changed. A gate -// fed by Revision reads no Assignments. -func (g *ToolchainGate) Revision(r meshwants.AssignmentRevision) bool { +// the placements come from the seat mesh-wants' assignment half, never the assigner's state) and answers whether the +// nodes the modules stand on changed. The first revision taken settles it. Placements fed by Revision read no Assignments. +func (g *PlacedNodes) Revision(r meshwants.AssignmentRevision) bool { g.mu.Lock() defer g.mu.Unlock() placed, ok := g.placed.(*meshwantsclient.Placed) @@ -141,20 +78,14 @@ func (g *ToolchainGate) Revision(r meshwants.AssignmentRevision) bool { if changed == nil { return false } - if !g.known || was != placed.Seeded() { - g.known = true - return g.all() - } - moved := false - for _, node := range slices.Sorted(maps.Keys(changed)) { - moved = g.again(node, changed[node]) || moved - } - return moved + first := !g.known + g.known = true + return first || was != placed.Seeded() || len(changed) > 0 } // Nodes are the nodes the assignments places any module on, sorted: the nodes the mesh counts (a node leaves the count // when the assignments no longer names it; ADR 0624 §1). None until the assignments is read. -func (g *ToolchainGate) Nodes() []string { +func (g *PlacedNodes) Nodes() []string { var out []string for node, modules := range g.placed.ByNode() { if len(modules) > 0 { @@ -165,95 +96,27 @@ func (g *ToolchainGate) Nodes() []string { return out } -// Settled says the assignments's history is read: from now on only the reports of placed nodes pass, once a node's -// assignments is held. It answers whether what it passed on changed. -func (g *ToolchainGate) Settled() bool { +// Settled says the assignments's history is read. +func (g *PlacedNodes) Settled() { g.mu.Lock() defer g.mu.Unlock() g.known = true - return g.all() -} - -// all passes on every report again. The caller holds mu. -func (g *ToolchainGate) all() bool { - 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 -} - -// again passes on every report of node by a build module among modules. The caller holds mu. -func (g *ToolchainGate) again(node string, modules []string) bool { - changed := false - for _, module := range modules { - if !g.modules[module] { - continue - } - for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { - if n, _, err := buildstep.ParseToolchainKey(k); err == nil && n == node { - changed = g.pass(module, k) || changed - } - } - } - return changed -} - -// pass hands one key on as it stands: its value when its node is placed and wrote it since (or the assignments is 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 := buildstep.ParseToolchainKey(key) - why := "" - if err == nil && g.known && g.placed.Seeded() { - 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)", 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: the catalogue does not place the module on its -// node, or the report was written before it did. 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, placed := g.placed.Since(node, module) - if !placed { - return fmt.Sprintf("the catalogue does not place %s on %s, so it is no builder", module, node) - } - tc, err := buildstep.DecodeToolchain(value) - if err == nil && tc.At.Before(since) { - return fmt.Sprintf("it was written at %s, before %s was placed 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 "" } // FollowAssignments reads the seat mesh-module-assigner's assignments with its history into gate, says it settled, and follows it; changed // is told whenever what the gate passes on changed. Where the state cannot be read or watched (not there, or this part // may not read it) it says so and the gate passes every report on: not fatal, since an indexer, builder or checker that // stops for it stops every delivery (issue 1945). An error of the watch after that ends on the channel. -func FollowAssignments(ctx context.Context, js process.BucketOpener, gate *ToolchainGate, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) { +func FollowAssignments(ctx context.Context, js process.BucketOpener, gate *PlacedNodes, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) { errs := make(chan error, 1) kv, err := js.KeyValue(ctx, AssignmentsBucket) if err != nil { - logf("the state %s cannot be read, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", AssignmentsBucket, err) + logf("the state %s cannot be read, so no node is counted by the assignments: %v", AssignmentsBucket, err) return errs, nil } w, err := kv.WatchAll(ctx, jetstream.IncludeHistory()) if err != nil { - logf("the state %s cannot be watched, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", AssignmentsBucket, err) + logf("the state %s cannot be watched, so no node is counted by the assignments: %v", AssignmentsBucket, err) return errs, nil } put := func(e jetstream.KeyValueEntry) bool { @@ -275,9 +138,8 @@ func FollowAssignments(ctx context.Context, js process.BucketOpener, gate *Toolc put(e) } } - if gate.Settled() { - changed(ctx) - } + gate.Settled() + changed(ctx) if gate.OnPlaced != nil { gate.OnPlaced() } diff --git a/modules/mesh-delivery-builder/buildrun/standing_test.go b/modules/mesh-delivery-builder/buildrun/standing_test.go index 5fca15d6..2b8da16f 100644 --- a/modules/mesh-delivery-builder/buildrun/standing_test.go +++ b/modules/mesh-delivery-builder/buildrun/standing_test.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "errors" - "fmt" "log" "strings" "testing" @@ -15,19 +14,8 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" - "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) -func gateReport(t *testing.T, node, version string, at time.Time) (string, []byte) { - t.Helper() - key, _ := buildstep.ToolchainKey(node, "typescript") - value, err := buildstep.EncodeToolchain(buildstep.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at}) - if err != nil { - t.Fatal(err) - } - return key, value -} - // placing is a value of the state `assignments` placing modules on node. func placing(t *testing.T, node string, modules ...string) (string, []byte) { t.Helper() @@ -43,219 +31,52 @@ func placing(t *testing.T, node string, modules ...string) (string, []byte) { return key, values[key] } -func agreedOf(set *ToolchainSet) (string, string) { - tc, none := buildstep.AgreedToolchain(set.Entries("typescript")) - return tc.Version, none -} - -// Issue 1940, the gate over the catalogue's assignments (issue 774, ADR 0549 §3): a build node's report counts only while -// the assignments places its build module on its node, and only once written since; before the assignments is settled, and -// while no node's assignments is held, every report counts. The live cases: mesh-build-node taken off ace left -// ace.typescript at v26.12.0 against v26.12.1 on the three others; put on ace again, its report of 2026-10-23 06:29 came -// back before the module ran there. -func TestToolchainGate(t *testing.T) { - old := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run - placed := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node placed on ace again +// The placements the builder counts nodes by (ADR 0624 §1): a settled assignments names the nodes its modules stand on, a +// revision of the assignment half the same; they hold no toolchain (design 47 G §546: a toolchain is forgotten on the +// node-engine seat's stopped, never on a placement). +func TestPlacedNodes(t *testing.T) { + old := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) three := []string{"g14", "novox", "shanks"} - start := func(t *testing.T, seeded bool) (*ToolchainGate, *ToolchainSet, *[]string) { - set := &ToolchainSet{} - said := &[]string{} - gate := NewToolchainGate([]string{GoBuildModule, NodeBuildModule}, set.Put, func(f string, a ...any) { *said = append(*said, fmt.Sprintf(f, a...)) }) + + t.Run("the assignments names the counted nodes once it is settled", func(t *testing.T) { + gate := NewPlacedNodes(func(string, ...any) {}) for _, node := range three { - if seeded { - key, value := placing(t, node, NodeBuildModule) - gate.Assignments(key, value, false, old.Add(-time.Hour)) + k, v := placing(t, node, NodeBuildModule) + if gate.Assignments(k, v, false, old) { + t.Fatal("an assignments before it is settled changed the counted nodes") } - key, value := gateReport(t, node, "v26.12.1", placed.Add(time.Minute)) - gate.Toolchain(NodeBuildModule, key, value, false) } - return gate, set, said - } - - t.Run("not placed on ace: its report is no builder's", func(t *testing.T) { - gate, set, said := start(t, true) - key, value := gateReport(t, "ace", "v26.12.0", old) - gate.Toolchain(NodeBuildModule, key, value, false) - if a := buildstep.ToolchainAgreement(set.Entries("typescript")); len(a.SittingOut) != 1 || a.SittingOut[0].Node != "ace" { // ADR 0584: the majority is agreed, ace's report counts as the one that sits out - t.Fatalf("before the assignments is settled every report counts; got %+v", a) - } - k, v := placing(t, "ace", GoBuildModule) // another build module on ace: no matter - gate.Assignments(k, v, false, old) - if !gate.Settled() { - t.Fatal("settling left ace's report out, so what was passed on changed") - } - if v, none := agreedOf(set); v != "v26.12.1" || none != "" { - t.Fatalf("the placed nodes agree on v26.12.1; got %q, %q", v, none) - } - if len(*said) != 1 || !strings.Contains((*said)[0], "does not place mesh-build-node on ace") { - t.Fatalf("ace's report is said left out once; said %q", *said) - } - if gate.Toolchain(NodeBuildModule, key, value, false) { - t.Fatal("a report of a node not placed changed what was passed on") - } - // Placed on ace with a report written since: it counts, and the builders really disagree. - k, v = placing(t, "ace", GoBuildModule, NodeBuildModule) - gate.Assignments(k, v, false, old.Add(-time.Minute)) - if a := buildstep.ToolchainAgreement(set.Entries("typescript")); a.Toolchain.Version != "v26.12.1" || len(a.SittingOut) != 1 || a.SittingOut[0].Version != "v26.12.0" { - t.Fatalf("builders that run different versions: the majority is agreed and ace sits out; got %+v", a) - } - // Taken off ace again (its key deleted): its report goes again. - if !gate.Assignments(k, nil, true, placed) { - t.Fatal("taking ace's assignments away passed nothing on") - } - if v, none := agreedOf(set); v != "v26.12.1" || none != "" { - t.Fatalf("got %q, %q", v, none) - } - // A report deleted from the bucket goes whether its node is placed or not. - key, _ = gateReport(t, "g14", "v26.12.1", placed) - if !gate.Toolchain(NodeBuildModule, key, nil, true) || len(set.Entries("typescript")) != 2 { - t.Fatalf("deleting g14's report: %d entries left, want novox and shanks", len(set.Entries("typescript"))) - } - }) - - t.Run("build modules given later pass their reports on when the assignments moves", func(t *testing.T) { - set := &ToolchainSet{} - gate := NewToolchainGate(nil, set.Put, func(string, ...any) {}) gate.Settled() - k, v := placing(t, "novox", GoBuildModule) - gate.Assignments(k, v, false, old) - key, value := gateReport(t, "ace", "v26.12.1", placed.Add(time.Minute)) - gate.Toolchain(NodeBuildModule, key, value, false) // ace is not placed: left out - k, v = placing(t, "ace", NodeBuildModule) - if gate.Assignments(k, v, false, placed) || len(set.Entries("typescript")) != 0 { - t.Fatalf("a module the gate was not given passed on at the assignments: %v", set.Entries("typescript")) + if got := strings.Join(gate.Nodes(), ","); got != "g14,novox,shanks" { + t.Fatalf("nodes = %q", got) } - if !gate.UseModules([]string{NodeBuildModule}) || len(set.Entries("typescript")) != 1 { - t.Fatalf("given mesh-build-node, ace's report placed since is passed on; entries %v", set.Entries("typescript")) + k, v := placing(t, "ace", GoBuildModule) + if !gate.Assignments(k, v, false, old) || len(gate.Nodes()) != 4 { + t.Fatalf("ace placed: nodes %v", gate.Nodes()) + } + if !gate.Assignments(k, nil, true, old.Add(time.Hour)) || len(gate.Nodes()) != 3 { + t.Fatalf("ace's assignments deleted: nodes %v", gate.Nodes()) } }) - for _, c := range []struct { - name string - at time.Time // when ace wrote its report - counts bool - }{ - {"a report left before the assignments", old, false}, - {"a report written since the assignments", placed.Add(30 * time.Second), true}, - } { - t.Run(c.name, func(t *testing.T) { - gate, set, said := start(t, true) - k, v := placing(t, "ace") // ace's key, first heard placing nothing - gate.Assignments(k, v, false, old.Add(-time.Hour)) - gate.Settled() - k, v = placing(t, "ace", NodeBuildModule) - if gate.Assignments(k, v, false, placed) { - t.Fatal("assignments with no report passed something on") - } - key, value := gateReport(t, "ace", "v26.12.0", c.at) - if got := gate.Toolchain(NodeBuildModule, key, value, false); got != c.counts { - t.Fatalf("ace's report passed on: %v, want %v", got, c.counts) - } - // ace's assignments written again with another module beside: its standing does not move, nothing is said again. - k, v = placing(t, "ace", GoBuildModule, NodeBuildModule) - if gate.Assignments(k, v, false, placed.Add(time.Minute)) { - t.Fatal("the same assignments of mesh-build-node changed what was passed on") - } - if c.counts { - return - } - if v, none := agreedOf(set); v != "v26.12.1" || none != "" { - t.Fatalf("ace's left report is no builder's; got %q, %q", v, none) - } - if len(*said) != 1 || !strings.Contains((*said)[0], "before mesh-build-node was placed on ace at 2026-10-25T19:56:45Z") { - t.Fatalf("said %q, want once why ace's report is left out", *said) - } - key, value = gateReport(t, "ace", "v26.12.1", placed.Add(time.Minute)) - if !gate.Toolchain(NodeBuildModule, key, value, false) || len(set.Entries("typescript")) != 4 { - t.Fatal("ace's new report was not passed on") - } - }) - } - - t.Run("no assignments held: every report counts", func(t *testing.T) { - gate, set, _ := start(t, false) - key, value := gateReport(t, "ace", "v26.12.0", old) - gate.Toolchain(NodeBuildModule, key, value, false) - gate.Settled() - if n := len(set.Entries("typescript")); n != 4 { - t.Fatalf("before the first assignments file every report counts; got %d", n) - } - k, v := placing(t, "g14", NodeBuildModule) - if !gate.Assignments(k, v, false, old.Add(-time.Hour)) || len(set.Entries("typescript")) != 1 { - t.Fatalf("the first assignments leaves every node it does not place out; got %d", len(set.Entries("typescript"))) - } - if !gate.Assignments(k, nil, true, placed) || len(set.Entries("typescript")) != 4 { - t.Fatal("the last assignments gone: every report counts again") - } - }) - - // Design 47 *Each part acts on its own entries*: fed by the assignment half's revisions, the gate settles at the first, - // leaves out a node no revision places the build module on, and counts a node placed in a later revision only by a - // report written since that revision's time. - t.Run("fed by the assignment half", func(t *testing.T) { - set := &ToolchainSet{} - said := &[]string{} - gate := NewToolchainGate([]string{GoBuildModule, NodeBuildModule}, set.Put, func(f string, a ...any) { *said = append(*said, fmt.Sprintf(f, a...)) }) - for _, node := range append([]string{"ace"}, three...) { - key, value := gateReport(t, node, "v26.12.1", placed.Add(time.Minute)) - if node == "ace" { - key, value = gateReport(t, node, "v26.12.0", old) - } - gate.Toolchain(NodeBuildModule, key, value, false) - } - revision := func(n int64, at time.Time, nodes ...string) meshwants.AssignmentRevision { + t.Run("the assignment half: the first revision settles, a later one moves the nodes", func(t *testing.T) { + gate := NewPlacedNodes(func(string, ...any) {}) + revision := func(n int64, nodes ...string) meshwants.AssignmentRevision { + at := old.Add(time.Duration(n) * time.Hour) r := meshwants.AssignmentRevision{Revision: n, Kind: "trunk", At: &at} for _, node := range nodes { r.Entries = append(r.Entries, meshwants.NodeEntryRef{Node: node, Module: NodeBuildModule, Digest: "d-" + node}) } return r } - if !gate.Revision(revision(7, old.Add(-time.Hour), three...)) || len(set.Entries("typescript")) != 3 { - t.Fatalf("the first revision settles the gate and leaves ace out; %d entries", len(set.Entries("typescript"))) + if !gate.Revision(revision(7, three...)) || len(gate.Nodes()) != 3 { + t.Fatalf("the first revision: nodes %v", gate.Nodes()) } - if gate.Revision(revision(7, old.Add(-time.Hour), three...)) { - t.Fatal("the same revision again changed what was passed on") + if gate.Revision(revision(7, three...)) { + t.Fatal("the same revision again changed the counted nodes") } - if gate.Revision(revision(8, placed, "ace", "g14", "novox", "shanks")); len(set.Entries("typescript")) != 3 { - t.Fatal("ace's report from before revision 8 placed it counts") - } - if len(*said) < 2 || !strings.Contains((*said)[len(*said)-1], "before mesh-build-node was placed on ace at 2026-10-25T19:56:45Z") { - t.Fatalf("said %q, want why ace's old report is left out", *said) - } - key, value := gateReport(t, "ace", "v26.12.1", placed.Add(time.Minute)) - if !gate.Toolchain(NodeBuildModule, key, value, false) || len(set.Entries("typescript")) != 4 { - t.Fatal("ace's report written since revision 8 is not passed on") - } - if !gate.Revision(revision(9, placed.Add(time.Hour), "ace", "novox", "shanks")) || len(set.Entries("typescript")) != 3 || len(gate.Nodes()) != 3 { - t.Fatalf("g14 taken off in revision 9: %d entries, nodes %v", len(set.Entries("typescript")), gate.Nodes()) - } - }) - - t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) { - p := meshmoduleassignerclient.NewPlaced() - // ace's node key is the first revision heard of it: its module stands since before. - k, v := placing(t, "ace", NodeBuildModule) - p.Put(k, v, false, old) - if _, changed, _ := p.Put("ace."+NodeBuildModule, []byte("{}"), false, placed); len(changed) != 0 { - t.Errorf("a module key says which modules stand: %v", changed) - } - k2, v2 := placing(t, "g14") - p.Put(k2, v2, false, old) - k2, v2 = placing(t, "g14", NodeBuildModule) - if _, changed, _ := p.Put(k2, v2, false, placed); strings.Join(changed, ",") != NodeBuildModule { - t.Errorf("a module placed says changed %v", changed) - } - if _, _, err := p.Put("novox.node", []byte("not json"), false, old); err == nil { - t.Error("a value that does not read was not said") - } - if s, ok := p.Since("ace", NodeBuildModule); !ok || !s.IsZero() { - t.Errorf("ace's module stands since %v, %v; want the zero time", s, ok) - } - if s, ok := p.Since("g14", NodeBuildModule); !ok || !s.Equal(placed) { - t.Errorf("g14's module stands since %v, %v; want the revision that placed it", s, ok) - } - if got := p.PlacedNodes(NodeBuildModule); strings.Join(got, ",") != "ace,g14" || len(p.ByNode()) != 2 || !p.Seeded() { - t.Errorf("PlacedNodes = %v, ByNode = %v", got, p.ByNode()) + if !gate.Revision(revision(9, "novox", "shanks")) || len(gate.Nodes()) != 2 { + t.Fatalf("g14 taken off: nodes %v", gate.Nodes()) } }) @@ -263,7 +84,7 @@ func TestToolchainGate(t *testing.T) { var logs bytes.Buffer logger := log.New(&logs, "", 0) refused := errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_assignments") - gate, _, _ := start(t, false) + gate := NewPlacedNodes(logger.Printf) for _, bus := range []refusingBus{{err: refused}, {refuseOpen: true, err: refused}} { if _, err := FollowAssignments(context.Background(), bus, gate, func(context.Context) {}, logger.Printf); err != nil { t.Errorf("a refused assignments stopped the start: %v", err) diff --git a/modules/mesh-delivery-builder/buildrun/toolchains.go b/modules/mesh-delivery-builder/buildrun/toolchains.go index a37200b9..ba2368c6 100644 --- a/modules/mesh-delivery-builder/buildrun/toolchains.go +++ b/modules/mesh-delivery-builder/buildrun/toolchains.go @@ -1,55 +1,54 @@ package buildrun import ( + "sort" "sync" "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) -// LanguageOf is the language each build module's toolchains state holds (the indexer's ToolchainModules, reversed). +// LanguageOf is the language each build module's toolchain reports hold. var LanguageOf = map[string]string{GoBuildModule: "go", NodeBuildModule: "typescript"} -// ToolchainSet holds what the build modules' toolchains buckets hold: each node's entry of a language, as the watches -// put them in. Safe for use from the watches and from a build at once. +// ToolchainsHeardState is the builder's old state section of the toolchains it heard (design 47 G §604, §546): one key +// ., the toolchain as the build seat said it. +const ( + ToolchainsHeardState = "toolchains-heard" + ToolchainsHeardBucket = "mesh-delivery-builder_" + ToolchainsHeardState +) + +// ToolchainSet holds each node's toolchain of a language, as the builder heard it (design 47 G §604). Safe for use from +// the consumer and from a build at once. type ToolchainSet struct { mu sync.Mutex by map[string]map[string]buildstep.Toolchain // language -> node -> entry } -// Put takes a change of a build module's toolchains bucket in: a put of a node's entry (key .) or its -// delete. It answers whether the language's entries changed (a node's version or recipe, or a node appearing or going). -// A key or a value that is not an entry of the module's language is left out. -func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool { - language, ok := LanguageOf[module] - if !ok { - return false - } - node, keyLanguage, err := buildstep.ParseToolchainKey(key) - if err != nil || keyLanguage != language { - return false - } +// Set takes a node's toolchain in and answers whether the language's entries changed (a node's version, recipe or +// digest, or a node appearing). +func (s *ToolchainSet) Set(node string, tc buildstep.Toolchain) bool { s.mu.Lock() defer s.mu.Unlock() - if deleted { - if _, had := s.by[language][node]; !had { - return false - } - delete(s.by[language], node) - return true - } - tc, err := buildstep.DecodeToolchain(value) - if err != nil || tc.Language != language { - return false - } if s.by == nil { s.by = map[string]map[string]buildstep.Toolchain{} } - if s.by[language] == nil { - s.by[language] = map[string]buildstep.Toolchain{} + if s.by[tc.Language] == nil { + s.by[tc.Language] = map[string]buildstep.Toolchain{} } - cur, had := s.by[language][node] - s.by[language][node] = tc - return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe + cur, had := s.by[tc.Language][node] + s.by[tc.Language][node] = tc + return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe || cur.Digest != tc.Digest +} + +// Delete takes a node's toolchain of a language out and answers whether it was held. +func (s *ToolchainSet) Delete(node, language string) bool { + s.mu.Lock() + defer s.mu.Unlock() + if _, had := s.by[language][node]; !had { + return false + } + delete(s.by[language], node) + return true } // Entries are the nodes' entries of a language, in no particular order. @@ -62,3 +61,33 @@ func (s *ToolchainSet) Entries(language string) []buildstep.NodeToolchain { } return out } + +// AgreedDigest is a digest of the agreed version and the nodes reporting it. +type AgreedDigest struct { + Digest string + Nodes []string +} + +// Digests are the digests the nodes reporting the agreed toolchain's version report, sorted by digest, each with its +// nodes sorted (design 47 G §472, §574). A node that reports no digest is left out. +func Digests(entries []buildstep.NodeToolchain, agreed buildstep.Toolchain) []AgreedDigest { + by := map[string][]string{} + for _, e := range entries { + if e.Version == agreed.Version && e.Recipe == agreed.Recipe && e.Digest != "" { + by[e.Digest] = append(by[e.Digest], e.Node) + } + } + out := make([]AgreedDigest, 0, len(by)) + for d, nodes := range by { + sort.Strings(nodes) + out = append(out, AgreedDigest{Digest: d, Nodes: nodes}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Digest < out[j].Digest }) + return out +} + +// The builder's tools over the toolchains it heard (design 47 G §604, §546). +const ( + ToolchainsTool = "builder_toolchains" + ToolchainsDropTool = "builder_toolchains_drop" +) diff --git a/modules/mesh-delivery-builder/buildrun/toolchains_test.go b/modules/mesh-delivery-builder/buildrun/toolchains_test.go index 72aeab92..82b78c5e 100644 --- a/modules/mesh-delivery-builder/buildrun/toolchains_test.go +++ b/modules/mesh-delivery-builder/buildrun/toolchains_test.go @@ -1,50 +1,49 @@ package buildrun import ( + "reflect" "testing" - "time" "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) -func tcValue(t *testing.T, language, version string) []byte { - t.Helper() - now := time.Now() - b, err := buildstep.EncodeToolchain(buildstep.Toolchain{Language: language, Version: version, Recipe: "1", At: now}) - if err != nil { - t.Fatal(err) - } - return b +func tc(language, version, digest string) buildstep.Toolchain { + return buildstep.Toolchain{Language: language, Version: version, Recipe: "1", Digest: digest} } -func TestToolchainSetKeepsEachNodesEntryByLanguageAndSaysWhenOneChanged(t *testing.T) { // 1607/AC6 +func TestToolchainSetKeepsEachNodesEntryByLanguageAndSaysWhenOneChanged(t *testing.T) { // 1607/AC6, design 47 G §604 var s ToolchainSet - if !s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + if !s.Set("novox", tc("go", "go1.27.1", "sha256:a")) { t.Error("a first entry did not count as a change") } - if s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + if s.Set("novox", tc("go", "go1.27.1", "sha256:a")) { t.Error("the same entry again counted as a change") } - s.Put(GoBuildModule, "ace.go", tcValue(t, "go", "go1.27.1"), false) - s.Put(NodeBuildModule, "ace.typescript", tcValue(t, "typescript", "tsc5"), false) + if !s.Set("novox", tc("go", "go1.27.1", "sha256:b")) { + t.Error("a new digest at one version did not count as a change") + } + s.Set("ace", tc("go", "go1.27.1", "sha256:b")) + s.Set("ace", tc("typescript", "tsc5", "")) if got := s.Entries("go"); len(got) != 2 { t.Errorf("go entries = %+v, want two", got) } if got := s.Entries("typescript"); len(got) != 1 || got[0].Node != "ace" || got[0].Version != "tsc5" { t.Errorf("typescript entries = %+v", got) } - if !s.Put(GoBuildModule, "ace.go", nil, true) || len(s.Entries("go")) != 1 { - t.Error("a delete did not remove the node's entry") + if !s.Delete("ace", "go") || len(s.Entries("go")) != 1 || s.Delete("ace", "go") { + t.Error("a delete did not remove the node's entry once") } - // What is not a toolchain of the module's language is left out. - s = ToolchainSet{} - if s.Put(GoBuildModule, "novox.typescript", tcValue(t, "typescript", "x"), false) { - t.Error("a typescript toolchain in the go build module's bucket counted") - } - if s.Put(GoBuildModule, "novox.go", []byte("not json"), false) || s.Put("other-module", "novox.go", tcValue(t, "go", "x"), false) { - t.Error("garbage or an unknown module counted") - } - if len(s.Entries("go")) != 0 { - t.Error("an entry was kept") +} + +// Design 47 G §472, §574: the digests are those the nodes reporting the agreed version report, sorted, with their nodes. +func TestDigestsOfTheAgreedVersion(t *testing.T) { + entries := []buildstep.NodeToolchain{ + {Node: "novox", Toolchain: tc("go", "1.26.0", "sha256:b")}, {Node: "ace", Toolchain: tc("go", "1.26.0", "sha256:a")}, + {Node: "g14", Toolchain: tc("go", "1.26.0", "sha256:a")}, {Node: "shanks", Toolchain: tc("go", "1.25.0", "sha256:c")}, + {Node: "old", Toolchain: tc("go", "1.26.0", "")}} + got := Digests(entries, tc("go", "1.26.0", "")) + want := []AgreedDigest{{Digest: "sha256:a", Nodes: []string{"ace", "g14"}}, {Digest: "sha256:b", Nodes: []string{"novox"}}} + if !reflect.DeepEqual(got, want) { + t.Errorf("digests = %+v, want %+v", got, want) } } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index bbd75eaf..7e201f0a 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -372,6 +372,7 @@ func main() { }} tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, fr, ofr) tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...) + tools = append(tools, toolchainsTools(stdio.State(buildrun.ToolchainsHeardState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) // The builder's expiring keys (state bounds): what each bounds and when it ends. tools = append(tools, stdio.Tool{Name: versions.BoundsTool, Description: versions.BoundsDescription, Input: map[string]any{"type": "object", "properties": map[string]any{}}, diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains.go new file mode 100644 index 00000000..06ec72b9 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains.go @@ -0,0 +1,82 @@ +package main + +import ( + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/sdk/buildstep" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// heardState is the builder's state toolchains-heard as the tools bundle reads and deletes in it. +type heardState interface { + Keys() ([]string, error) + Get(key string) (*stdio.StateEntry, error) + Delete(key string) error +} + +// toolchainsTools are builder_toolchains, which lists the toolchains the builder heard from the build seats, and +// builder_toolchains_drop, which deletes one key and writes it in the hand-act log (design 47 G §604, §546). The drop is +// of class act, with the default callers: the bus refuses another caller. +func toolchainsTools(st heardState, acts meshhandactsclient.Recorder, now func() time.Time) []stdio.Tool { + return []stdio.Tool{ + {Name: buildrun.ToolchainsTool, + Description: "The toolchains the builder heard from the build seats' toolchain-reported, one key . each: its version, " + + "recipe, the digest of its toolchain program and when it was said. The builder agrees a language's version from these, and forgets a node's " + + "when the node-engine seat says the build module stopped there. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + keys, err := st.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + rows := []map[string]any{} + for _, k := range keys { + e, err := st.Get(k) + if err != nil || e == nil { + continue + } + row := map[string]any{"key": k} + if tc, err := buildstep.DecodeToolchain(e.Value); err != nil { + row["error"] = err.Error() + } else { + row["version"], row["recipe"], row["digest"], row["said"] = tc.Version, tc.Recipe, tc.Digest, tc.At + } + rows = append(rows, row) + } + return map[string]any{"toolchains": rows}, nil + }}, + {Name: buildrun.ToolchainsDropTool, + Description: "Drop one node's toolchain of a language from what the builder heard, by hand: a build module that left a node and said nothing. " + + "The builder agrees without it; the node's build module says it again at its start and on each change. Written in the hand-act log (mesh-hand-acts).", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "., as builder_toolchains lists it"}, + "why": map[string]any{"type": "string", "description": "why, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + key, why := str("key"), str("why") + if _, _, err := buildstep.ParseToolchainKey(key); err != nil { + return nil, err + } + if why == "" { + return nil, fmt.Errorf("why is required") + } + if e, err := st.Get(key); err != nil || e == nil { + return nil, fmt.Errorf("%s holds no key %s", buildrun.ToolchainsHeardState, key) + } + if _, err := acts.Record(meshhandactsclient.Request{What: fmt.Sprintf("dropped the toolchain %s from what the builder heard", key), + Why: why, Cause: buildrun.ToolchainsDropTool, Verb: "toolchains-drop"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err) + } + if err := st.Delete(key); err != nil { + return nil, err + } + return map[string]any{"key": key, "dropped": true, "at": now().UTC()}, nil + }}, + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains_test.go new file mode 100644 index 00000000..70429914 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/toolchains_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/sdk/buildstep" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +type fakeHeard struct{ m map[string]json.RawMessage } + +func (f *fakeHeard) Keys() ([]string, error) { + var out []string + for k := range f.m { + out = append(out, k) + } + return out, nil +} +func (f *fakeHeard) Get(k string) (*stdio.StateEntry, error) { + v, ok := f.m[k] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: k, Value: v}, nil +} +func (f *fakeHeard) Delete(k string) error { delete(f.m, k); return nil } + +// Design 47 G §604, §546: builder_toolchains lists the keys; builder_toolchains_drop deletes one with its why, after the +// hand act is recorded, and nothing is deleted when it is not. +func TestToolchainsToolsListAndDropAKeyAfterTheHandAct(t *testing.T) { + data, err := buildstep.EncodeToolchain(buildstep.Toolchain{Language: "go", Version: "1.26.0", Recipe: "1", Digest: "sha256:a", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + st := &fakeHeard{m: map[string]json.RawMessage{"ace.go": data}} + asked := 0 + var failAct error + acts := meshhandactsclient.Recorder{Ask: func(string, any) (json.RawMessage, error) { + asked++ + return json.RawMessage(`{"id":"h1"}`), failAct + }} + tools := toolchainsTools(st, acts, time.Now) + run := func(name string, args map[string]any) (any, error) { + for _, tl := range tools { + if tl.Name == name { + return tl.Run(args) + } + } + t.Fatalf("no tool %s", name) + return nil, nil + } + got, err := run(buildrun.ToolchainsTool, nil) + rows := got.(map[string]any)["toolchains"].([]map[string]any) + if err != nil || len(rows) != 1 || rows[0]["key"] != "ace.go" || rows[0]["digest"] != "sha256:a" { + t.Fatalf("listed %v, %v", got, err) + } + if _, err := run(buildrun.ToolchainsDropTool, map[string]any{"key": "ace.go", "why": ""}); err == nil { + t.Fatal("a drop with no why was taken") + } + failAct = json.Unmarshal([]byte("x"), new(int)) + if _, err := run(buildrun.ToolchainsDropTool, map[string]any{"key": "ace.go", "why": "ace left"}); err == nil || !strings.Contains(err.Error(), "nothing was dropped") || len(st.m) != 1 { + t.Fatalf("a hand act not recorded still dropped: %v", err) + } + failAct = nil + if _, err := run(buildrun.ToolchainsDropTool, map[string]any{"key": "ace.go", "why": "ace left"}); err != nil || len(st.m) != 0 || asked < 2 { + t.Fatalf("the drop: %v, keys left %d", err, len(st.m)) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go index ec0783ec..4c50f3ca 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go @@ -69,7 +69,7 @@ func eventsConsumer(t *testing.T, b *bed, node string) { t.Fatal(err) } if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", - AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{sweep.SubjectArtifactsGone}}); err != nil { + AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{sweep.SubjectArtifactsGone, "mesh.seat.node-build-go.event.toolchain-reported.*", "mesh.seat.node-build-node.event.toolchain-reported.*"}}); err != nil { t.Fatal(err) } } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index c277bc6c..fbfc0f50 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -213,7 +213,7 @@ type bed struct { candidates jetstream.KeyValue source jetstream.KeyValue builds jetstream.KeyValue - tools map[string]jetstream.KeyValue + heard jetstream.KeyValue // the state toolchains-heard instances jetstream.KeyValue packages jetstream.KeyValue // the indexer's packages, served by its read tool and views family out *saidLog // the builder's process log, every instance's @@ -265,21 +265,21 @@ func startBedWith(t *testing.T, closed bool) *bed { } ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) t.Cleanup(cancel) - b := &bed{ctx: ctx, js: js, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL(), dir: dir} - for module, stream := range toolchainStreams { - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil { + b := &bed{ctx: ctx, js: js, srvURL: srv.ClientURL(), dir: dir} + // EVENTS, where the build seats say toolchain-reported (design 47 G §492), with each builder's durable consumer of it, and + // the builder's state toolchains-heard (§604). + if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { + t.Fatal(err) + } + for _, node := range []string{"n1", "n2"} { + if _, err := js.CreateOrUpdateConsumer(ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } - for _, node := range []string{"n1", "n2"} { - if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { - t.Fatal(err) - } - } - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buildrun.ToolchainsBucket(module), History: 1}) - if err != nil { - t.Fatal(err) - } - b.tools[module] = kv + } + if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buildrun.ToolchainsHeardBucket, History: 1}); err != nil { + t.Fatal(err) + } else { + b.heard = kv } mk := func(name string, history int) jetstream.KeyValue { kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)}) @@ -344,18 +344,15 @@ func (b *bed) logged(t *testing.T, module string) []builds.BuildLogEntry { return out } -// report writes a node's toolchains key of a build module and says so on the module's stream, as the module does. +// report says a node's toolchain of a language on the build module's seat, as the seat's holder does (design 47 G §492). func (b *bed) report(t *testing.T, module, node, language, version, recipe string) { t.Helper() - data, err := sdkstep.EncodeToolchain(sdkstep.Toolchain{Language: language, Version: version, Recipe: recipe, At: time.Now().UTC()}) + seat := map[string]string{"mesh-build-go": "node-build-go", "mesh-build-node": "node-build-node"}[module] + data, err := json.Marshal(map[string]string{"node": node, "language": language, "version": version, "recipe": recipe, "digest": "sha256:" + strings.Repeat("a", 64)}) if err != nil { t.Fatal(err) } - key, _ := sdkstep.ToolchainKey(node, language) - if _, err := b.tools[module].Put(b.ctx, key, data); err != nil { - t.Fatal(err) - } - if _, err := b.js.Publish(b.ctx, "mesh.stream."+module+".event.toolchain.reported", []byte(`{}`)); err != nil { + if _, err := b.js.Publish(b.ctx, "mesh.seat."+seat+".event.toolchain-reported."+node, data); err != nil { t.Fatal(err) } } 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 c04e2884..9a743179 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -31,6 +31,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/sdk/buildstep" @@ -247,12 +248,27 @@ 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 catalogue places the build module on count (issue 1940, ADR 0549 §3): the - // assignments 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 + // The toolchains the build seats say (design 47 G §492, §604): kept in the state toolchains-heard, read back at the start, so + // a build that starts from the first modules value finds the nodes; the assignments only counts the nodes (ADR 0624 §1). + gate := buildrun.NewPlacedNodes(logger.Printf) + heardTools := newToolchainsHeard(tools, b.NodesChanged, func(ctx context.Context, e meshbuilds.ToolchainAgreed) error { + return emitSeatEvent(context.WithoutCancel(ctx), js, cred, meshbuilds.SubjectToolchainAgreed, "toolchain-agreed-"+e.Language+"-"+e.Version, e) + }, logger) + process.AwaitBucket(ctx, js, conn, buildrun.ToolchainsHeardBucket, "the toolchains heard are kept in memory only, and builder_toolchains shows none", logger.Printf). + Ready(func(kv jetstream.KeyValue) { + heardTools.bind(ctx, kvHeard{kvKeeper{ctx: ctx, kv: kv}}) + heardTools.readRunning(ctx, js) + dropped := func(e jetstream.KeyValueEntry) { + if !opOf(e) { + heardTools.Dropped(ctx, e.Key()) + } + } + if w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()); err != nil { + process.NotWatched(logger.Printf, buildrun.ToolchainsHeardBucket, err, "builder_toolchains_drop is not heard until the builder starts again") + } else { + go func() { errs <- follow(ctx, w, dropped) }() + } + }) // What each counted node's engine reads (ADR 0618 §2, ADR 0624 §1): asked through the node-engine seat's verb as the // builder's own account, for every node the assignments names, and kept in the state engine-reads once it is bound. reads := newEngineReads(calls.CallKeptOn, gate, logger.Printf) @@ -284,7 +300,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // asked when the bound ends), is call-unanswered in the builder's findings (issue 1966, ADR 0668 §3, ADR 0596 §2). reads.callsOf(found, pubBoundsKV.KV, heardBy(calls.CallKept), logger.Printf) b.Forms = reads.built - // Until the assignments's first view is answered the gate passes every report on, as before the assignments is settled. + // The assignments names the nodes the builder counts, once its first view is answered. assignmentsErrs := make(chan error, 1) go func() { select { @@ -299,12 +315,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } assignmentsErrs <- <-pe }() - if err := watch.start(); err != nil { - return err - } - // A build module says each time it wrote its toolchains key: its stream, and the builder's durable consumer on it, are - // made by the controller. Where the stream is not there yet the builder says so and is started again with the build - // module. // Each consumer that is not there yet is the finding family-not-heard naming its stream until it binds at a later // start (heard.Writer; ADR 0615 §5); one there that cannot be heard stops the builder, as before. notHeard := found.notHeard(ctx) @@ -324,9 +334,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil }) } - for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} { - hear(toolchainStreams[module], module+"'s toolchain reports are not heard until the builder starts again", func(m jetstream.Msg) { watch.Reported(m) }) - } // The seat mesh-artifact-sweep says artifacts-gone when it found a key's blob gone (issue 1937, ADR 0571 §2): the // builder's durable consumer on EVENTS, made by the grants part from the manifest's consumes. Where it is not there yet the builder says // so; it still asks held at its start and at every module value. @@ -336,7 +343,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version. seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs, res: reserved} events := func(m jetstream.Msg) { - if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4) + if reads.heard(m, logger.Printf) || heardTools.Reported(ctx, m) || heardTools.Running(ctx, m) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4) return } heardGone(ctx, b, m, logger.Printf) @@ -430,8 +437,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-assignmentsErrs: return err - case err := <-watch.errs: - return err } } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go index 6b3b86c3..2c1549b5 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go @@ -26,7 +26,7 @@ func TestTheNamesAreTheManifests(t *testing.T) { } for _, read := range []string{ gateModule + ".source", - "mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", } { if !slices.Contains(m.Reads, read) { t.Errorf("the builder reads %s, its manifest's reads are %v", read, m.Reads) @@ -40,7 +40,7 @@ func TestTheNamesAreTheManifests(t *testing.T) { } for want, got := range map[string]string{ "mesh-delivery-gate_source": sourceBucket, "mesh-build-go_steps": buildrun.StepsBucket("mesh-build-go"), - "mesh-build-node_toolchains": buildrun.ToolchainsBucket("mesh-build-node"), "mesh-delivery-indexer_packages": packagesBucket, + buildrun.ToolchainsHeardBucket: "mesh-delivery-builder_toolchains-heard", "mesh-delivery-indexer_packages": packagesBucket, "mesh-delivery-assigner_assignments": buildrun.AssignmentsBucket, } { if want != got { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go index 1cc99503..cd9a1702 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go @@ -28,7 +28,7 @@ const engineReadsBucket = builds.BuilderModule + "_" + EngineReadsState // engineReads keeps the statements and the counted nodes for the builder's forms (build.Builder.Forms). type engineReads struct { keeper *enginereads.Keeper - gate *buildrun.ToolchainGate + gate *buildrun.PlacedNodes mu sync.Mutex counted []string // window is the old reference form's window over these statements (DLV-B9); nil keeps none. @@ -39,7 +39,7 @@ type engineReads struct { } // newEngineReads asks through call (process.Conn.CallKeptOn, a kept call; ADR 0668) and keeps each answer in the state once it is bound. -func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads { +func newEngineReads(call enginereads.Caller, gate *buildrun.PlacedNodes, logf func(string, ...any)) *engineReads { r := &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}} r.keeper.Changed = func() { r.stepWindow() 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 index d3d4bf22..cafc3e8a 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -34,24 +34,16 @@ import ( "mesh-delivery-builder/internal/versions" ) -// 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 seat's assignments, and its refusal, are mesh-sdk's (buildrun.ToolchainGate, +// 1945 (a part exits at start for a bucket it works without): a refused bucket it cannot work without is an error naming the bucket and +// its grant. The gate over the seat's assignments, and its refusal, are mesh-sdk's (buildrun.PlacedNodes, // issue 774). func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) { t.Run("a floor's wake is heard on the events consumer (design 51 The floor)", aFloorsWakeIsHeard) t.Run("a failed trunk build is a finding until a later done one (DLV-B7)", aFailedTrunkBuildIsAFinding) t.Run("a branch's end is acknowledged once its tagged versions are removed (design 51 Tagged versions)", aBranchsEndIsHeard) - 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()) - } + refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_modules")} 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) @@ -74,7 +66,7 @@ func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) { // away node keeps its last statement; a node the assignments no longer names leaves the count, its key removed; and // builder_engine_reads drops one node's key. func theCountedNodesEngineReads(t *testing.T) { - gate := buildrun.NewToolchainGate(nil, func(string, string, []byte, bool) bool { return false }, t.Logf) + gate := buildrun.NewPlacedNodes(t.Logf) place := func(nodes ...string) { for _, node := range []string{"n1", "n2"} { pl := meshmoduleassigner.Assignments{Node: node} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go index 82734a16..62c88a80 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -2,107 +2,290 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "log" "strings" "sync" - - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" - "git.novox.be/novox/mesh-sdk/go/process" + "time" "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/sdk/buildstep" ) -// toolchainStreams are the stream each build module carries its toolchain.reported event on, as the bus names it -// (declared by the build modules, heard by this module through consumes). -var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO", buildrun.NodeBuildModule: "TOOLCHAINS_NODE"} - -// toolchainWatch keeps the build modules' toolchains in a buildrun.ToolchainSet and tells the builder when one changed. A -// build module's bucket is made when it is assigned, after the builder started, and the builder holds no timer (ADR 0437 -// §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 - // 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 - mu sync.Mutex - watched map[string]bool +// toolchainReportedPrefixes are the subjects the three build seats say toolchain-reported on, a node its last token (design +// 47 G §492). The builder hears them through the consumer of its manifest's consumes, and keeps what it heard in its +// state toolchains-heard (§604). +var toolchainReportedPrefixes = []string{ + strings.TrimSuffix(nodebuildgo.SubjectToolchainReportedAll, "*"), + strings.TrimSuffix(nodebuildnode.SubjectToolchainReportedAll, "*"), } -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, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}} +// heardKept is the state toolchains-heard: one key . (design 47 G §604). +type heardKept interface { + Put(key string, value []byte) error + Delete(key string) error } -// start watches the buckets that exist now; a missing one is said and waits for its build module's event. -func (w *toolchainWatch) start() error { - for module := range buildrun.LanguageOf { - if err := w.ensure(module); err != nil { - return err +// heardState is the state as the builder reads it back at its start. +type heardState interface { + heardKept + Keys() ([]string, error) + Get(key string) ([]byte, error) +} + +// kvHeard is heardState over the bound bucket. +type kvHeard struct { + kvKeeper +} + +func (k kvHeard) Keys() ([]string, error) { + keys, err := k.kv.Keys(k.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +func (k kvHeard) Get(key string) ([]byte, error) { + e, err := k.kv.Get(k.ctx, key) + if err != nil { + return nil, err + } + return e.Value(), nil +} + +// toolchainsHeard keeps the toolchains the build seats said (design 47 G §492, §604), forgets a node's on the node-engine +// seat's stopped (§546), and says toolchain-agreed with its digests when a language's agreed version changes (§472, §574). +type toolchainsHeard struct { + set *buildrun.ToolchainSet + changed func(context.Context) + say func(ctx context.Context, e meshbuilds.ToolchainAgreed) error + now func() time.Time + logf func(string, ...any) + + mu sync.Mutex + kept heardKept // nil until the state is bound: what was heard is kept in memory only, and written then + agreed map[string]string // language -> the version last said + runAt map[string]time.Time // . -> the `at` of the newest running read +} + +func newToolchainsHeard(set *buildrun.ToolchainSet, changed func(context.Context), say func(context.Context, meshbuilds.ToolchainAgreed) error, logger *log.Logger) *toolchainsHeard { + return &toolchainsHeard{set: set, changed: changed, say: say, now: time.Now, logf: logger.Printf, agreed: map[string]string{}, runAt: map[string]time.Time{}} +} + +// bind takes the bound state in: the toolchains it holds are read back (a restart needs no node to report again), and +// what was heard before it was bound is written. +func (h *toolchainsHeard) bind(ctx context.Context, state heardState) { + keys, err := state.Keys() + if err != nil { + h.logf("%s cannot be read, so the toolchains of before this start are not known: %v", buildrun.ToolchainsHeardState, err) + } + for _, k := range keys { + value, err := state.Get(k) + if err != nil { + continue + } + node, language, err := buildstep.ParseToolchainKey(k) + tc, derr := buildstep.DecodeToolchain(value) + if err != nil || derr != nil || tc.Language != language { + h.logf("%s.%s holds what is not a toolchain, and is left: %v %v", buildrun.ToolchainsHeardState, k, err, derr) + continue + } + h.set.Set(node, tc) + } + h.mu.Lock() + h.kept = state + h.mu.Unlock() + for _, language := range buildrun.LanguageOf { + for _, e := range h.set.Entries(language) { + h.keep(e.Node, e.Toolchain) } } - return nil + h.agree(ctx) + h.changed(ctx) } -func (w *toolchainWatch) ensure(module string) error { - w.mu.Lock() - defer w.mu.Unlock() - if w.watched[module] { - return nil +func (h *toolchainsHeard) keep(node string, tc buildstep.Toolchain) { + h.mu.Lock() + kept := h.kept + h.mu.Unlock() + if kept == nil { + return } - name := buildrun.ToolchainsBucket(module) - kv, err := w.js.KeyValue(w.ctx, name) - switch { - case errors.Is(err, jetstream.ErrBucketNotFound): - w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) - return nil - case err != nil: - // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report. - w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module) - return nil - } - watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) }) + key, err := buildstep.ToolchainKey(node, tc.Language) if err != nil { - process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module)) - return nil - } - w.watched[module] = true - go func() { - w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { - if w.put(module, e.Key(), e.Value(), !opOf(e)) { - w.changed(w.ctx) - } - }) - }() - w.changed(w.ctx) // what the settle read is new to the builder - return nil -} - -// delivered is the part of a JetStream message the watch uses. -type delivered interface { - Subject() string - Ack() error - Term() error -} - -// Reported takes one toolchain.reported event of a build module in: the module is read off the subject, -// mesh.stream..event.toolchain.reported, and its bucket is watched if it is not. A subject of no build module is -// terminated; a failure stays unacknowledged, and the server delivers it again. -func (w *toolchainWatch) Reported(m delivered) { - module, _, _ := strings.Cut(strings.TrimPrefix(m.Subject(), "mesh.stream."), ".event.") - if _, known := buildrun.LanguageOf[module]; !known || !strings.HasPrefix(m.Subject(), "mesh.stream.") { - w.logger.Printf("%s: not a build module's event, dropped", m.Subject()) - _ = m.Term() return } - if err := w.ensure(module); err != nil { - w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) - return + data, err := buildstep.EncodeToolchain(tc) + if err == nil { + err = kept.Put(key, data) + } + if err != nil { + h.logf("%s.%s could not be written, so it is kept in memory only: %v", buildrun.ToolchainsHeardState, key, err) + } +} + +// Reported takes one toolchain-reported of a build seat in; it answers whether the message was one. The toolchain is +// kept in the state, and builds that waited for the nodes to agree are looked at again. +func (h *toolchainsHeard) Reported(ctx context.Context, m jetstream.Msg) bool { + subject := m.Subject() + var prefix string + for _, p := range toolchainReportedPrefixes { + if strings.HasPrefix(subject, p) { + prefix = p + } + } + if prefix == "" { + return false } _ = m.Ack() + var e nodebuildgo.ToolchainReported + if err := json.Unmarshal(m.Data(), &e); err != nil || e.Node == "" || e.Node != strings.TrimPrefix(subject, prefix) { + h.logf("%s holds no toolchain of its node, and is dropped: %v", subject, err) + return true + } + tc := buildstep.Toolchain{Language: e.Language, Version: e.Version, Recipe: e.Recipe, Digest: e.Digest, At: h.now().UTC()} + if _, err := buildstep.EncodeToolchain(tc); err != nil { + h.logf("%s is no toolchain, and is dropped: %v", subject, err) + return true + } + if h.set.Set(e.Node, tc) { + h.keep(e.Node, tc) + h.agree(ctx) + h.changed(ctx) + } + return true +} + +// Running takes one running of the node-engine seat in (design 47 G §546); it answers whether the message was one. A +// running with no fingerprint is stopped (§365): the key . of the build module of that language on that +// node is deleted. An event older by `at` than one taken before it changes nothing. +func (h *toolchainsHeard) Running(ctx context.Context, m jetstream.Msg) bool { + if !strings.HasPrefix(m.Subject(), strings.TrimSuffix(nodeengine.SubjectRunningAll, "*.*")) { + return false + } + _ = m.Ack() + h.running(ctx, m.Data()) + return true +} + +func (h *toolchainsHeard) running(ctx context.Context, data []byte) { + var r nodeengine.Running + if err := json.Unmarshal(data, &r); err != nil { + h.logf("a running that does not read is dropped: %v", err) + return + } + id := r.Node + "." + r.Module + h.mu.Lock() + if prev, ok := h.runAt[id]; ok && r.At.Before(prev) { + h.mu.Unlock() + return + } + h.runAt[id] = r.At + kept := h.kept + h.mu.Unlock() + language, build := buildrun.LanguageOf[r.Module] + if !build || r.Fingerprint != "" { + return + } + if !h.set.Delete(r.Node, language) { + return + } + key, _ := buildstep.ToolchainKey(r.Node, language) + if kept != nil { + if err := kept.Delete(key); err != nil { + h.logf("%s.%s could not be deleted: %v", buildrun.ToolchainsHeardState, key, err) + } + } + h.agree(ctx) + h.changed(ctx) +} + +// Dropped is the key's drop by a person (builder_toolchains_drop), taken from the state's watch. +func (h *toolchainsHeard) Dropped(ctx context.Context, key string) { + node, language, err := buildstep.ParseToolchainKey(key) + if err != nil { + return + } + if h.set.Delete(node, language) { + h.agree(ctx) + h.changed(ctx) + } +} + +// agree says toolchain-agreed for each language whose agreed version is not the one last said (design 47 G §472: said +// only when the agreed version changes, ADR 0584 §1). +func (h *toolchainsHeard) agree(ctx context.Context) { + for _, language := range buildrun.LanguageOf { + entries := h.set.Entries(language) + tc, none := buildstep.AgreedToolchain(entries) + if none != "" { + continue + } + h.mu.Lock() + same := h.agreed[language] == tc.Version + if !same { + h.agreed[language] = tc.Version + } + h.mu.Unlock() + if same { + continue + } + at := h.now().UTC() + e := meshbuilds.ToolchainAgreed{V: 1, At: &at, Language: language, Version: tc.Version, Digests: []meshbuilds.ToolchainAgreedDigestsItem{}} + for _, d := range buildrun.Digests(entries, tc) { + e.Digests = append(e.Digests, meshbuilds.ToolchainAgreedDigestsItem{Digest: d.Digest, Nodes: d.Nodes}) + } + if err := h.say(ctx, e); err != nil { + h.logf("saying %s of %s %s: %v", meshbuilds.EventToolchainAgreed, language, tc.Version, err) + h.mu.Lock() + delete(h.agreed, language) // said again at the next change + h.mu.Unlock() + } + } +} + +// lastPerSubject reads the last message of every subject of a stream kept last per subject, through an ordered consumer +// made for that read (design 47 G §556), and hands each to each; it answers when the last of them was handed. +func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte)) error { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return fmt.Errorf("reading %s: %w", stream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return fmt.Errorf("reading %s: %w", stream, err) + } + pending := int(info.NumPending) + if pending == 0 { + return nil + } + msgs, err := cons.Messages() + if err != nil { + return fmt.Errorf("reading %s: %w", stream, err) + } + defer msgs.Stop() + for ; pending > 0; pending-- { + m, err := msgs.Next() + if err != nil { + return fmt.Errorf("reading %s: %w", stream, err) + } + each(m.Subject(), m.Data()) + } + return nil +} + +// readRunning is the start's read of every node's module's last running (§556), before the consumer's. +func (h *toolchainsHeard) readRunning(ctx context.Context, js jetstream.JetStream) { + if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte) { h.running(ctx, data) }); err != nil { + h.logf("the running of the node-engine seat is not read at the start, so a build module stopped while this builder was away is forgotten at its next running: %v", err) + } } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains_test.go new file mode 100644 index 00000000..8018670c --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains_test.go @@ -0,0 +1,153 @@ +package main + +import ( + "context" + "encoding/json" + "io" + "log" + "sort" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" +) + +// memState is the state toolchains-heard in memory. +type memState struct { + mu sync.Mutex + m map[string][]byte +} + +func (s *memState) Put(k string, v []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.m == nil { + s.m = map[string][]byte{} + } + s.m[k] = v + return nil +} +func (s *memState) Delete(k string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.m, k) + return nil +} +func (s *memState) Get(k string) ([]byte, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.m[k], nil +} +func (s *memState) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + var out []string + for k := range s.m { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// msg is a delivered message as the builder's consumer hands it. +type msg struct { + jetstream.Msg + subject string + data []byte +} + +func (m msg) Subject() string { return m.subject } +func (m msg) Data() []byte { return m.data } +func (m msg) Ack() error { return nil } + +type heardRig struct { + h *toolchainsHeard + state *memState + said []meshbuilds.ToolchainAgreed + changed int +} + +func newHeardRig(t *testing.T, state *memState) *heardRig { + t.Helper() + r := &heardRig{state: state} + r.h = newToolchainsHeard(&buildrun.ToolchainSet{}, func(context.Context) { r.changed++ }, + func(_ context.Context, e meshbuilds.ToolchainAgreed) error { r.said = append(r.said, e); return nil }, log.New(io.Discard, "", 0)) + r.h.bind(context.Background(), state) + return r +} + +func (r *heardRig) report(t *testing.T, seat, node, language, version, digest string) { + t.Helper() + data, _ := json.Marshal(map[string]string{"node": node, "language": language, "version": version, "recipe": "1", "digest": digest}) + if !r.h.Reported(context.Background(), msg{subject: "mesh.seat." + seat + ".event.toolchain-reported." + node, data: data}) { + t.Fatal("a toolchain-reported was not taken") + } +} + +func (r *heardRig) running(t *testing.T, node, module, fingerprint string, at time.Time) { + t.Helper() + data, _ := json.Marshal(nodeengine.Running{Node: node, Module: module, Fingerprint: fingerprint, At: at}) + if !r.h.Running(context.Background(), msg{subject: nodeengine.SubjectRunning(node, module), data: data}) { + t.Fatal("a running was not taken") + } +} + +// Design 47 G §492, §604, §472, §574, §546: the builder agrees from the toolchain-reported it heard, keeps them in +// toolchains-heard, says toolchain-agreed with the digests only when the agreed version changes, and forgets a node's on +// the node-engine seat's stopped. +func TestTheBuilderKeepsTheToolchainsItHeardAndForgetsThemOnStopped(t *testing.T) { + da, db := "sha256:a", "sha256:b" + state := &memState{} + r := newHeardRig(t, state) + r.report(t, "node-build-go", "ace", "go", "1.26.0", da) + r.report(t, "node-build-go", "g14", "go", "1.26.0", da) + r.report(t, "node-build-go", "novox", "go", "1.26.0", db) + keys, _ := state.Keys() + if len(keys) != 3 || keys[0] != "ace.go" { + t.Fatalf("the state holds %v, want the three nodes' keys .go", keys) + } + if len(r.said) != 1 || r.said[0].Version != "1.26.0" || r.said[0].Language != "go" { + t.Fatalf("toolchain-agreed said %+v, want one for go 1.26.0 (said only when the version changes)", r.said) + } + // The first report agreed it with one digest; a later digest of the same version is not said again. + if len(r.said[0].Digests) != 1 { + t.Fatalf("the first agreement carried %+v", r.said[0].Digests) + } + // A restarted builder reads the keys back and agrees the same version with no new report; the digests it says are all three nodes'. + again := newHeardRig(t, state) + if len(again.said) != 1 || again.said[0].Version != "1.26.0" || len(again.said[0].Digests) != 2 || again.said[0].Digests[0].Nodes[1] != "g14" { + t.Fatalf("a restarted builder said %+v, want go 1.26.0 with the digests of its three nodes", again.said) + } + // A version change is said; the stopped build module of ace takes ace's key out, a placement taking nothing. + again.report(t, "node-build-go", "ace", "go", "1.27.0", da) + again.report(t, "node-build-go", "g14", "go", "1.27.0", da) + if len(again.said) != 2 || again.said[1].Version != "1.27.0" { + t.Fatalf("a changed agreed version was not said: %+v", again.said) + } + at := time.Now() + again.running(t, "ace", buildrun.GoBuildModule, "src2:x", at) // running: nothing is forgotten + if k, _ := state.Keys(); len(k) != 3 { + t.Fatalf("a running forgot a key: %v", k) + } + again.running(t, "ace", buildrun.GoBuildModule, "", at.Add(time.Second)) // stopped: no fingerprint + if k, _ := state.Keys(); len(k) != 2 || k[0] != "g14.go" { + t.Fatalf("stopped on ace left %v, want g14.go and novox.go", k) + } + // An older running read after (the start's read of the stream kept last per subject) changes nothing. + before := again.changed + again.running(t, "g14", buildrun.GoBuildModule, "", at) + again.running(t, "g14", buildrun.GoBuildModule, "src2:y", at.Add(-time.Hour)) + if k, _ := state.Keys(); len(k) != 1 || k[0] != "novox.go" || again.changed != before+1 { + t.Fatalf("after g14 stopped and an older running: %v", k) + } + // A key dropped by a person leaves the set. + again.h.Dropped(context.Background(), "novox.go") + if len(again.h.set.Entries("go")) != 0 { + t.Fatal("a dropped key stayed in the set") + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go index fbe610ac..ed1460ab 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go @@ -227,6 +227,25 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account return err } +// emitSeatEvent says an event of a seat the builder holds, on the seat's subject (mesh.seat..event.). +func emitSeatEvent(ctx context.Context, js jetstream.JetStream, cred process.Account, subject, id string, body any) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + msg := nats.NewMsg(subject) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + pub, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + _, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err +} + // The events on the builder's consumer of EVENTS that wake a floor (design 51 *The floor*, DLV-B3b). var ( manifestAcceptedSubject = meshcatalog.SubjectManifestAccepted diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 3d333a4d..cf46a190 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -13,6 +13,8 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/git v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-builder/internal/build/fakes_test.go b/modules/mesh-delivery-builder/internal/build/fakes_test.go index bd18bba5..fb490343 100644 --- a/modules/mesh-delivery-builder/internal/build/fakes_test.go +++ b/modules/mesh-delivery-builder/internal/build/fakes_test.go @@ -75,14 +75,9 @@ func presenceValue(t *testing.T, online bool) []byte { return b } -// tcValue is a toolchains value, as the build module's writer encodes it (mesh-sdk go/delivery EncodeToolchain). -func tcValue(t *testing.T, language, version string) []byte { - t.Helper() - b, err := buildstep.EncodeToolchain(buildstep.Toolchain{Language: language, Version: version, Recipe: "1", At: time.Now().UTC()}) - if err != nil { - t.Fatal(err) - } - return b +// tcOf is a node's toolchain of a language, as the builder keeps what it heard. +func tcOf(language, version string) buildstep.Toolchain { + return buildstep.Toolchain{Language: language, Version: version, Recipe: "1", At: time.Now().UTC()} } // tcs are the toolchain entries of a language on nodes, as the toolchains set gives them. diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 4d28e0a9..1c471d37 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -180,8 +180,8 @@ func newRig(t *testing.T, artifacts ...catalogue.Artifact) *rig { t.Helper() r := &rig{st: newMem(), bm: newBuildModules(), tools: &buildrun.ToolchainSet{}, mf: &fakeManifests{byDigest: map[string]catalogue.Manifest{mdigest('a'): {Module: "m", Build: &catalogue.Build{Artifacts: artifacts}}}}} for _, n := range []string{"ace", "novox"} { - r.tools.Put(buildrun.GoBuildModule, n+".go", tcValue(t, "go", "go1.27.1"), false) - r.tools.Put(buildrun.NodeBuildModule, n+".typescript", tcValue(t, "typescript", "tsc5"), false) + r.tools.Set(n, tcOf("go", "go1.27.1")) + r.tools.Set(n, tcOf("typescript", "tsc5")) } r.asks = &buildrun.FakeAsks{Answer: r.answer} r.b = &Builder{ @@ -235,8 +235,8 @@ func TestABundleAndAnArchiveAreBuiltOnOneNodeAndDone(t *testing.T) { // 1607/AC7 t.Run("ADR 0624: a done key keeps each form a counted node reads", aDoneKeyKeepsEachFormACountedNodeReads) r := newRig(t, goBundle, archiveA, tsBundle) // ADR 0584: a third node agrees with novox and ace's typescript is the odd one; the build is done and its step says ace sits out. - r.tools.Put(buildrun.NodeBuildModule, "g14.typescript", tcValue(t, "typescript", "tsc5"), false) - r.tools.Put(buildrun.NodeBuildModule, "ace.typescript", tcValue(t, "typescript", "tsc4"), false) + r.tools.Set("g14", tcOf("typescript", "tsc5")) + r.tools.Set("ace", tcOf("typescript", "tsc4")) defer func() { if st := r.st.vals[r.key]; !strings.Contains(string(st), "ace sits out: typescript tsc4, agreed tsc5") { t.Errorf("the key does not say ace sits out: %s", st) @@ -717,8 +717,8 @@ func TestAKeyIsBuiltOnlyWithTheToolchainItsFingerprintNamesAndNothingIsWrittenOt t.Errorf("keys %v and %d calls for a recipe that differs or a value that names no toolchain", r.st.vals, len(r.bm.calls)) } // The toolchain moves under a key that is not built yet: the old fingerprint stays unbuilt, the new one builds. - r.tools.Put(buildrun.GoBuildModule, "ace.go", tcValue(t, "go", "go1.27.2"), false) - r.tools.Put(buildrun.GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.2"), false) + r.tools.Set("ace", tcOf("go", "go1.27.2")) + r.tools.Set("novox", tcOf("go", "go1.27.2")) r.b.Reconsider(context.Background()) r.b.Wait() if len(r.st.vals) != 0 { @@ -740,11 +740,7 @@ func aRecipeThatMovesBuildsTheModuleAgainUnderItsNewFingerprint(t *testing.T) { calls := len(r.bm.calls) fp2 := "src2:" + hex64('2') for _, n := range []string{"ace", "novox"} { - b, err := buildstep.EncodeToolchain(buildstep.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "2", At: time.Now().UTC()}) - if err != nil { - t.Fatal(err) - } - r.tools.Put(buildrun.GoBuildModule, n+".go", b, false) + r.tools.Set(n, buildstep.Toolchain{Language: "go", Version: "go1.27.1", Recipe: "2", At: time.Now().UTC()}) } r.b.Reconsider(context.Background()) r.b.Wait() @@ -772,7 +768,7 @@ func aRecipeThatMovesBuildsTheModuleAgainUnderItsNewFingerprint(t *testing.T) { func TestNodesTheyDisagreeOnLeaveTheKeyUnwritten(t *testing.T) { // 1615/AC4 r := newRig(t, goBundle) - r.tools.Put(buildrun.GoBuildModule, "ace.go", tcValue(t, "go", "go1.26.0"), false) + r.tools.Set("ace", tcOf("go", "go1.26.0")) r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) r.b.Wait() if len(r.st.vals) != 0 || len(r.bm.calls) != 0 { diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 5c6c1068..0b5e7521 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -102,13 +102,13 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "toolchains-heard", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 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). Whether a // node's machine is heard it asks the watchdogs seat's presence (issue 1966, ADR 0596 §2, ADR 0668 §3). - if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", + if want := []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-delivery-assigner.assignments", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } @@ -137,18 +137,18 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 "seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want) } - // The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it - // starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the + // The builder hears the build seats' toolchain-reported and the node-engine seat's running (design 47 G §492, §546) and reads no + // build module's toolchains; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the // controller's no longer (1126/B9, ADR 0590). // Design 51 *The floor*: the indexer's manifest-accepted and the condition keeper's condition-cleared wake a silent floor. // DLV-B4: the four registry seats' outcomes, each a seat event `.` on the shared events stream. // DLV-M1: the indexer's and the assigner's views families of the states it follows. if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed", "go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed", - "mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled", + "mesh-artifact-sweep.artifacts-gone", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled", "mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-forge-bridge.forge.**", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", - "mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed", + "mesh-delivery-indexer.views.packages", "node-build-go.toolchain-reported", "node-build-node.toolchain-reported", "node-engine.manifest-reads-changed", "node-engine.running", "npm-package-registry.kept", "npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed", "oci-artifact-registry.kept", "oci-artifact-registry.published", "oci-artifact-registry.refused", "oci-artifact-registry.removed"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) @@ -161,7 +161,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 } // 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again. if want := []string{"builder_announce_again", "builder_ask_again", "builder_bounds", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_first_view_again", "builder_first_views", - "builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_versions", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) { + "builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_toolchains", "builder_toolchains_drop", "builder_versions", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } // Design 47 G §349: version.numbered is no longer said; old-form-closed on the family version, apart from the builds' ends. diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index f3d82244..3581b792 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -26,6 +26,8 @@ "builder_reserved_drop", "builder_seat_queue", "builder_stuck", + "builder_toolchains", + "builder_toolchains_drop", "builder_versions", "reserved", "builder_first_views", @@ -45,14 +47,15 @@ "go-package-registry.refused", "go-package-registry.removed", "mesh-artifact-sweep.artifacts-gone", - "mesh-build-go.toolchain.reported", - "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled", "mesh-conditions.condition-cleared", "mesh-delivery-forge-bridge.forge.**", + "node-build-go.toolchain-reported", + "node-build-node.toolchain-reported", "node-engine.manifest-reads-changed", + "node-engine.running", "npm-package-registry.kept", "npm-package-registry.published", "npm-package-registry.refused", @@ -103,6 +106,7 @@ { "name": "old-form", "history": 1 }, { "name": "bounds", "history": 1, "key-ttl": true }, { "name": "queues", "history": 1 }, + { "name": "toolchains-heard", "history": 1 }, { "name": "reserved", "history": 1 }, { "name": "reasks", "history": 1 }, { "name": "versions", "history": 1 }, @@ -113,10 +117,8 @@ ], "reads": [ "mesh-build-go.steps", - "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", - "mesh-build-node.toolchains", "mesh-delivery-assigner.assignments", "mesh-delivery-gate.source" ],