Merge upstream go-ethereum a6f7b4833 into luxfi/geth

Pulls 35 upstream commits with fixes and improvements:
- cmd/utils: fix DeveloperFlag handling when set to false
- core/stateless: cap witness depth metrics buckets
- eth/fetcher: add metadata validation in tx announcement
- triedb/pathdb: use copy instead of append to reduce memory alloc
- core/rawdb: fix size counting in memory freezer
- p2p/tracker: fix head detection in Fulfil
- eth/tracers/native: include SWAP16 in default ignored opcodes
- core/state: fix incorrect contract code state metrics
- eth/downloader: keep current syncmode in downloader only
- core/vm: fix PC increment for EIP-8024 opcodes
- common/bitutil: deprecate XORBytes in favor of stdlib crypto/subtle
- ethdb/pebble: change the Pebble database configuration
- eth/filters: change error code for invalid parameter errors
- beacon/types: update for fulu
- core: log detailed statistics for slow block

All luxfi packages preserved (crypto, node, ids).
This commit is contained in:
Zach Kelling
2025-12-12 16:38:48 -08:00
71 changed files with 1122 additions and 558 deletions
+1 -1
View File
@@ -52,7 +52,7 @@ func BlockFromJSON(forkName string, data []byte) (*BeaconBlock, error) {
obj = new(capella.BeaconBlock)
case "deneb":
obj = new(deneb.BeaconBlock)
case "electra":
case "electra", "fulu":
obj = new(electra.BeaconBlock)
default:
return nil, fmt.Errorf("unsupported fork: %s", forkName)
+1 -1
View File
@@ -45,7 +45,7 @@ func ExecutionHeaderFromJSON(forkName string, data []byte) (*ExecutionHeader, er
switch forkName {
case "capella":
obj = new(capella.ExecutionPayloadHeader)
case "deneb", "electra": // note: the payload type was not changed in electra
case "deneb", "electra", "fulu": // note: the payload type was not changed in electra/fulu
obj = new(deneb.ExecutionPayloadHeader)
default:
return nil, fmt.Errorf("unsupported fork: %s", forkName)
+8 -3
View File
@@ -343,7 +343,7 @@ func buildFlags(env build.Environment, staticLinking bool, buildTags []string) (
}
ld = append(ld, "-extldflags", "'"+strings.Join(extld, " ")+"'")
}
// TODO(gballet): revisit after the input api has been defined
// TODO(gballet): revisit after the input api has been defined
if runtime.GOARCH == "wasm" {
ld = append(ld, "-gcflags=all=-d=softfloat")
}
@@ -462,9 +462,14 @@ func doCheckGenerate() {
)
pathList := []string{filepath.Join(protocPath, "bin"), protocGenGoPath, os.Getenv("PATH")}
excludes := []string{"tests/testdata", "build/cache", ".git"}
for i := range excludes {
excludes[i] = filepath.FromSlash(excludes[i])
}
for _, mod := range goModules {
// Compute the origin hashes of all the files
hashes, err := build.HashFolder(mod, []string{"tests/testdata", "build/cache", ".git"})
hashes, err := build.HashFolder(mod, excludes)
if err != nil {
log.Fatal("Error computing hashes", "err", err)
}
@@ -474,7 +479,7 @@ func doCheckGenerate() {
c.Dir = mod
build.MustRun(c)
// Check if generate file hashes have changed
generated, err := build.HashFolder(mod, []string{"tests/testdata", "build/cache", ".git"})
generated, err := build.HashFolder(mod, excludes)
if err != nil {
log.Fatalf("Error re-computing hashes: %v", err)
}
+58 -13
View File
@@ -17,16 +17,18 @@
package main
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"maps"
"os"
"regexp"
"slices"
"github.com/luxfi/geth/common"
"github.com/luxfi/geth/core"
"github.com/luxfi/geth/core/rawdb"
"github.com/luxfi/geth/log"
"github.com/luxfi/geth/tests"
"github.com/urfave/cli/v2"
)
@@ -34,33 +36,52 @@ import (
var blockTestCommand = &cli.Command{
Action: blockTestCmd,
Name: "blocktest",
Usage: "Executes the given blockchain tests",
Usage: "Executes the given blockchain tests. Filenames can be fed via standard input (batch mode) or as an argument (one-off execution).",
ArgsUsage: "<path>",
Flags: slices.Concat([]cli.Flag{
DumpFlag,
HumanReadableFlag,
RunFlag,
WitnessCrossCheckFlag,
FuzzFlag,
}, traceFlags),
}
func blockTestCmd(ctx *cli.Context) error {
path := ctx.Args().First()
if len(path) == 0 {
return errors.New("path argument required")
// If path is provided, run the tests at that path.
if len(path) != 0 {
var (
collected = collectFiles(path)
results []testResult
)
for _, fname := range collected {
r, err := runBlockTest(ctx, fname)
if err != nil {
return err
}
results = append(results, r...)
}
report(ctx, results)
return nil
}
var (
collected = collectFiles(path)
results []testResult
)
for _, fname := range collected {
r, err := runBlockTest(ctx, fname)
// Otherwise, read filenames from stdin and execute back-to-back.
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
fname := scanner.Text()
if len(fname) == 0 {
return nil
}
results, err := runBlockTest(ctx, fname)
if err != nil {
return err
}
results = append(results, r...)
// During fuzzing, we report the result after every block
if !ctx.IsSet(FuzzFlag.Name) {
report(ctx, results)
}
}
report(ctx, results)
return nil
}
@@ -79,6 +100,11 @@ func runBlockTest(ctx *cli.Context, fname string) ([]testResult, error) {
}
tracer := tracerFromFlags(ctx)
// Suppress INFO logs during fuzzing
if ctx.IsSet(FuzzFlag.Name) {
log.SetDefault(log.NewLogger(log.DiscardHandler()))
}
// Pull out keys to sort and ensure tests are run in order.
keys := slices.Sorted(maps.Keys(tests))
@@ -88,16 +114,35 @@ func runBlockTest(ctx *cli.Context, fname string) ([]testResult, error) {
if !re.MatchString(name) {
continue
}
test := tests[name]
result := &testResult{Name: name, Pass: true}
if err := tests[name].Run(false, rawdb.PathScheme, ctx.Bool(WitnessCrossCheckFlag.Name), tracer, func(res error, chain *core.BlockChain) {
var finalRoot *common.Hash
if err := test.Run(false, rawdb.PathScheme, ctx.Bool(WitnessCrossCheckFlag.Name), tracer, func(res error, chain *core.BlockChain) {
if ctx.Bool(DumpFlag.Name) {
if s, _ := chain.State(); s != nil {
result.State = dump(s)
}
}
// Capture final state root for end marker
if chain != nil {
root := chain.CurrentBlock().Root
finalRoot = &root
}
}); err != nil {
result.Pass, result.Error = false, err.Error()
}
// Always assign fork (regardless of pass/fail or tracer)
result.Fork = test.Network()
// Assign root if test succeeded
if result.Pass && finalRoot != nil {
result.Root = finalRoot
}
// When fuzzing, write results after every block
if ctx.IsSet(FuzzFlag.Name) {
report(ctx, []testResult{*result})
}
results = append(results, *result)
}
return results, nil
+5
View File
@@ -55,6 +55,11 @@ var (
Usage: "benchmark the execution",
Category: flags.VMCategory,
}
FuzzFlag = &cli.BoolFlag{
Name: "fuzz",
Usage: "adapts output format for fuzzing",
Category: flags.VMCategory,
}
WitnessCrossCheckFlag = &cli.BoolFlag{
Name: "cross-check",
Aliases: []string{"xc"},
+1
View File
@@ -96,6 +96,7 @@ if one is set. Otherwise it prints the genesis from the datadir.`,
utils.CacheNoPrefetchFlag,
utils.CachePreimagesFlag,
utils.NoCompactionFlag,
utils.LogSlowBlockFlag,
utils.MetricsEnabledFlag,
utils.MetricsEnabledExpensiveFlag,
utils.MetricsHTTPFlag,
+2 -1
View File
@@ -39,8 +39,9 @@ const (
// child g gets a temporary data directory.
func runMinimalGeth(t *testing.T, args ...string) *testgeth {
// --holesky to make the 'writing genesis to disk' faster (no accounts)
// --networkid=1337 to avoid cache bump
// --syncmode=full to avoid allocating fast sync bloom
allArgs := []string{"--holesky", "--authrpc.port", "0", "--syncmode=full", "--port", "0",
allArgs := []string{"--holesky", "--networkid", "1337", "--authrpc.port", "0", "--syncmode=full", "--port", "0",
"--nat", "none", "--nodiscover", "--maxpeers", "0", "--cache", "64",
"--datadir.minfreedisk", "0"}
return runGeth(t, append(allArgs, args...)...)
+1
View File
@@ -157,6 +157,7 @@ var (
utils.BeaconGenesisTimeFlag,
utils.BeaconCheckpointFlag,
utils.BeaconCheckpointFileFlag,
utils.LogSlowBlockFlag,
}, utils.NetworkFlags, utils.DatabaseFlags)
rpcFlags = []cli.Flag{
+40 -18
View File
@@ -137,7 +137,7 @@ var (
}
NetworkIdFlag = &cli.Uint64Flag{
Name: "networkid",
Usage: "Explicitly set network id (integer)(For testnets: use --sepolia, --holesky, --hoodi instead)",
Usage: "Explicitly set network ID (integer)(For testnets: use --sepolia, --holesky, --hoodi instead)",
Value: ethconfig.Defaults.NetworkId,
Category: flags.EthCategory,
}
@@ -672,6 +672,12 @@ var (
Usage: "Disables db compaction after import",
Category: flags.LoggingCategory,
}
LogSlowBlockFlag = &cli.DurationFlag{
Name: "debug.logslowblock",
Usage: "Block execution time threshold beyond which detailed statistics will be logged (0 means disable)",
Value: ethconfig.Defaults.SlowBlockThreshold,
Category: flags.LoggingCategory,
}
// MISC settings
SyncTargetFlag = &cli.StringFlag{
@@ -864,14 +870,14 @@ var (
Aliases: []string{"discv4"},
Usage: "Enables the V4 discovery mechanism",
Category: flags.NetworkingCategory,
Value: true,
Value: node.DefaultConfig.P2P.DiscoveryV4,
}
DiscoveryV5Flag = &cli.BoolFlag{
Name: "discovery.v5",
Aliases: []string{"discv5"},
Usage: "Enables the V5 discovery mechanism",
Category: flags.NetworkingCategory,
Value: true,
Value: node.DefaultConfig.P2P.DiscoveryV5,
}
NetrestrictFlag = &cli.StringFlag{
Name: "netrestrict",
@@ -1368,13 +1374,17 @@ func SetP2PConfig(ctx *cli.Context, cfg *p2p.Config) {
cfg.MaxPendingPeers = ctx.Int(MaxPendingPeersFlag.Name)
}
if ctx.IsSet(NoDiscoverFlag.Name) {
cfg.NoDiscovery = true
cfg.NoDiscovery = ctx.Bool(NoDiscoverFlag.Name)
}
flags.CheckExclusive(ctx, DiscoveryV4Flag, NoDiscoverFlag)
flags.CheckExclusive(ctx, DiscoveryV5Flag, NoDiscoverFlag)
cfg.DiscoveryV4 = ctx.Bool(DiscoveryV4Flag.Name)
cfg.DiscoveryV5 = ctx.Bool(DiscoveryV5Flag.Name)
if ctx.IsSet(DiscoveryV4Flag.Name) {
cfg.DiscoveryV4 = ctx.Bool(DiscoveryV4Flag.Name)
}
if ctx.IsSet(DiscoveryV5Flag.Name) {
cfg.DiscoveryV5 = ctx.Bool(DiscoveryV5Flag.Name)
}
if netrestrict := ctx.String(NetrestrictFlag.Name); netrestrict != "" {
list, err := netutil.ParseNetlist(netrestrict)
@@ -1420,7 +1430,7 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
cfg.KeyStoreDir = ctx.String(KeyStoreDirFlag.Name)
}
if ctx.IsSet(DeveloperFlag.Name) {
cfg.UseLightweightKDF = true
cfg.UseLightweightKDF = ctx.Bool(DeveloperFlag.Name)
}
if ctx.IsSet(LightKDFFlag.Name) {
cfg.UseLightweightKDF = ctx.Bool(LightKDFFlag.Name)
@@ -1577,8 +1587,7 @@ func setMiner(ctx *cli.Context, cfg *miner.Config) {
cfg.Recommit = ctx.Duration(MinerNewPayloadTimeoutFlag.Name)
}
if ctx.IsSet(MinerMaxBlobsFlag.Name) {
maxBlobs := ctx.Int(MinerMaxBlobsFlag.Name)
cfg.MaxBlobsPerBlock = &maxBlobs
cfg.MaxBlobsPerBlock = ctx.Int(MinerMaxBlobsFlag.Name)
}
}
@@ -1612,8 +1621,8 @@ func setRequiredBlocks(ctx *cli.Context, cfg *ethconfig.Config) {
// SetEthConfig applies eth-related command line flags to the config.
func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
// Avoid conflicting network flags, don't allow network id override on preset networks
flags.CheckExclusive(ctx, MainnetFlag, DeveloperFlag, SepoliaFlag, HoleskyFlag, HoodiFlag, NetworkIdFlag, OverrideGenesisFlag)
// Avoid conflicting network flags
flags.CheckExclusive(ctx, MainnetFlag, DeveloperFlag, SepoliaFlag, HoleskyFlag, HoodiFlag, OverrideGenesisFlag)
flags.CheckExclusive(ctx, DeveloperFlag, ExternalSignerFlag) // Can't use both ephemeral unlocked and external signer
// Set configurations from CLI flags
@@ -1660,9 +1669,6 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
}
}
if ctx.IsSet(NetworkIdFlag.Name) {
cfg.NetworkId = ctx.Uint64(NetworkIdFlag.Name)
}
if ctx.IsSet(CacheFlag.Name) || ctx.IsSet(CacheDatabaseFlag.Name) {
cfg.DatabaseCache = ctx.Int(CacheFlag.Name) * ctx.Int(CacheDatabaseFlag.Name) / 100
}
@@ -1683,8 +1689,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
if ctx.IsSet(CacheNoPrefetchFlag.Name) {
cfg.NoPrefetch = ctx.Bool(CacheNoPrefetchFlag.Name)
}
// Read the value from the flag no matter if it's set or not.
cfg.Preimages = ctx.Bool(CachePreimagesFlag.Name)
if ctx.IsSet(CachePreimagesFlag.Name) {
cfg.Preimages = ctx.Bool(CachePreimagesFlag.Name)
}
if cfg.NoPruning && !cfg.Preimages {
cfg.Preimages = true
log.Info("Enabling recording of key preimages since archive mode is used")
@@ -1717,7 +1724,10 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
cfg.LogHistory = ctx.Uint64(LogHistoryFlag.Name)
}
if ctx.IsSet(LogNoHistoryFlag.Name) {
cfg.LogNoHistory = true
cfg.LogNoHistory = ctx.Bool(LogNoHistoryFlag.Name)
}
if ctx.IsSet(LogSlowBlockFlag.Name) {
cfg.SlowBlockThreshold = ctx.Duration(LogSlowBlockFlag.Name)
}
if ctx.IsSet(LogExportCheckpointsFlag.Name) {
cfg.LogExportCheckpoints = ctx.String(LogExportCheckpointsFlag.Name)
@@ -1912,10 +1922,18 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *ethconfig.Config) {
}
cfg.Genesis = genesis
default:
if cfg.NetworkId == 1 {
if ctx.Uint64(NetworkIdFlag.Name) == 1 {
SetDNSDiscoveryDefaults(cfg, params.MainnetGenesisHash)
}
}
if ctx.IsSet(NetworkIdFlag.Name) {
// Typically it's best to automatically set the network ID to the chainID,
// by not passing the --networkid flag at all. Emit a warning when set
// explicitly in case overriding the network ID is not the user's intention.
id := ctx.Uint64(NetworkIdFlag.Name)
log.Warn("Setting network ID with command-line flag", "id", id)
cfg.NetworkId = id
}
// Set any dangling config values
if ctx.String(CryptoKZGFlag.Name) != "gokzg" && ctx.String(CryptoKZGFlag.Name) != "ckzg" {
Fatalf("--%s flag must be 'gokzg' or 'ckzg'", CryptoKZGFlag.Name)
@@ -2290,6 +2308,7 @@ func MakeChain(ctx *cli.Context, stack *node.Node, readonly bool) (*core.BlockCh
Preimages: ctx.Bool(CachePreimagesFlag.Name),
StateScheme: scheme,
StateHistory: ctx.Uint64(StateHistoryFlag.Name),
// Disable transaction indexing/unindexing.
TxLookupLimit: -1,
@@ -2301,6 +2320,9 @@ func MakeChain(ctx *cli.Context, stack *node.Node, readonly bool) (*core.BlockCh
// Enable state size tracking if enabled
StateSizeTracking: ctx.Bool(StateSizeTrackingFlag.Name),
// Configure the slow block statistic logger
SlowBlockThreshold: ctx.Duration(LogSlowBlockFlag.Name),
}
if options.ArchiveMode && !options.Preimages {
options.Preimages = true
+10 -39
View File
@@ -8,6 +8,7 @@
package bitutil
import (
"crypto/subtle"
"runtime"
"unsafe"
)
@@ -17,46 +18,16 @@ const supportsUnaligned = runtime.GOARCH == "386" || runtime.GOARCH == "amd64" |
// XORBytes xors the bytes in a and b. The destination is assumed to have enough
// space. Returns the number of bytes xor'd.
//
// If dst does not have length at least n,
// XORBytes panics without writing anything to dst.
//
// dst and x or y may overlap exactly or not at all,
// otherwise XORBytes may panic.
//
// Deprecated: use crypto/subtle.XORBytes
func XORBytes(dst, a, b []byte) int {
if supportsUnaligned {
return fastXORBytes(dst, a, b)
}
return safeXORBytes(dst, a, b)
}
// fastXORBytes xors in bulk. It only works on architectures that support
// unaligned read/writes.
func fastXORBytes(dst, a, b []byte) int {
n := len(a)
if len(b) < n {
n = len(b)
}
w := n / wordSize
if w > 0 {
dw := *(*[]uintptr)(unsafe.Pointer(&dst))
aw := *(*[]uintptr)(unsafe.Pointer(&a))
bw := *(*[]uintptr)(unsafe.Pointer(&b))
for i := 0; i < w; i++ {
dw[i] = aw[i] ^ bw[i]
}
}
for i := n - n%wordSize; i < n; i++ {
dst[i] = a[i] ^ b[i]
}
return n
}
// safeXORBytes xors one by one. It works on all architectures, independent if
// it supports unaligned read/writes or not.
func safeXORBytes(dst, a, b []byte) int {
n := len(a)
if len(b) < n {
n = len(b)
}
for i := 0; i < n; i++ {
dst[i] = a[i] ^ b[i]
}
return n
return subtle.XORBytes(dst, a, b)
}
// ANDBytes ands the bytes in a and b. The destination is assumed to have enough
+14 -2
View File
@@ -29,7 +29,7 @@ func TestXOR(t *testing.T) {
d2 := make([]byte, 1023+alignD)[alignD:]
XORBytes(d1, p, q)
safeXORBytes(d2, p, q)
naiveXOR(d2, p, q)
if !bytes.Equal(d1, d2) {
t.Error("not equal", d1, d2)
}
@@ -38,6 +38,18 @@ func TestXOR(t *testing.T) {
}
}
// naiveXOR xors bytes one by one.
func naiveXOR(dst, a, b []byte) int {
n := len(a)
if len(b) < n {
n = len(b)
}
for i := 0; i < n; i++ {
dst[i] = a[i] ^ b[i]
}
return n
}
// Tests that bitwise AND works for various alignments.
func TestAND(t *testing.T) {
for alignP := 0; alignP < 2; alignP++ {
@@ -134,7 +146,7 @@ func benchmarkBaseXOR(b *testing.B, size int) {
p, q := make([]byte, size), make([]byte, size)
for i := 0; i < b.N; i++ {
safeXORBytes(p, p, q)
naiveXOR(p, p, q)
}
}
+97 -65
View File
@@ -199,6 +199,10 @@ type BlockChainConfig struct {
// StateSizeTracking indicates whether the state size tracking is enabled.
StateSizeTracking bool
// SlowBlockThreshold is the block execution time threshold beyond which
// detailed statistics will be logged.
SlowBlockThreshold time.Duration
}
// DefaultConfig returns the default config.
@@ -338,7 +342,8 @@ type BlockChain struct {
logger *tracing.Hooks
stateSizer *state.SizeTracker // State size tracking
lastForkReadyAlert time.Time // Last time there was a fork readiness print out
lastForkReadyAlert time.Time // Last time there was a fork readiness print out
slowBlockThreshold time.Duration // Block execution time threshold beyond which detailed statistics will be logged
}
// NewBlockChain returns a fully initialised block chain using information
@@ -378,19 +383,20 @@ func NewBlockChain(db ethdb.Database, genesis *Genesis, engine consensus.Engine,
}
bc := &BlockChain{
chainConfig: chainConfig,
cfg: cfg,
db: db,
triedb: triedb,
triegc: prque.New[int64, common.Hash](nil),
chainmu: syncx.NewClosableMutex(),
bodyCache: lru.NewCache[common.Hash, *types.Body](bodyCacheLimit),
bodyRLPCache: lru.NewCache[common.Hash, rlp.RawValue](bodyCacheLimit),
receiptsCache: lru.NewCache[common.Hash, []*types.Receipt](receiptsCacheLimit),
blockCache: lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
engine: engine,
logger: cfg.VmConfig.Tracer,
chainConfig: chainConfig,
cfg: cfg,
db: db,
triedb: triedb,
triegc: prque.New[int64, common.Hash](nil),
chainmu: syncx.NewClosableMutex(),
bodyCache: lru.NewCache[common.Hash, *types.Body](bodyCacheLimit),
bodyRLPCache: lru.NewCache[common.Hash, rlp.RawValue](bodyCacheLimit),
receiptsCache: lru.NewCache[common.Hash, []*types.Receipt](receiptsCacheLimit),
blockCache: lru.NewCache[common.Hash, *types.Block](blockCacheLimit),
txLookupCache: lru.NewCache[common.Hash, txLookup](txLookupCacheLimit),
engine: engine,
logger: cfg.VmConfig.Tracer,
slowBlockThreshold: cfg.SlowBlockThreshold,
}
bc.hc, err = NewHeaderChain(db, chainConfig, engine, bc.insertStopped)
if err != nil {
@@ -1665,15 +1671,22 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
if err := blockBatch.Write(); err != nil {
log.Crit("Failed to write block into disk", "err", err)
}
// Commit all cached state changes into underlying memory database.
root, stateUpdate, err := statedb.CommitWithUpdate(block.NumberU64(), bc.chainConfig.IsEIP158(block.Number()), bc.chainConfig.IsCancun(block.Number(), block.Time()))
var (
err error
root common.Hash
isEIP158 = bc.chainConfig.IsEIP158(block.Number())
isCancun = bc.chainConfig.IsCancun(block.Number(), block.Time())
)
if bc.stateSizer == nil {
root, err = statedb.Commit(block.NumberU64(), isEIP158, isCancun)
} else {
root, err = statedb.CommitAndTrack(block.NumberU64(), isEIP158, isCancun, bc.stateSizer)
}
if err != nil {
return err
}
// Emit the state update to the state sizestats if it's active
if bc.stateSizer != nil {
bc.stateSizer.Notify(stateUpdate)
}
// If node is running in path mode, skip explicit gc operation
// which is unnecessary in this mode.
if bc.triedb.Scheme() == rawdb.PathScheme {
@@ -1909,7 +1922,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
// still need re-execution to generate snapshots that are missing
case err != nil && !errors.Is(err, ErrKnownBlock):
stats.ignored += len(it.chain)
bc.reportBlock(block, nil, err)
bc.reportBadBlock(block, nil, err)
return nil, it.index, err
}
// Track the singleton witness from this chain insertion (if any)
@@ -1977,6 +1990,14 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool, makeWitness
if err != nil {
return nil, it.index, err
}
res.stats.reportMetrics()
// Log slow block only if a single block is inserted (usually after the
// initial sync) to not overwhelm the users.
if len(chain) == 1 {
res.stats.logSlow(block, bc.slowBlockThreshold)
}
// Report the import stats before returning the various results
stats.processed++
stats.usedGas += res.usedGas
@@ -2037,15 +2058,20 @@ type blockProcessingResult struct {
procTime time.Duration
status WriteStatus
witness *stateless.Witness
stats *ExecuteStats
}
func (bpr *blockProcessingResult) Witness() *stateless.Witness {
return bpr.witness
}
func (bpr *blockProcessingResult) Stats() *ExecuteStats {
return bpr.stats
}
// ProcessBlock executes and validates the given block. If there was no error
// it writes the block and associated state to database.
func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, setHead bool, makeWitness bool) (_ *blockProcessingResult, blockEndErr error) {
func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, setHead bool, makeWitness bool) (result *blockProcessingResult, blockEndErr error) {
var (
err error
startTime = time.Now()
@@ -2079,16 +2105,22 @@ func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, s
}
// Upload the statistics of reader at the end
defer func() {
stats := prefetch.GetStats()
accountCacheHitPrefetchMeter.Mark(stats.AccountHit)
accountCacheMissPrefetchMeter.Mark(stats.AccountMiss)
storageCacheHitPrefetchMeter.Mark(stats.StorageHit)
storageCacheMissPrefetchMeter.Mark(stats.StorageMiss)
stats = process.GetStats()
accountCacheHitMeter.Mark(stats.AccountHit)
accountCacheMissMeter.Mark(stats.AccountMiss)
storageCacheHitMeter.Mark(stats.StorageHit)
storageCacheMissMeter.Mark(stats.StorageMiss)
pStat := prefetch.GetStats()
accountCacheHitPrefetchMeter.Mark(pStat.AccountCacheHit)
accountCacheMissPrefetchMeter.Mark(pStat.AccountCacheMiss)
storageCacheHitPrefetchMeter.Mark(pStat.StorageCacheHit)
storageCacheMissPrefetchMeter.Mark(pStat.StorageCacheMiss)
rStat := process.GetStats()
accountCacheHitMeter.Mark(rStat.AccountCacheHit)
accountCacheMissMeter.Mark(rStat.AccountCacheMiss)
storageCacheHitMeter.Mark(rStat.StorageCacheHit)
storageCacheMissMeter.Mark(rStat.StorageCacheMiss)
if result != nil {
result.stats.StatePrefetchCacheStats = pStat
result.stats.StateReadCacheStats = rStat
}
}()
go func(start time.Time, throwaway *state.StateDB, block *types.Block) {
@@ -2145,14 +2177,14 @@ func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, s
pstart := time.Now()
res, err := bc.processor.Process(block, statedb, bc.cfg.VmConfig)
if err != nil {
bc.reportBlock(block, res, err)
bc.reportBadBlock(block, res, err)
return nil, err
}
ptime := time.Since(pstart)
vstart := time.Now()
if err := bc.validator.ValidateState(block, statedb, res, false); err != nil {
bc.reportBlock(block, res, err)
bc.reportBadBlock(block, res, err)
return nil, err
}
vtime := time.Since(vstart)
@@ -2186,26 +2218,28 @@ func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, s
}
}
xvtime := time.Since(xvstart)
proctime := time.Since(startTime) // processing + validation + cross validation
var (
xvtime = time.Since(xvstart)
proctime = time.Since(startTime) // processing + validation + cross validation
stats = &ExecuteStats{}
)
// Update the metrics touched during block processing and validation
accountReadTimer.Update(statedb.AccountReads) // Account reads are complete(in processing)
storageReadTimer.Update(statedb.StorageReads) // Storage reads are complete(in processing)
if statedb.AccountLoaded != 0 {
accountReadSingleTimer.Update(statedb.AccountReads / time.Duration(statedb.AccountLoaded))
}
if statedb.StorageLoaded != 0 {
storageReadSingleTimer.Update(statedb.StorageReads / time.Duration(statedb.StorageLoaded))
}
accountUpdateTimer.Update(statedb.AccountUpdates) // Account updates are complete(in validation)
storageUpdateTimer.Update(statedb.StorageUpdates) // Storage updates are complete(in validation)
accountHashTimer.Update(statedb.AccountHashes) // Account hashes are complete(in validation)
triehash := statedb.AccountHashes // The time spent on tries hashing
trieUpdate := statedb.AccountUpdates + statedb.StorageUpdates // The time spent on tries update
blockExecutionTimer.Update(ptime - (statedb.AccountReads + statedb.StorageReads)) // The time spent on EVM processing
blockValidationTimer.Update(vtime - (triehash + trieUpdate)) // The time spent on block validation
blockCrossValidationTimer.Update(xvtime) // The time spent on stateless cross validation
stats.AccountReads = statedb.AccountReads // Account reads are complete(in processing)
stats.StorageReads = statedb.StorageReads // Storage reads are complete(in processing)
stats.AccountUpdates = statedb.AccountUpdates // Account updates are complete(in validation)
stats.StorageUpdates = statedb.StorageUpdates // Storage updates are complete(in validation)
stats.AccountHashes = statedb.AccountHashes // Account hashes are complete(in validation)
stats.AccountLoaded = statedb.AccountLoaded
stats.AccountUpdated = statedb.AccountUpdated
stats.AccountDeleted = statedb.AccountDeleted
stats.StorageLoaded = statedb.StorageLoaded
stats.StorageUpdated = int(statedb.StorageUpdated.Load())
stats.StorageDeleted = int(statedb.StorageDeleted.Load())
stats.Execution = ptime - (statedb.AccountReads + statedb.StorageReads) // The time spent on EVM processing
stats.Validation = vtime - (statedb.AccountHashes + statedb.AccountUpdates + statedb.StorageUpdates) // The time spent on block validation
stats.CrossValidation = xvtime // The time spent on stateless cross validation
// Write the block to the chain and get the status.
var (
@@ -2227,24 +2261,22 @@ func (bc *BlockChain) ProcessBlock(parentRoot common.Hash, block *types.Block, s
}
// Update the metrics touched during block commit
accountCommitTimer.Update(statedb.AccountCommits) // Account commits are complete, we can mark them
storageCommitTimer.Update(statedb.StorageCommits) // Storage commits are complete, we can mark them
snapshotCommitTimer.Update(statedb.SnapshotCommits) // Snapshot commits are complete, we can mark them
triedbCommitTimer.Update(statedb.TrieDBCommits) // Trie database commits are complete, we can mark them
stats.AccountCommits = statedb.AccountCommits // Account commits are complete, we can mark them
stats.StorageCommits = statedb.StorageCommits // Storage commits are complete, we can mark them
stats.SnapshotCommit = statedb.SnapshotCommits // Snapshot commits are complete, we can mark them
stats.TrieDBCommit = statedb.TrieDBCommits // Trie database commits are complete, we can mark them
stats.BlockWrite = time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits
blockWriteTimer.Update(time.Since(wstart) - max(statedb.AccountCommits, statedb.StorageCommits) /* concurrent */ - statedb.SnapshotCommits - statedb.TrieDBCommits)
elapsed := time.Since(startTime) + 1 // prevent zero division
blockInsertTimer.Update(elapsed)
// TODO(rjl493456442) generalize the ResettingTimer
mgasps := float64(res.GasUsed) * 1000 / float64(elapsed)
chainMgaspsMeter.Update(time.Duration(mgasps))
stats.TotalTime = elapsed
stats.MgasPerSecond = float64(res.GasUsed) * 1000 / float64(elapsed)
return &blockProcessingResult{
usedGas: res.GasUsed,
procTime: proctime,
status: status,
witness: witness,
stats: stats,
}, nil
}
@@ -2729,8 +2761,8 @@ func (bc *BlockChain) skipBlock(err error, it *insertIterator) bool {
return false
}
// reportBlock logs a bad block error.
func (bc *BlockChain) reportBlock(block *types.Block, res *ProcessResult, err error) {
// reportBadBlock logs a bad block error.
func (bc *BlockChain) reportBadBlock(block *types.Block, res *ProcessResult, err error) {
var receipts types.Receipts
if res != nil {
receipts = res.Receipts
+138
View File
@@ -0,0 +1,138 @@
// Copyright 2025 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package core
import (
"fmt"
"strings"
"time"
"github.com/luxfi/geth/common"
"github.com/luxfi/geth/core/state"
"github.com/luxfi/geth/core/types"
"github.com/luxfi/geth/log"
)
// ExecuteStats includes all the statistics of a block execution in details.
type ExecuteStats struct {
// State read times
AccountReads time.Duration // Time spent on the account reads
StorageReads time.Duration // Time spent on the storage reads
AccountHashes time.Duration // Time spent on the account trie hash
AccountUpdates time.Duration // Time spent on the account trie update
AccountCommits time.Duration // Time spent on the account trie commit
StorageUpdates time.Duration // Time spent on the storage trie update
StorageCommits time.Duration // Time spent on the storage trie commit
AccountLoaded int // Number of accounts loaded
AccountUpdated int // Number of accounts updated
AccountDeleted int // Number of accounts deleted
StorageLoaded int // Number of storage slots loaded
StorageUpdated int // Number of storage slots updated
StorageDeleted int // Number of storage slots deleted
Execution time.Duration // Time spent on the EVM execution
Validation time.Duration // Time spent on the block validation
CrossValidation time.Duration // Optional, time spent on the block cross validation
SnapshotCommit time.Duration // Time spent on snapshot commit
TrieDBCommit time.Duration // Time spent on database commit
BlockWrite time.Duration // Time spent on block write
TotalTime time.Duration // The total time spent on block execution
MgasPerSecond float64 // The million gas processed per second
// Cache hit rates
StateReadCacheStats state.ReaderStats
StatePrefetchCacheStats state.ReaderStats
}
// reportMetrics uploads execution statistics to the metrics system.
func (s *ExecuteStats) reportMetrics() {
accountReadTimer.Update(s.AccountReads) // Account reads are complete(in processing)
storageReadTimer.Update(s.StorageReads) // Storage reads are complete(in processing)
if s.AccountLoaded != 0 {
accountReadSingleTimer.Update(s.AccountReads / time.Duration(s.AccountLoaded))
}
if s.StorageLoaded != 0 {
storageReadSingleTimer.Update(s.StorageReads / time.Duration(s.StorageLoaded))
}
accountUpdateTimer.Update(s.AccountUpdates) // Account updates are complete(in validation)
storageUpdateTimer.Update(s.StorageUpdates) // Storage updates are complete(in validation)
accountHashTimer.Update(s.AccountHashes) // Account hashes are complete(in validation)
accountCommitTimer.Update(s.AccountCommits) // Account commits are complete, we can mark them
storageCommitTimer.Update(s.StorageCommits) // Storage commits are complete, we can mark them
blockExecutionTimer.Update(s.Execution) // The time spent on EVM processing
blockValidationTimer.Update(s.Validation) // The time spent on block validation
blockCrossValidationTimer.Update(s.CrossValidation) // The time spent on stateless cross validation
snapshotCommitTimer.Update(s.SnapshotCommit) // Snapshot commits are complete, we can mark them
triedbCommitTimer.Update(s.TrieDBCommit) // Trie database commits are complete, we can mark them
blockWriteTimer.Update(s.BlockWrite) // The time spent on block write
blockInsertTimer.Update(s.TotalTime) // The total time spent on block execution
chainMgaspsMeter.Update(time.Duration(s.MgasPerSecond)) // TODO(rjl493456442) generalize the ResettingTimer
// Cache hit rates
accountCacheHitPrefetchMeter.Mark(s.StatePrefetchCacheStats.AccountCacheHit)
accountCacheMissPrefetchMeter.Mark(s.StatePrefetchCacheStats.AccountCacheMiss)
storageCacheHitPrefetchMeter.Mark(s.StatePrefetchCacheStats.StorageCacheHit)
storageCacheMissPrefetchMeter.Mark(s.StatePrefetchCacheStats.StorageCacheMiss)
accountCacheHitMeter.Mark(s.StateReadCacheStats.AccountCacheHit)
accountCacheMissMeter.Mark(s.StateReadCacheStats.AccountCacheMiss)
storageCacheHitMeter.Mark(s.StateReadCacheStats.StorageCacheHit)
storageCacheMissMeter.Mark(s.StateReadCacheStats.StorageCacheMiss)
}
// logSlow prints the detailed execution statistics if the block is regarded as slow.
func (s *ExecuteStats) logSlow(block *types.Block, slowBlockThreshold time.Duration) {
if slowBlockThreshold == 0 {
return
}
if s.TotalTime < slowBlockThreshold {
return
}
msg := fmt.Sprintf(`
########## SLOW BLOCK #########
Block: %v (%#x) txs: %d, mgasps: %.2f, elapsed: %v
EVM execution: %v
Validation: %v
Account read: %v(%d)
Storage read: %v(%d)
Account hash: %v
Storage hash: %v
DB commit: %v
Block write: %v
%s
##############################
`, block.Number(), block.Hash(), len(block.Transactions()), s.MgasPerSecond, common.PrettyDuration(s.TotalTime),
common.PrettyDuration(s.Execution), common.PrettyDuration(s.Validation+s.CrossValidation),
common.PrettyDuration(s.AccountReads), s.AccountLoaded,
common.PrettyDuration(s.StorageReads), s.StorageLoaded,
common.PrettyDuration(s.AccountHashes+s.AccountCommits+s.AccountUpdates),
common.PrettyDuration(s.StorageCommits+s.StorageUpdates),
common.PrettyDuration(s.TrieDBCommit+s.SnapshotCommit), common.PrettyDuration(s.BlockWrite),
s.StateReadCacheStats)
for _, line := range strings.Split(msg, "\n") {
if line == "" {
continue
}
log.Info(line)
}
}
+2 -2
View File
@@ -162,12 +162,12 @@ func testBlockChainImport(chain types.Blocks, blockchain *BlockChain) error {
}
res, err := blockchain.processor.Process(block, statedb, vm.Config{})
if err != nil {
blockchain.reportBlock(block, res, err)
blockchain.reportBadBlock(block, res, err)
return err
}
err = blockchain.validator.ValidateState(block, statedb, res, false)
if err != nil {
blockchain.reportBlock(block, res, err)
blockchain.reportBadBlock(block, res, err)
return err
}
+1 -1
View File
@@ -434,7 +434,7 @@ func (f *FilterMaps) safeDeleteWithLogs(deleteFn func(db ethdb.KeyValueStore, ha
lastLogPrinted = start
)
switch err := deleteFn(f.db, f.hashScheme, func(deleted bool) bool {
if deleted && !logPrinted || time.Since(lastLogPrinted) > time.Second*10 {
if deleted && (!logPrinted || time.Since(lastLogPrinted) > time.Second*10) {
log.Info(action+" in progress...", "elapsed", common.PrettyDuration(time.Since(start)))
logPrinted, lastLogPrinted = true, time.Now()
}
+14
View File
@@ -91,6 +91,13 @@ func (t *memoryTable) truncateHead(items uint64) error {
if items < t.offset {
return errors.New("truncation below tail")
}
for i := int(items - t.offset); i < len(t.data); i++ {
if t.size > uint64(len(t.data[i])) {
t.size -= uint64(len(t.data[i]))
} else {
t.size = 0
}
}
t.data = t.data[:items-t.offset]
t.items = items
return nil
@@ -108,6 +115,13 @@ func (t *memoryTable) truncateTail(items uint64) error {
if t.items < items {
return errors.New("truncation above head")
}
for i := uint64(0); i < items-t.offset; i++ {
if t.size > uint64(len(t.data[i])) {
t.size -= uint64(len(t.data[i]))
} else {
t.size = 0
}
}
t.data = t.data[items-t.offset:]
t.offset = items
return nil
+48 -16
View File
@@ -18,6 +18,7 @@ package state
import (
"errors"
"fmt"
"sync"
"sync/atomic"
@@ -39,6 +40,10 @@ import (
// ContractCodeReader defines the interface for accessing contract code.
type ContractCodeReader interface {
// Has returns the flag indicating whether the contract code with
// specified address and hash exists or not.
Has(addr common.Address, codeHash common.Hash) bool
// Code retrieves a particular contract's code.
//
// - Returns nil code along with nil error if the requested contract code
@@ -88,10 +93,29 @@ type Reader interface {
// ReaderStats wraps the statistics of reader.
type ReaderStats struct {
AccountHit int64
AccountMiss int64
StorageHit int64
StorageMiss int64
// Cache stats
AccountCacheHit int64
AccountCacheMiss int64
StorageCacheHit int64
StorageCacheMiss int64
}
// String implements fmt.Stringer, returning string format statistics.
func (s ReaderStats) String() string {
var (
accountCacheHitRate float64
storageCacheHitRate float64
)
if s.AccountCacheHit > 0 {
accountCacheHitRate = float64(s.AccountCacheHit) / float64(s.AccountCacheHit+s.AccountCacheMiss) * 100
}
if s.StorageCacheHit > 0 {
storageCacheHitRate = float64(s.StorageCacheHit) / float64(s.StorageCacheHit+s.StorageCacheMiss) * 100
}
msg := fmt.Sprintf("Reader statistics\n")
msg += fmt.Sprintf("account: hit: %d, miss: %d, rate: %.2f\n", s.AccountCacheHit, s.AccountCacheMiss, accountCacheHitRate)
msg += fmt.Sprintf("storage: hit: %d, miss: %d, rate: %.2f\n", s.StorageCacheHit, s.StorageCacheMiss, storageCacheHitRate)
return msg
}
// ReaderWithStats wraps the additional method to retrieve the reader statistics from.
@@ -150,6 +174,13 @@ func (r *cachingCodeReader) CodeSize(addr common.Address, codeHash common.Hash)
return len(code), nil
}
// Has returns the flag indicating whether the contract code with
// specified address and hash exists or not.
func (r *cachingCodeReader) Has(addr common.Address, codeHash common.Hash) bool {
code, _ := r.Code(addr, codeHash)
return len(code) > 0
}
// flatReader wraps a database state reader and is safe for concurrent access.
type flatReader struct {
reader database.StateReader
@@ -544,10 +575,11 @@ func (r *readerWithCache) Storage(addr common.Address, slot common.Hash) (common
type readerWithCacheStats struct {
*readerWithCache
accountHit atomic.Int64
accountMiss atomic.Int64
storageHit atomic.Int64
storageMiss atomic.Int64
accountCacheHit atomic.Int64
accountCacheMiss atomic.Int64
storageCacheHit atomic.Int64
storageCacheMiss atomic.Int64
}
// newReaderWithCacheStats constructs the reader with additional statistics tracked.
@@ -567,9 +599,9 @@ func (r *readerWithCacheStats) Account(addr common.Address) (*types.StateAccount
return nil, err
}
if incache {
r.accountHit.Add(1)
r.accountCacheHit.Add(1)
} else {
r.accountMiss.Add(1)
r.accountCacheMiss.Add(1)
}
return account, nil
}
@@ -585,9 +617,9 @@ func (r *readerWithCacheStats) Storage(addr common.Address, slot common.Hash) (c
return common.Hash{}, err
}
if incache {
r.storageHit.Add(1)
r.storageCacheHit.Add(1)
} else {
r.storageMiss.Add(1)
r.storageCacheMiss.Add(1)
}
return value, nil
}
@@ -595,9 +627,9 @@ func (r *readerWithCacheStats) Storage(addr common.Address, slot common.Hash) (c
// GetStats implements ReaderWithStats, returning the statistics of state reader.
func (r *readerWithCacheStats) GetStats() ReaderStats {
return ReaderStats{
AccountHit: r.accountHit.Load(),
AccountMiss: r.accountMiss.Load(),
StorageHit: r.storageHit.Load(),
StorageMiss: r.storageMiss.Load(),
AccountCacheHit: r.accountCacheHit.Load(),
AccountCacheMiss: r.accountCacheMiss.Load(),
StorageCacheHit: r.storageCacheHit.Load(),
StorageCacheMiss: r.storageCacheMiss.Load(),
}
}
+39 -3
View File
@@ -31,6 +31,7 @@ import (
"github.com/luxfi/geth/crypto"
"github.com/luxfi/geth/ethdb"
"github.com/luxfi/geth/log"
"github.com/luxfi/geth/metrics"
"github.com/luxfi/geth/triedb"
"golang.org/x/sync/errgroup"
)
@@ -48,6 +49,21 @@ var (
codeKeySize = int64(len(rawdb.CodePrefix) + common.HashLength)
)
// State size metrics
var (
stateSizeChainHeightGauge = metrics.NewRegisteredGauge("state/height", nil)
stateSizeAccountsCountGauge = metrics.NewRegisteredGauge("state/accounts/count", nil)
stateSizeAccountsBytesGauge = metrics.NewRegisteredGauge("state/accounts/bytes", nil)
stateSizeStoragesCountGauge = metrics.NewRegisteredGauge("state/storages/count", nil)
stateSizeStoragesBytesGauge = metrics.NewRegisteredGauge("state/storages/bytes", nil)
stateSizeAccountTrieNodesCountGauge = metrics.NewRegisteredGauge("state/trienodes/account/count", nil)
stateSizeAccountTrieNodesBytesGauge = metrics.NewRegisteredGauge("state/trienodes/account/bytes", nil)
stateSizeStorageTrieNodesCountGauge = metrics.NewRegisteredGauge("state/trienodes/storage/count", nil)
stateSizeStorageTrieNodesBytesGauge = metrics.NewRegisteredGauge("state/trienodes/storage/bytes", nil)
stateSizeContractsCountGauge = metrics.NewRegisteredGauge("state/contracts/count", nil)
stateSizeContractsBytesGauge = metrics.NewRegisteredGauge("state/contracts/bytes", nil)
)
// SizeStats represents either the current state size statistics or the size
// differences resulting from a state transition.
type SizeStats struct {
@@ -76,6 +92,20 @@ func (s SizeStats) String() string {
)
}
func (s SizeStats) publish() {
stateSizeChainHeightGauge.Update(int64(s.BlockNumber))
stateSizeAccountsCountGauge.Update(s.Accounts)
stateSizeAccountsBytesGauge.Update(s.AccountBytes)
stateSizeStoragesCountGauge.Update(s.Storages)
stateSizeStoragesBytesGauge.Update(s.StorageBytes)
stateSizeAccountTrieNodesCountGauge.Update(s.AccountTrienodes)
stateSizeAccountTrieNodesBytesGauge.Update(s.AccountTrienodeBytes)
stateSizeStorageTrieNodesCountGauge.Update(s.StorageTrienodes)
stateSizeStorageTrieNodesBytesGauge.Update(s.StorageTrienodeBytes)
stateSizeContractsCountGauge.Update(s.ContractCodes)
stateSizeContractsBytesGauge.Update(s.ContractCodeBytes)
}
// add applies the given state diffs and produces a new version of the statistics.
func (s SizeStats) add(diff SizeStats) SizeStats {
s.StateRoot = diff.StateRoot
@@ -213,12 +243,14 @@ func calSizeStats(update *stateUpdate) (SizeStats, error) {
}
}
// Measure code changes. Note that the reported contract code size may be slightly
// inaccurate due to database deduplication (code is stored by its hash). However,
// this deviation is negligible and acceptable for measurement purposes.
codeExists := make(map[common.Hash]struct{})
for _, code := range update.codes {
if _, ok := codeExists[code.hash]; ok || code.exists {
continue
}
stats.ContractCodes += 1
stats.ContractCodeBytes += codeKeySize + int64(len(code.blob))
codeExists[code.hash] = struct{}{}
}
return stats, nil
}
@@ -309,6 +341,10 @@ func (t *SizeTracker) run() {
stats[u.root] = stat
last = u.root
// Publish statistics to metric system
stat.publish()
// Evict the stale statistics
heap.Push(&h, stats[u.root])
for u.blockNumber-h[0].BlockNumber > statEvictThreshold {
delete(stats, h[0].StateRoot)
+8 -7
View File
@@ -58,7 +58,7 @@ func TestSizeTracker(t *testing.T) {
state.AddBalance(addr3, uint256.NewInt(3000), tracing.BalanceChangeUnspecified)
state.SetNonce(addr3, 3, tracing.NonceChangeUnspecified)
currentRoot, _, err := state.CommitWithUpdate(1, true, false)
currentRoot, err := state.Commit(1, true, false)
if err != nil {
t.Fatalf("Failed to commit initial state: %v", err)
}
@@ -83,7 +83,7 @@ func TestSizeTracker(t *testing.T) {
if i%3 == 0 {
newState.SetCode(testAddr, []byte{byte(i), 0x60, 0x80, byte(i + 1), 0x52}, tracing.CodeChangeUnspecified)
}
root, _, err := newState.CommitWithUpdate(blockNum, true, false)
root, err := newState.Commit(blockNum, true, false)
if err != nil {
t.Fatalf("Failed to commit state at block %d: %v", blockNum, err)
}
@@ -154,21 +154,22 @@ func TestSizeTracker(t *testing.T) {
if i%3 == 0 {
newState.SetCode(testAddr, []byte{byte(i), 0x60, 0x80, byte(i + 1), 0x52}, tracing.CodeChangeUnspecified)
}
root, update, err := newState.CommitWithUpdate(blockNum, true, false)
ret, err := newState.commitAndFlush(blockNum, true, false, true)
if err != nil {
t.Fatalf("Failed to commit state at block %d: %v", blockNum, err)
}
if err := tdb.Commit(root, false); err != nil {
tracker.Notify(ret)
if err := tdb.Commit(ret.root, false); err != nil {
t.Fatalf("Failed to commit trie at block %d: %v", blockNum, err)
}
diff, err := calSizeStats(update)
diff, err := calSizeStats(ret)
if err != nil {
t.Fatalf("Failed to calculate size stats for block %d: %v", blockNum, err)
}
trackedUpdates = append(trackedUpdates, diff)
tracker.Notify(update)
currentRoot = root
currentRoot = ret.root
}
finalRoot := rawdb.ReadSnapshotRoot(db)
+18 -12
View File
@@ -141,10 +141,11 @@ type StateDB struct {
witnessStats *stateless.WitnessStats
// Measurements gathered during execution for debugging purposes
AccountReads time.Duration
AccountHashes time.Duration
AccountUpdates time.Duration
AccountCommits time.Duration
AccountReads time.Duration
AccountHashes time.Duration
AccountUpdates time.Duration
AccountCommits time.Duration
StorageReads time.Duration
StorageUpdates time.Duration
StorageCommits time.Duration
@@ -1316,11 +1317,16 @@ func (s *StateDB) commit(deleteEmptyObjects bool, noStorageWiping bool, blockNum
// commitAndFlush is a wrapper of commit which also commits the state mutations
// to the configured data stores.
func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (*stateUpdate, error) {
func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorageWiping bool, dedupCode bool) (*stateUpdate, error) {
ret, err := s.commit(deleteEmptyObjects, noStorageWiping, block)
if err != nil {
return nil, err
}
if dedupCode {
ret.markCodeExistence(s.reader)
}
// Commit dirty contract code if any exists
if db := s.db.TrieDB().Disk(); db != nil && len(ret.codes) > 0 {
batch := db.NewBatch()
@@ -1375,21 +1381,21 @@ func (s *StateDB) commitAndFlush(block uint64, deleteEmptyObjects bool, noStorag
// no empty accounts left that could be deleted by EIP-158, storage wiping
// should not occur.
func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (common.Hash, error) {
ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping)
ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping, false)
if err != nil {
return common.Hash{}, err
}
return ret.root, nil
}
// CommitWithUpdate writes the state mutations and returns both the root hash and the state update.
// This is useful for tracking state changes at the blockchain level.
func (s *StateDB) CommitWithUpdate(block uint64, deleteEmptyObjects bool, noStorageWiping bool) (common.Hash, *stateUpdate, error) {
ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping)
// CommitAndTrack writes the state mutations and notifies the size tracker of the state changes.
func (s *StateDB) CommitAndTrack(block uint64, deleteEmptyObjects bool, noStorageWiping bool, sizer *SizeTracker) (common.Hash, error) {
ret, err := s.commitAndFlush(block, deleteEmptyObjects, noStorageWiping, true)
if err != nil {
return common.Hash{}, nil, err
return common.Hash{}, err
}
return ret.root, ret, nil
sizer.Notify(ret)
return ret.root, nil
}
// Prepare handles the preparatory steps for executing a state transition with.
+1 -1
View File
@@ -228,7 +228,7 @@ func (test *stateTest) run() bool {
} else {
state.IntermediateRoot(true) // call intermediateRoot at the transaction boundary
}
ret, err := state.commitAndFlush(0, true, false) // call commit at the block boundary
ret, err := state.commitAndFlush(0, true, false, false) // call commit at the block boundary
if err != nil {
panic(err)
}
+22 -2
View File
@@ -26,8 +26,9 @@ import (
// contractCode represents a contract code with associated metadata.
type contractCode struct {
hash common.Hash // hash is the cryptographic hash of the contract code.
blob []byte // blob is the binary representation of the contract code.
hash common.Hash // hash is the cryptographic hash of the contract code.
blob []byte // blob is the binary representation of the contract code.
exists bool // flag whether the code has been existent
}
// accountDelete represents an operation for deleting an Ethereum account.
@@ -190,3 +191,22 @@ func (sc *stateUpdate) stateSet() *triedb.StateSet {
RawStorageKey: sc.rawStorageKey,
}
}
// markCodeExistence determines whether each piece of contract code referenced
// in this state update actually exists.
//
// Note: This operation is expensive and not needed during normal state transitions.
// It is only required when SizeTracker is enabled to produce accurate state
// statistics.
func (sc *stateUpdate) markCodeExistence(reader ContractCodeReader) {
cache := make(map[common.Hash]bool)
for addr, code := range sc.codes {
if exists, ok := cache[code.hash]; ok {
code.exists = exists
continue
}
res := reader.Has(addr, code.hash)
cache[code.hash] = res
code.exists = res
}
}
+9 -2
View File
@@ -62,10 +62,17 @@ func (s *WitnessStats) Add(nodes map[string][]byte, owner common.Hash) {
// If current path is a prefix of the next path, it's not a leaf.
// The last path is always a leaf.
if i == len(paths)-1 || !strings.HasPrefix(paths[i+1], paths[i]) {
depth := len(path)
if owner == (common.Hash{}) {
s.accountTrieLeaves[len(path)] += 1
if depth >= len(s.accountTrieLeaves) {
depth = len(s.accountTrieLeaves) - 1
}
s.accountTrieLeaves[depth] += 1
} else {
s.storageTrieLeaves[len(path)] += 1
if depth >= len(s.storageTrieLeaves) {
depth = len(s.storageTrieLeaves) - 1
}
s.storageTrieLeaves[depth] += 1
}
}
}
+6 -1
View File
@@ -377,7 +377,12 @@ func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bo
// Filter returns whether the given transaction can be consumed by the blob pool.
func (p *BlobPool) Filter(tx *types.Transaction) bool {
return tx.Type() == types.BlobTxType
return p.FilterType(tx.Type())
}
// FilterType returns whether the blob pool supports the given transaction type.
func (p *BlobPool) FilterType(kind byte) bool {
return kind == types.BlobTxType
}
// Init sets the gas price needed to keep a transaction in the pool and the chain
+6 -1
View File
@@ -288,7 +288,12 @@ func New(config Config, chain BlockChain) *LegacyPool {
// Filter returns whether the given transaction can be consumed by the legacy
// pool, specifically, whether it is a Legacy, AccessList or Dynamic transaction.
func (pool *LegacyPool) Filter(tx *types.Transaction) bool {
switch tx.Type() {
return pool.FilterType(tx.Type())
}
// FilterType returns whether the legacy pool supports the given transaction type.
func (pool *LegacyPool) FilterType(kind byte) bool {
switch kind {
case types.LegacyTxType, types.AccessListTxType, types.DynamicFeeTxType, types.SetCodeTxType:
return true
default:
+3
View File
@@ -100,6 +100,9 @@ type SubPool interface {
// to this particular subpool.
Filter(tx *types.Transaction) bool
// FilterType returns whether the subpool supports the given transaction type.
FilterType(kind byte) bool
// Init sets the base parameters of the subpool, allowing it to load any saved
// transactions from disk and also permitting internal maintenance routines to
// start up.
+11
View File
@@ -489,3 +489,14 @@ func (p *TxPool) Clear() {
subpool.Clear()
}
}
// FilterType returns whether a transaction with the given type is supported
// (can be added) by the pool.
func (p *TxPool) FilterType(kind byte) bool {
for _, subpool := range p.subpools {
if subpool.FilterType(kind) {
return true
}
}
return false
}
+3 -3
View File
@@ -964,7 +964,7 @@ func opDupN(pc *uint64, evm *EVM, scope *ScopeContext) ([]byte, error) {
//The nth stack item is duplicated at the top of the stack.
scope.Stack.push(scope.Stack.Back(n - 1))
*pc += 2
*pc += 1
return nil, nil
}
@@ -993,7 +993,7 @@ func opSwapN(pc *uint64, evm *EVM, scope *ScopeContext) ([]byte, error) {
indexTop := scope.Stack.len() - 1
indexN := scope.Stack.len() - 1 - n
scope.Stack.data[indexTop], scope.Stack.data[indexN] = scope.Stack.data[indexN], scope.Stack.data[indexTop]
*pc += 2
*pc += 1
return nil, nil
}
@@ -1025,7 +1025,7 @@ func opExchange(pc *uint64, evm *EVM, scope *ScopeContext) ([]byte, error) {
indexN := scope.Stack.len() - 1 - n
indexM := scope.Stack.len() - 1 - m
scope.Stack.data[indexN], scope.Stack.data[indexM] = scope.Stack.data[indexM], scope.Stack.data[indexN]
*pc += 2
*pc += 1
return nil, nil
}
+8 -4
View File
@@ -1107,6 +1107,11 @@ func TestEIP8024_Execution(t *testing.T) {
codeHex: "e8", // no operand
wantErr: true,
},
{
name: "PC_INCREMENT",
codeHex: "600060006000e80115",
wantVals: []uint64{1, 0, 0},
},
}
for _, tc := range tests {
@@ -1123,17 +1128,15 @@ func TestEIP8024_Execution(t *testing.T) {
return
case 0x60:
_, err = opPush1(&pc, evm, scope)
pc++
case 0x80:
dup1 := makeDup(1)
_, err = dup1(&pc, evm, scope)
pc++
case 0x56:
_, err = opJump(&pc, evm, scope)
pc++
case 0x5b:
_, err = opJumpdest(&pc, evm, scope)
pc++
case 0x15:
_, err = opIszero(&pc, evm, scope)
case 0xe6:
_, err = opDupN(&pc, evm, scope)
case 0xe7:
@@ -1143,6 +1146,7 @@ func TestEIP8024_Execution(t *testing.T) {
default:
err = &ErrInvalidOpCode{opcode: OpCode(op)}
}
pc++
}
if tc.wantErr {
if err == nil {
-6
View File
@@ -499,17 +499,14 @@ func (api *DebugAPI) ExecutionWitness(bn rpc.BlockNumber) (*stateless.ExtWitness
if err != nil {
return &stateless.ExtWitness{}, fmt.Errorf("block number %v not found", bn)
}
parent := bc.GetHeader(block.ParentHash(), block.NumberU64()-1)
if parent == nil {
return &stateless.ExtWitness{}, fmt.Errorf("block number %v found, but parent missing", bn)
}
result, err := bc.ProcessBlock(parent.Root, block, false, true)
if err != nil {
return nil, err
}
return result.Witness().ToExtWitness(), nil
}
@@ -519,16 +516,13 @@ func (api *DebugAPI) ExecutionWitnessByHash(hash common.Hash) (*stateless.ExtWit
if block == nil {
return &stateless.ExtWitness{}, fmt.Errorf("block hash %x not found", hash)
}
parent := bc.GetHeader(block.ParentHash(), block.NumberU64()-1)
if parent == nil {
return &stateless.ExtWitness{}, fmt.Errorf("block number %x found, but parent missing", hash)
}
result, err := bc.ProcessBlock(parent.Root, block, false, true)
if err != nil {
return nil, err
}
return result.Witness().ToExtWitness(), nil
}
+1 -26
View File
@@ -244,6 +244,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
// - DATADIR/triedb/verkle.journal
TrieJournalDirectory: stack.ResolvePath("triedb"),
StateSizeTracking: config.EnableStateSizeTracking,
SlowBlockThreshold: config.SlowBlockThreshold,
}
)
if config.VMTrace != "" {
@@ -594,29 +595,3 @@ func (s *Ethereum) Stop() error {
return nil
}
// SyncMode retrieves the current sync mode, either explicitly set, or derived
// from the chain status.
func (s *Ethereum) SyncMode() ethconfig.SyncMode {
// If we're in snap sync mode, return that directly
if s.handler.snapSync.Load() {
return ethconfig.SnapSync
}
// We are probably in full sync, but we might have rewound to before the
// snap sync pivot, check if we should re-enable snap sync.
head := s.blockchain.CurrentBlock()
if pivot := rawdb.ReadLastPivotNumber(s.chainDb); pivot != nil {
if head.Number.Uint64() < *pivot {
return ethconfig.SnapSync
}
}
// We are in a full sync, but the associated head state is missing. To complete
// the head state, forcefully rerun the snap sync. Note it doesn't mean the
// persistent state is corrupted, just mismatch with the head block.
if !s.blockchain.HasState(head.Root) {
log.Info("Reenabled snap sync as chain is stateless")
return ethconfig.SnapSync
}
// Nope, we're really full syncing
return ethconfig.FullSync
}
+52 -32
View File
@@ -270,7 +270,7 @@ func (api *ConsensusAPI) forkchoiceUpdated(update engine.ForkchoiceStateV1, payl
}
}
log.Info("Forkchoice requested sync to new head", context...)
if err := api.eth.Downloader().BeaconSync(api.eth.SyncMode(), header, finalized); err != nil {
if err := api.eth.Downloader().BeaconSync(header, finalized); err != nil {
return engine.STATUS_SYNCING, err
}
return engine.STATUS_SYNCING, nil
@@ -402,10 +402,12 @@ func (api *ConsensusAPI) ExchangeTransitionConfigurationV1(config engine.Transit
// GetPayloadV1 returns a cached payload by id.
func (api *ConsensusAPI) GetPayloadV1(payloadID engine.PayloadID) (*engine.ExecutableData, error) {
if !payloadID.Is(engine.PayloadV1) {
return nil, engine.UnsupportedFork
}
data, err := api.getPayload(payloadID, false)
data, err := api.getPayload(
payloadID,
false,
[]engine.PayloadVersion{engine.PayloadV1},
nil,
)
if err != nil {
return nil, err
}
@@ -414,35 +416,34 @@ func (api *ConsensusAPI) GetPayloadV1(payloadID engine.PayloadID) (*engine.Execu
// GetPayloadV2 returns a cached payload by id.
func (api *ConsensusAPI) GetPayloadV2(payloadID engine.PayloadID) (*engine.ExecutionPayloadEnvelope, error) {
// executionPayload: ExecutionPayloadV1 | ExecutionPayloadV2 where:
//
// - ExecutionPayloadV1 MUST be returned if the payload timestamp is lower
// than the Shanghai timestamp
//
// - ExecutionPayloadV2 MUST be returned if the payload timestamp is greater
// or equal to the Shanghai timestamp
if !payloadID.Is(engine.PayloadV1, engine.PayloadV2) {
return nil, engine.UnsupportedFork
}
return api.getPayload(payloadID, false)
return api.getPayload(
payloadID,
false,
[]engine.PayloadVersion{engine.PayloadV1, engine.PayloadV2},
[]forks.Fork{forks.Shanghai},
)
}
// GetPayloadV3 returns a cached payload by id. This endpoint should only
// be used for the Cancun fork.
func (api *ConsensusAPI) GetPayloadV3(payloadID engine.PayloadID) (*engine.ExecutionPayloadEnvelope, error) {
if !payloadID.Is(engine.PayloadV3) {
return nil, engine.UnsupportedFork
}
return api.getPayload(payloadID, false)
return api.getPayload(
payloadID,
false,
[]engine.PayloadVersion{engine.PayloadV3},
[]forks.Fork{forks.Cancun},
)
}
// GetPayloadV4 returns a cached payload by id. This endpoint should only
// be used for the Prague fork.
func (api *ConsensusAPI) GetPayloadV4(payloadID engine.PayloadID) (*engine.ExecutionPayloadEnvelope, error) {
if !payloadID.Is(engine.PayloadV3) {
return nil, engine.UnsupportedFork
}
return api.getPayload(payloadID, false)
return api.getPayload(
payloadID,
false,
[]engine.PayloadVersion{engine.PayloadV3},
[]forks.Fork{forks.Prague},
)
}
// GetPayloadV5 returns a cached payload by id. This endpoint should only
@@ -451,18 +452,37 @@ func (api *ConsensusAPI) GetPayloadV4(payloadID engine.PayloadID) (*engine.Execu
// This method follows the same specification as engine_getPayloadV4 with
// changes of returning BlobsBundleV2 with BlobSidecar version 1.
func (api *ConsensusAPI) GetPayloadV5(payloadID engine.PayloadID) (*engine.ExecutionPayloadEnvelope, error) {
if !payloadID.Is(engine.PayloadV3) {
return nil, engine.UnsupportedFork
}
return api.getPayload(payloadID, false)
return api.getPayload(
payloadID,
false,
[]engine.PayloadVersion{engine.PayloadV3},
[]forks.Fork{
forks.Osaka,
forks.BPO1,
forks.BPO2,
forks.BPO3,
forks.BPO4,
forks.BPO5,
})
}
func (api *ConsensusAPI) getPayload(payloadID engine.PayloadID, full bool) (*engine.ExecutionPayloadEnvelope, error) {
// getPayload will retrieve the specified payload and verify it conforms to the
// endpoint's allowed payload versions and forks.
//
// Note passing nil `forks`, `versions` disables the respective check.
func (api *ConsensusAPI) getPayload(payloadID engine.PayloadID, full bool, versions []engine.PayloadVersion, forks []forks.Fork) (*engine.ExecutionPayloadEnvelope, error) {
log.Trace("Engine API request received", "method", "GetPayload", "id", payloadID)
if versions != nil && !payloadID.Is(versions...) {
return nil, engine.UnsupportedFork
}
data := api.localBlocks.get(payloadID, full)
if data == nil {
return nil, engine.UnknownPayload
}
if forks != nil && !api.checkFork(data.ExecutionPayload.Timestamp, forks...) {
return nil, engine.UnsupportedFork
}
return data, nil
}
@@ -744,7 +764,7 @@ func (api *ConsensusAPI) newPayload(params engine.ExecutableData, versionedHashe
// tries to make it import a block. That should be denied as pushing something
// into the database directly will conflict with the assumptions of snap sync
// that it has an empty db that it can fill itself.
if api.eth.SyncMode() != ethconfig.FullSync {
if api.eth.Downloader().ConfigSyncMode() == ethconfig.SnapSync {
return api.delayPayloadImport(block), nil
}
if !api.eth.BlockChain().HasBlockAndState(block.ParentHash(), block.NumberU64()-1) {
@@ -792,7 +812,7 @@ func (api *ConsensusAPI) delayPayloadImport(block *types.Block) engine.PayloadSt
// Although we don't want to trigger a sync, if there is one already in
// progress, try to extend it with the current payload request to relieve
// some strain from the forkchoice update.
err := api.eth.Downloader().BeaconExtend(api.eth.SyncMode(), block.Header())
err := api.eth.Downloader().BeaconExtend(block.Header())
if err == nil {
log.Debug("Payload accepted for sync extension", "number", block.NumberU64(), "hash", block.Hash())
return engine.PayloadStatusV1{Status: engine.SYNCING}
@@ -801,7 +821,7 @@ func (api *ConsensusAPI) delayPayloadImport(block *types.Block) engine.PayloadSt
// payload as non-integratable on top of the existing sync. We'll just
// have to rely on the beacon client to forcefully update the head with
// a forkchoice update request.
if api.eth.SyncMode() == ethconfig.FullSync {
if api.eth.Downloader().ConfigSyncMode() == ethconfig.FullSync {
// In full sync mode, failure to import a well-formed block can only mean
// that the parent state is missing and the syncer rejected extending the
// current cycle with the new payload.
+56 -35
View File
@@ -203,7 +203,7 @@ func TestEth2PrepareAndGetPayload(t *testing.T) {
BeaconRoot: blockParams.BeaconRoot,
Version: engine.PayloadV1,
}).Id()
execData, err := api.getPayload(payloadID, true)
execData, err := api.getPayload(payloadID, true, nil, nil)
if err != nil {
t.Fatalf("error getting payload, err=%v", err)
}
@@ -426,7 +426,7 @@ func TestEth2DeepReorg(t *testing.T) {
}
// startEthService creates a full node instance for testing.
func startEthService(t *testing.T, genesis *core.Genesis, blocks []*types.Block) (*node.Node, *eth.Ethereum) {
func startEthService(t testing.TB, genesis *core.Genesis, blocks []*types.Block) (*node.Node, *eth.Ethereum) {
t.Helper()
n, err := node.New(&node.Config{
@@ -636,7 +636,7 @@ func TestNewPayloadOnInvalidChain(t *testing.T) {
t.Fatalf("error preparing payload, invalid status: %v", resp.PayloadStatus.Status)
}
// give the payload some time to be built
if payload, err = api.getPayload(*resp.PayloadID, true); err != nil {
if payload, err = api.getPayload(*resp.PayloadID, true, nil, nil); err != nil {
t.Fatalf("can't get payload: %v", err)
}
if len(payload.ExecutionPayload.Transactions) > 0 {
@@ -1219,7 +1219,7 @@ func TestNilWithdrawals(t *testing.T) {
Random: test.blockParams.Random,
Version: payloadVersion,
}).Id()
execData, err := api.GetPayloadV2(payloadID)
execData, err := api.getPayload(payloadID, false, nil, nil)
if err != nil {
t.Fatalf("error getting payload, err=%v", err)
}
@@ -1674,7 +1674,7 @@ func TestWitnessCreationAndConsumption(t *testing.T) {
BeaconRoot: blockParams.BeaconRoot,
Version: engine.PayloadV3,
}).Id()
envelope, err := api.getPayload(payloadID, true)
envelope, err := api.getPayload(payloadID, true, nil, nil)
if err != nil {
t.Fatalf("error getting payload, err=%v", err)
}
@@ -1873,7 +1873,7 @@ func makeMultiBlobTx(chainConfig *params.ChainConfig, nonce uint64, blobCount in
return types.MustSignNewTx(key, types.LatestSigner(chainConfig), blobtx)
}
func newGetBlobEnv(t *testing.T, version byte) (*node.Node, *ConsensusAPI) {
func newGetBlobEnv(t testing.TB, version byte) (*node.Node, *ConsensusAPI) {
var (
// Create a database pre-initialize with a genesis block
config = *params.MergedTestChainConfig
@@ -2045,36 +2045,57 @@ func TestGetBlobsV2(t *testing.T) {
},
}
for i, suite := range suites {
// Fill the request for retrieving blobs
var (
vhashes []common.Hash
expect []*engine.BlobAndProofV2
)
// fill missing blob
if suite.fillRandom {
vhashes = append(vhashes, testrand.Hash())
}
for j := suite.start; j < suite.limit; j++ {
vhashes = append(vhashes, testBlobVHashes[j])
var cellProofs []hexutil.Bytes
for _, proof := range testBlobCellProofs[j] {
cellProofs = append(cellProofs, proof[:])
runGetBlobsV2(t, api, suite.start, suite.limit, suite.fillRandom, fmt.Sprintf("suite=%d", i))
}
}
// Benchmark GetBlobsV2 internals
// Note that this is not an RPC-level benchmark, so JSON-RPC overhead is not included.
func BenchmarkGetBlobsV2(b *testing.B) {
n, api := newGetBlobEnv(b, 1)
defer n.Close()
// for blobs in [1, 2, 4, 6], print string and run benchmark
for _, blobs := range []int{1, 2, 4, 6} {
name := fmt.Sprintf("blobs=%d", blobs)
b.Run(name, func(b *testing.B) {
for b.Loop() {
runGetBlobsV2(b, api, 0, blobs, false, name)
}
expect = append(expect, &engine.BlobAndProofV2{
Blob: testBlobs[j][:],
CellProofs: cellProofs,
})
}
result, err := api.GetBlobsV2(vhashes)
if err != nil {
t.Errorf("Unexpected error for case %d, %v", i, err)
}
// null is responded if any blob is missing
if suite.fillRandom {
expect = nil
}
if !reflect.DeepEqual(result, expect) {
t.Fatalf("Unexpected result for case %d", i)
})
}
}
func runGetBlobsV2(t testing.TB, api *ConsensusAPI, start, limit int, fillRandom bool, name string) {
// Fill the request for retrieving blobs
var (
vhashes []common.Hash
expect []*engine.BlobAndProofV2
)
// fill missing blob
if fillRandom {
vhashes = append(vhashes, testrand.Hash())
}
for j := start; j < limit; j++ {
vhashes = append(vhashes, testBlobVHashes[j])
var cellProofs []hexutil.Bytes
for _, proof := range testBlobCellProofs[j] {
cellProofs = append(cellProofs, proof[:])
}
expect = append(expect, &engine.BlobAndProofV2{
Blob: testBlobs[j][:],
CellProofs: cellProofs,
})
}
result, err := api.GetBlobsV2(vhashes)
if err != nil {
t.Errorf("Unexpected error for case %s, %v", name, err)
}
// null is responded if any blob is missing
if fillRandom {
expect = nil
}
if !reflect.DeepEqual(result, expect) {
t.Fatalf("Unexpected result for case %s", name)
}
}
+1 -1
View File
@@ -214,7 +214,7 @@ func (c *SimulatedBeacon) sealBlock(withdrawals []*types.Withdrawal, timestamp u
return nil
}
envelope, err := c.engineAPI.getPayload(*fcResponse.PayloadID, true)
envelope, err := c.engineAPI.getPayload(*fcResponse.PayloadID, true, nil, nil)
if err != nil {
return err
}
+1 -2
View File
@@ -22,7 +22,6 @@ import (
"github.com/luxfi/geth/common"
"github.com/luxfi/geth/eth"
"github.com/luxfi/geth/eth/ethconfig"
"github.com/luxfi/geth/log"
"github.com/luxfi/geth/node"
)
@@ -71,7 +70,7 @@ func (tester *FullSyncTester) Start() error {
// Trigger beacon sync with the provided block hash as trusted
// chain head.
err = tester.backend.Downloader().BeaconDevSync(ethconfig.FullSync, header)
err = tester.backend.Downloader().BeaconDevSync(header)
if err != nil {
log.Info("Failed to trigger beacon sync", "err", err)
}
+2 -2
View File
@@ -33,14 +33,14 @@ import (
// Note, this must not be used in live code. If the forkchcoice endpoint where
// to use this instead of giving us the payload first, then essentially nobody
// in the network would have the block yet that we'd attempt to retrieve.
func (d *Downloader) BeaconDevSync(mode SyncMode, header *types.Header) error {
func (d *Downloader) BeaconDevSync(header *types.Header) error {
// Be very loud that this code should not be used in a live node
log.Warn("----------------------------------")
log.Warn("Beacon syncing with hash as target", "number", header.Number, "hash", header.Hash())
log.Warn("This is unhealthy for a live node!")
log.Warn("This is incompatible with the consensus layer!")
log.Warn("----------------------------------")
return d.BeaconSync(mode, header, header)
return d.BeaconSync(header, header)
}
// GetHeader tries to retrieve the header with a given hash from a random peer.
+7 -41
View File
@@ -34,7 +34,6 @@ import (
// directed by the skeleton sync's head/tail events.
type beaconBackfiller struct {
downloader *Downloader // Downloader to direct via this callback implementation
syncMode SyncMode // Sync mode to use for backfilling the skeleton chains
success func() // Callback to run on successful sync cycle completion
filling bool // Flag whether the downloader is backfilling or not
filled *types.Header // Last header filled by the last terminated sync loop
@@ -92,7 +91,6 @@ func (b *beaconBackfiller) resume() {
b.filling = true
b.filled = nil
b.started = make(chan struct{})
mode := b.syncMode
b.lock.Unlock()
// Start the backfilling on its own thread since the downloader does not have
@@ -107,7 +105,7 @@ func (b *beaconBackfiller) resume() {
}()
// If the downloader fails, report an error as in beacon chain mode there
// should be no errors as long as the chain we're syncing to is valid.
if err := b.downloader.synchronise(mode, b.started); err != nil {
if err := b.downloader.synchronise(b.started); err != nil {
log.Error("Beacon backfilling failed", "err", err)
return
}
@@ -119,27 +117,6 @@ func (b *beaconBackfiller) resume() {
}()
}
// setMode updates the sync mode from the current one to the requested one. If
// there's an active sync in progress, it will be cancelled and restarted.
func (b *beaconBackfiller) setMode(mode SyncMode) {
// Update the old sync mode and track if it was changed
b.lock.Lock()
oldMode := b.syncMode
updated := oldMode != mode
filling := b.filling
b.syncMode = mode
b.lock.Unlock()
// If the sync mode was changed mid-sync, restart. This should never ever
// really happen, we just handle it to detect programming errors.
if !updated || !filling {
return
}
log.Error("Downloader sync mode changed mid-run", "old", oldMode.String(), "new", mode.String())
b.suspend()
b.resume()
}
// SetBadBlockCallback sets the callback to run when a bad block is hit by the
// block processor. This method is not thread safe and should be set only once
// on startup before system events are fired.
@@ -153,8 +130,8 @@ func (d *Downloader) SetBadBlockCallback(onBadBlock badBlockFn) {
//
// Internally backfilling and state sync is done the same way, but the header
// retrieval and scheduling is replaced.
func (d *Downloader) BeaconSync(mode SyncMode, head *types.Header, final *types.Header) error {
return d.beaconSync(mode, head, final, true)
func (d *Downloader) BeaconSync(head *types.Header, final *types.Header) error {
return d.beaconSync(head, final, true)
}
// BeaconExtend is an optimistic version of BeaconSync, where an attempt is made
@@ -163,8 +140,8 @@ func (d *Downloader) BeaconSync(mode SyncMode, head *types.Header, final *types.
//
// This is useful if a beacon client is feeding us large chunks of payloads to run,
// but is not setting the head after each.
func (d *Downloader) BeaconExtend(mode SyncMode, head *types.Header) error {
return d.beaconSync(mode, head, nil, false)
func (d *Downloader) BeaconExtend(head *types.Header) error {
return d.beaconSync(head, nil, false)
}
// beaconSync is the post-merge version of the chain synchronization, where the
@@ -173,20 +150,9 @@ func (d *Downloader) BeaconExtend(mode SyncMode, head *types.Header) error {
//
// Internally backfilling and state sync is done the same way, but the header
// retrieval and scheduling is replaced.
func (d *Downloader) beaconSync(mode SyncMode, head *types.Header, final *types.Header, force bool) error {
// When the downloader starts a sync cycle, it needs to be aware of the sync
// mode to use (full, snap). To keep the skeleton chain oblivious, inject the
// mode into the backfiller directly.
//
// Super crazy dangerous type cast. Should be fine (TM), we're only using a
// different backfiller implementation for skeleton tests.
d.skeleton.filler.(*beaconBackfiller).setMode(mode)
func (d *Downloader) beaconSync(head *types.Header, final *types.Header, force bool) error {
// Signal the skeleton sync to switch to a new head, however it wants
if err := d.skeleton.Sync(head, final, force); err != nil {
return err
}
return nil
return d.skeleton.Sync(head, final, force)
}
// findBeaconAncestor tries to locate the common ancestor link of the local chain
+24 -4
View File
@@ -97,8 +97,9 @@ type headerTask struct {
}
type Downloader struct {
mode atomic.Uint32 // Synchronisation mode defining the strategy used (per sync cycle), use d.getMode() to get the SyncMode
mux *event.TypeMux // Event multiplexer to announce sync operation events
mode atomic.Uint32 // Synchronisation mode defining the strategy used (per sync cycle), use d.getMode() to get the SyncMode
moder *syncModer // Sync mode management, deliver the appropriate sync mode choice for each cycle
mux *event.TypeMux // Event multiplexer to announce sync operation events
queue *queue // Scheduler for selecting the hashes to download
peers *peerSet // Set of active peers from which download can proceed
@@ -165,6 +166,9 @@ type BlockChain interface {
// HasHeader verifies a header's presence in the local chain.
HasHeader(common.Hash, uint64) bool
// HasState checks if state trie is fully present in the database or not.
HasState(root common.Hash) bool
// GetHeaderByHash retrieves a header from the local chain.
GetHeaderByHash(common.Hash) *types.Header
@@ -221,10 +225,11 @@ type BlockChain interface {
}
// New creates a new downloader to fetch hashes and blocks from remote peers.
func New(stateDb ethdb.Database, mux *event.TypeMux, chain BlockChain, dropPeer peerDropFn, success func()) *Downloader {
func New(stateDb ethdb.Database, mode ethconfig.SyncMode, mux *event.TypeMux, chain BlockChain, dropPeer peerDropFn, success func()) *Downloader {
cutoffNumber, cutoffHash := chain.HistoryPruningCutoff()
dl := &Downloader{
stateDB: stateDb,
moder: newSyncModer(mode, chain, stateDb),
mux: mux,
queue: newQueue(blockCacheMaxItems, blockCacheInitialItems),
peers: newPeerSet(),
@@ -331,7 +336,7 @@ func (d *Downloader) UnregisterPeer(id string) error {
// synchronise will select the peer and use it for synchronising. If an empty string is given
// it will use the best peer possible and synchronize if its TD is higher than our own. If any of the
// checks fail an error will be returned. This method is synchronous
func (d *Downloader) synchronise(mode SyncMode, beaconPing chan struct{}) error {
func (d *Downloader) synchronise(beaconPing chan struct{}) (err error) {
// The beacon header syncer is async. It will start this synchronization and
// will continue doing other tasks. However, if synchronization needs to be
// cancelled, the syncer needs to know if we reached the startup point (and
@@ -356,6 +361,13 @@ func (d *Downloader) synchronise(mode SyncMode, beaconPing chan struct{}) error
if d.notified.CompareAndSwap(false, true) {
log.Info("Block synchronisation started")
}
mode := d.moder.get()
defer func() {
if err == nil && mode == ethconfig.SnapSync {
d.moder.disableSnap()
log.Info("Disabled snap-sync after the initial sync cycle")
}
}()
if mode == ethconfig.SnapSync {
// Snap sync will directly modify the persistent state, making the entire
// trie database unusable until the state is fully synced. To prevent any
@@ -399,6 +411,7 @@ func (d *Downloader) synchronise(mode SyncMode, beaconPing chan struct{}) error
// Atomically set the requested sync mode
d.mode.Store(uint32(mode))
defer d.mode.Store(0)
if beaconPing != nil {
close(beaconPing)
@@ -406,10 +419,17 @@ func (d *Downloader) synchronise(mode SyncMode, beaconPing chan struct{}) error
return d.syncToHead()
}
// getMode returns the sync mode used within current cycle.
func (d *Downloader) getMode() SyncMode {
return SyncMode(d.mode.Load())
}
// ConfigSyncMode returns the sync mode configured for the node.
// The actual running sync mode can differ from this.
func (d *Downloader) ConfigSyncMode() SyncMode {
return d.moder.get()
}
// syncToHead starts a block synchronization based on the hash chain from
// the specified head hash.
func (d *Downloader) syncToHead() (err error) {
+20 -19
View File
@@ -30,6 +30,7 @@ import (
"github.com/luxfi/geth/core"
"github.com/luxfi/geth/core/rawdb"
"github.com/luxfi/geth/core/types"
"github.com/luxfi/geth/eth/ethconfig"
"github.com/luxfi/geth/eth/protocols/eth"
"github.com/luxfi/geth/eth/protocols/snap"
"github.com/luxfi/geth/event"
@@ -49,12 +50,12 @@ type downloadTester struct {
}
// newTester creates a new downloader test mocker.
func newTester(t *testing.T) *downloadTester {
return newTesterWithNotification(t, nil)
func newTester(t *testing.T, mode ethconfig.SyncMode) *downloadTester {
return newTesterWithNotification(t, mode, nil)
}
// newTesterWithNotification creates a new downloader test mocker.
func newTesterWithNotification(t *testing.T, success func()) *downloadTester {
func newTesterWithNotification(t *testing.T, mode ethconfig.SyncMode, success func()) *downloadTester {
db, err := rawdb.Open(rawdb.NewMemoryDatabase(), rawdb.OpenOptions{})
if err != nil {
panic(err)
@@ -75,7 +76,7 @@ func newTesterWithNotification(t *testing.T, success func()) *downloadTester {
chain: chain,
peers: make(map[string]*downloadTesterPeer),
}
tester.downloader = New(db, new(event.TypeMux), tester.chain, tester.dropPeer, success)
tester.downloader = New(db, mode, new(event.TypeMux), tester.chain, tester.dropPeer, success)
return tester
}
@@ -393,7 +394,7 @@ func TestCanonicalSynchronisation68Snap(t *testing.T) { testCanonSync(t, eth.ETH
func testCanonSync(t *testing.T, protocol uint, mode SyncMode) {
success := make(chan struct{})
tester := newTesterWithNotification(t, func() {
tester := newTesterWithNotification(t, mode, func() {
close(success)
})
defer tester.terminate()
@@ -403,7 +404,7 @@ func testCanonSync(t *testing.T, protocol uint, mode SyncMode) {
tester.newPeer("peer", protocol, chain.blocks[1:])
// Synchronise with the peer and make sure all relevant data was retrieved
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("failed to beacon-sync chain: %v", err)
}
select {
@@ -420,7 +421,7 @@ func TestThrottling68Full(t *testing.T) { testThrottling(t, eth.ETH68, FullSync)
func TestThrottling68Snap(t *testing.T) { testThrottling(t, eth.ETH68, SnapSync) }
func testThrottling(t *testing.T, protocol uint, mode SyncMode) {
tester := newTester(t)
tester := newTester(t, mode)
defer tester.terminate()
// Create a long block chain to download and the tester
@@ -437,7 +438,7 @@ func testThrottling(t *testing.T, protocol uint, mode SyncMode) {
// Start a synchronisation concurrently
errc := make(chan error, 1)
go func() {
errc <- tester.downloader.BeaconSync(mode, testChainBase.blocks[len(testChainBase.blocks)-1].Header(), nil)
errc <- tester.downloader.BeaconSync(testChainBase.blocks[len(testChainBase.blocks)-1].Header(), nil)
}()
// Iteratively take some blocks, always checking the retrieval count
for {
@@ -502,7 +503,7 @@ func testCancel(t *testing.T, protocol uint, mode SyncMode) {
success := func() {
close(complete)
}
tester := newTesterWithNotification(t, success)
tester := newTesterWithNotification(t, mode, success)
defer tester.terminate()
chain := testChainBase.shorten(MaxHeaderFetch)
@@ -514,7 +515,7 @@ func testCancel(t *testing.T, protocol uint, mode SyncMode) {
t.Errorf("download queue not idle")
}
// Synchronise with the peer, but cancel afterwards
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("failed to synchronise blocks: %v", err)
}
<-complete
@@ -534,7 +535,7 @@ func testMultiProtoSync(t *testing.T, protocol uint, mode SyncMode) {
success := func() {
close(complete)
}
tester := newTesterWithNotification(t, success)
tester := newTesterWithNotification(t, mode, success)
defer tester.terminate()
// Create a small enough block chain to download
@@ -543,7 +544,7 @@ func testMultiProtoSync(t *testing.T, protocol uint, mode SyncMode) {
// Create peers of every type
tester.newPeer("peer 68", eth.ETH68, chain.blocks[1:])
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("failed to start beacon sync: %v", err)
}
select {
@@ -570,7 +571,7 @@ func TestEmptyShortCircuit68Snap(t *testing.T) { testEmptyShortCircuit(t, eth.ET
func testEmptyShortCircuit(t *testing.T, protocol uint, mode SyncMode) {
success := make(chan struct{})
tester := newTesterWithNotification(t, func() {
tester := newTesterWithNotification(t, mode, func() {
close(success)
})
defer tester.terminate()
@@ -588,7 +589,7 @@ func testEmptyShortCircuit(t *testing.T, protocol uint, mode SyncMode) {
receiptsHave.Add(int32(len(headers)))
}
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("failed to synchronise blocks: %v", err)
}
select {
@@ -650,7 +651,7 @@ func testBeaconSync(t *testing.T, protocol uint, mode SyncMode) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
success := make(chan struct{})
tester := newTesterWithNotification(t, func() {
tester := newTesterWithNotification(t, mode, func() {
close(success)
})
defer tester.terminate()
@@ -662,7 +663,7 @@ func testBeaconSync(t *testing.T, protocol uint, mode SyncMode) {
if c.local > 0 {
tester.chain.InsertChain(chain.blocks[1 : c.local+1])
}
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("Failed to beacon sync chain %v %v", c.name, err)
}
select {
@@ -685,7 +686,7 @@ func TestSyncProgress68Snap(t *testing.T) { testSyncProgress(t, eth.ETH68, SnapS
func testSyncProgress(t *testing.T, protocol uint, mode SyncMode) {
success := make(chan struct{})
tester := newTesterWithNotification(t, func() {
tester := newTesterWithNotification(t, mode, func() {
success <- struct{}{}
})
defer tester.terminate()
@@ -700,7 +701,7 @@ func testSyncProgress(t *testing.T, protocol uint, mode SyncMode) {
faultyPeer.withholdBodies[header.Hash()] = struct{}{}
}
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)/2-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)/2-1].Header(), nil); err != nil {
t.Fatalf("failed to beacon-sync chain: %v", err)
}
select {
@@ -716,7 +717,7 @@ func testSyncProgress(t *testing.T, protocol uint, mode SyncMode) {
// Synchronise all the blocks and check continuation progress
tester.newPeer("peer-full", protocol, chain.blocks[1:])
if err := tester.downloader.BeaconSync(mode, chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
if err := tester.downloader.BeaconSync(chain.blocks[len(chain.blocks)-1].Header(), nil); err != nil {
t.Fatalf("failed to beacon-sync chain: %v", err)
}
startingBlock := uint64(len(chain.blocks)/2 - 1)
+111
View File
@@ -0,0 +1,111 @@
// Copyright 2025 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package downloader
import (
"sync"
"github.com/luxfi/geth/core/rawdb"
"github.com/luxfi/geth/eth/ethconfig"
"github.com/luxfi/geth/ethdb"
"github.com/luxfi/geth/log"
)
// syncModer is responsible for managing the downloader's sync mode. It takes the
// user's preference at startup and then determines the appropriate sync mode
// based on the current chain status.
type syncModer struct {
mode ethconfig.SyncMode
chain BlockChain
disk ethdb.KeyValueReader
lock sync.Mutex
}
func newSyncModer(mode ethconfig.SyncMode, chain BlockChain, disk ethdb.KeyValueReader) *syncModer {
if mode == ethconfig.FullSync {
// The database seems empty as the current block is the genesis. Yet the snap
// block is ahead, so snap sync was enabled for this node at a certain point.
// The scenarios where this can happen is
// * if the user manually (or via a bad block) rolled back a snap sync node
// below the sync point.
// * the last snap sync is not finished while user specifies a full sync this
// time. But we don't have any recent state for full sync.
// In these cases however it's safe to reenable snap sync.
fullBlock, snapBlock := chain.CurrentBlock(), chain.CurrentSnapBlock()
if fullBlock.Number.Uint64() == 0 && snapBlock.Number.Uint64() > 0 {
mode = ethconfig.SnapSync
log.Warn("Switching from full-sync to snap-sync", "reason", "snap-sync incomplete")
} else if !chain.HasState(fullBlock.Root) {
mode = ethconfig.SnapSync
log.Warn("Switching from full-sync to snap-sync", "reason", "head state missing")
} else {
// Grant the full sync mode
log.Info("Enabled full-sync", "head", fullBlock.Number, "hash", fullBlock.Hash())
}
} else {
head := chain.CurrentBlock()
if head.Number.Uint64() > 0 && chain.HasState(head.Root) {
mode = ethconfig.FullSync
log.Info("Switching from snap-sync to full-sync", "reason", "snap-sync complete")
} else {
// If snap sync was requested and our database is empty, grant it
log.Info("Enabled snap-sync", "head", head.Number, "hash", head.Hash())
}
}
return &syncModer{
mode: mode,
chain: chain,
disk: disk,
}
}
// get retrieves the current sync mode, either explicitly set, or derived
// from the chain status.
func (m *syncModer) get() ethconfig.SyncMode {
m.lock.Lock()
defer m.lock.Unlock()
// If we're in snap sync mode, return that directly
if m.mode == ethconfig.SnapSync {
return ethconfig.SnapSync
}
// We are probably in full sync, but we might have rewound to before the
// snap sync pivot, check if we should re-enable snap sync.
head := m.chain.CurrentBlock()
if pivot := rawdb.ReadLastPivotNumber(m.disk); pivot != nil {
if head.Number.Uint64() < *pivot {
log.Info("Reenabled snap-sync as chain is lagging behind the pivot", "head", head.Number, "pivot", pivot)
return ethconfig.SnapSync
}
}
// We are in a full sync, but the associated head state is missing. To complete
// the head state, forcefully rerun the snap sync. Note it doesn't mean the
// persistent state is corrupted, just mismatch with the head block.
if !m.chain.HasState(head.Root) {
log.Info("Reenabled snap-sync as chain is stateless")
return ethconfig.SnapSync
}
// Nope, we're really full syncing
return ethconfig.FullSync
}
// disableSnap disables the snap sync mode, usually it's called after a successful snap sync.
func (m *syncModer) disableSnap() {
m.lock.Lock()
m.mode = ethconfig.FullSync
m.lock.Unlock()
}
+5
View File
@@ -72,6 +72,7 @@ var Defaults = Config{
RPCTxFeeCap: 1, // 1 ether
TxSyncDefaultTimeout: 20 * time.Second,
TxSyncMaxTimeout: 1 * time.Minute,
SlowBlockThreshold: time.Second * 2,
}
//go:generate go run github.com/fjl/gencodec -type Config -formats toml -out gen_config.go
@@ -118,6 +119,10 @@ type Config struct {
// presence of these blocks for every new peer connection.
RequiredBlocks map[uint64]common.Hash `toml:"-"`
// SlowBlockThreshold is the block execution speed threshold (Mgas/s)
// below which detailed statistics are logged.
SlowBlockThreshold time.Duration `toml:",omitempty"`
// Database options
SkipBcVersionCheck bool `toml:"-"`
DatabaseHandles int `toml:"-"`
+6
View File
@@ -33,6 +33,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
StateHistory uint64 `toml:",omitempty"`
StateScheme string `toml:",omitempty"`
RequiredBlocks map[uint64]common.Hash `toml:"-"`
SlowBlockThreshold time.Duration `toml:",omitempty"`
SkipBcVersionCheck bool `toml:"-"`
DatabaseHandles int `toml:"-"`
DatabaseCache int
@@ -82,6 +83,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.StateHistory = c.StateHistory
enc.StateScheme = c.StateScheme
enc.RequiredBlocks = c.RequiredBlocks
enc.SlowBlockThreshold = c.SlowBlockThreshold
enc.SkipBcVersionCheck = c.SkipBcVersionCheck
enc.DatabaseHandles = c.DatabaseHandles
enc.DatabaseCache = c.DatabaseCache
@@ -135,6 +137,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
StateHistory *uint64 `toml:",omitempty"`
StateScheme *string `toml:",omitempty"`
RequiredBlocks map[uint64]common.Hash `toml:"-"`
SlowBlockThreshold *time.Duration `toml:",omitempty"`
SkipBcVersionCheck *bool `toml:"-"`
DatabaseHandles *int `toml:"-"`
DatabaseCache *int
@@ -219,6 +222,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.RequiredBlocks != nil {
c.RequiredBlocks = dec.RequiredBlocks
}
if dec.SlowBlockThreshold != nil {
c.SlowBlockThreshold = *dec.SlowBlockThreshold
}
if dec.SkipBcVersionCheck != nil {
c.SkipBcVersionCheck = *dec.SkipBcVersionCheck
}
+45 -38
View File
@@ -170,10 +170,10 @@ type TxFetcher struct {
alternates map[common.Hash]map[string]struct{} // In-flight transaction alternate origins if retrieval fails
// Callbacks
hasTx func(common.Hash) bool // Retrieves a tx from the local txpool
addTxs func([]*types.Transaction) []error // Insert a batch of transactions into local txpool
fetchTxs func(string, []common.Hash) error // Retrieves a set of txs from a remote peer
dropPeer func(string) // Drops a peer in case of announcement violation
validateMeta func(common.Hash, byte) error // Validate a tx metadata based on the local txpool
addTxs func([]*types.Transaction) []error // Insert a batch of transactions into local txpool
fetchTxs func(string, []common.Hash) error // Retrieves a set of txs from a remote peer
dropPeer func(string) // Drops a peer in case of announcement violation
step chan struct{} // Notification channel when the fetcher loop iterates
clock mclock.Clock // Monotonic clock or simulated clock for tests
@@ -183,36 +183,36 @@ type TxFetcher struct {
// NewTxFetcher creates a transaction fetcher to retrieve transaction
// based on hash announcements.
func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher {
return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil)
func NewTxFetcher(validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher {
return NewTxFetcherForTests(validateMeta, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil)
}
// NewTxFetcherForTests is a testing method to mock out the realtime clock with
// a simulated version and the internal randomness with a deterministic one.
func NewTxFetcherForTests(
hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string),
validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string),
clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher {
return &TxFetcher{
notify: make(chan *txAnnounce),
cleanup: make(chan *txDelivery),
drop: make(chan *txDrop),
quit: make(chan struct{}),
waitlist: make(map[common.Hash]map[string]struct{}),
waittime: make(map[common.Hash]mclock.AbsTime),
waitslots: make(map[string]map[common.Hash]*txMetadataWithSeq),
announces: make(map[string]map[common.Hash]*txMetadataWithSeq),
announced: make(map[common.Hash]map[string]struct{}),
fetching: make(map[common.Hash]string),
requests: make(map[string]*txRequest),
alternates: make(map[common.Hash]map[string]struct{}),
underpriced: lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
hasTx: hasTx,
addTxs: addTxs,
fetchTxs: fetchTxs,
dropPeer: dropPeer,
clock: clock,
realTime: realTime,
rand: rand,
notify: make(chan *txAnnounce),
cleanup: make(chan *txDelivery),
drop: make(chan *txDrop),
quit: make(chan struct{}),
waitlist: make(map[common.Hash]map[string]struct{}),
waittime: make(map[common.Hash]mclock.AbsTime),
waitslots: make(map[string]map[common.Hash]*txMetadataWithSeq),
announces: make(map[string]map[common.Hash]*txMetadataWithSeq),
announced: make(map[common.Hash]map[string]struct{}),
fetching: make(map[common.Hash]string),
requests: make(map[string]*txRequest),
alternates: make(map[common.Hash]map[string]struct{}),
underpriced: lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
validateMeta: validateMeta,
addTxs: addTxs,
fetchTxs: fetchTxs,
dropPeer: dropPeer,
clock: clock,
realTime: realTime,
rand: rand,
}
}
@@ -235,19 +235,26 @@ func (f *TxFetcher) Notify(peer string, types []byte, sizes []uint32, hashes []c
underpriced int64
)
for i, hash := range hashes {
switch {
case f.hasTx(hash):
err := f.validateMeta(hash, types[i])
if errors.Is(err, txpool.ErrAlreadyKnown) {
duplicate++
case f.isKnownUnderpriced(hash):
underpriced++
default:
unknownHashes = append(unknownHashes, hash)
// Transaction metadata has been available since eth68, and all
// legacy eth protocols (prior to eth68) have been deprecated.
// Therefore, metadata is always expected in the announcement.
unknownMetas = append(unknownMetas, txMetadata{kind: types[i], size: sizes[i]})
continue
}
if err != nil {
continue
}
if f.isKnownUnderpriced(hash) {
underpriced++
continue
}
unknownHashes = append(unknownHashes, hash)
// Transaction metadata has been available since eth68, and all
// legacy eth protocols (prior to eth68) have been deprecated.
// Therefore, metadata is always expected in the announcement.
unknownMetas = append(unknownMetas, txMetadata{kind: types[i], size: sizes[i]})
}
txAnnounceKnownMeter.Mark(duplicate)
txAnnounceUnderpricedMeter.Mark(underpriced)
+57 -28
View File
@@ -93,7 +93,7 @@ func TestTransactionFetcherWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -295,7 +295,7 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -385,7 +385,7 @@ func TestTransactionFetcherSingletonRequesting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -490,7 +490,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(origin string, hashes []common.Hash) error {
<-proceed
@@ -574,7 +574,7 @@ func TestTransactionFetcherCleanup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -618,7 +618,7 @@ func TestTransactionFetcherCleanupEmpty(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -661,7 +661,7 @@ func TestTransactionFetcherMissingRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -722,7 +722,7 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -771,7 +771,7 @@ func TestTransactionFetcherBroadcasts(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -827,7 +827,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -897,7 +897,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -975,7 +975,7 @@ func TestTransactionFetcherTimeoutTimerResets(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -1053,7 +1053,7 @@ func TestTransactionFetcherRateLimiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -1083,7 +1083,7 @@ func TestTransactionFetcherBandwidthLimiting(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -1200,7 +1200,7 @@ func TestTransactionFetcherDoSProtection(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -1267,7 +1267,7 @@ func TestTransactionFetcherUnderpricedDedup(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ {
@@ -1368,7 +1368,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) {
testTransactionFetcher(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ {
@@ -1400,7 +1400,7 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1459,7 +1459,7 @@ func TestTransactionFetcherDrop(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1533,7 +1533,7 @@ func TestTransactionFetcherDropRescheduling(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1579,7 +1579,7 @@ func TestInvalidAnnounceMetadata(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1662,7 +1662,7 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1690,7 +1690,7 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1720,7 +1720,7 @@ func TestTransactionFetcherFuzzCrash03(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1759,7 +1759,7 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1794,7 +1794,7 @@ func TestBlobTransactionAnnounce(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
nil,
func(string, []common.Hash) error { return nil },
nil,
@@ -1862,7 +1862,7 @@ func TestTransactionFetcherDropAlternates(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
@@ -1908,6 +1908,35 @@ func TestTransactionFetcherDropAlternates(t *testing.T) {
})
}
func TestTransactionFetcherWrongMetadata(t *testing.T) {
testTransactionFetcherParallel(t, txFetcherTest{
init: func() *TxFetcher {
return NewTxFetcher(
func(_ common.Hash, kind byte) error {
switch kind {
case types.LegacyTxType, types.AccessListTxType, types.DynamicFeeTxType, types.BlobTxType, types.SetCodeTxType:
return nil
}
return types.ErrTxTypeNotSupported
},
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
func(string, []common.Hash) error { return nil },
nil,
)
},
steps: []interface{}{
doTxNotify{peer: "A", hashes: []common.Hash{{0x01}, {0x02}}, types: []byte{0xff, types.LegacyTxType}, sizes: []uint32{111, 222}},
isWaiting(map[string][]announce{
"A": {
{common.Hash{0x02}, types.LegacyTxType, 222},
},
}),
},
})
}
func testTransactionFetcherParallel(t *testing.T, tt txFetcherTest) {
t.Parallel()
testTransactionFetcher(t, tt)
@@ -2245,7 +2274,7 @@ func TestTransactionForgotten(t *testing.T) {
}
fetcher := NewTxFetcherForTests(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
errs := make([]error, len(txs))
for i := 0; i < len(errs); i++ {
+17 -5
View File
@@ -35,17 +35,29 @@ import (
)
var (
errInvalidTopic = errors.New("invalid topic(s)")
errFilterNotFound = errors.New("filter not found")
errInvalidBlockRange = errors.New("invalid block range params")
errInvalidTopic = invalidParamsErr("invalid topic(s)")
errInvalidBlockRange = invalidParamsErr("invalid block range params")
errBlockRangeIntoFuture = invalidParamsErr("block range extends beyond current head block")
errBlockHashWithRange = invalidParamsErr("can't specify fromBlock/toBlock with blockHash")
errPendingLogsUnsupported = invalidParamsErr("pending logs are not supported")
errUnknownBlock = errors.New("unknown block")
errBlockHashWithRange = errors.New("can't specify fromBlock/toBlock with blockHash")
errPendingLogsUnsupported = errors.New("pending logs are not supported")
errFilterNotFound = errors.New("filter not found")
errExceedMaxTopics = errors.New("exceed max topics")
errExceedLogQueryLimit = errors.New("exceed max addresses or topics per search position")
errExceedMaxTxHashes = errors.New("exceed max number of transaction hashes allowed per transactionReceipts subscription")
)
type invalidParamsError struct {
err error
}
func (e invalidParamsError) Error() string { return e.err.Error() }
func (e invalidParamsError) ErrorCode() int { return -32602 }
func invalidParamsErr(format string, args ...any) error {
return invalidParamsError{fmt.Errorf(format, args...)}
}
const (
// The maximum number of topic criteria allowed, vm.LOG4 - vm.LOG0
maxTopics = 4
+4 -1
View File
@@ -221,9 +221,12 @@ func (s *searchSession) updateChainView() error {
if lastBlock == math.MaxUint64 {
lastBlock = head
}
if firstBlock > lastBlock || lastBlock > head {
if firstBlock > lastBlock {
return errInvalidBlockRange
}
if lastBlock > head {
return errBlockRangeIntoFuture
}
s.searchRange = common.NewRange(firstBlock, lastBlock+1-firstBlock)
// Trim existing match set in case a reorg may have invalidated some results
+22 -45
View File
@@ -92,6 +92,9 @@ type txPool interface {
// can decide whether to receive notifications only for newly seen transactions
// or also for reorged out ones.
SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription
// FilterType returns whether the given tx type is supported by the txPool.
FilterType(kind byte) bool
}
// handlerConfig is the collection of initialization parameters to create a full
@@ -111,9 +114,7 @@ type handlerConfig struct {
type handler struct {
nodeID enode.ID
networkID uint64
snapSync atomic.Bool // Flag whether snap sync is enabled (gets disabled if we already have blocks)
synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
synced atomic.Bool // Flag whether we're considered synchronised (enables transaction processing)
database ethdb.Database
txpool txPool
@@ -161,40 +162,13 @@ func newHandler(config *handlerConfig) (*handler, error) {
handlerDoneCh: make(chan struct{}),
handlerStartCh: make(chan struct{}),
}
if config.Sync == ethconfig.FullSync {
// The database seems empty as the current block is the genesis. Yet the snap
// block is ahead, so snap sync was enabled for this node at a certain point.
// The scenarios where this can happen is
// * if the user manually (or via a bad block) rolled back a snap sync node
// below the sync point.
// * the last snap sync is not finished while user specifies a full sync this
// time. But we don't have any recent state for full sync.
// In these cases however it's safe to reenable snap sync.
fullBlock, snapBlock := h.chain.CurrentBlock(), h.chain.CurrentSnapBlock()
if fullBlock.Number.Uint64() == 0 && snapBlock.Number.Uint64() > 0 {
h.snapSync.Store(true)
log.Warn("Switch sync mode from full sync to snap sync", "reason", "snap sync incomplete")
} else if !h.chain.HasState(fullBlock.Root) {
h.snapSync.Store(true)
log.Warn("Switch sync mode from full sync to snap sync", "reason", "head state missing")
}
} else {
head := h.chain.CurrentBlock()
if head.Number.Uint64() > 0 && h.chain.HasState(head.Root) {
log.Info("Switch sync mode from snap sync to full sync", "reason", "snap sync complete")
} else {
// If snap sync was requested and our database is empty, grant it
h.snapSync.Store(true)
log.Info("Enabled snap sync", "head", head.Number, "hash", head.Hash())
}
}
// Construct the downloader (long sync)
h.downloader = downloader.New(config.Database, config.Sync, h.eventMux, h.chain, h.removePeer, h.enableSyncedFeatures)
// If snap sync is requested but snapshots are disabled, fail loudly
if h.snapSync.Load() && (config.Chain.Snapshots() == nil && config.Chain.TrieDB().Scheme() == rawdb.HashScheme) {
if h.downloader.ConfigSyncMode() == ethconfig.SnapSync && (config.Chain.Snapshots() == nil && config.Chain.TrieDB().Scheme() == rawdb.HashScheme) {
return nil, errors.New("snap sync not supported with snapshots disabled")
}
// Construct the downloader (long sync)
h.downloader = downloader.New(config.Database, h.eventMux, h.chain, h.removePeer, h.enableSyncedFeatures)
fetchTx := func(peer string, hashes []common.Hash) error {
p := h.peers.peer(peer)
if p == nil {
@@ -205,7 +179,18 @@ func newHandler(config *handlerConfig) (*handler, error) {
addTxs := func(txs []*types.Transaction) []error {
return h.txpool.Add(txs, false)
}
h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, addTxs, fetchTx, h.removePeer)
validateMeta := func(tx common.Hash, kind byte) error {
if h.txpool.Has(tx) {
return txpool.ErrAlreadyKnown
}
if !h.txpool.FilterType(kind) {
return types.ErrTxTypeNotSupported
}
return nil
}
h.txFetcher = fetcher.NewTxFetcher(validateMeta, addTxs, fetchTx, h.removePeer)
return h, nil
}
@@ -267,7 +252,7 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
return err
}
reject := false // reserved peer slots
if h.snapSync.Load() {
if h.downloader.ConfigSyncMode() == ethconfig.SnapSync {
if snap == nil {
// If we are running snap-sync, we want to reserve roughly half the peer
// slots for peers supporting the snap protocol.
@@ -544,15 +529,7 @@ func (h *handler) txBroadcastLoop() {
// enableSyncedFeatures enables the post-sync functionalities when the initial
// sync is finished.
func (h *handler) enableSyncedFeatures() {
// Mark the local node as synced.
h.synced.Store(true)
// If we were running snap sync and it finished, disable doing another
// round on next sync cycle
if h.snapSync.Load() {
log.Info("Snap sync complete, auto disabling")
h.snapSync.Store(false)
}
}
// blockRangeState holds the state of the block range update broadcasting mechanism.
@@ -590,7 +567,7 @@ func (h *handler) blockRangeLoop(st *blockRangeState) {
if ev == nil {
continue
}
if _, ok := ev.Data.(downloader.StartEvent); ok && h.snapSync.Load() {
if _, ok := ev.Data.(downloader.StartEvent); ok && h.downloader.ConfigSyncMode() == ethconfig.SnapSync {
h.blockRangeWhileSnapSyncing(st)
}
case <-st.headCh:
+4 -5
View File
@@ -232,7 +232,7 @@ func testRecvTransactions(t *testing.T, protocol uint) {
t.Parallel()
// Create a message handler, configure it to accept transactions and watch them
handler := newTestHandler()
handler := newTestHandler(ethconfig.FullSync)
defer handler.close()
handler.handler.synced.Store(true) // mark synced to accept transactions
@@ -284,7 +284,7 @@ func testSendTransactions(t *testing.T, protocol uint) {
t.Parallel()
// Create a message handler and fill the pool with big transactions
handler := newTestHandler()
handler := newTestHandler(ethconfig.FullSync)
defer handler.close()
insert := make([]*types.Transaction, 100)
@@ -365,13 +365,12 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
// Create a source handler to send transactions from and a number of sinks
// to receive them. We need multiple sinks since a one-to-one peering would
// broadcast all transactions without announcement.
source := newTestHandler()
source.handler.snapSync.Store(false) // Avoid requiring snap, otherwise some will be dropped below
source := newTestHandler(ethconfig.FullSync)
defer source.close()
sinks := make([]*testHandler, 10)
for i := 0; i < len(sinks); i++ {
sinks[i] = newTestHandler()
sinks[i] = newTestHandler(ethconfig.FullSync)
defer sinks[i].close()
sinks[i].handler.synced.Store(true) // mark synced to accept transactions
+13 -4
View File
@@ -163,6 +163,15 @@ func (p *testTxPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bo
return p.txFeed.Subscribe(ch)
}
// FilterType should check whether the pool supports the given type of transactions.
func (p *testTxPool) FilterType(kind byte) bool {
switch kind {
case types.LegacyTxType, types.AccessListTxType, types.DynamicFeeTxType, types.BlobTxType, types.SetCodeTxType:
return true
}
return false
}
// testHandler is a live implementation of the Ethereum protocol handler, just
// preinitialized with some sane testing defaults and the transaction pool mocked
// out.
@@ -174,13 +183,13 @@ type testHandler struct {
}
// newTestHandler creates a new handler for testing purposes with no blocks.
func newTestHandler() *testHandler {
return newTestHandlerWithBlocks(0)
func newTestHandler(mode ethconfig.SyncMode) *testHandler {
return newTestHandlerWithBlocks(0, mode)
}
// newTestHandlerWithBlocks creates a new handler for testing purposes, with a
// given number of initial blocks.
func newTestHandlerWithBlocks(blocks int) *testHandler {
func newTestHandlerWithBlocks(blocks int, mode ethconfig.SyncMode) *testHandler {
// Create a database pre-initialize with a genesis block
db := rawdb.NewMemoryDatabase()
gspec := &core.Genesis{
@@ -200,7 +209,7 @@ func newTestHandlerWithBlocks(blocks int) *testHandler {
Chain: chain,
TxPool: txpool,
Network: 1,
Sync: ethconfig.SnapSync,
Sync: mode,
BloomCache: 1,
})
handler.Start(1000)
+4 -10
View File
@@ -36,17 +36,11 @@ func testSnapSyncDisabling(t *testing.T, ethVer uint, snapVer uint) {
t.Parallel()
// Create an empty handler and ensure it's in snap sync mode
empty := newTestHandler()
if !empty.handler.snapSync.Load() {
t.Fatalf("snap sync disabled on pristine blockchain")
}
empty := newTestHandler(ethconfig.SnapSync)
defer empty.close()
// Create a full handler and ensure snap sync ends up disabled
full := newTestHandlerWithBlocks(1024)
if full.handler.snapSync.Load() {
t.Fatalf("snap sync not disabled on non-empty blockchain")
}
full := newTestHandlerWithBlocks(1024, ethconfig.SnapSync)
defer full.close()
// Sync up the two handlers via both `eth` and `snap`
@@ -85,7 +79,7 @@ func testSnapSyncDisabling(t *testing.T, ethVer uint, snapVer uint) {
time.Sleep(250 * time.Millisecond)
// Check that snap sync was disabled
if err := empty.handler.downloader.BeaconSync(ethconfig.SnapSync, full.chain.CurrentBlock(), nil); err != nil {
if err := empty.handler.downloader.BeaconSync(full.chain.CurrentBlock(), nil); err != nil {
t.Fatal("sync failed:", err)
}
// Downloader internally has to wait for a timer (3s) to be expired before
@@ -96,7 +90,7 @@ func testSnapSyncDisabling(t *testing.T, ethVer uint, snapVer uint) {
case <-timeout:
t.Fatalf("snap sync not disabled after successful synchronisation")
case <-time.After(100 * time.Millisecond):
if !empty.handler.snapSync.Load() {
if empty.handler.downloader.ConfigSyncMode() == ethconfig.FullSync {
return
}
}
+5 -1
View File
@@ -130,7 +130,11 @@ func (s *Syncer) run() {
break
}
if resync {
req.errc <- s.backend.Downloader().BeaconDevSync(ethconfig.FullSync, target)
if mode := s.backend.Downloader().ConfigSyncMode(); mode != ethconfig.FullSync {
req.errc <- fmt.Errorf("unsupported syncmode %v, please relaunch geth with --syncmode full", mode)
} else {
req.errc <- s.backend.Downloader().BeaconDevSync(target)
}
}
case <-ticker.C:
+1 -1
View File
@@ -986,7 +986,7 @@ func (api *API) TraceCall(ctx context.Context, args ethapi.TransactionArgs, bloc
}
var (
msg = args.ToMessage(blockContext.BaseFee, true)
tx = args.ToTransaction(types.LegacyTxType)
tx = args.ToTransaction(types.DynamicFeeTxType)
traceConfig *TraceConfig
)
// Lower the basefee to 0 to avoid breaking EVM
+1 -1
View File
@@ -513,7 +513,7 @@ func defaultIgnoredOpcodes() []hexutil.Uint64 {
ignored := make([]hexutil.Uint64, 0, 64)
// Allow all PUSHx, DUPx and SWAPx opcodes as they have sequential codes
for op := vm.PUSH0; op < vm.SWAP16; op++ {
for op := vm.PUSH0; op <= vm.SWAP16; op++ {
ignored = append(ignored, hexutil.Uint64(op))
}
+21 -5
View File
@@ -205,8 +205,8 @@ func New(file string, cache int, handles int, namespace string, readonly bool) (
// limit unchanged allows writes to be flushed more smoothly. This helps
// avoid compaction spikes and mitigates write stalls caused by heavy
// compaction workloads.
memTableLimit := 4
memTableSize := cache * 1024 * 1024 / 2 / memTableLimit
memTableNumber := 4
memTableSize := cache * 1024 * 1024 / 2 / memTableNumber
// The memory table size is currently capped at maxMemTableSize-1 due to a
// known bug in the pebble where maxMemTableSize is not recognized as a
@@ -243,12 +243,16 @@ func New(file string, cache int, handles int, namespace string, readonly bool) (
// Note, there may have more than two memory tables in the system.
MemTableSize: uint64(memTableSize),
// MemTableStopWritesThreshold places a hard limit on the size
// MemTableStopWritesThreshold places a hard limit on the number
// of the existent MemTables(including the frozen one).
//
// Note, this must be the number of tables not the size of all memtables
// according to https://github.com/cockroachdb/pebble/blob/master/options.go#L738-L742
// and to https://github.com/cockroachdb/pebble/blob/master/db.go#L1892-L1903.
MemTableStopWritesThreshold: memTableLimit,
//
// MemTableStopWritesThreshold is set to twice the maximum number of
// allowed memtables to accommodate temporary spikes.
MemTableStopWritesThreshold: memTableNumber * 2,
// The default compaction concurrency(1 thread),
// Here use all available CPUs for faster compaction.
@@ -263,7 +267,9 @@ func New(file string, cache int, handles int, namespace string, readonly bool) (
{TargetFileSize: 16 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
{TargetFileSize: 32 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
{TargetFileSize: 64 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
{TargetFileSize: 128 * 1024 * 1024, FilterPolicy: bloom.FilterPolicy(10)},
// Pebble doesn't use the Bloom filter at level6 for read efficiency.
{TargetFileSize: 128 * 1024 * 1024},
},
ReadOnly: readonly,
EventListener: &pebble.EventListener{
@@ -298,6 +304,16 @@ func New(file string, cache int, handles int, namespace string, readonly bool) (
// for more details.
opt.Experimental.ReadSamplingMultiplier = -1
// These two settings define the conditions under which compaction concurrency
// is increased. Specifically, one additional compaction job will be enabled when:
// - there is one more overlapping sub-level0;
// - there is an additional 512 MB of compaction debt;
//
// The maximum concurrency is still capped by MaxConcurrentCompactions, but with
// these settings compactions can scale up more readily.
opt.Experimental.L0CompactionConcurrency = 1
opt.Experimental.CompactionDebtConcurrency = 1 << 28 // 256MB
// Open the db and recover any potential corruptions
innerDB, err := pebble.Open(file, opt)
if err != nil {
+7 -6
View File
@@ -21,20 +21,21 @@ import (
"io"
"os"
"path/filepath"
"sort"
"strings"
"slices"
)
// HashFolder iterates all files under the given directory, computing the hash
// of each.
func HashFolder(folder string, exlude []string) (map[string][32]byte, error) {
func HashFolder(folder string, excludes []string) (map[string][32]byte, error) {
res := make(map[string][32]byte)
err := filepath.WalkDir(folder, func(path string, d os.DirEntry, _ error) error {
// Skip anything that's exluded or not a regular file
for _, skip := range exlude {
if strings.HasPrefix(path, filepath.FromSlash(skip)) {
// Skip anything that's excluded or not a regular file
if slices.Contains(excludes, path) {
if d.IsDir() {
return filepath.SkipDir
}
return nil
}
if !d.Type().IsRegular() {
return nil
@@ -71,6 +72,6 @@ func DiffHashes(a map[string][32]byte, b map[string][32]byte) []string {
updates = append(updates, file)
}
}
sort.Strings(updates)
slices.Sort(updates)
return updates
}
+1 -1
View File
@@ -252,7 +252,7 @@ func (*HandlerT) SetGCPercent(v int) int {
// - Geth also allocates memory off-heap, particularly for fastCache and Pebble,
// which can be non-trivial (a few gigabytes by default).
func (*HandlerT) SetMemoryLimit(limit int64) int64 {
log.Info("Setting memory limit", "size", common.PrettyDuration(limit))
log.Info("Setting memory limit", "size", common.StorageSize(limit))
return debug.SetMemoryLimit(limit)
}
+15 -14
View File
@@ -374,9 +374,9 @@ func (api *BlockChainAPI) GetProof(ctx context.Context, address common.Address,
// Deserialize all keys. This prevents state access on invalid input.
for i, hexKey := range storageKeys {
var err error
keys[i], keyLengths[i], err = decodeHash(hexKey)
keys[i], keyLengths[i], err = decodeStorageKey(hexKey)
if err != nil {
return nil, err
return nil, &invalidParamsError{fmt.Sprintf("%v: %q", err, hexKey)}
}
}
statedb, header, err := api.b.StateAndHeaderByNumberOrHash(ctx, blockNrOrHash)
@@ -441,9 +441,10 @@ func (api *BlockChainAPI) GetProof(ctx context.Context, address common.Address,
}, statedb.Error()
}
// decodeHash parses a hex-encoded 32-byte hash. The input may optionally
// be prefixed by 0x and can have a byte length up to 32.
func decodeHash(s string) (h common.Hash, inputLength int, err error) {
// decodeStorageKey parses a hex-encoded 32-byte hash.
// For legacy compatibility reasons, we parse these keys leniently,
// with the 0x prefix being optional.
func decodeStorageKey(s string) (h common.Hash, inputLength int, err error) {
if strings.HasPrefix(s, "0x") || strings.HasPrefix(s, "0X") {
s = s[2:]
}
@@ -451,11 +452,11 @@ func decodeHash(s string) (h common.Hash, inputLength int, err error) {
s = "0" + s
}
if len(s) > 64 {
return common.Hash{}, len(s) / 2, errors.New("hex string too long, want at most 32 bytes")
return common.Hash{}, len(s) / 2, errors.New("storage key too long (want at most 32 bytes)")
}
b, err := hex.DecodeString(s)
if err != nil {
return common.Hash{}, 0, errors.New("hex string invalid")
return common.Hash{}, 0, errors.New("invalid hex in storage key")
}
return common.BytesToHash(b), len(b), nil
}
@@ -589,9 +590,9 @@ func (api *BlockChainAPI) GetStorageAt(ctx context.Context, address common.Addre
if state == nil || err != nil {
return nil, err
}
key, _, err := decodeHash(hexKey)
key, _, err := decodeStorageKey(hexKey)
if err != nil {
return nil, fmt.Errorf("unable to decode storage key: %s", err)
return nil, &invalidParamsError{fmt.Sprintf("%v: %q", err, hexKey)}
}
res := state.GetState(address, key)
return res[:], state.Error()
@@ -1606,7 +1607,7 @@ func (api *TransactionAPI) SendTransaction(ctx context.Context, args Transaction
return common.Hash{}, err
}
// Assemble the transaction and sign with the wallet
tx := args.ToTransaction(types.LegacyTxType)
tx := args.ToTransaction(types.DynamicFeeTxType)
signed, err := wallet.SignTx(account, tx, api.b.ChainConfig().ChainID)
if err != nil {
@@ -1628,7 +1629,7 @@ func (api *TransactionAPI) FillTransaction(ctx context.Context, args Transaction
return nil, err
}
// Assemble the transaction and obtain rlp
tx := args.ToTransaction(types.LegacyTxType)
tx := args.ToTransaction(types.DynamicFeeTxType)
data, err := tx.MarshalBinary()
if err != nil {
return nil, err
@@ -1824,7 +1825,7 @@ func (api *TransactionAPI) SignTransaction(ctx context.Context, args Transaction
return nil, err
}
// Before actually sign the transaction, ensure the transaction fee is reasonable.
tx := args.ToTransaction(types.LegacyTxType)
tx := args.ToTransaction(types.DynamicFeeTxType)
if err := checkTxFee(tx.GasPrice(), tx.Gas(), api.b.RPCTxFeeCap()); err != nil {
return nil, err
}
@@ -1880,7 +1881,7 @@ func (api *TransactionAPI) Resend(ctx context.Context, sendArgs TransactionArgs,
if err := sendArgs.setDefaults(ctx, api.b, sidecarConfig{}); err != nil {
return common.Hash{}, err
}
matchTx := sendArgs.ToTransaction(types.LegacyTxType)
matchTx := sendArgs.ToTransaction(types.DynamicFeeTxType)
// Before replacing the old transaction, ensure the _new_ transaction fee is reasonable.
price := matchTx.GasPrice()
@@ -1910,7 +1911,7 @@ func (api *TransactionAPI) Resend(ctx context.Context, sendArgs TransactionArgs,
if gasLimit != nil && *gasLimit != 0 {
sendArgs.Gas = gasLimit
}
signedTx, err := api.sign(sendArgs.from(), sendArgs.ToTransaction(types.LegacyTxType))
signedTx, err := api.sign(sendArgs.from(), sendArgs.ToTransaction(types.DynamicFeeTxType))
if err != nil {
return common.Hash{}, err
}
+1 -1
View File
@@ -233,7 +233,7 @@ func (sim *simulator) processBlock(ctx context.Context, block *simBlock, header,
if block.BlockOverrides.BlobBaseFee != nil {
blockContext.BlobBaseFee = block.BlockOverrides.BlobBaseFee.ToInt()
}
precompiles := sim.activePrecompiles(sim.base)
precompiles := sim.activePrecompiles(header)
// State overrides are applied prior to execution of a block
if err := block.StateOverrides.Apply(sim.state, precompiles); err != nil {
return nil, nil, nil, err
+1 -1
View File
@@ -48,7 +48,7 @@ type Config struct {
GasCeil uint64 // Target gas ceiling for mined blocks.
GasPrice *big.Int // Minimum gas price for mining a transaction
Recommit time.Duration // The time interval for miner to re-create mining work.
MaxBlobsPerBlock *int // Maximum number of blobs per block (unset uses protocol default)
MaxBlobsPerBlock int // Maximum number of blobs per block (0 for unset uses protocol default)
}
// DefaultConfig contains default settings for miner.
+2 -2
View File
@@ -47,8 +47,8 @@ var (
// Users can specify the maximum number of blobs per block if necessary.
func (miner *Miner) maxBlobsPerBlock(time uint64) int {
maxBlobs := eip4844.MaxBlobsPerBlock(miner.chainConfig, time)
if miner.config.MaxBlobsPerBlock != nil {
maxBlobs = *miner.config.MaxBlobsPerBlock
if miner.config.MaxBlobsPerBlock != 0 {
maxBlobs = miner.config.MaxBlobsPerBlock
}
return maxBlobs
}
+5 -3
View File
@@ -69,9 +69,11 @@ var DefaultConfig = Config{
BatchResponseMaxSize: 25 * 1000 * 1000,
GraphQLVirtualHosts: []string{"localhost"},
P2P: p2p.Config{
ListenAddr: ":30303",
MaxPeers: 50,
NAT: nat.Any(),
ListenAddr: ":30303",
MaxPeers: 50,
NAT: nat.Any(),
DiscoveryV4: true,
DiscoveryV5: true,
},
DBEngine: "", // Use whatever exists, will default to Pebble if non-existent and supported
}
+8 -8
View File
@@ -107,30 +107,30 @@ func (n *upnp) addAnyPortMapping(protocol string, extport, intport int, ip net.I
})
}
// For IGDv1 and v1 services we should first try to add with extport.
var lastErr error
for i := 0; i < retryCount+1; i++ {
err := n.withRateLimit(func() error {
lastErr = n.withRateLimit(func() error {
return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS)
})
if err == nil {
if lastErr == nil {
return uint16(extport), nil
}
log.Debug("Failed to add port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", err)
log.Debug("Failed to add port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", lastErr)
}
// If above fails, we retry with a random port.
// We retry several times because of possible port conflicts.
var err error
for i := 0; i < randomCount; i++ {
extport = n.randomPort()
err := n.withRateLimit(func() error {
lastErr = n.withRateLimit(func() error {
return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS)
})
if err == nil {
if lastErr == nil {
return uint16(extport), nil
}
log.Debug("Failed to add random port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", err)
log.Debug("Failed to add random port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", lastErr)
}
return 0, err
return 0, lastErr
}
func (n *upnp) randomPort() int {
+2 -2
View File
@@ -24,6 +24,7 @@ import (
"crypto/ecdsa"
"crypto/hmac"
"crypto/rand"
"crypto/subtle"
"encoding/binary"
"errors"
"fmt"
@@ -33,7 +34,6 @@ import (
"net"
"time"
"github.com/luxfi/geth/common/bitutil"
"github.com/luxfi/geth/crypto"
"github.com/luxfi/geth/crypto/ecies"
"github.com/luxfi/geth/rlp"
@@ -677,6 +677,6 @@ func exportPubkey(pub *ecies.PublicKey) []byte {
func xor(one, other []byte) (xor []byte) {
xor = make([]byte, len(one))
bitutil.XORBytes(xor, one, other)
subtle.XORBytes(xor, one, other)
return xor
}
+2 -1
View File
@@ -185,9 +185,10 @@ func (t *Tracker) Fulfil(peer string, version uint, code uint64, id uint64) {
return
}
// Everything matches, mark the request serviced and meter it
wasHead := req.expire.Prev() == nil
t.expire.Remove(req.expire)
delete(t.pending, id)
if req.expire.Prev() == nil {
if wasHead {
if t.wake.Stop() {
t.schedule()
}
+7 -1
View File
@@ -116,6 +116,7 @@ func (t *BlockTest) Run(snapshotter bool, scheme string, witness bool, tracer *t
if !ok {
return UnsupportedForkError{t.json.Network}
}
// import pre accounts & construct test genesis block & state root
// Commit genesis state
var (
@@ -203,6 +204,11 @@ func (t *BlockTest) Run(snapshotter bool, scheme string, witness bool, tracer *t
return t.validateImportedHeaders(chain, validBlocks)
}
// Network returns the network/fork name for this test.
func (t *BlockTest) Network() string {
return t.json.Network
}
func (t *BlockTest) genesis(config *params.ChainConfig) *core.Genesis {
return &core.Genesis{
Config: config,
@@ -260,7 +266,7 @@ func (t *BlockTest) insertBlocks(blockchain *core.BlockChain) ([]btBlock, error)
}
if b.BlockHeader == nil {
if data, err := json.MarshalIndent(cb.Header(), "", " "); err == nil {
fmt.Fprintf(os.Stderr, "block (index %d) insertion should have failed due to: %v:\n%v\n",
fmt.Fprintf(os.Stdout, "block (index %d) insertion should have failed due to: %v:\n%v\n",
bi, b.ExpectException, string(data))
}
return nil, fmt.Errorf("block (index %d) insertion should have failed due to: %v",
+1 -1
View File
@@ -78,7 +78,7 @@ func fuzz(input []byte) int {
rand := rand.New(rand.NewSource(0x3a29)) // Same used in package tests!!!
f := fetcher.NewTxFetcherForTests(
func(common.Hash) bool { return false },
func(common.Hash, byte) error { return nil },
func(txs []*types.Transaction) []error {
return make([]error, len(txs))
},
+1 -1
View File
@@ -275,7 +275,7 @@ func (dl *diskLayer) storage(accountHash, storageHash common.Hash, depth int) ([
// If the layer is being generated, ensure the requested storage slot
// has already been covered by the generator.
key := append(accountHash[:], storageHash[:]...)
key := storageKeySlice(accountHash, storageHash)
marker := dl.genMarker()
if marker != nil && bytes.Compare(key, marker) > 0 {
return nil, errNotCoveredYet
+3 -2
View File
@@ -116,15 +116,16 @@ func writeStates(batch ethdb.Batch, genMarker []byte, accountData map[common.Has
continue
}
slots += 1
key := storageKeySlice(addrHash, storageHash)
if len(blob) == 0 {
rawdb.DeleteStorageSnapshot(batch, addrHash, storageHash)
if clean != nil {
clean.Set(append(addrHash[:], storageHash[:]...), nil)
clean.Set(key, nil)
}
} else {
rawdb.WriteStorageSnapshot(batch, addrHash, storageHash, blob)
if clean != nil {
clean.Set(append(addrHash[:], storageHash[:]...), blob)
clean.Set(key, blob)
}
}
}
+3 -3
View File
@@ -23,9 +23,9 @@ import (
"sort"
"testing"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/luxfi/geth/common"
"github.com/luxfi/geth/core/rawdb"
"github.com/luxfi/geth/ethdb"
)
func makeTestIndexBlock(count int) ([]byte, []uint64) {
+1 -1
View File
@@ -55,7 +55,7 @@ func sanitizeRange(start, end uint64, freezer ethdb.AncientReader) (uint64, uint
last = end
}
// Make sure the range is valid
if first >= last {
if first > last {
return 0, 0, fmt.Errorf("range is invalid, first: %d, last: %d", first, last)
}
return first, last, nil
+7
View File
@@ -33,6 +33,13 @@ func storageKey(accountHash common.Hash, slotHash common.Hash) [64]byte {
return key
}
// storageKeySlice returns a key for uniquely identifying the storage slot in
// the slice format.
func storageKeySlice(accountHash common.Hash, slotHash common.Hash) []byte {
key := storageKey(accountHash, slotHash)
return key[:]
}
// lookup is an internal structure used to efficiently determine the layer in
// which a state entry resides.
type lookup struct {