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

Issue:
Subject filtered purge does not delete all data.
An easy way to test this is with object- seems to happen with big streams.

```
$ nats obj add BIGDATA
$  dd if=/dev/zero of=random.dat count=$((1024*1024)) bs=1024
$ nats obj put BIGDATA random.dat
$ nats s subjects OBJ_BIGDATA
        $O.BIGDATA.M.cmFuZG9tLmRhdA==: 1
  $O.BIGDATA.C.XeqSUUd063OcyuG9SAUx7u: 8,192
```

OK so we have a subject with 1 gig of data in it... lets purge it.

```
$ nats s purge OBJ_BIGDATA --subject '$O.BIGDATA.C.XeqSUUd063OcyuG9SAUx7u'
....
State:

             Messages: 64
                Bytes: 7.9 MiB
             FirstSeq: 63 @ 2022-10-06T17:42:41 UTC
              LastSeq: 8,193 @ 2022-10-06T17:42:45 UTC
     Deleted Messages: 8,067
     Active Consumers: 0
   Number of Subjects: 2
```

So thats too many messages, checking with `subjects`:

```
$ nats s subjects OBJ_BIGDATA
        $O.BIGDATA.M.cmFuZG9tLmRhdA==: 1
  $O.BIGDATA.C.XeqSUUd063OcyuG9SAUx7u: 63
```

All the data was not removed. Calling purge again will clear it.

Pull request description:
[FIXED] JetStream: Purge with additional options may leave some messages
While going over message blocks, when some blocks were removed there was a risk that a block is ignored, leaving some messages around.

Resolves #3528

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 881d0a0b..66f8c3fe 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -4513,7 +4513,10 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	var smv StoreMsg
 
 	fs.mu.Lock()
-	for _, mb := range fs.blks {
+	// We may remove blocks as we purge, so don't range directly on fs.blks
+	// otherwise we may jump over some (see https://github.com/nats-io/nats-server/issues/3528)
+	for i := 0; i < len(fs.blks); i++ {
+		mb := fs.blks[i]
 		mb.mu.Lock()
 		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
 			mb.mu.Unlock()
@@ -4555,6 +4558,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 					mb.selectNextFirst()
 					if mb.isEmpty() {
 						fs.removeMsgBlock(mb)
+						i--
 						firstSeqNeedsUpdate = seq == fs.state.FirstSeq
 					} else if seq == fs.state.FirstSeq {
 						fs.state.FirstSeq = mb.first.seq // new one.


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index f6bd449e..d97c37c4 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3971,19 +3971,27 @@ func TestFileStorePurgeExWithSubject(t *testing.T) {
 	defer removeDir(t, storeDir)
 
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "TEST", Subjects: []string{"foo"}, Storage: FileStorage},
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 1000},
+		StreamConfig{Name: "TEST", Subjects: []string{"foo.>"}, Storage: FileStorage},
 	)
 	require_NoError(t, err)
 	defer fs.Stop()
 
-	for i := 0; i < 100; i++ {
-		_, _, err = fs.StoreMsg("foo", nil, nil)
+	payload := make([]byte, 20)
+	total := 200
+	for i := 0; i < total; i++ {
+		_, _, err = fs.StoreMsg("foo.1", nil, payload)
 		require_NoError(t, err)
 	}
+	_, _, err = fs.StoreMsg("foo.2", nil, []byte("xxxxxx"))
+	require_NoError(t, err)
 
 	// This should purge all.
-	fs.PurgeEx("foo", 1, 0)
-	require_True(t, fs.State().Msgs == 0)
+	p, err := fs.PurgeEx("foo.1", 1, 0)
+	require_NoError(t, err)
+	require_True(t, int(p) == total)
+	require_True(t, int(p) == total)
+	require_True(t, fs.State().Msgs == 1)
 }
 
 // When the N.idx file is shorter than the previous write we could fail to recover the idx properly.