Wake a parked build puller on its capacity's own change, so a node started at capacity 0 takes asks once raised (issue 1887)

g14's Go builder sat idle at capacity 2 while delivery builds waited: a puller
with no room waited on its wake alone, which only a pause, a resume or a step's
end gives, and it read its capacity from the settings file only when woken.
The puller now follows the module's config value capacity through the
runtime's watch of what mesh-config resolved (stdio.WatchConfig), uses the
value it hands, and wakes on each change: no timer, no file watch, and no wait
for the node-engine to fill the file again. It says when it parks and when it
wakes, with its capacity.
This commit is contained in:
2026-10-25 17:23:28 +01:00
parent f9d11acf23
commit e6b2df6506
4 changed files with 186 additions and 12 deletions
+12 -3
View File
@@ -42,11 +42,20 @@ func NodeCapacity(path, fallback, node string, logf func(string, ...any)) func()
}
setting, _ = s[CapacitySetting].(string)
}
caps, err := buildrun.ParseCapacity(setting)
n, err := capacityOf(setting, node)
if err != nil {
say("%v, so %s pulls nothing", fmt.Errorf("the build capacity: %w", err), node)
say("%v, so %s pulls nothing", err, node)
return 0
}
return (&buildrun.Places{Capacity: caps}).CapacityOf(node)
return n
}
}
// capacityOf is node's capacity in a setting's `<node>=<asks>` pairs: DefaultCapacity for a node it does not name.
func capacityOf(setting, node string) (int, error) {
caps, err := buildrun.ParseCapacity(setting)
if err != nil {
return 0, fmt.Errorf("the build capacity: %w", err)
}
return (&buildrun.Places{Capacity: caps}).CapacityOf(node), nil
}
+82 -4
View File
@@ -9,6 +9,7 @@ import (
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -68,6 +69,9 @@ type Puller struct {
// node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives
// a restart.
Load func(key string) ([]byte, error)
// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
// for this node: stdio.WatchConfig in a module, a test's own double.
WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
mu sync.Mutex
inHand int
@@ -75,6 +79,9 @@ type Puller struct {
paused bool
wake chan struct{}
once sync.Once
// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
watched int
hasWatched bool
}
func (p *Puller) init() {
@@ -186,13 +193,29 @@ func (p *Puller) Running() []RunningAsk {
// room says whether the puller may take one more ask now.
func (p *Puller) room() bool {
capacity := 0
if p.Capacity != nil {
ok, _, _ := p.roomNow()
return ok
}
// roomNow is room, with the capacity and the asks in hand it was judged on.
func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
p.mu.Lock()
watched, has := p.watched, p.hasWatched
p.mu.Unlock()
if has {
capacity = watched
} else if p.Capacity != nil {
capacity = p.Capacity()
}
p.mu.Lock()
defer p.mu.Unlock()
return !p.paused && p.inHand < capacity
return !p.paused && p.inHand < capacity, capacity, p.inHand
}
// CapacityNow is the capacity the puller judges by: the config watch's last value, else Capacity.
func (p *Puller) CapacityNow() int {
_, capacity, _ := p.roomNow()
return capacity
}
// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
@@ -205,6 +228,7 @@ func (p *Puller) Pull(ctx context.Context) error {
if err := p.restore(); err != nil {
return err // pulling on a pause that cannot be read could build on a node paused on purpose
}
p.watchCapacity()
wait := p.PullWait
if wait <= 0 {
wait = DefaultPullWait
@@ -213,12 +237,17 @@ func (p *Puller) Pull(ctx context.Context) error {
if err := ctx.Err(); err != nil {
return err
}
if !p.room() {
if ok, capacity, inHand := p.roomNow(); !ok {
// Parked until something wakes it: a capacity change (the config watch), a resume, or a step's end. Said
// once each way, so a puller that sits idle says why (issue 1887).
p.logf("the %s puller on %s parks: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
select {
case <-ctx.Done():
return ctx.Err()
case <-p.wake:
}
_, capacity, inHand = p.roomNow()
p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
continue
}
msg, err := p.next(ctx, wait)
@@ -348,3 +377,52 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key
p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
}
}
// watchCapacity follows the module's config value `capacity` when WatchConfig is set (issue 1887, 2026-10-25: g14's
// puller started at capacity 0 sat parked with capacity 2 set, because nothing woke it to read its settings file
// again). The value comes as mesh-config resolved it for this node, the event of the change itself: the runtime hands
// the value now and then each change its machine's provisioner client says moved, so nothing polls, and the puller does
// not wait for the node-engine to fill its settings file again. Each value handed is kept, used in place of Capacity,
// and wakes the puller. A watch that cannot start leaves the puller on Capacity, said once.
func (p *Puller) watchCapacity() {
if p.WatchConfig == nil {
return
}
err := p.WatchConfig(CapacitySetting, func(c stdio.ProvisionChange) error {
n, why := p.capacityFrom(c)
p.mu.Lock()
before, had := p.watched, p.hasWatched
p.watched, p.hasWatched = n, true
p.mu.Unlock()
switch {
case why != "":
p.logf("the %s puller on %s pulls nothing: %s", p.Language, p.Node, why)
case !had || before != n:
p.logf("the %s build capacity of %s is %d, as mesh-config resolved it", p.Language, p.Node, n)
}
p.Wake()
return nil
})
if err != nil {
p.logf("the %s build capacity of %s cannot be followed on the bus, so it is read from the settings file when the "+
"puller wakes: %v", p.Language, p.Node, err)
}
}
// capacityFrom is the node's capacity in one value the watch handed, and why it is 0 when it cannot be read.
func (p *Puller) capacityFrom(c stdio.ProvisionChange) (int, string) {
if c.Refused != "" {
return 0, "its capacity is no longer answered: " + c.Refused
}
setting := ""
if len(c.Value) > 0 && string(c.Value) != "null" {
if err := json.Unmarshal(c.Value, &setting); err != nil {
return 0, fmt.Sprintf("its capacity %s is not a string", c.Value)
}
}
n, err := capacityOf(setting, p.Node)
if err != nil {
return 0, err.Error()
}
return n, ""
}
+91
View File
@@ -0,0 +1,91 @@
package buildstep
import (
"context"
"encoding/json"
"errors"
"sync"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// configWatch is a double of stdio.WatchConfig: it hands the value now at the start, as the runtime does, and a test
// hands each change after with set.
type configWatch struct {
mu sync.Mutex
value string
handle func(stdio.ProvisionChange) error
asked string
}
func (w *configWatch) watch(key string, handle func(stdio.ProvisionChange) error) error {
w.mu.Lock()
w.asked, w.handle = key, handle
value := w.value
w.mu.Unlock()
return handle(stdio.ProvisionChange{Kind: "config", Key: key, Value: json.RawMessage(`"` + value + `"`)})
}
func (w *configWatch) set(t *testing.T, value string) {
t.Helper()
w.mu.Lock()
handle, key := w.handle, w.asked
w.mu.Unlock()
if handle == nil {
t.Fatal("the puller watches no config value")
}
if err := handle(stdio.ProvisionChange{Kind: "config", Key: key, Value: json.RawMessage(`"` + value + `"`)}); err != nil {
t.Fatal(err)
}
}
// 1887/AC-wake (g14 idle at capacity 2, 2026-10-25): given a puller started at capacity 0 with an ask waiting, when the
// module's config value `capacity` moves to 2 for its node, then it takes the ask, with no pause, resume or step end to
// wake it.
func TestPullWakesWhenItsCapacityMoves(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
w := &configWatch{value: "shanks=3,g14=0"}
p := &Puller{Source: src, Language: delivery.AskGo, Node: "g14", Put: st.put, PullWait: 10 * time.Millisecond,
Capacity: func() int { return 0 }, // the settings file, not refilled yet
WatchConfig: w.watch,
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "the watch of the capacity", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.handle != nil })
time.Sleep(30 * time.Millisecond)
if src.Waiting(delivery.AskDelivery) != 1 {
t.Fatal("a node at capacity 0 took an ask")
}
if w.asked != CapacitySetting {
t.Fatalf("the puller watches %q, want %q", w.asked, CapacitySetting)
}
w.set(t, "shanks=3,g14=2")
waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
}
// 1887/AC-wake: a value the watch says is no longer answered pulls nothing; a watch that cannot start leaves the puller
// on its Capacity (the settings file).
func TestPullCapacityWatchRefusedOrUnstarted(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
p := &Puller{Source: src, Language: delivery.AskGo, Node: "g14", Put: st.put, Capacity: func() int { return 1 },
WatchConfig: func(key string, handle func(stdio.ProvisionChange) error) error {
return handle(stdio.ProvisionChange{Kind: "config", Key: key, Refused: "the module is no longer on g14"})
},
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
p.watchCapacity()
if p.room() {
t.Fatal("a refused capacity value left room")
}
q := &Puller{Source: src, Language: delivery.AskGo, Node: "g14", Put: st.put, Capacity: func() int { return 1 },
WatchConfig: func(string, func(stdio.ProvisionChange) error) error { return errors.New("no runtime") },
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
q.watchCapacity()
if !q.room() {
t.Fatal("a watch that could not start ignored the settings file's capacity of 1")
}
}
+1 -5
View File
@@ -14,11 +14,7 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
Run: func(map[string]any) (any, error) { return run() }}
}
current := func() (any, error) {
capacity := 0
if p.Capacity != nil {
capacity = p.Capacity()
}
return map[string]any{"node": p.Node, "capacity": capacity, "running": p.Running(), "paused": p.Paused()}, nil
return map[string]any{"node": p.Node, "capacity": p.CapacityNow(), "running": p.Running(), "paused": p.Paused()}, nil
}
return []stdio.Tool{
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+