apply bloom filter in api.StreamLogs() (iotexproject#2409)

Author: dustinxie

api/api.go

@@ -653,19 +653,19 @@ func (api *Server) GetLogs(
 	if in.GetFilter() == nil {
 		return nil, status.Error(codes.InvalidArgument, "empty filter")
 	}
+
+	var (
+		startBlock, count uint64
+		err               error
+	)
 	switch {
 	case in.GetByBlock() != nil:
+		count = 1
 		req := in.GetByBlock()
-		h, err := api.dao.GetBlockHeight(hash.BytesToHash256(req.BlockHash))
+		startBlock, err = api.dao.GetBlockHeight(hash.BytesToHash256(req.BlockHash))
 		if err != nil {
 			return nil, status.Error(codes.InvalidArgument, "invalid block hash")
 		}
-		filter, ok := NewLogFilter(in.GetFilter(), nil, nil).(*LogFilter)
-		if !ok {
-			return nil, status.Error(codes.Internal, "cannot convert to *LogFilter")
-		}
-		logs, err := api.getLogsInBlock(filter, h, 1)
-		return &iotexapi.GetLogsResponse{Logs: logs}, err
 	case in.GetByRange() != nil:
 		req := in.GetByRange()
 		if req.FromBlock > api.bc.TipHeight() {
@@ -674,15 +674,14 @@ func (api *Server) GetLogs(
 		if req.Count > 1000 {
 			return nil, status.Error(codes.InvalidArgument, "maximum query range is 1000 blocks")
 		}
-		filter, ok := NewLogFilter(in.GetFilter(), nil, nil).(*LogFilter)
-		if !ok {
-			return nil, status.Error(codes.Internal, "cannot convert to *LogFilter")
-		}
-		logs, err := api.getLogsInBlock(filter, req.FromBlock, req.Count)
-		return &iotexapi.GetLogsResponse{Logs: logs}, err
+		startBlock = req.FromBlock
+		count = req.Count
 	default:
 		return nil, status.Error(codes.InvalidArgument, "invalid GetLogsRequest type")
 	}
+
+	logs, err := api.getLogsInBlock(NewLogFilter(in.GetFilter(), nil, nil), startBlock, count)
+	return &iotexapi.GetLogsResponse{Logs: logs}, err
 }
 
 // StreamBlocks streams blocks

api/logfilter.go

@@ -3,6 +3,7 @@ package api
 import (
 	"bytes"
 
+	"github.com/iotexproject/go-pkgs/bloom"
 	"github.com/iotexproject/iotex-proto/golang/iotexapi"
 	"github.com/iotexproject/iotex-proto/golang/iotextypes"
 	"go.uber.org/zap"
@@ -31,7 +32,7 @@ type LogFilter struct {
 }
 
 // NewLogFilter returns a new log filter
-func NewLogFilter(in *iotexapi.LogsFilter, stream iotexapi.APIService_StreamLogsServer, errChan chan error) Responder {
+func NewLogFilter(in *iotexapi.LogsFilter, stream iotexapi.APIService_StreamLogsServer, errChan chan error) *LogFilter {
 	return &LogFilter{
 		stream:   stream,
 		errChan:  errChan,
@@ -41,6 +42,9 @@ func NewLogFilter(in *iotexapi.LogsFilter, stream iotexapi.APIService_StreamLogs
 
 // Respond to new block
 func (l *LogFilter) Respond(blk *block.Block) error {
+	if !l.ExistInBloomFilter(blk.LogsBloomfilter()) {
+		return nil
+	}
 	logs := l.MatchLogs(blk.Receipts)
 	if len(logs) == 0 {
 		return nil
@@ -117,3 +121,24 @@ func (l *LogFilter) match(log *iotextypes.Log) bool {
 	}
 	return true
 }
+
+// ExistInBloomFilter returns true if topics of filter exist in the bloom filter
+func (l *LogFilter) ExistInBloomFilter(filter bloom.BloomFilter) bool {
+	if filter == nil {
+		return true
+	}
+
+	for _, e := range l.pbFilter.Topics {
+		if e == nil || len(e.Topic) == 0 {
+			continue
+		}
+
+		for _, v := range e.Topic {
+			if filter.Exist(v) {
+				return true
+			}
+		}
+	}
+	// {} or nil matches any address or topic list
+	return len(l.pbFilter.Topics) == 0
+}

api/logfilter_test.go

@@ -3,6 +3,7 @@ package api
 import (
 	"testing"
 
+	"github.com/iotexproject/go-pkgs/bloom"
 	"github.com/iotexproject/go-pkgs/hash"
 	"github.com/iotexproject/iotex-proto/golang/iotexapi"
 	"github.com/stretchr/testify/require"
@@ -77,80 +78,98 @@ var (
 	testData = []struct {
 		log   *action.Log
 		match [5]bool
+		exist [5]bool
 	}{
 		{
 			&action.Log{
 				Address: "topicN",
 				Topics:  []hash.Hash256{dataN}, // both address and topic not exist
 			},
 			[5]bool{true, false, false, false, false},
+			[5]bool{true, true, false, false, false},
 		},
 		{
 			&action.Log{
 				Address: "topicN",
 				Topics:  []hash.Hash256{data1, data2, dataA}, // topic longer than log's topic list
 			},
 			[5]bool{true, false, false, false, false},
+			[5]bool{true, true, true, true, true},
 		},
 		{
 			&action.Log{
 				Address: "topicN",
 				Topics:  []hash.Hash256{data1, dataN}, // topic not match
 			},
 			[5]bool{true, false, false, false, false},
+			[5]bool{true, true, true, true, false},
 		},
 		{
 			&action.Log{
 				Address: "topic1",
 				Topics:  []hash.Hash256{dataN}, // topic not exist
 			},
 			[5]bool{true, true, false, false, false},
+			[5]bool{true, true, false, false, false},
 		},
 		{
 			&action.Log{
 				Address: "topic2",
 				Topics:  []hash.Hash256{dataA, dataB}, // topic not match
 			},
 			[5]bool{true, true, false, false, false},
+			[5]bool{true, true, true, false, true},
 		},
 		{
 			&action.Log{
 				Address: "topicN",
 				Topics:  []hash.Hash256{data1, dataB},
 			},
 			[5]bool{true, false, true, false, false},
+			[5]bool{true, true, true, true, true},
 		},
 		{
 			&action.Log{
 				Address: "topicN",
 				Topics:  []hash.Hash256{data2, dataA},
 			},
 			[5]bool{true, false, true, false, false},
+			[5]bool{true, true, true, true, true},
 		},
 		{
 			&action.Log{
 				Address: "topic1",
 				Topics:  []hash.Hash256{data1, dataN},
 			},
 			[5]bool{true, true, false, true, false},
+			[5]bool{true, true, true, true, false},
 		},
 		{
 			&action.Log{
 				Address: "topicB",
 				Topics:  []hash.Hash256{dataN, dataA},
 			},
 			[5]bool{true, true, false, false, true},
+			[5]bool{true, true, true, false, true},
 		},
 	}
 )
 
 func TestLogFilter_MatchBlock(t *testing.T) {
 	require := require.New(t)
 
+	f := NewLogFilter(testFilter[0], nil, nil)
+	require.True(f.ExistInBloomFilter(nil))
+
 	for i, q := range testFilter {
-		f, ok := NewLogFilter(q, nil, nil).(*LogFilter)
-		require.True(ok)
+		f = NewLogFilter(q, nil, nil)
 		for _, v := range testData {
+			bloom, err := bloom.NewBloomFilter(2048, 3)
+			require.NoError(err)
+			for _, topic := range v.log.Topics {
+				bloom.Add(topic[:])
+			}
+			require.Equal(f.ExistInBloomFilter(bloom), v.exist[i])
 			log := v.log.ConvertToLogPb()
 			require.Equal(f.match(log), v.match[i])
 		}