Merge pull request 'Wake a parked build puller when its capacity changes, so a node raised from capacity 0 takes asks (issue 1887)' (#80) from fix/puller-wakes-on-capacity into main

Reviewed-on: #80

go/buildstep/capacity.go

@@ -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
+}

go/buildstep/pull.go

@@ -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,10 @@ 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, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
+	// double. Until it hands a value the puller judges by Capacity.
+	WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
 
 	mu      sync.Mutex
 	inHand  int
@@ -75,6 +80,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 +194,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 +229,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
 	}
+	go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
 	wait := p.PullWait
 	if wait <= 0 {
 		wait = DefaultPullWait
@@ -213,12 +238,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. Every
+			// park logs once, 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 +378,59 @@ 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)
+	}
+}
+
+// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
+// pull before it serves, and a watch asked before that would be refused.
+func WatchConfigServed(key string, handle func(stdio.ProvisionChange) error) error {
+	<-stdio.Served()
+	return stdio.WatchConfig(key, handle)
+}
+
+// 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, ""
+}

go/buildstep/pull_capacity_test.go

@@ -0,0 +1,39 @@
+package buildstep
+
+import (
+	"encoding/json"
+	"sync"
+	"testing"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// 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)
+	}
+}

go/buildstep/pull_test.go

@@ -3,11 +3,13 @@ package buildstep
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
@@ -175,27 +177,76 @@ func (k *kv) load(key string) ([]byte, error) {
 	return k.vals[key], nil
 }
 
-// 1887/A2b, 1887/AC-zero: at capacity 0 the puller takes nothing; a capacity raised takes effect at the next pull.
+// 1887/A2b, 1887/AC-zero, 1887/AC-wake: at capacity 0 the puller takes nothing; a capacity raised takes effect at the
+// next pull, or, with the config watch (g14 idle at capacity 2, 2026-10-25), at once when the value moves, with no
+// pause, resume or step end to wake it; a refused value pulls nothing, and a watch that cannot start leaves the
+// settings file's capacity.
 func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
-	src, st := NewFakeAsks(), &kv{}
-	_ = src.PublishAsk(testAsk(delivery.AskDelivery))
-	var mu sync.Mutex
-	capacity := 0
-	p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
-		Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
-		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)
-	time.Sleep(50 * time.Millisecond)
-	if src.Waiting(delivery.AskDelivery) != 1 {
-		t.Fatal("a node at capacity 0 took an ask")
+	t.Run("settings file", func(t *testing.T) {
+		src, st := NewFakeAsks(), &kv{}
+		_ = src.PublishAsk(testAsk(delivery.AskDelivery))
+		var mu sync.Mutex
+		capacity := 0
+		p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
+			Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
+			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)
+		time.Sleep(50 * time.Millisecond)
+		if src.Waiting(delivery.AskDelivery) != 1 {
+			t.Fatal("a node at capacity 0 took an ask")
+		}
+		mu.Lock()
+		capacity = 1
+		mu.Unlock()
+		p.Wake()
+		waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
+	})
+	t.Run("config watch wakes", func(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 })
+	})
+	for _, c := range []struct {
+		name  string
+		watch func(string, func(stdio.ProvisionChange) error) error
+		room  bool
+	}{
+		{"refused value leaves no room", 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"})
+		}, false},
+		{"watch that cannot start keeps the settings file", func(string, func(stdio.ProvisionChange) error) error {
+			return errors.New("no runtime")
+		}, true},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			p := &Puller{Source: NewFakeAsks(), Language: delivery.AskGo, Node: "g14", Put: newSteps().put,
+				Capacity: func() int { return 1 }, WatchConfig: c.watch,
+				Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
+			p.watchCapacity()
+			if p.room() != c.room {
+				t.Fatalf("room = %v, want %v", p.room(), c.room)
+			}
+		})
 	}
-	mu.Lock()
-	capacity = 1
-	mu.Unlock()
-	p.Wake()
-	waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
 }
 
 // 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state

go/buildstep/seat_tools.go

@@ -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, "+

go/stdio.go

@@ -154,8 +154,15 @@ const CodeChanged = -32010
 var (
 	current   *channel
 	currentMu sync.Mutex
+	// served is closed once Serve has the runtime's channel (Served).
+	served     = make(chan struct{})
+	servedOnce sync.Once
 )
 
+// Served is closed once Serve has the runtime's channel, so code started before Serve can wait for it instead of
+// asking again until the runtime takes it: a watch of a config value, a subscribe.
+func Served() <-chan struct{} { return served }
+
 // Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the
 // module the runtime named in MESH_SERVED_MODULE.
 func Serve(name string, tools []Tool) error {
@@ -180,6 +187,7 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
 	currentMu.Lock()
 	current = c
 	currentMu.Unlock()
+	servedOnce.Do(func() { close(served) })
 	// What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so
 	// it is asked from beside the loop that reads the answer.
 	go func() { _ = announce1() }()

go/stdio_test.go

@@ -16,6 +16,22 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
 	t.Setenv(ServedAs, "chime")
 	t.Setenv("MESH_MODULE", "chime")
 	t.Setenv("MESH_NODE", "anchor")
+	// 1887/AC-wake: Served closes once Serve has the runtime's channel, not before; the globals are put back after.
+	oldServed := served
+	served, servedOnce = make(chan struct{}), sync.Once{}
+	t.Cleanup(func() {
+		served, servedOnce = oldServed, sync.Once{}
+		select {
+		case <-oldServed: // it was closed: its Once is spent too
+			servedOnce.Do(func() {})
+		default:
+		}
+	})
+	select {
+	case <-Served():
+		t.Fatal("Served closed before Serve")
+	default:
+	}
 	inR, inW := io.Pipe()
 	outR, outW := io.Pipe()
 	go func() {
@@ -50,6 +66,11 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
 	if got := next()["result"].(map[string]any)["serverInfo"].(map[string]any)["name"]; got != "chime" {
 		t.Fatalf("served as %v, want the module the runtime named", got)
 	}
+	select {
+	case <-Served():
+	case <-time.After(2 * time.Second):
+		t.Fatal("Served did not close once serving")
+	}
 	send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "ring", "arguments": map[string]any{}}})
 	asked := next()
 	if asked["method"] != "mesh/publish" {