Add verification tool for evm offchain replay

Makefile

@@ -203,10 +203,10 @@ generate-mocks: install-mock-generators
 	mockery --name 'API' --dir="./engine/protocol" --case=underscore --output="./engine/protocol/mock" --outpkg="mock"
 	mockery --name '.*' --dir="./engine/access/state_stream" --case=underscore --output="./engine/access/state_stream/mock" --outpkg="mock"
 	mockery --name 'BlockTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
+	mockery --name 'DataProvider' --dir="./engine/access/rest/websockets/data_provider" --case=underscore --output="./engine/access/rest/websockets/data_provider/mock"  --outpkg="mock"
 	mockery --name 'ExecutionDataTracker' --dir="./engine/access/subscription" --case=underscore --output="./engine/access/subscription/mock"  --outpkg="mock"
 	mockery --name 'ConnectionFactory' --dir="./engine/access/rpc/connection" --case=underscore --output="./engine/access/rpc/connection/mock" --outpkg="mock"
 	mockery --name 'Communicator' --dir="./engine/access/rpc/backend" --case=underscore --output="./engine/access/rpc/backend/mock" --outpkg="mock"
-
 	mockery --name '.*' --dir=model/fingerprint --case=underscore --output="./model/fingerprint/mock" --outpkg="mock"
 	mockery --name 'ExecForkActor' --structname 'ExecForkActorMock' --dir=module/mempool/consensus/mock/ --case=underscore --output="./module/mempool/consensus/mock/" --outpkg="mock"
 	mockery --name '.*' --dir=engine/verification/fetcher/ --case=underscore --output="./engine/verification/fetcher/mock" --outpkg="mockfetcher"

cmd/observer/node_builder/observer_builder.go

@@ -45,6 +45,7 @@ import (
 	restapiproxy "github.com/onflow/flow-go/engine/access/rest/apiproxy"
 	commonrest "github.com/onflow/flow-go/engine/access/rest/common"
 	"github.com/onflow/flow-go/engine/access/rest/router"
+	"github.com/onflow/flow-go/engine/access/rest/websockets"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	rpcConnection "github.com/onflow/flow-go/engine/access/rpc/connection"
@@ -168,6 +169,7 @@ type ObserverServiceConfig struct {
 	registerCacheSize                    uint
 	programCacheSize                     uint
 	registerDBPruneThreshold             uint64
+	websocketConfig                      websockets.Config
 }
 
 // DefaultObserverServiceConfig defines all the default values for the ObserverServiceConfig
@@ -252,6 +254,7 @@ func DefaultObserverServiceConfig() *ObserverServiceConfig {
 		registerCacheSize:        0,
 		programCacheSize:         0,
 		registerDBPruneThreshold: pruner.DefaultThreshold,
+		websocketConfig:          websockets.NewDefaultWebsocketConfig(),
 	}
 }
 

cmd/util/cmd/root.go

@@ -41,6 +41,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/cmd/snapshot"
 	system_addresses "github.com/onflow/flow-go/cmd/util/cmd/system-addresses"
 	truncate_database "github.com/onflow/flow-go/cmd/util/cmd/truncate-database"
+	verify_evm_offchain_replay "github.com/onflow/flow-go/cmd/util/cmd/verify-evm-offchain-replay"
 	"github.com/onflow/flow-go/cmd/util/cmd/version"
 	"github.com/onflow/flow-go/module/profiler"
 )
@@ -126,6 +127,7 @@ func addCommands() {
 	rootCmd.AddCommand(debug_script.Cmd)
 	rootCmd.AddCommand(generate_authorization_fixes.Cmd)
 	rootCmd.AddCommand(evm_state_exporter.Cmd)
+	rootCmd.AddCommand(verify_evm_offchain_replay.Cmd)
 }
 
 func initConfig() {

cmd/util/cmd/run-script/cmd.go

@@ -16,6 +16,7 @@ import (
 	"github.com/onflow/flow-go/cmd/util/ledger/util"
 	"github.com/onflow/flow-go/cmd/util/ledger/util/registers"
 	"github.com/onflow/flow-go/engine/access/rest"
+	"github.com/onflow/flow-go/engine/access/rest/websockets"
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
 	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/engine/execution/computation"
@@ -169,6 +170,7 @@ func run(*cobra.Command, []string) {
 			metrics.NewNoopCollector(),
 			nil,
 			backend.Config{},
+			websockets.NewDefaultWebsocketConfig(),
 		)
 		if err != nil {
 			log.Fatal().Err(err).Msg("failed to create server")

cmd/util/cmd/verify-evm-offchain-replay/main.go

@@ -0,0 +1,87 @@
+package verify
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+var (
+	flagDatadir          string
+	flagExecutionDataDir string
+	flagEVMStateGobDir   string
+	flagChain            string
+	flagFromTo           string
+)
+
+// usage example
+//
+//		./util verify-evm-offchain-replay --chain flow-testnet --from-to 211176671-211177000
+//	     --datadir /var/flow/data/protocol --execution_data_dir /var/flow/data/execution_data
+var Cmd = &cobra.Command{
+	Use:   "verify-evm-offchain-replay",
+	Short: "verify evm offchain replay with execution data",
+	Run:   run,
+}
+
+func init() {
+	Cmd.Flags().StringVar(&flagChain, "chain", "", "Chain name")
+	_ = Cmd.MarkFlagRequired("chain")
+
+	Cmd.Flags().StringVar(&flagDatadir, "datadir", "/var/flow/data/protocol",
+		"directory that stores the protocol state")
+
+	Cmd.Flags().StringVar(&flagExecutionDataDir, "execution_data_dir", "/var/flow/data/execution_data",
+		"directory that stores the execution state")
+
+	Cmd.Flags().StringVar(&flagFromTo, "from_to", "",
+		"the flow height range to verify blocks, i.e, 1-1000, 1000-2000, 2000-3000, etc.")
+
+	Cmd.Flags().StringVar(&flagEVMStateGobDir, "evm_state_gob_dir", "/var/flow/data/evm_state_gob",
+		"directory that stores the evm state gob files as checkpoint")
+}
+
+func run(*cobra.Command, []string) {
+	_ = flow.ChainID(flagChain).Chain()
+
+	from, to, err := parseFromTo(flagFromTo)
+	if err != nil {
+		log.Fatal().Err(err).Msg("could not parse from_to")
+	}
+
+	log.Info().Msgf("verifying range from %d to %d", from, to)
+	err = Verify(log.Logger, from, to, flow.Testnet, flagDatadir, flagExecutionDataDir, flagEVMStateGobDir)
+	if err != nil {
+		log.Fatal().Err(err).Msg("could not verify height")
+	}
+	log.Info().Msgf("successfully verified range from %d to %d", from, to)
+
+}
+
+func parseFromTo(fromTo string) (from, to uint64, err error) {
+	parts := strings.Split(fromTo, "-")
+	if len(parts) != 2 {
+		return 0, 0, fmt.Errorf("invalid format: expected 'from-to', got '%s'", fromTo)
+	}
+
+	from, err = strconv.ParseUint(strings.TrimSpace(parts[0]), 10, 64)
+	if err != nil {
+		return 0, 0, fmt.Errorf("invalid 'from' value: %w", err)
+	}
+
+	to, err = strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 64)
+	if err != nil {
+		return 0, 0, fmt.Errorf("invalid 'to' value: %w", err)
+	}
+
+	if from > to {
+		return 0, 0, fmt.Errorf("'from' value (%d) must be less than or equal to 'to' value (%d)", from, to)
+	}
+
+	return from, to, nil
+}

cmd/util/cmd/verify-evm-offchain-replay/verify.go

@@ -0,0 +1,144 @@
+package verify
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/dgraph-io/badger/v2"
+	badgerds "github.com/ipfs/go-ds-badger2"
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/cmd/util/cmd/common"
+	"github.com/onflow/flow-go/fvm/environment"
+	"github.com/onflow/flow-go/fvm/evm"
+	"github.com/onflow/flow-go/fvm/evm/offchain/utils"
+	"github.com/onflow/flow-go/fvm/evm/testutils"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/blobs"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
+	"github.com/onflow/flow-go/storage"
+)
+
+// Verify verifies the offchain replay of EVM blocks from the given height range
+// and updates the EVM state gob files with the latest state
+func Verify(log zerolog.Logger, from uint64, to uint64, chainID flow.ChainID, dataDir string, executionDataDir string, evmStateGobDir string) error {
+	log.Info().
+		Str("chain", chainID.String()).
+		Str("dataDir", dataDir).
+		Str("executionDataDir", executionDataDir).
+		Str("evmStateGobDir", evmStateGobDir).
+		Msgf("verifying range from %d to %d", from, to)
+
+	db, storages, executionDataStore, dsStore, err := initStorages(chainID, dataDir, executionDataDir)
+	if err != nil {
+		return fmt.Errorf("could not initialize storages: %w", err)
+	}
+
+	defer db.Close()
+	defer dsStore.Close()
+
+	var store *testutils.TestValueStore
+	isRoot := isEVMRootHeight(chainID, from)
+	if isRoot {
+		log.Info().Msgf("initializing EVM state for root height %d", from)
+
+		store = testutils.GetSimpleValueStore()
+		as := environment.NewAccountStatus()
+		rootAddr := evm.StorageAccountAddress(chainID)
+		err = store.SetValue(rootAddr[:], []byte(flow.AccountStatusKey), as.ToBytes())
+		if err != nil {
+			return err
+		}
+	} else {
+		prev := from - 1
+		log.Info().Msgf("loading EVM state from previous height %d", prev)
+
+		valueFileName, allocatorFileName := evmStateGobFileNamesByEndHeight(evmStateGobDir, prev)
+		values, err := testutils.DeserializeState(valueFileName)
+		if err != nil {
+			return fmt.Errorf("could not deserialize state %v: %w", valueFileName, err)
+		}
+
+		allocators, err := testutils.DeserializeAllocator(allocatorFileName)
+		if err != nil {
+			return fmt.Errorf("could not deserialize allocator %v: %w", allocatorFileName, err)
+		}
+		store = testutils.GetSimpleValueStorePopulated(values, allocators)
+	}
+
+	err = utils.OffchainReplayBackwardCompatibilityTest(
+		log,
+		chainID,
+		from,
+		to,
+		storages.Headers,
+		storages.Results,
+		executionDataStore,
+		store,
+	)
+
+	if err != nil {
+		return err
+	}
+
+	valueFileName, allocatorFileName := evmStateGobFileNamesByEndHeight(evmStateGobDir, to)
+	values, allocators := store.Dump()
+	err = testutils.SerializeState(valueFileName, values)
+	if err != nil {
+		return err
+	}
+	err = testutils.SerializeAllocator(allocatorFileName, allocators)
+	if err != nil {
+		return err
+	}
+
+	log.Info().Msgf("saved EVM state to %s and %s", valueFileName, allocatorFileName)
+
+	return nil
+}
+
+func initStorages(chainID flow.ChainID, dataDir string, executionDataDir string) (
+	*badger.DB,
+	*storage.All,
+	execution_data.ExecutionDataGetter,
+	io.Closer,
+	error,
+) {
+	db := common.InitStorage(dataDir)
+
+	storages := common.InitStorages(db)
+
+	datastoreDir := filepath.Join(executionDataDir, "blobstore")
+	err := os.MkdirAll(datastoreDir, 0700)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	dsOpts := &badgerds.DefaultOptions
+	ds, err := badgerds.NewDatastore(datastoreDir, dsOpts)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	executionDataBlobstore := blobs.NewBlobstore(ds)
+	executionDataStore := execution_data.NewExecutionDataStore(executionDataBlobstore, execution_data.DefaultSerializer)
+
+	return db, storages, executionDataStore, ds, nil
+}
+
+// EVM Root Height is the first block that has EVM Block Event where the EVM block height is 1
+func isEVMRootHeight(chainID flow.ChainID, flowHeight uint64) bool {
+	if chainID == flow.Testnet {
+		return flowHeight == 211176671
+	} else if chainID == flow.Mainnet {
+		return flowHeight == 85981136
+	}
+	return flowHeight == 1
+}
+
+func evmStateGobFileNamesByEndHeight(evmStateGobDir string, endHeight uint64) (string, string) {
+	valueFileName := filepath.Join(evmStateGobDir, fmt.Sprintf("values-%d.gob", endHeight))
+	allocatorFileName := filepath.Join(evmStateGobDir, fmt.Sprintf("allocators-%d.gob", endHeight))
+	return valueFileName, allocatorFileName
+}

engine/access/handle_irrecoverable_state_test.go

@@ -23,6 +23,7 @@ import (
 	accessmock "github.com/onflow/flow-go/engine/access/mock"
 	"github.com/onflow/flow-go/engine/access/rest"
 	"github.com/onflow/flow-go/engine/access/rest/router"
+	"github.com/onflow/flow-go/engine/access/rest/websockets"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	statestreambackend "github.com/onflow/flow-go/engine/access/state_stream/backend"
@@ -109,6 +110,7 @@ func (suite *IrrecoverableStateTestSuite) SetupTest() {
 		RestConfig: rest.Config{
 			ListenAddress: unittest.DefaultAddress,
 		},
+		WebSocketConfig: websockets.NewDefaultWebsocketConfig(),
 	}
 
 	// generate a server certificate that will be served by the GRPC server

engine/access/integration_unsecure_grpc_server_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/engine/access/index"
 	accessmock "github.com/onflow/flow-go/engine/access/mock"
+	"github.com/onflow/flow-go/engine/access/rest/websockets"
 	"github.com/onflow/flow-go/engine/access/rpc"
 	"github.com/onflow/flow-go/engine/access/rpc/backend"
 	"github.com/onflow/flow-go/engine/access/state_stream"
@@ -138,6 +139,7 @@ func (suite *SameGRPCPortTestSuite) SetupTest() {
 		UnsecureGRPCListenAddr: unittest.DefaultAddress,
 		SecureGRPCListenAddr:   unittest.DefaultAddress,
 		HTTPListenAddr:         unittest.DefaultAddress,
+		WebSocketConfig:        websockets.NewDefaultWebsocketConfig(),
 	}
 
 	blockCount := 5

engine/access/rest/router/router.go

@@ -2,6 +2,7 @@ package router
 
 import (
 	"fmt"
+	"net/http"
 	"regexp"
 	"strings"
 
@@ -10,8 +11,9 @@ import (
 
 	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/engine/access/rest/common/middleware"
-	"github.com/onflow/flow-go/engine/access/rest/http"
+	flowhttp "github.com/onflow/flow-go/engine/access/rest/http"
 	"github.com/onflow/flow-go/engine/access/rest/http/models"
+	"github.com/onflow/flow-go/engine/access/rest/websockets"
 	legacyws "github.com/onflow/flow-go/engine/access/rest/websockets/legacy"
 	"github.com/onflow/flow-go/engine/access/state_stream"
 	"github.com/onflow/flow-go/engine/access/state_stream/backend"
@@ -54,7 +56,7 @@ func (b *RouterBuilder) AddRestRoutes(
 ) *RouterBuilder {
 	linkGenerator := models.NewLinkGeneratorImpl(b.v1SubRouter)
 	for _, r := range Routes {
-		h := http.NewHandler(b.logger, backend, r.Handler, linkGenerator, chain, maxRequestSize)
+		h := flowhttp.NewHandler(b.logger, backend, r.Handler, linkGenerator, chain, maxRequestSize)
 		b.v1SubRouter.
 			Methods(r.Method).
 			Path(r.Pattern).
@@ -64,8 +66,8 @@ func (b *RouterBuilder) AddRestRoutes(
 	return b
 }
 
-// AddWsLegacyRoutes adds WebSocket routes to the router.
-func (b *RouterBuilder) AddWsLegacyRoutes(
+// AddLegacyWebsocketsRoutes adds WebSocket routes to the router.
+func (b *RouterBuilder) AddLegacyWebsocketsRoutes(
 	stateStreamApi state_stream.API,
 	chain flow.Chain,
 	stateStreamConfig backend.Config,
@@ -84,6 +86,23 @@ func (b *RouterBuilder) AddWsLegacyRoutes(
 	return b
 }
 
+func (b *RouterBuilder) AddWebsocketsRoute(
+	chain flow.Chain,
+	config websockets.Config,
+	streamApi state_stream.API,
+	streamConfig backend.Config,
+	maxRequestSize int64,
+) *RouterBuilder {
+	handler := websockets.NewWebSocketHandler(b.logger, config, chain, streamApi, streamConfig, maxRequestSize)
+	b.v1SubRouter.
+		Methods(http.MethodGet).
+		Path("/ws").
+		Name("ws").
+		Handler(handler)
+
+	return b
+}
+
 func (b *RouterBuilder) Build() *mux.Router {
 	return b.router
 }