Merge pull request 'Make the builder's build capacity a setting per node, hold a failed second-node build until the nodes change, and close keys left building for a vanished fingerprint (issues 1661, 1664, 1660)' (#14) from fix/1661-builder-follow-ups into main
This commit was merged in pull request #14.
This commit is contained in:
@@ -49,6 +49,14 @@ func (s kvStore) Update(ctx context.Context, key string, value []byte, revision
|
||||
return rev, conflict(err)
|
||||
}
|
||||
|
||||
func (s kvStore) Keys(ctx context.Context) ([]string, error) {
|
||||
keys, err := s.kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return keys, err
|
||||
}
|
||||
|
||||
func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut }
|
||||
|
||||
// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the
|
||||
|
||||
@@ -504,3 +504,21 @@ func (b *bed) moduleAt(t *testing.T, name, fixture, path, goVersion, at string)
|
||||
t.Helper()
|
||||
b.moduleCommit(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: goVersion, Recipe: "1"}}, at)
|
||||
}
|
||||
|
||||
// TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart: the builds bucket holds a key left building by another node for
|
||||
// a module the indexer no longer has; the program, at start, fails it naming why (issue 1660, ADR 0478 clause 6).
|
||||
func TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart(t *testing.T) {
|
||||
b := startBed(t)
|
||||
data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "n2.mesh-delivery-builder", Manifest: strings.Repeat("a", 64)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.builds.Put(b.ctx, mustKey(t, "ghost"), data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.start(t, "n1", nil)
|
||||
v := b.await(t, "ghost", isState(delivery.BuildFailed))
|
||||
if !strings.Contains(v.Error, "no longer has this fingerprint") || !strings.Contains(v.Error, "n2.mesh-delivery-builder") {
|
||||
t.Errorf("build = %+v, want it closed naming who left it", v)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -158,9 +158,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err)
|
||||
}
|
||||
}
|
||||
capacity, err := build.ParseCapacity(getenv("MESH_BUILDER_CAPACITY"))
|
||||
capacity, err := capacityFrom(strings.TrimSpace(getenv("MESH_BUILDER_SETTINGS")), getenv("MESH_BUILDER_CAPACITY"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("MESH_BUILDER_CAPACITY: %w", err)
|
||||
return fmt.Errorf("the build capacity setting: %w", err)
|
||||
}
|
||||
tools := &build.ToolchainSet{}
|
||||
presence := &build.PresenceSet{}
|
||||
@@ -177,7 +177,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Logf: logger.Printf,
|
||||
}
|
||||
// The toolchains first, so a build that starts from the first modules value finds the nodes.
|
||||
watch := newToolchainWatch(ctx, js, tools, b.Reconsider, logger)
|
||||
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
|
||||
if err := watch.start(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -214,7 +214,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() {
|
||||
presenceErrs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
|
||||
if presence.Put(e.Key(), e.Value(), !opOf(e)) {
|
||||
b.Reconsider(ctx) // a node came online or went: a build that waited for one may go on
|
||||
b.NodesChanged(ctx) // a node came online or went: a build that waited for one may go on
|
||||
}
|
||||
})
|
||||
}()
|
||||
@@ -235,7 +235,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() {
|
||||
haveErrs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) {
|
||||
if have.Put(e.Key(), e.Value(), !opOf(e)) {
|
||||
b.Reconsider(ctx) // a build module became healthy or stopped being: a build that waited may go on
|
||||
b.NodesChanged(ctx) // a build module became healthy or stopped being: a build that waited may go on
|
||||
}
|
||||
})
|
||||
}()
|
||||
@@ -251,6 +251,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// A key left building for a fingerprint its module no longer has is closed (issue 1660); the modules are all known now.
|
||||
if err := b.Sweep(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
logger.Printf("started on %s", cred.Node)
|
||||
errs := make(chan error, 1)
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
|
||||
// capacityFrom is the build capacity of each node (ADR 0478 clause 3): the module's setting `capacity`, `<node>=<steps at
|
||||
// once>` pairs separated by commas, laid into the settings file the controller composes (issue 1661); empty is one step at a
|
||||
// time on every node. With no settings file named, fallback is read instead, for a run outside the module's launch.
|
||||
func capacityFrom(path, fallback string) (map[string]int, error) {
|
||||
if path == "" {
|
||||
return build.ParseCapacity(fallback)
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the settings file: %w", err)
|
||||
}
|
||||
var s struct {
|
||||
Capacity string `json:"capacity"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return nil, fmt.Errorf("the settings file %s is not JSON: %w", path, err)
|
||||
}
|
||||
return build.ParseCapacity(s.Capacity)
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
|
||||
func settingsFile(t *testing.T, body string) string {
|
||||
t.Helper()
|
||||
p := filepath.Join(t.TempDir(), "settings.json")
|
||||
if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// 1661: the build capacity is the module's setting `capacity`, laid into the settings file: empty is one step at a time everywhere.
|
||||
func TestCapacityIsReadFromTheSettingsFile(t *testing.T) {
|
||||
for body, want := range map[string]map[string]int{
|
||||
`{"capacity": "shanks=3,ace=2"}`: {"shanks": 3, "ace": 2},
|
||||
`{"capacity": ""}`: {},
|
||||
`{}`: {},
|
||||
} {
|
||||
got, err := capacityFrom(settingsFile(t, body), "")
|
||||
if err != nil || len(got) != len(want) || got["shanks"] != want["shanks"] || got["ace"] != want["ace"] {
|
||||
t.Errorf("%s: capacity %v, %v, want %v", body, got, err, want)
|
||||
}
|
||||
}
|
||||
p := &build.Places{Capacity: mustCapacity(t, settingsFile(t, `{"capacity": ""}`))}
|
||||
if p.CapacityOf("shanks") != 1 || p.CapacityOf("ace") != 1 {
|
||||
t.Errorf("an empty setting gives %d and %d, want 1 everywhere", p.CapacityOf("shanks"), p.CapacityOf("ace"))
|
||||
}
|
||||
}
|
||||
|
||||
func mustCapacity(t *testing.T, path string) map[string]int {
|
||||
t.Helper()
|
||||
c, err := capacityFrom(path, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func TestACapacitySettingThatIsNotOneIsRefusedNamingIt(t *testing.T) {
|
||||
if _, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=-1"}`), ""); err == nil {
|
||||
t.Error("shanks=-1 accepted")
|
||||
}
|
||||
if c, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=3,ace=2,g14=0,novox=0"}`), ""); err != nil || c["g14"] != 0 || len(c) != 4 {
|
||||
t.Errorf("the starting value: %v, %v", c, err)
|
||||
}
|
||||
if _, err := capacityFrom(settingsFile(t, `not json`), ""); err == nil {
|
||||
t.Error("a file that is not JSON accepted")
|
||||
}
|
||||
if _, err := capacityFrom(filepath.Join(t.TempDir(), "none"), ""); err == nil {
|
||||
t.Error("a settings file that is not there accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// With no settings file named (a run outside the module's launch, as the tests do), the capacity is the environment's.
|
||||
func TestWithNoSettingsFileTheCapacityIsTheEnvironments(t *testing.T) {
|
||||
got, err := capacityFrom("", "ace=2")
|
||||
if err != nil || got["ace"] != 2 {
|
||||
t.Errorf("capacity %v, %v", got, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"capacity": "shanks=3,ace=2,g14=0,novox=0"
|
||||
}
|
||||
@@ -34,6 +34,9 @@ func ChooseAnyNodes(language string, entries []delivery.NodeToolchain, places *P
|
||||
}
|
||||
|
||||
func choose(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) {
|
||||
if places == nil {
|
||||
places = &Places{}
|
||||
}
|
||||
nodes := make([]string, 0, len(entries))
|
||||
var reporting []string
|
||||
seen := map[string]bool{}
|
||||
@@ -43,16 +46,13 @@ func choose(language string, entries []delivery.NodeToolchain, places *Places, q
|
||||
}
|
||||
seen[e.Node] = true
|
||||
reporting = append(reporting, e.Node)
|
||||
if qualifies == nil || qualifies(e.Node) {
|
||||
if places.CapacityOf(e.Node) > 0 && (qualifies == nil || qualifies(e.Node)) { // a capacity of 0 keeps the builder off a node
|
||||
nodes = append(nodes, e.Node)
|
||||
}
|
||||
}
|
||||
if len(nodes) == 0 {
|
||||
sort.Strings(reporting)
|
||||
return nil, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online with a healthy build module", language, strings.Join(reporting, ", "))
|
||||
}
|
||||
if places == nil {
|
||||
places = &Places{}
|
||||
return nil, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online with a healthy build module and a build capacity", language, strings.Join(reporting, ", "))
|
||||
}
|
||||
return places.choose(nodes, want), nil
|
||||
}
|
||||
|
||||
@@ -18,6 +18,8 @@ type Store interface {
|
||||
Get(ctx context.Context, key string) (value []byte, revision uint64, found bool, err error)
|
||||
Create(ctx context.Context, key string, value []byte) (uint64, error)
|
||||
Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
|
||||
// Keys are every key of the bucket.
|
||||
Keys(ctx context.Context) ([]string, error)
|
||||
}
|
||||
|
||||
// Claimer claims fingerprints for the builder instance By (ADR 0429 clause 5: a builder claims a fingerprint by creating
|
||||
@@ -57,7 +59,8 @@ type ClaimOptions struct {
|
||||
// time, and a build left building by a builder that is gone, on this node or another, is nobody's (ADR 0478 clause 6).
|
||||
TakeOver bool
|
||||
// MoreNodes says whether one of these build modules, which built a done key on one node only, has a second node that
|
||||
// qualifies now. Nil means none does.
|
||||
// qualifies now and the key may be built again (it is not held back after a failed second build, issue 1664). Nil means
|
||||
// none does.
|
||||
MoreNodes func(modules []string) bool
|
||||
}
|
||||
|
||||
|
||||
@@ -29,6 +29,16 @@ func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, err
|
||||
return v, m.revs[key], ok, nil
|
||||
}
|
||||
|
||||
func (m *memStore) Keys(_ context.Context) ([]string, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
var keys []string
|
||||
for k := range m.vals {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
return keys, nil
|
||||
}
|
||||
|
||||
func (m *memStore) hook() {
|
||||
if h := m.before; h != nil {
|
||||
m.before = nil
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
package build
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// 1664: a done key built on one node is not built again, over and over, while the second node's build keeps failing.
|
||||
func TestASecondNodeBuildThatFailsIsNotRetriedUntilTheNodesChange(t *testing.T) { // issue 1664
|
||||
r := newRig(t, goBundle)
|
||||
r.pres.Put("ace", presenceValue(t, false), false) // only novox qualifies
|
||||
first := r.consider(t, "modules/m")
|
||||
r.bm.fails["ace/build_go_fetch"] = "ace is out of disk"
|
||||
r.pres.Put("ace", presenceValue(t, true), false)
|
||||
r.b.NodesChanged(context.Background()) // ace came online: one more try
|
||||
r.b.Wait()
|
||||
tries := len(r.bm.callsOf("build_go_fetch"))
|
||||
if tries < 2 {
|
||||
t.Fatalf("%d fetches: the second node was never tried", tries)
|
||||
}
|
||||
if v := valueOf(t, r.st, r.key); !delivery.SameBuilderBuild(v, first) {
|
||||
t.Fatalf("the done key changed: %+v", v)
|
||||
}
|
||||
// Nothing about the nodes changed: a modules value, a Reconsider and a start of looking again build nothing.
|
||||
for i := 0; i < 3; i++ {
|
||||
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
|
||||
r.b.Reconsider(context.Background())
|
||||
r.b.Wait()
|
||||
}
|
||||
if n := len(r.bm.callsOf("build_go_fetch")); n != tries {
|
||||
t.Errorf("%d fetches after looking again, was %d: the failed second build was retried in a loop", n, tries)
|
||||
}
|
||||
// A node changing (ace goes and comes back) is a reason to try again, and ace now builds.
|
||||
delete(r.bm.fails, "ace/build_go_fetch")
|
||||
r.pres.Put("ace", presenceValue(t, false), false)
|
||||
r.b.NodesChanged(context.Background())
|
||||
r.pres.Put("ace", presenceValue(t, true), false)
|
||||
r.b.NodesChanged(context.Background())
|
||||
r.b.Wait()
|
||||
if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || strings.HasPrefix(v.Step, OneNodePrefix) || !strings.Contains(v.Step, "ace and novox") {
|
||||
t.Errorf("build = %+v, want it built on both nodes once ace could", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1660: at start the builder closes a key left building for a fingerprint the module no longer has.
|
||||
func TestAKeyLeftBuildingForAFingerprintTheModuleNoLongerHasIsClosedAtStart(t *testing.T) { // issue 1660
|
||||
r := newRig(t, goBundle)
|
||||
stale := seed(t, r.st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace.builder", Manifest: manifestD}) // m at fp1
|
||||
other, _ := delivery.BuilderBuildKey("gone", fp1)
|
||||
data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace.builder", Manifest: manifestD})
|
||||
r.st.put(other, data)
|
||||
// The module m has moved on to another fingerprint.
|
||||
moved := delivery.Module{Repository: "novox/mesh-catalog", Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", FP: "src1:" + hex64('f'),
|
||||
FP2: "src2:" + hex64('7'), Manifest: manifestD, Path: "modules/m", Toolchains: agreedToolchains}
|
||||
enc, _ := delivery.EncodeModule(moved)
|
||||
r.b.Module(context.Background(), "m", enc)
|
||||
r.b.Wait()
|
||||
if err := r.b.Sweep(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, k := range []string{stale, other} {
|
||||
v := valueOf(t, r.st, k)
|
||||
if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "no longer") || !strings.Contains(v.Error, "ace.builder") {
|
||||
t.Errorf("%s = %+v, want it closed as failed, naming why and who left it", k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepLeavesAKeyOfTheModulesCurrentFingerprintForTheBuildToTakeOver(t *testing.T) { // issue 1660
|
||||
r := newRig(t, goBundle)
|
||||
seed(t, r.st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace.builder", Manifest: manifestD})
|
||||
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) // takes it over and builds it
|
||||
r.b.Wait()
|
||||
if err := r.b.Sweep(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || v.By != "novox.builder" {
|
||||
t.Errorf("build = %+v, want done by novox.builder", v)
|
||||
}
|
||||
}
|
||||
@@ -14,7 +14,7 @@ const DefaultCapacity = 1
|
||||
|
||||
// ParseCapacity reads the builder's capacity setting: `node=steps` pairs separated by commas, such as
|
||||
// `shanks=3,ace=2`. A node without a pair has DefaultCapacity. A pair with no node, a count that is not a whole number of
|
||||
// at least 1, and a node named twice are refused.
|
||||
// at least 0, and a node named twice are refused. A count of 0 keeps the builder off the node: it is never chosen.
|
||||
func ParseCapacity(s string) (map[string]int, error) {
|
||||
out := map[string]int{}
|
||||
for _, pair := range strings.Split(s, ",") {
|
||||
@@ -25,8 +25,8 @@ func ParseCapacity(s string) (map[string]int, error) {
|
||||
node, count, ok := strings.Cut(pair, "=")
|
||||
node, count = strings.TrimSpace(node), strings.TrimSpace(count)
|
||||
n, err := strconv.Atoi(count)
|
||||
if !ok || node == "" || err != nil || n < 1 {
|
||||
return nil, fmt.Errorf("the capacity %q is not <node>=<steps at once, at least 1>", pair)
|
||||
if !ok || node == "" || err != nil || n < 0 {
|
||||
return nil, fmt.Errorf("the capacity %q is not <node>=<steps at once, 0 or more>", pair)
|
||||
}
|
||||
if _, dup := out[node]; dup {
|
||||
return nil, fmt.Errorf("the capacity names %s twice", node)
|
||||
@@ -52,9 +52,10 @@ type Places struct {
|
||||
changed chan struct{} // closed, and replaced, each time a lane ends
|
||||
}
|
||||
|
||||
// CapacityOf is how many steps the builder runs on the node at once.
|
||||
// CapacityOf is how many steps the builder runs on the node at once: DefaultCapacity for a node the setting does not
|
||||
// name, and 0 for one it names so, where the builder never builds.
|
||||
func (p *Places) CapacityOf(node string) int {
|
||||
if n, ok := p.Capacity[node]; ok && n >= 1 {
|
||||
if n, ok := p.Capacity[node]; ok && n >= 0 {
|
||||
return n
|
||||
}
|
||||
return DefaultCapacity
|
||||
|
||||
@@ -19,7 +19,7 @@ func TestParseCapacityReadsNodeEqualsCountPairs(t *testing.T) {
|
||||
if c, err := ParseCapacity(""); err != nil || len(c) != 0 {
|
||||
t.Errorf("empty: %v, %v", c, err)
|
||||
}
|
||||
for _, bad := range []string{"shanks", "shanks=0", "shanks=x", "=2", "a=1,a=2", "shanks=-1"} {
|
||||
for _, bad := range []string{"shanks", "shanks=x", "=2", "a=1,a=2", "shanks=-1"} {
|
||||
if _, err := ParseCapacity(bad); err == nil {
|
||||
t.Errorf("%q accepted", bad)
|
||||
}
|
||||
@@ -138,3 +138,49 @@ func TestTwoNodeAgreementStillTakesTwoDifferentNodesWhenCapacitiesAreSet(t *test
|
||||
t.Errorf("build = %+v, want done on ace and novox", v)
|
||||
}
|
||||
}
|
||||
|
||||
// The operator's decision of 2026-10-21 13:40: the builder builds on shanks and ace only; a capacity of 0 keeps it off a node.
|
||||
func TestACapacityOfZeroKeepsTheBuilderOffTheNodeAndANodeNotNamedKeepsOne(t *testing.T) { // issue 1661
|
||||
c, err := ParseCapacity("shanks=3,ace=2,g14=0,novox=0")
|
||||
if err != nil || c["g14"] != 0 || c["novox"] != 0 || c["shanks"] != 3 || len(c) != 4 {
|
||||
t.Fatalf("capacity = %v, %v", c, err)
|
||||
}
|
||||
p := &Places{Capacity: c}
|
||||
if p.CapacityOf("g14") != 0 || p.CapacityOf("novox") != 0 || p.CapacityOf("ace") != 2 || p.CapacityOf("other") != DefaultCapacity {
|
||||
t.Errorf("capacities = %d %d %d %d, want 0 0 2 and the default for a node not named", p.CapacityOf("g14"), p.CapacityOf("novox"), p.CapacityOf("ace"), p.CapacityOf("other"))
|
||||
}
|
||||
if _, err := ParseCapacity("shanks=-1"); err == nil {
|
||||
t.Error("a negative capacity accepted")
|
||||
}
|
||||
entries := tcs("go", "go1.27.1", "ace", "g14", "novox", "shanks", "other")
|
||||
for i := 0; i < 6; i++ {
|
||||
nodes, _, err := ChooseNodes("go", entries, p, nil, 2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, n := range nodes {
|
||||
if n == "g14" || n == "novox" {
|
||||
t.Fatalf("choice %d: %v includes a node of capacity 0", i, nodes)
|
||||
}
|
||||
}
|
||||
}
|
||||
// Only nodes of capacity 0 report: nothing qualifies, and the build says why.
|
||||
_, _, err = ChooseNodes("go", tcs("go", "go1.27.1", "g14", "novox"), p, nil, 2)
|
||||
if err == nil || !strings.Contains(err.Error(), "capacity") {
|
||||
t.Errorf("only capacity-0 nodes: error = %v, want it to name the capacity", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestABuildNeverRunsOnANodeOfCapacityZero(t *testing.T) { // issue 1661
|
||||
r := newRig(t, catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"}) // ace and novox report
|
||||
r.b.Places = &Places{Capacity: map[string]int{"novox": 0}}
|
||||
v := r.consider(t, "modules/m")
|
||||
if v.State != "done" || !strings.Contains(v.Step, "on one node, ace") {
|
||||
t.Fatalf("build = %+v, want done on ace alone", v)
|
||||
}
|
||||
for _, c := range r.bm.calls {
|
||||
if c.node == "novox" {
|
||||
t.Errorf("a step was called on novox, whose capacity is 0: %+v", c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package build
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -53,6 +54,10 @@ type Builder struct {
|
||||
again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built
|
||||
plans map[string]planned // by manifest digest
|
||||
tried map[string]bool // by build key: looked at since this process started
|
||||
// epoch counts the changes of the nodes (online, healthy, a toolchain) seen since this process started; a done key
|
||||
// whose second-node build failed is held back until it moves (issue 1664).
|
||||
epoch uint64
|
||||
secondAt map[string]uint64 // by build key: the epoch a second-node build of a one-node key failed at
|
||||
}
|
||||
|
||||
type planned struct {
|
||||
@@ -68,6 +73,7 @@ func (b *Builder) init() {
|
||||
}
|
||||
b.sem = make(chan struct{}, limit)
|
||||
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.secondAt = map[string]uint64{}
|
||||
if b.Places == nil {
|
||||
b.Places = &Places{}
|
||||
}
|
||||
@@ -98,6 +104,63 @@ func (b *Builder) Module(ctx context.Context, key string, value []byte) {
|
||||
b.consider(ctx, key, m)
|
||||
}
|
||||
|
||||
// NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or
|
||||
// stopped being, or a toolchain moved. A done key built on one node whose second-node build failed is tried again only
|
||||
// after such a change, not at every modules value (issue 1664).
|
||||
func (b *Builder) NodesChanged(ctx context.Context) {
|
||||
b.init()
|
||||
b.mu.Lock()
|
||||
b.epoch++
|
||||
b.mu.Unlock()
|
||||
b.Reconsider(ctx)
|
||||
}
|
||||
|
||||
// Sweep closes, at start, every key left building for a fingerprint its module no longer has (issue 1660, ADR 0478
|
||||
// clause 6): the key can never be built, since nothing asks for it, and would stay building for ever. It fails the key
|
||||
// naming who left it. A key of the fingerprint the module has now is not touched: the build takes it over. Call it
|
||||
// once the modules bucket has been read, so every module is known.
|
||||
func (b *Builder) Sweep(ctx context.Context) error {
|
||||
b.init()
|
||||
keys, err := b.Claimer.Store.Keys(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b.mu.Lock()
|
||||
current := map[string]bool{}
|
||||
for name, m := range b.known {
|
||||
if m.FP2 != "" {
|
||||
if k, err := delivery.BuilderBuildKey(name, m.FP2); err == nil {
|
||||
current[k] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
for _, key := range keys {
|
||||
if current[key] {
|
||||
continue
|
||||
}
|
||||
value, rev, found, err := b.Claimer.Store.Get(ctx, key)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
cur, err := delivery.DecodeBuilderBuild(value)
|
||||
if err != nil || cur.State != delivery.BuildBuilding {
|
||||
continue
|
||||
}
|
||||
closed := delivery.BuilderBuild{State: delivery.BuildFailed, By: b.Claimer.By, Manifest: cur.Manifest, Step: cur.Step,
|
||||
Error: fmt.Sprintf("closed at start: left building by %s, and the module no longer has this fingerprint, so nothing will build it", cur.By)}
|
||||
data, err := delivery.EncodeBuilderBuild(closed)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := b.Claimer.Store.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) {
|
||||
return fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
b.logf("%s: closed, left building by %s for a fingerprint the module no longer has", key, cur.By)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Forget takes a module out (its key was deleted).
|
||||
func (b *Builder) Forget(key string) {
|
||||
b.init()
|
||||
@@ -206,8 +269,14 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
|
||||
b.mu.Lock()
|
||||
first := !b.tried[key]
|
||||
b.tried[key] = true
|
||||
epoch := b.epoch
|
||||
failedAt, failedBefore := b.secondAt[key]
|
||||
b.mu.Unlock()
|
||||
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, MoreNodes: func(modules []string) bool { return b.moreNodes(modules) }})
|
||||
moreNodes := func(modules []string) bool {
|
||||
// A second build that failed is not tried again until the nodes change (issue 1664).
|
||||
return (!failedBefore || epoch > failedAt) && b.moreNodes(modules)
|
||||
}
|
||||
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, MoreNodes: moreNodes})
|
||||
if err != nil {
|
||||
b.logf("%s: %v", module, err)
|
||||
return
|
||||
@@ -228,6 +297,9 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
|
||||
// A done key built again on a second node keeps its done value when the new build fails for another reason than
|
||||
// the digests: the first build stands, and a later change of the nodes looks again.
|
||||
b.logf("%s: built again on a second node and failed, the done build stands: %s", cl.Key, result.Error)
|
||||
b.mu.Lock()
|
||||
b.secondAt[cl.Key] = epoch
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if err := b.Claimer.Finish(ctx, cl, result); err != nil {
|
||||
@@ -383,7 +455,7 @@ func (b *Builder) online(node string) bool { return b.Presence == nil || b.Prese
|
||||
|
||||
// qualifies is whether the node can build with the build module now: it is online and the build module on it is healthy.
|
||||
func (b *Builder) qualifies(node, bm string) bool {
|
||||
return b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm))
|
||||
return b.Places.CapacityOf(node) > 0 && b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm))
|
||||
}
|
||||
|
||||
// moreNodes is whether any of these build modules has two nodes that qualify now.
|
||||
|
||||
@@ -32,13 +32,19 @@ type artifact struct {
|
||||
}
|
||||
|
||||
type manifest struct {
|
||||
Module string `json:"module"`
|
||||
Version string `json:"version"`
|
||||
Slug string `json:"slug"`
|
||||
Consumes []string `json:"consumes"`
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Module string `json:"module"`
|
||||
Version string `json:"version"`
|
||||
Slug string `json:"slug"`
|
||||
Consumes []string `json:"consumes"`
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Config map[string]struct {
|
||||
Type string `json:"type"`
|
||||
Description string `json:"description"`
|
||||
Pattern string `json:"pattern"`
|
||||
Default string `json:"default"`
|
||||
} `json:"config"`
|
||||
OwnSecrets map[string]string `json:"own-secrets"`
|
||||
SecretsOwner string `json:"secrets-owner"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
@@ -123,16 +129,35 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1
|
||||
t.Errorf("process = %v", p)
|
||||
}
|
||||
env, _ := p["env"].(map[string]any)
|
||||
// The build capacity of each node: how many build steps the builder runs on it at once (a node not named runs one). No
|
||||
// part of delivery reads a per-node setting yet, so the value is the manifest's own; changing it is a change of this line.
|
||||
if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_CAPACITY"] != "shanks=3,ace=2" {
|
||||
// The build capacity of each node is the module's setting `capacity` (issue 1661, ADR 0478 clause 3), laid into the
|
||||
// settings file the program reads; empty is one step at a time everywhere. The environment holds no capacity.
|
||||
if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_SETTINGS"] != "${dir:state}/settings.json" {
|
||||
t.Errorf("env = %v", p["env"])
|
||||
}
|
||||
if !reflect.DeepEqual(p["restart-on"], []any{"settings"}) {
|
||||
t.Errorf("restart-on = %v, want the settings file, so a changed capacity reaches the running program", p["restart-on"])
|
||||
}
|
||||
c, ok := m.Config["capacity"]
|
||||
if !ok || len(m.Config) != 1 || c.Type != "string" || c.Default != "shanks=3,ace=2,g14=0,novox=0" || c.Description == "" {
|
||||
t.Errorf("config = %+v, want one string setting, capacity, shanks=3,ace=2,g14=0,novox=0 to start", m.Config)
|
||||
}
|
||||
var settings map[string]any
|
||||
for _, r := range m.Resources {
|
||||
if r["id"] == "settings" {
|
||||
settings = r
|
||||
}
|
||||
}
|
||||
if settings == nil || settings["type"] != "file" || settings["merge"] != "json" || !reflect.DeepEqual(settings["takes-config"], []any{"capacity"}) || settings["path"] != "${dir:state}/settings.json" {
|
||||
t.Errorf("settings resource = %v", settings)
|
||||
}
|
||||
if raw, err := os.ReadFile("files/settings.json"); err != nil || !strings.Contains(string(raw), `"capacity": "shanks=3,ace=2,g14=0,novox=0"`) {
|
||||
t.Errorf("files/settings.json = %q, %v, want the setting's starting value", raw, err)
|
||||
}
|
||||
if _, bus := p["bus"]; bus {
|
||||
t.Errorf("the process declares a bus: it runs on its own broker credential")
|
||||
}
|
||||
if len(m.Build.Artifacts) != 1 {
|
||||
t.Fatalf("artifacts = %+v", m.Build.Artifacts)
|
||||
if len(m.Build.Artifacts) != 2 || m.Build.Artifacts[1].Name != "files" || m.Build.Artifacts[1].Kind != "archive" || m.Build.Artifacts[1].From != "files" {
|
||||
t.Fatalf("artifacts = %+v, want the program and the files", m.Build.Artifacts)
|
||||
}
|
||||
a := m.Build.Artifacts[0]
|
||||
if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-builder" || !strings.HasSuffix(a.From, a.Binary) || a.Kind != "bundle" || a.Language != "go" {
|
||||
|
||||
@@ -25,6 +25,14 @@
|
||||
"mesh-build-node.build_node_build",
|
||||
"mesh-build-node.build_node_fetch"
|
||||
],
|
||||
"config": {
|
||||
"capacity": {
|
||||
"type": "string",
|
||||
"description": "How many build steps the builder runs at once on each node, as node=steps pairs separated by commas, such as shanks=3,ace=2 (ADR 0478 clause 3); a node set to 0 never builds, a node not named runs one at a time, and empty is one everywhere. The builder builds on shanks and ace only (the operator, 2026-10-21).",
|
||||
"pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
|
||||
"default": "shanks=3,ace=2,g14=0,novox=0"
|
||||
}
|
||||
},
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
@@ -45,6 +53,17 @@
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "settings",
|
||||
"trusted": false,
|
||||
"type": "file",
|
||||
"path": "${dir:state}/settings.json",
|
||||
"mode": "0600",
|
||||
"owner": "mesh-delivery-builder",
|
||||
"takes-config": ["capacity"],
|
||||
"merge": "json",
|
||||
"template": "files/settings.json"
|
||||
},
|
||||
{
|
||||
"id": "builder",
|
||||
"type": "process",
|
||||
@@ -55,8 +74,11 @@
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker",
|
||||
"MESH_BUILDER_CAPACITY": "shanks=3,ace=2"
|
||||
"MESH_BUILDER_SETTINGS": "${dir:state}/settings.json"
|
||||
},
|
||||
"restart-on": [
|
||||
"settings"
|
||||
],
|
||||
"user": "mesh-delivery-builder",
|
||||
"health": {
|
||||
"kind": "unit"
|
||||
@@ -72,6 +94,11 @@
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-delivery-builder",
|
||||
"binary": "mesh-delivery-builder"
|
||||
},
|
||||
{
|
||||
"name": "files",
|
||||
"kind": "archive",
|
||||
"from": "files"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user