diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ed41eb7..76b4183 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -170,7 +170,7 @@ type Conn struct { cred Credential mu sync.Mutex issued map[string]*Membership // module → membership; present with nil = followed, none issued - onNew []func(Membership) + onNew []*func(Membership) subs []*nats.Subscription Logf func(format string, args ...any) // answering is every subject this runtime answers on, so a subscription the bus refused can be @@ -368,7 +368,10 @@ func (c *Conn) Follow(module string) { } c.mu.Lock() c.issued[module] = &m - handlers := append([]func(Membership){}, c.onNew...) + handlers := make([]func(Membership), 0, len(c.onNew)) + for _, h := range c.onNew { + handlers = append(handlers, *h) + } c.mu.Unlock() c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node) for _, h := range handlers { @@ -410,10 +413,28 @@ func (c *Conn) Serving() []string { // OnMembership is called with every new membership any followed module is issued. func (c *Conn) OnMembership(h func(Membership)) { c.mu.Lock() - c.onNew = append(c.onNew, h) + c.onNew = append(c.onNew, &h) c.mu.Unlock() } +// onMembership is OnMembership until the answered function is called, which drops the handler. +func (c *Conn) onMembership(h func(Membership)) func() { + mine := &h + c.mu.Lock() + c.onNew = append(c.onNew, mine) + c.mu.Unlock() + return func() { + c.mu.Lock() + defer c.mu.Unlock() + for i, had := range c.onNew { + if had == mine { + c.onNew = append(c.onNew[:i:i], c.onNew[i+1:]...) + return + } + } + } +} + func (c *Conn) track(s *nats.Subscription) { c.mu.Lock() c.subs = append(c.subs, s) @@ -958,7 +979,7 @@ func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), e h := newHearing(c, js, module, deliver) // Bound by name, never created and never matched against a subject: the consumer and its // filters are the controller's (design 29 §3), exactly as the module's own runtime bound it. - events, doing, err := h.bind(EventsStream, durable) + events, doing, err := h.bind(EventsStream, durable, nil) if err != nil { return nil, fmt.Errorf("%s %s's consumer %s: %w", doing, module, durable, err) } diff --git a/node-tools/internal/bus/streams.go b/node-tools/internal/bus/streams.go index 831225e..34a03c5 100644 --- a/node-tools/internal/bus/streams.go +++ b/node-tools/internal/bus/streams.go @@ -2,6 +2,7 @@ package bus import ( "context" + "errors" "sort" "strings" "sync" @@ -89,25 +90,35 @@ type hearing struct { stopped bool bound map[HeardIssued]jetstream.ConsumeContext said map[HeardIssued]bool - kick chan struct{} - done chan struct{} + // why is the last error the bus gave a consumer's loop, said when the loop ends on it. + // Guarded by whyMu, not mu: the bus hands errors from its own goroutines, possibly while mu is held. + whyMu sync.Mutex + why map[HeardIssued]error + kick chan struct{} + done chan struct{} + unhear func() } func newHearing(c *Conn, js jetstream.JetStream, module string, deliver func(Envelope) error) *hearing { return &hearing{c: c, js: js, module: module, deliver: deliver, - bound: map[HeardIssued]jetstream.ConsumeContext{}, said: map[HeardIssued]bool{}, + bound: map[HeardIssued]jetstream.ConsumeContext{}, said: map[HeardIssued]bool{}, why: map[HeardIssued]error{}, kick: make(chan struct{}, 1), done: make(chan struct{})} } -// bind reads one durable consumer by name, and on failure says what it was doing. -func (h *hearing) bind(stream, durable string) (jetstream.ConsumeContext, string, error) { +// bind reads one durable consumer by name, and on failure says what it was doing. Errors the bus gives +// the loop afterwards are handed to onErr, when given. +func (h *hearing) bind(stream, durable string, onErr func(error)) (jetstream.ConsumeContext, string, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() consumer, err := h.js.Consumer(ctx, stream, durable) if err != nil { return nil, "binding", err } - reading, err := consumer.Consume(h.handle, jetstream.PullMaxMessages(1)) + opts := []jetstream.PullConsumeOpt{jetstream.PullMaxMessages(1)} + if onErr != nil { + opts = append(opts, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { onErr(err) })) + } + reading, err := consumer.Consume(h.handle, opts...) if err != nil { return nil, "reading", err } @@ -159,14 +170,11 @@ func (h *hearing) handle(msg jetstream.Msg) { // what it no longer names, and a consumer not bound yet is tried again every RebindEvery. func (h *hearing) start() { h.reconcile() - h.c.OnMembership(func(m Membership) { + h.unhear = h.c.onMembership(func(m Membership) { if m.Module != h.module { return } - select { - case h.kick <- struct{}{}: - default: - } + h.nudge() }) go func() { tick := time.NewTicker(RebindEvery) @@ -183,6 +191,21 @@ func (h *hearing) start() { }() } +// forget drops the last error a consumer's loop was given. +func (h *hearing) forget(heard HeardIssued) { + h.whyMu.Lock() + delete(h.why, heard) + h.whyMu.Unlock() +} + +// nudge asks the loop to reconcile now. +func (h *hearing) nudge() { + select { + case h.kick <- struct{}{}: + default: + } +} + // stop stops every consumer on a module stream this module reads; the one on EVENTS is its caller's. func (h *hearing) stop() { h.mu.Lock() @@ -192,6 +215,9 @@ func (h *hearing) stop() { } h.stopped = true close(h.done) + if h.unhear != nil { + h.unhear() + } for heard, reading := range h.bound { reading.Stop() delete(h.bound, heard) @@ -224,6 +250,26 @@ func (h *hearing) reconcile() { for heard := range h.said { if !want[heard] { delete(h.said, heard) + h.forget(heard) + } + } + // **A loop the bus ended is no longer reading** (nats.go ends a pull loop on its own when its consumer is + // deleted): the controller deletes and makes again under the same name a restricted consumer it cannot + // narrow (ADR 0363 §6), and the membership, unchanged, says nothing — so it is bound again here. + for heard, reading := range h.bound { + select { + case <-reading.Closed(): + delete(h.bound, heard) + h.whyMu.Lock() + why := h.why[heard] + h.whyMu.Unlock() + if why == nil { + why = errors.New("no reason given") + } + h.c.Logf("[mesh-tools] %s stopped hearing %s: the bus ended reading its consumer %s (%v); bound again as it exists", + h.module, heard.Stream, heard.Consumer, why) + h.said[heard] = true + default: } } missing := make([]HeardIssued, 0, len(want)) @@ -234,7 +280,13 @@ func (h *hearing) reconcile() { } sort.Slice(missing, func(i, j int) bool { return missing[i].Stream < missing[j].Stream }) for _, heard := range missing { - reading, doing, err := h.bind(heard.Stream, heard.Consumer) + heard := heard + h.forget(heard) + reading, doing, err := h.bind(heard.Stream, heard.Consumer, func(err error) { + h.whyMu.Lock() + h.why[heard] = err + h.whyMu.Unlock() + }) if err != nil { // Said once, not at every attempt; said again when it is bound. if !h.said[heard] { @@ -247,5 +299,13 @@ func (h *hearing) reconcile() { h.bound[heard] = reading delete(h.said, heard) h.c.Logf("[mesh-tools] %s hears %s on its consumer %s", h.module, heard.Stream, heard.Consumer) + // When the bus ends the loop, look again at once rather than at the next tick. + go func() { + select { + case <-reading.Closed(): + h.nudge() + case <-h.done: + } + }() } } diff --git a/node-tools/internal/bus/streams_test.go b/node-tools/internal/bus/streams_test.go index 317eeaf..da8bde3 100644 --- a/node-tools/internal/bus/streams_test.go +++ b/node-tools/internal/bus/streams_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "sync" "testing" "time" @@ -196,3 +197,84 @@ func TestAConsumerNamedBeforeItExistsIsBoundOnceItDoes(t *testing.T) { return nil }) } + +// novox/hq ADR 0363 §6 (issue 725): a consumer the controller deletes and makes again under the same name, its +// membership unchanged, is bound again, and what is published after is heard — the module does not go deaf. +func TestAConsumerDeletedAndMadeAgainIsHeardAgain(t *testing.T) { + restore := bus.RebindEvery + bus.RebindEvery = 200 * time.Millisecond + t.Cleanup(func() { bus.RebindEvery = restore }) + reader := mt.MembershipOf("reader", "anchor", false, nil) + reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}} + mesh, conn := streamMesh(t, reader) + mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second) + mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second) + logs := &syncLogs{} + conn.Logf = logs.Logf + + var mu sync.Mutex + heard := 0 + stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error { + if env.Key == "mail.mail.home.message.added" { + mu.Lock() + heard++ + mu.Unlock() + } + return nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + count := func() int { + mu.Lock() + defer mu.Unlock() + return heard + } + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1}) + mt.Until(t, func() error { + if count() != 1 { + return errors.New("the first event on MAIL was not heard") + } + return nil + }) + + mesh.RemoveConsumer(t, "MAIL", "anchor", "reader") + mt.Until(t, func() error { + if !logs.Has("stopped hearing MAIL") { + return errors.New("the end of the loop was not said") + } + return nil + }) + mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second) + mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 2}) + mt.Until(t, func() error { + if count() != 2 { + return fmt.Errorf("heard %d events on MAIL after its consumer was made again, want 2", count()) + } + return nil + }) +} + +// syncLogs keeps what the runtime says, from any goroutine. +type syncLogs struct { + mu sync.Mutex + lines []string +} + +func (l *syncLogs) Logf(format string, args ...any) { + l.mu.Lock() + defer l.mu.Unlock() + l.lines = append(l.lines, fmt.Sprintf(format, args...)) +} + +func (l *syncLogs) Has(fragment string) bool { + l.mu.Lock() + defer l.mu.Unlock() + for _, line := range l.lines { + if strings.Contains(line, fragment) { + return true + } + } + return false +} diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 9276e8f..9edec5e 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -317,3 +317,12 @@ func (m *Mesh) Messages(t *testing.T, stream string) uint64 { } return info.State.Msgs } + +// RemoveConsumer deletes a module's consumer on a machine on a stream, as the controller does when a grant +// it cannot narrow is withdrawn (novox/hq ADR 0363 §6). +func (m *Mesh) RemoveConsumer(t *testing.T, stream, node, module string) { + t.Helper() + if err := m.js.DeleteConsumer(stream, node+"_"+module); err != nil { + t.Fatal(err) + } +}