Implement StorageProvider interface

api/api.go

@@ -1101,7 +1101,11 @@ func (b *BlockChainAPI) getBlockNumber(blockNumberOrHash *rpc.BlockNumberOrHash)
 		return 0, err
 	}
 	if number, ok := blockNumberOrHash.Number(); ok {
-		return number.Int64(), nil
+		height, err := resolveBlockNumber(number, b.blocks)
+		if err != nil {
+			return 0, err
+		}
+		return int64(height), nil
 	}
 
 	if hash, ok := blockNumberOrHash.Hash(); ok {
@@ -1116,6 +1120,24 @@ func (b *BlockChainAPI) getBlockNumber(blockNumberOrHash *rpc.BlockNumberOrHash)
 	return 0, err
 }
 
+func resolveBlockNumber(
+	number rpc.BlockNumber,
+	blocksDB storage.BlockIndexer,
+) (uint64, error) {
+	height := number.Int64()
+
+	// if special values (latest) we return latest executed height
+	if height < 0 {
+		executed, err := blocksDB.LatestEVMHeight()
+		if err != nil {
+			return 0, err
+		}
+		height = int64(executed)
+	}
+
+	return uint64(height), nil
+}
+
 // handleError takes in an error and in case the error is of type ErrEntityNotFound
 // it returns nil instead of an error since that is according to the API spec,
 // if the error is not of type ErrEntityNotFound it will return the error and the generic

api/encode_transaction.go

@@ -1,12 +1,9 @@
 package api
 
 import (
-	"fmt"
 	"math/big"
 
 	"github.com/onflow/go-ethereum/core/types"
-
-	errs "github.com/onflow/flow-evm-gateway/models/errors"
 )
 
 const blockGasLimit uint64 = 120_000_000
@@ -16,7 +13,7 @@ const blockGasLimit uint64 = 120_000_000
 // `EVM.dryRun` inside Cadence scripts, meaning that no state change
 // will occur.
 // This is only useful for `eth_estimateGas` and `eth_call` endpoints.
-func encodeTxFromArgs(args TransactionArgs) ([]byte, error) {
+func encodeTxFromArgs(args TransactionArgs) (*types.LegacyTx, error) {
 	var data []byte
 	if args.Data != nil {
 		data = *args.Data
@@ -36,21 +33,12 @@ func encodeTxFromArgs(args TransactionArgs) ([]byte, error) {
 		value = args.Value.ToInt()
 	}
 
-	tx := types.NewTx(
-		&types.LegacyTx{
-			Nonce:    0,
-			To:       args.To,
-			Value:    value,
-			Gas:      gasLimit,
-			GasPrice: big.NewInt(0),
-			Data:     data,
-		},
-	)
-
-	enc, err := tx.MarshalBinary()
-	if err != nil {
-		return nil, fmt.Errorf("%w: %w", errs.ErrInvalid, err)
-	}
-
-	return enc, nil
+	return &types.LegacyTx{
+		Nonce:    0,
+		To:       args.To,
+		Value:    value,
+		Gas:      gasLimit,
+		GasPrice: big.NewInt(0),
+		Data:     data,
+	}, nil
 }

bootstrap/bootstrap.go

@@ -11,6 +11,9 @@ import (
 	"github.com/onflow/flow-go-sdk/access"
 	"github.com/onflow/flow-go-sdk/access/grpc"
 	"github.com/onflow/flow-go-sdk/crypto"
+	"github.com/onflow/flow-go/fvm/environment"
+	"github.com/onflow/flow-go/fvm/evm"
+	flowGo "github.com/onflow/flow-go/model/flow"
 	gethTypes "github.com/onflow/go-ethereum/core/types"
 	"github.com/onflow/go-ethereum/eth/tracers"
 	"github.com/rs/zerolog"
@@ -128,11 +131,13 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 		Uint64("missed-heights", latestCadenceBlock.Height-latestCadenceHeight).
 		Msg("indexing cadence height information")
 
+	chainID := b.config.FlowNetworkID
+
 	// create event subscriber
 	subscriber := ingestion.NewRPCEventSubscriber(
 		b.logger,
 		b.client,
-		b.config.FlowNetworkID,
+		chainID,
 		latestCadenceHeight,
 	)
 
@@ -146,9 +151,15 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 	}
 	blocksProvider := replayer.NewBlocksProvider(
 		b.storages.Blocks,
-		b.config.FlowNetworkID,
+		chainID,
 		tracer,
 	)
+	replayerConfig := replayer.Config{
+		ChainID:         chainID,
+		RootAddr:        evm.StorageAccountAddress(chainID),
+		Tracer:          tracer,
+		ValidateResults: true,
+	}
 
 	// initialize event ingestion engine
 	b.events = ingestion.NewEventIngestionEngine(
@@ -163,6 +174,7 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 		b.publishers.Logs,
 		b.logger,
 		b.collector,
+		replayerConfig,
 	)
 
 	StartEngine(ctx, b.events, l)
@@ -276,7 +288,23 @@ func (b *Bootstrap) StartAPIServer(ctx context.Context) error {
 		b.logger,
 	)
 
+	tracer, err := tracers.DefaultDirectory.New(
+		callTracerName,
+		&tracers.Context{},
+		json.RawMessage(callTracerConfig),
+	)
+	if err != nil {
+		return err
+	}
+	blocksProvider := replayer.NewBlocksProvider(
+		b.storages.Blocks,
+		b.config.FlowNetworkID,
+		tracer,
+	)
+
 	evm, err := requester.NewEVM(
+		b.storages.Storage,
+		blocksProvider,
 		b.client,
 		b.config,
 		signer,
@@ -532,6 +560,18 @@ func setupStorage(
 			return nil, fmt.Errorf("could not fetch provided cadence height, make sure it's correct: %w", err)
 		}
 
+		storageProvider := pebble.NewRegister(store, 0, nil)
+		storageAddress := evm.StorageAccountAddress(config.FlowNetworkID)
+		accountStatus := environment.NewAccountStatus()
+		err = storageProvider.SetValue(
+			storageAddress[:],
+			[]byte(flowGo.AccountStatusKey),
+			accountStatus.ToBytes(),
+		)
+		if err != nil {
+			return nil, fmt.Errorf("could not initialize state index: %w", err)
+		}
+
 		if err := blocks.InitHeights(cadenceHeight, cadenceBlock.ID); err != nil {
 			return nil, fmt.Errorf(
 				"failed to init the database for block height: %d and ID: %s, with : %w",

services/ingestion/engine.go

@@ -14,6 +14,8 @@ import (
 	"github.com/onflow/flow-evm-gateway/services/replayer"
 	"github.com/onflow/flow-evm-gateway/storage"
 	"github.com/onflow/flow-evm-gateway/storage/pebble"
+
+	"github.com/onflow/flow-go/fvm/evm/offchain/sync"
 )
 
 var _ models.Engine = &Engine{}
@@ -47,6 +49,7 @@ type Engine struct {
 	blocksPublisher *models.Publisher[*models.Block]
 	logsPublisher   *models.Publisher[[]*gethTypes.Log]
 	collector       metrics.Collector
+	replayerConfig  replayer.Config
 }
 
 func NewEventIngestionEngine(
@@ -61,6 +64,7 @@ func NewEventIngestionEngine(
 	logsPublisher *models.Publisher[[]*gethTypes.Log],
 	log zerolog.Logger,
 	collector metrics.Collector,
+	replayerConfig replayer.Config,
 ) *Engine {
 	log = log.With().Str("component", "ingestion").Logger()
 
@@ -78,6 +82,7 @@ func NewEventIngestionEngine(
 		blocksPublisher: blocksPublisher,
 		logsPublisher:   logsPublisher,
 		collector:       collector,
+		replayerConfig:  replayerConfig,
 	}
 }
 
@@ -158,8 +163,47 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 	batch := e.store.NewBatch()
 	defer batch.Close()
 
-	// we first index the block
-	err := e.indexBlock(
+	// Step 1: Re-execute all transactions on the latest EVM block
+
+	// Step 1.1: Notify the `BlocksProvider` of the newly received EVM block
+	if err := e.blocksProvider.OnBlockReceived(events.Block()); err != nil {
+		return err
+	}
+
+	storageProvider := pebble.NewRegister(
+		e.store,
+		events.Block().Height,
+		batch,
+	)
+	cr := sync.NewReplayer(
+		e.replayerConfig.ChainID,
+		e.replayerConfig.RootAddr,
+		storageProvider,
+		e.blocksProvider,
+		e.log,
+		e.replayerConfig.Tracer,
+		e.replayerConfig.ValidateResults,
+	)
+
+	// Step 1.2: Replay all block transactions
+	// If `ReplayBlock` returns any error, we abort the EVM events processing
+	res, err := cr.ReplayBlock(events.TxEventPayloads(), events.BlockEventPayload())
+	if err != nil {
+		return fmt.Errorf("failed to replay block on height: %d, with: %w", events.Block().Height, err)
+	}
+
+	// Step 2: Write all the necessary changes to each storage
+
+	// Step 2.1: Write all the EVM state changes to `StorageProvider`
+	for k, v := range res.StorageRegisterUpdates() {
+		err = storageProvider.SetValue([]byte(k.Owner), []byte(k.Key), v)
+		if err != nil {
+			return fmt.Errorf("failed to commit state changes on block: %d", events.Block().Height)
+		}
+	}
+
+	// Step 2.2: Write the latest EVM block to `Blocks` storage
+	err = e.indexBlock(
 		events.CadenceHeight(),
 		events.CadenceBlockID(),
 		events.Block(),
@@ -169,6 +213,8 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 		return fmt.Errorf("failed to index block %d event: %w", events.Block().Height, err)
 	}
 
+	// Step 2.3: Write all EVM transactions of the current block,
+	// to `Transactions` storage
 	for i, tx := range events.Transactions() {
 		receipt := events.Receipts()[i]
 
@@ -178,19 +224,13 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 		}
 	}
 
+	// Step 2.4: Write all EVM transaction receipts of the current block,
+	// to `Receipts` storage
 	err = e.indexReceipts(events.Receipts(), batch)
 	if err != nil {
 		return fmt.Errorf("failed to index receipts for block %d event: %w", events.Block().Height, err)
 	}
 
-	if err := e.blocksProvider.OnBlockReceived(events.Block()); err != nil {
-		return fmt.Errorf(
-			"failed to call OnBlockReceived for block %d, with: %w",
-			events.Block().Height,
-			err,
-		)
-	}
-
 	if err := batch.Commit(pebbleDB.Sync); err != nil {
 		return fmt.Errorf("failed to commit indexed data for Cadence block %d: %w", events.CadenceHeight(), err)
 	}