perf(db): cache Pos() result to avoid repeated disk reads

Cache the replication position computed from the latest L0 LTX file
using an atomic.Pointer. The cache is invalidated whenever L0 files
are created, deleted, or the L0 directory is cleared (reset, retention,
sync errors, replica fetch). On sync success, the cache is updated
directly from the encoder without a disk read.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: benbjohnson

db.go

@@ -16,6 +16,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -101,6 +102,10 @@ type DB struct {
 		m map[int]*ltx.FileInfo
 	}
 
+	// Cached position from the latest L0 LTX file.
+	// nil means cache is invalid; non-nil is the cached position.
+	posCache atomic.Pointer[ltx.Pos]
+
 	fileInfo os.FileInfo // db info cached during init
 	dirInfo  os.FileInfo // parent dir info cached during init
 
@@ -300,6 +305,8 @@ func (db *DB) ResetLocalState(ctx context.Context) error {
 	db.maxLTXFileInfos.m = make(map[int]*ltx.FileInfo)
 	db.maxLTXFileInfos.Unlock()
 
+	db.invalidatePosCache()
+
 	db.Logger.Info("local state reset complete, next sync will create fresh snapshot")
 	return nil
 }
@@ -330,6 +337,9 @@ func (db *DB) deleteLocalLTXFile(level int, minTXID, maxTXID ltx.TXID) error {
 	if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
 		return err
 	}
+	if level == 0 {
+		db.invalidatePosCache()
+	}
 	return nil
 }
 
@@ -364,7 +374,12 @@ func (db *DB) DirInfo() os.FileInfo {
 }
 
 // Pos returns the current replication position of the database.
+// The result is cached and invalidated when L0 LTX files change.
 func (db *DB) Pos() (ltx.Pos, error) {
+	if p := db.posCache.Load(); p != nil {
+		return *p, nil
+	}
+
 	minTXID, maxTXID, err := db.MaxLTX()
 	if err != nil {
 		return ltx.Pos{}, err
@@ -382,7 +397,18 @@ func (db *DB) Pos() (ltx.Pos, error) {
 	if err := dec.Verify(); err != nil {
 		return ltx.Pos{}, fmt.Errorf("ltx verification failed: %w", err)
 	}
-	return dec.PostApplyPos(), nil
+
+	pos := dec.PostApplyPos()
+	db.posCache.Store(&pos)
+
+	return pos, nil
+}
+
+// invalidatePosCache clears the cached position so the next call to Pos()
+// recomputes it from disk. Call this when L0 LTX files are deleted or
+// when the L0 directory is cleared.
+func (db *DB) invalidatePosCache() {
+	db.posCache.Store(nil)
 }
 
 // Notify returns a channel that closes when the shadow WAL changes.
@@ -1195,6 +1221,7 @@ func (db *DB) checkDatabaseBehindReplica(ctx context.Context) error {
 	if err := os.RemoveAll(l0Dir); err != nil && !os.IsNotExist(err) {
 		return fmt.Errorf("remove L0 directory: %w", err)
 	}
+	db.invalidatePosCache()
 	if err := internal.MkdirAll(l0Dir, db.dirInfo); err != nil {
 		return fmt.Errorf("recreate L0 directory: %w", err)
 	}
@@ -1235,6 +1262,7 @@ func (db *DB) checkDatabaseBehindReplica(ctx context.Context) error {
 	if err := os.Rename(tmpPath, localPath); err != nil {
 		return fmt.Errorf("rename L0 file: %w", err)
 	}
+	db.invalidatePosCache()
 
 	db.Logger.Info("fetched latest L0 file from replica",
 		"min_txid", minTXID,
@@ -1632,6 +1660,7 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) (sync
 		db.maxLTXFileInfos.Lock()
 		delete(db.maxLTXFileInfos.m, 0) // clear cache if in unknown state
 		db.maxLTXFileInfos.Unlock()
+		db.invalidatePosCache()
 		return false, fmt.Errorf("rename ltx file: %w", err)
 	}
 
@@ -1646,6 +1675,10 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) (sync
 	}
 	db.maxLTXFileInfos.Unlock()
 
+	// Update cached position from the encoder.
+	encPos := enc.PostApplyPos()
+	db.posCache.Store(&encPos)
+
 	// Track the logical end of WAL content for checkpoint decisions.
 	// This is the WALOffset + WALSize from the LTX we just created.
 	// Using this instead of file size prevents issue #997 where stale
@@ -2177,6 +2210,9 @@ func (db *DB) EnforceL0RetentionByTime(ctx context.Context) error {
 			db.Logger.Error("failed to remove local l0 file", "path", localPath, "error", err)
 		}
 	}
+	if len(deleted) > 0 {
+		db.invalidatePosCache()
+	}
 
 	db.Logger.Info("l0 retention enforced", "deleted_count", len(deleted), "max_l1_txid", maxL1TXID)
 

db_internal_test.go

@@ -656,6 +656,9 @@ func TestDB_Verify_WALOffsetAtHeader(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// Invalidate cached position since we wrote an L0 file directly.
+	db.invalidatePosCache()
+
 	// Now call verify - before the fix, this would fail with:
 	// "prev WAL offset is less than the header size: -4088"
 	info, err := db.verify(context.Background())
@@ -776,6 +779,9 @@ func TestDB_Verify_WALOffsetAtHeader_SaltMismatch(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// Invalidate cached position since we wrote an L0 file directly.
+	db.invalidatePosCache()
+
 	// Call verify - should succeed but indicate snapshotting due to salt mismatch
 	info, err := db.verify(context.Background())
 	if err != nil {