Implement Execution/Consensus interface over RPC

.github/workflows/_go-tests.yml

@@ -122,6 +122,12 @@ jobs:
           echo "One or more tests failed."
           exit 1
 
+      - name: run tests with consensus and execution nodes connected over json rpc
+        if: matrix.test-mode == 'defaults'
+        run: >-
+          ${{ github.workspace }}/.github/workflows/gotestsum.sh 
+          --tags cionly --timeout 60m --test_state_scheme hash --consensus_execution_in_same_process_use_rpc
+
       - name: run redis tests
         if: inputs.run-defaults-a
         run: >-

.github/workflows/gotestsum.sh

@@ -17,6 +17,7 @@ junitfile=""
 log=true
 race=false
 cover=false
+consensus_execution_in_same_process_use_rpc=false
 flaky=false
 while [[ $# -gt 0 ]]; do
   case $1 in
@@ -64,6 +65,10 @@ while [[ $# -gt 0 ]]; do
       cover=true
       shift
       ;;
+    --consensus_execution_in_same_process_use_rpc)
+     consensus_execution_in_same_process_use_rpc=true
+      shift
+      ;;
     --nolog)
       log=false
       shift
@@ -137,6 +142,10 @@ else
     cmd="$cmd -args -- --test_loglevel=8" # Use error log level, which is the value 8 in the slog level enum for tests.
 fi
 
+if [ "$consensus_execution_in_same_process_use_rpc" == true ]; then
+    cmd="$cmd --consensus_execution_in_same_process_use_rpc=true"
+fi
+
 if [ "$test_database_engine" != "" ]; then
     cmd="$cmd --test_database_engine=$test_database_engine"
 fi

arbnode/blockmetadata.go

@@ -69,8 +69,10 @@ type BlockMetadataFetcher struct {
 	stopwaiter.StopWaiter
 	config                 BlockMetadataFetcherConfig
 	db                     ethdb.Database
+	genesisBlockNum        uint64
 	client                 *rpcclient.RpcClient
 	exec                   execution.ExecutionClient
+	startBlockNum          uint64
 	trackBlockMetadataFrom arbutil.MessageIndex
 	expectedChainId        uint64
 
@@ -83,18 +85,12 @@ func NewBlockMetadataFetcher(
 	ctx context.Context,
 	c BlockMetadataFetcherConfig,
 	db ethdb.Database,
+	genesisBlockNum uint64,
 	exec execution.ExecutionClient,
-	startPos uint64,
+	startBlockNum uint64,
 	expectedChainId uint64,
 ) (*BlockMetadataFetcher, error) {
-	var trackBlockMetadataFrom arbutil.MessageIndex
 	var err error
-	if startPos != 0 {
-		trackBlockMetadataFrom, err = exec.BlockNumberToMessageIndex(startPos).Await(ctx)
-		if err != nil {
-			return nil, err
-		}
-	}
 	client := rpcclient.NewRpcClient(func() *rpcclient.ClientConfig { return &c.Source }, nil)
 	if err = client.Start(ctx); err != nil {
 		return nil, err
@@ -110,14 +106,15 @@ func NewBlockMetadataFetcher(
 	}
 
 	fetcher := &BlockMetadataFetcher{
-		config:                 c,
-		db:                     db,
-		client:                 client,
-		exec:                   exec,
-		trackBlockMetadataFrom: trackBlockMetadataFrom,
-		expectedChainId:        expectedChainId,
-		chainIdChecked:         chainIdChecked,
-		currentSyncInterval:    c.SyncInterval,
+		config:              c,
+		db:                  db,
+		genesisBlockNum:     genesisBlockNum,
+		client:              client,
+		exec:                exec,
+		startBlockNum:       startBlockNum,
+		expectedChainId:     expectedChainId,
+		chainIdChecked:      chainIdChecked,
+		currentSyncInterval: c.SyncInterval,
 	}
 	return fetcher, nil
 }
@@ -151,7 +148,7 @@ func (b *BlockMetadataFetcher) persistBlockMetadata(ctx context.Context, query [
 	batch := b.db.NewBatch()
 	queryMap := util.ArrayToSet(query)
 	for _, elem := range result {
-		pos, err := b.exec.BlockNumberToMessageIndex(elem.BlockNumber).Await(ctx)
+		pos, err := util.BlockNumberToMessageIndex(elem.BlockNumber, b.genesisBlockNum)
 		if err != nil {
 			return err
 		}
@@ -184,16 +181,8 @@ func (b *BlockMetadataFetcher) Update(ctx context.Context) time.Duration {
 	}
 
 	handleQuery := func(query []uint64) bool {
-		fromBlock, err := b.exec.MessageIndexToBlockNumber(arbutil.MessageIndex(query[0])).Await(ctx)
-		if err != nil {
-			log.Error("Error getting fromBlock", "err", err)
-			return false
-		}
-		toBlock, err := b.exec.MessageIndexToBlockNumber(arbutil.MessageIndex(query[len(query)-1])).Await(ctx)
-		if err != nil {
-			log.Error("Error getting toBlock", "err", err)
-			return false
-		}
+		fromBlock := util.MessageIndexToBlockNumber(arbutil.MessageIndex(query[0]), b.genesisBlockNum)
+		toBlock := util.MessageIndexToBlockNumber(arbutil.MessageIndex(query[len(query)-1]), b.genesisBlockNum)
 
 		result, err := b.fetch(
 			ctx,
@@ -247,9 +236,24 @@ func (b *BlockMetadataFetcher) Update(ctx context.Context) time.Duration {
 	return b.config.SyncInterval
 }
 
-func (b *BlockMetadataFetcher) Start(ctx context.Context) {
+func (b *BlockMetadataFetcher) InitializeTrackBlockMetadataFrom(ctx context.Context) error {
+	var err error
+	if b.startBlockNum != 0 {
+		b.trackBlockMetadataFrom, err = util.BlockNumberToMessageIndex(b.startBlockNum, b.genesisBlockNum)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (b *BlockMetadataFetcher) Start(ctx context.Context) error {
 	b.StopWaiter.Start(ctx, b)
+	if err := b.InitializeTrackBlockMetadataFrom(ctx); err != nil {
+		return err
+	}
 	b.CallIteratively(b.Update)
+	return nil
 }
 
 func (b *BlockMetadataFetcher) StopAndWait() {

arbnode/inbox_test.go

@@ -111,14 +111,6 @@ func (w *execClientWrapper) Start(ctx context.Context) error {
 	return nil
 }
 
-func (w *execClientWrapper) MessageIndexToBlockNumber(messageNum arbutil.MessageIndex) containers.PromiseInterface[uint64] {
-	return containers.NewReadyPromise(w.ExecutionEngine.MessageIndexToBlockNumber(messageNum), nil)
-}
-
-func (w *execClientWrapper) BlockNumberToMessageIndex(blockNum uint64) containers.PromiseInterface[arbutil.MessageIndex] {
-	return containers.NewReadyPromise(w.ExecutionEngine.BlockNumberToMessageIndex(blockNum))
-}
-
 func (w *execClientWrapper) ArbOSVersionForMessageIndex(msgIdx arbutil.MessageIndex) containers.PromiseInterface[uint64] {
 	return w.ExecutionEngine.ArbOSVersionForMessageIndex(msgIdx)
 }