Build puller: wake on a settings file write, retry a failed config follow at the next event, say a capacity of 0 and why

go/buildstep/capacity.go

@@ -22,20 +22,7 @@ func NodeCapacity(path, node string, logf func(string, ...any)) func() int {
 		}
 	}
 	return func() int {
-		if path == "" {
-			return 0
-		}
-		raw, err := os.ReadFile(path)
-		if err != nil {
-			say("the build capacity cannot be read, so %s pulls nothing: %v", node, err)
-			return 0
-		}
-		var s map[string]json.RawMessage
-		if err := json.Unmarshal(raw, &s); err != nil {
-			say("the settings file %s is not JSON, so %s pulls nothing: %v", path, node, err)
-			return 0
-		}
-		n, err := capacityOf(s[CapacitySetting])
+		n, err := ReadCapacity(path)
 		if err != nil {
 			say("%v, so %s pulls nothing", err, node)
 			return 0
@@ -44,6 +31,22 @@ func NodeCapacity(path, node string, logf func(string, ...any)) func() int {
 	}
 }
 
+// ReadCapacity is the capacity the settings file at path holds, and why it cannot be read when it cannot (then 0).
+func ReadCapacity(path string) (int, error) {
+	if path == "" {
+		return 0, nil
+	}
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return 0, fmt.Errorf("the build capacity cannot be read: %w", err)
+	}
+	var s map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &s); err != nil {
+		return 0, fmt.Errorf("the settings file %s is not JSON: %w", path, err)
+	}
+	return capacityOf(s[CapacitySetting])
+}
+
 // capacityOf is the capacity a config value holds: absent or null is 0; anything but a whole number of at least 0 is
 // refused, naming it.
 func capacityOf(value json.RawMessage) (int, error) {

go/buildstep/pull.go

@@ -98,6 +98,10 @@ type Puller struct {
 	// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
 	// does not take the asker's next copy at once. Nil has nothing.
 	Has func(need string) bool
+	// SettingsFile is the module's settings file. With it set the puller wakes whenever the file is written (a file event,
+	// WatchSettings), and reads Capacity from it itself, keeping why it cannot be read (CapacityNote); a parked puller at
+	// capacity 0 therefore wakes when the node-engine refills the file, and a failed config follow is tried again then.
+	SettingsFile string
 
 	mu      sync.Mutex
 	inHand  int
@@ -108,6 +112,14 @@ type Puller struct {
 	// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
 	watched    int
 	hasWatched bool
+	// capFollow and hoursFollow are the state of the two config follows: followed once WatchConfig took them, trying
+	// while a start is in flight; followFault is the last failure said, so each failure is said once.
+	capFollowed, capTrying, hoursFollowed, hoursTrying bool
+	followFault                                        string
+	// fileFault is why the settings file's capacity cannot be read, "" when it can.
+	fileFault string
+	// capWhy is why the config watch's last value could not be read, "" when it could.
+	capWhy string
 	// hours is the build window the config watch handed last, and hoursFault why its value cannot be read; used in place
 	// of BuildHours once hasHours.
 	hours      *BuildWindow
@@ -248,6 +260,21 @@ func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
 	return !p.paused && p.inHand < capacity, capacity, p.inHand
 }
 
+// CapacityNote says when the puller takes nothing for want of a capacity: why, and whether it is a fault (a setting that
+// cannot be read or is refused, so a person must look) and not a choice (a capacity set to 0). "" when it pulls.
+func (p *Puller) CapacityNote() (why string, fault bool) {
+	c := p.CapacityDetail()
+	if c.Effective > 0 {
+		return "", false
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.fileFault != "" && !p.hasWatched {
+		return "the " + p.Language + " build capacity of " + p.Node + " is 0 because it cannot be read: " + p.fileFault, true
+	}
+	return "the " + p.Language + " build capacity of " + p.Node + " is " + c.Why, p.hasWatched && p.capWhy != ""
+}
+
 // CapacityNow is the capacity the puller judges by (CapacityDetail's Effective).
 func (p *Puller) CapacityNow() int {
 	return p.CapacityDetail().Effective
@@ -265,6 +292,15 @@ func (p *Puller) CapacityDetail() delivery.Capacity {
 	switch {
 	case has:
 		configured, from = watched, "mesh-config, as the config watch handed it"
+	case p.SettingsFile != "":
+		n, err := ReadCapacity(p.SettingsFile)
+		configured = n
+		p.mu.Lock()
+		p.fileFault = ""
+		if err != nil {
+			p.fileFault = err.Error()
+		}
+		p.mu.Unlock()
 	case p.Capacity != nil:
 		configured = p.Capacity()
 	}
@@ -386,8 +422,8 @@ func (p *Puller) Pull(ctx context.Context) error {
 		// node kept from building is held by its capacity (0), not by a pause.
 		p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
 	}
-	go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
-	go p.watchHours()
+	p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
+	go p.watchSettings(ctx)
 	wait := p.PullWait
 	if wait <= 0 {
 		wait = DefaultPullWait
@@ -414,6 +450,7 @@ func (p *Puller) Pull(ctx context.Context) error {
 			case <-edge:
 			}
 			stop()
+			p.follow() // a follow that failed is tried again at the next event
 			_, 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
@@ -432,6 +469,7 @@ func (p *Puller) Pull(ctx context.Context) error {
 			case <-ctx.Done():
 				return ctx.Err()
 			case <-p.wake:
+				p.follow()
 			case <-time.After(wait):
 			}
 			continue
@@ -620,11 +658,18 @@ func (p *Puller) watchCapacity() {
 	if p.WatchConfig == nil {
 		return
 	}
+	p.mu.Lock()
+	if p.capFollowed || p.capTrying {
+		p.mu.Unlock()
+		return
+	}
+	p.capTrying = true
+	p.mu.Unlock()
 	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.watched, p.hasWatched, p.capWhy = n, true, why
 		p.mu.Unlock()
 		switch {
 		case why != "":
@@ -635,9 +680,31 @@ func (p *Puller) watchCapacity() {
 		p.Wake()
 		return nil
 	})
+	p.followed(&p.capFollowed, &p.capTrying, err, "build capacity")
+}
+
+// follow starts each config follow that is not running: at the start, and at every wake after (a file event, a resume, a
+// step's end), so a follow that failed is tried again by the next event and not given up for good.
+func (p *Puller) follow() {
+	go p.watchCapacity()
+	go p.watchHours()
+}
+
+// followed records the end of one follow's start. A failure is said once per distinct error, with that it is tried again.
+func (p *Puller) followed(done, trying *bool, err error, what string) {
+	p.mu.Lock()
+	*trying = false
+	*done = err == nil
+	said := err != nil && err.Error() == p.followFault
 	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)
+		p.followFault = err.Error()
+	} else {
+		p.followFault = ""
+	}
+	p.mu.Unlock()
+	if err != nil && !said {
+		p.logf("the %s %s of %s cannot be followed on the bus, so it is read from the settings file; the follow is tried "+
+			"again at the next event: %v", p.Language, what, p.Node, err)
 	}
 }
 
@@ -648,6 +715,13 @@ func (p *Puller) watchHours() {
 	if p.WatchConfig == nil || p.BuildHours == nil {
 		return
 	}
+	p.mu.Lock()
+	if p.hoursFollowed || p.hoursTrying {
+		p.mu.Unlock()
+		return
+	}
+	p.hoursTrying = true
+	p.mu.Unlock()
 	err := p.WatchConfig(BuildHoursSetting, func(c stdio.ProvisionChange) error {
 		setting := ""
 		var w *BuildWindow
@@ -672,9 +746,7 @@ func (p *Puller) watchHours() {
 		p.Wake()
 		return nil
 	})
-	if err != nil {
-		p.logf("the %s build hours of %s cannot be followed on the bus, so they are read from the settings file: %v", p.Language, p.Node, err)
-	}
+	p.followed(&p.hoursFollowed, &p.hoursTrying, err, "build hours")
 }
 
 // WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its

go/buildstep/pull_test.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -301,6 +303,47 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
 		w.set(t, "2")
 		waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
 	})
+	t.Run("settings refill wakes a parked puller and retries a failed follow", func(t *testing.T) {
+		// 2026-10-28: a replaced builder held the old "shanks=2,..." in its settings file, its config follow timed out, and
+		// the puller parked at 0 with nothing to wake it when the node-engine refilled the file.
+		src, st := NewFakeAsks(), newSteps()
+		_ = src.PublishAsk(testAsk(delivery.AskDelivery))
+		file := filepath.Join(t.TempDir(), "settings.json")
+		if err := os.WriteFile(file, []byte(`{"capacity":"shanks=2,ace=2"}`), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		var mu sync.Mutex
+		attempts := 0
+		watch := func(key string, handle func(stdio.ProvisionChange) error) error {
+			mu.Lock()
+			defer mu.Unlock()
+			attempts++
+			if attempts == 1 {
+				return errors.New("nats: timeout")
+			}
+			return nil // followed; no value handed yet
+		}
+		p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, PullWait: 10 * time.Millisecond,
+			SettingsFile: file, WatchConfig: 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 parked puller to say why", func() bool { why, fault := p.CapacityNote(); return fault && why != "" })
+		if src.Waiting(delivery.AskDelivery) != 1 {
+			t.Fatal("a node whose capacity cannot be read took an ask")
+		}
+		time.Sleep(50 * time.Millisecond) // let the watch start
+		tmp := file + ".new"
+		if err := os.WriteFile(tmp, []byte(`{"capacity":2}`), 0o600); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Rename(tmp, file); err != nil {
+			t.Fatal(err)
+		}
+		waitFor(t, "the ask taken once the file was refilled", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
+		waitFor(t, "the failed follow tried again", func() bool { mu.Lock(); defer mu.Unlock(); return attempts >= 2 })
+	})
 	for _, c := range []struct {
 		name  string
 		watch func(string, func(stdio.ProvisionChange) error) error

go/buildstep/watch_settings_linux.go

@@ -0,0 +1,58 @@
+//go:build linux
+
+package buildstep
+
+import (
+	"context"
+	"encoding/binary"
+	"os"
+	"path/filepath"
+	"syscall"
+)
+
+// watchSettings wakes the puller each time its settings file is written or replaced (inotify on the file's directory, so
+// a rename over it counts): the event the node-engine's refill of the file is, which a parked puller at capacity 0
+// otherwise never sees. No timer: it blocks on the kernel's event. A watch that cannot start is said once.
+func (p *Puller) watchSettings(ctx context.Context) {
+	if p.SettingsFile == "" {
+		return
+	}
+	fd, err := syscall.InotifyInit1(syscall.IN_CLOEXEC | syscall.IN_NONBLOCK)
+	if err != nil {
+		p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
+		return
+	}
+	f := os.NewFile(uintptr(fd), "inotify")
+	defer f.Close()
+	dir, base := filepath.Split(p.SettingsFile)
+	if _, err := syscall.InotifyAddWatch(fd, filepath.Clean(dir), syscall.IN_CLOSE_WRITE|syscall.IN_MOVED_TO|syscall.IN_CREATE); err != nil {
+		p.logf("the settings file %s cannot be watched, so a refill of it wakes the puller only at another event: %v", p.SettingsFile, err)
+		return
+	}
+	go func() { <-ctx.Done(); f.Close() }()
+	buf := make([]byte, 4096)
+	for {
+		n, err := f.Read(buf)
+		if err != nil {
+			return
+		}
+		hit := false
+		for off := 0; off+16 <= n; {
+			l := int(binary.LittleEndian.Uint32(buf[off+12:]))
+			name := buf[off+16 : off+16+l]
+			for i, c := range name {
+				if c == 0 {
+					name = name[:i]
+					break
+				}
+			}
+			if string(name) == base {
+				hit = true
+			}
+			off += 16 + l
+		}
+		if hit {
+			p.Wake()
+		}
+	}
+}

go/buildstep/watch_settings_other.go

@@ -0,0 +1,8 @@
+//go:build !linux
+
+package buildstep
+
+import "context"
+
+// watchSettings watches nothing off Linux: the puller wakes at its other events.
+func (p *Puller) watchSettings(ctx context.Context) {}