Repository: nats-io/nats-server
Pull request #4873 (2023-12-11 17:21:54)

Check all filterSubjects of consumers when purging a stream
Closes https://github.com/nats-io/nats-server/issues/4826



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index ef0eb7d2..68f55418 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -3350,7 +3350,7 @@ func (o *consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 	o.sendAdvisory(o.deliveryExcEventT, j)
 }
 
-// Check to see if the candidate subject matches a filter if its present.
+// Check if the candidate subject matches a filter if its present.
 // Lock should be held.
 func (o *consumer) isFilteredMatch(subj string) bool {
 	// No filter is automatic match.
@@ -3374,6 +3374,25 @@ func (o *consumer) isFilteredMatch(subj string) bool {
 	return false
 }
 
+// Check if the candidate filter subject is equal to or a subset match
+// of one of the filter subjects.
+// Lock should be held.
+func (o *consumer) isEqualOrSubsetMatch(subj string) bool {
+	for _, filter := range o.subjf {
+		if !filter.hasWildcard && subj == filter.subject {
+			return true
+		}
+	}
+	tsa := [32]string{}
+	tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
+	for _, filter := range o.subjf {
+		if isSubsetMatchTokenized(filter.tokenizedSubject, tts) {
+			return true
+		}
+	}
+	return false
+}
+
 var (
 	errMaxAckPending = errors.New("max ack pending reached")
 	errBadConsumer   = errors.New("consumer not valid")
diff --git a/server/stream.go b/server/stream.go
index f4c630d0..2241efdb 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2001,11 +2001,10 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 		// no subject was specified, we can purge all consumers sequences
 		doPurge := preq == nil ||
 			preq.Subject == _EMPTY_ ||
-			// or consumer filter subject is equal to purged subject
-			preq.Subject == o.cfg.FilterSubject ||
-			// or consumer subject is subset of purged subject,
+			// consumer filter subject is equal to purged subject
+			// or consumer filter subject is subset of purged subject,
 			// but not the other way around.
-			subjectIsSubsetMatch(o.cfg.FilterSubject, preq.Subject)
+			o.isEqualOrSubsetMatch(preq.Subject)
 		o.mu.RUnlock()
 		if doPurge {
 			o.purge(fseq, lseq)


Test patch:
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index ed3de3ad..ec8fdf12 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -871,6 +871,36 @@ func TestJetStreamConsumerIsFilteredMatch(t *testing.T) {
 	}
 }
 
+func TestJetStreamConsumerIsEqualOrSubsetMatch(t *testing.T) {
+	for _, test := range []struct {
+		name           string
+		filterSubjects []string
+		subject        string
+		result         bool
+	}{
+		{"no filter", []string{}, "foo.bar", false},
+		{"literal match", []string{"foo.baz", "foo.bar"}, "foo.bar", true},
+		{"literal mismatch", []string{"foo.baz", "foo.bar"}, "foo.ban", false},
+		{"literal match", []string{"bar.>", "foo.>"}, "foo.>", true},
+		{"subset match", []string{"bar.foo.>", "foo.bar.>"}, "bar.>", true},
+		{"subset mismatch", []string{"bar.>", "foo.>"}, "baz.foo.>", false},
+		{"literal match", filterSubjects(100), "foo.bar.*.xyz.abcdef", true},
+		{"subset match", filterSubjects(100), "foo.bar.>", true},
+	} {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			c := consumerWithFilterSubjects(test.filterSubjects)
+			if res := c.isEqualOrSubsetMatch(test.subject); res != test.result {
+				t.Fatalf("Subject %q subset match of %v, should be %v, got %v",
+					test.subject, test.filterSubjects, test.result, res)
+			}
+		})
+	}
+}
+
 func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
 	subject := "foo.bar.do.not.match.any.filter.subject"
 	for n := 1; n <= 1024; n *= 2 {