Update

Author: alpe

apps/evm/server/force_inclusion_test.go

@@ -18,7 +18,8 @@ import (
 
 // mockDA implements block/internal/da.Client for testing
 type mockDA struct {
-	submitFunc func(ctx context.Context, data [][]byte, gasPrice float64, namespace []byte, options []byte) da.ResultSubmit
+	submitFunc    func(ctx context.Context, data [][]byte, gasPrice float64, namespace []byte, options []byte) da.ResultSubmit
+	subscribeFunc func(ctx context.Context, namespace []byte) (<-chan da.ResultRetrieve, error)
 }
 
 func (m *mockDA) Submit(ctx context.Context, data [][]byte, gasPrice float64, namespace []byte, options []byte) da.ResultSubmit {
@@ -29,6 +30,13 @@ func (m *mockDA) Submit(ctx context.Context, data [][]byte, gasPrice float64, na
 	return da.ResultSubmit{BaseResult: da.BaseResult{Code: da.StatusSuccess, Height: 1}}
 }
 
+func (m *mockDA) Subscribe(ctx context.Context, namespace []byte) (<-chan da.ResultRetrieve, error) {
+	if m.subscribeFunc != nil {
+		return m.subscribeFunc(ctx, namespace)
+	}
+	return nil, nil
+}
+
 func (m *mockDA) Retrieve(ctx context.Context, height uint64, namespace []byte) da.ResultRetrieve {
 	return da.ResultRetrieve{}
 }

block/internal/da/client.go

@@ -487,10 +487,10 @@ func (c *client) Subscribe(ctx context.Context, namespace []byte) (<-chan datype
 				// Using 0 or Now() is a trade-off. Let's use Now() for liveness.
 				outCh <- datypes.ResultRetrieve{
 					BaseResult: datypes.BaseResult{
-						Code:      datypes.StatusSuccess,
-						IDs:       ids,
-						Height:    resp.Height,
-						Timestamp: time.Now(),
+						Code:   datypes.StatusSuccess,
+						IDs:    ids,
+						Height: resp.Height,
+						//Timestamp: // TODO: set proper value
 					},
 					Data: data,
 				}

block/internal/syncing/da_retriever.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
 	"sync"
 
 	"github.com/rs/zerolog"
@@ -22,7 +23,7 @@ import (
 // DARetriever defines the interface for retrieving events from the DA layer
 type DARetriever interface {
 	RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error)
-	Subscribe(ctx context.Context) (<-chan common.DAHeightEvent, error)
+	Subscribe(ctx context.Context, ch chan common.DAHeightEvent) error
 }
 
 // daRetriever handles DA retrieval operations for syncing
@@ -78,65 +79,68 @@ func (r *daRetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]co
 	return r.processBlobs(ctx, blobsResp.Data, daHeight), nil
 }
 
-// Subscribe subscribes to specific DA namespace and returns a channel of height events
-func (r *daRetriever) Subscribe(ctx context.Context) (<-chan common.DAHeightEvent, error) {
-	// Subscribe to header namespace
-	// Note: We currently only subscribe to header namespace.
-	// If header and data namespaces are different, we might need to subscribe to both or fetch data on demand.
-	// For now, assuming header subscription is sufficient to trigger catchup or process full blocks if namespaces are same.
-	// Actually, if we follow the "Subscribe" API from recent changes, we likely want to subscribe to *headers*.
-	// However, processBlobs expects blobs.
-	// Let's subscribe to the header namespace.
-
-	// Use combined channel for output
-	outCh := make(chan common.DAHeightEvent, 100)
-
-	subCh, err := r.client.Subscribe(ctx, r.client.GetHeaderNamespace())
+// Subscribe subscribes to specific DA namespace
+func (r *daRetriever) Subscribe(ctx context.Context, outCh chan common.DAHeightEvent) error {
+	subChHeader, err := r.client.Subscribe(ctx, r.client.GetHeaderNamespace())
 	if err != nil {
-		return nil, fmt.Errorf("failed to subscribe to headers: %w", err)
+		return fmt.Errorf("subscribe to headers: %w", err)
+	}
+
+	var subChData <-chan datypes.ResultRetrieve
+	if !bytes.Equal(r.client.GetHeaderNamespace(), r.client.GetDataNamespace()) {
+		var err error
+		subChData, err = r.client.Subscribe(ctx, r.client.GetDataNamespace())
+		if err != nil {
+			return fmt.Errorf("subscribe to data: %w", err)
+		}
 	}
 
 	go func() {
 		defer close(outCh)
 		for {
+			var blobs [][]byte
+			var height uint64
+			var errCode datypes.StatusCode
+
 			select {
 			case <-ctx.Done():
 				return
-			case res, ok := <-subCh:
+			case res, ok := <-subChHeader:
 				if !ok {
 					return
 				}
-				if res.Code != datypes.StatusSuccess {
-					r.logger.Error().Uint64("code", uint64(res.Code)).Msg("subscription error")
+				blobs = res.Data
+				height = res.Height
+				errCode = res.Code
+			case res, ok := <-subChData:
+				if subChData == nil {
 					continue
 				}
+				if !ok {
+					return
+				}
+				blobs = res.Data
+				height = res.Height
+				errCode = res.Code
+			}
 
-				// We received some blobs (headers) via subscription.
-				// We process them similar to RetrieveFromDA, but we might need to fetch data if namespaces differ.
-				// If namespaces differ, data might not be in the subscription result.
-				// For the "Follow" case, we want to emit an event.
-				// If we have just headers, we might need to fetch data corresponding to these headers.
-
-				// However, `processBlobs` handles pending headers/data.
-				// If we only get headers, they will be stored in pendingHeaders.
-				// If we need data, we might need to fetch it.
-
-				// IMPORTANT: processBlobs takes "blobs" [][]byte. `res.Data` is []byte (aliases to Blob).
-				// So we can pass res.Data directly.
-
-				events := r.processBlobs(ctx, res.Data, res.Height)
-				for _, ev := range events {
-					select {
-					case <-ctx.Done():
-						return
-					case outCh <- ev:
-					}
+			if errCode != datypes.StatusSuccess {
+				r.logger.Error().Uint64("code", uint64(errCode)).Msg("subscription error")
+				continue
+			}
+
+			events := r.processBlobs(ctx, blobs, height)
+			for _, ev := range events {
+				select {
+				case <-ctx.Done():
+					return
+				case outCh <- ev:
 				}
 			}
 		}
 	}()
 
-	return outCh, nil
+	return nil
 }
 
 // fetchBlobs retrieves blobs from both header and data namespaces
@@ -279,18 +283,17 @@ func (r *daRetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 		events = append(events, event)
 	}
 
+	// Sort events by height to match execution order
+	sort.Slice(events, func(i, j int) bool {
+		if events[i].DaHeight != events[j].DaHeight {
+			return events[i].DaHeight < events[j].DaHeight
+		}
+		return events[i].Header.Height() < events[j].Header.Height()
+	})
+
 	if len(events) > 0 {
 		startHeight := events[0].Header.Height()
-		endHeight := events[0].Header.Height()
-		for _, event := range events {
-			h := event.Header.Height()
-			if h < startHeight {
-				startHeight = h
-			}
-			if h > endHeight {
-				endHeight = h
-			}
-		}
+		endHeight := events[len(events)-1].Header.Height()
 		r.logger.Info().Uint64("da_height", daHeight).Uint64("start_height", startHeight).Uint64("end_height", endHeight).Msg("processed blocks from DA")
 	}
 

block/internal/syncing/da_retriever_mock.go

@@ -374,3 +374,32 @@ func Test_isEmptyDataExpected(t *testing.T) {
 	h.DataHash = common.DataHashForEmptyTxs
 	assert.True(t, isEmptyDataExpected(h))
 }
+
+func TestDARetriever_ProcessBlobs_Sorting(t *testing.T) {
+	addr, pub, signer := buildSyncTestSigner(t)
+	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr}
+	r := newTestDARetriever(t, nil, config.DefaultConfig(), gen)
+
+	// Event A: Block Height 10
+	// Event B: Block Height 5
+	// Although DaHeight is currently identical for all events in a single processBlobs call,
+	// this test ensures that the secondary sort key (Block Height) behaves correctly.
+
+	data1Bin, data1 := makeSignedDataBytes(t, gen.ChainID, 10, addr, pub, signer, 1)
+	data2Bin, data2 := makeSignedDataBytes(t, gen.ChainID, 5, addr, pub, signer, 1)
+
+	hdr1Bin, _ := makeSignedHeaderBytes(t, gen.ChainID, 10, addr, pub, signer, nil, &data1.Data, nil)
+	hdr2Bin, _ := makeSignedHeaderBytes(t, gen.ChainID, 5, addr, pub, signer, nil, &data2.Data, nil)
+
+	// Process blobs.
+	daHeight := uint64(100)
+	// We pass them in mixed order to ensure sorting happens.
+	events := r.processBlobs(context.Background(), [][]byte{hdr1Bin, data1Bin, hdr2Bin, data2Bin}, daHeight)
+
+	require.Len(t, events, 2)
+	assert.Equal(t, uint64(5), events[0].Header.Height(), "Events should be sorted by block height asc")
+	assert.Equal(t, uint64(10), events[1].Header.Height())
+
+	assert.Equal(t, daHeight, events[0].DaHeight)
+	assert.Equal(t, daHeight, events[1].DaHeight)
+}

block/internal/syncing/da_retriever_test.go

@@ -186,9 +186,7 @@ func (s *Syncer) Start(ctx context.Context) error {
 	}
 
 	// Initialize handlers
-	if s.daRetriever == nil {
-		s.daRetriever = NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
-	}
+	s.daRetriever = NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
 	s.fiRetriever = da.NewForcedInclusionRetriever(s.daClient, s.logger, s.genesis.DAStartHeight, s.genesis.DAEpochForcedInclusion)
 	s.p2pHandler = NewP2PHandler(s.headerStore.Store(), s.dataStore.Store(), s.cache, s.genesis, s.logger)
 	if currentHeight, err := s.store.Height(s.ctx); err != nil {
@@ -318,17 +316,15 @@ func (s *Syncer) daWorkerLoop() {
 
 	s.logger.Info().Msg("starting DA worker")
 	defer s.logger.Info().Msg("DA worker stopped")
+	backoff := s.config.DA.BlockTime.Duration
+	if backoff <= 0 {
+		backoff = 2 * time.Second
+	}
 
 	for {
 		// 1. Catch up mode: fetch sequentially until we are up to date
 		if err := s.fetchDAUntilCaughtUp(); err != nil {
 			s.logger.Error().Err(err).Msg("DA catchup failed, retrying after backoff")
-
-			backoff := s.config.DA.BlockTime.Duration
-			if backoff <= 0 {
-				backoff = 2 * time.Second
-			}
-
 			select {
 			case <-s.ctx.Done():
 				return
@@ -343,25 +339,30 @@ func (s *Syncer) daWorkerLoop() {
 			if errors.Is(err, context.Canceled) {
 				return
 			}
-			s.logger.Warn().Err(err).Msg("DA follow disrupted, switching to catchup")
+			s.logger.Warn().Err(err).Msg("DA follow disrupted")
 			// We don't need explicit backoff here as we'll switch to catchup immediately,
 			// checking if there are new blocks to fetch.
 		}
+		s.logger.Info().Msg("DA follow mode resumed")
 	}
 }
 
 // followDA subscribes to DA events and processes them until a gap is detected or error occurs
 func (s *Syncer) followDA() error {
 	s.logger.Info().Msg("entering DA follow mode")
-	subCh, err := s.daRetriever.Subscribe(s.ctx)
-	if err != nil {
-		return fmt.Errorf("failed to subscribe to DA: %w", err)
+
+	ctx, cancel := context.WithCancel(s.ctx)
+	defer cancel()
+	subCh := make(chan common.DAHeightEvent, 1)
+
+	if err := s.daRetriever.Subscribe(ctx, subCh); err != nil {
+		return fmt.Errorf("subscribe to DA: %w", err)
 	}
 
 	for {
 		select {
-		case <-s.ctx.Done():
-			return s.ctx.Err()
+		case <-ctx.Done():
+			return ctx.Err()
 		case event, ok := <-subCh:
 			if !ok {
 				return errors.New("DA subscription channel closed")