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

Issue:
Nats Server is ignoring Jetstreams limits in case of 0 byte files and grows uncontrolled
## Defect

We discovered that some Nats brokers are running over their Jetstream limits in our edge computing scenarios. We cannot completely avoid, that these devices are shut down properly, so sometimes there can occur 0 byte files. 

Somehow the Nats broker seems not to be able to clear the directory, when there is a 0 byte file inside for the .blk files as shown in this screenshot:
![CleanShot 2022-12-21 at 13 10 23@2x](https://user-images.githubusercontent.com/12080409/208902456-6b749058-488f-46a3-9046-5842e4a070ee.png)

We have a limit of 1GB for our Jetstream and the max is 5 GB. The metrics show a steady rise of data, without data being removed from the Jetstream.
<img width="945" alt="CleanShot 2022-12-21 at 13 12 16@2x" src="https://user-images.githubusercontent.com/12080409/208902653-1beb135c-6dcc-4f00-ad5b-6781696f2bb6.png">

Nats does nothing strange until the 5GB limit is reached:
<img width="1192" alt="CleanShot 2022-12-21 at 13 16 57@2x" src="https://user-images.githubusercontent.com/12080409/208903377-2467550c-a31c-44d0-9c60-7cf54ec76389.png">


#### Versions of `nats-server` and affected client libraries used:

#### OS/Container environment:
containerd - nats server in version: 2.9.10-alpine

```
nats.conf: |-
    pid_file: "/var/run/nats/nats.pid"
    http: 8222

    jetstream {
      store_dir: "/data/jetstream/store"
      max_file_store: 5G
    }
```

Stream:

```yaml
---
apiVersion: jetstream.nats.io/v1beta2
kind: Stream
metadata:
  name: telemetry
  namespace: cecu-broker
spec:
  maxBytes: 1073741824
  name: telemetry
  storage: file
  subjects:
  - telemetry.*
  - telemetry.command.*
```
#### Steps or code to reproduce the issue:
Create a 0 byte file in the messages directory and wait until Jetstream is running over.

#### Expected result:
Jetstream keeps the 1 GB limit for the stream as configured.

#### Actual result:
File system is running full until Nats runs into max limit (in our case 5GB) and fails to transfer any data.


Pull request description:
Fix for #3734
When a msg blk was not written correctly, but the idx file was, max bytes for a stream would no longer be honored since the deletion of any messages in that empty block were not being handled properly.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #3734 


/cc @nats-io/core


Patch:
diff --git a/server/filestore.go b/server/filestore.go
index baf39642..7e75d5a2 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -736,6 +736,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 			file.ReadAt(lchk[:], fi.Size()-checksumSize)
 		}
 	}
+
 	file.Close()
 
 	// Read our index file. Use this as source of truth if possible.
@@ -755,7 +756,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 
 	// If we get data loss rebuilding the message block state record that with the fs itself.
 	if ld, _ := mb.rebuildState(); ld != nil {
-		fs.rebuildStateLocked(ld)
+		fs.addLostData(ld)
 	}
 	if mb.msgs > 0 && !mb.noTrack && fs.psim != nil {
 		fs.populateGlobalPerSubjectInfo(mb)
@@ -787,7 +788,10 @@ func (fs *fileStore) rebuildState(ld *LostStreamData) {
 }
 
 // Lock should be held.
-func (fs *fileStore) rebuildStateLocked(ld *LostStreamData) {
+func (fs *fileStore) addLostData(ld *LostStreamData) {
+	if ld == nil {
+		return
+	}
 	if fs.ld != nil {
 		fs.ld.Msgs = append(fs.ld.Msgs, ld.Msgs...)
 		msgs := fs.ld.Msgs
@@ -796,6 +800,12 @@ func (fs *fileStore) rebuildStateLocked(ld *LostStreamData) {
 	} else {
 		fs.ld = ld
 	}
+}
+
+// Lock should be held.
+func (fs *fileStore) rebuildStateLocked(ld *LostStreamData) {
+	fs.addLostData(ld)
+
 	fs.state.Msgs, fs.state.Bytes = 0, 0
 	fs.state.FirstSeq, fs.state.LastSeq = 0, 0
 
@@ -923,16 +933,31 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 	startLastSeq := mb.last.seq
 
+	buf, err := mb.loadBlock(nil)
+	if err != nil || len(buf) == 0 {
+		var ld *LostStreamData
+		// No data to rebuild from here.
+		if mb.msgs > 0 {
+			// We need to declare lost data here.
+			ld = &LostStreamData{Msgs: make([]uint64, 0, mb.msgs), Bytes: mb.bytes}
+			for seq := mb.first.seq; seq <= mb.last.seq; seq++ {
+				if _, ok := mb.dmap[seq]; !ok {
+					ld.Msgs = append(ld.Msgs, seq)
+				}
+			}
+			// Clear invalid state. We will let this blk be added in here.
+			mb.msgs, mb.bytes, mb.rbytes, mb.fss = 0, 0, 0, nil
+			mb.dmap = nil
+			mb.first.seq = mb.last.seq + 1
+		}
+		return ld, err
+	}
+
 	// Clear state we need to rebuild.
 	mb.msgs, mb.bytes, mb.rbytes, mb.fss = 0, 0, 0, nil
 	mb.last.seq, mb.last.ts = 0, 0
 	firstNeedsSet := true
 
-	buf, err := mb.loadBlock(nil)
-	if err != nil {
-		return nil, err
-	}
-
 	// Check if we need to decrypt.
 	if mb.bek != nil && len(buf) > 0 {
 		// Recreate to reset counter.
@@ -1151,6 +1176,24 @@ func (fs *fileStore) recoverMsgs() error {
 		_, err = fs.newMsgBlockForWrite()
 	}
 
+	// Check if we encountered any lost data.
+	if fs.ld != nil {
+		var emptyBlks []*msgBlock
+		for _, mb := range fs.blks {
+			if mb.msgs == 0 && mb.rbytes == 0 {
+				if mb == fs.lmb {
+					mb.first.seq, mb.first.ts = mb.last.seq+1, 0
+					mb.closeAndKeepIndex()
+				} else {
+					emptyBlks = append(emptyBlks, mb)
+				}
+			}
+		}
+		for _, mb := range emptyBlks {
+			fs.removeMsgBlock(mb)
+		}
+	}
+
 	if err != nil {
 		return err
 	}
@@ -4252,11 +4295,20 @@ func (fs *fileStore) State() StreamState {
 	state.NumSubjects = fs.numSubjects()
 	state.Deleted = nil // make sure.
 
+	cur := fs.state.FirstSeq
+
 	for _, mb := range fs.blks {
 		mb.mu.Lock()
 		fseq := mb.first.seq
+		// Account for messages missing from the head.
+		if fseq > cur {
+			for seq := cur; seq < fseq; seq++ {
+				state.Deleted = append(state.Deleted, seq)
+			}
+		}
+		cur = mb.last.seq + 1 // Expected next first.
 		for seq := range mb.dmap {
-			if seq <= fseq {
+			if seq < fseq {
 				delete(mb.dmap, seq)
 			} else {
 				state.Deleted = append(state.Deleted, seq)
@@ -4497,7 +4549,7 @@ func (mb *msgBlock) genDeleteMap() []byte {
 	fseq, n := uint64(mb.first.seq), 0
 	for seq := range mb.dmap {
 		// This is for lazy cleanup as the first sequence moves up.
-		if seq <= fseq {
+		if seq < fseq {
 			delete(mb.dmap, seq)
 		} else {
 			n += binary.PutUvarint(buf[n:], seq-fseq)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 0dd49c52..8502d0f2 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4858,3 +4858,139 @@ func TestFileStoreCompactAllWithDanglingLMB(t *testing.T) {
 	_, _, err = fs.StoreMsg(subj, nil, msg)
 	require_NoError(t, err)
 }
+
+func TestFileStoreStateWithBlkFirstDeleted(t *testing.T) {
+	storeDir := t.TempDir()
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 4096},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	toStore := 500
+	for i := 0; i < toStore; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	// Delete some messages from the beginning of an interior block.
+	fs.mu.RLock()
+	require_True(t, len(fs.blks) > 2)
+	fseq := fs.blks[1].first.seq
+	fs.mu.RUnlock()
+
+	// Now start from first seq of second blk and delete 10 msgs
+	for seq := fseq; seq < fseq+10; seq++ {
+		removed, err := fs.RemoveMsg(seq)
+		require_NoError(t, err)
+		require_True(t, removed)
+	}
+
+	// This bug was in normal detailed state. But check fast state too.
+	var fstate StreamState
+	fs.FastState(&fstate)
+	require_True(t, fstate.NumDeleted == 10)
+	state := fs.State()
+	require_True(t, state.NumDeleted == 10)
+}
+
+func TestFileStoreMsgBlkFailOnKernelFaultLostDataReporting(t *testing.T) {
+	storeDir := t.TempDir()
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 4096},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	toStore := 500
+	for i := 0; i < toStore; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	// We want to make sure all of the scenarios report lost data properly.
+	// Will run 3 scenarios, 1st block, last block, interior block.
+
+	// First block
+	fs.mu.RLock()
+	require_True(t, fs.numMsgBlocks() > 0)
+	mfn := fs.blks[0].mfn
+	fs.mu.RUnlock()
+
+	fs.Stop()
+
+	err = os.WriteFile(mfn, nil, defaultFilePerms)
+	require_NoError(t, err)
+
+	// Restart.
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 4096},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	state := fs.State()
+	require_True(t, state.FirstSeq == 94)
+	require_True(t, state.Lost != nil)
+	require_True(t, len(state.Lost.Msgs) == 93)
+
+	// Last block
+	fs.mu.RLock()
+	require_True(t, fs.numMsgBlocks() > 0)
+	require_True(t, fs.lmb != nil)
+	mfn = fs.lmb.mfn
+	fs.mu.RUnlock()
+
+	fs.Stop()
+
+	err = os.WriteFile(mfn, nil, defaultFilePerms)
+	require_NoError(t, err)
+
+	// Restart.
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 4096},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	state = fs.State()
+	require_True(t, state.FirstSeq == 94)
+	require_True(t, state.LastSeq == 500)   // Make sure we do not lose last seq.
+	require_True(t, state.NumDeleted == 35) // These are interiors
+	require_True(t, state.Lost != nil)
+	require_True(t, len(state.Lost.Msgs) == 35)
+
+	// Interior block.
+	fs.mu.RLock()
+	require_True(t, fs.numMsgBlocks() > 3)
+	mfn = fs.blks[len(fs.blks)-3].mfn
+	fs.mu.RUnlock()
+
+	fs.Stop()
+
+	err = os.WriteFile(mfn, nil, defaultFilePerms)
+	require_NoError(t, err)
+
+	// Restart.
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 4096},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	state = fs.State()
+	require_True(t, state.FirstSeq == 94)
+	require_True(t, state.LastSeq == 500) // Make sure we do not lose last seq.
+	require_True(t, state.NumDeleted == 128)
+	require_True(t, state.Lost != nil)
+	require_True(t, len(state.Lost.Msgs) == 93)
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 84b13255..65381f1d 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -19215,3 +19215,82 @@ func TestJetStreamMetaDataFailOnKernelFault(t *testing.T) {
 	require_NoError(t, err)
 	require_True(t, si.State.Msgs == 10)
 }
+
+// https://github.com/nats-io/nats-server/issues/3734
+func TestJetStreamMsgBlkFailOnKernelFault(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		MaxBytes: 10 * 1024 * 1024, // 10MB
+	})
+	require_NoError(t, err)
+
+	msgSize := 1024 * 1024 // 1MB
+	msg := make([]byte, msgSize)
+	rand.Read(msg)
+
+	for i := 0; i < 20; i++ {
+		_, err := js.Publish("foo", msg)
+		require_NoError(t, err)
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_True(t, si.State.Bytes < uint64(si.Config.MaxBytes))
+
+	// Now emulate a kernel fault that fails to write the last blk properly.
+	mset, err := s.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+
+	mset.mu.RLock()
+	fs := mset.store.(*fileStore)
+	fs.mu.RLock()
+	require_True(t, len(fs.blks) > 2)
+	// Here we do not grab the last one, which we handle correctly. We grab an interior one near the end.
+	lmbf := fs.blks[len(fs.blks)-2].mfn
+	fs.mu.RUnlock()
+	mset.mu.RUnlock()
+
+	sd := s.JetStreamConfig().StoreDir
+	s.Shutdown()
+
+	// Zero out the last block.
+	err = os.WriteFile(lmbf, nil, defaultFilePerms)
+	require_NoError(t, err)
+
+	s = RunJetStreamServerOnPort(-1, sd)
+	defer s.Shutdown()
+
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_True(t, si.State.NumDeleted == 3)
+
+	// Test detailed version as well.
+	si, err = js.StreamInfo("TEST", &nats.StreamInfoRequest{DeletedDetails: true})
+	require_NoError(t, err)
+	require_True(t, si.State.NumDeleted == 3)
+	if !reflect.DeepEqual(si.State.Deleted, []uint64{16, 17, 18}) {
+		t.Fatalf("Expected deleted of %+v, got %+v", []uint64{16, 17, 18}, si.State.Deleted)
+	}
+
+	for i := 0; i < 20; i++ {
+		_, err := js.Publish("foo", msg)
+		require_NoError(t, err)
+	}
+
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	if si.State.Bytes > uint64(si.Config.MaxBytes) {
+		t.Fatalf("MaxBytes not enforced with empty interior msg blk, max %v, bytes %v",
+			friendlyBytes(si.Config.MaxBytes), friendlyBytes(int64(si.State.Bytes)))
+	}
+}