From 850469070a485cf66e0187ab8a721bf2fa2ee779 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Wed, 12 Aug 2026 14:06:39 +0800 Subject: [PATCH 1/3] fix(eth): terminate peer broadcasters on unregister peerSet.Unregister never closed p.term, so the broadcast goroutines of every removed peer leaked for the lifetime of the process. Close term under the peer set lock so the draining loops wind down on removal. --- eth/peer.go | 22 ++++++++++++----- eth/peer_test.go | 63 +++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 78 insertions(+), 7 deletions(-) diff --git a/eth/peer.go b/eth/peer.go index daf97d0b23e0..493764f61551 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -107,7 +107,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 + term chan struct{} // Termination channel to stop the broadcaster + broadcastWg sync.WaitGroup // Tracks the broadcaster goroutines so they can be awaited 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 @@ -847,24 +848,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 } diff --git a/eth/peer_test.go b/eth/peer_test.go index 95c43062db15..234df3445fd3 100644 --- a/eth/peer_test.go +++ b/eth/peer_test.go @@ -1,6 +1,16 @@ package eth -import "testing" +import ( + "crypto/rand" + "sync" + "testing" + "time" + + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/p2p" + "github.com/XinFinOrg/XDPoSChain/p2p/enode" +) func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) { peers := newPeerSet() @@ -20,3 +30,54 @@ 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) + } +} From 40a0fa983be9986432b20de7784891bf3ee0c5c5 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Wed, 12 Aug 2026 14:06:39 +0800 Subject: [PATCH 2/3] fix(eth): keep tx broadcaster draining after send failure broadcastTransactions and announceTransactions returned on the first network send error, leaving p.txBroadcast and p.txAnnounce without a reader while the peer was still registered; every later AsyncSendTransactions blocked forever. On mainnet that stalled txBroadcastLoop, filled pm.txsCh and pinned the transaction pool lock for hours. Stop sending after a failure but keep servicing the queue until p.term, discarding queued events once sending is hopeless. Buffer the fail channel so an in-flight sender whose error races the loop's term exit cannot block forever on an unbuffered send. --- eth/peer.go | 40 +++++++++---- eth/peer_test.go | 145 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 10 deletions(-) diff --git a/eth/peer.go b/eth/peer.go index 493764f61551..e173e70ae14d 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -167,9 +167,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 @@ -205,6 +206,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 { @@ -215,8 +220,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 @@ -229,9 +239,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 @@ -267,6 +278,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 { @@ -277,8 +292,13 @@ 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 diff --git a/eth/peer_test.go b/eth/peer_test.go index 234df3445fd3..2c635d143527 100644 --- a/eth/peer_test.go +++ b/eth/peer_test.go @@ -2,12 +2,15 @@ package eth 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" ) @@ -81,3 +84,145 @@ func waitBroadcasters(t *testing.T, wg *sync.WaitGroup) { t.Fatalf("waitBroadcasters: broadcast goroutines still running after %v, want them to terminate", gracePeriod) } } + +// 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 }) + var closeOnce sync.Once + terminate := func() { closeOnce.Do(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) + } +} From f2c9d10ad6dd2f2d4db04d29ffeedf857067320e Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Wed, 12 Aug 2026 14:06:39 +0800 Subject: [PATCH 3/3] fix(eth): make peer close idempotent Guard the close of term with a sync.Once so repeated or concurrent calls cannot panic on a double close. Ownership of term stays with peerSet.Unregister; the once hardens the exactly-once invariant. --- eth/peer.go | 6 ++++-- eth/peer_test.go | 37 +++++++++++++++++++++++++++++++++++-- 2 files changed, 39 insertions(+), 4 deletions(-) diff --git a/eth/peer.go b/eth/peer.go index e173e70ae14d..3771535cfe6b 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -109,6 +109,7 @@ type peer struct { 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 @@ -306,9 +307,10 @@ func (p *peer) announceTransactions() { } } -// 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. diff --git a/eth/peer_test.go b/eth/peer_test.go index 2c635d143527..e757634a0f90 100644 --- a/eth/peer_test.go +++ b/eth/peer_test.go @@ -85,6 +85,40 @@ func waitBroadcasters(t *testing.T, wg *sync.WaitGroup) { } } +// 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 { @@ -118,8 +152,7 @@ func newBroadcastTestPeer(t *testing.T, name string) (*peer, *types.Transaction, } counter := &countingMsgWriter{MsgReadWriter: net} p := newPeer(xdc100, p2p.NewPeer(id, name, nil), counter, func(common.Hash) *types.Transaction { return tx }) - var closeOnce sync.Once - terminate := func() { closeOnce.Do(p.close) } + terminate := p.close t.Cleanup(terminate) app.Close() return p, tx, counter, terminate