A consumer a machine is bound to keeps the subject that works
novox/hq 04-ISSUES/156. Issue 146 put the stream into a push consumer's delivery subject. The server will not move that subject while a subscriber is bound, and answers `consumer name already in use` — a message about the name, for a conflict about the subject. A node is bound to its declaration consumer the whole time it is up: that IS a node listening. So every node consumer in a running mesh became one the assertion could not bring to match, and the control plane crash-looped on the assertion it makes before it serves. A fresh mesh showed nothing, because nothing was bound. Kept rather than deleted and re-made. Re-making moves the subject, and a holder may not be allowed to subscribe to the new one yet: the wider grant travels in the bus's user list, which this same control plane composes and a machine applies minutes later. On the live mesh the nodes are granted `_DELIVER.<node>` and not `_DELIVER.<node>.>`, so re-making would have silenced every machine — worse than the collision it fixes, and harder to undo. Kept rather than fatal, which is what 146's change intended and did not do. The bare subject still delivers, and collides only where one holder has two consumers of one name. That is the controller's own pair, and the controller is not bound to them while it asserts, so those do move. Also: an existing consumer's deliver policy is carried across rather than reasserted, because the server refuses to change it and where a consumer starts is its history. Two tests against a real server: a consumer with a subscriber bound keeps its subject, is reported, and still delivers; one with nothing bound moves, so 146's fix still applies where it matters.
This commit is contained in:
@@ -26,6 +26,16 @@ import (
|
||||
type JetStream struct {
|
||||
conn *nats.Conn
|
||||
js nats.JetStreamContext
|
||||
// Note is how this says something it decided not to fail over. Nil is silent, which is only
|
||||
// right for a caller that has no way to report; the controller sets it.
|
||||
Note func(string, ...any)
|
||||
}
|
||||
|
||||
// note reports without requiring a caller to have set one.
|
||||
func (j *JetStream) note(format string, args ...any) {
|
||||
if j.Note != nil {
|
||||
j.Note(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// Dial connects and returns the controller's JetStream handle.
|
||||
@@ -200,9 +210,44 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
|
||||
want.DeliverSubject = DeliverSubjectFor(c)
|
||||
}
|
||||
|
||||
switch _, err := j.js.ConsumerInfo(c.Stream, c.Name); {
|
||||
switch have, err := j.js.ConsumerInfo(c.Stream, c.Name); {
|
||||
case err == nil:
|
||||
// Where an existing consumer starts is its history, not something an assertion may move:
|
||||
// the server refuses a changed deliver policy outright. Carried across, so asserting twice
|
||||
// is the no-op a restart depends on.
|
||||
want.DeliverPolicy = have.Config.DeliverPolicy
|
||||
want.OptStartSeq = have.Config.OptStartSeq
|
||||
want.OptStartTime = have.Config.OptStartTime
|
||||
|
||||
if _, err := j.js.UpdateConsumer(c.Stream, want); err != nil {
|
||||
// **A consumer that works is not replaced to make its name tidier**
|
||||
// (novox/hq 04-ISSUES/156).
|
||||
//
|
||||
// The server will not move a push consumer's delivery subject while a subscriber is
|
||||
// bound to it, and answers `consumer name already in use` — a message about the name,
|
||||
// for a conflict about the subject. A node is bound to its declaration consumer the
|
||||
// whole time it is up; that IS a node listening. So when 04-ISSUES/146 put the stream
|
||||
// into the subject, every node consumer in a running mesh became one this could not
|
||||
// bring to match, and the control plane crash-looped on the assertion it makes before
|
||||
// it serves. A fresh mesh showed nothing: nothing was bound.
|
||||
//
|
||||
// Kept rather than deleted and re-made. Re-making moves the subject, and a holder may
|
||||
// not be allowed to subscribe to the new one yet — the wider grant travels in the bus's
|
||||
// user list, which this same control plane composes and a machine applies minutes
|
||||
// later. Re-making here would have silenced every machine in the mesh, which is worse
|
||||
// than the collision it was fixing and harder to undo.
|
||||
//
|
||||
// Kept rather than fatal, which is what 146's change intended and did not do: the bare
|
||||
// subject it replaces still delivers, and it collides only where one holder has two
|
||||
// consumers of one name. That is the controller's own pair, and the controller is not
|
||||
// bound to them while it asserts, so those do move. A node has one consumer and nothing
|
||||
// to collide with.
|
||||
if have.Config.DeliverSubject != want.DeliverSubject {
|
||||
j.note("consumer %s on %s still delivers to %q and not %q: %v. It keeps working; "+
|
||||
"the subject moves on an assertion made while nothing is bound to it",
|
||||
c.Name, c.Stream, have.Config.DeliverSubject, want.DeliverSubject, err)
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("bringing consumer %s on %s to match: %w", c.Name, c.Stream, err)
|
||||
}
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user