fix(syncer): include in-flight and pending-cache work in PendingCount

Author: alpe

block/internal/cache/generic_cache.go

@@ -113,6 +113,11 @@ func (c *Cache[T]) getNextItem(height uint64) *T {
 	return item
 }
 
+// itemCount returns the number of items currently stored by height.
+func (c *Cache[T]) itemCount() int {
+	return c.itemsByHeight.Len()
+}
+
 // isSeen returns true if the hash has been seen.
 func (c *Cache[T]) isSeen(hash string) bool {
 	seen, ok := c.hashes.Get(hash)

block/internal/cache/manager.go

@@ -58,6 +58,7 @@ type CacheManager interface {
 	// Pending events syncing coordination
 	GetNextPendingEvent(blockHeight uint64) *common.DAHeightEvent
 	SetPendingEvent(blockHeight uint64, event *common.DAHeightEvent)
+	PendingEventsCount() int
 
 	// Store operations
 	SaveToStore() error
@@ -321,6 +322,10 @@ func (m *implementation) SetPendingEvent(height uint64, event *common.DAHeightEv
 	m.pendingEventsCache.setItem(height, event)
 }
 
+func (m *implementation) PendingEventsCount() int {
+	return m.pendingEventsCache.itemCount()
+}
+
 // 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 {

block/internal/syncing/syncer.go

@@ -72,6 +72,7 @@ type Syncer struct {
 	// Channels for coordination
 	heightInCh chan common.DAHeightEvent
 	errorCh    chan<- error // Channel to report critical execution client failures
+	inFlight   atomic.Int64
 
 	// Handlers
 	daRetriever   DARetriever
@@ -379,7 +380,9 @@ func (s *Syncer) processLoop(ctx context.Context) {
 			return
 		case heightEvent, ok := <-s.heightInCh:
 			if ok {
+				s.inFlight.Add(1)
 				s.processHeightEvent(ctx, &heightEvent)
+				s.inFlight.Add(-1)
 			}
 		}
 	}
@@ -403,7 +406,7 @@ func (s *Syncer) HasReachedDAHead() bool {
 
 // PendingCount returns the number of unprocessed height events in the pipeline.
 func (s *Syncer) PendingCount() int {
-	return len(s.heightInCh)
+	return len(s.heightInCh) + int(s.inFlight.Load()) + s.cache.PendingEventsCount()
 }
 
 func (s *Syncer) pendingWorkerLoop(ctx context.Context) {