Merge 0859bd75b7 into 8a171dce1f

eth/filters/api.go

@@ -28,9 +28,11 @@ import (
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/core/history"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/internal/ethapi"
+	logger "github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
@@ -43,6 +45,9 @@ var (
 	errPendingLogsUnsupported = errors.New("pending logs are not supported")
 	errExceedMaxTopics        = errors.New("exceed max topics")
 	errExceedMaxAddresses     = errors.New("exceed max addresses")
+	errInvalidFromBlock       = errors.New(`from block can be only a number, or "latest", "safe", "finalized"`)
+	errInvalidToBlock         = errors.New("log subscription does not support history block range")
+	errInvalidFromToBlock     = errors.New("invalid from and to block combination: from > to")
 )
 
 const (
@@ -52,6 +57,11 @@ const (
 	maxTopics = 4
 	// The maximum number of allowed topics within a topic criteria
 	maxSubTopics = 1000
+	// maxTrackedBlocks is the number of block hashes that will be tracked by subscription.
+	maxTrackedBlocks = 32 * 1024
+	// histLogChunkSize is the maximum number of blocks to process in a single chunk
+	// when fetching historical logs to improve responsiveness
+	histLogChunkSize = 4096
 )
 
 // filter is a helper struct that holds meta information over the filter type
@@ -261,38 +271,320 @@ func (api *FilterAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) {
 	return rpcSub, nil
 }
 
-// Logs creates a subscription that fires for all new log that match the given filter criteria.
+// notifier is used for broadcasting data(eg: logs) to rpc receivers
+// used in unit testing.
+type notifier interface {
+	Notify(id rpc.ID, data interface{}) error
+}
+
+// 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 {
 		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
 	}
+	rpcSub := notifier.CreateSubscription()
+	err := api.logs(ctx, notifier, rpcSub, crit)
+	return rpcSub, err
+}
 
-	var (
+// logs is the inner implementation of logs subscription.
-		rpcSub      = notifier.CreateSubscription()
+// The following criteria are valid:
-		matchedLogs = make(chan []*types.Log)
+// * from: nil | latest, 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 {
+		return errInvalidToBlock
+	}
+	if crit.FromBlock == nil {
+		return api.liveLogs(notifier, rpcSub, crit)
+	}
+	from := rpc.BlockNumber(crit.FromBlock.Int64())
+	switch from {
+	case rpc.PendingBlockNumber:
+		return errInvalidFromBlock
+	case rpc.LatestBlockNumber:
+		return api.liveLogs(notifier, rpcSub, crit)
+	case rpc.SafeBlockNumber, rpc.FinalizedBlockNumber:
+		header, err := api.sys.backend.HeaderByNumber(ctx, from)
+		if err != nil {
+			return err
+		}
+		from = rpc.BlockNumber(header.Number.Int64())
+	}
+	if from < 0 {
+		return errInvalidFromBlock
+	}
+	return api.histLogs(notifier, rpcSub, int64(from), crit)
+}
 
+// liveLogs only retrieves live logs.
+func (api *FilterAPI) liveLogs(notifier notifier, rpcSub *rpc.Subscription, crit FilterCriteria) error {
+	matchedLogs := make(chan []*types.Log)
 	logsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), matchedLogs)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	go func() {
-		defer logsSub.Unsubscribe()
 		for {
 			select {
 			case logs := <-matchedLogs:
-				for _, log := range logs {
+				notifyLogsIf(notifier, rpcSub.ID, logs, nil)
-					notifier.Notify(rpcSub.ID, &log)
+
+			case <-rpcSub.Err(): // client send an unsubscribe request
+				logsSub.Unsubscribe()
+				return
 			}
+		}
+	}()
+	return nil
+}
+
+// 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,
+	// we will first recv the old chain's deleted logs in descending order,
+	// and then the new chain's added logs in descending order
+	// see core/blockchain.go#reorg(oldHead *types.Header, newHead *types.Block) for more details
+	// if an reorg happened between `from` and `to`, we will need to think about some scenarios:
+	// 1. if a reorg occurs after the currently delivered block, then because this is happening in the future, has nothing to do with the current historical sync, we can just ignore it.
+	// 2. if a reorg occurs before the currently delivered block, then we need to stop the historical delivery, and send all replaced logs instead
+	var (
+		liveLogs = make(chan []*types.Log)
+		histLogs = make(chan []*types.Log)
+		histDone = make(chan error)
+	)
+	liveLogsSub, err := api.events.SubscribeLogs(ethereum.FilterQuery(crit), liveLogs)
+	if err != nil {
+		return err
+	}
+
+	// The original request ctx will be canceled as soon as the parent goroutine
+	// returns a subscription. Use a new context instead.
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		histDone <- api.doHistLogs(ctx, from, crit.Addresses, crit.Topics, histLogs)
+	}()
+
+	// Compose and notify the logs from liveLogs and histLogs
+	go func() {
+		defer func() {
+			liveLogsSub.Unsubscribe()
+			cancel()
+		}()
+		var (
+			// delivered is the block number of the last historical log delivered.
+			delivered uint64
+
+			// liveMode is true when either:
+			// - all historical logs are delivered.
+			// - or, during history processing a reorg is detected.
+			liveMode bool
+
+			// reorgedBlockHash is the block hash of the reorg point. It is set when
+			// a reorg is detected in the future. It is used to detect if the history
+			// processor is sending stale logs.
+			reorgedBlockHash common.Hash
+
+			// hashes is used to track the hashes of the blocks that have been delivered.
+			// It is used as a guard to prevent duplicate logs as well as inaccurate "removed"
+			// logs being delivered during a reorg.
+			hashes = lru.NewBasicLRU[common.Hash, struct{}](maxTrackedBlocks)
+		)
+		for {
+			select {
+			case err := <-histDone:
+				if err != nil {
+					logger.Warn("History logs delivery failed", "err", err)
+					return
+				}
+				// Else historical logs are all delivered, let's switch to live mode
+				logger.Info("History logs delivery finished, and now enter into live mode", "delivered", delivered)
+				// TODO: It's theoretically possible that we miss logs due to
+				// asynchrony between the history processor and the chain subscription.
+				liveMode = true
+				histLogs = nil
+
+			case logs := <-liveLogs:
+				if len(logs) == 0 {
+					continue
+				}
+				// TODO: further reorgs are possible during history processing.
+				if !liveMode && logs[0].BlockNumber <= delivered {
+					// History is being processed and a reorg is encountered.
+					// From this point we ignore historical logs coming in and
+					// only send logs from the chain subscription.
+					logger.Info("Reorg detected", "reorgBlock", logs[0].BlockNumber, "delivered", delivered)
+					liveMode = true
+				}
+				if !liveMode {
+					if logs[0].Removed && reorgedBlockHash == (common.Hash{}) {
+						// Reorg in future. Remember fork point.
+						reorgedBlockHash = logs[0].BlockHash
+					}
+					// Implicit cases:
+					// - there was a reorg in future and blockchain is sending logs from the new chain.
+					// - history is still being processed and blockchain sends logs from the tip.
+					continue
+				}
+				// Removed logs from reorged chain, replacing logs or logs from tip of the chain.
+				notifyLogsIf(notifier, rpcSub.ID, logs, &hashes)
+
+			case logs := <-histLogs:
+				if len(logs) == 0 {
+					continue
+				}
+				if liveMode {
+					continue
+				}
+				if logs[0].BlockHash == reorgedBlockHash {
+					// We have reached the fork point and the historical producer
+					// is emitting old logs because of delay. Restart the process
+					// from last delivered block.
+					logger.Info("Restarting historical logs delivery", "from", logs[0].BlockNumber, "delivered", delivered)
+					liveLogsSub.Unsubscribe()
+					// Stop hist logs fetcher
+					cancel()
+					if err := api.histLogs(notifier, rpcSub, int64(logs[0].BlockNumber), crit); err != nil {
+						logger.Warn("failed to restart historical logs delivery", "err", err)
+					}
+					return
+				}
+				notifyLogsIf(notifier, rpcSub.ID, logs, &hashes)
+				// Assuming batch = all logs of a single block
+				delivered = logs[0].BlockNumber
+
 			case <-rpcSub.Err(): // client send an unsubscribe request
 				return
 			}
 		}
 	}()
 
-	return rpcSub, nil
+	return nil
+}
+
+// doHistLogs retrieves the logs older than current header, and forward them to the histLogs channel.
+func (api *FilterAPI) doHistLogs(ctx context.Context, from int64, addrs []common.Address, topics [][]common.Hash, histLogs chan<- []*types.Log) error {
+	// Fetch logs from a range of blocks.
+	fetchRange := func(from, to int64) error {
+		f := api.sys.NewRangeFilter(from, to, addrs, topics)
+		logs, err := f.Logs(ctx)
+		if err != nil {
+			return err
+		}
+		// Group logs by block and send them in batches
+		var currentBlockLogs []*types.Log
+		var currentBlockNumber uint64
+
+		for _, log := range logs {
+			if currentBlockNumber != log.BlockNumber && len(currentBlockLogs) > 0 {
+				// Send the current block's logs
+				select {
+				case histLogs <- currentBlockLogs:
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+				currentBlockLogs = nil
+			}
+			currentBlockLogs = append(currentBlockLogs, log)
+			currentBlockNumber = log.BlockNumber
+		}
+		// Send remaining logs if any
+		if len(currentBlockLogs) > 0 {
+			select {
+			case histLogs <- currentBlockLogs:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		return nil
+	}
+
+	for {
+		// Get the latest block header.
+		header := api.sys.backend.CurrentHeader()
+		if header == nil {
+			return errors.New("unexpected error: no header block found")
+		}
+		head := header.Number.Int64()
+		if from > head {
+			logger.Info("Finish historical sync", "from", from, "head", head)
+			return nil
+		}
+
+		to := min(from+histLogChunkSize-1, head)
+
+		logger.Debug("Processing historical logs chunk", "from", from, "to", to, "head", head)
+		if err := fetchRange(from, to); err != nil {
+			if errors.Is(err, context.Canceled) {
+				logger.Warn("Historical logs delivery canceled", "from", from, "to", to)
+				return nil
+			}
+			return err
+		}
+
+		// Move forward to the next chunk
+		from = to + 1
+
+		// Allow other goroutines to run and check for cancellation
+		select {
+		case <-ctx.Done():
+			logger.Warn("Historical logs delivery canceled", "from", from, "to", to)
+			return ctx.Err()
+		default:
+		}
+	}
+}
+
+// notifyLogsIf sends logs to the notifier if the condition is met.
+// It assumes all logs of the same block are either all removed or all added.
+func notifyLogsIf(notifier notifier, id rpc.ID, logs []*types.Log, hashes *lru.BasicLRU[common.Hash, struct{}]) {
+	// Iterate logs and batch them by block hash.
+	type batch struct {
+		start   int
+		end     int
+		hash    common.Hash
+		removed bool
+	}
+	var (
+		batches = make([]batch, 0)
+		h       common.Hash
+	)
+	for i, log := range logs {
+		if h == log.BlockHash {
+			// Skip logs of seen block
+			continue
+		}
+		if len(batches) > 0 {
+			batches[len(batches)-1].end = i
+		}
+		batches = append(batches, batch{start: i, hash: log.BlockHash, removed: log.Removed})
+		h = log.BlockHash
+	}
+	// Close off last batch.
+	if batches[len(batches)-1].end == 0 {
+		batches[len(batches)-1].end = len(logs)
+	}
+	for _, batch := range batches {
+		if hashes != nil {
+			// During reorgs it's possible that logs from the new chain have been delivered.
+			// Avoid sending removed logs from the old chain and duplicate logs from new chain.
+			if batch.removed && !hashes.Contains(batch.hash) {
+				continue
+			}
+			if !batch.removed && hashes.Contains(batch.hash) {
+				continue
+			}
+			hashes.Add(batch.hash, struct{}{})
+		}
+		for _, log := range logs[batch.start:batch.end] {
+			notifier.Notify(id, log)
+		}
+	}
 }
 
 // FilterCriteria represents a request to create a new filter.

eth/filters/filter_system.go

@@ -152,6 +152,8 @@ const (
 	UnknownSubscription Type = iota
 	// LogsSubscription queries for new or removed (chain reorg) logs
 	LogsSubscription
+	// MinedAndPendingLogsSubscription queries for both mined and pending logs
+	MinedAndPendingLogsSubscription
 	// PendingTransactionsSubscription queries for pending transactions entering
 	// the pending state
 	PendingTransactionsSubscription
@@ -327,11 +329,32 @@ func (es *EventSystem) SubscribeLogs(crit ethereum.FilterQuery, logs chan []*typ
 	if from >= 0 && to >= 0 && to >= from {
 		return es.subscribeLogs(crit, logs), nil
 	}
+	// interested in mined logs from a specific block number, new logs and pending logs
+	if (from == rpc.LatestBlockNumber || from >= 0) && to == rpc.PendingBlockNumber {
+		return es.subscribeMinedPendingLogs(crit, logs), nil
+	}
 	// interested in logs from a specific block number to new mined blocks
 	if from >= 0 && to == rpc.LatestBlockNumber {
 		return es.subscribeLogs(crit, logs), nil
 	}
-	return nil, errInvalidBlockRange
+	return nil, errInvalidFromToBlock
+}
+
+// subscribeMinedPendingLogs creates a subscription that returned mined and
+// pending logs that match the given criteria.
+func (es *EventSystem) subscribeMinedPendingLogs(crit ethereum.FilterQuery, logs chan []*types.Log) *Subscription {
+	sub := &subscription{
+		id:        rpc.NewID(),
+		typ:       MinedAndPendingLogsSubscription,
+		logsCrit:  crit,
+		created:   time.Now(),
+		logs:      logs,
+		txs:       make(chan []*types.Transaction),
+		headers:   make(chan *types.Header),
+		installed: make(chan struct{}),
+		err:       make(chan error),
+	}
+	return es.subscribe(sub)
 }
 
 // subscribeLogs creates a subscription that will write all logs matching the

eth/filters/filter_system_test.go

@@ -19,6 +19,7 @@ package filters
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math/big"
 	"reflect"
 	"runtime"
@@ -31,11 +32,14 @@ import (
 	"github.com/ethereum/go-ethereum/core/filtermaps"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/internal/ethapi"
+	logger "github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/triedb"
 )
 
 type testBackend struct {
@@ -167,6 +171,19 @@ func (b *testBackend) NewMatcherBackend() filtermaps.MatcherBackend {
 	return b.fm.NewMatcherBackend()
 }
 
+func (b *testBackend) forwardLogEvents(logCh chan []*types.Log, removedLogCh chan core.RemovedLogsEvent) {
+	go func() {
+		for {
+			select {
+			case logs := <-logCh:
+				b.logsFeed.Send(logs)
+			case logs := <-removedLogCh:
+				b.rmLogsFeed.Send(logs)
+			}
+		}
+	}()
+}
+
 func (b *testBackend) startFilterMaps(history uint64, disabled bool, params filtermaps.Params) {
 	head := b.CurrentBlock()
 	chainView := filtermaps.NewChainView(b, head.Number.Uint64(), head.Hash())
@@ -694,3 +711,516 @@ func TestPendingTxFilterDeadlock(t *testing.T) {
 		}
 	}
 }
+
+type mockNotifier struct {
+	c chan interface{}
+}
+
+func newMockNotifier() *mockNotifier {
+	return &mockNotifier{c: make(chan interface{})}
+}
+
+func (n *mockNotifier) Notify(id rpc.ID, data interface{}) error {
+	n.c <- data
+	return nil
+}
+
+var (
+	signer   = types.HomesteadSigner{}
+	key, _   = crypto.GenerateKey()
+	addr     = crypto.PubkeyToAddress(key.PublicKey)
+	addrHash = common.HexToHash(addr.Hex())
+	contract = common.HexToAddress("0000000000000000000000000000000000031ec7")
+	// Transfer(address indexed from, address indexed to, uint256 value);
+	topic        = common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")
+	genesisAlloc = core.GenesisAlloc{
+		// // SPDX-License-Identifier: GPL-3.0
+		// pragma solidity >=0.7.0 <0.9.0;
+		//
+		// contract Token {
+		//     event Transfer(address indexed from, address indexed to, uint256 value);
+		//     function transfer(address to, uint256 value) public returns (bool) {
+		//         emit Transfer(msg.sender, to, value);
+		//         return true;
+		//     }
+		// }
+		contract: {Balance: big.NewInt(params.Ether), Code: common.FromHex("0x608060405234801561001057600080fd5b506004361061002b5760003560e01c8063a9059cbb14610030575b600080fd5b61004a6004803603810190610045919061016a565b610060565b60405161005791906101c5565b60405180910390f35b60008273ffffffffffffffffffffffffffffffffffffffff163373ffffffffffffffffffffffffffffffffffffffff167fddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef846040516100bf91906101ef565b60405180910390a36001905092915050565b600080fd5b600073ffffffffffffffffffffffffffffffffffffffff82169050919050565b6000610101826100d6565b9050919050565b610111816100f6565b811461011c57600080fd5b50565b60008135905061012e81610108565b92915050565b6000819050919050565b61014781610134565b811461015257600080fd5b50565b6000813590506101648161013e565b92915050565b60008060408385031215610181576101806100d1565b5b600061018f8582860161011f565b92505060206101a085828601610155565b9150509250929050565b60008115159050919050565b6101bf816101aa565b82525050565b60006020820190506101da60008301846101b6565b92915050565b6101e981610134565b82525050565b600060208201905061020460008301846101e0565b9291505056fea2646970667358221220b469033f4b77b9565ee84e0a2f04d496b18160d26034d54f9487e57788fd36d564736f6c63430008120033")},
+		addr:     {Balance: big.NewInt(params.Ether)},
+	}
+)
+
+// TestLogsSubscription tests if a rpc subscription receives the correct logs
+func TestLogsSubscription(t *testing.T) {
+	t.Parallel()
+
+	var (
+		db           = rawdb.NewMemoryDatabase()
+		tdb          = triedb.NewDatabase(db, triedb.HashDefaults)
+		genesis      = &core.Genesis{Config: params.TestChainConfig, Alloc: genesisAlloc}
+		backend, sys = newTestFilterSystem(db, Config{})
+		api          = NewFilterAPI(sys)
+	)
+
+	// Hack: GenerateChainWithGenesis creates a new db.
+	// Commit the genesis manually and use GenerateChain.
+	_, err := genesis.Commit(db, tdb)
+	if err != nil {
+		t.Fatal(err)
+	}
+	blocks, _ := core.GenerateChain(genesis.Config, genesis.ToBlock(), ethash.NewFaker(), db, 4, func(i int, b *core.BlockGen) {
+		tx := makeTx(i+1, i+11, b)
+		b.AddTx(tx)
+	})
+	bc, err := core.NewBlockChain(db, genesis, ethash.NewFaker(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Hack: FilterSystem is using mock backend, i.e. blockchain events will be not received
+	// by it. Forward them manually.
+	var (
+		bcLogsFeed   = make(chan []*types.Log)
+		bcRmLogsFeed = make(chan core.RemovedLogsEvent)
+	)
+	backend.forwardLogEvents(bcLogsFeed, bcRmLogsFeed)
+	bc.SubscribeLogsEvent(bcLogsFeed)
+
+	_, err = bc.InsertChain(blocks)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	backend.startFilterMaps(0, false, filtermaps.RangeTestParams)
+	defer backend.stopFilterMaps()
+
+	// Generate pending block, logs for which
+	// will be sent to subscription feed.
+	_, preceipts := core.GenerateChain(genesis.Config, blocks[len(blocks)-1], ethash.NewFaker(), db, 1, func(i int, gen *core.BlockGen) {
+		// transfer(address to, uint256 value)
+		data := fmt.Sprintf("0xa9059cbb%s%s", common.HexToHash(common.BigToAddress(big.NewInt(int64(i + 1))).Hex()).String()[2:], common.BytesToHash([]byte{byte(i + 11)}).String()[2:])
+		tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{Nonce: uint64(len(blocks) + i), To: &contract, Value: big.NewInt(0), Gas: 46000, GasPrice: gen.BaseFee(), Data: common.FromHex(data)}), signer, key)
+		gen.AddTx(tx)
+	})
+	liveLogs := preceipts[0][0].Logs
+
+	allLogs := []*types.Log{
+		{Address: contract, Topics: []common.Hash{topic, addrHash, i2h(1)}, Data: i2b(11), BlockNumber: 1, BlockTimestamp: 10, BlockHash: blocks[0].Hash(), TxHash: blocks[0].Transactions()[0].Hash()},
+		{Address: contract, Topics: []common.Hash{topic, addrHash, i2h(2)}, Data: i2b(12), BlockNumber: 2, BlockTimestamp: 20, BlockHash: blocks[1].Hash(), TxHash: blocks[1].Transactions()[0].Hash()},
+		{Address: contract, Topics: []common.Hash{topic, addrHash, i2h(3)}, Data: i2b(13), BlockNumber: 3, BlockTimestamp: 30, BlockHash: blocks[2].Hash(), TxHash: blocks[2].Transactions()[0].Hash()},
+		{Address: contract, Topics: []common.Hash{topic, addrHash, i2h(4)}, Data: i2b(14), BlockNumber: 4, BlockTimestamp: 40, BlockHash: blocks[3].Hash(), TxHash: blocks[3].Transactions()[0].Hash()},
+	}
+	testCases := []struct {
+		crit      FilterCriteria
+		expected  []*types.Log
+		notifier  *mockNotifier
+		sub       *rpc.Subscription
+		expectErr error
+		err       chan error
+	}{
+		// from 0
+		{
+			FilterCriteria{FromBlock: big.NewInt(0)},
+			append(allLogs, liveLogs...), newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
+		},
+		// from 1 to latest
+		{
+			FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())},
+			nil, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, errInvalidToBlock, nil,
+		},
+		// from 2
+		{
+			FilterCriteria{FromBlock: big.NewInt(2)},
+			append(allLogs[1:], liveLogs...), newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
+		},
+		// live logs
+		{
+			FilterCriteria{},
+			liveLogs, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
+		},
+		// from 1 to 3
+		{
+			FilterCriteria{FromBlock: big.NewInt(1), ToBlock: big.NewInt(3)},
+			nil, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, errInvalidToBlock, nil,
+		},
+		// from -101 to latest
+		{
+			FilterCriteria{FromBlock: big.NewInt(-101)},
+			nil, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, errInvalidFromBlock, nil,
+		},
+		// from "latest" - should work like live logs only
+		{
+			FilterCriteria{FromBlock: big.NewInt(rpc.LatestBlockNumber.Int64())},
+			liveLogs, newMockNotifier(), &rpc.Subscription{ID: rpc.NewID()}, nil, nil,
+		},
+	}
+
+	// subscribe logs
+	for i, tc := range testCases {
+		testCases[i].err = make(chan error)
+		err := api.logs(context.Background(), tc.notifier, tc.sub, tc.crit)
+		if tc.expectErr != nil {
+			if err == nil {
+				t.Errorf("test %d: want error %v, have nothing", i, tc.expectErr)
+				continue
+			}
+			if !errors.Is(err, tc.expectErr) {
+				t.Errorf("test %d: error mismatch, want %v, have %v", i, tc.expectErr, err)
+			}
+			continue
+		}
+		if err != nil {
+			t.Fatalf("SubscribeLogs %d failed: %v\n", i, err)
+		}
+	}
+
+	// receive logs
+	for n, test := range testCases {
+		i := n
+		tt := test
+		go func() {
+			var fetched []*types.Log
+
+			timeout := time.After(3 * time.Second)
+		fetchLoop:
+			for {
+				select {
+				case log := <-tt.notifier.c:
+					fetched = append(fetched, log.(*types.Log))
+				case <-timeout:
+					break fetchLoop
+				}
+			}
+
+			if len(fetched) != len(tt.expected) {
+				tt.err <- fmt.Errorf("invalid number of logs for case %d, want %d log(s), got %d", i, len(tt.expected), len(fetched))
+				return
+			}
+
+			for l := range fetched {
+				have, want := fetched[l], tt.expected[l]
+				if !reflect.DeepEqual(have, want) {
+					tt.err <- fmt.Errorf("invalid log on index %d for case %d have: %+v want: %+v\n", l, i, have, want)
+					return
+				}
+			}
+			tt.err <- nil
+		}()
+	}
+
+	// Wait for historical logs to be processed.
+	// The reason we need to wait is this test is artificial
+	// and no new block containing the logs is mined.
+	time.Sleep(1 * time.Second)
+	// Send live logs
+	backend.logsFeed.Send(liveLogs)
+
+	for i := range testCases {
+		if err := <-testCases[i].err; err != nil {
+			t.Errorf("test %d failed: %v", i, err)
+		}
+	}
+}
+
+func makeTx(to int, value int, b *core.BlockGen) *types.Transaction {
+	// transfer(address to, uint256 value)
+	data := fmt.Sprintf("0xa9059cbb%s%s", common.HexToHash(common.BigToAddress(big.NewInt(int64(to))).Hex()).String()[2:], common.BytesToHash([]byte{byte(value)}).String()[2:])
+	tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{Nonce: b.TxNonce(addr), To: &contract, Gas: 46000, GasPrice: b.BaseFee(), Data: common.FromHex(data)}), signer, key)
+	return tx
+}
+
+func i2h(i int) common.Hash            { return common.BigToHash(big.NewInt(int64(i))) }
+func a2h(a common.Address) common.Hash { return common.HexToHash(a.Hex()) }
+func i2b(i int) []byte                 { return i2h(i).Bytes() }
+
+func parseTransferLog(log *types.Log) (from, to common.Address, amount uint64) {
+	from = common.BytesToAddress(log.Topics[1].Bytes())
+	to = common.BytesToAddress(log.Topics[2].Bytes())
+	amount = common.BytesToHash(log.Data).Big().Uint64()
+	return
+}
+
+// calculateBalance calculates the address balances of all Transfer events
+// We check with the balance instead of logs because there is a gap between the ChainReorg occurred and detected,
+// so we can't fully control how many logs we'll receive.
+// By checking the balance, we can test with the token transfer amount.
+func calculateBalance(logs []*types.Log) map[common.Address]uint64 {
+	balances := make(map[common.Address]uint64)
+	for _, log := range logs {
+		from, to, amount := parseTransferLog(log)
+
+		if _, ok := balances[from]; !ok {
+			balances[from] = 0
+		}
+		if _, ok := balances[to]; !ok {
+			balances[to] = 0
+		}
+
+		if log.Removed { // revert
+			balances[from] += amount
+			balances[to] -= amount
+		} else {
+			balances[from] -= amount
+			balances[to] += amount
+		}
+	}
+	// Remove zero balances
+	for addr, balance := range balances {
+		if balance == 0 {
+			delete(balances, addr)
+		}
+	}
+	return balances
+}
+
+// TestLogsSubscriptionReorg tests that logs subscription works correctly in case of reorg.
+func testLogsSubscriptionReorg(t *testing.T, oldChainMaker, newChainMaker, pendingMaker func(i int, b *core.BlockGen), oldChainLen, newChainLen, forkAt, reorgAt int, expected []*types.Log) {
+	var (
+		db           = rawdb.NewMemoryDatabase()
+		tdb          = triedb.NewDatabase(db, triedb.HashDefaults)
+		backend, sys = newTestFilterSystem(db, Config{})
+		api          = NewFilterAPI(sys)
+		genesis      = &core.Genesis{Config: params.TestChainConfig, Alloc: genesisAlloc, GasLimit: 10000000000000}
+	)
+
+	// Hack: GenerateChainWithGenesis creates a new db.
+	// Commit the genesis manually and use GenerateChain.
+	_, err := genesis.Commit(db, tdb)
+	if err != nil {
+		t.Fatal(err)
+	}
+	oldChain, _ := core.GenerateChain(genesis.Config, genesis.ToBlock(), ethash.NewFaker(), db, oldChainLen, oldChainMaker)
+	bc, err := core.NewBlockChain(db, genesis, ethash.NewFaker(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Hack: FilterSystem is using mock backend, i.e. blockchain events will be not received
+	// by it. Forward them manually.
+	var (
+		bcLogsFeed   = make(chan []*types.Log)
+		bcRmLogsFeed = make(chan core.RemovedLogsEvent)
+	)
+	backend.forwardLogEvents(bcLogsFeed, bcRmLogsFeed)
+	bc.SubscribeLogsEvent(bcLogsFeed)
+	bc.SubscribeRemovedLogsEvent(bcRmLogsFeed)
+
+	_, err = bc.InsertChain(oldChain)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	backend.startFilterMaps(0, false, filtermaps.RangeTestParams)
+	defer backend.stopFilterMaps()
+
+	newChain, _ := core.GenerateChain(genesis.Config, oldChain[forkAt], ethash.NewFaker(), db, newChainLen, newChainMaker)
+	logger.Info("oldChain/newChain", "oldChain", fmt.Sprintf("%d..%d", oldChain[0].Number(), oldChain[len(oldChain)-1].Number()), "newChain", fmt.Sprintf("%d..%d", newChain[0].Number(), newChain[len(newChain)-1].Number()))
+
+	// Generate pending block, logs for which
+	// will be sent to subscription feed.
+	_, preceipts := core.GenerateChain(genesis.Config, newChain[len(newChain)-1], ethash.NewFaker(), db, 1, pendingMaker)
+	liveLogs := preceipts[0][0].Logs
+
+	expected = append(expected, liveLogs...)
+
+	expectedBalance := calculateBalance(expected)
+
+	// Subscribe to logs
+	var (
+		errc        = make(chan error)
+		notifier    = newMockNotifier()
+		sub         = &rpc.Subscription{ID: rpc.NewID()}
+		crit        = FilterCriteria{FromBlock: big.NewInt(1)}
+		reorgSignal = make(chan struct{})
+	)
+	err = api.logs(context.Background(), notifier, sub, crit)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		var (
+			fetched []*types.Log
+			timeout = time.After(3 * time.Second)
+			reorged = false
+		)
+	fetchLoop:
+		for {
+			select {
+			case log := <-notifier.c:
+				l := log.(*types.Log)
+
+				fetched = append(fetched, l)
+				// 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) && !reorged {
+					// signal reorg
+					reorgSignal <- struct{}{}
+					// wait for reorg to happen
+					<-reorgSignal
+					reorged = true
+				}
+			case <-timeout:
+				break fetchLoop
+			}
+		}
+
+		// for i, log := range fetched {
+		// 	logger.Info("Flog", "i", fmt.Sprintf("%02d", i), "blknum", log.BlockNumber, "index", log.Index, "removed", log.Removed, "to", common.BytesToAddress(log.Topics[2].Bytes()), "amount", common.BytesToHash(log.Data).Big().Uint64())
+		// }
+		//
+		// for i, log := range expected {
+		// 	logger.Debug("Elog", "i", fmt.Sprintf("%02d", i), "blknum", log.BlockNumber, "index", log.Index, "removed", log.Removed, "to", common.BytesToAddress(log.Topics[2].Bytes()), "amount", common.BytesToHash(log.Data).Big().Uint64())
+		// }
+
+		fetchedBalance := calculateBalance(fetched)
+		if len(fetchedBalance) != len(expectedBalance) {
+			errc <- fmt.Errorf("invalid number of balances, have %d, want %d", len(fetchedBalance), len(expectedBalance))
+			logger.Info("balance diff", "fetched", fetchedBalance, "expected", expectedBalance)
+			return
+		}
+
+		diffBalance := make(map[common.Address]struct{ have, want uint64 })
+		for addr, balance := range expectedBalance {
+			if fetchedBalance[addr] != balance {
+				diffBalance[addr] = struct{ have, want uint64 }{fetchedBalance[addr], balance}
+			}
+		}
+		if len(diffBalance) > 0 {
+			errc <- fmt.Errorf("invalid balance detected: %+v\n", diffBalance)
+			return
+		}
+
+		errc <- nil
+	}()
+	<-reorgSignal
+	if n, err := bc.InsertChain(newChain); err != nil {
+		t.Fatalf("failed to insert forked chain at %d: %v", n, err)
+	}
+	reorgSignal <- struct{}{}
+
+	// Wait for historical logs to be processed.
+	// The reason we need to wait is this test is artificial
+	// and no new block containing the logs is mined.
+	time.Sleep(2 * time.Second)
+	// Send live logs
+	backend.logsFeed.Send(liveLogs)
+
+	err = <-errc
+	if err != nil {
+		t.Fatalf("test failed: %v", err)
+	}
+}
+
+func TestLogsSubscriptionReorgedSimple(t *testing.T) {
+	t.Parallel()
+
+	expected := []*types.Log{
+		// 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 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},
+	}
+	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) {
+	t.Parallel()
+
+	txs := 520
+	expected := []*types.Log{}
+	// oldChain
+	for blknum := 1; blknum <= 2; blknum++ {
+		for i := 0; i < txs; i++ {
+			expected = append(expected, &types.Log{
+				Address: contract,
+				Topics:  []common.Hash{topic, a2h(addr), i2h(blknum + i)}, Data: i2h(10 + blknum).Bytes(),
+				BlockNumber: uint64(blknum),
+				Index:       uint(i),
+			})
+		}
+	}
+	expected = append(
+		expected,
+		// newChain
+		[]*types.Log{
+			{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) {
+		for j := 0; j < txs; j++ {
+			tx := makeTx(i+j+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)
+}
+
+func TestLogsSubscriptionReorgNewMore(t *testing.T) {
+	t.Parallel()
+
+	txs := 520
+	// oldChain
+	expected := []*types.Log{
+		{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},
+	}
+	// newChain
+	for blknum := 3; blknum <= 6; blknum++ {
+		for i := 0; i < txs; i++ {
+			expected = append(expected, &types.Log{
+				Address: contract,
+				Topics:  []common.Hash{topic, a2h(addr), i2h(blknum + i - 2)}, Data: i2h(100 + blknum).Bytes(),
+				BlockNumber: uint64(blknum),
+				Index:       uint(i),
+			})
+		}
+	}
+	testLogsSubscriptionReorg(t, func(i int, b *core.BlockGen) {
+		tx := makeTx(i+1, i+11, b)
+		b.AddTx(tx)
+	}, func(i int, b *core.BlockGen) {
+		for j := 0; j < txs; j++ {
+			tx := makeTx(i+j+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)
+}

eth/filters/filter_test.go

@@ -456,7 +456,7 @@ func TestRangeLogs(t *testing.T) {
 	)
 
 	expEvent := func(expEvent int, expFirst, expAfterLast uint64) {
-		exp := rangeLogsTestEvent{expEvent, common.NewRange[uint64](expFirst, expAfterLast-expFirst)}
+		exp := rangeLogsTestEvent{expEvent, common.NewRange(expFirst, expAfterLast-expFirst)}
 		event++
 		ev := <-filter.rangeLogsTestHook
 		if ev != exp {

ethclient/ethclient.go

@@ -476,6 +476,13 @@ 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
+	// For subscriptions, if FromBlock is nil, we want only live logs, not historical logs from genesis
+	if q.FromBlock == nil {
+		arg["fromBlock"] = "latest"
+	}
 	sub, err := ec.c.EthSubscribe(ctx, ch, "logs", arg)
 	if err != nil {
 		// Defensively prefer returning nil interface explicitly on error-path, instead
@@ -486,7 +493,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,