perf(pruning): skip empty shards + per-shard byte counters

CompactSharded256 and CompactPrefixHex256 sweep the full 256-byte
keyspace once `compaction_interval` is crossed. CometBFT key schemas
("tx.hash", "tx.height", "block.height", event-type prefixes) live in
the lowercase-ASCII zone (~30 of 256 byte values), so ~220 of those
shard sweeps target ranges with zero keys.

Each empty Compact still spins up goleveldb's compaction-scheduling
state and (on populated databases) can trigger background level work.
On a 45 GB tx_index.db with 23k SSTs we observed 275 GB written for
21,935 deleted entries (12.5 MB/entry) — the per-prune sweep was
amplifying ~6× the DB size into write traffic and ~33 GB of OS page
cache.

Three changes:

1. shardHasKeys probe: one IteratorWithOpts(DontFillCache=true).Valid()
   per shard, microseconds. Empty shards are skipped, populated ones
   proceed unchanged.

2. CompactAndLog now reads /proc/self/io read_bytes/write_bytes around
   db.Compact and logs the delta. Pins per-shard amplification cost in
   production logs so we can attribute the spike post hoc without a
   rebuild. On non-linux the helper returns 0/0 and the log line shows
   zero-deltas — no functional change off-linux.

3. WaitTimeBetweenCompactions: 50ms -> 200ms. With skip-empty the
   active shard count drops to ~30, so 30 * 200ms = 6s extra pacing
   per sweep — well inside the 3h pruner budget — gives the kernel
   real time to drain dirty pages between bursts.

Tests:
- TestCompactSharded256_SkipsEmptyShards: seed only 't.*' keys, assert
  exactly one shard fires.
- TestCompactSharded256_EmptyDB_NoShardsCompacted: empty DB, zero shards.
- TestCompactPrefixHex256_SkipsEmptyShards: same coverage for prefix-hex.

Existing CompactIntSharded tests untouched (path is height-bounded so
sharding is dense by construction; skip-empty there would never fire).

Author: lucca30

internal/db/db_utils.go

@@ -1,12 +1,15 @@
 package db
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
 	"log"
+	"os"
 	"strconv"
+	"strings"
 	"time"
 	"unicode/utf8"
 
@@ -24,11 +27,11 @@ const (
 	// WaitTimeBetweenCompactions throttles successive shard compactions to
 	// (a) yield to the scheduler so consensus goroutines aren't starved and
 	// (b) give the kernel time to drain dirty page-cache pages before the
-	// next shard re-fills it. 2ms (the prior value) wins (a) but is far
-	// below Linux's 5s vm.dirty_writeback_centisecs default; 50ms gives
-	// writeback a real window at negligible cycle-time cost
-	// (~75s extra on the heaviest 1500-shard pruner cycle).
-	WaitTimeBetweenCompactions = 50 * time.Millisecond
+	// next shard re-fills it. 200ms gives writeback well over Linux's 5s
+	// vm.dirty_writeback_centisecs default a real window between bursts.
+	// Combined with skip-empty-shard, sweeps now spend most time on the
+	// ~30 populated shards (out of 256) so total wall-time cost is small.
+	WaitTimeBetweenCompactions = 200 * time.Millisecond
 )
 
 var (
@@ -146,13 +149,18 @@ func CompactIntSharded(
 //
 // The final shard ends at "BH:fg" so that every key starting with "BH:ff"
 // compares < "BH:fg" (since 'g' is the next ASCII char after 'f').
+//
+// Shards that contain no keys are detected by a one-row Iterator probe and
+// skipped — on real workloads this prunes ~220 of 256 shards because key
+// schemas concentrate in the lowercase-ASCII zone.
 func CompactPrefixHex256(db dbm.DB, prefix string, label string) error {
 	startAll := time.Now()
 
 	if prefix == "" {
 		return fmt.Errorf("prefix must be non-empty")
 	}
 
+	var compacted, skipped int
 	for b := 0; b <= 0xFF; b++ {
 		start := []byte(fmt.Sprintf("%s%02x", prefix, b))
 
@@ -173,20 +181,36 @@ func CompactPrefixHex256(db dbm.DB, prefix string, label string) error {
 			shardLabel = fmt.Sprintf("%s %s ff-fg", label, prefix)
 		}
 
+		hasAny, err := shardHasKeys(db, start, end)
+		if err != nil {
+			return fmt.Errorf("compaction %s probe failed: %w", shardLabel, err)
+		}
+		if !hasAny {
+			skipped++
+			continue
+		}
+		compacted++
 		if err := compactAndLog(db, start, end, shardLabel); err != nil {
 			return err
 		}
 	}
 
-	log.Printf("compaction %s prefix %q ALL 256 SHARDS DONE in %s", label, prefix, time.Since(startAll))
+	log.Printf("compaction %s prefix %q DONE in %s (compacted=%d skipped_empty=%d)",
+		label, prefix, time.Since(startAll), compacted, skipped)
 	return nil
 }
 
 // CompactSharded256 compacts the DB into 256 ranges:
 // [0x00,0x01), [0x01,0x02), …, [0xFE,0xFF), [0xFF,∞)
+//
+// Shards that contain no keys are detected by a one-row Iterator probe and
+// skipped. CometBFT key schemas (e.g. "tx.hash", "block.height", event-type
+// prefixes) concentrate keys in the lowercase-ASCII zone, leaving ~220 of
+// the 256 shards empty on real workloads.
 func CompactSharded256(db dbm.DB, label string) error {
 	startAll := time.Now()
 
+	var compacted, skipped int
 	for b := 0; b < 256; b++ {
 		start := []byte{byte(b)}
 		var end []byte
@@ -203,32 +227,92 @@ func CompactSharded256(db dbm.DB, label string) error {
 			shardLabel = fmt.Sprintf("%s shard %02x-%02x", label, b, b+1)
 		}
 
+		hasAny, err := shardHasKeys(db, start, end)
+		if err != nil {
+			return fmt.Errorf("compaction %s probe failed: %w", shardLabel, err)
+		}
+		if !hasAny {
+			skipped++
+			continue
+		}
+		compacted++
 		if err := compactAndLog(db, start, end, shardLabel); err != nil {
 			return err
 		}
 	}
 
-	log.Printf("compaction %s ALL SHARDS DONE in %s", label, time.Since(startAll))
+	log.Printf("compaction %s DONE in %s (compacted=%d skipped_empty=%d)",
+		label, time.Since(startAll), compacted, skipped)
 	return nil
 }
 
-// CompactAndLog compacts [start, limit) and logs the range and duration.
+// shardHasKeys reports whether any key exists in [start, end). Cost is one
+// SeekGE — microseconds — versus a full CompactRange which on a populated DB
+// rewrites overlapping SSTs. Uses DontFillCache so the probe doesn't pollute
+// goleveldb's block cache.
+func shardHasKeys(db dbm.DB, start, end []byte) (bool, error) {
+	it, err := dbm.IteratorWithOpts(db, start, end, &dbm.ReadOptions{DontFillCache: true})
+	if err != nil {
+		return false, err
+	}
+	defer it.Close()
+	return it.Valid(), nil
+}
+
+// CompactAndLog compacts [start, limit) and logs the range, duration, and
+// the process-level read/write byte deltas observed during the call. The
+// byte deltas come from /proc/self/io and capture *all* I/O issued by the
+// process during the compact (including background goroutines), so they're
+// upper bounds — but they pin per-shard amplification cost in production.
 func CompactAndLog(db dbm.DB, start, limit []byte, label string) error {
 	time.Sleep(WaitTimeBetweenCompactions)
 
 	rng := fmt.Sprintf("[%s, %s)", prettyKey(start), prettyKey(limit))
 
+	rb0, wb0 := procIOBytes()
 	t0 := time.Now()
 	err := db.Compact(start, limit)
 	elapsed := time.Since(t0)
+	rb1, wb1 := procIOBytes()
 
 	if err != nil {
 		log.Printf("compaction %s range %s FAILED after %s: %v", label, rng, elapsed, err)
 		return err
 	}
+	log.Printf("compaction %s range %s done in %s dRead=%dB dWrite=%dB",
+		label, rng, elapsed, rb1-rb0, wb1-wb0)
 	return nil
 }
 
+// procIOBytes reads /proc/self/io and returns (read_bytes, write_bytes).
+// On non-linux or read failure, returns (0, 0); the resulting delta of 0
+// is logged unobtrusively rather than failing the compaction.
+func procIOBytes() (uint64, uint64) {
+	f, err := os.Open("/proc/self/io")
+	if err != nil {
+		return 0, 0
+	}
+	defer f.Close()
+	var rb, wb uint64
+	sc := bufio.NewScanner(f)
+	for sc.Scan() {
+		line := sc.Text()
+		switch {
+		case strings.HasPrefix(line, "read_bytes:"):
+			v, err := strconv.ParseUint(strings.TrimSpace(strings.TrimPrefix(line, "read_bytes:")), 10, 64)
+			if err == nil {
+				rb = v
+			}
+		case strings.HasPrefix(line, "write_bytes:"):
+			v, err := strconv.ParseUint(strings.TrimSpace(strings.TrimPrefix(line, "write_bytes:")), 10, 64)
+			if err == nil {
+				wb = v
+			}
+		}
+	}
+	return rb, wb
+}
+
 // prettyKey renders a key as quoted ASCII if possible, otherwise as hex.
 // nil renders as ∞ (end-of-keyspace).
 func prettyKey(b []byte) string {

internal/db/db_utils_test.go

@@ -163,3 +163,58 @@ func TestCompactIntSharded_ResumeFromStoredMeta_NoGaps(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, end2-1, last2)
 }
+
+// TestCompactSharded256_SkipsEmptyShards seeds keys only under the lowercase
+// 't' prefix and asserts that CompactSharded256 invokes compactAndLog only
+// for that single shard out of 256.
+func TestCompactSharded256_SkipsEmptyShards(t *testing.T) {
+	var calls [][2][]byte
+	restore := swapCompactAndLog(func(db dbm.DB, start, end []byte, lbl string) error {
+		calls = append(calls, [2][]byte{append([]byte(nil), start...), append([]byte(nil), end...)})
+		return nil
+	})
+	defer restore()
+
+	memdb := dbm.NewMemDB()
+	require.NoError(t, memdb.Set([]byte("tx.height/100"), []byte{1}))
+	require.NoError(t, memdb.Set([]byte("tx.hash/abcd"), []byte{1}))
+
+	require.NoError(t, CompactSharded256(memdb, "test"))
+
+	require.Len(t, calls, 1, "only the 't' shard (0x74) should be compacted")
+	require.Equal(t, byte('t'), calls[0][0][0], "compacted shard must start at 0x74 ('t')")
+}
+
+// TestCompactSharded256_EmptyDB_NoShardsCompacted asserts that on an empty
+// DB no shard is compacted at all.
+func TestCompactSharded256_EmptyDB_NoShardsCompacted(t *testing.T) {
+	var calls int
+	restore := swapCompactAndLog(func(db dbm.DB, start, end []byte, lbl string) error {
+		calls++
+		return nil
+	})
+	defer restore()
+
+	require.NoError(t, CompactSharded256(dbm.NewMemDB(), "empty"))
+	require.Zero(t, calls, "empty DB must yield zero compactions")
+}
+
+// TestCompactPrefixHex256_SkipsEmptyShards seeds two keys under "BH:" prefix
+// (heights 1 and 1000) so they fall in distinct hex shards (BH:31 and BH:31
+// — both '1' first byte). One shard should fire.
+func TestCompactPrefixHex256_SkipsEmptyShards(t *testing.T) {
+	var calls int
+	restore := swapCompactAndLog(func(db dbm.DB, start, end []byte, lbl string) error {
+		calls++
+		return nil
+	})
+	defer restore()
+
+	memdb := dbm.NewMemDB()
+	require.NoError(t, memdb.Set([]byte("BH:31"), []byte{1}))
+	require.NoError(t, memdb.Set([]byte("BH:31000"), []byte{1}))
+
+	require.NoError(t, CompactPrefixHex256(memdb, "BH:", "test"))
+
+	require.Equal(t, 1, calls, "only BH:31-32 shard should fire")
+}