Reintroduce torrent-wide PEX throttling

https://github.com/anacrolix/torrent/discussions/836

pex.go

@@ -4,6 +4,7 @@ import (
 	"net"
 	"net/netip"
 	"sync"
+	"time"
 
 	pp "github.com/anacrolix/torrent/peer_protocol"
 )
@@ -144,8 +145,11 @@ func (me *pexMsgFactory) PexMsg() *pp.PexMsg {
 // Per-torrent PEX state
 type pexState struct {
 	sync.RWMutex
-	tail *pexEvent     // event feed list
+	tail *pexEvent  // event feed list
-	hold []pexEvent    // delayed drops
+	hold []pexEvent // delayed drops
+	// Torrent-wide cooldown deadline on inbound. This exists to prevent PEX from drowning out other
+	// peer address sources, until that is fixed.
+	rest time.Time
 	nc   int           // net number of alive conns
 	msg0 pexMsgFactory // initial message
 }
@@ -157,6 +161,7 @@ func (s *pexState) Reset() {
 	s.tail = nil
 	s.hold = nil
 	s.nc = 0
+	s.rest = time.Time{}
 	s.msg0 = pexMsgFactory{}
 }
 

pexconn.go

@@ -143,9 +143,17 @@ func (s *pexConnState) Recv(payload []byte) error {
 	var peers peerInfos
 	peers.AppendFromPex(rx.Added6, rx.Added6Flags)
 	peers.AppendFromPex(rx.Added, rx.AddedFlags)
+	if time.Now().Before(s.torrent.pex.rest) {
+		s.dbg.Printf("in cooldown period, incoming PEX discarded")
+		return nil
+	}
 	added := s.torrent.addPeers(peers)
 	s.dbg.Printf("got %v peers over pex, added %v", len(peers), added)
 
+	if len(peers) > 0 {
+		s.torrent.pex.rest = time.Now().Add(pexInterval)
+	}
+
 	// one day we may also want to:
 	// - handle drops somehow
 	// - detect malicious peers