fix: fix da height bumping and improve retriever

Author: julienrbrt

block/internal/da/forced_inclusion_retriever.go

@@ -53,11 +53,11 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 
 	epochStart, epochEnd, currentEpochNumber := types.CalculateEpochBoundaries(daHeight, r.genesis.DAStartHeight, r.daEpochSize)
 
-	if daHeight != epochStart {
+	if daHeight != epochEnd {
 		r.logger.Debug().
 			Uint64("da_height", daHeight).
-			Uint64("epoch_start", epochStart).
-			Msg("not at epoch start - returning empty transactions")
+			Uint64("epoch_end", epochEnd).
+			Msg("not at epoch end - returning empty transactions")
 
 		return &ForcedInclusionEvent{
 			StartDaHeight: daHeight,
@@ -68,16 +68,10 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 
 	event := &ForcedInclusionEvent{
 		StartDaHeight: epochStart,
+		EndDaHeight:   epochEnd,
 		Txs:           [][]byte{},
 	}
 
-	r.logger.Debug().
-		Uint64("da_height", daHeight).
-		Uint64("epoch_start", epochStart).
-		Uint64("epoch_end", epochEnd).
-		Uint64("epoch_num", currentEpochNumber).
-		Msg("retrieving forced included transactions from DA")
-
 	epochEndResult := r.client.RetrieveForcedInclusion(ctx, epochEnd)
 	if epochEndResult.Code == coreda.StatusHeightFromFuture {
 		r.logger.Debug().
@@ -97,58 +91,59 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 		}
 	}
 
-	lastProcessedHeight := epochStart
+	r.logger.Debug().
+		Uint64("da_height", daHeight).
+		Uint64("epoch_start", epochStart).
+		Uint64("epoch_end", epochEnd).
+		Uint64("epoch_num", currentEpochNumber).
+		Msg("retrieving forced included transactions from DA")
 
-	if err := r.processForcedInclusionBlobs(event, &lastProcessedHeight, epochStartResult, epochStart); err != nil {
-		return nil, err
-	}
+	var processErrs error
+	err := r.processForcedInclusionBlobs(event, epochStartResult, epochStart)
+	processErrs = errors.Join(processErrs, err)
 
 	// Process heights between start and end (exclusive)
 	for epochHeight := epochStart + 1; epochHeight < epochEnd; epochHeight++ {
 		result := r.client.RetrieveForcedInclusion(ctx, epochHeight)
 
-		// If any intermediate height is from future, break early
-		if result.Code == coreda.StatusHeightFromFuture {
-			r.logger.Debug().
-				Uint64("epoch_height", epochHeight).
-				Uint64("last_processed", lastProcessedHeight).
-				Msg("reached future DA height within epoch - stopping")
-			break
-		}
-
-		if err := r.processForcedInclusionBlobs(event, &lastProcessedHeight, result, epochHeight); err != nil {
-			return nil, err
-		}
+		err = r.processForcedInclusionBlobs(event, result, epochHeight)
+		processErrs = errors.Join(processErrs, err)
 	}
 
 	// Process epoch end (only if different from start)
 	if epochEnd != epochStart {
-		if err := r.processForcedInclusionBlobs(event, &lastProcessedHeight, epochEndResult, epochEnd); err != nil {
-			return nil, err
-		}
+		err = r.processForcedInclusionBlobs(event, epochEndResult, epochEnd)
+		processErrs = errors.Join(processErrs, err)
 	}
 
-	event.EndDaHeight = lastProcessedHeight
+	// any error during process, need to retry at next call
+	if processErrs != nil {
+		r.logger.Warn().
+			Uint64("da_height", daHeight).
+			Uint64("epoch_start", epochStart).
+			Uint64("epoch_end", epochEnd).
+			Uint64("epoch_num", currentEpochNumber).
+			Err(processErrs).
+			Msg("Failed to retrieve DA epoch.. retrying next iteration")
 
-	r.logger.Info().
-		Uint64("epoch_start", epochStart).
-		Uint64("epoch_end", lastProcessedHeight).
-		Int("tx_count", len(event.Txs)).
-		Msg("retrieved forced inclusion transactions")
+		return &ForcedInclusionEvent{
+			StartDaHeight: daHeight,
+			EndDaHeight:   daHeight,
+			Txs:           [][]byte{},
+		}, nil
+	}
 
 	return event, nil
 }
 
 // processForcedInclusionBlobs processes blobs from a single DA height for forced inclusion.
 func (r *ForcedInclusionRetriever) processForcedInclusionBlobs(
 	event *ForcedInclusionEvent,
-	lastProcessedHeight *uint64,
 	result coreda.ResultRetrieve,
 	height uint64,
 ) error {
 	if result.Code == coreda.StatusNotFound {
 		r.logger.Debug().Uint64("height", height).Msg("no forced inclusion blobs at height")
-		*lastProcessedHeight = height
 		return nil
 	}
 
@@ -163,8 +158,6 @@ func (r *ForcedInclusionRetriever) processForcedInclusionBlobs(
 		}
 	}
 
-	*lastProcessedHeight = height
-
 	r.logger.Debug().
 		Uint64("height", height).
 		Int("blob_count", len(result.Data)).

sequencers/based/sequencer.go

@@ -67,7 +67,7 @@ func (s *BasedSequencer) SubmitBatchTxs(ctx context.Context, req coresequencer.S
 // GetNextBatch retrieves the next batch of transactions from the DA layer
 // It fetches forced inclusion transactions and returns them as the next batch
 func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextBatchRequest) (*coresequencer.GetNextBatchResponse, error) {
-	currentDAHeight := s.daHeight.Load()
+	currentDAHeight := s.GetDAHeight()
 
 	s.logger.Debug().Uint64("da_height", currentDAHeight).Msg("fetching forced inclusion transactions from DA")
 
@@ -86,13 +86,13 @@ func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.Get
 			s.logger.Error().Err(err).Uint64("da_height", currentDAHeight).Msg("failed to retrieve forced inclusion transactions")
 			return nil, err
 		}
-	}
-
-	// Update DA height based on the retrieved event
-	if forcedTxsEvent.EndDaHeight > currentDAHeight {
-		s.SetDAHeight(forcedTxsEvent.EndDaHeight)
-	} else if forcedTxsEvent.StartDaHeight > currentDAHeight {
-		s.SetDAHeight(forcedTxsEvent.StartDaHeight)
+	} else {
+		// Update DA height.
+		// If we are in between epochs, we still need to bump the da height.
+		// At the end of an epoch, we need to bump to go to the next epoch.
+		if forcedTxsEvent.EndDaHeight >= currentDAHeight {
+			s.SetDAHeight(forcedTxsEvent.EndDaHeight + 1)
+		}
 	}
 
 	// Add forced inclusion transactions to the queue with validation

sequencers/common/size_validation.go

@@ -12,7 +12,7 @@ const (
 // This checks against the DA layer limit, not the per-batch limit.
 // Returns true if the blob is within the absolute size limit, false otherwise.
 func ValidateBlobSize(blob []byte) bool {
-	return uint64(len(blob)) <= AbsoluteMaxBlobSize
+	return uint64(GetBlobSize(blob)) <= AbsoluteMaxBlobSize
 }
 
 // WouldExceedCumulativeSize checks if adding a blob would exceed the cumulative size limit for a batch.

sequencers/single/sequencer.go

@@ -130,9 +130,9 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 		return nil, ErrInvalidId
 	}
 
-	currentDAHeight := c.daHeight.Load()
+	currentDAHeight := c.GetDAHeight()
 
-	forcedEvent, err := c.fiRetriever.RetrieveForcedIncludedTxs(ctx, currentDAHeight)
+	forcedTxsEvent, err := c.fiRetriever.RetrieveForcedIncludedTxs(ctx, currentDAHeight)
 	if err != nil {
 		if errors.Is(err, coreda.ErrHeightFromFuture) {
 			c.logger.Debug().
@@ -143,25 +143,27 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 		}
 
 		// Still create an empty forced inclusion event
-		forcedEvent = &block.ForcedInclusionEvent{
+		forcedTxsEvent = &block.ForcedInclusionEvent{
 			Txs:           [][]byte{},
 			StartDaHeight: currentDAHeight,
 			EndDaHeight:   currentDAHeight,
 		}
+	} else {
+		// Update DA height.
+		// If we are in between epochs, we still need to bump the da height.
+		// At the end of an epoch, we need to bump to go to the next epoch.
+		if forcedTxsEvent.EndDaHeight >= currentDAHeight {
+			c.SetDAHeight(forcedTxsEvent.EndDaHeight + 1)
+		}
 	}
 
 	// Always try to process forced inclusion transactions (including pending from previous epochs)
-	forcedTxs := c.processForcedInclusionTxs(forcedEvent, req.MaxBytes)
-	if forcedEvent.EndDaHeight > currentDAHeight {
-		c.SetDAHeight(forcedEvent.EndDaHeight)
-	} else if forcedEvent.StartDaHeight > currentDAHeight {
-		c.SetDAHeight(forcedEvent.StartDaHeight)
-	}
+	forcedTxs := c.processForcedInclusionTxs(forcedTxsEvent, req.MaxBytes)
 
 	c.logger.Debug().
 		Int("tx_count", len(forcedTxs)).
-		Uint64("da_height_start", forcedEvent.StartDaHeight).
-		Uint64("da_height_end", forcedEvent.EndDaHeight).
+		Uint64("da_height_start", forcedTxsEvent.StartDaHeight).
+		Uint64("da_height_end", forcedTxsEvent.EndDaHeight).
 		Msg("retrieved forced inclusion transactions from DA")
 
 	// Calculate size used by forced inclusion transactions