eth/filters: Simple refactor of log filters to support iterative retrieval

eth/filters/filter.go

@@ -85,8 +85,7 @@ func (f *Filter) SetTopics(topics [][]common.Hash) {
 	f.topics = topics
 }
 
-// Run filters logs with the current parameters set
-func (f *Filter) Find(ctx context.Context) ([]*vm.Log, error) {
+func (f *Filter) FindOnce(ctx context.Context) ([]*vm.Log, error) {
 	head, _ := f.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
 	if head == nil {
 		return nil, nil
@@ -106,12 +105,28 @@ func (f *Filter) Find(ctx context.Context) ([]*vm.Log, error) {
 	// uses the mipmap bloom filters to check for fast inclusion and uses
 	// higher range probability in order to ensure at least a false positive
 	if !f.useMipMap || len(f.addresses) == 0 {
-		return f.getLogs(ctx, beginBlockNo, endBlockNo)
+		logs, blockNumber, err := f.getLogs(ctx, beginBlockNo, endBlockNo)
+		f.begin = int64(blockNumber + 1)
+		return logs, err
 	}
-	return f.mipFind(beginBlockNo, endBlockNo, 0), nil
+	
+	logs, blockNumber := f.mipFind(beginBlockNo, endBlockNo, 0)
+	f.begin = int64(blockNumber + 1)
+	return logs, nil
 }
 
-func (f *Filter) mipFind(start, end uint64, depth int) (logs []*vm.Log) {
+// Run filters logs with the current parameters set
+func (f *Filter) Find(ctx context.Context) (logs []*vm.Log, err error) {
+	for {
+		newLogs, err := f.FindOnce(ctx)
+		if len(newLogs) == 0 || err != nil {
+			return logs, err
+		}
+		logs = append(logs, newLogs...)
+	}
+}
+
+func (f *Filter) mipFind(start, end uint64, depth int) (logs []*vm.Log, blockNumber uint64) {
 	level := core.MIPMapLevels[depth]
 	// normalise numerator so we can work in level specific batches and
 	// work with the proper range checks
@@ -126,27 +141,24 @@ func (f *Filter) mipFind(start, end uint64, depth int) (logs []*vm.Log) {
 				start := uint64(math.Max(float64(num), float64(start)))
 				end := uint64(math.Min(float64(num+level-1), float64(end)))
 				if depth+1 == len(core.MIPMapLevels) {
-					l, _ := f.getLogs(context.Background(), start, end)
-					logs = append(logs, l...)
+					l, blockNumber, _ := f.getLogs(context.Background(), start, end)
+					return l, blockNumber
 				} else {
-					logs = append(logs, f.mipFind(start, end, depth+1)...)
+					return f.mipFind(start, end, depth+1)
 				}
-				// break so we don't check the same range for each
-				// possible address. Checks on multiple addresses
-				// are handled further down the stack.
-				break
 			}
 		}
 	}
 
-	return logs
+	return nil, end
 }
 
-func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []*vm.Log, err error) {
+func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []*vm.Log, blockNumber uint64, err error) {
 	for i := start; i <= end; i++ {
-		header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(i))
+		blockNumber := rpc.BlockNumber(i)
+		header, err := f.backend.HeaderByNumber(ctx, blockNumber)
 		if header == nil || err != nil {
-			return logs, err
+			return logs, end, err
 		}
 
 		// Use bloom filtering to see if this block is interesting given the
@@ -155,17 +167,20 @@ func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []*vm.Log
 			// Get the logs of the block
 			receipts, err := f.backend.GetReceipts(ctx, header.Hash())
 			if err != nil {
-				return nil, err
+				return nil, end, err
 			}
 			var unfiltered []*vm.Log
 			for _, receipt := range receipts {
 				unfiltered = append(unfiltered, ([]*vm.Log)(receipt.Logs)...)
 			}
-			logs = append(logs, filterLogs(unfiltered, nil, nil, f.addresses, f.topics)...)
+			logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics)
+			if len(logs) > 0 {
+				return logs, uint64(blockNumber), nil
+			}
 		}
 	}
 
-	return logs, nil
+	return logs, end, nil
 }
 
 func includes(addresses []common.Address, a common.Address) bool {