chore: pr feedback

Author: chatton

block/components.go

@@ -238,9 +238,6 @@ func newAggregatorComponents(
 	// error channel for critical failures
 	errorCh := make(chan error, 1)
 
-	// capture raw sequencer before tracing wrap for batch ack interface
-	rawSequencer := sequencer
-
 	// wrap sequencer with tracing if enabled
 	if config.Instrumentation.IsTracingEnabled() {
 		sequencer = telemetry.WithTracingSequencer(sequencer)
@@ -288,18 +285,6 @@ func newAggregatorComponents(
 		return nil, fmt.Errorf("failed to create reaper: %w", err)
 	}
 
-	// wire batch ack callback so drained queue entries are committed after block commit
-	type batchAcknowledger interface {
-		AckBatch(ctx context.Context) error
-	}
-	if acker, ok := rawSequencer.(batchAcknowledger); ok {
-		executor.SetOnBatchCommitted(acker.AckBatch)
-	} else if !config.Node.BasedSequencer {
-		// without an ack, drained queue entries are rolled back on every
-		// retrieval and the same transactions would be re-included each block
-		logger.Warn().Msg("sequencer does not implement AckBatch; drained batch entries will not be acknowledged after block commit")
-	}
-
 	if config.Node.BasedSequencer { // no submissions needed for based sequencer
 		return &Components{
 			Executor: executor,

block/internal/executing/executor.go

@@ -100,6 +100,12 @@ type Executor struct {
 	blockProducer BlockProducer
 }
 
+// batchAcknowledger is implemented by sequencers whose drained queue
+// entries must be acknowledged after the block is durably committed.
+type batchAcknowledger interface {
+	AckBatch(ctx context.Context) error
+}
+
 // NewExecutor creates a new block executor.
 // The executor is responsible for:
 // - Block production from sequencer batches
@@ -162,6 +168,17 @@ func NewExecutor(
 		logger:            logger.With().Str("component", "executor").Logger(),
 	}
 	e.blockProducer = e
+
+	// wire the batch ack so drained queue entries are committed after block
+	// commit. tracing wrappers forward AckBatch to the underlying sequencer.
+	if acker, ok := sequencer.(batchAcknowledger); ok {
+		e.onBatchCommitted = acker.AckBatch
+	} else if !config.Node.BasedSequencer {
+		// without an ack, drained queue entries are rolled back on every
+		// retrieval and the same transactions would be re-included each block
+		e.logger.Warn().Msg("sequencer does not implement AckBatch; drained batch entries will not be acknowledged after block commit")
+	}
+
 	return e, nil
 }
 
@@ -171,12 +188,6 @@ func (e *Executor) SetBlockProducer(bp BlockProducer) {
 	e.blockProducer = bp
 }
 
-// SetOnBatchCommitted registers a callback fired after each block commit.
-// If the callback fails, it is retried before the next block is produced.
-func (e *Executor) SetOnBatchCommitted(fn func(ctx context.Context) error) {
-	e.onBatchCommitted = fn
-}
-
 // ackCommittedBatch invokes the batch ack callback and tracks failures so
 // they can be retried before the next block is produced.
 func (e *Executor) ackCommittedBatch(ctx context.Context) error {

block/internal/executing/executor_logic_test.go

@@ -127,13 +127,13 @@ func TestProduceBlock_RetriesFailedAckBeforeNextBlock(t *testing.T) {
 	defer fx.Cancel()
 
 	var ackCalls int
-	fx.Exec.SetOnBatchCommitted(func(ctx context.Context) error {
+	fx.Exec.onBatchCommitted = func(ctx context.Context) error {
 		ackCalls++
 		if ackCalls <= 2 {
 			return assert.AnError
 		}
 		return nil
-	})
+	}
 
 	fx.MockSeq.EXPECT().GetNextBatch(mock.Anything, mock.AnythingOfType("sequencer.GetNextBatchRequest")).
 		RunAndReturn(func(ctx context.Context, req coreseq.GetNextBatchRequest) (*coreseq.GetNextBatchResponse, error) {

pkg/telemetry/sequencer_tracing.go

@@ -20,12 +20,35 @@ type tracedSequencer struct {
 	tracer trace.Tracer
 }
 
+// batchAcknowledger is implemented by sequencers that require an ack
+// after drained queue entries are durably committed in a block.
+type batchAcknowledger interface {
+	AckBatch(ctx context.Context) error
+}
+
 // WithTracingSequencer decorates the provided Sequencer with tracing spans.
+// If the inner sequencer implements AckBatch, the returned sequencer
+// forwards it so consumers can still detect and use the ack capability.
 func WithTracingSequencer(inner coresequencer.Sequencer) coresequencer.Sequencer {
-	return &tracedSequencer{
+	ts := &tracedSequencer{
 		inner:  inner,
 		tracer: otel.Tracer("ev-node/sequencer"),
 	}
+	if acker, ok := inner.(batchAcknowledger); ok {
+		return &tracedAckSequencer{tracedSequencer: ts, acker: acker}
+	}
+	return ts
+}
+
+// tracedAckSequencer is a tracedSequencer whose inner sequencer also
+// implements AckBatch.
+type tracedAckSequencer struct {
+	*tracedSequencer
+	acker batchAcknowledger
+}
+
+func (t *tracedAckSequencer) AckBatch(ctx context.Context) error {
+	return t.acker.AckBatch(ctx)
 }
 
 func (t *tracedSequencer) SubmitBatchTxs(ctx context.Context, req coresequencer.SubmitBatchTxsRequest) (*coresequencer.SubmitBatchTxsResponse, error) {