whisper: periodically rebuild bloom filter to remove stale topics

In the current version of whisper there is no way to remove topic from
bloom filter after it was added. It leads to unnecessary accumulation of
such topics over time and undesirable network traffic.

In this change I added an option to periodically rebuild bloom filter
from available filter topics as there is no way to simply revert an update
to bloom filter. But it would be possible if whisper used cuckoo filters.

Current interaction between peers allows to reset bloom filter to any
desirable state, allowing message from old bloom filter for certain time.
At the moment this time is 10s.

whisper/whisperv6/config.go

@@ -16,14 +16,18 @@
 
 package whisperv6
 
+import "time"
+
 // Config represents the configuration state of a whisper node.
 type Config struct {
-	MaxMessageSize     uint32  `toml:",omitempty"`
-	MinimumAcceptedPOW float64 `toml:",omitempty"`
+	MaxMessageSize           uint32        `toml:",omitempty"`
+	MinimumAcceptedPOW       float64       `toml:",omitempty"`
+	BloomFilterRebuildPeriod time.Duration `toml:",omitempty"`
 }
 
 // DefaultConfig represents (shocker!) the default configuration.
 var DefaultConfig = Config{
-	MaxMessageSize:     DefaultMaxMessageSize,
-	MinimumAcceptedPOW: DefaultMinimumPoW,
+	MaxMessageSize:           DefaultMaxMessageSize,
+	MinimumAcceptedPOW:       DefaultMinimumPoW,
+	BloomFilterRebuildPeriod: 0,
 }

whisper/whisperv6/filter.go

@@ -85,6 +85,15 @@ func (fs *Filters) Install(watcher *Filter) (string, error) {
 	return id, err
 }
 
+// Each is a thread safe iterator over all existing filters.
+func (fs *Filters) Each(f func(filter *Filter)) {
+	fs.mutex.RLock()
+	defer fs.mutex.RUnlock()
+	for _, filter := range fs.watchers {
+		f(filter)
+	}
+}
+
 // Uninstall will remove a filter whose id has been specified from
 // the filter collection
 func (fs *Filters) Uninstall(id string) bool {

whisper/whisperv6/whisper.go

@@ -49,12 +49,13 @@ type Statistics struct {
 }
 
 const (
-	maxMsgSizeIdx           = iota // Maximal message length allowed by the whisper node
-	overflowIdx                    // Indicator of message queue overflow
-	minPowIdx                      // Minimal PoW required by the whisper node
-	minPowToleranceIdx             // Minimal PoW tolerated by the whisper node for a limited time
-	bloomFilterIdx                 // Bloom filter for topics of interest for this node
-	bloomFilterToleranceIdx        // Bloom filter tolerated by the whisper node for a limited time
+	maxMsgSizeIdx               = iota // Maximal message length allowed by the whisper node
+	overflowIdx                        // Indicator of message queue overflow
+	minPowIdx                          // Minimal PoW required by the whisper node
+	minPowToleranceIdx                 // Minimal PoW tolerated by the whisper node for a limited time
+	bloomFilterIdx                     // Bloom filter for topics of interest for this node
+	bloomFilterToleranceIdx            // Bloom filter tolerated by the whisper node for a limited time
+	bloomFilterRebuildPeriodIdx        // Used to define period how often bloom filter is rebuild from scratch.
 )
 
 // Whisper represents a dark communication interface through the Ethereum
@@ -111,6 +112,7 @@ func New(cfg *Config) *Whisper {
 	whisper.settings.Store(minPowIdx, cfg.MinimumAcceptedPOW)
 	whisper.settings.Store(maxMsgSizeIdx, cfg.MaxMessageSize)
 	whisper.settings.Store(overflowIdx, false)
+	whisper.settings.Store(bloomFilterRebuildPeriodIdx, cfg.BloomFilterRebuildPeriod)
 
 	// p2p whisper sub protocol handler
 	whisper.protocol = p2p.Protocol{
@@ -236,7 +238,6 @@ func (whisper *Whisper) SetBloomFilter(bloom []byte) error {
 
 	b := make([]byte, bloomFilterSize)
 	copy(b, bloom)
-
 	whisper.settings.Store(bloomFilterIdx, b)
 	whisper.notifyPeersAboutBloomFilterChange(b)
 
@@ -553,6 +554,24 @@ func (whisper *Whisper) Subscribe(f *Filter) (string, error) {
 	return s, err
 }
 
+// rebuildBloomFilter makes a new bloom filter from currently subscribed watchers
+// if new bloom filter is different from current bloom filter - peers will be notified
+func (whisper *Whisper) rebuildBloomFilter() {
+	if isFullNode(whisper.BloomFilter()) {
+		return
+	}
+	aggregate := make([]byte, bloomFilterSize)
+	whisper.filters.Each(func(filter *Filter) {
+		for _, t := range filter.Topics {
+			top := BytesToTopic(t)
+			aggregate = addBloom(aggregate, TopicToBloom(top))
+		}
+	})
+	if !bloomFilterMatch(aggregate, whisper.BloomFilter()) {
+		whisper.SetBloomFilter(aggregate)
+	}
+}
+
 // updateBloomFilter recalculates the new value of bloom filter,
 // and informs the peers if necessary.
 func (whisper *Whisper) updateBloomFilter(f *Filter) {
@@ -868,13 +887,22 @@ func (whisper *Whisper) processQueue() {
 func (whisper *Whisper) update() {
 	// Start a ticker to check for expirations
 	expire := time.NewTicker(expirationCycle)
-
+	defer expire.Stop()
+	// bloomFilterRebuildPeriodIdx is always set in New
+	val, _ := whisper.settings.Load(bloomFilterRebuildPeriodIdx)
+	var rebuildTicker <-chan time.Time
+	if duration := val.(time.Duration); duration > 0 {
+		bloomRebuild := time.NewTicker(val.(time.Duration))
+		defer bloomRebuild.Stop()
+		rebuildTicker = bloomRebuild.C
+	}
 	// Repeat updates until termination is requested
 	for {
 		select {
 		case <-expire.C:
 			whisper.expire()
-
+		case <-rebuildTicker:
+			whisper.rebuildBloomFilter()
 		case <-whisper.quit:
 			return
 		}

whisper/whisperv6/whisper_test.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/crypto/pbkdf2"
 )
 
@@ -892,3 +893,39 @@ func TestBloom(t *testing.T) {
 		t.Fatalf("retireved wrong bloom filter")
 	}
 }
+
+// TestPeriodicFilterRebuild verifies that bloom filter will be periodically
+// rebuild to avoid consuming unnecessary traffic.
+func TestPeriodicBloomFilterRebuild(t *testing.T) {
+	InitSingleTest()
+	w := New(&Config{
+		MaxMessageSize:           DefaultMaxMessageSize,
+		MinimumAcceptedPOW:       DefaultMinimumPoW,
+		BloomFilterRebuildPeriod: 100 * time.Millisecond,
+	})
+	w.Start(nil)
+	defer w.Stop()
+
+	emptyBloom := make([]byte, bloomFilterSize)
+	require.NoError(t, w.SetBloomFilter(emptyBloom))
+
+	f1, err := generateFilter(t, true)
+	require.NoError(t, err)
+	f2, err := generateFilter(t, true)
+	require.NoError(t, err)
+	sub1, err := w.Subscribe(f1)
+	require.NoError(t, err)
+	bloomf1 := w.BloomFilter()
+	sub2, err := w.Subscribe(f2)
+	require.NoError(t, err)
+
+	require.NoError(t, w.Unsubscribe(sub2))
+	time.Sleep(200 * time.Millisecond)
+	require.Equal(t, bloomf1, w.BloomFilter())
+	// tolerance will be updated only after 10s
+	require.Nil(t, w.BloomFilterTolerance())
+
+	require.NoError(t, w.Unsubscribe(sub1))
+	time.Sleep(200 * time.Millisecond)
+	require.Equal(t, emptyBloom, w.BloomFilter())
+}