Merge branch 'main' into alex/adr023_ha-failover_raft11

* main:
  fix: inconsistent state detection and rollback (#2983)
  chore: improve graceful shutdown restarts (#2985)
  feat(submitting): add posting strategies (#2973)

Author: alpe

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `post-tx` command and force inclusion server to submit transaction directly to the DA layer. ([#2888](https://github.com/evstack/ev-node/pull/2888))
   Additionally, modified the core package to support marking transactions as forced included transactions.
   The execution client ought to perform basic validation on those transactions as they have skipped the execution client's mempool.
+- Add batching stategies (default stay time-based, unchanged with previous betas). Currently available strategies are `time`, `size`, `immediate` and `adaptive`.
 
 ### Changed
 

apps/evm/cmd/rollback.go

@@ -1,13 +1,18 @@
 package cmd
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"os"
 
+	"github.com/ethereum/go-ethereum/common"
+	ds "github.com/ipfs/go-datastore"
 	"github.com/spf13/cobra"
 
 	goheaderstore "github.com/celestiaorg/go-header/store"
+	"github.com/evstack/ev-node/execution/evm"
 	rollcmd "github.com/evstack/ev-node/pkg/cmd"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
@@ -42,7 +47,7 @@ func NewRollbackCmd() *cobra.Command {
 
 			defer func() {
 				if closeErr := rawEvolveDB.Close(); closeErr != nil {
-					fmt.Printf("Warning: failed to close evolve database: %v\n", closeErr)
+					cmd.Printf("Warning: failed to close evolve database: %v\n", closeErr)
 				}
 			}()
 
@@ -63,6 +68,17 @@ func NewRollbackCmd() *cobra.Command {
 				return fmt.Errorf("failed to rollback ev-node state: %w", err)
 			}
 
+			// rollback execution layer via EngineClient
+			engineClient, err := createRollbackEngineClient(cmd, rawEvolveDB)
+			if err != nil {
+				cmd.Printf("Warning: failed to create engine client, skipping EL rollback: %v\n", err)
+			} else {
+				if err := engineClient.Rollback(goCtx, height); err != nil {
+					return fmt.Errorf("failed to rollback execution layer: %w", err)
+				}
+				cmd.Printf("Rolled back execution layer to height %d\n", height)
+			}
+
 			// rollback ev-node goheader state
 			headerStore, err := goheaderstore.NewStore[*types.SignedHeader](
 				evolveDB,
@@ -101,7 +117,7 @@ func NewRollbackCmd() *cobra.Command {
 				errs = errors.Join(errs, fmt.Errorf("failed to rollback data sync service state: %w", err))
 			}
 
-			fmt.Printf("Rolled back ev-node state to height %d\n", height)
+			cmd.Printf("Rolled back ev-node state to height %d\n", height)
 			if syncNode {
 				fmt.Println("Restart the node with the `--evnode.clear_cache` flag")
 			}
@@ -113,5 +129,42 @@ func NewRollbackCmd() *cobra.Command {
 	cmd.Flags().Uint64Var(&height, "height", 0, "rollback to a specific height")
 	cmd.Flags().BoolVar(&syncNode, "sync-node", false, "sync node (no aggregator)")
 
+	// EVM flags for execution layer rollback
+	cmd.Flags().String(evm.FlagEvmEthURL, "http://localhost:8545", "URL of the Ethereum JSON-RPC endpoint")
+	cmd.Flags().String(evm.FlagEvmEngineURL, "http://localhost:8551", "URL of the Engine API endpoint")
+	cmd.Flags().String(evm.FlagEvmJWTSecretFile, "", "Path to file containing the JWT secret for authentication")
+
 	return cmd
 }
+
+func createRollbackEngineClient(cmd *cobra.Command, db ds.Batching) (*evm.EngineClient, error) {
+	ethURL, err := cmd.Flags().GetString(evm.FlagEvmEthURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get '%s' flag: %w", evm.FlagEvmEthURL, err)
+	}
+	engineURL, err := cmd.Flags().GetString(evm.FlagEvmEngineURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get '%s' flag: %w", evm.FlagEvmEngineURL, err)
+	}
+
+	jwtSecretFile, err := cmd.Flags().GetString(evm.FlagEvmJWTSecretFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get '%s' flag: %w", evm.FlagEvmJWTSecretFile, err)
+	}
+
+	if jwtSecretFile == "" {
+		return nil, fmt.Errorf("JWT secret file must be provided via --evm.jwt-secret-file for EL rollback")
+	}
+
+	secretBytes, err := os.ReadFile(jwtSecretFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read JWT secret from file '%s': %w", jwtSecretFile, err)
+	}
+	jwtSecret := string(bytes.TrimSpace(secretBytes))
+
+	if jwtSecret == "" {
+		return nil, fmt.Errorf("JWT secret file '%s' is empty", jwtSecretFile)
+	}
+
+	return evm.NewEngineExecutionClient(ethURL, engineURL, jwtSecret, common.Hash{}, common.Address{}, db, false)
+}

block/internal/cache/bench_test.go

@@ -72,7 +72,7 @@ func BenchmarkManager_GetPendingHeaders(b *testing.B) {
 			b.ReportAllocs()
 			b.ResetTimer()
 			for b.Loop() {
-				hs, err := m.GetPendingHeaders(ctx)
+				hs, _, err := m.GetPendingHeaders(ctx)
 				if err != nil {
 					b.Fatal(err)
 				}
@@ -93,7 +93,7 @@ func BenchmarkManager_GetPendingData(b *testing.B) {
 			b.ReportAllocs()
 			b.ResetTimer()
 			for b.Loop() {
-				ds, err := m.GetPendingData(ctx)
+				ds, _, err := m.GetPendingData(ctx)
 				if err != nil {
 					b.Fatal(err)
 				}

block/internal/cache/manager.go

@@ -78,8 +78,8 @@ type CacheManager interface {
 
 // PendingManager provides operations for managing pending headers and data
 type PendingManager interface {
-	GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, error)
-	GetPendingData(ctx context.Context) ([]*types.SignedData, error)
+	GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, [][]byte, error)
+	GetPendingData(ctx context.Context) ([]*types.SignedData, [][]byte, error)
 	SetLastSubmittedHeaderHeight(ctx context.Context, height uint64)
 	GetLastSubmittedHeaderHeight() uint64
 	SetLastSubmittedDataHeight(ctx context.Context, height uint64)
@@ -320,20 +320,21 @@ func (m *implementation) DeleteHeight(blockHeight uint64) {
 }
 
 // Pending operations
-func (m *implementation) GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, error) {
+func (m *implementation) GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, [][]byte, error) {
 	return m.pendingHeaders.GetPendingHeaders(ctx)
 }
 
-func (m *implementation) GetPendingData(ctx context.Context) ([]*types.SignedData, error) {
-	// Get pending raw data
-	dataList, err := m.pendingData.GetPendingData(ctx)
+func (m *implementation) GetPendingData(ctx context.Context) ([]*types.SignedData, [][]byte, error) {
+	// Get pending raw data with marshalled bytes
+	dataList, marshalledData, err := m.pendingData.GetPendingData(ctx)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	// Convert to SignedData (this logic was in manager.go)
 	signedDataList := make([]*types.SignedData, 0, len(dataList))
-	for _, data := range dataList {
+	marshalledSignedData := make([][]byte, 0, len(dataList))
+	for i, data := range dataList {
 		if len(data.Txs) == 0 {
 			continue // Skip empty data
 		}
@@ -344,9 +345,10 @@ func (m *implementation) GetPendingData(ctx context.Context) ([]*types.SignedDat
 			Data: *data,
 			// Signature and Signer will be set by executing component
 		})
+		marshalledSignedData = append(marshalledSignedData, marshalledData[i])
 	}
 
-	return signedDataList, nil
+	return signedDataList, marshalledSignedData, nil
 }
 
 func (m *implementation) GetLastSubmittedHeaderHeight() uint64 {

block/internal/cache/manager_test.go

@@ -183,14 +183,14 @@ func TestPendingHeadersAndData_Flow(t *testing.T) {
 	require.NoError(t, err)
 
 	// headers: all 3 should be pending initially
-	headers, err := cm.GetPendingHeaders(ctx)
+	headers, _, err := cm.GetPendingHeaders(ctx)
 	require.NoError(t, err)
 	require.Len(t, headers, 3)
 	assert.Equal(t, uint64(1), headers[0].Height())
 	assert.Equal(t, uint64(3), headers[2].Height())
 
 	// data: empty one filtered, so 2 and 3 only
-	signedData, err := cm.GetPendingData(ctx)
+	signedData, _, err := cm.GetPendingData(ctx)
 	require.NoError(t, err)
 	require.Len(t, signedData, 2)
 	assert.Equal(t, uint64(2), signedData[0].Height())
@@ -200,12 +200,12 @@ func TestPendingHeadersAndData_Flow(t *testing.T) {
 	cm.SetLastSubmittedHeaderHeight(ctx, 1)
 	cm.SetLastSubmittedDataHeight(ctx, 2)
 
-	headers, err = cm.GetPendingHeaders(ctx)
+	headers, _, err = cm.GetPendingHeaders(ctx)
 	require.NoError(t, err)
 	require.Len(t, headers, 2)
 	assert.Equal(t, uint64(2), headers[0].Height())
 
-	signedData, err = cm.GetPendingData(ctx)
+	signedData, _, err = cm.GetPendingData(ctx)
 	require.NoError(t, err)
 	require.Len(t, signedData, 1)
 	assert.Equal(t, uint64(3), signedData[0].Height())

block/internal/cache/pending_base.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	ds "github.com/ipfs/go-datastore"
@@ -22,6 +23,9 @@ type pendingBase[T any] struct {
 	metaKey    string
 	fetch      func(ctx context.Context, store store.Store, height uint64) (T, error)
 	lastHeight atomic.Uint64
+
+	// Marshalling cache to avoid redundant marshalling
+	marshalledCache sync.Map // key: uint64 (height), value: []byte
 }
 
 // newPendingBase constructs a new pendingBase for a given type.
@@ -84,6 +88,9 @@ func (pb *pendingBase[T]) setLastSubmittedHeight(ctx context.Context, newLastSub
 		if err != nil {
 			pb.logger.Error().Err(err).Msg("failed to store height of latest item submitted to DA")
 		}
+
+		// Clear marshalled cache for submitted heights
+		pb.clearMarshalledCacheUpTo(newLastSubmittedHeight)
 	}
 }
 
@@ -105,3 +112,26 @@ func (pb *pendingBase[T]) init() error {
 	pb.lastHeight.CompareAndSwap(0, lsh)
 	return nil
 }
+
+// getMarshalledForHeight returns cached marshalled bytes for a height, or nil if not cached
+func (pb *pendingBase[T]) getMarshalledForHeight(height uint64) []byte {
+	if val, ok := pb.marshalledCache.Load(height); ok {
+		return val.([]byte)
+	}
+	return nil
+}
+
+// setMarshalledForHeight caches marshalled bytes for a height
+func (pb *pendingBase[T]) setMarshalledForHeight(height uint64, marshalled []byte) {
+	pb.marshalledCache.Store(height, marshalled)
+}
+
+// clearMarshalledCacheUpTo removes cached marshalled bytes up to and including the given height
+func (pb *pendingBase[T]) clearMarshalledCacheUpTo(height uint64) {
+	pb.marshalledCache.Range(func(key, _ any) bool {
+		if h := key.(uint64); h <= height {
+			pb.marshalledCache.Delete(h)
+		}
+		return true
+	})
+}

block/internal/cache/pending_base_test.go

@@ -35,7 +35,7 @@ func TestPendingBase_ErrorConditions(t *testing.T) {
 	// ensure store height stays lower (0)
 	ph, err := NewPendingHeaders(st, logger)
 	require.NoError(t, err)
-	pending, err := ph.GetPendingHeaders(ctx)
+	pending, _, err := ph.GetPendingHeaders(ctx)
 	assert.Error(t, err)
 	assert.Len(t, pending, 0)
 

block/internal/cache/pending_data.go

@@ -3,6 +3,7 @@ package cache
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/rs/zerolog"
 
@@ -56,9 +57,42 @@ func (pd *PendingData) init() error {
 	return pd.base.init()
 }
 
-// GetPendingData returns a sorted slice of pending Data.
-func (pd *PendingData) GetPendingData(ctx context.Context) ([]*types.Data, error) {
-	return pd.base.getPending(ctx)
+// GetPendingData returns a sorted slice of pending Data along with their marshalled bytes.
+// It uses an internal cache to avoid re-marshalling data on subsequent calls.
+func (pd *PendingData) GetPendingData(ctx context.Context) ([]*types.Data, [][]byte, error) {
+	dataList, err := pd.base.getPending(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if len(dataList) == 0 {
+		return nil, nil, nil
+	}
+
+	marshalled := make([][]byte, len(dataList))
+	lastSubmitted := pd.base.lastHeight.Load()
+
+	for i, data := range dataList {
+		height := lastSubmitted + uint64(i) + 1
+
+		// Try to get from cache first
+		if cached := pd.base.getMarshalledForHeight(height); cached != nil {
+			marshalled[i] = cached
+			continue
+		}
+
+		// Marshal if not in cache
+		dataBytes, err := data.MarshalBinary()
+		if err != nil {
+			return nil, nil, fmt.Errorf("failed to marshal data at height %d: %w", height, err)
+		}
+		marshalled[i] = dataBytes
+
+		// Store in cache
+		pd.base.setMarshalledForHeight(height, dataBytes)
+	}
+
+	return dataList, marshalled, nil
 }
 
 func (pd *PendingData) NumPendingData() uint64 {

block/internal/cache/pending_data_test.go

@@ -39,7 +39,7 @@ func TestPendingData_BasicFlow(t *testing.T) {
 
 	// initially all 3 data items are pending, incl. empty
 	require.Equal(t, uint64(3), pendingData.NumPendingData())
-	pendingDataList, err := pendingData.GetPendingData(ctx)
+	pendingDataList, _, err := pendingData.GetPendingData(ctx)
 	require.NoError(t, err)
 	require.Len(t, pendingDataList, 3)
 	require.Equal(t, uint64(1), pendingDataList[0].Height())
@@ -53,7 +53,7 @@ func TestPendingData_BasicFlow(t *testing.T) {
 	require.Equal(t, uint64(1), binary.LittleEndian.Uint64(metadataRaw))
 
 	require.Equal(t, uint64(2), pendingData.NumPendingData())
-	pendingDataList, err = pendingData.GetPendingData(ctx)
+	pendingDataList, _, err = pendingData.GetPendingData(ctx)
 	require.NoError(t, err)
 	require.Len(t, pendingDataList, 2)
 	require.Equal(t, uint64(2), pendingDataList[0].Height())
@@ -97,7 +97,7 @@ func TestPendingData_GetPending_PropagatesFetchError(t *testing.T) {
 	require.NoError(t, err)
 
 	// fetching pending should propagate the not-found error from store
-	pending, err := pendingData.GetPendingData(ctx)
+	pending, _, err := pendingData.GetPendingData(ctx)
 	require.Error(t, err)
 	require.Empty(t, pending)
 }