Repository: nats-io/nats-server
Pull request #2080 (2021-04-07 21:21:52)

Fix bug that could cause panic and corruption on filestore compact.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 3d076cd5..96811d44 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2975,12 +2975,13 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 
 	smb.mu.Lock()
 	for mseq := smb.first.seq; mseq < seq; mseq++ {
-		if _, rl, _, err := smb.slotInfo(int(mseq - smb.cache.fseq)); err != nil {
-			smb.bytes -= uint64(rl)
+		if sm, _ := smb.cacheLookupWithLock(mseq); sm != nil && smb.msgs > 0 {
+			smb.bytes -= fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+			smb.msgs--
+			purged++
 		}
-		smb.msgs--
-		purged++
 	}
+
 	// Update first entry.
 	sm, _ := smb.cacheLookupWithLock(seq)
 	if sm != nil {
diff --git a/server/stream.go b/server/stream.go
index 3296a56d..1089a294 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1292,9 +1292,7 @@ func (mset *stream) processInboundMirrorMsg(m *inMsg) bool {
 
 func (mset *stream) setMirrorErr(err *ApiError) {
 	mset.mu.Lock()
-	if mset.mirror != nil {
-		mset.mirror.err = err
-	}
+	mset.mirror.err = err
 	mset.mu.Unlock()
 }
 
@@ -1423,22 +1421,16 @@ func (mset *stream) setupMirrorConsumer() error {
 			if ccr.Error != nil || ccr.ConsumerInfo == nil {
 				mset.cancelMirrorConsumer()
 			} else {
+				// Capture consumer name.
 				mset.mu.Lock()
-				// Mirror config has been removed.
-				if mset.mirror == nil {
-					mset.mu.Unlock()
-					mset.cancelMirrorConsumer()
-					return
+				if mset.mirror != nil {
+					mset.mirror.cname = ccr.ConsumerInfo.Name
 				}
 
-				// Capture consumer name.
-				mset.mirror.cname = ccr.ConsumerInfo.Name
-				msgs := mset.mirror.msgs
-
 				// Process inbound mirror messages from the wire.
 				sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
 					hdr, msg := c.msgParts(append(rmsg[:0:0], rmsg...)) // Need to copy.
-					mset.queueInbound(msgs, subject, reply, hdr, msg)
+					mset.queueInbound(mset.mirror.msgs, subject, reply, hdr, msg)
 				})
 				if err != nil {
 					mset.mirror.err = jsError(err)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index b2bf6310..4c384908 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -845,6 +845,47 @@ func TestFileStoreCompactLastPlusOne(t *testing.T) {
 	}
 }
 
+func TestFileStoreCompactMsgCountBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	for i := 0; i < 10; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+	if state := fs.State(); state.Msgs != 10 {
+		t.Fatalf("Expected 10 msgs, got %d", state.Msgs)
+	}
+	// Now delete 2,3,4.
+	fs.EraseMsg(2)
+	fs.EraseMsg(3)
+	fs.EraseMsg(4)
+
+	// Also delete 7,8, and 9.
+	fs.RemoveMsg(7)
+	fs.RemoveMsg(8)
+	fs.RemoveMsg(9)
+
+	n, err := fs.Compact(6)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// 1 & 5
+	if n != 2 {
+		t.Fatalf("Expected to have deleted 2 msgs, got %d", n)
+	}
+	if state := fs.State(); state.Msgs != 2 {
+		t.Fatalf("Expected to have 2 remaining, got %d", state.Msgs)
+	}
+}
+
 func TestFileStoreCompactPerf(t *testing.T) {
 	t.SkipNow()