Builder, checker and indexer count a build node's toolchain only where the catalogue's placement puts its build module, through mesh-sdk's one gate, and read the controller's assignments no more (issue 774, ADR 0549 §3, issue 1940)
This commit is contained in:
@@ -153,11 +153,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
// The toolchains first, so a build that starts from the first modules value finds the nodes.
|
||||
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
|
||||
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
|
||||
// assignments are settled first, so the toolchains settle through the gate already knowing them.
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
|
||||
// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
|
||||
// placement is settled first, so the toolchains settle through the gate already knowing it.
|
||||
gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
|
||||
watch.put = gate.Toolchain
|
||||
assignErrs, err := watchAssignments(ctx, js, gate, b.NodesChanged, logger)
|
||||
placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -254,7 +254,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
case err := <-assignErrs:
|
||||
case err := <-placementErrs:
|
||||
return err
|
||||
case err := <-watch.errs:
|
||||
return err
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
// 1945 (a part exits at start for a bucket it works without): a bus that refuses every toolchains bucket leaves the
|
||||
// watch unstarted and the part running, said; a refused bucket it cannot work without is an error naming the bucket and
|
||||
// its grant. The gate over the catalogue's placement, and its refusal, are mesh-sdk's (buildrun.ToolchainGate,
|
||||
// issue 774).
|
||||
func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var logs bytes.Buffer
|
||||
logger := log.New(&logs, "", 0)
|
||||
refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-build-node_toolchains")}
|
||||
if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil {
|
||||
t.Errorf("a refused toolchains bucket stopped the start: %v", err)
|
||||
}
|
||||
if !strings.Contains(logs.String(), "cannot be read") {
|
||||
t.Errorf("the refusals were not said: %q", logs.String())
|
||||
}
|
||||
if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil ||
|
||||
!strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") {
|
||||
t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err)
|
||||
}
|
||||
// The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk
|
||||
// head is told why and opens it again then.
|
||||
src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf}
|
||||
src.start()
|
||||
if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) {
|
||||
t.Errorf("a trunk head with the source refused: %v, want why", err)
|
||||
}
|
||||
if src.opened {
|
||||
t.Error("a refused source counts as opened, so it is never opened again")
|
||||
}
|
||||
}
|
||||
|
||||
// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.
|
||||
type refusingBus struct {
|
||||
refuseOpen bool
|
||||
err error
|
||||
}
|
||||
|
||||
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
|
||||
if r.refuseOpen {
|
||||
return nil, r.err
|
||||
}
|
||||
return refusedWatch{err: r.err}, nil
|
||||
}
|
||||
|
||||
type refusedWatch struct {
|
||||
jetstream.KeyValue
|
||||
err error
|
||||
}
|
||||
|
||||
func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
@@ -1,211 +0,0 @@
|
||||
package main
|
||||
|
||||
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
|
||||
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
|
||||
// toolchain agreement this file repairs.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
|
||||
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
|
||||
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
|
||||
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
|
||||
//
|
||||
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
|
||||
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
|
||||
// key <node>.<module>) hold an open assignment of its build module on its node, and only once it was written since that
|
||||
// assignment; it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report
|
||||
// written before the assignment is the one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node
|
||||
// assigned to ace again brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement, before the module ran
|
||||
// there, and the controller's take-in refused every module "builders disagree"). Until the assignments are settled (or where they cannot
|
||||
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
|
||||
// refusal is the guard of novox/hq ADR 0465, and it stays.
|
||||
type toolchainGate struct {
|
||||
mu sync.Mutex
|
||||
modules map[string]bool // the build modules whose toolchains pass the gate
|
||||
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
|
||||
logf func(format string, a ...any)
|
||||
known bool // the assignments are settled
|
||||
standing map[string]time.Time // <node>.<module> with an open assignment -> since when it stands
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]string // <module>/<key> -> why it was said left out, so each reason is said once
|
||||
}
|
||||
|
||||
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
|
||||
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]time.Time{},
|
||||
raw: map[string]map[string][]byte{}, left: map[string]string{}}
|
||||
for _, m := range modules {
|
||||
g.modules[m] = true
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.raw[module] == nil {
|
||||
g.raw[module] = map[string][]byte{}
|
||||
}
|
||||
if deleted {
|
||||
delete(g.raw[module], key)
|
||||
} else {
|
||||
g.raw[module][key] = value
|
||||
}
|
||||
return g.pass(module, key)
|
||||
}
|
||||
|
||||
// Assignment takes one change of the controller's assignments in (at is when the bus wrote that revision) and answers
|
||||
// whether what it passed on changed.
|
||||
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool, at time.Time) bool {
|
||||
node, module, ok := strings.Cut(key, ".")
|
||||
if !ok || !g.modules[module] {
|
||||
return false
|
||||
}
|
||||
assigned, open := openAssignment(value)
|
||||
open = open && !deleted
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
was, stood := g.standing[key]
|
||||
since := assigned
|
||||
if since.IsZero() {
|
||||
// A value with no time of its own (an assignment a module pulled in) stands since the revision that opened it.
|
||||
since = at
|
||||
if stood {
|
||||
since = was
|
||||
}
|
||||
}
|
||||
if open == stood && (!open || since.Equal(was)) {
|
||||
return false
|
||||
}
|
||||
if open {
|
||||
g.standing[key] = since
|
||||
} else {
|
||||
delete(g.standing, key)
|
||||
}
|
||||
if !g.known {
|
||||
return false
|
||||
}
|
||||
changed := false
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
|
||||
// passed on changed.
|
||||
func (g *toolchainGate) Settled() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.known = true
|
||||
changed := false
|
||||
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// pass hands one key on as it stands: its value when its node is assigned and wrote it since (or the assignments are not
|
||||
// known), deleted otherwise. The caller holds mu.
|
||||
func (g *toolchainGate) pass(module, key string) bool {
|
||||
value, held := g.raw[module][key]
|
||||
if !held {
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
node, _, err := delivery.ParseToolchainKey(key)
|
||||
why := ""
|
||||
if err == nil && g.known {
|
||||
why = g.leftOut(module, node, value)
|
||||
}
|
||||
if why == "" {
|
||||
delete(g.left, module+"/"+key)
|
||||
return g.forward(module, key, value, false)
|
||||
}
|
||||
if g.left[module+"/"+key] != why {
|
||||
g.left[module+"/"+key] = why
|
||||
g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why)
|
||||
}
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
|
||||
// leftOut is why a node's report is no builder's, or empty when it counts: its node has no open assignment of the module,
|
||||
// or the report was written before that assignment. A value that is no toolchain is passed on for its reader to refuse.
|
||||
// The caller holds mu.
|
||||
func (g *toolchainGate) leftOut(module, node string, value []byte) string {
|
||||
since, open := g.standing[node+"."+module]
|
||||
if !open {
|
||||
return fmt.Sprintf("%s has no assignment on %s, so it is no builder", module, node)
|
||||
}
|
||||
tc, err := delivery.DecodeToolchain(value)
|
||||
if err == nil && tc.At.Before(since) {
|
||||
return fmt.Sprintf("it was written at %s, before %s's assignment on %s at %s: it counts once %s writes it again",
|
||||
tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// openAssignment is whether a value of the controller's assignments holds an assignment that stands (one that has not
|
||||
// ended and has no state: a pending one is not on the machine yet; mesh-controller internal/chain Assignment), and when
|
||||
// it was assigned, zero where the value does not say.
|
||||
func openAssignment(value []byte) (time.Time, bool) {
|
||||
var a struct {
|
||||
Assigned time.Time `json:"assigned"`
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
if json.Unmarshal(value, &a) != nil {
|
||||
return time.Time{}, false
|
||||
}
|
||||
return a.Assigned, a.Ended.IsZero() && a.State == ""
|
||||
}
|
||||
|
||||
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
|
||||
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
|
||||
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
|
||||
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
|
||||
errs := make(chan error, 1)
|
||||
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
|
||||
if err != nil {
|
||||
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
|
||||
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
|
||||
return errs, nil
|
||||
}
|
||||
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) }
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
// Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read.
|
||||
logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
|
||||
return errs, nil
|
||||
}
|
||||
if gate.Settled() {
|
||||
changed(ctx)
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
|
||||
if gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) {
|
||||
changed(ctx)
|
||||
}
|
||||
})
|
||||
}()
|
||||
return errs, nil
|
||||
}
|
||||
@@ -1,228 +0,0 @@
|
||||
package main
|
||||
|
||||
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
func gateReport(t *testing.T, node, version string) (string, []byte) {
|
||||
t.Helper()
|
||||
key, err := delivery.ToolchainKey(node, "typescript")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, value
|
||||
}
|
||||
|
||||
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
|
||||
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
|
||||
return tc.Version, none
|
||||
}
|
||||
|
||||
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
|
||||
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
|
||||
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
|
||||
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var said []string
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
|
||||
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
|
||||
key, value := gateReport(t, node, version)
|
||||
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
|
||||
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
|
||||
}
|
||||
for _, node := range []string{"g14", "novox", "shanks"} {
|
||||
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false, time.Time{})
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false, time.Time{})
|
||||
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false, time.Time{}) // another module's: no matter
|
||||
if !gate.Settled() {
|
||||
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
|
||||
}
|
||||
if len(said) != 1 {
|
||||
t.Fatalf("ace's report is said left out once; said %v", said)
|
||||
}
|
||||
// ace reports again while unassigned (an old value seen again): still left out.
|
||||
key, value := gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an unassigned node's report changed what was passed on")
|
||||
}
|
||||
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false, time.Time{}) {
|
||||
t.Fatal("assigning ace again passed nothing on")
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
|
||||
t.Fatalf("builders that run different versions still disagree; got %q", none)
|
||||
}
|
||||
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false, time.Time{}) {
|
||||
t.Fatal("a pending assignment is no assignment on the machine")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", nil, true, time.Time{})
|
||||
if v, _ := agreedOf(set); v != "v26.12.1" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
// A report deleted from the bucket goes whether its node is assigned or not.
|
||||
key, _ = gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
|
||||
t.Fatal("deleting g14's report passed nothing on")
|
||||
}
|
||||
if n := len(set.Entries("typescript")); n != 2 {
|
||||
t.Fatalf("novox and shanks are left; got %d entries", n)
|
||||
}
|
||||
|
||||
// 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every
|
||||
// toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without
|
||||
// is an error naming the bucket and its grant.
|
||||
var logs bytes.Buffer
|
||||
logger := log.New(&logs, "", 0)
|
||||
refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")}
|
||||
if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil {
|
||||
t.Errorf("a refused watch of the assignments stopped the start: %v", err)
|
||||
}
|
||||
if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil {
|
||||
t.Errorf("a refused toolchains bucket stopped the start: %v", err)
|
||||
}
|
||||
if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") {
|
||||
t.Errorf("the refusals were not said: %q", logs.String())
|
||||
}
|
||||
if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil ||
|
||||
!strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") {
|
||||
t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err)
|
||||
}
|
||||
// The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk
|
||||
// head is told why and opens it again then.
|
||||
src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf}
|
||||
src.start()
|
||||
if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) {
|
||||
t.Errorf("a trunk head with the source refused: %v, want why", err)
|
||||
}
|
||||
if src.opened {
|
||||
t.Error("a refused source counts as opened, so it is never opened again")
|
||||
}
|
||||
}
|
||||
|
||||
// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.
|
||||
type refusingBus struct {
|
||||
refuseOpen bool
|
||||
err error
|
||||
}
|
||||
|
||||
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
|
||||
if r.refuseOpen {
|
||||
return nil, r.err
|
||||
}
|
||||
return refusedWatch{err: r.err}, nil
|
||||
}
|
||||
|
||||
type refusedWatch struct {
|
||||
jetstream.KeyValue
|
||||
err error
|
||||
}
|
||||
|
||||
func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
|
||||
// A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not
|
||||
// reported changes nothing, and the report an earlier run left on the bus (issue 1940, 2026-10-25 20:56: ace.typescript
|
||||
// v26.12.0 of 2026-10-23 06:29, mesh-build-node assigned to ace again at 20:56:45) is left out, said once, until the node
|
||||
// writes it again. An assignment whose value names no time stands since the revision that opened it.
|
||||
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
|
||||
left := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run
|
||||
assigned := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node assigned to ace again
|
||||
report := func(t *testing.T, node, version string, at time.Time) (string, []byte) {
|
||||
key, _ := gateReport(t, node, version)
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, value
|
||||
}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
assignment string // ace's assignment value
|
||||
revision time.Time // when the bus wrote it
|
||||
at time.Time // when ace wrote its report
|
||||
counts bool
|
||||
}{
|
||||
{"no report yet, then one", `{}`, time.Time{}, time.Now(), true},
|
||||
{"a report left before the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, left, false},
|
||||
{"a report left before a value with no time", `{"node":"ace","module":"mesh-build-node"}`, assigned, left, false},
|
||||
{"a report written since the assignment", `{"node":"ace","module":"mesh-build-node","assigned":"2026-10-25T19:56:45Z"}`, assigned, assigned.Add(30 * time.Second), true},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var said []string
|
||||
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) })
|
||||
for _, node := range []string{"g14", "novox", "shanks"} {
|
||||
gate.Assignment(node+".mesh-build-node", []byte(`{}`), false, time.Time{})
|
||||
key, value := gateReport(t, node, "v26.12.1")
|
||||
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
|
||||
}
|
||||
gate.Settled()
|
||||
if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision) {
|
||||
t.Fatal("an assignment with no report passed something on")
|
||||
}
|
||||
key, value := report(t, "ace", "v26.12.0", c.at)
|
||||
if got := gate.Toolchain(buildrun.NodeBuildModule, key, value, false); got != c.counts {
|
||||
t.Fatalf("ace's report passed on: %v, want %v", got, c.counts)
|
||||
}
|
||||
// A rewrite of the assignment that does not move it changes nothing, and nothing is said again.
|
||||
if gate.Assignment("ace.mesh-build-node", []byte(c.assignment), false, c.revision.Add(time.Minute)) {
|
||||
t.Fatal("the same assignment written again changed what was passed on")
|
||||
}
|
||||
v, none := agreedOf(set)
|
||||
if c.counts {
|
||||
if !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
|
||||
t.Fatalf("ace's own report counts and disagrees; got %q, %q", v, none)
|
||||
}
|
||||
return
|
||||
}
|
||||
if v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("ace's left report is no builder's, so g14, novox and shanks agree; got %q, %q", v, none)
|
||||
}
|
||||
if len(said) != 1 || !strings.Contains(said[0], "before mesh-build-node's assignment on ace at 2026-10-25T19:56:45Z") {
|
||||
t.Fatalf("said %q, want once why ace's report is left out", said)
|
||||
}
|
||||
// ace writes its report again since it was assigned: it counts.
|
||||
key, value = report(t, "ace", "v26.12.1", assigned.Add(time.Minute))
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("ace's new report was not passed on")
|
||||
}
|
||||
if n := len(set.Entries("typescript")); n != 4 {
|
||||
t.Fatalf("four builders report; got %d", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -92,8 +92,8 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
}
|
||||
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
|
||||
// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
||||
"mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
"mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
|
||||
@@ -32,11 +32,11 @@
|
||||
"mesh-build-image.steps",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules"
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement"
|
||||
],
|
||||
"uses": [
|
||||
"node-build-go",
|
||||
|
||||
Reference in New Issue
Block a user