refactor(block): add retries on SetFinal

Author: julienrbrt

apps/evm/single/cmd/rollback.go

@@ -93,12 +93,12 @@ func NewRollbackCmd() *cobra.Command {
 			}
 
 			if err := headerStore.Start(goCtx); err != nil {
-				return err
+				return fmt.Errorf("failed to start header store: %w", err)
 			}
 			defer headerStore.Stop(goCtx)
 
 			if err := dataStore.Start(goCtx); err != nil {
-				return err
+				return fmt.Errorf("failed to start data store: %w", err)
 			}
 			defer dataStore.Stop(goCtx)
 

block/internal/common/types.go block/internal/common/event.goblock/internal/common/types.go renamed to block/internal/common/event.go

@@ -0,0 +1,9 @@
+package common
+
+import "time"
+
+// MaxRetriesBeforeHalt is the maximum number of retries before halting.
+const MaxRetriesBeforeHalt = 3
+
+// MaxRetriesTimeout is the maximum time to wait before halting.
+const MaxRetriesTimeout = 10 * time.Second

block/internal/common/retry.go

@@ -59,9 +59,10 @@ type Submitter 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
 }
 
 // NewSubmitter creates a new DA submitter component
@@ -78,17 +79,18 @@ func NewSubmitter(
 	errorCh chan<- error,
 ) *Submitter {
 	return &Submitter{
-		store:       store,
-		exec:        exec,
-		cache:       cache,
-		metrics:     metrics,
-		config:      config,
-		genesis:     genesis,
-		daSubmitter: daSubmitter,
-		signer:      signer,
-		daStateMtx:  &sync.RWMutex{},
-		errorCh:     errorCh,
-		logger:      logger.With().Str("component", "submitter").Logger(),
+		store:             store,
+		exec:              exec,
+		cache:             cache,
+		metrics:           metrics,
+		config:            config,
+		genesis:           genesis,
+		daSubmitter:       daSubmitter,
+		signer:            signer,
+		daStateMtx:        &sync.RWMutex{},
+		errorCh:           errorCh,
+		logger:            logger.With().Str("component", "submitter").Logger(),
+		retriesBeforeHalt: make(map[uint64]uint64),
 	}
 }
 
@@ -217,12 +219,21 @@ func (s *Submitter) processDAInclusionLoop() {
 					break
 				}
 
+			retry:
 				// Set final height in executor
 				if err := s.exec.SetFinal(s.ctx, nextHeight); err != nil {
-					s.sendCriticalError(fmt.Errorf("failed to set final height: %w", err))
-					s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set final height")
-					break
+					s.retriesBeforeHalt[header.Height()]++
+					if s.retriesBeforeHalt[header.Height()] > common.MaxRetriesBeforeHalt {
+						s.sendCriticalError(fmt.Errorf("failed to set final height: %w", err))
+						s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("failed to set final height")
+						return
+					}
+
+					time.Sleep(common.MaxRetriesTimeout) // sleep before retrying
+					s.logger.Error().Err(err).Msgf("failed to set final height (retry %d / %d): %w", s.retriesBeforeHalt[header.Height()], common.MaxRetriesBeforeHalt, err)
+					goto retry
 				}
+				delete(s.retriesBeforeHalt, header.Height())
 
 				// Update DA included height
 				s.SetDAIncludedHeight(nextHeight)

block/internal/submitting/submitter.go

@@ -30,13 +30,6 @@ type p2pHandler interface {
 	ProcessDataRange(ctx context.Context, fromHeight, toHeight uint64) []common.DAHeightEvent
 }
 
-const (
-	// maxRetriesBeforeHalt is the maximum number of retries against the execution client before halting the syncer.
-	maxRetriesBeforeHalt = 3
-	// maxRetriesTimeout is the maximum time to wait for a retry before halting the syncer.
-	maxRetriesTimeout = time.Second * 10
-)
-
 // Syncer handles block synchronization from DA and P2P sources.
 type Syncer struct {
 	// Core components
@@ -493,13 +486,13 @@ func (s *Syncer) applyBlock(header types.Header, data *types.Data, currentState
 		header.Time(), currentState.AppHash)
 	if err != nil {
 		s.retriesBeforeHalt[header.Height()]++
-		if s.retriesBeforeHalt[header.Height()] > maxRetriesBeforeHalt {
+		if s.retriesBeforeHalt[header.Height()] > common.MaxRetriesBeforeHalt {
 			s.sendCriticalError(fmt.Errorf("failed to execute transactions: %w", err))
 			return types.State{}, fmt.Errorf("failed to execute transactions: %w", err)
 		}
 
-		time.Sleep(maxRetriesTimeout) // sleep before retrying
-		return types.State{}, fmt.Errorf("failed to execute transactions (retry %d / %d): %w", s.retriesBeforeHalt[header.Height()], maxRetriesBeforeHalt, err)
+		time.Sleep(common.MaxRetriesTimeout) // sleep before retrying
+		return types.State{}, fmt.Errorf("failed to execute transactions (retry %d / %d): %w", s.retriesBeforeHalt[header.Height()], common.MaxRetriesBeforeHalt, err)
 	}
 	delete(s.retriesBeforeHalt, header.Height())