Skip to content

Commit 220667d

Browse files
committed
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.
1 parent be73c53 commit 220667d

2 files changed

Lines changed: 39 additions & 4 deletions

File tree

‎eth/peer.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ type peer struct {
109109

110110
term chan struct{} // Termination channel to stop the broadcaster
111111
broadcastWg sync.WaitGroup // Tracks the broadcaster goroutines so they can be awaited
112+
closeOnce sync.Once // Ensures term is closed exactly once
112113

113114
knownVote mapset.Set[common.Hash] // Set of BFT Vote known to be known by this peer
114115
knownTimeout mapset.Set[common.Hash] // Set of BFT timeout known to be known by this peer
@@ -303,9 +304,10 @@ func (p *peer) announceTransactions() {
303304
}
304305
}
305306

306-
// close signals the broadcast goroutine to terminate.
307+
// close signals the broadcast goroutine to terminate. It is safe for
308+
// concurrent and repeated calls: only the first call closes term.
307309
func (p *peer) close() {
308-
close(p.term)
310+
p.closeOnce.Do(func() { close(p.term) })
309311
}
310312

311313
// Info gathers and returns a collection of metadata known about a peer.

‎eth/peer_test.go‎

Lines changed: 35 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,40 @@ func waitBroadcasters(t *testing.T, wg *sync.WaitGroup) {
8282
}
8383
}
8484

85+
// TestPeerCloseIsIdempotent verifies that close can be called repeatedly and
86+
// from concurrent goroutines without panicking on a double close of term. The
87+
// peer set owns term via Unregister, but making close idempotent removes the
88+
// risk of a "close of closed channel" panic from future callers.
89+
func TestPeerCloseIsIdempotent(t *testing.T) {
90+
app, net := p2p.MsgPipe()
91+
defer app.Close()
92+
var id enode.ID
93+
if _, err := rand.Read(id[:]); err != nil {
94+
t.Fatalf("failed to generate random peer id: %v", err)
95+
}
96+
p := newPeer(xdc100, p2p.NewPeer(id, "close", nil), net, func(common.Hash) *types.Transaction { return nil })
97+
98+
var wg sync.WaitGroup
99+
for i := 0; i < 8; i++ {
100+
wg.Add(1)
101+
go func() {
102+
defer wg.Done()
103+
p.close()
104+
}()
105+
}
106+
p.close()
107+
wg.Wait()
108+
109+
// All concurrent calls returned without panic, so idempotency holds. As a
110+
// final sanity check, the first call must have closed term; the default
111+
// case fails fast if close ever stops closing it.
112+
select {
113+
case <-p.term:
114+
default:
115+
t.Fatal("close did not terminate the peer")
116+
}
117+
}
118+
85119
// countingMsgWriter wraps a p2p.MsgReadWriter and counts every write attempt,
86120
// allowing tests to assert that a peer stops sending after a network error.
87121
type countingMsgWriter struct {
@@ -115,8 +149,7 @@ func newBroadcastTestPeer(t *testing.T, name string) (*peer, *types.Transaction,
115149
}
116150
counter := &countingMsgWriter{MsgReadWriter: net}
117151
p := newPeer(xdc100, p2p.NewPeer(id, name, nil), counter, func(common.Hash) *types.Transaction { return tx })
118-
var closeOnce sync.Once
119-
terminate := func() { closeOnce.Do(p.close) }
152+
terminate := p.close
120153
t.Cleanup(terminate)
121154
app.Close()
122155
return p, tx, counter, terminate

0 commit comments

Comments
 (0)