Repository: nats-io/nats-server
Pull request #6352 (2025-01-09 11:47:16)

[FIXED] Filestore doesn't spam compact when block contains mostly tombstones
If the raw message block size was large enough to compact and it contained mostly tombstones, compaction would not be very effective (as `mb.compact()` does not remove tombstones). In certain unfortunate scenarios we would spam compaction upon subsequent calls to `fs.RemoveMsg`, since every call would try to compact again.

When this condition was hit, it would decrease publisher throughput, slow down consumer acks if they remove messages due to WorkQueue/Interest retention, etc.

This fix checks if there was a compaction before, and if so there are potential savings.

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 901ea576..13bfce25 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -221,6 +221,7 @@ type msgBlock struct {
 	index      uint32
 	bytes      uint64 // User visible bytes count.
 	rbytes     uint64 // Total bytes (raw) including deleted. Used for rolling to new blk.
+	cbytes     uint64 // Bytes count after last compaction. 0 if no compaction happened yet.
 	msgs       uint64 // User visible message count.
 	fss        *stree.SubjectTree[SimpleState]
 	kfn        string
@@ -4548,9 +4549,10 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 
 // Tests whether we should try to compact this block while inline removing msgs.
 // We will want rbytes to be over the minimum and have a 2x potential savings.
+// If we compacted before but rbytes didn't improve much, guard against constantly compacting.
 // Lock should be held.
 func (mb *msgBlock) shouldCompactInline() bool {
-	return mb.rbytes > compactMinimum && mb.bytes*2 < mb.rbytes
+	return mb.rbytes > compactMinimum && mb.bytes*2 < mb.rbytes && (mb.cbytes == 0 || mb.bytes*2 < mb.cbytes)
 }
 
 // Tests whether we should try to compact this block while running periodic sync.
@@ -4682,6 +4684,7 @@ func (mb *msgBlock) compactWithFloor(floor uint64) {
 	} else {
 		mb.rbytes = rbytes
 	}
+	mb.cbytes = mb.bytes
 
 	// 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 f78b69a3..2369891c 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -8483,3 +8483,82 @@ func TestFileStoreMessageTTLRecoveredSingleMessageWithoutStreamState(t *testing.
 		require_Equal(t, ss.Msgs, 0)
 	})
 }
+
+func TestFileStoreDontSpamCompactWhenMostlyTombstones(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: defaultMediumBlockSize},
+		StreamConfig{Name: "TEST", Subjects: []string{"foo"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Store a bunch of messages, ensuring we get enough blocks.
+	msg := bytes.Repeat([]byte("X"), 100)
+	totalBlksAfterStore := 6
+	expectedLseq := uint64(31_536*(totalBlksAfterStore-1) + 2)
+	for seq := uint64(1); seq <= expectedLseq; seq++ {
+		_, _, err = fs.StoreMsg("foo", nil, msg, 0)
+		require_NoError(t, err)
+	}
+
+	// Confirms we have the required amount of blocks.
+	fs.mu.RLock()
+	lenBlks := len(fs.blks)
+	lmb := fs.lmb
+	fs.mu.RUnlock()
+	require_Len(t, lenBlks, totalBlksAfterStore)
+
+	// Ensure the last block contains the last two messages.
+	lmb.mu.RLock()
+	fseq, lseq := lmb.first.seq, lmb.last.seq
+	lmb.mu.RUnlock()
+	require_Equal(t, fseq, expectedLseq-1)
+	require_Equal(t, lseq, expectedLseq)
+
+	// Remove all messages before the last block, will fill up last block with tombstones.
+	for seq := uint64(1); seq < fseq; seq++ {
+		removed, err := fs.RemoveMsg(seq)
+		require_NoError(t, err)
+		require_True(t, removed)
+	}
+
+	// The last block will be filled with so many tombstones that
+	// a new message block will be created to store the rest.
+	fs.mu.RLock()
+	lenBlks = len(fs.blks)
+	fs.mu.RUnlock()
+	require_Len(t, lenBlks, 2)
+
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+
+	fmb := fs.blks[0]
+	fmb.mu.Lock()
+	defer fmb.mu.Unlock()
+
+	// We don't call fs.RemoveMsg as that would call into compact.
+	// Instead, we mark as deleted and check compaction ourselves.
+	fmb.dmap.Insert(expectedLseq)
+	fmb.bytes /= 2
+
+	// This message block takes up ~4MB but contains only one ~100 bytes message, and the rest is all tombstones.
+	// We should allow trying to compact.
+	require_Equal(t, fmb.bytes, 133)
+	require_Equal(t, fmb.cbytes, 0)
+	require_True(t, fmb.shouldCompactInline())
+
+	// Compact will be successful, but since it doesn't clean up tombstones it will be ineffective.
+	fmb.compact()
+
+	// We should not allow compacting again as we're not removing tombstones inline.
+	// Otherwise, we would spam compaction.
+	require_False(t, fmb.shouldCompactInline())
+
+	// Just checking fmb.cbytes is tracking the previous value of fmb.bytes,
+	// so it can block compaction until enough data has been removed.
+	require_Equal(t, fmb.bytes, 133)
+	require_Equal(t, fmb.cbytes, fmb.bytes)
+
+	// Simulate having removed sufficient bytes and being allowed to compact again.
+	fmb.bytes /= 2
+	require_True(t, fmb.shouldCompactInline())
+}