Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Handle edge case in catch up
  • Loading branch information
alpe committed Mar 13, 2026
commit 6552a8262d13c73755d67a05007a575cebe5d6b2
2 changes: 1 addition & 1 deletion apps/testapp/cmd/run.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
15 changes: 14 additions & 1 deletion block/internal/da/subscriber.go
Original file line number Diff line number Diff line change
Expand Up @@ -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{}

Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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.
Expand All @@ -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
}
Expand Down
14 changes: 9 additions & 5 deletions block/internal/da/subscriber_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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++
Expand All @@ -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())
})
}
Loading