DLV-W1: the watchman says an accepted entry of bus state the bus module said nothing of in five minutes, and a refused one, until created or no longer asked (ADR 0595 §4, design 25 § 12.2)

It consumes the indexer's manifest-accepted and the bus module's created and refused on its
durable consumers. The five-minute bound is a row of its signals table (a pulse key whose
expiry is the finding, read back at a restart), not a new state: the prepare's state
overdue would have kept the same keys with no expiry to act on.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

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

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 {

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 }

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: `<module>.<local>`.
+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
+}

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

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

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

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 },