Handle edge case in catch up

Author: alpe

apps/testapp/cmd/run.go

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

block/internal/da/subscriber.go

@@ -65,6 +65,11 @@ type Subscriber struct {
 	// headReached tracks whether the subscriber has caught up to DA head.
 	headReached atomic.Bool
 
+	// seenSubscriptionEvent tracks whether followLoop has observed at least one
+	// subscription event in this run. Without this, startup catchup can stop
+	// too early when highestSeen is still just the initial start height.
+	seenSubscriptionEvent atomic.Bool
+
 	// catchupSignal wakes catchupLoop when a new height is seen above localDAHeight.
 	catchupSignal chan struct{}
 
@@ -192,6 +197,7 @@ func (s *Subscriber) runSubscription(ctx context.Context) error {
 			if !ok {
 				return errors.New("subscription channel closed")
 			}
+			s.seenSubscriptionEvent.Store(true)
 			s.updateHighest(ev.Height)
 
 			local := s.localDAHeight.Load()
@@ -317,6 +323,13 @@ func (s *Subscriber) runCatchup(ctx context.Context) {
 		}
 
 		local := s.localDAHeight.Load()
+		highest := s.highestSeenDAHeight.Load()
+
+		if s.seenSubscriptionEvent.Load() && local > highest {
+			s.headReached.Store(true)
+			return
+		}
+
 		// CAS claims this height — prevents followLoop from inline-processing.
 		if !s.localDAHeight.CompareAndSwap(local, local+1) {
 			// followLoop already advanced past this height via inline processing.
@@ -326,7 +339,7 @@ func (s *Subscriber) runCatchup(ctx context.Context) {
 		if err := s.handler.HandleCatchup(ctx, local); err != nil {
 			// Roll back so we can retry after backoff.
 			s.localDAHeight.Store(local)
-			if errors.Is(err, datypes.ErrHeightFromFuture) {
+			if errors.Is(err, datypes.ErrHeightFromFuture) && local >= highest {
 				s.headReached.Store(true)
 				return
 			}

block/internal/da/subscriber_test.go

@@ -46,10 +46,11 @@ func TestSubscriber_RunCatchup(t *testing.T) {
 			DABlockTime: time.Millisecond,
 		})
 
-		// It should try 100, 101, then we return ErrHeightFromFuture at 102
+		// It should process observed heights [100..101] then stop when local passes highestSeen.
+		sub.updateHighest(101)
+		sub.seenSubscriptionEvent.Store(true)
 		mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).Return(nil).Once()
 		mockHandler.On("HandleCatchup", mock.Anything, uint64(101)).Return(nil).Once()
-		mockHandler.On("HandleCatchup", mock.Anything, uint64(102)).Return(datypes.ErrHeightFromFuture).Once()
 
 		sub.runCatchup(ctx)
 
@@ -76,6 +77,9 @@ func TestSubscriber_RunCatchup(t *testing.T) {
 
 		var callCount int
 
+		sub.updateHighest(100)
+		sub.seenSubscriptionEvent.Store(true)
+
 		mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).
 			Run(func(args mock.Arguments) {
 				callCount++
@@ -85,14 +89,14 @@ func TestSubscriber_RunCatchup(t *testing.T) {
 		mockHandler.On("HandleCatchup", mock.Anything, uint64(100)).
 			Run(func(args mock.Arguments) {
 				callCount++
-				cancel()
 			}).
-			Return(datypes.ErrHeightFromFuture).Once()
+			Return(nil).Once()
 
 		sub.runCatchup(ctx)
 
 		mockHandler.AssertExpectations(t)
 		assert.Equal(t, 2, callCount)
-		assert.Equal(t, uint64(100), sub.LocalDAHeight(), "should roll back to 100 on future error")
+		assert.Equal(t, uint64(101), sub.LocalDAHeight())
+		assert.True(t, sub.HasReachedHead())
 	})
 }