Merge branch 'julien/reaper' into julien/testapp

Author: julienrbrt

block/internal/executing/executor.go

@@ -162,14 +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/reaping/reaper.go

@@ -88,9 +88,9 @@ func (r *Reaper) reaperLoop() {
 	consecutiveFailures := 0
 
 	for {
-		submitted, err := r.drainMempool()
+		submitted, err := r.drainMempool(cleanupTicker.C)
 
-		if err != nil && !errors.Is(err, context.Canceled) {
+		if err != nil && r.ctx.Err() == nil {
 			consecutiveFailures++
 			backoff := r.interval * time.Duration(1<<min(consecutiveFailures, 5))
 			backoff = min(backoff, MaxBackoffInterval)
@@ -99,7 +99,7 @@ func (r *Reaper) reaperLoop() {
 				Int("consecutive_failures", consecutiveFailures).
 				Dur("next_retry_in", backoff).
 				Msg("reaper error, backing off")
-			if r.wait(backoff, nil) {
+			if r.wait(backoff) {
 				return
 			}
 			continue
@@ -115,29 +115,19 @@ func (r *Reaper) reaperLoop() {
 			continue
 		}
 
-		// Note: if the cleanup ticker fires before the idle interval elapses,
-		// the remaining idle duration is discarded. drainMempool() is called
-		// immediately and a fresh idle wait starts from scratch.
-		if r.wait(r.interval, cleanupTicker.C) {
+		if r.wait(r.interval) {
 			return
 		}
 	}
 }
 
 // wait blocks for the given duration. Returns true if the context was cancelled.
-// When cleanupCh is non-nil, processes cache cleanup if that channel fires first.
-func (r *Reaper) wait(d time.Duration, cleanupCh <-chan time.Time) bool {
+func (r *Reaper) wait(d time.Duration) bool {
 	timer := time.NewTimer(d)
 	defer timer.Stop()
 	select {
 	case <-r.ctx.Done():
 		return true
-	case <-cleanupCh:
-		removed := r.cache.CleanupOldTxs(cache.DefaultTxCacheRetention)
-		if removed > 0 {
-			r.logger.Info().Int("removed", removed).Msg("cleaned up old transaction hashes")
-		}
-		return false
 	case <-timer.C:
 		return false
 	}
@@ -158,7 +148,7 @@ type pendingTx struct {
 	hash string
 }
 
-func (r *Reaper) drainMempool() (bool, error) {
+func (r *Reaper) drainMempool(cleanupCh <-chan time.Time) (bool, error) {
 	var totalSubmitted int
 
 	defer func() {
@@ -168,6 +158,15 @@ func (r *Reaper) drainMempool() (bool, error) {
 	}()
 
 	for {
+		select {
+		case <-cleanupCh:
+			removed := r.cache.CleanupOldTxs(cache.DefaultTxCacheRetention)
+			if removed > 0 {
+				r.logger.Info().Int("removed", removed).Msg("cleaned up old transaction hashes")
+			}
+		default:
+		}
+
 		txs, err := r.exec.GetTxs(r.ctx)
 		if err != nil {
 			return totalSubmitted > 0, fmt.Errorf("failed to get txs from executor: %w", err)

block/internal/reaping/reaper_test.go

@@ -87,7 +87,7 @@ func TestReaper_NewTxs_SubmitsAndPersistsAndNotifies(t *testing.T) {
 			return &coresequencer.SubmitBatchTxsResponse{}, nil
 		}).Once()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 	assert.True(t, env.cache.IsTxSeen(hashTx(tx1)))
@@ -106,7 +106,7 @@ func TestReaper_AllSeen_NoSubmit(t *testing.T) {
 
 	env.execMock.EXPECT().GetTxs(mock.Anything).Return([][]byte{tx1, tx2}, nil).Once()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.False(t, submitted)
 	assert.False(t, env.wasNotified())
@@ -129,7 +129,7 @@ func TestReaper_PartialSeen_FiltersAndPersists(t *testing.T) {
 			return &coresequencer.SubmitBatchTxsResponse{}, nil
 		}).Once()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 	assert.True(t, env.cache.IsTxSeen(hashTx(txOld)))
@@ -147,7 +147,7 @@ func TestReaper_SequencerError_NoPersistence_NoNotify(t *testing.T) {
 	env.seqMock.EXPECT().SubmitBatchTxs(mock.Anything, mock.AnythingOfType("sequencer.SubmitBatchTxsRequest")).
 		Return((*coresequencer.SubmitBatchTxsResponse)(nil), assert.AnError).Once()
 
-	_, err := env.reaper.drainMempool()
+	_, err := env.reaper.drainMempool(nil)
 	assert.Error(t, err)
 	assert.False(t, env.cache.IsTxSeen(hashTx(tx)))
 	assert.False(t, env.wasNotified())
@@ -169,7 +169,7 @@ func TestReaper_DrainsMempoolInMultipleRounds(t *testing.T) {
 			return &coresequencer.SubmitBatchTxsResponse{}, nil
 		}).Twice()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 	assert.True(t, env.cache.IsTxSeen(hashTx(tx1)))
@@ -183,7 +183,7 @@ func TestReaper_EmptyMempool_NoAction(t *testing.T) {
 
 	env.execMock.EXPECT().GetTxs(mock.Anything).Return(nil, nil).Once()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.False(t, submitted)
 	assert.False(t, env.wasNotified())
@@ -201,7 +201,7 @@ func TestReaper_HashComputedOnce(t *testing.T) {
 	env.seqMock.EXPECT().SubmitBatchTxs(mock.Anything, mock.AnythingOfType("sequencer.SubmitBatchTxsRequest")).
 		Return(&coresequencer.SubmitBatchTxsResponse{}, nil).Once()
 
-	submitted, err := env.reaper.drainMempool()
+	submitted, err := env.reaper.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 	assert.True(t, env.cache.IsTxSeen(expectedHash))
@@ -227,7 +227,7 @@ func TestReaper_NilCallback_NoPanic(t *testing.T) {
 	mockSeq.EXPECT().SubmitBatchTxs(mock.Anything, mock.AnythingOfType("sequencer.SubmitBatchTxsRequest")).
 		Return(&coresequencer.SubmitBatchTxsResponse{}, nil).Once()
 
-	submitted, err := r.drainMempool()
+	submitted, err := r.drainMempool(nil)
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 }

block/internal/submitting/submitter.go

@@ -118,14 +118,21 @@ func NewSubmitter(
 }
 
 // Start begins the submitting component
-func (s *Submitter) Start(ctx context.Context) error {
+func (s *Submitter) Start(ctx context.Context) (err error) {
 	if s.cancel != nil {
 		return errors.New("submitter already started")
 	}
+
 	s.ctx, s.cancel = context.WithCancel(ctx)
+	defer func() { // if error during init cancel context
+		if err != nil {
+			s.cancel()
+			s.ctx, s.cancel = nil, nil
+		}
+	}()
 
 	// Initialize DA included height
-	if err := s.initializeDAIncludedHeight(ctx); err != nil {
+	if err = s.initializeDAIncludedHeight(ctx); err != nil {
 		return err
 	}
 

block/internal/syncing/syncer.go

@@ -162,6 +162,7 @@ func (s *Syncer) SetBlockSyncer(bs BlockSyncer) {
 }
 
 // Start begins the syncing component
+// The component should not be started after being stopped.
 func (s *Syncer) Start(ctx context.Context) (err error) {
 	if s.cancel != nil {
 		return errors.New("syncer already started")