Make the builder's capacity a module setting, hold back a failed second-node build until the nodes change, and close at start a key left building for a fingerprint the module no longer has (issues 1661, 1664, 1660)

This commit is contained in:
2026-10-21 13:34:39 +02:00
parent d68d7ef3d9
commit dcf62b0413
12 changed files with 367 additions and 20 deletions
@@ -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,65 @@
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=0"}`), ""); err == nil {
t.Error("shanks=0 accepted")
}
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": ""
}
@@ -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)
}
}
@@ -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 {
+37 -12
View File
@@ -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 != "" || c.Description == "" {
t.Errorf("config = %+v, want one string setting, capacity, empty by default", 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": ""`) {
t.Errorf("files/settings.json = %q, %v, want the setting's empty default", 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" {
+28 -1
View File
@@ -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 not named runs one at a time, and empty is one everywhere.",
"pattern": "|[a-z0-9_-]+=[1-9][0-9]*(,[a-z0-9_-]+=[1-9][0-9]*)*",
"default": ""
}
},
"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"
}
]
}