Test that an ask running when the watchman starts gets its pulse from the steps that stand (issue 1909)

2026-10-23 16:34:50 +02:00
parent e927469460
commit 76bc810bda
1 changed files with 81 additions and 0 deletions
@@ -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)
+}