fix(eth): keep the peer transaction broadcaster draining after a send error

eth/peer.go

@@ -107,7 +107,9 @@ 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
+	term        chan struct{}  // Termination channel to stop the broadcaster
+	broadcastWg sync.WaitGroup // Tracks the broadcaster goroutines so they can be awaited
+	closeOnce   sync.Once      // Ensures term is closed exactly once
 
 	knownVote     mapset.Set[common.Hash] // Set of BFT Vote known to be known by this peer
 	knownTimeout  mapset.Set[common.Hash] // Set of BFT timeout known to be known by this peer
@@ -166,9 +168,10 @@ func (p *peer) broadcastBlocks() {
 // node internals and at the same time rate limits queued data.
 func (p *peer) broadcastTransactions() {
 	var (
-		queue []common.Hash      // Queue of hashes to broadcast as full transactions
-		done  chan struct{}      // Non-nil if background broadcaster is running
-		fail  = make(chan error) // Channel used to receive network error
+		queue  []common.Hash         // Queue of hashes to broadcast as full transactions
+		done   chan struct{}         // Non-nil if background broadcaster is running
+		fail   = make(chan error, 1) // Channel used to receive network error
+		failed bool                  // Keep draining the queue once sending is hopeless
 	)
 	for {
 		// If there's no in-flight broadcast running, check if a new one is needed
@@ -204,6 +207,10 @@ func (p *peer) broadcastTransactions() {
 		// Transfer goroutine may or may not have been started, listen for events
 		select {
 		case hashes := <-p.txBroadcast:
+			// If the connection failed, discard all transaction events
+			if failed {
+				continue
+			}
 			// New batch of transactions to be broadcast, queue them (with cap)
 			queue = append(queue, hashes...)
 			if len(queue) > maxQueuedTxs {
@@ -214,8 +221,13 @@ func (p *peer) broadcastTransactions() {
 		case <-done:
 			done = nil
 
-		case <-fail:
-			return
+		case err := <-fail:
+			// p.term is only closed when the peer is removed from the peer set
+			// (removePeer -> Unregister -> close), which may lag the connection
+			// failure. Stay around as a reader, or
+			// AsyncSendTransactions would block forever.
+			failed, queue, done = true, nil, nil
+			p.Log().Debug("Transaction broadcast send failed, draining queued events", "err", err)
 
 		case <-p.term:
 			return
@@ -228,9 +240,10 @@ func (p *peer) broadcastTransactions() {
 // node internals and at the same time rate limits queued data.
 func (p *peer) announceTransactions() {
 	var (
-		queue []common.Hash      // Queue of hashes to announce as transaction stubs
-		done  chan struct{}      // Non-nil if background announcer is running
-		fail  = make(chan error) // Channel used to receive network error
+		queue  []common.Hash         // Queue of hashes to announce as transaction stubs
+		done   chan struct{}         // Non-nil if background announcer is running
+		fail   = make(chan error, 1) // Channel used to receive network error
+		failed bool                  // Keep draining the queue once sending is hopeless
 	)
 	for {
 		// If there's no in-flight announce running, check if a new one is needed
@@ -266,6 +279,10 @@ func (p *peer) announceTransactions() {
 		// Transfer goroutine may or may not have been started, listen for events
 		select {
 		case hashes := <-p.txAnnounce:
+			// If the connection failed, discard all transaction events
+			if failed {
+				continue
+			}
 			// New batch of transactions to be broadcast, queue them (with cap)
 			queue = append(queue, hashes...)
 			if len(queue) > maxQueuedTxAnns {
@@ -276,18 +293,24 @@ func (p *peer) announceTransactions() {
 		case <-done:
 			done = nil
 
-		case <-fail:
-			return
+		case err := <-fail:
+			// p.term is only closed when the peer is removed from the peer set
+			// (removePeer -> Unregister -> close), which may lag the connection
+			// failure. Stay around as a reader, or
+			// AsyncSendPooledTransactionHashes would block forever.
+			failed, queue, done = true, nil, nil
+			p.Log().Debug("Transaction announcement send failed, draining queued events", "err", err)
 
 		case <-p.term:
 			return
 		}
 	}
 }
 
-// close signals the broadcast goroutine to terminate.
+// close signals the broadcast goroutine to terminate. It is safe for
+// concurrent and repeated calls: only the first call closes term.
 func (p *peer) close() {
-	close(p.term)
+	p.closeOnce.Do(func() { close(p.term) })
 }
 
 // Info gathers and returns a collection of metadata known about a peer.
@@ -847,24 +870,33 @@ func (ps *peerSet) Register(p *peer) error {
 	}
 	ps.peers[p.id] = p
 
-	go p.broadcastBlocks()
-	go p.broadcastTransactions()
+	p.broadcastWg.Go(func() {
+		p.broadcastBlocks()
+	})
+	p.broadcastWg.Go(func() {
+		p.broadcastTransactions()
+	})
 	if p.version >= xdc165 {
-		go p.announceTransactions()
+		p.broadcastWg.Go(func() {
+			p.announceTransactions()
+		})
 	}
 	return nil
 }
 
 // Unregister removes a remote peer from the active set, disabling any further
-// actions to/from that particular entity.
+// actions to/from that particular entity. It also terminates the peer's
+// broadcast goroutines, so they cannot leak once the peer is removed.
 func (ps *peerSet) Unregister(id string) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if _, ok := ps.peers[id]; !ok {
+	p, ok := ps.peers[id]
+	if !ok {
 		return errNotRegistered
 	}
 	delete(ps.peers, id)
+	p.close()
 	return nil
 }
 

eth/peer_test.go

@@ -1,6 +1,19 @@
 package eth
 
-import "testing"
+import (
+	"crypto/rand"
+	"math/big"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/XinFinOrg/XDPoSChain/common"
+	"github.com/XinFinOrg/XDPoSChain/core/types"
+	"github.com/XinFinOrg/XDPoSChain/crypto"
+	"github.com/XinFinOrg/XDPoSChain/p2p"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
+)
 
 func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) {
 	peers := newPeerSet()
@@ -20,3 +33,229 @@ func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) {
 		t.Fatalf("registered peer replaced: got %p want %p", got, first)
 	}
 }
+
+// TestPeerSetUnregisterTerminatesBroadcasters ensures that Unregister closes
+// p.term, so the peer's broadcast goroutines wind down when the peer is removed.
+// The loops only exit via p.term (or a send error), so without this the
+// goroutines leak and retain the peer for the lifetime of the process.
+func TestPeerSetUnregisterTerminatesBroadcasters(t *testing.T) {
+	peers := newPeerSet()
+
+	app, net := p2p.MsgPipe()
+	defer app.Close()
+	defer net.Close()
+	var id enode.ID
+	if _, err := rand.Read(id[:]); err != nil {
+		t.Fatalf("failed to generate random peer id: %v", err)
+	}
+	// Use xdc165 so Register also starts the transaction announcer; the wait
+	// below then covers all three broadcast goroutines, not just two.
+	p := newPeer(xdc165, p2p.NewPeer(id, "unregister", nil), net, func(common.Hash) *types.Transaction { return nil })
+	if err := peers.Register(p); err != nil {
+		t.Fatalf("first register failed: %v", err)
+	}
+	if err := peers.Unregister(p.id); err != nil {
+		t.Fatalf("unregister failed: %v", err)
+	}
+	// Unregister only closes term to signal the broadcasters; wait until the
+	// goroutines have actually exited so the leak this guards against is
+	// detected, not merely the close of the channel.
+	waitBroadcasters(t, &p.broadcastWg)
+	// A second unregister must fail cleanly and must not close term again.
+	if err := peers.Unregister(p.id); err != errNotRegistered {
+		t.Fatalf("second unregister error mismatch: got %v want %v", err, errNotRegistered)
+	}
+}
+
+// waitBroadcasters blocks until the peer's broadcast goroutines have exited,
+// or fails the test if they are still running after the grace period.
+func waitBroadcasters(t *testing.T, wg *sync.WaitGroup) {
+	t.Helper()
+	const gracePeriod = 2 * time.Second
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(gracePeriod):
+		t.Fatalf("waitBroadcasters: broadcast goroutines still running after %v, want them to terminate", gracePeriod)
+	}
+}
+
+// TestPeerCloseIsIdempotent verifies that close can be called repeatedly and
+// from concurrent goroutines without panicking on a double close of term. The
+// peer set owns term via Unregister, but making close idempotent removes the
+// risk of a "close of closed channel" panic from future callers.
+func TestPeerCloseIsIdempotent(t *testing.T) {
+	app, net := p2p.MsgPipe()
+	defer app.Close()
+	var id enode.ID
+	if _, err := rand.Read(id[:]); err != nil {
+		t.Fatalf("failed to generate random peer id: %v", err)
+	}
+	p := newPeer(xdc100, p2p.NewPeer(id, "close", nil), net, func(common.Hash) *types.Transaction { return nil })
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			p.close()
+		}()
+	}
+	p.close()
+	wg.Wait()
+
+	// All concurrent calls returned without panic, so idempotency holds. As a
+	// final sanity check, the first call must have closed term; the default
+	// case fails fast if close ever stops closing it.
+	select {
+	case <-p.term:
+	default:
+		t.Fatal("close did not terminate the peer")
+	}
+}
+
+// countingMsgWriter wraps a p2p.MsgReadWriter and counts every write attempt,
+// allowing tests to assert that a peer stops sending after a network error.
+type countingMsgWriter struct {
+	p2p.MsgReadWriter
+	writes atomic.Int32
+}
+
+func (c *countingMsgWriter) WriteMsg(msg p2p.Msg) error {
+	c.writes.Add(1)
+	return c.MsgReadWriter.WriteMsg(msg)
+}
+
+// newBroadcastTestPeer assembles a peer whose network connection is already
+// broken, so the first send from any broadcast loop fails. The returned
+// terminate function closes the peer exactly once and is safe to call from
+// anywhere, including t.Cleanup and the test body itself.
+func newBroadcastTestPeer(t *testing.T, name string) (*peer, *types.Transaction, *countingMsgWriter, func()) {
+	t.Helper()
+	key, err := crypto.GenerateKey()
+	if err != nil {
+		t.Fatalf("failed to generate test key: %v", err)
+	}
+	tx, err := types.SignTx(types.NewTransaction(0, common.Address{}, big.NewInt(1), 100000, big.NewInt(1), nil), types.HomesteadSigner{}, key)
+	if err != nil {
+		t.Fatalf("failed to sign test transaction: %v", err)
+	}
+	app, net := p2p.MsgPipe()
+	var id enode.ID
+	if _, err := rand.Read(id[:]); err != nil {
+		t.Fatalf("failed to generate random peer id: %v", err)
+	}
+	counter := &countingMsgWriter{MsgReadWriter: net}
+	p := newPeer(xdc100, p2p.NewPeer(id, name, nil), counter, func(common.Hash) *types.Transaction { return tx })
+	terminate := p.close
+	t.Cleanup(terminate)
+	app.Close()
+	return p, tx, counter, terminate
+}
+
+// runDrainCheck performs repeated asynchronous sends through send and fails the
+// test if any of them blocks, which would indicate that the peer's broadcast
+// loop stopped draining its queue after a send error. On a blocked send it
+// terminates the peer first, so the stuck sender unblocks via p.term and cannot
+// leak after the test has aborted.
+func runDrainCheck(t *testing.T, what string, send, terminate func()) {
+	t.Helper()
+	for i := 0; i < 50; i++ {
+		sent := make(chan struct{})
+		go func() {
+			defer close(sent)
+			send()
+		}()
+		select {
+		case <-sent:
+		case <-time.After(5 * time.Second):
+			terminate()
+			// The blocked sender unblocks via p.term; join it before failing so
+			// the failure path cannot leak the goroutine.
+			<-sent
+			t.Fatalf("%s blocked on attempt %d: the peer stopped draining after a send error", what, i)
+		}
+	}
+}
+
+// waitForWrites waits until the wrapped connection has seen its first write
+// attempt and reports the total number of attempts observed so far. The
+// broadcast loops launch at most one sender at a time and never start a new one
+// after a failure, so once the count reaches one it must never grow again. The
+// test fails if no write attempt is observed before the deadline.
+func waitForWrites(t *testing.T, counter *countingMsgWriter) int32 {
+	t.Helper()
+	const deadline = 5 * time.Second
+
+	timeout := time.NewTimer(deadline)
+	defer timeout.Stop()
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	for counter.writes.Load() == 0 {
+		select {
+		case <-timeout.C:
+			t.Fatalf("waitForWrites: no write attempt observed within %v, want 1", deadline)
+		case <-ticker.C:
+		}
+	}
+	return counter.writes.Load()
+}
+
+// TestBroadcastTransactionsKeepsDrainingAfterSendFailure reproduces the mainnet freeze:
+// the broadcaster used to return on the first send error, leaving p.txBroadcast without a
+// reader. Since p.term is only closed once the peer handler unwinds, every later
+// AsyncSendTransactions blocked forever and took txBroadcastLoop, and through it the
+// transaction pool, down with it.
+func TestBroadcastTransactionsKeepsDrainingAfterSendFailure(t *testing.T) {
+	p, tx, counter, terminate := newBroadcastTestPeer(t, "broadcaster")
+	broadcastDone := make(chan struct{})
+	go func() {
+		defer close(broadcastDone)
+		p.broadcastTransactions()
+	}()
+	t.Cleanup(func() { terminate(); <-broadcastDone })
+
+	runDrainCheck(t, "AsyncSendTransactions", func() { p.AsyncSendTransactions([]common.Hash{tx.Hash()}) }, terminate)
+
+	// The connection is broken, so the very first broadcast fails. After that
+	// the broadcaster must only drain its queue and never attempt another write.
+	if got := waitForWrites(t, counter); got != 1 {
+		t.Fatalf("write attempts after failed send: got %d, want 1", got)
+	}
+	// Extra batches still have to drain without blocking or resending.
+	runDrainCheck(t, "AsyncSendTransactions after failure", func() { p.AsyncSendTransactions([]common.Hash{tx.Hash()}) }, terminate)
+	if got := counter.writes.Load(); got != 1 {
+		t.Fatalf("write attempts after further drain: got %d, want 1", got)
+	}
+}
+
+// TestAnnounceTransactionsKeepsDrainingAfterSendFailure is the announce-side
+// counterpart: the announcer has to keep servicing p.txAnnounce until p.term,
+// otherwise AsyncSendPooledTransactionHashes blocks forever.
+func TestAnnounceTransactionsKeepsDrainingAfterSendFailure(t *testing.T) {
+	p, tx, counter, terminate := newBroadcastTestPeer(t, "announcer")
+	broadcastDone := make(chan struct{})
+	go func() {
+		defer close(broadcastDone)
+		p.announceTransactions()
+	}()
+	t.Cleanup(func() { terminate(); <-broadcastDone })
+
+	runDrainCheck(t, "AsyncSendPooledTransactionHashes", func() { p.AsyncSendPooledTransactionHashes([]common.Hash{tx.Hash()}) }, terminate)
+
+	// The connection is broken, so the very first announcement fails. After that
+	// the announcer must only drain its queue and never attempt another write.
+	if got := waitForWrites(t, counter); got != 1 {
+		t.Fatalf("write attempts after failed send: got %d, want 1", got)
+	}
+	// Extra batches still have to drain without blocking or resending.
+	runDrainCheck(t, "AsyncSendPooledTransactionHashes after failure", func() { p.AsyncSendPooledTransactionHashes([]common.Hash{tx.Hash()}) }, terminate)
+	if got := counter.writes.Load(); got != 1 {
+		t.Fatalf("write attempts after further drain: got %d, want 1", got)
+	}
+}