feat: add inline blob processing to DAFollower for zero-latency follow mode

When the DA subscription delivers blobs at the current local DA height,
the followLoop now processes them inline via ProcessBlobs — avoiding
a round-trip re-fetch from the DA layer.

Architecture:
- followLoop: processes subscription blobs inline when caught up (fast path),
  falls through to catchupLoop when behind (slow path).
- catchupLoop: unchanged — sequential RetrieveFromDA() for bulk sync.

Changes:
- Add Blobs field to SubscriptionEvent for carrying raw blob data
- Add extractBlobData() to DA client Subscribe adapter
- Export ProcessBlobs on DARetriever interface
- Add handleSubscriptionEvent() to DAFollower with inline fast path
- Add TestDAFollower_InlineProcessing with 3 sub-tests

Author: alpe

block/internal/da/client.go

@@ -381,7 +381,10 @@ func (c *client) Subscribe(ctx context.Context, namespace []byte) (<-chan datype
 					continue
 				}
 				select {
-				case out <- datypes.SubscriptionEvent{Height: resp.Height}:
+				case out <- datypes.SubscriptionEvent{
+					Height: resp.Height,
+					Blobs:  extractBlobData(resp),
+				}:
 				case <-ctx.Done():
 					return
 				}
@@ -392,6 +395,26 @@ func (c *client) Subscribe(ctx context.Context, namespace []byte) (<-chan datype
 	return out, nil
 }
 
+// extractBlobData extracts raw byte slices from a subscription response,
+// filtering out nil blobs and empty data.
+func extractBlobData(resp *blobrpc.SubscriptionResponse) [][]byte {
+	if resp == nil || len(resp.Blobs) == 0 {
+		return nil
+	}
+	blobs := make([][]byte, 0, len(resp.Blobs))
+	for _, blob := range resp.Blobs {
+		if blob == nil {
+			continue
+		}
+		data := blob.Data()
+		if len(data) == 0 {
+			continue
+		}
+		blobs = append(blobs, data)
+	}
+	return blobs
+}
+
 // Get fetches blobs by their IDs. Used for visualization and fetching specific blobs.
 func (c *client) Get(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error) {
 	if len(ids) == 0 {

block/internal/syncing/da_follower.go

@@ -17,8 +17,9 @@ import (
 // DAFollower subscribes to DA blob events and drives sequential catchup.
 //
 // Architecture:
-//   - followLoop listens on the subscription channel and atomically updates
-//     highestSeenDAHeight.
+//   - followLoop listens on the subscription channel. When caught up, it processes
+//     subscription blobs inline (fast path, no DA re-fetch). Otherwise, it updates
+//     highestSeenDAHeight and signals the catchup loop.
 //   - catchupLoop sequentially retrieves from localDAHeight → highestSeenDAHeight,
 //     piping events to the Syncer's heightInCh.
 //
@@ -169,11 +170,41 @@ func (f *daFollower) runSubscription() error {
 			if !ok {
 				return errors.New("subscription channel closed")
 			}
-			f.updateHighest(ev.Height)
+			f.handleSubscriptionEvent(ev)
 		}
 	}
 }
 
+// handleSubscriptionEvent processes a subscription event. When the follower is
+// caught up (ev.Height == localDAHeight) and blobs are available, it processes
+// them inline — avoiding a DA re-fetch round trip. Otherwise, it just updates
+// highestSeenDAHeight and lets catchupLoop handle retrieval.
+func (f *daFollower) handleSubscriptionEvent(ev datypes.SubscriptionEvent) {
+	// Always record the highest height we've seen from the subscription.
+	f.updateHighest(ev.Height)
+
+	// Fast path: process blobs inline when caught up.
+	// Only fire when ev.Height == localDAHeight to avoid out-of-order processing.
+	if len(ev.Blobs) > 0 && ev.Height == f.localDAHeight.Load() {
+		events := f.retriever.ProcessBlobs(f.ctx, ev.Blobs, ev.Height)
+		for _, event := range events {
+			if err := f.pipeEvent(f.ctx, event); err != nil {
+				f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
+					Msg("failed to pipe inline event, catchup will retry")
+				return // catchupLoop already signaled via updateHighest
+			}
+		}
+		// Advance local height — we processed this height inline.
+		f.localDAHeight.Store(ev.Height + 1)
+		f.headReached.Store(true)
+		f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
+			Msg("processed subscription blobs inline (fast path)")
+		return
+	}
+
+	// Slow path: behind or no blobs — catchupLoop will handle via signal from updateHighest.
+}
+
 // updateHighest atomically bumps highestSeenDAHeight and signals catchup if needed.
 func (f *daFollower) updateHighest(height uint64) {
 	for {

block/internal/syncing/da_retriever.go

@@ -24,6 +24,9 @@ import (
 type DARetriever interface {
 	// RetrieveFromDA retrieves blocks from the specified DA height and returns height events
 	RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
+	// ProcessBlobs parses raw blob bytes at a given DA height into height events.
+	// Used by the DAFollower to process subscription blobs inline without re-fetching.
+	ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent
 	// QueuePriorityHeight queues a DA height for priority retrieval (from P2P hints).
 	// These heights take precedence over sequential fetching.
 	QueuePriorityHeight(daHeight uint64)
@@ -191,6 +194,12 @@ func (r *daRetriever) validateBlobResponse(res datypes.ResultRetrieve, daHeight
 	}
 }
 
+// ProcessBlobs processes raw blob bytes to extract headers and data and returns height events.
+// This is the public interface used by the DAFollower for inline subscription processing.
+func (r *daRetriever) ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
+	return r.processBlobs(ctx, blobs, daHeight)
+}
+
 // processBlobs processes retrieved blobs to extract headers and data and returns height events
 func (r *daRetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
 	// Decode all blobs

block/internal/syncing/da_retriever_mock.go

@@ -63,3 +63,7 @@ func (t *tracedDARetriever) QueuePriorityHeight(daHeight uint64) {
 func (t *tracedDARetriever) PopPriorityHeight() uint64 {
 	return t.inner.PopPriorityHeight()
 }
+
+func (t *tracedDARetriever) ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
+	return t.inner.ProcessBlobs(ctx, blobs, daHeight)
+}

block/internal/syncing/da_retriever_tracing.go

@@ -31,6 +31,10 @@ func (m *mockDARetriever) QueuePriorityHeight(daHeight uint64) {}
 
 func (m *mockDARetriever) PopPriorityHeight() uint64 { return 0 }
 
+func (m *mockDARetriever) ProcessBlobs(_ context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
+	return nil
+}
+
 func setupDARetrieverTrace(t *testing.T, inner DARetriever) (DARetriever, *tracetest.SpanRecorder) {
 	t.Helper()
 	sr := tracetest.NewSpanRecorder()

block/internal/syncing/da_retriever_tracing_test.go

@@ -256,6 +256,112 @@ func TestDAFollower_CatchupThenReachHead(t *testing.T) {
 	})
 }
 
+// TestDAFollower_InlineProcessing verifies the fast path: when the subscription
+// delivers blobs at the current localDAHeight, handleSubscriptionEvent processes
+// them inline via ProcessBlobs (not RetrieveFromDA).
+func TestDAFollower_InlineProcessing(t *testing.T) {
+	t.Run("processes_blobs_inline_when_caught_up", func(t *testing.T) {
+		daRetriever := NewMockDARetriever(t)
+
+		var pipedEvents []common.DAHeightEvent
+		pipeEvent := func(_ context.Context, ev common.DAHeightEvent) error {
+			pipedEvents = append(pipedEvents, ev)
+			return nil
+		}
+
+		follower := NewDAFollower(DAFollowerConfig{
+			Retriever:     daRetriever,
+			Logger:        zerolog.Nop(),
+			PipeEvent:     pipeEvent,
+			Namespace:     []byte("ns"),
+			StartDAHeight: 10,
+			DABlockTime:   500 * time.Millisecond,
+		}).(*daFollower)
+
+		follower.ctx, follower.cancel = context.WithCancel(t.Context())
+		defer follower.cancel()
+
+		blobs := [][]byte{[]byte("header-blob"), []byte("data-blob")}
+		expectedEvents := []common.DAHeightEvent{
+			{DaHeight: 10, Source: common.SourceDA},
+		}
+
+		// ProcessBlobs should be called (not RetrieveFromDA)
+		daRetriever.On("ProcessBlobs", mock.Anything, blobs, uint64(10)).
+			Return(expectedEvents).Once()
+
+		// Simulate subscription event at the current localDAHeight
+		follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
+			Height: 10,
+			Blobs:  blobs,
+		})
+
+		// 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)
+		assert.Equal(t, uint64(11), follower.localDAHeight.Load(), "localDAHeight should advance past processed height")
+		assert.True(t, follower.HasReachedHead(), "should mark head as reached after inline processing")
+	})
+
+	t.Run("falls_through_to_catchup_when_behind", func(t *testing.T) {
+		daRetriever := NewMockDARetriever(t)
+
+		pipeEvent := func(_ context.Context, _ common.DAHeightEvent) error { return nil }
+
+		follower := NewDAFollower(DAFollowerConfig{
+			Retriever:     daRetriever,
+			Logger:        zerolog.Nop(),
+			PipeEvent:     pipeEvent,
+			Namespace:     []byte("ns"),
+			StartDAHeight: 10,
+			DABlockTime:   500 * time.Millisecond,
+		}).(*daFollower)
+
+		follower.ctx, follower.cancel = context.WithCancel(t.Context())
+		defer follower.cancel()
+
+		// Subscription reports height 15 but local is at 10 — should NOT process inline
+		follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
+			Height: 15,
+			Blobs:  [][]byte{[]byte("blob")},
+		})
+
+		// ProcessBlobs should NOT have been called
+		daRetriever.AssertNotCalled(t, "ProcessBlobs", mock.Anything, mock.Anything, mock.Anything)
+		assert.Equal(t, uint64(10), follower.localDAHeight.Load(), "localDAHeight should not change")
+		assert.Equal(t, uint64(15), follower.highestSeenDAHeight.Load(), "highestSeen should be updated")
+	})
+
+	t.Run("falls_through_when_no_blobs", func(t *testing.T) {
+		daRetriever := NewMockDARetriever(t)
+
+		pipeEvent := func(_ context.Context, _ common.DAHeightEvent) error { return nil }
+
+		follower := NewDAFollower(DAFollowerConfig{
+			Retriever:     daRetriever,
+			Logger:        zerolog.Nop(),
+			PipeEvent:     pipeEvent,
+			Namespace:     []byte("ns"),
+			StartDAHeight: 10,
+			DABlockTime:   500 * time.Millisecond,
+		}).(*daFollower)
+
+		follower.ctx, follower.cancel = context.WithCancel(t.Context())
+		defer follower.cancel()
+
+		// Subscription at current height but no blobs — should fall through
+		follower.handleSubscriptionEvent(datypes.SubscriptionEvent{
+			Height: 10,
+			Blobs:  nil,
+		})
+
+		// ProcessBlobs should NOT have been called
+		daRetriever.AssertNotCalled(t, "ProcessBlobs", mock.Anything, mock.Anything, mock.Anything)
+		assert.Equal(t, uint64(10), follower.localDAHeight.Load(), "localDAHeight should not change")
+		assert.Equal(t, uint64(10), follower.highestSeenDAHeight.Load(), "highestSeen should be updated")
+	})
+}
+
 // backoffTestGenesis creates a test genesis for the backoff tests.
 func backoffTestGenesis(addr []byte) genesis.Genesis {
 	return genesis.Genesis{

block/internal/syncing/syncer_backoff_test.go

@@ -88,4 +88,7 @@ func SplitID(id []byte) (uint64, []byte, error) {
 type SubscriptionEvent struct {
 	// Height is the DA layer height at which the blob was finalized.
 	Height uint64
+	// Blobs contains the raw blob data from the subscription response.
+	// When non-nil, followers can process blobs inline without re-fetching from DA.
+	Blobs [][]byte
 }