From d8de9f4dc57f2c1414af622e75675dc7fa18961c Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 25 Apr 2025 21:22:52 +0200 Subject: [PATCH] eth/protocols/eth: pass initial block range into handshake --- eth/handler.go | 29 +++++++++++------- eth/handler_eth_test.go | 4 +-- eth/protocols/eth/handshake.go | 46 +++++++++++++++++------------ eth/protocols/eth/handshake_test.go | 2 +- eth/protocols/eth/protocol.go | 10 ++++--- 5 files changed, 55 insertions(+), 36 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index f26db857c5..c9927c2357 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -259,7 +259,7 @@ func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error { } // Execute the Ethereum handshake - if err := peer.Handshake(h.networkID, h.chain); err != nil { + if err := peer.Handshake(h.networkID, h.chain, h.blockRange.currentRange()); err != nil { peer.Log().Debug("Ethereum handshake failed", "err", err) return err } @@ -571,7 +571,7 @@ func (h *handler) enableSyncedFeatures() { // blockRangeState holds the state of the block range update broadcasting mechanism. type blockRangeState struct { prev eth.BlockRangeUpdatePacket - next eth.BlockRangeUpdatePacket + next atomic.Pointer[eth.BlockRangeUpdatePacket] headCh chan core.ChainHeadEvent headSub event.Subscription syncSub *event.TypeMuxSubscription @@ -587,7 +587,7 @@ func newBlockRangeState(chain *core.BlockChain, typeMux *event.TypeMux) *blockRa syncSub: syncSub, } st.update(chain, chain.CurrentBlock()) - st.prev = st.next + st.prev = *st.next.Load() return st } @@ -655,20 +655,22 @@ func (h *handler) broadcastBlockRange(state *blockRangeState) { if len(peerlist) == 0 { return } - msg := state.next + msg := state.currentRange() log.Debug("Sending BlockRangeUpdate", "peers", len(peerlist), "earliest", msg.EarliestBlock, "latest", msg.LatestBlock) for _, p := range peerlist { p.SendBlockRangeUpdate(msg) } - state.prev = state.next + state.prev = *state.next.Load() } // update assigns the values of the next block range update from the chain. func (st *blockRangeState) update(chain *core.BlockChain, latest *types.Header) { earliest, _ := chain.HistoryPruningCutoff() - st.next.EarliestBlock = min(latest.Number.Uint64(), earliest) - st.next.LatestBlock = latest.Number.Uint64() - st.next.LatestBlockHash = latest.Hash() + st.next.Store(ð.BlockRangeUpdatePacket{ + EarliestBlock: min(latest.Number.Uint64(), earliest), + LatestBlock: latest.Number.Uint64(), + LatestBlockHash: latest.Hash(), + }) } // shouldSend decides whether it is time to send a block range update. We don't want to @@ -676,11 +678,18 @@ func (st *blockRangeState) update(chain *core.BlockChain, latest *types.Header) // However, there is a special case: if the range would move back, i.e. due to SetHead, we // want to send it immediately. func (st *blockRangeState) shouldSend() bool { - return st.next.LatestBlock < st.prev.LatestBlock || - st.next.LatestBlock-st.prev.LatestBlock >= 32 + next := st.next.Load() + return next.LatestBlock < st.prev.LatestBlock || + next.LatestBlock-st.prev.LatestBlock >= 32 } func (st *blockRangeState) stop() { st.syncSub.Unsubscribe() st.headSub.Unsubscribe() } + +// currentRange returns the current block range. +// This is safe to call from any goroutine. +func (st *blockRangeState) currentRange() eth.BlockRangeUpdatePacket { + return *st.next.Load() +} diff --git a/eth/handler_eth_test.go b/eth/handler_eth_test.go index 8ce56e8373..2446c5fccb 100644 --- a/eth/handler_eth_test.go +++ b/eth/handler_eth_test.go @@ -256,7 +256,7 @@ func testRecvTransactions(t *testing.T, protocol uint) { return eth.Handle((*ethHandler)(handler.handler), peer) }) // Run the handshake locally to avoid spinning up a source handler - if err := src.Handshake(1, handler.chain); err != nil { + if err := src.Handshake(1, handler.chain, eth.BlockRangeUpdatePacket{}); err != nil { t.Fatalf("failed to run protocol handshake") } // Send the transaction to the sink and verify that it's added to the tx pool @@ -311,7 +311,7 @@ func testSendTransactions(t *testing.T, protocol uint) { return eth.Handle((*ethHandler)(handler.handler), peer) }) // Run the handshake locally to avoid spinning up a source handler - if err := sink.Handshake(1, handler.chain); err != nil { + if err := sink.Handshake(1, handler.chain, eth.BlockRangeUpdatePacket{}); err != nil { t.Fatalf("failed to run protocol handshake") } // After the handshake completes, the source handler should stream the sink diff --git a/eth/protocols/eth/handshake.go b/eth/protocols/eth/handshake.go index 14d0c4d93f..824e49fb2b 100644 --- a/eth/protocols/eth/handshake.go +++ b/eth/protocols/eth/handshake.go @@ -36,10 +36,10 @@ const ( // Handshake executes the eth protocol handshake, negotiating version number, // network IDs, difficulties, head and genesis blocks. -func (p *Peer) Handshake(networkID uint64, chain *core.BlockChain) error { +func (p *Peer) Handshake(networkID uint64, chain *core.BlockChain, rangeMsg BlockRangeUpdatePacket) error { switch p.version { case ETH69: - return p.handshake69(networkID, chain) + return p.handshake69(networkID, chain, rangeMsg) case ETH68: return p.handshake68(networkID, chain) default: @@ -92,14 +92,12 @@ func (p *Peer) readStatus68(networkID uint64, status *StatusPacket68, genesis co return nil } -func (p *Peer) handshake69(networkID uint64, chain *core.BlockChain) error { +func (p *Peer) handshake69(networkID uint64, chain *core.BlockChain, rangeMsg BlockRangeUpdatePacket) error { var ( - genesis = chain.Genesis() - latest = chain.CurrentBlock() - latestSnap = chain.CurrentSnapBlock() - forkID = forkid.NewID(chain.Config(), genesis, latest.Number.Uint64(), latest.Time) - forkFilter = forkid.NewFilter(chain) - earliest, _ = chain.HistoryPruningCutoff() + genesis = chain.Genesis() + latest = chain.CurrentBlock() + forkID = forkid.NewID(chain.Config(), genesis, latest.Number.Uint64(), latest.Time) + forkFilter = forkid.NewFilter(chain) ) errc := make(chan error, 2) @@ -109,14 +107,9 @@ func (p *Peer) handshake69(networkID uint64, chain *core.BlockChain) error { NetworkID: networkID, Genesis: genesis.Hash(), ForkID: forkID, - EarliestBlock: earliest, - } - if latestSnap != nil && latestSnap.Number.Uint64() > latest.Number.Uint64() { - pkt.LatestBlock = latestSnap.Number.Uint64() - pkt.LatestBlockHash = latestSnap.Hash() - } else { - pkt.LatestBlock = latest.Number.Uint64() - pkt.LatestBlockHash = latest.Hash() + EarliestBlock: rangeMsg.EarliestBlock, + LatestBlock: rangeMsg.LatestBlock, + LatestBlockHash: rangeMsg.LatestBlockHash, } errc <- p2p.Send(p.rw, StatusMsg, pkt) }() @@ -145,11 +138,15 @@ func (p *Peer) readStatus69(networkID uint64, status *StatusPacket69, genesis co return fmt.Errorf("%w: %v", errForkIDRejected, err) } // Handle initial block range. - p.lastRange.Store(&BlockRangeUpdatePacket{ + initRange := &BlockRangeUpdatePacket{ EarliestBlock: status.EarliestBlock, LatestBlock: status.LatestBlock, LatestBlockHash: status.LatestBlockHash, - }) + } + if err := initRange.Validate(); err != nil { + return fmt.Errorf("%w: %v", errInvalidBlockRange, err) + } + p.lastRange.Store(initRange) return nil } @@ -210,3 +207,14 @@ func markError(p *Peer, err error) { m.peerError.Mark(1) } } + +// Validate checks basic validity of a block range announcement. +func (p *BlockRangeUpdatePacket) Validate() error { + if p.EarliestBlock > p.LatestBlock { + return errors.New("earliest > latest") + } + if p.LatestBlockHash == (common.Hash{}) { + return errors.New("zero latest hash") + } + return nil +} diff --git a/eth/protocols/eth/handshake_test.go b/eth/protocols/eth/handshake_test.go index 05055b72c2..2fab3ea5a8 100644 --- a/eth/protocols/eth/handshake_test.go +++ b/eth/protocols/eth/handshake_test.go @@ -80,7 +80,7 @@ func testHandshake(t *testing.T, protocol uint) { // Send the junk test with one peer, check the handshake failure go p2p.Send(app, test.code, test.data) - err := peer.Handshake(1, backend.chain) + err := peer.Handshake(1, backend.chain, BlockRangeUpdatePacket{}) if err == nil { t.Errorf("test %d: protocol returned nil error, want %q", i, test.want) } else if !errors.Is(err, test.want) { diff --git a/eth/protocols/eth/protocol.go b/eth/protocols/eth/protocol.go index 0b99860e32..7c41e7a996 100644 --- a/eth/protocols/eth/protocol.go +++ b/eth/protocols/eth/protocol.go @@ -67,13 +67,15 @@ const ( ) var ( - errNoStatusMsg = errors.New("no status message") errMsgTooLarge = errors.New("message too long") errInvalidMsgCode = errors.New("invalid message code") errProtocolVersionMismatch = errors.New("protocol version mismatch") - errNetworkIDMismatch = errors.New("network ID mismatch") - errGenesisMismatch = errors.New("genesis mismatch") - errForkIDRejected = errors.New("fork ID rejected") + // handshake errors + errNoStatusMsg = errors.New("no status message") + errNetworkIDMismatch = errors.New("network ID mismatch") + errGenesisMismatch = errors.New("genesis mismatch") + errForkIDRejected = errors.New("fork ID rejected") + errInvalidBlockRange = errors.New("invalid block range in status") ) // Packet represents a p2p message in the `eth` protocol.