fix(shrex/peers): bound blacklisted hashes with an LRU cache

The peer manager stored blacklisted datahashes in a plain map that was never
pruned, so it grew for the entire lifetime of a node. A peer broadcasting
shrexsub notifications with arbitrary invalid datahashes could drive unbounded
memory growth.

Replace the map with a fixed-size LRU cache. The blacklist is a best-effort
spam filter, so evicting the least-recently-used entries is acceptable.

Closes #1926

Author: neyy91

share/shwap/p2p/shrex/peers/manager.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	lru "github.com/hashicorp/golang-lru/v2"
 	logging "github.com/ipfs/go-log/v2"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/event"
@@ -41,6 +42,11 @@ const (
 	// storedPoolsAmount is the amount of pools for recent headers that will be stored in the peer
 	// manager
 	storedPoolsAmount = 10
+
+	// blacklistedHashesCacheSize bounds the number of blacklisted datahashes kept in memory.
+	// The blacklist is a best-effort spam filter, so evicting the least-recently-used entries
+	// is acceptable and prevents unbounded growth from a stream of invalid datahashes.
+	blacklistedHashesCacheSize = 1024
 )
 
 type result string
@@ -74,8 +80,8 @@ type Manager struct {
 	// nodes collects nodes' peer.IDs found via discovery
 	nodes *pool
 
-	// hashes that are not in the chain
-	blacklistedHashes map[string]bool
+	// hashes that are not in the chain, bounded by an LRU to avoid unbounded growth
+	blacklistedHashes *lru.Cache[string, struct{}]
 
 	metrics *metrics
 
@@ -111,12 +117,17 @@ func NewManager(
 		return nil, err
 	}
 
+	blacklistedHashes, err := lru.New[string, struct{}](blacklistedHashesCacheSize)
+	if err != nil {
+		return nil, fmt.Errorf("shrex/peer-manager: creating blacklisted hashes cache: %w", err)
+	}
+
 	s := &Manager{
 		params:                params,
 		connGater:             connGater,
 		host:                  host,
 		pools:                 make(map[string]*syncPool),
-		blacklistedHashes:     make(map[string]bool),
+		blacklistedHashes:     blacklistedHashes,
 		headerSubDone:         make(chan struct{}),
 		disconnectedPeersDone: make(chan struct{}),
 		tag:                   tag,
@@ -445,7 +456,7 @@ func (m *Manager) isBlacklistedPeer(peerID peer.ID) bool {
 func (m *Manager) isBlacklistedHash(hash share.DataHash) bool {
 	m.lock.Lock()
 	defer m.lock.Unlock()
-	return m.blacklistedHashes[hash.String()]
+	return m.blacklistedHashes.Contains(hash.String())
 }
 
 func (m *Manager) validatedPool(hashStr string, height uint64) *syncPool {
@@ -520,7 +531,7 @@ func (m *Manager) cleanUp() []peer.ID {
 				"hash", h,
 				"peer_list", p.peersList)
 			// blacklist hash
-			m.blacklistedHashes[h] = true
+			m.blacklistedHashes.Add(h, struct{}{})
 
 			// blacklist peers
 			for _, peer := range p.peersList {

share/shwap/p2p/shrex/peers/manager_test.go

@@ -2,6 +2,7 @@ package peers
 
 import (
 	"context"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -287,7 +288,7 @@ func TestManager(t *testing.T) {
 		// check that no peers or hashes were blacklisted
 		manager.params.PoolValidationTimeout = 0
 		require.Len(t, blacklisted, 0)
-		require.Len(t, manager.blacklistedHashes, 0)
+		require.Equal(t, 0, manager.blacklistedHashes.Len())
 	})
 
 	t.Run("pools store window", func(t *testing.T) {
@@ -469,6 +470,27 @@ func TestIntegration(t *testing.T) {
 	})
 }
 
+// TestManager_blacklistedHashesBounded ensures the blacklisted hashes cache
+// does not grow without bound. Previously it was a plain map with no eviction,
+// so a peer spamming invalid datahashes could grow it indefinitely (#1926).
+func TestManager_blacklistedHashesBounded(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+
+	headerSub := newSubLock(testHeader(), nil)
+	manager, err := testManager(ctx, headerSub)
+	require.NoError(t, err)
+
+	// blacklist far more datahashes than the cache bound, as would happen on a
+	// long-running node fed a stream of invalid hashes.
+	for i := range blacklistedHashesCacheSize * 3 {
+		manager.blacklistedHashes.Add(strconv.Itoa(i), struct{}{})
+	}
+
+	require.Equal(t, blacklistedHashesCacheSize, manager.blacklistedHashes.Len(),
+		"blacklisted hashes cache must stay bounded")
+}
+
 func testManager(ctx context.Context, headerSub libhead.Subscriber[*header.ExtendedHeader]) (*Manager, error) {
 	host, err := mocknet.New().GenPeer()
 	if err != nil {

share/shwap/p2p/shrex/peers/metrics.go

@@ -277,6 +277,6 @@ func (m *Manager) shrexPools() map[poolStatus]int64 {
 		shrexPools[poolStatusValidated]++
 	}
 
-	shrexPools[poolStatusBlacklisted] = int64(len(m.blacklistedHashes))
+	shrexPools[poolStatusBlacklisted] = int64(m.blacklistedHashes.Len())
 	return shrexPools
 }