eth/protocols/eth: pass initial block range into handshake

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(&eth.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()
+}

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

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
+}

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) {

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.