State Cache Consolidation (PR erigontech#1 of the perf stack) (erigontech#21380)

This is **PR erigontech#1 of a 3-PR perf stack**. It consolidates state caching
across all modes — sync, tip-tracking, integration and testing — so the
state cache is either **on and completely trustworthy**, or **off to
measure** — never off because it unexpectedly breaks things.

The caches (the account/storage `StateCache` and the commitment
`BranchCache`) are an **internal implementation detail of
`SharedDomains`**. No external entity accesses or mutates them directly:
callers drive state through `Flush` / `Commit` / `GetLatest` /
`DomainPut`, and the full cache lifecycle (population, invalidation,
commit-gating) is owned inside `SharedDomains`.

## What this PR contains

### `BranchCache` — single aggregator-scope commitment cache
- Aggregator-lifetime cache: pinned root slot + bounded LRU tail, behind
the `sd.mem` chain so unwinds and fork-validations see consistent state.
- Wired into the trie read + encoder write paths.
- Tx-precise unwind invalidation: entries are stamped with their per-key
write `txNum`; `sd.Unwind` evicts everything above the unwind watermark
(`BranchCache.UnwindTo`).

### One switch for all caches
The `BranchCache` is a *type of* state cache, so it rides the existing
`USE_STATE_CACHE` toggle rather than getting its own env. One operator
switch turns **all** caching off — the relevant operation when bisecting
a state-root mismatch, where an operator shouldn't have to reason about
the interaction of several independent caches. (This is a deliberate
deviation from the review's "add a separate `BranchCache` kill-switch"
suggestion — flagged for confirmation.)

### The BranchCache reflects only committed state — by construction
The BranchCache can never hold a value a failed commit rolled back:
`SharedDomains.Commit` flushes the in-memory batch into the tx, commits,
and **only then** applies the flushed commitment branches to the cache
(the flush is implicit in committing; a failed commit applies nothing).
Plain `Flush` — callers that own their own commit, e.g. offline tools —
never touches the cache; read-through populates from committed files.
The caches are an internal detail of `SharedDomains`; nothing else
writes to them.

> **StateCache no-poisoning is erigontech#21386, not this PR.** Unlike the
BranchCache, the account/storage StateCache is an *in-flight,
cross-transaction* cache — it holds prior txs' not-yet-committed writes
within a batch, and later txs read them from it. So it can't be made
commit-safe by simply invalidating on write (that breaks cross-tx reads
in serial exec). Its no-poisoning is the txNum/epoch rework in erigontech#21386;
this PR keeps its existing ValidateAndPrepare/unwind invalidation.
### BUG erigontech#21138 — parallel-exec from-0 wrong trie root
`ResetExec` wipes the commitment DB table; the aggregator's in-memory
`BranchCache` could still reference the just-deleted trie nodes, so a
from-0 re-exec served stale entries when computing block 0's commitment
→ wrong root, dropping genesis-allocated balances no later block touched
(mainnet block 46147, `0xA1E4380A3B1f749673E270229993eE55F35663b4`).
Fix: `ResetExec` clears the aggregator's `BranchCache`.
`TestFromZero_GenesisAllocPreservedAfterResetReExec` passes on current
`main`; the test's value here is keeping *this PR's* cache safe across
reset, not fixing a live `main` bug.

## Follow-ups (the rest of the stack)
- **erigontech#21386 (PR erigontech#2 of the stack) — StateCache LRU + Mode rework:**
consistency + no performance drop-off at a 1 GB cache for long-running
nodes; re-adds the warm StateCache repopulation deferred above, under
its `txNum`/`epoch` model.
- **Pinning** (the stack's third step) — **no PR yet**; in progress and
under test on branch
[`mh/branch-cache-trunk-pin`](https://github.com/erigontech/erigon/tree/mh/branch-cache-trunk-pin),
to be **re-benchmarked before merge**.
- **erigontech#21739** — interface-unification follow-up: collapse the duck-typed
`GetLatest` variants into a single metered, `txNum`-returning
`GetLatest`.
## Testing
Behaves identically across parallel and serial exec — confirmed in CI
across both exec modes. Unit coverage for the BranchCache (tiers,
tx-precise `UnwindTo`, commit-gated population) plus the
engine/exec-module FCU commit paths.

Given today's changes (the commit-gating of both caches), we will do
another A/B performance run before merging.

---------

Co-authored-by: Mark Holt <erigon@dev-bm-e3-ethmainnet-n4.erigon.io>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

cmd/evm/staterunner.go

@@ -35,6 +35,7 @@ import (
 	"github.com/erigontech/erigon/common/log/v3"
 	"github.com/erigontech/erigon/db/datadir"
 	"github.com/erigontech/erigon/db/kv/temporal/temporaltest"
+	"github.com/erigontech/erigon/db/state/execctx"
 	"github.com/erigontech/erigon/execution/tests/testutil"
 	"github.com/erigontech/erigon/execution/tracing/tracers/logger"
 	"github.com/erigontech/erigon/execution/vm"
@@ -224,7 +225,15 @@ func runStateTest(ctx *cli.Context, cfg vm.Config, fname string) ([]testResult,
 				}
 				defer tx.Rollback()
 
-				statedb, root, err := test.Run(nil, tx, st, cfg, dirs)
+				// Per-subtest SD: closed without Flush so its writes never enter the branch cache.
+				sd, err := execctx.NewSharedDomains(context.Background(), tx, log.New())
+				if err != nil {
+					result.Pass, result.Error = false, err.Error()
+					return
+				}
+				defer sd.Close()
+
+				statedb, root, err := test.Run(nil, sd, tx, st, cfg, dirs)
 				if err != nil {
 					result.Pass, result.Error = false, err.Error()
 				}
@@ -238,8 +247,9 @@ func runStateTest(ctx *cli.Context, cfg vm.Config, fname string) ([]testResult,
 					}
 				}
 				if bench {
+					// Reuse the subtest's tx+sd: a second concurrent rwtx on the same env would deadlock.
 					_, stats, _ := timedExec(true, func() ([]byte, uint64, error) {
-						_, _, gasUsed, _ := test.RunNoVerify(nil, tx, st, cfg, dirs)
+						_, _, gasUsed, _ := test.RunNoVerify(nil, sd, tx, st, cfg, dirs)
 						return nil, gasUsed, nil
 					})
 					result.Stats = &stats

cmd/integration/commands/stages.go

@@ -758,13 +758,10 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 					return err
 				}
 			}
-			if err := doms.Flush(ctx, tx); err != nil {
-				return err
-			}
-			doms.ClearRam(true)
-			if err := tx.Commit(); err != nil {
+			if err := doms.Commit(ctx, tx); err != nil {
 				return err
 			}
+			doms.Close()
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
@@ -780,15 +777,20 @@ func stageExec(db kv.TemporalRwDB, ctx context.Context, logger log.Logger) error
 			if err := tx.Commit(); err != nil {
 				return err
 			}
+			if bn+1 >= block { // last block committed; nothing more to run
+				break
+			}
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
+			// Fresh SD for the next block: a committed SD is never reused.
+			if doms, err = execctx.NewSharedDomains(ctx, tx, logger); err != nil {
+				return err
+			}
+			doms.SetInMemHistoryReads(false)
 		}
-		if err := doms.Flush(ctx, tx); err != nil {
-			return err
-		}
-		doms.ClearRam(true)
-		return tx.Commit()
+		doms.Close()
+		return nil
 	}
 	agg := (db.(dbstate.HasAgg).Agg()).(*dbstate.Aggregator)
 	blockSnapBuildSema := semaphore.NewWeighted(int64(runtime.NumCPU()))

cmd/integration/commands/state_stages.go

@@ -169,7 +169,7 @@ func syncBySmallSteps(db kv.TemporalRwDB, builderConfig buildercfg.BuilderConfig
 	if err != nil {
 		return err
 	}
-	defer sd.Close()
+	defer func() { sd.Close() }() // closes whichever SD is current after the commit loop swaps it
 	sd.SetInMemHistoryReads(false)
 
 	var batchSize datasize.ByteSize
@@ -283,18 +283,20 @@ func syncBySmallSteps(db kv.TemporalRwDB, builderConfig buildercfg.BuilderConfig
 				return err
 			}
 
-			if err = sd.Flush(ctx, tx); err != nil {
-				return err
-			}
-			sd.ClearRam(true)
-			if err = tx.Commit(); err != nil {
+			if err = sd.Commit(ctx, tx); err != nil {
 				return err
 			}
+			sd.Close()
 
 			if tx, err = db.BeginTemporalRw(ctx); err != nil {
 				return err
 			}
 			defer tx.Rollback()
+			// Fresh SD: a committed SD is never reused.
+			if sd, err = execctx.NewSharedDomains(ctx, tx, logger1); err != nil {
+				return err
+			}
+			sd.SetInMemHistoryReads(false)
 		}
 
 		//receiptsInDB := rawdb.ReadReceiptsByNumber(tx, progress(tx, stages.Execution)+1)

common/maphash/maphash.go

@@ -163,3 +163,26 @@ func (l *LRU[V]) SetByHash(hash uint64, value V) {
 func (l *LRU[V]) ContainsByHash(hash uint64) bool {
 	return l.cache.Contains(hash)
 }
+
+// DeleteByHash removes the entry under the pre-computed hash. Use
+// alongside Range when the byte-key is unknown.
+func (l *LRU[V]) DeleteByHash(hash uint64) {
+	l.cache.Remove(hash)
+}
+
+// Range iterates over every (hash, value) pair without affecting LRU
+// recency (uses Peek under the hood). Iteration order is unspecified.
+// Return false from fn to stop early.
+//
+// The original byte-key is not recoverable; use the hash as identity (same key → same hash, collisions aside).
+func (l *LRU[V]) Range(fn func(hash uint64, v V) bool) {
+	for _, h := range l.cache.Keys() {
+		v, ok := l.cache.Peek(h)
+		if !ok {
+			continue
+		}
+		if !fn(h, v) {
+			return
+		}
+	}
+}

db/integrity/commitment_integrity.go

@@ -955,7 +955,6 @@ func checkCommitmentHistValBucket(ctx context.Context, tx kv.TemporalTx, br serv
 // checkCommitmentHistAtBlkWithIdx checks commitment for blockNum using the pre-built
 // per-domain key index from ChangedKeysPerBlockIdx.
 func checkCommitmentHistAtBlkWithIdx(ctx context.Context, tx kv.TemporalTx, sd *execctx.SharedDomains, br services.FullBlockReader, blockNum uint64, idx *ChangedKeysPerBlockIdx, lvl log.Lvl, logger log.Logger) error {
-	sd.ClearRam(true)
 	logger.Log(lvl, "checking commitment hist at block", "blockNum", blockNum)
 	header, err := br.HeaderByNumber(ctx, tx, blockNum)
 	if err != nil {
@@ -1131,19 +1130,22 @@ func CheckCommitmentHistAtBlkRange(ctx context.Context, sc SamplerCfg, db kv.Tem
 				return err
 			}
 			defer tx.Rollback()
-			sd, err := execctx.NewSharedDomains(wCtx, tx, logger, execctx.WithoutDeferredBranchUpdates())
-			if err != nil {
-				return err
-			}
-			defer sd.Close()
 			idx, err := NewChangedKeysPerBlockIdx(wCtx, tx, br, windowStart, windowEnd, logger)
 			if err != nil {
 				return fmt.Errorf("CheckCommitmentHistAtBlkRange: build index window=[%d,%d): %w", windowStart, windowEnd, err)
 			}
 			// Each goroutine needs its own Sampler — the RNG is not goroutine-safe.
 			sampler := sc.NewWindowSampler(windowStart)
 			for blockNum := range sampler.BlockNums(windowStart, windowEnd) {
-				if err := checkCommitmentHistAtBlkWithIdx(wCtx, tx, sd, br, blockNum, idx, log.LvlTrace, logger); err != nil {
+				// Fresh SharedDomains per block: an SD is committed-or-closed,
+				// never reset in place.
+				sd, err := execctx.NewSharedDomains(wCtx, tx, logger, execctx.WithoutDeferredBranchUpdates())
+				if err != nil {
+					return err
+				}
+				err = checkCommitmentHistAtBlkWithIdx(wCtx, tx, sd, br, blockNum, idx, log.LvlTrace, logger)
+				sd.Close()
+				if err != nil {
 					return fmt.Errorf("checkCommitmentHistAtBlk: %d, %w", blockNum, err)
 				}
 				checked.Add(1)

db/state/aggregator.go

@@ -342,6 +342,13 @@ func (a *Aggregator) ConfigureDomains() error {
 	}
 	a.configured = true
 
+	// Attach the aggregator-lifetime BranchCache to the commitment domain; gated by USE_STATE_CACHE, nil = disabled.
+	if dbg.UseStateCache {
+		if cd := a.d[kv.CommitmentDomain]; cd != nil && cd.branchCache == nil {
+			cd.branchCache = commitment.NewBranchCache(commitment.DefaultBranchCacheTailCapacity)
+		}
+	}
+
 	if a.disableFsync {
 		for _, d := range a.d {
 			if d != nil {
@@ -2353,6 +2360,14 @@ func (a *Aggregator) BeginFilesRo() *AggregatorRoTx {
 	return ac
 }
 
+// BranchCache attached to the commitment domain (implements commitment.BranchCacheProvider).
+func (at *AggregatorRoTx) BranchCache() *commitment.BranchCache {
+	if at.d[kv.CommitmentDomain] == nil {
+		return nil
+	}
+	return at.d[kv.CommitmentDomain].d.branchCache
+}
+
 func (at *AggregatorRoTx) Dirs() datadir.Dirs                  { return at.a.dirs }
 func (at *AggregatorRoTx) standaloneIIs() []*InvertedIndexRoTx { return at.iis[:at.iisCount] }
 
@@ -2401,31 +2416,50 @@ func (at *AggregatorRoTx) MeteredGetLatest(domain kv.Domain, k []byte, tx kv.Tx,
 	return at.getLatest(domain, k, tx, maxStep, metrics, start)
 }
 
+// MeteredGetLatestWithTxN returns the high-water txN alongside (value,
+// step) for tagging BranchCache entries so UnwindTo can evict by
+// watermark. Non-CommitmentDomain reads return txN=0.
+func (at *AggregatorRoTx) MeteredGetLatestWithTxN(domain kv.Domain, k []byte, tx kv.Tx, maxStep kv.Step, metrics *changeset.DomainMetrics, start time.Time) (v []byte, step kv.Step, txN uint64, ok bool, err error) {
+	return at.getLatestWithTxN(domain, k, tx, maxStep, metrics, start)
+}
+
 func (at *AggregatorRoTx) getLatest(domain kv.Domain, k []byte, tx kv.Tx, maxStep kv.Step, metrics *changeset.DomainMetrics, start time.Time) (v []byte, step kv.Step, ok bool, err error) {
+	v, step, _, ok, err = at.getLatestWithTxN(domain, k, tx, maxStep, metrics, start)
+	return v, step, ok, err
+}
+
+func (at *AggregatorRoTx) getLatestWithTxN(domain kv.Domain, k []byte, tx kv.Tx, maxStep kv.Step, metrics *changeset.DomainMetrics, start time.Time) (v []byte, step kv.Step, txN uint64, ok bool, err error) {
 	if domain != kv.CommitmentDomain {
-		return at.d[domain].getLatest(k, tx, maxStep, metrics, start)
+		v, step, ok, err = at.d[domain].getLatest(k, tx, maxStep, metrics, start)
+		return v, step, 0, ok, err
 	}
 
 	v, step, ok, err = at.d[domain].getLatestFromDb(k, tx)
 	if err != nil {
-		return nil, kv.Step(0), false, err
+		return nil, kv.Step(0), 0, false, err
 	}
 	if ok && step <= maxStep {
 		if metrics != nil && dbg.KVReadLevelledMetrics {
 			metrics.UpdateDbReads(domain, start)
 		}
-		return v, step, true, nil
+		// DB-sourced: tag with the step's high-water; the exact write
+		// txN isn't recoverable from the step-keyed record.
+		return v, step, lastTxNumOfStep(step, at.StepSize()), true, nil
 	}
 
 	v, found, fileStartTxNum, fileEndTxNum, err := at.d[domain].getLatestFromFiles(k, 0)
 	if !found {
-		return nil, kv.Step(0), false, err
+		return nil, kv.Step(0), 0, false, err
 	}
 	if metrics != nil && dbg.KVReadLevelledMetrics {
-		metrics.UpdateFileReads(domain, start)
+		// UpdateFileReadsUnique tracks total + distinct prefixes; the
+		// ratio is the read amplification factor (hot prefixes re-read).
+		metrics.UpdateFileReadsUnique(domain, k, start)
 	}
 	v, err = at.replaceShortenedKeysInBranch(k, commitment.BranchData(v), fileStartTxNum, fileEndTxNum)
-	return v, kv.Step(fileEndTxNum / at.StepSize()), found, err
+	// File-sourced: tag with fileEndTxNum; snapshots are immutable and
+	// unwind can't cross them, so this is always <= any legal watermark.
+	return v, kv.Step(fileEndTxNum / at.StepSize()), fileEndTxNum, found, err
 }
 
 func (at *AggregatorRoTx) DebugGetLatestFromDB(domain kv.Domain, key []byte, tx kv.Tx) ([]byte, kv.Step, bool, error) {