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

Issue:
panic: runtime error: makeslice: cap out of range
## Defect

Make sure that these boxes are checked before submitting your issue -- thank you!

 - [x] Included `nats-server -DV` output
 - [x] Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

#### Versions of `nats-server` and affected client libraries used:

```
[54] 2022/05/10 03:53:10.909500 [INF] Starting nats-server
[54] 2022/05/10 03:53:10.909527 [INF]   Version:  2.8.0
[54] 2022/05/10 03:53:10.909536 [INF]   Git:      [90721ee]
[54] 2022/05/10 03:53:10.909544 [DBG]   Go build: go1.17.9
[54] 2022/05/10 03:53:10.909553 [INF]   Name:     NCFLZXZ6MUOIW6H4BRRNMQMOHFYAENP6OEU3XUYYKU3WHHE4ZRDPFRBS
[54] 2022/05/10 03:53:10.909572 [INF]   ID:       NCFLZXZ6MUOIW6H4BRRNMQMOHFYAENP6OEU3XUYYKU3WHHE4ZRDPFRBS
[54] 2022/05/10 03:53:10.909595 [DBG] Created system account: "$SYS"
```

Messages were sent and consumed using nats.py 2.1.0.

#### OS/Container environment:
GKE

#### Steps or code to reproduce the issue:
- Install NATS with Helm, enable jetstream filestorage
-- Stored on a volume with a storage class type=pd-ssd and WaitForFirstConsumer
- Send tens of thousands of messages without consuming them.
- Attempt to scale down the node that nats is running on for an upgrade and recover on a new node.

#### Expected result:
Not to panic, to recover gracefully, or at least to provide a more instructive error message.

#### Actual result:
```
[3426] 2022/05/10 03:10:51.017008 [INF] Starting nats-server
[3426] 2022/05/10 03:10:51.017092 [INF]   Version:  2.8.0
[3426] 2022/05/10 03:10:51.017096 [INF]   Git:      [90721ee]
[3426] 2022/05/10 03:10:51.017100 [INF]   Name:     queue-nats-0
[3426] 2022/05/10 03:10:51.017110 [INF]   Node:     4pcPsiNg
[3426] 2022/05/10 03:10:51.017116 [INF]   ID:       NC6TCJ2D5KDDFEJ37JNOLTZIBEDERC6T3MMGN6ZMVNHDGFW6IM3ETFIJ
[3426] 2022/05/10 03:10:51.017146 [INF] Using configuration file: /etc/nats-config/nats.conf
[3426] 2022/05/10 03:10:51.018272 [INF] Starting http monitor on 0.0.0.0:8222
[3426] 2022/05/10 03:10:51.018355 [INF] Starting JetStream
[3426] 2022/05/10 03:10:51.018534 [INF]     _ ___ _____ ___ _____ ___ ___   _   __  __
[3426] 2022/05/10 03:10:51.018538 [INF]  _ | | __|_   _/ __|_   _| _ \ __| /_\ |  \/  |
[3426] 2022/05/10 03:10:51.018540 [INF] | || | _|  | | \__ \ | | |   / _| / _ \| |\/| |
[3426] 2022/05/10 03:10:51.018541 [INF]  \__/|___| |_| |___/ |_| |_|_\___/_/ \_\_|  |_|
[3426] 2022/05/10 03:10:51.018543 [INF] 
[3426] 2022/05/10 03:10:51.018545 [INF]          https://docs.nats.io/jetstream
[3426] 2022/05/10 03:10:51.018546 [INF] 
[3426] 2022/05/10 03:10:51.018548 [INF] ---------------- JETSTREAM ----------------
[3426] 2022/05/10 03:10:51.018556 [INF]   Max Memory:      1.00 GB
[3426] 2022/05/10 03:10:51.018559 [INF]   Max Storage:     100.00 GB
[3426] 2022/05/10 03:10:51.018561 [INF]   Store Directory: "/data/jetstream"
[3426] 2022/05/10 03:10:51.018562 [INF] -------------------------------------------
[3426] 2022/05/10 03:10:51.321612 [INF]   Restored 13,833 messages for stream '$G > INGEST'
[3426] 2022/05/10 03:10:51.421122 [INF] Server is ready
panic: runtime error: makeslice: cap out of range

goroutine 1 [running]:
github.com/nats-io/nats-server/v2/server.(*msgBlock).indexCacheBuf(0xc000292a80, {0xc00078c000, 0x186a22, 0x200000})
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:3165 +0x195
github.com/nats-io/nats-server/v2/server.(*msgBlock).loadMsgsWithLock(0xc000292a80)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:3466 +0x414
github.com/nats-io/nats-server/v2/server.(*msgBlock).generatePerSubjectInfo(0xc000292a80)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:4663 +0xf0
github.com/nats-io/nats-server/v2/server.(*msgBlock).readPerSubjectInfo(0xc000292a80)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:4716 +0x3b4
github.com/nats-io/nats-server/v2/server.(*fileStore).recoverMsgBlock(0xc000290280, {0xb1de38, 0xc000709790}, 0x3d)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:673 +0xff0
github.com/nats-io/nats-server/v2/server.(*fileStore).recoverMsgs(0xc000290280)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:958 +0x8bd
github.com/nats-io/nats-server/v2/server.newFileStoreWithCreated({{0xc000220990, 0x25}, _, _, _, _}, {{0xc000759020, 0xa}, {0x0, 0x0}, ...}, ...)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/filestore.go:315 +0x76c
github.com/nats-io/nats-server/v2/server.(*stream).setupStore(0xc00011c2c0, 0xc000160988)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/stream.go:2760 +0x2d5
github.com/nats-io/nats-server/v2/server.(*Account).addStreamWithAssignment(0xc000118900, 0xc0000f6108, 0x0, 0x0)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/stream.go:417 +0xee5
github.com/nats-io/nats-server/v2/server.(*Account).addStream(0xc000759030, 0xc000758ff0)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/stream.go:276 +0x1d
github.com/nats-io/nats-server/v2/server.(*Account).EnableJetStream(0xc000118900, 0xc00001f8c0)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/jetstream.go:1193 +0x2ecd
github.com/nats-io/nats-server/v2/server.(*Server).configJetStream(0xc0001ac000, 0xc000118900)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/jetstream.go:645 +0x9c
github.com/nats-io/nats-server/v2/server.(*Server).enableJetStreamAccounts(0xc0000d0160)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/jetstream.go:575 +0xb4
github.com/nats-io/nats-server/v2/server.(*Server).enableJetStream(0xc0001ac000, {0x40000000, 0x1900000000, {0xc0000c23e0, 0xf}, {0x0, 0x0}})
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/jetstream.go:392 +0x919
github.com/nats-io/nats-server/v2/server.(*Server).EnableJetStream(0xc0001ac000, 0xc000161e38)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/jetstream.go:204 +0x40c
github.com/nats-io/nats-server/v2/server.(*Server).Start(0xc0001ac000)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/server.go:1746 +0xf10
github.com/nats-io/nats-server/v2/server.Run(...)
        /home/travis/gopath/src/github.com/nats-io/nats-server/server/service.go:22
main.main()
        /home/travis/gopath/src/github.com/nats-io/nats-server/main.go:118 +0x2fa
```


Pull request description:
General improvements to accounting for the filestore.
 This in response to tracking issue #3114.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #3114

Hopefully ;)

/cc @nats-io/core


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 0bcaf2d7..808a792c 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -856,7 +856,10 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		// This is an old erased message, or a new one that we can track.
 		if seq == 0 || seq&ebit != 0 || seq < mb.first.seq {
 			seq = seq &^ ebit
-			addToDmap(seq)
+			// Only add to dmap if past recorded first seq and non-zero.
+			if seq != 0 && seq >= mb.first.seq {
+				addToDmap(seq)
+			}
 			index += rl
 			mb.last.seq = seq
 			mb.last.ts = ts
@@ -1094,10 +1097,14 @@ func (fs *fileStore) expireMsgsOnRecover() {
 						mb.dmap = nil
 					}
 				}
+				// Keep this update just in case since we are removing dmap entries.
+				mb.first.seq = seq
 				continue
 			}
 			// Break on other errors.
 			if err != nil || sm == nil {
+				// Keep this update just in case since we could have removed dmap entries.
+				mb.first.seq = seq
 				break
 			}
 
@@ -1112,11 +1119,13 @@ func (fs *fileStore) expireMsgsOnRecover() {
 			}
 
 			// Delete the message here.
-			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			mb.bytes -= sz
-			bytes += sz
-			mb.msgs--
-			purged++
+			if mb.msgs > 0 {
+				sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+				mb.bytes -= sz
+				bytes += sz
+				mb.msgs--
+				purged++
+			}
 			// Update fss
 			fs.removePerSubject(sm.subj)
 			mb.removeSeqPerSubject(sm.subj, seq, &smv)
@@ -1998,7 +2007,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 
 	mb.mu.Lock()
 
-	// See if the sequence numbers is still relevant.
+	// See if the sequence number is still relevant.
 	if seq < mb.first.seq {
 		mb.mu.Unlock()
 		fsUnlock()
@@ -2170,7 +2179,8 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 // writing new messages. We will silently bail on any issues with the underlying block and let someone else detect.
 // Write lock needs to be held.
 func (mb *msgBlock) compact() {
-	if mb.cacheNotLoaded() {
+	wasLoaded := mb.cacheAlreadyLoaded()
+	if !wasLoaded {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return
 		}
@@ -2252,11 +2262,12 @@ func (mb *msgBlock) compact() {
 
 	// We will write to a new file and mv/rename it in case of failure.
 	mfn := filepath.Join(filepath.Join(mb.fs.fcfg.StoreDir, msgDir), fmt.Sprintf(newScan, mb.index))
-	defer os.Remove(mfn)
 	if err := ioutil.WriteFile(mfn, nbuf, defaultFilePerms); err != nil {
+		os.Remove(mfn)
 		return
 	}
 	if err := os.Rename(mfn, mb.mfn); err != nil {
+		os.Remove(mfn)
 		return
 	}
 
@@ -2265,7 +2276,11 @@ func (mb *msgBlock) compact() {
 	mb.removeIndexFileLocked()
 	mb.deleteDmap()
 	mb.rebuildStateLocked()
-	mb.loadMsgsWithLock()
+
+	// If we entered with the msgs loaded make sure to reload them.
+	if wasLoaded {
+		mb.loadMsgsWithLock()
+	}
 }
 
 // Nil out our dmap.
@@ -2485,23 +2500,34 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 	var purged, bytes uint64
 
 	mb.mu.Lock()
+
 	checkDmap := len(mb.dmap) > 0
+	var smv StoreMsg
+
 	for seq := mb.last.seq; seq > sm.seq; seq-- {
 		if checkDmap {
 			if _, ok := mb.dmap[seq]; ok {
 				// Delete and skip to next.
 				delete(mb.dmap, seq)
+				if len(mb.dmap) == 0 {
+					mb.dmap = nil
+					checkDmap = false
+				}
 				continue
 			}
 		}
 		// We should have a valid msg to calculate removal stats.
-		_, rl, _, err := mb.slotInfo(int(seq - mb.cache.fseq))
-		if err != nil {
-			mb.mu.Unlock()
-			return 0, 0, err
+		if m, err := mb.cacheLookup(seq, &smv); err == nil {
+			if mb.msgs > 0 {
+				rl := fileStoreMsgSize(m.subj, m.hdr, m.msg)
+				mb.msgs--
+				mb.bytes -= rl
+				mb.rbytes -= rl
+				// For return accounting.
+				purged++
+				bytes += uint64(rl)
+			}
 		}
-		purged++
-		bytes += uint64(rl)
 	}
 
 	// Truncate our msgs and close file.
@@ -2517,11 +2543,6 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 		return 0, 0, fmt.Errorf("failed to truncate msg block %d, file not open", mb.index)
 	}
 
-	// Do local mb stat updates.
-	mb.msgs -= purged
-	mb.bytes -= bytes
-	mb.rbytes -= bytes
-
 	// Update our last msg.
 	mb.last.seq = sm.seq
 	mb.last.ts = sm.ts
@@ -3603,6 +3624,12 @@ func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 	if err != nil || fsm == nil {
 		return nil, err
 	}
+
+	// Deleted messages that are decoded return a 0 for seqeunce.
+	if fsm.seq == 0 {
+		return nil, errDeletedMsg
+	}
+
 	if seq != fsm.seq {
 		recycleMsgBlockBuf(mb.cache.buf)
 		mb.cache.buf = nil
@@ -4040,10 +4067,16 @@ func (mb *msgBlock) readIndexInfo() error {
 
 	// Check if this is a short write index file.
 	if bi < 0 || bi+checksumSize > len(buf) {
-		defer os.Remove(mb.ifn)
+		os.Remove(mb.ifn)
 		return fmt.Errorf("short index file")
 	}
 
+	// Check for consistency if accounting. If something is off bail and we will rebuild.
+	if mb.msgs != (mb.last.seq-mb.first.seq+1)-dmapLen {
+		os.Remove(mb.ifn)
+		return fmt.Errorf("accounting inconsistent")
+	}
+
 	// Checksum
 	copy(mb.lchk[0:], buf[bi:bi+checksumSize])
 	bi += checksumSize
@@ -4151,6 +4184,10 @@ func compareFn(subject string) func(string, string) bool {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
+	if sequence > 1 && keep > 0 {
+		return 0, ErrPurgeArgMismatch
+	}
+
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return fs.Purge()
@@ -4198,7 +4235,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 			mb.loadMsgsWithLock()
 			shouldExpire = true
 		}
-		if sequence > 0 && sequence <= l {
+		if sequence > 1 && sequence <= l {
 			l = sequence - 1
 		}
 
@@ -4207,10 +4244,13 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 				rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				// Do fast in place remove.
 				// Stats
-				fs.state.Msgs--
-				fs.state.Bytes -= rl
-				mb.msgs--
-				mb.bytes -= rl
+				if mb.msgs > 0 {
+					fs.state.Msgs--
+					fs.state.Bytes -= rl
+					mb.msgs--
+					mb.bytes -= rl
+					purged++
+				}
 				// FSS updates.
 				fs.removePerSubject(sm.subj)
 				mb.removeSeqPerSubject(sm.subj, seq, &smv)
@@ -4231,7 +4271,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 					}
 					mb.dmap[seq] = struct{}{}
 				}
-				purged++
+
 				if maxp > 0 && purged >= maxp {
 					break
 				}
@@ -4391,10 +4431,12 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			}
 		} else if sm != nil {
 			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			smb.bytes -= sz
-			bytes += sz
-			smb.msgs--
-			purged++
+			if smb.msgs > 0 {
+				smb.bytes -= sz
+				bytes += sz
+				smb.msgs--
+				purged++
+			}
 			// Update fss
 			fs.removePerSubject(sm.subj)
 			smb.removeSeqPerSubject(sm.subj, mseq, &smv)
diff --git a/server/memstore.go b/server/memstore.go
index d06cad4f..65493301 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -428,6 +428,10 @@ func (ms *memStore) expireMsgs() {
 // PurgeEx will remove messages based on subject filters, sequence and number of messages to keep.
 // Will return the number of purged messages.
 func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
+	if sequence > 1 && keep > 0 {
+		return 0, ErrPurgeArgMismatch
+	}
+
 	if subject == _EMPTY_ || subject == fwcs {
 		if keep == 0 && (sequence == 0 || sequence == 1) {
 			return ms.Purge()
@@ -455,7 +459,7 @@ func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint6
 			ss.Msgs -= keep
 		}
 		last := ss.Last
-		if sequence > 0 {
+		if sequence > 1 {
 			last = sequence - 1
 		}
 		ms.mu.Lock()
diff --git a/server/store.go b/server/store.go
index 10fc10a7..e6158a0c 100644
--- a/server/store.go
+++ b/server/store.go
@@ -61,6 +61,8 @@ var (
 	ErrInvalidSequence = errors.New("invalid sequence")
 	// ErrSequenceMismatch is returned when storing a raw message and the expected sequence is wrong.
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
+	// ErrPurgeArgMismatch is returned when PurgeEx is called with sequence > 1 and keep > 0.
+	ErrPurgeArgMismatch = errors.New("sequence > 1 && keep > 0 not allowed")
 )
 
 // StoreMsg is the stored message format for messages that are retained by the Store layer.


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index d03eaa68..9249199c 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -3825,3 +3825,85 @@ func TestFileStoreRememberLastMsgTime(t *testing.T) {
 	require_True(t, lt == fs.State().LastTime)
 	require_True(t, seq == 6)
 }
+
+func (fs *fileStore) getFirstBlock() *msgBlock {
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+	if len(fs.blks) == 0 {
+		return nil
+	}
+	return fs.blks[0]
+}
+
+func TestFileStoreRebuildStateDmapAccountingBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 1024 * 1024},
+		StreamConfig{Name: "TEST", Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 0; i < 100; i++ {
+		_, _, err = fs.StoreMsg("foo", nil, nil)
+		require_NoError(t, err)
+	}
+	// Delete 2-40.
+	for i := 2; i <= 40; i++ {
+		_, err := fs.RemoveMsg(uint64(i))
+		require_NoError(t, err)
+	}
+
+	mb := fs.getFirstBlock()
+	require_True(t, mb != nil)
+
+	check := func() {
+		t.Helper()
+		mb.mu.RLock()
+		defer mb.mu.RUnlock()
+		dmapLen := uint64(len(mb.dmap))
+		if mb.msgs != (mb.last.seq-mb.first.seq+1)-dmapLen {
+			t.Fatalf("Consistency check failed: %d != %d -> last %d first %d len(dmap) %d",
+				mb.msgs, (mb.last.seq-mb.first.seq+1)-dmapLen, mb.last.seq, mb.first.seq, dmapLen)
+		}
+	}
+
+	check()
+
+	mb.mu.Lock()
+	mb.compact()
+	mb.mu.Unlock()
+
+	// Now delete first.
+	_, err = fs.RemoveMsg(1)
+	require_NoError(t, err)
+
+	mb.mu.Lock()
+	_, err = mb.rebuildStateLocked()
+	require_NoError(t, err)
+	mb.mu.Unlock()
+
+	check()
+}
+
+func TestFileStorePurgeExWithSubject(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer removeDir(t, storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "TEST", Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 0; i < 100; i++ {
+		_, _, err = fs.StoreMsg("foo", nil, nil)
+		require_NoError(t, err)
+	}
+
+	// This should purge all.
+	fs.PurgeEx("foo", 1, 0)
+	require_True(t, fs.State().Msgs == 0)
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 413d39a7..92fa00c5 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -391,3 +391,17 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 	}
 }
+
+func TestMemStorePurgeExWithSubject(t *testing.T) {
+	ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
+	require_NoError(t, err)
+
+	for i := 0; i < 100; i++ {
+		_, _, err = ms.StoreMsg("foo", nil, nil)
+		require_NoError(t, err)
+	}
+
+	// This should purge all.
+	ms.PurgeEx("foo", 1, 0)
+	require_True(t, ms.State().Msgs == 0)
+}