Repository: nats-io/nats-server

Issue:
JetStream Consumer Subject Filtering Causes 99% Performance Degradation
### Observed behavior

JetStream consumers with subject filters experience severe performance degradation (99%+) compared to unfiltered consumers, even when filtering for exact literal subjects that represent the vast majority of messages in a stream.


### Expected behavior

Filtering for a subject that represents 99.99% of messages should have minimal performance impact, as the vast majority of messages match the filter anyway.


### Server and client version

Server: v2.11.4 (v2.11.5 as well, see update below)
CLI: built from `main` (with patch from https://github.com/nats-io/natscli/pull/1405 for `--filter` support in bench command)


### Host environment

3-node cluster on AWS EKS. NVMe storage.

### Steps to reproduce

1. Create a stream with wildcard subjects containing messages with skewed distribution:
```bash
nats stream create bench --subjects="bench.>" --storage=file --replicas=3 --defaults
```

2. Publish messages with heavily skewed subject distribution:
```bash
# 99.99% of messagess
nats bench js pub bench.common --stream=bench --msgs=999900

# 0.01% of messages
nats bench js pub bench.rare --stream=bench --msgs=100
```

3. Benchmark consumer without filter:
```bash
nats bench js --stream=bench consume --no-progress
# Result: 69,863 msgs/sec ~ 8.53 MB/sec
```

4. Benchmark consumer with exact literal filter for the common subject:
```bash
nats bench js --stream=bench consume --filter=events.common --no-progress
# Result: 305 msgs/sec ~ 38.20 KB/sec
```

UPDATE:

Same benchmark on v2.11.5 results filtered consumer to hang forever not returning a single message.

```bash
nats bench js --stream=bench consume --filter=events.common --no-progress
# Hangs
```


Discussion:
@MauriceVanVeen: Bisecting points to the regression being introduced in https://github.com/nats-io/nats-server/commit/5282516a1def57df55e4cac1dcd4c8bf534a3539.@mikluko: <DELETED>

**Edit:** Removed incorrect benchmarking data from this comment. Posted corrected benchmarks in a follow-up comment below. Thanks @neilalexander for catching the error!@neilalexander: Please re-test against #7015, it should help.@neilalexander: Also just to note that you are mixing `bench.common` and `events.common` subjects in your above snippets, which may explain why you are getting no messages.@jnmoyne: By default "consume" will expect 100k messages. However if you put 990,000 messages with one subject and 10,000 with another subject and then do "consume" with the default it will not complete because you only have 990,000 messages in the stream that match your filter. You need to also specify `-msgs` with the number of messages expected to match the filter. @mikluko: > By default "consume" will expect 100k messages. However if you put 990,000 messages with one subject and 10,000 with another subject and then do "consume" with the default it will not complete because you only have 990,000 messages in the stream that match your filter. You need to also specify `-msgs` with the number of messages expected to match the filter.

There are 1M messages total. 100 of them are "rare". Others are "common". I'm running benchmark which is supposed to fetch 100k of common ones from the stream. And it does, but with terrible performance. I don't see any logical gap here.

Just to clarify - I'm filtering for the "common" subject (999,900 messages), not the rare one. So plenty of messages to satisfy the 100k default. The benchmark completes and fetches 100k messages as expected, just with significant performance degradation compared to unfiltered consumer.@mikluko: > Also just to note that you are mixing `bench.common` and `events.common` subjects in your above snippets, which may explain why you are getting no messages.

Oh... Good catch, thank you and sorry for the confusion. Below is updated benchmarking for v 2.11.5:

```
debug-shell# nats account info | grep "Server Version:"
       Connected Server Version: 2.11.5
```

```
debug-shell# nats stream subjects bench
╭───────────────────────────────────────────────────────────────╮
│                   2 Subjects in stream bench                  │
├────────────┬───────┬──────────────┬─────────┬─────────┬───────┤
│ Subject    │ Count │ Subject      │ Count   │ Subject │ Count │
├────────────┼───────┼──────────────┼─────────┼─────────┼───────┤
│ bench.rare │ 100   │ bench.common │ 999,900 │         │       │
╰────────────┴───────┴──────────────┴─────────┴─────────┴───────╯
```

```
debug-shell# nats bench js --stream=bench consume --no-progress
06:44:11 Starting JetStream durable consumer (callback) benchmark [acks=explicit, batch=500, clients=1, consumer=nats-bench, double-acked=false, msg-size=128 B, msgs=100,000, purge=false, sleep=0s, stream=bench]
06:44:11 Starting subscriber, expecting 100,000 messages

Sub stats: 95,509 msgs/sec ~ 11.66 MB/sec
```

```
debug-shell# nats bench js --stream=bench consume --no-progress --filter=bench.common
06:44:48 Starting JetStream durable consumer (callback) benchmark [acks=explicit, batch=500, clients=1, consumer=nats-bench, double-acked=false, filter=bench.common, msg-size=128 B, msgs=100,000, purge=false, sleep=0s, stream=bench]
06:44:48 Starting subscriber, expecting 100,000 messages
Finished   2m52s 

Sub stats: 581 msgs/sec ~ 72.65 KB/sec
```

Bottomline: it doesn't hang as I originally reported, but disaster-grade performance drop is still there.

I will try #1705 build and let you know.@mikluko: ```
debug-shell# nats account info | grep "Server Version:"
       Connected Server Version: 2.12.0-dev
debug-shell# nats stream subjects bench
╭───────────────────────────────────────────────────────────────╮
│                   2 Subjects in stream bench                  │
├────────────┬───────┬──────────────┬─────────┬─────────┬───────┤
│ Subject    │ Count │ Subject      │ Count   │ Subject │ Count │
├────────────┼───────┼──────────────┼─────────┼─────────┼───────┤
│ bench.rare │ 100   │ bench.common │ 999,900 │         │       │
╰────────────┴───────┴──────────────┴─────────┴─────────┴───────╯

debug-shell# nats bench js --stream=bench consume --no-progress
07:59:19 Starting JetStream durable consumer (callback) benchmark [acks=explicit, batch=500, clients=1, consumer=nats-bench, double-acked=false, msg-size=128 B, msgs=100,000, purge=false, sleep=0s, stream=bench]
07:59:19 Starting subscriber, expecting 100,000 messages

Sub stats: 98,900 msgs/sec ~ 12.07 MB/sec

debug-shell# nats bench js --stream=bench consume --no-progress --filter=bench.common
07:59:40 Starting JetStream durable consumer (callback) benchmark [acks=explicit, batch=500, clients=1, consumer=nats-bench, double-acked=false, filter=bench.common, msg-size=128 B, msgs=100,000, purge=false, sleep=0s, stream=bench]
07:59:40 Starting subscriber, expecting 100,000 messages

Sub stats: 99,321 msgs/sec ~ 12.12 MB/sec
```

I can confirm #7015 resolves the issue. Thanks @neilalexander! Looking forward to see it released.@neilalexander: Thank you for confirming, that matches my own checks. Will merge it, thanks for the report!

Solution patch:
diff --git a/server/filestore.go b/server/filestore.go
index ee9f58e1..58b12274 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2418,10 +2418,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 				// mb is already loaded into the cache so should be fast-ish.
 				mb.recalculateForSubj(bytesToString(subj), ss)
 			}
-			first := ss.First
-			if start > first {
-				first = start
-			}
+			first := max(start, ss.First)
 			if first > ss.Last || first >= hseq {
 				// The start cutoff is after the last sequence for this subject,
 				// or we think we already know of a subject with an earlier msg
@@ -2463,7 +2460,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 			}
 		})
 		if hseq < uint64(math.MaxUint64) && sm != nil {
-			return sm, didLoad, nil
+			return sm, didLoad && start == lseq, nil
 		}
 	} else {
 		for seq := start; seq <= lseq; seq++ {



Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 17aeeae7..07e052d1 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -9687,3 +9687,41 @@ func BenchmarkFileStoreSubjectAccesses(b *testing.B) {
 		}
 	})
 }
+
+func TestFileStoreFirstMatchingMultiExpiry(t *testing.T) {
+	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
+		cfg := StreamConfig{Name: "zzz", Subjects: []string{"foo.>"}, Storage: FileStorage}
+		fs, err := newFileStoreWithCreated(fcfg, cfg, time.Now(), prf(&fcfg), nil)
+		require_NoError(t, err)
+		defer fs.Stop()
+
+		_, _, err = fs.StoreMsg("foo.foo", nil, []byte("A"), 0)
+		require_NoError(t, err)
+
+		_, _, err = fs.StoreMsg("foo.foo", nil, []byte("B"), 0)
+		require_NoError(t, err)
+
+		_, _, err = fs.StoreMsg("foo.foo", nil, []byte("C"), 0)
+		require_NoError(t, err)
+
+		fs.mu.RLock()
+		mb := fs.lmb
+		mb.expireCacheLocked()
+		fs.mu.RUnlock()
+
+		sl := NewSublistNoCache()
+		sl.Insert(&subscription{subject: []byte("foo.foo")})
+
+		_, didLoad, err := mb.firstMatchingMulti(sl, 1, nil)
+		require_NoError(t, err)
+		require_False(t, didLoad)
+
+		_, didLoad, err = mb.firstMatchingMulti(sl, 2, nil)
+		require_NoError(t, err)
+		require_False(t, didLoad)
+
+		_, didLoad, err = mb.firstMatchingMulti(sl, 3, nil)
+		require_NoError(t, err)
+		require_True(t, didLoad) // last message, should expire
+	})
+}