chore: addressing PR feedback

Author: chatton

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Drain the pending tx queue in merged batches with a durable WAL-backed ack, fixing severe queue backlog under heavy tx load. Tx dedup moved from the reaper cache into the sequencer queue [#3351](https://github.com/evstack/ev-node/pull/3351)
+
 ## v1.1.2
 
 ### Changes

block/components.go

@@ -276,25 +276,13 @@ func newAggregatorComponents(
 	}
 	pruner := pruner.New(logger, store, execPruner, config.Pruning, config.Node.BlockTime.Duration, config.DA.Address)
 
-	// expose the raw sequencer's monotonic enqueue count so the reaper can
-	// distinguish duplicate scrapes from newly queued entries, even when
-	// the sequencer is wrapped with tracing
-	type enqueueCounter interface {
-		TotalEnqueuedBatches() uint64
-	}
-	var totalEnqueuedBatches func() uint64
-	if counter, ok := rawSequencer.(enqueueCounter); ok {
-		totalEnqueuedBatches = counter.TotalEnqueuedBatches
-	}
-
 	reaper, err := reaping.NewReaper(
 		exec,
 		sequencer,
 		genesis,
 		logger,
 		config.Node.ScrapeInterval.Duration,
 		executor.NotifyNewTransactions,
-		totalEnqueuedBatches,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create reaper: %w", err)

block/internal/reaping/bench_test.go

@@ -97,7 +97,6 @@ func benchmarkReaperFlow(b *testing.B, batchSize int, txSize int, feedInterval t
 			zerolog.Nop(),
 			50*time.Millisecond,
 			func() { notified.Add(1) },
-			nil,
 		)
 		if err != nil {
 			b.Fatal(err)
@@ -181,7 +180,6 @@ func BenchmarkReaperFlow_Sustained(b *testing.B) {
 			zerolog.Nop(),
 			10*time.Millisecond,
 			func() { notified.Add(1) },
-			nil,
 		)
 		if err != nil {
 			b.Fatal(err)
@@ -240,7 +238,6 @@ func BenchmarkReaperFlow_StartStop(b *testing.B) {
 				zerolog.Nop(),
 				100*time.Millisecond,
 				func() {},
-				nil,
 			)
 			if err != nil {
 				b.Fatal(err)
@@ -281,7 +278,6 @@ func BenchmarkReaperFlow_StartStop(b *testing.B) {
 				zerolog.Nop(),
 				10*time.Millisecond,
 				func() {},
-				nil,
 			)
 			if err != nil {
 				b.Fatal(err)

block/internal/reaping/reaper.go

@@ -28,12 +28,6 @@ type Reaper struct {
 	interval       time.Duration
 	onTxsSubmitted func()
 
-	// totalEnqueuedBatches reports the sequencer's monotonic enqueue count,
-	// used to detect whether a submission actually enqueued new entries.
-	// monotonicity makes the before/after comparison immune to concurrent
-	// drains shrinking the queue. Optional.
-	totalEnqueuedBatches func() uint64
-
 	logger zerolog.Logger
 
 	ctx    context.Context
@@ -48,20 +42,18 @@ func NewReaper(
 	logger zerolog.Logger,
 	scrapeInterval time.Duration,
 	onTxsSubmitted func(),
-	totalEnqueuedBatches func() uint64,
 ) (*Reaper, error) {
 	if scrapeInterval == 0 {
 		return nil, errors.New("scrape interval cannot be empty")
 	}
 
 	return &Reaper{
-		exec:                 exec,
-		sequencer:            sequencer,
-		chainID:              genesis.ChainID,
-		interval:             scrapeInterval,
-		logger:               logger.With().Str("component", "reaper").Logger(),
-		onTxsSubmitted:       onTxsSubmitted,
-		totalEnqueuedBatches: totalEnqueuedBatches,
+		exec:           exec,
+		sequencer:      sequencer,
+		chainID:        genesis.ChainID,
+		interval:       scrapeInterval,
+		logger:         logger.With().Str("component", "reaper").Logger(),
+		onTxsSubmitted: onTxsSubmitted,
 	}, nil
 }
 
@@ -141,11 +133,6 @@ func (r *Reaper) drainMempool() error {
 		return nil
 	}
 
-	var before uint64
-	if r.totalEnqueuedBatches != nil {
-		before = r.totalEnqueuedBatches()
-	}
-
 	_, err = r.sequencer.SubmitBatchTxs(r.ctx, coresequencer.SubmitBatchTxsRequest{
 		Id:    []byte(r.chainID),
 		Batch: &coresequencer.Batch{Transactions: txs},
@@ -154,19 +141,15 @@ func (r *Reaper) drainMempool() error {
 		return fmt.Errorf("failed to submit txs to sequencer: %w", err)
 	}
 
-	// without an enqueue count we cannot tell duplicates apart, so assume queued
-	queued := true
-	if r.totalEnqueuedBatches != nil {
-		queued = r.totalEnqueuedBatches() > before
-	}
-
-	if r.onTxsSubmitted != nil && queued {
+	// the sequencer dedups resubmitted txs, so this may notify for txs that
+	// were already queued. at worst this triggers an unnecessary (possibly
+	// empty) block in lazy mode.
+	if r.onTxsSubmitted != nil {
 		r.onTxsSubmitted()
 	}
 
 	r.logger.Debug().
 		Int("seen_txs", len(txs)).
-		Bool("queued", queued).
 		Msg("drained mempool")
 
 	return nil

block/internal/reaping/reaper_test.go

@@ -39,7 +39,6 @@ func newTestEnv(t *testing.T) *testEnv {
 		zerolog.Nop(),
 		100*time.Millisecond,
 		env.notify,
-		nil,
 	)
 	require.NoError(t, err)
 	env.reaper = r
@@ -130,7 +129,6 @@ func TestReaper_NilCallback_NoPanic(t *testing.T) {
 		zerolog.Nop(),
 		100*time.Millisecond,
 		nil,
-		nil,
 	)
 	require.NoError(t, err)
 
@@ -143,31 +141,6 @@ func TestReaper_NilCallback_NoPanic(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-func TestReaper_DuplicateScrape_NoNotify(t *testing.T) {
-	mockExec := testmocks.NewMockExecutor(t)
-	mockSeq := testmocks.NewMockSequencer(t)
-
-	var notified atomic.Bool
-	// enqueue count never changes — submission was fully deduped
-	r, err := NewReaper(
-		mockExec, mockSeq,
-		genesis.Genesis{ChainID: "test-chain"},
-		zerolog.Nop(),
-		100*time.Millisecond,
-		func() { notified.Store(true) },
-		func() uint64 { return 1 },
-	)
-	require.NoError(t, err)
-
-	tx := []byte("dup-tx")
-	mockExec.EXPECT().GetTxs(mock.Anything).Return([][]byte{tx}, nil).Once()
-	mockSeq.EXPECT().SubmitBatchTxs(mock.Anything, mock.AnythingOfType("sequencer.SubmitBatchTxsRequest")).
-		Return(&coresequencer.SubmitBatchTxsResponse{}, nil).Once()
-
-	require.NoError(t, r.drainMempool())
-	assert.False(t, notified.Load())
-}
-
 func TestReaper_StopTerminates(t *testing.T) {
 	env := newTestEnv(t)
 	env.execMock.EXPECT().GetTxs(mock.Anything).Return(nil, nil).Maybe()

pkg/sequencers/single/queue.go

@@ -61,10 +61,6 @@ type BatchQueue struct {
 	// by maxQueueSize (unbounded when maxQueueSize is 0).
 	txSeen map[[32]byte]struct{}
 
-	// totalEnqueued counts batches ever enqueued via AddBatch. Monotonic,
-	// never decremented, so callers can detect new enqueues race-free.
-	totalEnqueued uint64
-
 	// Sequence numbers for generating new keys
 	nextAddSeq     uint64
 	nextPrependSeq uint64
@@ -142,7 +138,6 @@ func (bq *BatchQueue) AddBatch(ctx context.Context, batch coresequencer.Batch) e
 	bq.nextAddSeq++
 
 	bq.queue = append(bq.queue, queuedItem{Batch: batch, Key: key})
-	bq.totalEnqueued++
 
 	return nil
 }
@@ -537,15 +532,6 @@ func (bq *BatchQueue) Size() int {
 	return len(bq.queue) - bq.head + len(bq.inFlight)
 }
 
-// TotalEnqueued returns a monotonic count of batches ever enqueued via
-// AddBatch. Unlike Size it never decreases, so comparing two snapshots
-// reliably detects whether new batches were enqueued in between.
-func (bq *BatchQueue) TotalEnqueued() uint64 {
-	bq.mu.Lock()
-	defer bq.mu.Unlock()
-	return bq.totalEnqueued
-}
-
 // persistBatch persists a batch to the datastore with the given key
 func (bq *BatchQueue) persistBatch(ctx context.Context, batch coresequencer.Batch, key string) error {
 	pbBatch := &pb.Batch{

pkg/sequencers/single/sequencer.go

@@ -425,13 +425,6 @@ func (c *Sequencer) AckBatch(ctx context.Context) error {
 	return c.queue.Ack(ctx)
 }
 
-// TotalEnqueuedBatches reports a monotonic count of batches enqueued via
-// SubmitBatchTxs. It never decreases, so two snapshots reliably detect
-// whether new batches were enqueued in between.
-func (c *Sequencer) TotalEnqueuedBatches() uint64 {
-	return c.queue.TotalEnqueued()
-}
-
 // VerifyBatch implements sequencing.Sequencer.
 func (c *Sequencer) VerifyBatch(ctx context.Context, req coresequencer.VerifyBatchRequest) (*coresequencer.VerifyBatchResponse, error) {
 	if !c.isValid(req.Id) {