Bind a stream consumer again when the bus ends its loop, so a module is not left deaf when the controller deletes and makes again a consumer it cannot narrow (hq issue 725)

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

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

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