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

@@ -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 != "" {

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")
+	}
+}

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() }()