Bloom filter gossip

go.mod

@@ -4,7 +4,7 @@ go 1.19
 
 require (
 	github.com/VictoriaMetrics/fastcache v1.10.0
-	github.com/ava-labs/avalanchego v1.10.9-rc.4
+	github.com/ava-labs/avalanchego v1.10.10-rc.0
 	github.com/cespare/cp v0.1.0
 	github.com/cockroachdb/pebble v0.0.0-20230209160836-829675f94811
 	github.com/davecgh/go-spew v1.1.1

go.sum

@@ -55,8 +55,8 @@ github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
-github.com/ava-labs/avalanchego v1.10.9-rc.4 h1:vtavPfRiF6r1Zc6RV8/arEfVpe9GQsLWHbMfIWkHbMI=
-github.com/ava-labs/avalanchego v1.10.9-rc.4/go.mod h1:vTBLl1zK36olfLRA7IUfdbvphWqlkuarIoXxvZTHZVw=
+github.com/ava-labs/avalanchego v1.10.10-rc.0 h1:6VjkpwhAJ0tDNJK+UIUD8WIb5VelgH3w61mgk7JAkDQ=
+github.com/ava-labs/avalanchego v1.10.10-rc.0/go.mod h1:C8R5uiltpc8MQ62ixxgODR+15mesWF0aAw3H+Qrl9Iw=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=
 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=

gossip/bloom.go

@@ -0,0 +1,96 @@
+// Copyright (C) 2019-2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package gossip
+
+import (
+	"encoding/binary"
+	"hash"
+	"time"
+
+	safemath "github.com/ava-labs/avalanchego/utils/math"
+	bloomfilter "github.com/holiman/bloomfilter/v2"
+	"golang.org/x/exp/rand"
+)
+
+var _ hash.Hash64 = (*hasher)(nil)
+
+func NewBloomFilter(m uint64, p float64) (*BloomFilter, error) {
+	bloom, err := bloomfilter.NewOptimal(m, p)
+	if err != nil {
+		return nil, err
+	}
+
+	bloomFilter := &BloomFilter{
+		Bloom: bloom,
+		Salt:  randomSalt(),
+	}
+	return bloomFilter, nil
+}
+
+type BloomFilter struct {
+	Bloom *bloomfilter.Filter
+	// Salt is provided to eventually unblock collisions in Bloom. It's possible
+	// that conflicting Gossipable items collide in the bloom filter, so a salt
+	// is generated to eventually resolve collisions.
+	Salt []byte
+}
+
+func (b *BloomFilter) Add(gossipable Gossipable) {
+	salted := hasher{
+		hash: gossipable.GetHash(),
+		salt: b.Salt,
+	}
+	b.Bloom.Add(salted)
+}
+
+func (b *BloomFilter) Has(gossipable Gossipable) bool {
+	salted := hasher{
+		hash: gossipable.GetHash(),
+		salt: b.Salt,
+	}
+	return b.Bloom.Contains(salted)
+}
+
+// ResetBloomFilterIfNeeded resets a bloom filter if it breaches a ratio of
+// filled elements. Returns true if the bloom filter was reset.
+func ResetBloomFilterIfNeeded(
+	bloomFilter *BloomFilter,
+	maxFilledRatio float64,
+) bool {
+	if bloomFilter.Bloom.PreciseFilledRatio() < maxFilledRatio {
+		return false
+	}
+
+	// it's not possible for this to error assuming that the original
+	// bloom filter's parameters were valid
+	fresh, _ := bloomfilter.New(bloomFilter.Bloom.M(), bloomFilter.Bloom.K())
+	bloomFilter.Bloom = fresh
+	bloomFilter.Salt = randomSalt()
+	return true
+}
+
+func randomSalt() []byte {
+	salt := make([]byte, HashLength)
+	r := rand.New(rand.NewSource(uint64(time.Now().Nanosecond())))
+	_, _ = r.Read(salt)
+	return salt
+}
+
+type hasher struct {
+	hash.Hash64
+	hash Hash
+	salt []byte
+}
+
+func (h hasher) Sum64() uint64 {
+	for i := 0; i < safemath.Min(len(h.hash), len(h.salt)); i++ {
+		h.hash[i] ^= h.salt[i]
+	}
+
+	return binary.BigEndian.Uint64(h.hash[:])
+}
+
+func (h hasher) Size() int {
+	return HashLength
+}

gossip/bloom_test.go

@@ -0,0 +1,64 @@
+// Copyright (C) 2019-2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package gossip
+
+import (
+	"testing"
+
+	bloomfilter "github.com/holiman/bloomfilter/v2"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBloomFilterRefresh(t *testing.T) {
+	tests := []struct {
+		name         string
+		refreshRatio float64
+		add          []*testTx
+		expected     []*testTx
+	}{
+		{
+			name:         "no refresh",
+			refreshRatio: 1,
+			add: []*testTx{
+				{hash: Hash{0}},
+			},
+			expected: []*testTx{
+				{hash: Hash{0}},
+			},
+		},
+		{
+			name:         "refresh",
+			refreshRatio: 0.1,
+			add: []*testTx{
+				{hash: Hash{0}},
+				{hash: Hash{1}},
+			},
+			expected: []*testTx{
+				{hash: Hash{1}},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require := require.New(t)
+			b, err := bloomfilter.New(10, 1)
+			require.NoError(err)
+			bloom := BloomFilter{
+				Bloom: b,
+			}
+
+			for _, item := range tt.add {
+				_ = ResetBloomFilterIfNeeded(&bloom, tt.refreshRatio)
+				bloom.Add(item)
+			}
+
+			require.Equal(uint64(len(tt.expected)), bloom.Bloom.N())
+
+			for _, expected := range tt.expected {
+				require.True(bloom.Has(expected))
+			}
+		})
+	}
+}

gossip/gossip.go

@@ -0,0 +1,123 @@
+// Copyright (C) 2019-2023, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package gossip
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/ava-labs/avalanchego/codec"
+	"github.com/ava-labs/avalanchego/ids"
+	"github.com/ava-labs/avalanchego/network/p2p"
+)
+
+// GossipableAny exists to help create non-nil pointers to a concrete Gossipable
+type GossipableAny[T any] interface {
+	*T
+	Gossipable
+}
+
+type Config struct {
+	Frequency time.Duration
+	PollSize  int
+}
+
+func NewGossiper[T any, U GossipableAny[T]](
+	config Config,
+	set Set[U],
+	client *p2p.Client,
+	codec codec.Manager,
+	codecVersion uint16,
+) *Gossiper[T, U] {
+	return &Gossiper[T, U]{
+		config:       config,
+		set:          set,
+		client:       client,
+		codec:        codec,
+		codecVersion: codecVersion,
+	}
+}
+
+type Gossiper[T any, U GossipableAny[T]] struct {
+	config       Config
+	set          Set[U]
+	client       *p2p.Client
+	codec        codec.Manager
+	codecVersion uint16
+}
+
+func (g *Gossiper[T, U]) Gossip(shutdownChan chan struct{}, shutdownWg *sync.WaitGroup) {
+	gossipTicker := time.NewTicker(g.config.Frequency)
+	defer func() {
+		gossipTicker.Stop()
+		shutdownWg.Done()
+	}()
+
+	for {
+		select {
+		case <-gossipTicker.C:
+			if err := g.gossip(); err != nil {
+				log.Warn("failed to gossip", "error", err)
+			}
+		case <-shutdownChan:
+			log.Debug("shutting down gossip")
+			return
+		}
+	}
+}
+
+func (g *Gossiper[T, U]) gossip() error {
+	filter := g.set.GetFilter()
+	bloomBytes, err := filter.Bloom.MarshalBinary()
+	if err != nil {
+		return err
+	}
+
+	request := PullGossipRequest{
+		FilterBytes: bloomBytes,
+		SaltBytes:   filter.Salt,
+	}
+	msgBytes, err := g.codec.Marshal(g.codecVersion, request)
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < g.config.PollSize; i++ {
+		if err := g.client.AppRequestAny(context.TODO(), msgBytes, g.handleResponse); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (g *Gossiper[T, U]) handleResponse(nodeID ids.NodeID, responseBytes []byte, err error) {
+	if err != nil {
+		log.Debug("failed gossip request", "nodeID", nodeID, "error", err)
+		return
+	}
+
+	response := PullGossipResponse{}
+	if _, err := g.codec.Unmarshal(responseBytes, &response); err != nil {
+		log.Debug("failed to unmarshal gossip response", "error", err)
+		return
+	}
+
+	for _, gossipBytes := range response.GossipBytes {
+		gossipable := U(new(T))
+		if err := gossipable.Unmarshal(gossipBytes); err != nil {
+			log.Debug("failed to unmarshal gossip", "error", err, "nodeID", nodeID)
+			continue
+		}
+
+		log.Debug("received gossip", "nodeID", nodeID, "hash", gossipable.GetHash())
+		if err := g.set.Add(gossipable); err != nil {
+			log.Debug("failed to add gossip to the known set", "error", err, "nodeID", nodeID, "id", gossipable.GetHash())
+			continue
+		}
+	}
+}