cerc-io/plugeth
12
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

Merge commit '73b01f40c' geth v1.11.2 into merge/geth-v1.11.2

Browse Source
This commit is contained in:
philip-morlier
2023-02-22 08:13:38 -08:00
parent 78e00bf68d 73b01f40ce
commit 4415e7c7b8
63 changed files with 539 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
eth/catalyst/api.go
+5 -2
View File
@@ -448,7 +448,7 @@ func (api *ConsensusAPI) newPayload(params engine.ExecutableData) (engine.Payloa
block, err := engine.ExecutableDataToBlock(params)
if err != nil {
log.Debug("Invalid NewPayload params", "params", params, "error", err)
return engine.PayloadStatusV1{Status: engine.INVALIDBLOCKHASH}, nil
return engine.PayloadStatusV1{Status: engine.INVALID}, nil
}
// Stash away the last update to warn the user if the beacon client goes offline
api.lastNewPayloadLock.Lock()
@@ -785,9 +785,12 @@ func (api *ConsensusAPI) GetPayloadBodiesByHashV1(hashes []common.Hash) []*engin
// GetPayloadBodiesByRangeV1 implements engine_getPayloadBodiesByRangeV1 which allows for retrieval of a range
// of block bodies by the engine api.
func (api *ConsensusAPI) GetPayloadBodiesByRangeV1(start, count hexutil.Uint64) ([]*engine.ExecutionPayloadBodyV1, error) {
if start == 0 || count == 0 || count > 1024 {
if start == 0 || count == 0 {
return nil, engine.InvalidParams.With(fmt.Errorf("invalid start or count, start: %v count: %v", start, count))
}
if count > 1024 {
return nil, engine.TooLargeRequest.With(fmt.Errorf("requested count too large: %v", count))
}
// limit count up until current
current := api.eth.BlockChain().CurrentBlock().NumberU64()
last := uint64(start) + uint64(count) - 1
eth/catalyst/api_test.go
+13 -7
View File
@@ -864,8 +864,8 @@ func TestInvalidBloom(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if status.Status != engine.INVALIDBLOCKHASH {
t.Errorf("invalid status: expected VALID got: %v", status.Status)
if status.Status != engine.INVALID {
t.Errorf("invalid status: expected INVALID got: %v", status.Status)
}
}
@@ -1383,37 +1383,43 @@ func TestGetBlockBodiesByRangeInvalidParams(t *testing.T) {
node, eth, _ := setupBodies(t)
api := NewConsensusAPI(eth)
defer node.Close()
tests := []struct {
start hexutil.Uint64
count hexutil.Uint64
want *engine.EngineAPIError
}{
// Genesis
{
start: 0,
count: 1,
want: engine.InvalidParams,
},
// No block requested
{
start: 1,
count: 0,
want: engine.InvalidParams,
},
// Genesis & no block
{
start: 0,
count: 0,
want: engine.InvalidParams,
},
// More than 1024 blocks
{
start: 1,
count: 1025,
want: engine.TooLargeRequest,
},
}
for _, test := range tests {
result, err := api.GetPayloadBodiesByRangeV1(test.start, test.count)
for i, tc := range tests {
result, err := api.GetPayloadBodiesByRangeV1(tc.start, tc.count)
if err == nil {
t.Fatalf("expected error, got %v", result)
t.Fatalf("test %d: expected error, got %v", i, result)
}
if have, want := err.Error(), tc.want.Error(); have != want {
t.Fatalf("test %d: have %s, want %s", i, have, want)
}
}
}
eth/downloader/downloader.go
+64 -1
View File
@@ -154,6 +154,11 @@ type Downloader struct {
bodyFetchHook func([]*types.Header) // Method to call upon starting a block body fetch
receiptFetchHook func([]*types.Header) // Method to call upon starting a receipt fetch
chainInsertHook func([]*fetchResult) // Method to call upon inserting a chain of blocks (possibly in multiple invocations)
// Progress reporting metrics
syncStartBlock uint64 // Head snap block when Geth was started
syncStartTime time.Time // Time instance when chain sync started
syncLogTime time.Time // Time instance when status was last reported
}
// LightChain encapsulates functions required to synchronise a light chain.
@@ -231,7 +236,9 @@ func New(checkpoint uint64, stateDb ethdb.Database, mux *event.TypeMux, chain Bl
quitCh: make(chan struct{}),
SnapSyncer: snap.NewSyncer(stateDb, chain.TrieDB().Scheme()),
stateSyncStart: make(chan *stateSync),
syncStartBlock: chain.CurrentFastBlock().NumberU64(),
}
// Create the post-merge skeleton syncer and start the process
dl.skeleton = newSkeleton(stateDb, dl.peers, dropPeer, newBeaconBackfiller(dl, success))
go dl.stateFetcher()
@@ -1614,6 +1621,7 @@ func (d *Downloader) processSnapSyncContent() error {
if len(results) == 0 {
// If pivot sync is done, stop
if oldPivot == nil {
d.reportSnapSyncProgress(true)
return sync.Cancel()
}
// If sync failed, stop
@@ -1627,6 +1635,8 @@ func (d *Downloader) processSnapSyncContent() error {
if d.chainInsertHook != nil {
d.chainInsertHook(results)
}
d.reportSnapSyncProgress(false)
// If we haven't downloaded the pivot block yet, check pivot staleness
// notifications from the header downloader
d.pivotLock.RLock()
@@ -1739,7 +1749,7 @@ func (d *Downloader) commitSnapSyncData(results []*fetchResult, stateSync *state
}
default:
}
// Retrieve the a batch of results to import
// Retrieve the batch of results to import
first, last := results[0].Header, results[len(results)-1].Header
log.Debug("Inserting snap-sync blocks", "items", len(results),
"firstnum", first.Number, "firsthash", first.Hash(),
@@ -1820,3 +1830,56 @@ func (d *Downloader) readHeaderRange(last *types.Header, count int) []*types.Hea
}
return headers
}
// reportSnapSyncProgress calculates various status reports and provides it to the user.
func (d *Downloader) reportSnapSyncProgress(force bool) {
// Initialize the sync start time if it's the first time we're reporting
if d.syncStartTime.IsZero() {
d.syncStartTime = time.Now().Add(-time.Millisecond) // -1ms offset to avoid division by zero
}
// Don't report all the events, just occasionally
if !force && time.Since(d.syncLogTime) < 8*time.Second {
return
}
// Don't report anything until we have a meaningful progress
var (
headerBytes, _ = d.stateDB.AncientSize(rawdb.ChainFreezerHeaderTable)
bodyBytes, _ = d.stateDB.AncientSize(rawdb.ChainFreezerBodiesTable)
receiptBytes, _ = d.stateDB.AncientSize(rawdb.ChainFreezerReceiptTable)
)
syncedBytes := common.StorageSize(headerBytes + bodyBytes + receiptBytes)
if syncedBytes == 0 {
return
}
var (
header = d.blockchain.CurrentHeader()
block = d.blockchain.CurrentFastBlock()
)
syncedBlocks := block.NumberU64() - d.syncStartBlock
if syncedBlocks == 0 {
return
}
// Retrieve the current chain head and calculate the ETA
latest, _, err := d.skeleton.Bounds()
if err != nil {
// We're going to cheat for non-merged networks, but that's fine
latest = d.pivotHeader
}
if latest == nil {
// This should really never happen, but add some defensive code for now.
// TODO(karalabe): Remove it eventually if we don't see it blow.
log.Error("Nil latest block in sync progress report")
return
}
var (
left = latest.Number.Uint64() - block.NumberU64()
eta = time.Since(d.syncStartTime) / time.Duration(syncedBlocks) * time.Duration(left)
progress = fmt.Sprintf("%.2f%%", float64(block.NumberU64())*100/float64(latest.Number.Uint64()))
headers = fmt.Sprintf("%v@%v", log.FormatLogfmtUint64(header.Number.Uint64()), common.StorageSize(headerBytes).TerminalString())
bodies = fmt.Sprintf("%v@%v", log.FormatLogfmtUint64(block.NumberU64()), common.StorageSize(bodyBytes).TerminalString())
receipts = fmt.Sprintf("%v@%v", log.FormatLogfmtUint64(block.NumberU64()), common.StorageSize(receiptBytes).TerminalString())
)
log.Info("Syncing: chain download in progress", "synced", progress, "chain", syncedBytes, "headers", headers, "bodies", bodies, "receipts", receipts, "eta", common.PrettyDuration(eta))
d.syncLogTime = time.Now()
}
eth/downloader/queue.go
+5 -4
View File
@@ -144,7 +144,7 @@ type queue struct {
active *sync.Cond
closed bool
lastStatLog time.Time
logTime time.Time // Time instance when status was last reported
}
// newQueue creates a new download queue for scheduling block retrieval.
@@ -390,11 +390,12 @@ func (q *queue) Results(block bool) []*fetchResult {
}
}
// Log some info at certain times
if time.Since(q.lastStatLog) > 60*time.Second {
q.lastStatLog = time.Now()
if time.Since(q.logTime) >= 60*time.Second {
q.logTime = time.Now()
info := q.Stats()
info = append(info, "throttle", throttleThreshold)
log.Info("Downloader queue stats", info...)
log.Debug("Downloader queue stats", info...)
}
return results
}
eth/fetcher/block_fetcher.go
+1 -1
View File
@@ -599,7 +599,7 @@ func (f *BlockFetcher) loop() {
announce.time = task.time
// If the block is empty (header only), short circuit into the final import queue
if header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash {
if header.TxHash == types.EmptyTxsHash && header.UncleHash == types.EmptyUncleHash {
log.Trace("Block empty, skipping body retrieval", "peer", announce.origin, "number", header.Number, "hash", header.Hash())
block := types.NewBlockWithHeader(header)
eth/filters/api.go
+21 -5
View File
@@ -39,6 +39,7 @@ type filter struct {
typ Type
deadline *time.Timer // filter is inactive when deadline triggers
hashes []common.Hash
fullTx bool
txs []*types.Transaction
crit FilterCriteria
logs []*types.Log
@@ -103,14 +104,14 @@ func (api *FilterAPI) timeoutLoop(timeout time.Duration) {
//
// It is part of the filter package because this filter can be used through the
// `eth_getFilterChanges` polling method that is also used for log filters.
func (api *FilterAPI) NewPendingTransactionFilter() rpc.ID {
func (api *FilterAPI) NewPendingTransactionFilter(fullTx *bool) rpc.ID {
var (
pendingTxs = make(chan []*types.Transaction)
pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
)
api.filtersMu.Lock()
api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, deadline: time.NewTimer(api.timeout), txs: make([]*types.Transaction, 0), s: pendingTxSub}
api.filters[pendingTxSub.ID] = &filter{typ: PendingTransactionsSubscription, fullTx: fullTx != nil && *fullTx, deadline: time.NewTimer(api.timeout), txs: make([]*types.Transaction, 0), s: pendingTxSub}
api.filtersMu.Unlock()
go func() {
@@ -412,6 +413,9 @@ func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
api.filtersMu.Lock()
defer api.filtersMu.Unlock()
chainConfig := api.sys.backend.ChainConfig()
latest := api.sys.backend.CurrentHeader()
if f, found := api.filters[id]; found {
if !f.deadline.Stop() {
// timer expired but filter is not yet removed in timeout loop
@@ -426,9 +430,21 @@ func (api *FilterAPI) GetFilterChanges(id rpc.ID) (interface{}, error) {
f.hashes = nil
return returnHashes(hashes), nil
case PendingTransactionsSubscription:
txs := f.txs
f.txs = nil
return txs, nil
if f.fullTx {
txs := make([]*ethapi.RPCTransaction, 0, len(f.txs))
for _, tx := range f.txs {
txs = append(txs, ethapi.NewRPCPendingTransaction(tx, latest, chainConfig))
}
f.txs = nil
return txs, nil
} else {
hashes := make([]common.Hash, 0, len(f.txs))
for _, tx := range f.txs {
hashes = append(hashes, tx.Hash())
}
f.txs = nil
return hashes, nil
}
case LogsSubscription, MinedAndPendingLogsSubscription:
logs := f.logs
f.logs = nil
eth/filters/filter_system_test.go
+69 -10
View File
@@ -37,6 +37,7 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/internal/ethapi"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
)
@@ -52,11 +53,12 @@ type testBackend struct {
}
func (b *testBackend) ChainConfig() *params.ChainConfig {
panic("implement me")
return params.TestChainConfig
}
func (b *testBackend) CurrentHeader() *types.Header {
panic("implement me")
hdr, _ := b.HeaderByNumber(context.TODO(), rpc.LatestBlockNumber)
return hdr
}
func (b *testBackend) ChainDb() ethdb.Database {
@@ -256,10 +258,10 @@ func TestPendingTxFilter(t *testing.T) {
types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
}
txs []*types.Transaction
hashes []common.Hash
)
fid0 := api.NewPendingTransactionFilter()
fid0 := api.NewPendingTransactionFilter(nil)
time.Sleep(1 * time.Second)
backend.txFeed.Send(core.NewTxsEvent{Txs: transactions})
@@ -271,7 +273,64 @@ func TestPendingTxFilter(t *testing.T) {
t.Fatalf("Unable to retrieve logs: %v", err)
}
tx := results.([]*types.Transaction)
h := results.([]common.Hash)
hashes = append(hashes, h...)
if len(hashes) >= len(transactions) {
break
}
// check timeout
if time.Now().After(timeout) {
break
}
time.Sleep(100 * time.Millisecond)
}
if len(hashes) != len(transactions) {
t.Errorf("invalid number of transactions, want %d transactions(s), got %d", len(transactions), len(hashes))
return
}
for i := range hashes {
if hashes[i] != transactions[i].Hash() {
t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), hashes[i])
}
}
}
// TestPendingTxFilterFullTx tests whether pending tx filters retrieve all pending transactions that are posted to the event mux.
func TestPendingTxFilterFullTx(t *testing.T) {
t.Parallel()
var (
db = rawdb.NewMemoryDatabase()
backend, sys = newTestFilterSystem(t, db, Config{})
api = NewFilterAPI(sys, false)
transactions = []*types.Transaction{
types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
types.NewTransaction(1, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
types.NewTransaction(2, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
types.NewTransaction(3, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
types.NewTransaction(4, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), 0, new(big.Int), nil),
}
txs []*ethapi.RPCTransaction
)
fullTx := true
fid0 := api.NewPendingTransactionFilter(&fullTx)
time.Sleep(1 * time.Second)
backend.txFeed.Send(core.NewTxsEvent{Txs: transactions})
timeout := time.Now().Add(1 * time.Second)
for {
results, err := api.GetFilterChanges(fid0)
if err != nil {
t.Fatalf("Unable to retrieve logs: %v", err)
}
tx := results.([]*ethapi.RPCTransaction)
txs = append(txs, tx...)
if len(txs) >= len(transactions) {
break
@@ -289,8 +348,8 @@ func TestPendingTxFilter(t *testing.T) {
return
}
for i := range txs {
if txs[i].Hash() != transactions[i].Hash() {
t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), txs[i].Hash())
if txs[i].Hash != transactions[i].Hash() {
t.Errorf("hashes[%d] invalid, want %x, got %x", i, transactions[i].Hash(), txs[i].Hash)
}
}
}
@@ -854,15 +913,15 @@ func TestPendingTxFilterDeadlock(t *testing.T) {
// timeout either in 100ms or 200ms
fids := make([]rpc.ID, 20)
for i := 0; i < len(fids); i++ {
fid := api.NewPendingTransactionFilter()
fid := api.NewPendingTransactionFilter(nil)
fids[i] = fid
// Wait for at least one tx to arrive in filter
for {
txs, err := api.GetFilterChanges(fid)
hashes, err := api.GetFilterChanges(fid)
if err != nil {
t.Fatalf("Filter should exist: %v\n", err)
}
if len(txs.([]*types.Transaction)) > 0 {
if len(hashes.([]common.Hash)) > 0 {
break
}
runtime.Gosched()
eth/protocols/snap/handler.go
+2 -1
View File
@@ -23,6 +23,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
@@ -464,7 +465,7 @@ func ServiceGetByteCodesQuery(chain *core.BlockChain, req *GetByteCodesPacket) [
bytes uint64
)
for _, hash := range req.Hashes {
if hash == emptyCode {
if hash == types.EmptyCodeHash {
// Peers should not request the empty code, but if they do, at
// least sent them back a correct response without db lookups
codes = append(codes, []byte{})
eth/protocols/snap/sync.go
+2 -10
View File
@@ -46,14 +46,6 @@ import (
"golang.org/x/crypto/sha3"
)
var (
// emptyRoot is the known root hash of an empty trie.
emptyRoot = common.HexToHash("56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421")
// emptyCode is the known hash of the empty EVM bytecode.
emptyCode = crypto.Keccak256Hash(nil)
)
const (
// minRequestSize is the minimum number of bytes to request from a remote peer.
// This number is used as the low cap for account and storage range requests.
@@ -1833,7 +1825,7 @@ func (s *Syncer) processAccountResponse(res *accountResponse) {
res.task.pend = 0
for i, account := range res.accounts {
// Check if the account is a contract with an unknown code
if !bytes.Equal(account.CodeHash, emptyCode[:]) {
if !bytes.Equal(account.CodeHash, types.EmptyCodeHash.Bytes()) {
if !rawdb.HasCodeWithPrefix(s.db, common.BytesToHash(account.CodeHash)) {
res.task.codeTasks[common.BytesToHash(account.CodeHash)] = struct{}{}
res.task.needCode[i] = true
@@ -1841,7 +1833,7 @@ func (s *Syncer) processAccountResponse(res *accountResponse) {
}
}
// Check if the account is a contract with an unknown storage trie
if account.Root != emptyRoot {
if account.Root != types.EmptyRootHash {
if !rawdb.HasTrieNode(s.db, res.hashes[i], nil, account.Root, s.scheme) {
// If there was a previous large state retrieval in progress,
// don't restart it from scratch. This happens if a sync cycle
eth/protocols/snap/sync_test.go
+7 -7
View File
@@ -1354,7 +1354,7 @@ func getCodeHash(i uint64) []byte {
// getCodeByHash convenience function to lookup the code from the code hash
func getCodeByHash(hash common.Hash) []byte {
if hash == emptyCode {
if hash == types.EmptyCodeHash {
return nil
}
for i, h := range codehashes {
@@ -1376,7 +1376,7 @@ func makeAccountTrieNoStorage(n int) (string, *trie.Trie, entrySlice) {
value, _ := rlp.EncodeToBytes(&types.StateAccount{
Nonce: i,
Balance: big.NewInt(int64(i)),
Root: emptyRoot,
Root: types.EmptyRootHash,
CodeHash: getCodeHash(i),
})
key := key32(i)
@@ -1427,7 +1427,7 @@ func makeBoundaryAccountTrie(n int) (string, *trie.Trie, entrySlice) {
value, _ := rlp.EncodeToBytes(&types.StateAccount{
Nonce: uint64(0),
Balance: big.NewInt(int64(i)),
Root: emptyRoot,
Root: types.EmptyRootHash,
CodeHash: getCodeHash(uint64(i)),
})
elem := &kv{boundaries[i].Bytes(), value}
@@ -1439,7 +1439,7 @@ func makeBoundaryAccountTrie(n int) (string, *trie.Trie, entrySlice) {
value, _ := rlp.EncodeToBytes(&types.StateAccount{
Nonce: i,
Balance: big.NewInt(int64(i)),
Root: emptyRoot,
Root: types.EmptyRootHash,
CodeHash: getCodeHash(i),
})
elem := &kv{key32(i), value}
@@ -1472,7 +1472,7 @@ func makeAccountTrieWithStorageWithUniqueStorage(accounts, slots int, code bool)
// Create n accounts in the trie
for i := uint64(1); i <= uint64(accounts); i++ {
key := key32(i)
codehash := emptyCode[:]
codehash := types.EmptyCodeHash.Bytes()
if code {
codehash = getCodeHash(i)
}
@@ -1527,7 +1527,7 @@ func makeAccountTrieWithStorage(accounts, slots int, code, boundary bool) (strin
// Create n accounts in the trie
for i := uint64(1); i <= uint64(accounts); i++ {
key := key32(i)
codehash := emptyCode[:]
codehash := types.EmptyCodeHash.Bytes()
if code {
codehash = getCodeHash(i)
}
@@ -1678,7 +1678,7 @@ func verifyTrie(db ethdb.KeyValueStore, root common.Hash, t *testing.T) {
log.Crit("Invalid account encountered during snapshot creation", "err", err)
}
accounts++
if acc.Root != emptyRoot {
if acc.Root != types.EmptyRootHash {
id := trie.StorageTrieID(root, common.BytesToHash(accIt.Key), acc.Root)
storeTrie, err := trie.NewStateTrie(id, triedb)
if err != nil {