diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index a7b2943..5d56874 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -221,3 +221,84 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 } } } + +// TestAnAskRunningBeforeTheStartGetsItsPulse puts a running step in the build module's steps before the watchman starts, and +// sees the pulse of its ask once it is up: what stood at the start is mirrored as what happens later is (issue 1909). +func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "w") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + conf := filepath.Join(dir, "bus.conf") + if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatal(err) + } + opts.NoLog, opts.NoSigs = true, true + srv, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(ctl.Close) + js, err := jetstream.New(ctl) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + kvs := map[string]jetstream.KeyValue{} + for _, c := range []jetstream.KeyValueConfig{ + {Bucket: delivery.PresenceBucket, History: 20}, + {Bucket: delivery.StuckBucket, History: 20}, + {Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second}, + {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, + } { + kv, err := js.CreateKeyValue(ctx, c) + if err != nil { + t.Fatal(err) + } + kvs[c.Bucket] = kv + } + const id = "m0000009ffffff" + body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()}) + if err != nil { + t.Fatal(err) + } + if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil { + t.Fatal(err) + } + broker := filepath.Join(dir, "broker") + cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { stop(); <-done }) + stream, _ := delivery.AskStream(delivery.AskGo) + pk, err := delivery.PulseKey(stream, id) + if err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if _, err := kvs[delivery.PulseBucket].Get(ctx, pk); err == nil { + return + } + } + t.Fatalf("no pulse for the ask running before the start (%s)", pk) +}