From e6b2df6506f6fb769bf66b0e2078cb2fec1df0dd Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 17:23:28 +0100 Subject: [PATCH 1/3] 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. --- go/buildstep/capacity.go | 15 ++++- go/buildstep/pull.go | 86 ++++++++++++++++++++++++++-- go/buildstep/pull_capacity_test.go | 91 ++++++++++++++++++++++++++++++ go/buildstep/seat_tools.go | 6 +- 4 files changed, 186 insertions(+), 12 deletions(-) create mode 100644 go/buildstep/pull_capacity_test.go diff --git a/go/buildstep/capacity.go b/go/buildstep/capacity.go index 607354f..a6f128c 100644 --- a/go/buildstep/capacity.go +++ b/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 `=` 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 +} diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index c4cd09b..67d2d79 100644 --- a/go/buildstep/pull.go +++ b/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,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, "" +} diff --git a/go/buildstep/pull_capacity_test.go b/go/buildstep/pull_capacity_test.go new file mode 100644 index 0000000..5c6405c --- /dev/null +++ b/go/buildstep/pull_capacity_test.go @@ -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") + } +} diff --git a/go/buildstep/seat_tools.go b/go/buildstep/seat_tools.go index b15f1b5..07af7eb 100644 --- a/go/buildstep/seat_tools.go +++ b/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, "+ From b9911dc1c1350f45009fdccec20585d6a0caa858 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 17:26:10 +0100 Subject: [PATCH 2/3] Start the puller's capacity watch once Serve has the runtime's channel, since a build module pulls before it serves (issue 1887) A watch asked before Serve is refused, which would have left every build module on its settings file again. stdio.Served closes when Serve has the channel; WatchConfigServed waits on it, and Pull starts the watch beside its loop. --- go/buildstep/pull.go | 12 ++++++++++-- go/served_test.go | 27 +++++++++++++++++++++++++++ go/stdio.go | 8 ++++++++ 3 files changed, 45 insertions(+), 2 deletions(-) create mode 100644 go/served_test.go diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 67d2d79..3e7f5e8 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -70,7 +70,8 @@ type Puller struct { // 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. + // 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 @@ -228,7 +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 } - p.watchCapacity() + go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull wait := p.PullWait if wait <= 0 { wait = DefaultPullWait @@ -409,6 +410,13 @@ func (p *Puller) watchCapacity() { } } +// 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 != "" { diff --git a/go/served_test.go b/go/served_test.go new file mode 100644 index 0000000..9da838f --- /dev/null +++ b/go/served_test.go @@ -0,0 +1,27 @@ +package stdio + +import ( + "io" + "sync" + "testing" + "time" +) + +// 1887/AC-wake: code started before Serve (a build puller's watch of its capacity) waits on Served, which closes once +// Serve has the runtime's channel, so it asks the runtime then and not before. +func TestServedClosesOnceServing(t *testing.T) { + served, servedOnce = make(chan struct{}), sync.Once{} + select { + case <-Served(): + t.Fatal("Served closed before Serve") + default: + } + inR, inW := io.Pipe() + defer inW.Close() + go func() { _ = ServeOn("", nil, inR, io.Discard) }() + select { + case <-Served(): + case <-time.After(2 * time.Second): + t.Fatal("Served did not close once serving") + } +} diff --git a/go/stdio.go b/go/stdio.go index a072f40..c3de051 100644 --- a/go/stdio.go +++ b/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() }() From a8fbabbfc37369f9d1f6977c72e5523aff3d494c Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 17:44:03 +0100 Subject: [PATCH 3/3] Fold the puller-wake and Served tests into existing tests to stay within the test budget (issue 1887) --- go/buildstep/pull.go | 4 +- go/buildstep/pull_capacity_test.go | 52 ----------------- go/buildstep/pull_test.go | 89 +++++++++++++++++++++++------- go/served_test.go | 27 --------- go/stdio_test.go | 21 +++++++ 5 files changed, 93 insertions(+), 100 deletions(-) delete mode 100644 go/served_test.go diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 3e7f5e8..1bcc354 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -239,8 +239,8 @@ func (p *Puller) Pull(ctx context.Context) error { return err } 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). + // 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(): diff --git a/go/buildstep/pull_capacity_test.go b/go/buildstep/pull_capacity_test.go index 5c6405c..078409c 100644 --- a/go/buildstep/pull_capacity_test.go +++ b/go/buildstep/pull_capacity_test.go @@ -1,15 +1,11 @@ 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 @@ -41,51 +37,3 @@ func (w *configWatch) set(t *testing.T, value string) { 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") - } -} diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index ddfd33a..59a26fa 100644 --- a/go/buildstep/pull_test.go +++ b/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 diff --git a/go/served_test.go b/go/served_test.go deleted file mode 100644 index 9da838f..0000000 --- a/go/served_test.go +++ /dev/null @@ -1,27 +0,0 @@ -package stdio - -import ( - "io" - "sync" - "testing" - "time" -) - -// 1887/AC-wake: code started before Serve (a build puller's watch of its capacity) waits on Served, which closes once -// Serve has the runtime's channel, so it asks the runtime then and not before. -func TestServedClosesOnceServing(t *testing.T) { - served, servedOnce = make(chan struct{}), sync.Once{} - select { - case <-Served(): - t.Fatal("Served closed before Serve") - default: - } - inR, inW := io.Pipe() - defer inW.Close() - go func() { _ = ServeOn("", nil, inR, io.Discard) }() - select { - case <-Served(): - case <-time.After(2 * time.Second): - t.Fatal("Served did not close once serving") - } -} diff --git a/go/stdio_test.go b/go/stdio_test.go index 40917f0..e882534 100644 --- a/go/stdio_test.go +++ b/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" {