consolidate

Author: tac0turtle

apps/evm/cmd/run.go

@@ -77,6 +77,11 @@ var RunCmd = &cobra.Command{
 			return err
 		}
 
+		// Attach store to the EVM engine client for ExecMeta tracking (idempotent execution)
+		if ec, ok := executor.(*evm.EngineClient); ok {
+			ec.SetStore(store.New(datastore))
+		}
+
 		genesisPath := filepath.Join(filepath.Dir(nodeConfig.ConfigPath()), "genesis.json")
 		genesis, err := genesispkg.LoadGenesis(genesisPath)
 		if err != nil {
@@ -250,7 +255,7 @@ func createExecutionClient(cmd *cobra.Command) (execution.Executor, error) {
 	genesisHash := common.HexToHash(genesisHashStr)
 	feeRecipient := common.HexToAddress(feeRecipientStr)
 
-	return evm.NewEngineExecutionClient(ethURL, engineURL, jwtSecret, genesisHash, feeRecipient)
+	return evm.NewEngineExecutionClient(ethURL, engineURL, jwtSecret, genesisHash, feeRecipient, nil)
 }
 
 // addFlags adds flags related to the EVM execution client

block/internal/executing/executor.go

@@ -3,7 +3,6 @@ package executing
 import (
 	"bytes"
 	"context"
-	"crypto/sha256"
 	"errors"
 	"fmt"
 	"sync"
@@ -26,15 +25,6 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
-// payloadResumer is an optional interface that EVM execution clients can implement
-// to support resuming in-progress payload builds after crashes.
-// This is defined locally to avoid coupling the core interface to EVM-specific concepts.
-type payloadResumer interface {
-	// ResumePayload resumes an in-progress payload build using a stored payloadID.
-	// This allows crash recovery without creating sibling blocks.
-	ResumePayload(ctx context.Context, payloadID []byte) (stateRoot []byte, err error)
-}
-
 // Executor handles block production, transaction processing, and state management
 type Executor struct {
 	// Core components
@@ -380,25 +370,6 @@ func (e *Executor) produceBlock() error {
 			return fmt.Errorf("failed to save block data: %w", err)
 		}
 
-		// Save ExecMeta with Stage="started" for crash recovery and idempotent execution
-		execMeta := &store.ExecMeta{
-			Height:        newHeight,
-			Timestamp:     header.Time().Unix(),
-			Stage:         store.ExecStageStarted,
-			UpdatedAtUnix: time.Now().Unix(),
-		}
-		// Compute tx hash for sanity checks on retry
-		if len(data.Txs) > 0 {
-			h := sha256.New()
-			for _, tx := range data.Txs {
-				h.Write(tx)
-			}
-			execMeta.TxHash = h.Sum(nil)
-		}
-		if err = batch.SaveExecMeta(execMeta); err != nil {
-			return fmt.Errorf("failed to save exec meta: %w", err)
-		}
-
 		if err = batch.Commit(); err != nil {
 			return fmt.Errorf("failed to commit early save batch: %w", err)
 		}
@@ -452,18 +423,6 @@ func (e *Executor) produceBlock() error {
 		return fmt.Errorf("failed to update state: %w", err)
 	}
 
-	// Update ExecMeta to Stage="promoted" after successful execution
-	execMeta := &store.ExecMeta{
-		Height:        newHeight,
-		Timestamp:     header.Time().Unix(),
-		StateRoot:     newState.AppHash,
-		Stage:         store.ExecStagePromoted,
-		UpdatedAtUnix: time.Now().Unix(),
-	}
-	if err := batch.SaveExecMeta(execMeta); err != nil {
-		return fmt.Errorf("failed to update exec meta to promoted: %w", err)
-	}
-
 	if err := batch.Commit(); err != nil {
 		return fmt.Errorf("failed to commit batch: %w", err)
 	}
@@ -666,63 +625,11 @@ func (e *Executor) signHeader(header types.Header) (types.Signature, error) {
 }
 
 // executeTxsWithRetry executes transactions with retry logic.
-// It first checks ExecMeta for idempotent execution - if a block was already built
-// at this height, it returns the stored StateRoot instead of rebuilding.
-// If a payloadID exists (started but not promoted), it attempts to resume the payload
-// using the PayloadResumer interface if available.
-// NOTE: the function retries the execution client call regardless of the error. Some execution clients errors are irrecoverable, and will eventually halt the node, as expected.
+// Idempotent execution and crash recovery are handled by the execution client
+// (e.g., EngineClient) via ExecMeta tracking.
+// NOTE: the function retries the execution client call regardless of the error.
+// Some execution client errors are irrecoverable, and will eventually halt the node, as expected.
 func (e *Executor) executeTxsWithRetry(ctx context.Context, rawTxs [][]byte, header types.Header, currentState types.State) ([]byte, error) {
-	height := header.Height()
-
-	// Check ExecMeta for idempotent execution
-	// If we already have a promoted block at this height, return the stored StateRoot
-	execMeta, err := e.store.GetExecMeta(ctx, height)
-	if err == nil && execMeta != nil {
-		if execMeta.Stage == store.ExecStagePromoted && len(execMeta.StateRoot) > 0 {
-			e.logger.Info().
-				Uint64("height", height).
-				Str("stage", execMeta.Stage).
-				Msg("executeTxsWithRetry: reusing already-promoted execution (idempotent)")
-			return execMeta.StateRoot, nil
-		}
-
-		// If we have a started execution with a payloadID, try to resume
-		// This handles crash recovery where we got a payloadID but didn't complete the build
-		if execMeta.Stage == store.ExecStageStarted && len(execMeta.PayloadID) > 0 {
-			e.logger.Info().
-				Uint64("height", height).
-				Str("stage", execMeta.Stage).
-				Msg("executeTxsWithRetry: found in-progress execution with payloadID, attempting resume")
-
-			// Check if the executor implements payloadResumer (EVM-specific)
-			if resumer, ok := e.exec.(payloadResumer); ok {
-				stateRoot, err := resumer.ResumePayload(ctx, execMeta.PayloadID)
-				if err == nil {
-					e.logger.Info().
-						Uint64("height", height).
-						Msg("executeTxsWithRetry: successfully resumed payload")
-					return stateRoot, nil
-				}
-				// Resume failed - log and fall through to normal execution
-				// The EL-level idempotency check will handle if the block was already built
-				e.logger.Warn().Err(err).
-					Uint64("height", height).
-					Msg("executeTxsWithRetry: failed to resume payload, falling back to normal execution")
-			} else {
-				e.logger.Debug().
-					Uint64("height", height).
-					Msg("executeTxsWithRetry: executor does not support PayloadResumer, using normal execution")
-			}
-		} else if execMeta.Stage == store.ExecStageStarted {
-			// Started but no payloadID - log and proceed normally
-			// The EL-level idempotency check in ExecuteTxs will handle reusing the block
-			e.logger.Debug().
-				Uint64("height", height).
-				Str("stage", execMeta.Stage).
-				Msg("executeTxsWithRetry: found in-progress execution without payloadID, will attempt EL-level idempotency")
-		}
-	}
-
 	for attempt := 1; attempt <= common.MaxRetriesBeforeHalt; attempt++ {
 		newAppHash, _, err := e.exec.ExecuteTxs(ctx, rawTxs, header.Height(), header.Time(), currentState.AppHash)
 		if err != nil {

execution/evm/execution.go

@@ -2,6 +2,7 @@ package evm
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -20,6 +21,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/core/execution"
+	"github.com/evstack/ev-node/pkg/store"
 )
 
 const (
@@ -125,6 +127,10 @@ type EngineClient struct {
 	initialHeight uint64
 	feeRecipient  common.Address // Address to receive transaction fees
 
+	// store provides persistence for ExecMeta to enable idempotent execution
+	// and crash recovery. Optional - if nil, ExecMeta tracking is disabled.
+	store store.Store
+
 	mu                        sync.Mutex  // Mutex to protect concurrent access to block hashes
 	currentHeadBlockHash      common.Hash // Store last non-finalized HeadBlockHash
 	currentHeadHeight         uint64      // Height of the current head block (for safe lag calculation)
@@ -139,13 +145,16 @@ type EngineClient struct {
 	logger zerolog.Logger
 }
 
-// NewEngineExecutionClient creates a new instance of EngineAPIExecutionClient
+// NewEngineExecutionClient creates a new instance of EngineAPIExecutionClient.
+// The store parameter is optional - if provided, ExecMeta tracking is enabled
+// for idempotent execution and crash recovery.
 func NewEngineExecutionClient(
 	ethURL,
 	engineURL string,
 	jwtSecret string,
 	genesisHash common.Hash,
 	feeRecipient common.Address,
+	evStore store.Store,
 ) (*EngineClient, error) {
 	ethClient, err := ethclient.Dial(ethURL)
 	if err != nil {
@@ -178,6 +187,7 @@ func NewEngineExecutionClient(
 		ethClient:                 ethClient,
 		genesisHash:               genesisHash,
 		feeRecipient:              feeRecipient,
+		store:                     evStore,
 		currentHeadBlockHash:      genesisHash,
 		currentSafeBlockHash:      genesisHash,
 		currentFinalizedBlockHash: genesisHash,
@@ -190,6 +200,13 @@ func (c *EngineClient) SetLogger(l zerolog.Logger) {
 	c.logger = l
 }
 
+// SetStore allows callers to attach a store for ExecMeta tracking.
+// This enables idempotent execution and crash recovery features.
+// Must be called before ExecuteTxs if ExecMeta tracking is desired.
+func (c *EngineClient) SetStore(s store.Store) {
+	c.store = s
+}
+
 // InitChain initializes the blockchain with the given genesis parameters
 func (c *EngineClient) InitChain(ctx context.Context, genesisTime time.Time, initialHeight uint64, chainID string) ([]byte, uint64, error) {
 	if initialHeight != 1 {
@@ -263,11 +280,51 @@ func (c *EngineClient) GetTxs(ctx context.Context) ([][]byte, error) {
 // ExecuteTxs executes the given transactions at the specified block height and timestamp.
 // This method is serialized via executeMu to prevent concurrent block builds that could
 // create sibling blocks (fork explosion).
+//
+// ExecMeta tracking (if store is configured):
+// - Checks for already-promoted blocks to enable idempotent execution
+// - Saves ExecMeta with payloadID after forkchoiceUpdatedV3 for crash recovery
+// - Updates ExecMeta to "promoted" after successful execution
 func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight uint64, timestamp time.Time, prevStateRoot []byte) (updatedStateRoot []byte, maxBytes uint64, err error) {
 	// Serialize all ExecuteTxs calls to prevent concurrent sibling block creation
 	c.executeMu.Lock()
 	defer c.executeMu.Unlock()
 
+	// Check ExecMeta for idempotent execution (if store is configured)
+	if c.store != nil {
+		execMeta, err := c.store.GetExecMeta(ctx, blockHeight)
+		if err == nil && execMeta != nil {
+			// If we already have a promoted block at this height, return the stored StateRoot
+			if execMeta.Stage == store.ExecStagePromoted && len(execMeta.StateRoot) > 0 {
+				c.logger.Info().
+					Uint64("height", blockHeight).
+					Str("stage", execMeta.Stage).
+					Msg("ExecuteTxs: reusing already-promoted execution (idempotent)")
+				return execMeta.StateRoot, 0, nil
+			}
+
+			// If we have a started execution with a payloadID, try to resume
+			if execMeta.Stage == store.ExecStageStarted && len(execMeta.PayloadID) > 0 {
+				c.logger.Info().
+					Uint64("height", blockHeight).
+					Str("stage", execMeta.Stage).
+					Msg("ExecuteTxs: found in-progress execution with payloadID, attempting resume")
+
+				stateRoot, err := c.ResumePayload(ctx, execMeta.PayloadID)
+				if err == nil {
+					c.logger.Info().
+						Uint64("height", blockHeight).
+						Msg("ExecuteTxs: successfully resumed payload")
+					return stateRoot, 0, nil
+				}
+				// Resume failed - log and fall through to EL-level idempotency check
+				c.logger.Warn().Err(err).
+					Uint64("height", blockHeight).
+					Msg("ExecuteTxs: failed to resume payload, falling back to EL-level check")
+			}
+		}
+	}
+
 	// Idempotency check: if EL already has a block at this height with matching
 	// timestamp, return its state root instead of building a new block.
 	// This handles retries and crash recovery without creating sibling blocks.
@@ -279,14 +336,19 @@ func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight
 				Uint64("height", blockHeight).
 				Str("blockHash", existingBlockHash.Hex()).
 				Str("stateRoot", existingStateRoot.Hex()).
-				Msg("ExecuteTxs: reusing existing block at height (idempotency)")
+				Msg("ExecuteTxs: reusing existing block at height (EL idempotency)")
 
 			// Update head to point to this existing block
 			if err := c.setHead(ctx, existingBlockHash); err != nil {
 				c.logger.Warn().Err(err).Msg("ExecuteTxs: failed to update head to existing block")
 				// Continue anyway - the block exists and we can return its state root
 			}
 
+			// Update ExecMeta to promoted if store is configured
+			if c.store != nil {
+				c.saveExecMeta(ctx, blockHeight, timestamp.Unix(), nil, existingBlockHash[:], existingStateRoot.Bytes(), txs, store.ExecStagePromoted)
+			}
+
 			return existingStateRoot.Bytes(), 0, nil
 		}
 		// Timestamp mismatch - this is unexpected, log a warning but proceed
@@ -416,6 +478,12 @@ func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight
 		return nil, 0, err
 	}
 
+	// Save ExecMeta with payloadID for crash recovery (Stage="started")
+	// This allows resuming the payload build if we crash before completing
+	if c.store != nil {
+		c.saveExecMeta(ctx, blockHeight, timestamp.Unix(), payloadID[:], nil, nil, txs, store.ExecStageStarted)
+	}
+
 	// get payload
 	var payloadResult engine.ExecutionPayloadEnvelope
 	err = c.engineClient.CallContext(ctx, &payloadResult, "engine_getPayloadV4", *payloadID)
@@ -459,6 +527,11 @@ func (c *EngineClient) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight
 		return nil, 0, err
 	}
 
+	// Update ExecMeta to "promoted" after successful execution
+	if c.store != nil {
+		c.saveExecMeta(ctx, blockHeight, timestamp.Unix(), payloadID[:], blockHash[:], payloadResult.ExecutionPayload.StateRoot.Bytes(), txs, store.ExecStagePromoted)
+	}
+
 	return payloadResult.ExecutionPayload.StateRoot.Bytes(), payloadResult.ExecutionPayload.GasUsed, nil
 }
 
@@ -689,6 +762,11 @@ func (c *EngineClient) ResumePayload(ctx context.Context, payloadIDBytes []byte)
 		Uint64("blockHeight", blockHeight).
 		Msg("ResumePayload: successfully resumed payload")
 
+	// Update ExecMeta to "promoted" after successful resume
+	if c.store != nil {
+		c.saveExecMeta(ctx, blockHeight, int64(payloadResult.ExecutionPayload.Timestamp), payloadIDBytes, blockHash[:], payloadResult.ExecutionPayload.StateRoot.Bytes(), nil, store.ExecStagePromoted)
+	}
+
 	return payloadResult.ExecutionPayload.StateRoot.Bytes(), nil
 }
 
@@ -706,6 +784,54 @@ func (c *EngineClient) getBlockInfo(ctx context.Context, height uint64) (common.
 	return header.Hash(), header.Root, header.GasLimit, header.Time, nil
 }
 
+// saveExecMeta persists execution metadata to the store for crash recovery.
+// This is a best-effort operation - failures are logged but don't fail the execution.
+func (c *EngineClient) saveExecMeta(ctx context.Context, height uint64, timestamp int64, payloadID []byte, blockHash []byte, stateRoot []byte, txs [][]byte, stage string) {
+	if c.store == nil {
+		return
+	}
+
+	execMeta := &store.ExecMeta{
+		Height:        height,
+		Timestamp:     timestamp,
+		PayloadID:     payloadID,
+		BlockHash:     blockHash,
+		StateRoot:     stateRoot,
+		Stage:         stage,
+		UpdatedAtUnix: time.Now().Unix(),
+	}
+
+	// Compute tx hash for sanity checks on retry
+	if len(txs) > 0 {
+		h := sha256.New()
+		for _, tx := range txs {
+			h.Write(tx)
+		}
+		execMeta.TxHash = h.Sum(nil)
+	}
+
+	batch, err := c.store.NewBatch(ctx)
+	if err != nil {
+		c.logger.Warn().Err(err).Uint64("height", height).Msg("saveExecMeta: failed to create batch")
+		return
+	}
+
+	if err := batch.SaveExecMeta(execMeta); err != nil {
+		c.logger.Warn().Err(err).Uint64("height", height).Msg("saveExecMeta: failed to save exec meta")
+		return
+	}
+
+	if err := batch.Commit(); err != nil {
+		c.logger.Warn().Err(err).Uint64("height", height).Msg("saveExecMeta: failed to commit batch")
+		return
+	}
+
+	c.logger.Debug().
+		Uint64("height", height).
+		Str("stage", stage).
+		Msg("saveExecMeta: saved execution metadata")
+}
+
 // GetLatestHeight returns the current block height of the execution layer
 func (c *EngineClient) GetLatestHeight(ctx context.Context) (uint64, error) {
 	header, err := c.ethClient.HeaderByNumber(ctx, nil) // nil = latest block