test: Fix TestBatchQueueThrottlingWithDAFailure (#2663)

## Overview
Fixes TestBatchQueueThrottlingWithDAFailure test

Minor updates

Author: alpe

block/components.go

@@ -238,6 +238,7 @@ func NewAggregatorComponents(
 		genesis,
 		logger,
 		executor,
+		reaping.DefaultInterval,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create reaper: %w", err)

block/internal/reaping/reaper.go

@@ -42,11 +42,20 @@ type Reaper struct {
 }
 
 // NewReaper creates a new Reaper instance with persistent seenTx storage.
-func NewReaper(exec coreexecutor.Executor, sequencer coresequencer.Sequencer, genesis genesis.Genesis, logger zerolog.Logger, executor *executing.Executor) (*Reaper, error) {
+func NewReaper(
+	exec coreexecutor.Executor,
+	sequencer coresequencer.Sequencer,
+	genesis genesis.Genesis,
+	logger zerolog.Logger,
+	executor *executing.Executor,
+	scrapeInterval time.Duration,
+) (*Reaper, error) {
 	if executor == nil {
 		return nil, errors.New("executor cannot be nil")
 	}
-
+	if scrapeInterval == 0 {
+		return nil, errors.New("scrape interval cannot be empty")
+	}
 	store, err := store.NewDefaultInMemoryKVStore()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create reaper store: %w", err)
@@ -56,7 +65,7 @@ func NewReaper(exec coreexecutor.Executor, sequencer coresequencer.Sequencer, ge
 		exec:      exec,
 		sequencer: sequencer,
 		chainID:   genesis.ChainID,
-		interval:  DefaultInterval, // eventually this can be made configurable via config.Node
+		interval:  scrapeInterval,
 		logger:    logger.With().Str("component", "reaper").Logger(),
 		seenStore: store,
 		executor:  executor,
@@ -110,6 +119,10 @@ func (r *Reaper) SubmitTxs() {
 		r.logger.Error().Err(err).Msg("failed to get txs from executor")
 		return
 	}
+	if len(txs) == 0 {
+		r.logger.Debug().Msg("no new txs")
+		return
+	}
 
 	var newTxs [][]byte
 	for _, tx := range txs {

block/internal/reaping/reaper_test.go

@@ -61,7 +61,7 @@ func newTestExecutor(t *testing.T) *executing.Executor {
 func newTestReaper(t *testing.T, chainID string, execMock *testmocks.MockExecutor, seqMock *testmocks.MockSequencer, e *executing.Executor) *Reaper {
 	t.Helper()
 
-	r, err := NewReaper(execMock, seqMock, genesis.Genesis{ChainID: chainID}, zerolog.Nop(), e)
+	r, err := NewReaper(execMock, seqMock, genesis.Genesis{ChainID: chainID}, zerolog.Nop(), e, 100*time.Millisecond)
 	require.NoError(t, err)
 
 	return r

node/single_sequencer_integration_test.go

@@ -308,8 +308,8 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 	// Set up configuration with low limits to trigger throttling quickly
 	config := getTestConfig(t, 1)
 	config.Node.MaxPendingHeadersAndData = 3 // Low limit to quickly reach pending limit after DA failure
-	config.Node.BlockTime = evconfig.DurationWrapper{Duration: 100 * time.Millisecond}
-	config.DA.BlockTime = evconfig.DurationWrapper{Duration: 1 * time.Second} // Longer DA time to ensure blocks are produced first
+	config.Node.BlockTime = evconfig.DurationWrapper{Duration: 25 * time.Millisecond}
+	config.DA.BlockTime = evconfig.DurationWrapper{Duration: 100 * time.Millisecond} // Longer DA time to ensure blocks are produced first
 
 	// Create test components
 	executor, sequencer, dummyDA, p2pClient, ds, _, stopDAHeightTicker := createTestComponents(t, config)
@@ -327,23 +327,22 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 	node, cleanup := createNodeWithCustomComponents(t, config, executor, sequencer, dummyDAImpl, p2pClient, ds, func() {})
 	defer cleanup()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(t.Context())
 	defer cancel()
 
 	var runningWg sync.WaitGroup
 	startNodeInBackground(t, []*FullNode{node}, []context.Context{ctx}, &runningWg, 0)
 
 	// Wait for the node to start producing blocks
-	require.NoError(waitForFirstBlock(node, Store))
+	waitForBlockN(t, 1, node, config.Node.BlockTime.Duration)
 
 	// Inject some initial transactions to get the system working
 	for i := 0; i < 5; i++ {
 		dummyExecutor.InjectTx([]byte(fmt.Sprintf("initial-tx-%d", i)))
 	}
 
-	// Wait for at least 5 blocks to be produced before simulating DA failure
-	require.NoError(waitForAtLeastNBlocks(node, 5, Store))
-	t.Log("Initial 5 blocks produced successfully")
+	waitForBlockN(t, 2, node, config.Node.BlockTime.Duration)
+	t.Log("Initial blocks produced successfully")
 
 	// Get the current height before DA failure
 	initialHeight, err := getNodeHeight(node, Store)
@@ -366,7 +365,7 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 				return
 			default:
 				dummyExecutor.InjectTx([]byte(fmt.Sprintf("tx-after-da-failure-%d", i)))
-				time.Sleep(10 * time.Millisecond) // Inject faster than block time
+				time.Sleep(config.Node.BlockTime.Duration / 2) // Inject faster than block time
 			}
 		}
 	}()
@@ -405,7 +404,17 @@ func TestBatchQueueThrottlingWithDAFailure(t *testing.T) {
 
 	t.Log("NOTE: This test uses DummySequencer. In a real deployment with SingleSequencer,")
 	t.Log("the batch queue would fill up and return ErrQueueFull, providing backpressure.")
+}
 
-	// Shutdown
-	shutdownAndWait(t, []context.CancelFunc{cancel}, &runningWg, 10*time.Second)
+// waitForBlockN waits for the node to produce a block with height >= n.
+func waitForBlockN(t *testing.T, n uint64, node *FullNode, blockInterval time.Duration, timeout ...time.Duration) {
+	t.Helper()
+	if len(timeout) == 0 {
+		timeout = []time.Duration{time.Duration(n+1)*blockInterval + time.Second/2}
+	}
+	require.Eventually(t, func() bool {
+		got, err := getNodeHeight(node, Store)
+		require.NoError(t, err)
+		return got >= n
+	}, timeout[0], blockInterval/2)
 }