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

Issue:
After changing the replica of a stream: The non-leaders have more messages
## Defect

Make sure that these boxes are checked before submitting your issue -- thank you!

 - [x] Included `nats-server -DV` output
```
[124] 2023/02/06 08:14:04.050458 [INF] Starting nats-server
[124] 2023/02/06 08:14:04.050483 [INF]   Version:  2.9.11
[124] 2023/02/06 08:14:04.050485 [INF]   Git:      [23ffc16]
[124] 2023/02/06 08:14:04.050491 [DBG]   Go build: go1.19.4
[124] 2023/02/06 08:14:04.050496 [INF]   Name:     NBS3DHEIAJA2NGY2ZQ6IL2MK4YLIDGIZNF3IKORWUNMVAQEJOGJHY3EZ
[124] 2023/02/06 08:14:04.050498 [INF]   ID:       NBS3DHEIAJA2NGY2ZQ6IL2MK4YLIDGIZNF3IKORWUNMVAQEJOGJHY3EZ
```
 - [x] Included a Minimal, Complete, and Verifiable example: Please see section _"Steps or code to reproduce the issue"_

#### Versions of `nats-server` and affected client libraries used:

Reproducible in different versions and setups.

#### OS/Container environment:

Reproducible for various operating systems and environments (k8s, locally mac, locally linux, …).

#### Steps or code to reproduce the issue:

<details>
  <summary><code>docker-compose.yml</code></summary>

  ```yml
  version: '3'
  services:
    nats-1:
      container_name: nats-1
      image: nats:2.9.11
      entrypoint: /nats-server
      command: --name nats-1 --cluster_name NATS --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats-1:4245,nats://nats-2:4245,nats://nats-3:4245 -p 4222 --http_port 8222
      networks:
        - nats
      ports:
        - 14222:4222
        - 18222:8222
  
    nats-2:
      container_name: nats-2
      image: nats:2.9.11
      entrypoint: /nats-server
      command: --name nats-2 --cluster_name NATS --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats-1:4245,nats://nats-2:4245,nats://nats-3:4245 -p 4222 --http_port 8222
      networks:
        - nats
      ports:
        - 24222:4222
        - 28222:8222
  
    nats-3:
      container_name: nats-3
      image: nats:2.9.11
      entrypoint: /nats-server
      command: --name nats-3 --cluster_name NATS --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats-1:4245,nats://nats-2:4245,nats://nats-3:4245 -p 4222 --http_port 8222
      networks:
        - nats
      ports:
        - 34222:4222
        - 38222:8222
  
    nats-setup:
      container_name: nats_setup
      depends_on:
        - nats-1
        - nats-2
        - nats-3
      build:
        context: .
      environment:
        - NATS_URL=nats://host.docker.internal:14222
      networks:
        - nats
  
  networks:
    nats: {}
  ```
</details>

<details>
  <summary><code>Dockerfile</code></summary>

  ```yml
  FROM alpine:3.17
  
  RUN apk add --no-cache wget rclone curl jq && \
      wget -P ~ https://github.com/nats-io/natscli/releases/download/v0.0.35/nats-0.0.35-linux-arm64.zip && \
      unzip -d ~ ~/nats-0.0.35-linux-arm64.zip && \
      cp ~/nats-0.0.35-linux-arm64/nats /usr/local/bin && \
      adduser -D user
  
  WORKDIR /workdir
  
  COPY stream.conf .
  COPY script.sh .
  RUN chmod +x script.sh
  
  USER user
  
  ENTRYPOINT ["/workdir/script.sh"]
  ```
</details>

<details>
  <summary><code>script.sh</code></summary>

  ```sh
  #!/bin/sh
  
  function fetch_stream_information() {
    N1=$(curl --silent "http://nats-1:8222/jsz?consumers=1" | jq ".messages")
    N2=$(curl --silent "http://nats-2:8222/jsz?consumers=1" | jq ".messages")
    N3=$(curl --silent "http://nats-3:8222/jsz?consumers=1" | jq ".messages")
  
    echo "=================="
    echo "nats-1: $N1"
    echo "nats-2: $N2"
    echo "nats-3: $N3"
    echo "=================="
  }
  
  nats stream ls
  nats stream create --config=stream.conf
  nats stream info test
  
  nats pub test.a --count 10 "Message {{Count}} @ {{Time}}"
  sleep 1
  nats pub test.a --count 10 "Message {{Count}} @ {{Time}}"
  sleep 1
  nats pub test.a --count 10 "Message {{Count}} @ {{Time}}"
  sleep 1
  nats pub test.a --count 10 "Message {{Count}} @ {{Time}}"
  sleep 1
  nats pub test.a --count 10 "Message {{Count}} @ {{Time}}"
  sleep 1
  
  fetch_stream_information
  
  nats stream edit --replicas=3 --force test
  
  sleep 2
  fetch_stream_information
  sleep 2
  fetch_stream_information
  sleep 2
  fetch_stream_information
  
  nats pub test.a --count 4 "Message {{Count}} @ {{Time}}"
  
  sleep 1
  fetch_stream_information
  sleep 1
  fetch_stream_information
  sleep 1
  fetch_stream_information
  sleep 1
  fetch_stream_information
  sleep 1
  fetch_stream_information
  ```
</details>

<details>
  <summary><code>stream.conf</code></summary>

  ```json
  {
    "config":{
      "name":"test",
      "subjects":[
        "test.*"
      ],
      "retention":"limits",
      "max_consumers":-1,
      "max_msgs":100,
      "max_bytes":-1,
      "max_age":10000000000,
      "max_msg_size":-1,
      "storage":"file",
      "discard":"old",
      "num_replicas":1,
      "duplicate_window":5000000000,
      "deny_delete":false,
      "deny_purge":false,
      "allow_rollup_hdrs":true
    },
    "created":"2021-02-27T23:49:36.700424Z",
    "state":{
      "messages":0,
      "bytes":0,
      "first_seq":0,
      "first_ts":"0001-01-01T00:00:00Z",
      "last_seq":0,
      "last_ts":"0001-01-01T00:00:00Z",
      "consumer_count":0
    }
  }
  ```
</details>

1. download all files
2. run `docker compose up --build --force-recreate --remove-orphans --renew-anon-volumes`
3. observe logs

Short explanation what this scenario does:

1. Creation of a stream with `replica: 1` and a time based retention
2. Publish 50 messages
3. Change stream configuration to `replica: 3`
4. Wait
5. Publish 4 messages
6. ⚠️ The replicas have more messages than the leader

#### Expected result:

After changing the replica size of a stream and the first retention of a message the messages on all replicas are the same.

#### Actual result:

After changing the replica size of a stream and the first retention of a message the messages on the leader are the correct ones, but the replicas have old messages.


**Thanks in advance! 🙏** 

Cheers
Stefan

Pull request description:
[FIXED] Filtered Consumers return wrong messages.
When a block's subject meta state was swapped out and subsequently loaded back in with only one subject present, but other messages with different subjects were added later, a filtered get could return the wrong result.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #3848

/cc @nats-io/core


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index fb63c026..e9e08a86 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -147,7 +147,6 @@ type msgBlock struct {
 	rbytes  uint64 // Total bytes (raw) including deleted. Used for rolling to new blk.
 	msgs    uint64 // User visible message count.
 	fss     map[string]*SimpleState
-	sfilter string // Single subject filter
 	sfn     string
 	kfn     string
 	lwits   int64
@@ -419,18 +418,6 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 		fs.ageChk.Stop()
 		fs.ageChk = nil
 	}
-
-	// Update our sfilter for the last block.
-	if lmb := fs.lmb; lmb != nil {
-		lmb.mu.Lock()
-		if len(fs.cfg.Subjects) == 1 {
-			lmb.sfilter = fs.cfg.Subjects[0]
-		} else {
-			lmb.sfilter = _EMPTY_
-		}
-		lmb.mu.Unlock()
-	}
-
 	fs.mu.Unlock()
 
 	if cfg.MaxAge != 0 {
@@ -1103,13 +1090,6 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		mb.last.seq = mb.first.seq - 1
 	}
 
-	// If we only have one subject registered we can optimize filtered lookups here.
-	if len(mb.fss) == 1 {
-		for sfilter := range mb.fss {
-			mb.sfilter = sfilter
-		}
-	}
-
 	return nil, nil
 }
 
@@ -1159,12 +1139,6 @@ func (fs *fileStore) recoverMsgs() error {
 	if len(fs.blks) > 0 {
 		sort.Slice(fs.blks, func(i, j int) bool { return fs.blks[i].index < fs.blks[j].index })
 		fs.lmb = fs.blks[len(fs.blks)-1]
-		// Update our sfilter for the last block since we could have only see one subject during recovery.
-		if len(fs.cfg.Subjects) == 1 {
-			fs.lmb.sfilter = fs.cfg.Subjects[0]
-		} else {
-			fs.lmb.sfilter = _EMPTY_
-		}
 	} else {
 		_, err = fs.newMsgBlockForWrite()
 	}
@@ -1406,8 +1380,12 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		return nil, false, err
 	}
 
-	fseq, isAll, subs := start, filter == _EMPTY_ || filter == mb.sfilter || filter == fwcs, []string{filter}
+	fseq, isAll, subs := start, filter == _EMPTY_ || filter == fwcs, []string{filter}
 
+	// If we only have 1 subject currently and it matches our filter we can also set isAll.
+	if !isAll && len(mb.fss) == 1 {
+		_, isAll = mb.fss[filter]
+	}
 	// Skip scan of mb.fss if number of messages in the block are less than
 	// 1/2 the number of subjects in mb.fss. Or we have a wc and lots of fss entries.
 	const linearScanMaxFSS = 32
@@ -1796,12 +1774,6 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 
 	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire, noTrack: fs.noTrackSubjects()}
 
-	// Optimize lookups if we have one subject.
-	// TODO(dlc) - Make work for mirrors?
-	if !mb.noTrack && len(fs.cfg.Subjects) == 1 {
-		mb.sfilter = fs.cfg.Subjects[0]
-	}
-
 	// Lock should be held to quiet race detector.
 	mb.mu.Lock()
 	mb.setupWriteCache(rbuf)
@@ -5249,19 +5221,6 @@ func (mb *msgBlock) readPerSubjectInfo(hasLock bool) error {
 		return nil
 	}
 
-	defer func() {
-		if !hasLock {
-			mb.mu.Lock()
-			defer mb.mu.Unlock()
-		}
-		// If we only have one subject registered we can optimize filtered lookups here.
-		if len(mb.fss) == 1 {
-			for sfilter := range mb.fss {
-				mb.sfilter = sfilter
-			}
-		}
-	}()
-
 	buf, err := mb.loadPerSubjectInfo()
 	// On failure re-generate.
 	if err != nil {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index b7e7a30b..f6bd449e 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4623,3 +4623,50 @@ func TestFileStoreFSSExpireNumPendingBug(t *testing.T) {
 		t.Fatalf("Expected only 1 msg, got %d", fss.Msgs)
 	}
 }
+
+// https://github.com/nats-io/nats-server/issues/3484
+func TestFileStoreFilteredFirstMatchingBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.>"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	_, _, err = fs.StoreMsg("foo.foo", nil, []byte("A"))
+	require_NoError(t, err)
+
+	_, _, err = fs.StoreMsg("foo.foo", nil, []byte("B"))
+	require_NoError(t, err)
+
+	_, _, err = fs.StoreMsg("foo.foo", nil, []byte("C"))
+	require_NoError(t, err)
+
+	fs.mu.RLock()
+	mb := fs.lmb
+	fs.mu.RUnlock()
+
+	mb.mu.Lock()
+	// Simulate swapping out the fss state and reading it back in with only one subject
+	// present in the block.
+	if mb.fss != nil {
+		mb.writePerSubjectInfo()
+		mb.fss = nil
+	}
+	// Now load info back in.
+	mb.readPerSubjectInfo(true)
+	mb.mu.Unlock()
+
+	// Now add in a different subject.
+	_, _, err = fs.StoreMsg("foo.bar", nil, []byte("X"))
+	require_NoError(t, err)
+
+	// Now see if a filtered load would incorrectly succeed.
+	sm, _, err := fs.LoadNextMsg("foo.foo", false, 4, nil)
+	if err == nil || sm != nil {
+		t.Fatalf("Loaded filtered message with wrong subject, wanted %q got %q", "foo.foo", sm.subj)
+	}
+}