Merge branch 'ramtin/evm-offchain-part2' into ramtin/evm-offchain-part3

fvm/evm/offchain/blocks/provider.go

@@ -0,0 +1,59 @@
+package blocks
+
+import (
+	"github.com/onflow/flow-go/fvm/evm/events"
+	"github.com/onflow/flow-go/fvm/evm/types"
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// BasicProvider implements a ledger based block provider
+type BasicProvider struct {
+	blks *Blocks
+}
+
+var _ types.BlockSnapshotProvider = (*BasicProvider)(nil)
+
+func NewBasicProvider(
+	chainID flow.ChainID,
+	storage types.BackendStorage,
+	rootAddr flow.Address,
+) (*BasicProvider, error) {
+	blks, err := NewBlocks(chainID, rootAddr, storage)
+	if err != nil {
+		return nil, err
+	}
+	return &BasicProvider{blks: blks}, nil
+}
+
+func (p *BasicProvider) GetSnapshotAt(height uint64) (
+	types.BlockSnapshot,
+	error,
+) {
+	return p.blks, nil
+}
+
+// OnBlockReceived should be called before
+// executing blocks.
+func (p *BasicProvider) OnBlockReceived(blockEvent *events.BlockEventPayload) error {
+	// prepare blocks
+	// push the new block meta
+	// it should be done before execution so block context creation
+	// can be done properly
+	return p.blks.PushBlockMeta(
+		NewMeta(
+			blockEvent.Height,
+			blockEvent.Timestamp,
+			blockEvent.PrevRandao,
+		),
+	)
+}
+
+func (p *BasicProvider) OnBlockExecuted(blockEvent *events.BlockEventPayload) error {
+	// push block hash
+	// we push the block hash after execution, so the behaviour of the blockhash is
+	// identical to the evm.handler.
+	return p.blks.PushBlockHash(
+		blockEvent.Height,
+		blockEvent.Hash,
+	)
+}

fvm/evm/offchain/sync/replayer.go

@@ -5,7 +5,6 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/onflow/flow-go/fvm/evm/events"
-	"github.com/onflow/flow-go/fvm/evm/offchain/blocks"
 	"github.com/onflow/flow-go/fvm/evm/offchain/storage"
 	"github.com/onflow/flow-go/fvm/evm/types"
 	"github.com/onflow/flow-go/model/flow"
@@ -19,6 +18,7 @@ type Replayer struct {
 	rootAddr        flow.Address
 	logger          zerolog.Logger
 	storageProvider types.StorageProvider
+	blockProvider   types.BlockSnapshotProvider
 	tracer          *gethTracers.Tracer
 	validateResults bool
 }
@@ -28,6 +28,7 @@ func NewReplayer(
 	chainID flow.ChainID,
 	rootAddr flow.Address,
 	sp types.StorageProvider,
+	bp types.BlockSnapshotProvider,
 	logger zerolog.Logger,
 	tracer *gethTracers.Tracer,
 	validateResults bool,
@@ -36,6 +37,7 @@ func NewReplayer(
 		chainID:         chainID,
 		rootAddr:        rootAddr,
 		storageProvider: sp,
+		blockProvider:   bp,
 		logger:          logger,
 		tracer:          tracer,
 		validateResults: validateResults,
@@ -44,12 +46,8 @@ func NewReplayer(
 
 // OnBlockReceived is called when a new block is received
 // (including all the related transaction executed events)
-//
-// currently this version of replayer requires
-// sequential calls to the OnBlockReceived
-// in the future if move the blocks logic to outside
-// we can pass blockSnapshotProvider and have the ability
-// to call OnBlockReceived concurrently.
+// this method can be called concurrently if underlying storage
+// tracer and block snapshot provider support concurrency.
 func (cr *Replayer) OnBlockReceived(
 	transactionEvents []events.TransactionEventPayload,
 	blockEvent *events.BlockEventPayload,
@@ -63,22 +61,8 @@ func (cr *Replayer) OnBlockReceived(
 	// create storage
 	state := storage.NewEphemeralStorage(storage.NewReadOnlyStorage(st))
 
-	// prepare blocks
-	blks, err := blocks.NewBlocks(cr.chainID, cr.rootAddr, state)
-	if err != nil {
-		return nil, err
-	}
-
-	// push the new block meta
-	// it should be done before execution so block context creation
-	// can be done properly
-	err = blks.PushBlockMeta(
-		blocks.NewMeta(
-			blockEvent.Height,
-			blockEvent.Timestamp,
-			blockEvent.PrevRandao,
-		),
-	)
+	// get block snapshot
+	bs, err := cr.blockProvider.GetSnapshotAt(blockEvent.Height)
 	if err != nil {
 		return nil, err
 	}
@@ -88,7 +72,7 @@ func (cr *Replayer) OnBlockReceived(
 		cr.chainID,
 		cr.rootAddr,
 		state,
-		blks,
+		bs,
 		cr.tracer,
 		transactionEvents,
 		blockEvent,
@@ -98,16 +82,5 @@ func (cr *Replayer) OnBlockReceived(
 		return nil, err
 	}
 
-	// push block hash
-	// we push the block hash after execution, so the behaviour of the blockhash is
-	// identical to the evm.handler.
-	err = blks.PushBlockHash(
-		blockEvent.Height,
-		blockEvent.Hash,
-	)
-	if err != nil {
-		return nil, err
-	}
-
 	return state, nil
 }

fvm/evm/offchain/sync/replayer_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/onflow/flow-go/fvm/evm"
 	"github.com/onflow/flow-go/fvm/evm/events"
+	"github.com/onflow/flow-go/fvm/evm/offchain/blocks"
 	"github.com/onflow/flow-go/fvm/evm/offchain/storage"
 	"github.com/onflow/flow-go/fvm/evm/offchain/sync"
 	. "github.com/onflow/flow-go/fvm/evm/testutils"
@@ -154,9 +155,19 @@ func TestChainReplay(t *testing.T) {
 						require.Len(t, txEventPayloads, totalTxCount)
 
 						// check replay
+
+						bp, err := blocks.NewBasicProvider(chainID, snapshot, rootAddr)
+						require.NoError(t, err)
+
+						err = bp.OnBlockReceived(blockEventPayload)
+						require.NoError(t, err)
+
 						sp := newTestStorageProvider(snapshot, 1)
-						cr := sync.NewReplayer(chainID, rootAddr, sp, zerolog.Logger{}, nil, true)
-						_, err := cr.OnBlockReceived(txEventPayloads, blockEventPayload)
+						cr := sync.NewReplayer(chainID, rootAddr, sp, bp, zerolog.Logger{}, nil, true)
+						_, err = cr.OnBlockReceived(txEventPayloads, blockEventPayload)
+						require.NoError(t, err)
+
+						err = bp.OnBlockExecuted(blockEventPayload)
 						require.NoError(t, err)
 
 						// TODO: verify the state delta