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 },