fix: drain pending tx queue in batches with durable ack (#3351)

* fix: drain pending tx queue in batches with durable ack

Resolves slow pending-tx drain caused by one-queue-entry-per-block
consumption and seen-on-enqueue dedup poisoning:

- single sequencer drains multiple queue entries per block up to
  MaxBytes; WAL entries are deleted only on ack after block commit,
  with retry before the next batch and startup reconciliation against
  the last committed block (DropIncluded)
- tx dedup moves from the 30-min cache-manager tx cache into the batch
  queue itself, keyed by sha256, released on ack; dead cache tx-seen
  subsystem removed
- postponed txs are requeued durably during ack and stay deduped
- Load cleans duplicate/stale WAL entries so restarts cannot
  resurrect committed txs
- reaper submits one scrape per interval and notifies the executor
  only when new entries were actually queued (explicit pending count
  wiring, immune to tracing wrappers)

* fix: address review findings in batch queue drain/ack path

- defer postponed entry requeue until ack fully succeeds so a drain
  rollback after a failed ack neither loses nor duplicates postponed txs
- replace fmt.Printf with structured logging in BatchQueue
- use monotonic enqueue counter in reaper to detect new submissions
  race-free against concurrent drain/ack

* chore: fix import grouping in reaping bench test

* fix: address PR review feedback on batch queue durability

- fail Load on datastore read errors instead of dropping txs
- detach rollback WAL cleanup from drain context so it runs on shutdown
- raise queue load timeout for large WAL backlogs
- merge postponed-tx collection into the filter loop
- document DropIncluded aliasing, SetPostponed contract, txSeen bound
- add tests: executor ack retry, reconcile crash recovery, bulk-prepend rollback

* chore: addressing PR feedback

* chore: 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

@@ -278,7 +278,6 @@ func newAggregatorComponents(
 		sequencer,
 		genesis,
 		logger,
-		cacheManager,
 		config.Node.ScrapeInterval.Duration,
 		executor.NotifyNewTransactions,
 	)

block/internal/cache/generic_cache.go

@@ -58,19 +58,6 @@ func (c *Cache) isSeen(hash string) bool {
 	return c.hashes[hash]
 }
 
-// areSeen checks which hashes have been seen. Returns a boolean slice
-// parallel to the input where result[i] is true if hashes[i] is in the
-// cache. Acquires the read lock once for the entire batch.
-func (c *Cache) areSeen(hashes []string) []bool {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	result := make([]bool, len(hashes))
-	for i, h := range hashes {
-		result[i] = c.hashes[h]
-	}
-	return result
-}
-
 func (c *Cache) setSeen(hash string, height uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -82,37 +69,6 @@ func (c *Cache) setSeen(hash string, height uint64) {
 	c.hashByHeight[height] = hash
 }
 
-func (c *Cache) removeSeen(hash string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	delete(c.hashes, hash)
-}
-
-// setSeenBatch marks all hashes as seen under a single write lock.
-// For height 0 (transactions), the hashByHeight bookkeeping is skipped
-// since all txs share the same sentinel height — the map lookup and
-// overwrite on every entry is pure overhead with no benefit.
-func (c *Cache) setSeenBatch(hashes []string, height uint64) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if height == 0 {
-		for _, h := range hashes {
-			c.hashes[h] = true
-		}
-		return
-	}
-
-	// currently not used, but there for completeness against setSeen
-	for _, h := range hashes {
-		if existing, ok := c.hashByHeight[height]; ok && existing == h {
-			c.hashes[existing] = true
-			continue
-		}
-		c.hashes[h] = true
-		c.hashByHeight[height] = h
-	}
-}
-
 func (c *Cache) getDAIncluded(hash string) (uint64, bool) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()

block/internal/cache/generic_cache_test.go

@@ -199,8 +199,6 @@ func TestCache_BasicOperations(t *testing.T) {
 	assert.False(t, c.isSeen("hash1"))
 	c.setSeen("hash1", 1)
 	assert.True(t, c.isSeen("hash1"))
-	c.removeSeen("hash1")
-	assert.False(t, c.isSeen("hash1"))
 
 	_, ok := c.getDAIncluded("hash2")
 	assert.False(t, ok)

block/internal/cache/manager.go

@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/rs/zerolog"
 
@@ -24,10 +23,6 @@ const (
 
 	// DataDAIncludedPrefix is the store key prefix for data DA inclusion tracking.
 	DataDAIncludedPrefix = "cache/data-da-included/"
-
-	// DefaultTxCacheRetention is the default time to keep transaction hashes in cache.
-	// Keeping a too high value can lead to OOM during heavy transaction load.
-	DefaultTxCacheRetention = 30 * time.Minute
 )
 
 // CacheManager provides thread-safe cache operations for tracking seen blocks
@@ -51,13 +46,6 @@ type CacheManager interface {
 	SetDataDAIncluded(daCommitmentHash string, daHeight uint64, blockHeight uint64)
 	RemoveDataDAIncluded(hash string)
 
-	// Transaction operations
-	IsTxSeen(hash string) bool
-	AreTxsSeen(hashes []string) []bool
-	SetTxSeen(hash string)
-	SetTxsSeen(hashes []string)
-	CleanupOldTxs(olderThan time.Duration) int
-
 	// Pending events syncing coordination
 	GetNextPendingEvent(blockHeight uint64) *common.DAHeightEvent
 	SetPendingEvent(blockHeight uint64, event *common.DAHeightEvent)
@@ -94,8 +82,6 @@ var _ Manager = (*implementation)(nil)
 type implementation struct {
 	headerCache    *Cache
 	dataCache      *Cache
-	txCache        *Cache
-	txTimestamps   *sync.Map // map[string]time.Time
 	pendingEvents  map[uint64]*common.DAHeightEvent
 	pendingMu      sync.Mutex
 	pendingHeaders *PendingHeaders
@@ -109,7 +95,6 @@ type implementation struct {
 func NewManager(cfg config.Config, st store.Store, logger zerolog.Logger) (Manager, error) {
 	headerCache := NewCache(st, HeaderDAIncludedPrefix)
 	dataCache := NewCache(st, DataDAIncludedPrefix)
-	txCache := NewCache(nil, "")
 
 	pendingHeaders, err := NewPendingHeaders(st, logger)
 	if err != nil {
@@ -124,8 +109,6 @@ func NewManager(cfg config.Config, st store.Store, logger zerolog.Logger) (Manag
 	impl := &implementation{
 		headerCache:    headerCache,
 		dataCache:      dataCache,
-		txCache:        txCache,
-		txTimestamps:   new(sync.Map),
 		pendingEvents:  make(map[uint64]*common.DAHeightEvent),
 		pendingHeaders: pendingHeaders,
 		pendingData:    pendingData,
@@ -202,59 +185,6 @@ func (m *implementation) RemoveDataDAIncluded(hash string) {
 	m.dataCache.removeDAIncluded(hash)
 }
 
-func (m *implementation) IsTxSeen(hash string) bool {
-	return m.txCache.isSeen(hash)
-}
-
-func (m *implementation) AreTxsSeen(hashes []string) []bool {
-	return m.txCache.areSeen(hashes)
-}
-
-func (m *implementation) SetTxSeen(hash string) {
-	// Use 0 as height since transactions don't have a block height yet
-	m.txCache.setSeen(hash, 0)
-	// Track timestamp for cleanup purposes
-	m.txTimestamps.Store(hash, time.Now())
-}
-
-func (m *implementation) SetTxsSeen(hashes []string) {
-	m.txCache.setSeenBatch(hashes, 0)
-	now := time.Now()
-	for _, hash := range hashes {
-		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 {
-	if olderThan <= 0 {
-		olderThan = DefaultTxCacheRetention
-	}
-
-	cutoff := time.Now().Add(-olderThan)
-	removed := 0
-
-	m.txTimestamps.Range(func(key, value any) bool {
-		hash, ok := key.(string)
-		if !ok {
-			return true
-		}
-		timestamp, ok := value.(time.Time)
-		if !ok {
-			return true
-		}
-		if timestamp.Before(cutoff) {
-			m.txCache.removeSeen(hash)
-			m.txTimestamps.Delete(hash)
-			removed++
-		}
-		return true
-	})
-
-	return removed
-}
-
 // DeleteHeight removes from all caches the given height.
 // This can be done when a height has been da included.
 func (m *implementation) DeleteHeight(blockHeight uint64) {
@@ -263,12 +193,6 @@ func (m *implementation) DeleteHeight(blockHeight uint64) {
 	m.pendingMu.Lock()
 	delete(m.pendingEvents, blockHeight)
 	m.pendingMu.Unlock()
-
-	// Note: txCache is intentionally NOT deleted here because:
-	// 1. Transactions are tracked by hash, not by block height (they use height 0)
-	// 2. A transaction seen at one height may be resubmitted at a different height
-	// 3. The cache prevents duplicate submissions across block heights
-	// 4. Cleanup is handled separately via CleanupOldTxs() based on time, not height
 }
 
 // Pending operations
@@ -363,7 +287,7 @@ func (m *implementation) SaveToStore() error {
 		return fmt.Errorf("failed to save data cache to store: %w", err)
 	}
 
-	// TX cache and pending events are ephemeral - not persisted
+	// pending events are ephemeral - not persisted
 	return nil
 }
 
@@ -406,7 +330,6 @@ func (m *implementation) ClearFromStore() error {
 
 	m.headerCache = NewCache(m.store, HeaderDAIncludedPrefix)
 	m.dataCache = NewCache(m.store, DataDAIncludedPrefix)
-	m.txCache = NewCache(nil, "")
 	m.pendingEvents = make(map[uint64]*common.DAHeightEvent)
 
 	// Initialize DA height from store metadata to ensure DaHeight() is never 0.