Repository: nats-io/nats-server
Pull request #2201 (2021-05-10 22:26:01)

Fix for failing test
We needed to account for consumers having filtered subjects configured but not really being filtered.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/auth.go b/server/auth.go
index 1d7b45e4..72cd94df 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -40,7 +40,7 @@ type Authentication interface {
 // ClientAuthentication is an interface for client authentication
 type ClientAuthentication interface {
 	// Get options associated with a client
-	GetOpts() *ClientOpts
+	GetOpts() *clientOpts
 	// If TLS is enabled, TLS ConnectionState, nil otherwise
 	GetTLSConnectionState() *tls.ConnectionState
 	// Optionally map a user after auth.
diff --git a/server/client.go b/server/client.go
index e5aba72e..2470657a 100644
--- a/server/client.go
+++ b/server/client.go
@@ -222,7 +222,7 @@ type client struct {
 	perms *permissions
 	in    readCache
 	parseState
-	opts       ClientOpts
+	opts       clientOpts
 	rrTracking *rrTracking
 	mpay       int32
 	msubs      int32
@@ -434,7 +434,7 @@ func (c *client) GetName() string {
 }
 
 // GetOpts returns the client options provided by the application.
-func (c *client) GetOpts() *ClientOpts {
+func (c *client) GetOpts() *clientOpts {
 	return &c.opts
 }
 
@@ -508,7 +508,7 @@ func (s *subscription) isClosed() bool {
 	return atomic.LoadInt32(&s.closed) == 1
 }
 
-type ClientOpts struct {
+type clientOpts struct {
 	Echo         bool   `json:"echo"`
 	Verbose      bool   `json:"verbose"`
 	Pedantic     bool   `json:"pedantic"`
@@ -533,8 +533,8 @@ type ClientOpts struct {
 	Export *SubjectPermission `json:"export,omitempty"`
 }
 
-var defaultOpts = ClientOpts{Verbose: true, Pedantic: true, Echo: true}
-var internalOpts = ClientOpts{Verbose: false, Pedantic: false, Echo: false}
+var defaultOpts = clientOpts{Verbose: true, Pedantic: true, Echo: true}
+var internalOpts = clientOpts{Verbose: false, Pedantic: false, Echo: false}
 
 func (c *client) setTraceLevel() {
 	if c.kind == SYSTEM && !(atomic.LoadInt32(&c.srv.logging.traceSysAcc) != 0) {
diff --git a/server/consumer.go b/server/consumer.go
index 358d3647..2d2a9932 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1586,6 +1586,25 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	}
 }
 
+// Determine if this is a truly filtered consumer. Modern clients will place filtered subjects
+// even if the stream only has a single non-wildcard subject designation.
+// Read lock should be held.
+func (o *consumer) isFiltered() bool {
+	if o.cfg.FilterSubject == _EMPTY_ {
+		return false
+	}
+	// If we are here we want to check if the filtered subject is
+	// a direct match for our only listed subject.
+	mset := o.mset
+	if mset == nil {
+		return true
+	}
+	if len(mset.cfg.Subjects) > 1 {
+		return true
+	}
+	return o.cfg.FilterSubject != mset.cfg.Subjects[0]
+}
+
 // Check if we need an ack for this store seq.
 // This is called for interest based retention streams to remove messages.
 func (o *consumer) needAck(sseq uint64) bool {
@@ -1604,9 +1623,9 @@ func (o *consumer) needAck(sseq uint64) bool {
 		state, err := o.store.State()
 		if err != nil || state == nil {
 			// Fall back to what we track internally for now.
-			needsAck := sseq > o.asflr && o.cfg.FilterSubject == _EMPTY_
+			needAck := sseq > o.asflr && !o.isFiltered()
 			o.mu.RUnlock()
-			return needsAck
+			return needAck
 		}
 		asflr, osseq = state.AckFloor.Stream, o.sseq
 		pending = state.Pending
@@ -2908,7 +2927,7 @@ func (o *consumer) setInitialPending() {
 	if !notFiltered {
 		// Check to see if we directly match the configured stream.
 		// Many clients will always send a filtered subject.
-		cfg := mset.cfg
+		cfg := &mset.cfg
 		if len(cfg.Subjects) == 1 && cfg.Subjects[0] == o.cfg.FilterSubject {
 			notFiltered = true
 		}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 34ae2e36..1c44232d 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3378,9 +3378,12 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 	cfg := &ccfg
 
 	// Check for stream limits here before proposing. These need to be tracked from meta layer, not jsa.
-	jsa.mu.RLock()
+	js.mu.RLock()
 	asa := cc.streams[acc.Name]
 	numStreams := len(asa)
+	js.mu.RUnlock()
+
+	jsa.mu.RLock()
 	exceeded := jsa.limits.MaxStreams > 0 && numStreams >= jsa.limits.MaxStreams
 	jsa.mu.RUnlock()
 
diff --git a/server/route.go b/server/route.go
index f7bbdadd..fa2c0a69 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1292,7 +1292,7 @@ func (s *Server) createRoute(conn net.Conn, rURL *url.URL) *client {
 		}
 	}
 
-	c := &client{srv: s, nc: conn, opts: ClientOpts{}, kind: ROUTER, msubs: -1, mpay: -1, route: r}
+	c := &client{srv: s, nc: conn, opts: clientOpts{}, kind: ROUTER, msubs: -1, mpay: -1, route: r}
 
 	// Grab server variables
 	s.mu.Lock()


Test patch:
diff --git a/server/client_test.go b/server/client_test.go
index c4c3f3d0..ed3873a1 100644
--- a/server/client_test.go
+++ b/server/client_test.go
@@ -489,7 +489,7 @@ func TestClientConnect(t *testing.T) {
 	if c.state != OP_START {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
-	if !reflect.DeepEqual(c.opts, ClientOpts{Verbose: true, Pedantic: true, Echo: false}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Verbose: true, Pedantic: true, Echo: false}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 
@@ -503,7 +503,7 @@ func TestClientConnect(t *testing.T) {
 	if c.state != OP_START {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
-	if !reflect.DeepEqual(c.opts, ClientOpts{Echo: true, Verbose: true, Pedantic: true, Username: "derek", Password: "foo"}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Echo: true, Verbose: true, Pedantic: true, Username: "derek", Password: "foo"}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 
@@ -518,7 +518,7 @@ func TestClientConnect(t *testing.T) {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
 
-	if !reflect.DeepEqual(c.opts, ClientOpts{Echo: true, Verbose: true, Pedantic: true, Username: "derek", Password: "foo", Name: "router"}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Echo: true, Verbose: true, Pedantic: true, Username: "derek", Password: "foo", Name: "router"}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 
@@ -533,7 +533,7 @@ func TestClientConnect(t *testing.T) {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
 
-	if !reflect.DeepEqual(c.opts, ClientOpts{Echo: true, Verbose: true, Pedantic: true, Token: "YZZ222", Name: "router"}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Echo: true, Verbose: true, Pedantic: true, Token: "YZZ222", Name: "router"}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 }
@@ -551,7 +551,7 @@ func TestClientConnectProto(t *testing.T) {
 	if c.state != OP_START {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
-	if !reflect.DeepEqual(c.opts, ClientOpts{Echo: true, Verbose: true, Pedantic: true, Protocol: ClientProtoZero}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Echo: true, Verbose: true, Pedantic: true, Protocol: ClientProtoZero}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 
@@ -564,7 +564,7 @@ func TestClientConnectProto(t *testing.T) {
 	if c.state != OP_START {
 		t.Fatalf("Expected state of OP_START vs %d\n", c.state)
 	}
-	if !reflect.DeepEqual(c.opts, ClientOpts{Echo: true, Verbose: true, Pedantic: true, Protocol: ClientProtoInfo}) {
+	if !reflect.DeepEqual(c.opts, clientOpts{Echo: true, Verbose: true, Pedantic: true, Protocol: ClientProtoInfo}) {
 		t.Fatalf("Did not parse connect options correctly: %+v\n", c.opts)
 	}
 	if c.opts.Protocol != ClientProtoInfo {
@@ -1685,7 +1685,7 @@ func TestClientUserInfo(t *testing.T) {
 	pnkey := "UD6AYQSOIN2IN5OGC6VQZCR4H3UFMIOXSW6NNS6N53CLJA4PB56CEJJI"
 	c := &client{
 		cid: 1024,
-		opts: ClientOpts{
+		opts: clientOpts{
 			Nkey: pnkey,
 		},
 	}
@@ -1697,7 +1697,7 @@ func TestClientUserInfo(t *testing.T) {
 
 	c = &client{
 		cid: 1024,
-		opts: ClientOpts{
+		opts: clientOpts{
 			Username: "foo",
 		},
 	}
@@ -1709,7 +1709,7 @@ func TestClientUserInfo(t *testing.T) {
 
 	c = &client{
 		cid:  1024,
-		opts: ClientOpts{},
+		opts: clientOpts{},
 	}
 	got = c.getAuthUser()
 	expected = `User "N/A"`
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 5fccb5c4..330373aa 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -2072,9 +2072,6 @@ func TestJetStreamClusterMirrorAndSourceWorkQueues(t *testing.T) {
 }
 
 func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
-	// FIXME(dlc) - Flaky on Travis, skip for now.
-	skip(t)
-
 	c := createJetStreamClusterExplicit(t, "WQ", 3)
 	defer c.shutdown()
 
@@ -2083,9 +2080,9 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 	defer nc.Close()
 
 	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "WQ22",
+		Name:      "IP22",
 		Subjects:  []string{"foo"},
-		Replicas:  2,
+		Replicas:  3,
 		Retention: nats.InterestPolicy,
 	})
 	if err != nil {
@@ -2095,7 +2092,7 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:     "M",
 		Replicas: 2,
-		Mirror:   &nats.StreamSource{Name: "WQ22"},
+		Mirror:   &nats.StreamSource{Name: "IP22"},
 	})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -2104,7 +2101,7 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:     "S",
 		Replicas: 2,
-		Sources:  []*nats.StreamSource{{Name: "WQ22"}},
+		Sources:  []*nats.StreamSource{{Name: "IP22"}},
 	})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -2118,8 +2115,8 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		// This one will be 0 since no other interest exists.
-		if si, _ := js.StreamInfo("WQ22"); si.State.Msgs != 0 {
-			return fmt.Errorf("Expected no msgs for %q, got %d", "WQ22", si.State.Msgs)
+		if si, _ := js.StreamInfo("IP22"); si.State.Msgs != 0 {
+			return fmt.Errorf("Expected no msgs for %q, got %d", "IP22", si.State.Msgs)
 		}
 		if si, _ := js.StreamInfo("M"); si.State.Msgs != 1 {
 			return fmt.Errorf("Expected 1 msg for %q, got %d", "M", si.State.Msgs)
@@ -2130,7 +2127,7 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 		return nil
 	})
 
-	// Now create other interest on WQ22.
+	// Now create other interest on IP22.
 	sub, err := js.SubscribeSync("foo")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -2145,8 +2142,8 @@ func TestJetStreamClusterMirrorAndSourceInterestPolicyStream(t *testing.T) {
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
 		// This one will be 0 since no other interest exists.
-		if si, _ := js.StreamInfo("WQ22"); si.State.Msgs != 1 {
-			return fmt.Errorf("Expected 1 msg for %q, got %d", "WQ22", si.State.Msgs)
+		if si, _ := js.StreamInfo("IP22"); si.State.Msgs != 1 {
+			return fmt.Errorf("Expected 1 msg for %q, got %d", "IP22", si.State.Msgs)
 		}
 		if si, _ := js.StreamInfo("M"); si.State.Msgs != 2 {
 			return fmt.Errorf("Expected 2 msgs for %q, got %d", "M", si.State.Msgs)