diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 51270c8..20591a8 100644 --- a/node-tools/internal/bus/bus.go +++ b/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) } diff --git a/node-tools/internal/bus/streams.go b/node-tools/internal/bus/streams.go index b0ce8e7..fe94e92 100644 --- a/node-tools/internal/bus/streams.go +++ b/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.