engine_server: use chainRW instead of DB (#19296)

Requires erigontech/interfaces#279

Author: shohamc1

execution/engineapi/engine_api_methods.go

@@ -19,14 +19,15 @@ package engineapi
 import (
 	"context"
 	"encoding/binary"
+	"fmt"
 
 	"github.com/erigontech/erigon/cl/clparams"
 	"github.com/erigontech/erigon/common"
 	"github.com/erigontech/erigon/common/hexutil"
-	"github.com/erigontech/erigon/db/kv"
-	"github.com/erigontech/erigon/db/rawdb"
 	"github.com/erigontech/erigon/db/version"
+	"github.com/erigontech/erigon/execution/engineapi/engine_helpers"
 	"github.com/erigontech/erigon/execution/engineapi/engine_types"
+	"github.com/erigontech/erigon/rpc"
 )
 
 var ourCapabilities = []string{
@@ -186,42 +187,11 @@ func (e *EngineServer) GetPayloadBodiesByHashV1(ctx context.Context, hashes []co
 // including blockAccessList sidecars from DB.
 // See https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_getpayloadbodiesbyhashv2
 func (e *EngineServer) GetPayloadBodiesByHashV2(ctx context.Context, hashes []common.Hash) ([]*engine_types.ExecutionPayloadBodyV2, error) {
-	bodies, err := e.getPayloadBodiesByHash(ctx, hashes)
-	if err != nil {
-		return nil, err
-	}
-
-	bodyHashes := make([]common.Hash, len(bodies))
-	bodyNumbers := make([]*uint64, len(bodies))
-	for i := range bodies {
-		if i >= len(hashes) {
-			break
-		}
-		bodyHashes[i] = hashes[i]
-		number, err := e.chainRW.HeaderNumber(ctx, hashes[i])
-		if err != nil {
-			return nil, err
-		}
-		bodyNumbers[i] = number
+	if len(hashes) > 1024 {
+		return nil, &engine_helpers.TooLargeRequestErr
 	}
-
-	blockAccessLists, err := e.readBlockAccessLists(ctx, bodyHashes, bodyNumbers)
-	if err != nil {
-		return nil, err
-	}
-
-	resp := make([]*engine_types.ExecutionPayloadBodyV2, len(bodies))
-	for i := range bodies {
-		if bodies[i] == nil {
-			continue
-		}
-		resp[i] = &engine_types.ExecutionPayloadBodyV2{
-			Transactions:    bodies[i].Transactions,
-			Withdrawals:     bodies[i].Withdrawals,
-			BlockAccessList: blockAccessLists[i],
-		}
-	}
-	return resp, nil
+	e.engineLogSpamer.RecordRequest()
+	return e.chainRW.GetPayloadBodiesByHash(ctx, hashes)
 }
 
 // Returns an ordered (as per canonical chain) array of execution payload bodies, with corresponding execution block numbers from "start", up to "count"
@@ -234,67 +204,13 @@ func (e *EngineServer) GetPayloadBodiesByRangeV1(ctx context.Context, start, cou
 // including blockAccessList sidecars from DB.
 // See https://github.com/ethereum/execution-apis/blob/main/src/engine/amsterdam.md#engine_getpayloadbodiesbyrangev2
 func (e *EngineServer) GetPayloadBodiesByRangeV2(ctx context.Context, start, count hexutil.Uint64) ([]*engine_types.ExecutionPayloadBodyV2, error) {
-	bodies, err := e.getPayloadBodiesByRange(ctx, uint64(start), uint64(count))
-	if err != nil {
-		return nil, err
+	if uint64(start) == 0 || uint64(count) == 0 {
+		return nil, &rpc.InvalidParamsError{Message: fmt.Sprintf("invalid start or count, start: %v count: %v", start, count)}
 	}
-
-	bodyHashes := make([]common.Hash, len(bodies))
-	bodyNumbers := make([]*uint64, len(bodies))
-	for i := range bodies {
-		blockNumber := uint64(start) + uint64(i)
-		header := e.chainRW.GetHeaderByNumber(ctx, blockNumber)
-		if header == nil {
-			continue
-		}
-		bodyHashes[i] = header.Hash()
-		number := blockNumber
-		bodyNumbers[i] = &number
-	}
-
-	blockAccessLists, err := e.readBlockAccessLists(ctx, bodyHashes, bodyNumbers)
-	if err != nil {
-		return nil, err
-	}
-
-	resp := make([]*engine_types.ExecutionPayloadBodyV2, len(bodies))
-	for i := range bodies {
-		if bodies[i] == nil {
-			continue
-		}
-		resp[i] = &engine_types.ExecutionPayloadBodyV2{
-			Transactions:    bodies[i].Transactions,
-			Withdrawals:     bodies[i].Withdrawals,
-			BlockAccessList: blockAccessLists[i],
-		}
-	}
-	return resp, nil
-}
-
-func (e *EngineServer) readBlockAccessLists(ctx context.Context, hashes []common.Hash, numbers []*uint64) ([]hexutil.Bytes, error) {
-	resp := make([]hexutil.Bytes, len(hashes))
-	if e.db == nil {
-		return resp, nil
-	}
-	if err := e.db.View(ctx, func(tx kv.Tx) error {
-		for i := range hashes {
-			if numbers[i] == nil {
-				continue
-			}
-			balBytes, err := rawdb.ReadBlockAccessListBytes(tx, hashes[i], *numbers[i])
-			if err != nil {
-				return err
-			}
-			if len(balBytes) == 0 {
-				continue
-			}
-			resp[i] = append([]byte(nil), balBytes...)
-		}
-		return nil
-	}); err != nil {
-		return nil, err
+	if uint64(count) > 1024 {
+		return nil, &engine_helpers.TooLargeRequestErr
 	}
-	return resp, nil
+	return e.chainRW.GetPayloadBodiesByRange(ctx, uint64(start), uint64(count))
 }
 
 // Returns the node's code and commit details in a slice

execution/engineapi/engine_server.go

@@ -76,7 +76,6 @@ type EngineServer struct {
 	caplin           bool // we need to send errors for caplin.
 	executionService executionproto.ExecutionClient
 	txpool           txpoolproto.TxpoolClient // needed for getBlobs
-	db               kv.TemporalRoDB
 
 	chainRW chainreader.ChainReaderWriterEth1
 	lock    sync.Mutex
@@ -131,8 +130,6 @@ func (e *EngineServer) Start(
 	eth rpchelper.ApiBackend,
 	mining txpoolproto.MiningClient,
 ) error {
-	e.db = db
-
 	var eg errgroup.Group
 	if !e.caplin {
 		eg.Go(func() error {

execution/engineapi/engine_server_test.go

@@ -281,7 +281,6 @@ func TestGetPayloadBodiesByHashV2(t *testing.T) {
 	executionRpc := direct.NewExecutionClientDirect(mockSentry.Eth1ExecutionService)
 	maxReorgDepth := ethconfig.Defaults.MaxReorgDepth
 	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, nil, ethconfig.Defaults.FcuTimeout, maxReorgDepth)
-	engineServer.db = mockSentry.DB
 
 	const blockNum = 1
 	blockHash := canonicalHashAt(t, mockSentry.DB, blockNum)
@@ -315,7 +314,6 @@ func TestGetPayloadBodiesByRangeV2(t *testing.T) {
 	executionRpc := direct.NewExecutionClientDirect(mockSentry.Eth1ExecutionService)
 	maxReorgDepth := ethconfig.Defaults.MaxReorgDepth
 	engineServer := NewEngineServer(mockSentry.Log, mockSentry.ChainConfig, executionRpc, nil, false, false, true, nil, ethconfig.Defaults.FcuTimeout, maxReorgDepth)
-	engineServer.db = mockSentry.DB
 
 	const (
 		start = 1

execution/execmodule/chainreader/chain_reader.go

@@ -30,6 +30,7 @@ import (
 	"github.com/erigontech/erigon/cl/cltypes/solid"
 	"github.com/erigontech/erigon/cl/utils"
 	"github.com/erigontech/erigon/common"
+	"github.com/erigontech/erigon/common/hexutil"
 	"github.com/erigontech/erigon/common/log/v3"
 	"github.com/erigontech/erigon/execution/chain"
 	"github.com/erigontech/erigon/execution/engineapi/engine_types"
@@ -250,6 +251,50 @@ func (c ChainReaderWriterEth1) GetBodiesByRange(ctx context.Context, start, coun
 	return ret, nil
 }
 
+func (c ChainReaderWriterEth1) GetPayloadBodiesByHash(ctx context.Context, hashes []common.Hash) ([]*engine_types.ExecutionPayloadBodyV2, error) {
+	grpcHashes := make([]*typesproto.H256, len(hashes))
+	for i := range grpcHashes {
+		grpcHashes[i] = gointerfaces.ConvertHashToH256(hashes[i])
+	}
+	resp, err := c.executionModule.GetPayloadBodiesByHash(ctx, &executionproto.GetPayloadBodiesByHashRequest{
+		Hashes: grpcHashes,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return convertPayloadBodiesFromRpc(resp.Bodies), nil
+}
+
+func (c ChainReaderWriterEth1) GetPayloadBodiesByRange(ctx context.Context, start, count uint64) ([]*engine_types.ExecutionPayloadBodyV2, error) {
+	resp, err := c.executionModule.GetPayloadBodiesByRange(ctx, &executionproto.GetPayloadBodiesByRangeRequest{
+		Start: start,
+		Count: count,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return convertPayloadBodiesFromRpc(resp.Bodies), nil
+}
+
+func convertPayloadBodiesFromRpc(bodies []*typesproto.ExecutionPayloadBody) []*engine_types.ExecutionPayloadBodyV2 {
+	result := make([]*engine_types.ExecutionPayloadBodyV2, len(bodies))
+	for i, body := range bodies {
+		if body == nil {
+			continue
+		}
+		txs := make([]hexutil.Bytes, len(body.Transactions))
+		for j, tx := range body.Transactions {
+			txs[j] = tx
+		}
+		result[i] = &engine_types.ExecutionPayloadBodyV2{
+			Transactions:    txs,
+			Withdrawals:     moduleutil.ConvertWithdrawalsFromRpc(body.Withdrawals),
+			BlockAccessList: body.BlockAccessList,
+		}
+	}
+	return result
+}
+
 func (c ChainReaderWriterEth1) Ready(ctx context.Context) (bool, error) {
 	resp, err := c.executionModule.Ready(ctx, &emptypb.Empty{})
 	if err != nil {

execution/execmodule/getters.go

@@ -17,6 +17,7 @@
 package execmodule
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -214,6 +215,115 @@ func (e *EthereumExecutionModule) GetBodiesByRange(ctx context.Context, req *exe
 	}, nil
 }
 
+func (e *EthereumExecutionModule) GetPayloadBodiesByHash(ctx context.Context, req *executionproto.GetPayloadBodiesByHashRequest) (*executionproto.GetPayloadBodiesBatchResponse, error) {
+	tx, err := e.db.BeginRo(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByHash: could not begin database tx %w", err)
+	}
+	defer tx.Rollback()
+
+	bodies := make([]*typesproto.ExecutionPayloadBody, 0, len(req.Hashes))
+	for _, hash := range req.Hashes {
+		h := gointerfaces.ConvertH256ToHash(hash)
+		number, err := e.blockReader.HeaderNumber(ctx, tx, h)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByHash: HeaderNumber error %w", err)
+		}
+		if number == nil {
+			bodies = append(bodies, nil)
+			continue
+		}
+		body, err := e.getBody(ctx, tx, h, *number)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByHash: getBody error %w", err)
+		}
+		if body == nil {
+			bodies = append(bodies, nil)
+			continue
+		}
+		txs, err := types.MarshalTransactionsBinary(body.Transactions)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByHash: MarshalTransactionsBinary error %w", err)
+		}
+		balBytes, err := rawdb.ReadBlockAccessListBytes(tx, h, *number)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByHash: ReadBlockAccessListBytes error %w", err)
+		}
+		var bal []byte
+		if len(balBytes) > 0 {
+			bal = bytes.Clone(balBytes)
+		}
+
+		bodies = append(bodies, &typesproto.ExecutionPayloadBody{
+			Transactions:    txs,
+			Withdrawals:     moduleutil.ConvertWithdrawalsToRpc(body.Withdrawals),
+			BlockAccessList: bal,
+		})
+	}
+
+	return &executionproto.GetPayloadBodiesBatchResponse{Bodies: bodies}, nil
+}
+
+func (e *EthereumExecutionModule) GetPayloadBodiesByRange(ctx context.Context, req *executionproto.GetPayloadBodiesByRangeRequest) (*executionproto.GetPayloadBodiesBatchResponse, error) {
+	tx, err := e.db.BeginRo(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByRange: could not begin database tx %w", err)
+	}
+	defer tx.Rollback()
+
+	bodies := make([]*typesproto.ExecutionPayloadBody, 0, req.Count)
+
+	for i := uint64(0); i < req.Count; i++ {
+		blockNum := req.Start + i
+		hash, err := e.canonicalHash(ctx, tx, blockNum)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByRange: ReadCanonicalHash error %w", err)
+		}
+		if hash == (common.Hash{}) {
+			break
+		}
+
+		body, err := e.getBody(ctx, tx, hash, blockNum)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByRange: getBody error %w", err)
+		}
+		if body == nil {
+			bodies = append(bodies, nil)
+			continue
+		}
+
+		txs, err := types.MarshalTransactionsBinary(body.Transactions)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByRange: MarshalTransactionsBinary error %w", err)
+		}
+		balBytes, err := rawdb.ReadBlockAccessListBytes(tx, hash, blockNum)
+		if err != nil {
+			return nil, fmt.Errorf("ethereumExecutionModule.GetPayloadBodiesByRange: ReadBlockAccessListBytes error %w", err)
+		}
+		var bal []byte
+		if len(balBytes) > 0 {
+			bal = bytes.Clone(balBytes)
+		}
+
+		bodies = append(bodies, &typesproto.ExecutionPayloadBody{
+			Transactions:    txs,
+			Withdrawals:     moduleutil.ConvertWithdrawalsToRpc(body.Withdrawals),
+			BlockAccessList: bal,
+		})
+	}
+
+	// Remove trailing nil values
+	for i := len(bodies) - 1; i >= 0; i-- {
+		if bodies[i] == nil {
+			bodies = bodies[:i]
+		} else {
+			break
+		}
+	}
+
+	return &executionproto.GetPayloadBodiesBatchResponse{Bodies: bodies}, nil
+}
+
 func (e *EthereumExecutionModule) GetHeaderHashNumber(ctx context.Context, req *typesproto.H256) (*executionproto.GetHeaderHashNumberResponse, error) {
 	tx, err := e.db.BeginRo(ctx)
 	if err != nil {

node/direct/execution_client.go

@@ -46,6 +46,14 @@ func (s *ExecutionClientDirect) GetBodiesByRange(ctx context.Context, in *execut
 	return s.server.GetBodiesByRange(ctx, in)
 }
 
+func (s *ExecutionClientDirect) GetPayloadBodiesByHash(ctx context.Context, in *executionproto.GetPayloadBodiesByHashRequest, opts ...grpc.CallOption) (*executionproto.GetPayloadBodiesBatchResponse, error) {
+	return s.server.GetPayloadBodiesByHash(ctx, in)
+}
+
+func (s *ExecutionClientDirect) GetPayloadBodiesByRange(ctx context.Context, in *executionproto.GetPayloadBodiesByRangeRequest, opts ...grpc.CallOption) (*executionproto.GetPayloadBodiesBatchResponse, error) {
+	return s.server.GetPayloadBodiesByRange(ctx, in)
+}
+
 func (s *ExecutionClientDirect) HasBlock(ctx context.Context, in *executionproto.GetSegmentRequest, opts ...grpc.CallOption) (*executionproto.HasBlockResponse, error) {
 	return s.server.HasBlock(ctx, in)
 }