Show what the watchman says: offline presence, stuck keys and unreadable values, in the view and its command (issue 1168)

2026-10-19 00:55:19 +02:00
parent f032b384c9
commit 954728b390
8 changed files with 351 additions and 7 deletions
@@ -45,7 +45,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 	kvs := map[string]jetstream.KeyValue{}
-	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} {
+	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
 		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
 		if err != nil {
 			t.Fatal(err)
@@ -122,3 +122,89 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
 	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK))
 	rowsMatch(func(s view.Snapshot) bool { r := byKey(s, "a", "x"); return r.Phase == "running" && r.Health == "ok" })
 }
+
+// TestProcessWholeShowsWhatTheWatchmanSays starts the program whole and writes an offline presence and a stuck key to the
+// watchman's buckets, sees both shown, then writes them online and moved and sees both gone.
+func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5
+	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	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)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	kvs := map[string]jetstream.KeyValue{}
+	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[b] = kv
+	}
+	presence := func(key string, online bool) {
+		t.Helper()
+		b, _ := delivery.EncodePresence(delivery.Presence{Online: online})
+		if _, err := kvs[delivery.PresenceBucket].Put(ctx, key, b); err != nil {
+			t.Fatal(err)
+		}
+	}
+	stuck := func(moved bool) {
+		t.Helper()
+		b, _ := delivery.EncodeStuck(delivery.Stuck{Bucket: delivery.HaveBucket, Key: "a.x", Waited: "a report from a", Since: time.Now(), Moved: moved})
+		if _, err := kvs[delivery.StuckBucket].Put(ctx, delivery.HaveBucket+".a.x", b); err != nil {
+			t.Fatal(err)
+		}
+	}
+	presence("a", false)
+	presence("part.mesh-delivery-view.i1", false)
+	stuck(false)
+
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	socket := filepath.Join(dir, "run", "cli.sock")
+	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
+	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 })
+
+	wait := func(check func(view.Snapshot) bool) {
+		t.Helper()
+		var last view.Snapshot
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
+			if s, err := view.Ask(socket); err == nil {
+				last = s
+				if check(s) {
+					return
+				}
+			}
+		}
+		t.Fatalf("the socket never showed the expected lists; last: %+v", last)
+	}
+	wait(func(s view.Snapshot) bool { return len(s.Offline) == 2 && len(s.Stuck) == 1 && s.Stuck[0].Key == "a.x" })
+	presence("a", true)
+	presence("part.mesh-delivery-view.i1", true)
+	stuck(true)
+	wait(func(s view.Snapshot) bool { return len(s.Offline) == 0 && len(s.Stuck) == 0 })
+}
@@ -136,7 +136,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} {
+	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
 		go follow(ctx, js, bucket, model, logger)
 	}
 	select {