diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 01f0ac3..51580bf 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -61,6 +61,21 @@ type fakeBuildModules struct { calls map[string]int // by `.` 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) + } +}