fix(p2p/pex): de-flake TestPEXReactorRunning by ignoring duplicate-peer dial rejections

p2p/errors.go

@@ -106,6 +106,18 @@ func (e ErrRejected) IsAuthFailure() bool { return e.isAuthFailure }
 // IsDuplicate when Peer ID or IP are present already.
 func (e ErrRejected) IsDuplicate() bool { return e.isDuplicate }
 
+// DuplicatePeerID returns the peer ID and true when the rejection was caused
+// by a duplicate peer ID. Returns "", false for duplicate-IP rejections or
+// non-duplicate errors. Callers that need to know *why* a peer was deemed a
+// duplicate (e.g. to decide whether to penalize the dialed address) should
+// use this in preference to IsDuplicate.
+func (e ErrRejected) DuplicatePeerID() (ID, bool) {
+	if e.isDuplicate && e.id != "" {
+		return e.id, true
+	}
+	return "", false
+}
+
 // IsFiltered when Peer ID or IP was filtered.
 func (e ErrRejected) IsFiltered() bool { return e.isFiltered }
 

p2p/pex/pex_reactor.go

@@ -560,6 +560,23 @@ func (r *Reactor) dialPeer(addr *p2p.NetAddress) error {
 		if _, ok := err.(p2p.ErrCurrentlyDialingOrExistingAddress); ok {
 			return err
 		}
+		// If the rejection came from a duplicate *peer ID* matching the peer
+		// we tried to dial, treat the dial as success-equivalent: the peer is
+		// already connected (typically via an inbound connection raced against
+		// our outbound dial), so we shouldn't penalize this address with the
+		// 30s dial backoff or increment its attempt counter. Duplicate-IP
+		// rejections take the normal failure path, since the colliding peer
+		// may not be the one we wanted (e.g. an unrelated peer sharing an IP).
+		if e, ok := err.(p2p.ErrRejected); ok {
+			if id, ok := e.DuplicatePeerID(); ok && id == addr.ID {
+				r.attemptsToDial.Delete(addr.DialString())
+				return err
+			}
+		}
+		if e, ok := err.(p2p.ErrSwitchDuplicatePeerID); ok && e.ID == addr.ID {
+			r.attemptsToDial.Delete(addr.DialString())
+			return err
+		}
 
 		markAddrInBookBasedOnErr(addr, r.book, err)
 		switch err.(type) {

p2p/pex/pex_reactor_test.go

@@ -111,7 +111,7 @@ func TestPEXReactorRunning(t *testing.T) {
 		require.Nil(t, err)
 	}
 
-	assertPeersWithTimeout(t, switches, 10*time.Millisecond, 10*time.Second, N-1)
+	assertPeersWithTimeout(t, switches, 10*time.Millisecond, 30*time.Second, N-1)
 
 	// stop them
 	for _, s := range switches {