fix wrong epoch check and improve time check

Author: julienrbrt

block/internal/syncing/syncer.go

@@ -753,7 +753,7 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 	// P2P nodes aren't actually able to verify forced inclusions txs as DA inclusion happens later (so DA hints are not available) and DA hints cannot be trusted. This is a known limitation described in the ADR.
 	// In the future we should verify force inclusion txs completely asynchronously, while still waiting for block n-1 execution.
 	if event.Source == common.SourceDA {
-		if err := s.VerifyForcedInclusionTxs(ctx, currentState.DAHeight, data); err != nil {
+		if err := s.VerifyForcedInclusionTxs(ctx, event.DaHeight, data); err != nil {
 			s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("forced inclusion verification failed")
 			if errors.Is(err, errMaliciousProposer) {
 				// remove header as da included from cache
@@ -772,25 +772,16 @@ func (s *Syncer) TrySyncNextBlock(ctx context.Context, event *common.DAHeightEve
 	}
 
 	// Update DA height if needed
-	// This height is only updated when a height is processed from DA as P2P
-	// events do not contain DA height information.
+	// Update state.DAHeight to track where we are in DA.
+	// state.DAHeight is used for state persistence and restart recovery, not for
+	// forced inclusion verification (which now uses event.DaHeight directly).
 	//
-	// When a sequencer restarts after extended downtime, it produces "catch-up"
-	// blocks containing forced inclusion transactions from missed DA epochs and
-	// submits them to DA at the current (much higher) DA height. This creates a
-	// gap between the state's DAHeight (tracking forced inclusion epoch progress)
-	// and event.DaHeight (the DA submission height).
-	//
-	// If we jump state.DAHeight directly to event.DaHeight, subsequent calls to
-	// VerifyForcedInclusionTxs would check the wrong epoch (the submission epoch
-	// instead of the next forced-inclusion epoch), causing valid catch-up blocks
-	// to be incorrectly flagged as malicious.
-	//
-	// To handle this, when the gap exceeds one DA epoch, we compute the epoch
-	// start that contains event.DaHeight and set DAHeight to that epoch start.
-	// This ensures all blocks from the same DA epoch verify against the same
-	// epoch. Once the sequencer finishes catching up and the gap closes,
-	// DAHeight converges to event.DaHeight.
+	// When a sequencer restarts after extended downtime it submits catch-up blocks
+	// all at the current (much higher) DA height, creating a large gap between
+	// state.DAHeight and event.DaHeight. Jumping straight to event.DaHeight would
+	// cause the next VerifyForcedInclusionTxs call to skip over intermediate epochs.
+	// Instead we advance state.DAHeight to the epoch start that contains
+	// event.DaHeight, so it converges gradually as catch-up completes.
 	if event.DaHeight > newState.DAHeight {
 		epochSize := s.genesis.DAEpochForcedInclusion
 		gap := event.DaHeight - newState.DAHeight

pkg/sequencers/based/sequencer.go

@@ -37,6 +37,8 @@ type BasedSequencer struct {
 	currentBatchTxs [][]byte
 	// DA epoch end time for timestamp calculation
 	currentDAEndTime time.Time
+	// Total number of transactions in the current DA epoch (used for timestamp jitter)
+	currentEpochTxCount uint64
 }
 
 // NewBasedSequencer creates a new based sequencer instance
@@ -113,6 +115,7 @@ func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.Get
 		if daEndTime.After(s.currentDAEndTime) {
 			s.currentDAEndTime = daEndTime
 		}
+		s.currentEpochTxCount = uint64(len(s.currentBatchTxs))
 	}
 
 	// Get remaining transactions from checkpoint position
@@ -156,9 +159,13 @@ func (s *BasedSequencer) GetNextBatch(ctx context.Context, req coresequencer.Get
 	}
 doneProcessing:
 
-	// Update checkpoint based on consumed transactions
+	// Update checkpoint based on consumed transactions.
+	// txIndexForTimestamp is captured before the epoch-boundary reset so the
+	// final block of an epoch lands exactly on daEndTime.
+	var txIndexForTimestamp uint64
 	if daHeight > 0 || len(batchTxs) > 0 {
 		s.checkpoint.TxIndex += consumedCount
+		txIndexForTimestamp = s.checkpoint.TxIndex
 
 		// If we've consumed all transactions from this DA epoch, move to next DA epoch
 		if s.checkpoint.TxIndex >= uint64(len(s.currentBatchTxs)) {
@@ -179,14 +186,13 @@ doneProcessing:
 			Msg("updated checkpoint after processing batch")
 	}
 
-	// Calculate timestamp based on remaining transactions after this batch
-	// timestamp corresponds to the last block time of a DA epoch, based on the remaining transactions to be executed
-	// this is done in order to handle the case where a DA epoch must fit in multiple blocks
-	var remainingTxs uint64
-	if len(s.currentBatchTxs) > 0 {
-		remainingTxs = uint64(len(s.currentBatchTxs)) - s.checkpoint.TxIndex
-	}
-	timestamp := s.currentDAEndTime.Add(-time.Duration(remainingTxs) * time.Millisecond)
+	// Spread blocks across the DA epoch window to produce monotonically increasing timestamps:
+	//   epochStart     = daEndTime - totalEpochTxs * 1ms
+	//   blockTimestamp = epochStart + txIndexForTimestamp * 1ms
+	// The last block of an epoch lands exactly on daEndTime; the first block of
+	// the next epoch starts at nextDaEndTime - N*1ms >= prevDaEndTime.
+	epochStart := s.currentDAEndTime.Add(-time.Duration(s.currentEpochTxCount) * time.Millisecond)
+	timestamp := epochStart.Add(time.Duration(txIndexForTimestamp) * time.Millisecond)
 
 	return &coresequencer.GetNextBatchResponse{
 		Batch:     &coresequencer.Batch{Transactions: validTxs},

pkg/sequencers/based/sequencer_test.go

@@ -693,8 +693,10 @@ func TestBasedSequencer_GetNextBatch_EmptyDABatch_IncreasesDAHeight(t *testing.T
 }
 
 func TestBasedSequencer_GetNextBatch_TimestampAdjustment(t *testing.T) {
-	// Test that timestamp is adjusted based on the number of transactions in the batch
-	// The timestamp should be: daEndTime - (len(batch.Transactions) * 1ms)
+	// Test that timestamp is adjusted based on the position within the DA epoch.
+	// The formula is: epochStart + txIndexForTimestamp * 1ms
+	// where epochStart = daEndTime - totalEpochTxs * 1ms
+	// and txIndexForTimestamp is captured before any checkpoint reset.
 
 	testBlobs := [][]byte{[]byte("tx1"), []byte("tx2"), []byte("tx3")}
 	daEndTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
@@ -726,15 +728,18 @@ func TestBasedSequencer_GetNextBatch_TimestampAdjustment(t *testing.T) {
 	require.NotNil(t, resp.Batch)
 	assert.Equal(t, 3, len(resp.Batch.Transactions))
 
-	// After taking all 3 txs, there are 0 remaining, so timestamp = daEndTime - 0ms = daEndTime
+	// epochStart = T - 3ms; all 3 txs consumed → txIndexForTimestamp=3 (pre-reset)
+	// timestamp = T - 3ms + 3ms = daEndTime
 	expectedTimestamp := daEndTime
 	assert.Equal(t, expectedTimestamp, resp.Timestamp)
 
 	mockRetriever.AssertExpectations(t)
 }
 
 func TestBasedSequencer_GetNextBatch_TimestampAdjustment_PartialBatch(t *testing.T) {
-	// Test timestamp adjustment when MaxBytes limits the batch size
+	// Test timestamp adjustment when filtering limits the batch size.
+	// Formula: epochStart + txIndexForTimestamp * 1ms
+	// where epochStart = daEndTime - totalEpochTxs * 1ms
 	tx1 := make([]byte, 100)
 	tx2 := make([]byte, 150)
 	tx3 := make([]byte, 200)
@@ -769,7 +774,8 @@ func TestBasedSequencer_GetNextBatch_TimestampAdjustment_PartialBatch(t *testing
 	require.NotNil(t, resp.Batch)
 	assert.Equal(t, 2, len(resp.Batch.Transactions))
 
-	// After taking 2 txs, there is 1 remaining, so timestamp = daEndTime - 1ms
+	// epochStart = T - 3ms; 2 txs consumed → txIndexForTimestamp=2
+	// timestamp = T - 3ms + 2ms = daEndTime - 1ms
 	expectedTimestamp := daEndTime.Add(-1 * time.Millisecond)
 	assert.Equal(t, expectedTimestamp, resp.Timestamp)
 
@@ -786,7 +792,8 @@ func TestBasedSequencer_GetNextBatch_TimestampAdjustment_PartialBatch(t *testing
 	require.NotNil(t, resp.Batch)
 	assert.Equal(t, 1, len(resp.Batch.Transactions))
 
-	// After taking this 1 tx, there are 0 remaining, so timestamp = daEndTime - 0ms = daEndTime
+	// epochStart = T - 3ms; 3 txs consumed total → txIndexForTimestamp=3 (pre-reset)
+	// timestamp = T - 3ms + 3ms = daEndTime
 	expectedTimestamp2 := daEndTime
 	assert.Equal(t, expectedTimestamp2, resp.Timestamp)
 

pkg/sequencers/single/sequencer.go

@@ -64,6 +64,8 @@ type Sequencer struct {
 	catchUpState atomic.Int32
 	// currentDAEndTime is the DA epoch end timestamp, used during catch-up
 	currentDAEndTime time.Time
+	// currentEpochTxCount is the total number of txs in the current DA epoch (used for timestamp jitter)
+	currentEpochTxCount uint64
 }
 
 // NewSequencer creates a new Single Sequencer
@@ -311,10 +313,14 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 		}
 	}
 
-	// Update checkpoint after consuming forced inclusion transactions
+	// Update checkpoint after consuming forced inclusion transactions.
+	// txIndexForTimestamp is captured before the epoch-boundary reset so the
+	// final block of an epoch lands exactly on daEndTime.
+	var txIndexForTimestamp uint64
 	if daHeight > 0 || len(forcedTxs) > 0 {
 		// Advance TxIndex by the number of consumed forced transactions
 		c.checkpoint.TxIndex += forcedTxConsumedCount
+		txIndexForTimestamp = c.checkpoint.TxIndex
 
 		if c.checkpoint.TxIndex >= uint64(len(c.cachedForcedInclusionTxs)) {
 			// All forced txs were consumed (OK or Remove), move to next DA epoch
@@ -341,14 +347,16 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 	batchTxs = append(batchTxs, validForcedTxs...)
 	batchTxs = append(batchTxs, validMempoolTxs...)
 
-	// Use DA epoch timestamp during catch-up
+	// Spread catch-up blocks across the DA epoch window for monotonically increasing timestamps:
+	//   epochStart     = daEndTime - totalEpochTxs * 1ms
+	//   blockTimestamp = epochStart + txIndexForTimestamp * 1ms
+	// The last block of an epoch lands exactly on daEndTime; the first block of
+	// the next epoch starts at nextDaEndTime - N*1ms >= prevDaEndTime.
+	// During normal operation, use wall-clock time instead.
 	timestamp := time.Now()
 	if c.catchUpState.Load() == catchUpInProgress && !c.currentDAEndTime.IsZero() {
-		var remainingForcedTxs uint64
-		if len(c.cachedForcedInclusionTxs) > 0 {
-			remainingForcedTxs = uint64(len(c.cachedForcedInclusionTxs)) - c.checkpoint.TxIndex
-		}
-		timestamp = c.currentDAEndTime.Add(-time.Duration(remainingForcedTxs) * time.Millisecond)
+		epochStart := c.currentDAEndTime.Add(-time.Duration(c.currentEpochTxCount) * time.Millisecond)
+		timestamp = epochStart.Add(time.Duration(txIndexForTimestamp) * time.Millisecond)
 	}
 
 	return &coresequencer.GetNextBatchResponse{
@@ -447,6 +455,8 @@ func (c *Sequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64) (uint
 	if !forcedTxsEvent.Timestamp.IsZero() {
 		c.currentDAEndTime = forcedTxsEvent.Timestamp.UTC()
 	}
+	// Record total tx count for the epoch so the timestamp jitter can be computed
+	// after oversized txs are filtered out below.
 
 	// Filter out oversized transactions
 	validTxs := make([][]byte, 0, len(forcedTxsEvent.Txs))
@@ -473,6 +483,7 @@ func (c *Sequencer) fetchNextDAEpoch(ctx context.Context, maxBytes uint64) (uint
 		Msg("fetched forced inclusion transactions from DA")
 
 	c.cachedForcedInclusionTxs = validTxs
+	c.currentEpochTxCount = uint64(len(validTxs))
 
 	return forcedTxsEvent.EndDaHeight, nil
 }

pkg/sequencers/single/sequencer_test.go

@@ -1912,8 +1912,12 @@ func TestSequencer_CatchUp_CheckpointAdvancesDuringCatchUp(t *testing.T) {
 func TestSequencer_CatchUp_MonotonicTimestamps(t *testing.T) {
 	// When a single DA epoch has more forced txs than fit in one block,
 	// catch-up must produce strictly monotonic timestamps across the
-	// resulting blocks.  This uses the same jitter scheme as the based
-	// sequencer: timestamp = DAEndTime - (remainingForcedTxs * 1ms).
+	// resulting blocks.  The jitter scheme is:
+	//   epochStart     = daEndTime - totalEpochTxs * 1ms
+	//   blockTimestamp = epochStart + txIndexForTimestamp * 1ms
+	// where txIndexForTimestamp is the cumulative consumed-tx count
+	// captured *before* the checkpoint resets at an epoch boundary.
+	// The final block of an epoch therefore lands exactly on daEndTime.
 	ctx := context.Background()
 	logger := zerolog.New(zerolog.NewTestWriter(t))
 
@@ -2016,10 +2020,11 @@ func TestSequencer_CatchUp_MonotonicTimestamps(t *testing.T) {
 			i, timestamps[i], i-1, timestamps[i-1])
 	}
 
-	// Verify exact jitter values:
-	//   Block 0: 3 txs total, 1 consumed → 2 remaining → T - 2ms
-	//   Block 1: 1 consumed → 1 remaining → T - 1ms
-	//   Block 2: 1 consumed → 0 remaining → T
+	// Verify exact jitter values using epochStart + txIndexForTimestamp formula:
+	//   epochStart = T - 3ms  (3 total txs in epoch)
+	//   Block 0: 1 consumed → txIndex=1 → epochStart + 1ms = T - 2ms
+	//   Block 1: 1 consumed → txIndex=2 → epochStart + 2ms = T - 1ms
+	//   Block 2: 1 consumed → txIndex=3 (pre-reset) → epochStart + 3ms = T
 	assert.Equal(t, epochTimestamp.Add(-2*time.Millisecond), timestamps[0], "block 0: T - 2ms")
 	assert.Equal(t, epochTimestamp.Add(-1*time.Millisecond), timestamps[1], "block 1: T - 1ms")
 	assert.Equal(t, epochTimestamp, timestamps[2], "block 2: T (exact epoch end time)")
@@ -2032,6 +2037,7 @@ func TestSequencer_CatchUp_MonotonicTimestamps(t *testing.T) {
 	assert.True(t, resp4.Timestamp.After(timestamps[2]),
 		"epoch 101 timestamp (%v) must be after epoch 100 last timestamp (%v)",
 		resp4.Timestamp, timestamps[2])
+	// epoch 101 has 1 tx: epochStart = T2 - 1ms, txIndexForTimestamp=1 → T2 - 1ms + 1ms = T2
 	assert.Equal(t, epoch2Timestamp, resp4.Timestamp, "single-tx epoch gets exact DA end time")
 }