refactor(block): better sync/process loops (#2688)

## Overview

Moves processPendingEvents into the producer loop.
Minor refactorings

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

.mockery.yaml

@@ -46,3 +46,15 @@ packages:
           dir: ./test/mocks
           pkgname: mocks
           filename: external/hstore.go
+  github.com/evstack/ev-node/block/internal/syncing:
+    interfaces:
+      daRetriever:
+        config:
+          dir: ./block/internal/syncing
+          pkgname: syncing
+          filename: syncer_mock.go
+      p2pHandler:
+        config:
+          dir: ./block/internal/syncing
+          pkgname: syncing
+          filename: syncer_mock.go

apps/evm/single/go.mod

@@ -277,6 +277,7 @@ require (
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
 	github.com/spf13/viper v1.21.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/supranational/blst v0.3.14 // indirect

apps/grpc/single/go.mod

@@ -130,6 +130,7 @@ require (
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
 	github.com/spf13/viper v1.21.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect

apps/testapp/go.mod

@@ -139,6 +139,7 @@ require (
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
 	github.com/spf13/viper v1.21.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
 	github.com/wlynxg/anet v0.0.5 // indirect

block/internal/syncing/syncer.go

@@ -22,6 +22,14 @@ import (
 	"github.com/evstack/ev-node/types"
 )
 
+type daRetriever interface {
+	RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
+}
+type p2pHandler interface {
+	ProcessHeaderRange(ctx context.Context, fromHeight, toHeight uint64) []common.DAHeightEvent
+	ProcessDataRange(ctx context.Context, fromHeight, toHeight uint64) []common.DAHeightEvent
+}
+
 // Syncer handles block synchronization from DA and P2P sources.
 type Syncer struct {
 	// Core components
@@ -51,14 +59,12 @@ type Syncer struct {
 	dataStore   goheader.Store[*types.Data]
 
 	// Channels for coordination
-	heightInCh    chan common.DAHeightEvent
-	headerStoreCh chan struct{}
-	dataStoreCh   chan struct{}
-	errorCh       chan<- error // Channel to report critical execution client failures
+	heightInCh chan common.DAHeightEvent
+	errorCh    chan<- error // Channel to report critical execution client failures
 
 	// Handlers
-	daRetriever *DARetriever
-	p2pHandler  *P2PHandler
+	daRetriever daRetriever
+	p2pHandler  p2pHandler
 
 	// Logging
 	logger zerolog.Logger
@@ -85,23 +91,21 @@ func NewSyncer(
 	errorCh chan<- error,
 ) *Syncer {
 	return &Syncer{
-		store:         store,
-		exec:          exec,
-		da:            da,
-		cache:         cache,
-		metrics:       metrics,
-		config:        config,
-		genesis:       genesis,
-		options:       options,
-		headerStore:   headerStore,
-		dataStore:     dataStore,
-		lastStateMtx:  &sync.RWMutex{},
-		daStateMtx:    &sync.RWMutex{},
-		heightInCh:    make(chan common.DAHeightEvent, 10000),
-		headerStoreCh: make(chan struct{}, 1),
-		dataStoreCh:   make(chan struct{}, 1),
-		errorCh:       errorCh,
-		logger:        logger.With().Str("component", "syncer").Logger(),
+		store:        store,
+		exec:         exec,
+		da:           da,
+		cache:        cache,
+		metrics:      metrics,
+		config:       config,
+		genesis:      genesis,
+		options:      options,
+		headerStore:  headerStore,
+		dataStore:    dataStore,
+		lastStateMtx: &sync.RWMutex{},
+		daStateMtx:   &sync.RWMutex{},
+		heightInCh:   make(chan common.DAHeightEvent, 10_000),
+		errorCh:      errorCh,
+		logger:       logger.With().Str("component", "syncer").Logger(),
 	}
 }
 
@@ -212,20 +216,10 @@ func (s *Syncer) processLoop() {
 	s.logger.Info().Msg("starting process loop")
 	defer s.logger.Info().Msg("process loop stopped")
 
-	blockTicker := time.NewTicker(s.config.Node.BlockTime.Duration)
-	defer blockTicker.Stop()
-
 	for {
-		// Process pending events from cache on every iteration
-		s.processPendingEvents()
-
 		select {
 		case <-s.ctx.Done():
 			return
-		case <-blockTicker.C:
-			// Signal P2P stores to check for new data
-			s.sendNonBlockingSignal(s.headerStoreCh, "header_store")
-			s.sendNonBlockingSignal(s.dataStoreCh, "data_store")
 		case heightEvent := <-s.heightInCh:
 			s.processHeightEvent(&heightEvent)
 		}
@@ -250,15 +244,19 @@ func (s *Syncer) syncLoop() {
 	var hffDelay time.Duration
 	var nextDARequestAt time.Time
 
+	blockTicker := time.NewTicker(s.config.Node.BlockTime.Duration)
+	defer blockTicker.Stop()
+
 	// TODO: we should request to see what the head of the chain is at
 	// then we know if we are falling behind or in sync mode
-syncLoop:
 	for {
 		select {
 		case <-s.ctx.Done():
 			return
 		default:
 		}
+		// Process pending events from cache on every iteration
+		s.processPendingEvents()
 
 		now := time.Now()
 		// Respect backoff window if set
@@ -291,9 +289,7 @@ syncLoop:
 					select {
 					case s.heightInCh <- event:
 					default:
-						s.logger.Warn().Msg("height channel full, dropping DA event")
-						time.Sleep(10 * time.Millisecond)
-						continue syncLoop
+						s.cache.SetPendingEvent(event.Header.Height(), &event)
 					}
 				}
 
@@ -303,34 +299,34 @@ syncLoop:
 			}
 		}
 
+		// Opportunistically process any P2P signals
 		select {
-		case <-s.headerStoreCh:
+		case <-blockTicker.C:
 			newHeaderHeight := s.headerStore.Height()
 			if newHeaderHeight > lastHeaderHeight {
 				events := s.p2pHandler.ProcessHeaderRange(s.ctx, lastHeaderHeight+1, newHeaderHeight)
 				for _, event := range events {
 					select {
 					case s.heightInCh <- event:
 					default:
-						s.logger.Warn().Msg("height channel full, dropping P2P header event")
-						time.Sleep(10 * time.Millisecond)
-						continue syncLoop
+						s.cache.SetPendingEvent(event.Header.Height(), &event)
 					}
 				}
-
 				lastHeaderHeight = newHeaderHeight
 			}
-		case <-s.dataStoreCh:
+
 			newDataHeight := s.dataStore.Height()
+			if newDataHeight == newHeaderHeight {
+				lastDataHeight = newDataHeight
+				continue
+			}
 			if newDataHeight > lastDataHeight {
 				events := s.p2pHandler.ProcessDataRange(s.ctx, lastDataHeight+1, newDataHeight)
 				for _, event := range events {
 					select {
 					case s.heightInCh <- event:
 					default:
-						s.logger.Warn().Msg("height channel full, dropping P2P data event")
-						time.Sleep(10 * time.Millisecond)
-						continue syncLoop
+						s.cache.SetPendingEvent(event.Header.Height(), &event)
 					}
 				}
 				lastDataHeight = newDataHeight