refactor(block): speed-up cache by retrieving only next pending event (#2702)

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

Speed-up cache by retrieving only next pending event

<!-- 
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/bench_test.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"math/rand/v2"
 	"testing"
 
 	"github.com/rs/zerolog"
@@ -107,10 +108,9 @@ func BenchmarkManager_PendingEventsSnapshot(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for b.Loop() {
-		ev := m.GetPendingEvents()
-		if len(ev) == 0 {
-			b.Fatal("unexpected empty events")
-		}
+		// Test getting next pending event at various heights
+		height := rand.N(uint64(50_000)) + 1 //nolint:gosec // this is a benchmark test
+		_ = m.GetNextPendingEvent(height)
 	}
 }
 

block/internal/cache/generic_cache.go

@@ -46,11 +46,6 @@ func (c *Cache[T]) setItem(height uint64, item *T) {
 	c.itemsByHeight.Store(height, item)
 }
 
-// deleteItem deletes an item from the cache by height
-func (c *Cache[T]) deleteItem(height uint64) {
-	c.itemsByHeight.Delete(height)
-}
-
 // 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) {
@@ -67,6 +62,20 @@ func (c *Cache[T]) rangeByHeight(fn func(height uint64, item *T) bool) {
 	})
 }
 
+// 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 {
+	item, loaded := c.itemsByHeight.LoadAndDelete(height)
+	if !loaded {
+		return nil
+	}
+	val, ok := item.(*T)
+	if !ok {
+		return nil
+	}
+	return val
+}
+
 // isSeen returns true if the hash has been seen
 func (c *Cache[T]) isSeen(hash string) bool {
 	seen, ok := c.hashes.Load(hash)

block/internal/cache/generic_cache_test.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"sync"
 	"testing"
 )
 
@@ -15,32 +14,6 @@ func init() {
 	gob.Register(&testItem{})
 }
 
-// TestCache_ConcurrentOperations exercises concurrent Set, Delete, and Range operations.
-func TestCache_ConcurrentOperations(t *testing.T) {
-	c := NewCache[testItem]()
-
-	const N = 2000
-	var wg sync.WaitGroup
-
-	// writers
-	writer := func(start int) {
-		defer wg.Done()
-		for i := start; i < N; i += 2 {
-			v := &testItem{V: i}
-			c.setItem(uint64(i), v)
-			if i%10 == 0 {
-				// randomly delete some keys
-				c.deleteItem(uint64(i))
-			}
-		}
-	}
-
-	wg.Add(2)
-	go writer(0)
-	go writer(1)
-	wg.Wait()
-}
-
 // TestCache_TypeSafety ensures methods gracefully handle invalid underlying types.
 func TestCache_TypeSafety(t *testing.T) {
 	c := NewCache[testItem]()
@@ -120,6 +93,6 @@ func TestCache_LargeDataset(t *testing.T) {
 	}
 	// Delete a range in the middle
 	for i := 5000; i < 10000; i += 2 {
-		c.deleteItem(uint64(i))
+		c.getNextItem(uint64(i))
 	}
 }

block/internal/cache/manager.go

@@ -58,9 +58,8 @@ type Manager interface {
 	NumPendingData() uint64
 
 	// Pending events syncing coordination
+	GetNextPendingEvent(height uint64) *common.DAHeightEvent
 	SetPendingEvent(height uint64, event *common.DAHeightEvent)
-	GetPendingEvents() map[uint64]*common.DAHeightEvent
-	DeletePendingEvent(height uint64)
 
 	// Cleanup operations
 	SaveToDisk() error
@@ -194,23 +193,15 @@ func (m *implementation) NumPendingData() uint64 {
 	return m.pendingData.NumPendingData()
 }
 
-// Pending events operations
+// SetPendingEvent sets the event at the specified height.
 func (m *implementation) SetPendingEvent(height uint64, event *common.DAHeightEvent) {
 	m.pendingEventsCache.setItem(height, event)
 }
 
-func (m *implementation) GetPendingEvents() map[uint64]*common.DAHeightEvent {
-
-	events := make(map[uint64]*common.DAHeightEvent)
-	m.pendingEventsCache.rangeByHeight(func(height uint64, event *common.DAHeightEvent) bool {
-		events[height] = event
-		return true
-	})
-	return events
-}
-
-func (m *implementation) DeletePendingEvent(height uint64) {
-	m.pendingEventsCache.deleteItem(height)
+// GetNextPendingEvent efficiently retrieves and removes the event at the specified height.
+// Returns nil if no event exists at that height.
+func (m *implementation) GetNextPendingEvent(height uint64) *common.DAHeightEvent {
+	return m.pendingEventsCache.getNextItem(height)
 }
 
 func (m *implementation) SaveToDisk() error {

block/internal/cache/manager_test.go

@@ -68,18 +68,22 @@ func TestManager_PendingEventsCRUD(t *testing.T) {
 	m.SetPendingEvent(1, evt1)
 	m.SetPendingEvent(3, evt3)
 
-	got := m.GetPendingEvents()
-	require.Len(t, got, 3)
-	assert.Equal(t, evt1, got[1])
-	assert.Equal(t, evt3, got[3])
-	assert.Equal(t, evt5, got[5])
-
-	// delete and re-check
-	m.DeletePendingEvent(3)
-	got = m.GetPendingEvents()
-	require.Len(t, got, 2)
-	_, ok := got[3]
-	assert.False(t, ok)
+	// Test getting specific events
+	got1 := m.GetNextPendingEvent(1)
+	require.NotNil(t, got1)
+	assert.Equal(t, evt1.DaHeight, got1.DaHeight)
+
+	got3 := m.GetNextPendingEvent(3)
+	require.NotNil(t, got3)
+	assert.Equal(t, evt3.DaHeight, got3.DaHeight)
+
+	got5 := m.GetNextPendingEvent(5)
+	require.NotNil(t, got5)
+	assert.Equal(t, evt5.DaHeight, got5.DaHeight)
+
+	// Events should be removed after GetNextPendingEvent
+	got1Again := m.GetNextPendingEvent(1)
+	assert.Nil(t, got1Again)
 }
 
 func TestManager_SaveAndLoadFromDisk(t *testing.T) {
@@ -120,9 +124,10 @@ func TestManager_SaveAndLoadFromDisk(t *testing.T) {
 	_, ok2 := m2.GetDataDAIncluded("D2")
 	assert.True(t, ok2)
 
-	events := m2.GetPendingEvents()
-	require.Len(t, events, 1)
-	assert.Equal(t, uint64(2), events[2].Header.Height())
+	// Verify pending event was loaded
+	loadedEvent := m2.GetNextPendingEvent(2)
+	require.NotNil(t, loadedEvent)
+	assert.Equal(t, uint64(2), loadedEvent.Header.Height())
 
 	// directories exist under cfg.RootDir/data/cache/...
 	base := filepath.Join(cfg.RootDir, "data", "cache")
@@ -131,6 +136,19 @@ func TestManager_SaveAndLoadFromDisk(t *testing.T) {
 	assert.DirExists(t, filepath.Join(base, "pending_da_events"))
 }
 
+func TestManager_GetNextPendingEvent_NonExistent(t *testing.T) {
+	t.Parallel()
+	cfg := tempConfig(t)
+	st := memStore(t)
+
+	m, err := NewManager(cfg, st, zerolog.Nop())
+	require.NoError(t, err)
+
+	// Try to get non-existent event
+	event := m.GetNextPendingEvent(999)
+	assert.Nil(t, event)
+}
+
 func TestPendingHeadersAndData_Flow(t *testing.T) {
 	t.Parallel()
 	st := memStore(t)

block/internal/syncing/syncer.go

@@ -371,25 +371,25 @@ func (s *Syncer) processHeightEvent(event *common.DAHeightEvent) {
 	// If this is not the next block in sequence, store as pending event
 	// This check is crucial as trySyncNextBlock simply attempts to sync the next block
 	if height != currentHeight+1 {
-		// Create a DAHeightEvent that matches the cache interface
-		pendingEvent := &common.DAHeightEvent{
-			Header:                 event.Header,
-			Data:                   event.Data,
-			DaHeight:               event.DaHeight,
-			HeaderDaIncludedHeight: event.HeaderDaIncludedHeight,
-		}
-		s.cache.SetPendingEvent(height, pendingEvent)
+		s.cache.SetPendingEvent(height, event)
 		s.logger.Debug().Uint64("height", height).Uint64("current_height", currentHeight).Msg("stored as pending event")
 		return
 	}
 
 	// Try to sync the next block
 	if err := s.trySyncNextBlock(event); err != nil {
 		s.logger.Error().Err(err).Msg("failed to sync next block")
+		// If the error is not due to an validation error, re-store the event as pending
+		if !errors.Is(err, errInvalidBlock) {
+			s.cache.SetPendingEvent(height, event)
+		}
 		return
 	}
 }
 
+// errInvalidBlock is returned when a block is failing validation
+var errInvalidBlock = errors.New("invalid block")
+
 // trySyncNextBlock attempts to sync the next available block
 // the event is always the next block in sequence as processHeightEvent ensures it.
 func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
@@ -410,7 +410,7 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 	// here only the previous block needs to be applied to proceed to the verification.
 	// The header validation must be done before applying the block to avoid executing gibberish
 	if err := s.validateBlock(currentState, header, data); err != nil {
-		return fmt.Errorf("failed to validate block: %w", err)
+		return errors.Join(errInvalidBlock, fmt.Errorf("failed to validate block: %w", err))
 	}
 
 	// Mark as DA included
@@ -551,34 +551,30 @@ func (s *Syncer) isHeightFromFutureError(err error) bool {
 
 // processPendingEvents fetches and processes pending events from cache
 func (s *Syncer) processPendingEvents() {
-	pendingEvents := s.cache.GetPendingEvents()
+	currentHeight, err := s.store.Height(s.ctx)
+	if err != nil {
+		s.logger.Error().Err(err).Msg("failed to get current height for pending events")
+		return
+	}
 
-	for height, event := range pendingEvents {
-		currentHeight, err := s.store.Height(s.ctx)
-		if err != nil {
-			s.logger.Error().Err(err).Msg("failed to get current height for pending events")
-			continue
+	// Try to get the next processable event (currentHeight + 1)
+	nextHeight := currentHeight + 1
+	if event := s.cache.GetNextPendingEvent(nextHeight); event != nil {
+		heightEvent := common.DAHeightEvent{
+			Header:                 event.Header,
+			Data:                   event.Data,
+			DaHeight:               event.DaHeight,
+			HeaderDaIncludedHeight: event.HeaderDaIncludedHeight,
 		}
 
-		// Only process events for blocks we haven't synced yet
-		if height > currentHeight {
-			heightEvent := common.DAHeightEvent{
-				Header:                 event.Header,
-				Data:                   event.Data,
-				DaHeight:               event.DaHeight,
-				HeaderDaIncludedHeight: event.HeaderDaIncludedHeight,
-			}
-
-			select {
-			case s.heightInCh <- heightEvent:
-				// Remove from pending events once sent
-				s.cache.DeletePendingEvent(height)
-			case <-s.ctx.Done():
-				return
-			}
-		} else {
-			// Clean up events for blocks we've already processed
-			s.cache.DeletePendingEvent(height)
+		select {
+		case s.heightInCh <- heightEvent:
+			// Event was successfully sent and already removed by GetNextPendingEvent
+			s.logger.Debug().Uint64("height", nextHeight).Msg("sent pending event to processing")
+		case <-s.ctx.Done():
+			s.cache.SetPendingEvent(nextHeight, event)
+		default:
+			s.cache.SetPendingEvent(nextHeight, event)
 		}
 	}
 }

block/internal/syncing/syncer_benchmark_test.go

@@ -52,8 +52,11 @@ func BenchmarkSyncerIO(b *testing.B) {
 				}, 5*time.Second, 50*time.Microsecond)
 				fixt.s.cancel()
 
-				// Ensure clean end-state per iteration
-				require.Len(b, fixt.s.cache.GetPendingEvents(), 0)
+				// Ensure clean end-state per iteration - verify no pending events remain
+				for i := uint64(1); i <= spec.heights; i++ {
+					event := fixt.s.cache.GetNextPendingEvent(i)
+					require.Nil(b, event, "expected no pending event at height %d", i)
+				}
 				require.Len(b, fixt.s.heightInCh, 0)
 
 				assert.Equal(b, spec.heights+daHeightOffset, fixt.s.daHeight)

block/internal/syncing/syncer_test.go

@@ -144,8 +144,9 @@ func TestSyncer_processPendingEvents(t *testing.T) {
 		t.Fatal("expected a forwarded pending event")
 	}
 
-	remaining := cm.GetPendingEvents()
-	assert.Len(t, remaining, 0)
+	// Verify the event was removed by trying to get it again
+	remaining := cm.GetNextPendingEvent(2)
+	assert.Nil(t, remaining)
 }
 
 func TestSyncLoopPersistState(t *testing.T) {
@@ -230,8 +231,11 @@ func TestSyncLoopPersistState(t *testing.T) {
 	daRtrMock.AssertExpectations(t)
 	p2pHndlMock.AssertExpectations(t)
 
-	// and all processed
-	assert.Len(t, syncerInst1.cache.GetPendingEvents(), 0)
+	// and all processed - verify no events remain at heights we tested
+	event1 := syncerInst1.cache.GetNextPendingEvent(1)
+	assert.Nil(t, event1)
+	event2 := syncerInst1.cache.GetNextPendingEvent(2)
+	assert.Nil(t, event2)
 	assert.Len(t, syncerInst1.heightInCh, 0)
 
 	// and when new instance is up on restart