refactor(reaping): use cache for seen hashes instead of in memory store (#2811)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

Use cache instead of in memory store for reaper

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->

Author: julienrbrt

CHANGELOG.md

@@ -1,4 +1,3 @@
-
 # Changelog
 
 <!--
@@ -10,9 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- Use cache instead of in memory store for reaper. Persist cache on reload. Autoclean after 24 hours. ([#2811](https://github.com/evstack/ev-node/pull/2811))
+
+## v1.0.0-beta.9
+
 ### Added
 
 <!-- New features or capabilities -->
+
 - Added automated upgrade test for the `evm-single` app that verifies compatibility when moving from v1.0.0-beta.8 to HEAD in CI ([#2780](https://github.com/evstack/ev-node/pull/2780))
 - Added execution-layer replay mechanism so nodes can resynchronize by replaying missed batches against the executor ([#2771](https://github.com/evstack/ev-node/pull/2771))
 - Added cache-pruning logic that evicts entries once heights are finalized to keep node memory usage bounded ([#2761](https://github.com/evstack/ev-node/pull/2761))
@@ -24,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 <!-- Changes to existing functionality -->
+
 - Hardened signer CLI and block pipeline per security audit: passphrases must be provided via `--evnode.signer.passphrase_file`, JWT secrets must be provided via `--evm.jwt-secret-file`, data/header validation enforces metadata and timestamp checks, and the reaper backs off on failures (BREAKING) ([#2764](https://github.com/evstack/ev-node/pull/2764))
 - Added retries around executor `ExecuteTxs` calls to better tolerate transient execution errors ([#2784](https://github.com/evstack/ev-node/pull/2784))
 - Increased default `ReadinessMaxBlocksBehind` from 3 to 30 blocks so `/health/ready` stays true during normal batch sync ([#2779](https://github.com/evstack/ev-node/pull/2779))
@@ -32,11 +39,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Deprecated
 
 <!-- Features that will be removed in future versions -->
--
 
 ### Removed
 
 <!-- Features that were removed -->
+
 - Removed `LastCommitHash`, `ConsensusHash`, and `LastResultsHash` from the canonical header representation in favor of slim headers (BREAKING; legacy hashes now live under `Header.Legacy`) ([#2766](https://github.com/evstack/ev-node/pull/2766))
 
 ### Fixed
@@ -46,7 +53,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Security
 
 <!-- Security vulnerability fixes -->
--
 
 <!--
 ## Category Guidelines:
@@ -117,4 +123,4 @@ Pre-release versions: 0.x.y (anything may change)
 -->
 
 <!-- Links -->
-[Unreleased]: https://github.com/evstack/ev-node/compare/v1.0.0-beta.1...HEAD
+- [Unreleased]: https://github.com/evstack/ev-node/compare/v1.0.0-beta.1...HEAD

block/components.go

@@ -236,6 +236,7 @@ func NewAggregatorComponents(
 		genesis,
 		logger,
 		executor,
+		cacheManager,
 		reaping.DefaultInterval,
 	)
 	if err != nil {

block/internal/cache/bench_test.go

@@ -28,7 +28,7 @@ ok  	github.com/evstack/ev-node/block/internal/cache	25.834s
 */
 
 func benchSetupStore(b *testing.B, n int, txsPer int, chainID string) store.Store {
-	ds, err := store.NewDefaultInMemoryKVStore()
+	ds, err := store.NewTestInMemoryKVStore()
 	if err != nil {
 		b.Fatal(err)
 	}

block/internal/cache/manager.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
+	"time"
 
 	"github.com/rs/zerolog"
 
@@ -21,6 +22,10 @@ var (
 	headerCacheDir        = filepath.Join(cacheDir, "header")
 	dataCacheDir          = filepath.Join(cacheDir, "data")
 	pendingEventsCacheDir = filepath.Join(cacheDir, "pending_da_events")
+	txCacheDir            = filepath.Join(cacheDir, "tx")
+
+	// DefaultTxCacheRetention is the default time to keep transaction hashes in cache
+	DefaultTxCacheRetention = 24 * time.Hour
 )
 
 // gobRegisterOnce ensures gob type registration happens exactly once process-wide.
@@ -51,6 +56,11 @@ type Manager interface {
 	GetDataDAIncluded(hash string) (uint64, bool)
 	SetDataDAIncluded(hash string, daHeight uint64, blockHeight uint64)
 
+	// Transaction operations
+	IsTxSeen(hash string) bool
+	SetTxSeen(hash string)
+	CleanupOldTxs(olderThan time.Duration) int
+
 	// Pending operations
 	GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, error)
 	GetPendingData(ctx context.Context) ([]*types.SignedData, error)
@@ -78,6 +88,8 @@ var _ Manager = (*implementation)(nil)
 type implementation struct {
 	headerCache        *Cache[types.SignedHeader]
 	dataCache          *Cache[types.Data]
+	txCache            *Cache[struct{}]
+	txTimestamps       *sync.Map // map[string]time.Time - tracks when each tx was seen
 	pendingEventsCache *Cache[common.DAHeightEvent]
 	pendingHeaders     *PendingHeaders
 	pendingData        *PendingData
@@ -90,6 +102,7 @@ func NewManager(cfg config.Config, store store.Store, logger zerolog.Logger) (Ma
 	// Initialize caches
 	headerCache := NewCache[types.SignedHeader]()
 	dataCache := NewCache[types.Data]()
+	txCache := NewCache[struct{}]()
 	pendingEventsCache := NewCache[common.DAHeightEvent]()
 
 	// Initialize pending managers
@@ -107,6 +120,8 @@ func NewManager(cfg config.Config, store store.Store, logger zerolog.Logger) (Ma
 	impl := &implementation{
 		headerCache:        headerCache,
 		dataCache:          dataCache,
+		txCache:            txCache,
+		txTimestamps:       new(sync.Map),
 		pendingEventsCache: pendingEventsCache,
 		pendingHeaders:     pendingHeaders,
 		pendingData:        pendingData,
@@ -167,12 +182,71 @@ func (m *implementation) SetDataDAIncluded(hash string, daHeight uint64, blockHe
 	m.dataCache.setDAIncluded(hash, daHeight, blockHeight)
 }
 
+// Transaction operations
+func (m *implementation) IsTxSeen(hash string) bool {
+	return m.txCache.isSeen(hash)
+}
+
+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())
+}
+
+// CleanupOldTxs removes transaction hashes older than the specified duration.
+// Returns the number of transactions removed.
+// This prevents unbounded growth of the transaction cache.
+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) {
+			// Remove from both caches
+			m.txCache.hashes.Delete(hash)
+			m.txTimestamps.Delete(hash)
+			removed++
+		}
+		return true
+	})
+
+	if removed > 0 {
+		m.logger.Debug().
+			Int("removed", removed).
+			Dur("older_than", olderThan).
+			Msg("cleaned up old transaction hashes from cache")
+	}
+
+	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) {
 	m.headerCache.deleteAllForHeight(blockHeight)
 	m.dataCache.deleteAllForHeight(blockHeight)
 	m.pendingEventsCache.deleteAllForHeight(blockHeight)
+
+	// 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
@@ -246,10 +320,18 @@ func (m *implementation) SaveToDisk() error {
 		return fmt.Errorf("failed to save data cache to disk: %w", err)
 	}
 
+	if err := m.txCache.SaveToDisk(filepath.Join(cfgDir, txCacheDir)); err != nil {
+		return fmt.Errorf("failed to save tx cache to disk: %w", err)
+	}
+
 	if err := m.pendingEventsCache.SaveToDisk(filepath.Join(cfgDir, pendingEventsCacheDir)); err != nil {
 		return fmt.Errorf("failed to save pending events cache to disk: %w", err)
 	}
 
+	// Note: txTimestamps are not persisted to disk intentionally.
+	// On restart, all cached transactions will be treated as "new" for cleanup purposes,
+	// which is acceptable as they will be cleaned up on the next cleanup cycle if old enough.
+
 	return nil
 }
 
@@ -267,10 +349,24 @@ func (m *implementation) LoadFromDisk() error {
 		return fmt.Errorf("failed to load data cache from disk: %w", err)
 	}
 
+	if err := m.txCache.LoadFromDisk(filepath.Join(cfgDir, txCacheDir)); err != nil {
+		return fmt.Errorf("failed to load tx cache from disk: %w", err)
+	}
+
 	if err := m.pendingEventsCache.LoadFromDisk(filepath.Join(cfgDir, pendingEventsCacheDir)); err != nil {
 		return fmt.Errorf("failed to load pending events cache from disk: %w", err)
 	}
 
+	// After loading tx cache from disk, initialize timestamps for loaded transactions
+	// Set them to current time so they won't be immediately cleaned up
+	now := time.Now()
+	m.txCache.hashes.Range(func(key, value any) bool {
+		if hash, ok := key.(string); ok {
+			m.txTimestamps.Store(hash, now)
+		}
+		return true
+	})
+
 	return nil
 }