diff --git a/eth/peer.go b/eth/peer.go index daf97d0b23e..3771535cfe6 100644 --- a/eth/peer.go +++ b/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,8 +293,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 @@ -285,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. @@ -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 } diff --git a/eth/peer_test.go b/eth/peer_test.go index 95c43062db1..e757634a0f9 100644 --- a/eth/peer_test.go +++ b/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) + } +}