From 495ffc62abf4c7fc1d8eb4fa4377cead81b08de5 Mon Sep 17 00:00:00 2001 From: yperbasis Date: Thu, 16 Jul 2026 16:27:53 +0200 Subject: [PATCH 1/2] db, execution, rpcdaemon: coherent version-keyed state cache Split out of #21293. Announce the post-commit PlainStateVersion from the pre-commit dispatch, rework the Coherent kvcache around fresh version-keyed roots with reader-shaped keys, and wire the standalone rpcdaemon to the coherent cache at every budget. The execmoduletester waits for the FCU background commit after InsertChain so bg-commit tests read committed state. --- cmd/rpcdaemon/cli/config.go | 22 +- cmd/rpcdaemon/cli/config_test.go | 61 +++ db/kv/kvcache/cache.go | 186 ++++----- db/kv/kvcache/cache_test.go | 393 +++++++++++++++--- db/kv/kvcache/simple.go | 1 + .../docs/fundamentals/modules/rpc-daemon.md | 2 +- docs/site/static/llms-full.txt | 2 +- execution/execmodule/exec_module_test.go | 101 ++++- .../execmoduletester/exec_module_tester.go | 11 +- execution/execmodule/forkchoice.go | 1 + .../execmodule/notification_dispatcher.go | 8 + execution/stagedsync/stageloop/stageloop.go | 7 +- llms-full.txt | 2 +- rpc/rpchelper/helper.go | 13 +- 14 files changed, 603 insertions(+), 207 deletions(-) diff --git a/cmd/rpcdaemon/cli/config.go b/cmd/rpcdaemon/cli/config.go index c45bdacfda7..30310d2af9a 100644 --- a/cmd/rpcdaemon/cli/config.go +++ b/cmd/rpcdaemon/cli/config.go @@ -138,7 +138,7 @@ func RootCommand() (*cobra.Command, *httpcfg.HttpCfg) { rootCmd.PersistentFlags().BoolVar(&cfg.GethCompatibility, "rpc.gethcompat", false, "Enables Geth-compatible storage iteration order for debug_storageRangeAt (sorted by keccak256 hash). Disabled by default for performance.") rootCmd.PersistentFlags().StringVar(&cfg.TxPoolApiAddr, "txpool.api.addr", "", "txpool api network address, for example: 127.0.0.1:9090 (default: use value of --private.api.addr)") - rootCmd.PersistentFlags().StringVar(&stateCacheStr, "state.cache", "0MB", "Amount of data to store in StateCache (enabled if no --datadir set). Set 0 to disable StateCache. Defaults to 0MB RAM") + rootCmd.PersistentFlags().StringVar(&stateCacheStr, "state.cache", "0MB", "Amount of data to store in the version-keyed StateCache (an equally-sized code cache is budgeted on top). Set 0 to disable entry retention while preserving snapshot-consistent reads") rootCmd.PersistentFlags().BoolVar(&cfg.GRPCServerEnabled, "grpc", false, "Enable GRPC server") rootCmd.PersistentFlags().StringVar(&cfg.GRPCListenAddress, "grpc.addr", nodecfg.DefaultGRPCHost, "GRPC server listening interface") rootCmd.PersistentFlags().IntVar(&cfg.GRPCPort, "grpc.port", nodecfg.DefaultGRPCPort, "GRPC server listening port") @@ -242,6 +242,13 @@ type StateChangesClient interface { StateChanges(ctx context.Context, in *remoteproto.StateChangeRequest, opts ...grpc.CallOption) (remoteproto.KV_StateChangesClient, error) } +func newRemoteStateCache(cfg kvcache.CoherentConfig) kvcache.Cache { + if cfg.CacheSize == 0 && cfg.CodeCacheSize == 0 { + cfg.WaitForNewBlock = false + } + return kvcache.New(cfg) +} + func subscribeToStateChangesLoop(ctx context.Context, client StateChangesClient, cache kvcache.Cache) { go func() { for { @@ -334,11 +341,8 @@ func EmbeddedServices(ctx context.Context, // the overlay is always current, has zero memory overhead, and // doesn't need the StateChanges gRPC stream to stay coherent. stateCache = stateCacheCfg.LocalCache - } else if stateCacheCfg.CacheSize > 0 { - // Remote RPCDaemon: use coherent cache fed by StateChanges stream. - stateCache = kvcache.New(stateCacheCfg) } else { - stateCache = kvcache.NewSimple() + stateCache = newRemoteStateCache(stateCacheCfg) } subscribeToStateChangesLoop(ctx, stateDiffClient, stateCache) @@ -530,7 +534,6 @@ func RemoteServices(ctx context.Context, cfg *httpcfg.HttpCfg, logger log.Logger if err != nil { return nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, err } - stateCache = kvcache.NewSimple() } // If DB can't be configured - used PrivateApiAddr as remote DB if db == nil { @@ -538,14 +541,11 @@ func RemoteServices(ctx context.Context, cfg *httpcfg.HttpCfg, logger log.Logger } if !cfg.WithDatadir { - if cfg.StateCache.CacheSize > 0 { - stateCache = kvcache.New(cfg.StateCache) - } else { - stateCache = kvcache.NewSimple() - } logger.Info("if you run RPCDaemon on same machine with Erigon add --datadir option") } + stateCache = newRemoteStateCache(cfg.StateCache) + subscribeToStateChangesLoop(ctx, remoteKvClient, stateCache) txpoolConn := conn diff --git a/cmd/rpcdaemon/cli/config_test.go b/cmd/rpcdaemon/cli/config_test.go index 2c6105c669e..f5732042ce2 100644 --- a/cmd/rpcdaemon/cli/config_test.go +++ b/cmd/rpcdaemon/cli/config_test.go @@ -24,11 +24,20 @@ import ( "github.com/holiman/uint256" "github.com/stretchr/testify/require" + "github.com/erigontech/erigon/common" "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/kvcache" + "github.com/erigontech/erigon/db/kv/temporal/temporaltest" + "github.com/erigontech/erigon/db/state/execctx" "github.com/erigontech/erigon/execution/chain" "github.com/erigontech/erigon/execution/protocol/rules/ethash" "github.com/erigontech/erigon/execution/protocol/rules/merge" "github.com/erigontech/erigon/execution/types" + "github.com/erigontech/erigon/execution/types/accounts" + "github.com/erigontech/erigon/node/gointerfaces" + "github.com/erigontech/erigon/node/gointerfaces/remoteproto" ) // TestIsWebsocket tests if an incoming websocket upgrade request is detected properly. @@ -70,3 +79,55 @@ func TestRemoteRulesEngineFinalizeDelegates(t *testing.T) { require.NoError(t, err) }) } + +func TestZeroBudgetRemoteCachePinsCommittedState(t *testing.T) { + cfg := kvcache.DefaultCoherentConfig + cfg.CacheSize = 0 + cfg.CodeCacheSize = 0 + cache := newRemoteStateCache(cfg) + + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + addr := common.Address{1} + committedAccount := accounts.Account{Nonce: 1, Balance: *uint256.NewInt(1), CodeHash: accounts.EmptyCodeHash} + announcedAccount := committedAccount + announcedAccount.Nonce = 2 + committedData := accounts.SerialiseV3(&committedAccount) + announcedData := accounts.SerialiseV3(&announcedAccount) + + require.NoError(t, db.UpdateTemporal(t.Context(), func(tx kv.TemporalRwTx) error { + domains, err := execctx.NewSharedDomains(t.Context(), tx, log.New()) + if err != nil { + return err + } + defer domains.Close() + if err := domains.DomainPut(kv.AccountsDomain, tx, addr[:], committedData, 0, nil); err != nil { + return err + } + return domains.Flush(t.Context(), tx) + })) + + tx, err := db.BeginTemporalRo(t.Context()) + require.NoError(t, err) + defer tx.Rollback() + stateVersion, err := tx.ReadSequence(string(kv.PlainStateVersion)) + require.NoError(t, err) + + cache.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: stateVersion + 1, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_UPSERT, + Address: gointerfaces.ConvertAddressToH160(addr), + Data: announcedData, + }}, + }}, + }) + require.Zero(t, cache.Len()) + + view, err := cache.View(t.Context(), tx) + require.NoError(t, err) + data, err := view.Get(addr[:]) + require.NoError(t, err) + require.Equal(t, committedData, data) +} diff --git a/db/kv/kvcache/cache.go b/db/kv/kvcache/cache.go index 0a7096b97da..11d4e8476f9 100644 --- a/db/kv/kvcache/cache.go +++ b/db/kv/kvcache/cache.go @@ -20,16 +20,13 @@ import ( "bytes" "context" - "encoding/binary" "fmt" - "hash" "sync" "sync/atomic" "time" "github.com/c2h5oh/datasize" - keccak "github.com/erigontech/fastkeccak" btree2 "github.com/tidwall/btree" "github.com/erigontech/erigon/common" @@ -58,6 +55,7 @@ type Cache interface { } type CacheView interface { Get(k []byte) ([]byte, error) + GetAsOf(key []byte, ts uint64) (v []byte, ok bool, err error) GetCode(k []byte) ([]byte, error) HasStorage(address common.Address) (bool, error) } @@ -66,24 +64,19 @@ type CacheView interface { // provide "Serializable Isolation Level" semantic: all data form consistent db view at moment // when read transaction started, read data are immutable until end of read transaction, reader can't see newer updates // -// Every time a new state change comes, we do the following: -// - Check that prevBlockHeight and prevBlockHash match what is the top values we have, and if they don't we -// invalidate the cache, because we missed some messages and cannot consider the cache coherent anymore. -// - Clone the cache pointer (such that the previous pointer is still accessible, but new one shared the content with it), -// apply state updates to the cloned cache pointer and save under the new identified made from blockHeight and blockHash. -// - If there is a conditional variable corresponding to the identifier, remove it from the map and notify conditional -// variable, waking up the read-only transaction waiting on it. -// -// On the other hand, whenever we have a cache miss (by looking at the top cache), we do the following: -// - Once read the current block height and block hash (canonical) from underlying db transaction -// - Construct the identifier from the current block height and block hash -// - Look for the constructed identifier in the cache. If the identifier is found, use the corresponding -// cache in conjunction with this read-only transaction (it will be consistent with it). If the identifier is -// not found, it means that the transaction has been committed in Erigon, but the state update has not -// arrived yet (as shown in the picture on the right). Insert conditional variable for this identifier and wait on -// it until either cache with the given identifier appears, or timeout (indicating that the cache update -// mechanism is broken and cache is likely invalidated). +// Roots are keyed by PlainStateVersion. OnNewBlock creates the canonical root +// for the announced version from that batch's changes alone; a reader whose +// version has no root yet waits up to NewBlockWait for the batch, then +// proceeds uncached. On a cache miss the reader consults its own transaction +// and, when its version is the latest known one, inserts the result for other +// same-version readers. // +// A canonical root deliberately does not inherit entries from its predecessor: +// the state-change producers do not announce every mutation (see +// https://github.com/erigontech/erigon/issues/22276), so carried entries could +// go stale with no later batch to correct them. Fresh roots bound any producer +// gap to one version — and a missed batch only costs cache warmth, never +// coherency, because the version gap simply leaves that root unfed. // Pair.Value == nil - is a marker of absense key in db @@ -91,18 +84,7 @@ type CacheView interface { // High-level guaranties: // - Keys/Values returned by cache are valid/immutable until end of db transaction // - CacheView is always coherent with given db transaction - -// -// Rules of set view.isCanonical value: -// - method View can't parent.Clone() - because parent view is not coherent with current kv.Tx -// - only OnNewBlock method may do parent.Clone() and apply StateChanges to create coherent view of kv.Tx -// - parent.Clone() can't be called if parent.isCanonical=false -// - only OnNewBlock method can set view.isCanonical=true -// -// Rules of filling cache.stateEvict: -// - changes in Canonical View SHOULD reflect in stateEvict -// - changes in Non-Canonical View SHOULD NOT reflect in stateEvict type Coherent struct { - hasher hash.Hash codeEvictLen metrics.Gauge codeKeys metrics.Gauge keys metrics.Gauge @@ -128,7 +110,6 @@ type CoherentRoot struct { ready chan struct{} // close when ready readyChanClosed atomic.Bool // quick check if ready channel is closed closeOnce sync.Once // protecting `ready` field from double-close - isCanonical bool } // CoherentView - dumb object, which proxy all requests to Coherent object. @@ -142,6 +123,7 @@ type CoherentView struct { func (c *CoherentView) Get(k []byte) ([]byte, error) { return c.cache.Get(k, c.tx, c.stateVersionID) } + func (c *CoherentView) GetAsOf(key []byte, ts uint64) (v []byte, ok bool, err error) { return nil, false, nil } @@ -198,7 +180,6 @@ func New(cfg CoherentConfig) *Coherent { roots: map[uint64]*CoherentRoot{}, stateEvict: &ThreadSafeEvictionList{l: NewList()}, codeEvict: &ThreadSafeEvictionList{l: NewList()}, - hasher: keccak.NewFastKeccak(), cfg: cfg, miss: metrics.GetOrCreateCounter(fmt.Sprintf(`cache_total{result="miss",name="%s"}`, cfg.MetricsLabel)), hits: metrics.GetOrCreateCounter(fmt.Sprintf(`cache_total{result="hit",name="%s"}`, cfg.MetricsLabel)), @@ -212,6 +193,14 @@ func New(cfg CoherentConfig) *Coherent { } } +func newCoherentRoot() *CoherentRoot { + return &CoherentRoot{ + ready: make(chan struct{}), + cache: btree2.NewBTreeG(Less), + codeCache: btree2.NewBTreeG(Less), + } +} + // selectOrCreateRoot - used for usual getting root func (c *Coherent) selectOrCreateRoot(versionID uint64) *CoherentRoot { c.lock.Lock() @@ -221,11 +210,7 @@ func (c *Coherent) selectOrCreateRoot(versionID uint64) *CoherentRoot { return r } - r = &CoherentRoot{ - ready: make(chan struct{}), - cache: btree2.NewBTreeG(Less), - codeCache: btree2.NewBTreeG(Less), - } + r = newCoherentRoot() c.roots[versionID] = r return r } @@ -240,38 +225,21 @@ func (c *Coherent) advanceRoot(stateVersionID uint64) (r *CoherentRoot) { } if !rootExists { - r = &CoherentRoot{ready: make(chan struct{})} + r = newCoherentRoot() c.roots[stateVersionID] = r } - if prevView, ok := c.roots[stateVersionID-1]; ok && prevView.isCanonical { - //log.Info("advance: clone", "from", viewID-1, "to", viewID) - r.cache = prevView.cache.Copy() - r.codeCache = prevView.codeCache.Copy() - } else { - c.stateEvict.Init() - c.codeEvict.Init() - if r.cache == nil { - //log.Info("advance: new", "to", viewID) - r.cache = btree2.NewBTreeG(Less) - r.codeCache = btree2.NewBTreeG(Less) - } else { - r.cache.Walk(func(items []*Element) bool { - for _, i := range items { - c.stateEvict.PushFront(i) - } - return true - }) - r.codeCache.Walk(func(items []*Element) bool { - for _, i := range items { - c.codeEvict.PushFront(i) - } - return true - }) - } + // No carry-over from the previous canonical root: the state-change + // producers don't announce every mutation (account deletions, code on + // unwind — https://github.com/erigontech/erigon/issues/22276), so + // inherited entries could stay stale forever. Fresh roots bound any + // producer gap to one version. + for _, root := range c.roots { + root.cache.Clear() + root.codeCache.Clear() } - r.isCanonical = true - + c.stateEvict.Init() + c.codeEvict.Init() c.evictRoots() c.latestStateVersionID = stateVersionID c.latestStateView = r @@ -292,40 +260,31 @@ func (c *Coherent) OnNewBlock(stateChanges *remoteproto.StateChangeBatch) { for _, sc := range stateChanges.ChangeBatch { for i := range sc.Changes { + // Code and storage keys must match what readers look up: code is + // keyed by account address (the E3 CodeDomain key) and storage by + // address+location — see state.CachedReader3. + addr := gointerfaces.ConvertH160toAddress(sc.Changes[i].Address) switch sc.Changes[i].Action { case remoteproto.Action_UPSERT: - addr := gointerfaces.ConvertH160toAddress(sc.Changes[i].Address) - v := sc.Changes[i].Data - c.add(addr[:], v, r, id) + c.add(addr[:], sc.Changes[i].Data, r, id) case remoteproto.Action_UPSERT_CODE: - addr := gointerfaces.ConvertH160toAddress(sc.Changes[i].Address) - v := sc.Changes[i].Data - c.add(addr[:], v, r, id) - c.hasher.Reset() - c.hasher.Write(sc.Changes[i].Code) - k := c.hasher.Sum(nil) - c.addCode(k, sc.Changes[i].Code, r, id) + c.add(addr[:], sc.Changes[i].Data, r, id) + c.addCode(addr[:], sc.Changes[i].Code, r, id) case remoteproto.Action_REMOVE: - addr := gointerfaces.ConvertH160toAddress(sc.Changes[i].Address) c.add(addr[:], nil, r, id) case remoteproto.Action_STORAGE: //skip, will check later case remoteproto.Action_CODE: - c.hasher.Reset() - c.hasher.Write(sc.Changes[i].Code) - k := c.hasher.Sum(nil) - c.addCode(k, sc.Changes[i].Code, r, id) + c.addCode(addr[:], sc.Changes[i].Code, r, id) default: panic("not implemented yet") } if c.cfg.WithStorage && len(sc.Changes[i].StorageChanges) > 0 { - addr := gointerfaces.ConvertH160toAddress(sc.Changes[i].Address) for _, change := range sc.Changes[i].StorageChanges { loc := gointerfaces.ConvertH256ToHash(change.Location) - k := make([]byte, 20+8+32) + k := make([]byte, 20+32) copy(k, addr[:]) - binary.BigEndian.PutUint64(k[20:], sc.Changes[i].Incarnation) - copy(k[20+8:], loc[:]) + copy(k[20:], loc[:]) c.add(k, change.Data, r, id) } } @@ -367,7 +326,10 @@ func (c *Coherent) View(ctx context.Context, tx kv.TemporalTx) (CacheView, error } } -func (c *Coherent) getFromCache(k []byte, id uint64, domain kv.Domain) (*Element, *CoherentRoot, error) { +// getFromCache returns a nil root when the view's root was already evicted +// (the view outlived KeepViews version advances): the caller then reads +// through its own tx snapshot without caching. +func (c *Coherent) getFromCache(k []byte, id uint64, domain kv.Domain) (*Element, *CoherentRoot) { // using the full lock here rather than RLock as RLock causes a lot of calls to runtime.usleep degrading // performance under load c.lock.Lock() @@ -375,7 +337,7 @@ func (c *Coherent) getFromCache(k []byte, id uint64, domain kv.Domain) (*Element r, ok := c.roots[id] if !ok { - return nil, r, fmt.Errorf("too old ViewID: %d, latestStateVersionID=%d", id, c.latestStateVersionID) + return nil, nil } isLatest := c.latestStateVersionID == id @@ -386,19 +348,25 @@ func (c *Coherent) getFromCache(k []byte, id uint64, domain kv.Domain) (*Element it, _ = r.cache.Get(&Element{K: k}) } if it != nil && isLatest { - c.stateEvict.MoveToFront(it) + if domain == kv.CodeDomain { + c.codeEvict.MoveToFront(it) + } else { + c.stateEvict.MoveToFront(it) + } } - return it, r, nil + return it, r } func (c *Coherent) Get(k []byte, tx kv.TemporalTx, id uint64) (v []byte, err error) { //TODO: Get must accept from user Domain parameter - it, r, err := c.getFromCache(k, id, kv.AccountsDomain) - if err != nil { - return nil, err + var it *Element + var r *CoherentRoot + // A zero budget retains nothing: skip the lookup and its global lock; + // leaving r nil also skips the add below. + if c.cfg.CacheSize != 0 { + it, r = c.getFromCache(k, id, kv.AccountsDomain) } if it != nil { - //fmt.Printf("from cache: %#x,%x\n", k, it.(*Element).V) c.hits.Inc() return it.V, nil } @@ -416,7 +384,9 @@ func (c *Coherent) Get(k []byte, tx kv.TemporalTx, id uint64) (v []byte, err err if len(v) == 0 { return v, nil } - //fmt.Printf("from db: %#x,%x\n", k, v) + if r == nil { + return v, nil + } c.lock.Lock() defer c.lock.Unlock() @@ -426,13 +396,14 @@ func (c *Coherent) Get(k []byte, tx kv.TemporalTx, id uint64) (v []byte, err err } func (c *Coherent) GetCode(k []byte, tx kv.TemporalTx, id uint64) (v []byte, err error) { - it, r, err := c.getFromCache(k, id, kv.CodeDomain) - if err != nil { - return nil, err + var it *Element + var r *CoherentRoot + // see Get + if c.cfg.CodeCacheSize != 0 { + it, r = c.getFromCache(k, id, kv.CodeDomain) } if it != nil { - //fmt.Printf("from cache: %#x,%x\n", k, it.(*Element).V) c.codeHits.Inc() return it.V, nil } @@ -442,7 +413,9 @@ func (c *Coherent) GetCode(k []byte, tx kv.TemporalTx, id uint64) (v []byte, err if err != nil { return nil, err } - //fmt.Printf("from db: %#x,%x\n", k, v) + if r == nil { + return v, nil + } c.lock.Lock() defer c.lock.Unlock() @@ -466,11 +439,14 @@ func (c *Coherent) removeOldestCode(r *CoherentRoot) { func (c *Coherent) add(k, v []byte, r *CoherentRoot, id uint64) *Element { it := &Element{K: k, V: v} - replaced, _ := r.cache.Set(it) - if c.latestStateVersionID != id { - //fmt.Printf("add to non-last viewID: %d<%d\n", c.latestViewID, id) + // Non-latest roots bypass eviction accounting, so growing them would be + // unbounded (e.g. with no state-change stream feeding OnNewBlock); the + // caller's tx read is authoritative for its snapshot, skip caching. + // A zero budget would evict the entry immediately — also skip. + if c.latestStateVersionID != id || c.cfg.CacheSize == 0 { return it } + replaced, _ := r.cache.Set(it) if replaced != nil { c.stateEvict.Remove(replaced) } @@ -485,11 +461,11 @@ func (c *Coherent) add(k, v []byte, r *CoherentRoot, id uint64) *Element { } func (c *Coherent) addCode(k, v []byte, r *CoherentRoot, id uint64) *Element { it := &Element{K: k, V: v} - replaced, _ := r.codeCache.Set(it) - if c.latestStateVersionID != id { - //fmt.Printf("add to non-last viewID: %d<%d\n", c.latestViewID, id) + // see add + if c.latestStateVersionID != id || c.cfg.CodeCacheSize == 0 { return it } + replaced, _ := r.codeCache.Set(it) if replaced != nil { c.codeEvict.Remove(replaced) } diff --git a/db/kv/kvcache/cache_test.go b/db/kv/kvcache/cache_test.go index 0ca66836d66..94711d1a93d 100644 --- a/db/kv/kvcache/cache_test.go +++ b/db/kv/kvcache/cache_test.go @@ -25,7 +25,6 @@ import ( "testing" "time" - keccak "github.com/erigontech/fastkeccak" "github.com/holiman/uint256" "github.com/stretchr/testify/require" @@ -50,7 +49,6 @@ func TestEvictionInUnexpectedOrder(t *testing.T) { c.selectOrCreateRoot(2) require.Len(c.roots, 1) require.Zero(int(c.latestStateVersionID)) - require.False(c.roots[2].isCanonical) c.add([]byte{1}, nil, c.roots[2], 2) require.Zero(c.stateEvict.Len()) @@ -58,7 +56,6 @@ func TestEvictionInUnexpectedOrder(t *testing.T) { c.advanceRoot(2) require.Len(c.roots, 1) require.Equal(2, int(c.latestStateVersionID)) - require.True(c.roots[2].isCanonical) c.add([]byte{1}, nil, c.roots[2], 2) require.Equal(1, c.stateEvict.Len()) @@ -66,7 +63,6 @@ func TestEvictionInUnexpectedOrder(t *testing.T) { c.selectOrCreateRoot(5) require.Len(c.roots, 2) require.Equal(2, int(c.latestStateVersionID)) - require.False(c.roots[5].isCanonical) c.add([]byte{2}, nil, c.roots[5], 5) // not added to evict list require.Equal(1, c.stateEvict.Len()) @@ -76,32 +72,26 @@ func TestEvictionInUnexpectedOrder(t *testing.T) { c.selectOrCreateRoot(6) require.Len(c.roots, 3) require.Equal(2, int(c.latestStateVersionID)) - require.False(c.roots[6].isCanonical) // parrent exists, but parent has isCanonical=false c.advanceRoot(3) require.Len(c.roots, 4) require.Equal(3, int(c.latestStateVersionID)) - require.True(c.roots[3].isCanonical) c.advanceRoot(4) require.Len(c.roots, 5) require.Equal(4, int(c.latestStateVersionID)) - require.True(c.roots[4].isCanonical) c.selectOrCreateRoot(5) require.Len(c.roots, 5) require.Equal(4, int(c.latestStateVersionID)) - require.False(c.roots[5].isCanonical) c.advanceRoot(5) require.Len(c.roots, 5) require.Equal(5, int(c.latestStateVersionID)) - require.True(c.roots[5].isCanonical) c.advanceRoot(100) require.Len(c.roots, 6) require.Equal(100, int(c.latestStateVersionID)) - require.True(c.roots[100].isCanonical) //c.add([]byte{1}, nil, c.roots[2], 2) require.Equal(0, c.latestStateView.cache.Len()) @@ -169,6 +159,338 @@ func TestEviction(t *testing.T) { require.Equal(int(cfg.CacheSize.Bytes()), c.stateEvict.Size()) } +// Canonical roots must start from their own batch only: the state-change +// producers do not announce every mutation (e.g. account deletions), so +// entries inherited from the previous root could stay stale forever. +func TestCanonicalRootsStartFresh(t *testing.T) { + require := require.New(t) + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + + k1 := [20]byte{1} + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: 2, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_UPSERT, + Address: gointerfaces.ConvertAddressToH160(k1), + Data: []byte{1}, + }}, + }}, + }) + require.Equal(1, c.roots[2].cache.Len()) + + c.OnNewBlock(&remoteproto.StateChangeBatch{StateVersionId: 3}) + require.Zero(c.roots[3].cache.Len()) +} + +func TestRetainedRootsShareCacheBudgets(t *testing.T) { + require := require.New(t) + cfg := DefaultCoherentConfig + cfg.CacheSize = 21 + cfg.CodeCacheSize = 21 + cfg.NewBlockWait = 0 + c := New(cfg) + + addVersion := func(version uint64, addr [20]byte) { + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: version, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_UPSERT_CODE, + Address: gointerfaces.ConvertAddressToH160(addr), + Data: []byte{byte(version)}, + Code: []byte{byte(version)}, + }}, + }}, + }) + } + + addVersion(1, [20]byte{1}) + addVersion(2, [20]byte{2}) + require.Len(c.roots, 2) + + var stateSize, codeSize int + for _, root := range c.roots { + root.cache.Scan(func(element *Element) bool { + stateSize += element.Size() + return true + }) + root.codeCache.Scan(func(element *Element) bool { + codeSize += element.Size() + return true + }) + } + require.LessOrEqual(stateSize, int(cfg.CacheSize.Bytes())) + require.LessOrEqual(codeSize, int(cfg.CodeCacheSize.Bytes())) +} + +// Batch-fed storage entries must be stored under the key shape readers use: +// address+location (see state.CachedReader3.ReadAccountStorage). +func TestOnNewBlockStorageKeysMatchReaders(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + + addr, loc := [20]byte{1}, [32]byte{2} + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: 2, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_STORAGE, + Address: gointerfaces.ConvertAddressToH160(addr), + StorageChanges: []*remoteproto.StorageChange{{ + Location: gointerfaces.ConvertHashToH256(loc), + Data: []byte{42}, + }}, + }}, + }}, + }) + + err := db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + k := append(addr[:], loc[:]...) + v, err := c.Get(k, tx, 2) + require.NoError(err) + require.Equal([]byte{42}, v) + return nil + }) + require.NoError(err) +} + +// Batch-fed code entries must be stored under the key shape readers use: +// the account address, which is the E3 CodeDomain key +// (see state.CachedReader3.ReadAccountCode). +func TestOnNewBlockCodeKeysMatchReaders(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + + addr := [20]byte{1} + code := []byte{0x60, 0x60} + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: 2, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_CODE, + Address: gointerfaces.ConvertAddressToH160(addr), + Code: code, + }}, + }}, + }) + + err := db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + v, err := c.GetCode(addr[:], tx, 2) + require.NoError(err) + require.Equal(code, v) + return nil + }) + require.NoError(err) +} + +// A cache hit on the code domain must refresh the entry's position in the +// code eviction list — otherwise hot code is evicted in insertion order. +func TestCodeHitRefreshesCodeEvictLRU(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + + addr1, addr2 := [20]byte{1}, [20]byte{2} + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: 2, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{ + {Action: remoteproto.Action_CODE, Address: gointerfaces.ConvertAddressToH160(addr1), Code: []byte{1}}, + {Action: remoteproto.Action_CODE, Address: gointerfaces.ConvertAddressToH160(addr2), Code: []byte{2}}, + }, + }}, + }) + require.Equal(addr1[:], c.codeEvict.Oldest().K) + + err := db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + _, err := c.GetCode(addr1[:], tx, 2) + return err + }) + require.NoError(err) + require.Equal(addr2[:], c.codeEvict.Oldest().K) +} + +// A request whose cache view outlives KeepViews state-version advances (e.g. a +// long eth_call) must fall back to its own tx snapshot, not error out. +func TestViewSurvivesRootEviction(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + k1 := [20]byte{1} + + err := db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + cacheView, err := c.View(ctx, tx) + require.NoError(err) + view := cacheView.(*CoherentView) + + for i := uint64(1); i <= cfg.KeepViews+2; i++ { + c.OnNewBlock(&remoteproto.StateChangeBatch{StateVersionId: view.stateVersionID + i}) + } + _, rootAlive := c.roots[view.stateVersionID] + require.False(rootAlive, "root must be evicted for this test to be meaningful") + + v, err := c.Get(k1[:], tx, view.stateVersionID) + require.NoError(err) + require.Empty(v) + + code, err := c.GetCode(k1[:], tx, view.stateVersionID) + require.NoError(err) + require.Empty(code) + return nil + }) + require.NoError(err) +} + +// Reads through a view whose version is not the latest (pre-commit window, or +// no state-change stream at all) bypass eviction accounting, so they must not +// grow the root either — otherwise memory is unbounded. +func TestNonLatestViewReadsAreNotCached(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.NewBlockWait = 0 + c := New(cfg) + + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + k1 := [20]byte{1} + acc := accounts.Account{Nonce: 1, Balance: *uint256.NewInt(11), CodeHash: accounts.EmptyCodeHash} + accEnc := accounts.SerialiseV3(&acc) + + err := db.UpdateTemporal(ctx, func(tx kv.TemporalRwTx) error { + d, err := execctx.NewSharedDomains(ctx, tx, log.New()) + if err != nil { + return err + } + defer d.Close() + if err := d.DomainPut(kv.AccountsDomain, tx, k1[:], accEnc, 0, nil); err != nil { + return err + } + return d.Flush(ctx, tx) + }) + require.NoError(err) + + err = db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + cacheView, err := c.View(ctx, tx) + require.NoError(err) + view := cacheView.(*CoherentView) + require.NotEqual(c.latestStateVersionID, view.stateVersionID) + + v, err := c.Get(k1[:], tx, view.stateVersionID) + require.NoError(err) + require.Equal(accEnc, v) + + require.Zero(c.roots[view.stateVersionID].cache.Len()) + require.Zero(c.stateEvict.Len()) + return nil + }) + require.NoError(err) +} + +// A zero budget must never retain entries — batch-fed or read-through, even at +// the latest version — and reads must resolve on the caller's tx snapshot, +// never on announced batch data. +func TestZeroBudgetRetainsNothing(t *testing.T) { + require, ctx := require.New(t), t.Context() + cfg := DefaultCoherentConfig + cfg.CacheSize = 0 + cfg.CodeCacheSize = 0 + cfg.NewBlockWait = 0 + c := New(cfg) + + db := temporaltest.NewTestDB(t, datadir.New(t.TempDir())) + addr, loc := [20]byte{1}, [32]byte{2} + committedAcc := accounts.Account{Nonce: 1, Balance: *uint256.NewInt(11), CodeHash: accounts.EmptyCodeHash} + committedAccEnc := accounts.SerialiseV3(&committedAcc) + committedCode := []byte{0x60, 0x01} + committedSlot := []byte{7} + storageKey := append(addr[:], loc[:]...) + + err := db.UpdateTemporal(ctx, func(tx kv.TemporalRwTx) error { + d, err := execctx.NewSharedDomains(ctx, tx, log.New()) + if err != nil { + return err + } + defer d.Close() + if err := d.DomainPut(kv.AccountsDomain, tx, addr[:], committedAccEnc, 0, nil); err != nil { + return err + } + if err := d.DomainPut(kv.CodeDomain, tx, addr[:], committedCode, 0, nil); err != nil { + return err + } + if err := d.DomainPut(kv.StorageDomain, tx, storageKey, committedSlot, 0, nil); err != nil { + return err + } + return d.Flush(ctx, tx) + }) + require.NoError(err) + + err = db.ViewTemporal(ctx, func(tx kv.TemporalTx) error { + stateVersion, err := tx.ReadSequence(string(kv.PlainStateVersion)) + require.NoError(err) + + announcedAcc := committedAcc + announcedAcc.Nonce = 2 + c.OnNewBlock(&remoteproto.StateChangeBatch{ + StateVersionId: stateVersion, + ChangeBatch: []*remoteproto.StateChange{{ + Direction: remoteproto.Direction_FORWARD, + Changes: []*remoteproto.AccountChange{{ + Action: remoteproto.Action_UPSERT_CODE, + Address: gointerfaces.ConvertAddressToH160(addr), + Data: accounts.SerialiseV3(&announcedAcc), + Code: []byte{0x60, 0x02}, + StorageChanges: []*remoteproto.StorageChange{{ + Location: gointerfaces.ConvertHashToH256(loc), + Data: []byte{42}, + }}, + }}, + }}, + }) + + cacheView, err := c.View(ctx, tx) + require.NoError(err) + view := cacheView.(*CoherentView) + require.Equal(c.latestStateVersionID, view.stateVersionID) + + v, err := c.Get(addr[:], tx, view.stateVersionID) + require.NoError(err) + require.Equal(committedAccEnc, v) + + v, err = c.Get(storageKey, tx, view.stateVersionID) + require.NoError(err) + require.Equal(committedSlot, v) + + code, err := c.GetCode(addr[:], tx, view.stateVersionID) + require.NoError(err) + require.Equal(committedCode, code) + + require.Zero(c.roots[view.stateVersionID].cache.Len()) + require.Zero(c.roots[view.stateVersionID].codeCache.Len()) + require.Zero(c.stateEvict.Len()) + require.Zero(c.codeEvict.Len()) + return nil + }) + require.NoError(err) +} + func TestAPI(t *testing.T) { require := require.New(t) @@ -463,57 +785,6 @@ func TestAPI(t *testing.T) { } } -func TestOnNewBlockCodeHashKey(t *testing.T) { - require := require.New(t) - cfg := DefaultCoherentConfig - cfg.NewBlockWait = 0 - c := New(cfg) - - code := []byte{0x01, 0x02, 0x03, 0x04} - addr := common.Address{0xAA} - - batch := &remoteproto.StateChangeBatch{ - StateVersionId: 1, - ChangeBatch: []*remoteproto.StateChange{ - { - Direction: remoteproto.Direction_FORWARD, - Changes: []*remoteproto.AccountChange{ - { - Action: remoteproto.Action_CODE, - Address: gointerfaces.ConvertAddressToH160(addr), - Code: code, - }, - }, - }, - }, - } - - c.OnNewBlock(batch) - - c.lock.Lock() - defer c.lock.Unlock() - - require.NotNil(c.latestStateView) - require.Equal(uint64(1), c.latestStateVersionID) - - var elems []*Element - c.latestStateView.codeCache.Walk(func(items []*Element) bool { - if len(items) > 0 { - elems = append(elems, items...) - } - return true - }) - - require.Len(elems, 1) - - h := keccak.NewFastKeccak() - h.Write(code) - expectedKey := h.Sum(nil) - - require.Equal(expectedKey, elems[0].K) - require.Equal(code, elems[0].V) -} - func TestCode(t *testing.T) { require, ctx := require.New(t), t.Context() c := New(DefaultCoherentConfig) diff --git a/db/kv/kvcache/simple.go b/db/kv/kvcache/simple.go index 205e6ef0c46..87202b350fd 100644 --- a/db/kv/kvcache/simple.go +++ b/db/kv/kvcache/simple.go @@ -105,6 +105,7 @@ type SimpleView struct { } func (c *SimpleView) Get(k []byte) ([]byte, error) { return c.cache.Get(k, c.tx, 0) } + func (c *SimpleView) GetAsOf(key []byte, ts uint64) (v []byte, ok bool, err error) { return nil, false, nil } diff --git a/docs/site/docs/fundamentals/modules/rpc-daemon.md b/docs/site/docs/fundamentals/modules/rpc-daemon.md index 3d3f3fa07bb..019b52e0c3a 100644 --- a/docs/site/docs/fundamentals/modules/rpc-daemon.md +++ b/docs/site/docs/fundamentals/modules/rpc-daemon.md @@ -117,7 +117,7 @@ Flags: --rpc.txfeecap float Sets a cap on transaction fee (in ether) that can be sent via the RPC APIs (0 = no cap) (default 1) --socket.enabled Enable IPC server --socket.url string IPC server listening url. prefix supported are tcp, unix (default "unix:///var/run/erigon.sock") - --state.cache string Amount of data to store in StateCache (enabled if no --datadir set). Set 0 to disable StateCache. Defaults to 0MB RAM (default "0MB") + --state.cache string Amount of data to store in the version-keyed StateCache (an equally-sized code cache is budgeted on top). Set 0 to disable entry retention while preserving snapshot-consistent reads (default "0MB") --tls.cacert string CA certificate for client side TLS handshake for GRPC --tls.cert string certificate for client side TLS handshake for GRPC --tls.key string key file for client side TLS handshake for GRPC diff --git a/docs/site/static/llms-full.txt b/docs/site/static/llms-full.txt index 9812bf8eec5..6f549b09bc0 100644 --- a/docs/site/static/llms-full.txt +++ b/docs/site/static/llms-full.txt @@ -4121,7 +4121,7 @@ Flags: --rpc.txfeecap float Sets a cap on transaction fee (in ether) that can be sent via the RPC APIs (0 = no cap) (default 1) --socket.enabled Enable IPC server --socket.url string IPC server listening url. prefix supported are tcp, unix (default "unix:///var/run/erigon.sock") - --state.cache string Amount of data to store in StateCache (enabled if no --datadir set). Set 0 to disable StateCache. Defaults to 0MB RAM (default "0MB") + --state.cache string Amount of data to store in the version-keyed StateCache (an equally-sized code cache is budgeted on top). Set 0 to disable entry retention while preserving snapshot-consistent reads (default "0MB") --tls.cacert string CA certificate for client side TLS handshake for GRPC --tls.cert string certificate for client side TLS handshake for GRPC --tls.key string key file for client side TLS handshake for GRPC diff --git a/execution/execmodule/exec_module_test.go b/execution/execmodule/exec_module_test.go index 08cab900c92..8db74f314eb 100644 --- a/execution/execmodule/exec_module_test.go +++ b/execution/execmodule/exec_module_test.go @@ -33,6 +33,7 @@ import ( "github.com/cenkalti/backoff/v4" "github.com/holiman/uint256" "github.com/stretchr/testify/require" + "google.golang.org/grpc" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/crypto" @@ -55,6 +56,7 @@ import ( "github.com/erigontech/erigon/execution/tests/blockgen" "github.com/erigontech/erigon/execution/types" "github.com/erigontech/erigon/execution/types/accounts" + "github.com/erigontech/erigon/node/gointerfaces/remoteproto" "github.com/erigontech/erigon/node/gointerfaces/txpoolproto" ) @@ -697,6 +699,26 @@ func insertValidateAndUfc1By1(ctx context.Context, exec *execmodule.ExecModule, return nil } +// insertAndUfcBatched inserts all blocks and drives a single FCU to the top, +// so the whole batch executes under one forkchoice run. +func insertAndUfcBatched(ctx context.Context, exec *execmodule.ExecModule, blocks []*types.Block) error { + ir, err := insertBlocks(ctx, exec, blocks) + if err != nil { + return err + } + if ir != execmodule.ExecutionStatusSuccess { + return fmt.Errorf("unexpected insertBlocks status: %s", ir) + } + ur, err := updateForkChoice(ctx, exec, blocks[len(blocks)-1].Header()) + if err != nil { + return err + } + if ur.Status != execmodule.ExecutionStatusSuccess { + return fmt.Errorf("unexpected updateForkChoice status: %s", ur.Status) + } + return nil +} + func assembleBlock(ctx context.Context, exec *execmodule.ExecModule, params *builder.Parameters) (uint64, error) { return retryBusy(ctx, func() (uint64, bool, error) { r, err := exec.AssembleBlock(ctx, params) @@ -1335,6 +1357,68 @@ func TestAssembleBlockAmsterdamForkTransition(t *testing.T) { require.NoError(t, err) } +// TestStateChangeVersionMatchesCommitted pins the contract the Coherent kvcache +// relies on: the StateVersionId announced in a state-change batch equals the +// PlainStateVersion a committed read tx observes once that batch's commit lands. +// If they diverge, version-keyed cache roots never match any reader. +func TestStateChangeVersionMatchesCommitted(t *testing.T) { + for _, mode := range []struct { + name string + opts []execmoduletester.Option + }{ + {name: "fg-commit"}, + {name: "bg-commit", opts: []execmoduletester.Option{execmoduletester.WithFcuBackgroundCommit()}}, + } { + // 1by1 flushes once per block; batched executes all blocks under one + // FCU and crosses the initial-cycle threshold, so mid-FCU CommitCycle + // commits bump the version before the single announce. + for _, ins := range []struct { + name string + blocks int + insert func(context.Context, *execmodule.ExecModule, []*types.Block) error + }{ + {name: "1by1", blocks: 3, insert: insertValidateAndUfc1By1}, + {name: "batched", blocks: 20, insert: insertAndUfcBatched}, + } { + t.Run(mode.name+"/"+ins.name, func(t *testing.T) { + ctx := t.Context() + m := execmoduletester.New(t, mode.opts...) + exec := m.ExecModule + + streamCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + stream, err := m.StateChangesClient().StateChanges(streamCtx, &remoteproto.StateChangeRequest{}, grpc.WaitForReady(true)) + require.NoError(t, err) + + chainPack, err := blockgen.GenerateChain(m.ChainConfig, m.Genesis, m.Engine, m.DB, ins.blocks, nil) + require.NoError(t, err) + require.NoError(t, ins.insert(ctx, exec, chainPack.Blocks)) + + topBlock := chainPack.TopBlock.NumberU64() + var lastAnnounced uint64 + for found := false; !found; { + batch, err := stream.Recv() + require.NoError(t, err) + for _, cb := range batch.ChangeBatch { + if cb.Direction == remoteproto.Direction_FORWARD && cb.BlockHeight == topBlock { + lastAnnounced = batch.StateVersionId + found = true + } + } + } + + exec.WaitIdle(ctx) + var committed uint64 + require.NoError(t, m.DB.View(ctx, func(tx kv.Tx) error { + committed, err = rawdb.GetStateVersion(tx) + return err + })) + require.Equal(t, committed, lastAnnounced, "announced StateVersionId must equal committed PlainStateVersion") + }) + } + } +} + // TestGetPayloadBodiesRegenerateBlockAccessLists verifies the payload-bodies // getters serve stored BALs as-is and, once the stored rows are pruned (kept // only for the reorg window), regenerate them by re-execution — @@ -1489,17 +1573,14 @@ func TestNotificationDispatchForegroundCommit(t *testing.T) { // commit enabled, notifications are still dispatched before FCU returns, // even though the DB commit happens asynchronously. // -// Note: with background commit, subsequent blocks may fail validation -// because the DB state hasn't caught up yet (the commit is async). This -// test only processes the genesis → block 1 transition to verify that -// notification dispatch works correctly in the background commit path. +// Successive FCUs are correctly serialized via the ExecModule semaphore +// (see updateForkChoice / runPostForkchoice): the bg goroutine releases +// the semaphore only after Flush+Commit, so FCU N+1 always reads the +// committed state of FCU N. This test exercises one genesis → block 1 +// transition; multi-block bg-commit coverage lives in +// TestReorgBackAndForwardIntoCanonicalChain (bg-commit mode) and +// TestInsertBlocksWithBatchedFCU_BadBlockRecovery_Background. func TestNotificationDispatchBackgroundCommit(t *testing.T) { - // Background commit creates a race: FCU N returns before commit finishes, - // so FCU N+1 reads stale state from DB. This is the known limitation that - // the API-layer "latest head pointer" coordination is designed to solve. - // Once that's implemented, remove this skip and verify the full flow. - t.Skip("background commit requires API-layer coordination (latest head pointer) to work correctly") - m := execmoduletester.New(t, execmoduletester.WithFcuBackgroundCommit()) headerCh, unsub := m.Notifications.Events.AddHeaderSubscription() diff --git a/execution/execmodule/execmoduletester/exec_module_tester.go b/execution/execmodule/execmoduletester/exec_module_tester.go index e0e4344f463..6b309bf2b23 100644 --- a/execution/execmodule/execmoduletester/exec_module_tester.go +++ b/execution/execmodule/execmoduletester/exec_module_tester.go @@ -830,6 +830,8 @@ func (emt *ExecModuleTester) EnableLogs() { func (emt *ExecModuleTester) Cfg() ethconfig.Config { return emt.cfg } +func (emt *ExecModuleTester) StateChangesClient() StateChangesClient { return emt.stateChangesClient } + func (emt *ExecModuleTester) insertPoSBlocks(chain *blockgen.ChainPack) error { wr := chainreader.NewChainReaderEth1(emt.ChainConfig, emt.ExecModule, time.Hour) @@ -867,9 +869,9 @@ func (emt *ExecModuleTester) insertPoSBlocks(chain *blockgen.ChainPack) error { return fmt.Errorf("insertion failed for block %d, code: %s", chain.Blocks[chain.Length()-1].NumberU64(), status.String()) } - // UpdateForkChoice calls commit asyncronously so we need to - // wait for confimation that the headers are processed before - // returning to the caller + // Wait for the state-change dispatcher to fire for all inserted blocks. + // This only confirms dispatch — commit completion is ensured separately + // (WaitIdle in InsertChain). lastSeenBlock := chain.Headers[0].Number.Uint64() for len(insertedBlocks) > 0 { @@ -902,6 +904,9 @@ func (emt *ExecModuleTester) InsertChain(chain *blockgen.ChainPack) error { if err := emt.insertPoSBlocks(chain); err != nil { return err } + // UpdateForkChoice can return before the MDBX commit lands; wait so the + // reads below see committed state. + emt.ExecModule.WaitIdle(emt.Ctx) roTx, err := emt.DB.BeginRo(emt.Ctx) if err != nil { return err diff --git a/execution/execmodule/forkchoice.go b/execution/execmodule/forkchoice.go index 280959cb907..8212f38fad3 100644 --- a/execution/execmodule/forkchoice.go +++ b/execution/execmodule/forkchoice.go @@ -843,6 +843,7 @@ func (e *ExecModule) dispatchNotificationsFromOverlay(sd *execctx.SharedDomains, finishProgressBefore, finishProgressAfter, e.pipelineExecutor.Sync().PrevUnwindPoint(), + true, // pre-commit: the overlay's flush+commit runs after this dispatch ); err != nil { return err } diff --git a/execution/execmodule/notification_dispatcher.go b/execution/execmodule/notification_dispatcher.go index ddfc9929849..408366e6cf1 100644 --- a/execution/execmodule/notification_dispatcher.go +++ b/execution/execmodule/notification_dispatcher.go @@ -79,6 +79,7 @@ func NewDispatcher( // - finishProgressBefore: Finish stage progress before the sync run // - finishProgressAfter: Finish stage progress after the sync run // - prevUnwindPoint: previous unwind point from the pipeline (may be nil) +// - preCommit: tx is an overlay whose flush+commit has not happened yet func (d *Dispatcher) Dispatch( ctx context.Context, tx kv.Tx, @@ -87,6 +88,7 @@ func (d *Dispatcher) Dispatch( finishProgressBefore uint64, finishProgressAfter uint64, prevUnwindPoint *uint64, + preCommit bool, ) error { // Update the accumulator with the current plain state version so downstream // consumers (e.g. state cache) know state has moved on. @@ -95,6 +97,12 @@ func (d *Dispatcher) Dispatch( if err != nil { return err } + if preCommit { + // The flush that follows bumps PlainStateVersion exactly once + // (TemporalMemBatch.flushLocked); announce the post-commit value so + // version-keyed caches match what committed readers will observe. + plainStateVersion++ + } accumulator.SetStateID(plainStateVersion) } diff --git a/execution/stagedsync/stageloop/stageloop.go b/execution/stagedsync/stageloop/stageloop.go index 67f98000697..ec0a647ab4a 100644 --- a/execution/stagedsync/stageloop/stageloop.go +++ b/execution/stagedsync/stageloop/stageloop.go @@ -52,7 +52,7 @@ import ( // an implementation defined in another package (e.g. execmodule.Dispatcher) // without creating a circular import. type NotificationSender interface { - Dispatch(ctx context.Context, tx kv.Tx, accumulator *shards.Accumulator, recentReceipts *shards.RecentReceipts, finishProgressBefore, finishProgressAfter uint64, prevUnwindPoint *uint64) error + Dispatch(ctx context.Context, tx kv.Tx, accumulator *shards.Accumulator, recentReceipts *shards.RecentReceipts, finishProgressBefore, finishProgressAfter uint64, prevUnwindPoint *uint64, preCommit bool) error } type Hook struct { @@ -120,8 +120,8 @@ func (h *Hook) BeforeRun(tx kv.Tx, inSync bool) error { } // SendNotifications dispatches all pending notifications (state changes, -// headers, logs, receipts) via the Dispatcher. The tx is the data source — -// either the SD's blockOverlay (pre-commit) or a committed DB tx. +// headers, logs, receipts) via the Dispatcher. The tx must be a committed DB +// tx (pre-commit overlay dispatch goes through the Dispatcher directly). // // All call sites follow the same pattern: // @@ -145,6 +145,7 @@ func (h *Hook) SendNotifications(tx kv.Tx, finishProgressBefore uint64) error { finishProgressBefore, finishStageAfterSync, h.sync.PrevUnwindPoint(), + false, ) } diff --git a/llms-full.txt b/llms-full.txt index 9812bf8eec5..6f549b09bc0 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -4121,7 +4121,7 @@ Flags: --rpc.txfeecap float Sets a cap on transaction fee (in ether) that can be sent via the RPC APIs (0 = no cap) (default 1) --socket.enabled Enable IPC server --socket.url string IPC server listening url. prefix supported are tcp, unix (default "unix:///var/run/erigon.sock") - --state.cache string Amount of data to store in StateCache (enabled if no --datadir set). Set 0 to disable StateCache. Defaults to 0MB RAM (default "0MB") + --state.cache string Amount of data to store in the version-keyed StateCache (an equally-sized code cache is budgeted on top). Set 0 to disable entry retention while preserving snapshot-consistent reads (default "0MB") --tls.cacert string CA certificate for client side TLS handshake for GRPC --tls.cert string certificate for client side TLS handshake for GRPC --tls.key string key file for client side TLS handshake for GRPC diff --git a/rpc/rpchelper/helper.go b/rpc/rpchelper/helper.go index ce61d683eef..24f893aaa7f 100644 --- a/rpc/rpchelper/helper.go +++ b/rpc/rpchelper/helper.go @@ -23,7 +23,6 @@ import ( "github.com/holiman/uint256" "github.com/erigontech/erigon/common" - "github.com/erigontech/erigon/common/dbg" "github.com/erigontech/erigon/db/dbservices" "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/kv/kvcache" @@ -220,15 +219,7 @@ func CreateLatestCachedStateReader(cache kvcache.CacheView, tx kv.TemporalTx) st return state.NewCachedReader3(cache, tx) } -type asOfView interface { - GetAsOf(key []byte, ts uint64) (v []byte, ok bool, err error) -} - func CreateHistoryCachedStateReader(ctx context.Context, cache kvcache.CacheView, tx kv.TemporalTx, blockNumber uint64, txnIndex int, txNumsReader rawdbv3.TxNumsReader) (state.StateReader, error) { - asOfView, ok := cache.(asOfView) - if !ok { - return nil, fmt.Errorf("%T does not implement GetAsOf at: %s", cache, dbg.Stack()) - } minTxNum, err := txNumsReader.Min(ctx, tx, blockNumber) if err != nil { return nil, err @@ -238,14 +229,14 @@ func CreateHistoryCachedStateReader(ctx context.Context, cache kvcache.CacheView return nil, fmt.Errorf("%w: block tx: %d, min tx: %d", state.PrunedError, txNum, minHistoryTxNum) } return &cachedHistoryReaderV3{ - cache: asOfView, + cache: cache, reader: state.NewHistoryReaderV3(tx, txNum), composite: make([]byte, 0, len(common.Address{})+len(common.Hash{})), }, nil } type cachedHistoryReaderV3 struct { - cache asOfView + cache kvcache.CacheView reader *state.HistoryReaderV3 composite []byte } From 6c006e5516ee90a4d2d2de7e6291001cb6805dbc Mon Sep 17 00:00:00 2001 From: yperbasis Date: Fri, 17 Jul 2026 17:13:39 +0200 Subject: [PATCH 2/2] db/kv/kvcache: dedup coherent root comment --- db/kv/kvcache/cache.go | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/db/kv/kvcache/cache.go b/db/kv/kvcache/cache.go index 11d4e8476f9..33571359996 100644 --- a/db/kv/kvcache/cache.go +++ b/db/kv/kvcache/cache.go @@ -229,11 +229,8 @@ func (c *Coherent) advanceRoot(stateVersionID uint64) (r *CoherentRoot) { c.roots[stateVersionID] = r } - // No carry-over from the previous canonical root: the state-change - // producers don't announce every mutation (account deletions, code on - // unwind — https://github.com/erigontech/erigon/issues/22276), so - // inherited entries could stay stale forever. Fresh roots bound any - // producer gap to one version. + // No carry-over from the previous canonical root — producers don't + // announce every mutation; see the type comment. for _, root := range c.roots { root.cache.Clear() root.codeCache.Clear()