Prove on the real bus that a builder started again while its asks waited puts no second ask (issue 1887 AC3, review round 2 item 2)

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go

@@ -61,6 +61,21 @@ type fakeBuildModules struct {
 	calls      map[string]int // by `<module>.<tool>`
 	fetches    []map[string]any
 	builds     []map[string]any // the arguments of every build call
+	// closed, while set, keeps every node's capacity at 0: no ask is pulled until open is called (a test of the asks
+	// waiting in their streams).
+	closed  atomic.Bool
+	pullers []*buildstep.Puller
+}
+
+// open lets the nodes pull again and wakes their pullers.
+func (f *fakeBuildModules) open() {
+	f.closed.Store(false)
+	f.mu.Lock()
+	ps := append([]*buildstep.Puller(nil), f.pullers...)
+	f.mu.Unlock()
+	for _, p := range ps {
+		p.Wake()
+	}
 }
 
 func (f *fakeBuildModules) count(tool string) int {
@@ -133,8 +148,16 @@ func (f *fakeBuildModules) serve(t *testing.T, module string, nodes ...string) {
 	steps := buildstep.AskSteps{Fetch: call("fetch"), Build: call("build"), Copy: call("copy")}
 	for _, node := range nodes {
 		p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: node, Put: put,
-			Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
+			Capacity: func() int {
+				if f.closed.Load() {
+					return 0
+				}
+				return 2
+			}, PullWait: 200 * time.Millisecond,
 			Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }}
+		f.mu.Lock()
+		f.pullers = append(f.pullers, p)
+		f.mu.Unlock()
 		go func() { _ = p.Pull(f.ctx) }()
 	}
 }
@@ -754,3 +777,41 @@ func (b *bed) asksPut(t *testing.T, nc *nats.Conn) func() []delivery.StepAsk {
 		return append([]delivery.StepAsk(nil), asks...)
 	}
 }
+
+// 1887/AC3: a builder that started again while its asks waited in the streams (no node had a free slot) puts no second
+// ask: it waits for the asks its building key names, and the key is done once a node took them.
+func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
+	b := startBed(t)
+	b.fake.closed.Store(true)
+	b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
+	sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.Close()
+	asked := b.asksPut(t, sub)
+	b.start(t, "n1", nil)
+	v := b.await(t, "netcheck", func(v delivery.BuilderBuild) bool { return v.State == delivery.BuildBuilding && len(v.Asks) == 2 })
+	for deadline := time.Now().Add(5 * time.Second); len(asked()) < 2 && time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) {
+	}
+	b.stop()
+	b.start(t, "n1", nil) // the same builder, started again, while both asks wait
+	time.Sleep(500 * time.Millisecond)
+	b.fake.open()
+	done := b.await(t, "netcheck", isState(delivery.BuildDone))
+	if got := asked(); len(got) != 2 {
+		t.Fatalf("asks put = %d (%+v), want the two the key named and none again", len(got), got)
+	}
+	for lang, id := range v.Asks {
+		found := false
+		for _, a := range asked() {
+			found = found || a.ID == id
+		}
+		if !found {
+			t.Errorf("the %s ask %s the key named was never put", lang, id)
+		}
+	}
+	if len(done.Artifacts) != 2 {
+		t.Errorf("done = %+v", done)
+	}
+}