From b9911dc1c1350f45009fdccec20585d6a0caa858 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 17:26:10 +0100 Subject: [PATCH] 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() }()