Apply review feedback

Author: alpe

apps/evm/server/force_inclusion_test.go

@@ -50,7 +50,7 @@ func (m *mockDA) Get(ctx context.Context, ids []da.ID, namespace []byte) ([]da.B
 	return nil, nil
 }
 
-func (m *mockDA) Subscribe(_ context.Context, _ []byte, _ bool) (<-chan da.SubscriptionEvent, error) {
+func (m *mockDA) Subscribe(_ context.Context, _ []byte) (<-chan da.SubscriptionEvent, error) {
 	// Not needed in these tests; return a closed channel.
 	ch := make(chan da.SubscriptionEvent)
 	close(ch)

block/internal/da/async_block_retriever.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync/atomic"
 	"time"
 
 	ds "github.com/ipfs/go-datastore"
@@ -44,9 +45,11 @@ type asyncBlockRetriever struct {
 	cache ds.Batching
 
 	// Current DA height tracking (accessed atomically via subscriber).
-	// Updated externally via UpdateCurrentHeight.
 	daStartHeight uint64
 
+	// Tracks DA height consumed by the sequencer to trigger cache cleanups.
+	consumedHeight atomic.Uint64
+
 	// Prefetch window - how many blocks ahead to speculatively fetch.
 	prefetchWindow uint64
 }
@@ -108,17 +111,17 @@ func (f *asyncBlockRetriever) Stop() {
 	f.subscriber.Stop()
 }
 
-// UpdateCurrentHeight updates the current DA height.
+// UpdateCurrentHeight updates the consumed DA height and triggers cache cleanup.
 func (f *asyncBlockRetriever) UpdateCurrentHeight(height uint64) {
 	for {
-		current := f.subscriber.LocalDAHeight()
+		current := f.consumedHeight.Load()
 		if height <= current {
 			return
 		}
-		if f.subscriber.CompareAndSwapLocalHeight(current, height) {
+		if f.consumedHeight.CompareAndSwap(current, height) {
 			f.logger.Debug().
 				Uint64("new_height", height).
-				Msg("updated current DA height")
+				Msg("updated consumed DA height for cleanup")
 			f.cleanupOldBlocks(context.Background(), height)
 			return
 		}
@@ -168,11 +171,10 @@ func (f *asyncBlockRetriever) GetCachedBlock(ctx context.Context, daHeight uint6
 	return block, nil
 }
 
-// HandleEvent caches blobs from the subscription inline.
+// HandleEvent caches blobs from the subscription inline, even empty ones,
+// to record that the DA height was seen and has 0 blobs.
 func (f *asyncBlockRetriever) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent) {
-	if len(ev.Blobs) > 0 {
-		f.cacheBlock(ctx, ev.Height, ev.Timestamp, ev.Blobs)
-	}
+	f.cacheBlock(ctx, ev.Height, ev.Timestamp, ev.Blobs)
 }
 
 // HandleCatchup fetches a single height via Retrieve and caches it.
@@ -181,8 +183,7 @@ func (f *asyncBlockRetriever) HandleCatchup(ctx context.Context, height uint64)
 	f.fetchAndCacheBlock(ctx, height)
 
 	// Speculatively prefetch ahead.
-	highest := f.subscriber.HighestSeenDAHeight()
-	target := highest + f.prefetchWindow
+	target := height + f.prefetchWindow
 	for h := height + 1; h <= target; h++ {
 		if err := ctx.Err(); err != nil {
 			return err

block/internal/da/async_block_retriever_test.go

@@ -156,9 +156,7 @@ func TestAsyncBlockRetriever_HeightFromFuture(t *testing.T) {
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
 	// Subscription delivers height 100 with no blobs.
-	subCh := make(chan datypes.SubscriptionEvent, 1)
-	subCh <- datypes.SubscriptionEvent{Height: 100}
-
+	subCh := make(chan datypes.SubscriptionEvent)
 	client.On("Subscribe", mock.Anything, fiNs, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(subCh), nil).Once()
 	blockCh := make(chan datypes.SubscriptionEvent)
 	client.On("Subscribe", mock.Anything, fiNs, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(blockCh), nil).Maybe()
@@ -169,15 +167,17 @@ func TestAsyncBlockRetriever_HeightFromFuture(t *testing.T) {
 	}).Maybe()
 
 	logger := zerolog.Nop()
-	fetcher := NewAsyncBlockRetriever(client, logger, fiNs, 100*time.Millisecond, 100, 10)
+	fetcher := NewAsyncBlockRetriever(client, logger, fiNs, time.Millisecond, 100, 10)
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(t.Context())
 	defer cancel()
 	fetcher.Start(ctx)
 	defer fetcher.Stop()
 
 	// Wait a bit for catchup to attempt fetches.
-	time.Sleep(250 * time.Millisecond)
+	require.Eventually(t, func() bool {
+		return fetcher.(*asyncBlockRetriever).subscriber.HasReachedHead()
+	}, 1250*time.Millisecond, time.Millisecond)
 
 	// Cache should be empty since all heights are from the future.
 	block, err := fetcher.GetCachedBlock(ctx, 100)

block/internal/da/subscriber.go

@@ -87,6 +87,9 @@ func NewSubscriber(cfg SubscriberConfig) *Subscriber {
 		daBlockTime:         cfg.DABlockTime,
 		fetchBlockTimestamp: cfg.FetchBlockTimestamp,
 	}
+	if len(s.namespaces) == 0 {
+		s.logger.Warn().Msg("no namespaces configured, subscriber will stay idle")
+	}
 	return s
 }
 
@@ -98,7 +101,7 @@ func (s *Subscriber) SetStartHeight(height uint64) {
 // Start begins the follow and catchup goroutines.
 func (s *Subscriber) Start(ctx context.Context) error {
 	if len(s.namespaces) == 0 {
-		return errors.New("no namespaces configured")
+		return nil
 	}
 
 	ctx, s.cancel = context.WithCancel(ctx)
@@ -131,11 +134,6 @@ func (s *Subscriber) HasReachedHead() bool {
 	return s.headReached.Load()
 }
 
-// SetHeadReached marks the subscriber as having reached DA head.
-func (s *Subscriber) SetHeadReached() {
-	s.headReached.Store(true)
-}
-
 // CompareAndSwapLocalHeight attempts a CAS on localDAHeight.
 // Used by handlers that want to claim exclusive processing of a height.
 func (s *Subscriber) CompareAndSwapLocalHeight(old, new uint64) bool {
@@ -343,23 +341,19 @@ 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 !s.waitOnCatchupError(ctx, err, local) {
+			if errors.Is(err, datypes.ErrHeightFromFuture) {
+				s.headReached.Store(true)
+				return
+			}
+			if !s.shouldContinueCatchup(ctx, err, local) {
 				return
 			}
-			continue
 		}
 	}
 }
 
-// ErrCaughtUp is a sentinel used to signal that the catchup loop has reached DA head.
-var ErrCaughtUp = errors.New("caught up with DA head")
-
-// waitOnCatchupError logs the error and backs off before retrying.
-func (s *Subscriber) waitOnCatchupError(ctx context.Context, err error, daHeight uint64) bool {
-	if errors.Is(err, ErrCaughtUp) {
-		s.logger.Debug().Uint64("da_height", daHeight).Msg("DA catchup reached head, waiting for subscription signal")
-		return false
-	}
+// shouldContinueCatchup logs the error and backs off before retrying.
+func (s *Subscriber) shouldContinueCatchup(ctx context.Context, err error, daHeight uint64) bool {
 	if ctx.Err() != nil {
 		return false
 	}

block/internal/syncing/da_follower.go

@@ -116,7 +116,6 @@ func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEve
 			}
 		}
 		if len(events) != 0 {
-			f.subscriber.SetHeadReached()
 			f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
 				Msg("processed subscription blobs inline (fast path)")
 		} else {
@@ -131,38 +130,63 @@ func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEve
 
 // HandleCatchup retrieves events at a single DA height and pipes them
 // to the event sink. Checks priority heights first.
+//
+// ErrHeightFromFuture is handled here rather than in fetchAndPipeHeight because
+// the two call sites need different behaviour:
+//   - Normal catchup: mark head reached and return da.ErrCaughtUp to stop the loop.
+//   - Priority hint:  the hint points to a future height — ignore it without
+//     skipping the current daHeight or stopping catchup.
 func (f *daFollower) HandleCatchup(ctx context.Context, daHeight uint64) error {
-	// Check for priority heights from P2P hints first.
-	if priorityHeight := f.popPriorityHeight(); priorityHeight > 0 {
-		if priorityHeight >= daHeight {
-			f.logger.Debug().
-				Uint64("da_height", priorityHeight).
-				Msg("fetching priority DA height from P2P hint")
-			if err := f.fetchAndPipeHeight(ctx, priorityHeight); err != nil {
-				return err
+	// 1. Drain stale or future priority heights from P2P hints
+	for {
+		priorityHeight := f.popPriorityHeight()
+		if priorityHeight == 0 {
+			break
+		}
+		if priorityHeight < daHeight {
+			continue // skip stale hints without yielding back to the catchup loop
+		}
+
+		f.logger.Debug().
+			Uint64("da_height", priorityHeight).
+			Msg("fetching priority DA height from P2P hint")
+
+		if err := f.fetchAndPipeHeight(ctx, priorityHeight); err != nil {
+			if errors.Is(err, datypes.ErrHeightFromFuture) {
+				// Priority hint points to a future height — silently ignore.
+				f.logger.Debug().Uint64("priority_da_height", priorityHeight).
+					Msg("priority hint is from future, ignoring")
+				continue
 			}
+			// Roll back so daHeight is attempted again next cycle after backoff.
+			f.subscriber.SetLocalHeight(daHeight)
+			return err
 		}
-		// Re-queue the current height by rolling back (the subscriber already advanced).
+
+		// We successfully handled a priority height (we didn't actually process `daHeight`)
+		// Roll back so daHeight is attempted again next cycle.
 		f.subscriber.SetLocalHeight(daHeight)
 		return nil
 	}
 
-	return f.fetchAndPipeHeight(ctx, daHeight)
+	// 2. Normal sequential fetch
+	if err := f.fetchAndPipeHeight(ctx, daHeight); err != nil {
+		return err
+	}
+	return nil
 }
 
 // fetchAndPipeHeight retrieves events at a single DA height and pipes them.
+// It does NOT handle ErrHeightFromFuture — callers must decide how to react
+// because the correct response depends on whether this is a normal sequential
+// catchup or a priority-hint fetch.
 func (f *daFollower) fetchAndPipeHeight(ctx context.Context, daHeight uint64) error {
 	events, err := f.retriever.RetrieveFromDA(ctx, daHeight)
 	if err != nil {
-		switch {
-		case errors.Is(err, datypes.ErrBlobNotFound):
+		if errors.Is(err, datypes.ErrBlobNotFound) {
 			return nil
-		case errors.Is(err, datypes.ErrHeightFromFuture):
-			f.subscriber.SetHeadReached()
-			return err
-		default:
-			return err
 		}
+		return err
 	}
 
 	for _, event := range events {
@@ -200,5 +224,8 @@ func (f *daFollower) popPriorityHeight() uint64 {
 	}
 	height := f.priorityHeights[0]
 	f.priorityHeights = f.priorityHeights[1:]
+	if len(f.priorityHeights) == 0 {
+		f.priorityHeights = nil
+	}
 	return height
 }

block/internal/syncing/syncer_backoff_test.go

@@ -25,6 +25,7 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
 		daBlockTime    time.Duration
 		error          error
 		expectsBackoff bool
+		exitsCatchup   bool // ErrCaughtUp → clean exit, no backoff, no advance
 		description    string
 	}{
 		"generic_error_triggers_backoff": {
@@ -33,11 +34,11 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
 			expectsBackoff: true,
 			description:    "Generic DA errors should trigger backoff",
 		},
-		"height_from_future_triggers_backoff": {
-			daBlockTime:    500 * time.Millisecond,
-			error:          datypes.ErrHeightFromFuture,
-			expectsBackoff: true,
-			description:    "Height from future should trigger backoff",
+		"height_from_future_stops_catchup": {
+			daBlockTime:  500 * time.Millisecond,
+			error:        datypes.ErrHeightFromFuture,
+			exitsCatchup: true,
+			description:  "Height from future should stop catchup (da.ErrCaughtUp), not backoff",
 		},
 		"blob_not_found_no_backoff": {
 			daBlockTime:    1 * time.Second,
@@ -92,7 +93,10 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
 					}).
 					Return(nil, tc.error).Once()
 
-				if tc.expectsBackoff {
+				if tc.exitsCatchup {
+					// ErrCaughtUp causes runCatchup to exit cleanly after 1 call.
+					// No second mock needed.
+				} else if tc.expectsBackoff {
 					daRetriever.On("RetrieveFromDA", mock.Anything, uint64(100)).
 						Run(func(args mock.Arguments) {
 							callTimes = append(callTimes, time.Now())
@@ -110,10 +114,20 @@ func TestDAFollower_BackoffOnCatchupError(t *testing.T) {
 						Return(nil, datypes.ErrBlobNotFound).Once()
 				}
 
-				go sub.RunCatchupForTest(ctx)
-				<-ctx.Done()
+				done := make(chan struct{})
+				go func() {
+					sub.RunCatchupForTest(ctx)
+					close(done)
+				}()
+				select {
+				case <-ctx.Done():
+				case <-done:
+				}
 
-				if tc.expectsBackoff {
+				if tc.exitsCatchup {
+					assert.Equal(t, 1, callCount, "should exit after single call (ErrCaughtUp)")
+					assert.True(t, follower.HasReachedHead(), "should mark head as reached")
+				} else if tc.expectsBackoff {
 					require.Len(t, callTimes, 2, "should make exactly 2 calls with backoff")
 
 					timeBetweenCalls := callTimes[1].Sub(callTimes[0])
@@ -291,12 +305,16 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 		daRetriever.On("ProcessBlobs", mock.Anything, blobs, uint64(10)).
 			Return(expectedEvents).Once()
 
-		// Simulate subscription event at the current localDAHeight
+		// Simulate subscription event: update highest seen, then handle event
+		follower.subscriber.UpdateHighestForTest(10)
 		follower.HandleEvent(t.Context(), datypes.SubscriptionEvent{
 			Height: 10,
 			Blobs:  blobs,
 		})
 
+		// Simulate catchup loop waking up to see if local > highest
+		follower.subscriber.RunCatchupForTest(t.Context())
+
 		// Verify: ProcessBlobs was called, events were piped, height advanced
 		require.Len(t, pipedEvents, 1, "should pipe 1 event from inline processing")
 		assert.Equal(t, uint64(10), pipedEvents[0].DaHeight)