diff --git a/go/buildstep/capacity.go b/go/buildstep/capacity.go index 4e041a4..5b4fad5 100644 --- a/go/buildstep/capacity.go +++ b/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) { diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 80c0a02..0474145 100644 --- a/go/buildstep/pull.go +++ b/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 diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 165e673..30c6b66 100644 --- a/go/buildstep/pull_test.go +++ b/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 diff --git a/go/buildstep/watch_settings_linux.go b/go/buildstep/watch_settings_linux.go new file mode 100644 index 0000000..f9d776d --- /dev/null +++ b/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() + } + } +} diff --git a/go/buildstep/watch_settings_other.go b/go/buildstep/watch_settings_other.go new file mode 100644 index 0000000..932ad97 --- /dev/null +++ b/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) {}