From da787ae36048f130a9c561e683a5a438aedb9e65 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:44:18 +0100 Subject: [PATCH] Watchman: hold mesh-watchdogs and write its watchdogs as findings, so health can leave the controller (issue 1952 WM1-WM5) The signals rows S1, S8, S9, S10, S11, S21 and the haves' unhealthy and needs-operator are pulses whose expiry is a finding; stuck keys and presence are its own findings; it keeps the dead letters from the lease write on its own consumer, lists and shows them, and refuses deliver and drop until the hand-act log's home (issue 1953). Tools read every state it keeps and start resync. Pins mesh-sdk 2a04bf0. --- .../cmd/mesh-delivery-watchman-tools/main.go | 170 ++++++ .../mesh-delivery-watchman-tools/main_test.go | 144 +++++ .../integration_test.go | 102 +++- .../cmd/mesh-delivery-watchman/main.go | 58 +- .../cmd/mesh-delivery-watchman/signals.go | 185 +++++++ modules/mesh-delivery-watchman/go.mod | 2 +- modules/mesh-delivery-watchman/go.sum | 4 +- .../internal/deadletters/deadletters.go | 358 ++++++++++++ .../internal/deadletters/deadletters_test.go | 64 +++ .../internal/socket/socket.go | 125 +++++ .../internal/watch/advisories.go | 56 ++ .../internal/watch/dead.go | 78 +++ .../internal/watch/finding.go | 159 ++++++ .../internal/watch/findings_test.go | 200 +++++++ .../internal/watch/mirror.go | 2 + .../internal/watch/signals.go | 524 ++++++++++++++++++ .../internal/watch/watch.go | 61 +- .../internal/watch/watch_test.go | 50 +- .../mesh-delivery-watchman/manifest_test.go | 64 ++- modules/mesh-delivery-watchman/module.json | 39 +- 20 files changed, 2401 insertions(+), 44 deletions(-) create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main_test.go create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/signals.go create mode 100644 modules/mesh-delivery-watchman/internal/deadletters/deadletters.go create mode 100644 modules/mesh-delivery-watchman/internal/deadletters/deadletters_test.go create mode 100644 modules/mesh-delivery-watchman/internal/socket/socket.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/advisories.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/dead.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/finding.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/findings_test.go create mode 100644 modules/mesh-delivery-watchman/internal/watch/signals.go diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go new file mode 100644 index 0000000..1ae2d78 --- /dev/null +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go @@ -0,0 +1,170 @@ +// Command mesh-delivery-watchman-tools serves the seat mesh-watchdogs' verbs and the watchman's read tools (novox/hq ADR 0564 +// §4, §5, §12, issue 1952 WM1): the runtime launches it for a call. It reads every state the watchman keeps (presence, pulse, +// stuck, findings: its keys, one key, how many it holds) and asks the watchman's program, on its socket, for what only the +// program reads or does: DEAD_LETTERS, a stream, and resync. A process never serves (ADR 0436). +package main + +import ( + "encoding/json" + "fmt" + "os" + "sort" + "strings" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-watchman/internal/socket" + "mesh-delivery-watchman/internal/watch" +) + +// Seat is the seat the watchman holds. +const Seat = "mesh-watchdogs" + +// The seat's verbs and the module's own tools, as the manifest names them. +const ( + VerbSignals = "signals" + VerbStuck = "stuck" + VerbPresence = "presence" + VerbDeadLetters = "dead-letters" + VerbResync = "resync" + ToolFindings = "watchman_findings" +) + +// Verbs are the seat's verbs, in the manifest's order. +var Verbs = []string{VerbSignals, VerbStuck, VerbPresence, VerbDeadLetters, VerbResync} + +// ModuleTools are the module's own tools, in the manifest's order. +var ModuleTools = []string{ToolFindings} + +// Reads names, per tool, the states it reads whole; Acts, per tool, the program's act it starts. The clause-12 test holds +// that every state the manifest keeps is read by one, and every act the program takes by hand is started by one. +var ( + Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}} + Acts = map[string][]string{VerbDeadLetters: {socket.AskDeadLetters}, VerbResync: {socket.AskResync}} +) + +// state is what the tools need of a state: the runtime's, or a test's. +type state interface { + Keys() ([]string, error) + Get(key string) (*stdio.StateEntry, error) +} + +// keyed is one key as a read answers it. +type keyed struct { + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` +} + +// readAll answers a state whole, or one key, with how many keys it holds; prefix narrows the keys. +func readAll(st state, name, key, prefix string) (any, error) { + if key != "" { + e, err := st.Get(key) + if err != nil { + return nil, fmt.Errorf("%s %s cannot be read: %w", name, key, err) + } + if e == nil { + return nil, fmt.Errorf("%s holds no key %s", name, key) + } + return keyed{Key: key, Value: e.Value, Revision: e.Revision}, nil + } + keys, err := st.Keys() + if err != nil { + return nil, fmt.Errorf("%s cannot be listed: %w", name, err) + } + sort.Strings(keys) + out := []keyed{} + for _, k := range keys { + if prefix != "" && !strings.HasPrefix(k, prefix) { + continue + } + e, err := st.Get(k) + if err != nil { + return nil, fmt.Errorf("%s %s cannot be read: %w", name, k, err) + } + if e != nil { + out = append(out, keyed{Key: k, Value: e.Value, Revision: e.Revision}) + } + } + return map[string]any{"state": name, "holds": len(keys), "keys": out}, nil +} + +func str(args map[string]any, name string) string { + v, _ := args[name].(string) + return strings.TrimSpace(v) +} + +var keyArgs = map[string]any{"type": "object", "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "one key only"}, + "prefix": map[string]any{"type": "string", "description": "only the keys that begin with it"}}} + +// Tools are the seat's verbs and the module's tools, over the states states gives and the program asked through ask. +func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, error)) []stdio.Tool { + read := func(name string) func(map[string]any) (any, error) { + return func(args map[string]any) (any, error) { + return readAll(states(name), name, str(args, "key"), str(args, "prefix")) + } + } + signals := func(args map[string]any) (any, error) { + pulses, err := readAll(states("pulse"), "pulse", str(args, "key"), str(args, "prefix")) + if err != nil { + return nil, err + } + return map[string]any{"rows": watch.SignalRows, "pulses": pulses}, nil + } + dead := func(args map[string]any) (any, error) { + a := socket.Ask{Ask: socket.AskDeadLetters, Consumer: str(args, "consumer"), Act: str(args, "act"), Why: str(args, "why")} + if v, ok := args["limit"].(float64); ok { + a.Limit = int(v) + } + if v, ok := args["id"].(float64); ok && v > 0 { + a.ID = uint64(v) + } + return ask(a) + } + resync := func(map[string]any) (any, error) { return ask(socket.Ask{Ask: socket.AskResync}) } + verb := func(v string) string { return Seat + "." + v } + return []stdio.Tool{ + {Name: verb(VerbSignals), Input: keyArgs, Run: signals, + Description: "The signals table's rows the watchman holds (S1, S8, S9, S10, S11, S21 and the haves' health), each a pulse key " + + "whose time to live is its bound and whose expiry is its finding (novox/hq ADR 0564 §4), and every pulse the watchman " + + "keeps: the moves it mirrors (a build step, a want, a presence) and the signals (signal.., with what their " + + "expiry does). Give key for one, prefix to narrow. Read-only."}, + {Name: verb(VerbStuck), Input: keyArgs, Run: read("stuck"), + Description: "What stopped moving: every key of the watchman's stuck bucket, . with what it waited for since " + + "when, and whether it moved since (never deleted). A key not moved is also the watchman's finding " + + "delivery..stuck, a node's presence machine..silent. Give key for one, prefix to narrow. Read-only."}, + {Name: verb(VerbPresence), Input: keyArgs, Run: read("presence"), + Description: "Who is online on the bus as the watchman hears it: each node's agent () and each delivery part's instance " + + "(part..), online or not since the entry's time. Give key for one, prefix to narrow. Read-only."}, + {Name: verb(VerbDeadLetters), Run: dead, + Description: "What a consumer gave up on, kept in DEAD_LETTERS (novox/hq ADR 0564 §5, ADR 0264, ADR 0376): newest first, " + + "for one consumer (its name or .), at most limit; one whole by id (a restricted one's body " + + "withheld). Who keeps them now: the controller until the lease names no controller, the watchman from then. " + + "Delivering again and dropping (act deliver or drop, with why) are refused here until the hand-act log's home " + + "(issue 1953): before the lease write use mesh-controller.dead-letters.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "consumer": map[string]any{"type": "string", "description": "only this consumer's, its name or ."}, + "limit": map[string]any{"type": "integer", "description": "at most this many, newest first"}, + "id": map[string]any{"type": "integer", "description": "one dead letter whole, by its id"}, + "act": map[string]any{"type": "string", "enum": []string{"deliver", "drop"}, "description": "deliver again or drop (refused until issue 1953)"}, + "why": map[string]any{"type": "string", "description": "why, with act"}}}}, + {Name: verb(VerbResync), Run: resync, + Description: "Rebuild what the watchman holds by hand (novox/hq ADR 0564 §12): presence from the bus's open connections, " + + "the dead letters' count from DEAD_LETTERS and whether to take the notices, and every finding it observes written " + + "again to its bucket. Answers how many findings it wrote."}, + {Name: ToolFindings, Input: keyArgs, Run: read("findings"), + Description: "The watchman's findings (novox/hq ADR 0564 §2): each thing it observes wrong now, .., with its " + + "severity, subject, evidence and the row that observed it; the condition keeper raises and clears the conditions from " + + "them. Give key for one, prefix to narrow. Read-only."}, + } +} + +func main() { + path := socket.PathOf(os.Getenv) + tools := Tools(func(name string) state { return stdio.State(name) }, func(a socket.Ask) (json.RawMessage, error) { return socket.On(path, a) }) + if err := stdio.Serve("", tools); err != nil { + fmt.Fprintln(os.Stderr, "mesh-delivery-watchman-tools:", err) + os.Exit(1) + } +} diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main_test.go new file mode 100644 index 0000000..97a9262 --- /dev/null +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main_test.go @@ -0,0 +1,144 @@ +package main + +import ( + "encoding/json" + "errors" + "os" + "slices" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-watchman/internal/socket" +) + +type fakeState map[string]string + +func (f fakeState) Keys() ([]string, error) { + out := []string{} + for k := range f { + out = append(out, k) + } + return out, nil +} + +func (f fakeState) Get(key string) (*stdio.StateEntry, error) { + v, ok := f[key] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: key, Value: json.RawMessage(v), Revision: 3}, nil +} + +// 1952/WM1, ADR 0564 §12: every state the manifest keeps is read by a tool, every act the program starts by hand is +// started by one, and every verb the claim serves and every tool the manifest names is served; and each answers. +func TestEveryStateAndActHasATool(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Claims []struct { + Name string `json:"name"` + Serves []string `json:"serves"` + } `json:"claims"` + Tools []string `json:"tools"` + State []struct { + Name string `json:"name"` + } `json:"state"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + asked := []socket.Ask{} + states := map[string]fakeState{"findings": {"machine.ace.silent": `{"v":1}`, "core.doctor.self-check-silent": `{"v":1}`}} + tools := Tools(func(name string) state { return states[name] }, func(a socket.Ask) (json.RawMessage, error) { + asked = append(asked, a) + if a.Act != "" { + return nil, errors.New("refused") + } + return json.RawMessage(`{}`), nil + }) + served := map[string]stdio.Tool{} + for _, tool := range tools { + served[tool.Name] = tool + } + var names []string + for _, c := range m.Claims { + if c.Name != Seat || !slices.Equal(c.Serves, Verbs) { + t.Errorf("claim %+v, want %s serving %v", c, Seat, Verbs) + } + for _, v := range c.Serves { + names = append(names, Seat+"."+v) + } + } + if !slices.Equal(m.Tools, ModuleTools) { + t.Errorf("tools = %v, want %v", m.Tools, ModuleTools) + } + names = append(names, m.Tools...) + if len(served) != len(names) { + t.Errorf("served %d tools, the manifest names %v", len(served), names) + } + for _, n := range names { + if _, ok := served[n]; !ok { + t.Errorf("%s is named and not served", n) + } + } + read := map[string]bool{} + for _, r := range Reads { + for _, s := range r { + read[s] = true + } + } + for _, s := range m.State { + if !read[s.Name] { + t.Errorf("the state %s is read by no tool", s.Name) + } + } + acts := map[string]bool{} + for _, a := range Acts { + for _, s := range a { + acts[s] = true + } + } + for _, a := range []string{socket.AskDeadLetters, socket.AskResync} { + if !acts[a] { + t.Errorf("the program's act %s is started by no tool", a) + } + } + t.Run("a read answers every key and how many", func(t *testing.T) { + out, err := served[ToolFindings].Run(map[string]any{"prefix": "machine."}) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(out) + var got struct { + Holds int `json:"holds"` + Keys []struct { + Key string `json:"key"` + } `json:"keys"` + } + _ = json.Unmarshal(b, &got) + if got.Holds != 2 || len(got.Keys) != 1 || got.Keys[0].Key != "machine.ace.silent" { + t.Fatalf("answer = %s", b) + } + if _, err := served[ToolFindings].Run(map[string]any{"key": "no.such.key"}); err == nil { + t.Fatal("a key not held is answered") + } + }) + t.Run("dead-letters and resync ask the program", func(t *testing.T) { + if _, err := served[Seat+"."+VerbDeadLetters].Run(map[string]any{"consumer": "ace_web", "limit": float64(3), "id": float64(7)}); err != nil { + t.Fatal(err) + } + if _, err := served[Seat+"."+VerbDeadLetters].Run(map[string]any{"act": "drop", "why": "x"}); err == nil { + t.Fatal("a refused act answered") + } + if _, err := served[Seat+"."+VerbResync].Run(nil); err != nil { + t.Fatal(err) + } + want := []socket.Ask{{Ask: socket.AskDeadLetters, Consumer: "ace_web", Limit: 3, ID: 7}, {Ask: socket.AskDeadLetters, Act: "drop", Why: "x"}, {Ask: socket.AskResync}} + if !slices.Equal(asked, want) { + t.Fatalf("asked %+v", asked) + } + }) +} diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index 27cf64c..b32baa3 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -17,8 +17,13 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-watchman/internal/deadletters" + "mesh-delivery-watchman/internal/socket" + "mesh-delivery-watchman/internal/watch" ) // busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system @@ -103,6 +108,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, + {Bucket: watch.FindingsBucket, History: 5}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -111,6 +117,31 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 kvs[c.Bucket] = kv } + // The controller's lease, its consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one + // message a consumer will give up on (1952/WM5). + lease, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buscompose.LeaseBucket}) + if err != nil { + t.Fatal(err) + } + for _, c := range []jetstream.StreamConfig{ + {Name: buscompose.DeadLetterNoticesStream, Subjects: []string{buscompose.MaxDeliveriesAdvisories}, Retention: jetstream.WorkQueuePolicy}, + {Name: buscompose.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}, Duplicates: 24 * time.Hour, AllowMsgTTL: true}, + {Name: "SRC", Subjects: []string{"src.>"}}, + } { + if _, err := js.CreateStream(ctx, c); err != nil { + t.Fatal(err) + } + } + if _, err := js.CreateConsumer(ctx, buscompose.DeadLetterNoticesStream, jetstream.ConsumerConfig{Durable: buscompose.ControllerName, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + given := nats.NewMsg("src.ask") + given.Header.Set("X-Asker", "n1") + given.Data = []byte("do it") + if _, err := js.PublishMsg(ctx, given); err != nil { + t.Fatal(err) + } + // A node already connected when the watchman starts: its presence comes from the one connections request. connect("node.n0") @@ -119,7 +150,13 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } - env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_PULSE_BOUND": "3s"} + // A unix socket's path is at most 108 bytes, which a test's TMPDIR may pass. + sockDir, err := os.MkdirTemp("/tmp", "ws") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(sockDir) }) + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_PULSE_BOUND": "3s", "MESH_WATCHMAN_SOCKET": filepath.Join(sockDir, "tools.sock")} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() @@ -224,6 +261,63 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 t.Errorf("a stuck key %s was written for a move that happened before the bound", k) } } + + // 1952/WM4: what stopped moving is the watchman's own finding: the ask no node ended, and node n1 silent. + finding := func(key string) bool { _, ok := has(watch.FindingsBucket, key); return ok } + until("the stuck ask's finding", func() bool { + return finding(watch.FindingKey(watch.ScopeDelivery, stuckOf(askStream, ask1), watch.KindStuck)) + }) + until("node n1's silence", func() bool { return finding("machine.n1.silent") }) + + // 1952/WM5: while the lease names a controller the watchman takes no notice; handed away while the controller's consumer + // stands, it says the notices wait; the consumer removed (the lease write's step), it makes its own and keeps a notice + // with its headers. + if _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer); err == nil { + t.Fatal("the watchman made its consumer while the lease names a controller") + } + if _, err := lease.Put(ctx, buscompose.ControllerLeaseKey, []byte(`{"handed-to":"mesh-vault","handed-by":"operator","why":"the lease write"}`)); err != nil { + t.Fatal(err) + } + waits := watch.FindingKey(watch.ScopeBus, "DEAD_LETTER_NOTICES.controller", watch.KindNoticesWaiting) + until("the notices said waiting behind the controller's consumer", func() bool { return finding(waits) }) + if _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer); err == nil { + t.Fatal("the watchman made its consumer while the controller's stands") + } + if err := js.DeleteConsumer(ctx, buscompose.DeadLetterNoticesStream, buscompose.ControllerName); err != nil { + t.Fatal(err) + } + until("the watchman's own consumer on the notices", func() bool { + _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, deadletters.Consumer) + return err == nil + }) + until("the wait no longer said", func() bool { return !finding(waits) }) + notice, _ := json.Marshal(map[string]any{"stream": "SRC", "consumer": "c1", "stream_seq": 1, "deliveries": 5, "timestamp": time.Now().UTC()}) + if err := ctl.Publish(buscompose.MaxDeliveriesAdvisories[:len(buscompose.MaxDeliveriesAdvisories)-1]+"SRC.c1", notice); err != nil { + t.Fatal(err) + } + dl, err := js.Stream(ctx, buscompose.DeadLettersStream) + if err != nil { + t.Fatal(err) + } + var kept *jetstream.RawStreamMsg + until("the notice kept in DEAD_LETTERS", func() bool { + kept, err = dl.GetLastMsgForSubject(ctx, buscompose.DeadLetterSubject("SRC", "c1")) + return err == nil + }) + if kept.Header.Get(deadletters.HeaderStream) != "SRC" || kept.Header.Get(deadletters.HeaderSequence) != "1" || + kept.Header.Get(deadletters.HeaderSubject) != "src.ask" || kept.Header.Get("X-Asker") != "n1" || string(kept.Data) != "do it" { + t.Fatalf("kept = %v %q", kept.Header, kept.Data) + } + until("the consumer's dead letters said", func() bool { return finding("bus.SRC.c1.max-deliveries") }) + // The tools' socket lists it, and refuses to drop it. + answer, err := socket.On(env["MESH_WATCHMAN_SOCKET"], socket.Ask{Ask: socket.AskDeadLetters, Consumer: "c1"}) + if err != nil || !strings.Contains(string(answer), `"total":1`) || !strings.Contains(string(answer), "the watchman") { + t.Fatalf("dead-letters = %s, %v", answer, err) + } + if _, err := socket.On(env["MESH_WATCHMAN_SOCKET"], socket.Ask{Ask: socket.AskDeadLetters, ID: kept.Sequence, Act: "drop", Why: "test"}); err == nil || + !strings.Contains(err.Error(), "issue 1953") { + t.Fatalf("a drop was not refused naming the hand-act log's home: %v", err) + } } // TestAnAskRunningBeforeTheStartGetsItsPulse puts a running step in the build module's steps before the watchman starts, and @@ -271,6 +365,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, + {Bucket: watch.FindingsBucket, History: 5}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -291,7 +386,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } - env := map[string]string{"MESH_BROKER_FILE": broker} + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_WATCHMAN_SOCKET": filepath.Join(dir, "tools.sock")} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() @@ -355,6 +450,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { {Bucket: delivery.StepsBucket("mesh-build-go"), History: 5}, {Bucket: delivery.AssignerWantsBucket, History: 20}, {Bucket: delivery.HaveBucket, History: 20}, + {Bucket: watch.FindingsBucket, History: 5}, } { kv, err := js.CreateKeyValue(ctx, c) if err != nil { @@ -393,7 +489,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } - env := map[string]string{"MESH_BROKER_FILE": broker} + env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_WATCHMAN_SOCKET": filepath.Join(dir, "tools.sock")} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index c349da7..af4d80b 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -22,8 +22,10 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" + "mesh-delivery-watchman/internal/socket" "mesh-delivery-watchman/internal/watch" ) @@ -44,12 +46,19 @@ func main() { } } -// kvOut writes the watchman's three buckets. +// kvOut writes the watchman's four buckets. type kvOut struct { - ctx context.Context - presence, pulse, stuk jetstream.KeyValue + ctx context.Context + presence, pulse, stuk, findings jetstream.KeyValue } +func (o *kvOut) PutFinding(key string, value []byte) error { + _, err := o.findings.Put(o.ctx, key, value) + return err +} + +func (o *kvOut) DeleteFinding(key string) error { return o.findings.Delete(o.ctx, key) } + func (o *kvOut) PutPresence(key string, online bool) error { value, err := delivery.EncodePresence(delivery.Presence{Online: online}) if err != nil { @@ -161,11 +170,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants // and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build // seats' steps, and says that moves are not watched (issue 1945). - var kvs struct{ presence, pulse, stuck, wants, haves jetstream.KeyValue } + var kvs struct{ presence, pulse, stuck, findings, wants, haves, lease jetstream.KeyValue } for _, b := range []struct { name string into *jetstream.KeyValue - }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} { + }{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}, {watch.FindingsBucket, &kvs.findings}} { if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil { return err } @@ -176,7 +185,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it } - w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck}, time.Now) + kvs.lease = process.OptionalBucket(ctx, js, buscompose.LeaseBucket, "the lease is read as naming a controller, so the dead letters stay the controller's", logger.Printf) + w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now) w.Logf(logger.Printf) // A test shortens how long a move may take; the mesh never sets it. if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" { @@ -197,6 +207,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro w.SeedPresence(e.Key(), p.Online, e.Created()) } }}, + {kvs.findings, func(e jetstream.KeyValueEntry) { + if opOf(e) == watch.Put { + w.SeedFinding(e.Key(), e.Value()) + } + }}, {kvs.stuck, func(e jetstream.KeyValueEntry) { if opOf(e) == watch.Put { w.SeedStuck(e.Key(), e.Value()) @@ -212,6 +227,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro pulses, err := settle(ctx, kvs.pulse, func(e jetstream.KeyValueEntry) { if opOf(e) == watch.Put { w.SeedPulse(e.Key()) + w.SeedSignal(e.Key(), e.Value()) } }) if err != nil { @@ -281,9 +297,39 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := w.Rebuild(ask); err != nil { return err } + dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf} + dead.count() + dead.check() w.Reconcile() + if err := hear(conn, w, dead, logger.Printf); err != nil { + return err + } errs := make(chan error, 8) + if kvs.lease != nil { + lw, err := kvs.lease.Watch(ctx, buscompose.ControllerLeaseKey) + if err != nil { + process.NotWatched(logger.Printf, buscompose.LeaseBucket, err, "the lease is read as naming a controller, so the dead letters stay the controller's") + } else { + go func() { + errs <- follow(ctx, lw, func(e jetstream.KeyValueEntry) { + var raw []byte + if opOf(e) == watch.Put { + raw = e.Value() + } + on, _ := buscompose.LeaseHandedAway(raw, len(raw) > 0, nil) + dead.lease(on) + }) + }() + } + } + if l, err := socket.Listen(socket.PathOf(getenv)); err != nil { + logger.Printf("the tools' socket cannot be opened (%v): dead-letters and resync answer nothing until the watchman starts again", err) + } else { + go func() { + errs <- socket.Serve(ctx, l, func(a socket.Ask) (any, error) { return answer(a, w, dead, ask) }) + }() + } go func() { errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) }) }() diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/signals.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/signals.go new file mode 100644 index 0000000..6f7f742 --- /dev/null +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/signals.go @@ -0,0 +1,185 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/buscompose" + + "mesh-delivery-watchman/internal/deadletters" + "mesh-delivery-watchman/internal/socket" + "mesh-delivery-watchman/internal/watch" +) + +// The subjects the watchman hears for the signals table's rows (novox/hq ADR 0564 §4, Limits 3 and 11), each on the bus, +// none in the controller's store. Its grant names them (mesh-sdk buscompose, the watchman's principal). +const ( + PowerSubject = "mesh.mod.power.event.*" + FailingSubject = "mesh.mod.*.event.provisioner.failing" + RecoveredSubject = "mesh.mod.*.event.provisioner.recovered" + ToolsAliveSubject = "mesh.control.*.tools-alive" + CheckedSubject = "mesh.seat.mesh-self-check.event.checked" + DeadKeptSubject = "mesh.events.dead.>" +) + +// HeardSubjects is every subject the watchman subscribes for its rows, beside the bus's connect and disconnect. +var HeardSubjects = []string{PowerSubject, FailingSubject, RecoveredSubject, ToolsAliveSubject, CheckedSubject, + watch.ConsumerDeletedAdvisories, DeadKeptSubject} + +// heard hands one message of a row's subject to the watchman. +func heard(w *watch.Watchman, dead *deadKeeper, m *nats.Msg, logf func(string, ...any)) { + switch { + case strings.HasPrefix(m.Subject, "mesh.mod.power.event."): + w.Power(m.Header.Get("x-node"), strings.TrimPrefix(m.Subject, "mesh.mod.power.event.")) + case strings.HasSuffix(m.Subject, ".event.provisioner.failing") || strings.HasSuffix(m.Subject, ".event.provisioner.recovered"): + st, err := watch.ReadStanding(m.Subject, m.Data) + if err != nil { + logf("%v; left", err) + return + } + w.Stood(st) + case strings.HasSuffix(m.Subject, ".tools-alive"): + var body struct { + Every float64 `json:"every"` + } + _ = json.Unmarshal(m.Data, &body) + w.ToolsAlive(watch.ToolsNode(m.Subject), time.Duration(body.Every*float64(time.Second))) + case m.Subject == CheckedSubject: + w.Checked(m.Header.Get("x-node")) + case strings.HasPrefix(m.Subject, "$JS.EVENT.ADVISORY."): + a, ok := watch.ReadAdvisory(m.Subject, m.Data, deadletters.ConsumerInWords) + if !ok { + return + } + if dead.isTheControllersNotices(a.ID) { + dead.check() // the lease write's step removed it: the watchman may take the notices now + return + } + w.Advised(a) + case strings.HasPrefix(m.Subject, "mesh.events.dead."): + dead.count() + } +} + +// hear subscribes every row's subject; a subject the bus refuses stops the start, naming it. +func hear(conn *nats.Conn, w *watch.Watchman, dead *deadKeeper, logf func(string, ...any)) error { + for _, subject := range HeardSubjects { + if _, err := conn.Subscribe(subject, func(m *nats.Msg) { heard(w, dead, m, logf) }); err != nil { + return fmt.Errorf("the watchman cannot hear %s: %w", subject, err) + } + } + return conn.Flush() +} + +// deadKeeper keeps what consumers gave up on from the lease write (ADR 0564 §5): it counts DEAD_LETTERS into findings, +// and takes DEAD_LETTER_NOTICES on its own consumer only once the lease names no controller and `controller` is gone. +type deadKeeper struct { + ctx context.Context + js jetstream.JetStream + w *watch.Watchman + logf func(string, ...any) + + mu sync.Mutex + handedAway bool + taking jetstream.ConsumeContext +} + +// count reads DEAD_LETTERS' count per consumer into the findings. +func (d *deadKeeper) count() { + held, err := deadletters.Held(d.ctx, d.js) + if err != nil { + d.logf("the dead letters cannot be counted, their findings are left as they are: %v", err) + return + } + d.w.DeadLettersHeld(held, deadletters.ConsumerInWords) +} + +// lease is the lease's change heard: handed away or not. +func (d *deadKeeper) lease(handedAway bool) { + d.mu.Lock() + d.handedAway = handedAway + d.mu.Unlock() + d.check() +} + +func (d *deadKeeper) away() bool { + d.mu.Lock() + defer d.mu.Unlock() + return d.handedAway +} + +// isTheControllersNotices says an advisory is about the controller's consumer on DEAD_LETTER_NOTICES once the lease is +// handed away: its removal is the lease write's step, not a loss. +func (d *deadKeeper) isTheControllersNotices(id string) bool { + return d.away() && id == buscompose.DeadLetterNoticesStream+"."+buscompose.ControllerName +} + +// check takes the notices, or says they wait, as Gate has it. +func (d *deadKeeper) check() { + away := d.away() + stands := true + if away { + var err error + if stands, err = deadletters.ControllerStands(d.ctx, d.js); err != nil { + d.logf("whether the controller's consumer on %s stands cannot be read (%v): it is taken to stand", buscompose.DeadLetterNoticesStream, err) + stands = true + } + } + take, waiting := deadletters.Gate(away, stands) + d.w.NoticesWaiting(waiting) + d.mu.Lock() + defer d.mu.Unlock() + if !take || d.taking != nil { + return + } + cc, err := deadletters.TakeNotices(d.ctx, d.js, func(deadletters.DeadLetter) { go d.count() }, d.logf) + if err != nil { + d.logf("the dead-letter notices cannot be taken yet: %v", err) + return + } + d.taking = cc + d.logf("the lease names no controller and its consumer on %s is gone: the watchman keeps the dead letters", buscompose.DeadLetterNoticesStream) +} + +// answer is the program's answer to the tools' ask on its socket. +func answer(a socket.Ask, w *watch.Watchman, dead *deadKeeper, ask func([]byte) ([]byte, error)) (any, error) { + switch a.Ask { + case socket.AskResync: + if err := w.Rebuild(ask); err != nil { + return nil, err + } + dead.count() + dead.check() + return map[string]any{"findings-written": w.Rewrite(), "presence": "rebuilt from the bus's connections"}, nil + case socket.AskDeadLetters: + switch { + case a.Act == "deliver" || a.Act == "drop": + if strings.TrimSpace(a.Why) == "" { + return nil, errors.New("a dead letter is delivered again or dropped only with why") + } + return nil, deadletters.Refusal(dead.away(), a.Act) + case a.Act != "": + return nil, fmt.Errorf("%q is not deliver or drop", a.Act) + case a.ID > 0: + return deadletters.Named(dead.ctx, dead.js, a.ID) + } + list, total, err := deadletters.List(dead.ctx, dead.js, a.Consumer, a.Limit) + if err != nil { + return nil, err + } + keeper := "the controller (it holds its lease)" + if dead.away() { + keeper = "the watchman" + } + return map[string]any{"dead-letters": list, "total": total, "kept-by": keeper}, nil + } + return nil, fmt.Errorf("refused: the watchman's socket answers %q and %q", socket.AskDeadLetters, socket.AskResync) +} diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index d42c521..1390d78 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-watchman go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-watchman/go.sum b/modules/mesh-delivery-watchman/go.sum index 084deeb..e796a10 100644 --- a/modules/mesh-delivery-watchman/go.sum +++ b/modules/mesh-delivery-watchman/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf h1:Yy51rUWqyk+cZPAcSX02dnK4JVSTUJIfLa9j+741+1Y= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-watchman/internal/deadletters/deadletters.go b/modules/mesh-delivery-watchman/internal/deadletters/deadletters.go new file mode 100644 index 0000000..4e4a228 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/deadletters/deadletters.go @@ -0,0 +1,358 @@ +// Package deadletters is what a consumer gave up on, kept by the watchman from the lease write (novox/hq ADR 0564 §5, +// ADR 0264, ADR 0376, issue 1952 WM5): ported from mesh-controller internal/link/deadletters.go at e04846e9 with nothing +// changed in what is kept, aged or withheld, on the jetstream API. +// +// DEAD_LETTER_NOTICES is a work queue that holds one consumer. The controller's consumer `controller` never expires, so +// the watchman makes its own (Consumer) only when the lease names no controller and `controller` is gone (Gate); while +// `controller` still stands after the lease write it says so as a finding. Removing `controller` is a step of the lease +// write, with the bus module's tool. Before the lease write the watchman only lists and shows DEAD_LETTERS, which the +// controller still fills, and refuses to deliver or drop, naming the controller's verb; from it, delivering and dropping +// wait for the hand-act log's home (issue 1953), so they are refused naming that. +package deadletters + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/buscompose" +) + +// Consumer is the watchman's own consumer on DEAD_LETTER_NOTICES, made only by Gate's leave. +const Consumer = "mesh-delivery-watchman" + +// The headers a kept message carries beside its own (mesh-controller link.DeadHeader*: the same names, so the +// controller's copies and the watchman's are read alike). +const ( + HeaderStream = "Mesh-Dead-Stream" + HeaderConsumer = "Mesh-Dead-Consumer" + HeaderSequence = "Mesh-Dead-Sequence" + HeaderSubject = "Mesh-Dead-Subject" + HeaderDeliveries = "Mesh-Dead-Deliveries" + HeaderGaveUp = "Mesh-Dead-Gave-Up" + HeaderPublished = "Mesh-Dead-Published" + HeaderLost = "Mesh-Dead-Lost" + HeaderMsgID = "Mesh-Dead-Msg-Id" + HeaderRestricted = "Mesh-Dead-Restricted" + HeaderAgesOut = "Mesh-Dead-Ages-Out" + HeaderWorkQueue = "Mesh-Dead-Work-Queue" +) + +// DeadLetter is one message a consumer gave up on, as DEAD_LETTERS keeps it (the controller's JSON). +type DeadLetter struct { + ID uint64 `json:"id"` + Stream string `json:"stream"` + Consumer string `json:"consumer"` + Who string `json:"who"` + Subject string `json:"subject"` + Sequence uint64 `json:"sequence"` + Deliveries uint64 `json:"deliveries"` + GaveUp time.Time `json:"gave_up"` + Published time.Time `json:"published,omitzero"` + Lost string `json:"lost,omitempty"` + Size int `json:"size"` + Restricted bool `json:"restricted,omitempty"` + AgesOut time.Time `json:"ages_out,omitzero"` + Withheld string `json:"withheld,omitempty"` + WorkQueue bool `json:"work-queue,omitempty"` + Body string `json:"body,omitempty"` + Headers map[string][]string `json:"headers,omitempty"` +} + +// Now is the clock dead letters age by: a variable, so a test can move it. +var Now = time.Now + +// WithheldWords is what is said in place of a restricted dead letter's body. +const WithheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " + + "the modules its grant names may read (novox/hq ADR 0376)" + +// ConsumerInWords is a durable consumer as the mesh says it: whose, and for what. +func ConsumerInWords(stream, name string) string { + switch { + case name == buscompose.ControllerName || name == buscompose.MergesConsumerName: + return "the controller's consumer on " + stream + case stream == "NODES": + return "how " + name + " hears what it should be (its declaration consumer)" + case strings.HasPrefix(stream, "SEAT_") && strings.HasSuffix(name, "_worker"): + seat := strings.ToLower(strings.ReplaceAll(strings.TrimPrefix(stream, "SEAT_"), "_", "-")) + return "the worker every holder of " + seat + " takes its asks from" + case stream == buscompose.EventsStream: + if node, module, ok := strings.Cut(name, "_"); ok { + return "how " + module + " on " + node + " hears what it consumes" + } + } + return "the consumer " + name + " on " + stream +} + +// sourceAge is whether a stream is restricted, a module's declared work queue, and its maximum age; a stream gone is +// neither. +func sourceAge(ctx context.Context, js jetstream.JetStream, stream string) (restricted, workQueue bool, maxAge time.Duration, err error) { + s, err := js.Stream(ctx, stream) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return false, false, 0, nil + } + if err != nil { + return false, false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) + } + cfg := s.CachedInfo().Config + return cfg.Metadata[buscompose.RestrictedMetadata] == "true", cfg.Metadata[buscompose.WorkQueueMetadata] == "true", cfg.MaxAge, nil +} + +// notice is the part of the server's maximum-deliveries advisory a kept message is made from. +type notice struct { + Stream string `json:"stream"` + Consumer string `json:"consumer"` + StreamSeq uint64 `json:"stream_seq"` + Deliveries uint64 `json:"deliveries"` + Timestamp time.Time `json:"timestamp"` +} + +// Keep keeps the message one maximum-deliveries notice is about. An error keeps nothing and the notice is to be taken +// again; a source that no longer holds the message is no error: the record is kept with why. +func Keep(ctx context.Context, js jetstream.JetStream, body []byte) (DeadLetter, error) { + var a notice + if err := json.Unmarshal(body, &a); err != nil || a.Stream == "" || a.Consumer == "" || a.StreamSeq == 0 { + return DeadLetter{}, fmt.Errorf("not a maximum-deliveries notice the mesh can read: %.200s", body) + } + gaveUp := a.Timestamp + if gaveUp.IsZero() { + gaveUp = Now() + } + kept := &nats.Msg{Subject: buscompose.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}} + restricted, workQueue, maxAge, err := sourceAge(ctx, js, a.Stream) + if err != nil { + return DeadLetter{}, err + } + if restricted { + kept.Header.Set(HeaderRestricted, "true") + } + if workQueue { + kept.Header.Set(HeaderWorkQueue, "true") + } + var raw *jetstream.RawStreamMsg + src, err := js.Stream(ctx, a.Stream) + if err == nil { + raw, err = src.GetMsg(ctx, a.StreamSeq) + } + agesOut := time.Time{} + if err == nil && restricted && maxAge > 0 { + agesOut = raw.Time.Add(maxAge) + kept.Header.Set(HeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano)) + } + switch { + case err == nil && !agesOut.IsZero() && agesOut.Sub(Now()) < time.Second: + kept.Header.Set(HeaderSubject, raw.Subject) + kept.Header.Set(HeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) + kept.Header.Set(HeaderLost, fmt.Sprintf("message %d of %s had passed its stream's maximum age of %s when the notice "+ + "was taken, so it is kept without its body (novox/hq ADR 0376)", a.StreamSeq, a.Stream, maxAge)) + case err == nil: + for k, v := range raw.Header { + if !strings.HasPrefix(k, "Nats-") { + kept.Header[k] = v + } + } + if id := raw.Header.Get(nats.MsgIdHdr); id != "" { + kept.Header.Set(HeaderMsgID, id) + } + kept.Header.Set(HeaderSubject, raw.Subject) + kept.Header.Set(HeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) + kept.Data = raw.Data + if !agesOut.IsZero() { + kept.Header.Set(nats.MsgTTLHdr, strconv.FormatInt(int64(agesOut.Sub(Now())/time.Second), 10)+"s") + } + case errors.Is(err, jetstream.ErrMsgNotFound) || errors.Is(err, jetstream.ErrStreamNotFound): + kept.Header.Set(HeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v", a.Stream, a.StreamSeq, err)) + default: + return DeadLetter{}, fmt.Errorf("message %d of %s could not be read: %w", a.StreamSeq, a.Stream, err) + } + kept.Header.Set(HeaderStream, a.Stream) + kept.Header.Set(HeaderConsumer, a.Consumer) + kept.Header.Set(HeaderSequence, strconv.FormatUint(a.StreamSeq, 10)) + kept.Header.Set(HeaderDeliveries, strconv.FormatUint(a.Deliveries, 10)) + kept.Header.Set(HeaderGaveUp, gaveUp.UTC().Format(time.RFC3339Nano)) + // One copy per message given up, however often its notice is taken, by whichever keeper takes it. + kept.Header.Set(nats.MsgIdHdr, fmt.Sprintf("%s.%s.%d", a.Stream, a.Consumer, a.StreamSeq)) + ack, err := js.PublishMsg(ctx, kept) + if err != nil { + return DeadLetter{}, fmt.Errorf("message %d of %s could not be kept in %s: %w", a.StreamSeq, a.Stream, buscompose.DeadLettersStream, err) + } + return Of(ack.Sequence, kept.Subject, kept.Header, kept.Data, false), nil +} + +// Of reads a kept message; whole gives its body and its own headers. +func Of(id uint64, subject string, h nats.Header, data []byte, whole bool) DeadLetter { + d := DeadLetter{ID: id, Stream: h.Get(HeaderStream), Consumer: h.Get(HeaderConsumer), Subject: h.Get(HeaderSubject), + Lost: h.Get(HeaderLost), Size: len(data)} + if d.Stream == "" || d.Consumer == "" { + d.Stream, d.Consumer, _ = buscompose.DeadLetterOf(subject) + } + d.Who = ConsumerInWords(d.Stream, d.Consumer) + d.Sequence, _ = strconv.ParseUint(h.Get(HeaderSequence), 10, 64) + d.Deliveries, _ = strconv.ParseUint(h.Get(HeaderDeliveries), 10, 64) + d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderGaveUp)) + d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderPublished)) + d.Restricted = h.Get(HeaderRestricted) == "true" + d.WorkQueue = h.Get(HeaderWorkQueue) == "true" + d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(HeaderAgesOut)) + if whole { + d.Body = string(data) + d.Headers = map[string][]string{} + for k, v := range h { + if !strings.HasPrefix(k, "Mesh-Dead-") && !strings.HasPrefix(k, "Nats-") { + d.Headers[k] = append([]string(nil), v...) + } + } + } + return d +} + +// Held is how many messages DEAD_LETTERS holds for each consumer, by `.`. +func Held(ctx context.Context, js jetstream.JetStream) (map[string]int, error) { + s, err := js.Stream(ctx, buscompose.DeadLettersStream) + if err != nil { + return nil, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err) + } + info, err := s.Info(ctx, jetstream.WithSubjectFilter(">")) + if err != nil { + return nil, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err) + } + held := map[string]int{} + for subject, n := range info.State.Subjects { + if stream, consumer, ok := buscompose.DeadLetterOf(subject); ok && n > 0 { + held[stream+"."+consumer] += int(n) + } + } + return held, nil +} + +// List is what DEAD_LETTERS holds, newest first, at most most (all when most is not positive); for one consumer when +// consumer names one (its name, or `.`). A restricted one's body is withheld. The total is the stream's +// own count. +func List(ctx context.Context, js jetstream.JetStream, consumer string, most int) ([]DeadLetter, int, error) { + held, err := Held(ctx, js) + if err != nil { + return nil, 0, err + } + total := 0 + for key, n := range held { + if consumer == "" || key == consumer || strings.HasSuffix(key, "."+consumer) { + total += n + } + } + if total == 0 { + return nil, 0, nil + } + s, err := js.Stream(ctx, buscompose.DeadLettersStream) + if err != nil { + return nil, 0, err + } + info, err := s.Info(ctx) + if err != nil { + return nil, 0, fmt.Errorf("%s cannot be read: %w", buscompose.DeadLettersStream, err) + } + var out []DeadLetter + for seq := info.State.LastSeq; seq >= info.State.FirstSeq && seq > 0; seq-- { + if most > 0 && len(out) >= most || len(out) >= total { + break + } + raw, err := s.GetMsg(ctx, seq) + if errors.Is(err, jetstream.ErrMsgNotFound) { + continue + } + if err != nil { + return out, total, fmt.Errorf("dead letter %d cannot be read: %w", seq, err) + } + d := Of(seq, raw.Subject, raw.Header, raw.Data, false) + if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer { + continue + } + if d.Restricted { + d.Withheld = WithheldWords + } + out = append(out, d) + } + return out, total, nil +} + +// ErrNoDeadLetter is an id DEAD_LETTERS does not hold. +var ErrNoDeadLetter = errors.New("no such dead letter") + +// Named is one kept message whole, but for a restricted one, whose body and own headers are never shown. +func Named(ctx context.Context, js jetstream.JetStream, id uint64) (DeadLetter, error) { + s, err := js.Stream(ctx, buscompose.DeadLettersStream) + if err != nil { + return DeadLetter{}, err + } + raw, err := s.GetMsg(ctx, id) + if errors.Is(err, jetstream.ErrMsgNotFound) { + return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, or never kept; "+ + "dead-letters lists what it holds", ErrNoDeadLetter, buscompose.DeadLettersStream, id) + } + if err != nil { + return DeadLetter{}, fmt.Errorf("dead letter %d cannot be read: %w", id, err) + } + d := Of(id, raw.Subject, raw.Header, raw.Data, true) + if d.Restricted { + d.Body, d.Headers, d.Withheld = "", nil, WithheldWords + } + return d, nil +} + +// Refusal is why the watchman delivers nothing again and drops nothing: before the lease write the controller keeps them +// and its verb acts; from it the act waits for the hand-act log's home (issue 1953), where a person's act is recorded. +// act is deliver or drop. +func Refusal(handedAway bool, act string) error { + done := map[string]string{"deliver": "delivered again", "drop": "dropped"}[act] + if done == "" { + return fmt.Errorf("%q is not deliver or drop", act) + } + if !handedAway { + return fmt.Errorf("the controller still holds its lease and keeps the dead letters: %s it with "+ + "mesh-controller.dead-letters (novox/hq ADR 0564 §5); nothing is %s here", act, done) + } + return fmt.Errorf("a dead letter is %s only as a person's act recorded in the hand-act log, whose home is not "+ + "delivered yet (issue 1953): nothing is %s", done, done) +} + +// Gate is whether the watchman takes the notices itself: only when the lease names no controller (handedAway) and the +// controller's consumer is gone. waiting says the lease is handed away while `controller` still stands, which is the +// watchman's finding: each notice waits in the queue until the lease write's step removes it. +func Gate(handedAway, controllerStands bool) (take, waiting bool) { + return handedAway && !controllerStands, handedAway && controllerStands +} + +// ControllerStands asks the bus whether the controller's consumer on DEAD_LETTER_NOTICES stands. +func ControllerStands(ctx context.Context, js jetstream.JetStream) (bool, error) { + _, err := js.Consumer(ctx, buscompose.DeadLetterNoticesStream, buscompose.ControllerName) + if errors.Is(err, jetstream.ErrConsumerNotFound) { + return false, nil + } + return err == nil, err +} + +// TakeNotices makes the watchman's own durable consumer on DEAD_LETTER_NOTICES (never touching `controller`) and keeps +// each notice it takes; a notice not kept is handed back to be taken again. kept hears each one kept. +func TakeNotices(ctx context.Context, js jetstream.JetStream, kept func(DeadLetter), logf func(string, ...any)) (jetstream.ConsumeContext, error) { + c, err := js.CreateOrUpdateConsumer(ctx, buscompose.DeadLetterNoticesStream, jetstream.ConsumerConfig{ + Durable: Consumer, AckPolicy: jetstream.AckExplicitPolicy, AckWait: time.Minute}) + if err != nil { + return nil, fmt.Errorf("the consumer %s on %s cannot be made: %w", Consumer, buscompose.DeadLetterNoticesStream, err) + } + return c.Consume(func(m jetstream.Msg) { + d, err := Keep(ctx, js, m.Data()) + if err != nil { + logf("a dead-letter notice was not kept, offered again in a minute: %v", err) + _ = m.NakWithDelay(time.Minute) + return + } + _ = m.Ack() + kept(d) + }) +} diff --git a/modules/mesh-delivery-watchman/internal/deadletters/deadletters_test.go b/modules/mesh-delivery-watchman/internal/deadletters/deadletters_test.go new file mode 100644 index 0000000..2926eca --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/deadletters/deadletters_test.go @@ -0,0 +1,64 @@ +package deadletters + +import ( + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// 1952/WM5, ADR 0564 §5: the watchman makes no consumer while `controller` exists or the lease names a controller, makes +// one once both are false, and says the wait meanwhile; it delivers and drops nothing, naming who does; and a kept message +// reads as the controller's copies do, a restricted one's age and work queue kept. +func TestTheWatchmanKeepsTheDeadLettersOnlyFromTheLeaseWrite(t *testing.T) { + for _, c := range []struct { + handedAway, controllerStands, take, waiting bool + }{ + {false, true, false, false}, + {false, false, false, false}, + {true, true, false, true}, + {true, false, true, false}, + } { + take, waiting := Gate(c.handedAway, c.controllerStands) + if take != c.take || waiting != c.waiting { + t.Errorf("Gate(%v, %v) = %v, %v", c.handedAway, c.controllerStands, take, waiting) + } + } + for _, c := range []struct { + away bool + act string + says string + }{ + {false, "deliver", "mesh-controller.dead-letters"}, + {false, "drop", "nothing is dropped"}, + {true, "drop", "issue 1953"}, + {true, "deliver", "delivered again"}, + {true, "send", "not deliver or drop"}, + } { + if err := Refusal(c.away, c.act); err == nil || !strings.Contains(err.Error(), c.says) { + t.Errorf("Refusal(%v, %s) = %v, want it to say %q", c.away, c.act, err, c.says) + } + } + at := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC) + h := nats.Header{} + h.Set(HeaderStream, "EVENTS") + h.Set(HeaderConsumer, "ace_web") + h.Set(HeaderSequence, "41") + h.Set(HeaderDeliveries, "5") + h.Set(HeaderGaveUp, at.Format(time.RFC3339Nano)) + h.Set(HeaderRestricted, "true") + h.Set(HeaderWorkQueue, "true") + h.Set(HeaderAgesOut, at.Add(time.Hour).Format(time.RFC3339Nano)) + h.Set("X-Own", "kept") + h.Set("Nats-Msg-Id", "EVENTS.ace_web.41") + d := Of(9, "mesh.events.dead.EVENTS.ace_web", h, []byte("body"), true) + if d.ID != 9 || d.Sequence != 41 || d.Deliveries != 5 || !d.GaveUp.Equal(at) || !d.Restricted || !d.WorkQueue || + !d.AgesOut.Equal(at.Add(time.Hour)) || d.Who != "how web on ace hears what it consumes" || + d.Body != "body" || len(d.Headers) != 1 || d.Headers["X-Own"][0] != "kept" { + t.Fatalf("dead letter = %+v", d) + } + if d := Of(10, "mesh.events.dead.SEAT_X.y_worker", nats.Header{}, nil, false); d.Stream != "SEAT_X" || d.Consumer != "y_worker" || d.Body != "" { + t.Fatalf("a copy without headers reads its stream and consumer from its subject: %+v", d) + } +} diff --git a/modules/mesh-delivery-watchman/internal/socket/socket.go b/modules/mesh-delivery-watchman/internal/socket/socket.go new file mode 100644 index 0000000..42cf799 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/socket/socket.go @@ -0,0 +1,125 @@ +// Package socket is how the watchman's tools ask its program what only the program reads (novox/hq ADR 0564 §12, as the +// view's tools ask the view, mesh-delivery-view internal/view/serve.go): DEAD_LETTERS is a stream, which the tool runtime +// grants no bundle, and resync is an act of the program. One line of JSON each way, on a unix socket on the program's +// machine; the socket keeps nothing. +package socket + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "os" + "path/filepath" + "strings" + "time" +) + +// Path is where the program listens unless MESH_WATCHMAN_SOCKET says otherwise. +const Path = "/var/lib/mesh-delivery-watchman-tools/tools.sock" + +// PathOf is the socket's path as getenv says it. +func PathOf(getenv func(string) string) string { + if p := strings.TrimSpace(getenv("MESH_WATCHMAN_SOCKET")); p != "" { + return p + } + return Path +} + +// The asks the program answers. +const ( + AskDeadLetters = "dead-letters" + AskResync = "resync" +) + +// Ask is one ask of the tools: dead-letters (consumer, limit, id; deliver or drop with why) or resync. +type Ask struct { + Ask string `json:"ask"` + Consumer string `json:"consumer,omitempty"` + Limit int `json:"limit,omitempty"` + ID uint64 `json:"id,omitempty"` + Act string `json:"act,omitempty"` + Why string `json:"why,omitempty"` +} + +// Listen listens on the unix socket at path, in a directory of mode 0755, the socket 0666. +func Listen(path string) (net.Listener, error) { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, err + } + l, err := net.Listen("unix", path) + if err != nil { + return nil, err + } + if err := os.Chmod(path, 0o666); err != nil { + l.Close() + return nil, err + } + return l, nil +} + +// Serve answers each ask on l with answer until ctx ends. +func Serve(ctx context.Context, l net.Listener, answer func(Ask) (any, error)) error { + go func() { <-ctx.Done(); l.Close() }() + for { + c, err := l.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + go func() { + defer c.Close() + _ = c.SetDeadline(time.Now().Add(30 * time.Second)) + line, err := bufio.NewReader(io.LimitReader(c, 4096)).ReadBytes('\n') + if err != nil && len(line) == 0 { + return + } + var a Ask + enc := json.NewEncoder(c) + if json.Unmarshal(line, &a) != nil { + _ = enc.Encode(map[string]any{"error": "refused: an ask is one line of JSON"}) + return + } + out, err := answer(a) + if err != nil { + _ = enc.Encode(map[string]any{"error": err.Error()}) + return + } + _ = enc.Encode(map[string]any{"answer": out}) + }() + } +} + +// On asks the program at path and returns its answer. +func On(path string, a Ask) (json.RawMessage, error) { + c, err := net.DialTimeout("unix", path, 5*time.Second) + if err != nil { + return nil, fmt.Errorf("cannot reach the watchman at %s: %w", path, err) + } + defer c.Close() + _ = c.SetDeadline(time.Now().Add(30 * time.Second)) + line, _ := json.Marshal(a) + if _, err := c.Write(append(line, '\n')); err != nil { + return nil, err + } + var reply struct { + Answer json.RawMessage `json:"answer"` + Error string `json:"error"` + } + if err := json.NewDecoder(c).Decode(&reply); err != nil { + return nil, fmt.Errorf("the watchman's answer cannot be read: %w", err) + } + if reply.Error != "" { + return nil, errors.New(reply.Error) + } + return reply.Answer, nil +} diff --git a/modules/mesh-delivery-watchman/internal/watch/advisories.go b/modules/mesh-delivery-watchman/internal/watch/advisories.go new file mode 100644 index 0000000..d4d0057 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/advisories.go @@ -0,0 +1,56 @@ +package watch + +import ( + "encoding/json" + "fmt" + "strings" +) + +// The bus's advisories the watchman hears (S9, mesh-controller internal/link/advisories.go at e04846e9): a durable consumer +// the mesh defines deleted is `consumer-lost`. A message given up on is said from DEAD_LETTERS, where it is kept, for as +// long as it is kept (DeadLettersHeld), not from its advisory. The controller's own slow consumer and refused subjects end +// with it (ADR 0564 §4). + +// Advisory subjects the watchman subscribes. +const ( + ConsumerDeletedAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>" + MaxDeliveriesAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>" +) + +// KindConsumerLost is a durable consumer of the mesh deleted. +const KindConsumerLost = "consumer-lost" + +// ReadAdvisory is one JetStream advisory as the mesh says it; false for one the watchman does not say. who words a +// consumer. +func ReadAdvisory(subject string, body []byte, who func(stream, consumer string) string) (Advisory, bool) { + var a struct { + Stream string `json:"stream"` + Consumer string `json:"consumer"` + } + if err := json.Unmarshal(body, &a); err != nil || a.Stream == "" || a.Consumer == "" { + return Advisory{}, false + } + if !strings.HasPrefix(subject, "$JS.EVENT.ADVISORY.CONSUMER.DELETED.") || !MeshNamed(a.Stream, a.Consumer) { + return Advisory{}, false + } + return Advisory{Kind: KindConsumerLost, ID: a.Stream + "." + a.Consumer, Machine: consumerMachine(a.Stream, a.Consumer), + Said: fmt.Sprintf("%s was deleted from the bus", who(a.Stream, a.Consumer))}, true +} + +// MeshNamed says a consumer is one of the durable consumers the mesh defines, by its name's shape; every other is a +// reader's own, named at random and deleted when its read is done. +func MeshNamed(stream, name string) bool { + switch { + case strings.HasPrefix(stream, "KV_") || stream == "ASSIGNMENTS": + return false + case stream == "NODES": + return true + case strings.HasPrefix(stream, "SEAT_"): + return strings.HasSuffix(name, "_worker") + case name == "controller" || name == "controller-merges": + return true + case stream == "EVENTS": + return strings.Contains(name, "_") + } + return false +} diff --git a/modules/mesh-delivery-watchman/internal/watch/dead.go b/modules/mesh-delivery-watchman/internal/watch/dead.go new file mode 100644 index 0000000..8919dd6 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/dead.go @@ -0,0 +1,78 @@ +package watch + +import ( + "fmt" + "sort" + "strings" +) + +// The dead letters' findings (novox/hq ADR 0564 §5, issue 330): a consumer DEAD_LETTERS holds a message for is said for as +// long as it holds any, `bus...max-deliveries`, and the notices waiting behind the controller's consumer +// after the lease write are said as `bus.DEAD_LETTER_NOTICES.controller.notices-waiting`. + +// KindNoticesWaiting is the finding of notices waiting while the controller's consumer stands after the lease write. +const KindNoticesWaiting = "notices-waiting" + +// DeadLettersHeld is DEAD_LETTERS' count per consumer, `.`, read whole: each consumer held writes its +// finding, each one no longer held has it deleted. who words a consumer. +func (w *Watchman) DeadLettersHeld(held map[string]int, who func(stream, consumer string) string) { + w.mu.Lock() + defer w.mu.Unlock() + want := map[string]bool{} + keys := make([]string, 0, len(held)) + for k := range held { + keys = append(keys, k) + } + sort.Strings(keys) + for _, key := range keys { + n := held[key] + stream, consumer, ok := strings.Cut(key, ".") + if !ok || n <= 0 { + continue + } + messages, them := "a message", "it is" + if n > 1 { + messages, them = fmt.Sprintf("%d messages", n), "they are" + } + f := Finding{Scope: ScopeBus, ID: key, Kind: KindDeadLetters, Severity: Warning, Source: "S9", + Machine: consumerMachine(stream, consumer), + Summary: fmt.Sprintf("%s gave up on %s; %s kept in DEAD_LETTERS until delivered again or dropped, with why", + who(stream, consumer), messages, them), + Said: fmt.Sprintf("%d held for %s", n, key), + Needs: "deliver them again or drop them, from the mesh MCP server."} + want[f.Key()] = true + w.find(f) + } + for key := range w.findings { + if strings.HasPrefix(key, ScopeBus+".") && strings.HasSuffix(key, "."+KindDeadLetters) && !want[key] { + w.unfind(key) + } + } +} + +// consumerMachine is the machine a module's consumer on EVENTS is on; empty for the others. +func consumerMachine(stream, consumer string) string { + if stream == "EVENTS" { + if node, _, ok := strings.Cut(consumer, "_"); ok { + return node + } + } + return "" +} + +// NoticesWaiting says whether the lease is handed away while the controller's consumer on DEAD_LETTER_NOTICES still +// stands: every notice then waits in the queue, kept a week, until the lease write's step removes the consumer. +func (w *Watchman) NoticesWaiting(waiting bool) { + w.mu.Lock() + defer w.mu.Unlock() + f := Finding{Scope: ScopeBus, ID: "DEAD_LETTER_NOTICES.controller", Kind: KindNoticesWaiting, Severity: Warning, Source: "dead-letters", + Summary: "the lease names no controller but the controller's consumer on DEAD_LETTER_NOTICES still stands, so every " + + "notice of a message given up on waits in the queue, kept a week, and nothing keeps it in DEAD_LETTERS", + Said: "the consumer controller on DEAD_LETTER_NOTICES stands after the lease write", + Needs: "remove the consumer controller on DEAD_LETTER_NOTICES with the bus module's tool, the lease write's step."} + if waiting { + w.find(f) + return + } + w.unfind(f.Key()) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/finding.go b/modules/mesh-delivery-watchman/internal/watch/finding.go new file mode 100644 index 0000000..fce809b --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/finding.go @@ -0,0 +1,159 @@ +package watch + +import ( + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The watchman is an observer of the seat mesh-watchdogs (novox/hq ADR 0564 §2, §4): what it observes wrong it writes as a +// finding, one key `..` of its own bucket `findings` in mesh-sdk's layout (health.Finding), only when what +// it says changes, and it deletes the key when it no longer observes it. The condition keeper (mesh-conditions) raises and +// clears the conditions from them; the watchman writes no condition and reads none (Limits 11). + +// FindingsBucket is the watchman's findings. +var FindingsBucket = health.FindingsBucket(health.WatchmanModule) + +// Observer is the name a finding of the watchman carries. +const Observer = health.WatchmanModule + +// The scopes and severities a finding of the watchman names, as the controller's conditions name them, so the keeper's +// `compare` meets the same keys. +const ( + ScopeMachine = health.ScopeMachine + ScopeDelivery = health.ScopeDelivery + ScopeModule = health.ScopeModule + ScopeProvider = health.ScopeProvider + ScopeBus = health.ScopeBus + ScopeCore = health.ScopeCore + + Urgent = string(health.Urgent) + Warning = string(health.Warning) +) + +// Finding is what the watchman builds of one thing it observes wrong: its subject (scope, id, the machine and the others +// it concerns), kind and severity, its evidence (Summary, and Said with Since, its one observation), the words the +// operator reads (ADR 0253; empty leaves the keeper's wording of the kind), and the row or fact it comes from (Source). +// It is written as health.Finding. +type Finding struct { + Scope, ID, Token, Kind, Machine string + Also []string + Severity string + Summary, Said string + Headline, Explanation string + Needs, Resolved string + Source string + Observer string + // Since is when what it says began, as the watchman saw it: the finding's time and its observation's, so the same + // thing observed again writes the same bytes. + Since time.Time +} + +// FindingKey composes a finding's key as the controller composes a condition's (health.Key). +func FindingKey(scope, id, token string) string { return health.FindingKey(scope, id, token) } + +// Key is the finding's key: its token, or its kind when it has none. +func (f Finding) Key() string { + token := f.Token + if token == "" { + token = f.Kind + } + return FindingKey(f.Scope, f.ID, token) +} + +// EncodeFinding is f in mesh-sdk's layout, refusing one the keeper could not make a condition of. +func EncodeFinding(f Finding) ([]byte, error) { + said := f.Said + if said == "" { + said = f.Summary + } + return health.EncodeFinding(health.Finding{Key: f.Key(), Kind: f.Kind, Token: f.Token, + Subject: health.Subject{Scope: f.Scope, ID: f.ID, Machine: f.Machine, Also: f.Also}, + Severity: health.Severity(f.Severity), Summary: f.Summary, Headline: f.Headline, Explanation: f.Explanation, + Resolved: f.Resolved, Needs: f.Needs, Evidence: []health.Evidence{{At: f.Since.UTC(), Said: said}}, + Source: f.Source, Observer: Observer, At: f.Since.UTC()}) +} + +// DecodeFinding reads a finding the watchman wrote. +func DecodeFinding(data []byte) (Finding, error) { + h, err := health.DecodeFinding(data) + if err != nil { + return Finding{}, err + } + f := Finding{Scope: h.Subject.Scope, ID: h.Subject.ID, Token: h.Token, Kind: h.Kind, Machine: h.Subject.Machine, + Also: h.Subject.Also, Severity: string(h.Severity), Summary: h.Summary, Headline: h.Headline, Explanation: h.Explanation, + Needs: h.Needs, Resolved: h.Resolved, Source: h.Source, Observer: h.Observer, Since: h.At} + if len(h.Evidence) > 0 { + f.Said = h.Evidence[0].Said + } + return f, nil +} + +// find writes a finding when what it says differs from what the bucket holds (ADR 0289 §1). +func (w *Watchman) find(f Finding) { + value, err := EncodeFinding(f) + if err != nil { + w.say("no finding: %v", err) + return + } + key := f.Key() + delete(w.seededFindings, key) + if string(w.findings[key]) == string(value) { + return + } + if err := w.out.PutFinding(key, value); err != nil { + w.say("cannot write the finding %s: %v", key, err) + return + } + w.findings[key] = value +} + +// unfind deletes a finding the watchman no longer observes. +func (w *Watchman) unfind(key string) { + if _, ok := w.findings[key]; !ok { + return + } + if err := w.out.DeleteFinding(key); err != nil { + w.say("cannot delete the finding %s: %v", key, err) + return + } + delete(w.findings, key) +} + +// SeedFinding tells the watchman a finding that stands in its bucket at its start; ReconcileFindings deletes it unless what +// the watchman reads at its start observes it again. +func (w *Watchman) SeedFinding(key string, value []byte) { + w.mu.Lock() + defer w.mu.Unlock() + w.findings[key] = value + w.seededFindings[key] = true +} + +// Findings is every finding the watchman holds now, by key. +func (w *Watchman) Findings() map[string]Finding { + w.mu.Lock() + defer w.mu.Unlock() + out := map[string]Finding{} + for k, v := range w.findings { + if f, err := DecodeFinding(v); err == nil { + out[k] = f + } + } + return out +} + +// Rewrite writes every finding the watchman holds again (resync, ADR 0564 §12): a bucket cleared or a key lost by hand is +// written back as the watchman observes it. It answers how many it wrote. +func (w *Watchman) Rewrite() int { + w.mu.Lock() + defer w.mu.Unlock() + n := 0 + for key, value := range w.findings { + if err := w.out.PutFinding(key, value); err != nil { + w.say("cannot write the finding %s again: %v", key, err) + continue + } + n++ + } + return n +} diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go new file mode 100644 index 0000000..030d477 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -0,0 +1,200 @@ +package watch + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func haveOf(t *testing.T, phase delivery.Phase, health delivery.Health, detail string) []byte { + t.Helper() + b, err := delivery.EncodeHave(delivery.Have{FP: fpNew, Phase: phase, Health: health, Detail: detail}) + if err != nil { + t.Fatal(err) + } + return b +} + +// expire hears the expiry of the pulse at pk. +func expire(w *Watchman, c *clock, pk string, after time.Duration) { + c.t = c.t.Add(after) + w.Pulse(pk, Expired, c.t) +} + +// The watchman's findings (novox/hq ADR 0564 §2, §4, §5, issue 1952 WM2-WM5): per row of the signals table a key not renewed +// writes its finding and a renewed one none; the haves' rows; stuck keys and presence as its own findings; the dead letters. +func TestTheWatchmansFindings(t *testing.T) { + cases := []struct { + name string + run func(t *testing.T, w *Watchman, o *fakeOut, c *clock) + // want are the finding keys standing at the end, with their severity. + want map[string]string + }{ + {"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Step("go", "n1.ask1", runningStep("ask1", t0), Put) + expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second) + k := FindingKey(ScopeDelivery, stuckKeyOf(t, checksKey, "ask1"), KindStuck) + if f := o.findings[k]; f.Kind != KindStuck || !strings.Contains(f.Summary, "the ask ask1 to end on n1") { + t.Fatalf("findings = %v", o.findings) + } + w.Step("go", "n1.ask1", endedStep("ask1", delivery.StepPassed, t0), Put) + }, map[string]string{}}, + {"WM2 S1: a node offline past its bound is silent, and back online it is not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Connect("node.ace") + w.Disconnect("node.ace") + expire(w, c, pulseKeyOf(t, presenceNS, "ace"), 120*time.Second) + if _, ok := o.findings["machine.ace.silent"]; !ok { + t.Fatalf("findings = %v", o.findings) + } + w.Connect("node.ace") + }, map[string]string{}}, + {"WM2 S1: a node that said it sleeps is not silent", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Connect("node.lap") + w.Power("lap", "sleeping") + w.Disconnect("node.lap") + expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second) + }, map[string]string{}}, + {"WM2 S1: a node back from sleep and still offline is silent after its bound", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Connect("node.lap") + w.Power("lap", "sleeping") + w.Disconnect("node.lap") + expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second) + w.Power("lap", "woke") + expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second) + }, map[string]string{"machine.lap.silent": Warning}}, + {"WM2 S11: a tool runner not heard for three intervals is silent; a heartbeat ends it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.ToolsAlive("ace", 0) + pk := signalKey("S11", "ace") + if o.ttls[pk] != 3*HeartbeatEvery { + t.Fatalf("ttl = %v", o.ttls[pk]) + } + expire(w, c, pk, 3*HeartbeatEvery) + if _, ok := o.findings["machine.ace.tools-silent"]; !ok { + t.Fatalf("findings = %v", o.findings) + } + w.ToolsAlive("ace", 20*time.Second) + if o.ttls[pk] != time.Minute { + t.Fatalf("ttl = %v, want three of the 20 s it says", o.ttls[pk]) + } + w.ToolsAlive("g14", 0) // renewed, never expired: no finding + }, map[string]string{}}, + {"WM2 S10: the self-check not heard for three runs is urgent; not before its first run", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + if len(o.pulses) != 0 { + t.Fatal("a pulse before the first run") + } + w.Checked("novox") + expire(w, c, signalKey("S10", "doctor"), 3*SelfCheckEvery) + }, map[string]string{"core.doctor.self-check-silent": Urgent}}, + {"WM3 S8: a failing word is its finding; not said again for 30 min it is silent; recovered ends both", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + st, err := ReadStanding("mesh.mod.keycloak.event.provisioner.failing", + []byte(`{"provider":"oidc","provider-node":"ace","consumer":"photos","node":"g14","class":"refused","attempts":4,"since":"2026-10-18T11:00:00Z"}`)) + if err != nil { + t.Fatal(err) + } + w.Stood(st) + if f := o.findings["provider.keycloak.ace.photos.failing"]; f.Kind != KindProviderFailing || len(f.Also) != 1 { + t.Fatalf("findings = %v", o.findings) + } + expire(w, c, signalKey("S8", "keycloak.ace.photos"), ProviderSaysAgainWithin) + if _, ok := o.findings["provider.keycloak.ace.photos.silent"]; !ok { + t.Fatalf("findings = %v", o.findings) + } + st.Failing = false + w.Stood(st) + }, map[string]string{}}, + {"WM3 S9: an advisory is a finding counted, ended by an hour's quiet", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + a, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.DELETED.EVENTS.ace_web", []byte(`{"stream":"EVENTS","consumer":"ace_web"}`), + func(s, c string) string { return c + " on " + s }) + if !ok { + t.Fatal("a mesh consumer's deletion is not read") + } + if _, ok := ReadAdvisory("$JS.EVENT.ADVISORY.CONSUMER.DELETED.KV_x.abc", []byte(`{"stream":"KV_x","consumer":"abc"}`), nil); ok { + t.Fatal("a reader's own consumer is said") + } + w.Advised(a) + w.Advised(a) + k := "bus.EVENTS.ace_web.consumer-lost" + if f := o.findings[k]; !strings.Contains(f.Summary, "2 times") || f.Machine != "ace" { + t.Fatalf("findings = %v", o.findings) + } + expire(w, c, signalKey("S9", k), AdvisoryQuiet) + }, map[string]string{}}, + {"WM3 S21: a have failed for one apply interval is apply-failing; applied again it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Have("ace.web", haveOf(t, delivery.PhaseFailed, delivery.HealthUnknown, "refused: the package manager (pacman) does not answer"), Put) + expire(w, c, signalKey("S21", "ace.web"), ApplyEvery) + if f := o.findings["machine.ace.web.apply-failing"]; !strings.Contains(f.Said, "pacman") { + t.Fatalf("findings = %v", o.findings) + } + w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put) + w.Have("ace.mail", haveOf(t, delivery.PhaseFailed, delivery.HealthUnknown, "refused"), Put) + w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put) + expire(w, c, signalKey("S21", "ace.mail"), ApplyEvery) // applied before its bound: nothing + }, map[string]string{}}, + {"WM3: unhealthy only at the second write in a row, urgent after four hours", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "the unit failed"), Put) + if len(o.findings) != 0 { + t.Fatalf("one write raised %v", o.findings) + } + w.Have("ace.web", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "the unit failed"), Put) + if o.findings["module.web.ace.unhealthy"].Severity != Warning { + t.Fatalf("findings = %v", o.findings) + } + expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter) + w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "a part waits for the secret token"), Put) + }, map[string]string{"module.web.ace.unhealthy": Urgent, "module.mail.ace.needs-operator": Warning}}, + {"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "waits"), Put) + w.Have("ace.mail", haveOf(t, delivery.PhaseRemoved, delivery.HealthUnknown, ""), Put) + }, map[string]string{}}, + {"WM4: resync after a cleared bucket writes the same findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "waits"), Put) + o.findings = map[string]Finding{} + if n := w.Rewrite(); n != 1 { + t.Fatalf("rewrote %d", n) + } + }, map[string]string{"module.mail.ace.needs-operator": Warning}}, + {"WM4: a finding read at the start and not observed again is deleted; a signal's stays", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + for _, f := range []Finding{ + {Scope: ScopeDelivery, ID: "x.y", Kind: KindStuck, Severity: Warning, Summary: "old", Source: "stuck"}, + {Scope: ScopeMachine, ID: "ace", Kind: KindToolsSilent, Severity: Warning, Summary: "silent", Source: "S11"}, + } { + v, _ := EncodeFinding(f) + o.findings[f.Key()], _ = DecodeFinding(v) + w.SeedFinding(f.Key(), v) + } + w.Reconcile() + }, map[string]string{"machine.ace.tools-silent": Warning}}, + {"WM5: a consumer DEAD_LETTERS holds for is its finding until it holds none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + who := func(s, c string) string { return c } + w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 2, "SEAT_X.y_worker": 1}, who) + if f := o.findings["bus.EVENTS.ace_web.max-deliveries"]; f.Machine != "ace" || !strings.Contains(f.Summary, "2 messages") { + t.Fatalf("findings = %v", o.findings) + } + w.DeadLettersHeld(map[string]int{"SEAT_X.y_worker": 1}, who) + w.NoticesWaiting(true) + }, map[string]string{"bus.SEAT_X.y_worker.max-deliveries": Warning, "bus.DEAD_LETTER_NOTICES.controller.notices-waiting": Warning}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w, o, c := setup() + tc.run(t, w, o, c) + got := map[string]string{} + for k, f := range o.findings { + got[k] = f.Severity + if f.Key() != k || f.Observer != Observer { + t.Errorf("finding %s = %+v", k, f) + } + } + if len(got) != len(tc.want) { + t.Fatalf("findings = %v, want %v", got, tc.want) + } + for k, sev := range tc.want { + if got[k] != sev { + t.Fatalf("findings = %v, want %v", got, tc.want) + } + } + }) + } +} diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 78e54c5..4eee236 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -116,6 +116,7 @@ func (w *Watchman) Have(key string, value []byte, op Op) { return case Delete: delete(w.haves, key) + w.haveFindings(key, node, module, nil) return } have, err := delivery.DecodeHave(value) @@ -124,6 +125,7 @@ func (w *Watchman) Have(key string, value []byte, op Op) { return } w.haves[key] = have.FP + w.haveFindings(key, node, module, &have) wk, err := delivery.WantKey(node, module) if err != nil { return diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go new file mode 100644 index 0000000..949d6d5 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -0,0 +1,524 @@ +package watch + +import ( + "encoding/json" + "fmt" + "regexp" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The rows of the signals table the watchman holds (novox/hq ADR 0564 §4, Limits 3 and 11). Each is a key of the pulse +// bucket whose time to live is the row's bound, renewed by the signal's own event: its expiry is the finding, so no loop +// looks every half minute. A signal's pulse is kept under the pulse bucket's pseudo-bucket `signal`, `signal..`, +// and its value says what its expiry does (signalDue), so a pulse that stood across a restart still acts when it expires. +// +// What each row hears is on the bus, never in the controller's store: +// - S1, a machine silent: the watchman's own presence of `node.` (its stuck key), not while the machine said it +// sleeps or shuts down (the power seat's events, ADR 0211); +// - S8, a provider's failing word repeated: the providers' `provisioner.failing` and `provisioner.recovered` (ADR 0224); +// the failing word is itself the finding `provider-failing`, and its pulse of 30 minutes, renewed by each repeat, is S8; +// - S9, the bus's advisories: each occurrence a finding, cleared by the expiry of an hour's pulse renewed by the next; +// - S10, the self-check's `checked`: a pulse of three runs' length, from the first run heard; +// - S11, a machine's tool runner heartbeat on `mesh.control..tools-alive`: a pulse of three intervals; +// - S21, a module's have failed: a pulse of one apply interval from the first failure, whose expiry while it still fails +// is the finding `apply-failing`; +// - a module's have unhealthy in two writes in a row, `unhealthy`, urgent after four hours; and a have that waits for +// the operator, `needs-operator` (ADR 0240 §12, §17). + +// The bounds of the rows. +const ( + // ProviderSaysAgainWithin is how long a provider's failing word stays current without being said again (S8): twice + // the quarter of an hour it repeats it at. + ProviderSaysAgainWithin = 30 * time.Minute + // AdvisoryQuiet is how long the bus must be quiet about a thing before its advisory clears (S9). + AdvisoryQuiet = time.Hour + // SelfCheckEvery is the self-check's run interval; S10's bound is three of them. + SelfCheckEvery = 5 * time.Minute + // HeartbeatEvery is a tool runner's interval when it says none; S11's bound is three of them. + HeartbeatEvery = 60 * time.Second + // ApplyEvery is one apply interval (S21): a failure still there after it is not one the next apply clears. + ApplyEvery = 5 * time.Minute + // UnhealthyUrgentAfter is how long a module stays unhealthy before it is urgent. + UnhealthyUrgentAfter = 4 * time.Hour +) + +// signalBucket is the pseudo-bucket of a signal's pulse key. +const signalBucket = "signal" + +// signalDue is what a signal's pulse does when it expires: raise the finding (find), or end it (end, an advisory's quiet +// hour). Kept as the pulse's Owes, so it is read back at a restart. +type signalDue struct { + Do string `json:"do"` + Finding Finding `json:"finding"` + // While names what must still stand at the expiry for a find to be written: a have's key still failing or unhealthy; + // empty for a row whose expiry alone is the finding. + While string `json:"while,omitempty"` +} + +const ( + dueFind = "find" + dueEnd = "end" +) + +// signalKey is the pulse key of a row's signal about id. +func signalKey(row, id string) string { + parts := []string{signalBucket, row} + for _, p := range strings.Split(id, ".") { + if p = unsafeKey.ReplaceAllString(strings.TrimSpace(p), "_"); p == "" { + p = "_" + } + parts = append(parts, p) + } + return strings.Join(parts, ".") +} + +// unsafeKey is what a key's token may not hold: the bus takes letters, digits and `-_/=`. +var unsafeKey = regexp.MustCompile(`[^A-Za-z0-9_=/-]`) + +// signal writes (or renews) the pulse of a row's signal, whose expiry does due. +func (w *Watchman) signal(row, id string, ttl time.Duration, due signalDue) { + pk := signalKey(row, id) + owes, err := json.Marshal(due) + if err != nil { + w.say("no signal %s: %v", pk, err) + return + } + value, err := delivery.EncodePulse(delivery.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)}) + if err != nil { + w.say("no signal %s: %v", pk, err) + return + } + if err := w.out.PutPulse(pk, value, ttl); err != nil { + w.say("cannot write the signal %s: %v", pk, err) + return + } + w.signals[pk] = due +} + +// unsignal ends a row's signal about id: its pulse deleted, so it never expires. +func (w *Watchman) unsignal(row, id string) { + pk := signalKey(row, id) + if _, ok := w.signals[pk]; !ok { + return + } + delete(w.signals, pk) + if err := w.out.DeletePulse(pk); err != nil { + w.say("cannot delete the signal %s: %v", pk, err) + } +} + +// SeedSignal tells the watchman a signal's pulse that stands at its start, with its value: it is kept, since its event +// renews it and its expiry still acts. +func (w *Watchman) SeedSignal(key string, value []byte) { + w.mu.Lock() + defer w.mu.Unlock() + if due, ok := dueOf(value); ok { + w.signals[key] = due + } +} + +func dueOf(value []byte) (signalDue, bool) { + p, err := delivery.DecodePulse(value) + if err != nil || p.Bucket != signalBucket { + return signalDue{}, false + } + var due signalDue + if json.Unmarshal([]byte(p.Owes), &due) != nil || (due.Do != dueFind && due.Do != dueEnd) { + return signalDue{}, false + } + return due, true +} + +// signalExpired is a signal's pulse expired: its due is done. +func (w *Watchman) signalExpired(pk string) { + due, ok := w.signals[pk] + if !ok { + return + } + delete(w.signals, pk) + switch due.Do { + case dueEnd: + w.unfind(due.Finding.Key()) + case dueFind: + if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) { + return + } + w.find(due.Finding) + } +} + +// stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for +// unhealthy. +func (w *Watchman) stillWrong(haveKey, kind string) bool { + h, ok := w.haveState[haveKey] + if !ok { + return false + } + switch kind { + case KindApplyFailing: + return h.Phase == delivery.PhaseFailed + case KindUnhealthy: + return h.Health == delivery.HealthFailed + } + return false +} + +// The kinds the watchman's rows write, as the controller's conditions name them. +const ( + KindSilent = "silent" + KindStuck = "stuck" + KindProviderFailing = "provider-failing" + KindProviderSilent = "provider-silent" + KindSelfCheckSilent = "self-check-silent" + KindToolsSilent = "tools-silent" + KindApplyFailing = "apply-failing" + KindUnhealthy = "module-unhealthy" + KindNeedsOperator = "needs-operator" + KindDeadLetters = "max-deliveries" +) + +func stamp(t time.Time) string { return t.UTC().Format("2006-01-02 15:04 MST") } + +// ---- S1 and the power seat ---- + +// PowerEvents are the power seat's events that say whether a machine is about to be away or is back (ADR 0211). +var PowerEvents = map[string]bool{"sleeping": true, "shutting-down": true, "booted": true, "woke": true} + +// Power is a machine's power event: sleeping and shutting-down say it is away, booted and woke that it is back. A machine +// that said it is away is not silent; one back that is still offline is watched again from now. +func (w *Watchman) Power(node, state string) { + if !PowerEvents[state] || node == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + away := state == "sleeping" || state == "shutting-down" + w.away[node] = away + key, err := delivery.PresenceNodeKey(node) + if err != nil { + return + } + if away { + w.unfind(FindingKey(ScopeMachine, node, KindSilent)) + return + } + if p, ok := w.presence[key]; ok && !p.online { + p.since = w.now() + w.presence[key] = p + w.ensureOffline(key) + } +} + +// stuckFinding is the finding of a stuck key as it stands (ADR 0433 §2, §3, now the watchman's own): a node's presence not +// moved is S1, the machine silent, unless it said it is away; any other key not moved is `delivery..stuck`; a key +// moved ends its finding. +func (w *Watchman) stuckFinding(sk string, s delivery.Stuck) { + if s.Bucket == delivery.PresenceBucket { + if node, part, _, err := delivery.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" { + key := FindingKey(ScopeMachine, node, KindSilent) + if s.Moved || w.away[node] { + w.unfind(key) + return + } + w.find(Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1", + Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), delivery.PulseMoveBound), + Said: fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since}) + return + } + } + key := FindingKey(ScopeDelivery, sk, KindStuck) + if s.Moved { + w.unfind(key) + return + } + waited := s.Waited + if waited == "" { + waited = "a move" + } + w.find(Finding{Scope: ScopeDelivery, ID: sk, Kind: KindStuck, Severity: Warning, Source: "stuck", + Summary: fmt.Sprintf("%s has not moved since %s: it waited for %s", sk, stamp(s.Since), waited), + Said: fmt.Sprintf("%s: waiting for %s since %s", sk, waited, s.Since.UTC().Format(time.RFC3339)), Since: s.Since}) +} + +// ---- S8 ---- + +// Standing is a provider's word about one consumer (ADR 0224): Module is the emitter, read from the subject. +type Standing struct { + Module string `json:"-"` + Failing bool `json:"-"` + Provider string `json:"provider"` + ProviderNode string `json:"provider-node"` + Consumer string `json:"consumer"` + Node string `json:"node"` + Class string `json:"class,omitempty"` + Error string `json:"error,omitempty"` + Since time.Time `json:"since"` + Attempts int `json:"attempts"` + Why string `json:"why,omitempty"` +} + +// ReadStanding reads a provider's word from its subject, `mesh.mod..event.provisioner.failing` or `.recovered`. +func ReadStanding(subject string, body []byte) (Standing, error) { + rest, ok := strings.CutPrefix(subject, "mesh.mod.") + if !ok { + return Standing{}, fmt.Errorf("%s is not a provider's word", subject) + } + module, event, ok := strings.Cut(rest, ".event.") + if !ok || module == "" || (event != "provisioner.failing" && event != "provisioner.recovered") { + return Standing{}, fmt.Errorf("%s is not a provider's word", subject) + } + var st Standing + if err := json.Unmarshal(body, &st); err != nil { + return Standing{}, fmt.Errorf("%s's word could not be read: %w", module, err) + } + if st.Consumer == "" { + return Standing{}, fmt.Errorf("%s's word names no consumer", module) + } + st.Module, st.Failing = module, event == "provisioner.failing" + return st, nil +} + +// Stood is a provider's word: failing writes `provider....failing` and renews S8's pulse, whose +// expiry writes `provider-silent`; recovered deletes both (Limits 11). +func (w *Watchman) Stood(st Standing) { + w.mu.Lock() + defer w.mu.Unlock() + id := st.Module + "." + st.ProviderNode + "." + st.Consumer + failing := FindingKey(ScopeProvider, id, "failing") + silentKey := FindingKey(ScopeProvider, id, "silent") + if !st.Failing { + w.unfind(failing) + w.unfind(silentKey) + w.unsignal("S8", id) + return + } + class := st.Class + if class == "" { + class = "failing" + } + whom := st.Consumer + if st.Node != "" { + whom += " on " + st.Node + } + said := class + if st.Error != "" { + said += ": " + firstLine(st.Error) + } + var also []string + if st.Node != "" && st.Node != st.ProviderNode { + also = []string{st.Node} + } + w.find(Finding{Scope: ScopeProvider, ID: id, Token: "failing", Kind: KindProviderFailing, Machine: st.ProviderNode, Also: also, + Severity: Warning, Source: "provisioner.failing", Since: st.Since, Said: said, + Summary: fmt.Sprintf("%s on %s keeps failing %s: %s, %d attempt(s) since %s", st.Module, st.ProviderNode, whom, class, + st.Attempts, stamp(st.Since))}) + w.unfind(silentKey) + now := w.now() + w.signal("S8", id, ProviderSaysAgainWithin, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeProvider, ID: id, Token: "silent", + Kind: KindProviderSilent, Machine: st.ProviderNode, Severity: Warning, Source: "S8", Since: now, + Summary: fmt.Sprintf("%s on %s said it keeps failing %s and has said nothing since %s: it stopped saying anything, so "+ + "its last word is all the mesh has", st.Module, st.ProviderNode, st.Consumer, stamp(now)), + Said: fmt.Sprintf("not said again within %s", ProviderSaysAgainWithin)}}) +} + +func firstLine(s string) string { + s, _, _ = strings.Cut(strings.TrimSpace(s), "\n") + return s +} + +// ---- S9 ---- + +// Advisory is one thing the bus said (S9): its kind (slow-consumer, connection-closed, max-deliveries, refused, +// consumer-lost), what it is about, its words, and the machine when it names one. +type Advisory struct { + Kind, ID, Said, Machine string +} + +// Advised is an advisory heard: its finding `bus..` is written at once, counting the times since its first, and +// an hour's pulse renewed by each, whose expiry ends it. +func (w *Watchman) Advised(a Advisory) { + if a.Kind == "" || a.ID == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + f := Finding{Scope: ScopeBus, ID: a.ID, Kind: a.Kind, Machine: a.Machine, Severity: Warning, Source: "S9", Said: a.Said} + key := f.Key() + n := w.advised[key] + 1 + if _, open := w.findings[key]; !open { + n = 1 + w.advisedSince[key] = w.now() + } + w.advised[key] = n + since := w.advisedSince[key] + f.Since, f.Summary = since, a.Said + if n > 1 { + f.Summary = fmt.Sprintf("%s (%d times since %s)", a.Said, n, since.UTC().Format("15:04 MST")) + } + w.find(f) + w.signal("S9", key, AdvisoryQuiet, signalDue{Do: dueEnd, Finding: f}) +} + +// ---- S10 ---- + +// Checked is the self-check's `checked` heard: its pulse of three runs' length is renewed, and its silence ended. Before +// the first run heard there is no pulse, so a self-check not yet placed is not said silent. +func (w *Watchman) Checked(machine string) { + w.mu.Lock() + defer w.mu.Unlock() + w.unfind(FindingKey(ScopeCore, "doctor", KindSelfCheckSilent)) + now := w.now() + w.signal("S10", "doctor", 3*SelfCheckEvery, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeCore, ID: "doctor", + Kind: KindSelfCheckSilent, Machine: machine, Severity: Urgent, Source: "S10", Since: now, + Summary: fmt.Sprintf("the self-check has not finished a run since %s (it runs every %s): the mesh's invariants are not "+ + "being checked", stamp(now), SelfCheckEvery), + Said: fmt.Sprintf("no run within %s", 3*SelfCheckEvery)}}) +} + +// ---- S11 ---- + +// ToolsAlive is a machine's tool runner heartbeat, saying its interval (zero: HeartbeatEvery): its pulse of three +// intervals is renewed and its silence ended. +func (w *Watchman) ToolsAlive(node string, every time.Duration) { + if node == "" { + return + } + if every <= 0 { + every = HeartbeatEvery + } + w.mu.Lock() + defer w.mu.Unlock() + w.unfind(FindingKey(ScopeMachine, node, KindToolsSilent)) + if w.away[node] { + w.unsignal("S11", node) + return + } + now := w.now() + w.signal("S11", node, 3*every, signalDue{Do: dueFind, Finding: Finding{Scope: ScopeMachine, ID: node, Kind: KindToolsSilent, + Machine: node, Severity: Warning, Source: "S11", Since: now, + Summary: fmt.Sprintf("the node tools on %s have not said they are there since %s (bound %s): nothing can ask that "+ + "machine anything", node, stamp(now), 3*every), + Said: fmt.Sprintf("silent for more than %s", 3*every)}}) +} + +// ToolsNode is the machine a tool runner heartbeat's subject names, `mesh.control..tools-alive`. +func ToolsNode(subject string) string { + rest, ok := strings.CutPrefix(subject, "mesh.control.") + if !ok { + return "" + } + node, ok := strings.CutSuffix(rest, ".tools-alive") + if !ok || strings.Contains(node, ".") { + return "" + } + return node +} + +// ---- S21, unhealthy and needs-operator, from the haves ---- + +// haveFindings reads a have's change for the rows the haves feed: S21 (failed for one apply interval), unhealthy (two +// writes in a row, urgent after four hours) and needs-operator (it waits for the operator). A have removed or deleted ends +// them all. +func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Have) { + applyKey := FindingKey(ScopeMachine, node+"."+module, KindApplyFailing) + unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy") + waitsKey := FindingKey(ScopeModule, module+"."+node, KindNeedsOperator) + if h == nil || h.Phase == delivery.PhaseRemoved { + delete(w.haveState, key) + delete(w.unhealthy, key) + w.unsignal("S21", node+"."+module) + w.unsignal("unhealthy", module+"."+node) + w.unfind(applyKey) + w.unfind(unhealthyKey) + w.unfind(waitsKey) + return + } + was, had := w.haveState[key] + w.haveState[key] = *h + now := w.now() + + // S21: a have refused or applied in part says phase failed, with the node's words in its detail. + if h.Phase == delivery.PhaseFailed { + if !had || was.Phase != delivery.PhaseFailed { + w.signal("S21", node+"."+module, ApplyEvery, signalDue{Do: dueFind, While: key, Finding: Finding{Scope: ScopeMachine, + ID: node + "." + module, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: now, + Summary: fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, module, stamp(now), ApplyEvery), + Said: firstLine(orWords(h.Detail, "no reason given")), + Headline: node + " is not applying its update", Resolved: node + " applies what it is sent again"}}) + } + } else { + w.unsignal("S21", node+"."+module) + w.unfind(applyKey) + } + + // unhealthy: two writes in a row that say it failed. + if h.Health == delivery.HealthFailed { + w.unhealthy[key]++ + if w.unhealthy[key] == 2 { + f := Finding{Scope: ScopeModule, ID: module + "." + node, Token: "unhealthy", Kind: KindUnhealthy, Machine: node, + Severity: Warning, Source: "health", Since: now, + Summary: fmt.Sprintf("%s on %s is not healthy: %s", module, node, orWords(h.Detail, "its check failed")), + Said: firstLine(orWords(h.Detail, "failed")), + Headline: fmt.Sprintf("%s not working on %s", module, node)} + w.find(f) + f.Severity = Urgent + w.signal("unhealthy", module+"."+node, UnhealthyUrgentAfter, signalDue{Do: dueFind, While: key, Finding: f}) + } + } else { + delete(w.unhealthy, key) + w.unsignal("unhealthy", module+"."+node) + w.unfind(unhealthyKey) + } + + // needs-operator: a part waits for the operator's secret or setting. + if h.Health == delivery.HealthWaits { + w.find(Finding{Scope: ScopeModule, ID: module + "." + node, Kind: KindNeedsOperator, Machine: node, Severity: Warning, + Source: "health", Since: now, + Summary: fmt.Sprintf("%s on %s waits for the operator: %s", module, node, orWords(h.Detail, "a part waits for a secret or a setting")), + Said: firstLine(orWords(h.Detail, "waits"))}) + } else { + w.unfind(waitsKey) + } +} + +func orWords(s, otherwise string) string { + if strings.TrimSpace(s) == "" { + return otherwise + } + return s +} + +// SignalRows are the rows the watchman holds, in the table's order, for `signals`. +var SignalRows = []struct{ Row, Signal, Bound, Kind string }{ + {"S1", "a machine's presence (not while it said it sleeps or shuts down)", delivery.PulseMoveBound.String(), KindSilent}, + {"S8", "a provider's failing word repeated", ProviderSaysAgainWithin.String(), KindProviderSilent}, + {"S9", "the bus's advisories", "any occurrence; clears after " + AdvisoryQuiet.String() + " without another", "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost"}, + {"S10", "the self-check's checked", (3 * SelfCheckEvery).String(), KindSelfCheckSilent}, + {"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent}, + {"S21", "a have refused or applied in part", ApplyEvery.String(), KindApplyFailing}, + {"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy}, + {"health", "a have that waits for the operator", "at once", KindNeedsOperator}, +} + +// Signal is one signal's pulse as `signals` shows it. +type Signal struct { + Key string `json:"key"` + Do string `json:"do"` + Finding Finding `json:"finding"` +} + +// Signals is every signal's pulse the watchman holds, sorted by key. +func (w *Watchman) Signals() []Signal { + w.mu.Lock() + defer w.mu.Unlock() + out := make([]Signal, 0, len(w.signals)) + for k, d := range w.signals { + out = append(out, Signal{Key: k, Do: d.Do, Finding: d.Finding}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) + return out +} diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index e81613b..50a42e6 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -31,6 +31,9 @@ type Out interface { // DeletePulse deletes the pulse key (a delete, not an expiry). DeletePulse(key string) error PutStuck(key string, value []byte) error + // PutFinding writes one of the watchman's findings; DeleteFinding deletes one it no longer observes. + PutFinding(key string, value []byte) error + DeleteFinding(key string) error } type presence struct { @@ -60,6 +63,20 @@ type Watchman struct { wants map[string]string // the fingerprint of each want that stands, by want key haves map[string]string // the fingerprint each node runs of a module, by have key bound time.Duration + // findings is every finding written, by key, as written; seededFindings those read at the start and not yet observed + // again. + findings map[string][]byte + seededFindings map[string]bool + // signals is every signal's pulse standing, by pulse key, with what its expiry does. + signals map[string]signalDue + // away is each machine that said it sleeps or shuts down, until it says it is back. + away map[string]bool + // haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each + // advisory's count while its finding stands and advisedSince since when. + haveState map[string]delivery.Have + unhealthy map[string]int + advised map[string]int + advisedSince map[string]time.Time // stepBound is how long a build step may run once a node took the ask; a test shortens it. stepBound time.Duration } @@ -67,7 +84,9 @@ type Watchman struct { // New is a watchman writing to out, reading the time from now. func New(out Out, now func() time.Time) *Watchman { return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{}, - pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound} + pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, bound: delivery.PulseMoveBound, stepBound: StepBound, + findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{}, + haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{}} } // Logf sets where a write that failed is said; it is silent until set. @@ -178,6 +197,7 @@ func (w *Watchman) putStuck(sk string, s delivery.Stuck) { } s.V = delivery.Version w.stuck[sk] = s + w.stuckFinding(sk, s) } // Pulse is a change heard on the pulse bucket. Only the server's expiry marker of a pulse the watchman made writes a @@ -188,6 +208,10 @@ func (w *Watchman) Pulse(key string, op Op, at time.Time) { if op != Expired { return } + if _, ok := w.signals[key]; ok { + w.signalExpired(key) + return + } p, ok := w.pulses[key] if !ok || p.seeded { return @@ -202,6 +226,7 @@ func (w *Watchman) SeedStuck(key string, value []byte) { defer w.mu.Unlock() if s, err := delivery.DecodeStuck(value); err == nil { w.stuck[key] = s + w.stuckFinding(key, s) } } @@ -211,7 +236,7 @@ func (w *Watchman) SeedPulse(key string) { w.mu.Lock() defer w.mu.Unlock() bucket, k, err := delivery.ParsePulseKey(key) - if err != nil { + if err != nil || bucket == signalBucket { return } w.pulses[key] = &pulse{bucket: bucket, key: k, seeded: true} @@ -227,6 +252,38 @@ func (w *Watchman) Reconcile() { w.deletePulse(pk) } } + w.reconcileFindings() +} + +// reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not +// observe again is deleted, unless a signal still standing keeps it (a provider's failing word, an advisory's quiet hour, a +// machine's silent tools, a silent self-check): those end by their own events or expiry. +func (w *Watchman) reconcileFindings() { + kept := map[string]bool{} + for _, due := range w.signals { + kept[due.Finding.Key()] = true + } + for key := range w.seededFindings { + delete(w.seededFindings, key) + if kept[key] || signalKind(w.findings[key]) { + continue + } + w.unfind(key) + } +} + +// signalKind says a finding is one a signal row writes and ends by its own event (S8, S9, S10, S11): it outlives a +// restart, since what ends it is heard, not read. +func signalKind(value []byte) bool { + f, err := DecodeFinding(value) + if err != nil { + return false + } + switch f.Source { + case "provisioner.failing", "S8", "S9", "S10", "S11": + return true + } + return false } func describe(format string, args ...any) string { return fmt.Sprintf(format, args...) } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 28ea6f0..edcdec9 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -19,10 +19,31 @@ type fakeOut struct { pulseDeletes []string stuck map[string]delivery.Stuck stuckWrites int + findings map[string]Finding + findingWrites int } func newOut() *fakeOut { - return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{}} + return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{}, + findings: map[string]Finding{}} +} + +func (o *fakeOut) PutFinding(key string, value []byte) error { + f, err := DecodeFinding(value) + if err != nil { + return err + } + if f.Key() != key { + return errors.New("a finding written under another key than its own: " + key) + } + o.findings[key] = f + o.findingWrites++ + return nil +} + +func (o *fakeOut) DeleteFinding(key string) error { + delete(o.findings, key) + return nil } func (o *fakeOut) PutPresence(key string, online bool) error { @@ -116,7 +137,12 @@ func stuckKeyOf(t *testing.T, bucket, key string) string { return k } -func TestPresenceFromEvents(t *testing.T) { // 1135/AC1 +func TestPresenceFromEventsWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1 + t.Run("from events", presenceFromEvents) + t.Run("written only when it differs", presenceWrittenOnlyWhenItDiffers) +} + +func presenceFromEvents(t *testing.T) { // 1135/AC1 w, o, _ := setup() w.Connect("node.ace") w.Connect("ace.mesh-delivery-view") @@ -135,7 +161,7 @@ func TestPresenceFromEvents(t *testing.T) { // 1135/AC1 } } -func TestPresenceWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1 +func presenceWrittenOnlyWhenItDiffers(t *testing.T) { // 1135/AC1 w, o, _ := setup() w.Connect("node.ace") w.Connect("node.ace") @@ -178,7 +204,12 @@ func connzAnswer(users ...string) []byte { return b } -func TestRebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2 +func TestRebuildAsksConnectionsAndTakesWhoIsGone(t *testing.T) { // 1135/AC2 + t.Run("asks the connections with auth", rebuildAsksConnectionsWithAuth) + t.Run("takes the presence of who is gone", rebuildTakesThePresenceOfWhoIsGone) +} + +func rebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2 w, o, _ := setup() var body []byte err := w.Rebuild(func(b []byte) ([]byte, error) { @@ -201,7 +232,7 @@ func TestRebuildAsksConnectionsWithAuth(t *testing.T) { // 1135/AC2 } } -func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2 +func rebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2 w, o, c := setup() w.SeedPresence("ace", true, t0.Add(-time.Hour)) w.SeedPresence("g14", true, t0.Add(-time.Hour)) @@ -416,7 +447,12 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4 } } -func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 +func TestOnlyTheExpiryOfItsOwnPulseWritesStuck(t *testing.T) { // 1135/AC5 + t.Run("the expiry marker writes stuck", expiryMarkerWritesStuck) + t.Run("its own delete of a pulse writes nothing", ownDeleteOfAPulseWritesNothing) +} + +func expiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 w, o, c := setup() w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) c.t = t0.Add(120 * time.Second) @@ -430,7 +466,7 @@ func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5 } } -func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5 +func ownDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5 w, o, c := setup() w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) pk := pulseKeyOf(t, checksKey, "askid1") diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 58e2171..3b8a9a0 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -25,19 +25,28 @@ type state struct { } type artifact struct { - Name string `json:"name"` - Kind string `json:"kind"` - Language string `json:"language"` - System string `json:"system"` - From string `json:"from"` - Binary string `json:"binary"` + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` + Loads []string `json:"loads"` Context *struct { Repository string `json:"repository"` } `json:"context"` } +type claim struct { + Name string `json:"name"` + Scope string `json:"scope"` + Serves []string `json:"serves"` +} + type manifest struct { Module string `json:"module"` + Claims []claim `json:"claims"` + Tools []string `json:"tools"` Moved bool `json:"moved"` Version string `json:"version"` Slug string `json:"slug"` @@ -75,16 +84,24 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 if m.Module != "mesh-delivery-watchman" || m.Moved || m.Slug != "watchman" { t.Errorf("module %q moved %v (it has no template, so it is not moved) slug %q", m.Module, m.Moved, m.Slug) } - want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}} + want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}} if !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } reads := slices.Clone(m.Reads) slices.Sort(reads) - if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-delivery-assigner.wants", "node-engine.have"}) { + if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-controller.lease", + "mesh-delivery-assigner.wants", "node-engine.have"}) { t.Errorf("reads = %v", reads) } - for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} { + // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). + if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) { + t.Errorf("claims = %+v", m.Claims) + } + if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) { + t.Errorf("tools = %v", m.Tools) + } + for _, k := range []string{"emits", "context", "items", "consumes", "provides"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the watchman has none", k) } @@ -97,25 +114,28 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 } } -func TestManifestIsWholeInItsRepository(t *testing.T) { // 1135/AC6 - m, _ := load(t) - if len(m.Build.Artifacts) != 1 { - t.Fatalf("artifacts = %+v, want the one bundle program", m.Build.Artifacts) +// The manifest is whole in its repository: the program, and the tools bundle the runtime loads (1952/WM1). +func manifestIsWholeInItsRepository(t *testing.T, m manifest) { // 1135/AC6 + if len(m.Build.Artifacts) != 2 { + t.Fatalf("artifacts = %+v, want the program and the tools bundle", m.Build.Artifacts) } - a := m.Build.Artifacts[0] - if a.Name != "program" || a.Kind != "bundle" || a.Language != "go" || a.System != "arch" || a.Context != nil { - t.Errorf("artifact = %+v", a) - } - if st, err := os.Stat(a.From); err != nil || !st.IsDir() { - t.Errorf("the artifact builds from %q, which is not a directory of this module", a.From) - } - if a.Binary != "mesh-delivery-watchman" || !strings.HasSuffix(a.From, a.Binary) { - t.Errorf("binary %q from %q", a.Binary, a.From) + for i, name := range []string{"program", "tools"} { + a := m.Build.Artifacts[i] + if a.Name != name || a.Kind != "bundle" || a.Language != "go" || a.System != "arch" || a.Context != nil { + t.Errorf("artifact = %+v", a) + } + if st, err := os.Stat(a.From); err != nil || !st.IsDir() { + t.Errorf("the artifact builds from %q, which is not a directory of this module", a.From) + } + if !strings.HasSuffix(a.From, a.Binary) || (name == "tools") != reflect.DeepEqual(a.Loads, []string{a.Binary}) { + t.Errorf("binary %q from %q loads %v", a.Binary, a.From, a.Loads) + } } } func TestManifestResources(t *testing.T) { // 1135/AC6 m, _ := load(t) + manifestIsWholeInItsRepository(t, m) byType := map[string][]map[string]any{} for _, r := range m.Resources { ty, _ := r["type"].(string) diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index c1c128f..45e46eb 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -2,17 +2,35 @@ "module": "mesh-delivery-watchman", "version": "1", "slug": "watchman", + "claims": [ + { + "name": "mesh-watchdogs", + "scope": "mesh", + "serves": [ + "signals", + "stuck", + "presence", + "dead-letters", + "resync" + ] + } + ], "reads": [ "mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps", + "mesh-controller.lease", "mesh-delivery-assigner.wants", "node-engine.have" ], + "tools": [ + "watchman_findings" + ], "state": [ { "name": "presence", "history": 20 }, { "name": "pulse", "key-ttl": true }, - { "name": "stuck", "history": 20 } + { "name": "stuck", "history": 20 }, + { "name": "findings", "history": 5 } ], "own-secrets": { "broker": "${dir:state}/broker" @@ -34,6 +52,14 @@ "place": ".", "unknown": "program" }, + { + "id": "socket-dir", + "type": "directory", + "path": "/var/lib/mesh-delivery-watchman-tools", + "mode": "0755", + "owner": "mesh-delivery-watchman", + "unknown": "program" + }, { "id": "watchman", "type": "process", @@ -60,6 +86,17 @@ "system": "arch", "from": "cmd/mesh-delivery-watchman", "binary": "mesh-delivery-watchman" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-watchman-tools", + "binary": "mesh-delivery-watchman-tools", + "loads": [ + "mesh-delivery-watchman-tools" + ] } ] }