From 5824a53a4c5c2499909207dd0b9c777053f34aad Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 16:15:40 +0200 Subject: [PATCH 01/16] execution, db: state-cache review follow-ups; wire frozen-block catchup into the apply stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups to #22444, addressing the remaining post-approval review points, plus the fix for #22925 built on the same machinery. - Commit routes pending state updates through Applier.ApplyAll: the admission write lock is taken once per 4096-update chunk instead of once per key (main's walk had no global lock, so per-key locking was a regression; chunking bounds how long concurrent fills wait). - Flush returns an error on a cache-attached SD: a plain Flush would leave the cache serving pre-flush values forever. The memo test moved into Commit's validate window; an end-to-end test pins the incoherence the rejection prevents. - kv.TemporalRwDB carries Agg() any and the visibility guard became StateCache.BindAggregator; SetStateCache asserts the binding, so no wiring site can forget the load-bearing guard. - Frontier lookups tolerate a tx whose Debug() is nil: no exact frontier, no fill, reads unaffected. - Fill admission outcomes (admitted/rejected) are counted and reported by PrintStatsAndReset. - Apply-only mode (STATE_CACHE_FILLS=false) no longer binds a frontier on the miss path just for Fill to no-op; CanFill means what it says. - ProcessFrozenBlocks' SharedDomains are wired to the state cache: catchup commits apply post-commit and advance the admission frontier, so pre-catchup read views cannot refill stale state — admission is the fence. Closes #22925. - Per-domain admission invariant stated at appliedEnd; duplicated rationale trimmed from view.go; dead domain field dropped from the branch stash. --- cmd/integration/commands/stages.go | 2 +- db/kv/kv_interface.go | 3 + db/kv/membatchwithdb/memory_mutation.go | 2 + db/state/execctx/domain_shared.go | 135 +++++------ db/state/execctx/statecache_readfill_test.go | 223 ++++++++++++++++-- execution/cache/apply_all_test.go | 156 ++++++++++++ execution/cache/state_cache.go | 85 ++++++- execution/cache/view.go | 50 ++-- execution/execmodule/exec_module.go | 4 +- .../execmodule/exec_module_internal_test.go | 47 ++++ execution/execmodule/executor.go | 24 +- 11 files changed, 602 insertions(+), 129 deletions(-) create mode 100644 execution/cache/apply_all_test.go diff --git a/cmd/integration/commands/stages.go b/cmd/integration/commands/stages.go index e55f94ca9b8..76f49c264dd 100644 --- a/cmd/integration/commands/stages.go +++ b/cmd/integration/commands/stages.go @@ -844,9 +844,9 @@ func execBlocksBatch(ctx context.Context, db kv.TemporalRwDB, st *stagedsync.Syn } defer doms.Close() doms.SetInMemHistoryReads(false) + stateCache.BindAggregator(db) doms.SetStateCache(stateCache) doms.SetCodeStore(codeStore) - execctx.GuardAggregatorForCache(db, stateCache) s, err := st.StageState(stages.Execution, tx, initialCycle, false) if err != nil { diff --git a/db/kv/kv_interface.go b/db/kv/kv_interface.go index 78c4fdd76ef..7bf751c4565 100644 --- a/db/kv/kv_interface.go +++ b/db/kv/kv_interface.go @@ -658,6 +658,9 @@ type TemporalRwDB interface { BeginTemporalRw(ctx context.Context) (TemporalRwTx, error) BeginTemporalRwNosync(ctx context.Context) (TemporalRwTx, error) UpdateTemporal(ctx context.Context, f func(tx TemporalRwTx) error) error + // Agg returns the DB's state-files aggregator as `any` (the concrete type + // lives above the kv layer); nil when the DB has none. + Agg() any } // ---- non-important utilities diff --git a/db/kv/membatchwithdb/memory_mutation.go b/db/kv/membatchwithdb/memory_mutation.go index 1d24f6ce070..46415baf39d 100644 --- a/db/kv/membatchwithdb/memory_mutation.go +++ b/db/kv/membatchwithdb/memory_mutation.go @@ -1317,6 +1317,8 @@ func (td temporaldb) BeginTemporalRwNosync(ctx context.Context) (kv.TemporalRwTx return td.memoryMutation, nil } +func (td temporaldb) Agg() any { return nil } + func (td temporaldb) Debug() kv.TemporalDebugDB { panic("not implemented") } diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index 8398bacf452..9c83ca2a905 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -134,7 +134,7 @@ func (m *domainVisibleEndMemo) load(tx kv.TemporalTx, domain kv.Domain, viewID u state = 0 m.viewID.Store(viewID) } - end, ok := tx.Debug().DomainVisibleEnd(domain) + end, ok := debugDomainVisibleEnd(tx, domain) m.ends[domain].Store(end) state |= loadedBit if ok { @@ -157,7 +157,17 @@ func (sd *SharedDomains) domainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (u if _, ok := tx.(kv.TemporalRwTx); ok { return sd.visibleEnds.get(tx, domain) } - return tx.Debug().DomainVisibleEnd(domain) + return debugDomainVisibleEnd(tx, domain) +} + +// debugDomainVisibleEnd tolerates txs without a debug backend (MemoryMutation +// over a nil db): no exact frontier means no fills, reads still work. +func debugDomainVisibleEnd(tx kv.TemporalTx, domain kv.Domain) (uint64, bool) { + dbgTx := tx.Debug() + if dbgTx == nil { + return 0, false + } + return dbgTx.DomainVisibleEnd(domain) } // sdFrontier adapts one (SharedDomains, tx) pair to cache.Frontier: writable @@ -848,33 +858,13 @@ func (sd *SharedDomains) SetStateCache(stateCache *cache.StateCache) { if !dbg.UseStateCache || stateCache == nil { return } + if stateCache.FillsEnabled() && !stateCache.AggregatorBound() { + panic("assert: fill-enabled StateCache wired before BindAggregator — the visibility-lowering guard is not bound") + } sd.stateCache = stateCache sd.cacheApplier = stateCache.Applier() } -// GuardAggregatorForCache forbids visibility lowering on db's aggregator when -// sc is a fill-enabled StateCache: fill admission relies on view frontiers -// never decreasing. This is the one place that binds the invariant — call it -// wherever a fill-enabled cache is wired over a DB. Duck-typed so the storage -// layer need not know the cache type (and vice versa) — but load-bearing, so -// a db that cannot produce its aggregator fails loudly instead of silently -// dropping the guard. A nil or apply-only cache needs no guard. -func GuardAggregatorForCache(db any, sc *cache.StateCache) { - if sc == nil || !sc.FillsEnabled() { - return - } - h, ok := db.(interface{ Agg() any }) - if !ok { - panic(fmt.Sprintf("assert: fill-enabled StateCache wired over %T, which cannot produce its aggregator — the visibility-lowering guard would be silently dropped", db)) - } - agg := h.Agg() - f, ok := agg.(interface{ ForbidVisibilityLowering() }) - if !ok { - panic(fmt.Sprintf("assert: aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg)) - } - f.ForbidVisibilityLowering() -} - // SetCodeStore sets the persistent codehash-keyed code cache. func (sd *SharedDomains) SetCodeStore(codeStore *cache.CodeStore) { sd.codeStore = codeStore @@ -979,22 +969,14 @@ func (sd *SharedDomains) Close() { // admission. // Flush writes the in-memory batch into tx without committing. It deliberately -// does NOT touch the caches: plain Flush leaves the commit to the caller (who -// may still roll back), so it must not warm a cache with state that could be -// rolled back. Cache entries are populated elsewhere — by Commit after a -// successful commit, and by reads (GetLatest) — each stamped with a -// conservative upper-bound txNum. It is that txNum stamp, not population -// timing, that keeps the cache correct: an unwind lowers the floor so every -// entry reflecting a now-dead fork is evicted, and mem-first masking means a -// later in-memory write shadows a stale cached read. -// -// An SD with an attached state cache must route every flush through Commit: -// Flush neither applies nor invalidates, so a populated cache would keep -// serving pre-flush values for the flushed keys after the caller's own -// commit — and Commit collects its cache updates only from its own flush, so -// an earlier plain Flush's keys would never be applied. Cache-less callers -// may Flush and commit themselves. +// does not touch the caches — the caller may still roll back. An SD with a +// state cache must route every flush through Commit: a plain Flush would +// leave the cache serving pre-flush values for the flushed keys forever, so +// it is rejected here. func (sd *SharedDomains) Flush(ctx context.Context, tx kv.RwTx) error { + if sd.stateCache != nil { + return errors.New("SharedDomains with a state cache must flush through Commit") + } defer mxFlushTook.ObserveDuration(time.Now()) return sd.flushMem(ctx, tx) } @@ -1016,12 +998,11 @@ func (sd *SharedDomains) flushMem(ctx context.Context, tx kv.RwTx, opts ...kv.Fl return sd.mem.Flush(ctx, tx, opts...) } -type cacheUpdate struct { - domain kv.Domain - key []byte - val []byte - step kv.Step - txN uint64 +type branchUpdate struct { + key []byte + val []byte + step kv.Step + txN uint64 } // Commit flushes the in-memory batch into tx, commits tx, and only then applies @@ -1066,24 +1047,31 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun // no cache apply ever runs ahead of durable MDBX state. (Reads through // this SD between flush and a failed commit can still fill flushed // values; a failed commit is fatal, so they die with the process.) - var pending []cacheUpdate - stash := func(domain kv.Domain) kv.FlushOption { + var pendingBranch []branchUpdate + var pendingState []cache.Update + stashState := func(domain kv.Domain) kv.FlushOption { return kv.WithFlushCallback(domain, func(k []byte, v []byte, step kv.Step, txNum uint64) { - pending = append(pending, cacheUpdate{ - domain: domain, - key: append([]byte(nil), k...), - val: append([]byte(nil), v...), - step: step, - txN: txNum, + pendingState = append(pendingState, cache.Update{ + Domain: domain, + Key: append([]byte(nil), k...), + Val: append([]byte(nil), v...), + TxNum: txNum, }) }) } var opts []kv.FlushOption if sd.branchCache != nil { - opts = append(opts, stash(kv.CommitmentDomain)) + opts = append(opts, kv.WithFlushCallback(kv.CommitmentDomain, func(k []byte, v []byte, step kv.Step, txNum uint64) { + pendingBranch = append(pendingBranch, branchUpdate{ + key: append([]byte(nil), k...), + val: append([]byte(nil), v...), + step: step, + txN: txNum, + }) + })) } if sd.stateCache != nil { - opts = append(opts, stash(kv.AccountsDomain), stash(kv.StorageDomain)) + opts = append(opts, stashState(kv.AccountsDomain), stashState(kv.StorageDomain)) } // CodeDomain flush stashes state-cache updates and collects code for the // persistent store. The code-store MDBX write is deferred to after flushMem — @@ -1096,12 +1084,11 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun codeStoreWrites = append(codeStoreWrites, [2][]byte{crypto.Keccak256(v), append([]byte(nil), v...)}) } if sd.stateCache != nil { - pending = append(pending, cacheUpdate{ - domain: kv.CodeDomain, - key: append([]byte(nil), k...), - val: append([]byte(nil), v...), - step: step, - txN: txNum, + pendingState = append(pendingState, cache.Update{ + Domain: kv.CodeDomain, + Key: append([]byte(nil), k...), + Val: append([]byte(nil), v...), + TxNum: txNum, }) } })) @@ -1172,18 +1159,15 @@ func (sd *SharedDomains) Commit(ctx context.Context, tx kv.RwTx, validate ...fun if err := tx.Commit(); err != nil { return err } - for i := range pending { - u := &pending[i] - if u.domain == kv.CommitmentDomain { - if len(u.val) == 0 { - sd.branchCache.Invalidate(u.key) - } else { - sd.branchCache.Put(u.key, u.val, uint64(u.step), u.txN) - } - continue + for i := range pendingBranch { + u := &pendingBranch[i] + if len(u.val) == 0 { + sd.branchCache.Invalidate(u.key) + } else { + sd.branchCache.Put(u.key, u.val, uint64(u.step), u.txN) } - sd.cacheApplier.Apply(u.domain, u.key, u.val, u.txN) } + sd.cacheApplier.ApplyAll(pendingState) return nil } @@ -1345,8 +1329,9 @@ func (sd *SharedDomains) getLatestMetered(domain kv.Domain, tx kv.TemporalTx, k } // View freshness is rechecked while the fill is serialized against - // committed cache updates. - if sd.stateCache != nil && sd.stateCache.Caches(domain) { + // committed cache updates. Apply-only mode skips the block: binding a + // frontier for a fill that will no-op is a wasted allocation. + if sd.stateCache != nil && sd.stateCache.FillsEnabled() && sd.stateCache.Caches(domain) { readTxNum := (uint64(step)+1)*sd.StepSize() - 1 fillView := view if !fillView.CanFill() { @@ -1541,7 +1526,7 @@ func (sd *SharedDomains) codeHashForAddr(tx kv.TemporalTx, view cache.ReadView, } h, fromReadView := resolve() - if fromReadView && sd.stateCache != nil { + if fromReadView && sd.stateCache != nil && sd.stateCache.FillsEnabled() { var fixed [32]byte if len(h) == 32 { copy(fixed[:], h) diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 7c198a3034e..382501b8b6d 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -147,17 +147,97 @@ func TestReadFill_MemoizesWritableVisibleEndUntilFlush(t *testing.T) { written[0] = 4 domains.SetTxNum(20) require.NoError(t, domains.DomainPut(kv.AccountsDomain, rwTx, written, encAccount(2), 20, nil)) - require.NoError(t, domains.Flush(ctx, rwTx)) - missing := make([]byte, 20) - missing[0] = 5 - value, _, err := domains.GetLatest(kv.AccountsDomain, rwTx, missing) - require.NoError(t, err) - require.Empty(t, value) + // The memo must re-derive inside Commit's validate window (after the + // internal flush, before the tx commits): reads here already see the + // advanced frontier. + require.NoError(t, domains.Commit(ctx, rwTx, func(kv.RwTx) error { + missing := make([]byte, 20) + missing[0] = 5 + value, _, err := domains.GetLatest(kv.AccountsDomain, rwTx, missing) + require.NoError(t, err) + require.Empty(t, value) + return nil + })) require.Equal(t, uint64(2), debug.calls) require.Greater(t, debug.last, initialEnd) } +// An SD with a state cache must route every flush through Commit: a plain +// Flush neither applies nor invalidates, so the cache would keep serving +// pre-flush values for the flushed keys forever. +func TestFlushRejectsCacheAttachedSD(t *testing.T) { + t.Parallel() + + ctx := t.Context() + db := newTestDb(t, 16) + + rwTx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer rwTx.Rollback() + domains, err := execctx.NewSharedDomains(ctx, rwTx, log.New()) + require.NoError(t, err) + defer domains.Close() + + require.NoError(t, domains.Flush(ctx, rwTx), "cache-less SDs may flush and commit themselves") + + stateCache := newSmallStateCache() + t.Cleanup(stateCache.Close) + domains.SetStateCacheForTest(stateCache) + require.Error(t, domains.Flush(ctx, rwTx)) +} + +// The incoherence the Flush rejection prevents, end to end: after v1 is +// committed (the cache holds it), flushing v2 through another cache-attached +// SD and committing the tx would leave the cache serving v1 while MDBX holds +// v2. The rejection fires at exactly that step; routing through Commit keeps +// the cache coherent. +func TestFlushRejectionPreventsStaleCachedReads(t *testing.T) { + t.Parallel() + + ctx := t.Context() + db := newTestDb(t, 16) + stateCache := newSmallStateCache() + t.Cleanup(stateCache.Close) + + slot := make([]byte, 52) + slot[0] = 1 + v1, v2 := []byte{1}, []byte{2} + + tx1, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer tx1.Rollback() + sd1, err := execctx.NewSharedDomains(ctx, tx1, log.New()) + require.NoError(t, err) + defer sd1.Close() + sd1.SetStateCacheForTest(stateCache) + sd1.SetTxNum(10) + require.NoError(t, sd1.DomainPut(kv.StorageDomain, tx1, slot, v1, 10, nil)) + require.NoError(t, sd1.Commit(ctx, tx1)) + sd1.Close() + + got, ok := stateCache.View(nil).Get(kv.StorageDomain, slot) + require.True(t, ok) + require.Equal(t, v1, got) + + tx2, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer tx2.Rollback() + sd2, err := execctx.NewSharedDomains(ctx, tx2, log.New()) + require.NoError(t, err) + defer sd2.Close() + sd2.SetStateCacheForTest(stateCache) + sd2.SetTxNum(20) + require.NoError(t, sd2.DomainPut(kv.StorageDomain, tx2, slot, v2, 20, nil)) + require.Error(t, sd2.Flush(ctx, tx2), + "the step that would split the cache (v1) from MDBX (v2) must be rejected") + + require.NoError(t, sd2.Commit(ctx, tx2)) + got, ok = stateCache.View(nil).Get(kv.StorageDomain, slot) + require.True(t, ok) + require.Equal(t, v2, got, "Commit keeps the cache coherent with MDBX") +} + // 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 @@ -349,41 +429,132 @@ type fakeForbidder struct{ called bool } func (f *fakeForbidder) ForbidVisibilityLowering() { f.called = true } -type fakeHasAgg struct{ f *fakeForbidder } - -func (h fakeHasAgg) Agg() any { return h.f } - -type fakeHasBadAgg struct{} +// fakeTemporalDB satisfies kv.TemporalRwDB by embedding (the interface now +// carries Agg, so a DB shape without it no longer compiles); only Agg is +// implemented — the guard must not touch anything else. +type fakeTemporalDB struct { + kv.TemporalRwDB + agg any +} -func (fakeHasBadAgg) Agg() any { return struct{}{} } +func (d fakeTemporalDB) Agg() any { return d.agg } -// The guard is load-bearing: for a fill-enabled cache it must either bind the -// invariant or fail loudly — never silently drop it on a DB shape mismatch. -// A nil or apply-only cache needs no guard at all. -func TestGuardAggregatorForCache(t *testing.T) { +// The binding is load-bearing: for a fill-enabled cache it must either bind +// the invariant or fail loudly — never silently drop it. A nil or apply-only +// cache needs no binding at all. +func TestBindAggregator(t *testing.T) { sc := newSmallStateCache() t.Cleanup(sc.Close) f := &fakeForbidder{} - execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc) + sc.BindAggregator(fakeTemporalDB{agg: f}) require.True(t, f.called) + require.True(t, sc.AggregatorBound()) + + var nilCache *cache.StateCache + require.NotPanics(t, func() { nilCache.BindAggregator(fakeTemporalDB{}) }, + "no cache, no invariant to bind — the aggregator is never consulted") + sc2 := newSmallStateCache() + t.Cleanup(sc2.Close) + require.Panics(t, func() { sc2.BindAggregator(fakeTemporalDB{agg: struct{}{}}) }, + "an aggregator without ForbidVisibilityLowering must fail loudly, not drop the binding") +} + +type nilDebugRwTx struct { + kv.TemporalRwTx +} - require.NotPanics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, nil) }, - "no cache, no invariant to bind — shape is irrelevant") - require.Panics(t, func() { execctx.GuardAggregatorForCache(struct{}{}, sc) }, - "a db that cannot produce its aggregator must fail loudly, not drop the guard") - require.Panics(t, func() { execctx.GuardAggregatorForCache(fakeHasBadAgg{}, sc) }, - "an aggregator without ForbidVisibilityLowering must fail loudly, not drop the guard") +func (nilDebugRwTx) Debug() kv.TemporalDebugTx { return nil } + +// A tx without a debug backend (MemoryMutation over a nil db) has no exact +// frontier: reads must still work and simply never fill. +func TestReadFill_NilDebugTxSkipsFills(t *testing.T) { + t.Parallel() + + ctx := t.Context() + db := newTestDb(t, 16) + + baseTx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer baseTx.Rollback() + domains, err := execctx.NewSharedDomains(ctx, baseTx, log.New()) + require.NoError(t, err) + defer domains.Close() + stateCache := newSmallStateCache() + t.Cleanup(stateCache.Close) + domains.SetStateCacheForTest(stateCache) + + missing := make([]byte, 20) + missing[0] = 7 + value, _, err := domains.GetLatest(kv.AccountsDomain, nilDebugRwTx{TemporalRwTx: baseTx}, missing) + require.NoError(t, err) + require.Empty(t, value) + + _, ok := stateCache.View(nil).Get(kv.AccountsDomain, missing) + require.False(t, ok, "no exact frontier means no fill") +} + +// The binding is asserted at the real wiring point, so no future call site +// can wire a fill-enabled cache while forgetting the aggregator guard. +func TestSetStateCacheRequiresBoundAggregator(t *testing.T) { + ctx := t.Context() + db := newTestDb(t, 16) + rwTx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer rwTx.Rollback() + domains, err := execctx.NewSharedDomains(ctx, rwTx, log.New()) + require.NoError(t, err) + defer domains.Close() + + unbound := newSmallStateCache() + t.Cleanup(unbound.Close) + require.Panics(t, func() { domains.SetStateCache(unbound) }, + "wiring a fill-enabled cache without a bound aggregator must fail loudly") + + bound := newSmallStateCache() + t.Cleanup(bound.Close) + f := &fakeForbidder{} + bound.BindAggregator(fakeTemporalDB{agg: f}) + require.True(t, f.called) + require.NotPanics(t, func() { domains.SetStateCache(bound) }) +} + +// Apply-only mode must not pay for fills it will never make: the plain miss +// path used to box a frontier only for the fill to no-op. +func TestApplyOnlyMissPathBindsNoFrontier(t *testing.T) { + t.Setenv("STATE_CACHE_FILLS", "false") + + ctx := t.Context() + db := newTestDb(t, 16) + rwTx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer rwTx.Rollback() + domains, err := execctx.NewSharedDomains(ctx, rwTx, log.New()) + require.NoError(t, err) + defer domains.Close() + stateCache := newSmallStateCache() + t.Cleanup(stateCache.Close) + domains.SetStateCacheForTest(stateCache) + + missing := make([]byte, 20) + missing[0] = 7 + allocs := testing.AllocsPerRun(100, func() { + v, _, err := domains.GetLatest(kv.AccountsDomain, rwTx, missing) + if err != nil || len(v) != 0 { + t.Fatalf("expected a clean negative read, got %x %v", v, err) + } + }) + require.Zero(t, allocs, "an apply-only cache must not bind a frontier on the miss path") } // An apply-only cache (STATE_CACHE_FILLS=false) has no fills for a lowered -// frontier to poison, so the guard must not constrain the aggregator. -func TestGuardAggregatorForCache_ApplyOnlySkips(t *testing.T) { +// frontier to poison, so the binding must not constrain the aggregator. +func TestBindAggregator_ApplyOnlySkips(t *testing.T) { t.Setenv("STATE_CACHE_FILLS", "false") sc := newSmallStateCache() t.Cleanup(sc.Close) f := &fakeForbidder{} - execctx.GuardAggregatorForCache(fakeHasAgg{f}, sc) + sc.BindAggregator(fakeTemporalDB{agg: f}) require.False(t, f.called) } diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go new file mode 100644 index 00000000000..2003a06553a --- /dev/null +++ b/execution/cache/apply_all_test.go @@ -0,0 +1,156 @@ +// 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 cache + +import ( + "encoding/binary" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/crypto" + "github.com/erigontech/erigon/db/kv" +) + +func applyAllTestCache(t *testing.T) *StateCache { + t.Helper() + c := NewStateCache(1<<20, 1<<20, 1<<20, 1<<20) + t.Cleanup(c.Close) + return c +} + +// ApplyAll must be observationally identical to per-key Apply: same entries, +// same deletions and cascades, same frontier advance (so the same fills are +// rejected afterwards). Only the locking is batched. +func TestApplierApplyAllMatchesPerKeyApply(t *testing.T) { + t.Parallel() + + addr := make([]byte, 20) + addr[0] = 1 + deleted := make([]byte, 20) + deleted[0] = 2 + slot := make([]byte, 52) + slot[0] = 3 + code := []byte{0x60, 0x00, 0x60, 0x00} + + updates := []Update{ + {Domain: kv.AccountsDomain, Key: append([]byte(nil), addr...), Val: []byte{1}, TxNum: 30}, + {Domain: kv.AccountsDomain, Key: append([]byte(nil), deleted...), Val: nil, TxNum: 31}, + {Domain: kv.StorageDomain, Key: append([]byte(nil), slot...), Val: []byte{7}, TxNum: 32}, + {Domain: kv.CodeDomain, Key: append([]byte(nil), addr...), Val: append([]byte(nil), code...), TxNum: 33}, + } + + perKey := applyAllTestCache(t) + for _, u := range updates { + perKey.Applier().Apply(u.Domain, u.Key, u.Val, u.TxNum) + } + batched := applyAllTestCache(t) + batched.Applier().ApplyAll(append([]Update(nil), updates...)) + + for name, c := range map[string]*StateCache{"per-key": perKey, "batched": batched} { + v, ok := c.View(nil).Get(kv.AccountsDomain, addr) + require.True(t, ok, name) + require.Equal(t, []byte{1}, v, name) + _, ok = c.View(nil).Get(kv.AccountsDomain, deleted) + require.False(t, ok, name) + v, ok = c.View(nil).Get(kv.StorageDomain, slot) + require.True(t, ok, name) + require.Equal(t, []byte{7}, v, name) + gotCode, ok := c.View(nil).GetCodeByHash(crypto.Keccak256(code)) + require.True(t, ok, name) + require.Equal(t, code, gotCode, name) + + staleView := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 20, true })) + staleKey := make([]byte, 20) + staleKey[0] = 9 + staleView.Fill(kv.AccountsDomain, staleKey, []byte{9}, 5) + _, ok = c.View(nil).Get(kv.AccountsDomain, staleKey) + require.False(t, ok, "%s: the batch apply must advance the frontier and reject stale fills", name) + } +} + +// One batch may span several chunks; entries on both sides of the chunk +// boundary must land. +func TestApplierApplyAllCrossesChunkBoundary(t *testing.T) { + t.Parallel() + + c := applyAllTestCache(t) + n := applyChunkSize + 3 + updates := make([]Update, 0, n) + for i := range n { + key := make([]byte, 20) + binary.BigEndian.PutUint32(key, uint32(i)) + updates = append(updates, Update{Domain: kv.AccountsDomain, Key: key, Val: []byte{1}, TxNum: uint64(i)}) + } + c.Applier().ApplyAll(updates) + + for _, i := range []int{0, applyChunkSize - 1, applyChunkSize, n - 1} { + key := make([]byte, 20) + binary.BigEndian.PutUint32(key, uint32(i)) + _, ok := c.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok, "index %d", i) + } +} + +// The admission counters distinguish surviving reader warming from rejected +// stale fills. +func TestFillAdmissionCounters(t *testing.T) { + t.Parallel() + + c := applyAllTestCache(t) + fresh := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 100, true })) + key := make([]byte, 20) + key[0] = 1 + fresh.Fill(kv.AccountsDomain, key, []byte{1}, 50) + require.EqualValues(t, 1, c.fillsAdmitted.Load()) + require.EqualValues(t, 0, c.fillsRejected.Load()) + + c.Applier().Apply(kv.AccountsDomain, key, []byte{2}, 200) + stale := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 100, true })) + stale.Fill(kv.AccountsDomain, key, []byte{1}, 50) + require.EqualValues(t, 1, c.fillsAdmitted.Load()) + require.EqualValues(t, 1, c.fillsRejected.Load()) +} + +func BenchmarkApplierApply(b *testing.B) { + for _, batched := range []bool{false, true} { + b.Run(fmt.Sprintf("batched=%t", batched), func(b *testing.B) { + c := NewStateCache(64<<20, 64<<20, 64<<20, 64<<20) + defer c.Close() + const n = 100_000 + updates := make([]Update, 0, n) + for i := range n { + key := make([]byte, 20) + binary.BigEndian.PutUint32(key, uint32(i)) + updates = append(updates, Update{Domain: kv.AccountsDomain, Key: key, Val: key[:8], TxNum: uint64(i)}) + } + applier := c.Applier() + b.ResetTimer() + for b.Loop() { + if batched { + applier.ApplyAll(updates) + } else { + for _, u := range updates { + applier.Apply(u.Domain, u.Key, u.Val, u.TxNum) + } + } + } + b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/n, "ns/update") + }) + } +} diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 276871b07cc..d9983b8e70f 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -18,9 +18,11 @@ package cache import ( "bytes" + "fmt" "math" "strings" "sync" + "sync/atomic" "github.com/c2h5oh/datasize" @@ -61,7 +63,18 @@ type StateCache struct { // admissionMu makes Apply's frontier advance + cache mutation atomic // against concurrent read-fills, which recheck freshness under RLock. admissionMu sync.RWMutex - appliedEnd [kv.DomainLen]uint64 + // appliedEnd is per domain, necessarily: a domain's frontier advances only + // on its own writes, so a single global applied end would reject every + // quiet domain's fills. + appliedEnd [kv.DomainLen]uint64 + // fillsAdmitted/fillsRejected count admission-gate outcomes, reported by + // PrintStatsAndReset — the lens on how much reader warming survives at a + // given commit cadence. + fillsAdmitted atomic.Uint64 + fillsRejected atomic.Uint64 + // aggBound records that BindAggregator ran; SetStateCache asserts it + // before wiring a fill-enabled cache. + aggBound atomic.Bool // disableFills (STATE_CACHE_FILLS=false) turns off every reader fill // (including the content-addressed ones), leaving applies as the only // writer ("apply-only" mode) — an A/B lever and an operational kill switch. @@ -260,8 +273,10 @@ func (c *StateCache) fillIfFresh(domain kv.Domain, key []byte, value []byte, rea c.admissionMu.RLock() defer c.admissionMu.RUnlock() if visibleEnd < c.appliedEnd[domain] { + c.fillsRejected.Add(1) return } + c.fillsAdmitted.Add(1) cache.PutIfAbsent(key, cloned, readTxNum) } @@ -280,8 +295,10 @@ func (c *StateCache) fillCodeIfFresh(key []byte, value []byte, readTxNum, visibl c.admissionMu.RLock() defer c.admissionMu.RUnlock() if visibleEnd < c.appliedEnd[kv.CodeDomain] || accountsVisibleEnd < c.appliedEnd[kv.AccountsDomain] { + c.fillsRejected.Add(1) return } + c.fillsAdmitted.Add(1) codeCache.PutWithCodeHashIfAbsent(key, cloned, codeHash, readTxNum) } @@ -311,6 +328,45 @@ func (c *StateCache) apply(domain kv.Domain, key, value []byte, txNum uint64) { c.admissionMu.Lock() defer c.admissionMu.Unlock() + c.applyLocked(cache, domain, key, value, txNum, codeHash) +} + +// applyChunkSize bounds one exclusive critical section of applyAll, so a huge +// batch apply never starves concurrent fills for its whole duration. +const applyChunkSize = 4096 + +func (c *StateCache) applyAll(updates []Update) { + for start := 0; start < len(updates); start += applyChunkSize { + chunk := updates[start:min(start+applyChunkSize, len(updates))] + var codeHashes [][]byte + for i := range chunk { + u := &chunk[i] + if u.Domain == kv.CodeDomain && len(u.Val) > 0 { + if codeHashes == nil { + codeHashes = make([][]byte, len(chunk)) + } + u.Val = bytes.Clone(u.Val) + codeHashes[i] = crypto.Keccak256(u.Val) + } + } + c.admissionMu.Lock() + for i := range chunk { + u := &chunk[i] + cache := c.caches[u.Domain] + if cache == nil { + continue + } + var codeHash []byte + if codeHashes != nil { + codeHash = codeHashes[i] + } + c.applyLocked(cache, u.Domain, u.Key, u.Val, u.TxNum, codeHash) + } + c.admissionMu.Unlock() + } +} + +func (c *StateCache) applyLocked(cache Cache, domain kv.Domain, key, value []byte, txNum uint64, codeHash []byte) { c.noteApplied(domain, txNum) switch domain { @@ -368,6 +424,29 @@ func (c *StateCache) clear() { } } +// BindAggregator forbids visibility lowering on db's aggregator for a +// fill-enabled cache: fill admission relies on view frontiers never +// decreasing. SharedDomains.SetStateCache asserts this binding, so wiring +// cannot forget it. The aggregator side is duck-typed (the concrete type +// lives in db/state, above this package) but load-bearing: an aggregator +// without the forbid fails loudly. A nil or apply-only cache needs no +// binding. +func (c *StateCache) BindAggregator(db kv.TemporalRwDB) { + if c == nil || !c.FillsEnabled() { + return + } + agg := db.Agg() + f, ok := agg.(interface{ ForbidVisibilityLowering() }) + if !ok { + panic(fmt.Sprintf("assert: fill-enabled StateCache bound to a DB whose aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg)) + } + f.ForbidVisibilityLowering() + c.aggBound.Store(true) +} + +// AggregatorBound reports whether BindAggregator ran. +func (c *StateCache) AggregatorBound() bool { return c.aggBound.Load() } + // Close releases every sub-cache's slot in the shared memory envelope so later // caches size against real concurrency. Idempotent. func (c *StateCache) Close() { @@ -413,6 +492,10 @@ func (c *StateCache) PrintStatsAndReset() { if c == nil { return } + admitted, rejected := c.fillsAdmitted.Swap(0), c.fillsRejected.Swap(0) + if admitted+rejected > 0 { + log.Info("[cache] fill admission", "admitted", admitted, "rejected", rejected) + } if acc, ok := c.caches[kv.AccountsDomain].(*DomainCache); ok { acc.PrintStatsAndReset("Account") } diff --git a/execution/cache/view.go b/execution/cache/view.go index 0de781920d6..c5727b1dbfc 100644 --- a/execution/cache/view.go +++ b/execution/cache/view.go @@ -21,15 +21,12 @@ import ( ) // Frontier reports the exclusive txNum bound of one transaction's read view -// per domain. ok=false means the view has no exact frontier for the domain -// (remote or history-disabled backends, dependency-clamped values views); -// fills sourced from such a view are skipped. +// per domain. ok=false means the backend has no exact frontier for the domain +// (remote, history-disabled); fills sourced from such a view are skipped. // -// An implementation may report a stale-low bound only for a coherent, -// monotonically extended view — then it merely over-rejects fills. A view -// serving mixed-age reads has no exact frontier and must answer ok=false. -// Overstating what the tx can currently read is never safe: admission rests -// on that. +// An implementation may report a stale-low bound — that only over-rejects +// fills — but must never overstate what its tx can currently read: admission +// safety rests on that. type Frontier interface { DomainVisibleEnd(domain kv.Domain) (visibleEnd uint64, ok bool) } @@ -47,11 +44,8 @@ func (f FrontierFunc) DomainVisibleEnd(domain kv.Domain) (uint64, bool) { return // inert: reads miss, fills no-op. // // A ReadView does not isolate reads: the cache holds latest-applied state, so -// a hit can be newer than the view — the same direction the exec overlay -// already serves. In the forward direction the cache's invariant is -// monotonicity (content never regresses behind the applied frontier), -// enforced on the fill side; unwinds invalidate by epoch and floor. -// Snapshot-isolated caching is kvcache's job (node/shards). +// a hit can be newer than the view. Snapshot-isolated caching is kvcache's +// job (node/shards). type ReadView struct { c *StateCache frontier Frontier @@ -105,16 +99,15 @@ func (v ReadView) GetAddrCodeHash(addr []byte) ([32]byte, bool) { return v.c.getAddrCodeHash(addr) } -// CanFill reports whether this view carries a frontier, i.e. Fill and -// SeedAddrCodeHash can admit values through it. -func (v ReadView) CanFill() bool { return v.c != nil && v.frontier != nil } +// CanFill reports whether fills can go through this view: it carries a +// frontier and fills are enabled. A frontier answering ok=false for a domain +// is still decided at fill time. +func (v ReadView) CanFill() bool { return v.c != nil && !v.c.disableFills && v.frontier != nil } // Fill offers a value read from this view without replacing an authoritative // entry. Admission is checked against the view's frontier for the domain; // views without an exact frontier skip the fill. A code fill also checks the -// accounts frontier: an addr-keyed code entry derives from the account — an -// account deletion drops it without advancing the code frontier — so a view -// that predates the deletion must not refill it (mirrors SeedAddrCodeHash). +// accounts frontier — see fillCodeIfFresh for why. func (v ReadView) Fill(domain kv.Domain, key []byte, value []byte, readTxNum uint64) { if v.c == nil || v.c.disableFills || v.frontier == nil { return @@ -178,6 +171,25 @@ func (a Applier) Apply(domain kv.Domain, key, value []byte, txNum uint64) { a.c.apply(domain, key, value, txNum) } +// Update is one authoritative committed tuple for ApplyAll. +type Update struct { + Domain kv.Domain + Key []byte + Val []byte + TxNum uint64 +} + +// ApplyAll is Apply over a batch: the write lock is taken once per chunk +// instead of once per key, bounding how long concurrent fills wait. Code +// values are cloned (and hashed) outside the lock; the updates slice is +// consumed and may be rewritten in place. +func (a Applier) ApplyAll(updates []Update) { + if a.c == nil { + return + } + a.c.applyAll(updates) +} + // Unwind invalidates, across all caches, entries reflecting state above // unwindToTxNum on a now-dead fork, and lowers the applied frontiers. func (a Applier) Unwind(unwindToTxNum uint64) { diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go index 9a5b3204633..fa43aef4f6b 100644 --- a/execution/execmodule/exec_module.go +++ b/execution/execmodule/exec_module.go @@ -255,7 +255,7 @@ func NewExecModule( stopNode func() error, ) *ExecModule { domainCache := newDomainStateCache(stateCacheBudget) - execctx.GuardAggregatorForCache(db, domainCache) + domainCache.BindAggregator(db) var codeStore *cache.CodeStore if dbg.UseCodeStore { codeStore = cache.NewCodeStore(cache.DefaultCodeStoreMemBytes, cache.DefaultCodeStoreTableBytes) @@ -702,7 +702,7 @@ func (e *ExecModule) Start(ctx context.Context, hook *stageloop.Hook) { } defer e.semaphore.Release(1) - if err := e.pipelineExecutor.ProcessFrozenBlocks(ctx, hook, e.onlySnapDownloadOnStart); err != nil { + if err := e.pipelineExecutor.ProcessFrozenBlocks(ctx, hook, e.onlySnapDownloadOnStart, e.stateCache, e.codeStore); err != nil { if !errors.Is(err, context.Canceled) { e.logger.Error("Could not start execution service", "err", err) } diff --git a/execution/execmodule/exec_module_internal_test.go b/execution/execmodule/exec_module_internal_test.go index f2420c0ff57..9892a600520 100644 --- a/execution/execmodule/exec_module_internal_test.go +++ b/execution/execmodule/exec_module_internal_test.go @@ -23,6 +23,11 @@ import ( "github.com/stretchr/testify/require" "github.com/erigontech/erigon/common/dbg" + "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/datadir" + "github.com/erigontech/erigon/db/kv" + "github.com/erigontech/erigon/db/kv/temporal/temporaltest" + "github.com/erigontech/erigon/execution/cache" ) // The module is the one owner of the domain state cache: callers pass a byte @@ -44,3 +49,45 @@ func TestNewDomainStateCacheRespectsUseStateCache(t *testing.T) { require.NotNil(t, scDefault, "zero budget means the production default, not no cache") scDefault.Close() } + +// Frozen-block startup processing must advance state through the cache like +// every other writer: its post-commit applies overwrite pre-catchup entries +// and advance the admission frontier, so a read view opened before catchup +// cannot refill stale values (issue 22925). +func TestFrozenBlocksSDWiredToStateCache(t *testing.T) { + t.Parallel() + + ctx := t.Context() + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + sc := cache.NewStateCache(1<<20, 1<<20, 1<<20, 1<<20) + t.Cleanup(sc.Close) + sc.BindAggregator(db) + + addr := make([]byte, 20) + addr[0] = 1 + stale := []byte{1} + sc.Applier().Apply(kv.AccountsDomain, addr, stale, 5) + + tx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer tx.Rollback() + pe := &PipelineExecutor{logger: log.New()} + sd, err := pe.newFrozenBlocksSD(ctx, tx, sc, nil) + require.NoError(t, err) + defer sd.Close() + + fresh := []byte{2} + sd.SetTxNum(20) + require.NoError(t, sd.DomainPut(kv.AccountsDomain, tx, addr, fresh, 20, nil)) + require.NoError(t, sd.Commit(ctx, tx)) + + got, ok := sc.View(nil).Get(kv.AccountsDomain, addr) + require.True(t, ok) + require.Equal(t, fresh, got, "catchup applies must reach the cache") + + preCatchup := sc.View(cache.FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })) + preCatchup.Fill(kv.AccountsDomain, addr, stale, 5) + got, ok = sc.View(nil).Get(kv.AccountsDomain, addr) + require.True(t, ok) + require.Equal(t, fresh, got, "a pre-catchup read view must not refill stale state") +} diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index c172ce51913..838594f261c 100644 --- a/execution/execmodule/executor.go +++ b/execution/execmodule/executor.go @@ -28,6 +28,7 @@ import ( "github.com/erigontech/erigon/db/kv" dbstate "github.com/erigontech/erigon/db/state" "github.com/erigontech/erigon/db/state/execctx" + "github.com/erigontech/erigon/execution/cache" "github.com/erigontech/erigon/execution/chain" "github.com/erigontech/erigon/execution/protocol/rules" "github.com/erigontech/erigon/execution/stagedsync" @@ -177,10 +178,25 @@ func (pe *PipelineExecutor) RunLoop(ctx context.Context, sd *execctx.SharedDomai return tx, sd, nil } +// newFrozenBlocksSD builds a SharedDomains for frozen-block processing wired +// to the module's caches: its post-commit applies overwrite pre-catchup cache +// entries and advance the admission frontier, so read views opened before +// catchup cannot refill stale state. +func (pe *PipelineExecutor) newFrozenBlocksSD(ctx context.Context, tx kv.TemporalRwTx, stateCache *cache.StateCache, codeStore *cache.CodeStore) (*execctx.SharedDomains, error) { + sd, err := execctx.NewSharedDomains(ctx, tx, pe.logger) + if err != nil { + return nil, err + } + sd.SetInMemHistoryReads(inMemHistoryReads) + sd.SetStateCache(stateCache) + sd.SetCodeStore(codeStore) + return sd, nil +} + // ProcessFrozenBlocks runs the pipeline over snapshot blocks at startup. // It downloads block files, then executes them in a hasMore loop until // all frozen blocks are processed. -func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stageloop.Hook, onlySnapDownload bool) error { +func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stageloop.Hook, onlySnapDownload bool, stateCache *cache.StateCache, codeStore *cache.CodeStore) error { sawZeroBlocksTimes := 0 tx, err := pe.db.BeginTemporalRw(ctx) if err != nil { @@ -203,12 +219,11 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage return tx.Commit() } - doms, err := execctx.NewSharedDomains(ctx, tx, pe.logger) + doms, err := pe.newFrozenBlocksSD(ctx, tx, stateCache, codeStore) if err != nil { return err } defer func() { doms.Close() }() // RunLoop rotates doms; close whichever is current at exit - doms.SetInMemHistoryReads(inMemHistoryReads) var finishStageBeforeSync uint64 if hook != nil { @@ -247,11 +262,10 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage return nil, nil, err } tx = newTx - newSD, err := execctx.NewSharedDomains(ctx, newTx, pe.logger) + newSD, err := pe.newFrozenBlocksSD(ctx, newTx, stateCache, codeStore) if err != nil { return nil, nil, err } - newSD.SetInMemHistoryReads(inMemHistoryReads) hook.NotifySyncState(newTx) return newTx, newSD, nil }, From b05da15a8416a1614f5987ef990d52e8853629a9 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 16:30:57 +0200 Subject: [PATCH 02/16] execution/cache: name the nil-aggregator case in BindAggregator's assert A DB whose Agg() returns nil (membatchwithdb's temporaldb) panicked with the type-mismatch message, reading 'aggregator lacks ForbidVisibilityLowering'. Same failure, clearer diagnosis. --- db/state/execctx/statecache_readfill_test.go | 4 ++++ execution/cache/state_cache.go | 3 +++ 2 files changed, 7 insertions(+) diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 382501b8b6d..9e6d4225840 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -458,6 +458,10 @@ func TestBindAggregator(t *testing.T) { t.Cleanup(sc2.Close) require.Panics(t, func() { sc2.BindAggregator(fakeTemporalDB{agg: struct{}{}}) }, "an aggregator without ForbidVisibilityLowering must fail loudly, not drop the binding") + require.PanicsWithValue(t, + "assert: fill-enabled StateCache bound to a DB without an aggregator — the visibility-lowering guard would be silently dropped", + func() { sc2.BindAggregator(fakeTemporalDB{}) }, + "a DB without an aggregator must name that case, not report a nil type mismatch") } type nilDebugRwTx struct { diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index d9983b8e70f..0a5b544dda2 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -436,6 +436,9 @@ func (c *StateCache) BindAggregator(db kv.TemporalRwDB) { return } agg := db.Agg() + if agg == nil { + panic("assert: fill-enabled StateCache bound to a DB without an aggregator — the visibility-lowering guard would be silently dropped") + } f, ok := agg.(interface{ ForbidVisibilityLowering() }) if !ok { panic(fmt.Sprintf("assert: fill-enabled StateCache bound to a DB whose aggregator %T lacks ForbidVisibilityLowering — the visibility-lowering guard would be silently dropped", agg)) From 5fa7d08fad6eb3868d7c4b76f294c90afac1b3c5 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 17:19:48 +0200 Subject: [PATCH 03/16] execution/cache, execution/execmodule: absorb snapshot publication into the admission fence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Downloaded state files publish through agg.OpenFolder without a single Apply: appliedEnd does not move, and even if it did, admission only rejects future fills — it cannot evict entries already inside. A cache entry filled before startup catchup and untouched by later execution served stale state indefinitely, and a plain clear is not enough because a read view opened before publication refills the cleared slot past a cold gate. Applier.AbsorbFilesExtension does both halves under one admission lock: advance appliedEnd to the new file visibility and drop every entry. ProcessFrozenBlocks calls it right after RunSnapshots, covering the execution loop, the onlySnapDownload return and the IsDomainAheadOfBlocks early return. File publication that stays within applied ranges (local segment building) is a strict no-op, so the every-merge path never churns the cache. --- execution/cache/apply_all_test.go | 53 +++++++++++++++++++++++++++++++ execution/cache/state_cache.go | 27 ++++++++++++++++ execution/cache/view.go | 12 +++++++ execution/execmodule/executor.go | 5 +++ 4 files changed, 97 insertions(+) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index 2003a06553a..687b1a60ae8 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -107,6 +107,59 @@ func TestApplierApplyAllCrossesChunkBoundary(t *testing.T) { } } +// Snapshot publication brings state that never flows through Apply, so +// nothing can overwrite entries it invalidates. Absorbing the extension must +// drop every entry and advance the admission frontiers, so pre-publication +// views cannot refill what was just dropped. +func TestAbsorbFilesExtension(t *testing.T) { + t.Parallel() + + c := applyAllTestCache(t) + key := make([]byte, 20) + key[0] = 1 + preView := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 10, true })) + preView.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok := c.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok, "pre-publication fill lands on a cold cache") + + c.Applier().AbsorbFilesExtension(FrontierFunc(func(kv.Domain) (uint64, bool) { return 50, true })) + + _, ok = c.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok, "absorbing the extension must drop pre-publication entries") + + preView.Fill(kv.AccountsDomain, key, []byte{1}, 5) + _, ok = c.View(nil).Get(kv.AccountsDomain, key) + require.False(t, ok, "a pre-publication view must not refill past the absorbed extension") + + postView := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 50, true })) + postView.Fill(kv.AccountsDomain, key, []byte{2}, 45) + got, ok := c.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok, "a post-publication view fills normally") + require.Equal(t, []byte{2}, got) +} + +// An extension that does not pass the applied frontier (files built from +// already-applied state) must not churn the cache. +func TestAbsorbFilesExtensionNoOpWhenCovered(t *testing.T) { + t.Parallel() + + c := applyAllTestCache(t) + key := make([]byte, 20) + key[0] = 1 + c.Applier().Apply(kv.AccountsDomain, key, []byte{1}, 100) + + c.Applier().AbsorbFilesExtension(FrontierFunc(func(d kv.Domain) (uint64, bool) { + if d == kv.AccountsDomain { + return 50, true + } + return 0, false + })) + + got, ok := c.View(nil).Get(kv.AccountsDomain, key) + require.True(t, ok, "an already-covered extension must not drop applied entries") + require.Equal(t, []byte{1}, got) +} + // The admission counters distinguish surviving reader warming from rejected // stale fills. func TestFillAdmissionCounters(t *testing.T) { diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 0a5b544dda2..66bee6c45ae 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -417,6 +417,10 @@ func (c *StateCache) noteApplied(domain kv.Domain, txNum uint64) { func (c *StateCache) clear() { c.admissionMu.Lock() defer c.admissionMu.Unlock() + c.clearLocked() +} + +func (c *StateCache) clearLocked() { for _, cache := range c.caches { if cache != nil { cache.Clear() @@ -424,6 +428,29 @@ func (c *StateCache) clear() { } } +// absorbFilesExtension reconciles the cache with state published by files +// rather than applies (snapshot download): entries invalidated that way are +// never overwritten, so when visibility passes a domain's applied frontier, +// drop every entry and advance the frontiers — pre-publication views cannot +// refill what was dropped. +func (c *StateCache) absorbFilesExtension(f Frontier) { + c.admissionMu.Lock() + defer c.admissionMu.Unlock() + extended := false + for _, domain := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain} { + if c.caches[domain] == nil { + continue + } + if end, ok := f.DomainVisibleEnd(domain); ok && end > c.appliedEnd[domain] { + c.appliedEnd[domain] = end + extended = true + } + } + if extended { + c.clearLocked() + } +} + // BindAggregator forbids visibility lowering on db's aggregator for a // fill-enabled cache: fill admission relies on view frontiers never // decreasing. SharedDomains.SetStateCache asserts this binding, so wiring diff --git a/execution/cache/view.go b/execution/cache/view.go index c5727b1dbfc..961113084aa 100644 --- a/execution/cache/view.go +++ b/execution/cache/view.go @@ -190,6 +190,18 @@ func (a Applier) ApplyAll(updates []Update) { a.c.applyAll(updates) } +// AbsorbFilesExtension reconciles the cache with state published by files +// rather than applies (snapshot download): when visibility passes a domain's +// applied frontier, every entry is dropped and the frontiers advance, so +// pre-publication views cannot refill them. A no-op when visibility stays +// within what applies covered. +func (a Applier) AbsorbFilesExtension(f Frontier) { + if a.c == nil || f == nil { + return + } + a.c.absorbFilesExtension(f) +} + // Unwind invalidates, across all caches, entries reflecting state above // unwindToTxNum on a now-dead fork, and lowers the applied frontiers. func (a Applier) Unwind(unwindToTxNum uint64) { diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index 838594f261c..f0c7e6a6389 100644 --- a/execution/execmodule/executor.go +++ b/execution/execmodule/executor.go @@ -210,6 +210,11 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage if err := pe.sync.RunSnapshots(nil, tx); err != nil { return err } + // Downloaded state files publish without applies; reconcile the cache + // before anything reads through it. + stateCache.Applier().AbsorbFilesExtension(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) { + return tx.Debug().DomainVisibleEnd(domain) + })) if onlySnapDownload { return nil } From 75728a28e30d7da2076471f9ccbc0a83942231db Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 17:25:02 +0200 Subject: [PATCH 04/16] execution/execmodule: evict the code store on the catch-up prune path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Attaching the CodeStore to frozen-block catchup commits writes every deployed contract's code into TblCodeCache, and Evict is the only cap enforcement — previously it ran only on the FCU prune path, which a node does not reach until catchup completes, so a full-chain catchup could grow the table far past its byte cap. Mirror the forkchoice prune callback's eviction in the catch-up PruneFn. No new test: the eviction mechanics, including the restart re-seeding of the byte counter that a long catchup exercises, are pinned by TestCodeStore_TwoTierAndEvict; the call site mirrors the proven forkchoice pattern, and pinning it directly would need an injectable table cap plus a full pipeline harness. --- execution/execmodule/executor.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index f0c7e6a6389..81c0b4f3322 100644 --- a/execution/execmodule/executor.go +++ b/execution/execmodule/executor.go @@ -244,6 +244,11 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage tx, doms, err = pe.RunLoop(ctx, doms, tx, RunLoopConfig{ InitialCycle: true, PruneFn: func(ctx context.Context, initialCycle bool, rwtx kv.TemporalRwTx, sd *execctx.SharedDomains) error { + if codeStore != nil { + if err := codeStore.Evict(rwtx); err != nil { + return err + } + } return pe.sync.RunPrune(ctx, rwtx, initialCycle, 0) }, CommitCycle: func(ctx context.Context, hasMore bool, sd *execctx.SharedDomains) (kv.TemporalRwTx, *execctx.SharedDomains, error) { From 72bbd6b007a50bc928c7c14f91c7c70fd4313b92 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 17:27:23 +0200 Subject: [PATCH 05/16] execution/cache: count addr-codehash seed admission outcomes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit seedAddrCodeHash runs the same accounts-frontier admission check as the fill functions but reported nothing, so code-heavy workloads could reject seeds in volume while the stats showed zero rejections — the exact silent signal the counters exist to reveal. Count both outcomes where the decision is made; FillCodeSize stays uncounted since it is content-addressed and makes no admission decision. --- execution/cache/apply_all_test.go | 6 ++++++ execution/cache/state_cache.go | 2 ++ 2 files changed, 8 insertions(+) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index 687b1a60ae8..8cd070b8ed4 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -178,6 +178,12 @@ func TestFillAdmissionCounters(t *testing.T) { stale.Fill(kv.AccountsDomain, key, []byte{1}, 50) require.EqualValues(t, 1, c.fillsAdmitted.Load()) require.EqualValues(t, 1, c.fillsRejected.Load()) + + stale.SeedAddrCodeHash(key, [32]byte{7}, 50) + require.EqualValues(t, 2, c.fillsRejected.Load(), "a rejected addr-codehash seed must count") + fresh2 := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 300, true })) + fresh2.SeedAddrCodeHash(key, [32]byte{7}, 250) + require.EqualValues(t, 2, c.fillsAdmitted.Load(), "an admitted addr-codehash seed must count") } func BenchmarkApplierApply(b *testing.B) { diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 66bee6c45ae..a94aa76b00b 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -228,8 +228,10 @@ func (c *StateCache) seedAddrCodeHash(addr []byte, h [32]byte, txNum, visibleEnd c.admissionMu.RLock() defer c.admissionMu.RUnlock() if visibleEnd < c.appliedEnd[kv.AccountsDomain] { + c.fillsRejected.Add(1) return } + c.fillsAdmitted.Add(1) cc.PutAddrCodeHash(addr, h, txNum) } From 67177e063d27c13ba226792703b5c1968457ebfb Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 17:40:41 +0200 Subject: [PATCH 06/16] execution/cache: keep the fill counters off the hot read line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every fill RMWs a counter, and the counters landed on the cache line holding appliedEnd[Storage] and appliedEnd[Code] — turning a line that concurrent fills only read into one that ping-pongs between cores, handing part of the batched-apply contention win back to telemetry. Group the fields the fill path reads (appliedEnd, disableFills, aggBound) ahead of a 64-byte pad and put the write-hot counters behind it. A layout test pins the separation with unsafe.Offsetof so a field reorder cannot silently reintroduce the coupling. --- execution/cache/apply_all_test.go | 19 +++++++++++++++++++ execution/cache/state_cache.go | 13 ++++++++----- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index 8cd070b8ed4..29c5e7e5ce5 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -20,6 +20,7 @@ import ( "encoding/binary" "fmt" "testing" + "unsafe" "github.com/stretchr/testify/require" @@ -213,3 +214,21 @@ func BenchmarkApplierApply(b *testing.B) { }) } } + +// Every fill RMWs the admission counters; every fill also reads appliedEnd +// and disableFills. If they share a cache line, the counters invalidate it +// for every concurrent fill — handing back the contention the batched apply +// work bought. +func TestFillCountersLiveOnTheirOwnCacheLine(t *testing.T) { + var c StateCache + const line = 64 + countersFirst := unsafe.Offsetof(c.fillsAdmitted) / line + countersLast := (unsafe.Offsetof(c.fillsRejected) + 7) / line + appliedEndFirst := unsafe.Offsetof(c.appliedEnd) / line + appliedEndLast := (unsafe.Offsetof(c.appliedEnd) + uintptr(len(c.appliedEnd))*8 - 1) / line + require.True(t, countersFirst > appliedEndLast || countersLast < appliedEndFirst, + "fill counters must not share a cache line with appliedEnd") + disableFillsLine := unsafe.Offsetof(c.disableFills) / line + require.True(t, disableFillsLine < countersFirst || disableFillsLine > countersLast, + "fill counters must not share a cache line with disableFills") +} diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index a94aa76b00b..8d75ce63550 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -67,11 +67,6 @@ type StateCache struct { // on its own writes, so a single global applied end would reject every // quiet domain's fills. appliedEnd [kv.DomainLen]uint64 - // fillsAdmitted/fillsRejected count admission-gate outcomes, reported by - // PrintStatsAndReset — the lens on how much reader warming survives at a - // given commit cadence. - fillsAdmitted atomic.Uint64 - fillsRejected atomic.Uint64 // aggBound records that BindAggregator ran; SetStateCache asserts it // before wiring a fill-enabled cache. aggBound atomic.Bool @@ -79,6 +74,14 @@ type StateCache struct { // (including the content-addressed ones), leaving applies as the only // writer ("apply-only" mode) — an A/B lever and an operational kill switch. disableFills bool + // The pad keeps the counters — RMWed by every fill — off the cache line + // of the read-mostly fields above, which every fill reads. + _ [64]byte + // fillsAdmitted/fillsRejected count admission-gate outcomes, reported by + // PrintStatsAndReset — the lens on how much reader warming survives at a + // given commit cadence. + fillsAdmitted atomic.Uint64 + fillsRejected atomic.Uint64 } // NewStateCache creates a new StateCache with the specified byte capacities. From 91ccda89ea1acbb3c788337377ed9288c27edf6b Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 20:26:27 +0200 Subject: [PATCH 07/16] db/state/execctx: assert the apply-only miss path as a difference MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit require.Zero pinned the whole read path's allocation count to the frontier-boxing claim: any future allocation anywhere in the miss path would fail with a misleading message. Measure the same negative read with and without an apply-only cache and assert the difference is zero — only the cache attachment itself can fail it. Still red without the FillsEnabled gate (verified by reverting it). --- db/state/execctx/statecache_readfill_test.go | 44 ++++++++++++-------- 1 file changed, 26 insertions(+), 18 deletions(-) diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 9e6d4225840..548c0f21426 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -524,31 +524,39 @@ func TestSetStateCacheRequiresBoundAggregator(t *testing.T) { } // Apply-only mode must not pay for fills it will never make: the plain miss -// path used to box a frontier only for the fill to no-op. +// path used to box a frontier only for the fill to no-op. Asserted as a +// difference against the cache-less read, so unrelated allocations elsewhere +// in the read path cannot fail this test. func TestApplyOnlyMissPathBindsNoFrontier(t *testing.T) { t.Setenv("STATE_CACHE_FILLS", "false") ctx := t.Context() db := newTestDb(t, 16) - rwTx, err := db.BeginTemporalRw(ctx) - require.NoError(t, err) - defer rwTx.Rollback() - domains, err := execctx.NewSharedDomains(ctx, rwTx, log.New()) - require.NoError(t, err) - defer domains.Close() - stateCache := newSmallStateCache() - t.Cleanup(stateCache.Close) - domains.SetStateCacheForTest(stateCache) - missing := make([]byte, 20) - missing[0] = 7 - allocs := testing.AllocsPerRun(100, func() { - v, _, err := domains.GetLatest(kv.AccountsDomain, rwTx, missing) - if err != nil || len(v) != 0 { - t.Fatalf("expected a clean negative read, got %x %v", v, err) + missAllocs := func(withCache bool) float64 { + rwTx, err := db.BeginTemporalRw(ctx) + require.NoError(t, err) + defer rwTx.Rollback() + domains, err := execctx.NewSharedDomains(ctx, rwTx, log.New()) + require.NoError(t, err) + defer domains.Close() + if withCache { + stateCache := newSmallStateCache() + t.Cleanup(stateCache.Close) + domains.SetStateCacheForTest(stateCache) } - }) - require.Zero(t, allocs, "an apply-only cache must not bind a frontier on the miss path") + missing := make([]byte, 20) + missing[0] = 7 + return testing.AllocsPerRun(100, func() { + v, _, err := domains.GetLatest(kv.AccountsDomain, rwTx, missing) + if err != nil || len(v) != 0 { + t.Fatalf("expected a clean negative read, got %x %v", v, err) + } + }) + } + + require.Equal(t, missAllocs(false), missAllocs(true), + "an apply-only cache must add no allocations to the miss path") } // An apply-only cache (STATE_CACHE_FILLS=false) has no fills for a lowered From 1890f349ed67207798d68f8c66180547e67389cf Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 20:29:20 +0200 Subject: [PATCH 08/16] execution/cache: assert only eviction-safe indices in the chunk-boundary test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test assumed capacity for all 4099 entries, but the LRU grows into the process-global cachebudget envelope: under pressure (CI memory, race-detector inflation, parallel tests holding reservations) Reserve is denied, the cache stays near its start size and the oldest entries are legitimately evicted — index 0 failed across CI shards while the chunking was correct. Assert the seam-spanning tail indices, which are inserted last and survive any plausible capacity. --- execution/cache/apply_all_test.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index 29c5e7e5ce5..a86d3ca63d6 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -86,7 +86,10 @@ func TestApplierApplyAllMatchesPerKeyApply(t *testing.T) { } // One batch may span several chunks; entries on both sides of the chunk -// boundary must land. +// boundary must land. Only the last-inserted indices are asserted: the LRU +// grows into the process-global memory envelope, so under pressure (CI, race +// detector, parallel tests) early entries can be legitimately evicted — +// presence of index 0 is not a property of chunking. func TestApplierApplyAllCrossesChunkBoundary(t *testing.T) { t.Parallel() @@ -100,7 +103,7 @@ func TestApplierApplyAllCrossesChunkBoundary(t *testing.T) { } c.Applier().ApplyAll(updates) - for _, i := range []int{0, applyChunkSize - 1, applyChunkSize, n - 1} { + for _, i := range []int{applyChunkSize - 1, applyChunkSize, n - 1} { key := make([]byte, 20) binary.BigEndian.PutUint32(key, uint32(i)) _, ok := c.View(nil).Get(kv.AccountsDomain, key) From 9a4888437eef876edbf5bb978d219121f796edf8 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 21:01:12 +0200 Subject: [PATCH 09/16] execution/cache: ApplyAll no longer rewrites the caller's slice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The code-value clone landed in the caller's Update via u.Val = bytes.Clone(u.Val), forcing a 'may be rewritten in place' contract on an exported method. Carry the clone in a codeVals slice parallel to codeHashes instead; the caller's slice is never written, and the contract clause is gone. The clone stays even though Commit already deep-copies (code values are copied twice on that path): dropping it would trade a copy of rare, small data for an aliasing obligation on every ApplyAll caller. Pinned by pointer identity — require.Same on unsafe.SliceData, since require.Equal dereferences and passes on equal pointees. --- execution/cache/apply_all_test.go | 5 ++++- execution/cache/state_cache.go | 15 ++++++++------- execution/cache/view.go | 4 ++-- 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index a86d3ca63d6..fcde044abb2 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -61,7 +61,10 @@ func TestApplierApplyAllMatchesPerKeyApply(t *testing.T) { perKey.Applier().Apply(u.Domain, u.Key, u.Val, u.TxNum) } batched := applyAllTestCache(t) - batched.Applier().ApplyAll(append([]Update(nil), updates...)) + batchedUpdates := append([]Update(nil), updates...) + batched.Applier().ApplyAll(batchedUpdates) + require.Same(t, unsafe.SliceData(updates[3].Val), unsafe.SliceData(batchedUpdates[3].Val), + "ApplyAll must not rewrite the caller's updates") for name, c := range map[string]*StateCache{"per-key": perKey, "batched": batched} { v, ok := c.View(nil).Get(kv.AccountsDomain, addr) diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 8d75ce63550..5940cfe906f 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -343,15 +343,16 @@ const applyChunkSize = 4096 func (c *StateCache) applyAll(updates []Update) { for start := 0; start < len(updates); start += applyChunkSize { chunk := updates[start:min(start+applyChunkSize, len(updates))] - var codeHashes [][]byte + var codeVals, codeHashes [][]byte for i := range chunk { u := &chunk[i] if u.Domain == kv.CodeDomain && len(u.Val) > 0 { if codeHashes == nil { + codeVals = make([][]byte, len(chunk)) codeHashes = make([][]byte, len(chunk)) } - u.Val = bytes.Clone(u.Val) - codeHashes[i] = crypto.Keccak256(u.Val) + codeVals[i] = bytes.Clone(u.Val) + codeHashes[i] = crypto.Keccak256(codeVals[i]) } } c.admissionMu.Lock() @@ -361,11 +362,11 @@ func (c *StateCache) applyAll(updates []Update) { if cache == nil { continue } - var codeHash []byte - if codeHashes != nil { - codeHash = codeHashes[i] + val, codeHash := u.Val, []byte(nil) + if codeHashes != nil && codeHashes[i] != nil { + val, codeHash = codeVals[i], codeHashes[i] } - c.applyLocked(cache, u.Domain, u.Key, u.Val, u.TxNum, codeHash) + c.applyLocked(cache, u.Domain, u.Key, val, u.TxNum, codeHash) } c.admissionMu.Unlock() } diff --git a/execution/cache/view.go b/execution/cache/view.go index 961113084aa..59186bd2b89 100644 --- a/execution/cache/view.go +++ b/execution/cache/view.go @@ -181,8 +181,8 @@ type Update struct { // ApplyAll is Apply over a batch: the write lock is taken once per chunk // instead of once per key, bounding how long concurrent fills wait. Code -// values are cloned (and hashed) outside the lock; the updates slice is -// consumed and may be rewritten in place. +// values are cloned (and hashed) outside the lock; the caller's slice is +// not modified. func (a Applier) ApplyAll(updates []Update) { if a.c == nil { return From 8a4852916fb11309817f79fb77531f8f0f7d316b Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 21:08:23 +0200 Subject: [PATCH 10/16] db/state/execctx: Flush on a cache-attached SD panics like the neighbouring assert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flush answered one wiring bug with an error while SetStateCache answers the adjacent one with a panic. Both are programmer errors that silently corrupt cache coherence if allowed to proceed, and the error variant can be swallowed by errcheck suppression or a log-and-continue — converting a loud first-CI-run failure back into silent stale reads. Escalate the wiring branch to a panic; Flush keeps its error return for the real flushMem error paths. --- db/state/execctx/domain_shared.go | 5 +++-- db/state/execctx/statecache_readfill_test.go | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index 9c83ca2a905..1a2e85ead37 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -972,10 +972,11 @@ func (sd *SharedDomains) Close() { // does not touch the caches — the caller may still roll back. An SD with a // state cache must route every flush through Commit: a plain Flush would // leave the cache serving pre-flush values for the flushed keys forever, so -// it is rejected here. +// it panics here, like the SetStateCache assert for the neighbouring wiring +// bug — an error return can be swallowed. func (sd *SharedDomains) Flush(ctx context.Context, tx kv.RwTx) error { if sd.stateCache != nil { - return errors.New("SharedDomains with a state cache must flush through Commit") + panic("assert: SharedDomains with a state cache must flush through Commit") } defer mxFlushTook.ObserveDuration(time.Now()) return sd.flushMem(ctx, tx) diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 548c0f21426..0d92941053a 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -184,7 +184,8 @@ func TestFlushRejectsCacheAttachedSD(t *testing.T) { stateCache := newSmallStateCache() t.Cleanup(stateCache.Close) domains.SetStateCacheForTest(stateCache) - require.Error(t, domains.Flush(ctx, rwTx)) + require.Panics(t, func() { _ = domains.Flush(ctx, rwTx) }, + "a wiring bug must fail loudly, like the SetStateCache assert — an error can be swallowed") } // The incoherence the Flush rejection prevents, end to end: after v1 is @@ -229,7 +230,7 @@ func TestFlushRejectionPreventsStaleCachedReads(t *testing.T) { sd2.SetStateCacheForTest(stateCache) sd2.SetTxNum(20) require.NoError(t, sd2.DomainPut(kv.StorageDomain, tx2, slot, v2, 20, nil)) - require.Error(t, sd2.Flush(ctx, tx2), + require.Panics(t, func() { _ = sd2.Flush(ctx, tx2) }, "the step that would split the cache (v1) from MDBX (v2) must be rejected") require.NoError(t, sd2.Commit(ctx, tx2)) From c1dd19e2a7aa1a53631f123bafb4389346400ac8 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 21:14:57 +0200 Subject: [PATCH 11/16] execution/cache: count fill attempts dying on an inexact frontier admitted+rejected read as all fill attempts but undercounted: an attempt dies before the admission compare when the frontier answers ok=false (remote, history-disabled, dependency-clamped views), so the stats could show healthy admission while fills died wholesale one step earlier. A third bucket counts those at the three early returns; attempts that never happen (fills disabled, no frontier bound) stay uncounted by design. The counter sits behind the telemetry pad, so no new false sharing. --- execution/cache/apply_all_test.go | 8 ++++++++ execution/cache/state_cache.go | 18 ++++++++++-------- execution/cache/view.go | 3 +++ 3 files changed, 21 insertions(+), 8 deletions(-) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index fcde044abb2..254c5e4c499 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -191,6 +191,14 @@ func TestFillAdmissionCounters(t *testing.T) { fresh2 := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 300, true })) fresh2.SeedAddrCodeHash(key, [32]byte{7}, 250) require.EqualValues(t, 2, c.fillsAdmitted.Load(), "an admitted addr-codehash seed must count") + + inexact := c.View(FrontierFunc(func(kv.Domain) (uint64, bool) { return 0, false })) + inexact.Fill(kv.AccountsDomain, key, []byte{1}, 50) + inexact.SeedAddrCodeHash(key, [32]byte{7}, 50) + require.EqualValues(t, 2, c.fillsNoFrontier.Load(), + "fills dying on an inexact frontier must count — admitted+rejected alone undercounts attempts") + require.EqualValues(t, 2, c.fillsAdmitted.Load()) + require.EqualValues(t, 2, c.fillsRejected.Load()) } func BenchmarkApplierApply(b *testing.B) { diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 5940cfe906f..f2269699fa7 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -77,11 +77,13 @@ type StateCache struct { // The pad keeps the counters — RMWed by every fill — off the cache line // of the read-mostly fields above, which every fill reads. _ [64]byte - // fillsAdmitted/fillsRejected count admission-gate outcomes, reported by - // PrintStatsAndReset — the lens on how much reader warming survives at a - // given commit cadence. - fillsAdmitted atomic.Uint64 - fillsRejected atomic.Uint64 + // Fill-attempt outcomes, reported by PrintStatsAndReset — the lens on how + // much reader warming survives at a given commit cadence. noFrontier + // counts attempts dying on an inexact frontier (ok=false) before the + // admission compare; without it admitted+rejected undercounts attempts. + fillsAdmitted atomic.Uint64 + fillsRejected atomic.Uint64 + fillsNoFrontier atomic.Uint64 } // NewStateCache creates a new StateCache with the specified byte capacities. @@ -528,9 +530,9 @@ func (c *StateCache) PrintStatsAndReset() { if c == nil { return } - admitted, rejected := c.fillsAdmitted.Swap(0), c.fillsRejected.Swap(0) - if admitted+rejected > 0 { - log.Info("[cache] fill admission", "admitted", admitted, "rejected", rejected) + admitted, rejected, noFrontier := c.fillsAdmitted.Swap(0), c.fillsRejected.Swap(0), c.fillsNoFrontier.Swap(0) + if admitted+rejected+noFrontier > 0 { + log.Info("[cache] fill admission", "admitted", admitted, "rejected", rejected, "noFrontier", noFrontier) } if acc, ok := c.caches[kv.AccountsDomain].(*DomainCache); ok { acc.PrintStatsAndReset("Account") diff --git a/execution/cache/view.go b/execution/cache/view.go index 59186bd2b89..9875f9e0477 100644 --- a/execution/cache/view.go +++ b/execution/cache/view.go @@ -114,11 +114,13 @@ func (v ReadView) Fill(domain kv.Domain, key []byte, value []byte, readTxNum uin } visibleEnd, ok := v.frontier.DomainVisibleEnd(domain) if !ok { + v.c.fillsNoFrontier.Add(1) return } if domain == kv.CodeDomain { accountsEnd, ok := v.frontier.DomainVisibleEnd(kv.AccountsDomain) if !ok { + v.c.fillsNoFrontier.Add(1) return } v.c.fillCodeIfFresh(key, value, readTxNum, visibleEnd, accountsEnd) @@ -136,6 +138,7 @@ func (v ReadView) SeedAddrCodeHash(addr []byte, h [32]byte, txNum uint64) { } visibleEnd, ok := v.frontier.DomainVisibleEnd(kv.AccountsDomain) if !ok { + v.c.fillsNoFrontier.Add(1) return } v.c.seedAddrCodeHash(addr, h, txNum, visibleEnd) From 00cec03abce856a037e776f97537058479ce0f23 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 21:19:41 +0200 Subject: [PATCH 12/16] execution/cache: make AggregatorBound as nil-safe as BindAggregator --- db/state/execctx/statecache_readfill_test.go | 2 ++ execution/cache/state_cache.go | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/db/state/execctx/statecache_readfill_test.go b/db/state/execctx/statecache_readfill_test.go index 0d92941053a..11476a2b902 100644 --- a/db/state/execctx/statecache_readfill_test.go +++ b/db/state/execctx/statecache_readfill_test.go @@ -455,6 +455,8 @@ func TestBindAggregator(t *testing.T) { var nilCache *cache.StateCache require.NotPanics(t, func() { nilCache.BindAggregator(fakeTemporalDB{}) }, "no cache, no invariant to bind — the aggregator is never consulted") + require.NotPanics(t, func() { require.False(t, nilCache.AggregatorBound()) }, + "the query must be as nil-safe as the binding") sc2 := newSmallStateCache() t.Cleanup(sc2.Close) require.Panics(t, func() { sc2.BindAggregator(fakeTemporalDB{agg: struct{}{}}) }, diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index f2269699fa7..37adc1fa9f1 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -483,7 +483,7 @@ func (c *StateCache) BindAggregator(db kv.TemporalRwDB) { } // AggregatorBound reports whether BindAggregator ran. -func (c *StateCache) AggregatorBound() bool { return c.aggBound.Load() } +func (c *StateCache) AggregatorBound() bool { return c != nil && c.aggBound.Load() } // Close releases every sub-cache's slot in the shared memory envelope so later // caches size against real concurrency. Idempotent. From 5991d056cac4e6395599519aef2f73efa787774c Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:28:57 +0200 Subject: [PATCH 13/16] execution/commitment: add the missed BranchCache absorb unit test The watermark test was written with the boundary-hook commit but never staged (git add -u skips new files); it sat untracked, breaking compilation on sibling branches. --- .../commitment/branch_cache_absorb_test.go | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 execution/commitment/branch_cache_absorb_test.go diff --git a/execution/commitment/branch_cache_absorb_test.go b/execution/commitment/branch_cache_absorb_test.go new file mode 100644 index 00000000000..b215fb3bdc6 --- /dev/null +++ b/execution/commitment/branch_cache_absorb_test.go @@ -0,0 +1,49 @@ +// 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 commitment + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// Commitment files built from the process's own writes cover txNums at or +// below the put watermark and must not churn the cache; files from a snapshot +// download exceed it and must clear everything — a stale branch restores the +// trie to the wrong state. +func TestBranchCacheAbsorbFilesExtension(t *testing.T) { + t.Parallel() + + c := NewBranchCache(64) + t.Cleanup(c.Close) + prefix := []byte{0x01} + c.Put(prefix, []byte{0xbb}, 0, 100) + + c.AbsorbFilesExtension(101) + _, _, ok := c.Get(prefix) + require.True(t, ok, "files covering the process's own writes must not clear the cache") + + c.AbsorbFilesExtension(150) + _, _, ok = c.Get(prefix) + require.False(t, ok, "files beyond the put watermark carry foreign state — clear") + + c.Put(prefix, []byte{0xcc}, 0, 200) + c.AbsorbFilesExtension(150) + _, _, ok = c.Get(prefix) + require.True(t, ok, "an already-absorbed extension must not clear again") +} From eb6a44f318be54fa3535acdef9ddecdd3d096408 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:29:33 +0200 Subject: [PATCH 14/16] Revert "execution/commitment: add the missed BranchCache absorb unit test" This reverts commit 5991d056cac4e6395599519aef2f73efa787774c. --- .../commitment/branch_cache_absorb_test.go | 49 ------------------- 1 file changed, 49 deletions(-) delete mode 100644 execution/commitment/branch_cache_absorb_test.go diff --git a/execution/commitment/branch_cache_absorb_test.go b/execution/commitment/branch_cache_absorb_test.go deleted file mode 100644 index b215fb3bdc6..00000000000 --- a/execution/commitment/branch_cache_absorb_test.go +++ /dev/null @@ -1,49 +0,0 @@ -// 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 commitment - -import ( - "testing" - - "github.com/stretchr/testify/require" -) - -// Commitment files built from the process's own writes cover txNums at or -// below the put watermark and must not churn the cache; files from a snapshot -// download exceed it and must clear everything — a stale branch restores the -// trie to the wrong state. -func TestBranchCacheAbsorbFilesExtension(t *testing.T) { - t.Parallel() - - c := NewBranchCache(64) - t.Cleanup(c.Close) - prefix := []byte{0x01} - c.Put(prefix, []byte{0xbb}, 0, 100) - - c.AbsorbFilesExtension(101) - _, _, ok := c.Get(prefix) - require.True(t, ok, "files covering the process's own writes must not clear the cache") - - c.AbsorbFilesExtension(150) - _, _, ok = c.Get(prefix) - require.False(t, ok, "files beyond the put watermark carry foreign state — clear") - - c.Put(prefix, []byte{0xcc}, 0, 200) - c.AbsorbFilesExtension(150) - _, _, ok = c.Get(prefix) - require.True(t, ok, "an already-absorbed extension must not clear again") -} From 9ca2938c957a7f168b6e343a23e45730627e91e1 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:50:23 +0200 Subject: [PATCH 15/16] execution/cache: cover the third counter in the layout test; cache-driven absorb loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cache-line test derived its counter span from fillsRejected, so fillsNoFrontier — added one commit later — sat outside the asserted range; dormant while the counters trail the struct, but a reorder could overlap it with the hot line unnoticed. absorbFilesExtension now iterates every cached domain instead of a hardcoded trio: a future fill-capable domain missing from the list would silently not be fenced at publication. --- execution/cache/apply_all_test.go | 2 +- execution/cache/state_cache.go | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/execution/cache/apply_all_test.go b/execution/cache/apply_all_test.go index 254c5e4c499..9c76633d283 100644 --- a/execution/cache/apply_all_test.go +++ b/execution/cache/apply_all_test.go @@ -237,7 +237,7 @@ func TestFillCountersLiveOnTheirOwnCacheLine(t *testing.T) { var c StateCache const line = 64 countersFirst := unsafe.Offsetof(c.fillsAdmitted) / line - countersLast := (unsafe.Offsetof(c.fillsRejected) + 7) / line + countersLast := (unsafe.Offsetof(c.fillsNoFrontier) + 7) / line appliedEndFirst := unsafe.Offsetof(c.appliedEnd) / line appliedEndLast := (unsafe.Offsetof(c.appliedEnd) + uintptr(len(c.appliedEnd))*8 - 1) / line require.True(t, countersFirst > appliedEndLast || countersLast < appliedEndFirst, diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index 37adc1fa9f1..e4f8c9938f2 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -445,7 +445,8 @@ func (c *StateCache) absorbFilesExtension(f Frontier) { c.admissionMu.Lock() defer c.admissionMu.Unlock() extended := false - for _, domain := range []kv.Domain{kv.AccountsDomain, kv.StorageDomain, kv.CodeDomain} { + for d := range kv.DomainLen { + domain := kv.Domain(d) if c.caches[domain] == nil { continue } From f730805bea69830ae1ff53f60d0e2990fa4e0ec0 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Wed, 5 Aug 2026 23:56:39 +0200 Subject: [PATCH 16/16] execution/cache, execution/execmodule: three review nits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clone code bytes before hashing in fillCodeIfFresh, matching the apply paths, so stored bytes and codeHash cannot diverge under a reused caller buffer. Tolerate a nil debug backend in the absorb frontier (unreachable from the module's own DB; consistency with the execctx frontier lookups). Log fill-admission stats at Debug like the sibling cache stats — PrintCacheStats runs per commit cycle. --- execution/cache/state_cache.go | 6 ++++-- execution/execmodule/executor.go | 6 +++++- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/execution/cache/state_cache.go b/execution/cache/state_cache.go index e4f8c9938f2..4571612491c 100644 --- a/execution/cache/state_cache.go +++ b/execution/cache/state_cache.go @@ -297,8 +297,10 @@ func (c *StateCache) fillCodeIfFresh(key []byte, value []byte, readTxNum, visibl if !ok || len(value) == 0 { return } - codeHash := crypto.Keccak256(value) + // Clone before hashing, like the apply paths: the stored bytes and their + // codeHash cannot diverge even if the caller's buffer is reused mid-call. cloned := bytes.Clone(value) + codeHash := crypto.Keccak256(cloned) c.admissionMu.RLock() defer c.admissionMu.RUnlock() if visibleEnd < c.appliedEnd[kv.CodeDomain] || accountsVisibleEnd < c.appliedEnd[kv.AccountsDomain] { @@ -533,7 +535,7 @@ func (c *StateCache) PrintStatsAndReset() { } admitted, rejected, noFrontier := c.fillsAdmitted.Swap(0), c.fillsRejected.Swap(0), c.fillsNoFrontier.Swap(0) if admitted+rejected+noFrontier > 0 { - log.Info("[cache] fill admission", "admitted", admitted, "rejected", rejected, "noFrontier", noFrontier) + log.Debug("[cache] fill admission", "admitted", admitted, "rejected", rejected, "noFrontier", noFrontier) } if acc, ok := c.caches[kv.AccountsDomain].(*DomainCache); ok { acc.PrintStatsAndReset("Account") diff --git a/execution/execmodule/executor.go b/execution/execmodule/executor.go index 81c0b4f3322..04cb29b1343 100644 --- a/execution/execmodule/executor.go +++ b/execution/execmodule/executor.go @@ -213,7 +213,11 @@ func (pe *PipelineExecutor) ProcessFrozenBlocks(ctx context.Context, hook *stage // Downloaded state files publish without applies; reconcile the cache // before anything reads through it. stateCache.Applier().AbsorbFilesExtension(cache.FrontierFunc(func(domain kv.Domain) (uint64, bool) { - return tx.Debug().DomainVisibleEnd(domain) + dbgTx := tx.Debug() + if dbgTx == nil { + return 0, false + } + return dbgTx.DomainVisibleEnd(domain) })) if onlySnapDownload { return nil