process: one EventHub pulls the module's one durable consumer, handlers by subject pattern; the views reader registers on it instead of pulling (class fix of B7c finding 1; ADR 0635 §5; issue 1966)

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

go/process/event_hub.go

@@ -0,0 +1,146 @@
+package process
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// EventHub is the one pull of a module's durable consumer on the shared events stream (novox/hq ADR 0635 §5): a module
+// has one durable, `<node>_<module>`, so two loops pulling it each take messages the other needs. Every part of the
+// module that hears events registers a handler by subject pattern, and the hub pulls once and hands each message to the
+// handler whose pattern it matches. A message no handler takes is acknowledged and said, never left to be redelivered.
+// The hub never makes the stream or the consumer: they are the controller's, and a bus without them is an error of Start.
+//
+// Register every handler, then Start once: a message heard before its handler is registered would be taken by none.
+type EventHub struct {
+	logf func(string, ...any)
+
+	mu       sync.RWMutex
+	routes   []hubRoute
+	started  bool
+	cons     jetstream.ConsumeContext
+	startErr error
+}
+
+type hubRoute struct {
+	pattern string
+	handle  func(jetstream.Msg)
+}
+
+// NewEventHub is a hub saying through logf (nil says nothing).
+func NewEventHub(logf func(string, ...any)) *EventHub {
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
+	return &EventHub{logf: logf}
+}
+
+// Handle registers handle for the subjects matching pattern (`*` one token, `>` the rest). The handler acknowledges
+// the message itself. A pattern that overlaps one registered (one matches every subject of the other) is refused,
+// so a message always has one taker.
+func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
+	if pattern == "" || handle == nil {
+		return fmt.Errorf("an event handler needs a subject pattern and a handler")
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for _, r := range h.routes {
+		if subjectsOverlap(r.pattern, pattern) {
+			return fmt.Errorf("the events %q overlap %q, already handled: a message has one taker", pattern, r.pattern)
+		}
+	}
+	h.routes = append(h.routes, hubRoute{pattern, handle})
+	return nil
+}
+
+// Patterns are the registered subject patterns, in order.
+func (h *EventHub) Patterns() []string {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	out := make([]string, len(h.routes))
+	for i, r := range h.routes {
+		out[i] = r.pattern
+	}
+	return out
+}
+
+// Start binds the module's durable on the shared events stream and pulls it; later calls do nothing and return the
+// first one's result. It makes no stream and no consumer.
+func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Account) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.started {
+		return h.startErr
+	}
+	h.started = true
+	cons, err := cred.EventConsumer(ctx, js, EventsStream)
+	if err == nil {
+		h.cons, err = HearEvents(cons, h.hear, h.logf)
+	}
+	if err != nil {
+		h.startErr = fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
+	}
+	return h.startErr
+}
+
+// Stop ends the pull.
+func (h *EventHub) Stop() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.cons != nil {
+		h.cons.Stop()
+		h.cons = nil
+	}
+}
+
+func (h *EventHub) hear(m jetstream.Msg) {
+	h.mu.RLock()
+	var take func(jetstream.Msg)
+	for _, r := range h.routes {
+		if subjectMatches(r.pattern, m.Subject()) {
+			take = r.handle
+			break
+		}
+	}
+	h.mu.RUnlock()
+	if take == nil {
+		h.logf("events: %s is taken by no handler of this module, acknowledged", m.Subject())
+		_ = m.Ack()
+		return
+	}
+	take(m)
+}
+
+// subjectMatches says whether subject is one the pattern (`*` one token, `>` the rest) names.
+func subjectMatches(pattern, subject string) bool {
+	pt, st := strings.Split(pattern, "."), strings.Split(subject, ".")
+	for i, t := range pt {
+		if t == ">" {
+			return len(st) > i
+		}
+		if i >= len(st) || (t != "*" && t != st[i]) {
+			return false
+		}
+	}
+	return len(pt) == len(st)
+}
+
+// subjectsOverlap says whether some subject matches both patterns.
+func subjectsOverlap(a, b string) bool {
+	at, bt := strings.Split(a, "."), strings.Split(b, ".")
+	for i := 0; ; i++ {
+		if i >= len(at) || i >= len(bt) {
+			return len(at) == len(bt)
+		}
+		if at[i] == ">" || bt[i] == ">" {
+			return true
+		}
+		if at[i] != "*" && bt[i] != "*" && at[i] != bt[i] {
+			return false
+		}
+	}
+}

go/process/event_hub_test.go

@@ -0,0 +1,38 @@
+package process
+
+import (
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+func TestTheHubMatchesSubjectsAndRefusesOverlappingPatterns(t *testing.T) {
+	for _, c := range []struct {
+		a, b           string
+		match, overlap bool
+	}{
+		{"mesh.mod.*.event.views.*", "mesh.mod.o.event.views.x", true, true},
+		{"mesh.mod.p.event.>", "mesh.mod.p.event", false, false},
+		{"mesh.mod.p.event.a", "mesh.mod.q.event.a", false, false},
+		{"mesh.mod.p.event.>", "mesh.mod.*.event.a", false, true},
+		{"mesh.mod.p.event.a", "mesh.mod.p.event.a.b", false, false},
+	} {
+		if got := subjectMatches(c.a, c.b); got != c.match && c.match {
+			t.Errorf("%s should match %s", c.a, c.b)
+		}
+		if got := subjectsOverlap(c.a, c.b); got != c.overlap {
+			t.Errorf("overlap(%s, %s) = %v", c.a, c.b, got)
+		}
+	}
+	h := NewEventHub(nil)
+	nop := func(jetstream.Msg) {}
+	if err := h.Handle("mesh.mod.p.event.>", nop); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.Handle("mesh.mod.*.event.a", nop); err == nil {
+		t.Fatal("an overlapping pattern was accepted")
+	}
+	if err := h.Handle("mesh.mod.q.event.>", nop); err != nil {
+		t.Fatal(err)
+	}
+}

go/views/process_reader.go

@@ -46,6 +46,11 @@ type ProcessReader struct {
 	// views event, so a module that hears other events through the same consumer (a module has one) handles them there.
 	// Nil says them in the log, acknowledged: the module hears views events alone.
 	Other func(jetstream.Msg)
+	// Hub is the module's one pull of its durable consumer (process.EventHub), when other parts of the module hear events
+	// too: the reader registers its views events on it and does not pull; the host starts the hub once, after every part
+	// registered. Nil: the reader pulls the durable itself and takes every message (Other the rest), for a module that
+	// hears nothing else.
+	Hub *process.EventHub
 
 	mu    sync.Mutex
 	pics  map[string]*Picture // by bucket name <owner>_<state>
@@ -528,14 +533,21 @@ func (e entry) Created() time.Time              { return e.at }
 func (e entry) Delta() uint64                   { return 0 }
 func (e entry) Operation() jetstream.KeyValueOp { return e.op }
 
+// EventsPattern is the subjects of every owner's views events on the shared events stream.
+const EventsPattern = "mesh.mod.*.event." + EventPrefix + "*"
+
 // Start hears the owners' views events on the process's durable consumer of the shared events stream (its manifest's
 // `consumes` `<owner>.views.<state>`), keeps each status in the process's state `first-views` and hears the hand asks the
 // tools bundle's `<reader>_first_view_again` writes there (StatusTools), then asks every first view (Ask). A consumer the
 // bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand.
 func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) {
-	cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
-	if err == nil {
-		_, err = process.HearEvents(cons, r.Hear(ctx), r.logf)
+	hub, pattern, own := r.Hub, EventsPattern, false
+	if hub == nil {
+		hub, pattern, own = process.NewEventHub(r.logf), ">", true
+	}
+	err := hub.Handle(pattern, r.Hear(ctx))
+	if err == nil && own {
+		err = hub.Start(ctx, js, cred)
 	}
 	if err != nil {
 		r.logf("the consumer %s on %s is not heard (%v): the states followed are the first views' alone, asked again by hand, until the %s starts again",

go/views/process_reader_integration_test.go

@@ -78,3 +78,82 @@ func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) {
 		t.Fatalf("other got %s", got)
 	}
 }
+
+// A module has one durable consumer on the shared events stream, so the loops that hear events in one module must share
+// one pull (process.EventHub): two handlers and the views reader on the hub each get their own messages and no other's,
+// and a message nobody takes is acknowledged.
+func TestTheViewsReaderAndTwoHandlersShareTheModulesOneDurable(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cred := process.Account{Node: "n1", Module: "reader"}
+	kv := viewstest.Serve(t, ctx, js, nc, "o", "modules", "o.o_read_modules", cred.EventConsumerName())
+	hub := process.NewEventHub(t.Logf)
+	got := map[string]chan string{"a": make(chan string, 8), "b": make(chan string, 8)}
+	for name, pattern := range map[string]string{"a": "mesh.mod.p.event.a", "b": "mesh.mod.q.event.b.*"} {
+		if err := hub.Handle(pattern, func(m jetstream.Msg) { _ = m.Ack(); got[name] <- m.Subject() }); err != nil {
+			t.Fatal(err)
+		}
+	}
+	r := &views.ProcessReader{Module: "reader", Hub: hub, Logf: t.Logf,
+		Call: func(tool string, body any) (json.RawMessage, error) {
+			raw, _ := json.Marshal(body)
+			m, err := nc.Request("mesh.mod.o.call.o_read_modules.n0", raw, 5*time.Second)
+			if err != nil {
+				return nil, err
+			}
+			return m.Data, nil
+		}}
+	if _, err := r.Follow("o", "modules", "o.o_read_modules"); err != nil {
+		t.Fatal(err)
+	}
+	r.Start(ctx, js, cred, nil)
+	if err := hub.Start(ctx, js, cred); err != nil {
+		t.Fatal(err)
+	}
+	<-r.Ready()
+	pic, err := r.KeyValue(ctx, "o_modules")
+	if err != nil {
+		t.Fatal(err)
+	}
+	w, err := pic.Watch(ctx, "b")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, s := range []string{"mesh.mod.p.event.a", "mesh.mod.q.event.b.c", "mesh.mod.z.event.nobody", "mesh.mod.p.event.a"} {
+		if _, err := js.Publish(ctx, s, []byte("{}")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := kv.Put(ctx, "b", []byte(`{"n":2}`)); err != nil {
+		t.Fatal(err)
+	}
+	for e := range w.Updates() {
+		if e != nil && e.Key() == "b" {
+			break
+		}
+	}
+	for _, want := range []struct{ who, subject string }{{"a", "mesh.mod.p.event.a"}, {"b", "mesh.mod.q.event.b.c"}, {"a", "mesh.mod.p.event.a"}} {
+		if s := <-got[want.who]; s != want.subject {
+			t.Fatalf("handler %s got %s, wanted %s", want.who, s, want.subject)
+		}
+	}
+	if len(got["a"])+len(got["b"]) != 0 {
+		t.Fatal("a handler heard a message that was another's")
+	}
+	cons, err := js.Consumer(ctx, process.EventsStream, cred.EventConsumerName())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info, err := cons.Info(ctx); err != nil || info.NumAckPending != 0 {
+		t.Fatalf("a message was left unacknowledged: %+v %v", info, err)
+	}
+}