implement checkpoint for single sequencer

Author: julienrbrt

apps/evm/cmd/run.go

@@ -169,7 +169,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, da, datastore, genesis, logger)
+		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

apps/grpc/cmd/run.go

@@ -131,7 +131,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, da, datastore, genesis, logger)
+		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

apps/testapp/cmd/run.go

@@ -131,7 +131,7 @@ func createSequencer(
 			return nil, fmt.Errorf("based sequencer mode requires aggregator mode to be enabled")
 		}
 
-		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, da, datastore, genesis, logger)
+		basedSeq, err := based.NewBasedSequencer(ctx, fiRetriever, datastore, genesis, logger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create based sequencer: %w", err)
 		}

sequencers/based/sequencer.go

@@ -29,8 +29,6 @@ var _ coresequencer.Sequencer = (*BasedSequencer)(nil)
 // It uses DA as a queue and only persists a checkpoint of where it is in processing.
 type BasedSequencer struct {
 	fiRetriever ForcedInclusionRetriever
-	da          coreda.DA
-	genesis     genesis.Genesis
 	logger      zerolog.Logger
 
 	daHeight        atomic.Uint64
@@ -45,15 +43,12 @@ type BasedSequencer struct {
 func NewBasedSequencer(
 	ctx context.Context,
 	fiRetriever ForcedInclusionRetriever,
-	da coreda.DA,
 	db ds.Batching,
 	genesis genesis.Genesis,
 	logger zerolog.Logger,
 ) (*BasedSequencer, error) {
 	bs := &BasedSequencer{
 		fiRetriever:     fiRetriever,
-		da:              da,
-		genesis:         genesis,
 		logger:          logger.With().Str("component", "based_sequencer").Logger(),
 		checkpointStore: seqcommon.NewCheckpointStore(db, ds.NewKey("/based/checkpoint")),
 	}
@@ -96,11 +91,15 @@ func (s *BasedSequencer) SubmitBatchTxs(ctx context.Context, req coresequencer.S
 // It treats DA as a queue and only persists where it is in processing
 func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextBatchRequest) (*coresequencer.GetNextBatchResponse, error) {
 	// If we have no cached transactions or we've consumed all from the current DA block,
-	// fetch the next DA block
+	// fetch the next DA epoch
+	daHeight := s.GetDAHeight()
 	if len(s.currentBatchTxs) == 0 || s.checkpoint.TxIndex >= uint64(len(s.currentBatchTxs)) {
-		if err := s.fetchNextDABatch(ctx); err != nil {
+		daEndHeight, err := s.fetchNextDAEpoch(ctx)
+		if err != nil {
 			return nil, err
 		}
+
+		daHeight = daEndHeight
 	}
 
 	// Create batch from current position up to MaxBytes
@@ -113,7 +112,7 @@ func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.Get
 
 		// If we've consumed all transactions from this DA block, move to next
 		if s.checkpoint.TxIndex >= uint64(len(s.currentBatchTxs)) {
-			s.checkpoint.DAHeight++
+			s.checkpoint.DAHeight = daHeight + 1
 			s.checkpoint.TxIndex = 0
 			s.currentBatchTxs = nil
 
@@ -135,8 +134,8 @@ func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.Get
 	}, nil
 }
 
-// fetchNextDABatch fetches transactions from the next DA block
-func (s *BasedSequencer) fetchNextDABatch(ctx context.Context) error {
+// fetchNextDAEpoch fetches transactions from the next DA epoch
+func (s *BasedSequencer) fetchNextDAEpoch(ctx context.Context) (uint64, error) {
 	currentDAHeight := s.checkpoint.DAHeight
 
 	s.logger.Debug().
@@ -148,17 +147,17 @@ func (s *BasedSequencer) fetchNextDABatch(ctx context.Context) error {
 	if err != nil {
 		// Check if forced inclusion is not configured
 		if errors.Is(err, block.ErrForceInclusionNotConfigured) {
-			return errors.New("forced inclusion not configured")
+			return currentDAHeight, block.ErrForceInclusionNotConfigured
 		} else if errors.Is(err, coreda.ErrHeightFromFuture) {
 			// If we get a height from future error, stay at current position
 			// We'll retry the same height on the next call until DA produces that block
 			s.logger.Debug().
 				Uint64("da_height", currentDAHeight).
 				Msg("DA height from future, waiting for DA to produce block")
-			return nil
+			return currentDAHeight, nil
 		}
 		s.logger.Error().Err(err).Uint64("da_height", currentDAHeight).Msg("failed to retrieve forced inclusion transactions")
-		return err
+		return currentDAHeight, err
 	}
 
 	// Validate and filter transactions
@@ -184,18 +183,18 @@ func (s *BasedSequencer) fetchNextDABatch(ctx context.Context) error {
 		Uint64("da_height_end", forcedTxsEvent.EndDaHeight).
 		Msg("fetched forced inclusion transactions from DA")
 
-	// Cache the transactions for this DA block
+	// Cache the transactions for this DA epoch
 	s.currentBatchTxs = validTxs
 
 	// If we had a non-zero tx index, we're resuming from a crash mid-block
 	// The transactions starting from that index are what we need
 	if s.checkpoint.TxIndex > 0 {
 		s.logger.Info().
 			Uint64("tx_index", s.checkpoint.TxIndex).
-			Msg("resuming from checkpoint within DA block")
+			Msg("resuming from checkpoint within DA epoch")
 	}
 
-	return nil
+	return forcedTxsEvent.EndDaHeight, nil
 }
 
 // createBatchFromCheckpoint creates a batch from the current checkpoint position respecting MaxBytes

sequencers/based/sequencer_test.go

@@ -90,7 +90,7 @@ func createTestSequencer(t *testing.T, mockDA *MockDA, cfg config.Config, gen ge
 	// Create in-memory datastore
 	db := syncds.MutexWrap(ds.NewMapDatastore())
 
-	seq, err := NewBasedSequencer(context.Background(), fiRetriever, mockDA, db, gen, zerolog.Nop())
+	seq, err := NewBasedSequencer(context.Background(), fiRetriever, db, gen, zerolog.Nop())
 	require.NoError(t, err)
 	return seq
 }
@@ -223,7 +223,7 @@ func TestBasedSequencer_GetNextBatch_NotConfigured(t *testing.T) {
 	// Create in-memory datastore
 	db := syncds.MutexWrap(ds.NewMapDatastore())
 
-	seq, err := NewBasedSequencer(context.Background(), fiRetriever, mockDA, db, gen, zerolog.Nop())
+	seq, err := NewBasedSequencer(context.Background(), fiRetriever, db, gen, zerolog.Nop())
 	require.NoError(t, err)
 
 	req := coresequencer.GetNextBatchRequest{
@@ -580,7 +580,7 @@ func TestBasedSequencer_CheckpointPersistence(t *testing.T) {
 	daClient := block.NewDAClient(mockDA, cfg, zerolog.Nop())
 	fiRetriever := block.NewForcedInclusionRetriever(daClient, gen, zerolog.Nop())
 
-	seq1, err := NewBasedSequencer(context.Background(), fiRetriever, mockDA, db, gen, zerolog.Nop())
+	seq1, err := NewBasedSequencer(context.Background(), fiRetriever, db, gen, zerolog.Nop())
 	require.NoError(t, err)
 
 	req := coresequencer.GetNextBatchRequest{
@@ -595,7 +595,7 @@ func TestBasedSequencer_CheckpointPersistence(t *testing.T) {
 	assert.Equal(t, 2, len(resp.Batch.Transactions))
 
 	// Create a new sequencer with the same datastore (simulating restart)
-	seq2, err := NewBasedSequencer(context.Background(), fiRetriever, mockDA, db, gen, zerolog.Nop())
+	seq2, err := NewBasedSequencer(context.Background(), fiRetriever, db, gen, zerolog.Nop())
 	require.NoError(t, err)
 
 	// Checkpoint should be loaded from DB

sequencers/common/size_validation.go

@@ -12,16 +12,5 @@ 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(GetBlobSize(blob)) <= AbsoluteMaxBlobSize
-}
-
-// WouldExceedCumulativeSize checks if adding a blob would exceed the cumulative size limit for a batch.
-// Returns true if adding the blob would exceed the limit, false otherwise.
-func WouldExceedCumulativeSize(currentSize int, blobSize int, maxBytes uint64) bool {
-	return uint64(currentSize)+uint64(blobSize) > maxBytes
-}
-
-// GetBlobSize returns the size of a blob in bytes.
-func GetBlobSize(blob []byte) int {
-	return len(blob)
+	return uint64(len(blob)) <= AbsoluteMaxBlobSize
 }

sequencers/common/size_validation_test.go

@@ -47,95 +47,3 @@ func TestValidateBlobSize(t *testing.T) {
 		})
 	}
 }
-
-func TestWouldExceedCumulativeSize(t *testing.T) {
-	tests := []struct {
-		name        string
-		currentSize int
-		blobSize    int
-		maxBytes    uint64
-		want        bool
-	}{
-		{
-			name:        "empty batch, small blob",
-			currentSize: 0,
-			blobSize:    50,
-			maxBytes:    100,
-			want:        false,
-		},
-		{
-			name:        "would fit exactly",
-			currentSize: 50,
-			blobSize:    50,
-			maxBytes:    100,
-			want:        false,
-		},
-		{
-			name:        "would exceed by one byte",
-			currentSize: 50,
-			blobSize:    51,
-			maxBytes:    100,
-			want:        true,
-		},
-		{
-			name:        "far exceeds",
-			currentSize: 80,
-			blobSize:    100,
-			maxBytes:    100,
-			want:        true,
-		},
-		{
-			name:        "zero max bytes",
-			currentSize: 0,
-			blobSize:    1,
-			maxBytes:    0,
-			want:        true,
-		},
-		{
-			name:        "current already at limit",
-			currentSize: 100,
-			blobSize:    1,
-			maxBytes:    100,
-			want:        true,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := WouldExceedCumulativeSize(tt.currentSize, tt.blobSize, tt.maxBytes)
-			assert.Equal(t, tt.want, got)
-		})
-	}
-}
-
-func TestGetBlobSize(t *testing.T) {
-	tests := []struct {
-		name     string
-		blobSize int
-		want     int
-	}{
-		{
-			name:     "empty blob",
-			blobSize: 0,
-			want:     0,
-		},
-		{
-			name:     "small blob",
-			blobSize: 42,
-			want:     42,
-		},
-		{
-			name:     "large blob",
-			blobSize: 1024 * 1024,
-			want:     1024 * 1024,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			blob := make([]byte, tt.blobSize)
-			got := GetBlobSize(blob)
-			assert.Equal(t, tt.want, got)
-		})
-	}
-}