Provide alternative to gossip.BloomFilter

Author: StephenButtolph

network/p2p/gossip/bloom.go

@@ -18,6 +18,8 @@ import (
 //
 // Invariant: The returned bloom filter is not safe to reset concurrently with
 // other operations. However, it is otherwise safe to access concurrently.
+//
+// Deprecated: [SetWithBloomFilter] should be used to manage bloom filters.
 func NewBloomFilter(
 	registerer prometheus.Registerer,
 	namespace string,
@@ -45,6 +47,7 @@ func NewBloomFilter(
 	return filter, err
 }
 
+// Deprecated: [SetWithBloomFilter] should be used to manage bloom filters.
 type BloomFilter struct {
 	minTargetElements              int
 	targetFalsePositiveProbability float64
@@ -85,6 +88,8 @@ func (b *BloomFilter) Marshal() ([]byte, []byte) {
 // the same [targetFalsePositiveProbability].
 //
 // Returns true if the bloom filter was reset.
+//
+// Deprecated: [SetWithBloomFilter] should be used to manage bloom filters.
 func ResetBloomFilterIfNeeded(
 	bloomFilter *BloomFilter,
 	targetElements int,

network/p2p/gossip/gossip.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/network/p2p"
 	"github.com/ava-labs/avalanchego/snow/engine/common"
+	"github.com/ava-labs/avalanchego/utils/bloom"
 	"github.com/ava-labs/avalanchego/utils/buffer"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/set"
@@ -40,6 +41,7 @@ const (
 var (
 	_ Gossiper = (*ValidatorGossiper)(nil)
 	_ Gossiper = (*PullGossiper[Gossipable])(nil)
+	_ Gossiper = (*PushGossiper[Gossipable])(nil)
 
 	ioTypeLabels   = []string{ioLabel, typeLabel}
 	sentPushLabels = prometheus.Labels{
@@ -75,6 +77,17 @@ var (
 	ErrInvalidRegossipFrequency = errors.New("re-gossip frequency cannot be negative")
 )
 
+// Gossipable is an item that can be gossiped across the network
+type Gossipable interface {
+	GossipID() ids.ID
+}
+
+// Marshaller handles parsing logic for a concrete Gossipable type
+type Marshaller[T Gossipable] interface {
+	MarshalGossip(T) ([]byte, error)
+	UnmarshalGossip([]byte) (T, error)
+}
+
 // Gossiper gossips Gossipables to other nodes
 type Gossiper interface {
 	// Gossip runs a cycle of gossip. Returns an error if we failed to gossip.
@@ -191,7 +204,7 @@ func (v ValidatorGossiper) Gossip(ctx context.Context) error {
 func NewPullGossiper[T Gossipable](
 	log logging.Logger,
 	marshaller Marshaller[T],
-	set Set[T],
+	set PullGossiperSet[T],
 	client *p2p.Client,
 	metrics Metrics,
 	pollSize int,
@@ -206,17 +219,27 @@ func NewPullGossiper[T Gossipable](
 	}
 }
 
+// PullGossiperSet exposes the current bloom filter and allows adding new items
+// that were not included in the filter.
+type PullGossiperSet[T Gossipable] interface {
+	// Add adds a value to the set. Returns an error if v was not added.
+	Add(v T) error
+	// BloomFilter returns the bloom filter and its corresponding salt.
+	BloomFilter() (bloom *bloom.Filter, salt ids.ID)
+}
+
 type PullGossiper[T Gossipable] struct {
 	log        logging.Logger
 	marshaller Marshaller[T]
-	set        Set[T]
+	set        PullGossiperSet[T]
 	client     *p2p.Client
 	metrics    Metrics
 	pollSize   int
 }
 
 func (p *PullGossiper[_]) Gossip(ctx context.Context) error {
-	msgBytes, err := MarshalAppRequest(p.set.GetFilter())
+	bf, salt := p.set.BloomFilter()
+	msgBytes, err := MarshalAppRequest(bf.Marshal(), salt[:])
 	if err != nil {
 		return err
 	}
@@ -293,7 +316,7 @@ func (p *PullGossiper[_]) handleResponse(
 // NewPushGossiper returns an instance of PushGossiper
 func NewPushGossiper[T Gossipable](
 	marshaller Marshaller[T],
-	mempool Set[T],
+	set PushGossiperSet,
 	validators p2p.ValidatorSubset,
 	client *p2p.Client,
 	metrics Metrics,
@@ -320,7 +343,7 @@ func NewPushGossiper[T Gossipable](
 
 	return &PushGossiper[T]{
 		marshaller:           marshaller,
-		set:                  mempool,
+		set:                  set,
 		validators:           validators,
 		client:               client,
 		metrics:              metrics,
@@ -336,10 +359,16 @@ func NewPushGossiper[T Gossipable](
 	}, nil
 }
 
+// PushGossiperSet exposes whether hashes are still included in a set.
+type PushGossiperSet interface {
+	// Has returns true if the hash is in the set.
+	Has(h ids.ID) bool
+}
+
 // PushGossiper broadcasts gossip to peers randomly in the network
 type PushGossiper[T Gossipable] struct {
 	marshaller Marshaller[T]
-	set        Set[T]
+	set        PushGossiperSet
 	validators p2p.ValidatorSubset
 	client     *p2p.Client
 	metrics    Metrics

network/p2p/gossip/gossip_test.go

@@ -20,7 +20,6 @@ import (
 	"github.com/ava-labs/avalanchego/snow/engine/enginetest"
 	"github.com/ava-labs/avalanchego/snow/validators"
 	"github.com/ava-labs/avalanchego/snow/validators/validatorstest"
-	"github.com/ava-labs/avalanchego/utils/bloom"
 	"github.com/ava-labs/avalanchego/utils/constants"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/set"
@@ -69,38 +68,35 @@ func (marshaller) UnmarshalGossip(bytes []byte) (tx, error) {
 
 type setDouble struct {
 	txs   set.Set[tx]
-	bloom *BloomFilter
 	onAdd func(tx tx)
 }
 
-func (s *setDouble) Add(gossipable tx) error {
-	if s.txs.Contains(gossipable) {
-		return fmt.Errorf("%s already present", ids.ID(gossipable))
+func (s *setDouble) Add(t tx) error {
+	if s.txs.Contains(t) {
+		return fmt.Errorf("%s already present", t)
 	}
 
-	s.txs.Add(gossipable)
-	s.bloom.Add(gossipable)
+	s.txs.Add(t)
 	if s.onAdd != nil {
-		s.onAdd(gossipable)
+		s.onAdd(t)
 	}
-
 	return nil
 }
 
-func (s *setDouble) Has(gossipID ids.ID) bool {
-	return s.txs.Contains(tx(gossipID))
+func (s *setDouble) Has(id ids.ID) bool {
+	return s.txs.Contains(tx(id))
 }
 
-func (s *setDouble) Iterate(f func(gossipable tx) bool) {
+func (s *setDouble) Iterate(f func(t tx) bool) {
 	for tx := range s.txs {
 		if !f(tx) {
 			return
 		}
 	}
 }
 
-func (s *setDouble) GetFilter() ([]byte, []byte) {
-	return s.bloom.Marshal()
+func (s *setDouble) Len() int {
+	return s.txs.Len()
 }
 
 func TestGossiperGossip(t *testing.T) {
@@ -175,13 +171,18 @@ func TestGossiperGossip(t *testing.T) {
 			)
 			require.NoError(err)
 
-			responseBloom, err := NewBloomFilter(prometheus.NewRegistry(), "", 1000, 0.01, 0.05)
+			var serverSet setDouble
+			serverMempool, err := NewSetWithBloomFilter(
+				&serverSet,
+				prometheus.NewRegistry(),
+				"",
+				1000,
+				0.01,
+				0.05,
+			)
 			require.NoError(err)
-			responseSet := &setDouble{
-				bloom: responseBloom,
-			}
 			for _, item := range tt.responder {
-				require.NoError(responseSet.Add(item))
+				require.NoError(serverMempool.Add(item))
 			}
 
 			metrics, err := NewMetrics(prometheus.NewRegistry(), "")
@@ -196,7 +197,7 @@ func TestGossiperGossip(t *testing.T) {
 			handler := NewHandler[tx](
 				logging.NoLog{},
 				marshaller,
-				responseSet,
+				serverMempool,
 				metrics,
 				tt.targetResponseSize,
 			)
@@ -218,13 +219,18 @@ func TestGossiperGossip(t *testing.T) {
 			require.NoError(err)
 			require.NoError(requestNetwork.Connected(t.Context(), ids.EmptyNodeID, nil))
 
-			bloom, err := NewBloomFilter(prometheus.NewRegistry(), "", 1000, 0.01, 0.05)
+			var requestSet setDouble
+			clientMempool, err := NewSetWithBloomFilter(
+				&requestSet,
+				prometheus.NewRegistry(),
+				"",
+				1000,
+				0.01,
+				0.05,
+			)
 			require.NoError(err)
-			requestSet := &setDouble{
-				bloom: bloom,
-			}
 			for _, item := range tt.requester {
-				require.NoError(requestSet.Add(item))
+				require.NoError(clientMempool.Add(item))
 			}
 
 			requestClient := requestNetwork.NewClient(
@@ -236,7 +242,7 @@ func TestGossiperGossip(t *testing.T) {
 			gossiper := NewPullGossiper[tx](
 				logging.NoLog{},
 				marshaller,
-				requestSet,
+				clientMempool,
 				requestClient,
 				metrics,
 				1,
@@ -457,20 +463,10 @@ func TestPushGossiperNew(t *testing.T) {
 	}
 }
 
-type fullSet[T Gossipable] struct{}
-
-func (fullSet[T]) Add(T) error {
-	return nil
-}
-
-func (fullSet[T]) Has(ids.ID) bool {
-	return true
-}
-
-func (fullSet[T]) Iterate(func(gossipable T) bool) {}
+type hasFunc func(h ids.ID) bool
 
-func (fullSet[_]) GetFilter() ([]byte, []byte) {
-	return bloom.FullFilter.Marshal(), ids.Empty[:]
+func (f hasFunc) Has(h ids.ID) bool {
+	return f(h)
 }
 
 // Tests that the outgoing gossip is equivalent to what was accumulated
@@ -593,7 +589,9 @@ func TestPushGossiper(t *testing.T) {
 
 			gossiper, err := NewPushGossiper[tx](
 				marshaller,
-				fullSet[tx]{},
+				hasFunc(func(ids.ID) bool {
+					return true // Never remove the items from the set
+				}),
 				validators,
 				client,
 				metrics,

network/p2p/gossip/gossipable.go

@@ -18,10 +18,19 @@ import (
 
 var _ p2p.Handler = (*Handler[Gossipable])(nil)
 
+// HandlerSet exposes the ability to add new values to the set in response to
+// pushed information and for responding to pull requests.
+type HandlerSet[T Gossipable] interface {
+	// Add adds a value to the set. Returns an error if v was not added.
+	Add(v T) error
+	// Iterate iterates over elements until f returns false.
+	Iterate(f func(v T) bool)
+}
+
 func NewHandler[T Gossipable](
 	log logging.Logger,
 	marshaller Marshaller[T],
-	set Set[T],
+	set HandlerSet[T],
 	metrics Metrics,
 	targetResponseSize int,
 ) *Handler[T] {
@@ -39,7 +48,7 @@ type Handler[T Gossipable] struct {
 	p2p.Handler
 	marshaller         Marshaller[T]
 	log                logging.Logger
-	set                Set[T]
+	set                HandlerSet[T]
 	metrics            Metrics
 	targetResponseSize int
 }