forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 14:46:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix tests

Browse source
This commit is contained in:
Arpit Temani 2023-09-22 15:03:24 +05:30
parent 4e3bcbc90d
commit 11eb53be2b
14 changed files with 31 additions and 289 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
core/rawdb/accessors_chain.go
View file

@ -37,17 +37,14 @@ import (
// ReadCanonicalHash retrieves the hash assigned to a canonical block number.
func ReadCanonicalHash(db ethdb.Reader, number uint64) common.Hash {
var data []byte
_ = db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
data, _ = reader.Ancient(ChainFreezerHashTable, number)
if len(data) == 0 {
// Get it by hash from leveldb
data, _ = db.Get(headerHashKey(number))
}
return nil
})
return common.BytesToHash(data)
}

81
core/rawdb/freezer.go
View file

@ -26,12 +26,11 @@ import (
"sync/atomic"
"time"
"github.com/gofrs/flock"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/gofrs/flock"
)
var (
@ -102,7 +101,6 @@ func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize ui
return nil, errSymlinkDatadir
}
}
flockFile := filepath.Join(datadir, "FLOCK")
if err := os.MkdirAll(filepath.Dir(flockFile), 0755); err != nil {
return nil, err
@ -127,30 +125,13 @@ func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize ui
table, err := newTable(datadir, name, readMeter, writeMeter, sizeGauge, maxTableSize, disableSnappy, readonly)
if err != nil {
for _, table := range freezer.tables {
_ = table.Close()
table.Close()
}
_ = lock.Unlock()
lock.Unlock()
return nil, err
}
freezer.tables[name] = table
}
// Adjust table length for bor-receipt freezer for already synced nodes.
//
// Since, table only supports sequential data, this will fill empty-data upto current
// synced block (till current total header number).
//
// This way they don't have to sync again from block 0 and still be compatible
// for block logs for future blocks. Note that already synced nodes
// won't have past block logs. Newly synced node will have all the data.
if _, ok := freezer.tables[freezerBorReceiptTable]; ok {
if err := freezer.tables[freezerBorReceiptTable].Fill(freezer.tables[ChainFreezerHeaderTable].items.Load()); err != nil {
return nil, err
}
}
var err error
if freezer.readonly {
// In readonly mode only validate, don't truncate.
@ -160,13 +141,11 @@ func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize ui
// Truncate all tables to common length.
err = freezer.repair()
}
if err != nil {
for _, table := range freezer.tables {
_ = table.Close()
table.Close()
}
_ = lock.Unlock()
lock.Unlock()
return nil, err
}
@ -174,7 +153,6 @@ func NewFreezer(datadir string, namespace string, readonly bool, maxTableSize ui
freezer.writeBatch = newFreezerBatch(freezer)
log.Info("Opened ancient database", "database", datadir, "readonly", readonly)
return freezer, nil
}
@ -184,7 +162,6 @@ func (f *Freezer) Close() error {
defer f.writeLock.Unlock()
var errs []error
f.closeOnce.Do(func() {
for _, table := range f.tables {
if err := table.Close(); err != nil {
@ -195,11 +172,9 @@ func (f *Freezer) Close() error {
errs = append(errs, err)
}
})
if errs != nil {
return fmt.Errorf("%v", errs)
}
return nil
}
@ -209,7 +184,6 @@ func (f *Freezer) HasAncient(kind string, number uint64) (bool, error) {
if table := f.tables[kind]; table != nil {
return table.has(number), nil
}
return false, nil
}
@ -218,7 +192,6 @@ func (f *Freezer) Ancient(kind string, number uint64) ([]byte, error) {
if table := f.tables[kind]; table != nil {
return table.Retrieve(number)
}
return nil, errUnknownTable
}
@ -232,7 +205,6 @@ func (f *Freezer) AncientRange(kind string, start, count, maxBytes uint64) ([][]
if table := f.tables[kind]; table != nil {
return table.RetrieveItems(start, count, maxBytes)
}
return nil, errUnknownTable
}
@ -256,7 +228,6 @@ func (f *Freezer) AncientSize(kind string) (uint64, error) {
if table := f.tables[kind]; table != nil {
return table.size()
}
return 0, errUnknownTable
}
@ -274,13 +245,11 @@ func (f *Freezer) ModifyAncients(fn func(ethdb.AncientWriteOp) error) (writeSize
if f.readonly {
return 0, errReadOnly
}
f.writeLock.Lock()
defer f.writeLock.Unlock()
// Roll back all tables to the starting position in case of error.
prevItem := f.frozen.Load()
defer func() {
if err != nil {
// The write operation has failed. Go back to the previous item position.
@ -294,18 +263,14 @@ func (f *Freezer) ModifyAncients(fn func(ethdb.AncientWriteOp) error) (writeSize
}()
f.writeBatch.reset()
if err := fn(f.writeBatch); err != nil {
return 0, err
}
item, writeSize, err := f.writeBatch.commit()
if err != nil {
return 0, err
}
f.frozen.Store(item)
return writeSize, nil
}
@ -315,7 +280,6 @@ func (f *Freezer) TruncateHead(items uint64) (uint64, error) {
if f.readonly {
return 0, errReadOnly
}
f.writeLock.Lock()
defer f.writeLock.Unlock()
@ -323,13 +287,11 @@ func (f *Freezer) TruncateHead(items uint64) (uint64, error) {
if oitems <= items {
return oitems, nil
}
for _, table := range f.tables {
if err := table.truncateHead(items); err != nil {
return 0, err
}
}
f.frozen.Store(items)
return oitems, nil
}
@ -339,7 +301,6 @@ func (f *Freezer) TruncateTail(tail uint64) (uint64, error) {
if f.readonly {
return 0, errReadOnly
}
f.writeLock.Lock()
defer f.writeLock.Unlock()
@ -347,13 +308,11 @@ func (f *Freezer) TruncateTail(tail uint64) (uint64, error) {
if old >= tail {
return old, nil
}
for _, table := range f.tables {
if err := table.truncateTail(tail); err != nil {
return 0, err
}
}
f.tail.Store(tail)
return old, nil
}
@ -361,17 +320,14 @@ func (f *Freezer) TruncateTail(tail uint64) (uint64, error) {
// Sync flushes all data tables to disk.
func (f *Freezer) Sync() error {
var errs []error
for _, table := range f.tables {
if err := table.Sync(); err != nil {
errs = append(errs, err)
}
}
if errs != nil {
return fmt.Errorf("%v", errs)
}
return nil
}
@ -381,7 +337,6 @@ func (f *Freezer) validate() error {
if len(f.tables) == 0 {
return nil
}
var (
head uint64
tail uint64
@ -392,7 +347,6 @@ func (f *Freezer) validate() error {
head = table.items.Load()
tail = table.itemHidden.Load()
name = kind
break
}
// Now check every table against those boundaries.
@ -400,15 +354,12 @@ func (f *Freezer) validate() error {
if head != table.items.Load() {
return fmt.Errorf("freezer tables %s and %s have differing head: %d != %d", kind, name, table.items.Load(), head)
}
if tail != table.itemHidden.Load() {
return fmt.Errorf("freezer tables %s and %s have differing tail: %d != %d", kind, name, table.itemHidden.Load(), tail)
}
}
f.frozen.Store(head)
f.tail.Store(tail)
return nil
}
@ -418,32 +369,26 @@ func (f *Freezer) repair() error {
head = uint64(math.MaxUint64)
tail = uint64(0)
)
for _, table := range f.tables {
items := table.items.Load()
if head > items {
head = items
}
hidden := table.itemHidden.Load()
if hidden > tail {
tail = hidden
}
}
for _, table := range f.tables {
if err := table.truncateHead(head); err != nil {
return err
}
if err := table.truncateTail(tail); err != nil {
return err
}
}
f.frozen.Store(head)
f.tail.Store(tail)
return nil
}
@ -453,12 +398,10 @@ type convertLegacyFn = func([]byte) ([]byte, error)
// MigrateTable processes the entries in a given table in sequence
// converting them to a new format if they're of an old format.
// nolint:gocognit
func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
if f.readonly {
return errReadOnly
}
f.writeLock.Lock()
defer f.writeLock.Unlock()
@ -475,26 +418,21 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
batchSize = uint64(1024)
maxBytes = uint64(1024 * 1024)
)
for i := offset; i < items; {
if i+batchSize > items {
batchSize = items - i
}
data, err := t.RetrieveItems(i, batchSize, maxBytes)
if err != nil {
return err
}
for j, item := range data {
if err := fn(i+uint64(j), item); err != nil {
return err
}
}
i += uint64(len(data))
}
return nil
}
// TODO(s1na): This is a sanity-check since as of now no process does tail-deletion. But the migration
@ -502,17 +440,14 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
if table.itemOffset.Load() > 0 || table.itemHidden.Load() > 0 {
return errors.New("migration not supported for tail-deleted freezers")
}
ancientsPath := filepath.Dir(table.index.Name())
// Set up new dir for the migrated table, the content of which
// we'll at the end move over to the ancients dir.
migrationPath := filepath.Join(ancientsPath, "migration")
newTable, err := newFreezerTable(migrationPath, kind, table.noCompression, false)
if err != nil {
return err
}
var (
batch = newTable.newBatch()
out []byte
@ -520,7 +455,6 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
logged = time.Now()
offset = newTable.items.Load()
)
if offset > 0 {
log.Info("found previous migration attempt", "migrated", offset)
}
@ -534,7 +468,6 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
if err != nil {
return err
}
if err := batch.AppendRaw(i, out); err != nil {
return err
}
@ -542,11 +475,9 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
}); err != nil {
return err
}
if err := batch.commit(); err != nil {
return err
}
log.Info("Replacing old table files with migrated ones", "elapsed", common.PrettyDuration(time.Since(start)))
// Release and delete old table files. Note this won't
// delete the index file.
@ -555,7 +486,6 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
if err := newTable.Close(); err != nil {
return err
}
files, err := os.ReadDir(migrationPath)
if err != nil {
return err
@ -571,6 +501,5 @@ func (f *Freezer) MigrateTable(kind string, convert convertLegacyFn) error {
if err := os.Remove(migrationPath); err != nil {
return err
}
return nil
}

56
core/txpool/txpool.go
View file

@ -70,7 +70,8 @@ type TxPool struct {
reservations map[common.Address]SubPool // Map with the account to pool reservations
reserveLock sync.Mutex // Lock protecting the account reservations
quit chan chan error // Quit channel to tear down the head updater
subs event.SubscriptionScope // Subscription scope to unscubscribe all on shutdown
quit chan chan error // Quit channel to tear down the head updater
}
// New creates a new transaction pool to gather, sort and filter inbound
@ -312,7 +313,6 @@ func (p *TxPool) Pending(enforceTips bool) map[common.Address][]*LazyTransaction
txs[addr] = set
}
}
return txs
}
@ -323,25 +323,9 @@ func (p *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscrip
for i, subpool := range p.subpools {
subs[i] = subpool.SubscribeTransactions(ch)
}
return subs[0]
return p.subs.Track(event.JoinSubscriptions(subs...))
}
// // validateTxBasics checks whether a transaction is valid according to the consensus
// // rules, but does not check state-dependent validation such as sufficient balance.
// // This check is meant as an early check which only needs to be performed once,
// // and does not require the pool mutex to be held.
// // nolint:gocognit
// func (pool *TxPool) validateTxBasics(tx *types.Transaction, local bool) error {
// pool.currentStateMutex.Lock()
// defer pool.currentStateMutex.Unlock()
// // Accept only legacy transactions until EIP-2718/2930 activates.
// if !pool.eip2718.Load() && tx.Type() != types.LegacyTxType {
// return core.ErrTxTypeNotSupported
// }
// return p.subs.Track(event.JoinSubscriptions(subs...))
// }
// Nonce returns the next nonce of an account, with all transactions executable
// by the pool already applied on top.
func (p *TxPool) Nonce(addr common.Address) uint64 {
@ -370,40 +354,6 @@ func (p *TxPool) Stats() (int, int) {
return runnable, blocked
}
// // Stats retrieves the current pool stats, namely the number of pending and the
// // number of queued (non-executable) transactions.
// func (p *TxPool) Stats() (int, int) {
// var runnable, blocked int
// for _, subpool := range p.subpools {
// run, block := subpool.Stats()
// }
// runnable += run
// blocked += block
// // }
// return runnable, blocked
// }
// // validateTx checks whether a transaction is valid according to the consensus
// // rules and adheres to some heuristic limits of the local node (price and size).
// func (pool *TxPool) validateTx(tx *types.Transaction, _ bool) error {
// pool.currentStateMutex.Lock()
// defer pool.currentStateMutex.Unlock()
// // Signature has been checked already, this cannot error.
// from, _ := types.Sender(pool.signer, tx)
// // Ensure the transaction adheres to nonce ordering
// if pool.currentState.GetNonce(from) > tx.Nonce() {
// return core.ErrNonceTooLow
// }
// // Transactor should have enough funds to cover the costs
// // cost == V + GP * GL
// balance := pool.currentState.GetBalance(from)
// if balance.Cmp(tx.Cost()) < 0 {
// return core.ErrInsufficientFunds
// }
// }
// Content retrieves the data content of the transaction pool, returning all the
// pending as well as queued transactions, grouped by account and sorted by nonce.
func (p *TxPool) Content() (map[common.Address][]*types.Transaction, map[common.Address][]*types.Transaction) {

2
eth/api_debug.go
View file

@ -118,7 +118,7 @@ func (api *DebugAPI) GetBadBlocks(ctx context.Context) ([]*BadBlockArgs, error)
} else {
blockRlp = fmt.Sprintf("%#x", rlpBytes)
}
blockJSON = ethapi.RPCMarshalBlock(block, true, true, api.eth.APIBackend.ChainConfig(), api.eth.ChainDb())
blockJSON = ethapi.RPCMarshalBlock(block, true, true, api.eth.APIBackend.ChainConfig(), api.eth.chainDb)
results = append(results, &BadBlockArgs{
Hash: block.Hash(),
RLP: blockRlp,

8
eth/handler.go
View file

@ -374,13 +374,11 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
number = head.Number.Uint64()
td = h.chain.GetTd(hash, number)
)
forkID := forkid.NewID(h.chain.Config(), genesis.Hash(), number, head.Time)
if err := peer.Handshake(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
peer.Log().Debug("Ethereum handshake failed", "err", err)
return err
}
reject := false // reserved peer slots
if h.snapSync.Load() {
if snap == nil {
@ -398,7 +396,6 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
return p2p.DiscTooManyPeers
}
}
peer.Log().Debug("Ethereum peer connected", "name", peer.Name())
// Register the peer locally
@ -417,7 +414,6 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
peer.Log().Error("Failed to register peer in eth syncer", "err", err)
return err
}
if snap != nil {
if err := h.downloader.SnapSyncer.Register(snap); err != nil {
peer.Log().Error("Failed to register peer in snap syncer", "err", err)
@ -442,7 +438,6 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
if err != nil {
return err
}
go func(number uint64, hash common.Hash, req *eth.Request) {
// Ensure the request gets cancelled in case of error/drop
defer req.Close()
@ -464,14 +459,11 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
res.Done <- errors.New("too many headers in required block response")
return
}
if headers[0].Number.Uint64() != number || headers[0].Hash() != hash {
peer.Log().Info("Required block mismatch, dropping peer", "number", number, "hash", headers[0].Hash(), "want", hash)
res.Done <- errors.New("required block mismatch")
return
}
peer.Log().Debug("Peer required block verified", "number", number, "hash", hash)
res.Done <- nil
case <-timeout.C:

44
eth/handler_eth_test.go
View file

@ -323,21 +323,12 @@ func testRecvTransactions(t *testing.T, protocol uint) {
}
// This test checks that pending transactions are sent.
func TestSendTransactions66(t *testing.T) {
t.Parallel()
testSendTransactions(t, eth.ETH66)
}
func TestSendTransactions67(t *testing.T) {
t.Parallel()
testSendTransactions(t, eth.ETH67)
}
func TestSendTransactions68(t *testing.T) {
t.Parallel()
testSendTransactions(t, eth.ETH68)
}
func TestSendTransactions66(t *testing.T) { testSendTransactions(t, eth.ETH66) }
func TestSendTransactions67(t *testing.T) { testSendTransactions(t, eth.ETH67) }
func TestSendTransactions68(t *testing.T) { testSendTransactions(t, eth.ETH68) }
func testSendTransactions(t *testing.T, protocol uint) {
t.Helper()
t.Parallel()
// Create a message handler and fill the pool with big transactions
handler := newTestHandler()
@ -360,7 +351,6 @@ func testSendTransactions(t *testing.T, protocol uint) {
src := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{1}, "", nil, p2pSrc), p2pSrc, handler.txpool)
sink := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{2}, "", nil, p2pSink), p2pSink, handler.txpool)
defer src.Close()
defer sink.Close()
@ -373,7 +363,6 @@ func testSendTransactions(t *testing.T, protocol uint) {
head = handler.chain.CurrentBlock()
td = handler.chain.GetTd(head.Hash(), head.Number.Uint64())
)
if err := sink.Handshake(1, td, head.Hash(), genesis.Hash(), forkid.NewIDWithChain(handler.chain), forkid.NewFilter(handler.chain)); err != nil {
t.Fatalf("failed to run protocol handshake")
}
@ -382,12 +371,10 @@ func testSendTransactions(t *testing.T, protocol uint) {
backend := new(testEthHandler)
anns := make(chan []common.Hash)
annSub := backend.txAnnounces.Subscribe(anns)
defer annSub.Unsubscribe()
bcasts := make(chan []*types.Transaction)
bcastSub := backend.txBroadcasts.Subscribe(bcasts)
defer bcastSub.Unsubscribe()
@ -404,7 +391,6 @@ func testSendTransactions(t *testing.T, protocol uint) {
if _, ok := seen[hash]; ok {
t.Errorf("duplicate transaction announced: %x", hash)
}
seen[hash] = struct{}{}
}
case <-bcasts:
@ -415,7 +401,6 @@ func testSendTransactions(t *testing.T, protocol uint) {
panic("unsupported protocol, please extend test")
}
}
for _, tx := range insert {
if _, ok := seen[tx.Tx.Hash()]; !ok {
t.Errorf("missing transaction: %x", tx.Tx.Hash())
@ -425,21 +410,12 @@ func testSendTransactions(t *testing.T, protocol uint) {
// Tests that transactions get propagated to all attached peers, either via direct
// broadcasts or via announcements/retrievals.
func TestTransactionPropagation66(t *testing.T) {
t.Parallel()
testTransactionPropagation(t, eth.ETH66)
}
func TestTransactionPropagation67(t *testing.T) {
t.Parallel()
testTransactionPropagation(t, eth.ETH67)
}
func TestTransactionPropagation68(t *testing.T) {
t.Parallel()
testTransactionPropagation(t, eth.ETH68)
}
func TestTransactionPropagation66(t *testing.T) { testTransactionPropagation(t, eth.ETH66) }
func TestTransactionPropagation67(t *testing.T) { testTransactionPropagation(t, eth.ETH67) }
func TestTransactionPropagation68(t *testing.T) { testTransactionPropagation(t, eth.ETH68) }
func testTransactionPropagation(t *testing.T, protocol uint) {
t.Helper()
t.Parallel()
// 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
@ -457,7 +433,7 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
}
// Interconnect all the sink handlers with the source handler
for i, sink := range sinks {
sink := sink // Closure for gorotuine below
sink := sink // Closure for goroutine below
sourcePipe, sinkPipe := p2p.MsgPipe()
defer sourcePipe.Close()
@ -465,7 +441,6 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
sourcePeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{byte(i + 1)}, "", nil, sourcePipe), sourcePipe, source.txpool)
sinkPeer := eth.NewPeer(protocol, p2p.NewPeerPipe(enode.ID{0}, "", nil, sinkPipe), sinkPipe, sink.txpool)
defer sourcePeer.Close()
defer sinkPeer.Close()
@ -502,7 +477,6 @@ func testTransactionPropagation(t *testing.T, protocol uint) {
arrived += len(event.Txs)
case <-time.After(2 * time.Second):
t.Errorf("sink %d: transaction propagation timed out: have %d, want %d", i, arrived, len(txs))
timeout = true
}
}

40
eth/protocols/eth/broadcast.go
View file

@ -46,14 +46,12 @@ func (p *Peer) broadcastBlocks() {
if err := p.SendNewBlock(prop.block, prop.td); err != nil {
return
}
p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
case block := <-p.queuedBlockAnns:
if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {
return
}
p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
case <-p.term:
@ -72,7 +70,6 @@ func (p *Peer) broadcastTransactions() {
fail = make(chan error, 1) // Channel used to receive network error
failed bool // Flag whether a send failed, discard everything onward
)
for {
// If there's no in-flight broadcast running, check if a new one is needed
if done == nil && len(queue) > 0 {
@ -82,21 +79,14 @@ func (p *Peer) broadcastTransactions() {
txs []*types.Transaction
size common.StorageSize
)
for i := 0; i < len(queue) && size < maxTxPacketSize; i++ {
// TODO - Arpit
// tx := p.txpool.Get(queue[i])
// Skip EIP-4337 bundled transactions
// if tx != nil && tx.GetOptions() == nil {
// txs = append(txs, tx)
// size += common.StorageSize(tx.Size())
// }
// TODO - Arpit Pratik changes
if tx := p.txpool.Get(queue[i]); tx != nil {
txs = append(txs, tx.Tx)
size += common.StorageSize(tx.Tx.Size())
}
hashesCount++
}
queue = queue[:copy(queue, queue[hashesCount:])]
// If there's anything available to transfer, fire up an async writer
@ -107,7 +97,6 @@ func (p *Peer) broadcastTransactions() {
fail <- err
return
}
close(done)
p.Log().Trace("Sent transactions", "count", len(txs))
}()
@ -149,7 +138,6 @@ func (p *Peer) announceTransactions() {
fail = make(chan error, 1) // Channel used to receive network error
failed bool // Flag whether a send failed, discard everything onward
)
for {
// If there's no in-flight announce running, check if a new one is needed
if done == nil && len(queue) > 0 {
@ -161,18 +149,15 @@ func (p *Peer) announceTransactions() {
pendingSizes []uint32
size common.StorageSize
)
for count = 0; count < len(queue) && size < maxTxPacketSize; count++ {
// TODO - Arpit
// tx := p.txpool.Get(queue[count])
// // Skip EIP-4337 bundled transactions
// if tx != nil && tx.GetOptions() == nil {
// pending = append(pending, queue[count])
// pendingTypes = append(pendingTypes, tx.Tx.Type())
// pendingSizes = append(pendingSizes, uint32(tx.Tx.Size()))
// size += common.HashLength
// }
// TODO - Arpit Pratik changes
if tx := p.txpool.Get(queue[count]); tx != nil {
pending = append(pending, queue[count])
pendingTypes = append(pendingTypes, tx.Tx.Type())
pendingSizes = append(pendingSizes, uint32(tx.Tx.Size()))
size += common.HashLength
}
}
// Shift and trim queue
queue = queue[:copy(queue, queue[count:])]
@ -192,7 +177,6 @@ func (p *Peer) announceTransactions() {
return
}
}
close(done)
p.Log().Trace("Sent transaction announcements", "count", len(pending))
}()

7
eth/protocols/eth/handler.go
View file

@ -96,7 +96,6 @@ type TxPool interface {
// MakeProtocols constructs the P2P protocol definitions for `eth`.
func MakeProtocols(backend Backend, network uint64, dnsdisc enode.Iterator) []p2p.Protocol {
protocols := make([]p2p.Protocol, len(ProtocolVersions))
for i, version := range ProtocolVersions {
version := version // Closure
@ -122,7 +121,6 @@ func MakeProtocols(backend Backend, network uint64, dnsdisc enode.Iterator) []p2
DialCandidates: dnsdisc,
}
}
return protocols
}
@ -223,18 +221,15 @@ func handleMessage(backend Backend, peer *Peer) error {
if err != nil {
return err
}
if msg.Size > maxMessageSize {
return fmt.Errorf("%w: %v > %v", errMsgTooLarge, msg.Size, maxMessageSize)
}
defer msg.Discard()
var handlers = eth66
if peer.Version() == ETH67 {
handlers = eth67
}
if peer.Version() >= ETH68 {
handlers = eth68
}
@ -251,10 +246,8 @@ func handleMessage(backend Backend, peer *Peer) error {
metrics.GetOrRegisterHistogramLazy(h, nil, sampler).Update(time.Since(start).Microseconds())
}(time.Now())
}
if handler := handlers[msg.Code]; handler != nil {
return handler(backend, msg, peer)
}
return fmt.Errorf("%w: %v", errInvalidMsgCode, msg.Code)
}

21
eth/protocols/eth/peer.go
View file

@ -22,7 +22,6 @@ import (
"sync"
mapset "github.com/deckarep/golang-set/v2"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/p2p"
@ -62,7 +61,6 @@ func max(a, b int) int {
if a > b {
return a
}
return b
}
@ -146,7 +144,6 @@ func (p *Peer) Head() (hash common.Hash, td *big.Int) {
defer p.lock.RUnlock()
copy(hash[:], p.head[:])
return hash, new(big.Int).Set(p.td)
}
@ -197,7 +194,6 @@ func (p *Peer) SendTransactions(txs types.Transactions) error {
for _, tx := range txs {
p.knownTxs.Add(tx.Hash())
}
return p2p.Send(p.rw, TransactionsMsg, txs)
}
@ -275,7 +271,6 @@ func (p *Peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error
request[i].Hash = hashes[i]
request[i].Number = numbers[i]
}
return p2p.Send(p.rw, NewBlockHashesMsg, request)
}
@ -296,7 +291,6 @@ func (p *Peer) AsyncSendNewBlockHash(block *types.Block) {
func (p *Peer) SendNewBlock(block *types.Block, td *big.Int) error {
// Mark all the block hash as known, but ensure we don't overflow our limits
p.knownBlocks.Add(block.Hash())
return p2p.Send(p.rw, NewBlockMsg, &NewBlockPacket{
Block: block,
TD: td,
@ -352,7 +346,6 @@ func (p *Peer) ReplyReceiptsRLP(id uint64, receipts []rlp.RawValue) error {
// single header. It is used solely by the fetcher.
func (p *Peer) RequestOneHeader(hash common.Hash, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching single header", "hash", hash)
id := rand.Uint64()
req := &Request{
@ -373,7 +366,6 @@ func (p *Peer) RequestOneHeader(hash common.Hash, sink chan *Response) (*Request
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
@ -381,7 +373,6 @@ func (p *Peer) RequestOneHeader(hash common.Hash, sink chan *Response) (*Request
// specified header query, based on the hash of an origin block.
func (p *Peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse)
id := rand.Uint64()
req := &Request{
@ -402,7 +393,6 @@ func (p *Peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, re
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
@ -410,7 +400,6 @@ func (p *Peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, re
// specified header query, based on the number of an origin block.
func (p *Peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse)
id := rand.Uint64()
req := &Request{
@ -431,7 +420,6 @@ func (p *Peer) RequestHeadersByNumber(origin uint64, amount int, skip int, rever
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
@ -439,7 +427,6 @@ func (p *Peer) RequestHeadersByNumber(origin uint64, amount int, skip int, rever
// specified.
func (p *Peer) RequestBodies(hashes []common.Hash, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching batch of block bodies", "count", len(hashes))
id := rand.Uint64()
req := &Request{
@ -455,7 +442,6 @@ func (p *Peer) RequestBodies(hashes []common.Hash, sink chan *Response) (*Reques
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
@ -463,7 +449,6 @@ func (p *Peer) RequestBodies(hashes []common.Hash, sink chan *Response) (*Reques
// data, corresponding to the specified hashes.
func (p *Peer) RequestNodeData(hashes []common.Hash, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching batch of state data", "count", len(hashes))
id := rand.Uint64()
req := &Request{
@ -479,14 +464,12 @@ func (p *Peer) RequestNodeData(hashes []common.Hash, sink chan *Response) (*Requ
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
// RequestReceipts fetches a batch of transaction receipts from a remote node.
func (p *Peer) RequestReceipts(hashes []common.Hash, sink chan *Response) (*Request, error) {
p.Log().Debug("Fetching batch of receipts", "count", len(hashes))
id := rand.Uint64()
req := &Request{
@ -502,18 +485,15 @@ func (p *Peer) RequestReceipts(hashes []common.Hash, sink chan *Response) (*Requ
if err := p.dispatchRequest(req); err != nil {
return nil, err
}
return req, nil
}
// RequestTxs fetches a batch of transactions from a remote node.
func (p *Peer) RequestTxs(hashes []common.Hash) error {
p.Log().Debug("Fetching batch of transactions", "count", len(hashes))
id := rand.Uint64()
requestTracker.Track(p.id, p.version, GetPooledTransactionsMsg, PooledTransactionsMsg, id)
return p2p.Send(p.rw, GetPooledTransactionsMsg, &GetPooledTransactionsPacket66{
RequestId: id,
GetPooledTransactionsPacket: hashes,
@ -539,7 +519,6 @@ func (k *knownCache) Add(hashes ...common.Hash) {
for k.hashes.Cardinality() > max(0, k.max-len(hashes)) {
k.hashes.Pop()
}
for _, hash := range hashes {
k.hashes.Add(hash)
}

12
event/feed.go
View file

@ -72,12 +72,10 @@ func (f *Feed) init(etype reflect.Type) {
// Slow subscribers are not dropped.
func (f *Feed) Subscribe(channel interface{}) Subscription {
chanval := reflect.ValueOf(channel)
chantyp := chanval.Type()
if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 {
panic(errBadChannel)
}
sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}
f.once.Do(func() { f.init(chantyp.Elem()) })
@ -91,7 +89,6 @@ func (f *Feed) Subscribe(channel interface{}) Subscription {
// The next Send will add it to f.sendCases.
cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
f.inbox = append(f.inbox, cas)
return sub
}
@ -99,14 +96,11 @@ func (f *Feed) remove(sub *feedSub) {
// Delete from inbox first, which covers channels
// that have not been added to f.sendCases yet.
ch := sub.channel.Interface()
f.mu.Lock()
index := f.inbox.find(ch)
if index != -1 {
f.inbox = f.inbox.delete(index)
f.mu.Unlock()
return
}
f.mu.Unlock()
@ -148,7 +142,6 @@ func (f *Feed) Send(value interface{}) (nsent int) {
// of sendCases. When a send succeeds, the corresponding case moves to the end of
// 'cases' and it shrinks by one element.
cases := f.sendCases
for {
// Fast path: try sending without blocking before adding to the select set.
// This should usually succeed if subscribers are fast enough and have free
@ -160,7 +153,6 @@ func (f *Feed) Send(value interface{}) (nsent int) {
i--
}
}
if len(cases) == firstSubSendCase {
break
}
@ -169,7 +161,6 @@ func (f *Feed) Send(value interface{}) (nsent int) {
if chosen == 0 /* <-f.removeSub */ {
index := f.sendCases.find(recv.Interface())
f.sendCases = f.sendCases.delete(index)
if index >= 0 && index < len(cases) {
// Shrink 'cases' too because the removed case was still active.
cases = f.sendCases[:len(cases)-1]
@ -185,7 +176,6 @@ func (f *Feed) Send(value interface{}) (nsent int) {
f.sendCases[i].Send = reflect.Value{}
}
f.sendLock <- struct{}{}
return nsent
}
@ -216,7 +206,6 @@ func (cs caseList) find(channel interface{}) int {
return i
}
}
return -1
}
@ -229,7 +218,6 @@ func (cs caseList) delete(index int) caseList {
func (cs caseList) deactivate(index int) caseList {
last := len(cs) - 1
cs[index], cs[last] = cs[last], cs[index]
return cs[:last]
}

1
node/rpcstack.go
View file

@ -42,7 +42,6 @@ type httpConfig struct {
CorsAllowedOrigins []string
Vhosts []string
prefix string // path prefix on which to mount http handler
jwtSecret []byte // optional JWT secret
// Execution pool config
executionPoolSize uint64

33
p2p/peer.go
View file

@ -130,13 +130,11 @@ func NewPeer(id enode.ID, name string, caps []Cap) *Peer {
protos[i].Name = cap.Name
protos[i].Version = cap.Version
}
pipe, _ := net.Pipe()
node := enode.SignNull(new(enr.Record), id)
conn := &conn{fd: pipe, transport: nil, node: node, caps: caps, name: name}
peer := newPeer(log.Root(), conn, protos)
close(peer.closed) // ensures Disconnect doesn't block
return peer
}
@ -146,7 +144,6 @@ func NewPeer(id enode.ID, name string, caps []Cap) *Peer {
func NewPeerPipe(id enode.ID, name string, caps []Cap, pipe *MsgPipeRW) *Peer {
p := NewPeer(id, name, caps)
p.testPipe = pipe
return p
}
@ -166,7 +163,6 @@ func (p *Peer) Name() string {
if len(s) > 20 {
return s[:20] + "..."
}
return s
}
@ -192,7 +188,6 @@ func (p *Peer) RunningCap(protocol string, versions []uint) bool {
}
}
}
return false
}
@ -242,7 +237,6 @@ func newPeer(log log.Logger, conn *conn, protocols []Protocol) *Peer {
pingRecv: make(chan struct{}, 16),
log: log.New("id", conn.node.ID(), "conn", conn.flags),
}
return p
}
@ -257,9 +251,7 @@ func (p *Peer) run() (remoteRequested bool, err error) {
readErr = make(chan error, 1)
reason DiscReason // sent to the peer
)
p.wg.Add(2)
go p.readLoop(readErr)
go p.pingLoop()
@ -299,7 +291,6 @@ loop:
close(p.closed)
p.rw.close(reason)
p.wg.Wait()
return remoteRequested, err
}
@ -316,7 +307,6 @@ func (p *Peer) pingLoop() {
p.protoErr <- err
return
}
ping.Reset(pingInterval)
case <-p.pingRecv:
@ -330,14 +320,12 @@ func (p *Peer) pingLoop() {
func (p *Peer) readLoop(errc chan<- error) {
defer p.wg.Done()
for {
msg, err := p.rw.ReadMsg()
if err != nil {
errc <- err
return
}
msg.ReceivedAt = time.Now()
if err = p.handle(msg); err != nil {
errc <- err
@ -358,9 +346,7 @@ func (p *Peer) handle(msg Msg) error {
// This is the last message. We don't need to discard or
// check errors because, the connection will be closed after it.
var m struct{ R DiscReason }
rlp.Decode(msg.Payload, &m)
return m.R
case msg.Code < baseProtocolLength:
// ignore other base protocol messages
@ -371,7 +357,6 @@ func (p *Peer) handle(msg Msg) error {
if err != nil {
return fmt.Errorf("msg code out of range: %v", msg.Code)
}
if metrics.Enabled {
m := fmt.Sprintf("%s/%s/%d/%#02x", ingressMeterName, proto.Name, proto.Version, msg.Code-proto.offset)
metrics.GetOrRegisterMeter(m, nil).Mark(int64(msg.meterSize))
@ -384,13 +369,11 @@ func (p *Peer) handle(msg Msg) error {
return io.EOF
}
}
return nil
}
func countMatchingProtocols(protocols []Protocol, caps []Cap) int {
n := 0
for _, cap := range caps {
for _, proto := range protocols {
if proto.Name == cap.Name && proto.Version == cap.Version {
@ -398,7 +381,6 @@ func countMatchingProtocols(protocols []Protocol, caps []Cap) int {
}
}
}
return n
}
@ -424,34 +406,26 @@ outer:
}
}
}
return result
}
func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error) {
p.wg.Add(len(p.running))
for _, proto := range p.running {
proto := proto
proto.closed = p.closed
proto.wstart = writeStart
proto.werr = writeErr
var rw MsgReadWriter = proto
if p.events != nil {
rw = newMsgEventer(rw, p.events, p.ID(), proto.Name, p.Info().Network.RemoteAddress, p.Info().Network.LocalAddress)
}
p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
go func() {
defer p.wg.Done()
err := proto.Run(p, rw)
if err == nil {
p.log.Trace(fmt.Sprintf("Protocol %s/%d returned", proto.Name, proto.Version))
err = errProtocolReturned
} else if !errors.Is(err, io.EOF) {
p.log.Trace(fmt.Sprintf("Protocol %s/%d failed", proto.Name, proto.Version), "err", err)
@ -469,7 +443,6 @@ func (p *Peer) getProto(code uint64) (*protoRW, error) {
return proto, nil
}
}
return nil, newPeerError(errInvalidMsgCode, "%d", code)
}
@ -487,7 +460,6 @@ func (rw *protoRW) WriteMsg(msg Msg) (err error) {
if msg.Code >= rw.Length {
return newPeerError(errInvalidMsgCode, "not handled")
}
msg.meterCap = rw.cap()
msg.meterCode = msg.Code
@ -504,7 +476,6 @@ func (rw *protoRW) WriteMsg(msg Msg) (err error) {
case <-rw.closed:
err = ErrShuttingDown
}
return err
}
@ -555,7 +526,6 @@ func (p *Peer) Info() *PeerInfo {
if p.Node().Seq() > 0 {
info.ENR = p.Node().String()
}
info.Network.LocalAddress = p.LocalAddr().String()
info.Network.RemoteAddress = p.RemoteAddr().String()
info.Network.Inbound = p.rw.is(inboundConn)
@ -565,7 +535,6 @@ func (p *Peer) Info() *PeerInfo {
// Gather all the running protocol infos
for _, proto := range p.running {
protoInfo := interface{}("unknown")
if query := proto.Protocol.PeerInfo; query != nil {
if metadata := query(p.ID()); metadata != nil {
protoInfo = metadata
@ -573,9 +542,7 @@ func (p *Peer) Info() *PeerInfo {
protoInfo = "handshake"
}
}
info.Protocols[proto.Name] = protoInfo
}
return info
}

3
rpc/client.go
View file

@ -347,7 +347,6 @@ func (c *Client) CallContext(ctx context.Context, result interface{}, method str
if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
}
msg, err := c.newMessage(method, args...)
if err != nil {
return err
@ -362,7 +361,6 @@ func (c *Client) CallContext(ctx context.Context, result interface{}, method str
} else {
err = c.send(ctx, op, msg)
}
if err != nil {
return err
}
@ -382,7 +380,6 @@ func (c *Client) CallContext(ctx context.Context, result interface{}, method str
if result == nil {
return nil
}
return json.Unmarshal(resp.Result, result)
}
}

7
rpc/http.go
View file

@ -171,12 +171,10 @@ func newClientTransportHTTP(endpoint string, cfg *clientConfig) reconnectFunc {
func (c *Client) sendHTTP(ctx context.Context, op *requestOp, msg interface{}) error {
hc := c.writeConn.(*httpConn)
respBody, err := hc.doRequest(ctx, msg)
if err != nil {
return err
}
defer respBody.Close()
var resp jsonrpcMessage
@ -211,12 +209,10 @@ func (hc *httpConn) doRequest(ctx context.Context, msg interface{}) (io.ReadClos
if err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, hc.url, io.NopCloser(bytes.NewReader(body)))
if err != nil {
return nil, err
}
req.ContentLength = int64(len(body))
req.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(bytes.NewReader(body)), nil }
@ -237,10 +233,8 @@ func (hc *httpConn) doRequest(ctx context.Context, msg interface{}) (io.ReadClos
if err != nil {
return nil, err
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
var buf bytes.Buffer
var body []byte
if _, err := buf.ReadFrom(resp.Body); err == nil {
body = buf.Bytes()
@ -252,7 +246,6 @@ func (hc *httpConn) doRequest(ctx context.Context, msg interface{}) (io.ReadClos
Body: body,
}
}
return resp.Body, nil
}