align subscribe with forced inclusion

Author: tac0turtle

block/internal/common/subscription.go

@@ -0,0 +1,24 @@
+package common
+
+import blobrpc "github.com/evstack/ev-node/pkg/da/jsonrpc"
+
+// BlobsFromSubscription returns non-empty blob data from a subscription response.
+func BlobsFromSubscription(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
+}

block/internal/da/async_block_retriever.go

@@ -25,6 +25,7 @@ type AsyncBlockRetriever interface {
 	Stop()
 	GetCachedBlock(ctx context.Context, daHeight uint64) (*BlockData, error)
 	UpdateCurrentHeight(height uint64)
+	StoreBlock(ctx context.Context, height uint64, blobs [][]byte, timestamp time.Time)
 }
 
 // BlockData contains data retrieved from a single DA height
@@ -125,6 +126,68 @@ func (f *asyncBlockRetriever) UpdateCurrentHeight(height uint64) {
 	}
 }
 
+// StoreBlock caches a block's blobs, favoring existing data to avoid churn.
+func (f *asyncBlockRetriever) StoreBlock(ctx context.Context, height uint64, blobs [][]byte, timestamp time.Time) {
+	if len(f.namespace) == 0 {
+		return
+	}
+	if height < f.daStartHeight {
+		return
+	}
+	if len(blobs) == 0 {
+		return
+	}
+
+	filtered := make([][]byte, 0, len(blobs))
+	for _, blob := range blobs {
+		if len(blob) > 0 {
+			filtered = append(filtered, blob)
+		}
+	}
+	if len(filtered) == 0 {
+		return
+	}
+
+	if timestamp.IsZero() {
+		timestamp = time.Now().UTC()
+	}
+
+	key := newBlockDataKey(height)
+	if existing, err := f.cache.Get(ctx, key); err == nil {
+		var pbBlock pb.BlockData
+		if err := proto.Unmarshal(existing, &pbBlock); err == nil && len(pbBlock.Blobs) > 0 {
+			return
+		}
+	}
+
+	pbBlock := &pb.BlockData{
+		Height:    height,
+		Timestamp: timestamp.Unix(),
+		Blobs:     filtered,
+	}
+	data, err := proto.Marshal(pbBlock)
+	if err != nil {
+		f.logger.Error().
+			Err(err).
+			Uint64("height", height).
+			Msg("failed to marshal block for caching")
+		return
+	}
+
+	if err := f.cache.Put(ctx, key, data); err != nil {
+		f.logger.Error().
+			Err(err).
+			Uint64("height", height).
+			Msg("failed to cache block")
+		return
+	}
+
+	f.logger.Debug().
+		Uint64("height", height).
+		Int("blob_count", len(filtered)).
+		Msg("cached block from subscription")
+}
+
 func newBlockDataKey(height uint64) ds.Key {
 	return ds.NewKey(fmt.Sprintf("/block/%d", height))
 }

block/internal/da/forced_inclusion_retriever.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
 
+	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/pkg/config"
+	blobrpc "github.com/evstack/ev-node/pkg/da/jsonrpc"
 	datypes "github.com/evstack/ev-node/pkg/da/types"
 	"github.com/evstack/ev-node/types"
 )
@@ -23,6 +26,12 @@ type ForcedInclusionRetriever struct {
 	daEpochSize   uint64
 	daStartHeight uint64
 	asyncFetcher  AsyncBlockRetriever
+
+	mu                    sync.Mutex
+	pendingEpochStart     uint64
+	pendingEpochEnd       uint64
+	lastProcessedEpochEnd uint64
+	hasProcessedEpoch     bool
 }
 
 // ForcedInclusionEvent contains forced inclusion transactions retrieved from DA.
@@ -68,6 +77,25 @@ func (r *ForcedInclusionRetriever) Stop() {
 	r.asyncFetcher.Stop()
 }
 
+// HandleSubscriptionResponse caches forced inclusion blobs from subscription updates.
+func (r *ForcedInclusionRetriever) HandleSubscriptionResponse(resp *blobrpc.SubscriptionResponse) {
+	if resp == nil {
+		return
+	}
+	if !r.client.HasForcedInclusionNamespace() {
+		return
+	}
+
+	r.asyncFetcher.UpdateCurrentHeight(resp.Height)
+
+	blobs := common.BlobsFromSubscription(resp)
+	if len(blobs) == 0 {
+		return
+	}
+
+	r.asyncFetcher.StoreBlock(context.Background(), resp.Height, blobs, time.Now().UTC())
+}
+
 // RetrieveForcedIncludedTxs retrieves forced inclusion transactions at the given DA height.
 // It respects epoch boundaries and only fetches at epoch end.
 // It tries to get blocks from the async fetcher cache first, then falls back to sync fetching.
@@ -86,12 +114,50 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 	// Update the async fetcher's current height so it knows what to prefetch
 	r.asyncFetcher.UpdateCurrentHeight(daHeight)
 
+	r.mu.Lock()
+	pendingStart := r.pendingEpochStart
+	pendingEnd := r.pendingEpochEnd
+	lastProcessed := r.lastProcessedEpochEnd
+	hasProcessed := r.hasProcessedEpoch
+	r.mu.Unlock()
+
+	if pendingEnd != 0 {
+		if daHeight < pendingEnd {
+			return &ForcedInclusionEvent{
+				StartDaHeight: daHeight,
+				EndDaHeight:   daHeight,
+				Txs:           [][]byte{},
+			}, nil
+		}
+
+		event, err := r.retrieveEpoch(ctx, pendingStart, pendingEnd)
+		if err != nil {
+			return nil, err
+		}
+
+		r.mu.Lock()
+		r.pendingEpochStart = 0
+		r.pendingEpochEnd = 0
+		r.lastProcessedEpochEnd = pendingEnd
+		r.hasProcessedEpoch = true
+		r.mu.Unlock()
+
+		return event, nil
+	}
+
 	if daHeight != epochEnd {
 		r.logger.Debug().
 			Uint64("da_height", daHeight).
 			Uint64("epoch_end", epochEnd).
 			Msg("not at epoch end - returning empty transactions")
+		return &ForcedInclusionEvent{
+			StartDaHeight: daHeight,
+			EndDaHeight:   daHeight,
+			Txs:           [][]byte{},
+		}, nil
+	}
 
+	if hasProcessed && epochEnd <= lastProcessed {
 		return &ForcedInclusionEvent{
 			StartDaHeight: daHeight,
 			EndDaHeight:   daHeight,
@@ -106,6 +172,24 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 		Uint64("epoch_num", currentEpochNumber).
 		Msg("retrieving forced included transactions from DA epoch")
 
+	event, err := r.retrieveEpoch(ctx, epochStart, epochEnd)
+	if err != nil {
+		r.mu.Lock()
+		r.pendingEpochStart = epochStart
+		r.pendingEpochEnd = epochEnd
+		r.mu.Unlock()
+		return nil, err
+	}
+
+	r.mu.Lock()
+	r.lastProcessedEpochEnd = epochEnd
+	r.hasProcessedEpoch = true
+	r.mu.Unlock()
+
+	return event, nil
+}
+
+func (r *ForcedInclusionRetriever) retrieveEpoch(ctx context.Context, epochStart, epochEnd uint64) (*ForcedInclusionEvent, error) {
 	event := &ForcedInclusionEvent{
 		StartDaHeight: epochStart,
 		EndDaHeight:   epochEnd,
@@ -211,18 +295,12 @@ func (r *ForcedInclusionRetriever) RetrieveForcedIncludedTxs(ctx context.Context
 	// any error during process, need to retry at next call
 	if processErrs != nil {
 		r.logger.Warn().
-			Uint64("da_height", daHeight).
 			Uint64("epoch_start", epochStart).
 			Uint64("epoch_end", epochEnd).
-			Uint64("epoch_num", currentEpochNumber).
 			Err(processErrs).
-			Msg("Failed to retrieve DA epoch.. retrying next iteration")
+			Msg("failed to retrieve DA epoch")
 
-		return &ForcedInclusionEvent{
-			StartDaHeight: daHeight,
-			EndDaHeight:   daHeight,
-			Txs:           [][]byte{},
-		}, nil
+		return nil, processErrs
 	}
 
 	return event, nil

block/internal/da/forced_inclusion_retriever_test.go

@@ -224,11 +224,9 @@ func TestForcedInclusionRetriever_RetrieveForcedIncludedTxs_ErrorHandling(t *tes
 	defer retriever.Stop()
 	ctx := context.Background()
 
-	// Should return empty event with no error (errors are logged and retried later)
-	event, err := retriever.RetrieveForcedIncludedTxs(ctx, 100)
-	assert.NilError(t, err)
-	assert.Assert(t, event != nil)
-	assert.Equal(t, len(event.Txs), 0)
+	// Should return error so caller can retry without skipping the epoch
+	_, err := retriever.RetrieveForcedIncludedTxs(ctx, 100)
+	assert.Assert(t, err != nil)
 }
 
 func TestForcedInclusionRetriever_RetrieveForcedIncludedTxs_EmptyBlobsSkipped(t *testing.T) {

block/internal/syncing/syncer.go

@@ -254,6 +254,9 @@ func (s *Syncer) Stop() error {
 	if s.cancel != nil {
 		s.cancel()
 	}
+	if s.fiRetriever != nil {
+		s.fiRetriever.Stop()
+	}
 	s.cancelP2PWait(0)
 	s.wg.Wait()
 	s.logger.Info().Msg("syncer stopped")
@@ -524,12 +527,10 @@ func (s *Syncer) subscribeAndFollow() error {
 			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")
+			// Cache forced inclusion blobs for epoch retrieval.
+			if s.fiRetriever != nil {
+				s.fiRetriever.HandleSubscriptionResponse(resp)
+			}
 
 		case <-time.After(watchdogTimeout):
 			// Watchdog: if no events for watchdogTimeout, recheck mode
@@ -544,7 +545,7 @@ func (s *Syncer) subscribeAndFollow() error {
 
 // processSubscriptionResponse processes a subscription response and sends events to the processing channel.
 func (s *Syncer) processSubscriptionResponse(resp *blobrpc.SubscriptionResponse) error {
-	if resp == nil || len(resp.Blobs) == 0 {
+	if resp == nil {
 		return nil
 	}
 
@@ -554,9 +555,9 @@ func (s *Syncer) processSubscriptionResponse(resp *blobrpc.SubscriptionResponse)
 		Msg("processing subscription response")
 
 	// Convert blobs to raw byte slices for processing
-	blobs := make([][]byte, len(resp.Blobs))
-	for i, blob := range resp.Blobs {
-		blobs[i] = blob.Data()
+	blobs := common.BlobsFromSubscription(resp)
+	if len(blobs) == 0 {
+		return nil
 	}
 
 	// Process blobs using the DA retriever's ProcessBlobs method
@@ -836,7 +837,11 @@ func (s *Syncer) trySyncNextBlock(event *common.DAHeightEvent) error {
 
 	// Verify forced inclusion transactions if configured
 	if event.Source == common.SourceDA {
-		if err := s.verifyForcedInclusionTxs(currentState, data); err != nil {
+		verifyState := currentState
+		if event.DaHeight > verifyState.DAHeight {
+			verifyState.DAHeight = event.DaHeight
+		}
+		if err := s.verifyForcedInclusionTxs(verifyState, data); err != nil {
 			s.logger.Error().Err(err).Uint64("height", nextHeight).Msg("forced inclusion verification failed")
 			if errors.Is(err, errMaliciousProposer) {
 				s.cache.RemoveHeaderDAIncluded(headerHash)

block/internal/syncing/syncer_forced_inclusion_test.go

@@ -851,18 +851,6 @@ func TestVerifyForcedInclusionTxs_DeferralWithinEpoch(t *testing.T) {
 		return true
 	})
 
-	// Mock DA for second verification at same epoch (height 104 - epoch end)
-	for height := uint64(101); height <= 104; height++ {
-		client.On("Retrieve", mock.Anything, height, []byte("nsForcedInclusion")).Return(datypes.ResultRetrieve{
-			BaseResult: datypes.BaseResult{Code: datypes.StatusNotFound, Timestamp: time.Now()},
-		}).Once()
-	}
-
-	client.On("Retrieve", mock.Anything, uint64(100), []byte("nsForcedInclusion")).Return(datypes.ResultRetrieve{
-		BaseResult: datypes.BaseResult{Code: datypes.StatusSuccess, IDs: [][]byte{[]byte("fi1"), []byte("fi2")}, Timestamp: time.Now()},
-		Data:       [][]byte{dataBin1, dataBin2},
-	}).Once()
-
 	// Second block includes BOTH the previously included dataBin1 AND the deferred dataBin2
 	// This simulates the block containing both forced inclusion txs
 	data2 := makeData(gen.ChainID, 2, 2)