BL8: the builder hears the build seats' toolchain-reported, keeps them in toolchains-heard and says toolchain-agreed with digests

The builder agrees a language's toolchain from the three build seats'
toolchain-reported, kept in its state toolchains-heard (read back at its start),
forgets a node's on the node-engine seat's running with no fingerprint (stopped)
and never on a placement, says toolchain-agreed with its digests list only when
the agreed version changes; builder_toolchains and builder_toolchains_drop;
the manifest's reads of the build modules' toolchains are gone. The gate over
the assignments is now PlacedNodes, which only counts nodes.

Builds: design 47 G §492, §604, §546, §472, §574
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 20:49:02 +01:00
parent bd3727a900
commit 18f978e852
22 changed files with 818 additions and 610 deletions
@@ -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" }
@@ -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 <node>.<language> 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 // <module>/<key> -> 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()
}
@@ -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)
@@ -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
// <node>.<language>, 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 <node>.<language>) 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"
)
@@ -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)
}
}
@@ -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{}},
@@ -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 <node>.<language> 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": "<node>.<language>, 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
}},
}
}
@@ -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))
}
}
@@ -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)
}
}
@@ -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)
}
}
@@ -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
}
}
@@ -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 {
@@ -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()
@@ -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}
@@ -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 <node>.<language> (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 // <node>.<module> -> 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.<module>.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 <node>.<language> 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)
}
}
@@ -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 <node>.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")
}
}
@@ -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.<seat>.event.<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
+2
View File
@@ -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
@@ -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.
@@ -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 {
@@ -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 `<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.
+6 -4
View File
@@ -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"
],