Updates

Author: alpe

block/internal/da/forced_inclusion_retriever.go

@@ -18,6 +18,7 @@ var ErrForceInclusionNotConfigured = errors.New("forced inclusion namespace not
 // ForcedInclusionRetriever defines the interface for retrieving forced inclusion transactions from DA.
 type ForcedInclusionRetriever interface {
 	RetrieveForcedIncludedTxs(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error)
+	Start(ctx context.Context)
 	Stop()
 }
 
@@ -59,7 +60,6 @@ func NewForcedInclusionRetriever(
 		daStartHeight,
 		daEpochSize*2, // prefetch window: 2x epoch size
 	)
-	asyncFetcher.Start(ctx)
 
 	base := &forcedInclusionRetriever{
 		client:        client,
@@ -74,6 +74,11 @@ func NewForcedInclusionRetriever(
 	return base
 }
 
+// Start begins the background prefetcher.
+func (r *forcedInclusionRetriever) Start(ctx context.Context) {
+	r.asyncFetcher.Start(ctx)
+}
+
 // Stop stops the background prefetcher.
 func (r *forcedInclusionRetriever) Stop() {
 	r.asyncFetcher.Stop()

block/internal/da/forced_inclusion_tracing.go

@@ -49,6 +49,10 @@ func (t *tracedForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.C
 	return event, nil
 }
 
+func (t *tracedForcedInclusionRetriever) Start(ctx context.Context) {
+	t.inner.Start(ctx)
+}
+
 func (t *tracedForcedInclusionRetriever) Stop() {
 	t.inner.Stop()
 }

block/internal/da/subscriber.go

@@ -201,12 +201,12 @@ func (s *Subscriber) runSubscription(ctx context.Context) error {
 
 			err = s.handler.HandleEvent(ctx, ev, isInline)
 			if isInline {
-				if err != nil {
+				if err == nil {
+					s.headReached.Store(true)
+				} else {
 					s.localDAHeight.Store(local)
 					s.logger.Debug().Err(err).Uint64("da_height", ev.Height).
 						Msg("inline processing skipped/failed, rolling back")
-				} else {
-					s.headReached.Store(true)
 				}
 			}
 

block/internal/syncing/da_follower.go

@@ -98,8 +98,11 @@ func (f *daFollower) HasReachedHead() bool {
 // them inline — avoiding a DA re-fetch round trip. Otherwise, it just lets
 // the catchup loop handle retrieval.
 func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error {
-	if !isInline || len(ev.Blobs) == 0 {
-		return nil // skip
+	if !isInline {
+		return nil // skip: let subscriber just update highestSeenDAHeight
+	}
+	if len(ev.Blobs) == 0 {
+		return errors.New("skip inline: no blobs") // subscriber rolls back, catch-up loop will retry
 	}
 
 	events := f.retriever.ProcessBlobs(ctx, ev.Blobs, ev.Height)
@@ -122,12 +125,6 @@ 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 {
 	// 1. Drain stale or future priority heights from P2P hints
 	for priorityHeight := f.popPriorityHeight(); priorityHeight != 0; priorityHeight = f.popPriorityHeight() {

block/internal/syncing/syncer.go

@@ -182,6 +182,7 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 	}
 
 	s.fiRetriever = da.NewForcedInclusionRetriever(ctx, s.daClient, s.logger, s.config.DA.BlockTime.Duration, s.config.Instrumentation.IsTracingEnabled(), s.genesis.DAStartHeight, s.genesis.DAEpochForcedInclusion)
+	s.fiRetriever.Start(ctx)
 	s.p2pHandler = NewP2PHandler(s.headerStore, s.dataStore, s.cache, s.genesis, s.logger)
 
 	currentHeight, initErr := s.store.Height(ctx)

block/internal/syncing/syncer_backoff_test.go

@@ -315,7 +315,8 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 		}
 
 		daRetriever.On("ProcessBlobs", mock.Anything, blobs, uint64(10)).Return(expectedEvents).Once()
-		daRetriever.On("RetrieveFromDA", mock.Anything, uint64(10)).Return(expectedEvents, nil).Maybe()
+		// Mock RetrieveFromDA for height 10 in case background catch-up fires before HandleEvent.
+		daRetriever.On("RetrieveFromDA", mock.Anything, uint64(10)).Return(nil, datypes.ErrHeightFromFuture).Maybe()
 
 		// Simulate catchup loop to verify it advances naturally
 		require.NoError(t, follower.Start(t.Context()))
@@ -367,7 +368,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 
 		pipeEvent := func(_ context.Context, _ common.DAHeightEvent) error { return nil }
 
-		daClient, _ := setupMockDAClient(t)
+		daClient, eventCh := setupMockDAClient(t)
 		follower := NewDAFollower(DAFollowerConfig{
 			Client:        daClient,
 			Retriever:     daRetriever,
@@ -377,15 +378,23 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 			StartDAHeight: 10,
 			DABlockTime:   500 * time.Millisecond,
 		}).(*daFollower)
+
+		// Mock RetrieveFromDA because falling through inline processing triggers catchup.
+		daRetriever.On("RetrieveFromDA", mock.Anything, uint64(10)).Return(nil, datypes.ErrHeightFromFuture).Maybe()
+		require.NoError(t, follower.Start(t.Context()))
+
 		// HandleEvent at current height but no blobs — should fall through
 		// Subscriber.HandleEvent updates highest internally even if not processed.
-		follower.HandleEvent(t.Context(), datypes.SubscriptionEvent{
+		eventCh <- datypes.SubscriptionEvent{
 			Height: 10,
 			Blobs:  nil,
-		}, true)
+		}
 
 		// ProcessBlobs should NOT have been called
 		daRetriever.AssertNotCalled(t, "ProcessBlobs", mock.Anything, mock.Anything, mock.Anything)
+		require.Eventually(t, func() bool {
+			return follower.subscriber.LocalDAHeight() == 10 && follower.subscriber.HighestSeenDAHeight() == 10
+		}, 100*time.Millisecond, 2*time.Millisecond)
 		assert.Equal(t, uint64(10), follower.subscriber.LocalDAHeight(), "localDAHeight should not change")
 		assert.Equal(t, uint64(10), follower.subscriber.HighestSeenDAHeight(), "highestSeen should be updated")
 	})

block/public.go

@@ -78,6 +78,7 @@ type ForcedInclusionEvent = da.ForcedInclusionEvent
 type ForcedInclusionRetriever interface {
 	RetrieveForcedIncludedTxs(ctx context.Context, daHeight uint64) (*ForcedInclusionEvent, error)
 	Stop()
+	Start(ctx context.Context)
 }
 
 // NewForcedInclusionRetriever creates a new forced inclusion retriever.

pkg/sequencers/common/forced_inclusion_retriever_mock.go

@@ -203,6 +203,7 @@ func (c *Sequencer) GetNextBatch(ctx context.Context, req coresequencer.GetNextB
 			c.fiRetriever.Stop()
 		}
 		c.fiRetriever = block.NewForcedInclusionRetriever(ctx, c.daClient, c.cfg, c.logger, c.getInitialDAStartHeight(ctx), c.genesis.DAEpochForcedInclusion)
+		c.fiRetriever.Start(ctx)
 	}
 
 	// If we have no cached transactions or we've consumed all from the current epoch,