feat: add ProcessBlobs method to DARetriever interface and implement in daRetriever; update tests and syncer for subscription handling

Author: tac0turtle

block/internal/syncing/da_retriever.go

@@ -21,6 +21,9 @@ import (
 // DARetriever defines the interface for retrieving events from the DA layer
 type DARetriever interface {
 	RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
+	// ProcessBlobs processes raw blobs from subscription and returns height events.
+	// Used by follow mode to process real-time blob notifications.
+	ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent
 }
 
 // daRetriever handles DA retrieval operations for syncing
@@ -72,7 +75,7 @@ func (r *daRetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]co
 	}
 
 	r.logger.Debug().Int("blobs", len(blobsResp.Data)).Uint64("da_height", daHeight).Msg("retrieved blob data")
-	return r.processBlobs(ctx, blobsResp.Data, daHeight), nil
+	return r.ProcessBlobs(ctx, blobsResp.Data, daHeight), nil
 }
 
 // fetchBlobs retrieves blobs from both header and data namespaces
@@ -148,8 +151,9 @@ func (r *daRetriever) validateBlobResponse(res datypes.ResultRetrieve, 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 {
+// ProcessBlobs processes retrieved blobs to extract headers and data and returns height events.
+// This method implements the DARetriever interface and is used by both polling and subscription modes.
+func (r *daRetriever) ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
 	// Decode all blobs
 	for _, bz := range blobs {
 		if len(bz) == 0 {

block/internal/syncing/da_retriever_mock.go

@@ -148,7 +148,7 @@ func TestDARetriever_ProcessBlobs_HeaderAndData_Success(t *testing.T) {
 	dataBin, data := makeSignedDataBytes(t, gen.ChainID, 2, addr, pub, signer, 2)
 	hdrBin, _ := makeSignedHeaderBytes(t, gen.ChainID, 2, addr, pub, signer, nil, &data.Data, nil)
 
-	events := r.processBlobs(context.Background(), [][]byte{hdrBin, dataBin}, 77)
+	events := r.ProcessBlobs(context.Background(), [][]byte{hdrBin, dataBin}, 77)
 	require.Len(t, events, 1)
 	assert.Equal(t, uint64(2), events[0].Header.Height())
 	assert.Equal(t, uint64(2), events[0].Data.Height())
@@ -172,7 +172,7 @@ func TestDARetriever_ProcessBlobs_HeaderOnly_EmptyDataExpected(t *testing.T) {
 	// Header with no data hash present should trigger empty data creation (per current logic)
 	hb, _ := makeSignedHeaderBytes(t, gen.ChainID, 3, addr, pub, signer, nil, nil, nil)
 
-	events := r.processBlobs(context.Background(), [][]byte{hb}, 88)
+	events := r.ProcessBlobs(context.Background(), [][]byte{hb}, 88)
 	require.Len(t, events, 1)
 	assert.Equal(t, uint64(3), events[0].Header.Height())
 	assert.NotNil(t, events[0].Data)
@@ -282,14 +282,14 @@ func TestDARetriever_ProcessBlobs_CrossDAHeightMatching(t *testing.T) {
 	hdrBin, _ := makeSignedHeaderBytes(t, gen.ChainID, 5, addr, pub, signer, nil, &data.Data, nil)
 
 	// Process header from DA height 100 first
-	events1 := r.processBlobs(context.Background(), [][]byte{hdrBin}, 100)
+	events1 := r.ProcessBlobs(context.Background(), [][]byte{hdrBin}, 100)
 	require.Len(t, events1, 0, "should not create event yet - data is missing")
 
 	// Verify header is stored in pending headers
 	require.Contains(t, r.pendingHeaders, uint64(5), "header should be stored as pending")
 
 	// Process data from DA height 102
-	events2 := r.processBlobs(context.Background(), [][]byte{dataBin}, 102)
+	events2 := r.ProcessBlobs(context.Background(), [][]byte{dataBin}, 102)
 	require.Len(t, events2, 1, "should create event when matching data arrives")
 
 	event := events2[0]
@@ -319,7 +319,7 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	hdr5Bin, _ := makeSignedHeaderBytes(t, gen.ChainID, 5, addr, pub, signer, nil, &data5.Data, nil)
 
 	// Process multiple headers from DA height 200 - should be stored as pending
-	events1 := r.processBlobs(context.Background(), [][]byte{hdr3Bin, hdr4Bin, hdr5Bin}, 200)
+	events1 := r.ProcessBlobs(context.Background(), [][]byte{hdr3Bin, hdr4Bin, hdr5Bin}, 200)
 	require.Len(t, events1, 0, "should not create events yet - all data is missing")
 
 	// Verify all headers are stored in pending
@@ -328,7 +328,7 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	require.Contains(t, r.pendingHeaders, uint64(5), "header 5 should be pending")
 
 	// Process some data from DA height 203 - should create partial events
-	events2 := r.processBlobs(context.Background(), [][]byte{data3Bin, data5Bin}, 203)
+	events2 := r.ProcessBlobs(context.Background(), [][]byte{data3Bin, data5Bin}, 203)
 	require.Len(t, events2, 2, "should create events for heights 3 and 5")
 
 	// Sort events by height for consistent testing
@@ -352,7 +352,7 @@ func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testin
 	require.NotContains(t, r.pendingHeaders, uint64(5), "header 5 should be removed from pending")
 
 	// Process remaining data from DA height 205
-	events3 := r.processBlobs(context.Background(), [][]byte{data4Bin}, 205)
+	events3 := r.ProcessBlobs(context.Background(), [][]byte{data4Bin}, 205)
 	require.Len(t, events3, 1, "should create event for height 4")
 
 	// Verify final event for height 4

block/internal/syncing/da_retriever_test.go

@@ -446,7 +446,7 @@ func (s *Syncer) runFollowMode() {
 }
 
 // subscribeAndFollow uses the DA subscription API to receive real-time blob notifications.
-// It subscribes to both header and data namespaces and processes incoming blobs.
+// It subscribes to header, data, and forced inclusion namespaces and processes incoming blobs.
 // Returns when subscription fails, context is cancelled, or node falls behind.
 func (s *Syncer) subscribeAndFollow() error {
 	// Get namespaces
@@ -472,6 +472,20 @@ func (s *Syncer) subscribeAndFollow() error {
 		}
 	}
 
+	// Subscribe to forced inclusion namespace if configured
+	var forcedInclusionCh <-chan *blobrpc.SubscriptionResponse
+	if s.daClient.HasForcedInclusionNamespace() {
+		fiNS := s.daClient.GetForcedInclusionNamespace()
+		// Only subscribe if it's different from both header and data namespaces
+		if !bytes.Equal(fiNS, headerNS) && !bytes.Equal(fiNS, dataNS) {
+			forcedInclusionCh, err = s.daClient.Subscribe(subCtx, fiNS)
+			if err != nil {
+				return fmt.Errorf("failed to subscribe to forced inclusion namespace: %w", err)
+			}
+			s.logger.Info().Msg("subscribed to forced inclusion namespace for follow mode")
+		}
+	}
+
 	s.logger.Info().Msg("subscribed to DA namespaces for follow mode")
 
 	// Calculate watchdog timeout
@@ -481,6 +495,7 @@ func (s *Syncer) subscribeAndFollow() error {
 	}
 
 	// Process subscription events
+	// Note: Select on a nil channel blocks forever, so nil channels are effectively disabled
 	for {
 		select {
 		case <-s.ctx.Done():
@@ -495,17 +510,26 @@ func (s *Syncer) subscribeAndFollow() error {
 			}
 
 		case resp, ok := <-dataCh:
-			if dataCh == nil {
-				// Data channel not used (same namespace), continue
-				continue
-			}
+			// Note: if dataCh is nil (same namespace as header), this case never fires
 			if !ok {
 				return errors.New("data subscription closed")
 			}
 			if err := s.processSubscriptionResponse(resp); err != nil {
 				s.logger.Error().Err(err).Uint64("height", resp.Height).Msg("failed to process data subscription")
 			}
 
+		case resp, ok := <-forcedInclusionCh:
+			// Note: if forcedInclusionCh is nil (not configured), this case never fires
+			if !ok {
+				return errors.New("forced inclusion subscription closed")
+			}
+			// Forced inclusion responses are logged but not processed through processSubscriptionResponse
+			// They are handled separately by the forced inclusion retriever during block verification
+			s.logger.Debug().
+				Uint64("da_height", resp.Height).
+				Int("blobs", len(resp.Blobs)).
+				Msg("received forced inclusion subscription notification")
+
 		case <-time.After(watchdogTimeout):
 			// Watchdog: if no events for watchdogTimeout, recheck mode
 			// Might have fallen behind due to network issues
@@ -534,8 +558,8 @@ func (s *Syncer) processSubscriptionResponse(resp *blobrpc.SubscriptionResponse)
 		blobs[i] = blob.Data()
 	}
 
-	// Process blobs using the DA retriever's processBlobs method
-	events := s.daRetriever.(*daRetriever).processBlobs(s.ctx, blobs, resp.Height)
+	// Process blobs using the DA retriever's ProcessBlobs method
+	events := s.daRetriever.ProcessBlobs(s.ctx, blobs, resp.Height)
 
 	// Send events to the processing channel
 	for _, event := range events {

block/internal/syncing/syncer.go

@@ -859,23 +859,23 @@ func TestSyncer_modeSwitching(t *testing.T) {
 // mockCacheManager is a minimal implementation for testing
 type mockCacheManager struct{}
 
-func (m *mockCacheManager) DaHeight() uint64                        { return 0 }
+func (m *mockCacheManager) DaHeight() uint64 { return 0 }
 func (m *mockCacheManager) SetHeaderSeen(hash string, height uint64) {
 }
 func (m *mockCacheManager) IsHeaderSeen(hash string) bool { return false }
 func (m *mockCacheManager) SetDataSeen(hash string, height uint64) {
 }
-func (m *mockCacheManager) IsDataSeen(hash string) bool                            { return false }
+func (m *mockCacheManager) IsDataSeen(hash string) bool { return false }
 func (m *mockCacheManager) SetHeaderDAIncluded(hash string, daHeight, height uint64) {
 }
 func (m *mockCacheManager) GetHeaderDAIncluded(hash string) (uint64, bool) { return 0, false }
 func (m *mockCacheManager) RemoveHeaderDAIncluded(hash string)             {}
 func (m *mockCacheManager) SetDataDAIncluded(hash string, daHeight, height uint64) {
 }
-func (m *mockCacheManager) GetDataDAIncluded(hash string) (uint64, bool)             { return 0, false }
-func (m *mockCacheManager) IsTxSeen(hash string) bool                                { return false }
-func (m *mockCacheManager) SetTxSeen(hash string)                                    {}
-func (m *mockCacheManager) CleanupOldTxs(olderThan time.Duration) int                { return 0 }
+func (m *mockCacheManager) GetDataDAIncluded(hash string) (uint64, bool) { return 0, false }
+func (m *mockCacheManager) IsTxSeen(hash string) bool                    { return false }
+func (m *mockCacheManager) SetTxSeen(hash string)                        {}
+func (m *mockCacheManager) CleanupOldTxs(olderThan time.Duration) int    { return 0 }
 func (m *mockCacheManager) SetPendingEvent(height uint64, event *common.DAHeightEvent) {
 }
 func (m *mockCacheManager) GetNextPendingEvent(height uint64) *common.DAHeightEvent { return nil }