Repository: nats-io/nats-server
Pull request #5831 (2024-08-26 20:29:22)

[IMPROVED] When a block can not be compressed, mark it as such to avoid syncBlocks trying continously
We had reports of disk spikes and memory pops queued off of syncBlocks, and looks like it was coming from blocks that were no longer compressible, but would load each time and go through the process.

Resolves: #5595 

Signed-off-by: Derek Collison <derek@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index ee285ef0..5fe928c4 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -239,6 +239,7 @@ type msgBlock struct {
 	noTrack    bool
 	needSync   bool
 	syncAlways bool
+	noCompact  bool
 	closed     bool
 
 	// Used to mock write failures.
@@ -4072,6 +4073,9 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 		mb.bytes = 0
 	}
 
+	// Allow us to check compaction again.
+	mb.noCompact = false
+
 	// Mark as dirty for stream state.
 	fs.dirty++
 
@@ -4188,7 +4192,7 @@ func (mb *msgBlock) shouldCompactInline() bool {
 // Ignores 2MB minimum.
 // Lock should be held.
 func (mb *msgBlock) shouldCompactSync() bool {
-	return mb.bytes*2 < mb.rbytes
+	return mb.bytes*2 < mb.rbytes && !mb.noCompact
 }
 
 // This will compact and rewrite this block. This should only be called when we know we want to rewrite this block.
@@ -4297,7 +4301,12 @@ func (mb *msgBlock) compact() {
 	mb.needSync = true
 
 	// Capture the updated rbytes.
-	mb.rbytes = uint64(len(nbuf))
+	if rbytes := uint64(len(nbuf)); rbytes == mb.rbytes {
+		// No change, so set our noCompact bool here to avoid attempting to continually compress in syncBlocks.
+		mb.noCompact = true
+	} else {
+		mb.rbytes = rbytes
+	}
 
 	// Remove any seqs from the beginning of the blk.
 	for seq, nfseq := fseq, atomic.LoadUint64(&mb.first.seq); seq < nfseq; seq++ {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index dc073c2a..b6fb16e2 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -7480,6 +7480,83 @@ func TestFileStoreCheckSkipFirstBlockNotLoadOldBlocks(t *testing.T) {
 	require_Equal(t, loaded, 1)
 }
 
+func TestFileStoreSyncCompressOnlyIfDirty(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 256, SyncInterval: 250 * time.Millisecond},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("hello")
+
+	// 6 msgs per block.
+	// Fill 2 blocks.
+	for i := 0; i < 12; i++ {
+		fs.StoreMsg("foo.BB", nil, msg)
+	}
+	// Create third block with just one message in it.
+	fs.StoreMsg("foo.BB", nil, msg)
+
+	// Should have created 3 blocks.
+	require_Equal(t, fs.numMsgBlocks(), 3)
+
+	// Now delete a bunch that will will fill up 3 block with tombstones.
+	for _, seq := range []uint64{2, 3, 4, 5, 8, 9, 10, 11} {
+		_, err = fs.RemoveMsg(seq)
+		require_NoError(t, err)
+	}
+	// Now make sure we add 4th block so syncBlocks will try to compress.
+	for i := 0; i < 6; i++ {
+		fs.StoreMsg("foo.BB", nil, msg)
+	}
+	require_Equal(t, fs.numMsgBlocks(), 4)
+
+	// All should have compact set.
+	fs.mu.Lock()
+	// Only check first 3 blocks.
+	for i := 0; i < 3; i++ {
+		mb := fs.blks[i]
+		mb.mu.Lock()
+		shouldCompact := mb.shouldCompactSync()
+		mb.mu.Unlock()
+		if !shouldCompact {
+			fs.mu.Unlock()
+			t.Fatalf("Expected should compact to be true for %d, got false", mb.getIndex())
+		}
+	}
+	fs.mu.Unlock()
+
+	// Let sync run.
+	time.Sleep(300 * time.Millisecond)
+
+	// We want to make sure the last block, which is filled with tombstones and is not compactable, returns false now.
+	fs.mu.Lock()
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		shouldCompact := mb.shouldCompactSync()
+		mb.mu.Unlock()
+		if shouldCompact {
+			fs.mu.Unlock()
+			t.Fatalf("Expected should compact to be false for %d, got true", mb.getIndex())
+		}
+	}
+	fs.mu.Unlock()
+
+	// Now remove some from block 3 and verify that compact is not suppressed.
+	_, err = fs.RemoveMsg(13)
+	require_NoError(t, err)
+
+	fs.mu.Lock()
+	mb := fs.blks[2] // block 3.
+	mb.mu.Lock()
+	noCompact := mb.noCompact
+	mb.mu.Unlock()
+	fs.mu.Unlock()
+	// Verify that since we deleted a message we should be considered for compaction again in syncBlocks().
+	require_False(t, noCompact)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////