feat(tracing): adding block production tracing (#2980)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

Follows the same pattern as before and extracts a `BlockProducer`
interface and introduces a tracing implementation which wraps the real
one and is wrapped when telemetry is enabled.

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->

Author: chatton

block/components.go

@@ -220,6 +220,10 @@ func NewAggregatorComponents(
 		return nil, fmt.Errorf("failed to create executor: %w", err)
 	}
 
+	if config.Instrumentation.IsTracingEnabled() {
+		executor.SetBlockProducer(executing.WithTracingBlockProducer(executor))
+	}
+
 	reaper, err := reaping.NewReaper(
 		exec,
 		sequencer,

block/internal/executing/block_producer.go

@@ -0,0 +1,27 @@
+package executing
+
+import (
+	"context"
+
+	"github.com/evstack/ev-node/types"
+)
+
+// BlockProducer defines operations for block production that can be traced.
+// The Executor implements this interface, and a tracing decorator can wrap it
+// to add OpenTelemetry spans to each operation.
+type BlockProducer interface {
+	// ProduceBlock creates, validates, and broadcasts a new block.
+	ProduceBlock(ctx context.Context) error
+
+	// RetrieveBatch gets the next batch of transactions from the sequencer.
+	RetrieveBatch(ctx context.Context) (*BatchData, error)
+
+	// CreateBlock constructs a new block from the given batch data.
+	CreateBlock(ctx context.Context, height uint64, batchData *BatchData) (*types.SignedHeader, *types.Data, error)
+
+	// ApplyBlock executes the block transactions and returns the new state.
+	ApplyBlock(ctx context.Context, header types.Header, data *types.Data) (types.State, error)
+
+	// ValidateBlock validates block structure and state transitions.
+	ValidateBlock(ctx context.Context, lastState types.State, header *types.SignedHeader, data *types.Data) error
+}

block/internal/executing/executor.go

@@ -25,6 +25,8 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
+var _ BlockProducer = (*Executor)(nil)
+
 // Executor handles block production, transaction processing, and state management
 type Executor struct {
 	// Core components
@@ -60,6 +62,10 @@ type Executor struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
+
+	// blockProducer is the interface used for block production operations.
+	// defaults to self, but can be wrapped with tracing.
+	blockProducer BlockProducer
 }
 
 // NewExecutor creates a new block executor.
@@ -101,7 +107,7 @@ func NewExecutor(
 		}
 	}
 
-	return &Executor{
+	e := &Executor{
 		store:             store,
 		exec:              exec,
 		sequencer:         sequencer,
@@ -117,7 +123,15 @@ func NewExecutor(
 		txNotifyCh:        make(chan struct{}, 1),
 		errorCh:           errorCh,
 		logger:            logger.With().Str("component", "executor").Logger(),
-	}, nil
+	}
+	e.blockProducer = e
+	return e, nil
+}
+
+// SetBlockProducer sets the block producer interface, allowing injection of
+// a tracing wrapper or other decorator.
+func (e *Executor) SetBlockProducer(bp BlockProducer) {
+	e.blockProducer = bp
 }
 
 // Start begins the execution component
@@ -279,7 +293,7 @@ func (e *Executor) executionLoop() {
 				continue
 			}
 
-			if err := e.produceBlock(); err != nil {
+			if err := e.blockProducer.ProduceBlock(e.ctx); err != nil {
 				e.logger.Error().Err(err).Msg("failed to produce block")
 			}
 			txsAvailable = false
@@ -288,7 +302,7 @@ func (e *Executor) executionLoop() {
 
 		case <-lazyTimerCh:
 			e.logger.Debug().Msg("Lazy timer triggered block production")
-			if err := e.produceBlock(); err != nil {
+			if err := e.blockProducer.ProduceBlock(e.ctx); err != nil {
 				e.logger.Error().Err(err).Msg("failed to produce block from lazy timer")
 			}
 			// Reset lazy timer
@@ -300,8 +314,8 @@ func (e *Executor) executionLoop() {
 	}
 }
 
-// produceBlock creates, validates, and stores a new block
-func (e *Executor) produceBlock() error {
+// ProduceBlock creates, validates, and stores a new block.
+func (e *Executor) ProduceBlock(ctx context.Context) error {
 	start := time.Now()
 	defer func() {
 		if e.metrics.OperationDuration["block_production"] != nil {
@@ -338,7 +352,7 @@ func (e *Executor) produceBlock() error {
 
 	// Check if there's an already stored block at the newHeight
 	// If there is use that instead of creating a new block
-	pendingHeader, pendingData, err := e.store.GetBlockData(e.ctx, newHeight)
+	pendingHeader, pendingData, err := e.store.GetBlockData(ctx, newHeight)
 	if err == nil {
 		e.logger.Info().Uint64("height", newHeight).Msg("using pending block")
 		header = pendingHeader
@@ -347,7 +361,7 @@ func (e *Executor) produceBlock() error {
 		return fmt.Errorf("failed to get block data: %w", err)
 	} else {
 		// get batch from sequencer
-		batchData, err = e.retrieveBatch(e.ctx)
+		batchData, err = e.blockProducer.RetrieveBatch(ctx)
 		if errors.Is(err, common.ErrNoBatch) {
 			e.logger.Debug().Msg("no batch available")
 			return nil
@@ -357,13 +371,13 @@ func (e *Executor) produceBlock() error {
 			return fmt.Errorf("failed to retrieve batch: %w", err)
 		}
 
-		header, data, err = e.createBlock(e.ctx, newHeight, batchData)
+		header, data, err = e.blockProducer.CreateBlock(ctx, newHeight, batchData)
 		if err != nil {
 			return fmt.Errorf("failed to create block: %w", err)
 		}
 
 		// saved early for crash recovery, will be overwritten later with the final signature
-		batch, err := e.store.NewBatch(e.ctx)
+		batch, err := e.store.NewBatch(ctx)
 		if err != nil {
 			return fmt.Errorf("failed to create batch for early save: %w", err)
 		}
@@ -378,12 +392,12 @@ func (e *Executor) produceBlock() error {
 	// Pass force-included mask through context for execution optimization
 	// Force-included txs (from DA) MUST be validated as they're from untrusted sources
 	// Mempool txs can skip validation as they were validated when added to mempool
-	ctx := e.ctx
+	applyCtx := ctx
 	if batchData != nil && batchData.Batch != nil && batchData.ForceIncludedMask != nil {
-		ctx = coreexecutor.WithForceIncludedMask(ctx, batchData.ForceIncludedMask)
+		applyCtx = coreexecutor.WithForceIncludedMask(applyCtx, batchData.ForceIncludedMask)
 	}
 
-	newState, err := e.applyBlock(ctx, header.Header, data)
+	newState, err := e.blockProducer.ApplyBlock(applyCtx, header.Header, data)
 	if err != nil {
 		return fmt.Errorf("failed to apply block: %w", err)
 	}
@@ -400,13 +414,13 @@ func (e *Executor) produceBlock() error {
 	}
 	header.Signature = signature
 
-	if err := e.validateBlock(currentState, header, data); err != nil {
+	if err := e.blockProducer.ValidateBlock(ctx, currentState, header, data); err != nil {
 		e.sendCriticalError(fmt.Errorf("failed to validate block: %w", err))
 		e.logger.Error().Err(err).Msg("CRITICAL: Permanent block validation error - halting block production")
 		return fmt.Errorf("failed to validate block: %w", err)
 	}
 
-	batch, err := e.store.NewBatch(e.ctx)
+	batch, err := e.store.NewBatch(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to create batch: %w", err)
 	}
@@ -431,9 +445,9 @@ func (e *Executor) produceBlock() error {
 	e.setLastState(newState)
 
 	// broadcast header and data to P2P network
-	g, ctx := errgroup.WithContext(e.ctx)
-	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(ctx, header) })
-	g.Go(func() error { return e.dataBroadcaster.WriteToStoreAndBroadcast(ctx, data) })
+	g, broadcastCtx := errgroup.WithContext(ctx)
+	g.Go(func() error { return e.headerBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, header) })
+	g.Go(func() error { return e.dataBroadcaster.WriteToStoreAndBroadcast(broadcastCtx, data) })
 	if err := g.Wait(); err != nil {
 		e.logger.Error().Err(err).Msg("failed to broadcast header and/data")
 		// don't fail block production on broadcast error
@@ -449,8 +463,8 @@ func (e *Executor) produceBlock() error {
 	return nil
 }
 
-// retrieveBatch gets the next batch of transactions from the sequencer
-func (e *Executor) retrieveBatch(ctx context.Context) (*BatchData, error) {
+// RetrieveBatch gets the next batch of transactions from the sequencer.
+func (e *Executor) RetrieveBatch(ctx context.Context) (*BatchData, error) {
 	req := coresequencer.GetNextBatchRequest{
 		Id:            []byte(e.genesis.ChainID),
 		MaxBytes:      common.DefaultMaxBlobSize,
@@ -481,8 +495,8 @@ func (e *Executor) retrieveBatch(ctx context.Context) (*BatchData, error) {
 	}, nil
 }
 
-// createBlock creates a new block from the given batch
-func (e *Executor) createBlock(ctx context.Context, height uint64, batchData *BatchData) (*types.SignedHeader, *types.Data, error) {
+// CreateBlock creates a new block from the given batch.
+func (e *Executor) CreateBlock(ctx context.Context, height uint64, batchData *BatchData) (*types.SignedHeader, *types.Data, error) {
 	currentState := e.getLastState()
 	headerTime := uint64(e.genesis.StartTime.UnixNano())
 
@@ -581,8 +595,8 @@ func (e *Executor) createBlock(ctx context.Context, height uint64, batchData *Ba
 	return header, data, nil
 }
 
-// applyBlock applies the block to get the new state
-func (e *Executor) applyBlock(ctx context.Context, header types.Header, data *types.Data) (types.State, error) {
+// ApplyBlock applies the block to get the new state.
+func (e *Executor) ApplyBlock(ctx context.Context, header types.Header, data *types.Data) (types.State, error) {
 	currentState := e.getLastState()
 
 	// Prepare transactions
@@ -654,8 +668,8 @@ func (e *Executor) executeTxsWithRetry(ctx context.Context, rawTxs [][]byte, hea
 	return nil, nil
 }
 
-// validateBlock validates the created block
-func (e *Executor) validateBlock(lastState types.State, header *types.SignedHeader, data *types.Data) error {
+// ValidateBlock validates the created block.
+func (e *Executor) ValidateBlock(_ context.Context, lastState types.State, header *types.SignedHeader, data *types.Data) error {
 	// Set custom verifier for aggregator node signature
 	header.SetCustomVerifierForAggregator(e.options.AggregatorNodeSignatureBytesProvider)
 

block/internal/executing/executor_lazy_test.go

@@ -94,8 +94,8 @@ func TestLazyMode_ProduceBlockLogic(t *testing.T) {
 
 	mockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	// Direct call to produceBlock should work (this is what lazy timer does)
-	err = exec.produceBlock()
+	// Direct call to ProduceBlock should work (this is what lazy timer does)
+	err = exec.ProduceBlock(exec.ctx)
 	require.NoError(t, err)
 
 	h1, err := memStore.Height(context.Background())
@@ -118,7 +118,7 @@ func TestLazyMode_ProduceBlockLogic(t *testing.T) {
 
 	mockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	err = exec.produceBlock()
+	err = exec.ProduceBlock(exec.ctx)
 	require.NoError(t, err)
 
 	h2, err := memStore.Height(context.Background())
@@ -209,7 +209,7 @@ func TestRegularMode_ProduceBlockLogic(t *testing.T) {
 
 	mockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	err = exec.produceBlock()
+	err = exec.ProduceBlock(exec.ctx)
 	require.NoError(t, err)
 
 	h1, err := memStore.Height(context.Background())

block/internal/executing/executor_logic_test.go

@@ -117,7 +117,7 @@ func TestProduceBlock_EmptyBatch_SetsEmptyDataHash(t *testing.T) {
 	mockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
 	// produce one block
-	err = exec.produceBlock()
+	err = exec.ProduceBlock(exec.ctx)
 	require.NoError(t, err)
 
 	// Verify height and stored block
@@ -202,14 +202,14 @@ func TestPendingLimit_SkipsProduction(t *testing.T) {
 
 	mockSeq.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	require.NoError(t, exec.produceBlock())
+	require.NoError(t, exec.ProduceBlock(exec.ctx))
 	h1, err := memStore.Height(context.Background())
 	require.NoError(t, err)
 	assert.Equal(t, uint64(1), h1)
 
 	// With limit=1 and lastSubmitted default 0, pending >= 1 so next production should be skipped
-	// No new expectations; produceBlock should return early before hitting sequencer
-	require.NoError(t, exec.produceBlock())
+	// No new expectations; ProduceBlock should return early before hitting sequencer
+	require.NoError(t, exec.ProduceBlock(exec.ctx))
 	h2, err := memStore.Height(context.Background())
 	require.NoError(t, err)
 	assert.Equal(t, h1, h2, "height should not change when production is skipped")

block/internal/executing/executor_restart_test.go

@@ -95,7 +95,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 
 	mockSeq1.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	err = exec1.produceBlock()
+	err = exec1.ProduceBlock(exec1.ctx)
 	require.NoError(t, err)
 
 	// Verify first block was produced
@@ -214,7 +214,7 @@ func TestExecutor_RestartUsesPendingHeader(t *testing.T) {
 
 	// Note: mockSeq2 should NOT receive GetNextBatch calls because pending block should be used
 
-	err = exec2.produceBlock()
+	err = exec2.ProduceBlock(exec2.ctx)
 	require.NoError(t, err)
 
 	// Verify height advanced to 2
@@ -316,7 +316,7 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 
 	mockSeq1.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	err = exec1.produceBlock()
+	err = exec1.ProduceBlock(exec1.ctx)
 	require.NoError(t, err)
 
 	// Stop first executor
@@ -372,7 +372,7 @@ func TestExecutor_RestartNoPendingHeader(t *testing.T) {
 
 	mockSeq2.EXPECT().GetDAHeight().Return(uint64(0)).Once()
 
-	err = exec2.produceBlock()
+	err = exec2.ProduceBlock(exec2.ctx)
 	require.NoError(t, err)
 
 	// Verify normal operation