diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index 3ff9167f..89b8796f 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -7,6 +7,7 @@ package main import ( + "context" "encoding/json" "fmt" "os" @@ -17,6 +18,8 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-delivery/sdk/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -70,8 +73,10 @@ func main() { node := os.Getenv("MESH_NODE") w := NewWatcher(tg, time.Now, logf, node) l := &listening{now: "not yet: starting"} - go run(w, l, node) - if err := stdio.Serve("", tools(w, l)); err != nil { + placement := &views.Reader{Module: "mesh-watcher", Logf: logf} + placement.Follow(delivery.AssignerModule, delivery.SeatHoldersState, SeatHoldersTool) + go run(w, l, node, placement) + if err := stdio.Serve("", append(tools(w, l), placement.Tools("watcher")...)); err != nil { logf("%v", err) os.Exit(1) } @@ -79,7 +84,7 @@ func main() { // run keeps time, makes the round trips, reads where the controller is, and listens for heartbeats. // The clock runs whatever the bus does: it is what notices the bus is gone. -func run(w *Watcher, l *listening, node string) { +func run(w *Watcher, l *listening, node string, placement *views.Reader) { if err := w.Telegram.Ready(); err != nil { logf("[mesh-watcher] BLIND until given: %v", err) } @@ -89,17 +94,17 @@ func run(w *Watcher, l *listening, node string) { w.Tick() } }() - go func() { - time.Sleep(2 * time.Second) - for { - if nodes, err := controlNodes(); err == nil { - w.Placed(nodes) - } else { - logf("[mesh-watcher] cannot read where the bus runs (%v); asking again in ten minutes", err) - } - time.Sleep(10 * time.Minute) + // Where the bus runs: the assigner's seat-holders, asked once and then followed by its events (ADR 0615), never read + // from its state and never on a timer. + placed := func() { + if nodes, err := controlNodes(placement); err == nil { + w.Placed(nodes) + } else { + logf("[mesh-watcher] cannot read where the bus runs (%v)", err) } - }() + } + placement.Changed = func(string, string, json.RawMessage, bool) { placed() } + placement.Start(context.Background()) handle := func(e stdio.Envelope) error { if !isHeartbeat(e.Key) { return nil @@ -146,24 +151,17 @@ func roundTrip(node string) error { } } +// SeatHoldersTool is the assigner's read tool of the state of who holds which seat, which the manifest invokes; the +// state is followed as a view (ADR 0615), named as the manifest named it before: SeatHoldersRead. +const SeatHoldersTool = delivery.AssignerModule + ".assigner_read_seat_holders" + // controlNodes reads which machines hold the bus, from the placement seat's state of who holds which seat (novox/hq // ADR 0571 clause 4). The controller is not placed (ADR 0590), so the bus's machines are the control node. -func controlNodes() ([]string, error) { - st := stdio.State(delivery.SeatHoldersRead) - names, err := st.Keys() +func controlNodes(seats *views.Reader) ([]string, error) { + keys, err := seats.Values(delivery.SeatHoldersRead) if err != nil { return nil, err } - keys := map[string][]byte{} - for _, k := range names { - e, err := st.Get(k) - if err != nil { - return nil, err - } - if e != nil { - keys[k] = e.Value - } - } return holderNodes(keys, "mesh-broker") } diff --git a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go index 14107f86..4ef52ac0 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go @@ -9,6 +9,8 @@ import ( "strings" "testing" "time" + + "git.novox.be/novox/mesh-delivery/sdk/views" ) // The manifest says what the code does: it consumes the heartbeat and nothing else, calls its own @@ -24,6 +26,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { Module string `json:"module"` Consumes []string `json:"consumes"` Invokes []string `json:"invokes"` + Reads []string `json:"reads"` Claims []any `json:"claims"` Own map[string]string `json:"own-secrets"` Tools []string `json:"tools"` @@ -34,11 +37,12 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - if !reflect.DeepEqual(m.Consumes, []string{"mesh-self-check.checked"}) { + // Where the bus runs is the assigner's seat-holders, asked once and followed by its events (ADR 0615), never read. + if !reflect.DeepEqual(m.Consumes, []string{"mesh-self-check.checked", "mesh-delivery-assigner.views.seat-holders"}) { t.Fatalf("consumes %v", m.Consumes) } - if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping"}) { - t.Fatalf("invokes %v", m.Invokes) + if !reflect.DeepEqual(m.Invokes, []string{SeatHoldersTool, m.Module + ".watcher_ping"}) || len(m.Reads) != 0 { + t.Fatalf("invokes %v, reads %v", m.Invokes, m.Reads) } if len(m.Claims) != 0 { t.Fatalf("a watcher holds no seat: %v", m.Claims) @@ -48,7 +52,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { t.Fatalf("token: own %v, env %v", m.Own, env) } var served []string - for _, tool := range tools(NewWatcher(&fakeTelegram{}, time.Now, t.Logf, ""), &listening{}) { + for _, tool := range append(tools(NewWatcher(&fakeTelegram{}, time.Now, t.Logf, ""), &listening{}), (&views.Reader{}).Tools("watcher")...) { served = append(served, tool.Name) if strings.TrimSpace(tool.Description) == "" { t.Errorf("%s has no description", tool.Name) diff --git a/modules/mesh-watcher/module.json b/modules/mesh-watcher/module.json index f4163457..39cb6c2a 100644 --- a/modules/mesh-watcher/module.json +++ b/modules/mesh-watcher/module.json @@ -3,12 +3,11 @@ "version": "1", "slug": "watch", "consumes": [ - "mesh-self-check.checked" - ], - "reads": [ - "mesh-delivery-assigner.seat-holders" + "mesh-self-check.checked", + "mesh-delivery-assigner.views.seat-holders" ], "invokes": [ + "mesh-delivery-assigner.assigner_read_seat_holders", "mesh-watcher.watcher_ping" ], "own-secrets": { @@ -18,7 +17,9 @@ "watcher_status", "watcher_last_heard", "watcher_test", - "watcher_ping" + "watcher_ping", + "watcher_first_views", + "watcher_first_view_again" ], "config": { "telegram-chat-id": {