[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.20240105233604-41587de9c530
 	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.20240105233604-41587de9c530 h1:hIuoG2qeGhEuLXtspStQBidJPLl3jBrTxVe8yXIuXp4=
+github.com/ava-labs/avalanchego v1.10.18-rc.5.0.20240105233604-41587de9c530/go.mod h1:IbHVBINt8W+725XChBtmiUHHCGlVbakvjQYsYaHM4xs=
 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(txGossipBloomMinTargetItems, 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()
+			optimalSize := g.mempool.PendingSize() + len(pendingTxs.Txs) // should never come close to overflowing
 			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, optimalSize)
 				if err != nil {
 					log.Error("failed to reset bloom filter", "err", err)
 					continue

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(txGossipBloomMinTargetItems, 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())
 	if err != nil {
 		return err
 	}

plugin/evm/tx_gossip_test.go

@@ -97,7 +97,7 @@ func TestEthTxGossip(t *testing.T) {
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
-	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
+	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMinTargetItems, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	require.NoError(err)
 	emptyBloomFilterBytes, _, err := emptyBloomFilter.Marshal()
 	require.NoError(err)
@@ -225,7 +225,7 @@ func TestAtomicTxGossip(t *testing.T) {
 	}
 
 	// Ask the VM for any new transactions. We should get nothing at first.
-	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMaxItems, txGossipBloomFalsePositiveRate)
+	emptyBloomFilter, err := gossip.NewBloomFilter(txGossipBloomMinTargetItems, txGossipBloomTargetFalsePositiveRate, txGossipBloomResetFalsePositiveRate)
 	require.NoError(err)
 	emptyBloomFilterBytes, _, err := emptyBloomFilter.Marshal()
 	require.NoError(err)

plugin/evm/vm.go

@@ -141,15 +141,15 @@ 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
+	txGossipBloomMinTargetItems          = 1024
+	txGossipBloomTargetFalsePositiveRate = 0.01
+	txGossipBloomResetFalsePositiveRate  = 0.05
+	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