Repository: nats-io/nats-server (go)

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
```


Pull request description:
Fix performance issue in `firstMatchingMulti`
We should only return that the cache can be expired if the start sequence is the end of the block, otherwise we will repeatedly load and expire the cache of the same block in `LoadNextMsgMulti`. This results in a noticeable slowdown.

Fixes #7014.

Signed-off-by: Neil Twigg <neil@nats.io>

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
+	})
+}