Logs checking

erigon-lib/txpool/fetch.go

@@ -174,6 +174,7 @@ func (f *Fetch) receiveMessage(ctx context.Context, sentryClient sentry.SentryCl
 		}
 		return err
 	}
+
 	var req *sentry.InboundMessage
 	for req, err = stream.Recv(); ; req, err = stream.Recv() {
 		if err != nil {

turbo/jsonrpc/erigon_receipts.go

@@ -17,23 +17,22 @@
 package jsonrpc
 
 import (
-	"bytes"
 	"context"
-	"encoding/binary"
 	"fmt"
+	"github.com/erigontech/erigon-lib/kv/order"
+	"github.com/erigontech/erigon-lib/kv/rawdbv3"
+	"github.com/erigontech/erigon-lib/log/v3"
+	"github.com/erigontech/erigon/cmd/state/exec3"
 
 	"github.com/RoaringBitmap/roaring"
 	bortypes "github.com/erigontech/erigon/polygon/bor/types"
 
 	"github.com/erigontech/erigon-lib/common"
-	"github.com/erigontech/erigon-lib/common/hexutility"
 	"github.com/erigontech/erigon-lib/kv"
-	"github.com/erigontech/erigon-lib/kv/bitmapdb"
 	"github.com/erigontech/erigon/core/rawdb"
 	"github.com/erigontech/erigon/core/types"
 	"github.com/erigontech/erigon/eth/ethutils"
 	"github.com/erigontech/erigon/eth/filters"
-	"github.com/erigontech/erigon/ethdb/cbor"
 	"github.com/erigontech/erigon/rpc"
 	"github.com/erigontech/erigon/turbo/rpchelper"
 )
@@ -114,99 +113,8 @@ func (api *ErigonImpl) GetLogs(ctx context.Context, crit filters.FilterCriteria)
 	if end > roaring.MaxUint32 {
 		return nil, fmt.Errorf("end (%d) > MaxUint32", end)
 	}
-	blockNumbers := bitmapdb.NewBitmap()
-	defer bitmapdb.ReturnToPool(blockNumbers)
-	if err := applyFilters(blockNumbers, tx, begin, end, crit); err != nil {
-		return nil, err
-	}
-	if blockNumbers.IsEmpty() {
-		return erigonLogs, nil
-	}
-
-	addrMap := make(map[common.Address]struct{}, len(crit.Addresses))
-	for _, v := range crit.Addresses {
-		addrMap[v] = struct{}{}
-	}
-	iter := blockNumbers.Iterator()
-	for iter.HasNext() {
-		if err := ctx.Err(); err != nil {
-			return nil, err
-		}
-
-		blockNumber := uint64(iter.Next())
-		var logIndex uint
-		var txIndex uint
-		var blockLogs []*types.Log
-		it, err := tx.Prefix(kv.Log, hexutility.EncodeTs(blockNumber))
-		if err != nil {
-			return nil, err
-		}
-		for it.HasNext() {
-			k, v, err := it.Next()
-			if err != nil {
-				return erigonLogs, err
-			}
-			var logs types.Logs
-			if err := cbor.Unmarshal(&logs, bytes.NewReader(v)); err != nil {
-				return erigonLogs, fmt.Errorf("receipt unmarshal failed:  %w", err)
-			}
-			for _, log := range logs {
-				log.Index = logIndex
-				logIndex++
-			}
-			filtered := logs.Filter(addrMap, crit.Topics, 0)
-			if len(filtered) == 0 {
-				continue
-			}
-			txIndex = uint(binary.BigEndian.Uint32(k[8:]))
-			for _, log := range filtered {
-				log.TxIndex = txIndex
-			}
-			blockLogs = append(blockLogs, filtered...)
-		}
-		it.Close()
-		if len(blockLogs) == 0 {
-			continue
-		}
 
-		header, err := api._blockReader.HeaderByNumber(ctx, tx, blockNumber)
-		if err != nil {
-			return nil, err
-		}
-		if header == nil {
-			return nil, fmt.Errorf("block header not found: %d", blockNumber)
-		}
-		timestamp := header.Time
-
-		blockHash := header.Hash()
-		body, err := api._blockReader.BodyWithTransactions(ctx, tx, blockHash, blockNumber)
-		if err != nil {
-			return nil, err
-		}
-		if body == nil {
-			return nil, fmt.Errorf("block not found %d", blockNumber)
-		}
-		for _, log := range blockLogs {
-			erigonLog := &types.ErigonLog{}
-			erigonLog.BlockNumber = blockNumber
-			erigonLog.BlockHash = blockHash
-			if log.TxIndex == uint(len(body.Transactions)) {
-				erigonLog.TxHash = bortypes.ComputeBorTxHash(blockNumber, blockHash)
-			} else {
-				erigonLog.TxHash = body.Transactions[log.TxIndex].Hash()
-			}
-			erigonLog.Timestamp = timestamp
-			erigonLog.Address = log.Address
-			erigonLog.Topics = log.Topics
-			erigonLog.Data = log.Data
-			erigonLog.Index = log.Index
-			erigonLog.Removed = log.Removed
-			erigonLog.TxIndex = log.TxIndex
-			erigonLogs = append(erigonLogs, erigonLog)
-		}
-	}
-
-	return erigonLogs, nil
+	return api.getLogsV3(ctx, tx.(kv.TemporalTx), begin, end, crit)
 }
 
 // GetLatestLogs implements erigon_getLatestLogs.
@@ -228,11 +136,14 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri
 		logOptions = filters.DefaultLogFilterOptions()
 	}
 	erigonLogs := types.ErigonLogs{}
-	tx, beginErr := api.db.BeginRo(ctx)
+	db, beginErr := api.db.BeginRo(ctx)
 	if beginErr != nil {
 		return erigonLogs, beginErr
 	}
-	defer tx.Rollback()
+	defer db.Rollback()
+
+	tx := db.(kv.TemporalTx)
+
 	var err error
 	var begin, end uint64 // Filter range: begin-end(from-to). Two limits are included in the filter
 
@@ -274,13 +185,15 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri
 		return nil, fmt.Errorf("end (%d) < begin (%d)", end, begin)
 	}
 
-	blockNumbers := bitmapdb.NewBitmap()
-	defer bitmapdb.ReturnToPool(blockNumbers)
-	if err := applyFilters(blockNumbers, tx, begin, end, crit); err != nil {
-		return erigonLogs, err
+	chainConfig, err := api.chainConfig(ctx, tx)
+	if err != nil {
+		return nil, err
 	}
-	if blockNumbers.IsEmpty() {
-		return erigonLogs, nil
+	exec := exec3.NewTraceWorker(tx, chainConfig, api.engine(), api._blockReader, nil)
+
+	txNumbers, err := applyFiltersV3(tx, begin, end, crit)
+	if err != nil {
+		return erigonLogs, err
 	}
 
 	addrMap := make(map[common.Address]struct{}, len(crit.Addresses))
@@ -295,91 +208,97 @@ func (api *ErigonImpl) GetLatestLogs(ctx context.Context, crit filters.FilterCri
 	}
 
 	// latest logs that match the filter crit
-	iter := blockNumbers.ReverseIterator()
-	var logCount, blockCount uint64
-	for iter.HasNext() {
+	it := rawdbv3.TxNums2BlockNums(tx, txNumbers, order.Asc)
+	defer it.Close()
+
+	var blockHash common.Hash
+	var header *types.Header
+	var logCount, blockCount, timestamp uint64
+
+	for it.HasNext() {
 		if err = ctx.Err(); err != nil {
 			return nil, err
 		}
-
-		blockNumber := uint64(iter.Next())
-		var logIndex uint
-		var txIndex uint
-		var blockLogs []*types.Log
-		it, err := tx.Prefix(kv.Log, hexutility.EncodeTs(blockNumber))
+		txNum, blockNum, txIndex, isFinalTxn, blockNumChanged, err := it.Next()
 		if err != nil {
 			return nil, err
 		}
-		for it.HasNext() {
-			k, v, err := it.Next()
-			if err != nil {
-				return erigonLogs, err
-			}
-			var logs types.Logs
-			if err := cbor.Unmarshal(&logs, bytes.NewReader(v)); err != nil {
-				return erigonLogs, fmt.Errorf("receipt unmarshal failed:  %w", err)
-			}
-			for _, log := range logs {
-				log.Index = logIndex
-				logIndex++
-			}
-			var filtered types.Logs
-			var maxLogCount uint64
-			maxLogCount = 0
-			if logOptions.LogCount != 0 {
-				maxLogCount = logOptions.LogCount - logCount
-			}
+		if isFinalTxn {
+			continue
+		}
 
-			if logOptions.IgnoreTopicsOrder {
-				filtered = logs.CointainTopics(addrMap, topicsMap, maxLogCount)
-			} else {
-				filtered = logs.Filter(addrMap, crit.Topics, maxLogCount)
+		// if block number changed, calculate all related field
+		if blockNumChanged {
+			if header, err = api._blockReader.HeaderByNumber(ctx, tx, blockNum); err != nil {
+				return nil, err
 			}
-			if len(filtered) == 0 {
+			if header == nil {
+				log.Warn("[rpc] header is nil", "blockNum", blockNum)
 				continue
 			}
-			txIndex = uint(binary.BigEndian.Uint32(k[8:]))
-			for i := range filtered {
-				filtered[i].TxIndex = txIndex
-			}
-			for i := len(filtered) - 1; i >= 0; i-- {
-				blockLogs = append(blockLogs, filtered[i])
-				logCount++
-			}
-			if logOptions.LogCount != 0 && logOptions.LogCount <= logCount {
-				break
-			}
+			blockHash = header.Hash()
+			exec.ChangeBlock(header)
+			timestamp = header.Time
 		}
-		it.Close()
-		blockCount++
-		if len(blockLogs) == 0 {
+		var logIndex uint
+		var blockLogs types.Logs
+
+		txn, err := api._txnReader.TxnByIdxInBlock(ctx, tx, blockNum, txIndex)
+		if err != nil {
+			return nil, err
+		}
+		if txn == nil {
 			continue
 		}
 
-		header, err := api._blockReader.HeaderByNumber(ctx, tx, blockNumber)
+		_, err = exec.ExecTxn(txNum, txIndex, txn)
 		if err != nil {
 			return nil, err
 		}
-		if header == nil {
-			return nil, fmt.Errorf("block header not found: %d", blockNumber)
+		blockLogs = exec.GetLogs(txIndex, txn)
+		for _, log := range blockLogs {
+			log.Index = logIndex
+			logIndex++
+		}
+		var filtered types.Logs
+		var maxLogCount uint64
+		maxLogCount = 0
+		if logOptions.LogCount != 0 {
+			maxLogCount = logOptions.LogCount - logCount
+		}
+		if logOptions.IgnoreTopicsOrder {
+			filtered = blockLogs.CointainTopics(addrMap, topicsMap, maxLogCount)
+		} else {
+			filtered = blockLogs.Filter(addrMap, crit.Topics, maxLogCount)
+		}
+		if len(filtered) == 0 {
+			continue
+		}
+		for i := range filtered {
+			filtered[i].TxIndex = uint(txIndex)
+			logCount++
 		}
-		timestamp := header.Time
 
-		blockHash := header.Hash()
+		blockCount++
+
+		if len(blockLogs) == 0 {
+			continue
+		}
 
-		body, err := api._blockReader.BodyWithTransactions(ctx, tx, blockHash, blockNumber)
+		body, err := api._blockReader.BodyWithTransactions(ctx, tx, blockHash, blockNum)
 		if err != nil {
 			return nil, err
 		}
 		if body == nil {
-			return nil, fmt.Errorf("block not found %d", blockNumber)
+			return nil, fmt.Errorf("block not found %d", blockNum)
 		}
-		for _, log := range blockLogs {
+		for _, log := range filtered {
+			println("append to blocklogs")
 			erigonLog := &types.ErigonLog{}
-			erigonLog.BlockNumber = blockNumber
+			erigonLog.BlockNumber = blockNum
 			erigonLog.BlockHash = blockHash
 			if log.TxIndex == uint(len(body.Transactions)) {
-				erigonLog.TxHash = bortypes.ComputeBorTxHash(blockNumber, blockHash)
+				erigonLog.TxHash = bortypes.ComputeBorTxHash(blockNum, blockHash)
 			} else {
 				erigonLog.TxHash = body.Transactions[log.TxIndex].Hash()
 			}

turbo/jsonrpc/erigon_receipts_test.go

@@ -19,6 +19,7 @@ package jsonrpc
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"math/big"
 	"testing"
 
@@ -80,7 +81,7 @@ func TestErigonGetLatestLogs(t *testing.T) {
 	api := NewErigonAPI(newBaseApiForTest(m), db, nil)
 	expectedLogs, _ := api.GetLogs(m.Ctx, filters.FilterCriteria{FromBlock: big.NewInt(0), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())})
 
-	expectedErigonLogs := make([]*types.ErigonLog, 0)
+	expectedErigonLogs := make(types.ErigonLogs, 0)
 	for i := len(expectedLogs) - 1; i >= 0; i-- {
 		expectedErigonLogs = append(expectedErigonLogs, &types.ErigonLog{
 			Address:     expectedLogs[i].Address,
@@ -95,14 +96,29 @@ func TestErigonGetLatestLogs(t *testing.T) {
 			Timestamp:   expectedLogs[i].Timestamp,
 		})
 	}
-	actual, err := api.GetLatestLogs(m.Ctx, filters.FilterCriteria{}, filters.LogFilterOptions{
+	actual, err := api.GetLatestLogs(m.Ctx, filters.FilterCriteria{FromBlock: big.NewInt(0), ToBlock: big.NewInt(rpc.LatestBlockNumber.Int64())}, filters.LogFilterOptions{
 		LogCount: uint64(len(expectedLogs)),
 	})
 	if err != nil {
 		t.Errorf("calling erigon_getLatestLogs: %v", err)
 	}
 	require.NotNil(t, actual)
 	assert.EqualValues(expectedErigonLogs, actual)
+
+	expectedLog := &types.ErigonLog{
+		Address:     libcommon.HexToAddress("0x3CB5b6E26e0f37F2514D45641F15Bd6fEC2E0c4c"),
+		Topics:      []libcommon.Hash{libcommon.HexToHash("0x68f6a0f063c25c6678c443b9a484086f15ba8f91f60218695d32a5251f2050eb")},
+		Data:        []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 151, 160, 176, 241, 203, 220, 75, 75, 222, 127, 170, 33, 171, 34, 107, 143, 20, 185, 234, 201},
+		BlockNumber: 10,
+		TxHash:      libcommon.HexToHash("0xb6449d8e167a8826d050afe4c9f07095236ff769a985f02649b1023c2ded2059"),
+		TxIndex:     0,
+		BlockHash:   libcommon.HexToHash("0x6804117de2f3e6ee32953e78ced1db7b20214e0d8c745a03b8fecf7cc8ee76ef"),
+		Index:       0,
+		Removed:     false,
+		Timestamp:   100,
+	}
+	assert.Equal(expectedLog, actual[0])
+	println(fmt.Sprintf("%+v\n%+v\n", expectedLog, actual[0]))
 }
 
 func TestErigonGetLatestLogsIgnoreTopics(t *testing.T) {