From e648fe53741f34db456556ec4fc72ec8c9397aab Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 15:07:39 +0100 Subject: [PATCH] =?UTF-8?q?process:=20one=20EventHub=20pulls=20the=20modul?= =?UTF-8?q?e's=20one=20durable=20consumer,=20handlers=20by=20subject=20pat?= =?UTF-8?q?tern;=20the=20views=20reader=20registers=20on=20it=20instead=20?= =?UTF-8?q?of=20pulling=20(class=20fix=20of=20B7c=20finding=201;=20ADR=200?= =?UTF-8?q?635=20=C2=A75;=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/process/event_hub.go | 146 ++++++++++++++++++++ go/process/event_hub_test.go | 38 +++++ go/views/process_reader.go | 18 ++- go/views/process_reader_integration_test.go | 79 +++++++++++ 4 files changed, 278 insertions(+), 3 deletions(-) create mode 100644 go/process/event_hub.go create mode 100644 go/process/event_hub_test.go diff --git a/go/process/event_hub.go b/go/process/event_hub.go new file mode 100644 index 0000000..7d0a6a8 --- /dev/null +++ b/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, `_`, 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 + } + } +} diff --git a/go/process/event_hub_test.go b/go/process/event_hub_test.go new file mode 100644 index 0000000..790e3f5 --- /dev/null +++ b/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) + } +} diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 1ab7611..9cc1d74 100644 --- a/go/views/process_reader.go +++ b/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 _ @@ -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` `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the // tools bundle's `_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", diff --git a/go/views/process_reader_integration_test.go b/go/views/process_reader_integration_test.go index ab740da..e8310e0 100644 --- a/go/views/process_reader_integration_test.go +++ b/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) + } +}