updates

Author: julienrbrt

apps/evm/cmd/run.go

@@ -194,13 +194,8 @@ func createSequencer(
 		return nil, fmt.Errorf("failed to create single sequencer: %w", err)
 	}
 
-	// Configure DA transaction filter if executor supports it
-	if filter, ok := executor.(execution.DATransactionFilter); ok {
-		if infoProvider, ok := executor.(execution.ExecutionInfoProvider); ok {
-			sequencer.SetDATransactionFilter(filter, infoProvider)
-			logger.Info().Msg("DA transaction filter configured for gas-based filtering")
-		}
-	}
+	// Configure executor for DA transaction gas-based filtering
+	sequencer.SetExecutor(executor)
 
 	logger.Info().
 		Str("forced_inclusion_namespace", nodeConfig.DA.GetForcedInclusionNamespace()).

apps/testapp/kv/kvexecutor.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/evstack/ev-node/core/execution"
 	"github.com/evstack/ev-node/pkg/store"
 	ds "github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/query"
@@ -422,3 +423,30 @@ func (k *KVExecutor) Rollback(ctx context.Context, height uint64) error {
 func getTxKey(height uint64, txKey string) ds.Key {
 	return heightKeyPrefix.Child(ds.NewKey(fmt.Sprintf("%d/%s", height, txKey)))
 }
+
+// GetExecutionInfo returns execution layer parameters.
+// For KVExecutor, returns MaxGas=0 indicating no gas-based filtering.
+func (k *KVExecutor) GetExecutionInfo(ctx context.Context, height uint64) (execution.ExecutionInfo, error) {
+	return execution.ExecutionInfo{MaxGas: 0}, nil
+}
+
+// FilterDATransactions validates and filters force-included transactions from DA.
+// For KVExecutor, all transactions are considered valid (no gas-based filtering).
+// Invalid transactions (not in key=value format) are filtered out.
+func (k *KVExecutor) FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
+	// KVExecutor doesn't do gas filtering but does basic validation
+	validTxs := make([][]byte, 0, len(txs))
+	for _, tx := range txs {
+		if len(tx) == 0 {
+			continue // Skip empty transactions
+		}
+		// Basic format validation: must be key=value
+		parts := strings.SplitN(string(tx), "=", 2)
+		if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" {
+			continue // Filter out malformed transactions
+		}
+		validTxs = append(validTxs, tx)
+	}
+	// No gas-based filtering, so no remaining transactions
+	return validTxs, nil, nil
+}

core/execution/execution.go

@@ -142,3 +142,19 @@ type HeightProvider interface {
 	// - error: Any errors during height retrieval
 	GetLatestHeight(ctx context.Context) (uint64, error)
 }
+
+// ExecutionInfoProvider is an interface for components that can provide execution layer parameters.
+// This is useful for type assertions when an Executor implementation supports gas-based filtering.
+type ExecutionInfoProvider interface {
+	// GetExecutionInfo returns current execution layer parameters.
+	// See Executor.GetExecutionInfo for full documentation.
+	GetExecutionInfo(ctx context.Context, height uint64) (ExecutionInfo, error)
+}
+
+// DATransactionFilter is an interface for components that can filter DA transactions.
+// This is useful for type assertions when an Executor implementation supports gas-based filtering.
+type DATransactionFilter interface {
+	// FilterDATransactions validates and filters force-included transactions from DA.
+	// See Executor.FilterDATransactions for full documentation.
+	FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) (validTxs [][]byte, remainingTxs [][]byte, err error)
+}

execution/grpc/client.go

@@ -112,3 +112,41 @@ func (c *Client) SetFinal(ctx context.Context, blockHeight uint64) error {
 
 	return nil
 }
+
+// GetExecutionInfo returns current execution layer parameters.
+//
+// This method retrieves execution parameters such as the block gas limit
+// from the remote execution service.
+func (c *Client) GetExecutionInfo(ctx context.Context, height uint64) (execution.ExecutionInfo, error) {
+	req := connect.NewRequest(&pb.GetExecutionInfoRequest{
+		Height: height,
+	})
+
+	resp, err := c.client.GetExecutionInfo(ctx, req)
+	if err != nil {
+		return execution.ExecutionInfo{}, fmt.Errorf("grpc client: failed to get execution info: %w", err)
+	}
+
+	return execution.ExecutionInfo{
+		MaxGas: resp.Msg.MaxGas,
+	}, nil
+}
+
+// FilterDATransactions validates and filters force-included transactions from DA.
+//
+// This method sends DA transactions to the remote execution service for validation
+// and gas-based filtering. It returns transactions that are valid and fit within
+// the gas limit, plus any remaining valid transactions for re-queuing.
+func (c *Client) FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
+	req := connect.NewRequest(&pb.FilterDATransactionsRequest{
+		Txs:    txs,
+		MaxGas: maxGas,
+	})
+
+	resp, err := c.client.FilterDATransactions(ctx, req)
+	if err != nil {
+		return nil, nil, fmt.Errorf("grpc client: failed to filter DA transactions: %w", err)
+	}
+
+	return resp.Msg.ValidTxs, resp.Msg.RemainingTxs, nil
+}

execution/grpc/server.go

@@ -136,3 +136,39 @@ func (s *Server) SetFinal(
 
 	return connect.NewResponse(&pb.SetFinalResponse{}), nil
 }
+
+// GetExecutionInfo handles the GetExecutionInfo RPC request.
+//
+// It returns current execution layer parameters such as the block gas limit.
+func (s *Server) GetExecutionInfo(
+	ctx context.Context,
+	req *connect.Request[pb.GetExecutionInfoRequest],
+) (*connect.Response[pb.GetExecutionInfoResponse], error) {
+	info, err := s.executor.GetExecutionInfo(ctx, req.Msg.Height)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to get execution info: %w", err))
+	}
+
+	return connect.NewResponse(&pb.GetExecutionInfoResponse{
+		MaxGas: info.MaxGas,
+	}), nil
+}
+
+// FilterDATransactions handles the FilterDATransactions RPC request.
+//
+// It validates and filters force-included transactions from DA, returning
+// transactions that are valid and fit within the gas limit.
+func (s *Server) FilterDATransactions(
+	ctx context.Context,
+	req *connect.Request[pb.FilterDATransactionsRequest],
+) (*connect.Response[pb.FilterDATransactionsResponse], error) {
+	validTxs, remainingTxs, err := s.executor.FilterDATransactions(ctx, req.Msg.Txs, req.Msg.MaxGas)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to filter DA transactions: %w", err))
+	}
+
+	return connect.NewResponse(&pb.FilterDATransactionsResponse{
+		ValidTxs:     validTxs,
+		RemainingTxs: remainingTxs,
+	}), nil
+}

pkg/sequencers/single/sequencer.go

@@ -52,10 +52,10 @@ type Sequencer struct {
 	// Cached forced inclusion transactions from the current epoch
 	cachedForcedInclusionTxs [][]byte
 
-	// DA transaction filtering support (optional)
+	// Executor for DA transaction filtering (optional)
 	// When set, forced inclusion transactions are filtered by gas limit
-	txFilter     execution.DATransactionFilter
-	infoProvider execution.ExecutionInfoProvider
+	// using the executor's GetExecutionInfo and FilterDATransactions methods
+	executor execution.Executor
 }
 
 // NewSequencer creates a new Single Sequencer
@@ -122,13 +122,12 @@ func NewSequencer(
 	return s, nil
 }
 
-// SetDATransactionFilter sets the optional DA transaction filter and execution info provider.
+// SetExecutor sets the optional executor for DA transaction filtering.
 // When set, forced inclusion transactions will be filtered by gas limit before being included in batches.
 // This should be called after NewSequencer and before Start if filtering is desired.
-func (c *Sequencer) SetDATransactionFilter(filter execution.DATransactionFilter, infoProvider execution.ExecutionInfoProvider) {
-	c.txFilter = filter
-	c.infoProvider = infoProvider
-	c.logger.Info().Msg("DA transaction filter configured for gas-based filtering")
+func (c *Sequencer) SetExecutor(executor execution.Executor) {
+	c.executor = executor
+	c.logger.Info().Msg("Executor configured for DA transaction gas-based filtering")
 }
 
 // getInitialDAStartHeight retrieves the DA height of the first included chain height from store.
@@ -216,18 +215,18 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 	// Process forced inclusion transactions from checkpoint position
 	forcedTxs := c.processForcedInclusionTxsFromCheckpoint(req.MaxBytes)
 
-	// Apply gas-based filtering if filter is configured
+	// Apply gas-based filtering if executor is configured
 	var filteredForcedTxs [][]byte
 	var remainingGasFilteredTxs [][]byte
-	if c.txFilter != nil && c.infoProvider != nil && len(forcedTxs) > 0 {
+	if c.executor != nil && len(forcedTxs) > 0 {
 		// Get current gas limit from execution layer
-		info, err := c.infoProvider.GetExecutionInfo(ctx, 0) // 0 = latest/next block
+		info, err := c.executor.GetExecutionInfo(ctx, 0) // 0 = latest/next block
 		if err != nil {
 			c.logger.Warn().Err(err).Msg("failed to get execution info for gas filtering, proceeding without gas filter")
 			filteredForcedTxs = forcedTxs
 		} else if info.MaxGas > 0 {
 			// Filter by gas limit
-			filteredForcedTxs, remainingGasFilteredTxs, err = c.txFilter.FilterDATransactions(ctx, forcedTxs, info.MaxGas)
+			filteredForcedTxs, remainingGasFilteredTxs, err = c.executor.FilterDATransactions(ctx, forcedTxs, info.MaxGas)
 			if err != nil {
 				c.logger.Warn().Err(err).Msg("failed to filter DA transactions by gas, proceeding without gas filter")
 				filteredForcedTxs = forcedTxs

pkg/sequencers/single/sequencer_test.go

@@ -1010,26 +1010,40 @@ func TestSequencer_GetNextBatch_EmptyDABatch_IncreasesDAHeight(t *testing.T) {
 	assert.Equal(t, uint64(0), seq.checkpoint.TxIndex)
 }
 
-// mockDATransactionFilter is a mock implementation of execution.DATransactionFilter
-type mockDATransactionFilter struct {
-	filterFunc func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error)
+// mockExecutor is a mock implementation of execution.Executor for testing gas filtering
+type mockExecutor struct {
+	maxGas          uint64
+	getInfoErr      error
+	filterFunc      func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error)
+	filterCallCount int
 }
 
-func (m *mockDATransactionFilter) FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
-	if m.filterFunc != nil {
-		return m.filterFunc(ctx, txs, maxGas)
-	}
-	return txs, nil, nil
+func (m *mockExecutor) InitChain(ctx context.Context, genesisTime time.Time, initialHeight uint64, chainID string) ([]byte, error) {
+	return []byte("state-root"), nil
+}
+
+func (m *mockExecutor) GetTxs(ctx context.Context) ([][]byte, error) {
+	return nil, nil
 }
 
-// mockExecutionInfoProvider is a mock implementation of execution.ExecutionInfoProvider
-type mockExecutionInfoProvider struct {
-	maxGas uint64
-	err    error
+func (m *mockExecutor) ExecuteTxs(ctx context.Context, txs [][]byte, blockHeight uint64, timestamp time.Time, prevStateRoot []byte) ([]byte, error) {
+	return []byte("new-state-root"), nil
 }
 
-func (m *mockExecutionInfoProvider) GetExecutionInfo(ctx context.Context, height uint64) (execution.ExecutionInfo, error) {
-	return execution.ExecutionInfo{MaxGas: m.maxGas}, m.err
+func (m *mockExecutor) SetFinal(ctx context.Context, blockHeight uint64) error {
+	return nil
+}
+
+func (m *mockExecutor) GetExecutionInfo(ctx context.Context, height uint64) (execution.ExecutionInfo, error) {
+	return execution.ExecutionInfo{MaxGas: m.maxGas}, m.getInfoErr
+}
+
+func (m *mockExecutor) FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
+	m.filterCallCount++
+	if m.filterFunc != nil {
+		return m.filterFunc(ctx, txs, maxGas)
+	}
+	return txs, nil, nil
 }
 
 func TestSequencer_GetNextBatch_WithGasFiltering(t *testing.T) {
@@ -1070,11 +1084,10 @@ func TestSequencer_GetNextBatch_WithGasFiltering(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Configure the gas filter mock
-	filterCallCount := 0
-	mockFilter := &mockDATransactionFilter{
+	// Configure the executor mock
+	mockExec := &mockExecutor{
+		maxGas: 1000000, // 1M gas limit
 		filterFunc: func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
-			filterCallCount++
 			// Simulate: first 2 txs fit, third one doesn't
 			if len(txs) >= 3 {
 				return txs[:2], txs[2:], nil
@@ -1083,12 +1096,8 @@ func TestSequencer_GetNextBatch_WithGasFiltering(t *testing.T) {
 		},
 	}
 
-	mockInfoProvider := &mockExecutionInfoProvider{
-		maxGas: 1000000, // 1M gas limit
-	}
-
-	// Set the filter
-	seq.SetDATransactionFilter(mockFilter, mockInfoProvider)
+	// Set the executor
+	seq.SetExecutor(mockExec)
 
 	// Manually set up cached forced txs to simulate DA fetch
 	seq.cachedForcedInclusionTxs = forcedTxs
@@ -1118,11 +1127,10 @@ func TestSequencer_GetNextBatch_WithGasFiltering(t *testing.T) {
 	assert.Equal(t, uint64(0), seq.checkpoint.TxIndex) // Reset because we replaced the cache
 
 	// Filter should have been called
-	assert.Equal(t, 1, filterCallCount)
+	assert.Equal(t, 1, mockExec.filterCallCount)
 
 	// Second call should return the remaining tx
-	mockFilter.filterFunc = func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
-		filterCallCount++
+	mockExec.filterFunc = func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
 		// Now all remaining txs fit
 		return txs, nil, nil
 	}
@@ -1167,18 +1175,15 @@ func TestSequencer_GetNextBatch_GasFilterError(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Configure filter that returns error
-	mockFilter := &mockDATransactionFilter{
+	// Configure executor that returns filter error
+	mockExec := &mockExecutor{
+		maxGas: 1000000,
 		filterFunc: func(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
 			return nil, nil, errors.New("filter error")
 		},
 	}
 
-	mockInfoProvider := &mockExecutionInfoProvider{
-		maxGas: 1000000,
-	}
-
-	seq.SetDATransactionFilter(mockFilter, mockInfoProvider)
+	seq.SetExecutor(mockExec)
 
 	// Set up cached txs
 	forcedTxs := [][]byte{[]byte("tx1"), []byte("tx2")}

pkg/telemetry/executor_tracing.go

@@ -106,18 +106,14 @@ func (t *tracedExecutor) GetLatestHeight(ctx context.Context) (uint64, error) {
 	return height, err
 }
 
-// If the underlying executor implements ExecutionInfoProvider, forward it while tracing.
+// GetExecutionInfo forwards to the inner executor with tracing.
 func (t *tracedExecutor) GetExecutionInfo(ctx context.Context, height uint64) (coreexec.ExecutionInfo, error) {
-	eip, ok := t.inner.(coreexec.ExecutionInfoProvider)
-	if !ok {
-		return coreexec.ExecutionInfo{}, nil
-	}
 	ctx, span := t.tracer.Start(ctx, "Executor.GetExecutionInfo",
 		trace.WithAttributes(attribute.Int64("height", int64(height))),
 	)
 	defer span.End()
 
-	info, err := eip.GetExecutionInfo(ctx, height)
+	info, err := t.inner.GetExecutionInfo(ctx, height)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
@@ -127,13 +123,8 @@ func (t *tracedExecutor) GetExecutionInfo(ctx context.Context, height uint64) (c
 	return info, err
 }
 
-// If the underlying executor implements DATransactionFilter, forward it while tracing.
+// FilterDATransactions forwards to the inner executor with tracing.
 func (t *tracedExecutor) FilterDATransactions(ctx context.Context, txs [][]byte, maxGas uint64) ([][]byte, [][]byte, error) {
-	filter, ok := t.inner.(coreexec.DATransactionFilter)
-	if !ok {
-		// If not implemented, return all transactions as valid (no filtering)
-		return txs, nil, nil
-	}
 	ctx, span := t.tracer.Start(ctx, "Executor.FilterDATransactions",
 		trace.WithAttributes(
 			attribute.Int("input_tx_count", len(txs)),
@@ -142,7 +133,7 @@ func (t *tracedExecutor) FilterDATransactions(ctx context.Context, txs [][]byte,
 	)
 	defer span.End()
 
-	validTxs, remainingTxs, err := filter.FilterDATransactions(ctx, txs, maxGas)
+	validTxs, remainingTxs, err := t.inner.FilterDATransactions(ctx, txs, maxGas)
 	if err != nil {
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())