feat: handle execution proposer rotation

Author: randygrok

execution/evm/eth_rpc_client.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"math/big"
 
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
 )
@@ -30,3 +32,19 @@ func (e *ethRPCClient) GetTxs(ctx context.Context) ([]string, error) {
 	}
 	return result, nil
 }
+
+func (e *ethRPCClient) GetNextProposer(ctx context.Context, number *big.Int) (common.Hash, error) {
+	var result common.Hash
+	err := e.client.Client().CallContext(ctx, &result, "evolve_getNextProposer", blockNumberArg(number))
+	if err != nil {
+		return common.Hash{}, err
+	}
+	return result, nil
+}
+
+func blockNumberArg(number *big.Int) string {
+	if number == nil {
+		return "latest"
+	}
+	return hexutil.EncodeBig(number)
+}

execution/evm/eth_rpc_tracing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"math/big"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -82,3 +83,29 @@ func (t *tracedEthRPCClient) GetTxs(ctx context.Context) ([]string, error) {
 
 	return result, nil
 }
+
+func (t *tracedEthRPCClient) GetNextProposer(ctx context.Context, number *big.Int) (common.Hash, error) {
+	blockNumber := "latest"
+	if number != nil {
+		blockNumber = number.String()
+	}
+
+	ctx, span := t.tracer.Start(ctx, "Evolve.GetNextProposer",
+		trace.WithAttributes(
+			attribute.String("method", "evolve_getNextProposer"),
+			attribute.String("block_number", blockNumber),
+		),
+	)
+	defer span.End()
+
+	result, err := t.inner.GetNextProposer(ctx, number)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return common.Hash{}, err
+	}
+
+	span.SetAttributes(attribute.String("next_proposer", result.Hex()))
+
+	return result, nil
+}

execution/evm/eth_rpc_tracing_test.go

@@ -6,6 +6,7 @@ import (
 	"math/big"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel"
@@ -40,8 +41,9 @@ func setupTestEthRPCTracing(t *testing.T, mockClient EthRPCClient) (EthRPCClient
 
 // mockEthRPCClient is a simple mock for testing
 type mockEthRPCClient struct {
-	headerByNumberFn func(ctx context.Context, number *big.Int) (*types.Header, error)
-	getTxsFn         func(ctx context.Context) ([]string, error)
+	headerByNumberFn  func(ctx context.Context, number *big.Int) (*types.Header, error)
+	getTxsFn          func(ctx context.Context) ([]string, error)
+	getNextProposerFn func(ctx context.Context, number *big.Int) (common.Hash, error)
 }
 
 func (m *mockEthRPCClient) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
@@ -58,6 +60,13 @@ func (m *mockEthRPCClient) GetTxs(ctx context.Context) ([]string, error) {
 	return nil, nil
 }
 
+func (m *mockEthRPCClient) GetNextProposer(ctx context.Context, number *big.Int) (common.Hash, error) {
+	if m.getNextProposerFn != nil {
+		return m.getNextProposerFn(ctx, number)
+	}
+	return common.Hash{}, nil
+}
+
 func TestTracedEthRPCClient_HeaderByNumber_Success(t *testing.T) {
 	expectedHeader := &types.Header{
 		GasLimit: 30000000,

execution/evm/execution.go

@@ -160,6 +160,9 @@ type EthRPCClient interface {
 
 	// GetTxs retrieves pending transactions from the transaction pool.
 	GetTxs(ctx context.Context) ([]string, error)
+
+	// GetNextProposer retrieves the proposer selected by the execution layer.
+	GetNextProposer(ctx context.Context, number *big.Int) (common.Hash, error)
 }
 
 // EngineClient represents a client that interacts with an Ethereum execution engine
@@ -353,15 +356,15 @@ func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight
 		// Continue execution on error, as it might be transient
 	} else if found {
 		if stateRoot != nil {
-			return execution.ExecuteResult{UpdatedStateRoot: stateRoot}, nil
+			return c.executionResultAfterBlock(ctx, stateRoot, blockHeight)
 		}
 		if payloadID != nil {
 			// Found in-progress execution, attempt to resume
 			stateRoot, err := c.processPayload(ctx, *payloadID, txs)
 			if err != nil {
 				return execution.ExecuteResult{}, err
 			}
-			return execution.ExecuteResult{UpdatedStateRoot: stateRoot}, nil
+			return c.executionResultAfterBlock(ctx, stateRoot, blockHeight)
 		}
 	}
 
@@ -461,7 +464,7 @@ func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight
 	if err != nil {
 		return execution.ExecuteResult{}, err
 	}
-	return execution.ExecuteResult{UpdatedStateRoot: stateRoot}, nil
+	return c.executionResultAfterBlock(ctx, stateRoot, blockHeight)
 }
 
 // setHead updates the head block hash without changing safe or finalized.
@@ -858,21 +861,81 @@ func (c *EngineClient) filterTransactions(txs [][]byte) []string {
 
 // GetExecutionInfo returns current execution layer parameters.
 func (c *EngineClient) GetExecutionInfo(ctx context.Context) (execution.ExecutionInfo, error) {
+	var info execution.ExecutionInfo
 	if cached := c.cachedExecutionInfo.Load(); cached != nil {
-		return *cached, nil
+		info.MaxGas = cached.MaxGas
+	} else {
+		header, err := c.ethClient.HeaderByNumber(ctx, nil) // nil = latest
+		if err != nil {
+			return execution.ExecutionInfo{}, fmt.Errorf("failed to get latest block: %w", err)
+		}
+
+		info.MaxGas = header.GasLimit
+		c.cachedExecutionInfo.Store(&execution.ExecutionInfo{MaxGas: info.MaxGas})
 	}
 
-	header, err := c.ethClient.HeaderByNumber(ctx, nil) // nil = latest
+	nextProposer, err := c.ethClient.GetNextProposer(ctx, nil)
 	if err != nil {
-		return execution.ExecutionInfo{}, fmt.Errorf("failed to get latest block: %w", err)
+		if !isRPCMethodNotFound(err) {
+			return execution.ExecutionInfo{}, fmt.Errorf("failed to get next proposer: %w", err)
+		}
+		return info, nil
+	}
+	if nextProposer != (common.Hash{}) {
+		info.NextProposerAddress = nextProposer.Bytes()
 	}
-
-	info := execution.ExecutionInfo{MaxGas: header.GasLimit}
-	c.cachedExecutionInfo.Store(&info)
 
 	return info, nil
 }
 
+func isRPCMethodNotFound(err error) bool {
+	var rpcErr rpc.Error
+	return errors.As(err, &rpcErr) && rpcErr.ErrorCode() == -32601
+}
+
+func (c *EngineClient) executionResultAfterBlock(ctx context.Context, stateRoot []byte, executedBlockHeight uint64) (execution.ExecuteResult, error) {
+	nextProposer, err := c.nextProposerChangeAfterBlock(ctx, executedBlockHeight)
+	if err != nil {
+		return execution.ExecuteResult{}, err
+	}
+	return execution.ExecuteResult{
+		UpdatedStateRoot:    stateRoot,
+		NextProposerAddress: nextProposer,
+	}, nil
+}
+
+func (c *EngineClient) nextProposerChangeAfterBlock(ctx context.Context, executedBlockHeight uint64) ([]byte, error) {
+	if executedBlockHeight == 0 {
+		return nil, nil
+	}
+
+	before, supported, err := c.nextProposerAtBlock(ctx, executedBlockHeight-1)
+	if err != nil || !supported {
+		return nil, err
+	}
+
+	after, supported, err := c.nextProposerAtBlock(ctx, executedBlockHeight)
+	if err != nil || !supported {
+		return nil, err
+	}
+
+	if after == (common.Hash{}) || after == before {
+		return nil, nil
+	}
+	return after.Bytes(), nil
+}
+
+func (c *EngineClient) nextProposerAtBlock(ctx context.Context, blockHeight uint64) (common.Hash, bool, error) {
+	proposer, err := c.ethClient.GetNextProposer(ctx, new(big.Int).SetUint64(blockHeight))
+	if err != nil {
+		if isRPCMethodNotFound(err) {
+			return common.Hash{}, false, nil
+		}
+		return common.Hash{}, true, fmt.Errorf("failed to get next proposer at block %d: %w", blockHeight, err)
+	}
+	return proposer, true, nil
+}
+
 // FilterTxs validates force-included transactions and applies gas and size filtering for all passed txs.
 // If hasForceIncludedTransaction is false, skip filtering entirely - mempool batch is already filtered.
 // Returns a slice of FilterStatus for each transaction.

execution/evm/execution_reconcile_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/beacon/engine"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	ds "github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
@@ -123,3 +124,7 @@ func (mockReconcileEthRPCClient) HeaderByNumber(_ context.Context, _ *big.Int) (
 func (mockReconcileEthRPCClient) GetTxs(_ context.Context) ([]string, error) {
 	return nil, errors.New("unexpected GetTxs call")
 }
+
+func (mockReconcileEthRPCClient) GetNextProposer(_ context.Context, _ *big.Int) (common.Hash, error) {
+	return common.Hash{}, errors.New("unexpected GetNextProposer call")
+}

execution/evm/proposer_test.go

@@ -0,0 +1,132 @@
+package evm
+
+import (
+	"context"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/beacon/engine"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	ds "github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-datastore/sync"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+type proposerEthRPCClient struct {
+	headerByNumberFn   func(ctx context.Context, number *big.Int) (*types.Header, error)
+	getTxsFn           func(ctx context.Context) ([]string, error)
+	getNextProposerFn  func(ctx context.Context, number *big.Int) (common.Hash, error)
+	nextProposerBlocks []*big.Int
+}
+
+func (m *proposerEthRPCClient) HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error) {
+	if m.headerByNumberFn != nil {
+		return m.headerByNumberFn(ctx, number)
+	}
+	return &types.Header{GasLimit: 30_000_000}, nil
+}
+
+func (m *proposerEthRPCClient) GetTxs(ctx context.Context) ([]string, error) {
+	if m.getTxsFn != nil {
+		return m.getTxsFn(ctx)
+	}
+	return nil, nil
+}
+
+func (m *proposerEthRPCClient) GetNextProposer(ctx context.Context, number *big.Int) (common.Hash, error) {
+	m.nextProposerBlocks = append(m.nextProposerBlocks, number)
+	if m.getNextProposerFn != nil {
+		return m.getNextProposerFn(ctx, number)
+	}
+	return common.Hash{}, nil
+}
+
+func TestGetExecutionInfoIncludesNextProposer(t *testing.T) {
+	nextProposer := common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+	ethClient := &proposerEthRPCClient{
+		getNextProposerFn: func(ctx context.Context, number *big.Int) (common.Hash, error) {
+			require.Nil(t, number)
+			return nextProposer, nil
+		},
+	}
+	client := &EngineClient{ethClient: ethClient}
+
+	info, err := client.GetExecutionInfo(t.Context())
+
+	require.NoError(t, err)
+	require.Equal(t, uint64(30_000_000), info.MaxGas)
+	require.Equal(t, nextProposer.Bytes(), info.NextProposerAddress)
+	require.Len(t, ethClient.nextProposerBlocks, 1)
+}
+
+func TestExecuteTxsReturnsNextProposerWhenChanged(t *testing.T) {
+	timestamp := time.Unix(1_700_000_000, 0)
+	stateRoot := common.HexToHash("0x01")
+	prevProposer := common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+	nextProposer := common.HexToHash("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+
+	store := NewEVMStore(dssync.MutexWrap(ds.NewMapDatastore()))
+	require.NoError(t, store.SaveExecMeta(t.Context(), &ExecMeta{
+		Height:    2,
+		StateRoot: stateRoot.Bytes(),
+		Timestamp: timestamp.Unix(),
+		Stage:     ExecStagePromoted,
+	}))
+
+	ethClient := &proposerEthRPCClient{
+		headerByNumberFn: func(ctx context.Context, number *big.Int) (*types.Header, error) {
+			require.Equal(t, int64(2), number.Int64())
+			return &types.Header{
+				Number:   big.NewInt(2),
+				Time:     uint64(timestamp.Unix()),
+				Root:     stateRoot,
+				GasLimit: 30_000_000,
+			}, nil
+		},
+		getNextProposerFn: func(ctx context.Context, number *big.Int) (common.Hash, error) {
+			switch number.Uint64() {
+			case 1:
+				return prevProposer, nil
+			case 2:
+				return nextProposer, nil
+			default:
+				t.Fatalf("unexpected proposer block %v", number)
+				return common.Hash{}, nil
+			}
+		},
+	}
+	client := &EngineClient{
+		engineClient:              proposerEngineRPCClient{},
+		ethClient:                 ethClient,
+		store:                     store,
+		currentSafeBlockHash:      common.HexToHash("0x10"),
+		currentFinalizedBlockHash: common.HexToHash("0x10"),
+		logger:                    zerolog.Nop(),
+	}
+
+	result, err := client.ExecuteTxs(t.Context(), nil, 2, timestamp, nil)
+
+	require.NoError(t, err)
+	require.Equal(t, stateRoot.Bytes(), result.UpdatedStateRoot)
+	require.Equal(t, nextProposer.Bytes(), result.NextProposerAddress)
+	require.Len(t, ethClient.nextProposerBlocks, 2)
+}
+
+type proposerEngineRPCClient struct{}
+
+func (proposerEngineRPCClient) ForkchoiceUpdated(context.Context, engine.ForkchoiceStateV1, map[string]any) (*engine.ForkChoiceResponse, error) {
+	return &engine.ForkChoiceResponse{
+		PayloadStatus: engine.PayloadStatusV1{Status: engine.VALID},
+	}, nil
+}
+
+func (proposerEngineRPCClient) GetPayload(context.Context, engine.PayloadID) (*engine.ExecutionPayloadEnvelope, error) {
+	return nil, nil
+}
+
+func (proposerEngineRPCClient) NewPayload(context.Context, *engine.ExecutableData, []string, string, [][]byte) (*engine.PayloadStatusV1, error) {
+	return nil, nil
+}