catch up fixes

Author: julienrbrt

pkg/sequencers/based/sequencer.go

@@ -17,6 +17,7 @@ import (
 	datypes "github.com/evstack/ev-node/pkg/da/types"
 	"github.com/evstack/ev-node/pkg/genesis"
 	seqcommon "github.com/evstack/ev-node/pkg/sequencers/common"
+	"github.com/evstack/ev-node/pkg/store"
 )
 
 var _ coresequencer.Sequencer = (*BasedSequencer)(nil)
@@ -39,6 +40,10 @@ type BasedSequencer struct {
 	currentDAEndTime time.Time
 	// Total number of transactions in the current DA epoch (used for timestamp jitter)
 	currentEpochTxCount uint64
+	// lastTimestamp is the floor for timestamps to guarantee monotonicity
+	// after a restart on a node that already had blocks produced with wall-clock time.
+	// Initialised from the last block time in the store at construction.
+	lastTimestamp time.Time
 }
 
 // NewBasedSequencer creates a new based sequencer instance
@@ -56,6 +61,20 @@ func NewBasedSequencer(
 		executor:         executor,
 		currentDAEndTime: genesis.StartTime,
 	}
+
+	// Read the last block time from the store so that timestamps are
+	// guaranteed to be strictly after any previously produced or synced block.
+	// This handles the case where a node that already had blocks (produced with
+	// wall-clock time) restarts as a based sequencer.
+	initCtx, initCancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer initCancel()
+	s := store.New(store.NewEvNodeKVStore(db))
+	if state, err := s.GetState(initCtx); err == nil && !state.LastBlockTime.IsZero() {
+		bs.lastTimestamp = state.LastBlockTime
+		bs.logger.Debug().
+			Time("last_block_time", state.LastBlockTime).
+			Msg("initialized timestamp floor from last block time")
+	}
 	// based sequencers need community consensus about the da start height given no submission are done
 	bs.SetDAHeight(genesis.DAStartHeight)
 
@@ -194,6 +213,14 @@ doneProcessing:
 	epochStart := s.currentDAEndTime.Add(-time.Duration(s.currentEpochTxCount) * time.Millisecond)
 	timestamp := epochStart.Add(time.Duration(txIndexForTimestamp) * time.Millisecond)
 
+	// Clamp: the DA-derived timestamp may predate blocks that were
+	// produced or synced with wall-clock time before the node restarted
+	// as a based sequencer.  Ensure strict monotonicity.
+	if !s.lastTimestamp.IsZero() && !timestamp.After(s.lastTimestamp) {
+		timestamp = s.lastTimestamp.Add(time.Millisecond)
+	}
+	s.lastTimestamp = timestamp
+
 	return &coresequencer.GetNextBatchResponse{
 		Batch:     &coresequencer.Batch{Transactions: validTxs},
 		Timestamp: timestamp,

pkg/sequencers/single/sequencer.go

@@ -66,6 +66,10 @@ type Sequencer struct {
 	currentDAEndTime time.Time
 	// currentEpochTxCount is the total number of txs in the current DA epoch (used for timestamp jitter)
 	currentEpochTxCount uint64
+	// lastCatchUpTimestamp is the floor for catch-up timestamps to guarantee
+	// monotonicity after a restart.  Initialised from the last block time in
+	// the store when catch-up mode is entered.
+	lastCatchUpTimestamp time.Time
 }
 
 // NewSequencer creates a new Single Sequencer
@@ -357,6 +361,14 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 	if c.catchUpState.Load() == catchUpInProgress && !c.currentDAEndTime.IsZero() {
 		epochStart := c.currentDAEndTime.Add(-time.Duration(c.currentEpochTxCount) * time.Millisecond)
 		timestamp = epochStart.Add(time.Duration(txIndexForTimestamp) * time.Millisecond)
+
+		// Clamp: the DA-derived timestamp may predate blocks that were
+		// produced with time.Now() before the sequencer was restarted.
+		// Ensure strict monotonicity relative to the last produced block.
+		if !c.lastCatchUpTimestamp.IsZero() && !timestamp.After(c.lastCatchUpTimestamp) {
+			timestamp = c.lastCatchUpTimestamp.Add(time.Millisecond)
+		}
+		c.lastCatchUpTimestamp = timestamp
 	}
 
 	return &coresequencer.GetNextBatchResponse{
@@ -534,7 +546,18 @@ func (c *Sequencer) updateCatchUpState(ctx context.Context) {
 		return
 	}
 
-	// More than one epoch behind - enter catch-up mode
+	// More than one epoch behind - enter catch-up mode.
+	// Read the last block time from the store so that catch-up timestamps
+	// are guaranteed to be strictly after any previously produced block.
+	s := store.New(store.NewEvNodeKVStore(c.db))
+	state, err := s.GetState(ctx)
+	if err == nil && !state.LastBlockTime.IsZero() {
+		c.lastCatchUpTimestamp = state.LastBlockTime
+		c.logger.Debug().
+			Time("last_block_time", state.LastBlockTime).
+			Msg("initialized catch-up timestamp floor from last block time")
+	}
+
 	c.catchUpState.Store(catchUpInProgress)
 	c.logger.Warn().
 		Uint64("checkpoint_da_height", currentDAHeight).