Merge branch 'main' into alex/2803_best_2worlds_fi_rebased2

* main:
  fix(syncer): include in-flight and pending-cache work in PendingCount (#3162)
  chore: re-add replaces for development (#3161)

Author: alpe

CHANGELOG.md

@@ -2,7 +2,7 @@ module github.com/evstack/ev-node/apps/testapp
 
 go 1.25.6
 
-//replace github.com/evstack/ev-node => ../../.
+replace github.com/evstack/ev-node => ../../.
 
 require (
 	github.com/evstack/ev-node v1.0.0

apps/testapp/go.mod

@@ -367,8 +367,6 @@ github.com/envoyproxy/protoc-gen-validate v0.10.0/go.mod h1:DRjgyB0I43LtJapqN6Ni
 github.com/envoyproxy/protoc-gen-validate v0.10.1/go.mod h1:DRjgyB0I43LtJapqN6NiRwroiAU2PaFuvk/vjgh61ss=
 github.com/envoyproxy/protoc-gen-validate v1.0.1/go.mod h1:0vj8bNkYbSTNS2PIyH87KZaeN4x9zpL9Qt8fQC7d+vs=
 github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
-github.com/evstack/ev-node v1.0.0 h1:m3e51fo4Dk9Z32XRV56GJKEeAiqvjiJ9n3SRjG7C5n8=
-github.com/evstack/ev-node v1.0.0/go.mod h1:85H7BPvvRoA+uPfCiIcyWMBN728Wv1uLNhOsjLifJgw=
 github.com/evstack/ev-node/core v1.0.0 h1:s0Tx0uWHme7SJn/ZNEtee4qNM8UO6PIxXnHhPbbKTz8=
 github.com/evstack/ev-node/core v1.0.0/go.mod h1:n2w/LhYQTPsi48m6lMj16YiIqsaQw6gxwjyJvR+B3sY=
 github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=

apps/testapp/go.sum

@@ -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/generic_cache.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/cache/manager.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
@@ -380,7 +381,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)
 			}
 		}
 	}
@@ -404,7 +407,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) {