feat(cache): cache pruning (#2761)

<!--
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

Supersed #2748
Except it is fully contained in the cache.

We delete all the cache for the height that just been marked as
included.

<!-- 
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

block/internal/cache/generic_cache.go

@@ -16,6 +16,8 @@ type Cache[T any] struct {
 	hashes *sync.Map
 	// daIncluded tracks the DA inclusion height for a given hash
 	daIncluded *sync.Map
+	// hashByHeight tracks the hash associated with each height for pruning
+	hashByHeight *sync.Map
 }
 
 // NewCache returns a new Cache struct
@@ -24,6 +26,7 @@ func NewCache[T any]() *Cache[T] {
 		itemsByHeight: new(sync.Map),
 		hashes:        new(sync.Map),
 		daIncluded:    new(sync.Map),
+		hashByHeight:  new(sync.Map),
 	}
 }
 
@@ -46,22 +49,6 @@ func (c *Cache[T]) setItem(height uint64, item *T) {
 	c.itemsByHeight.Store(height, item)
 }
 
-// rangeByHeight iterates over items keyed by height in an unspecified order and calls fn for each.
-// If fn returns false, iteration stops early.
-func (c *Cache[T]) rangeByHeight(fn func(height uint64, item *T) bool) {
-	c.itemsByHeight.Range(func(k, v any) bool {
-		height, ok := k.(uint64)
-		if !ok {
-			return true
-		}
-		it, ok := v.(*T)
-		if !ok {
-			return true
-		}
-		return fn(height, it)
-	})
-}
-
 // getNextItem returns the item at the specified height and removes it from cache if found.
 // Returns nil if not found.
 func (c *Cache[T]) getNextItem(height uint64) *T {
@@ -85,9 +72,10 @@ func (c *Cache[T]) isSeen(hash string) bool {
 	return seen.(bool)
 }
 
-// setSeen sets the hash as seen
-func (c *Cache[T]) setSeen(hash string) {
+// setSeen sets the hash as seen and tracks its height for pruning
+func (c *Cache[T]) setSeen(hash string, height uint64) {
 	c.hashes.Store(hash, true)
+	c.hashByHeight.Store(height, hash)
 }
 
 // getDAIncluded returns the DA height if the hash has been DA-included, otherwise it returns 0.
@@ -99,16 +87,28 @@ func (c *Cache[T]) getDAIncluded(hash string) (uint64, bool) {
 	return daIncluded.(uint64), true
 }
 
-// setDAIncluded sets the hash as DA-included with the given DA height
-func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64) {
+// setDAIncluded sets the hash as DA-included with the given DA height and tracks block height for pruning
+func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64, blockHeight uint64) {
 	c.daIncluded.Store(hash, daHeight)
+	c.hashByHeight.Store(blockHeight, hash)
 }
 
 // removeDAIncluded removes the DA-included status of the hash
 func (c *Cache[T]) removeDAIncluded(hash string) {
 	c.daIncluded.Delete(hash)
 }
 
+// deleteAll removes all items and their associated data from the cache at the given height
+func (c *Cache[T]) deleteAllForHeight(height uint64) {
+	c.itemsByHeight.Delete(height)
+	hash, ok := c.hashByHeight.Load(height)
+	if ok {
+		c.hashes.Delete(hash)
+		c.hashByHeight.Delete(height)
+		// c.daIncluded.Delete(hash) // we actually do not want to delete the DA-included status here
+	}
+}
+
 const (
 	itemsByHeightFilename = "items_by_height.gob"
 	hashesFilename        = "hashes.gob"

block/internal/cache/generic_cache_test.go

@@ -14,31 +14,14 @@ func init() {
 	gob.Register(&testItem{})
 }
 
-// TestCache_TypeSafety ensures methods gracefully handle invalid underlying types.
-func TestCache_TypeSafety(t *testing.T) {
-	c := NewCache[testItem]()
-
-	// Inject invalid value types directly into maps (bypassing typed methods)
-	c.itemsByHeight.Store(uint64(1), "not-a-*testItem")
-
-	if got := c.getItem(1); got != nil {
-		t.Fatalf("expected nil for invalid stored type, got %#v", got)
-	}
-
-	// Range should skip invalid entries and not panic
-	ran := false
-	c.rangeByHeight(func(_ uint64, _ *testItem) bool { ran = true; return true })
-	_ = ran // ensure no panic
-}
-
 // TestCache_SaveLoad_ErrorPaths covers SaveToDisk and LoadFromDisk error scenarios.
 func TestCache_SaveLoad_ErrorPaths(t *testing.T) {
 	c := NewCache[testItem]()
 	for i := 0; i < 5; i++ {
 		v := &testItem{V: i}
 		c.setItem(uint64(i), v)
-		c.setSeen(fmt.Sprintf("s%d", i))
-		c.setDAIncluded(fmt.Sprintf("d%d", i), uint64(i))
+		c.setSeen(fmt.Sprintf("s%d", i), uint64(i))
+		c.setDAIncluded(fmt.Sprintf("d%d", i), uint64(i), uint64(i))
 	}
 
 	// Normal save/load roundtrip

block/internal/cache/manager.go

@@ -40,16 +40,16 @@ func registerGobTypes() {
 type Manager interface {
 	// Header operations
 	IsHeaderSeen(hash string) bool
-	SetHeaderSeen(hash string)
+	SetHeaderSeen(hash string, blockHeight uint64)
 	GetHeaderDAIncluded(hash string) (uint64, bool)
-	SetHeaderDAIncluded(hash string, daHeight uint64)
+	SetHeaderDAIncluded(hash string, daHeight uint64, blockHeight uint64)
 	RemoveHeaderDAIncluded(hash string)
 
 	// Data operations
 	IsDataSeen(hash string) bool
-	SetDataSeen(hash string)
+	SetDataSeen(hash string, blockHeight uint64)
 	GetDataDAIncluded(hash string) (uint64, bool)
-	SetDataDAIncluded(hash string, daHeight uint64)
+	SetDataDAIncluded(hash string, daHeight uint64, blockHeight uint64)
 
 	// Pending operations
 	GetPendingHeaders(ctx context.Context) ([]*types.SignedHeader, error)
@@ -60,13 +60,16 @@ type Manager interface {
 	NumPendingData() uint64
 
 	// Pending events syncing coordination
-	GetNextPendingEvent(height uint64) *common.DAHeightEvent
-	SetPendingEvent(height uint64, event *common.DAHeightEvent)
+	GetNextPendingEvent(blockHeight uint64) *common.DAHeightEvent
+	SetPendingEvent(blockHeight uint64, event *common.DAHeightEvent)
 
-	// Cleanup operations
+	// Disk operations
 	SaveToDisk() error
 	LoadFromDisk() error
 	ClearFromDisk() error
+
+	// Cleanup operations
+	DeleteHeight(blockHeight uint64)
 }
 
 var _ Manager = (*implementation)(nil)
@@ -100,6 +103,7 @@ func NewManager(cfg config.Config, store store.Store, logger zerolog.Logger) (Ma
 		return nil, fmt.Errorf("failed to create pending data: %w", err)
 	}
 
+	registerGobTypes()
 	impl := &implementation{
 		headerCache:        headerCache,
 		dataCache:          dataCache,
@@ -130,16 +134,16 @@ func (m *implementation) IsHeaderSeen(hash string) bool {
 	return m.headerCache.isSeen(hash)
 }
 
-func (m *implementation) SetHeaderSeen(hash string) {
-	m.headerCache.setSeen(hash)
+func (m *implementation) SetHeaderSeen(hash string, blockHeight uint64) {
+	m.headerCache.setSeen(hash, blockHeight)
 }
 
 func (m *implementation) GetHeaderDAIncluded(hash string) (uint64, bool) {
 	return m.headerCache.getDAIncluded(hash)
 }
 
-func (m *implementation) SetHeaderDAIncluded(hash string, daHeight uint64) {
-	m.headerCache.setDAIncluded(hash, daHeight)
+func (m *implementation) SetHeaderDAIncluded(hash string, daHeight uint64, blockHeight uint64) {
+	m.headerCache.setDAIncluded(hash, daHeight, blockHeight)
 }
 
 func (m *implementation) RemoveHeaderDAIncluded(hash string) {
@@ -151,16 +155,24 @@ func (m *implementation) IsDataSeen(hash string) bool {
 	return m.dataCache.isSeen(hash)
 }
 
-func (m *implementation) SetDataSeen(hash string) {
-	m.dataCache.setSeen(hash)
+func (m *implementation) SetDataSeen(hash string, blockHeight uint64) {
+	m.dataCache.setSeen(hash, blockHeight)
 }
 
 func (m *implementation) GetDataDAIncluded(hash string) (uint64, bool) {
 	return m.dataCache.getDAIncluded(hash)
 }
 
-func (m *implementation) SetDataDAIncluded(hash string, daHeight uint64) {
-	m.dataCache.setDAIncluded(hash, daHeight)
+func (m *implementation) SetDataDAIncluded(hash string, daHeight uint64, blockHeight uint64) {
+	m.dataCache.setDAIncluded(hash, daHeight, blockHeight)
+}
+
+// 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)
 }
 
 // Pending operations

block/internal/cache/manager_test.go

@@ -39,13 +39,13 @@ func TestManager_HeaderDataOperations(t *testing.T) {
 	require.NoError(t, err)
 
 	// seen & DA included flags
-	m.SetHeaderSeen("h1")
-	m.SetDataSeen("d1")
+	m.SetHeaderSeen("h1", 1)
+	m.SetDataSeen("d1", 1)
 	assert.True(t, m.IsHeaderSeen("h1"))
 	assert.True(t, m.IsDataSeen("d1"))
 
-	m.SetHeaderDAIncluded("h1", 10)
-	m.SetDataDAIncluded("d1", 11)
+	m.SetHeaderDAIncluded("h1", 10, 2)
+	m.SetDataDAIncluded("d1", 11, 2)
 	_, ok := m.GetHeaderDAIncluded("h1")
 	assert.True(t, ok)
 	_, ok = m.GetDataDAIncluded("d1")
@@ -102,10 +102,10 @@ func TestManager_SaveAndLoadFromDisk(t *testing.T) {
 	// populate caches
 	hdr := &types.SignedHeader{Header: types.Header{BaseHeader: types.BaseHeader{ChainID: "c", Height: 2}}}
 	dat := &types.Data{Metadata: &types.Metadata{ChainID: "c", Height: 2}}
-	m1.SetHeaderSeen("H2")
-	m1.SetDataSeen("D2")
-	m1.SetHeaderDAIncluded("H2", 100)
-	m1.SetDataDAIncluded("D2", 101)
+	m1.SetHeaderSeen("H2", 2)
+	m1.SetDataSeen("D2", 2)
+	m1.SetHeaderDAIncluded("H2", 100, 2)
+	m1.SetDataDAIncluded("D2", 101, 2)
 	m1.SetPendingEvent(2, &common.DAHeightEvent{Header: hdr, Data: dat, DaHeight: 99})
 
 	// persist

block/internal/submitting/da_submitter.go

@@ -225,7 +225,7 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
 		},
 		func(submitted []*types.SignedHeader, res *coreda.ResultSubmit, gasPrice float64) {
 			for _, header := range submitted {
-				cache.SetHeaderDAIncluded(header.Hash().String(), res.Height)
+				cache.SetHeaderDAIncluded(header.Hash().String(), res.Height, header.Height())
 			}
 			// Update last submitted height
 			if l := len(submitted); l > 0 {
@@ -270,7 +270,7 @@ func (s *DASubmitter) SubmitData(ctx context.Context, cache cache.Manager, signe
 		},
 		func(submitted []*types.SignedData, res *coreda.ResultSubmit, gasPrice float64) {
 			for _, sd := range submitted {
-				cache.SetDataDAIncluded(sd.Data.DACommitment().String(), res.Height)
+				cache.SetDataDAIncluded(sd.Data.DACommitment().String(), res.Height, sd.Height())
 			}
 			if l := len(submitted); l > 0 {
 				lastHeight := submitted[l-1].Height()

block/internal/submitting/submitter.go

@@ -234,6 +234,10 @@ func (s *Submitter) processDAInclusionLoop() {
 				if err := s.store.SetMetadata(s.ctx, store.DAIncludedHeightKey, bz); err != nil {
 					s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to persist DA included height")
 				}
+
+				// Delete height cache for that height
+				// This can only be performed after the height has been persisted to store
+				s.cache.DeleteHeight(nextHeight)
 			}
 		}
 	}