Fixup for TestSendTransactions65 (#838)

build/ci.go

@@ -341,7 +341,6 @@ func doTest(cmdline []string) {
 	// and some tests run into timeouts under load.
 	gotest := goTool("test", buildFlags(env)...)
 	gotest.Args = append(gotest.Args, "-p", "1")
-	*verbose = true
 	if *coverage {
 		gotest.Args = append(gotest.Args, "-covermode=atomic", "-cover")
 	}

eth/handler.go

@@ -426,37 +426,50 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		number  = head.Number.Uint64()
 		td      = pm.blockchain.GetTd(hash, number)
 	)
+
 	if err := p.Handshake(pm.networkID, td, hash, genesis.Hash(), forkid.NewID(pm.chainConfig, genesis.Hash(), number), pm.forkFilter); err != nil {
 		p.Log().Debug("Ethereum handshake failed", "err", err)
 		return err
 	}
+
+	// Make sure that we first exchange headers and only then announce transactions
+	p.HandshakeOrderMux.Lock()
 	// Register the peer locally
 	if err := pm.peers.Register(p); err != nil {
 		p.Log().Error("Ethereum peer registration failed", "err", err)
+		p.HandshakeOrderMux.Lock()
 		return err
 	}
 	defer pm.removePeer(p.id)
 
 	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
 	if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+		p.HandshakeOrderMux.Unlock()
 		return err
 	}
 	pm.chainSync.handlePeerEvent(p)
 
 	// Propagate existing transactions. new transactions appearing
 	// after this will be sent via broadcasts.
-	pm.syncTransactions(p)
-
 	// Send request for the head header
 	peerHeadHash, _ := p.Head()
 	if err := p.RequestHeadersByHash(peerHeadHash, 1, 0, false); err != nil {
+		p.HandshakeOrderMux.Unlock()
 		return err
 	}
+
 	// Handle one message to prevent two peers deadlocking each other
 	if err := pm.handleMsg(p); err != nil {
 		p.Log().Debug("Ethereum message handling failed", "err", err)
+		p.HandshakeOrderMux.Unlock()
 		return err
 	}
+
+	// Allow to handle transaction ordering
+	p.HandshakeOrderMux.Unlock()
+
+	pm.syncTransactions(p)
+
 	// If we have a trusted CHT, reject all peers below that (avoid fast sync eclipse)
 	if pm.checkpointHash != (common.Hash{}) {
 		// Request the peer's checkpoint header for chain height/weight validation

eth/peer.go

@@ -108,6 +108,8 @@ type peer struct {
 	getPooledTx func(common.Hash) *types.Transaction // Callback used to retrieve transaction from txpool
 
 	term chan struct{} // Termination channel to stop the broadcaster
+
+	HandshakeOrderMux sync.Mutex // This mutex enforces the order of operations when registering new peer on eth65+
 }
 
 func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter, getPooledTx func(hash common.Hash) *types.Transaction) *peer {
@@ -222,6 +224,14 @@ func (p *peer) announceTransactions() {
 		done  chan struct{}         // Non-nil if background announcer is running
 		fail  = make(chan error, 1) // Channel used to receive network error
 	)
+
+	// Making sure that we don't announce transactions too early.
+	// It this lock is set, it means that we are in process of exchanging latest block headers.
+	p.HandshakeOrderMux.Lock()
+	// this causes a false-positive SA2001: empty critical section, so we intentionally ignore it
+	//nolint: staticcheck
+	p.HandshakeOrderMux.Unlock()
+
 	for {
 		// If there's no in-flight announce running, check if a new one is needed
 		if done == nil && len(queue) > 0 {