feedback

Author: julienrbrt

block/internal/reaping/reaper.go

@@ -96,7 +96,9 @@ func (r *Reaper) reaperLoop() {
 				Int("consecutive_failures", consecutiveFailures).
 				Dur("backoff", backoff).
 				Msg("reaper error, backing off")
-			r.wait(backoff, cleanupTicker.C)
+			if r.wait(backoff, nil) {
+				return
+			}
 			continue
 		}
 
@@ -109,21 +111,28 @@ func (r *Reaper) reaperLoop() {
 			continue
 		}
 
-		r.wait(r.interval, cleanupTicker.C)
+		if r.wait(r.interval, cleanupTicker.C) {
+			return
+		}
 	}
 }
 
-func (r *Reaper) wait(d time.Duration, cleanupCh <-chan time.Time) {
+// 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 {
 	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
 	}
 }
 
@@ -156,7 +165,7 @@ func (r *Reaper) drainMempool() (bool, error) {
 
 		filtered := r.filterNewTxs(txs)
 		if len(filtered) == 0 {
-			continue
+			break
 		}
 
 		n, err := r.submitFiltered(filtered)

block/internal/reaping/reaper_test.go

@@ -105,7 +105,6 @@ func TestReaper_AllSeen_NoSubmit(t *testing.T) {
 	env.cache.SetTxSeen(hashTx(tx2))
 
 	env.execMock.EXPECT().GetTxs(mock.Anything).Return([][]byte{tx1, tx2}, nil).Once()
-	env.execMock.EXPECT().GetTxs(mock.Anything).Return(nil, nil).Once()
 
 	submitted, err := env.reaper.drainMempool()
 	assert.NoError(t, err)
@@ -232,3 +231,22 @@ func TestReaper_NilCallback_NoPanic(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, submitted)
 }
+
+func TestReaper_StopTerminates(t *testing.T) {
+	env := newTestEnv(t)
+	env.execMock.EXPECT().GetTxs(mock.Anything).Return(nil, nil).Maybe()
+
+	require.NoError(t, env.reaper.Start(context.Background()))
+
+	done := make(chan struct{})
+	go func() {
+		env.reaper.Stop()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Stop() did not return in time")
+	}
+}