Review feedback

Author: alpe

block/internal/da/subscriber.go

@@ -80,6 +80,7 @@ type Subscriber struct {
 	fetchBlockTimestamp bool
 
 	// lifecycle
+	lifecycleMu sync.Mutex
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
 }
@@ -113,8 +114,17 @@ func (s *Subscriber) Start(ctx context.Context) error {
 		return nil
 	}
 
-	ctx, s.cancel = context.WithCancel(ctx)
+	s.lifecycleMu.Lock()
+	if s.cancel != nil {
+		s.lifecycleMu.Unlock()
+		return nil
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	s.cancel = cancel
 	s.wg.Add(2)
+	s.lifecycleMu.Unlock()
+
 	go s.followLoop(ctx)
 	go s.catchupLoop(ctx)
 
@@ -123,8 +133,13 @@ func (s *Subscriber) Start(ctx context.Context) error {
 
 // Stop gracefully stops the background goroutines.
 func (s *Subscriber) Stop() {
-	if s.cancel != nil {
-		s.cancel()
+	s.lifecycleMu.Lock()
+	cancel := s.cancel
+	s.cancel = nil
+	s.lifecycleMu.Unlock()
+
+	if cancel != nil {
+		cancel()
 	}
 	s.wg.Wait()
 }
@@ -206,7 +221,12 @@ func (s *Subscriber) runSubscription(ctx context.Context) error {
 			err = s.handler.HandleEvent(subCtx, ev, isInline)
 			if isInline {
 				if err == nil {
-					s.headReached.Store(true)
+					highest := s.highestSeenDAHeight.Load()
+					// Mark head reached only if this inline event is at or beyond
+					// the currently observed head; otherwise catchup is still pending.
+					if ev.Height >= highest {
+						s.headReached.Store(true)
+					}
 				} else {
 					s.localDAHeight.Store(local)
 					s.logger.Debug().Err(err).Uint64("da_height", ev.Height).

block/internal/da/subscriber_test.go

@@ -100,3 +100,48 @@ func TestSubscriber_RunCatchup(t *testing.T) {
 		assert.True(t, sub.HasReachedHead())
 	})
 }
+
+func TestSubscriber_RunSubscription_InlineDoesNotPrematurelyReachHead(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	mockHandler := new(MockSubscriberHandler)
+	mockClient := testmocks.NewMockClient(t)
+
+	sub := NewSubscriber(SubscriberConfig{
+		Client:      mockClient,
+		Logger:      zerolog.Nop(),
+		Handler:     mockHandler,
+		Namespaces:  [][]byte{[]byte("ns")},
+		StartHeight: 100,
+		DABlockTime: time.Hour,
+	})
+
+	subCh := make(chan datypes.SubscriptionEvent, 2)
+	mockClient.EXPECT().
+		Subscribe(mock.Anything, []byte("ns"), false).
+		Return((<-chan datypes.SubscriptionEvent)(subCh), nil).
+		Once()
+
+	mockHandler.On("HandleEvent", mock.Anything, datypes.SubscriptionEvent{
+		Height: 101,
+		Blobs:  [][]byte{[]byte("h101")},
+	}, false).Return(nil).Once()
+	mockHandler.On("HandleEvent", mock.Anything, datypes.SubscriptionEvent{
+		Height: 100,
+		Blobs:  [][]byte{[]byte("h100")},
+	}, true).Return(nil).Once()
+
+	subCh <- datypes.SubscriptionEvent{Height: 101, Blobs: [][]byte{[]byte("h101")}}
+	subCh <- datypes.SubscriptionEvent{Height: 100, Blobs: [][]byte{[]byte("h100")}}
+	close(subCh)
+
+	err := sub.runSubscription(ctx)
+	assert.Error(t, err)
+	if err != nil {
+		assert.Contains(t, err.Error(), "subscription channel closed")
+	}
+	assert.False(t, sub.HasReachedHead())
+	assert.Equal(t, uint64(101), sub.LocalDAHeight())
+	assert.Equal(t, uint64(101), sub.HighestSeenDAHeight())
+}

block/internal/syncing/da_follower.go

@@ -36,6 +36,8 @@ type daFollower struct {
 	priorityHeights []uint64
 }
 
+const maxPriorityHeights = 1024
+
 // DAFollowerConfig holds configuration for creating a DAFollower.
 type DAFollowerConfig struct {
 	Client        da.Client
@@ -183,6 +185,16 @@ func (f *daFollower) QueuePriorityHeight(daHeight uint64) {
 	if found {
 		return
 	}
+
+	// Keep the queue bounded. When full, prefer lower (sooner) heights.
+	if len(f.priorityHeights) >= maxPriorityHeights {
+		last := f.priorityHeights[len(f.priorityHeights)-1]
+		if daHeight >= last {
+			return
+		}
+		f.priorityHeights = f.priorityHeights[:len(f.priorityHeights)-1]
+	}
+
 	f.priorityHeights = slices.Insert(f.priorityHeights, idx, daHeight)
 }
 

block/internal/syncing/da_follower_test.go

@@ -206,3 +206,48 @@ func TestDAFollower_HandleCatchup(t *testing.T) {
 		})
 	}
 }
+
+func TestDAFollower_QueuePriorityHeight(t *testing.T) {
+	specs := map[string]struct {
+		initial []uint64
+		queue   []uint64
+		want    []uint64
+	}{
+		"sorts_and_deduplicates": {
+			initial: []uint64{5, 10},
+			queue:   []uint64{7, 10, 3},
+			want:    []uint64{3, 5, 7, 10},
+		},
+		"bounded_drops_largest_when_smaller_arrives": {
+			initial: makeRange(1, maxPriorityHeights),
+			queue:   []uint64{maxPriorityHeights + 1, 0},
+			want:    append([]uint64{0}, makeRange(1, maxPriorityHeights-1)...),
+		},
+	}
+
+	for name, spec := range specs {
+		t.Run(name, func(t *testing.T) {
+			follower := &daFollower{
+				logger:          zerolog.Nop(),
+				priorityHeights: append([]uint64(nil), spec.initial...),
+			}
+
+			for _, daHeight := range spec.queue {
+				follower.QueuePriorityHeight(daHeight)
+			}
+
+			assert.Equal(t, spec.want, follower.priorityHeights)
+		})
+	}
+}
+
+func makeRange(start, end uint64) []uint64 {
+	if end < start {
+		return nil
+	}
+	out := make([]uint64, 0, end-start+1)
+	for v := start; v <= end; v++ {
+		out = append(out, v)
+	}
+	return out
+}