refactor(syncing): remove retry da retriever (#2711)

## Overview

The da height is never bumped if a blob has not been fetched.
It is bumped at successful blob retrieval (see syncer): an empty blob or
a valid blobs bumps it.

Additionally we already have a backoff for height in the future in the
syncer.
This system has been applied to all errors, which makes retries faster,
as it waits for DA block time.

We should remove the retry logic within the retriever, as we do still
want to check p2p prior to that.
We should automatically retry already at the next loop iteration

-
https://github.com/evstack/ev-node/blob/v1.0.0-beta.5/block/internal/syncing/syncer.go#L281-L283
-
https://github.com/evstack/ev-node/blob/v1.0.0-beta.5/block/internal/syncing/syncer.go#L302-L303

Author: julienrbrt

block/internal/syncing/da_retriever.go

@@ -20,10 +20,7 @@ import (
 	pb "github.com/evstack/ev-node/types/pb/evnode/v1"
 )
 
-const (
-	dAFetcherTimeout = 30 * time.Second
-	dAFetcherRetries = 10
-)
+const dAFetcherTimeout = 10 * time.Second
 
 // DARetriever handles DA retrieval operations for syncing
 type DARetriever struct {
@@ -70,49 +67,34 @@ func NewDARetriever(
 // RetrieveFromDA retrieves blocks from the specified DA height and returns height events
 func (r *DARetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
 	r.logger.Debug().Uint64("da_height", daHeight).Msg("retrieving from DA")
+	ctx, cancel := context.WithTimeout(ctx, dAFetcherTimeout)
+	defer cancel()
 
-	var err error
-	for retry := 0; retry < dAFetcherRetries; retry++ {
-		select {
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		default:
-		}
-
-		blobsResp, fetchErr := r.fetchBlobs(ctx, daHeight)
-		if fetchErr == nil {
-			if blobsResp.Code == coreda.StatusNotFound {
-				r.logger.Debug().Uint64("da_height", daHeight).Msg("no blob data found")
-				return nil, coreda.ErrBlobNotFound
-			}
-
-			r.logger.Debug().Int("blobs", len(blobsResp.Data)).Uint64("da_height", daHeight).Msg("retrieved blob data")
-			events := r.processBlobs(ctx, blobsResp.Data, daHeight)
-			return events, nil
-		}
-
-		if strings.Contains(fetchErr.Error(), coreda.ErrHeightFromFuture.Error()) {
+	blobsResp, err := r.fetchBlobs(ctx, daHeight)
+	if err != nil {
+		if strings.Contains(err.Error(), coreda.ErrHeightFromFuture.Error()) {
 			return nil, fmt.Errorf("%w: height from future", coreda.ErrHeightFromFuture)
 		}
 
-		err = errors.Join(err, fetchErr)
+		return nil, err
+	}
 
-		// Delay before retrying
-		select {
-		case <-ctx.Done():
-			return nil, err
-		case <-time.After(100 * time.Millisecond):
-		}
+	// Check for context cancellation upfront
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+
+	if blobsResp.Code == coreda.StatusNotFound {
+		r.logger.Debug().Uint64("da_height", daHeight).Msg("no blob data found")
+		return nil, coreda.ErrBlobNotFound
 	}
 
-	return nil, err
+	r.logger.Debug().Int("blobs", len(blobsResp.Data)).Uint64("da_height", daHeight).Msg("retrieved blob data")
+	return r.processBlobs(ctx, blobsResp.Data, daHeight), nil
 }
 
 // fetchBlobs retrieves blobs from the DA layer
 func (r *DARetriever) fetchBlobs(ctx context.Context, daHeight uint64) (coreda.ResultRetrieve, error) {
-	ctx, cancel := context.WithTimeout(ctx, dAFetcherTimeout)
-	defer cancel()
-
 	// Retrieve from both namespaces
 	headerRes := types.RetrieveWithHelpers(ctx, r.da, r.logger, daHeight, r.namespaceBz)
 

block/internal/syncing/da_retriever_test.go

@@ -105,6 +105,61 @@ func TestDARetriever_RetrieveFromDA_HeightFromFuture(t *testing.T) {
 	assert.Nil(t, events)
 }
 
+func TestDARetriever_RetrieveFromDA_Timeout(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+
+	mockDA := testmocks.NewMockDA(t)
+
+	// Mock GetIDs to hang longer than the timeout
+	mockDA.EXPECT().GetIDs(mock.Anything, mock.Anything, mock.Anything).
+		Run(func(ctx context.Context, height uint64, namespace []byte) {
+			<-ctx.Done()
+		}).
+		Return(nil, context.DeadlineExceeded).Maybe()
+
+	r := NewDARetriever(mockDA, cm, config.DefaultConfig(), genesis.Genesis{}, common.DefaultBlockOptions(), zerolog.Nop())
+
+	start := time.Now()
+	events, err := r.RetrieveFromDA(context.Background(), 42)
+	duration := time.Since(start)
+
+	// Verify error is returned and contains deadline exceeded information
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "DA retrieval failed")
+	assert.Contains(t, err.Error(), "context deadline exceeded")
+	assert.Len(t, events, 0)
+
+	// Verify timeout occurred approximately at expected time (with some tolerance)
+	assert.Greater(t, duration, 9*time.Second, "should timeout after approximately 10 seconds")
+	assert.Less(t, duration, 12*time.Second, "should not take much longer than timeout")
+}
+
+func TestDARetriever_RetrieveFromDA_TimeoutFast(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+
+	mockDA := testmocks.NewMockDA(t)
+
+	// Mock GetIDs to immediately return context deadline exceeded
+	mockDA.EXPECT().GetIDs(mock.Anything, mock.Anything, mock.Anything).
+		Return(nil, context.DeadlineExceeded).Maybe()
+
+	r := NewDARetriever(mockDA, cm, config.DefaultConfig(), genesis.Genesis{}, common.DefaultBlockOptions(), zerolog.Nop())
+
+	events, err := r.RetrieveFromDA(context.Background(), 42)
+
+	// Verify error is returned and contains deadline exceeded information
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "DA retrieval failed")
+	assert.Contains(t, err.Error(), "context deadline exceeded")
+	assert.Len(t, events, 0)
+}
+
 func TestDARetriever_ProcessBlobs_HeaderAndData_Success(t *testing.T) {
 	ds := dssync.MutexWrap(datastore.NewMapDatastore())
 	st := store.New(ds)

block/internal/syncing/syncer.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	goheader "github.com/celestiaorg/go-header"
@@ -30,6 +31,9 @@ type p2pHandler interface {
 	ProcessDataRange(ctx context.Context, fromHeight, toHeight uint64) []common.DAHeightEvent
 }
 
+// maxRetriesBeforeHalt is the maximum number of retries against the execution client before halting the syncer.
+const maxRetriesBeforeHalt = 3
+
 // Syncer handles block synchronization from DA and P2P sources.
 type Syncer struct {
 	// Core components
@@ -51,8 +55,7 @@ type Syncer struct {
 	lastStateMtx *sync.RWMutex
 
 	// DA state
-	daHeight   uint64
-	daStateMtx *sync.RWMutex
+	daHeight uint64
 
 	// P2P stores
 	headerStore goheader.Store[*types.SignedHeader]
@@ -70,9 +73,10 @@ type Syncer struct {
 	logger zerolog.Logger
 
 	// Lifecycle
-	ctx    context.Context
-	cancel context.CancelFunc
-	wg     sync.WaitGroup
+	ctx               context.Context
+	cancel            context.CancelFunc
+	wg                sync.WaitGroup
+	retriesBeforeHalt map[uint64]uint64
 }
 
 // NewSyncer creates a new block syncer
@@ -91,21 +95,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, 10_000),
-		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{},
+		heightInCh:        make(chan common.DAHeightEvent, 10_000),
+		errorCh:           errorCh,
+		logger:            logger.With().Str("component", "syncer").Logger(),
+		retriesBeforeHalt: make(map[uint64]uint64),
 	}
 }
 
@@ -166,16 +170,12 @@ func (s *Syncer) SetLastState(state types.State) {
 
 // GetDAHeight returns the current DA height
 func (s *Syncer) GetDAHeight() uint64 {
-	s.daStateMtx.RLock()
-	defer s.daStateMtx.RUnlock()
-	return s.daHeight
+	return atomic.LoadUint64(&s.daHeight)
 }
 
 // SetDAHeight updates the DA height
 func (s *Syncer) SetDAHeight(height uint64) {
-	s.daStateMtx.Lock()
-	defer s.daStateMtx.Unlock()
-	s.daHeight = height
+	atomic.StoreUint64(&s.daHeight, height)
 }
 
 // initializeState loads the current sync state
@@ -246,15 +246,13 @@ func (s *Syncer) syncLoop() {
 	lastHeaderHeight := initialHeight
 	lastDataHeight := initialHeight
 
-	// Backoff control when DA replies with height-from-future
+	// Backoff control when DA replies with errors
 	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
 	for {
 		select {
 		case <-s.ctx.Done():
@@ -265,26 +263,34 @@ func (s *Syncer) syncLoop() {
 		s.processPendingEvents()
 
 		now := time.Now()
+		daHeight := s.GetDAHeight()
+
 		// Respect backoff window if set
 		if nextDARequestAt.IsZero() || now.After(nextDARequestAt) || now.Equal(nextDARequestAt) {
 			// Retrieve from DA as fast as possible (unless throttled by HFF)
-			events, err := s.daRetriever.RetrieveFromDA(s.ctx, s.GetDAHeight())
+			// DaHeight is only increased on successful retrieval, it will retry on failure at the next iteration
+			events, err := s.daRetriever.RetrieveFromDA(s.ctx, daHeight)
 			if err != nil {
-				if s.isHeightFromFutureError(err) {
+				if errors.Is(err, coreda.ErrBlobNotFound) {
+					// no data at this height, increase DA height
+					// we do still want to check p2p
+					s.SetDAHeight(daHeight + 1)
+
+					// Reset backoff on success
+					nextDARequestAt = time.Time{}
+				} else {
 					// Back off exactly by DA block time to avoid overloading
 					hffDelay = s.config.DA.BlockTime.Duration
 					if hffDelay <= 0 {
 						hffDelay = 2 * time.Second
 					}
-					s.logger.Debug().Dur("delay", hffDelay).Uint64("da_height", s.GetDAHeight()).Msg("height from future; backing off DA requests")
 					nextDARequestAt = now.Add(hffDelay)
-				} else if errors.Is(err, coreda.ErrBlobNotFound) {
-					// no data at this height, increase DA height
-					s.SetDAHeight(s.GetDAHeight() + 1)
-				} else {
-					// Non-HFF errors: do not backoff artificially
-					nextDARequestAt = time.Time{}
-					s.logger.Error().Err(err).Msg("failed to retrieve from DA")
+
+					if s.isHeightFromFutureError(err) {
+						s.logger.Debug().Dur("delay", hffDelay).Uint64("da_height", daHeight).Msg("height from future; backing off DA requests")
+					} else {
+						s.logger.Error().Err(err).Dur("delay", hffDelay).Uint64("da_height", daHeight).Msg("failed to retrieve from DA; backing off DA requests")
+					}
 				}
 			} else {
 				// Reset backoff on success
@@ -300,8 +306,8 @@ func (s *Syncer) syncLoop() {
 				}
 
 				// increment DA height on successful retrieval and continue immediately
-				s.SetDAHeight(s.GetDAHeight() + 1)
-				continue
+				s.SetDAHeight(daHeight + 1)
+				continue // event sent, no need to check p2p
 			}
 		}
 
@@ -469,9 +475,15 @@ func (s *Syncer) applyBlock(header types.Header, data *types.Data, currentState
 	newAppHash, _, err := s.exec.ExecuteTxs(ctx, rawTxs, header.Height(),
 		header.Time(), currentState.AppHash)
 	if err != nil {
-		s.sendCriticalError(fmt.Errorf("failed to execute transactions: %w", err))
-		return types.State{}, fmt.Errorf("failed to execute transactions: %w", err)
+		s.retriesBeforeHalt[header.Height()]++
+		if s.retriesBeforeHalt[header.Height()] > maxRetriesBeforeHalt {
+			s.sendCriticalError(fmt.Errorf("failed to execute transactions: %w", err))
+			return types.State{}, fmt.Errorf("failed to execute transactions: %w", err)
+		}
+
+		return types.State{}, fmt.Errorf("failed to execute transactions (retry %d / %d): %w", s.retriesBeforeHalt[header.Height()], maxRetriesBeforeHalt, err)
 	}
+	delete(s.retriesBeforeHalt, header.Height())
 
 	// Create new state
 	newState, err := currentState.NextState(header, newAppHash)