refactors

Signed-off-by: jsvisa <delweng@gmail.com>

eth/filters/api.go

@@ -251,13 +251,14 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
 }
 
 // notifier is used for broadcasting data(eg: logs) to rpc receivers
-// used in unit testing
+// used in unit testing.
 type notifier interface {
 	Notify(id rpc.ID, data interface{}) error
 	Closed() <-chan interface{}
 }
 
-// Logs creates a subscription that fires for all new log that match the given filter criteria.
+// Logs creates a subscription that fires for all historical
+// and new logs that match the given filter criteria.
 func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subscription, error) {
 	notifier, supported := rpc.NotifierFromContext(ctx)
 	if !supported {
@@ -268,12 +269,13 @@ func (api *FilterAPI) Logs(ctx context.Context, crit FilterCriteria) (*rpc.Subsc
 	return rpcSub, err
 }
 
-// logs is the inner implementation of logs filtering.
+// logs is the inner implementation of logs subscription.
-// from: nil, to: nil -> only live mode.
+// The following criteria are valid:
-// from: blockNum | safe | finalized, to: nil -> historical beginning at `from` to head, then live mode.
+// * from: nil, to: nil -> yield live logs.
-// Every other case should fail with an error.
+// * from: blockNum | safe | finalized, to: nil -> historical beginning at `from` to head, then live logs.
+// * Every other case should fail with an error.
 func (api *FilterAPI) logs(ctx context.Context, notifier notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
-	if crit.ToBlock != nil && rpc.BlockNumber(crit.ToBlock.Int64()) != rpc.LatestBlockNumber {
+	if crit.ToBlock != nil {
 		return errInvalidToBlock
 	}
 	if crit.FromBlock == nil {
@@ -296,7 +298,7 @@ func (api *FilterAPI) logs(ctx context.Context, notifier notifier, rpcSub *rpc.S
 	return api.histLogs(notifier, rpcSub, int64(from), crit)
 }
 
-// liveLogs only retrieves live logs
+// liveLogs only retrieves live logs.
 func (api *FilterAPI) liveLogs(notify notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
 	matchedLogs := make(chan []*types.Log)
 	logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs)
@@ -324,7 +326,7 @@ func (api *FilterAPI) liveLogs(notify notifier, rpcSub *rpc.Subscription, crit F
 	return nil
 }
 
-// histLogs retrieves the logs older than current header.
+// histLogs retrieves logs older than current header.
 func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from int64, crit FilterCriteria) error {
 	// Subscribe the Live logs
 	// if an ChainReorg occurred,
@@ -352,10 +354,9 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
 	// Compose and notify the logs from liveLogs and histLogs
 	go func() {
 		var (
-			delivered  uint64
+			delivered uint64
-			reorgBlock uint64
+			liveOnly  bool
-			liveOnly   bool
+			reorged   bool
-			reorged    bool
 		)
 		for {
 			select {
@@ -380,9 +381,9 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
 					}
 					continue
 				}
-				reorgBlock = logs[0].BlockNumber
+				reorgBlock := logs[0].BlockNumber
-				if reorgBlock <= delivered {
+				if !reorged && reorgBlock <= delivered {
-					logger.Info("Reorg detected", "reorgBlock", reorgBlock, "delivered", delivered)
+					logger.Debug("Reorg detected", "reorgBlock", reorgBlock, "delivered", delivered)
 					reorged = true
 				}
 				if !reorged {
@@ -430,7 +431,29 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from
 func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<- *types.Log, closeC chan struct{}) error {
 	// The original request ctx will be canceled as soon as the parent goroutine
 	// returns a subscription. Use a new context instead.
-	ctx := context.Background()
+	ctx, cancel := context.WithCancel(context.Background())
+
+	// Fetch logs from a range of blocks.
+	fetchRange := func(from, to int64) error {
+		f := api.sys.NewRangeFilter(from, to, crit.Addresses, crit.Topics)
+		logsCh, errChan := f.rangeLogsAsync(ctx)
+		for {
+			select {
+			case <-closeC:
+				cancel()
+				return nil
+			case log := <-logsCh:
+				histLogs <- log
+			case err := <-errChan:
+				if err != nil {
+					logger.Error("Error while fetching historical logs", "err", err)
+					return err
+				}
+				// Range filter is done.
+				return nil
+			}
+		}
+	}
 
 	for {
 		// Get the latest block header.
@@ -440,31 +463,10 @@ func (api *FilterAPI) doHistLogs(from int64, crit FilterCriteria, histLogs chan<
 		}
 		head := header.Number.Int64()
 		if from > head {
-			logger.Info("Finish historical sync", "from", from, "head", head)
+			logger.Debug("Finish historical sync", "from", from, "head", head)
 			return nil
 		}
-
+		fetchRange(from, head)
-		// Do historical sync beginning at `from` to head.
-		f := api.sys.NewRangeFilter(from, head, crit.Addresses, crit.Topics)
-		logsCh, errChan := f.rangeLogsAsync(ctx)
-
-	FORLOOP:
-		for {
-			select {
-			case <-closeC:
-				return nil
-			case log := <-logsCh:
-				histLogs <- log
-			case err := <-errChan:
-				// Range filter is done or error, let's also stop the reorgLogs subscribe
-				if err != nil {
-					logger.Error("Error while fetching historical logs", "err", err)
-					return err
-				}
-				break FORLOOP
-			}
-		}
-
 		// Move forward to the next batch
 		from = head + 1
 	}

eth/filters/filter_system_test.go

@@ -793,7 +793,7 @@ func TestLogsSubscription(t *testing.T) {
 		expectErr error
 		err       chan error
 	}{
-		// from 0 to latest
+		// from 0
 		{
 			FilterCriteria{FromBlock: big.NewInt(0)},
 			append(allLogs, liveLogs...), newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
@@ -803,12 +803,12 @@ func TestLogsSubscription(t *testing.T) {
 			FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())},
 			nil, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, errInvalidToBlock, nil,
 		},
-		// from 2 to latest
+		// from 2
 		{
 			FilterCriteria{FromBlock: big.NewInt(2)},
 			append(allLogs[1:], liveLogs...), newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
 		},
-		// from latest to latest
+		// live logs
 		{
 			FilterCriteria{},
 			liveLogs, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
@@ -1027,9 +1027,11 @@ func testLogsSubscriptionReorg(t *testing.T, oldChainMaker, newChainMaker, pendi
 	}
 
 	go func() {
-		var fetched []*types.Log
+		var (
-
+			fetched []*types.Log
-		timeout := time.After(3 * time.Second)
+			timeout = time.After(3 * time.Second)
+			reorged = false
+		)
 	fetchLoop:
 		for {
 			select {
@@ -1040,11 +1042,12 @@ func testLogsSubscriptionReorg(t *testing.T, oldChainMaker, newChainMaker, pendi
 				// We halt the sender by blocking Notify(). However sender will already prepare
 				// logs for next block and send as soon as Notify is released. So we do reorg
 				// one block earlier than we intend.
-				if l.BlockNumber == uint64(reorgAt) && l.Index == 0 {
+				if l.BlockNumber == uint64(reorgAt) && !reorged {
 					// signal reorg
 					reorgSignal <- struct{}{}
 					// wait for reorg to happen
 					<-reorgSignal
+					reorged = true
 				}
 			case <-timeout:
 				break fetchLoop
@@ -1102,26 +1105,40 @@ func TestLogsSubscriptionReorgedSimple(t *testing.T) {
 	t.Parallel()
 
 	expected := []*types.Log{
-		// Original chain until block 3
+		// Original chain until block 2
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(1)}, Data: i2h(11).Bytes(), BlockNumber: 1, Index: 0},
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(2)}, Data: i2h(12).Bytes(), BlockNumber: 2, Index: 0},
 
-		// New logs for 4 onwards
+		// New logs for 3 onwards
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(1)}, Data: i2h(103).Bytes(), BlockNumber: 3, Index: 0},
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(2)}, Data: i2h(104).Bytes(), BlockNumber: 4, Index: 0},
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(3)}, Data: i2h(105).Bytes(), BlockNumber: 5, Index: 0},
 		{Address: contract, Topics: []common.Hash{topic, a2h(addr), i2h(4)}, Data: i2h(106).Bytes(), BlockNumber: 6, Index: 0},
 	}
-	testLogsSubscriptionReorg(t, func(i int, b *core.BlockGen) {
+	t.Run("reorg-during-historical", func(t *testing.T) {
-		tx := makeTx(i+1, i+11, b)
+		testLogsSubscriptionReorg(t, func(i int, b *core.BlockGen) {
-		b.AddTx(tx)
+			tx := makeTx(i+1, i+11, b)
-	}, func(i int, b *core.BlockGen) {
+			b.AddTx(tx)
-		tx := makeTx(i+1, i+103, b)
+		}, func(i int, b *core.BlockGen) {
-		b.AddTx(tx)
+			tx := makeTx(i+1, i+103, b)
-	}, func(i int, b *core.BlockGen) {
+			b.AddTx(tx)
-		tx := makeTx(i+1, i+21, b)
+		}, func(i int, b *core.BlockGen) {
-		b.AddTx(tx)
+			tx := makeTx(i+1, i+21, b)
-	}, 5, 4, 1, 2, expected)
+			b.AddTx(tx)
+		}, 5, 4, 1, 2, expected)
+	})
+	t.Run("reorg-after-historical", func(t *testing.T) {
+		testLogsSubscriptionReorg(t, func(i int, b *core.BlockGen) {
+			tx := makeTx(i+1, i+11, b)
+			b.AddTx(tx)
+		}, func(i int, b *core.BlockGen) {
+			tx := makeTx(i+1, i+103, b)
+			b.AddTx(tx)
+		}, func(i int, b *core.BlockGen) {
+			tx := makeTx(i+1, i+21, b)
+			b.AddTx(tx)
+		}, 5, 4, 1, 5, expected)
+	})
 }
 
 func TestLogsSubscriptionReorgOldMore(t *testing.T) {

ethclient/ethclient.go

@@ -476,6 +476,9 @@ func (ec *Client) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuer
 	if err != nil {
 		return nil, err
 	}
+	// TODO(s1na): Log subscription prohibits ToBlock being set.
+	// This will be later re-allowed once it supports historical-only queries.
+	arg["toBlock"] = nil
 	sub, err := ec.c.EthSubscribe(ctx, ch, "logs", arg)
 	if err != nil {
 		// Defensively prefer returning nil interface explicitly on error-path, instead
@@ -486,7 +489,7 @@ func (ec *Client) SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuer
 	return sub, nil
 }
 
-func toFilterArg(q ethereum.FilterQuery) (interface{}, error) {
+func toFilterArg(q ethereum.FilterQuery) (map[string]interface{}, error) {
 	arg := map[string]interface{}{
 		"address": q.Addresses,
 		"topics":  q.Topics,