Repository: nats-io/nats-server
Pull request #5048 (2024-02-07 16:54:53)

FIXED: MQTT retained message consumer creation
- Cleaned up the retained message consumer name so that it does not cause problems
- Per @derekcollison's recommendation, use an ephemeral consumer for retained messages


Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 5c501cd4..03c55ea5 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -277,6 +277,9 @@ const (
 	// JSAdvisoryStreamRestoreCompletePre notification that a restore was completed.
 	JSAdvisoryStreamRestoreCompletePre = "$JS.EVENT.ADVISORY.STREAM.RESTORE_COMPLETE"
 
+	// JSAdvisoryDomainLeaderElectedPre notification that a jetstream domain has elected a leader.
+	JSAdvisoryDomainLeaderElected = "$JS.EVENT.ADVISORY.DOMAIN.LEADER_ELECTED"
+
 	// JSAdvisoryStreamLeaderElectedPre notification that a replicated stream has elected a leader.
 	JSAdvisoryStreamLeaderElectedPre = "$JS.EVENT.ADVISORY.STREAM.LEADER_ELECTED"
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index e7756cd1..ee3576fe 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5333,6 +5333,31 @@ func (js *jetStream) stopUpdatesSub() {
 	}
 }
 
+func (s *Server) sendDomainLeaderElectAdvisory() {
+	js, cc := s.getJetStreamCluster()
+	if js == nil || cc == nil {
+		return
+	}
+
+	js.mu.RLock()
+	node := cc.meta
+	js.mu.RUnlock()
+
+	adv := &JSDomainLeaderElectedAdvisory{
+		TypedEvent: TypedEvent{
+			Type: JSDomainLeaderElectedAdvisoryType,
+			ID:   nuid.Next(),
+			Time: time.Now().UTC(),
+		},
+		Leader:   node.GroupLeader(),
+		Replicas: s.replicas(node),
+		Cluster:  s.cachedClusterName(),
+		Domain:   s.getOpts().JetStreamDomain,
+	}
+
+	s.publishAdvisory(nil, JSAdvisoryDomainLeaderElected, adv)
+}
+
 func (js *jetStream) processLeaderChange(isLeader bool) {
 	if js == nil {
 		return
@@ -5346,6 +5371,7 @@ func (js *jetStream) processLeaderChange(isLeader bool) {
 
 	if isLeader {
 		s.Noticef("Self is new JetStream cluster metadata leader")
+		s.sendDomainLeaderElectAdvisory()
 	} else {
 		var node string
 		if meta := js.getMetaGroup(); meta != nil {
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 7ea14ef8..1852811b 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -193,10 +193,22 @@ const JSRestoreCompleteAdvisoryType = "io.nats.jetstream.advisory.v1.restore_com
 
 // Clustering specific.
 
-// JSStreamLeaderElectedAdvisoryType is sent when the system elects a leader for a stream.
+// JSClusterLeaderElectedAdvisoryType is sent when the system elects a new meta leader.
+const JSDomainLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.domain_leader_elected"
+
+// JSClusterLeaderElectedAdvisory indicates that a domain has elected a new leader.
+type JSDomainLeaderElectedAdvisory struct {
+	TypedEvent
+	Leader   string      `json:"leader"`
+	Replicas []*PeerInfo `json:"replicas"`
+	Cluster  string      `json:"cluster"`
+	Domain   string      `json:"domain,omitempty"`
+}
+
+// JSStreamLeaderElectedAdvisoryType is sent when the system elects a new leader for a stream.
 const JSStreamLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.stream_leader_elected"
 
-// JSStreamLeaderElectedAdvisory indicates that a stream has lost quorum and is stalled.
+// JSStreamLeaderElectedAdvisory indicates that a stream has elected a new leader.
 type JSStreamLeaderElectedAdvisory struct {
 	TypedEvent
 	Account  string      `json:"account,omitempty"`
@@ -222,7 +234,7 @@ type JSStreamQuorumLostAdvisory struct {
 // JSConsumerLeaderElectedAdvisoryType is sent when the system elects a leader for a consumer.
 const JSConsumerLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_leader_elected"
 
-// JSConsumerLeaderElectedAdvisory indicates that a stream has lost quorum and is stalled.
+// JSConsumerLeaderElectedAdvisory indicates that a consumer has elected a new leader.
 type JSConsumerLeaderElectedAdvisory struct {
 	TypedEvent
 	Account  string      `json:"account,omitempty"`
diff --git a/server/mqtt.go b/server/mqtt.go
index f64c160e..a1abcb3e 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -1451,21 +1451,12 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	rmLegacyDurName := mqttRetainedMsgsStreamName + "_" + jsa.id
 	jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmLegacyDurName)
 
-	// Using ephemeral consumer is too risky because if this server were to be
-	// disconnected from the rest for few seconds, then the leader would remove
-	// the consumer, so even after a reconnect, we would no longer receive
-	// retained messages.
-	//
-	// So we use a durable consumer, and create a new one each time we start.
-	// The old one should expire and get deleted due to inactivity. The name for
-	// the durable is $MQTT_rmsgs_{uuid}_{server-name}, the server name is just
-	// for readability.
-	rmDurName := mqttRetainedMsgsStreamName + "_" + nuid.Next() + "_" + s.String()
-
+	// Create a new, uniquely names consumer for retained messages for this
+	// server. The prior one will expire eventually.
 	ccfg := &CreateConsumerRequest{
 		Stream: mqttRetainedMsgsStreamName,
 		Config: ConsumerConfig{
-			Durable:           rmDurName,
+			Name:              mqttRetainedMsgsStreamName + "_" + nuid.Next(),
 			FilterSubject:     mqttRetainedMsgsStreamSubject + ">",
 			DeliverSubject:    rmsubj,
 			ReplayPolicy:      ReplayInstant,
@@ -1473,7 +1464,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 			InactiveThreshold: 5 * time.Minute,
 		},
 	}
-	if _, err := jsa.createConsumer(ccfg); err != nil {
+	if _, err := jsa.createEphemeralConsumer(ccfg); err != nil {
 		return nil, fmt.Errorf("create retained messages consumer for account %q: %v", accName, err)
 	}
 
@@ -1653,17 +1644,26 @@ func (jsa *mqttJSA) sendAck(ackSubject string) {
 	jsa.sendq.push(&mqttJSPubMsg{subj: ackSubject, hdr: -1})
 }
 
-func (jsa *mqttJSA) createConsumer(cfg *CreateConsumerRequest) (*JSApiConsumerCreateResponse, error) {
+func (jsa *mqttJSA) createEphemeralConsumer(cfg *CreateConsumerRequest) (*JSApiConsumerCreateResponse, error) {
 	cfgb, err := json.Marshal(cfg)
 	if err != nil {
 		return nil, err
 	}
-	var subj string
-	if cfg.Config.Durable != _EMPTY_ {
-		subj = fmt.Sprintf(JSApiDurableCreateT, cfg.Stream, cfg.Config.Durable)
-	} else {
-		subj = fmt.Sprintf(JSApiConsumerCreateT, cfg.Stream)
+	subj := fmt.Sprintf(JSApiConsumerCreateT, cfg.Stream)
+	ccri, err := jsa.newRequest(mqttJSAConsumerCreate, subj, 0, cfgb)
+	if err != nil {
+		return nil, err
+	}
+	ccr := ccri.(*JSApiConsumerCreateResponse)
+	return ccr, ccr.ToError()
+}
+
+func (jsa *mqttJSA) createDurableConsumer(cfg *CreateConsumerRequest) (*JSApiConsumerCreateResponse, error) {
+	cfgb, err := json.Marshal(cfg)
+	if err != nil {
+		return nil, err
 	}
+	subj := fmt.Sprintf(JSApiDurableCreateT, cfg.Stream, cfg.Config.Durable)
 	ccri, err := jsa.newRequest(mqttJSAConsumerCreate, subj, 0, cfgb)
 	if err != nil {
 		return nil, err
@@ -4941,7 +4941,7 @@ func (sess *mqttSession) ensurePubRelConsumerSubscription(c *client) error {
 		if opts.MQTT.ConsumerInactiveThreshold > 0 {
 			ccr.Config.InactiveThreshold = opts.MQTT.ConsumerInactiveThreshold
 		}
-		if _, err := sess.jsa.createConsumer(ccr); err != nil {
+		if _, err := sess.jsa.createDurableConsumer(ccr); err != nil {
 			c.Errorf("Unable to add JetStream consumer for PUBREL for client %q: err=%v", id, err)
 			return err
 		}
@@ -5047,7 +5047,7 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 		if opts.MQTT.ConsumerInactiveThreshold > 0 {
 			ccr.Config.InactiveThreshold = opts.MQTT.ConsumerInactiveThreshold
 		}
-		if _, err := sess.jsa.createConsumer(ccr); err != nil {
+		if _, err := sess.jsa.createDurableConsumer(ccr); err != nil {
 			c.Errorf("Unable to add JetStream consumer for subscription on %q: err=%v", subject, err)
 			return nil, nil, err
 		}


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index a269523c..64335c73 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -5995,3 +5995,39 @@ func TestJetStreamClusterStreamResetPreacks(t *testing.T) {
 		return nil
 	})
 }
+
+func TestJetStreamClusterDomainAdvisory(t *testing.T) {
+	tmpl := strings.Replace(jsClusterAccountsTempl, "store_dir:", "domain: NGS, store_dir:", 1)
+	c := createJetStreamCluster(t, tmpl, "R3S", _EMPTY_, 3, 18033, true)
+	defer c.shutdown()
+
+	// Connect to system account.
+	nc, _ := jsClientConnect(t, c.randomServer(), nats.UserInfo("admin", "s3cr3t!"))
+	defer nc.Close()
+
+	sub, err := nc.SubscribeSync(JSAdvisoryDomainLeaderElected)
+	require_NoError(t, err)
+
+	// Ask meta leader to move and make sure we get an advisory.
+	nc.Request(JSApiLeaderStepDown, nil, time.Second)
+	c.waitOnLeader()
+
+	checkSubsPending(t, sub, 1)
+
+	m, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+
+	var adv JSDomainLeaderElectedAdvisory
+	require_NoError(t, json.Unmarshal(m.Data, &adv))
+
+	ml := c.leader()
+	js, cc := ml.getJetStreamCluster()
+	js.mu.RLock()
+	peer := cc.meta.ID()
+	js.mu.RUnlock()
+
+	require_Equal(t, adv.Leader, peer)
+	require_Equal(t, adv.Domain, "NGS")
+	require_Equal(t, adv.Cluster, "R3S")
+	require_Equal(t, len(adv.Replicas), 3)
+}