[Indexer] Pebble storage implementation

integration/generator_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/onflow/flow-evm-gateway/services/logs"
 	"math/big"
+	"os"
 	"testing"
 	"time"
 
@@ -40,8 +41,13 @@ func TestIntegration_TransferValue(t *testing.T) {
 	srv, err := startEmulator()
 	require.NoError(t, err)
 
+	dbDir := "./db-test"
+	defer func() {
+		_ = os.Remove(dbDir)
+	}()
+
 	ctx, cancelIngestion := context.WithCancel(context.Background())
-	blocks, receipts, txs, err := startEventIngestionEngine(ctx)
+	blocks, receipts, txs, err := startEventIngestionEngine(ctx, dbDir)
 	require.NoError(t, err)
 
 	defer func() {
@@ -131,8 +137,13 @@ func TestIntegration_DeployCallContract(t *testing.T) {
 	srv, err := startEmulator()
 	require.NoError(t, err)
 
+	dbDir := "./db-test"
+	defer func() {
+		_ = os.Remove(dbDir)
+	}()
+
 	ctx, cancelIngestion := context.WithCancel(context.Background())
-	blocks, receipts, txs, err := startEventIngestionEngine(ctx)
+	blocks, receipts, txs, err := startEventIngestionEngine(ctx, dbDir)
 	require.NoError(t, err)
 
 	defer func() {

integration/helpers.go

@@ -15,7 +15,7 @@ import (
 	"github.com/onflow/flow-emulator/server"
 	"github.com/onflow/flow-evm-gateway/services/events"
 	"github.com/onflow/flow-evm-gateway/storage"
-	"github.com/onflow/flow-evm-gateway/storage/memory"
+	"github.com/onflow/flow-evm-gateway/storage/pebble"
 	sdk "github.com/onflow/flow-go-sdk"
 	"github.com/onflow/flow-go-sdk/access/grpc"
 	"github.com/onflow/flow-go-sdk/crypto"
@@ -73,7 +73,7 @@ func startEmulator() (*server.EmulatorServer, error) {
 // listening for events.
 // todo for now we return index storage as a way to check the data if it was correctly
 // indexed this will be in future replaced with evm gateway API access
-func startEventIngestionEngine(ctx context.Context) (
+func startEventIngestionEngine(ctx context.Context, dbDir string) (
 	storage.BlockIndexer,
 	storage.ReceiptIndexer,
 	storage.TransactionIndexer,
@@ -90,11 +90,18 @@ func startEventIngestionEngine(ctx context.Context) (
 	}
 
 	subscriber := events.NewRPCSubscriber(client)
-	blocks := memory.NewBlockStorage(memory.WithLatestHeight(blk.Height))
-	receipts := memory.NewReceiptStorage()
-	txs := memory.NewTransactionStorage()
 
-	log := logger.With().Str("component", "ingestion").Logger()
+	log := logger.With().Str("component", "database").Logger()
+	db, err := pebble.New(dbDir, log)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	blocks, err := pebble.NewBlocks(db, pebble.WithInitHeight(blk.Height))
+	receipts := pebble.NewReceipts(db)
+	txs := pebble.NewTransactions(db)
+
+	log = logger.With().Str("component", "ingestion").Logger()
 	engine := events.NewEventIngestionEngine(subscriber, blocks, receipts, txs, log)
 
 	go func() {

models/receipt.go

@@ -0,0 +1,32 @@
+package models
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	gethTypes "github.com/ethereum/go-ethereum/core/types"
+	"math/big"
+)
+
+// StorageReceipt is a receipt representation for storage.
+//
+// This struct copies the geth.Receipt type found here: https://github.com/ethereum/go-ethereum/blob/9bbb9df18549d6f81c3d1f4fc6c65f71bc92490d/core/types/receipt.go#L52
+// the reason is if we use geth.Receipt some values will be skipped when RLP encoding which is because
+// geth node has the data locally, but we don't in evm gateway, so we can not reproduce those values
+// and we need to store them
+type StorageReceipt struct {
+	Type              uint8
+	PostState         []byte
+	Status            uint64
+	CumulativeGasUsed uint64
+	// todo we could skip bloom to optimize storage and dynamically recalculate it
+	Bloom             gethTypes.Bloom
+	Logs              []*gethTypes.Log
+	TxHash            common.Hash
+	ContractAddress   common.Address
+	GasUsed           uint64
+	EffectiveGasPrice *big.Int
+	BlobGasUsed       uint64
+	BlobGasPrice      *big.Int
+	BlockHash         common.Hash
+	BlockNumber       *big.Int
+	TransactionIndex  uint
+}

services/logs/filter.go

@@ -62,7 +62,7 @@ func (r *RangeFilter) Match() ([]*gethTypes.Log, error) {
 
 	logs := make([]*gethTypes.Log, 0)
 	for i, bloom := range blooms {
-		if !bloomMatch(bloom, r.criteria) {
+		if !bloomMatch(*bloom, r.criteria) {
 			continue
 		}
 

storage/index.go

@@ -52,10 +52,11 @@ type ReceiptIndexer interface {
 	GetByBlockHeight(height *big.Int) (*gethTypes.Receipt, error)
 
 	// BloomsForBlockRange returns slice of bloom values and a slice of block heights
-	// corresponding to each item in the bloom slice.
+	// corresponding to each item in the bloom slice. It only matches the blooms between
+	// inclusive start and end block height.
 	// Expected errors:
 	// - errors.InvalidRange if the block by the height was not indexed or if the end and start values are invalid.
-	BloomsForBlockRange(start, end *big.Int) ([]gethTypes.Bloom, []*big.Int, error)
+	BloomsForBlockRange(start, end *big.Int) ([]*gethTypes.Bloom, []*big.Int, error)
 }
 
 type TransactionIndexer interface {

storage/index_testsuite.go

@@ -5,6 +5,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/onflow/flow-evm-gateway/storage/errors"
+	"github.com/onflow/flow-evm-gateway/storage/mocks"
 	"github.com/stretchr/testify/suite"
 	"math/big"
 )
@@ -17,7 +18,7 @@ type BlockTestSuite struct {
 func (b *BlockTestSuite) TestGet() {
 	b.Run("existing block", func() {
 		height := uint64(1)
-		block := newBlock(height)
+		block := mocks.NewBlock(height)
 		err := b.Blocks.Store(block)
 		b.Require().NoError(err)
 
@@ -33,41 +34,70 @@ func (b *BlockTestSuite) TestGet() {
 	})
 
 	b.Run("non-existing block", func() {
-		retBlock, err := b.Blocks.GetByID(common.HexToHash("0x10"))
-		b.Require().Nil(retBlock)
+		// non-existing id
+		bl, err := b.Blocks.GetByID(common.HexToHash("0x10"))
+		b.Require().Nil(bl)
+		b.Require().ErrorIs(err, errors.NotFound)
+
+		// non-existing height
+		bl, err = b.Blocks.GetByHeight(uint64(200))
+		b.Require().Nil(bl)
 		b.Require().ErrorIs(err, errors.NotFound)
 	})
 }
 
 func (b *BlockTestSuite) TestStore() {
-	block := newBlock(10)
+	block := mocks.NewBlock(10)
 
 	b.Run("success", func() {
 		err := b.Blocks.Store(block)
 		b.Require().NoError(err)
+
+		// we allow overwriting blocks to make the actions idempotent
+		err = b.Blocks.Store(block)
+		b.Require().NoError(err)
 	})
 
-	b.Run("failed to store same block", func() {
-		err := b.Blocks.Store(block)
-		b.Require().ErrorIs(err, errors.Duplicate)
+	b.Run("store multiple blocks, and get one", func() {
+		for i := 0; i < 10; i++ {
+			err := b.Blocks.Store(mocks.NewBlock(uint64(10 + i)))
+			b.Require().NoError(err)
+		}
+
+		bl, err := b.Blocks.GetByHeight(15)
+		b.Require().NoError(err)
+
+		id, err := bl.Hash()
+		b.Require().NoError(err)
+		blId, err := b.Blocks.GetByID(id)
+		b.Require().Equal(bl, blId)
 	})
 }
 
 func (b *BlockTestSuite) TestHeights() {
 	b.Run("first height", func() {
-		first, err := b.Blocks.FirstHeight()
-		b.Require().NoError(err)
-		b.Require().Equal(uint64(1), first)
+		for i := 0; i < 5; i++ {
+			first, err := b.Blocks.FirstHeight()
+			b.Require().NoError(err)
+			b.Require().Equal(uint64(1), first)
+
+			// shouldn't affect first height
+			lastHeight := uint64(100 + i)
+			err = b.Blocks.Store(mocks.NewBlock(lastHeight))
+			b.Require().NoError(err)
+		}
 	})
 
 	b.Run("last height", func() {
-		lastHeight := uint64(100)
-		err := b.Blocks.Store(newBlock(lastHeight))
-		b.Require().NoError(err)
-
-		last, err := b.Blocks.LatestHeight()
-		b.Require().NoError(err)
-		b.Require().Equal(lastHeight, last)
+		for i := 0; i < 5; i++ {
+			lastHeight := uint64(100 + i)
+			err := b.Blocks.Store(mocks.NewBlock(lastHeight))
+			b.Require().NoError(err)
+
+			last, err := b.Blocks.LatestHeight()
+			b.Require().NoError(err)
+			b.Require().Equal(lastHeight, last)
+		}
 	})
 }
 
@@ -77,28 +107,23 @@ type ReceiptTestSuite struct {
 }
 
 func (s *ReceiptTestSuite) TestStoreReceipt() {
-	receipt := newReceipt(1, common.HexToHash("0xf1"))
+	receipt := mocks.NewReceipt(1, common.HexToHash("0xf1"))
 
 	s.Run("store receipt successfully", func() {
 		err := s.ReceiptIndexer.Store(receipt)
 		s.Require().NoError(err)
 	})
-
-	s.Run("store duplicate receipt", func() {
-		err := s.ReceiptIndexer.Store(receipt)
-		s.Require().ErrorIs(err, errors.Duplicate)
-	})
 }
 
 func (s *ReceiptTestSuite) TestGetReceiptByTransactionID() {
 	s.Run("existing transaction ID", func() {
-		receipt := newReceipt(2, common.HexToHash("0xf2"))
+		receipt := mocks.NewReceipt(2, common.HexToHash("0xf2"))
 		err := s.ReceiptIndexer.Store(receipt)
 		s.Require().NoError(err)
 
 		retReceipt, err := s.ReceiptIndexer.GetByTransactionID(receipt.TxHash)
 		s.Require().NoError(err)
-		s.Require().Equal(receipt, retReceipt)
+		s.compareReceipts(receipt, retReceipt)
 	})
 
 	s.Run("non-existing transaction ID", func() {
@@ -111,16 +136,16 @@ func (s *ReceiptTestSuite) TestGetReceiptByTransactionID() {
 
 func (s *ReceiptTestSuite) TestGetReceiptByBlockID() {
 	s.Run("existing block ID", func() {
-		receipt := newReceipt(3, common.HexToHash("0x1"))
+		receipt := mocks.NewReceipt(3, common.HexToHash("0x1"))
 		err := s.ReceiptIndexer.Store(receipt)
 		s.Require().NoError(err)
 
 		retReceipt, err := s.ReceiptIndexer.GetByBlockHeight(receipt.BlockNumber)
 		s.Require().NoError(err)
-		s.Require().Equal(receipt, retReceipt)
+		s.compareReceipts(receipt, retReceipt)
 	})
 
-	s.Run("non-existing block ID", func() {
+	s.Run("non-existing block height", func() {
 		retReceipt, err := s.ReceiptIndexer.GetByBlockHeight(big.NewInt(1337))
 		s.Require().Nil(retReceipt)
 		s.Require().ErrorIs(err, errors.NotFound)
@@ -135,7 +160,7 @@ func (s *ReceiptTestSuite) TestBloomsForBlockRange() {
 		testBlooms := make([]*types.Bloom, 0)
 
 		for i := start.Uint64(); i < end.Uint64(); i++ {
-			r := newReceipt(i, common.HexToHash(fmt.Sprintf("0xf1%d", i)))
+			r := mocks.NewReceipt(i, common.HexToHash(fmt.Sprintf("0xf1%d", i)))
 			testBlooms = append(testBlooms, &r.Bloom)
 			err := s.ReceiptIndexer.Store(r)
 			s.Require().NoError(err)
@@ -146,6 +171,8 @@ func (s *ReceiptTestSuite) TestBloomsForBlockRange() {
 		s.Require().Len(blooms, len(testBlooms))
 		s.Require().Len(heights, len(testBlooms))
 		s.Require().Equal(testBlooms, blooms)
+
+		// todo smaller block range
 	})
 
 	s.Run("invalid block range", func() {
@@ -161,40 +188,73 @@ func (s *ReceiptTestSuite) TestBloomsForBlockRange() {
 		start := big.NewInt(100)
 		end := big.NewInt(105)
 		blooms, heights, err := s.ReceiptIndexer.BloomsForBlockRange(start, end)
-		s.Require().NoError(err)
+		s.Require().ErrorIs(err, errors.InvalidRange)
 		s.Require().Nil(blooms)
 		s.Require().Nil(heights)
 	})
 }
 
+func (s *ReceiptTestSuite) compareReceipts(expected *types.Receipt, actual *types.Receipt) {
+	s.Require().Equal(expected.BlockNumber, actual.BlockNumber)
+	s.Require().Equal(expected.TxHash, actual.TxHash)
+	s.Require().Equal(expected.Type, actual.Type)
+	s.Require().Equal(expected.PostState, actual.PostState)
+	s.Require().Equal(expected.Status, actual.Status)
+	s.Require().Equal(expected.CumulativeGasUsed, actual.CumulativeGasUsed)
+	s.Require().Equal(expected.Bloom, actual.Bloom)
+	s.Require().Equal(len(expected.Logs), len(actual.Logs))
+	for i := range expected.Logs {
+		s.Require().Equal(expected.Logs[i], actual.Logs[i])
+	}
+	s.Require().Equal(expected.TxHash, actual.TxHash)
+	s.Require().Equal(expected.ContractAddress, actual.ContractAddress)
+	s.Require().Equal(expected.GasUsed, actual.GasUsed)
+	s.Require().Equal(expected.EffectiveGasPrice, actual.EffectiveGasPrice)
+	s.Require().Equal(expected.BlobGasUsed, actual.BlobGasUsed)
+	s.Require().Equal(expected.BlockHash, actual.BlockHash)
+	s.Require().Equal(expected.BlockNumber, actual.BlockNumber)
+	s.Require().Equal(expected.TransactionIndex, actual.TransactionIndex)
+}
+
 type TransactionTestSuite struct {
 	suite.Suite
 	TransactionIndexer TransactionIndexer
 }
 
 func (s *TransactionTestSuite) TestStoreTransaction() {
-	tx := newTransaction(0)
+	tx := mocks.NewTransaction(0)
 
 	s.Run("store transaction successfully", func() {
 		err := s.TransactionIndexer.Store(tx)
 		s.Require().NoError(err)
 	})
-
-	s.Run("store duplicate transaction", func() {
-		err := s.TransactionIndexer.Store(tx)
-		s.Require().ErrorIs(err, errors.Duplicate)
-	})
 }
 
 func (s *TransactionTestSuite) TestGetTransaction() {
 	s.Run("existing transaction", func() {
-		tx := newTransaction(1)
+		tx := mocks.NewTransaction(1)
 		err := s.TransactionIndexer.Store(tx)
 		s.Require().NoError(err)
 
 		retTx, err := s.TransactionIndexer.Get(tx.Hash())
 		s.Require().NoError(err)
-		s.Require().Equal(tx, retTx)
+		s.Require().Equal(tx.Hash(), retTx.Hash()) // if hashes are equal the data must be equal
+
+		// allow same transaction overwrites
+		s.Require().NoError(s.TransactionIndexer.Store(retTx))
+	})
+
+	s.Run("store multiple transactions and get single", func() {
+		var tx *types.Transaction
+		for i := 0; i < 10; i++ {
+			tx = mocks.NewTransaction(uint64(10 + i))
+			err := s.TransactionIndexer.Store(tx)
+			s.Require().NoError(err)
+		}
+
+		t, err := s.TransactionIndexer.Get(tx.Hash())
+		s.Require().Equal(tx.Hash(), t.Hash())
+		s.Require().NoError(err)
 	})
 
 	s.Run("non-existing transaction", func() {