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

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

@@ -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 })
+}

modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go

@@ -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 {

modules/mesh-delivery-view/internal/view/serve.go

@@ -143,13 +143,44 @@ func Table(w io.Writer, s Snapshot) {
 			row([]string{n.Node, fmt.Sprint(n.Windows), fmt.Sprint(n.Busy), fmt.Sprint(n.Strays), since(n.Since)})
 		}
 	}
+	if len(s.Offline) > 0 {
+		fmt.Fprintln(tw)
+		row([]string{"OFFLINE", "NAME", "STATE", "SINCE"})
+		for _, o := range s.Offline {
+			row([]string{o.Kind, o.name(), "offline", since(o.Since)})
+		}
+	}
+	if len(s.Stuck) > 0 {
+		fmt.Fprintln(tw)
+		row([]string{"STUCK", "KEY", "WAITED FOR", "SINCE"})
+		for _, k := range s.Stuck {
+			row([]string{k.Bucket, k.Key, k.Waited, since(k.Since)})
+		}
+	}
+	if len(s.Unreadable) > 0 {
+		fmt.Fprintln(tw)
+		row([]string{"UNREADABLE", "KEY", "WHY"})
+		for _, u := range s.Unreadable {
+			row([]string{u.Bucket, u.Key, u.Error})
+		}
+	}
 	tw.Flush()
 }
 
+func (o OfflineRow) name() string {
+	if o.Kind == "part" {
+		return o.Part + "." + o.Instance
+	}
+	return o.Node
+}
+
 var page = template.Must(template.New("p").Parse(`<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="5">
 <title>Delivery</title><style>body{font:14px sans-serif;margin:1em}table{border-collapse:collapse}td,th{padding:2px 10px;text-align:left}</style></head>
 <body><h1>What each node runs</h1><table><tr>{{range .Head}}<th>{{.}}</th>{{end}}</tr>
-{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table></body></html>`))
+{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
+{{if .Offline}}<h2>OFFLINE</h2><table>{{range .Offline}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
+{{if .Stuck}}<h2>STUCK</h2><table><tr><th>BUCKET</th><th>KEY</th><th>WAITED FOR</th><th>SINCE</th></tr>{{range .Stuck}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
+{{if .Unreadable}}<h2>UNREADABLE</h2><table>{{range .Unreadable}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}</body></html>`))
 
 // Web is the read-only web view of the same rows, refreshed by the browser.
 func Web(m *Model) http.Handler {
@@ -168,6 +199,17 @@ func Web(m *Model) http.Handler {
 			rows = append(rows, rowText(row))
 		}
 		w.Header().Set("Content-Type", "text/html; charset=utf-8")
-		page.Execute(w, map[string]any{"Head": head, "Rows": rows})
+		snap := m.Snapshot()
+		var off, stuck, bad [][]string
+		for _, o := range snap.Offline {
+			off = append(off, []string{o.Kind, o.name(), "offline", since(o.Since)})
+		}
+		for _, k := range snap.Stuck {
+			stuck = append(stuck, []string{k.Bucket, k.Key, k.Waited, since(k.Since)})
+		}
+		for _, u := range snap.Unreadable {
+			bad = append(bad, []string{u.Bucket, u.Key, u.Error})
+		}
+		page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Offline": off, "Stuck": stuck, "Unreadable": bad})
 	})
 }

modules/mesh-delivery-view/internal/view/serve_test.go

@@ -110,3 +110,44 @@ func TestWebIsTheSameRowsAndReadOnly(t *testing.T) { // 1134/AC6
 		t.Fatalf("POST = %d, want 405", rec.Code)
 	}
 }
+
+func watchmanSnapshot(t *testing.T) *Model {
+	t.Helper()
+	m := New()
+	fill(m, presenceE(t, "a", false, t0), stuckE(t, "node-engine_have", "a.x", "a report from a", t0, false),
+		Entry{Bucket: delivery.PresenceBucket, Key: "c", Value: []byte(`{"v":9}`), At: t0})
+	return m
+}
+
+func TestTableAndWebShowTheWatchmansLists(t *testing.T) { // 1168/AC4
+	m := watchmanSnapshot(t)
+	var out strings.Builder
+	Table(&out, m.Snapshot())
+	rec := httptest.NewRecorder()
+	Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
+	for name, text := range map[string]string{"table": out.String(), "web": rec.Body.String()} {
+		for _, w := range []string{"OFFLINE", "STUCK", "UNREADABLE", "node-engine_have", "a.x", "a report from a", "offline", "c"} {
+			if !strings.Contains(text, w) {
+				t.Errorf("%s lacks %q:\n%s", name, w, text)
+			}
+		}
+	}
+}
+
+func TestSocketCarriesTheWatchmansLists(t *testing.T) { // 1168/AC4
+	m := watchmanSnapshot(t)
+	dir, _ := os.MkdirTemp(os.Getenv("TMPDIR"), "w")
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	path := filepath.Join(dir, "cli.sock")
+	l, err := ListenSocket(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go ServeSocket(ctx, l, m)
+	got, err := Ask(path)
+	if err != nil || len(got.Offline) != 1 || len(got.Stuck) != 1 || len(got.Unreadable) != 1 {
+		t.Fatalf("socket = %+v, %v", got, err)
+	}
+}

modules/mesh-delivery-view/internal/view/view.go

@@ -51,6 +51,36 @@ type NodeRow struct {
 type Snapshot struct {
 	Rows  []Row     `json:"rows"`
 	Nodes []NodeRow `json:"nodes"`
+	// Offline is every node and part instance whose presence is offline (the watchman's word).
+	Offline []OfflineRow `json:"offline"`
+	// Stuck is every key that stopped moving and has not moved again.
+	Stuck []StuckRow `json:"stuck"`
+	// Unreadable is every watchman value the view could not read, named by its key: never dropped silently.
+	Unreadable []UnreadableRow `json:"unreadable"`
+}
+
+// OfflineRow is a node ("node") or a part's instance ("part") that is offline, since the presence entry's time.
+type OfflineRow struct {
+	Kind     string    `json:"kind"`
+	Node     string    `json:"node,omitempty"`
+	Part     string    `json:"part,omitempty"`
+	Instance string    `json:"instance,omitempty"`
+	Since    time.Time `json:"since,omitzero"`
+}
+
+// StuckRow is a key that stopped moving: its bucket and key, what it waited for and since when.
+type StuckRow struct {
+	Bucket string    `json:"bucket"`
+	Key    string    `json:"key"`
+	Waited string    `json:"waited,omitempty"`
+	Since  time.Time `json:"since,omitzero"`
+}
+
+// UnreadableRow is a value of the watchman's buckets that did not decode, with the reader's error.
+type UnreadableRow struct {
+	Bucket string `json:"bucket"`
+	Key    string `json:"key"`
+	Error  string `json:"error"`
 }
 
 // Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values.
@@ -61,6 +91,20 @@ type Model struct {
 	newest  map[string]string // module -> id
 	haves   map[string]haveEntry
 	nodes   map[string]nodeEntry
+	// the watchman's buckets, by key
+	presence map[string]presenceEntry
+	stuck    map[string]stuckEntry
+}
+
+type presenceEntry struct {
+	value delivery.Presence
+	err   string
+	at    time.Time
+}
+
+type stuckEntry struct {
+	value delivery.Stuck
+	err   string
 }
 
 // assignment holds only the fields of the controller's Assignment the view reads.
@@ -107,7 +151,8 @@ type nodeEntry struct {
 // New is an empty model.
 func New() *Model {
 	return &Model{assigns: map[string]assignment{}, builds: map[string]build{}, newest: map[string]string{},
-		haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}}
+		haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
+		presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
 }
 
 // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed.
@@ -136,6 +181,20 @@ func (m *Model) Apply(e Entry) {
 		}
 		h, err := delivery.DecodeHave(e.Value)
 		m.haves[key] = haveEntry{have: h, err: errText(err), at: e.At}
+	case delivery.PresenceBucket:
+		if e.Deleted {
+			delete(m.presence, e.Key)
+			return
+		}
+		p, err := delivery.DecodePresence(e.Value)
+		m.presence[e.Key] = presenceEntry{value: p, err: errText(err), at: e.At}
+	case delivery.StuckBucket:
+		if e.Deleted {
+			delete(m.stuck, e.Key)
+			return
+		}
+		st, err := delivery.DecodeStuck(e.Value)
+		m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)}
 	case AssignmentsBucket:
 		if e.Deleted {
 			delete(m.assigns, e.Key)
@@ -238,7 +297,47 @@ func (m *Model) Snapshot() Snapshot {
 			Strays: count(n.node.Strays), Since: n.at, Error: n.err})
 	}
 	sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node })
-	return Snapshot{Rows: rows, Nodes: nodes}
+	offline, stuck, unreadable := []OfflineRow{}, []StuckRow{}, []UnreadableRow{}
+	for key, p := range m.presence {
+		if p.err != "" {
+			unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: p.err})
+			continue
+		}
+		if p.value.Online {
+			continue
+		}
+		node, part, instance, err := delivery.ParsePresenceKey(key)
+		if err != nil {
+			unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: err.Error()})
+			continue
+		}
+		if part != "" {
+			offline = append(offline, OfflineRow{Kind: "part", Part: part, Instance: instance, Since: p.at})
+		} else {
+			offline = append(offline, OfflineRow{Kind: "node", Node: node, Since: p.at})
+		}
+	}
+	for key, st := range m.stuck {
+		if st.err != "" {
+			unreadable = append(unreadable, UnreadableRow{Bucket: delivery.StuckBucket, Key: key, Error: st.err})
+			continue
+		}
+		if st.value.Moved {
+			continue
+		}
+		stuck = append(stuck, StuckRow{Bucket: st.value.Bucket, Key: st.value.Key, Waited: st.value.Waited, Since: st.value.Since})
+	}
+	sort.Slice(offline, func(i, j int) bool {
+		a, b := offline[i], offline[j]
+		return a.Kind+a.Node+a.Part+a.Instance < b.Kind+b.Node+b.Part+b.Instance
+	})
+	sort.Slice(stuck, func(i, j int) bool {
+		return stuck[i].Bucket+"."+stuck[i].Key < stuck[j].Bucket+"."+stuck[j].Key
+	})
+	sort.Slice(unreadable, func(i, j int) bool {
+		return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key
+	})
+	return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable}
 }
 
 // count is the number of items of a carried JSON array or object, and 0 for anything else.

modules/mesh-delivery-view/internal/view/view_test.go

@@ -220,3 +220,77 @@ func TestNodeRow(t *testing.T) { // 1134/AC1 (the node's own key gives a node ro
 		t.Fatalf("nodes = %+v", n)
 	}
 }
+
+func presenceE(t *testing.T, key string, online bool, at time.Time) Entry {
+	t.Helper()
+	b, err := delivery.EncodePresence(delivery.Presence{Online: online})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return Entry{Bucket: delivery.PresenceBucket, Key: key, Value: b, At: at}
+}
+
+func stuckE(t *testing.T, bucket, key, waited string, since time.Time, moved bool) Entry {
+	t.Helper()
+	b, err := delivery.EncodeStuck(delivery.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since, Moved: moved})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return Entry{Bucket: delivery.StuckBucket, Key: bucket + "." + key, Value: b, At: since}
+}
+
+func TestOfflineAndStuckAreListed(t *testing.T) { // 1168/AC2
+	m := New()
+	fill(m, base(t)...)
+	fill(m,
+		presenceE(t, "a", false, t0),
+		presenceE(t, "b", true, t0),
+		presenceE(t, "part.mesh-delivery-view.i1", false, t0.Add(time.Minute)),
+		stuckE(t, "node-engine_have", "a.x", "a report from a", t0.Add(2*time.Minute), false),
+		stuckE(t, "node-engine_have", "b.x", "a report from b", t0, true),
+	)
+	s := m.Snapshot()
+	want := []OfflineRow{{Kind: "node", Node: "a", Since: t0}, {Kind: "part", Part: "mesh-delivery-view", Instance: "i1", Since: t0.Add(time.Minute)}}
+	if !reflect.DeepEqual(s.Offline, want) {
+		t.Fatalf("offline = %+v, want %+v", s.Offline, want)
+	}
+	wantStuck := []StuckRow{{Bucket: "node-engine_have", Key: "a.x", Waited: "a report from a", Since: t0.Add(2 * time.Minute)}}
+	if !reflect.DeepEqual(s.Stuck, wantStuck) {
+		t.Fatalf("stuck = %+v, want %+v", s.Stuck, wantStuck)
+	}
+	// back online and moved leave the lists
+	fill(m, presenceE(t, "a", true, t0.Add(5*time.Minute)), presenceE(t, "part.mesh-delivery-view.i1", true, t0.Add(5*time.Minute)),
+		stuckE(t, "node-engine_have", "a.x", "a report from a", t0.Add(2*time.Minute), true))
+	s = m.Snapshot()
+	if len(s.Offline) != 0 || len(s.Stuck) != 0 {
+		t.Fatalf("offline %+v stuck %+v, want both empty", s.Offline, s.Stuck)
+	}
+}
+
+func TestUnknownWatchmanVersionIsUnreadableNotDropped(t *testing.T) { // 1168/AC3
+	m := New()
+	fill(m,
+		Entry{Bucket: delivery.PresenceBucket, Key: "c", Value: []byte(`{"v":9,"online":false}`), At: t0},
+		Entry{Bucket: delivery.StuckBucket, Key: "node-engine_have.c.x", Value: []byte(`{"v":9}`), At: t0},
+		presenceE(t, "a", false, t0),
+	)
+	s := m.Snapshot()
+	if len(s.Unreadable) != 2 || len(s.Offline) != 1 {
+		t.Fatalf("unreadable %+v offline %+v", s.Unreadable, s.Offline)
+	}
+	_, sdkErr := delivery.DecodePresence([]byte(`{"v":9,"online":false}`))
+	var got *UnreadableRow
+	for i := range s.Unreadable {
+		if s.Unreadable[i].Key == "c" {
+			got = &s.Unreadable[i]
+		}
+	}
+	if got == nil || got.Bucket != delivery.PresenceBucket || got.Error != sdkErr.Error() {
+		t.Fatalf("unreadable = %+v, want key c with the SDK's error %q", s.Unreadable, sdkErr)
+	}
+	// a good value later replaces the unreadable one
+	fill(m, presenceE(t, "c", true, t0))
+	if len(m.Snapshot().Unreadable) != 1 {
+		t.Fatal("a readable value must clear the unreadable mark")
+	}
+}

modules/mesh-delivery-view/manifest_test.go

@@ -74,7 +74,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	if m.Module != "mesh-delivery-view" {
 		t.Errorf("module = %q", m.Module)
 	}
-	want := []string{"mesh-controller.assignments", "mesh-controller.builds", "node-engine.have"}
+	want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // 1168/AC1
 	got := slices.Clone(m.Reads)
 	slices.Sort(got)
 	if !reflect.DeepEqual(got, want) {

modules/mesh-delivery-view/module.json

@@ -6,6 +6,8 @@
   "reads": [
     "mesh-controller.assignments",
     "mesh-controller.builds",
+    "mesh-delivery-watchman.presence",
+    "mesh-delivery-watchman.stuck",
     "node-engine.have"
   ],
   "own-secrets": {