refactor: split reaper (#2660)

Author: julienrbrt

block/components.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/evstack/ev-node/block/internal/cache"
 	"github.com/evstack/ev-node/block/internal/executing"
+	"github.com/evstack/ev-node/block/internal/reaping"
 	"github.com/evstack/ev-node/block/internal/submitting"
 	"github.com/evstack/ev-node/block/internal/syncing"
 	coreda "github.com/evstack/ev-node/core/da"
@@ -25,6 +26,7 @@ import (
 // Components represents the block-related components
 type Components struct {
 	Executor  *executing.Executor
+	Reaper    *reaping.Reaper
 	Syncer    *syncing.Syncer
 	Submitter *submitting.Submitter
 	Cache     cache.Manager
@@ -65,6 +67,11 @@ func (bc *Components) Start(ctx context.Context) error {
 			return fmt.Errorf("failed to start executor: %w", err)
 		}
 	}
+	if bc.Reaper != nil {
+		if err := bc.Reaper.Start(ctxWithCancel); err != nil {
+			return fmt.Errorf("failed to start reaper: %w", err)
+		}
+	}
 	if bc.Syncer != nil {
 		if err := bc.Syncer.Start(ctxWithCancel); err != nil {
 			return fmt.Errorf("failed to start syncer: %w", err)
@@ -96,6 +103,11 @@ func (bc *Components) Stop() error {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop executor: %w", err))
 		}
 	}
+	if bc.Reaper != nil {
+		if err := bc.Reaper.Stop(); err != nil {
+			errs = errors.Join(errs, fmt.Errorf("failed to stop reaper: %w", err))
+		}
+	}
 	if bc.Syncer != nil {
 		if err := bc.Syncer.Stop(); err != nil {
 			errs = errors.Join(errs, fmt.Errorf("failed to stop syncer: %w", err))
@@ -220,6 +232,17 @@ func NewAggregatorComponents(
 		return nil, fmt.Errorf("failed to create executor: %w", err)
 	}
 
+	reaper, err := reaping.NewReaper(
+		exec,
+		sequencer,
+		genesis,
+		logger,
+		executor,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create reaper: %w", err)
+	}
+
 	// Create DA submitter for aggregator nodes (with signer for submission)
 	daSubmitter := submitting.NewDASubmitter(da, config, genesis, blockOpts, logger)
 	submitter := submitting.NewSubmitter(
@@ -237,6 +260,7 @@ func NewAggregatorComponents(
 
 	return &Components{
 		Executor:  executor,
+		Reaper:    reaper,
 		Submitter: submitter,
 		Cache:     cacheManager,
 		errorCh:   errorCh,

block/internal/executing/executor.go

@@ -22,11 +22,6 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
-const (
-	// DefaultInterval is the default reaper interval
-	DefaultInterval = 1 * time.Second
-)
-
 // broadcaster interface for P2P broadcasting
 type broadcaster[T any] interface {
 	WriteToStoreAndBroadcast(ctx context.Context, payload T) error
@@ -61,9 +56,6 @@ type Executor struct {
 	txNotifyCh chan struct{}
 	errorCh    chan<- error // Channel to report critical execution client failures
 
-	// Reaper for transaction processing
-	reaper *Reaper
-
 	// Logging
 	logger zerolog.Logger
 
@@ -126,27 +118,13 @@ func (e *Executor) Start(ctx context.Context) error {
 		return fmt.Errorf("failed to initialize state: %w", err)
 	}
 
-	// Initialize reaper
-	reaperStore, err := store.NewDefaultInMemoryKVStore()
-	if err != nil {
-		return fmt.Errorf("failed to create reaper store: %w", err)
-	}
-	e.reaper = NewReaper(e.ctx, e.exec, e.sequencer, e.genesis.ChainID, DefaultInterval, e.logger, reaperStore, e)
-
 	// Start execution loop
 	e.wg.Add(1)
 	go func() {
 		defer e.wg.Done()
 		e.executionLoop()
 	}()
 
-	// Start reaper
-	e.wg.Add(1)
-	go func() {
-		defer e.wg.Done()
-		e.reaper.Start(e.ctx)
-	}()
-
 	e.logger.Info().Msg("executor started")
 	return nil
 }
@@ -157,6 +135,7 @@ func (e *Executor) Stop() error {
 		e.cancel()
 	}
 	e.wg.Wait()
+
 	e.logger.Info().Msg("executor stopped")
 	return nil
 }
@@ -657,11 +636,6 @@ func (e *Executor) recordBlockMetrics(data *types.Data) {
 	e.metrics.CommittedHeight.Set(float64(data.Metadata.Height))
 }
 
-// GetCoreExecutor returns the underlying core executor for testing purposes
-func (e *Executor) GetCoreExecutor() coreexecutor.Executor {
-	return e.exec
-}
-
 // BatchData represents batch data from sequencer
 type BatchData struct {
 	*coresequencer.Batch

block/internal/executing/utils.go

@@ -0,0 +1,19 @@
+package executing
+
+import coreexecutor "github.com/evstack/ev-node/core/execution"
+
+// GetCoreExecutor returns the underlying core executor for testing purposes
+func (e *Executor) GetCoreExecutor() coreexecutor.Executor {
+	return e.exec
+}
+
+// HasPendingTxNotification checks if there's a pending transaction notification
+// It is used for testing purposes
+func (e *Executor) HasPendingTxNotification() bool {
+	select {
+	case <-e.txNotifyCh:
+		return true
+	default:
+		return false
+	}
+}

block/internal/executing/reaper.go block/internal/reaping/reaper.goblock/internal/executing/reaper.go renamed to block/internal/reaping/reaper.go

@@ -1,67 +1,108 @@
-package executing
+package reaping
 
 import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
+	"fmt"
+	"sync"
 	"time"
 
 	ds "github.com/ipfs/go-datastore"
 	"github.com/rs/zerolog"
 
+	"github.com/evstack/ev-node/block/internal/executing"
 	coreexecutor "github.com/evstack/ev-node/core/execution"
 	coresequencer "github.com/evstack/ev-node/core/sequencer"
+	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/pkg/store"
 )
 
+// DefaultInterval is the default reaper interval
+const DefaultInterval = 1 * time.Second
+
 // Reaper is responsible for periodically retrieving transactions from the executor,
 // filtering out already seen transactions, and submitting new transactions to the sequencer.
 type Reaper struct {
 	exec      coreexecutor.Executor
 	sequencer coresequencer.Sequencer
 	chainID   string
 	interval  time.Duration
-	logger    zerolog.Logger
-	ctx       context.Context
 	seenStore ds.Batching
-	executor  *Executor
+	executor  *executing.Executor
+
+	// shared components
+	logger zerolog.Logger
+
+	// Lifecycle
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
 }
 
 // NewReaper creates a new Reaper instance with persistent seenTx storage.
-func NewReaper(ctx context.Context, exec coreexecutor.Executor, sequencer coresequencer.Sequencer, chainID string, interval time.Duration, logger zerolog.Logger, store ds.Batching, executor *Executor) *Reaper {
-	if interval <= 0 {
-		interval = DefaultInterval
+func NewReaper(exec coreexecutor.Executor, sequencer coresequencer.Sequencer, genesis genesis.Genesis, logger zerolog.Logger, executor *executing.Executor) (*Reaper, error) {
+	if executor == nil {
+		return nil, errors.New("executor cannot be nil")
+	}
+
+	store, err := store.NewDefaultInMemoryKVStore()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create reaper store: %w", err)
 	}
+
 	return &Reaper{
 		exec:      exec,
 		sequencer: sequencer,
-		chainID:   chainID,
-		interval:  interval,
+		chainID:   genesis.ChainID,
+		interval:  DefaultInterval, // eventually this can be made configurable via config.Node
 		logger:    logger.With().Str("component", "reaper").Logger(),
-		ctx:       ctx,
 		seenStore: store,
 		executor:  executor,
-	}
+	}, nil
 }
 
-// Start begins the reaping process at the specified interval.
-func (r *Reaper) Start(ctx context.Context) {
-	r.ctx = ctx
-	ticker := time.NewTicker(r.interval)
-	defer ticker.Stop()
+// Start begins the execution component
+func (r *Reaper) Start(ctx context.Context) error {
+	r.ctx, r.cancel = context.WithCancel(ctx)
+
+	// Start repear loop
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		r.reaperLoop()
+	}()
 
 	r.logger.Info().Dur("interval", r.interval).Msg("reaper started")
+	return nil
+}
+
+func (r *Reaper) reaperLoop() {
+	ticker := time.NewTicker(r.interval)
+	defer ticker.Stop()
 
 	for {
 		select {
-		case <-ctx.Done():
-			r.logger.Info().Msg("reaper stopped")
+		case <-r.ctx.Done():
 			return
 		case <-ticker.C:
 			r.SubmitTxs()
 		}
 	}
 }
 
+// Stop shuts down the reaper component
+func (r *Reaper) Stop() error {
+	if r.cancel != nil {
+		r.cancel()
+	}
+	r.wg.Wait()
+
+	r.logger.Info().Msg("reaper stopped")
+	return nil
+}
+
 // SubmitTxs retrieves transactions from the executor and submits them to the sequencer.
 func (r *Reaper) SubmitTxs() {
 	txs, err := r.exec.GetTxs(r.ctx)
@@ -109,14 +150,19 @@ func (r *Reaper) SubmitTxs() {
 	}
 
 	// Notify the executor that new transactions are available
-	if r.executor != nil && len(newTxs) > 0 {
+	if len(newTxs) > 0 {
 		r.logger.Debug().Msg("notifying executor of new transactions")
 		r.executor.NotifyNewTransactions()
 	}
 
 	r.logger.Debug().Msg("successfully submitted txs")
 }
 
+// SeenStore returns the datastore used to track seen transactions.
+func (r *Reaper) SeenStore() ds.Datastore {
+	return r.seenStore
+}
+
 func hashTx(tx []byte) string {
 	hash := sha256.Sum256(tx)
 	return hex.EncodeToString(hash[:])