diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go new file mode 100644 index 0000000..4ccc867 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go @@ -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.. 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")) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 6e9d5db..b146faa 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing.go @@ -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 . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; 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 // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / 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 +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/standing_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go new file mode 100644 index 0000000..3f2bbef --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go @@ -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) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go index 95b6fc5..69faff2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -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) } }) diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index d5b8384..615cce0 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -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 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 2a3e9f3..d648ca0 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -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= diff --git a/modules/mesh-delivery-builder/internal/build/claim.go b/modules/mesh-delivery-builder/internal/build/claim.go index 21b1316..406a62c 100644 --- a/modules/mesh-delivery-builder/internal/build/claim.go +++ b/modules/mesh-delivery-builder/internal/build/claim.go @@ -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:////blobs/`): 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 } diff --git a/modules/mesh-delivery-builder/internal/build/gone_test.go b/modules/mesh-delivery-builder/internal/build/gone_test.go new file mode 100644 index 0000000..499cdbc --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/gone_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index f4f2035..1d140d7 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -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 `@` 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 ` `, 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. diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 930b0d8..e5dd046 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 798cb79..0333c9a 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -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" diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 34db5d7..297cfce 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -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 } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go @@ -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 . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; 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 // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / 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 +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go index 7b650b0..bc848ca 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go @@ -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) } }) diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 4a9f21b..c6af543 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -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) diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 1dd6ce1..67c97f2 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -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", diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 110ca94..d9142a5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index d281800..fbf6ff3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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: diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go new file mode 100644 index 0000000..5bb7d77 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go @@ -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 . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of +// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace +// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments, +// key .) hold an open assignment of its build module on its node; 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 // . with an open assignment + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]bool // / 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 +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go new file mode 100644 index 0000000..16ee063 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 0c28769..b8c3fc5 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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 diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 755eaea..bd5a90e 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -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"