[utils/bloom] Dynamic Bloom Filter

go.mod

@@ -4,7 +4,7 @@ go 1.20
 
 require (
 	github.com/VictoriaMetrics/fastcache v1.10.0
-	github.com/ava-labs/avalanchego v1.10.18-rc.15
+	github.com/ava-labs/avalanchego v1.10.18-rc.5.0.20240110070430-a905aaea3b06
 	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.18-rc.15 h1:8HVOLBbaZnvE9YNL89DzUk+na9zluAz+GGbAwTXZZQs=
-github.com/ava-labs/avalanchego v1.10.18-rc.15/go.mod h1:FsejaXWTz6rgjpk1dEbXUJvhX/Ip6ADmD4Q8WZHRNuY=
+github.com/ava-labs/avalanchego v1.10.18-rc.5.0.20240110070430-a905aaea3b06 h1:AI9vcuMKNpzMY2iZgOajWZi+kvUDIo+Kn+Gvo5Obg9M=
+github.com/ava-labs/avalanchego v1.10.18-rc.5.0.20240110070430-a905aaea3b06/go.mod h1:EaYuItYMqNAnMqnWBVfhdCuBmL/MXQc87fid8i94h1Y=
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

plugin/evm/gossip.go

@@ -109,7 +109,7 @@ func (tx *GossipAtomicTx) GossipID() ids.ID {
 }
 
 func NewGossipEthTxPool(mempool *txpool.TxPool) (*GossipEthTxPool, error) {
-	bloom, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
+	bloom, err := gossip.NewBloomFilter(txGossipBloomMinTargetElements, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize bloom filter: %w", err)
 	}
@@ -139,10 +139,11 @@ func (g *GossipEthTxPool) Subscribe(ctx context.Context) {
 			return
 		case pendingTxs := <-g.pendingTxs:
 			g.lock.Lock()
+			optimalElements := (g.mempool.PendingSize() + len(pendingTxs.Txs)) * txGossipBloomChurnMultiplier
 			for _, pendingTx := range pendingTxs.Txs {
 				tx := &GossipEthTx{Tx: pendingTx}
 				g.bloom.Add(tx)
-				reset, err := gossip.ResetBloomFilterIfNeeded(g.bloom, txGossipMaxFalsePositiveRate)
+				reset, err := gossip.ResetBloomFilterIfNeeded(g.bloom, optimalElements)
 				if err != nil {
 					log.Error("failed to reset bloom filter", "err", err)
 					continue
@@ -174,7 +175,7 @@ func (g *GossipEthTxPool) Iterate(f func(tx *GossipEthTx) bool) {
 	})
 }
 
-func (g *GossipEthTxPool) GetFilter() ([]byte, []byte, error) {
+func (g *GossipEthTxPool) GetFilter() ([]byte, []byte) {
 	g.lock.RLock()
 	defer g.lock.RUnlock()
 

plugin/evm/mempool.go

@@ -86,7 +86,7 @@ type Mempool struct {
 
 // NewMempool returns a Mempool with [maxSize]
 func NewMempool(ctx *snow.Context, maxSize int, verify func(tx *Tx) error) (*Mempool, error) {
-	bloom, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
+	bloom, err := gossip.NewBloomFilter(txGossipBloomMinTargetElements, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize bloom filter: %w", err)
 	}
@@ -340,7 +340,7 @@ func (m *Mempool) addTx(tx *Tx, force bool) error {
 	}
 
 	m.bloom.Add(&GossipAtomicTx{Tx: tx})
-	reset, err := gossip.ResetBloomFilterIfNeeded(m.bloom, txGossipMaxFalsePositiveRate)
+	reset, err := gossip.ResetBloomFilterIfNeeded(m.bloom, m.length()*txGossipBloomChurnMultiplier)
 	if err != nil {
 		return err
 	}
@@ -375,7 +375,7 @@ func (m *Mempool) Iterate(f func(tx *GossipAtomicTx) bool) {
 	}
 }
 
-func (m *Mempool) GetFilter() ([]byte, []byte, error) {
+func (m *Mempool) GetFilter() ([]byte, []byte) {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
 

plugin/evm/tx_gossip_test.go

@@ -97,10 +97,9 @@ func TestEthTxGossip(t *testing.T) {
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
-	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
-	require.NoError(err)
-	emptyBloomFilterBytes, _, err := emptyBloomFilter.Marshal()
+	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMinTargetElements, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	require.NoError(err)
+	emptyBloomFilterBytes, _ := emptyBloomFilter.Marshal()
 	request := &sdk.PullGossipRequest{
 		Filter: emptyBloomFilterBytes,
 		Salt:   agoUtils.RandomBytes(32),
@@ -225,10 +224,9 @@ func TestAtomicTxGossip(t *testing.T) {
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
-	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
-	require.NoError(err)
-	emptyBloomFilterBytes, _, err := emptyBloomFilter.Marshal()
+	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMinTargetElements, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	require.NoError(err)
+	emptyBloomFilterBytes, _ := emptyBloomFilter.Marshal()
 	request := &sdk.PullGossipRequest{
 		Filter: emptyBloomFilterBytes,
 		Salt:   agoUtils.RandomBytes(32),

plugin/evm/vm.go

@@ -141,15 +141,16 @@ const (
 	atomicTxGossipProtocol = 0x1
 
 	// gossip constants
-	txGossipBloomMaxItems          = 8 * 1024
-	txGossipBloomFalsePositiveRate = 0.01
-	txGossipMaxFalsePositiveRate   = 0.05
-	txGossipTargetMessageSize      = 20 * units.KiB
-	maxValidatorSetStaleness       = time.Minute
-	txGossipThrottlingPeriod       = 10 * time.Second
-	txGossipThrottlingLimit        = 2
-	gossipFrequency                = 10 * time.Second
-	txGossipPollSize               = 10
+	txGossipBloomMinTargetElements       = 1024
+	txGossipBloomTargetFalsePositiveRate = 0.01
+	txGossipBloomResetFalsePositiveRate  = 0.05
+	txGossipBloomChurnMultiplier         = 3
+	txGossipTargetMessageSize            = 20 * units.KiB
+	maxValidatorSetStaleness             = time.Minute
+	txGossipThrottlingPeriod             = 10 * time.Second
+	txGossipThrottlingLimit              = 2
+	gossipFrequency                      = 10 * time.Second
+	txGossipPollSize                     = 10
 )
 
 // Define the API endpoints for the VM

scripts/versions.sh

@@ -1,4 +1,4 @@
 #!/usr/bin/env bash
 
 # Don't export them as they're used in the context of other calls
-avalanche_version=${AVALANCHE_VERSION:-'v1.10.18-rc.15'}
+avalanche_version=${AVALANCHE_VERSION:-'a905aaea3b060d719951920b4d2d5cce38fcd5d2'}