diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go
index ed76fc775aa..10a0627e8a7 100644
--- a/db/state/execctx/domain_shared.go
+++ b/db/state/execctx/domain_shared.go
@@ -1082,18 +1082,6 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun
return nil
}
-// DetachBranchCache makes this SharedDomains ignore the aggregator-scope
-// BranchCache: commitment branch reads go straight to sd.mem/overlay/MDBX and
-// no read populates the shared cache. Used for fork-validation SDs, which read
-// transient fork state — sharing the canonical BranchCache let them read stale
-// committed branches (wrong trie root → INVALID payload) and pollute the cache
-// with fork-transient branches. Only sd.branchCache (the read/populate path,
-// domain_shared GetLatest) is consulted, so nil-ing it fully detaches; the
-// canonical SDs keep their warm cache.
-func (sd *SharedDomains) DetachBranchCache() {
- sd.branchCache = nil
-}
-
// TemporalDomain satisfaction. Collects no read metrics — see
// temporalGetter.GetLatest for why there is no process-wide accumulator.
func (sd *SharedDomains) GetLatest(domain kv.Domain, tx kv.TemporalTx, k []byte) (v []byte, step kv.Step, err error) {
@@ -1109,6 +1097,16 @@ func (sd *SharedDomains) GetLatestContext(ctx context.Context, domain kv.Domain,
return sd.getLatestMetered(domain, tx, k, kvmetrics.MetricsFromContext(ctx))
}
+// servableUnderBound gates a cached entry against an in-flight unwind's
+// per-key maxStep: a hit above the bound would diverge from the bounded read
+// the cache-disabled path takes (the epoch floor usually drops such entries
+// already; the gate keeps the two paths identical regardless). Callers convert
+// their unit first — the StateCache stamps txNums (divide by step size), the
+// BranchCache stores step indices (no divide).
+func servableUnderBound(cStep, maxStep kv.Step) bool {
+ return cStep <= maxStep
+}
+
// getLatestMetered is the read implementation. wm is the caller's lock-free
// per-task/per-worker metrics accumulator (nil disables metrics for the call).
// No global metrics lock is taken on this hot path — accumulators are combined
@@ -1171,16 +1169,9 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
// that haven't been flushed to DB yet. Early return keeps correctness AND performance.
if sd.stateCache != nil {
v, cTxNum, ok := sd.stateCache.GetWithTxNum(domain, k)
+ // The cache stamps txNums — divide to get the step the entry reflects.
cStep := kv.Step(cTxNum / sd.StepSize())
- // Respect maxStep, mirroring the BranchCache gate below. sd.mem /
- // sd.parent.mem lowered maxStep above when an in-flight unwind re-bound
- // this key to an earlier step (the per-key unwindChangeset signal). A
- // cached entry from a higher step would diverge from the (maxStep-bounded)
- // DB read the cache-disabled path takes, so treat it as a miss and fall
- // through; the Put below refreshes it. For direct domains the (txNum,epoch)
- // floor in Get usually already drops such entries — this keeps the two
- // read paths identical regardless.
- if ok && cStep > maxStep {
+ if ok && !servableUnderBound(cStep, maxStep) {
ok = false
}
if dbg.KVReadLevelledMetrics {
@@ -1191,7 +1182,11 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
}
}
if ok {
- if dbg.AssertStateCache {
+ // The divergence assert is skipped while the mem overlay bounds this
+ // key (in-flight unwind): MDBX still holds the not-yet-deleted dying
+ // rows inside the bound, so the "authoritative" read can return
+ // dead-fork bytes and blame the cache for a legitimate hit.
+ if dbg.AssertStateCache && maxStep == kv.Step(math.MaxUint64) {
// Fetch authoritative value from the backing tx and panic on any divergence.
// sd.mem and sd.parent.mem were already checked above and missed, so the
// backing tx is the single source of truth for this key at this point.
@@ -1225,18 +1220,11 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
// with a newer MDBX state.
if domain == kv.CommitmentDomain && sd.branchCache != nil {
if cv, cStepU64, ok := sd.branchCache.Get(k); ok {
- // Respect maxStep. sd.mem / sd.parent.mem lowered maxStep above when an
- // unwound key reports its restored step via unwindChangeset (the per-key
- // in-mem unwind signal). The cache's global epoch/floor is coarser than
- // that per-key signal, so a cached entry below the floor can still belong
- // to a step the unwind re-bound away. Serving it then diverges from the
- // (maxStep-bounded) DB read the cache-disabled path takes. Fall through to
- // the bounded read in that case; the Put below refreshes the entry.
// Get returns the on-disk step index directly — do NOT divide by
// StepSize (that double-division collapsed cStep to ~0, defeating the
// gate).
cStep := kv.Step(cStepU64)
- if cStep <= maxStep {
+ if servableUnderBound(cStep, maxStep) {
return cv, cStep, nil
}
}
@@ -1257,11 +1245,13 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
return nil, 0, fmt.Errorf("storage %x read error: %w", k, err)
}
- // Populate state cache on successful read. Stamp with an upper bound on the
- // value's write txNum (the read only gives us the file/step it came from):
- // the last txNum of that step. An unwind below this bound can't leave the
- // value stale, so frozen-step reads stay warm across unwinds while
- // recent-step reads are dropped.
+ // Populate the cache with if-absent semantics: a read-fill never carries
+ // newer information than a flush-apply, so it must not overwrite one
+ // (e.g. an embedded-RPC read straddling an FCU commit). Stamp with the
+ // last txNum of the step the value came from — an upper bound on its
+ // write txNum — so an unwind below it can't leave the entry stale. A
+ // negative carries no step; stamp it with the domain's progress at
+ // observation time so any unwind drops it.
if sd.stateCache != nil {
readTxNum := (uint64(step)+1)*sd.StepSize() - 1
if domain == kv.CodeDomain {
@@ -1269,14 +1259,16 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k
// This SD getter is the single place that populates the code cache
// on a read. Key the content-addressed entry by the code's OWN hash,
// keccak(v) — NEVER a separately-read account codeHash, which under
- // parallel exec can be a skewed/cross-account value and would poison
- // the shared codeHash->code map for every account sharing that hash.
- // keccak(v) makes every cached entry self-consistent, so a skewed
- // account read can never produce a bad entry.
- sd.stateCache.PutCodeWithHash(k, v, crypto.Keccak256(v), readTxNum)
+ // parallel exec can be a skewed or cross-account value and would
+ // poison the shared codeHash→code map for every account sharing
+ // that hash.
+ sd.stateCache.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), readTxNum)
}
} else {
- sd.stateCache.Put(domain, k, v, readTxNum)
+ if len(v) == 0 && sd.stateCache.GetCache(domain) != nil {
+ readTxNum = tx.Debug().DomainProgress(domain)
+ }
+ sd.stateCache.PutIfAbsent(domain, k, v, readTxNum)
}
}
// Only cache a branch when the read's txN is known: a txN=0 entry would
@@ -1415,11 +1407,11 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, addr []byte, txNum ui
// on flush. Route mem-first; the LRU is a committed-state layer that may only
// answer once mem has missed.
if v, _, ok := sd.mem.GetLatest(kv.AccountsDomain, addr); ok {
- return decodeAccountCodeHash(v)
+ return accounts.DeserialiseV3CodeHash(v)
}
if sd.parent != nil {
if v, _, ok := sd.parent.mem.GetLatest(kv.AccountsDomain, addr); ok {
- return decodeAccountCodeHash(v)
+ return accounts.DeserialiseV3CodeHash(v)
}
}
@@ -1440,14 +1432,14 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, addr []byte, txNum ui
resolve := func() []byte {
if sd.stateCache != nil {
if v, ok := sd.stateCache.Get(kv.AccountsDomain, addr); ok {
- return decodeAccountCodeHash(v)
+ return accounts.DeserialiseV3CodeHash(v)
}
}
v, _, err := tx.GetLatest(kv.AccountsDomain, addr)
if err != nil || len(v) == 0 {
return nil
}
- return decodeAccountCodeHash(v)
+ return accounts.DeserialiseV3CodeHash(v)
}
h := resolve()
@@ -1465,28 +1457,6 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, addr []byte, txNum ui
return h
}
-// decodeAccountCodeHash extracts the codeHash from an account's encoded
-// (DecodeForStorage) bytes. Returns nil on decode error or when the account
-// has no code (empty codeHash).
-func decodeAccountCodeHash(enc []byte) []byte {
- if len(enc) == 0 {
- return nil
- }
- var acc accounts.Account
- // AccountsDomain values are SerialiseV3-encoded, so they must be decoded
- // with DeserialiseV3. DecodeForStorage is the legacy MDBX bitmask format
- // with an incompatible binary layout; applied to V3 bytes it silently
- // misparses and leaves CodeHash empty.
- if err := accounts.DeserialiseV3(&acc, enc); err != nil {
- return nil
- }
- if acc.CodeHash.IsEmpty() {
- return nil
- }
- h := acc.CodeHash.Value()
- return h[:]
-}
-
func (sd *SharedDomains) Metrics() *kvmetrics.DomainMetrics {
return &sd.metrics
}
diff --git a/db/state/execctx/statecache_readfill_bench_test.go b/db/state/execctx/statecache_readfill_bench_test.go
new file mode 100644
index 00000000000..a5868cf2933
--- /dev/null
+++ b/db/state/execctx/statecache_readfill_bench_test.go
@@ -0,0 +1,99 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package execctx_test
+
+import (
+ "encoding/binary"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/db/kv"
+ "github.com/erigontech/erigon/db/state/execctx"
+)
+
+// benchSeedDb commits one account so the domain tables are non-empty for
+// cold-negative probes.
+func benchSeedDb(b *testing.B) kv.TemporalRwDB {
+ b.Helper()
+ const stepSize = uint64(16)
+ ctx := b.Context()
+ db := newTestDb(b, stepSize)
+
+ rwTx, err := db.BeginTemporalRw(ctx)
+ require.NoError(b, err)
+ defer rwTx.Rollback()
+ sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
+ require.NoError(b, err)
+ defer sd.Close()
+ written := make([]byte, 20)
+ written[0] = 0x01
+ sd.SetTxNum(100)
+ require.NoError(b, sd.DomainPut(kv.AccountsDomain, rwTx, written, encAccount(7), 100, nil))
+ require.NoError(b, sd.Commit(ctx, rwTx))
+ return db
+}
+
+// BenchmarkDomainProgress isolates the negative-stamp source: one
+// files.EndTxNum read plus an MDBX LastKey on the domain's keys table.
+func BenchmarkDomainProgress(b *testing.B) {
+ db := benchSeedDb(b)
+ roTx, err := db.BeginTemporalRo(b.Context())
+ require.NoError(b, err)
+ defer roTx.Rollback()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _ = roTx.Debug().DomainProgress(kv.AccountsDomain)
+ }
+}
+
+// benchColdNegativeReads drives the full cold-negative SD read: the whole
+// miss stack, plus — when a cache is wired — the progress stamp and the
+// if-absent fill.
+func benchColdNegativeReads(b *testing.B, withCache bool) {
+ db := benchSeedDb(b)
+ ctx := b.Context()
+ roTx, err := db.BeginTemporalRo(ctx)
+ require.NoError(b, err)
+ defer roTx.Rollback()
+ sd, err := execctx.NewSharedDomains(ctx, roTx, log.New())
+ require.NoError(b, err)
+ defer sd.Close()
+ if withCache {
+ sd.SetStateCacheForTest(newSmallStateCache())
+ }
+
+ key := make([]byte, 20)
+ key[0] = 0x02
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ binary.BigEndian.PutUint64(key[12:], uint64(i)+1)
+ v, _, err := sd.GetLatest(kv.AccountsDomain, roTx, key)
+ if err != nil {
+ b.Fatal(err)
+ }
+ if len(v) != 0 {
+ b.Fatalf("expected a negative, got %x", v)
+ }
+ }
+}
+
+func BenchmarkGetLatestColdNegative(b *testing.B) { benchColdNegativeReads(b, true) }
+
+// The baseline the stamp+fill cost adds to.
+func BenchmarkGetLatestColdNegativeNoCache(b *testing.B) { benchColdNegativeReads(b, false) }
diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go
new file mode 100644
index 00000000000..7209a1cac21
--- /dev/null
+++ b/db/state/execctx/statecache_readfill_test.go
@@ -0,0 +1,211 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package execctx_test
+
+import (
+ "encoding/binary"
+ "testing"
+
+ "github.com/c2h5oh/datasize"
+ "github.com/holiman/uint256"
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common/dbg"
+ "github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/db/kv"
+ "github.com/erigontech/erigon/db/state/execctx"
+ "github.com/erigontech/erigon/execution/cache"
+ "github.com/erigontech/erigon/execution/types/accounts"
+)
+
+func encAccount(nonce uint64) []byte {
+ a := accounts.Account{Nonce: nonce, Balance: *uint256.NewInt(nonce * 1000)}
+ return accounts.SerialiseV3(&a)
+}
+
+// twoStepRows commits two versions of one account key so MDBX holds rows at
+// step 0 (txNum 5, v1) and step 1 (txNum 20, v2), and returns a delete-only
+// unwind diff for the step-1 row — the legacy-changeset shape that makes the
+// mem overlay publish a per-key maxStep bound while MDBX still holds the
+// dying row.
+func twoStepRows(t *testing.T, db kv.TemporalRwDB, sc *cache.StateCache) (key, v1, v2 []byte, diffs [kv.DomainLen][]kv.DomainEntryDiff) {
+ t.Helper()
+ ctx := t.Context()
+ key = make([]byte, 20)
+ key[0] = 0xaa
+ v1, v2 = encAccount(1), encAccount(2)
+
+ rwTx, err := db.BeginTemporalRw(ctx)
+ require.NoError(t, err)
+ defer rwTx.Rollback()
+
+ sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
+ require.NoError(t, err)
+ defer sd.Close()
+ sd.SetStateCacheForTest(sc)
+
+ sd.SetTxNum(5)
+ require.NoError(t, sd.DomainPut(kv.AccountsDomain, rwTx, key, v1, 5, nil))
+ sd.SetTxNum(20)
+ require.NoError(t, sd.DomainPut(kv.AccountsDomain, rwTx, key, v2, 20, v1))
+ require.NoError(t, sd.Commit(ctx, rwTx))
+
+ stepBytes := make([]byte, 8)
+ binary.BigEndian.PutUint64(stepBytes, ^uint64(1))
+ diffs[kv.AccountsDomain] = []kv.DomainEntryDiff{{Key: string(key) + string(stepBytes), Value: nil}}
+ return key, v1, v2, diffs
+}
+
+func newSmallStateCache() *cache.StateCache {
+ b := 1 * datasize.MB
+ return cache.NewStateCache(b, b, b, b)
+}
+
+// During an in-flight unwind the mem overlay bounds reads of an affected key
+// by maxStep while MDBX still holds the not-yet-deleted dying row inside that
+// bound. A cache hit legitimately below the unwind floor then diverges from
+// the maxStep-bounded DB read, and the ASSERT_STATE_CACHE comparison must not
+// blame the cache for it.
+func TestAssertStateCache_NoFalsePanicDuringInFlightUnwind(t *testing.T) {
+ if testing.Short() {
+ t.Skip()
+ }
+ // Mutates dbg.AssertStateCache — must not run in parallel with tests that
+ // read it on the SD read path.
+
+ const stepSize = uint64(16)
+ ctx := t.Context()
+ db := newTestDb(t, stepSize)
+ sc := newSmallStateCache()
+ key, v1, _, diffs := twoStepRows(t, db, sc)
+
+ roTx, err := db.BeginTemporalRo(ctx)
+ require.NoError(t, err)
+ defer roTx.Rollback()
+
+ sd, err := execctx.NewSharedDomains(ctx, roTx, log.New())
+ require.NoError(t, err)
+ defer sd.Close()
+ sd.SetStateCacheForTest(sc)
+
+ // A live cache entry below the unwind floor: the restored (correct) value.
+ sc.Put(kv.AccountsDomain, key, v1, 5)
+ sd.Unwind(10, &diffs) // in-flight: mem publishes maxStep=1; MDBX still holds the step-1 row
+
+ old := dbg.AssertStateCache
+ dbg.AssertStateCache = true
+ t.Cleanup(func() { dbg.AssertStateCache = old })
+
+ var v []byte
+ require.NotPanics(t, func() {
+ v, _, err = sd.GetLatest(kv.AccountsDomain, roTx, key)
+ }, "assert must not fire on a legitimately-bounded cache hit during an in-flight unwind")
+ require.NoError(t, err)
+ require.Equal(t, v1, v, "the cache serves the restored value")
+}
+
+// The read-fill after a fall-through read must not replace a live cache
+// entry: it never carries newer information than a flush-apply, and during an
+// in-flight unwind the bounded DB read can even return the not-yet-deleted
+// dying row.
+func TestReadFill_DoesNotClobberLiveEntry(t *testing.T) {
+ if testing.Short() {
+ t.Skip()
+ }
+ t.Parallel()
+
+ const stepSize = uint64(16)
+ ctx := t.Context()
+ db := newTestDb(t, stepSize)
+ sc := newSmallStateCache()
+ key, _, v2, diffs := twoStepRows(t, db, sc)
+
+ roTx, err := db.BeginTemporalRo(ctx)
+ require.NoError(t, err)
+ defer roTx.Rollback()
+
+ sd, err := execctx.NewSharedDomains(ctx, roTx, log.New())
+ require.NoError(t, err)
+ defer sd.Close()
+ sd.SetStateCacheForTest(sc)
+
+ sd.Unwind(10, &diffs)
+ // A live (current-epoch) entry above the read bound: the maxStep gate turns
+ // the hit into a miss, so the read falls through to the bounded DB read.
+ v3 := encAccount(3)
+ sc.Put(kv.AccountsDomain, key, v3, 40)
+
+ v, _, err := sd.GetLatest(kv.AccountsDomain, roTx, key)
+ require.NoError(t, err)
+ require.Equal(t, v2, v, "fall-through read serves the maxStep-bounded DB row")
+
+ got, ok := sc.Get(kv.AccountsDomain, key)
+ require.True(t, ok)
+ require.Equal(t, v3, got, "read-fill must not clobber the live entry")
+}
+
+// Negative results (missing account) must be stamped with the domain's
+// progress at observation time, not a synthetic step-0 bound that survives
+// every unwind.
+func TestReadFill_NegativeStampedWithProgress(t *testing.T) {
+ if testing.Short() {
+ t.Skip()
+ }
+ t.Parallel()
+
+ const stepSize = uint64(16)
+ ctx := t.Context()
+ db := newTestDb(t, stepSize)
+ sc := newSmallStateCache()
+
+ rwTx, err := db.BeginTemporalRw(ctx)
+ require.NoError(t, err)
+ defer rwTx.Rollback()
+
+ sd, err := execctx.NewSharedDomains(ctx, rwTx, log.New())
+ require.NoError(t, err)
+ defer sd.Close()
+ sd.SetStateCacheForTest(sc)
+
+ written := make([]byte, 20)
+ written[0] = 0x01
+ sd.SetTxNum(100)
+ require.NoError(t, sd.DomainPut(kv.AccountsDomain, rwTx, written, encAccount(7), 100, nil))
+ require.NoError(t, sd.Commit(ctx, rwTx))
+
+ roTx, err := db.BeginTemporalRo(ctx)
+ require.NoError(t, err)
+ defer roTx.Rollback()
+ sd2, err := execctx.NewSharedDomains(ctx, roTx, log.New())
+ require.NoError(t, err)
+ defer sd2.Close()
+ sd2.SetStateCacheForTest(sc)
+
+ missing := make([]byte, 20)
+ missing[0] = 0x02
+ v, _, err := sd2.GetLatest(kv.AccountsDomain, roTx, missing)
+ require.NoError(t, err)
+ require.Empty(t, v)
+ _, ok := sc.Get(kv.AccountsDomain, missing)
+ require.True(t, ok, "the negative result must be cached")
+
+ // The domain's progress is 100 (the committed write), so any unwind at or
+ // below it must drop the negative instead of letting it outlive the fact.
+ sc.Unwind(50)
+ _, ok = sc.Get(kv.AccountsDomain, missing)
+ require.False(t, ok, "a negative observed at progress 100 must not survive an unwind to 50")
+}
diff --git a/execution/cache/cache_test.go b/execution/cache/cache_test.go
index 3051a4f365c..a3abba6ac6c 100644
--- a/execution/cache/cache_test.go
+++ b/execution/cache/cache_test.go
@@ -27,6 +27,7 @@ import (
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/crypto"
+ "github.com/erigontech/erigon/common/dbg"
"github.com/erigontech/erigon/db/kv"
)
@@ -866,11 +867,10 @@ func TestCodeCache_PutWithCodeHashIfAbsent(t *testing.T) {
// clobber it — the prefetch-vs-flush staleness this cache guards against.
func TestDomainCache_PutIfAbsentAtomicWithPut(t *testing.T) {
c := NewDomainCacheMode(1*datasize.MB, ModeEvictLRU)
- addr := makeAddr(1)
fresh := []byte("fresh")
stale := []byte("stale")
for round := 0; round < 20000; round++ {
- c.Delete(addr)
+ addr := makeAddr(round)
var wg sync.WaitGroup
wg.Add(2)
go func() { defer wg.Done(); c.Put(addr, fresh, 20) }()
@@ -881,3 +881,111 @@ func TestDomainCache_PutIfAbsentAtomicWithPut(t *testing.T) {
require.Equal(t, fresh, v, "round %d: PutIfAbsent raced past a concurrent Put", round)
}
}
+
+// A Delete racing an update-in-place put must not double-subtract the
+// displaced entry's size: freelru's OnEvict subtracts it for the Remove, and
+// put's update delta subtracts it again unless the two writers share the
+// key's stripe.
+func TestDomainCache_DeleteAtomicWithPut_NoSizeDrift(t *testing.T) {
+ c := NewDomainCacheMode(1*datasize.MB, ModeEvictLRU)
+ addr := makeAddr(1)
+ v1 := []byte("value-one")
+ v2 := []byte("value-two")
+ for round := 0; round < 20000; round++ {
+ c.Put(addr, v1, 10)
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() { defer wg.Done(); c.Put(addr, v2, 20) }()
+ go func() { defer wg.Done(); c.Delete(addr) }()
+ wg.Wait()
+ c.Delete(addr)
+ require.Zero(t, c.SizeBytes(), "round %d: size accounting drifted", round)
+ }
+}
+
+// The lazy stale-drop inside GetWithTxNum removes entries; an unstriped
+// Remove racing put's read-modify-write double-subtracts the displaced
+// entry's size. Exactly one live entry remains after every round, so drift
+// shows as a size mismatch.
+func TestDomainCache_StaleDropAtomicWithPut_NoSizeDrift(t *testing.T) {
+ c := NewDomainCacheMode(1*datasize.MB, ModeEvictLRU)
+ addr := makeAddr(1)
+ v1 := []byte("value-one")
+ v2 := []byte("value-two")
+ wantSize := int64(len(addr) + len(v1) + 24)
+ for round := 0; round < 20000; round++ {
+ c.Put(addr, v1, 10)
+ c.Unwind(5)
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() { defer wg.Done(); c.Put(addr, v2, 20) }()
+ go func() { defer wg.Done(); c.Get(addr) }()
+ wg.Wait()
+ require.Equal(t, wantSize, c.SizeBytes(), "round %d: size accounting drifted", round)
+ }
+}
+
+// A Clear racing a put must not leave phantom bytes: unless Clear excludes
+// writers via the put stripes, a put that loaded the retiring generation
+// lands its entry where no reader sees it and adds the entry's size after
+// Clear zeroed the counter — inflating SizeBytes for an invisible entry.
+func TestDomainCache_ClearAtomicWithPut_NoSizeDrift(t *testing.T) {
+ c := NewDomainCacheMode(1*datasize.MB, ModeEvictLRU)
+ addr := makeAddr(1)
+ v1 := []byte("value-one")
+ entrySize := int64(len(addr) + len(v1) + 24)
+ for round := 0; round < 20000; round++ {
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() { defer wg.Done(); c.Put(addr, v1, 10) }()
+ go func() { defer wg.Done(); c.Clear() }()
+ wg.Wait()
+ wantSize := int64(0)
+ if _, ok := c.Get(addr); ok {
+ wantSize = entrySize
+ }
+ require.Equal(t, wantSize, c.SizeBytes(), "round %d: size accounting drifted", round)
+ }
+}
+
+func TestCodeCache_ContainsLive(t *testing.T) {
+ cc := NewCodeCache(1*datasize.MB, 1*datasize.MB)
+ addr := makeAddr(1)
+ code := []byte{0xaa, 1, 2, 3}
+ require.False(t, cc.ContainsLive(addr), "absent addr")
+
+ cc.PutWithCodeHash(addr, code, crypto.Keccak256(code), 10)
+ require.True(t, cc.ContainsLive(addr))
+
+ cc.Unwind(5)
+ require.False(t, cc.ContainsLive(addr), "stale binding must not read as live")
+
+ // A bound addr whose content entry was evicted is not live: the binding
+ // alone cannot serve the bytes.
+ cc2 := NewCodeCache(1*datasize.MB, 1*datasize.MB)
+ cc2.PutWithCodeHash(addr, code, crypto.Keccak256(code), 10)
+ cc2.hashToCode.Purge()
+ require.False(t, cc2.ContainsLive(addr), "binding without content bytes is not servable")
+}
+
+// The drain-before-unwind convention (drainReadAhead ordered before every
+// epoch bump) is enforced here: a cache-populating warmup still in flight at
+// Unwind time can stamp a dead-fork value with the post-unwind epoch.
+func TestStateCache_UnwindAssertsWarmupInFlight(t *testing.T) {
+ old := dbg.AssertStateCache
+ dbg.AssertStateCache = true
+ t.Cleanup(func() { dbg.AssertStateCache = old })
+
+ b := 1 * datasize.MB
+ sc := NewStateCache(b, b, b, b)
+ sc.WarmupStarted()
+ require.Panics(t, func() { sc.Unwind(10) }, "epoch bump with a warmup in flight must fail loud")
+ sc.WarmupDone()
+ require.NotPanics(t, func() { sc.Unwind(10) })
+
+ // Without the assert flag the gauge is inert.
+ dbg.AssertStateCache = false
+ sc.WarmupStarted()
+ defer sc.WarmupDone()
+ require.NotPanics(t, func() { sc.Unwind(10) })
+}
diff --git a/execution/cache/code_cache.go b/execution/cache/code_cache.go
index 48c3a266a0a..2c1c442bdc4 100644
--- a/execution/cache/code_cache.go
+++ b/execution/cache/code_cache.go
@@ -361,6 +361,22 @@ func (c *CodeCache) putCode(addr []byte, code []byte, keyHash [32]byte, txNum ui
&c.coh, &c.codeSize, 8, &c.putStripes[uint8(codeID)])
}
+// ContainsLive reports whether addr resolves to live code bytes through the
+// addr→code binding, without touching hit/miss counters or LRU recency.
+// Prefetchers probe it to skip the keccak+copy work of preparing a conditional
+// put that a live binding would no-op; advisory only.
+func (c *CodeCache) ContainsLive(addr []byte) bool {
+ vID, ok := c.addrToHash.Peek(common.BytesToAddress(addr))
+ if !ok || c.isStale(vID.txNum, vID.epoch) {
+ return false
+ }
+ ce, ok := c.hashToCode.Peek(vID.addrID)
+ if !ok || len(ce.code) == 0 || c.isStale(ce.txNum, ce.epoch) {
+ return false
+ }
+ return vID.codeHash == ([32]byte{}) || ce.keyHash == vID.codeHash
+}
+
// GetAddrCodeHash returns the Ethereum codeHash for addr if cached. Lets
// SharedDomains.codeHashForAddr skip a cold AccountsDomain read when the
// EVM-known codeHash is already known. Eviction is LRU; freshly seen addrs
diff --git a/execution/cache/code_cache_concurrency_test.go b/execution/cache/code_cache_concurrency_test.go
index 109a01874fa..fb2a63fc12d 100644
--- a/execution/cache/code_cache_concurrency_test.go
+++ b/execution/cache/code_cache_concurrency_test.go
@@ -17,6 +17,7 @@
package cache
import (
+ "encoding/binary"
"sync"
"testing"
@@ -130,7 +131,7 @@ func TestCodeCache_PutIfAbsentAtomicWithPut(t *testing.T) {
fresh := []byte{0xaa, 1, 2, 3}
stale := []byte{0xbb, 4, 5, 6}
for round := 0; round < 20000; round++ {
- cc.Delete(addr)
+ binary.BigEndian.PutUint64(addr[1:], uint64(round))
var wg sync.WaitGroup
wg.Add(2)
go func() { defer wg.Done(); cc.Put(addr, fresh, 20) }()
diff --git a/execution/cache/generic_cache.go b/execution/cache/generic_cache.go
index 4d261f686f7..e3545978c72 100644
--- a/execution/cache/generic_cache.go
+++ b/execution/cache/generic_cache.go
@@ -20,6 +20,7 @@ import (
"bytes"
"sync"
"sync/atomic"
+ "time"
"github.com/c2h5oh/datasize"
"github.com/elastic/go-freelru"
@@ -59,9 +60,10 @@ type entry[T any] struct {
// GenericCache is a sharded, LRU-evicting bounded cache for key-value
// data. Eviction mode is fixed at construction (see policy.go).
type GenericCache[T any] struct {
- // data is the sharded LRU, replaced wholesale on a jump-grow. Load it once per
- // operation; a write racing a resize may land in the LRU about to be replaced
- // and be dropped — a benign miss (the value is re-read from the domain).
+ // data is the sharded LRU, replaced wholesale only with every put stripe
+ // held — on a jump-grow (fully copied generation) and on Clear (fresh
+ // empty one) — so no write lands in a retired generation and no reader
+ // sees a partial copy (see maybeGrow, Clear).
data atomic.Pointer[freelru.ShardedLRU[uint64, entry[T]]]
capacityB datasize.ByteSize
mode Mode
@@ -172,7 +174,11 @@ func newGenericCacheEntries[T any](capacityBytes datasize.ByteSize, capacityEntr
}
// newShards builds a sharded LRU of the given capacity with this cache's evict
-// callback wired, so currentSize follows capacity-driven eviction and Remove.
+// callback wired. The callback is the sole subtractor of currentSize — every
+// removal (capacity eviction, Remove) accounts through it. Freelru picks
+// eviction victims per shard (hash bits 16+), which the put stripes (bits 0-7)
+// don't cover, so any subtraction computed outside the callback races a
+// cross-stripe eviction of the same entry.
func (c *GenericCache[T]) newShards(capacity uint32) *freelru.ShardedLRU[uint64, entry[T]] {
lru, err := freelru.NewSharded[uint64, entry[T]](capacity, u64identity)
if err != nil {
@@ -187,7 +193,15 @@ func (c *GenericCache[T]) newShards(capacity uint32) *freelru.ShardedLRU[uint64,
// maybeGrow jump-resizes the LRU one step larger when it is full, the ceiling
// hasn't been reached, and the shared envelope can fund the step. Otherwise the
-// LRU keeps its size and freelru evicts within it. Called with no lock held.
+// LRU keeps its size and freelru evicts within it. Must not be called with a
+// stripe held (it takes them all).
+//
+// The copy runs with every put stripe held: writers (and the striped
+// stale-drop) are excluded, so no write can land in the generation being
+// retired and a conditional put never sees a mid-resize gap it could fill
+// with a stale value; readers stay on the retiring generation until the swap
+// and never miss. Grows are a handful of steps per cache lifetime, so the
+// writer stall is a bounded one-off.
func (c *GenericCache[T]) maybeGrow() {
c.resizeMu.Lock()
defer c.resizeMu.Unlock()
@@ -202,15 +216,27 @@ func (c *GenericCache[T]) maybeGrow() {
if !cachebudget.Global.Reserve(delta) {
return
}
- next := c.newShards(newCap)
+ start := time.Now()
+ next := c.newShards(newCap) // allocate before excluding writers
+ fenceStart := time.Now()
+ for i := range c.putStripes {
+ c.putStripes[i].Lock()
+ }
+ copied := 0
for _, k := range old.Keys() {
if v, ok := old.Get(k); ok {
next.Add(k, v)
+ copied++
}
}
c.data.Store(next)
c.curCap.Store(newCap)
+ for i := range c.putStripes {
+ c.putStripes[i].Unlock()
+ }
c.reservedBytes += delta
+ log.Debug("[cache] jump-grow", "fromSlots", curCap, "toSlots", newCap, "copied", copied,
+ "alloc", fenceStart.Sub(start), "fenced", time.Since(fenceStart))
}
// DomainCache wraps GenericCache[[]byte] to implement the Cache interface.
@@ -271,7 +297,7 @@ func (c *GenericCache[T]) GetWithTxNum(key []byte) (T, uint64, bool) {
// tx — and must be dropped; >= not > (the surviving block's last txNum is
// floor-1, so this never drops a live entry).
if c.coh.IsStale(e.txNum, e.epoch) {
- lru.Remove(h)
+ c.dropStale(h, key)
c.staleEvicted.Add(1)
c.misses.Add(1)
var zero T
@@ -296,6 +322,17 @@ func (c *GenericCache[T]) PutIfAbsent(key []byte, value T, txNum uint64) {
}
func (c *GenericCache[T]) put(key []byte, value T, txNum uint64, overwrite bool) {
+ if c.putLocked(key, value, txNum, overwrite) {
+ // Grow outside the stripe — maybeGrow takes every stripe.
+ c.maybeGrow()
+ }
+}
+
+// putLocked performs the write under the key's stripe and reports whether the
+// insert landed in a full LRU with ceiling headroom, i.e. the caller should
+// grow. Detection stays on the insert path — Len locks every shard, too costly
+// per warm update.
+func (c *GenericCache[T]) putLocked(key []byte, value T, txNum uint64, overwrite bool) bool {
h := maphash.Hash(key)
valBytes := c.sizeFunc(value)
newSize := len(key) + valBytes + 24
@@ -308,16 +345,17 @@ func (c *GenericCache[T]) put(key []byte, value T, txNum uint64, overwrite bool)
lru := c.data.Load()
existing, hasExisting := lru.Get(h)
- // Existing key — update in place. Reuse the stored key buffer to
- // avoid an extra allocation; the freshly-decoded value replaces the
- // old one.
+ // Existing key — update by remove-then-add (see newShards for why a size
+ // delta would be wrong). Reuse the stored key buffer to avoid an extra
+ // allocation; the freshly-decoded value replaces the old one.
if hasExisting && bytes.Equal(existing.key, key) {
if !overwrite && !c.coh.IsStale(existing.txNum, existing.epoch) {
- return
+ return false
}
+ c.removeLocked(lru, h)
lru.Add(h, entry[T]{key: existing.key, val: value, size: newSize, txNum: txNum, epoch: ep})
- c.currentSize.Add(int64(newSize - existing.size))
- return
+ c.currentSize.Add(int64(newSize))
+ return false
}
if c.mode == ModeNoOp {
@@ -325,16 +363,16 @@ func (c *GenericCache[T]) put(key []byte, value T, txNum uint64, overwrite bool)
// entry-count cap, which ModeNoOp ("drop new keys when full") must not do.
if c.currentSize.Load()+int64(newSize) > int64(c.capacityB) || lru.Len() >= int(c.maxCap) {
c.dropped.Add(1)
- return
+ return false
}
}
- // ModeEvictLRU: grow toward the ceiling before inserting into a full LRU, so a
- // busy cache expands into its budget rather than evicting at the start size.
- if curCap := c.curCap.Load(); c.mode != ModeNoOp && curCap < c.maxCap && lru.Len() >= int(curCap) {
- c.maybeGrow()
- lru = c.data.Load()
- }
+ curCap := c.curCap.Load()
+ // The insert lands before the grow (which must run outside the stripe), so
+ // it and any racers until the swap evict at the pre-grow cap — a transient
+ // bounded by the grow window, not a regression of the grow-first ordering.
+ needGrow := c.mode != ModeNoOp && curCap < c.maxCap && lru.Len() >= int(curCap)
+
// In ModeEvictLRU the byte budget is enforced through the entry-count cap,
// not a separate currentSize check: capacityEntries is derived from
// capacityB (capacityB/avgBytesPerEntry, see NewGenericCache /
@@ -348,24 +386,51 @@ func (c *GenericCache[T]) put(key []byte, value T, txNum uint64, overwrite bool)
// balcache.go / db/state/cache.go accept.
// hasExisting here means a 64-bit maphash collision (different key, same
- // hash): freelru.Add replaces the colliding entry in place WITHOUT firing
- // OnEvict, so subtract the displaced size now — otherwise currentSize drifts
- // up by it permanently.
+ // hash): remove the colliding entry first so OnEvict accounts for it —
+ // freelru.Add would replace it in place without firing OnEvict.
if hasExisting {
- c.currentSize.Add(-int64(existing.size))
+ c.removeLocked(lru, h)
}
keyCopy := common.Copy(key)
lru.Add(h, entry[T]{key: keyCopy, val: value, size: newSize, txNum: txNum, epoch: ep})
c.currentSize.Add(int64(newSize))
c.inserts.Add(1)
+ return needGrow
}
-// Delete removes the data for the given key.
+// removeLocked removes h under the caller-held stripe, deferring the size
+// subtraction to OnEvict (see newShards). The evictions metric is compensated:
+// an intentional removal is not a capacity eviction.
+func (c *GenericCache[T]) removeLocked(lru *freelru.ShardedLRU[uint64, entry[T]], h uint64) {
+ if lru.Remove(h) {
+ c.evictions.Add(^uint64(0))
+ }
+}
+
+// Delete removes the data for the given key. Runs under the key's put stripe
+// so the check-then-remove is atomic against same-key puts and excluded from
+// generation swaps (maybeGrow, Clear), which fence via the stripes.
func (c *GenericCache[T]) Delete(key []byte) {
h := maphash.Hash(key)
+ mu := &c.putStripes[h&(putStripeCount-1)]
+ mu.Lock()
+ defer mu.Unlock()
lru := c.data.Load()
if existing, ok := lru.Get(h); ok && bytes.Equal(existing.key, key) {
- lru.Remove(h)
+ c.removeLocked(lru, h)
+ }
+}
+
+// dropStale removes key's entry under its put stripe: the re-check keeps an
+// entry a concurrent put revived, and the stripe keeps the removal out of
+// generation swaps.
+func (c *GenericCache[T]) dropStale(h uint64, key []byte) {
+ mu := &c.putStripes[h&(putStripeCount-1)]
+ mu.Lock()
+ defer mu.Unlock()
+ lru := c.data.Load()
+ if e, ok := lru.Get(h); ok && bytes.Equal(e.key, key) && c.coh.IsStale(e.txNum, e.epoch) {
+ c.removeLocked(lru, h)
}
}
@@ -373,10 +438,10 @@ func (c *GenericCache[T]) Delete(key []byte) {
// unwindFloor) coherence pair: with no entries left, no stale (txNum, epoch)
// can survive, so a fresh floor keeps subsequent Puts at the live epoch
// serviceable. Mirrors CodeCache.Clear (which already did this — the two had
-// drifted).
+// drifted). The counter reset and the generation swap run with every put
+// stripe held — like maybeGrow's — so a racing put can neither land in the
+// retired generation nor add its size after the reset.
func (c *GenericCache[T]) Clear() {
- c.currentSize.Store(0)
- c.coh.Init()
// Shrink back to the start size and return the grown budget to the envelope,
// keeping the cache adaptive across fork-validation/reset (it regrows on
// demand). A no-op Purge would leave the grown slot array resident.
@@ -386,8 +451,17 @@ func (c *GenericCache[T]) Clear() {
cachebudget.Global.Release(c.reservedBytes - int64(c.startCap)*c.avgEntryBytes)
c.reservedBytes = int64(c.startCap) * c.avgEntryBytes
}
+ next := c.newShards(c.startCap) // allocate before excluding writers
+ for i := range c.putStripes {
+ c.putStripes[i].Lock()
+ }
+ c.currentSize.Store(0)
+ c.coh.Init()
c.curCap.Store(c.startCap)
- c.data.Store(c.newShards(c.startCap))
+ c.data.Store(next)
+ for i := range c.putStripes {
+ c.putStripes[i].Unlock()
+ }
}
// Close returns this cache's envelope reservation so later caches can grow into
diff --git a/execution/cache/generic_cache_concurrency_test.go b/execution/cache/generic_cache_concurrency_test.go
index 6e27e997385..e3e7a9d6d35 100644
--- a/execution/cache/generic_cache_concurrency_test.go
+++ b/execution/cache/generic_cache_concurrency_test.go
@@ -19,9 +19,13 @@ package cache
import (
"encoding/binary"
"sync"
+ "sync/atomic"
"testing"
"github.com/c2h5oh/datasize"
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common/maphash"
)
// TestGenericCache_ConcurrentPutAcrossGrow guards the jump-grow data race:
@@ -51,3 +55,168 @@ func TestGenericCache_ConcurrentPutAcrossGrow(t *testing.T) {
}
wg.Wait()
}
+
+// A same-key put serialized by its stripe must never be undone by a grow: with
+// copy-then-swap migration, a writer that loaded the old generation before the
+// swap landed its write in the abandoned generation, and the migrated (older)
+// value resurfaced as live — a stale serve, not a benign miss. The writer
+// self-verifies each put and a reader checks the hot key's monotonically
+// increasing value never goes backward.
+func TestGenericCache_PutNotLostAcrossGrow(t *testing.T) {
+ value := func(n uint64) []byte {
+ b := make([]byte, 8)
+ binary.BigEndian.PutUint64(b, n)
+ return b
+ }
+ for round := 0; round < 50; round++ {
+ c := NewGenericCache[[]byte](64*datasize.MB, func(v []byte) int { return len(v) }, ModeEvictLRU)
+ hot := []byte("hot-key")
+ c.Put(hot, value(0), 1)
+
+ stop := make(chan struct{})
+ var regressed atomic.Bool
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ for n := uint64(1); ; n++ {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ c.Put(hot, value(n), n)
+ if v, ok := c.Get(hot); ok {
+ if got := binary.BigEndian.Uint64(v); got < n {
+ regressed.Store(true)
+ return
+ }
+ }
+ }
+ }()
+ go func() {
+ defer wg.Done()
+ last := uint64(0)
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ if v, ok := c.Get(hot); ok {
+ if n := binary.BigEndian.Uint64(v); n < last {
+ regressed.Store(true)
+ return
+ } else {
+ last = n
+ }
+ }
+ }
+ }()
+
+ // Cross the grow threshold so maybeGrow swaps the generation while the
+ // hot-key writer runs.
+ key := make([]byte, 8)
+ for i := 0; i < 3*genericCacheStartCapacity; i++ {
+ binary.BigEndian.PutUint64(key, uint64(1+i))
+ c.Put(key, []byte{1}, 1)
+ }
+
+ close(stop)
+ wg.Wait()
+ c.Close()
+ require.False(t, regressed.Load(), "round %d: a striped put was lost across a grow (older value resurfaced)", round)
+ }
+}
+
+// A conditional put must keep deferring to a live entry across a grow: if the
+// resize ever publishes a generation the entry hasn't reached yet, a
+// PutIfAbsent arriving in that gap finds the key absent and inserts its
+// (stale) value — the writer class the if-absent semantics exist to close.
+// The prober watches for the generation swap and bursts conditional puts the
+// moment it lands, mimicking a fill thread that starts a put mid-resize.
+//
+// The cache is seeded below any capacity pressure with the hot key inserted
+// last — the LRU victim is always an older seed key, so the hot key cannot be
+// evicted and a stale value at the end can only have come through a resize
+// gap. The grow is forced by lowering curCap: reaching Len >= startCap
+// organically needs every freelru shard full, which would make the hot key
+// evictable and the signal ambiguous.
+func TestGenericCache_PutIfAbsentDefersAcrossGrow(t *testing.T) {
+ fresh := []byte("fresh-value")
+ stale := []byte("stale-value")
+ for round := 0; round < 50; round++ {
+ c := NewGenericCache[[]byte](64*datasize.MB, func(v []byte) int { return len(v) }, ModeEvictLRU)
+ key := make([]byte, 8)
+ for i := 0; i < 512; i++ {
+ binary.BigEndian.PutUint64(key, uint64(1+i))
+ c.Put(key, []byte{1}, 1)
+ }
+ hot := []byte("hot-key")
+ c.Put(hot, fresh, 10)
+
+ stop := make(chan struct{})
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ before := c.data.Load()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ if c.data.Load() != before {
+ for j := 0; j < 4096; j++ {
+ c.PutIfAbsent(hot, stale, 5)
+ }
+ return
+ }
+ }
+ }()
+
+ c.curCap.Store(uint32(c.Len()))
+ binary.BigEndian.PutUint64(key, 0)
+ c.Put(key, []byte{1}, 1) // insert at the lowered cap → triggers the grow
+
+ close(stop)
+ wg.Wait()
+ v, ok := c.Get(hot)
+ require.True(t, ok, "round %d: hot key missing", round)
+ require.Equal(t, fresh, v, "round %d: PutIfAbsent bypassed the live entry across a grow", round)
+ c.Close()
+ }
+}
+
+// A capacity eviction is a size-subtracting writer the put stripes cannot
+// serialize: freelru picks its victim per shard (hash bits 16+), so an insert
+// on one stripe can evict a key whose own update — on another stripe — is
+// between its Get and Add; delta accounting against the pre-eviction size then
+// double-subtracts. Capacity 1 collapses freelru to a single shard, making any
+// two keys same-shard; the keys are chosen to differ in their put stripe. Each
+// hit leaks negative size; drift accumulates and shows after the settle
+// deletes.
+func TestGenericCache_CapacityEvictionAtomicWithPut_NoSizeDrift(t *testing.T) {
+ c := newGenericCacheEntries(1*datasize.MB, 1, func(v []byte) int { return len(v) }, ModeEvictLRU)
+ a := makeAddr(1)
+ var b []byte
+ for i := 2; ; i++ {
+ b = makeAddr(i)
+ if maphash.Hash(a)&(putStripeCount-1) != maphash.Hash(b)&(putStripeCount-1) {
+ break
+ }
+ }
+ v := []byte("value-one")
+ for round := 0; round < 100000; round++ {
+ c.Put(b, v, 10)
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() { defer wg.Done(); c.Put(a, v, 10) }() // insert → evicts b (cap 1)
+ go func() { defer wg.Done(); c.Put(b, v, 20) }() // same-key update path
+ wg.Wait()
+ }
+ c.Delete(a)
+ c.Delete(b)
+ require.Zero(t, c.SizeBytes(), "capacity eviction raced the update-path delta")
+}
diff --git a/execution/cache/grow_lru.go b/execution/cache/grow_lru.go
index b2455f5f5c1..1e1b1897695 100644
--- a/execution/cache/grow_lru.go
+++ b/execution/cache/grow_lru.go
@@ -33,8 +33,14 @@ import (
// pre-commits its full configured capacity — the same demand-growth the state
// caches use — reused across the CodeCache's content and size layers.
//
-// A write racing a resize may land in the LRU about to be replaced and be
-// dropped; that is a benign cache miss (the value is re-read from the DB).
+// Generation swaps (maybeGrow, Purge) are not fenced against writers — safe
+// only for content-addressed layers, where a key's payload never changes: a
+// write lost in a retired generation is a benign miss, and an entry whose
+// removal a racing copy undid serves correct bytes until its stale stamp
+// drops it on the next read. Do not reuse for mutable-per-key values — those
+// need GenericCache's fenced swap. The onEvict-maintained counters are
+// approximate across grow windows (a lost write is counted but never
+// evicted; a raced removal can subtract twice).
type growLRU[V any] struct {
cur atomic.Pointer[freelru.ShardedLRU[uint64, V]]
onEvict func(uint64, V)
@@ -78,6 +84,9 @@ func (g *growLRU[V]) newShards(capacity uint32) *freelru.ShardedLRU[uint64, V] {
func (g *growLRU[V]) Get(key uint64) (V, bool) { return g.cur.Load().Get(key) }
+// Peek is Get without the LRU recency bump, for side-effect-free probes.
+func (g *growLRU[V]) Peek(key uint64) (V, bool) { return g.cur.Load().Peek(key) }
+
func (g *growLRU[V]) Add(key uint64, value V) {
lru := g.cur.Load()
if curCap := g.curCap.Load(); curCap < g.maxCap && lru.Len() >= int(curCap) {
diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go
index 1d215a350e4..28721b082ae 100644
--- a/execution/cache/state_cache.go
+++ b/execution/cache/state_cache.go
@@ -18,7 +18,9 @@ package cache
import (
"bytes"
+ "fmt"
"strings"
+ "sync/atomic"
"github.com/c2h5oh/datasize"
@@ -53,6 +55,12 @@ const (
// Code uses CodeCache (two-level for deduplication).
type StateCache struct {
caches [kv.DomainLen]Cache
+
+ // warmupsInFlight counts fire-and-forget cache-populating prefetches
+ // (WarmupStarted/WarmupDone). Unwind asserts it is zero: a prefetch put
+ // racing the epoch bump could stamp a dead-fork value with the post-unwind
+ // epoch and have it served as canonical.
+ warmupsInFlight atomic.Int64
}
// NewStateCache creates a new StateCache with the specified byte capacities.
@@ -170,6 +178,13 @@ func (c *StateCache) putCodeWithHash(addr, code, codeHash []byte, txNum uint64,
}
}
+// HasLiveCode reports whether addr resolves to live code bytes; see
+// CodeCache.ContainsLive.
+func (c *StateCache) HasLiveCode(addr []byte) bool {
+ cc, ok := c.caches[kv.CodeDomain].(*CodeCache)
+ return ok && cc.ContainsLive(addr)
+}
+
// GetCodeSizeByHash returns the size of code by its Ethereum codeHash
// without loading the bytes. Returns (0, false) when the size-only layer
// is not populated for this hash.
@@ -284,7 +299,15 @@ func (c *StateCache) Close() {
// GenericCaches and the CodeCache, all layers) bumps an epoch + lowers a floor
// and drops stale entries lazily on read. This is the sole cache-invalidation
// path on unwind — the executor never touches the cache during forward execution.
+//
+// Callers must drain any in-flight cache-populating warmup first (see
+// WarmupStarted); the assert converts that convention into a loud failure.
func (c *StateCache) Unwind(unwindToTxNum uint64) {
+ if dbg.AssertStateCache {
+ if n := c.warmupsInFlight.Load(); n != 0 {
+ panic(fmt.Sprintf("StateCache.Unwind with %d cache-populating warmup(s) in flight — missing drain before the epoch bump", n))
+ }
+ }
for _, cache := range c.caches {
if cache != nil {
cache.Unwind(unwindToTxNum)
@@ -292,6 +315,13 @@ func (c *StateCache) Unwind(unwindToTxNum uint64) {
}
}
+// WarmupStarted and WarmupDone bracket a fire-and-forget cache-populating
+// prefetch; see warmupsInFlight.
+func (c *StateCache) WarmupStarted() { c.warmupsInFlight.Add(1) }
+
+// WarmupDone is the counterpart of WarmupStarted.
+func (c *StateCache) WarmupDone() { c.warmupsInFlight.Add(-1) }
+
// GetCache returns the cache for the given domain.
// Returns nil if the domain is not supported.
func (c *StateCache) GetCache(domain kv.Domain) Cache {
diff --git a/execution/exec/blocks_read_ahead.go b/execution/exec/blocks_read_ahead.go
index 57eb3a6fdb0..c95b974df22 100644
--- a/execution/exec/blocks_read_ahead.go
+++ b/execution/exec/blocks_read_ahead.go
@@ -85,6 +85,14 @@ type cachePopulatingGetter struct {
g kv.TemporalGetter
sc *cache.StateCache
stepSize uint64 // for the read txNum upper bound (last txNum of the read's step)
+ // progress returns the domain's max committed txNum in the read snapshot;
+ // it stamps negative results, whose miss carries no step to derive a
+ // bound from.
+ progress func(kv.Domain) uint64
+}
+
+func newCachePopulatingGetter(ttx kv.TemporalTx, sc *cache.StateCache) *cachePopulatingGetter {
+ return &cachePopulatingGetter{g: ttx, sc: sc, stepSize: ttx.Debug().StepSize(), progress: ttx.Debug().DomainProgress}
}
func (cpg *cachePopulatingGetter) GetLatest(name kv.Domain, k []byte) ([]byte, kv.Step, error) {
@@ -93,23 +101,34 @@ func (cpg *cachePopulatingGetter) GetLatest(name kv.Domain, k []byte) ([]byte, k
// If-absent writes only: this runs in a fire-and-forget goroutine over a
// committed snapshot, so an unconditional Put racing an FCU flush's
// cache-apply could replace the flushed value with the pre-flush one.
- if name == kv.CodeDomain && len(v) > 0 {
- // Key the content cache by the code's OWN hash, never a separately
- // read account codeHash: under parallel/speculative exec that hash
- // can be skewed or cross-account, and a (hash, code) pair that
- // doesn't satisfy keccak(code)==hash poisons every account sharing
- // the hash. keccak(v) makes each entry self-consistent.
- cpg.sc.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), (uint64(step)+1)*cpg.stepSize-1)
+ if name == kv.CodeDomain {
+ // A live binding makes the conditional put a no-op — skip before
+ // paying the keccak+copy below. Code negatives end here too: they
+ // are not cacheable (CodeCache drops zero-length puts).
+ if len(v) > 0 && !cpg.sc.HasLiveCode(k) {
+ // Key the content cache by keccak(v), the code's own hash — never
+ // a separately read account codeHash, which parallel exec can skew
+ // (see the code-domain read-fill in SharedDomains.getLatestMetered).
+ cpg.sc.PutCodeWithHashIfAbsent(k, v, crypto.Keccak256(v), (uint64(step)+1)*cpg.stepSize-1)
+ }
} else {
- // Cache including nil/empty results: a probe returning no
- // bytes is a valid negative answer (missing account, empty
- // storage slot; empty code lands here too but CodeCache drops
- // zero-length puts) and caching it lets repeated probes
- // skip the file accessor stack. Mirrors revm's CacheAccount
- // { account: None, status: LoadedNotExisting } pattern.
- // Stamp with an upper bound on the value's write txNum (last txNum
- // of the step it came from) so unwind invalidation is correct.
- cpg.sc.PutIfAbsent(name, k, v, (uint64(step)+1)*cpg.stepSize-1)
+ // Cache including nil/empty results: a probe returning no bytes is
+ // a valid negative answer (missing account, empty storage slot) and
+ // caching it lets repeated probes skip the file accessor stack —
+ // revm's CacheAccount { account: None, status: LoadedNotExisting }
+ // pattern. Stamp with the last txNum of the value's step; a
+ // negative has no step — use the domain's progress at observation
+ // time so any unwind drops it.
+ txNum := (uint64(step)+1)*cpg.stepSize - 1
+ if len(v) == 0 {
+ if cpg.progress == nil {
+ // No progress oracle → no honest stamp; skip rather than
+ // cache an unwind-immortal negative.
+ return v, step, err
+ }
+ txNum = cpg.progress(name)
+ }
+ cpg.sc.PutIfAbsent(name, k, v, txNum)
}
}
return v, step, err
@@ -132,18 +151,34 @@ func (bra *BlockReadAheader) AddHeaderAndBody(ctx context.Context, db kv.RoDB, h
if !bra.warming.CompareAndSwap(false, true) {
return
}
+ // Ordering makes "WaitForWarmup drained ⟹ gauge is zero" hold on its
+ // own: WarmupStarted only after warmWg.Add, WarmupDone before
+ // warmWg.Done (defers run LIFO). StateCache.Unwind asserts on the gauge.
+ // The cache pointer is captured once so the Started/Done pair and the
+ // warmup's puts all bind to the same gauge even if SetStateCache races
+ // the launch.
bra.warmWg.Add(1)
+ sc := bra.stateCache
+ if sc != nil {
+ sc.WarmupStarted()
+ }
go func() {
defer bra.warmWg.Done()
- bra.warmBody(ctx, db, header, body, 8) // use 8 workers for warming
+ if sc != nil {
+ defer sc.WarmupDone()
+ }
+ bra.warmBody(ctx, db, sc, header, body, 8) // use 8 workers for warming
}()
}
}
-// WaitForWarmup blocks until any in-flight warmBody goroutine finishes or
-// the context is cancelled. Call before closing the database to avoid
-// waitTxsAllDoneOnClose hangs.
-func (bra *BlockReadAheader) WaitForWarmup(ctx context.Context) {
+// WaitForWarmup blocks until any in-flight warmBody goroutine finishes or the
+// context is cancelled, reporting whether the warmup fully drained. False
+// means a warmup may still be running — callers about to bump the cache epoch
+// or Clear must treat it as a failed precondition. Call before closing the
+// database to avoid waitTxsAllDoneOnClose hangs (that caller may ignore the
+// result: it only needs a bounded wait).
+func (bra *BlockReadAheader) WaitForWarmup(ctx context.Context) bool {
done := make(chan struct{})
go func() {
bra.warmWg.Wait()
@@ -151,7 +186,9 @@ func (bra *BlockReadAheader) WaitForWarmup(ctx context.Context) {
}()
select {
case <-done:
+ return true
case <-ctx.Done():
+ return false
}
}
@@ -166,7 +203,9 @@ func (bra *BlockReadAheader) AddSenders(senders []byte, blockHash common.Hash) {
// It reads: To accounts, To account code, To account storage from access lists,
// and block-level access lists. Each worker creates its own transaction.
// Only one warmBody can run at a time - concurrent calls are no-ops.
-func (bra *BlockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *types.Header, body *types.Body, workers int) {
+// sc is the launch-time cache snapshot (see AddHeaderAndBody), nil to warm the
+// OS page cache only.
+func (bra *BlockReadAheader) warmBody(ctx context.Context, db kv.RoDB, sc *cache.StateCache, header *types.Header, body *types.Body, workers int) {
defer bra.warming.Store(false)
if !dbg.ReadAhead {
@@ -228,8 +267,8 @@ func (bra *BlockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *t
return nil
}
var getter kv.TemporalGetter = ttx
- if bra.stateCache != nil {
- getter = &cachePopulatingGetter{g: ttx, sc: bra.stateCache, stepSize: ttx.Debug().StepSize()}
+ if sc != nil {
+ getter = newCachePopulatingGetter(ttx, sc)
}
stateReader := state.NewReaderV3(getter)
@@ -299,8 +338,8 @@ func (bra *BlockReadAheader) warmBody(ctx context.Context, db kv.RoDB, header *t
}
var getter kv.TemporalGetter = ttx
var cpg *cachePopulatingGetter
- if bra.stateCache != nil {
- cpg = &cachePopulatingGetter{g: ttx, sc: bra.stateCache, stepSize: ttx.Debug().StepSize()}
+ if sc != nil {
+ cpg = newCachePopulatingGetter(ttx, sc)
getter = cpg
}
stateReader := state.NewReaderV3(getter)
diff --git a/execution/exec/blocks_read_ahead_test.go b/execution/exec/blocks_read_ahead_test.go
index f3a88e4d15c..6ecfcb74132 100644
--- a/execution/exec/blocks_read_ahead_test.go
+++ b/execution/exec/blocks_read_ahead_test.go
@@ -17,6 +17,7 @@
package exec
import (
+ "context"
"testing"
"github.com/c2h5oh/datasize"
@@ -27,6 +28,22 @@ import (
"github.com/erigontech/erigon/execution/cache"
)
+// A cancelled wait can return while a warmup is still in flight — the gauge
+// convention only holds for a drained return, so callers about to bump the
+// cache epoch (or Clear) must be able to tell the two apart.
+func TestWaitForWarmupReportsDrained(t *testing.T) {
+ bra := &BlockReadAheader{}
+ require.True(t, bra.WaitForWarmup(context.Background()), "nothing in flight — drained")
+
+ bra.warmWg.Add(1)
+ cancelled, cancel := context.WithCancel(context.Background())
+ cancel()
+ require.False(t, bra.WaitForWarmup(cancelled), "cancelled wait with a live warmup must report undrained")
+
+ bra.warmWg.Done()
+ require.True(t, bra.WaitForWarmup(context.Background()), "drained after the warmup finished")
+}
+
// stubTemporalGetter stands in for the committed-state snapshot a warmup
// goroutine reads: every GetLatest returns the same fixed value.
type stubTemporalGetter struct {
@@ -98,7 +115,7 @@ func TestCachePopulatingGetterWarmsColdKeys(t *testing.T) {
for _, domain := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain} {
sc := newTestStateCache()
- cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: val}, sc: sc, stepSize: 1_562_500}
+ cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: val}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
_, _, err := cpg.GetLatest(domain, key)
require.NoError(t, err)
got, ok := sc.Get(domain, key)
@@ -107,7 +124,7 @@ func TestCachePopulatingGetterWarmsColdKeys(t *testing.T) {
}
sc := newTestStateCache()
- cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: code}, sc: sc, stepSize: 1_562_500}
+ cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: code}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
_, _, err := cpg.GetLatest(kv.CodeDomain, key)
require.NoError(t, err)
got, ok := sc.Get(kv.CodeDomain, key)
@@ -116,10 +133,65 @@ func TestCachePopulatingGetterWarmsColdKeys(t *testing.T) {
// Negative results (missing account, empty slot) are cached as nil hits.
sc = newTestStateCache()
- cpg = &cachePopulatingGetter{g: stubTemporalGetter{v: nil}, sc: sc, stepSize: 1_562_500}
+ cpg = &cachePopulatingGetter{g: stubTemporalGetter{v: nil}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
_, _, err = cpg.GetLatest(kv.AccountsDomain, key)
require.NoError(t, err)
got, ok = sc.Get(kv.AccountsDomain, key)
require.True(t, ok)
require.Empty(t, got)
}
+
+// With a live addr binding the prefetch must be a full no-op: not even the
+// content layers may be populated for its (superseded) snapshot code, because
+// the liveness pre-check exists to skip the keccak+copy for that code
+// entirely.
+func TestCachePopulatingGetterSkipsContentForLiveBinding(t *testing.T) {
+ addr := []byte("\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff\x00\x11\x22\x33\x44")
+ freshCode := []byte{0xaa, 0x01, 0x02, 0x03}
+ staleCode := []byte{0xbb, 0x04, 0x05, 0x06}
+ sc := newTestStateCache()
+ sc.PutCodeWithHash(addr, freshCode, crypto.Keccak256(freshCode), 54)
+ cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: staleCode}, sc: sc, stepSize: 1_562_500, progress: zeroProgress}
+
+ _, _, err := cpg.GetLatest(kv.CodeDomain, addr)
+ require.NoError(t, err)
+
+ _, ok := sc.GetCodeByHash(crypto.Keccak256(staleCode))
+ require.False(t, ok, "live binding: prefetch must not populate content for the snapshot code")
+}
+
+// Negative results are stamped with the domain's progress at observation time,
+// not a synthetic step-0 bound — a synthetic stamp far below any real unwind
+// floor would make the negative immortal.
+func TestCachePopulatingGetterNegativeDropsOnUnwind(t *testing.T) {
+ key := []byte("\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff\x00\x11\x22\x33\x44")
+ sc := newTestStateCache()
+ cpg := &cachePopulatingGetter{
+ g: stubTemporalGetter{v: nil}, sc: sc, stepSize: 1_562_500,
+ progress: func(kv.Domain) uint64 { return 10_000_000 },
+ }
+ _, _, err := cpg.GetLatest(kv.AccountsDomain, key)
+ require.NoError(t, err)
+ _, ok := sc.Get(kv.AccountsDomain, key)
+ require.True(t, ok)
+
+ sc.Unwind(5_000_000)
+ _, ok = sc.Get(kv.AccountsDomain, key)
+ require.False(t, ok, "a negative observed at txNum 10M must not survive an unwind to 5M")
+}
+
+// A getter constructed without a progress oracle must skip caching negatives
+// (an honest stamp is impossible), not panic.
+func TestCachePopulatingGetterNilProgressSkipsNegative(t *testing.T) {
+ key := []byte("\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff\x00\x11\x22\x33\x44")
+ sc := newTestStateCache()
+ cpg := &cachePopulatingGetter{g: stubTemporalGetter{v: nil}, sc: sc, stepSize: 1_562_500}
+ require.NotPanics(t, func() {
+ _, _, err := cpg.GetLatest(kv.AccountsDomain, key)
+ require.NoError(t, err)
+ })
+ _, ok := sc.Get(kv.AccountsDomain, key)
+ require.False(t, ok, "no progress oracle — the negative must not be cached")
+}
+
+func zeroProgress(kv.Domain) uint64 { return 0 }
diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go
index 009beaaad53..96841069546 100644
--- a/execution/execmodule/exec_module.go
+++ b/execution/execmodule/exec_module.go
@@ -381,22 +381,24 @@ func (e *ExecModule) canonicalHash(ctx context.Context, tx kv.Tx, blockNumber ui
return canonical, nil
}
-// drainReadAhead blocks until any in-flight block-assembly warmup finishes.
-// warmBody is fire-and-forget and populates the shared state/branch caches; if
-// it is still running when an unwind bumps the cache epoch, it can Put a
-// pre-unwind (dead-fork) value stamped with the post-unwind epoch — IsStale then
-// returns false and the stale value is served as canonical (wrong root). A
-// laggard Put can likewise land after a flush's cache-apply and pin the
-// pre-flush snapshot. Call before any unwind epoch-bump or flush cache-apply.
-func (e *ExecModule) drainReadAhead() {
+// drainReadAhead blocks until any in-flight block-assembly warmup finishes,
+// reporting whether it fully drained — false only when the module context is
+// cancelled (shutdown). warmBody is fire-and-forget and populates the shared
+// state/branch caches; if it is still running when an unwind bumps the cache
+// epoch, it can Put a pre-unwind (dead-fork) value stamped with the post-unwind
+// epoch — IsStale then returns false and the stale value is served as canonical
+// (wrong root). A laggard Put can likewise land after a flush's cache-apply and
+// pin the pre-flush snapshot. Call before any unwind epoch-bump or flush
+// cache-apply, and do not proceed to them on false.
+func (e *ExecModule) drainReadAhead() bool {
if e.readAheader == nil {
- return
+ return true
}
ctx := e.bacgroundCtx
if ctx == nil {
ctx = context.Background()
}
- e.readAheader.WaitForWarmup(ctx)
+ return e.readAheader.WaitForWarmup(ctx)
}
func (e *ExecModule) unwindToCommonCanonical(sd *execctx.SharedDomains, tx kv.TemporalRwTx, header *types.Header) error {
@@ -427,7 +429,9 @@ func (e *ExecModule) unwindToCommonCanonical(sd *execctx.SharedDomains, tx kv.Te
return err
}
- e.drainReadAhead()
+ if !e.drainReadAhead() {
+ return fmt.Errorf("read-ahead drain interrupted before unwind: %w", e.bacgroundCtx.Err())
+ }
if err := e.pipelineExecutor.UnwindTo(unwindPoint, stagedsync.ExecUnwind, tx); err != nil {
return err
}
@@ -687,6 +691,19 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) {
}
defer e.semaphore.Release(1)
+ // Engine servers are live before Start, so an early payload validation may
+ // already have warmed the state cache with pre-catchup state. Frozen-block
+ // processing advances state without touching the cache (its SDs are not
+ // wired to it), so such entries would be served stale afterwards — drain
+ // any in-flight warmup and clear before it runs. An interrupted drain
+ // means shutdown: return rather than Clear under a live warmup.
+ if !e.drainReadAhead() {
+ return
+ }
+ if e.stateCache != nil {
+ e.stateCache.Clear()
+ }
+
if err := e.pipelineExecutor.ProcessFrozenBlocks(ctx, hook, e.onlySnapDownloadOnStart); err != nil {
if !errors.Is(err, context.Canceled) {
e.logger.Error("Could not start execution service", "err", err)
diff --git a/execution/execmodule/forkchoice.go b/execution/execmodule/forkchoice.go
index 280959cb907..16f1cfbfa6d 100644
--- a/execution/execmodule/forkchoice.go
+++ b/execution/execmodule/forkchoice.go
@@ -363,8 +363,10 @@ func (e *ExecModule) updateForkChoice(ctx context.Context, originalBlockHash, sa
// Drain any warmup a preceding newPayload spawned: its Puts reflect a
// pre-FCU snapshot and must land before this FCU's unwind epoch-bump and
// flush cache-apply, not after them (no new warmup starts while we hold
- // the semaphore).
- e.drainReadAhead()
+ // the semaphore). An interrupted drain means shutdown — bail.
+ if !e.drainReadAhead() {
+ return sendForkchoiceErrorWithoutWaiting(e.logger, outcomeCh, e.bacgroundCtx.Err(), false)
+ }
var validationError string
diff --git a/execution/execmodule/set_head.go b/execution/execmodule/set_head.go
index 9a9c11c2004..e0ce6581297 100644
--- a/execution/execmodule/set_head.go
+++ b/execution/execmodule/set_head.go
@@ -113,8 +113,10 @@ func (e *ExecModule) SetHead(ctx context.Context, targetBlock uint64) error {
// Drain in-flight warmup before the unwind bumps the cache epoch, so a
// fire-and-forget warmup can't Put a dead-fork value stamped with the new
- // epoch (cross-fork contamination).
- e.drainReadAhead()
+ // epoch (cross-fork contamination). An interrupted drain means shutdown.
+ if !e.drainReadAhead() {
+ return fmt.Errorf("read-ahead drain interrupted before unwind: %w", e.bacgroundCtx.Err())
+ }
// Set the unwind point and run the unwind
if err := e.pipelineExecutor.UnwindTo(targetBlock, stagedsync.StagedUnwind, tx); err != nil {
diff --git a/execution/types/accounts/account.go b/execution/types/accounts/account.go
index 69378516720..5251e376237 100644
--- a/execution/types/accounts/account.go
+++ b/execution/types/accounts/account.go
@@ -17,6 +17,7 @@
package accounts
import (
+ "bytes"
"fmt"
"io"
"math/bits"
@@ -26,6 +27,7 @@ import (
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/empty"
+ "github.com/erigontech/erigon/common/length"
"github.com/erigontech/erigon/execution/rlp"
)
@@ -641,6 +643,38 @@ func DeserialiseV3(a *Account, enc []byte) error {
return nil
}
+// DeserialiseV3CodeHash extracts just the codeHash field from a
+// SerialiseV3-encoded account, skipping the full decode (balance parse,
+// codeHash interning) that DeserialiseV3 pays. It parses only up to and
+// including the codeHash field — later fields are not validated. Returns a
+// subslice of enc — valid only while enc is — or nil when the record is
+// malformed up to that field or the account has no code (including
+// non-canonical spellings of the empty or zero sentinel, which
+// CodeHash.IsEmpty treats as no-code).
+func DeserialiseV3CodeHash(enc []byte) []byte {
+ pos := 0
+ for range 2 { // skip the length-prefixed nonce and balance fields
+ if pos >= len(enc) {
+ return nil
+ }
+ pos += 1 + int(enc[pos])
+ }
+ if pos >= len(enc) {
+ return nil
+ }
+ codeHashBytes := int(enc[pos])
+ pos++
+ if codeHashBytes != length.Hash || pos+codeHashBytes > len(enc) {
+ return nil
+ }
+ h := enc[pos : pos+codeHashBytes]
+ var zero common.Hash
+ if bytes.Equal(h, zero[:]) || bytes.Equal(h, empty.CodeHash[:]) {
+ return nil
+ }
+ return h
+}
+
func SerialiseV3(a *Account) []byte {
var l int
l++
diff --git a/execution/types/accounts/account_test.go b/execution/types/accounts/account_test.go
index eb25a8a8e64..57885f5a08f 100644
--- a/execution/types/accounts/account_test.go
+++ b/execution/types/accounts/account_test.go
@@ -17,6 +17,7 @@
package accounts
import (
+ "bytes"
"testing"
"github.com/holiman/uint256"
@@ -76,6 +77,85 @@ func TestEmptyAccount_BufferStrangeBehaviour(t *testing.T) {
isIncarnationEqual(t, a.Incarnation, decodedAcc.Incarnation)
}
+func TestDeserialiseV3CodeHash(t *testing.T) {
+ t.Parallel()
+ balances := []uint256.Int{{}, *uint256.NewInt(1), *uint256.NewInt(1e18), *new(uint256.Int).Lsh(uint256.NewInt(1), 200)}
+ nonces := []uint64{0, 1, 255, 1 << 40}
+ codeHashes := []CodeHash{EmptyCodeHash, InternCodeHash(common.BytesToHash(crypto.Keccak256([]byte{1, 2, 3})))}
+ incarnations := []uint64{0, 7}
+
+ for _, nonce := range nonces {
+ for i := range balances {
+ for _, ch := range codeHashes {
+ for _, inc := range incarnations {
+ a := Account{Nonce: nonce, Balance: balances[i], CodeHash: ch, Incarnation: inc}
+ enc := SerialiseV3(&a)
+
+ var full Account
+ if err := DeserialiseV3(&full, enc); err != nil {
+ t.Fatal(err)
+ }
+ got := DeserialiseV3CodeHash(enc)
+ if full.CodeHash.IsEmpty() {
+ if got != nil {
+ t.Fatalf("empty codeHash must extract as nil, got %x (acc %+v)", got, a)
+ }
+ } else {
+ want := full.CodeHash.Value()
+ if !bytes.Equal(got, want[:]) {
+ t.Fatalf("extracted %x, want %x (acc %+v)", got, want, a)
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+func TestDeserialiseV3CodeHashMalformed(t *testing.T) {
+ t.Parallel()
+ a := Account{
+ Nonce: 255,
+ Balance: *uint256.NewInt(1e18),
+ CodeHash: InternCodeHash(common.BytesToHash(crypto.Keccak256([]byte{1, 2, 3}))),
+ Incarnation: 4,
+ }
+ enc := SerialiseV3(&a)
+ // [1+nonce][1+balance][1+codeHash]... — the codeHash field is complete at:
+ codeHashEnd := 1 + int(enc[0]) + 1
+ codeHashEnd += int(enc[codeHashEnd-1]) + 1
+ codeHashEnd += int(enc[codeHashEnd-1])
+ // Any truncation cutting into (or before) the codeHash must yield nil,
+ // never an out-of-bounds read; beyond it the codeHash is extractable.
+ for cut := 0; cut <= len(enc); cut++ {
+ got := DeserialiseV3CodeHash(enc[:cut])
+ if cut < codeHashEnd && got != nil {
+ t.Fatalf("cut=%d (codeHash complete at %d): expected nil, got %x", cut, codeHashEnd, got)
+ }
+ if cut >= codeHashEnd && got == nil {
+ t.Fatalf("cut=%d (codeHash complete at %d): expected hash, got nil", cut, codeHashEnd)
+ }
+ }
+ if got := DeserialiseV3CodeHash(nil); got != nil {
+ t.Fatalf("nil input: expected nil, got %x", got)
+ }
+ // A record claiming a non-32-byte codeHash is malformed for extraction.
+ odd := append([]byte{0, 0, 31}, make([]byte, 31)...)
+ if got := DeserialiseV3CodeHash(odd); got != nil {
+ t.Fatalf("non-32-byte codeHash field: expected nil, got %x", got)
+ }
+ // Non-canonical records spelling out the no-code sentinels (canonical
+ // SerialiseV3 writes length 0 instead) must extract as nil, matching
+ // CodeHash.IsEmpty.
+ for _, sentinel := range [][]byte{make([]byte, 32), empty.CodeHash[:]} {
+ rec := append([]byte{0, 0, 32}, sentinel...)
+ rec = append(rec, 0)
+ if got := DeserialiseV3CodeHash(rec); got != nil {
+ t.Fatalf("sentinel codeHash %x: expected nil, got %x", sentinel, got)
+ }
+ }
+}
+
func TestAccountEncodeWithCode(t *testing.T) {
t.Parallel()
a := Account{