CAT-RD7: mesh-watcher asks the assigner's seat-holders once and follows its events, no reads
This commit is contained in:
@@ -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")
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Reference in New Issue
Block a user