diff --git a/eth/filters/api.go b/eth/filters/api.go index 95b950ead4..f844b90006 100644 --- a/eth/filters/api.go +++ b/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. -// from: nil, to: nil -> only live mode. -// from: blockNum | safe | finalized, to: nil -> historical beginning at `from` to head, then live mode. -// Every other case should fail with an error. +// logs is the inner implementation of logs subscription. +// The following criteria are valid: +// * from: nil, to: nil -> yield live logs. +// * 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 - reorgBlock uint64 - liveOnly bool - reorged bool + delivered uint64 + liveOnly bool + reorged bool ) for { select { @@ -380,9 +381,9 @@ func (api *FilterAPI) histLogs(notifier notifier, rpcSub *rpc.Subscription, from } continue } - reorgBlock = logs[0].BlockNumber - if reorgBlock <= delivered { - logger.Info("Reorg detected", "reorgBlock", reorgBlock, "delivered", delivered) + reorgBlock := logs[0].BlockNumber + if !reorged && reorgBlock <= 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 } - - // 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 - } - } - + fetchRange(from, head) // Move forward to the next batch from = head + 1 } diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index 036b9e82d3..a361487aa7 100644 --- a/eth/filters/filter_system_test.go +++ b/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 - - timeout := time.After(3 * time.Second) + var ( + fetched []*types.Log + 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) { - 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, 2, expected) + t.Run("reorg-during-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, 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) { diff --git a/ethclient/ethclient.go b/ethclient/ethclient.go index 1195929f7d..f4b4a33ee1 100644 --- a/ethclient/ethclient.go +++ b/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,