novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge main into feat/r4g-finish (#33, #34)

Browse Source
This commit is contained in:
jschoubben
2026-10-24 13:26:24 +02:00
parent 2259d98c5b 42d880f075
commit ce7dcccb1d
25 changed files with 1623 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go
+185
View File
@@ -0,0 +1,185 @@
//go:build integration
package main
import (
"encoding/json"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A done key whose blob the artifact store no longer holds is built again, by the builder whole against the bus
// (novox/hq issue 1937): on 2026-10-24 the store's 03:30 collection deleted the bare blobs of three done keys the
// controller had never taken in, and the builder kept them done.
// heldVerb answers the controller seat's verb held on the bus, as the serving controller does: {result: {output, ok,
// answer}}; a reference is gone while gone says so.
type heldVerb struct {
mu sync.Mutex
gone func(reference string) bool
asked []string
}
func (h *heldVerb) serve(t *testing.T, b *bed) {
t.Helper()
sub, err := b.js.Conn().Subscribe("mesh.seat.mesh-controller.call.held.>", func(m *nats.Msg) {
var args struct {
References []string `json:"references"`
}
_ = json.Unmarshal(m.Data, &args)
a := delivery.HeldAnswer{Store: "store.test:5100", Held: []string{}, Gone: []string{}}
h.mu.Lock()
for _, r := range args.References {
h.asked = append(h.asked, r)
if h.gone(r) {
a.Gone = append(a.Gone, r)
} else {
a.Held = append(a.Held, r)
}
}
h.mu.Unlock()
reply, _ := json.Marshal(map[string]any{"result": map[string]any{"output": "held", "ok": true, "answer": a}, "node": "ctl"})
_ = m.Respond(reply)
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
func (h *heldVerb) asks() int {
h.mu.Lock()
defer h.mu.Unlock()
return len(h.asked)
}
// eventsConsumer makes EVENTS and the builder's durable consumer of artifacts-gone on it, as the controller makes them
// from the manifest's consumes.
func eventsConsumer(t *testing.T, b *bed, node string) {
t.Helper()
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
t.Fatal(err)
}
if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder",
AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
t.Fatal(err)
}
}
// doneKey writes builds.<module>.<fp> done, naming one archive at reference, and answers its revision.
func doneKey(t *testing.T, b *bed, module, manifest, reference, digest string) uint64 {
t.Helper()
key, _ := delivery.BuilderBuildKey(module, fp)
data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "n2.mesh-delivery-builder",
Manifest: manifest, Step: "built mesh-build-go on n2",
Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Digest: digest, Reference: reference}}})
if err != nil {
t.Fatal(err)
}
rev, err := b.builds.Put(b.ctx, key, data)
if err != nil {
t.Fatal(err)
}
return rev
}
func revisionOf(t *testing.T, b *bed, module string) uint64 {
t.Helper()
key, _ := delivery.BuilderBuildKey(module, fp)
e, err := b.builds.Get(b.ctx, key)
if err != nil {
t.Fatal(err)
}
return e.Revision()
}
// 1937/AC3 (integration): the builder, started whole, finds the done key of the trunk's fingerprint naming a blob held
// answers gone, builds it again as a new revision whose step says so, and leaves the rebuilt key done.
func TestADoneKeyWhoseBlobIsGoneIsBuiltAgainAtStart(t *testing.T) {
b := startBed(t)
eventsConsumer(t, b, "n1")
manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
old := digestOf("collected last night")
oldRef := "artifact-store://ca-trust/files/blobs/" + old
rev := doneKey(t, b, "ca-trust", manifest, oldRef, old)
h := &heldVerb{gone: func(r string) bool { return r == oldRef }}
h.serve(t, b)
b.start(t, "n1", nil)
v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
return v.State == delivery.BuildDone && len(v.Artifacts) == 1 && v.Artifacts[0].Digest == digestOf("files")
})
if !strings.Contains(v.Step, "built again: the artifact store no longer held "+oldRef) {
t.Errorf("the rebuilt key's step %q does not say why it was built again", v.Step)
}
if got := revisionOf(t, b, "ca-trust"); got <= rev {
t.Errorf("revision %d, was %d: want a new revision", got, rev)
}
if n := b.fake.count("mesh-build-go.build_go_build"); n != 1 {
t.Errorf("the archive was built %d times, want once", n)
}
if h.asks() == 0 {
t.Error("the builder never asked held")
}
}
// 1937/AC4 (integration): a done key whose blob held answers held at start stays done; when the controller says
// artifacts-gone for it, the builder asks again and builds it again.
func TestArtifactsGoneLeadsTheBuilderToBuildTheKeyAgain(t *testing.T) {
b := startBed(t)
eventsConsumer(t, b, "n1")
manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
d := digestOf("files")
ref := "artifact-store://ca-trust/files/blobs/" + d
rev := doneKey(t, b, "ca-trust", manifest, ref, d)
var mu sync.Mutex
collected := false
h := &heldVerb{gone: func(r string) bool {
mu.Lock()
defer mu.Unlock()
if r == ref && collected {
collected = false // gone once: the rebuild pushes it again
return true
}
return false
}}
h.serve(t, b)
b.start(t, "n1", nil)
for deadline := time.Now().Add(20 * time.Second); h.asks() == 0 && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
}
if h.asks() == 0 {
t.Fatal("the builder never asked held at its start")
}
time.Sleep(300 * time.Millisecond)
if got := revisionOf(t, b, "ca-trust"); got != rev || b.fake.count("mesh-build-go.build_go_build") != 0 {
t.Fatalf("a key whose blob is held was built again (revision %d, was %d)", got, rev)
}
mu.Lock()
collected = true
mu.Unlock()
key, _ := delivery.BuilderBuildKey("ca-trust", fp)
body, err := delivery.EncodeArtifactsGone(delivery.ArtifactsGone{Module: "ca-trust", Key: key, Fingerprint: fp,
Gone: []string{ref}, At: time.Now().UTC()})
if err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, delivery.ArtifactsGoneSubject, body); err != nil {
t.Fatal(err)
}
v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
return v.State == delivery.BuildDone && strings.Contains(v.Step, "built again")
})
if got := revisionOf(t, b, "ca-trust"); got <= rev || b.fake.count("mesh-build-go.build_go_build") != 1 {
t.Errorf("build = %+v at revision %d (was %d), %d builds: want built again once on the event", v, got, rev,
b.fake.count("mesh-build-go.build_go_build"))
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+38
View File
@@ -26,6 +26,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-builder/internal/build"
)
@@ -153,6 +154,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
asks := buildrun.JetStreamAsks{JS: js}
// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
account, err := process.ReadAccount(path)
if err != nil {
return err
}
calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
if err != nil {
return err
}
defer calls.Close()
b := &build.Builder{
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
@@ -160,6 +172,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Toolchains: tools,
Manifests: manifestReader{manifests},
Logf: logger.Printf,
Stored: storedVia(calls.Call),
}
// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
@@ -183,6 +196,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
// The toolchains first, so a build that starts from the first modules value finds the nodes.
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
// assignments are settled first, so the toolchains settle through the gate already knowing them.
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
watch.put = gate.Toolchain
assignErrs, err := watchAssignments(ctx, js, gate, b.NodesChanged, logger)
if err != nil {
return err
}
if err := watch.start(); err != nil {
return err
}
@@ -203,6 +224,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
defer cc.Stop()
}
// The controller says artifacts-gone when its take-in found a key's blob gone (issue 1937): the builder's durable
// consumer on EVENTS, made by the controller 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.
switch cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module); {
case err != nil:
logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
default:
cc, err := cons.Consume(func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) },
jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
if err != nil {
return err
}
defer cc.Stop()
}
each := func(e jetstream.KeyValueEntry) {
if opOf(e) {
b.Module(ctx, e.Key(), e.Value())
@@ -240,6 +276,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
case err := <-errs:
return err
case err := <-assignErrs:
return err
case err := <-watch.errs:
return err
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go
+174
View File
@@ -0,0 +1,174 @@
package main
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
// toolchain agreement this file repairs.
import (
"context"
"encoding/json"
"fmt"
"log"
"maps"
"slices"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provider"
)
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
//
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
// refusal is the guard of novox/hq ADR 0465, and it stays.
type toolchainGate struct {
mu sync.Mutex
modules map[string]bool // the build modules whose toolchains pass the gate
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
logf func(format string, a ...any)
known bool // the assignments are settled
standing map[string]bool // <node>.<module> with an open assignment
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
left map[string]bool // <module>/<key> said as left out, so it is said once
}
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
for _, m := range modules {
g.modules[m] = true
}
return g
}
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
g.mu.Lock()
defer g.mu.Unlock()
if g.raw[module] == nil {
g.raw[module] = map[string][]byte{}
}
if deleted {
delete(g.raw[module], key)
} else {
g.raw[module][key] = value
}
return g.pass(module, key)
}
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
node, module, ok := strings.Cut(key, ".")
if !ok || !g.modules[module] {
return false
}
open := !deleted && openAssignment(value)
g.mu.Lock()
defer g.mu.Unlock()
if g.standing[key] == open {
return false
}
if open {
g.standing[key] = true
} else {
delete(g.standing, key)
}
if !g.known {
return false
}
changed := false
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
changed = g.pass(module, k) || changed
}
}
return changed
}
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
// passed on changed.
func (g *toolchainGate) Settled() bool {
g.mu.Lock()
defer g.mu.Unlock()
g.known = true
changed := false
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
changed = g.pass(module, k) || changed
}
}
return changed
}
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
// otherwise. The caller holds mu.
func (g *toolchainGate) pass(module, key string) bool {
value, held := g.raw[module][key]
if !held {
return g.forward(module, key, nil, true)
}
node, _, err := delivery.ParseToolchainKey(key)
if err != nil || !g.known || g.standing[node+"."+module] {
delete(g.left, module+"/"+key)
return g.forward(module, key, value, false)
}
if !g.left[module+"/"+key] {
g.left[module+"/"+key] = true
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
delivery.ToolchainsBucket(module), key, module, node)
}
return g.forward(module, key, nil, true)
}
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
func openAssignment(value []byte) bool {
var a struct {
Ended time.Time `json:"ended"`
State string `json:"state"`
}
if json.Unmarshal(value, &a) != nil {
return false
}
return a.Ended.IsZero() && a.State == ""
}
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
errs := make(chan error, 1)
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
if err != nil {
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
return errs, nil
}
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
w, err := settle(ctx, kv, each)
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
}
if gate.Settled() {
changed(ctx)
}
go func() {
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
changed(ctx)
}
})
}()
return errs, nil
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go
+112
View File
@@ -0,0 +1,112 @@
package main
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
import (
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
)
func gateReport(t *testing.T, node, version string) (string, []byte) {
t.Helper()
key, err := delivery.ToolchainKey(node, "typescript")
if err != nil {
t.Fatal(err)
}
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
if err != nil {
t.Fatal(err)
}
return key, value
}
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
return tc.Version, none
}
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
set := &buildrun.ToolchainSet{}
var said []string
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
key, value := gateReport(t, node, version)
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
}
for _, node := range []string{"g14", "novox", "shanks"} {
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
}
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
if !gate.Settled() {
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
}
if len(said) != 1 {
t.Fatalf("ace's report is said left out once; said %v", said)
}
// ace reports again while unassigned (an old value seen again): still left out.
key, value := gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an unassigned node's report changed what was passed on")
}
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
t.Fatal("assigning ace again passed nothing on")
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
t.Fatalf("builders that run different versions still disagree; got %q", none)
}
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
t.Fatal("a pending assignment is no assignment on the machine")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
gate.Assignment("ace.mesh-build-node", nil, true)
if v, _ := agreedOf(set); v != "v26.12.1" {
t.Fatalf("got %q", v)
}
// A report deleted from the bucket goes whether its node is assigned or not.
key, _ = gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
t.Fatal("deleting g14's report passed nothing on")
}
if n := len(set.Entries("typescript")); n != 2 {
t.Fatalf("novox and shanks are left; got %d entries", n)
}
}
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
set := &buildrun.ToolchainSet{}
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
gate.Settled()
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
t.Fatal("an assignment with no report passed something on")
}
key, value := gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an assigned node's report was not passed on")
}
key, value = gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go
+61
View File
@@ -0,0 +1,61 @@
package main
import (
"context"
"encoding/json"
"fmt"
"sort"
"strings"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-builder/internal/build"
)
// eventsStream is the stream every module's and role's events are on, where the controller makes the builder's durable
// consumer of artifacts-gone (mesh-controller internal/broker EventsStream).
const eventsStream = "EVENTS"
// toolCaller calls a tool or a seat verb as the builder's own account (mesh-sdk process.Conn.Call).
type toolCaller func(key string, body any) (json.RawMessage, error)
// storedVia is build.Builder's Stored: it asks the controller's verb held (mesh-sdk delivery.HeldVerb; novox/hq issue
// 1937) and answers the references the store no longer holds. One it could not ask about, with none gone, is an error:
// the key is then left done, and the controller's take-in still refuses a gone blob.
func storedVia(call toolCaller) func(ctx context.Context, references []string) ([]string, error) {
return func(_ context.Context, references []string) ([]string, error) {
raw, err := call(delivery.HeldVerb, map[string]any{"references": references})
if err != nil {
return nil, fmt.Errorf("asking %s: %w", delivery.HeldVerb, err)
}
a, err := delivery.DecodeHeldAnswer(raw, references)
if err != nil {
return nil, err
}
if len(a.Gone) == 0 && len(a.Unknown) > 0 {
var why []string
for r, w := range a.Unknown {
why = append(why, r+": "+w)
}
sort.Strings(why)
return nil, fmt.Errorf("the artifact store could not be asked about %s", strings.Join(why, "; "))
}
return a.Gone, nil
}
}
// heardGone takes one artifacts-gone event in: acknowledged, then handed to the builder, which asks the store and builds
// the key again. A body that is not one is said and acknowledged, so it is not handed over again.
func heardGone(ctx context.Context, b *build.Builder, m jetstream.Msg, logf func(string, ...any)) {
e, err := delivery.DecodeArtifactsGone(m.Data())
if ackErr := m.Ack(); ackErr != nil {
logf("acknowledging %s: %v", m.Subject(), ackErr)
}
if err != nil {
logf("%s holds what is not %s, left: %v", m.Subject(), delivery.ArtifactsGoneEvent, err)
return
}
b.ArtifactsGone(ctx, e)
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go
+8 -6
View File
@@ -23,9 +23,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO"
// §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key
// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched.
type toolchainWatch struct {
ctx context.Context
js bucketOpener
set *buildrun.ToolchainSet
ctx context.Context
js bucketOpener
set *buildrun.ToolchainSet
// put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940).
put func(module, key string, value []byte, deleted bool) bool
changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again
logger *log.Logger
errs chan error
@@ -34,7 +36,7 @@ type toolchainWatch struct {
}
func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch {
return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
}
// start watches the buckets that exist now; a missing one is said and waits for its build module's event.
@@ -62,14 +64,14 @@ func (w *toolchainWatch) ensure(module string) error {
case err != nil:
return fmt.Errorf("the bucket %s: %w", name, err)
}
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) })
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
if err != nil {
return err
}
w.watched[module] = true
go func() {
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) {
if w.put(module, e.Key(), e.Value(), !opOf(e)) {
w.changed(w.ctx)
}
})
modules/mesh-delivery-builder/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-builder
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-builder/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-builder/internal/build/claim.go
+54
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"slices"
"sort"
"strings"
@@ -37,6 +38,11 @@ type Claim struct {
Key string
Rev uint64
Build delivery.BuilderBuild
// Gone are the references of a done key the artifact store no longer holds, for which it was claimed again
// (ClaimOptions.Stored; issue 1937): said in the log and the key's step.
Gone []string
// Unasked is why the store could not be asked about a done key taken as it stands to publish: said in the log.
Unasked error
}
// ClaimOptions say what the builder knows about the moment of a claim.
@@ -55,6 +61,30 @@ type ClaimOptions struct {
// a package's reference. A key queued waiting for a version (BuilderBuild.Waits) is taken only once it is (ADR 0500
// clause 6). Nil means none is.
Published func(ref string) bool
// Stored answers which of the references given the artifact store no longer holds (the controller's verb held,
// mesh-sdk delivery.HeldVerb; novox/hq issue 1937). A done key of the trunk's fingerprint is asked before it is left
// done: the store's nightly collector deletes a blob nothing holds, and a key the controller never took in holds
// nothing, so a key whose blob is gone is claimed building as a new revision and built again. Nil asks nothing.
Stored func(ctx context.Context, references []string) (gone []string, err error)
}
// StoredReferences are the references of a done build's bundles and archives that name a blob in the artifact store
// (`artifact-store://<module>/<artifact>/blobs/<digest>`): what the store's collector can delete. An image or an upstream
// copy is a manifest, which the collector keeps, and a package lives in the package registry.
func StoredReferences(b delivery.BuilderBuild) []string {
var out []string
for _, a := range b.Artifacts {
if strings.HasPrefix(a.Reference, "artifact-store://") && strings.Contains(a.Reference, "/blobs/") &&
!slices.Contains(out, a.Reference) {
out = append(out, a.Reference)
}
}
return out
}
// GoneStep is the step a key claimed again for gone blobs says (issue 1937).
func GoneStep(gone []string) string {
return "claimed again: the artifact store no longer holds " + strings.Join(gone, ", ")
}
// Unpublished is whether a done build holds an image built without a push or a package built without a publish: it has
@@ -118,6 +148,8 @@ func Unreferenced(b delivery.BuilderBuild) bool {
// names it (ClaimOptions.Trunk): that is taken as it stands, with the done value, and its builder writes the new
// revision when the publishing build ends (ADR 0498 clause 6). From R4e one node builds each fingerprint (ADR 0478
// clause 4), so a key built on one node is never built again for a second;
// - a done key of the trunk's fingerprint whose bundle or archive blob the artifact store no longer holds
// (ClaimOptions.Stored, issue 1937) is claimed building as a new revision and built whole, its step saying which;
// - a failed key is taken again as a new revision of the same key (ADR 0429 clause 18), except one failed for a kind no
// builder builds, for an upstream pinned by no digest or a package version taken (delivery.FailedUntilChanged: ADR 0478
// clause 2, ADR 0498 clauses 4 and 5) or for digests that differ (buildrun.DigestsDifferPrefix, clause 4),
@@ -149,6 +181,28 @@ func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string
if Unreferenced(cur) {
return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
}
if opt.Trunk && opt.Stored != nil {
if refs := StoredReferences(cur); len(refs) > 0 {
gone, err := opt.Stored(ctx, refs)
if err != nil {
err = fmt.Errorf("%s: whether the artifact store still holds %s cannot be asked, so the key is "+
"left done (the controller's take-in refuses it if one is gone): %w", key, strings.Join(refs, ", "), err)
if len(publishable(cur, opt.ImageBuilder)) > 0 {
return &Claim{Key: key, Rev: rev, Build: cur, Unasked: err}, nil
}
return nil, err
}
if len(gone) > 0 {
again := building
again.Step = GoneStep(gone)
cl, err := c.write(ctx, key, again, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
if cl != nil {
cl.Gone = gone
}
return cl, err
}
}
}
if opt.Trunk && len(publishable(cur, opt.ImageBuilder)) > 0 {
return &Claim{Key: key, Rev: rev, Build: cur}, nil
}
modules/mesh-delivery-builder/internal/build/gone_test.go
+204
View File
@@ -0,0 +1,204 @@
package build
import (
"context"
"errors"
"fmt"
"slices"
"strings"
"sync"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A done key whose blob the artifact store no longer holds is built again (novox/hq issue 1937): on 2026-10-24 the
// store's 03:30 collection deleted the bare blobs of node-tools', mesh-host's and the forge bridge's done keys, which the
// controller had never taken in, and the builder kept them done.
// fakeHeld answers the controller's verb held: the references in gone are gone.
type fakeHeld struct {
mu sync.Mutex
gone map[string]bool
err error
asked [][]string
}
func (f *fakeHeld) stored(_ context.Context, refs []string) ([]string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.asked = append(f.asked, slices.Clone(refs))
if f.err != nil {
return nil, f.err
}
var out []string
for _, r := range refs {
if f.gone[r] {
out = append(out, r)
}
}
return out, nil
}
func (f *fakeHeld) asks() int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.asked)
}
// logged keeps what the builder says.
type logged struct {
mu sync.Mutex
lines []string
}
func (l *logged) logf(format string, args ...any) {
l.mu.Lock()
defer l.mu.Unlock()
l.lines = append(l.lines, fmt.Sprintf(format, args...))
}
func (l *logged) has(words ...string) bool {
l.mu.Lock()
defer l.mu.Unlock()
for _, line := range l.lines {
all := true
for _, w := range words {
all = all && strings.Contains(line, w)
}
if all {
return true
}
}
return false
}
const goneRef = "artifact-store://m/tool/blobs/" + digest
func doneKeyNaming(ref string) delivery.BuilderBuild {
return delivery.BuilderBuild{State: delivery.BuildDone, By: "shanks.builder", Manifest: manifestD, Step: "built mesh-build-go on shanks",
Artifacts: []delivery.BuiltArtifact{{Name: "tool", Kind: "bundle", Digest: digest, Reference: ref}}}
}
// 1937/AC3 (unit, the claim): a done key of the trunk's fingerprint whose blob is gone is claimed building as a new
// revision, its step naming the gone reference; a candidate's done key is not asked about.
func TestADoneTrunkKeyWhoseBlobIsGoneIsClaimedAgain(t *testing.T) {
st := newMem()
c := Claimer{Store: st, By: "novox.builder"}
key, _ := delivery.BuilderBuildKey("m", fp1)
rev := putBuild(t, st, key, doneKeyNaming(goneRef))
held := &fakeHeld{gone: map[string]bool{goneRef: true}}
cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Stored: held.stored})
if err != nil || cl != nil || held.asks() != 0 {
t.Fatalf("a candidate's done key: claim %+v, %v, %d asks; want nothing and no ask", cl, err, held.asks())
}
cl, err = c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored})
if err != nil || cl == nil {
t.Fatalf("claim = %+v, %v; want the key claimed again", cl, err)
}
if cl.Rev <= rev || cl.Build.State != delivery.BuildBuilding || !slices.Equal(cl.Gone, []string{goneRef}) {
t.Errorf("claim = %+v (revision was %d); want building as a new revision naming the gone reference", cl, rev)
}
v := valueOf(t, st, key)
if v.State != delivery.BuildBuilding || !strings.Contains(v.Step, "no longer holds") || !strings.Contains(v.Step, goneRef) {
t.Errorf("the key holds %+v; want building, its step naming the gone reference", v)
}
}
// 1937/AC5 (unit, the claim): a done trunk key whose blobs are all held stays done; a store that cannot be asked leaves
// it done and says why; images and packages are not asked about.
func TestADoneTrunkKeyStaysDoneWhileTheStoreHoldsItsBlobs(t *testing.T) {
st := newMem()
c := Claimer{Store: st, By: "novox.builder"}
key, _ := delivery.BuilderBuildKey("m", fp1)
b := doneKeyNaming(goneRef)
b.Artifacts = append(b.Artifacts, delivery.BuiltArtifact{Name: "img", Kind: "image", Digest: digest, Reference: "registry.example/m/img@" + digest})
rev := putBuild(t, st, key, b)
held := &fakeHeld{gone: map[string]bool{}}
if cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored}); err != nil || cl != nil {
t.Fatalf("claim = %+v, %v; want nothing: the store holds the blob", cl, err)
}
if len(held.asked) != 1 || !slices.Equal(held.asked[0], []string{goneRef}) {
t.Errorf("asked %v; want the bundle's reference alone", held.asked)
}
held.err = errors.New("the controller did not answer")
cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored})
if cl != nil || err == nil || !strings.Contains(err.Error(), "cannot be asked") || !strings.Contains(err.Error(), "left done") {
t.Errorf("claim = %+v, %v; want the key left done and said", cl, err)
}
if st.revs[key] != rev {
t.Errorf("the key moved to revision %d (was %d)", st.revs[key], rev)
}
}
// 1937/AC3 (unit, the builder): the builder's look at a module whose done key's blob is gone builds it again, says so
// in a log line and in the done key's step, and asks the store once per process while it answers held.
func TestTheBuilderBuildsAgainADoneKeyWhoseBlobIsGone(t *testing.T) {
r := newRig(t, goBundle)
log := &logged{}
r.b.Logf = log.logf
held := &fakeHeld{gone: map[string]bool{goneRef: true}}
r.b.Stored = held.stored
rev := putBuild(t, r.st, r.key, doneKeyNaming(goneRef))
v := r.consider(t, "modules/m")
if v.State != delivery.BuildDone || r.st.revs[r.key] <= rev || len(r.bm.callsOf("build_go_build")) == 0 {
t.Fatalf("build = %+v at revision %d (was %d); want built again", v, r.st.revs[r.key], rev)
}
if !strings.Contains(v.Step, "built again: the artifact store no longer held "+goneRef) {
t.Errorf("the done key's step %q does not say it was built again for the gone blob", v.Step)
}
if !log.has(r.key, "built again", goneRef) {
t.Errorf("no log line says the key was built again: %v", log.lines)
}
// The new revision's blob is held: asked once, then not again in this process.
held.mu.Lock()
held.gone = map[string]bool{}
held.mu.Unlock()
rev = r.st.revs[r.key]
asks := held.asks()
r.consider(t, "modules/m")
r.consider(t, "modules/m")
if r.st.revs[r.key] != rev {
t.Errorf("a key whose blob is held was built again")
}
if held.asks() != asks+1 {
t.Errorf("asked the store %d times for two looks at a held key; want once", held.asks()-asks)
}
}
// 1937/AC4 (unit, the builder): artifacts-gone for the module's fingerprint makes the builder ask again, even for a key
// it found held before, and build it again; one for another fingerprint builds nothing.
func TestArtifactsGoneMakesTheBuilderAskAndBuildAgain(t *testing.T) {
r := newRig(t, goBundle)
log := &logged{}
r.b.Logf = log.logf
held := &fakeHeld{gone: map[string]bool{}}
r.b.Stored = held.stored
putBuild(t, r.st, r.key, doneKeyNaming(goneRef))
if v := r.consider(t, "modules/m"); v.State != delivery.BuildDone || len(r.bm.callsOf("build_go_build")) != 0 {
t.Fatalf("a key whose blob is held was built: %+v", v)
}
rev := r.st.revs[r.key]
held.mu.Lock()
held.gone[goneRef] = true // collected after the builder's look; the controller's take-in found it
held.mu.Unlock()
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
Fingerprint: "src2:" + strings.Repeat("9", 64), Gone: []string{goneRef}})
r.b.Wait()
if r.st.revs[r.key] != rev || !log.has("a fingerprint the module does not have now") {
t.Fatalf("an event for another fingerprint moved the key (revision %d, was %d): %v", r.st.revs[r.key], rev, log.lines)
}
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
Key: r.key, Fingerprint: fp1, Gone: []string{goneRef}})
r.b.Wait()
v := valueOf(t, r.st, r.key)
if v.State != delivery.BuildDone || r.st.revs[r.key] <= rev || len(r.bm.callsOf("build_go_build")) == 0 {
t.Fatalf("build = %+v at revision %d (was %d); want built again on the event", v, r.st.revs[r.key], rev)
}
if !log.has("heard artifacts-gone") || !log.has(r.key, "built again", goneRef) {
t.Errorf("the builder did not say what it heard and did: %v", log.lines)
}
}
modules/mesh-delivery-builder/internal/build/run.go
+68 -2
View File
@@ -50,6 +50,9 @@ type Builder struct {
// Answered is told the build module whose step answered a reference (issue 1886): what the builder says it writes
// rests on it. Nil tells nobody.
Answered func(buildModule string)
// Stored answers which references the artifact store no longer holds (the controller's verb held; issue 1937): a done
// key of the trunk's fingerprint is asked before it is left done (ClaimOptions.Stored). Nil asks nothing.
Stored func(ctx context.Context, references []string) (gone []string, err error)
once sync.Once
sem chan struct{}
@@ -66,6 +69,9 @@ type Builder struct {
// published are the package versions `<name>@<version>` a done key of this bucket names as a package's reference: a key
// waiting for one is built once it is here (ADR 0500 clause 6).
published map[string]bool
// held are the done keys' references the store answered held, by `<key> <references>`, asked once per process: a
// blob collected later is found by the take-in, whose artifacts-gone forgets the key here (issue 1937).
held map[string]bool
}
type planned struct {
@@ -82,7 +88,7 @@ func (b *Builder) init() {
b.sem = make(chan struct{}, limit)
b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{}
b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{}
b.published = map[string]bool{}
b.published, b.held = map[string]bool{}, map[string]bool{}
})
}
@@ -202,6 +208,55 @@ func (b *Builder) Forget(key string) {
b.mu.Unlock()
}
// storedOnce is Stored for one key, asked once per process while the store answers every reference held.
func (b *Builder) storedOnce(key string) func(ctx context.Context, references []string) ([]string, error) {
if b.Stored == nil {
return nil
}
return func(ctx context.Context, references []string) ([]string, error) {
id := key + " " + strings.Join(references, " ")
b.mu.Lock()
known := b.held[id]
b.mu.Unlock()
if known {
return nil, nil
}
gone, err := b.Stored(ctx, references)
if err == nil && len(gone) == 0 {
b.mu.Lock()
b.held[id] = true
b.mu.Unlock()
}
return gone, err
}
}
// ArtifactsGone takes the controller's event artifacts-gone in (issue 1937): a take-in found that the store does not
// hold what the module's key names. The module is looked at again, which asks the store (ClaimOptions.Stored) and
// builds the key again; an event for a fingerprint the module no longer has is said and left. The event only wakes the
// builder: what is built rests on the store's answer, never on the event alone.
func (b *Builder) ArtifactsGone(ctx context.Context, e delivery.ArtifactsGone) {
b.init()
b.mu.Lock()
m, ok := b.known[e.Module]
if key, err := delivery.BuilderBuildKey(e.Module, e.Fingerprint); err == nil {
for id := range b.held {
if strings.HasPrefix(id, key+" ") {
delete(b.held, id)
}
}
}
b.mu.Unlock()
if !ok || m.FP2 != e.Fingerprint {
b.logf("%s: heard %s for %s, a fingerprint the module does not have now: nothing to build", e.Module,
delivery.ArtifactsGoneEvent, e.Fingerprint)
return
}
b.logf("%s: heard %s (%s): asking the artifact store and building the key again", e.Module,
delivery.ArtifactsGoneEvent, strings.Join(e.Gone, ", "))
b.consider(ctx, e.Module, m)
}
// Reconsider looks at every module again: the toolchains changed, so a build that failed for them may build now.
func (b *Builder) Reconsider(ctx context.Context) {
b.init()
@@ -312,7 +367,8 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
first := !b.tried[key]
b.tried[key] = true
b.mu.Unlock()
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished})
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
Stored: b.storedOnce(key)})
if err != nil {
b.logf("%s: %v", module, err)
return
@@ -320,6 +376,13 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
if cl == nil {
return
}
if cl.Unasked != nil {
b.logf("%v", cl.Unasked)
}
if len(cl.Gone) > 0 {
// Said aloud (issue 1937): the key was done, and the store's collector deleted what it names.
b.logf("%s: built again: the artifact store no longer holds %s", cl.Key, strings.Join(cl.Gone, ", "))
}
// A done key taken again builds only what is missing: the image or the package built unpublished as a candidate's; the
// other artifacts stay as the done key has them (ADR 0487 clauses 5 and 8, ADR 0498 clause 6).
var prior *delivery.BuilderBuild
@@ -350,6 +413,9 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
return
}
result.By, result.Manifest = b.Claimer.By, m.Manifest
if len(cl.Gone) > 0 && result.State == delivery.BuildDone {
result.Step += " (built again: the artifact store no longer held " + strings.Join(cl.Gone, ", ") + ")"
}
if publishing && result.State == delivery.BuildFailed {
// The trunk's image or package was built only as a candidate's and its push or publish failed: the key says so, so the builds view, the
// checker and the failed-key retake see it, and a look does not build the image again for nothing.
modules/mesh-delivery-builder/manifest_test.go
+9 -6
View File
@@ -89,18 +89,21 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
}
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
}
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
t.Errorf("uses = %v, want the three build seats", m.Uses)
}
if len(m.Invokes) != 0 || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want none: no tool call from the builder and no capacity", m.Invokes, m.Config)
// 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no
// other tool; it holds no capacity.
if want := []string{"seat:mesh-controller.held"}; !reflect.DeepEqual(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.
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), 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 controller's artifacts-gone, a take-in refused for a gone blob (1937/AC4).
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-controller.artifacts-gone"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -109,7 +112,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if m.SecretsOwner != "mesh-delivery-builder" {
t.Errorf("secrets-owner = %q", m.SecretsOwner)
}
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "invokes", "config"} {
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "config"} {
if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the builder has none", k)
}
modules/mesh-delivery-builder/module.json
+6 -1
View File
@@ -4,7 +4,11 @@
"slug": "builder",
"consumes": [
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported"
"mesh-build-node.toolchain.reported",
"mesh-controller.artifacts-gone"
],
"invokes": [
"seat:mesh-controller.held"
],
"state": [
{ "name": "builds", "history": 5 },
@@ -16,6 +20,7 @@
"mesh-build-image.steps",
"mesh-build-node.steps",
"mesh-build-node.toolchains",
"mesh-controller.assignments",
"mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules"
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+10
View File
@@ -257,6 +257,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The toolchains first, so a check that starts from the first candidate finds the nodes.
watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger)
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
// assignments are settled first, so the toolchains settle through the gate already knowing them.
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
watch.put = gate.Toolchain
assignErrs, err := watchAssignments(ctx, js, gate, ch.NodesChanged, logger)
if err != nil {
return err
}
if err := watch.start(); err != nil {
return err
}
@@ -369,6 +377,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
case err := <-errs:
return err
case err := <-assignErrs:
return err
case err := <-watch.errs:
return err
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go
+174
View File
@@ -0,0 +1,174 @@
package main
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
// toolchain agreement this file repairs.
import (
"context"
"encoding/json"
"fmt"
"log"
"maps"
"slices"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provider"
)
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
//
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
// refusal is the guard of novox/hq ADR 0465, and it stays.
type toolchainGate struct {
mu sync.Mutex
modules map[string]bool // the build modules whose toolchains pass the gate
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
logf func(format string, a ...any)
known bool // the assignments are settled
standing map[string]bool // <node>.<module> with an open assignment
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
left map[string]bool // <module>/<key> said as left out, so it is said once
}
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
for _, m := range modules {
g.modules[m] = true
}
return g
}
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
g.mu.Lock()
defer g.mu.Unlock()
if g.raw[module] == nil {
g.raw[module] = map[string][]byte{}
}
if deleted {
delete(g.raw[module], key)
} else {
g.raw[module][key] = value
}
return g.pass(module, key)
}
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
node, module, ok := strings.Cut(key, ".")
if !ok || !g.modules[module] {
return false
}
open := !deleted && openAssignment(value)
g.mu.Lock()
defer g.mu.Unlock()
if g.standing[key] == open {
return false
}
if open {
g.standing[key] = true
} else {
delete(g.standing, key)
}
if !g.known {
return false
}
changed := false
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
changed = g.pass(module, k) || changed
}
}
return changed
}
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
// passed on changed.
func (g *toolchainGate) Settled() bool {
g.mu.Lock()
defer g.mu.Unlock()
g.known = true
changed := false
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
changed = g.pass(module, k) || changed
}
}
return changed
}
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
// otherwise. The caller holds mu.
func (g *toolchainGate) pass(module, key string) bool {
value, held := g.raw[module][key]
if !held {
return g.forward(module, key, nil, true)
}
node, _, err := delivery.ParseToolchainKey(key)
if err != nil || !g.known || g.standing[node+"."+module] {
delete(g.left, module+"/"+key)
return g.forward(module, key, value, false)
}
if !g.left[module+"/"+key] {
g.left[module+"/"+key] = true
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
delivery.ToolchainsBucket(module), key, module, node)
}
return g.forward(module, key, nil, true)
}
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
func openAssignment(value []byte) bool {
var a struct {
Ended time.Time `json:"ended"`
State string `json:"state"`
}
if json.Unmarshal(value, &a) != nil {
return false
}
return a.Ended.IsZero() && a.State == ""
}
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
errs := make(chan error, 1)
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
if err != nil {
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
return errs, nil
}
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
w, err := settle(ctx, kv, each)
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
}
if gate.Settled() {
changed(ctx)
}
go func() {
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
changed(ctx)
}
})
}()
return errs, nil
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go
+112
View File
@@ -0,0 +1,112 @@
package main
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
import (
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
)
func gateReport(t *testing.T, node, version string) (string, []byte) {
t.Helper()
key, err := delivery.ToolchainKey(node, "typescript")
if err != nil {
t.Fatal(err)
}
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
if err != nil {
t.Fatal(err)
}
return key, value
}
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
return tc.Version, none
}
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
set := &buildrun.ToolchainSet{}
var said []string
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
key, value := gateReport(t, node, version)
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
}
for _, node := range []string{"g14", "novox", "shanks"} {
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
}
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
if !gate.Settled() {
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
}
if len(said) != 1 {
t.Fatalf("ace's report is said left out once; said %v", said)
}
// ace reports again while unassigned (an old value seen again): still left out.
key, value := gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an unassigned node's report changed what was passed on")
}
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
t.Fatal("assigning ace again passed nothing on")
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
t.Fatalf("builders that run different versions still disagree; got %q", none)
}
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
t.Fatal("a pending assignment is no assignment on the machine")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
gate.Assignment("ace.mesh-build-node", nil, true)
if v, _ := agreedOf(set); v != "v26.12.1" {
t.Fatalf("got %q", v)
}
// A report deleted from the bucket goes whether its node is assigned or not.
key, _ = gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
t.Fatal("deleting g14's report passed nothing on")
}
if n := len(set.Entries("typescript")); n != 2 {
t.Fatalf("novox and shanks are left; got %d entries", n)
}
}
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
set := &buildrun.ToolchainSet{}
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
gate.Settled()
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
t.Fatal("an assignment with no report passed something on")
}
key, value := gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an assigned node's report was not passed on")
}
key, value = gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go
+8 -6
View File
@@ -24,9 +24,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO"
// §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key
// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched.
type toolchainWatch struct {
ctx context.Context
js bucketOpener
set *buildrun.ToolchainSet
ctx context.Context
js bucketOpener
set *buildrun.ToolchainSet
// put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940).
put func(module, key string, value []byte, deleted bool) bool
changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again
logger *log.Logger
errs chan error
@@ -35,7 +37,7 @@ type toolchainWatch struct {
}
func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch {
return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
}
// start watches the buckets that exist now; a missing one is said and waits for its build module's event.
@@ -63,14 +65,14 @@ func (w *toolchainWatch) ensure(module string) error {
case err != nil:
return fmt.Errorf("the bucket %s: %w", name, err)
}
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) })
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
if err != nil {
return err
}
w.watched[module] = true
go func() {
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) {
if w.put(module, e.Key(), e.Value(), !opOf(e)) {
w.changed(w.ctx)
}
})
modules/mesh-delivery-checker/manifest_test.go
+1 -1
View File
@@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
t.Errorf("state = %+v, want %+v", m.State, want)
}
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
"mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
modules/mesh-delivery-checker/module.json
+1
View File
@@ -21,6 +21,7 @@
"mesh-build-go.toolchains",
"mesh-build-node.steps",
"mesh-build-node.toolchains",
"mesh-controller.assignments",
"mesh-delivery-assigner.compositions",
"mesh-delivery-builder.builds",
"mesh-delivery-gate.source",
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+85
View File
@@ -549,3 +549,88 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC
t.Errorf("avahi = fp2 %q none2 %q within 5 s of the start, want a fingerprint and no reason", got.FP2, got.None2)
}
}
// TestIndexerCountsOnlyTheAssignedBuildNodesToolchains (issue 1940): the toolchains bucket holds the report of n2, where
// mesh-build-go was unassigned (its assignment ended) at another version than n1's, as ace's v26.12.0 stayed beside the
// v26.12.1 of the nodes mesh-build-node runs on. The indexer gives the merged module a fingerprint from n1's toolchain
// alone; when the controller assigns mesh-build-go to n2 again, its report counts and the builders disagree (the guard
// stays); when the assignment ends again, the fingerprint comes back. No restart, no timer: the assignments' watch.
func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) {
manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
if err != nil {
t.Fatal(err)
}
const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"
var assignments jetstream.KeyValue
js, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
if err != nil {
t.Fatal(err)
}
for node, version := range map[string]string{"n1": "go1.27.3", "n2": "go1.27.2"} {
key, _ := delivery.ToolchainKey(node, "go")
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: version, Recipe: "1", At: time.Now()})
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put(ctx, key, value); err != nil {
t.Fatal(err)
}
}
// The controller's assignments mirror, in its layout (mesh-controller internal/chain Assignment).
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_assignments", History: 1}); err != nil {
t.Fatal(err)
}
for key, value := range map[string]string{
"n1.mesh-build-go": `{"node":"n1","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z"}`,
"n2.mesh-build-go": `{"node":"n2","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z","ended":"2026-10-23T08:00:00Z"}`,
} {
if _, err := assignments.Put(ctx, key, []byte(value)); err != nil {
t.Fatal(err)
}
}
})
data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 302,
Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0",
Trees: map[string]string{"modules/avahi": "newtree"},
Manifests: map[string]string{"modules/avahi": string(manifest)},
Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}})
if err != nil {
t.Fatal(err)
}
if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.merged", data); err != nil {
t.Fatal(err)
}
read := func(ok func(delivery.Module) bool) delivery.Module {
var got delivery.Module
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := modules.Get(ctx, "avahi"); err == nil {
if m, err := delivery.DecodeModule(e.Value()); err == nil {
if got = m; ok(m) {
break
}
}
}
}
return got
}
agreed := func(m delivery.Module) bool { return m.FP2 != "" && m.Toolchains["go"].Version == "go1.27.3" }
if m := read(agreed); !agreed(m) {
t.Fatalf("within 5 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", m.FP2, m.None2, m.Toolchains)
}
if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","assigned":"2026-10-24T08:00:00Z"}`)); err != nil {
t.Fatal(err)
}
disagree := func(m delivery.Module) bool {
return m.FP2 == "" && strings.HasPrefix(m.None2, "builders disagree: go1.27.2 recipe 1 on n2")
}
if m := read(disagree); !disagree(m) {
t.Fatalf("within 5 s of n2's assignment avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2)
}
if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","ended":"2026-10-24T09:00:00Z"}`)); err != nil {
t.Fatal(err)
}
if m := read(agreed); !agreed(m) {
t.Fatalf("within 5 s of n2's unassignment avahi has fp2 %q, none2 %q", m.FP2, m.None2)
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+22 -1
View File
@@ -313,7 +313,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
go func() { sourceErrs <- follow(ctx, sw, each) }()
}
tools := newToolchainWatch(ctx, js, toolchainSink{put: ix.PutToolchain, apply: ix.ApplyToolchain, recompute: ix.Recompute}, logger)
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
// assignments are settled first, so the toolchains settle through the gate already knowing them.
gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf)
assignErrs, err := watchAssignments(ctx, js, gate, func(ctx context.Context) {
if err := ix.Recompute(ctx); err != nil {
logger.Printf("writing the modules again after an assignment of a build module changed: %v", err)
}
}, logger)
if err != nil {
return err
}
tools := newToolchainWatch(ctx, js, toolchainSink{
put: gate.Toolchain,
apply: func(ctx context.Context, module, key string, value []byte, deleted bool) error {
if !gate.Toolchain(module, key, value, deleted) {
return nil
}
return ix.Recompute(ctx)
},
recompute: ix.Recompute}, logger)
if err := tools.start(); err != nil {
return err
}
@@ -377,6 +396,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
case err := <-sourceErrs:
return err
case err := <-assignErrs:
return err
case err := <-tools.errs:
return err
case err := <-previews.errs:
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go
+174
View File
@@ -0,0 +1,174 @@
package main
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
// toolchain agreement this file repairs.
import (
"context"
"encoding/json"
"fmt"
"log"
"maps"
"slices"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provider"
)
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
//
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
// refusal is the guard of novox/hq ADR 0465, and it stays.
type toolchainGate struct {
mu sync.Mutex
modules map[string]bool // the build modules whose toolchains pass the gate
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
logf func(format string, a ...any)
known bool // the assignments are settled
standing map[string]bool // <node>.<module> with an open assignment
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
left map[string]bool // <module>/<key> said as left out, so it is said once
}
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
for _, m := range modules {
g.modules[m] = true
}
return g
}
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
g.mu.Lock()
defer g.mu.Unlock()
if g.raw[module] == nil {
g.raw[module] = map[string][]byte{}
}
if deleted {
delete(g.raw[module], key)
} else {
g.raw[module][key] = value
}
return g.pass(module, key)
}
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
node, module, ok := strings.Cut(key, ".")
if !ok || !g.modules[module] {
return false
}
open := !deleted && openAssignment(value)
g.mu.Lock()
defer g.mu.Unlock()
if g.standing[key] == open {
return false
}
if open {
g.standing[key] = true
} else {
delete(g.standing, key)
}
if !g.known {
return false
}
changed := false
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
changed = g.pass(module, k) || changed
}
}
return changed
}
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
// passed on changed.
func (g *toolchainGate) Settled() bool {
g.mu.Lock()
defer g.mu.Unlock()
g.known = true
changed := false
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
changed = g.pass(module, k) || changed
}
}
return changed
}
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
// otherwise. The caller holds mu.
func (g *toolchainGate) pass(module, key string) bool {
value, held := g.raw[module][key]
if !held {
return g.forward(module, key, nil, true)
}
node, _, err := delivery.ParseToolchainKey(key)
if err != nil || !g.known || g.standing[node+"."+module] {
delete(g.left, module+"/"+key)
return g.forward(module, key, value, false)
}
if !g.left[module+"/"+key] {
g.left[module+"/"+key] = true
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
delivery.ToolchainsBucket(module), key, module, node)
}
return g.forward(module, key, nil, true)
}
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
func openAssignment(value []byte) bool {
var a struct {
Ended time.Time `json:"ended"`
State string `json:"state"`
}
if json.Unmarshal(value, &a) != nil {
return false
}
return a.Ended.IsZero() && a.State == ""
}
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
errs := make(chan error, 1)
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
if err != nil {
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
return errs, nil
}
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
w, err := settle(ctx, kv, each)
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
}
if gate.Settled() {
changed(ctx)
}
go func() {
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
changed(ctx)
}
})
}()
return errs, nil
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go
+112
View File
@@ -0,0 +1,112 @@
package main
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
import (
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
)
func gateReport(t *testing.T, node, version string) (string, []byte) {
t.Helper()
key, err := delivery.ToolchainKey(node, "typescript")
if err != nil {
t.Fatal(err)
}
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
if err != nil {
t.Fatal(err)
}
return key, value
}
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
return tc.Version, none
}
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
set := &buildrun.ToolchainSet{}
var said []string
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
key, value := gateReport(t, node, version)
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
}
for _, node := range []string{"g14", "novox", "shanks"} {
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
}
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
if !gate.Settled() {
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
}
if len(said) != 1 {
t.Fatalf("ace's report is said left out once; said %v", said)
}
// ace reports again while unassigned (an old value seen again): still left out.
key, value := gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an unassigned node's report changed what was passed on")
}
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
t.Fatal("assigning ace again passed nothing on")
}
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
t.Fatalf("builders that run different versions still disagree; got %q", none)
}
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
t.Fatal("a pending assignment is no assignment on the machine")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
gate.Assignment("ace.mesh-build-node", nil, true)
if v, _ := agreedOf(set); v != "v26.12.1" {
t.Fatalf("got %q", v)
}
// A report deleted from the bucket goes whether its node is assigned or not.
key, _ = gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
t.Fatal("deleting g14's report passed nothing on")
}
if n := len(set.Entries("typescript")); n != 2 {
t.Fatalf("novox and shanks are left; got %d entries", n)
}
}
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
set := &buildrun.ToolchainSet{}
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
gate.Settled()
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
t.Fatal("an assignment with no report passed something on")
}
key, value := gateReport(t, "g14", "v26.12.1")
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("an assigned node's report was not passed on")
}
key, value = gateReport(t, "ace", "v26.12.0")
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
}
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
t.Fatalf("got %q, %q", v, none)
}
}
modules/mesh-delivery-indexer/manifest_test.go
+1 -1
View File
@@ -81,7 +81,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) {
t.Errorf("state = %+v, want %+v", m.State, want)
}
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
}
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
modules/mesh-delivery-indexer/module.json
+1
View File
@@ -16,6 +16,7 @@
"reads": [
"mesh-build-go.toolchains",
"mesh-build-node.toolchains",
"mesh-controller.assignments",
"mesh-controller.builds",
"mesh-delivery-checker.previews",
"mesh-delivery-gate.source"