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/cmd/utils/app/import_cmd.go b/cmd/utils/app/import_cmd.go index 78bb0fef479..67905370c72 100644 --- a/cmd/utils/app/import_cmd.go +++ b/cmd/utils/app/import_cmd.go @@ -417,7 +417,9 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool } // UpdateForkChoice has an async commit so we need to wait to make sure - // it is completed before assuming all state changes etc are inserted + // it is completed before assuming all state changes etc are inserted. + // State-change events are dispatched pre-commit, so waiting on the stream + // only ensures the dispatcher fired — not that MDBX is flushed. var lastSeenBlock uint64 for len(insertedBlocks) > 0 { req, err := stream.Recv() @@ -445,6 +447,10 @@ func InsertChain(ethereum *eth.Ethereum, chain *blockgen.ChainPack, setHead bool } } + // Wait for the FCU background commit so HeadBlockHash can't land in MDBX + // ahead of the header it points to. + ethereum.ExecutionModule().WaitIdle(ethereum.SentryCtx()) + return ethereum.ChainDB().Update(ethereum.SentryCtx(), func(tx kv.RwTx) error { rawdb.WriteHeadBlockHash(tx, lvh) return nil diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 54a35192174..2a38acb36a1 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -1143,7 +1143,7 @@ var ( } FcuBackgroundCommitFlag = cli.BoolFlag{ Name: "fcu.background.commit", - Usage: "Enables background flush and commit", + Usage: "Return FCU response before MDBX flush+commit lands (commit runs in background; remote rpcdaemon 'latest' stays consistent but can lag for the commit duration)", Value: ethconfig.Defaults.FcuBackgroundCommit, } MCPDisableFlag = cli.BoolFlag{ 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/db/kv/membatchwithdb/memory_mutation.go b/db/kv/membatchwithdb/memory_mutation.go index e227c98fef4..d82322982d7 100644 --- a/db/kv/membatchwithdb/memory_mutation.go +++ b/db/kv/membatchwithdb/memory_mutation.go @@ -29,8 +29,6 @@ import ( "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/db/kv" - "github.com/erigontech/erigon/db/kv/dbcfg" - "github.com/erigontech/erigon/db/kv/mdbx" "github.com/erigontech/erigon/db/kv/order" "github.com/erigontech/erigon/db/kv/stream" "github.com/erigontech/erigon/db/snapshotsync/blocksnapshots" @@ -46,7 +44,7 @@ type MemoryMutation struct { // Read views created via NewReadView share this pointer so they synchronize // with the parent's writers. mu *sync.RWMutex - memTx kv.RwTx + memTx *memStore // concrete type is load-bearing — see newReadViewMut memDb kv.RwDB deletedEntries map[string]map[string]struct{} deletedDups map[string]map[string]map[string]struct{} @@ -82,36 +80,6 @@ func NewMemoryBatch(tx kv.TemporalTx, tmpDir string, logger log.Logger) (*Memory }, nil } -// NewMemoryBatchMDBX creates an MDBX-backed in-memory batch. The MDBX write -// transaction pins the goroutine to an OS thread via runtime.LockOSThread(), -// so this variant must not be held across goroutine migrations. -func NewMemoryBatchMDBX(tx kv.TemporalTx, tmpDir string, logger log.Logger) (mm *MemoryMutation, err error) { - tmpDB := mdbx.New(dbcfg.TemporaryDB, logger).InMem(nil, tmpDir).GrowthStep(64 * datasize.MB).MapSize(512 * datasize.GB).MustOpen() - defer func() { - if err != nil { - tmpDB.Close() - } - }() - memTx, err := tmpDB.BeginRw(context.Background()) // nolint:gocritic - if err != nil { - return nil, fmt.Errorf("NewMemoryBatchMDBX: begin tx: %w", err) - } - if err = initSequences(tx, memTx); err != nil { - memTx.Rollback() - return nil, fmt.Errorf("NewMemoryBatchMDBX: init sequences: %w", err) - } - - return &MemoryMutation{ - mu: &sync.RWMutex{}, - db: tx, - memDb: tmpDB, - memTx: memTx, - deletedEntries: make(map[string]map[string]struct{}), - deletedDups: map[string]map[string]map[string]struct{}{}, - clearedTables: make(map[string]struct{}), - }, nil -} - func (m *MemoryMutation) UnderlyingTx() kv.TemporalTx { return m.db } @@ -576,6 +544,8 @@ func (m *MemoryMutation) Commit() error { return nil } +// Safe to close while read views are still iterating: the memStore backing +// makes Rollback a no-op on the data (see newReadViewMut). func (m *MemoryMutation) Rollback() { m.memTx.Rollback() m.memDb.Close() @@ -1042,13 +1012,19 @@ func (m *MemoryMutation) Unwind(ctx context.Context, txNumUnwindTo uint64, chang // // The returned kv.TemporalTx only exposes read methods. Callers cannot write // to the overlay through this view. The caller must not Close the returned -// view (it doesn't own the memDb). +// view (it doesn't own the memDb). Safe under a concurrent parent Close — +// see newReadViewMut. func (m *MemoryMutation) NewReadView(tx kv.Tx) kv.TemporalTx { return m.newReadViewMut(tx) } // newReadViewMut is the internal constructor that returns the full // *MemoryMutation. Used by NewTemporalReadView which needs to embed it. +// +// Read views stay safe under a concurrent parent Close only because the +// pure-Go memStore's Rollback/Close are no-ops on its data — memTx's type +// enforces that backing. A real-DB-backed memTx would invalidate cursors +// mid-iteration and need refcount/drain logic at the parent's Close. func (m *MemoryMutation) newReadViewMut(tx kv.Tx) *MemoryMutation { var dbTx kv.TemporalTx if t, ok := tx.(kv.TemporalTx); ok { diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index 2d994f32aa9..ea429e7749f 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -814,6 +814,11 @@ func (sd *SharedDomains) IteratePrefix(domain kv.Domain, prefix []byte, roTx kv. return sd.mem.IteratePrefix(domain, prefix, roTx, it) } +// Close releases this SD's in-memory state. Idempotent. +// +// Safe to call while readers still hold block-overlay views: the overlay's +// memStore backing keeps their data alive (see MemoryMutation.newReadViewMut), +// and sd.mem is never exposed to those views. func (sd *SharedDomains) Close() { if sd.sdCtx == nil { //idempotency return diff --git a/docs/site/docs/fundamentals/configuring-erigon.mdx b/docs/site/docs/fundamentals/configuring-erigon.mdx index d12064c9808..5544f7b456b 100644 --- a/docs/site/docs/fundamentals/configuring-erigon.mdx +++ b/docs/site/docs/fundamentals/configuring-erigon.mdx @@ -417,7 +417,7 @@ Flags for configuring Fork Choice Update behavior. * Default: `1s` * `--fcu.background.prune`: Enables background pruning after FCU. * Default: `true` -* `--fcu.background.commit`: Enables background flush and commit after FCU. +* `--fcu.background.commit`: Returns the FCU response before MDBX flush+commit lands (commit runs in background; remote `rpcdaemon` `latest` stays consistent but can lag for the commit duration). * Default: `false` ### Execution 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..4e4f7076023 100644 --- a/docs/site/static/llms-full.txt +++ b/docs/site/static/llms-full.txt @@ -2319,7 +2319,7 @@ Flags for configuring Fork Choice Update behavior. * Default: `1s` * `--fcu.background.prune`: Enables background pruning after FCU. * Default: `true` -* `--fcu.background.commit`: Enables background flush and commit after FCU. +* `--fcu.background.commit`: Returns the FCU response before MDBX flush+commit lands (commit runs in background; remote `rpcdaemon` `latest` stays consistent but can lag for the commit duration). * Default: `false` ### Execution @@ -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/engineapi/engine_block_downloader/block_downloader.go b/execution/engineapi/engine_block_downloader/block_downloader.go index 386bf9c589a..8cfbaa00d46 100644 --- a/execution/engineapi/engine_block_downloader/block_downloader.go +++ b/execution/engineapi/engine_block_downloader/block_downloader.go @@ -271,8 +271,31 @@ func (e *EngineBlockDownloader) downloadBlocks(ctx context.Context, req Backward return nil } +// retryBusy re-invokes call while it reports ExecutionStatusBusy (a background +// FCU commit briefly holds the exec semaphore), polling every 50ms and logging +// periodically so a stuck commit surfaces instead of hanging silently. +func (e *EngineBlockDownloader) retryBusy(ctx context.Context, label string, call func() (execmodule.ExecutionStatus, *string, common.Hash, error)) (execmodule.ExecutionStatus, *string, common.Hash, error) { + status, validationErr, lastValidHash, err := call() + logEvery := time.NewTicker(5 * time.Second) + defer logEvery.Stop() + for err == nil && status == execmodule.ExecutionStatusBusy { + if err := common.Sleep(ctx, 50*time.Millisecond); err != nil { + return status, validationErr, lastValidHash, err + } + select { + case <-logEvery.C: + e.logger.Debug("[EngineBlockDownloader] execution busy - retrying", "label", label) + default: + } + status, validationErr, lastValidHash, err = call() + } + return status, validationErr, lastValidHash, err +} + func (e *EngineBlockDownloader) execDownloadedBatch(ctx context.Context, block *types.Block, requested common.Hash) error { - status, validationErr, lastValidHash, err := e.chainRW.ValidateChain(ctx, block.Hash(), block.NumberU64()) + status, validationErr, lastValidHash, err := e.retryBusy(ctx, "ValidateChain", func() (execmodule.ExecutionStatus, *string, common.Hash, error) { + return e.chainRW.ValidateChain(ctx, block.Hash(), block.NumberU64()) + }) if err != nil { return err } @@ -297,7 +320,9 @@ func (e *EngineBlockDownloader) execDownloadedBatch(ctx context.Context, block * lastValidHash, ) } - fcuStatus, _, lastValidHash, err := e.chainRW.UpdateForkChoice(ctx, block.Hash(), common.Hash{}, common.Hash{}, 0) + fcuStatus, _, lastValidHash, err := e.retryBusy(ctx, "UpdateForkChoice", func() (execmodule.ExecutionStatus, *string, common.Hash, error) { + return e.chainRW.UpdateForkChoice(ctx, block.Hash(), common.Hash{}, common.Hash{}, 0) + }) if err != nil { return err } 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..4e4f7076023 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -2319,7 +2319,7 @@ Flags for configuring Fork Choice Update behavior. * Default: `1s` * `--fcu.background.prune`: Enables background pruning after FCU. * Default: `true` -* `--fcu.background.commit`: Enables background flush and commit after FCU. +* `--fcu.background.commit`: Returns the FCU response before MDBX flush+commit lands (commit runs in background; remote `rpcdaemon` `latest` stays consistent but can lag for the commit duration). * Default: `false` ### Execution @@ -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/node/ethconfig/config.go b/node/ethconfig/config.go index 6c795df23a5..78db4d8dd96 100644 --- a/node/ethconfig/config.go +++ b/node/ethconfig/config.go @@ -111,9 +111,13 @@ var Defaults = Config{ ProduceE2: true, ProduceE3: true, }, - FcuTimeout: 1 * time.Second, - FcuBackgroundPrune: true, - FcuBackgroundCommit: false, // to enable, we need to 1) have rawdb API go via execctx and 2) revive Coherent cache for rpcdaemon + FcuTimeout: 1 * time.Second, + FcuBackgroundPrune: true, + // FcuBackgroundCommit returns the FCU response before the MDBX commit + // lands; the commit runs in a background goroutine and successive FCUs are + // serialized by the ExecModule semaphore. "Latest" state reads can lag the + // announced head by one block for the commit's duration. + FcuBackgroundCommit: false, ExperimentalBAL: false, WarmupKzgCtxOnInit: true, } diff --git a/rpc/jsonrpc/bor_api_impl.go b/rpc/jsonrpc/bor_api_impl.go index f748b0733cf..d47f0ea997b 100644 --- a/rpc/jsonrpc/bor_api_impl.go +++ b/rpc/jsonrpc/bor_api_impl.go @@ -21,7 +21,6 @@ import ( "errors" "github.com/erigontech/erigon/common" - "github.com/erigontech/erigon/db/rawdb" "github.com/erigontech/erigon/execution/types" "github.com/erigontech/erigon/execution/types/accounts" "github.com/erigontech/erigon/polygon/heimdall" @@ -58,12 +57,11 @@ func (api *BorImpl) GetSnapshot(number *rpc.BlockNumber) (*Snapshot, error) { defer tx.Rollback() // Retrieve the requested block number (or current if none requested) - var header *types.Header - if number == nil || *number == rpc.LatestBlockNumber { - header = rawdb.ReadCurrentHeader(tx) - } else { - header, _ = api.headerByNumber(ctx, *number, tx) + blockNr := rpc.LatestBlockNumber + if number != nil { + blockNr = *number } + header, _ := api.headerByNumber(ctx, blockNr, tx) // Ensure we have an actually valid block if header == nil { return nil, errUnknownBlock @@ -99,22 +97,12 @@ func (api *BorImpl) GetAuthor(blockNrOrHash *rpc.BlockNumberOrHash) (accounts.Ad // Retrieve the requested block number (or current if none requested) var header *types.Header - - //nolint:nestif if blockNrOrHash == nil { - latestBlockNum, err2 := rpchelper.GetLatestBlockNumber(tx) - if err2 != nil { - return accounts.NilAddress, err2 - } - header, err = api._blockReader.HeaderByNumber(ctx, tx, latestBlockNum) - } else { - if blockNr, ok := blockNrOrHash.Number(); ok { - header, err = api._blockReader.HeaderByNumber(ctx, tx, uint64(blockNr)) - } else { - if blockHash, ok := blockNrOrHash.Hash(); ok { - header, err = api._blockReader.HeaderByHash(ctx, tx, blockHash) - } - } + header, err = api.headerByNumber(ctx, rpc.LatestBlockNumber, tx) + } else if blockNr, ok := blockNrOrHash.Number(); ok { + header, err = api.headerByNumber(ctx, blockNr, tx) + } else if blockHash, ok := blockNrOrHash.Hash(); ok { + header, err = api.headerByHash(ctx, blockHash, tx) } // Ensure we have an actually valid block and return its snapshot @@ -172,12 +160,11 @@ func (api *BorImpl) GetSigners(number *rpc.BlockNumber) ([]common.Address, error defer tx.Rollback() // Retrieve the requested block number (or current if none requested) - var header *types.Header - if number == nil || *number == rpc.LatestBlockNumber { - header = rawdb.ReadCurrentHeader(tx) - } else { - header, _ = api.headerByNumber(ctx, *number, tx) + blockNr := rpc.LatestBlockNumber + if number != nil { + blockNr = *number } + header, _ := api.headerByNumber(ctx, blockNr, tx) // Ensure we have an actually valid block if header == nil { return nil, errUnknownBlock @@ -298,7 +285,7 @@ func (api *BorImpl) getLatestBlockNum(ctx context.Context) (uint64, error) { } defer tx.Rollback() - return rpchelper.GetLatestBlockNumber(tx) + return rpchelper.GetLatestBlockNumber(api.filters.WithOverlay(tx)) } // GetSnapshotProposer retrieves the in-turn signer at a given block. @@ -312,21 +299,12 @@ func (api *BorImpl) GetSnapshotProposer(blockNrOrHash *rpc.BlockNumberOrHash) (c defer tx.Rollback() var header *types.Header - //nolint:nestif if blockNrOrHash == nil { - header = rawdb.ReadCurrentHeader(tx) - } else { - if blockNr, ok := blockNrOrHash.Number(); ok { - if blockNr == rpc.LatestBlockNumber { - header = rawdb.ReadCurrentHeader(tx) - } else { - header, err = api.headerByNumber(ctx, blockNr, tx) - } - } else { - if blockHash, ok := blockNrOrHash.Hash(); ok { - header, err = api.headerByHash(ctx, blockHash, tx) - } - } + header, err = api.headerByNumber(ctx, rpc.LatestBlockNumber, tx) + } else if blockNr, ok := blockNrOrHash.Number(); ok { + header, err = api.headerByNumber(ctx, blockNr, tx) + } else if blockHash, ok := blockNrOrHash.Hash(); ok { + header, err = api.headerByHash(ctx, blockHash, tx) } if header == nil || err != nil { @@ -352,19 +330,11 @@ func (api *BorImpl) GetSnapshotProposerSequence(blockNrOrHash *rpc.BlockNumberOr // Retrieve the requested block number (or current if none requested) var header *types.Header if blockNrOrHash == nil { - header = rawdb.ReadCurrentHeader(tx) - } else { - if blockNr, ok := blockNrOrHash.Number(); ok { - if blockNr == rpc.LatestBlockNumber { - header = rawdb.ReadCurrentHeader(tx) - } else { - header, err = api.headerByNumber(ctx, blockNr, tx) - } - } else { - if blockHash, ok := blockNrOrHash.Hash(); ok { - header, err = api.headerByHash(ctx, blockHash, tx) - } - } + header, err = api.headerByNumber(ctx, rpc.LatestBlockNumber, tx) + } else if blockNr, ok := blockNrOrHash.Number(); ok { + header, err = api.headerByNumber(ctx, blockNr, tx) + } else if blockHash, ok := blockNrOrHash.Hash(); ok { + header, err = api.headerByHash(ctx, blockHash, tx) } // Ensure we have an actually valid block diff --git a/rpc/jsonrpc/debug_api.go b/rpc/jsonrpc/debug_api.go index f8515afde2d..050aecb8d99 100644 --- a/rpc/jsonrpc/debug_api.go +++ b/rpc/jsonrpc/debug_api.go @@ -106,7 +106,9 @@ func (api *DebugAPIImpl) SetHead(ctx context.Context, number hexutil.Uint64) err } defer tx.Rollback() - currentHead, err := rpchelper.GetLatestBlockNumber(tx) + // Overlay-aware head, so setHead(N) isn't rejected as future while N's + // commit is still in flight. + currentHead, err := rpchelper.GetLatestBlockNumber(api.filters.WithOverlay(tx)) if err != nil { return err } @@ -137,7 +139,9 @@ func (api *DebugAPIImpl) StorageRangeAt(ctx context.Context, blockHash common.Ha } blockNrOrHash := rpc.BlockNumberOrHashWithHash(blockHash, true) - blockNumber, _, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, api.filters) + // nil filters: resolve on the committed view — the storage-range scan reads + // temporal data through the same plain tx (see rpchelper.GetBlockNumber). + blockNumber, _, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, nil) if err != nil { if errors.As(err, &rpc.BlockNotFoundErr{}) { return StorageRangeResult{}, nil @@ -232,7 +236,9 @@ func (api *DebugAPIImpl) AccountRange(ctx context.Context, blockNrOrHash rpc.Blo } } else if _, ok := blockNrOrHash.Hash(); ok { - bn, _, _, err2 := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, api.filters) + // nil filters: resolve on the committed view — the dumper reads temporal + // data through the same plain tx (see rpchelper.GetBlockNumber). + bn, _, _, err2 := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, nil) if err2 != nil { return state.IteratorDump{}, err2 } @@ -556,31 +562,33 @@ func (api *DebugAPIImpl) AccountAt(ctx context.Context, blockHash common.Hash, t } defer tx.Rollback() - header, err := api.headerByHash(ctx, blockHash, tx) + // Committed view: the canonical-hash check and GetAsOf reads below use the + // same plain tx (an overlay-resolved head would have no committed history). + blockNumber, err := api._blockReader.HeaderNumber(ctx, tx, blockHash) if err != nil { - return &AccountResult{}, err + return nil, err } - if header == nil { + if blockNumber == nil { return nil, nil // not error, see https://github.com/erigontech/erigon/issues/1645 } - canonicalHash, ok, err := api._blockReader.CanonicalHash(ctx, tx, header.Number.Uint64()) + canonicalHash, ok, err := api._blockReader.CanonicalHash(ctx, tx, *blockNumber) if err != nil { return nil, err } if !ok { - return nil, fmt.Errorf("canonical hash not found %d", header.Number.Uint64()) + return nil, fmt.Errorf("canonical hash not found %d", *blockNumber) } isCanonical := canonicalHash == blockHash if !isCanonical { return nil, errors.New("block hash is not canonical") } - err = api.BaseAPI.checkPruneHistory(ctx, tx, header.Number.Uint64()) + err = api.BaseAPI.checkPruneHistory(ctx, tx, *blockNumber) if err != nil { return nil, err } - minTxNum, err := api._txNumReader.Min(ctx, tx, header.Number.Uint64()) + minTxNum, err := api._txNumReader.Min(ctx, tx, *blockNumber) if err != nil { return nil, err } @@ -619,19 +627,25 @@ type AccountResult struct { // GetRawHeader implements debug_getRawHeader - returns a an RLP-encoded header, given a block number or hash func (api *DebugAPIImpl) GetRawHeader(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (hexutil.Bytes, error) { + if number, ok := blockNrOrHash.Number(); ok && number == rpc.PendingBlockNumber { + if block := api.pendingBlock(); block != nil { + return rlp.EncodeToBytes(block.Header()) + } + } tx, err := api.db.BeginTemporalRo(ctx) if err != nil { return nil, err } defer tx.Rollback() - n, h, _, err := rpchelper.GetBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, api.filters) + overlayTx := api.filters.WithOverlay(tx) + n, h, _, err := rpchelper.GetBlockNumber(ctx, blockNrOrHash, overlayTx, api._blockReader, nil) if err != nil { if errors.As(err, &rpc.BlockNotFoundErr{}) { return nil, nil // waiting for spec: not error, see Geth and https://github.com/erigontech/erigon/issues/1645 } return nil, err } - header, err := api._blockReader.Header(ctx, tx, h, n) + header, err := api._blockReader.Header(ctx, overlayTx, h, n) if err != nil { return nil, err } diff --git a/rpc/jsonrpc/erigon_block.go b/rpc/jsonrpc/erigon_block.go index 7b75cd7cea1..65ff099bbed 100644 --- a/rpc/jsonrpc/erigon_block.go +++ b/rpc/jsonrpc/erigon_block.go @@ -91,6 +91,8 @@ func (api *ErigonImpl) GetBlockByTimestamp(ctx context.Context, timeStamp rpc.Ti return nil, err } defer tx.Rollback() + // Everything here is a block-table read, so one overlay view keeps the + // head, the search bounds, and the lookups consistent. overlayTx := api.filters.WithOverlay(tx) uintTimestamp := timeStamp.TurnIntoUint64() diff --git a/rpc/jsonrpc/eth_api.go b/rpc/jsonrpc/eth_api.go index b72af9adaa5..116c4bcd48b 100644 --- a/rpc/jsonrpc/eth_api.go +++ b/rpc/jsonrpc/eth_api.go @@ -402,7 +402,8 @@ func (api *BaseAPI) headerByHash(ctx context.Context, hash common.Hash, tx kv.Tx } } - number, err := api._blockReader.HeaderNumber(ctx, tx, hash) + overlayTx := api.filters.WithOverlay(tx) + number, err := api._blockReader.HeaderNumber(ctx, overlayTx, hash) if err != nil { return nil, err } @@ -410,7 +411,7 @@ func (api *BaseAPI) headerByHash(ctx context.Context, hash common.Hash, tx kv.Tx if number == nil { return nil, nil } - return api._blockReader.Header(ctx, tx, hash, *number) + return api._blockReader.Header(ctx, overlayTx, hash, *number) } // checks the pruning state to see if we would hold information about this diff --git a/rpc/jsonrpc/eth_block.go b/rpc/jsonrpc/eth_block.go index 76a0255ce0d..743e4e11087 100644 --- a/rpc/jsonrpc/eth_block.go +++ b/rpc/jsonrpc/eth_block.go @@ -343,7 +343,8 @@ func (api *APIImpl) GetBlockTransactionCountByNumber(ctx context.Context, blockN return &n, nil } - blockNum, blockHash, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNr), tx, api._blockReader, api.filters) + overlayTx := api.filters.WithOverlay(tx) + blockNum, blockHash, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNr), overlayTx, api._blockReader, nil) if err != nil { if errors.As(err, &rpc.BlockNotFoundErr{}) { return nil, nil // not error, see https://github.com/erigontech/erigon/issues/1645 @@ -356,7 +357,7 @@ func (api *APIImpl) GetBlockTransactionCountByNumber(ctx context.Context, blockN return nil, err } - latestBlockNumber, err := rpchelper.GetLatestBlockNumber(tx) + latestBlockNumber, err := rpchelper.GetLatestBlockNumber(overlayTx) if err != nil { return nil, err } @@ -365,7 +366,7 @@ func (api *APIImpl) GetBlockTransactionCountByNumber(ctx context.Context, blockN return nil, nil } - body, txCount, err := api._blockReader.Body(ctx, tx, blockHash, blockNum) + body, txCount, err := api._blockReader.Body(ctx, overlayTx, blockHash, blockNum) if err != nil { return nil, err } @@ -399,7 +400,8 @@ func (api *APIImpl) GetBlockTransactionCountByHash(ctx context.Context, blockHas } defer tx.Rollback() - blockNum, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHash{BlockHash: &blockHash}, tx, api._blockReader, nil) + overlayTx := api.filters.WithOverlay(tx) + blockNum, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHash{BlockHash: &blockHash}, overlayTx, api._blockReader, nil) if err != nil { // (Compatibility) Every other node just return `null` for when the block does not exist. log.Debug("eth_getBlockTransactionCountByHash GetBlockNumber failed", "err", err) @@ -411,7 +413,7 @@ func (api *APIImpl) GetBlockTransactionCountByHash(ctx context.Context, blockHas return nil, err } - _, txCount, err := api._blockReader.Body(ctx, tx, blockHash, blockNum) + _, txCount, err := api._blockReader.Body(ctx, overlayTx, blockHash, blockNum) if err != nil { return nil, err } diff --git a/rpc/jsonrpc/eth_call.go b/rpc/jsonrpc/eth_call.go index 9be27296e01..5a1d25870b7 100644 --- a/rpc/jsonrpc/eth_call.go +++ b/rpc/jsonrpc/eth_call.go @@ -82,8 +82,9 @@ func (api *APIImpl) Call(ctx context.Context, args ethapi2.CallArgs, requestedBl } defer roTx.Rollback() - // Use the block overlay if available — reads uncommitted data from the - // pre-commit overlay so consumers don't need to wait for DB commit. + // The overlay exposes block tables only: "latest" resolves to the + // pre-commit head while temporal state reads still see the last committed + // block (see ethconfig.Defaults.FcuBackgroundCommit). var tx kv.TemporalTx = roTx if api.filters != nil { if sd := api.filters.LatestSD(); sd != nil { @@ -427,14 +428,16 @@ func (api *APIImpl) GetProof(ctx context.Context, address common.Address, storag } defer roTx.Rollback() - requestedBlockNr, _, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, roTx, api._blockReader, api.filters) + // nil filters: resolve on the committed view — getProof gates on and reads + // the same plain roTx (see rpchelper.GetBlockNumber). + blockNumber, _, isLatest, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, roTx, api._blockReader, nil) if err != nil { return nil, err - } else if requestedBlockNr == 0 { + } else if blockNumber == 0 { return nil, errors.New("block not found") } - err = api.BaseAPI.checkPruneHistory(ctx, roTx, uint64(requestedBlockNr)) + err = api.BaseAPI.checkPruneHistory(ctx, roTx, blockNumber) if err != nil { return nil, err } @@ -444,10 +447,10 @@ func (api *APIImpl) GetProof(ctx context.Context, address common.Address, storag storageKeysConverted[i].Hash.SetBytes(s) storageKeysConverted[i].KeyLength = len(s) } - return api.getProof(ctx, roTx, address, storageKeysConverted, rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(requestedBlockNr)), api.logger) + return api.getProof(ctx, roTx, address, storageKeysConverted, blockNumber, isLatest, api.logger) } -func (api *APIImpl) getProof(ctx context.Context, roTx kv.TemporalTx, address common.Address, storageKeys []StorageKeysInfo, blockNrOrHash rpc.BlockNumberOrHash, logger log.Logger) (*accounts.AccProofResult, error) { +func (api *APIImpl) getProof(ctx context.Context, roTx kv.TemporalTx, address common.Address, storageKeys []StorageKeysInfo, blockNumber uint64, isLatest bool, logger log.Logger) (*accounts.AccProofResult, error) { // Output key encoding is a bit special: if the input was a 32-byte hash, it is // returned as such. Otherwise, we apply the QUANTITY encoding mandated by the @@ -464,38 +467,29 @@ func (api *APIImpl) getProof(ctx context.Context, roTx kv.TemporalTx, address co return outputKey } - tx, err := api.db.BeginTemporalRo(ctx) - if err != nil { - return nil, err - } - defer tx.Rollback() // get the root hash from header to validate proofs along the way - header, err := api._blockReader.HeaderByNumber(ctx, roTx, blockNrOrHash.BlockNumber.Uint64()) + header, err := api._blockReader.HeaderByNumber(ctx, roTx, blockNumber) if err != nil { return nil, err } + if header == nil { + return nil, fmt.Errorf("header not found for block %d", blockNumber) + } - domains, err := execctx.NewSharedDomains(ctx, tx, log.New(), execctx.WithoutDeferredBranchUpdates(), execctx.WithSequentialCommitment()) + domains, err := execctx.NewSharedDomains(ctx, roTx, log.New(), execctx.WithoutDeferredBranchUpdates(), execctx.WithSequentialCommitment()) if err != nil { return nil, err } defer domains.Close() sdCtx := domains.GetCommitmentContext() - latestBlock, err := rpchelper.GetLatestBlockNumber(roTx) - if err != nil { - return nil, err - } - if latestBlock < blockNrOrHash.BlockNumber.Uint64() { - return nil, fmt.Errorf("block number is in the future latest=%d requested=%d", latestBlock, blockNrOrHash.BlockNumber.Uint64()) - } - if blockNrOrHash.BlockNumber.Uint64() < latestBlock { + if !isLatest { // Get first txnum of blockNumber+1 to ensure that correct state root will be restored as of blockNumber has been executed - lastTxnInBlock, err := api._txNumReader.Min(ctx, tx, blockNrOrHash.BlockNumber.Uint64()+1) + lastTxnInBlock, err := api._txNumReader.Min(ctx, roTx, blockNumber+1) if err != nil { return nil, err } - commitmentStartingTxNum := tx.Debug().HistoryStartFrom(kv.CommitmentDomain) + commitmentStartingTxNum := roTx.Debug().HistoryStartFrom(kv.CommitmentDomain) if lastTxnInBlock < commitmentStartingTxNum { return nil, fmt.Errorf("%w: commitment start: %d, last tx: %d", state.PrunedError, commitmentStartingTxNum, lastTxnInBlock) } @@ -575,9 +569,14 @@ func (api *APIImpl) getProof(ctx context.Context, roTx kv.TemporalTx, address co } } - reader, err := rpchelper.CreateStateReader(ctx, tx, api._blockReader, blockNrOrHash, 0, api.filters, api.stateCache, api._txNumReader) - if err != nil { - return nil, err + var reader state.StateReader + if isLatest { + reader = rpchelper.NewLatestStateReader(roTx) + } else { + reader, err = rpchelper.CreateHistoryStateReader(ctx, roTx, blockNumber+1, 0, api._txNumReader) + if err != nil { + return nil, err + } } // get storage key proofs @@ -652,7 +651,9 @@ func (api *BaseAPI) getWitness(ctx context.Context, db kv.TemporalRoDB, blockNrO } defer tx.Rollback() - blockNr, hash, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, api.filters) // DoCall cannot be executed on non-canonical blocks + // nil filters: resolve on the committed view — the witness computation reads + // temporal data through the same plain tx (see rpchelper.GetBlockNumber). + blockNr, hash, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, nil) // DoCall cannot be executed on non-canonical blocks if err != nil { return nil, err } diff --git a/rpc/jsonrpc/eth_call_test.go b/rpc/jsonrpc/eth_call_test.go index d645232f900..27657f3c848 100644 --- a/rpc/jsonrpc/eth_call_test.go +++ b/rpc/jsonrpc/eth_call_test.go @@ -39,13 +39,16 @@ import ( "github.com/erigontech/erigon/common/crypto" "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/common/log/v3" + "github.com/erigontech/erigon/db/dbservices" "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/kv/kvcache" "github.com/erigontech/erigon/db/kv/rawdbv3" "github.com/erigontech/erigon/db/rawdb" + "github.com/erigontech/erigon/db/state/execctx" "github.com/erigontech/erigon/db/state/statecfg" "github.com/erigontech/erigon/execution/chain" "github.com/erigontech/erigon/execution/commitment/trie" + "github.com/erigontech/erigon/execution/execmodule" "github.com/erigontech/erigon/execution/execmodule/execmoduletester" "github.com/erigontech/erigon/execution/protocol" "github.com/erigontech/erigon/execution/protocol/params" @@ -429,6 +432,79 @@ func TestGetProof(t *testing.T) { } } +type missingHeaderBlockReader struct { + dbservices.FullBlockReader +} + +func (missingHeaderBlockReader) HeaderByNumber(context.Context, kv.Getter, uint64) (*types.Header, error) { + return nil, nil +} + +func TestGetProofMissingHeader(t *testing.T) { + previousSchema := statecfg.Schema + statecfg.EnableHistoricalCommitment() + t.Cleanup(func() { + statecfg.Schema = previousSchema + }) + + m, bankAddr, _, _ := chainWithDeployedContract(t) + base := newBaseApiForTest(m) + base._blockReader = missingHeaderBlockReader{FullBlockReader: base._blockReader} + api := newEthApiForTest(base, m.DB, nil, nil) + + proof, err := api.GetProof( + context.Background(), + bankAddr, + nil, + bnhPtr(rpc.BlockNumberOrHashWithNumber(6)), + ) + require.EqualError(t, err, "header not found for block 6") + require.Nil(t, proof) +} + +func TestGetProofPinsReadSnapshot(t *testing.T) { + previousSchema := statecfg.Schema + statecfg.EnableHistoricalCommitment() + t.Cleanup(func() { + statecfg.Schema = previousSchema + }) + + m, _, contractAddress, _ := chainWithDeployedContract(t) + + roTx, err := m.DB.BeginTemporalRo(m.Ctx) + require.NoError(t, err) + defer roTx.Rollback() + + publishedDomains, err := execctx.NewSharedDomains(m.Ctx, roTx, m.Log) + require.NoError(t, err) + defer publishedDomains.Close() + + storageKey := common.Hash{} + compositeKey := make([]byte, 0, len(contractAddress)+len(storageKey)) + compositeKey = append(compositeKey, contractAddress[:]...) + compositeKey = append(compositeKey, storageKey[:]...) + require.NoError(t, publishedDomains.DomainPut(kv.StorageDomain, roTx, compositeKey, []byte{3}, 1, nil)) + + stateCache := &execmodule.Cache{} + stateCache.SetPublishedSD(func() *execctx.SharedDomains { return publishedDomains }) + base := newBaseApiForTest(m) + base.stateCache = stateCache + api := newEthApiForTest(base, m.DB, nil, nil) + + proof, err := api.getProof( + m.Ctx, + roTx, + contractAddress, + []StorageKeysInfo{{Hash: storageKey, KeyLength: len(storageKey)}}, + 6, + true, + log.New(), + ) + require.NoError(t, err) + require.NotNil(t, proof) + require.Equal(t, uint64(2), (*big.Int)(proof.StorageProof[0].Value).Uint64()) +} + func TestGetBlockByTimestampLatestTime(t *testing.T) { ctx := context.Background() m, _, _ := rpcdaemontest.CreateTestExecModule(t) diff --git a/rpc/jsonrpc/eth_receipts.go b/rpc/jsonrpc/eth_receipts.go index 5332e4dda11..4c58f381218 100644 --- a/rpc/jsonrpc/eth_receipts.go +++ b/rpc/jsonrpc/eth_receipts.go @@ -140,19 +140,18 @@ func exceedsLogQueryLimit(crit filters.FilterCriteria, limit int) bool { // resolveLogsRange resolves a filter's block range. A BlockHash pins the range to that // block; otherwise negative tags are resolved against the chain, defaulting to the // latest executed block. With checkFuture, ranges past the latest executed block are -// rejected as they are resolved. +// rejected as they are resolved. Tags resolve on the committed view of tx (nil +// filters — see rpchelper.GetBlockNumber): callers scan logs through the same tx. func (api *BaseAPI) resolveLogsRange(ctx context.Context, tx kv.Tx, crit filters.FilterCriteria, checkFuture bool) (begin, end uint64, err error) { if crit.BlockHash != nil { - block, err := api.blockByHashWithSenders(ctx, tx, *crit.BlockHash) + number, err := api._blockReader.HeaderNumber(ctx, tx, *crit.BlockHash) if err != nil { return 0, 0, err } - if block == nil { + if number == nil { return 0, 0, fmt.Errorf("block not found: %x", *crit.BlockHash) } - - num := block.NumberU64() - return num, num, nil + return *number, *number, nil } latest, _, _, err := rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(rpc.LatestExecutedBlockNumber), tx, api._blockReader, nil) @@ -167,7 +166,7 @@ func (api *BaseAPI) resolveLogsRange(ctx context.Context, tx kv.Tx, crit filters begin = uint64(fromBlock) } else { blockNum := rpc.BlockNumber(fromBlock) - begin, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) + begin, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, nil) if err != nil { return 0, 0, err } @@ -184,7 +183,7 @@ func (api *BaseAPI) resolveLogsRange(ctx context.Context, tx kv.Tx, crit filters end = uint64(toBlock) } else { blockNum := rpc.BlockNumber(toBlock) - end, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, api.filters) + end, _, _, err = rpchelper.GetBlockNumber(ctx, rpc.BlockNumberOrHashWithNumber(blockNum), tx, api._blockReader, nil) if err != nil { return 0, 0, err } @@ -228,6 +227,7 @@ func (api *APIImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria) (t return nil, &rpc.CustomError{Message: errInvalidBlockRange, Code: rpc.ErrCodeInvalidParams} } if end > roaring.MaxUint32 { + // Committed view: must agree with the scan below. latest, err := rpchelper.GetLatestBlockNumber(tx) if err != nil { return nil, err diff --git a/rpc/jsonrpc/eth_simulation.go b/rpc/jsonrpc/eth_simulation.go index be67abb6cb2..90e6bacfac7 100644 --- a/rpc/jsonrpc/eth_simulation.go +++ b/rpc/jsonrpc/eth_simulation.go @@ -125,7 +125,9 @@ func (api *APIImpl) SimulateV1(ctx context.Context, req SimulationRequest, block return nil, err } - blockNumber, blockHash, _, err := rpchelper.GetBlockNumber(ctx, blockParameter, tx, api._blockReader, api.filters) + // nil filters: resolve on the committed view — the gate and the simulator + // below read the same plain tx (see rpchelper.GetBlockNumber). + blockNumber, blockHash, _, err := rpchelper.GetBlockNumber(ctx, blockParameter, tx, api._blockReader, nil) if err != nil { return nil, err } diff --git a/rpc/jsonrpc/graphql_api.go b/rpc/jsonrpc/graphql_api.go index 8b88d2f1926..e275ce1b20d 100644 --- a/rpc/jsonrpc/graphql_api.go +++ b/rpc/jsonrpc/graphql_api.go @@ -89,7 +89,7 @@ func (api *GraphQLAPIImpl) GetLatestBlockNumber(ctx context.Context) (uint64, er return 0, err } defer tx.Rollback() - return rpchelper.GetLatestBlockNumber(tx) + return rpchelper.GetLatestBlockNumber(api.filters.WithOverlay(tx)) } func (api *GraphQLAPIImpl) GetBlockNumberForTx(ctx context.Context, hash common.Hash) (uint64, bool, error) { diff --git a/rpc/jsonrpc/overlay_api.go b/rpc/jsonrpc/overlay_api.go index efb9e5fd87c..325bc6256ee 100644 --- a/rpc/jsonrpc/overlay_api.go +++ b/rpc/jsonrpc/overlay_api.go @@ -555,6 +555,7 @@ func getBeginEnd(ctx context.Context, tx kv.Tx, api *OverlayAPIImpl, crit filter return 0, 0, fmt.Errorf("end (%d) < begin (%d)", end, begin) } if end > roaring.MaxUint32 { + // Committed view: must agree with the scan. latest, err := rpchelper.GetLatestBlockNumber(tx) if err != nil { return 0, 0, err diff --git a/rpc/jsonrpc/overlay_race_test.go b/rpc/jsonrpc/overlay_race_test.go index b37b3133fa1..322bd02641a 100644 --- a/rpc/jsonrpc/overlay_race_test.go +++ b/rpc/jsonrpc/overlay_race_test.go @@ -19,6 +19,7 @@ package jsonrpc import ( "bytes" "context" + "fmt" "strconv" "testing" @@ -29,6 +30,8 @@ import ( "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/hexutil" + "github.com/erigontech/erigon/db/dbservices" + "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/kv/kvcache" "github.com/erigontech/erigon/db/rawdb" "github.com/erigontech/erigon/db/state/execctx" @@ -41,6 +44,7 @@ import ( "github.com/erigontech/erigon/node/gointerfaces/txpoolproto" "github.com/erigontech/erigon/node/shards" "github.com/erigontech/erigon/rpc" + "github.com/erigontech/erigon/rpc/filters" "github.com/erigontech/erigon/rpc/rpchelper" ) @@ -59,6 +63,11 @@ const ( // misc.CalcBaseFee leaves BaseFee unchanged, making overlayRaceBaseFee a // reliable, deterministic fingerprint for "the code read the overlay head". func newOverlayAheadTestAPI(t *testing.T) (base *BaseAPI, m *execmoduletester.ExecModuleTester, overlayHeader *types.Header) { + base, m, overlayHeader, _ = newOverlayAheadTestAPIWithEvents(t) + return base, m, overlayHeader +} + +func newOverlayAheadTestAPIWithEvents(t *testing.T) (base *BaseAPI, m *execmoduletester.ExecModuleTester, overlayHeader *types.Header, events *shards.Events) { t.Helper() var cfg chain.Config @@ -98,15 +107,44 @@ func newOverlayAheadTestAPI(t *testing.T) (base *BaseAPI, m *execmoduletester.Ex // enough for the reader paths under test to resolve this header as current. require.NoError(t, rawdb.WriteHeader(overlay, overlayHeader)) require.NoError(t, rawdb.WriteHeadHeaderHash(overlay, hash)) + rawdb.WriteForkchoiceHead(overlay, hash) require.NoError(t, rawdb.WriteCanonicalHash(overlay, hash, overlayNumber)) require.NoError(t, rawdb.WriteBody(overlay, hash, overlayNumber, &types.Body{})) - events := shards.NewEvents() + events = shards.NewEvents() events.PublishOverlay(doms) filters := rpchelper.New(ctx, rpchelper.DefaultFiltersConfig, nil, nil, nil, func() {}, m.Log, events) stateCache := kvcache.New(kvcache.DefaultCoherentConfig) base = newBaseApiWithFiltersForTest(filters, stateCache, m) + return base, m, overlayHeader, events +} + +type unpublishOverlayBlockReader struct { + dbservices.FullBlockReader + events *shards.Events + blockNumber uint64 +} + +func (r *unpublishOverlayBlockReader) CanonicalHash(ctx context.Context, tx kv.Getter, blockNum uint64) (common.Hash, bool, error) { + hash, ok, err := r.FullBlockReader.CanonicalHash(ctx, tx, blockNum) + if err == nil && ok && blockNum == r.blockNumber { + r.events.PublishOverlay(nil) + } + return hash, ok, err +} + +func newOverlayUnpublishTestAPI(t *testing.T) (*BaseAPI, *execmoduletester.ExecModuleTester, *types.Header) { + t.Helper() + base, m, overlayHeader, events := newOverlayAheadTestAPIWithEvents(t) + overlay := events.LatestSD().BlockOverlay() + txn := signOverlayRaceTestTx(t, m, 1) + require.NoError(t, rawdb.WriteBody(overlay, overlayHeader.Hash(), overlayHeader.Number.Uint64(), &types.Body{Transactions: []types.Transaction{txn}})) + base._blockReader = &unpublishOverlayBlockReader{ + FullBlockReader: base._blockReader, + events: events, + blockNumber: overlayHeader.Number.Uint64(), + } return base, m, overlayHeader } @@ -211,6 +249,82 @@ func TestTxPoolContent_UsesOverlayHead(t *testing.T) { "pending tx gas price must be derived from the overlay head's base fee, not the stale MDBX head") } +// TestGetBlockTransactionCountByHash_SeesOverlayHead pins that the by-hash +// count resolves the overlay head exactly like its by-number twin: the same +// in-flight block must be visible through both, not null through one of them. +func TestGetBlockTransactionCountByHash_SeesOverlayHead(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayAheadTestAPI(t) + api := newEthApiForTest(base, m.DB, nil, nil) + + byNumber, err := api.GetBlockTransactionCountByNumber(m.Ctx, rpc.BlockNumber(overlayHeader.Number.Uint64())) + require.NoError(t, err) + require.NotNil(t, byNumber) + + byHash, err := api.GetBlockTransactionCountByHash(m.Ctx, overlayHeader.Hash()) + require.NoError(t, err) + require.NotNil(t, byHash, "by-hash count must see the overlay head the by-number count sees") + require.Equal(t, *byNumber, *byHash) +} + +func TestGetBlockTransactionCountByNumber_PinsOverlayView(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayUnpublishTestAPI(t) + api := newEthApiForTest(base, m.DB, nil, nil) + + count, err := api.GetBlockTransactionCountByNumber(m.Ctx, rpc.BlockNumber(overlayHeader.Number.Uint64())) + require.NoError(t, err) + require.NotNil(t, count) + require.Equal(t, hexutil.Uint(1), *count) +} + +func TestGetBlockTransactionCountByHash_PinsOverlayView(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayUnpublishTestAPI(t) + api := newEthApiForTest(base, m.DB, nil, nil) + + count, err := api.GetBlockTransactionCountByHash(m.Ctx, overlayHeader.Hash()) + require.NoError(t, err) + require.NotNil(t, count) + require.Equal(t, hexutil.Uint(1), *count) +} + +func TestGetRawHeader_PinsOverlayView(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayUnpublishTestAPI(t) + api := NewPrivateDebugAPI(base, m.DB, nil, 0, false) + + header, err := api.GetRawHeader(m.Ctx, rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(overlayHeader.Number.Uint64()))) + require.NoError(t, err) + require.NotNil(t, header) +} + +// TestDebugAccountAt_OverlayHeadHash_CommittedView pins that debug_accountAt +// resolves the block hash on the committed view: its GetAsOf history reads can +// only see committed data, so an overlay-published head must read as an +// unknown block (null) — not resolve to a header whose canonical-hash check +// then fails. +func TestDebugAccountAt_OverlayHeadHash_CommittedView(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayAheadTestAPI(t) + api := NewPrivateDebugAPI(base, m.DB, nil, 0, false) + + result, err := api.AccountAt(m.Ctx, overlayHeader.Hash(), 0, m.Address) + require.NoError(t, err, "an in-flight (uncommitted) head hash must read as unknown, not error") + require.Nil(t, result) +} + +func TestGetLogsBlockHashUsesCommittedView(t *testing.T) { + t.Parallel() + base, m, overlayHeader := newOverlayAheadTestAPI(t) + api := newEthApiForTest(base, m.DB, nil, nil) + hash := overlayHeader.Hash() + + logs, err := api.GetLogs(m.Ctx, filters.FilterCriteria{BlockHash: &hash}) + require.EqualError(t, err, fmt.Sprintf("block not found: %x", hash)) + require.Nil(t, logs) +} + // TestTxPoolContentFrom_UsesOverlayHead pins that txpool_contentFrom reads the // current header through the block overlay, matching TestTxPoolContent_UsesOverlayHead. func TestTxPoolContentFrom_UsesOverlayHead(t *testing.T) { diff --git a/rpc/jsonrpc/parity_api.go b/rpc/jsonrpc/parity_api.go index 749e364e185..36fa1d9819b 100644 --- a/rpc/jsonrpc/parity_api.go +++ b/rpc/jsonrpc/parity_api.go @@ -73,7 +73,12 @@ func (api *ParityAPIImpl) ListStorageKeys(ctx context.Context, account common.Ad return nil, errors.New("acc not found") } + // Committed view: bn must match the state version the RangeAsOf scan + // below can see (the overlay exposes block tables, not domain data). bn := rawdb.ReadCurrentBlockNumber(tx) + if bn == nil { + return nil, errors.New("current block number not found") + } minTxNum, err := api._txNumReader.Min(ctx, tx, *bn) if err != nil { return nil, err diff --git a/rpc/jsonrpc/trace_filtering.go b/rpc/jsonrpc/trace_filtering.go index b491e890a04..4dc8a424534 100644 --- a/rpc/jsonrpc/trace_filtering.go +++ b/rpc/jsonrpc/trace_filtering.go @@ -324,10 +324,12 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, gas var fromBlock uint64 var toBlock uint64 var err error + // nil filters: resolve tags on the committed view filterV3 scans + // (see rpchelper.GetBlockNumber). if req.FromBlock == nil { fromBlock = 0 } else { - fromBlock, _, _, err = rpchelper.GetBlockNumber(ctx, *req.FromBlock, dbtx, api._blockReader, api.filters) + fromBlock, _, _, err = rpchelper.GetBlockNumber(ctx, *req.FromBlock, dbtx, api._blockReader, nil) if err != nil { if errors.As(err, &rpc.BlockNotFoundErr{}) { stream.WriteEmptyArray() @@ -342,9 +344,12 @@ func (api *TraceAPIImpl) Filter(ctx context.Context, req TraceFilterRequest, gas if err != nil { return err } + if headNumber == nil { + return errors.New("head header not found") + } toBlock = *headNumber } else { - toBlock, _, _, err = rpchelper.GetBlockNumber(ctx, *req.ToBlock, dbtx, api._blockReader, api.filters) + toBlock, _, _, err = rpchelper.GetBlockNumber(ctx, *req.ToBlock, dbtx, api._blockReader, nil) if err != nil { if errors.As(err, &rpc.BlockNotFoundErr{}) { stream.WriteEmptyArray() diff --git a/rpc/jsonrpc/tracing.go b/rpc/jsonrpc/tracing.go index 9c5541403cc..ab41df775fe 100644 --- a/rpc/jsonrpc/tracing.go +++ b/rpc/jsonrpc/tracing.go @@ -58,7 +58,9 @@ func (api *DebugAPIImpl) traceBlock(ctx context.Context, blockNrOrHash rpc.Block } defer tx.Rollback() - blockNumber, hash, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, api.filters) + // nil filters: resolve on the committed view — the replay below reads + // temporal data through the same plain tx (see rpchelper.GetBlockNumber). + blockNumber, hash, _, err := rpchelper.GetCanonicalBlockNumber(ctx, blockNrOrHash, tx, api._blockReader, nil) if err != nil { return err } diff --git a/rpc/rpchelper/filters.go b/rpc/rpchelper/filters.go index fa0beceeb6e..52d952a6ba9 100644 --- a/rpc/rpchelper/filters.go +++ b/rpc/rpchelper/filters.go @@ -1079,9 +1079,12 @@ func (ff *Filters) LatestSD() *execctx.SharedDomains { // WithOverlay returns a read view backed by the latest block overlay if one // is available, otherwise returns the given tx unchanged. The read view uses -// the overlay's in-memory data for table lookups, falling back to the caller's tx -// for data not in the overlay. +// the overlay's in-memory data for table lookups, falling back to the caller's +// tx for data not in the overlay. // Safe to call on a nil receiver. +// +// The view stays readable if the publisher closes the SD concurrently (see +// MemoryMutation.newReadViewMut); it must not outlive the caller's tx. func (ff *Filters) WithOverlay(tx kv.Tx) kv.Tx { if ff == nil { return tx @@ -1098,6 +1101,8 @@ func (ff *Filters) WithOverlay(tx kv.Tx) kv.Tx { // WithTemporalOverlay is like WithOverlay but returns kv.TemporalTx directly, // avoiding repeated type assertions at callsites that need temporal access. +// The same concurrent-close safety property as WithOverlay applies (see the +// concurrency note there). func (ff *Filters) WithTemporalOverlay(tx kv.TemporalTx) kv.TemporalTx { if ff == nil { return tx diff --git a/rpc/rpchelper/helper.go b/rpc/rpchelper/helper.go index ce61d683eef..34397f3fbd6 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" @@ -57,6 +56,16 @@ func CheckBlockExecuted(tx kv.Tx, blockNumber uint64) error { return nil } +// GetBlockNumber resolves a block number, hash, or tag to a concrete block number and hash. +// +// Tags resolve against the view tx exposes. Passing the API's Filters +// additionally wraps tx in the block overlay (which includes a head whose +// commit is still in flight) and lets "pending" resolve via LastPendingBlock. +// With nil filters tx is used exactly as passed — a plain tx for +// committed-view resolution, required when the caller then scans data through +// that same tx so the bounds and the scan agree, or a tx the caller already +// overlay-wrapped to pin one view for all its reads; "pending" then falls +// back to the latest executed block. func GetBlockNumber(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash, tx kv.Tx, br dbservices.FullBlockReader, filters *Filters) (uint64, common.Hash, bool, error) { bn, bh, latest, found, err := _GetBlockNumber(ctx, blockNrOrHash.RequireCanonical, blockNrOrHash, tx, br, filters) if err != nil { @@ -120,9 +129,9 @@ func _GetBlockNumber(ctx context.Context, requireCanonical bool, blockNrOrHash r return 0, common.Hash{}, false, false, err } case rpc.PendingBlockNumber: + // nil filters (committed-view resolution) = no pending block known. if filters != nil { - pendingBlock := filters.LastPendingBlock() - if pendingBlock != nil { + if pendingBlock := filters.LastPendingBlock(); pendingBlock != nil { return pendingBlock.NumberU64(), pendingBlock.Hash(), false, true, nil } } @@ -220,15 +229,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 +239,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 }