Merge pull request 'node-tools: a consumer deleted or made again while no pull waits is noticed at the next rebind, so a module is not deaf for a heartbeat (issue 1945)' (#43) from fix/bus-consumer-remade-flake into main

node-tools/internal/bus/bus.go

@@ -999,7 +999,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, nil)
+	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

@@ -93,7 +93,10 @@ type hearing struct {
 	mu      sync.Mutex
 	stopped bool
 	bound   map[HeardIssued]jetstream.ConsumeContext
-	said    map[HeardIssued]bool
+	// made is when the bus made each bound consumer: a consumer deleted and made again under its name has
+	// another time, and a loop bound to the old one is not reading the new one.
+	made map[HeardIssued]time.Time
+	said map[HeardIssued]bool
 	// 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
@@ -105,18 +108,22 @@ type hearing struct {
 
 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{}, why: map[HeardIssued]error{},
+		bound: map[HeardIssued]jetstream.ConsumeContext{}, made: map[HeardIssued]time.Time{}, 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. 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) {
+func (h *hearing) bind(stream, durable string, onErr func(error)) (jetstream.ConsumeContext, time.Time, 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
+		return nil, time.Time{}, "binding", err
+	}
+	var created time.Time
+	if info := consumer.CachedInfo(); info != nil {
+		created = info.Created
 	}
 	opts := []jetstream.PullConsumeOpt{jetstream.PullMaxMessages(1)}
 	if onErr != nil {
@@ -124,9 +131,9 @@ func (h *hearing) bind(stream, durable string, onErr func(error)) (jetstream.Con
 	}
 	reading, err := consumer.Consume(h.handle, opts...)
 	if err != nil {
-		return nil, "reading", err
+		return nil, time.Time{}, "reading", err
 	}
-	return reading, "", nil
+	return reading, created, "", nil
 }
 
 // handle hands one delivered event to the module: acknowledged when it took it, offered again after NakDelay
@@ -236,9 +243,26 @@ func (h *hearing) stop() {
 	for heard, reading := range h.bound {
 		reading.Stop()
 		delete(h.bound, heard)
+		delete(h.made, heard)
 	}
 }
 
+// replaced says whether the consumer a loop is bound to is gone from the bus or was made again since. A bus
+// that cannot be asked right now says no: the loop is left as it is.
+func (h *hearing) replaced(heard HeardIssued) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	consumer, err := h.js.Consumer(ctx, heard.Stream, heard.Consumer)
+	if errors.Is(err, jetstream.ErrConsumerNotFound) {
+		return true
+	}
+	if err != nil || consumer.CachedInfo() == nil {
+		return false
+	}
+	created := h.made[heard]
+	return !created.IsZero() && !consumer.CachedInfo().Created.Equal(created)
+}
+
 // reconcile makes what is bound what the module's membership names now.
 func (h *hearing) reconcile() {
 	want := map[HeardIssued]bool{}
@@ -258,6 +282,7 @@ func (h *hearing) reconcile() {
 		if !want[heard] {
 			reading.Stop()
 			delete(h.bound, heard)
+			delete(h.made, heard)
 			h.c.Logf("[mesh-tools] %s no longer hears %s: its membership stopped naming its consumer %s",
 				h.module, heard.Stream, heard.Consumer)
 		}
@@ -271,10 +296,25 @@ func (h *hearing) reconcile() {
 	// **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.
+	// **A loop the bus did not end is asked about**: nats.go learns of a deleted consumer only from the answer
+	// to a pull it has pending, and a consumer deleted and made again between two pulls, or while none is
+	// pending, leaves the loop waiting out its heartbeat (30 seconds) on a consumer that is gone. So each
+	// bound consumer is looked up by name: gone, or made at another time, ends its loop here.
+	for heard, reading := range h.bound {
+		if h.replaced(heard) {
+			h.whyMu.Lock()
+			if h.why[heard] == nil {
+				h.why[heard] = errors.New("the consumer was deleted or made again")
+			}
+			h.whyMu.Unlock()
+			reading.Stop()
+		}
+	}
 	for heard, reading := range h.bound {
 		select {
 		case <-reading.Closed():
 			delete(h.bound, heard)
+			delete(h.made, heard)
 			h.whyMu.Lock()
 			why := h.why[heard]
 			h.whyMu.Unlock()
@@ -297,7 +337,7 @@ func (h *hearing) reconcile() {
 	for _, heard := range missing {
 		heard := heard
 		h.forget(heard)
-		reading, doing, err := h.bind(heard.Stream, heard.Consumer, func(err error) {
+		reading, created, doing, err := h.bind(heard.Stream, heard.Consumer, func(err error) {
 			h.whyMu.Lock()
 			h.why[heard] = err
 			h.whyMu.Unlock()
@@ -312,6 +352,7 @@ func (h *hearing) reconcile() {
 			continue
 		}
 		h.bound[heard] = reading
+		h.made[heard] = created
 		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.