add caching

Author: julienrbrt

block/manager.go

@@ -106,8 +106,9 @@ type Manager struct {
 	headerStore goheader.Store[*types.SignedHeader]
 	dataStore   goheader.Store[*types.Data]
 
-	headerCache *cache.Cache[types.SignedHeader]
-	dataCache   *cache.Cache[types.Data]
+	headerCache        *cache.Cache[types.SignedHeader]
+	dataCache          *cache.Cache[types.Data]
+	pendingEventsCache *cache.Cache[pendingDAEvent]
 
 	// headerStoreCh is used to notify sync goroutine (HeaderStoreRetrieveLoop) that it needs to retrieve headers from headerStore
 	headerStoreCh chan struct{}
@@ -385,6 +386,7 @@ func NewManager(
 		lastBatchData:                      lastBatchData,
 		headerCache:                        cache.NewCache[types.SignedHeader](),
 		dataCache:                          cache.NewCache[types.Data](),
+		pendingEventsCache:                 cache.NewCache[pendingDAEvent](),
 		retrieveCh:                         make(chan struct{}, 1),
 		daIncluderCh:                       make(chan struct{}, 1),
 		logger:                             logger,
@@ -1060,9 +1062,10 @@ func (m *Manager) NotifyNewTransactions() {
 }
 
 var (
-	cacheDir       = "cache"
-	headerCacheDir = filepath.Join(cacheDir, "header")
-	dataCacheDir   = filepath.Join(cacheDir, "data")
+	cacheDir              = "cache"
+	headerCacheDir        = filepath.Join(cacheDir, "header")
+	dataCacheDir          = filepath.Join(cacheDir, "data")
+	pendingEventsCacheDir = filepath.Join(cacheDir, "pending_da_events")
 )
 
 // HeaderCache returns the headerCache used by the manager.
@@ -1079,6 +1082,7 @@ func (m *Manager) DataCache() *cache.Cache[types.Data] {
 func (m *Manager) LoadCache() error {
 	gob.Register(&types.SignedHeader{})
 	gob.Register(&types.Data{})
+	gob.Register(&pendingDAEvent{})
 
 	cfgDir := filepath.Join(m.config.RootDir, "data")
 
@@ -1090,6 +1094,10 @@ func (m *Manager) LoadCache() error {
 		return fmt.Errorf("failed to load data 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")
+	}
+
 	return nil
 }
 
@@ -1104,6 +1112,11 @@ func (m *Manager) SaveCache() error {
 	if err := m.dataCache.SaveToDisk(filepath.Join(cfgDir, dataCacheDir)); err != nil {
 		return fmt.Errorf("failed to save data 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)
+	}
+
 	return nil
 }
 

block/retriever_da.go

@@ -21,11 +21,11 @@ const (
 	dAFetcherRetries  = 10
 )
 
-// daRetriever encapsulates DA retrieval with pending events management
+// daRetriever encapsulates DA retrieval with pending events management.
+// Pending events are persisted via Manager.pendingEventsCache to avoid data loss on retries or restarts.
 type daRetriever struct {
-	manager       *Manager
-	mutex         sync.RWMutex // mutex for pendingEvents
-	pendingEvents map[uint64]pendingDAEvent
+	manager *Manager
+	mutex   sync.RWMutex // mutex for pendingEvents
 }
 
 // pendingDAEvent represents a DA event waiting for processing
@@ -38,8 +38,7 @@ type pendingDAEvent struct {
 // newDARetriever creates a new DA retriever
 func newDARetriever(manager *Manager) *daRetriever {
 	return &daRetriever{
-		manager:       manager,
-		pendingEvents: make(map[uint64]pendingDAEvent),
+		manager: manager,
 	}
 }
 
@@ -51,6 +50,9 @@ func (m *Manager) DARetrieveLoop(ctx context.Context) {
 
 // run executes the main DA retrieval loop
 func (dr *daRetriever) run(ctx context.Context) {
+	// attempt to process any pending events loaded from disk before starting retrieval loop.
+	dr.processPendingEvents(ctx)
+
 	// blobsFoundCh is used to track when we successfully found a header so
 	// that we can continue to try and find headers that are in the next DA height.
 	// This enables syncing faster than the DA block time.
@@ -292,17 +294,23 @@ func (dr *daRetriever) sendHeightEventIfValid(ctx context.Context, header *types
 	dr.processEvent(ctx, header, data, daHeight)
 }
 
-// queuePendingEvent queues a DA event that cannot be processed immediately
+// queuePendingEvent queues a DA event that cannot be processed immediately.
+// The event is persisted via pendingEventsCache to survive restarts.
 func (dr *daRetriever) queuePendingEvent(header *types.SignedHeader, data *types.Data, daHeight uint64) {
 	dr.mutex.Lock()
 	defer dr.mutex.Unlock()
 
+	if dr.manager.pendingEventsCache == nil {
+		return
+	}
+
 	height := header.Height()
-	dr.pendingEvents[height] = pendingDAEvent{
+	event := pendingDAEvent{
 		header:   header,
 		data:     data,
 		daHeight: daHeight,
 	}
+	dr.manager.pendingEventsCache.SetItem(height, &event)
 
 	dr.manager.logger.Debug().
 		Uint64("height", height).
@@ -341,10 +349,12 @@ func (dr *daRetriever) processEvent(ctx context.Context, header *types.SignedHea
 			Uint64("daHeight", daHeight).
 			Msg("sent complete height event with header and data")
 	default:
+		// Channel full: keep event in pending cache for retry
+		dr.queuePendingEvent(header, data, daHeight)
 		dr.manager.logger.Warn().
 			Uint64("height", header.Height()).
 			Uint64("daHeight", daHeight).
-			Msg("heightInCh backlog full, dropping complete event")
+			Msg("heightInCh backlog full, re-queued event to pending cache")
 	}
 
 	// Try to process any pending events that might now be ready
@@ -353,6 +363,10 @@ func (dr *daRetriever) processEvent(ctx context.Context, header *types.SignedHea
 
 // processPendingEvents tries to process queued DA events that might now be ready
 func (dr *daRetriever) processPendingEvents(ctx context.Context) {
+	if dr.manager.pendingEventsCache == nil {
+		return
+	}
+
 	currentHeight, err := dr.manager.GetStoreHeight(ctx)
 	if err != nil {
 		dr.manager.logger.Debug().Err(err).Msg("failed to get store height for pending DA events")
@@ -362,15 +376,21 @@ func (dr *daRetriever) processPendingEvents(ctx context.Context) {
 	dr.mutex.Lock()
 	defer dr.mutex.Unlock()
 
-	for height, event := range dr.pendingEvents {
+	toDelete := make([]uint64, 0)
+	dr.manager.pendingEventsCache.RangeByHeight(func(height uint64, event *pendingDAEvent) bool {
 		if height <= currentHeight+1 {
 			dr.manager.logger.Debug().
 				Uint64("height", height).
 				Uint64("daHeight", event.daHeight).
 				Msg("processing previously queued DA event")
 			go dr.processEvent(ctx, event.header, event.data, event.daHeight)
-			delete(dr.pendingEvents, height)
+			toDelete = append(toDelete, height)
 		}
+		return true
+	})
+
+	for _, h := range toDelete {
+		dr.manager.pendingEventsCache.DeleteItem(h)
 	}
 }
 

pkg/cache/cache.go

@@ -44,6 +44,23 @@ func (c *Cache[T]) DeleteItem(height uint64) {
 	c.items.Delete(height)
 }
 
+// RangeByHeight iterates over all items keyed by uint64 height and calls fn for each.
+// If fn returns false, iteration stops early.
+// Non-uint64 keys (e.g. string hash entries) are ignored.
+func (c *Cache[T]) RangeByHeight(fn func(height uint64, item *T) bool) {
+	c.items.Range(func(k, v interface{}) bool {
+		height, ok := k.(uint64)
+		if !ok {
+			return true
+		}
+		item, ok := v.(*T)
+		if !ok {
+			return true
+		}
+		return fn(height, item)
+	})
+}
+
 // IsSeen returns true if the hash has been seen
 func (c *Cache[T]) IsSeen(hash string) bool {
 	seen, ok := c.hashes.Load(hash)