Run at most a node's build capacity of build steps on it at once, and hold a step back for a place with no timer (issue 1607, round 3)

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

@@ -158,6 +158,10 @@ 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"))
+	if err != nil {
+		return fmt.Errorf("MESH_BUILDER_CAPACITY: %w", err)
+	}
 	tools := &build.ToolchainSet{}
 	presence := &build.PresenceSet{}
 	b := &build.Builder{
@@ -165,6 +169,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Runner:     build.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: bound},
 		Toolchains: tools,
 		Presence:   presence,
+		Places:     &build.Places{Capacity: capacity},
 		Manifests:  manifestReader{manifests},
 		Nodes:      nodesPerFingerprint,
 		Logf:       logger.Printf,

modules/mesh-delivery-builder/internal/build/choose.go

@@ -9,13 +9,14 @@ import (
 )
 
 // ChooseNodes picks up to want build nodes for a language (ADR 0478 clause 3): the nodes whose toolchains key reports the
-// version the nodes agree on, the ones with the fewest steps the builder has running there first, a tie broken by the
-// node's name. A node qualifies only when online is true of it (the watchman's presence of the node; nil means every
+// version the nodes agree on, the ones with room under their build capacity first (places), then the fewest steps the
+// builder has running there, a tie broken by the node's name. A node at its capacity is still chosen when the others are
+// too (fewer than want have room): its steps wait for a place (Places.Acquire). A node qualifies only when online is true of it (the watchman's presence of the node; nil means every
 // node is). It answers the toolchain all the reporting nodes agree on, online or not, as the indexer computes it. When
 // nothing is reported, or the nodes report different versions (so the indexer agreed none), it fails naming the language
 // and why; when no reporting node is online it fails naming the nodes. A node whose build module refuses on its toolchain
 // refuses the step, and the build fails naming it.
-func ChooseNodes(language string, entries []delivery.NodeToolchain, running map[string]int, online func(node string) bool, want int) ([]string, delivery.Toolchain2, error) {
+func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Places, online func(node string) bool, want int) ([]string, delivery.Toolchain2, error) {
 	tc, none := delivery.AgreedToolchain(entries)
 	if none != "" {
 		return nil, delivery.Toolchain2{}, fmt.Errorf("no %s build node qualifies: %s", language, none)
@@ -37,12 +38,10 @@ func ChooseNodes(language string, entries []delivery.NodeToolchain, running map[
 		sort.Strings(reporting)
 		return nil, tc, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online", language, strings.Join(reporting, ", "))
 	}
-	sort.Slice(nodes, func(i, j int) bool {
-		if running[nodes[i]] != running[nodes[j]] {
-			return running[nodes[i]] < running[nodes[j]]
-		}
-		return nodes[i] < nodes[j]
-	})
+	if places == nil {
+		places = &Places{}
+	}
+	places.order(nodes) // nodes with room first, then the fewest steps running, then by name
 	if len(nodes) > want {
 		nodes = nodes[:want]
 	}

modules/mesh-delivery-builder/internal/build/choose_test.go

@@ -17,7 +17,7 @@ func tcs(language, version string, nodes ...string) []delivery.NodeToolchain {
 
 func TestChooseTakesTheAgreedNodesWithFewestRunningStepsTiesByName(t *testing.T) { // 1607/AC4, ADR 0478 clause 3
 	entries := tcs("go", "go1.27.1", "novox", "ace", "g14", "shanks")
-	nodes, tc, err := ChooseNodes("go", entries, map[string]int{"novox": 3, "ace": 1, "g14": 1, "shanks": 0}, nil, 2)
+	nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 3, "ace": 1, "g14": 1, "shanks": 0}), nil, 2)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -65,7 +65,7 @@ func TestAgreeNeedsEqualDigestsAndNamesBothNodesWhenTheyDiffer(t *testing.T) { /
 func TestChooseSkipsNodesThatAreNotOnlineAndFailsNamingThemWhenNoneIs(t *testing.T) { // 1607 review 3, ADR 0478 clause 3
 	entries := tcs("go", "go1.27.1", "ace", "g14", "novox")
 	online := func(n string) bool { return n != "g14" } // a sleeping laptop
-	nodes, tc, err := ChooseNodes("go", entries, map[string]int{"novox": 5}, online, 2)
+	nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 5}), online, 2)
 	if err != nil || len(nodes) != 2 || nodes[0] != "ace" || nodes[1] != "novox" || tc.Version != "go1.27.1" {
 		t.Errorf("nodes = %v, toolchain %+v, err %v, want [ace novox] with g14 left out although it runs nothing", nodes, tc, err)
 	}
@@ -77,3 +77,19 @@ func TestChooseSkipsNodesThatAreNotOnlineAndFailsNamingThemWhenNoneIs(t *testing
 		t.Errorf("none online: the agreed toolchain %+v is still answered", tc)
 	}
 }
+
+// placesRunning is Places with these steps counted as running, for a test of the choice.
+func placesRunning(running map[string]int) *Places {
+	return &Places{running: running, changed: make(chan struct{})}
+}
+
+func TestChoosePrefersNodesBelowTheirCapacity(t *testing.T) { // round 3
+	entries := tcs("go", "go1.27.1", "ace", "novox", "shanks")
+	// shanks has capacity 3 with 2 running: room; ace has the default 1 with 1 running: none; novox has none running.
+	p := placesRunning(map[string]int{"shanks": 2, "ace": 1})
+	p.Capacity = map[string]int{"shanks": 3}
+	nodes, _, err := ChooseNodes("go", entries, p, nil, 2)
+	if err != nil || len(nodes) != 2 || nodes[0] != "novox" || nodes[1] != "shanks" {
+		t.Errorf("nodes = %v, err %v, want [novox shanks]: the nodes with room, ace at capacity left out", nodes, err)
+	}
+}

modules/mesh-delivery-builder/internal/build/places.go

@@ -0,0 +1,118 @@
+package build
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+// DefaultCapacity is the build capacity of a node that has no setting: one build step at a time.
+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.
+func ParseCapacity(s string) (map[string]int, error) {
+	out := map[string]int{}
+	for _, pair := range strings.Split(s, ",") {
+		pair = strings.TrimSpace(pair)
+		if pair == "" {
+			continue
+		}
+		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 _, dup := out[node]; dup {
+			return nil, fmt.Errorf("the capacity names %s twice", node)
+		}
+		out[node] = n
+	}
+	return out, nil
+}
+
+// Places counts the build steps the builder runs on each node and holds a step back while its node is at its
+// capacity (the operator's decision of 2026-10-21; ADR 0478 clause 3). A step waits for a place by watching the ends of
+// the other steps, not by a timer: ending a step wakes the waiting ones.
+type Places struct {
+	// Capacity is the number of steps at once by node; a node not in it has DefaultCapacity.
+	Capacity map[string]int
+
+	mu      sync.Mutex
+	running map[string]int
+	changed chan struct{} // closed, and replaced, each time a step ends
+}
+
+// CapacityOf is how many steps the builder runs on the node at once.
+func (p *Places) CapacityOf(node string) int {
+	if n, ok := p.Capacity[node]; ok && n >= 1 {
+		return n
+	}
+	return DefaultCapacity
+}
+
+// Running is the steps in flight now, by node.
+func (p *Places) Running() map[string]int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	out := make(map[string]int, len(p.running))
+	for n, c := range p.running {
+		if c > 0 {
+			out[n] = c
+		}
+	}
+	return out
+}
+
+// Acquire takes a place on the node, waiting while the node is at its capacity, and answers the function that gives it
+// back. It returns an error when ctx ends first.
+func (p *Places) Acquire(ctx context.Context, node string) (release func(), err error) {
+	for {
+		p.mu.Lock()
+		if p.running == nil {
+			p.running = map[string]int{}
+			p.changed = make(chan struct{})
+		}
+		if p.running[node] < p.CapacityOf(node) {
+			p.running[node]++
+			p.mu.Unlock()
+			var once sync.Once
+			return func() {
+				once.Do(func() {
+					p.mu.Lock()
+					p.running[node]--
+					close(p.changed)
+					p.changed = make(chan struct{})
+					p.mu.Unlock()
+				})
+			}, nil
+		}
+		wait := p.changed
+		p.mu.Unlock()
+		select {
+		case <-wait:
+		case <-ctx.Done():
+			return nil, fmt.Errorf("waiting for a place on %s: %w", node, ctx.Err())
+		}
+	}
+}
+
+// order orders the nodes for a choice: those under their capacity first, then fewest running, then by name.
+func (p *Places) order(nodes []string) {
+	running := p.Running()
+	sort.SliceStable(nodes, func(i, j int) bool {
+		ri, rj := running[nodes[i]] < p.CapacityOf(nodes[i]), running[nodes[j]] < p.CapacityOf(nodes[j])
+		if ri != rj {
+			return ri
+		}
+		if running[nodes[i]] != running[nodes[j]] {
+			return running[nodes[i]] < running[nodes[j]]
+		}
+		return nodes[i] < nodes[j]
+	})
+}

modules/mesh-delivery-builder/internal/build/places_test.go

@@ -0,0 +1,140 @@
+package build
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+func TestParseCapacityReadsNodeEqualsCountPairs(t *testing.T) {
+	c, err := ParseCapacity(" shanks=3, ace=2 ")
+	if err != nil || c["shanks"] != 3 || c["ace"] != 2 || len(c) != 2 {
+		t.Errorf("capacity = %v, %v", c, err)
+	}
+	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"} {
+		if _, err := ParseCapacity(bad); err == nil {
+			t.Errorf("%q accepted", bad)
+		}
+	}
+}
+
+func acquireAsync(p *Places, node string) chan func() {
+	got := make(chan func(), 1)
+	go func() {
+		release, err := p.Acquire(context.Background(), node)
+		if err != nil {
+			close(got)
+			return
+		}
+		got <- release
+	}()
+	return got
+}
+
+func TestThreeStepsOnANodeOfCapacityTwoRunTwoAtOnceAndTheThirdStartsWhenOneEnds(t *testing.T) { // round 3
+	p := &Places{Capacity: map[string]int{"ace": 2}}
+	a, b := acquireAsync(p, "ace"), acquireAsync(p, "ace")
+	ra, rb := <-a, <-b
+	third := acquireAsync(p, "ace")
+	select {
+	case <-third:
+		t.Fatal("a third step started on a node of capacity 2 with two running")
+	case <-time.After(50 * time.Millisecond):
+	}
+	if p.Running()["ace"] != 2 {
+		t.Errorf("running = %v, want 2 on ace", p.Running())
+	}
+	ra()
+	select {
+	case r := <-third:
+		r()
+	case <-time.After(2 * time.Second):
+		t.Fatal("the third step did not start when one ended")
+	}
+	rb()
+}
+
+func TestANodeWithNoSettingRunsOneStepAtATime(t *testing.T) { // round 3
+	p := &Places{Capacity: map[string]int{"shanks": 3}}
+	first := <-acquireAsync(p, "g14")
+	second := acquireAsync(p, "g14")
+	select {
+	case <-second:
+		t.Fatal("two steps on a node without a setting")
+	case <-time.After(50 * time.Millisecond):
+	}
+	first()
+	r := <-second
+	r()
+	if p.CapacityOf("g14") != 1 || p.CapacityOf("shanks") != 3 {
+		t.Errorf("capacities = %d, %d", p.CapacityOf("g14"), p.CapacityOf("shanks"))
+	}
+}
+
+func TestAWaitingStepStopsWhenItsContextEnds(t *testing.T) { // round 3
+	p := &Places{}
+	r := <-acquireAsync(p, "ace")
+	defer r()
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { _, err := p.Acquire(ctx, "ace"); done <- err }()
+	cancel()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Error("Acquire returned no error for a cancelled wait")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("a cancelled wait did not return")
+	}
+}
+
+func TestStepsOfSeveralBuildsNeverExceedANodesCapacity(t *testing.T) { // round 3, through the builder
+	r := newRig(t, catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"})
+	r.b.Places = &Places{Capacity: map[string]int{"ace": 2}} // novox has none: one at a time
+	var now, most atomic.Int32
+	var mu sync.Mutex
+	perNode := map[string]int{}
+	r.bm.steps.during = func(key string) func() {
+		node, _, _ := strings.Cut(key, ".")
+		mu.Lock()
+		perNode[node]++
+		if int32(perNode[node]) > most.Load() && node == "ace" {
+			most.Store(int32(perNode[node]))
+		}
+		if node == "novox" && perNode[node] > 1 {
+			t.Errorf("%d steps at once on novox, which has capacity 1", perNode[node])
+		}
+		mu.Unlock()
+		now.Add(1)
+		time.Sleep(15 * time.Millisecond)
+		return func() { mu.Lock(); perNode[node]--; mu.Unlock(); now.Add(-1) }
+	}
+	for _, m := range []string{"m1", "m2", "m3", "m4"} {
+		r.b.Module(context.Background(), m, moduleValue(t, "modules/"+m))
+	}
+	r.b.Wait()
+	if most.Load() > 2 {
+		t.Errorf("%d steps at once on ace, which has capacity 2", most.Load())
+	}
+	if most.Load() < 2 {
+		t.Errorf("at most %d steps at once on ace: the capacity was not used", most.Load())
+	}
+}
+
+func TestTwoNodeAgreementStillTakesTwoDifferentNodesWhenCapacitiesAreSet(t *testing.T) { // round 3, ADR 0478 clause 4
+	r := newRig(t, catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"})
+	r.b.Places = &Places{Capacity: map[string]int{"ace": 2, "novox": 3}}
+	v := r.consider(t, "modules/m")
+	if v.State != "done" || !strings.Contains(v.Step, "ace and novox") {
+		t.Errorf("build = %+v, want done on ace and novox", v)
+	}
+}

modules/mesh-delivery-builder/internal/build/run.go

@@ -31,6 +31,8 @@ type Builder struct {
 	// Presence says which nodes are online (the watchman's); a node that is not is not chosen.
 	Presence  interface{ Online(node string) bool }
 	Manifests Manifests
+	// Places holds a step back while its node is at its build capacity; a Builder with none runs one step at a time on each node.
+	Places *Places
 	// Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4).
 	Nodes int
 	// Limit is how many builds run at once; four when 0.
@@ -44,7 +46,6 @@ type Builder struct {
 	known    map[string]delivery.Module // by modules key
 	inflight map[string]bool            // by build key
 	again    map[string]delivery.Module // by build key: the newest value that arrived while the key was being built
-	running  map[string]int             // steps in flight, by `<build module>/<node>`
 	plans    map[string]planned         // by manifest digest
 }
 
@@ -60,7 +61,10 @@ func (b *Builder) init() {
 			limit = 4
 		}
 		b.sem = make(chan struct{}, limit)
-		b.known, b.inflight, b.again, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]int{}, map[string]planned{}
+		b.known, b.inflight, b.again, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}
+		if b.Places == nil {
+			b.Places = &Places{}
+		}
 	})
 }
 
@@ -261,7 +265,7 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
 	var steps []string
 	for _, bm := range plan.Modules() {
 		language := LanguageOf[bm]
-		nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.runningOn(bm), b.online, b.Nodes)
+		nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.Places, b.online, b.Nodes)
 		if err != nil {
 			return failed("%s", err)
 		}
@@ -320,25 +324,13 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p
 
 func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) }
 
-func (b *Builder) runningOn(buildModule string) map[string]int {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	out := map[string]int{}
-	for k, n := range b.running {
-		if mod, node, _ := strings.Cut(k, "/"); mod == buildModule {
-			out[node] = n
-		}
-	}
-	return out
-}
-
-// step runs one step of a build module, counting it as running on the node while it does.
+// step runs one step of a build module, once its node has a place for it.
 func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) {
-	k := bm + "/" + node
-	b.mu.Lock()
-	b.running[k]++
-	b.mu.Unlock()
-	defer func() { b.mu.Lock(); b.running[k]--; b.mu.Unlock() }()
+	release, err := b.Places.Acquire(ctx, node)
+	if err != nil {
+		return nil, err
+	}
+	defer release()
 	return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args)
 }
 

modules/mesh-delivery-builder/internal/build/steps_test.go

@@ -36,13 +36,19 @@ func (f *fakeCaller) CallOn(node, key string, body any) (json.RawMessage, error)
 type fakeSteps struct {
 	mu    sync.Mutex
 	steps map[string]delivery.Step
+	// during, when set, is called as a wait starts and the function it answers as the wait ends.
+	during func(key string) func()
 }
 
 func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step, error) {
 	f.mu.Lock()
 	s, ok := f.steps[module+"/"+key]
+	during := f.during
 	f.mu.Unlock()
 	if ok {
+		if during != nil {
+			defer during(key)()
+		}
 		return s, nil
 	}
 	<-ctx.Done()

modules/mesh-delivery-builder/manifest_test.go

@@ -123,7 +123,9 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1
 		t.Errorf("process = %v", p)
 	}
 	env, _ := p["env"].(map[string]any)
-	if len(env) != 1 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" {
+	// 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" {
 		t.Errorf("env = %v", p["env"])
 	}
 	if _, bus := p["bus"]; bus {

modules/mesh-delivery-builder/module.json

@@ -53,7 +53,8 @@
         "./mesh-delivery-builder"
       ],
       "env": {
-        "MESH_BROKER_FILE": "${dir:state}/broker"
+        "MESH_BROKER_FILE": "${dir:state}/broker",
+        "MESH_BUILDER_CAPACITY": "shanks=3,ace=2"
       },
       "user": "mesh-delivery-builder",
       "health": {