use receipt indexer to fetch receipts

Author: CoderZhi

action/receipt.go

@@ -219,6 +219,64 @@ func (receipt *Receipt) TransferLogs(accountContractAddr string, logIndex uint32
 	return logs, nil
 }
 
+func (receipt *Receipt) CloneFixed() *Receipt {
+	var cls []*Log
+	if receipt.logs != nil {
+		cls = make([]*Log, len(receipt.logs))
+		for i, l := range receipt.logs {
+			data := make([]byte, len(l.Data))
+			copy(data, l.Data)
+			topics := make([]hash.Hash256, len(l.Topics))
+			for j, topic := range l.Topics {
+				copy(topics[j][:], topic[:])
+			}
+			cls[i] = &Log{
+				Address:     l.Address,
+				Topics:      topics,
+				Data:        data,
+				BlockHeight: l.BlockHeight,
+				ActionHash:  l.ActionHash,
+				Index:       l.Index,
+				TxIndex:     l.TxIndex,
+			}
+		}
+	}
+	var ctls []*TransactionLog
+	if receipt.transactionLogs != nil {
+		ctls = make([]*TransactionLog, len(receipt.transactionLogs))
+		for i, tl := range receipt.transactionLogs {
+			ctls[i] = &TransactionLog{
+				Type:      tl.Type,
+				Amount:    new(big.Int).Set(tl.Amount),
+				Sender:    tl.Sender,
+				Recipient: tl.Recipient,
+			}
+		}
+	}
+	var blobGasPrice *big.Int
+	if receipt.BlobGasPrice != nil {
+		blobGasPrice = new(big.Int).Set(receipt.BlobGasPrice)
+	}
+	var effectiveGasPrice *big.Int
+	if receipt.EffectiveGasPrice != nil {
+		effectiveGasPrice = new(big.Int).Set(receipt.EffectiveGasPrice)
+	}
+	return &Receipt{
+		Status:             receipt.Status,
+		BlockHeight:        receipt.BlockHeight,
+		ActionHash:         receipt.ActionHash,
+		GasConsumed:        receipt.GasConsumed,
+		BlobGasUsed:        receipt.BlobGasUsed,
+		BlobGasPrice:       blobGasPrice,
+		ContractAddress:    receipt.ContractAddress,
+		TxIndex:            receipt.TxIndex,
+		EffectiveGasPrice:  effectiveGasPrice,
+		logs:               cls,
+		transactionLogs:    ctls,
+		executionRevertMsg: receipt.executionRevertMsg,
+	}
+}
+
 // ConvertToLogPb converts a Log to protobuf's Log
 func (log *Log) ConvertToLogPb() *iotextypes.Log {
 	l := &iotextypes.Log{}

blockchain/blockdao/blockdao.go

@@ -63,17 +63,18 @@ type (
 	}
 
 	blockDAO struct {
-		blockStore   BlockStore
-		blobStore    BlobStore
-		indexers     []BlockIndexer
-		timerFactory *prometheustimer.TimerFactory
-		lifecycle    lifecycle.Lifecycle
-		headerCache  cache.LRUCache
-		footerCache  cache.LRUCache
-		receiptCache cache.LRUCache
-		blockCache   cache.LRUCache
-		txLogCache   cache.LRUCache
-		tipHeight    uint64
+		blockStore     BlockStore
+		blobStore      BlobStore
+		receiptIndexer *ReceiptIndexer
+		indexers       []BlockIndexer
+		timerFactory   *prometheustimer.TimerFactory
+		lifecycle      lifecycle.Lifecycle
+		headerCache    cache.LRUCache
+		footerCache    cache.LRUCache
+		receiptCache   cache.LRUCache
+		blockCache     cache.LRUCache
+		txLogCache     cache.LRUCache
+		tipHeight      uint64
 	}
 )
 
@@ -85,6 +86,13 @@ func WithBlobStore(bs BlobStore) Option {
 	}
 }
 
+// WithReceiptIndexer adds receipt indexer to block DAO
+func WithReceiptIndexer(ri *ReceiptIndexer) Option {
+	return func(dao *blockDAO) {
+		dao.receiptIndexer = ri
+	}
+}
+
 // NewBlockDAOWithIndexersAndCache returns a BlockDAO with indexers which will consume blocks appended, and
 // caches which will speed up reading
 func NewBlockDAOWithIndexersAndCache(blkStore BlockStore, indexers []BlockIndexer, cacheSize int, opts ...Option) BlockDAO {
@@ -103,6 +111,9 @@ func NewBlockDAOWithIndexersAndCache(blkStore BlockStore, indexers []BlockIndexe
 	if blockDAO.blobStore != nil {
 		blockDAO.lifecycle.Add(blockDAO.blobStore)
 	}
+	if blockDAO.receiptIndexer != nil {
+		blockDAO.lifecycle.Add(blockDAO.receiptIndexer)
+	}
 	for _, indexer := range indexers {
 		blockDAO.lifecycle.Add(indexer)
 	}
@@ -307,9 +318,23 @@ func (dao *blockDAO) GetReceipts(height uint64) ([]*action.Receipt, error) {
 	_cacheMtc.WithLabelValues("miss_receipts").Inc()
 	timer := dao.timerFactory.NewTimer("get_receipt")
 	defer timer.End()
-	receipts, err := dao.blockStore.GetReceipts(height)
-	if err != nil {
-		return nil, err
+	var receipts []*action.Receipt
+	var err error
+	if dao.receiptIndexer != nil {
+		receipts, err = dao.receiptIndexer.Receipts(height)
+		switch errors.Cause(err) {
+		case nil:
+		case ErrIndexOutOfRange, db.ErrNotExist, db.ErrBucketNotExist:
+			receipts = nil
+		default:
+			return nil, err
+		}
+	}
+	if receipts == nil {
+		receipts, err = dao.blockStore.GetReceipts(height)
+		if err != nil {
+			return nil, err
+		}
 	}
 	tlogs, err := dao.blockStore.TransactionLogs(height)
 	if err != nil {

blockchain/blockdao/receiptindexer.go

@@ -0,0 +1,142 @@
+// Copyright (c) 2024 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package blockdao
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/iotexproject/iotex-core/v2/action"
+	"github.com/iotexproject/iotex-core/v2/blockchain/block"
+	"github.com/iotexproject/iotex-core/v2/db"
+	"github.com/iotexproject/iotex-core/v2/db/batch"
+	"github.com/iotexproject/iotex-core/v2/pkg/util/byteutil"
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+	"github.com/pkg/errors"
+	"google.golang.org/protobuf/proto"
+)
+
+type (
+	// ReceiptIndexer defines a struct to store correct receipts for poluted blocks
+	ReceiptIndexer struct {
+		mu        sync.RWMutex
+		height    uint64
+		endHeight uint64
+		kvstore   db.KVStore
+	}
+)
+
+const (
+	_receiptsNS    = "receipts"
+	_receiptMetaNS = "meta"
+)
+
+var _heightKey = []byte("height")
+
+// ErrIndexOutOfRange indicates the receipt does not exist in the indexer
+var ErrIndexOutOfRange = errors.New("index out of range")
+
+// NewReceiptIndexer creates a new receipt indexer
+func NewReceiptIndexer(kvstore db.KVStore, endHeight uint64) *ReceiptIndexer {
+	return &ReceiptIndexer{
+		endHeight: endHeight,
+		kvstore:   kvstore,
+	}
+}
+
+// Start starts the receipt indexer
+func (ri *ReceiptIndexer) Start(ctx context.Context) error {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	if err := ri.kvstore.Start(ctx); err != nil {
+		return err
+	}
+	value, err := ri.kvstore.Get(_receiptMetaNS, _heightKey)
+	switch errors.Cause(err) {
+	case nil:
+	case db.ErrNotExist, db.ErrBucketNotExist:
+		return nil
+	default:
+		return err
+	}
+	ri.height = byteutil.BytesToUint64(value)
+	return nil
+}
+
+// Stop stops the receipt indexer
+func (ri *ReceiptIndexer) Stop(ctx context.Context) error {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+
+	return ri.kvstore.Stop(ctx)
+}
+
+// Height returns the end height of the receipt indexer
+func (ri *ReceiptIndexer) Height() (uint64, error) {
+	ri.mu.RLock()
+	defer ri.mu.RUnlock()
+	return ri.height, nil
+}
+
+// PutBlock puts the receipts of the block into kvstore
+func (ri *ReceiptIndexer) PutBlock(ctx context.Context, blk *block.Block) error {
+	height := blk.Height()
+	if height > ri.endHeight {
+		return nil
+	}
+	key := byteutil.Uint64ToBytes(height)
+	if blk.Receipts == nil {
+		return errors.Errorf("receipts of block %d is nil", height)
+	}
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	logIndex := uint32(0)
+	receipts := iotextypes.Receipts{}
+	for i, receipt := range blk.Receipts {
+		cr := receipt.CloneFixed()
+		logIndex = cr.UpdateIndex(uint32(i), logIndex)
+		receipts.Receipts = append(receipts.Receipts, cr.ConvertToReceiptPb())
+	}
+	receiptsBytes, err := proto.Marshal(&receipts)
+	if err != nil {
+		return err
+	}
+
+	b := batch.NewBatch()
+	b.Put(_receiptsNS, key, receiptsBytes, fmt.Sprintf("failed to write receipts for block %d", height))
+	if height > ri.height {
+		ri.height = height
+		b.Put(_receiptMetaNS, _heightKey, key, "failed to write receipt indexer height")
+	}
+	return ri.kvstore.WriteBatch(b)
+}
+
+// Receipts returns the receipts of the block at the given height
+func (ri *ReceiptIndexer) Receipts(height uint64) ([]*action.Receipt, error) {
+	key := byteutil.Uint64ToBytes(height)
+	ri.mu.RLock()
+	defer ri.mu.RUnlock()
+	if height > ri.endHeight {
+		return nil, errors.Wrapf(ErrIndexOutOfRange, "height %d > endHeight %d", height, ri.endHeight)
+	}
+	value, err := ri.kvstore.Get(_receiptsNS, key)
+	if err != nil {
+		return nil, err
+	}
+	receiptsPb := &iotextypes.Receipts{}
+	if err := proto.Unmarshal(value, receiptsPb); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal block receipts")
+	}
+
+	var receipts []*action.Receipt
+	for _, receiptPb := range receiptsPb.Receipts {
+		receipt := &action.Receipt{}
+		receipt.ConvertFromReceiptPb(receiptPb)
+		receipts = append(receipts, receipt)
+	}
+	return receipts, nil
+}

blockchain/blockdao/receiptindexer_test.go

@@ -0,0 +1,105 @@
+// Copyright (c) 2024 IoTeX Foundation
+// This source code is provided 'as is' and no warranties are given as to title or non-infringement, merchantability
+// or fitness for purpose and, to the extent permitted by law, all liability for your use of the code is disclaimed.
+// This source code is governed by Apache License 2.0 that can be found in the LICENSE file.
+
+package blockdao
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/iotexproject/go-pkgs/hash"
+	"github.com/iotexproject/iotex-core/v2/action"
+	"github.com/iotexproject/iotex-core/v2/blockchain/block"
+	"github.com/iotexproject/iotex-core/v2/db"
+	"github.com/iotexproject/iotex-core/v2/test/identityset"
+	"github.com/iotexproject/iotex-core/v2/testutil"
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+)
+
+func TestReceiptIndexer(t *testing.T) {
+	r := require.New(t)
+	ctx := context.Background()
+	testPath, err := testutil.PathOfTempFile("test-receipt-indexer")
+	r.NoError(err)
+	defer func() {
+		testutil.CleanupPath(testPath)
+	}()
+	cfg := db.DefaultConfig
+	cfg.DbPath = testPath
+	kvs := db.NewBoltDB(cfg)
+	ri := NewReceiptIndexer(kvs, 3)
+	r.NoError(ri.Start(ctx))
+	for i := uint64(1); i <= 3; i++ {
+		blk1 := createTestBlock(r, i, int(i))
+		r.NoError(ri.PutBlock(ctx, blk1))
+		height, err := ri.Height()
+		r.NoError(err)
+		r.EqualValues(i, height)
+		receipts, err := ri.Receipts(i)
+		r.NoError(err)
+		r.Equal(int(i), len(receipts))
+		for j, rcpt := range receipts {
+			r.Equal(len(blk1.Receipts[j].Logs()), len(rcpt.Logs()))
+			for k, lg := range rcpt.Logs() {
+				r.NotEqual(blk1.Receipts[j].Logs()[k].NotFixTopicCopyBug, lg.NotFixTopicCopyBug)
+				for m, n := range blk1.Receipts[j].Logs()[k].Topics {
+					r.Equal(n, lg.Topics[m])
+				}
+				r.Equal(blk1.Receipts[j].Logs()[k].Data, lg.Data)
+			}
+		}
+	}
+	r.NoError(ri.PutBlock(ctx, createTestBlock(r, 4, 4)))
+	r.NoError(ri.Stop(ctx))
+	r.NoError(ri.Start(ctx))
+	r.EqualValues(3, ri.height)
+	r.NoError(ri.Stop(ctx))
+}
+
+func createTestBlock(require *require.Assertions, height uint64, numOfReceipts int) *block.Block {
+	pb := &iotextypes.BlockHeader{
+		Core: &iotextypes.BlockHeaderCore{
+			Height:    height,
+			Timestamp: timestamppb.Now(),
+		},
+		ProducerPubkey: identityset.PrivateKey(1).PublicKey().Bytes(),
+	}
+	blk := &block.Block{}
+	require.NoError(blk.LoadFromBlockHeaderProto(pb))
+	receipts := make([]*action.Receipt, numOfReceipts)
+	for i := 0; i < numOfReceipts; i++ {
+		actionHash := hash.BytesToHash256([]byte{byte(i)})
+		receipts[i] = &action.Receipt{
+			Status:            uint64(iotextypes.ReceiptStatus_Success),
+			BlockHeight:       height,
+			ActionHash:        actionHash,
+			GasConsumed:       uint64(10000 + i),
+			BlobGasUsed:       0,
+			BlobGasPrice:      big.NewInt(0),
+			ContractAddress:   "io",
+			TxIndex:           uint32(i),
+			EffectiveGasPrice: big.NewInt(int64(i)),
+		}
+		topics := make([]hash.Hash256, i+1)
+		for j := 0; j <= i; j++ {
+			topics[j] = hash.BytesToHash256([]byte{byte(10*i + j)})
+		}
+		receipts[i].AddLogs(&action.Log{
+			Address:            "io",
+			Topics:             topics,
+			Data:               []byte{byte(i)},
+			BlockHeight:        height,
+			ActionHash:         actionHash,
+			Index:              uint32(i),
+			NotFixTopicCopyBug: true,
+		})
+	}
+	blk.Receipts = receipts
+	return blk
+}