Repository: nats-io/nats-server
Pull request #2432 (2021-08-11 22:36:05)

JetStream additions
This adds two items.

1. Domain information for PubAcks.
2. Active state for push consumers that details active subject and optionally queue group.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index e9ad5543..bfe24394 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -43,6 +43,7 @@ type ConsumerInfo struct {
 	NumRedelivered int             `json:"num_redelivered"`
 	NumWaiting     int             `json:"num_waiting"`
 	NumPending     uint64          `json:"num_pending"`
+	Active         *PushActive     `json:"active,omitempty"`
 	Cluster        *ClusterInfo    `json:"cluster,omitempty"`
 }
 
@@ -69,6 +70,11 @@ type ConsumerConfig struct {
 	Direct bool `json:"direct,omitempty"`
 }
 
+type PushActive struct {
+	Subject string `json:"subject"`
+	Queue   string `json:"queue,omitempty"`
+}
+
 type CreateConsumerRequest struct {
 	Stream string         `json:"stream_name"`
 	Config ConsumerConfig `json:"config"`
@@ -190,6 +196,7 @@ type consumer struct {
 	asflr             uint64
 	sgap              uint64
 	dsubj             string
+	qgroup            string
 	lss               *lastSeqSkipList
 	rlimit            *rate.Limiter
 	reqSub            *subscription
@@ -688,6 +695,27 @@ func (o *consumer) setConsumerAssignment(ca *consumerAssignment) {
 	}
 }
 
+// checkInterest will check on our interest's queue group status.
+// Lock should be held.
+func (o *consumer) checkQueueInterest() {
+	if !o.active || o.cfg.DeliverSubject == _EMPTY_ {
+		return
+	}
+	subj := o.dsubj
+	if subj == _EMPTY_ {
+		subj = o.cfg.DeliverSubject
+	}
+
+	if rr := o.acc.sl.Match(subj); len(rr.qsubs) > 0 {
+		// Just grab first
+		if qsubs := rr.qsubs[0]; len(qsubs) > 0 {
+			if sub := rr.qsubs[0][0]; len(sub.queue) > 0 {
+				o.qgroup = string(sub.queue)
+			}
+		}
+	}
+}
+
 // Lock should be held.
 func (o *consumer) isLeader() bool {
 	if o.node != nil {
@@ -763,6 +791,8 @@ func (o *consumer) setLeader(isLeader bool) {
 					stopAndClearTimer(&o.gwdtmr)
 					o.gwdtmr = time.AfterFunc(time.Second, func() { o.watchGWinterest() })
 				}
+			} else {
+				o.checkQueueInterest()
 			}
 		}
 
@@ -968,7 +998,12 @@ func (o *consumer) updateDeliveryInterest(localInterest bool) bool {
 	if interest && !o.active {
 		o.signalNewMessages()
 	}
-	o.active = interest
+	// Update active status, if not active clear any queue group we captured.
+	if o.active = interest; !o.active {
+		o.qgroup = _EMPTY_
+	} else {
+		o.checkQueueInterest()
+	}
 
 	// If the delete timer has already been set do not clear here and return.
 	if o.dtmr != nil && !o.isDurable() && !interest {
@@ -1433,6 +1468,8 @@ func (o *consumer) writeStoreState() error {
 
 // Info returns our current consumer state.
 func (o *consumer) info() *ConsumerInfo {
+	var pa *PushActive
+
 	o.mu.RLock()
 	mset := o.mset
 	if mset == nil || mset.srv == nil {
@@ -1440,6 +1477,12 @@ func (o *consumer) info() *ConsumerInfo {
 		return nil
 	}
 	js := o.js
+	if o.active {
+		pa = &PushActive{
+			Subject: o.dsubj,
+			Queue:   o.qgroup,
+		}
+	}
 	o.mu.RUnlock()
 
 	if js == nil {
@@ -1468,6 +1511,7 @@ func (o *consumer) info() *ConsumerInfo {
 		NumAckPending:  len(o.pending),
 		NumRedelivered: len(o.rdc),
 		NumPending:     o.adjustedPending(),
+		Active:         pa,
 		Cluster:        ci,
 	}
 	// If we are a pull mode consumer, report on number of waiting requests.
diff --git a/server/stream.go b/server/stream.go
index e8bae834..69fbd92b 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"math"
 	"os"
 	"path"
 	"path/filepath"
@@ -79,6 +78,7 @@ func (r *JSPubAckResponse) ToError() error {
 type PubAck struct {
 	Stream    string `json:"stream"`
 	Sequence  uint64 `json:"seq"`
+	Domain    string `json:"domain,omitempty"`
 	Duplicate bool   `json:"duplicate,omitempty"`
 }
 
@@ -382,10 +382,11 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	}
 
 	// Create our pubAck template here. Better than json marshal each time on success.
-	b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: cfg.Name, Sequence: math.MaxUint64}})
-	end := bytes.Index(b, []byte(strconv.FormatUint(math.MaxUint64, 10)))
-	// We need to force cap here to make sure this is a copy when sending a response.
-	mset.pubAck = b[:end:end]
+	if domain := s.getOpts().JetStreamDomain; domain != _EMPTY_ {
+		mset.pubAck = []byte(fmt.Sprintf("{%q:%q, %q:%q, %q:", "stream", cfg.Name, "domain", domain, "seq"))
+	} else {
+		mset.pubAck = []byte(fmt.Sprintf("{%q:%q, %q:", "stream", cfg.Name, "seq"))
+	}
 
 	// Rebuild dedupe as needed.
 	mset.rebuildDedupe()


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 2ae97422..314e119e 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -554,7 +554,7 @@ func TestJetStreamClusterConsumerRedeliveredInfo(t *testing.T) {
 }
 
 func TestJetStreamClusterConsumerState(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	c := createJetStreamClusterExplicit(t, "R3S", 5)
 	defer c.shutdown()
 
 	s := c.randomServer()
@@ -578,15 +578,27 @@ func TestJetStreamClusterConsumerState(t *testing.T) {
 		}
 	}
 
+	// Make sure we are not connected to any of the stream servers so that we do not do client reconnect
+	// when we take out the consumer leader.
+	if s.JetStreamIsStreamAssigned("$G", "TEST") {
+		nc.Close()
+		for _, ns := range c.servers {
+			if !ns.JetStreamIsStreamAssigned("$G", "TEST") {
+				s = ns
+				nc, js = jsClientConnect(t, s)
+				defer nc.Close()
+				break
+			}
+		}
+	}
+
 	sub, err := js.PullSubscribe("foo", "dlc")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
 	// Pull 5 messages and ack.
-	for i := 0; i < 5; i++ {
-		msgs := fetchMsgs(t, sub, 1, 5*time.Second)
-		m := msgs[0]
+	for _, m := range fetchMsgs(t, sub, 5, 5*time.Second) {
 		m.Ack()
 	}
 
@@ -618,11 +630,10 @@ func TestJetStreamClusterConsumerState(t *testing.T) {
 
 	// Now make sure we can receive new messages.
 	// Pull last 5.
-	for i := 0; i < 5; i++ {
-		msgs := fetchMsgs(t, sub, 1, 5*time.Second)
-		m := msgs[0]
+	for _, m := range fetchMsgs(t, sub, 5, 5*time.Second) {
 		m.Ack()
 	}
+
 	nci, _ = sub.ConsumerInfo()
 	if nci.Delivered.Consumer != 10 || nci.Delivered.Stream != 10 {
 		t.Fatalf("Received bad delivered: %+v", nci.Delivered)
@@ -1835,6 +1846,9 @@ func TestJetStreamClusterExtendedStreamInfo(t *testing.T) {
 		return nil
 	})
 
+	nc, js = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
 	// Now do consumer.
 	sub, err := js.PullSubscribe("foo", "dlc")
 	if err != nil {
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 0acd0f12..ebc8b758 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -11791,6 +11791,145 @@ func TestJetStreamServerDomainConfigButDisabled(t *testing.T) {
 	}
 }
 
+func TestJetStreamDomainInPubAck(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		jetstream: {domain: "HUB"}
+	`))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	config := s.JetStreamConfig()
+	if config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Storage:  nats.MemoryStorage,
+		Subjects: []string{"foo"},
+	}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Check by hand for now til it makes its way into Go client.
+	am, err := nc.Request("foo", nil, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var pa PubAck
+	json.Unmarshal(am.Data, &pa)
+	if pa.Domain != "HUB" {
+		t.Fatalf("Expected PubAck to have domain of %q, got %q", "HUB", pa.Domain)
+	}
+}
+
+func TestJetStreamPushConsumerInfo(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	config := s.JetStreamConfig()
+	if config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Storage:  nats.MemoryStorage,
+		Subjects: []string{"foo"},
+	}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// We want to test extended consumer info for push based consumers.
+	// We need to do these by hand for now.
+	createConsumer := func(name, deliver string) {
+		t.Helper()
+		creq := CreateConsumerRequest{
+			Stream: "TEST",
+			Config: ConsumerConfig{
+				Durable:        name,
+				DeliverSubject: deliver,
+			},
+		}
+		req, err := json.Marshal(creq)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		resp, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, "TEST", name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var ccResp JSApiConsumerCreateResponse
+		if err := json.Unmarshal(resp.Data, &ccResp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ccResp.ConsumerInfo == nil || ccResp.Error != nil {
+			t.Fatalf("Got a bad response %+v", ccResp)
+		}
+	}
+
+	consumerInfo := func(name string) *ConsumerInfo {
+		t.Helper()
+		resp, err := nc.Request(fmt.Sprintf(JSApiConsumerInfoT, "TEST", name), nil, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var cinfo JSApiConsumerInfoResponse
+		if err := json.Unmarshal(resp.Data, &cinfo); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if cinfo.ConsumerInfo == nil || cinfo.Error != nil {
+			t.Fatalf("Got a bad response %+v", cinfo)
+		}
+		return cinfo.ConsumerInfo
+	}
+
+	// First create a durable push and make sure we show now active status.
+	createConsumer("dlc", "d.X")
+	if ci := consumerInfo("dlc"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+	// Now bind the deliver subject.
+	sub, _ := nc.SubscribeSync("d.X")
+	nc.Flush() // Make sure it registers.
+	// Check that its reported.
+	if ci := consumerInfo("dlc"); ci.Active == nil || ci.Active.Subject != "d.X" {
+		t.Fatalf("Expected active to be set and have subject %q, got %+v\n", "d.X", ci.Active)
+	}
+	sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	if ci := consumerInfo("dlc"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+
+	// Now make sure we have queue groups indictated as needed.
+	createConsumer("ik", "d.Z")
+	// Now bind the deliver subject with a queue group.
+	sub, _ = nc.QueueSubscribeSync("d.Z", "g22")
+	defer sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	// Check that queue group reported.
+	if ci := consumerInfo("ik"); ci.Active == nil || ci.Active.Subject != "d.Z" || ci.Active.Queue != "g22" {
+		t.Fatalf("Expected active to be set and have subject %q and queue %q, got %+v\n", "d.Z", "g22", ci.Active)
+	}
+	sub.Unsubscribe()
+	nc.Flush() // Make sure it registers.
+	if ci := consumerInfo("ik"); ci.Active != nil {
+		t.Fatalf("Expected active to be nil, got %+v\n", ci.Active)
+	}
+}
+
 // Issue #2213
 func TestJetStreamDirectConsumersBeingReported(t *testing.T) {
 	s := RunBasicJetStreamServer()