refactor: reaper to drain mempool (#3236)

* refactor: reaper to drain mempool

* feedback

* fix partial drain

* cleanup old readme

* Prevent multiple Start() calls across components

* fix unwanted log

* lock

* updates

* remove redundant

* changelog

* feedback

* feedback

* feedback

* add bench

Author: julienrbrt

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changes
+
+- Improve reaper to sustain txs burst better [#3236](https://github.com/evstack/ev-node/pull/3236)
+
 ## v1.1.0
 
 No changes from v1.1.0-rc.2.

block/components.go

@@ -278,9 +278,9 @@ func newAggregatorComponents(
 		sequencer,
 		genesis,
 		logger,
-		executor,
 		cacheManager,
 		config.Node.ScrapeInterval.Duration,
+		executor.NotifyNewTransactions,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create reaper: %w", err)

block/internal/cache/manager.go

@@ -53,6 +53,7 @@ type CacheManager interface {
 	// Transaction operations
 	IsTxSeen(hash string) bool
 	SetTxSeen(hash string)
+	SetTxsSeen(hashes []string)
 	CleanupOldTxs(olderThan time.Duration) int
 
 	// Pending events syncing coordination
@@ -210,6 +211,14 @@ func (m *implementation) SetTxSeen(hash string) {
 	m.txTimestamps.Store(hash, time.Now())
 }
 
+func (m *implementation) SetTxsSeen(hashes []string) {
+	now := time.Now()
+	for _, hash := range hashes {
+		m.txCache.setSeen(hash, 0)
+		m.txTimestamps.Store(hash, now)
+	}
+}
+
 // CleanupOldTxs removes transaction hashes older than olderThan and returns
 // the count removed. Defaults to DefaultTxCacheRetention if olderThan <= 0.
 func (m *implementation) CleanupOldTxs(olderThan time.Duration) int {
@@ -237,13 +246,6 @@ func (m *implementation) CleanupOldTxs(olderThan time.Duration) int {
 		return true
 	})
 
-	if removed > 0 {
-		m.logger.Debug().
-			Int("removed", removed).
-			Dur("older_than", olderThan).
-			Msg("cleaned up old transaction hashes from cache")
-	}
-
 	return removed
 }
 

block/internal/executing/executor.go

@@ -162,11 +162,20 @@ func (e *Executor) SetBlockProducer(bp BlockProducer) {
 }
 
 // Start begins the execution component
-func (e *Executor) Start(ctx context.Context) error {
+func (e *Executor) Start(ctx context.Context) (err error) {
+	if e.cancel != nil {
+		return errors.New("executor already started")
+	}
 	e.ctx, e.cancel = context.WithCancel(ctx)
+	defer func() { // if error during init cancel context
+		if err != nil {
+			e.cancel()
+			e.ctx, e.cancel = nil, nil
+		}
+	}()
 
 	// Initialize state
-	if err := e.initializeState(); err != nil {
+	if err = e.initializeState(); err != nil {
 		return fmt.Errorf("failed to initialize state: %w", err)
 	}
 

block/internal/pruner/pruner.go

@@ -52,6 +52,9 @@ func New(
 
 // Start begins the pruning loop.
 func (p *Pruner) Start(ctx context.Context) error {
+	if p.cancel != nil {
+		return errors.New("pruner already started")
+	}
 	if !p.cfg.IsPruningEnabled() {
 		p.logger.Info().Msg("pruning is disabled, not starting pruner")
 		return nil