Skip to content

testBroadcastBlock / testEthHandlerPoW

Source (upstream pre-purge) Current
File eth/handler_eth_test.go handler_eth_pow_test.go
Symbol testBroadcastBlock testBroadcastBlock
Ref f4d53133f~1 etc/v1.17.4-full-node

The block propagation test #29169 deleted, revived with the TestBroadcastBlockNPeers table verbatim. It is the only end-to-end coverage of the PoW block path: a real block crosses a real p2p.MsgPipe as a NewBlockMsg, the sink runs the real protocol handler, so the message goes through handleNewBlockPow's decode, item caps and body verification, and the assembled block is asserted to arrive at the backend with the square-root propagation count. Adapted to the current signatures: newTestHandlerWithBlocks gained a SyncMode parameter, eth.NewPeer gained a chainConfig parameter, and the sink handshake calls Handshake68PoW. Two pieces could not be revived into their upstream homes and are fork-owned instead. testEthHandlerPoW embeds the upstream testEthHandler and adds back the blockBroadcasts feed and the *eth.NewBlockPacket case that #33511 removed, since the upstream struct cannot be extended from another file. newTestHandlerPoWWithBlocks mirrors newTestHandlerWithBlocks with IsPow set: the flag decides what newHandler wires up and which handshake runEthPeer runs, so it cannot be flipped after construction, and the upstream constructor never sets it.

3-way merge — purge → getc ← upstream

pre-purge≈ adapted (origin inferred by similarity)
func testBroadcastBlock(t *testing.T, peers, bcasts int) {
t.Parallel()
// Create a source handler to broadcast blocks from and a number of sinks
// to receive them.
source := newTestHandlerPoWWithBlocks(1)
defer source.close()
sinks := make([]*testEthHandlerPoW, peers)
for i := 0; i < len(sinks); i++ {
sinks[i] = new(testEthHandlerPoW)
}
// Interconnect all the sink handlers with the source handler
var (
genesis = source.chain.Genesis()
td = source.chain.GetTd(genesis.Hash(), genesis.NumberU64())
)
for i, sink := range sinks {
sourcePipe, sinkPipe := p2p.MsgPipe()
defer sourcePipe.Close()
defer sinkPipe.Close()
sourcePeer := eth.NewPeer(eth.ETH68, p2p.NewPeerPipe(enode.ID{byte(i)}, "", nil, sourcePipe), sourcePipe, nil, nil)
sinkPeer := eth.NewPeer(eth.ETH68, p2p.NewPeerPipe(enode.ID{0}, "", nil, sinkPipe), sinkPipe, nil, nil)
defer sourcePeer.Close()
defer sinkPeer.Close()
go source.handler.runEthPeer(sourcePeer, func(peer *eth.Peer) error {
return eth.Handle((*ethHandler)(source.handler), peer)
})
if err := sinkPeer.Handshake68PoW(1, td, genesis.Hash(), genesis.Hash(), forkid.NewIDWithChain(source.chain), forkid.NewFilter(source.chain)); err != nil {
t.Fatalf("failed to run protocol handshake")
}
go eth.Handle(sink, sinkPeer)
}
// Subscribe to all the transaction pools
blockChs := make([]chan *types.Block, len(sinks))
for i := 0; i < len(sinks); i++ {
blockChs[i] = make(chan *types.Block, 1)
defer close(blockChs[i])
sub := sinks[i].blockBroadcasts.Subscribe(blockChs[i])
defer sub.Unsubscribe()
}
// Initiate a block propagation across the peers
time.Sleep(100 * time.Millisecond)
header := source.chain.CurrentBlock()
source.handler.BroadcastBlock(source.chain.GetBlock(header.Hash(), header.Number.Uint64()), true)
// Iterate through all the sinks and ensure the correct number got the block
done := make(chan struct{}, peers)
for _, ch := range blockChs {
go func() {
<-ch
done <- struct{}{}
}()
}
var received int
for {
select {
case <-done:
received++
case <-time.After(100 * time.Millisecond):
if received != bcasts {
t.Errorf("broadcast count mismatch: have %d, want %d", received, bcasts)
}
return
}
}
}

← Eth Service