diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go new file mode 100644 index 00000000..f6788e91 --- /dev/null +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -0,0 +1,110 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/membership" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-watchman/internal/watch" +) + +// **What the watchman consumes** (its manifest's `consumes`; novox/hq ADR 0595 §4, design 25 § 12.2; DLV-W1): the indexer's +// manifest-accepted, on its durable consumer on EVENTS, and the bus module's created and refused, on its durable consumer +// on the bus module's own stream, which its membership names (Hears). Each consumer is the one the grants part makes; one +// not there yet is opened when the watchman's membership next changes, and one opened is never opened twice. An event is +// acknowledged once handed to the watchman, so one given while it was down is heard when it is back. + +// Consumed event names, as the consumes name them (the emitter is read from the subject's emitter token). +const ( + eventAccepted = "manifest-accepted" + eventCreated = "created" + eventRefused = "refused" +) + +// consumed hands one consumed event to the watchman, by its subject: manifest-accepted from EVENTS, created and refused +// from the bus module's stream. Anything else is left. +func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, ...any)) { + switch { + case strings.HasSuffix(subject, ".event."+eventAccepted) && strings.HasPrefix(subject, "mesh.mod."): + var a delivery.ManifestAccepted + if err := json.Unmarshal(body, &a); err != nil || a.Module == "" { + logf("a manifest-accepted that does not read is left: %v", err) + return + } + w.Accepted(a) + case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventCreated): + if s, err := watch.ReadBrokerSaid(body); err != nil { + logf("%v; left", err) + } else { + w.BrokerCreated(s) + } + case strings.HasPrefix(subject, "mesh.stream.") && strings.HasSuffix(subject, ".event."+eventRefused): + if s, err := watch.ReadBrokerSaid(body); err != nil { + logf("%v; left", err) + } else { + w.BrokerRefused(s) + } + } +} + +// hearConsumed opens the watchman's durable consumers: on EVENTS now, and on each module stream its membership names as +// that membership arrives. A consumer the bus does not hold yet is said and tried again at the membership's next change. +func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, w *watch.Watchman, logf func(string, ...any)) { + var mu sync.Mutex + opened := map[string]bool{} + open := func(stream, name string) { + mu.Lock() + defer mu.Unlock() + if opened[stream] { + return + } + cons, err := js.Consumer(ctx, stream, name) + if err != nil { + logf("the consumer %s on %s cannot be bound yet (%v): what it carries is heard once the grants part has made it", name, stream, err) + return + } + if _, err := process.HearEvents(cons, func(m jetstream.Msg) { + consumed(w, m.Subject(), m.Data(), logf) + _ = m.Ack() + }, logf); err != nil { + logf("the consumer %s on %s cannot be heard (%v)", name, stream, err) + return + } + opened[stream] = true + } + open("EVENTS", cred.EventConsumerName()) + kv, err := js.KeyValue(ctx, membership.GrantsBucket) + if err != nil { + logf("its own membership cannot be read (%v): the bus module's created and refused are not heard until the watchman starts again", err) + return + } + watcher, err := kv.Watch(ctx, membership.GrantsKey(cred.Node, cred.Module)) + if err != nil { + logf("its own membership cannot be watched (%v): the bus module's created and refused are not heard until the watchman starts again", err) + return + } + go func() { + defer watcher.Stop() + for e := range watcher.Updates() { + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + m, err := membership.DecodeGrantsMembership(e.Value()) + if err != nil { + logf("its own membership does not read (%v): left", err) + continue + } + open("EVENTS", cred.EventConsumerName()) + for _, h := range m.Hears { + open(h.Stream, h.Consumer) + } + } + }() +} 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 8ec5142d..3d509711 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -379,6 +379,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hear(conn, w, dead, logger.Printf); err != nil { return err } + hearConsumed(ctx, js, cred, w, logger.Printf) errs := make(chan error, 16) if healthWatch != nil { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index a5804e47..7ce9b82c 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -7,9 +7,12 @@ import ( "io" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/process" + "mesh-delivery-watchman/internal/watch" + "github.com/nats-io/nats.go/jetstream" ) @@ -45,6 +48,24 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 } } }) + t.Run("DLV-W1: the consumed events reach the watchman by their subjects", func(t *testing.T) { + out := &findingsOut{findings: map[string]bool{}} + w := watch.New(out, time.Now) + say := func(string, ...any) {} + consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l","why":"no room"}`), say) + if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindBusStateRefused+"-l")] { + t.Fatalf("refused reached no finding: %v", out.findings) + } + consumed(w, "mesh.mod.mesh-delivery-indexer.event.manifest-accepted", []byte(`{"module":"m","digest":"d","bus-state":[]}`), say) + if len(out.findings) != 0 { + t.Fatalf("an accepted manifest no longer asking for l left %v", out.findings) + } + consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l"}`), say) + consumed(w, "mesh.stream.nats.event.created", []byte(`{"module":"m","local":"l"}`), say) + if len(out.findings) != 0 { + t.Fatalf("created left %v", out.findings) + } + }) err := run(context.Background(), func(string) string { return "" }, io.Discard) if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err) @@ -55,3 +76,13 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 type refusingBus struct{ err error } func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err } + +// findingsOut keeps which findings stand. +type findingsOut struct{ findings map[string]bool } + +func (o *findingsOut) PutPresence(string, bool) error { return nil } +func (o *findingsOut) PutPulse(string, []byte, time.Duration) error { return nil } +func (o *findingsOut) DeletePulse(string) error { return nil } +func (o *findingsOut) PutStuck(string, []byte) error { return nil } +func (o *findingsOut) PutFinding(key string, _ []byte) error { o.findings[key] = true; return nil } +func (o *findingsOut) DeleteFinding(key string) error { delete(o.findings, key); return nil } diff --git a/modules/mesh-delivery-watchman/internal/watch/busstate.go b/modules/mesh-delivery-watchman/internal/watch/busstate.go new file mode 100644 index 00000000..ff1cc5bf --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/busstate.go @@ -0,0 +1,159 @@ +package watch + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A refused or overdue creation of bus state is loud** (novox/hq ADR 0595 §4; design 25 § 12.2; DLV-W1). The watchman +// hears, through its `consumes`, the indexer's `manifest-accepted` (which carries the module's bus-state entries whole) +// and the bus module's own `created` and `refused`. An entry an accepted manifest asks for that has no `created` or +// `refused` within OverdueBound is the finding bus-state-overdue, naming the module and the entry; a `refused` is the +// finding bus-state-refused, with why. A later `created` clears both, and so does a later accepted manifest of that +// module that no longer asks for the entry. Never the bus module's to raise: it cannot while it is down. +// +// The bound is a row of the signals table: a pulse key per entry with OverdueBound to live, whose expiry is the finding, +// so no loop looks and a pulse that stood across a restart still acts. Both findings outlive a restart (signalKind), +// since what ends them is heard, not read. + +// OverdueBound is how long an accepted entry may wait for the bus module's created or refused. +const OverdueBound = 5 * time.Minute + +// The kinds, and the signals table's row. +const ( + KindBusStateOverdue = "bus-state-overdue" + KindBusStateRefused = "bus-state-refused" + rowBusState = "bus-state" +) + +// BrokerSaid is the part of the bus module's created or refused the watchman reads: the module, the entry, and why a +// create was refused. It stands until the seat mesh-broker's definition is generated in mesh-contracts. +type BrokerSaid struct { + Module string `json:"module"` + Local string `json:"local"` + Why string `json:"why,omitempty"` +} + +// busStateID is the signal's and the findings' id of one entry: `.`. +func busStateID(module, local string) string { return module + "." + local } + +func overdueKey(module, local string) string { + return FindingKey(ScopeModule, module, KindBusStateOverdue+"-"+local) +} + +func refusedKey(module, local string) string { + return FindingKey(ScopeModule, module, KindBusStateRefused+"-"+local) +} + +// Accepted is the indexer's manifest-accepted heard: each entry it asks for waits OverdueBound for the bus module's word, +// and every entry of the module this watchman waits on or holds a finding of that it no longer asks for is ended. +func (w *Watchman) Accepted(a delivery.ManifestAccepted) { + w.mu.Lock() + defer w.mu.Unlock() + asked := map[string]bool{} + for _, e := range a.BusState { + if e.Name == "" { + continue + } + asked[e.Name] = true + // One already said created or refused since an earlier verdict waits again: a new verdict asks the bus module again. + w.signal(rowBusState, busStateID(a.Module, e.Name), OverdueBound, signalDue{Do: dueFind, Finding: Finding{ + Scope: ScopeModule, ID: a.Module, Token: KindBusStateOverdue + "-" + e.Name, Kind: KindBusStateOverdue, + Severity: Warning, Source: rowBusState, Since: w.now().Add(OverdueBound).UTC().Truncate(time.Second), + Summary: fmt.Sprintf("the bus module said nothing of %s's %s (%s) within %s of its accepted manifest %s", a.Module, + e.Name, e.Storage, OverdueBound, short(a.Digest)), + Headline: "The bus state " + e.Name + " of " + a.Module + " was not created", + Explanation: "The indexer accepted a manifest of " + a.Module + " asking the bus module for " + e.Name + ", and the bus " + + "module said neither created nor refused within " + OverdueBound.String() + ": it is down, or behind. " + a.Module + + " fails at its start while the entry's binding is missing (novox/hq ADR 0595 §1).", + Needs: "a look at the bus module's health and its process's log; nats_first_view_again asks it to read the verdicts again", + Resolved: "the bus module said " + e.Name + " of " + a.Module + " created, or a later manifest no longer asks for it"}}) + } + for _, local := range w.busStateEntriesOf(a.Module) { + if asked[local] { + continue + } + w.unsignal(rowBusState, busStateID(a.Module, local)) + w.unfind(overdueKey(a.Module, local)) + w.unfind(refusedKey(a.Module, local)) + } +} + +// BrokerCreated is the bus module's created heard: the entry's wait ends, and its findings with it. +func (w *Watchman) BrokerCreated(s BrokerSaid) { + w.mu.Lock() + defer w.mu.Unlock() + w.unsignal(rowBusState, busStateID(s.Module, s.Local)) + w.unfind(overdueKey(s.Module, s.Local)) + w.unfind(refusedKey(s.Module, s.Local)) +} + +// BrokerRefused is the bus module's refused heard: the entry's wait ends in the finding bus-state-refused, until a later +// created. +func (w *Watchman) BrokerRefused(s BrokerSaid) { + w.mu.Lock() + defer w.mu.Unlock() + w.unsignal(rowBusState, busStateID(s.Module, s.Local)) + w.unfind(overdueKey(s.Module, s.Local)) + why := strings.TrimSpace(s.Why) + if why == "" { + why = "it gave no reason" + } + w.find(Finding{Scope: ScopeModule, ID: s.Module, Token: KindBusStateRefused + "-" + s.Local, Kind: KindBusStateRefused, + Severity: Urgent, Source: rowBusState, Since: w.now().UTC().Truncate(time.Second), Said: why, + Summary: fmt.Sprintf("the bus module refused to create %s's %s: %s", s.Module, s.Local, why), + Headline: "The bus state " + s.Local + " of " + s.Module + " was refused", + Explanation: "The bus module could not create " + s.Local + " for " + s.Module + ", so no binding is handed and " + s.Module + + " fails at its start, naming the entry (novox/hq ADR 0595 §1): " + why, + Needs: "a pull request that mends the entry in the manifest of " + s.Module + ", or the bus module's cause fixed", + Resolved: "the bus module said " + s.Local + " of " + s.Module + " created"}) +} + +// busStateEntriesOf are the entries of module this watchman waits on or holds a finding of, by local name, sorted. +func (w *Watchman) busStateEntriesOf(module string) []string { + seen := map[string]bool{} + prefix := signalKey(rowBusState, module) + "." + for pk := range w.signals { + if local, ok := strings.CutPrefix(pk, prefix); ok && !strings.Contains(local, ".") { + seen[local] = true + } + } + for _, kind := range []string{KindBusStateOverdue, KindBusStateRefused} { + p := FindingKey(ScopeModule, module, kind+"-") + for key := range w.findings { + if local, ok := strings.CutPrefix(key, p); ok { + seen[local] = true + } + } + } + out := make([]string, 0, len(seen)) + for l := range seen { + out = append(out, l) + } + sort.Strings(out) + return out +} + +// ReadBrokerSaid reads the bus module's created or refused. +func ReadBrokerSaid(body []byte) (BrokerSaid, error) { + var s BrokerSaid + if err := json.Unmarshal(body, &s); err != nil { + return BrokerSaid{}, fmt.Errorf("the bus module's word cannot be read: %w", err) + } + if s.Module == "" || s.Local == "" { + return BrokerSaid{}, fmt.Errorf("the bus module's word names no module or no entry") + } + return s, nil +} + +func short(digest string) string { + if len(digest) > 12 { + return digest[:12] + } + return digest +} diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 40e308b2..d9c84281 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -6,10 +6,20 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" ) +func mustEncode(t *testing.T, f Finding) []byte { + t.Helper() + v, err := EncodeFinding(f) + if err != nil { + t.Fatal(err) + } + return v +} + 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}) @@ -72,6 +82,52 @@ func TestTheWatchmansFindings(t *testing.T) { } w.ModuleVerdict("nats", value(""), Put) }, map[string]string{}}, + {"DLV-W1, 1966/AC37-AC38 (ADR 0595 §4, design 25 § 12.2): an accepted entry the bus module says nothing of in five minutes is overdue until created; a refused one stands until created; a later verdict no longer asking ends both", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + accepted := func(locals ...string) delivery.ManifestAccepted { + a := delivery.ManifestAccepted{Module: "mesh-knowledge", Digest: strings.Repeat("d", 64)} + for _, l := range locals { + a.BusState = append(a.BusState, catalogue.BusStateEntry{Name: l, Storage: catalogue.StorageLedger}) + } + return a + } + w.Accepted(accepted("records", "drafts", "kept")) + pk := func(local string) string { return signalKey(rowBusState, busStateID("mesh-knowledge", local)) } + if o.ttls[pk("records")] != OverdueBound { + t.Fatalf("no five-minute pulse for records: %v", o.ttls) + } + w.BrokerCreated(BrokerSaid{Module: "mesh-knowledge", Local: "drafts"}) + expire(w, c, pk("drafts"), 5*time.Minute) // created in time: no finding + expire(w, c, pk("records"), 0) + overdue := o.findings[overdueKey("mesh-knowledge", "records")] + if overdue.Kind != KindBusStateOverdue || !strings.Contains(overdue.Summary, "mesh-knowledge's records") { + t.Fatalf("findings = %v", o.findings) + } + w.BrokerCreated(BrokerSaid{Module: "mesh-knowledge", Local: "records"}) + if _, open := o.findings[overdueKey("mesh-knowledge", "records")]; open { + t.Fatal("created did not clear overdue") + } + w.BrokerRefused(BrokerSaid{Module: "mesh-knowledge", Local: "kept", Why: "the cap is above the mesh's bound"}) + if f := o.findings[refusedKey("mesh-knowledge", "kept")]; f.Severity != Urgent || !strings.Contains(f.Summary, "cap is above") { + t.Fatalf("findings = %v", o.findings) + } + // A restart keeps both kinds: what ends them is heard. + if !signalKind(mustEncode(t, o.findings[refusedKey("mesh-knowledge", "kept")])) { + t.Error("a refused creation would be dropped at a restart") + } + // A later verdict asking for kept alone again waits on it again; one no longer asking for it ends it. + w.Accepted(accepted("kept", "fresh")) + w.Accepted(accepted("fresh")) + if _, open := o.findings[refusedKey("mesh-knowledge", "kept")]; open { + t.Fatal("a verdict no longer asking for kept left its finding") + } + if _, waits := o.pulses[pk("kept")]; waits { + t.Fatal("a verdict no longer asking for kept left its wait") + } + w.BrokerCreated(BrokerSaid{Module: "mesh-knowledge", Local: "fresh"}) + // A refused create stands until created. + w.Accepted(accepted("gone")) + w.BrokerRefused(BrokerSaid{Module: "mesh-knowledge", Local: "gone"}) + }, map[string]string{refusedKey("mesh-knowledge", "gone"): Urgent}}, {"ADR 0601 §4: what a node keeps of a module as it runs, for a manifest that does not say it yet, is a warning until the have says none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { have := func(kept ...string) []byte { b, err := delivery.EncodeHave(delivery.Have{FP: "src2:" + strings.Repeat("a", 64), Phase: delivery.PhaseRunning, Health: delivery.HealthOK, Kept: kept}) diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 0e8b7925..7b32fe28 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -366,7 +366,7 @@ func signalKind(value []byte) bool { return false } switch f.Source { - case "provisioner.failing", "S8", "S9", "S10", "S11": + case "provisioner.failing", "S8", "S9", "S10", "S11", rowBusState: return true } return false diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index a577361e..8529b4c1 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -47,6 +47,7 @@ type manifest struct { Module string `json:"module"` Claims []claim `json:"claims"` Tools []string `json:"tools"` + Consumes []string `json:"consumes"` Moved bool `json:"moved"` Version string `json:"version"` Slug string `json:"slug"` @@ -101,7 +102,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) { t.Errorf("tools = %v", m.Tools) } - for _, k := range []string{"emits", "context", "items", "consumes", "provides"} { + // DLV-W1 (ADR 0595 §4): it hears the accepted verdicts and the bus module's created and refused, durably. + if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.manifest-accepted", "nats.created", "nats.refused"}) { + t.Errorf("consumes = %v", m.Consumes) + } + for _, k := range []string{"emits", "context", "items", "provides"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the watchman has none", k) } diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 4c85f5d7..d8d87a1a 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -29,6 +29,11 @@ "tools": [ "watchman_findings" ], + "consumes": [ + "mesh-delivery-indexer.manifest-accepted", + "nats.created", + "nats.refused" + ], "state": [ { "name": "presence", "history": 20 }, { "name": "pulse", "key-ttl": true },