Optimize compaction under heavy KV use

Signed-off-by: Derek Collison <derek@nats.io>

server/filestore.go

@@ -1809,7 +1809,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 		}
 	} else {
 		// Check if we are empty first, as long as not the last message block.
-		if isLast := mb != fs.lmb; isLast && mb.msgs == 0 {
+		if notLast := mb != fs.lmb; notLast && mb.msgs == 0 {
 			fs.removeMsgBlock(mb)
 			firstSeqNeedsUpdate = seq == fs.state.FirstSeq
 		} else {
@@ -1819,9 +1819,9 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 				mb.dmap = make(map[uint64]struct{})
 			}
 			mb.dmap[seq] = struct{}{}
-			// Check if <50% utilization and minimum size met.
-			if mb.rbytes > compactMinimum && mb.rbytes>>1 > mb.bytes {
-				// FIXME(dlc) - Might want this out of band.
+			// Check if <25% utilization and minimum size met.
+
+			if notLast && mb.rbytes > compactMinimum && mb.rbytes>>2 > mb.bytes {
 				mb.compact()
 			}
 		}

server/norace_test.go

@@ -3720,3 +3720,29 @@ func TestNoRaceJetStreamPullConsumerAPIOutUnlock(t *testing.T) {
 	}
 	nc.Flush()
 }
+
+// Reports of high cpu on compaction for a KV store.
+func TestNoRaceJetStreamKeyValueCompaction(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
+		Bucket:   "COMPACT",
+		Replicas: 3,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	value := strings.Repeat("A", 128*1024)
+	for i := 0; i < 5_000; i++ {
+		key := fmt.Sprintf("K-%d", rand.Intn(256)+1)
+		if _, err := kv.PutString(key, value); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+}