Review feedback

Author: alpe

apps/evm/cmd/run.go

@@ -60,7 +60,7 @@ var RunCmd = &cobra.Command{
 			return err
 		}
 
-		blobClient, err := blobrpc.NewWSClient(context.Background(), nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
+		blobClient, err := blobrpc.NewWSClient(cmd.Context(), nodeConfig.DA.Address, nodeConfig.DA.AuthToken, "")
 		if err != nil {
 			return fmt.Errorf("failed to create blob client: %w", err)
 		}

block/internal/syncing/da_follower.go

@@ -22,7 +22,7 @@ import (
 //   - 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,
+//   - catchupLoop sequentially retrieves from localNextDAHeight → highestSeenDAHeight,
 //     piping events to the Syncer's heightInCh.
 //
 // The two goroutines share only atomic state; no mutexes needed.
@@ -51,9 +51,9 @@ type daFollower struct {
 	// share the same namespace). When different, we subscribe to both and merge.
 	dataNamespace []byte
 
-	// localDAHeight is only written by catchupLoop and read by followLoop
+	// localNextDAHeight is only written by catchupLoop and read by followLoop
 	// to determine whether a catchup is needed.
-	localDAHeight atomic.Uint64
+	localNextDAHeight atomic.Uint64
 
 	// highestSeenDAHeight is written by followLoop and read by catchupLoop.
 	highestSeenDAHeight atomic.Uint64
@@ -62,7 +62,7 @@ type daFollower struct {
 	headReached atomic.Bool
 
 	// catchupSignal is sent by followLoop to wake catchupLoop when a new
-	// height is seen that is above localDAHeight.
+	// height is seen that is above localNextDAHeight.
 	catchupSignal chan struct{}
 
 	// daBlockTime is used as a backoff before retrying after errors.
@@ -101,7 +101,7 @@ func NewDAFollower(cfg DAFollowerConfig) DAFollower {
 		catchupSignal: make(chan struct{}, 1),
 		daBlockTime:   cfg.DABlockTime,
 	}
-	f.localDAHeight.Store(cfg.StartDAHeight)
+	f.localNextDAHeight.Store(cfg.StartDAHeight)
 	return f
 }
 
@@ -114,7 +114,7 @@ func (f *daFollower) Start(ctx context.Context) error {
 	go f.catchupLoop(ctx)
 
 	f.logger.Info().
-		Uint64("start_da_height", f.localDAHeight.Load()).
+		Uint64("start_da_height", f.localNextDAHeight.Load()).
 		Msg("DA follower started")
 	return nil
 }
@@ -142,7 +142,7 @@ func (f *daFollower) signalCatchup() {
 }
 
 // followLoop subscribes to DA blob events and keeps highestSeenDAHeight up to date.
-// When a new height appears above localDAHeight, it wakes the catchup loop.
+// When a new height appears above localNextDAHeight, it wakes the catchup loop.
 func (f *daFollower) followLoop(ctx context.Context) {
 	defer f.wg.Done()
 
@@ -244,11 +244,11 @@ func (f *daFollower) mergeSubscriptions(
 }
 
 // handleSubscriptionEvent processes a subscription event. When the follower is
-// caught up (ev.Height == localDAHeight) and blobs are available, it processes
+// caught up (ev.Height == localNextDAHeight) 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.
 //
-// Uses CAS on localDAHeight to claim exclusive access to processBlobs,
+// Uses CAS on localNextDAHeight to claim exclusive access to processBlobs,
 // preventing concurrent map access with catchupLoop.
 func (f *daFollower) handleSubscriptionEvent(ctx context.Context, ev datypes.SubscriptionEvent) {
 	// Always record the highest height we've seen from the subscription.
@@ -257,24 +257,26 @@ func (f *daFollower) handleSubscriptionEvent(ctx context.Context, ev datypes.Sub
 	// Fast path: try to claim this height for inline processing.
 	// CAS(N, N+1) ensures only one goroutine (followLoop or catchupLoop)
 	// can enter processBlobs for height N.
-	if len(ev.Blobs) > 0 && f.localDAHeight.CompareAndSwap(ev.Height, ev.Height+1) {
+	if len(ev.Blobs) > 0 && f.localNextDAHeight.CompareAndSwap(ev.Height, ev.Height+1) {
 		events := f.retriever.ProcessBlobs(ctx, ev.Blobs, ev.Height)
 		for _, event := range events {
 			if err := f.pipeEvent(ctx, event); err != nil {
 				// Roll back so catchupLoop can retry this height.
-				f.localDAHeight.Store(ev.Height)
+				f.localNextDAHeight.Store(ev.Height)
 				f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
 					Msg("failed to pipe inline event, catchup will retry")
 				return
 			}
 		}
 		if len(events) != 0 {
-			f.headReached.Store(true)
+			if !f.headReached.Load() && f.localNextDAHeight.Load() > f.highestSeenDAHeight.Load() {
+				f.headReached.Store(true)
+			}
 			f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
 				Msg("processed subscription blobs inline (fast path)")
 		} else {
 			// No complete events (split namespace, waiting for other half).
-			f.localDAHeight.Store(ev.Height)
+			f.localNextDAHeight.Store(ev.Height)
 		}
 		return
 	}
@@ -297,7 +299,7 @@ func (f *daFollower) updateHighest(height uint64) {
 }
 
 // catchupLoop waits for signals and sequentially retrieves DA heights
-// from localDAHeight up to highestSeenDAHeight.
+// from localNextDAHeight up to highestSeenDAHeight.
 func (f *daFollower) catchupLoop(ctx context.Context) {
 	defer f.wg.Done()
 
@@ -314,7 +316,7 @@ func (f *daFollower) catchupLoop(ctx context.Context) {
 	}
 }
 
-// runCatchup sequentially retrieves from localDAHeight to highestSeenDAHeight.
+// runCatchup sequentially retrieves from localNextDAHeight to highestSeenDAHeight.
 // It handles priority heights first, then sequential heights.
 func (f *daFollower) runCatchup(ctx context.Context) {
 	for {
@@ -324,8 +326,8 @@ func (f *daFollower) runCatchup(ctx context.Context) {
 
 		// Check for priority heights from P2P hints first.
 		if priorityHeight := f.retriever.PopPriorityHeight(); priorityHeight > 0 {
-			currentHeight := f.localDAHeight.Load()
-			if priorityHeight < currentHeight {
+			nextHeight := f.localNextDAHeight.Load()
+			if priorityHeight < nextHeight {
 				continue
 			}
 			f.logger.Debug().
@@ -340,7 +342,7 @@ func (f *daFollower) runCatchup(ctx context.Context) {
 		}
 
 		// Sequential catchup.
-		local := f.localDAHeight.Load()
+		local := f.localNextDAHeight.Load()
 		highest := f.highestSeenDAHeight.Load()
 
 		if local > highest {
@@ -350,14 +352,14 @@ func (f *daFollower) runCatchup(ctx context.Context) {
 		}
 
 		// CAS claims this height prevents followLoop from inline-processing
-		if !f.localDAHeight.CompareAndSwap(local, local+1) {
+		if !f.localNextDAHeight.CompareAndSwap(local, local+1) {
 			// followLoop already advanced past this height via inline processing.
 			continue
 		}
 
 		if err := f.fetchAndPipeHeight(ctx, local); err != nil {
 			// Roll back so we can retry after backoff.
-			f.localDAHeight.Store(local)
+			f.localNextDAHeight.Store(local)
 			if !f.waitOnCatchupError(ctx, err, local) {
 				return
 			}

block/internal/syncing/da_retriever.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"slices"
 	"sync"
+	"sync/atomic"
 
 	"github.com/rs/zerolog"
 	"google.golang.org/protobuf/proto"
@@ -43,12 +44,13 @@ type daRetriever struct {
 
 	// transient cache, only full event need to be passed to the syncer
 	// on restart, will be refetch as da height is updated by syncer
+	pendingMu      sync.Mutex
 	pendingHeaders map[uint64]*types.SignedHeader
 	pendingData    map[uint64]*types.Data
 
 	// strictMode indicates if the node has seen a valid DAHeaderEnvelope
 	// and should now reject all legacy/unsigned headers.
-	strictMode bool
+	strictMode atomic.Bool
 
 	// priorityMu protects priorityHeights from concurrent access
 	priorityMu sync.Mutex
@@ -71,7 +73,6 @@ func NewDARetriever(
 		logger:          logger.With().Str("component", "da_retriever").Logger(),
 		pendingHeaders:  make(map[uint64]*types.SignedHeader),
 		pendingData:     make(map[uint64]*types.Data),
-		strictMode:      false,
 		priorityHeights: make([]uint64, 0),
 	}
 }
@@ -198,41 +199,55 @@ func (r *daRetriever) validateBlobResponse(res datypes.ResultRetrieve, daHeight
 // This is the public interface used by the DAFollower for inline subscription processing.
 //
 // NOT thread-safe: the caller (DAFollower) must ensure exclusive access via CAS
-// on localDAHeight before calling this method.
+// on localNextDAHeight before calling this method.
 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
+	var decodedHeaders []*types.SignedHeader
+	var decodedData []*types.Data
+
+	// Decode all blobs first without holding the lock
 	for _, bz := range blobs {
 		if len(bz) == 0 {
 			continue
 		}
 
 		if header := r.tryDecodeHeader(bz, daHeight); header != nil {
-			if _, ok := r.pendingHeaders[header.Height()]; ok {
-				// a (malicious) node may have re-published valid header to another da height (should never happen)
-				// we can already discard it, only the first one is valid
-				r.logger.Debug().Uint64("height", header.Height()).Uint64("da_height", daHeight).Msg("header blob already exists for height, discarding")
-				continue
-			}
-
-			r.pendingHeaders[header.Height()] = header
+			decodedHeaders = append(decodedHeaders, header)
 			continue
 		}
 
 		if data := r.tryDecodeData(bz, daHeight); data != nil {
-			if _, ok := r.pendingData[data.Height()]; ok {
-				// a (malicious) node may have re-published valid data to another da height (should never happen)
-				// we can already discard it, only the first one is valid
-				r.logger.Debug().Uint64("height", data.Height()).Uint64("da_height", daHeight).Msg("data blob already exists for height, discarding")
-				continue
-			}
+			decodedData = append(decodedData, data)
+		}
+	}
 
-			r.pendingData[data.Height()] = data
+	r.pendingMu.Lock()
+	defer r.pendingMu.Unlock()
+
+	for _, header := range decodedHeaders {
+		if _, ok := r.pendingHeaders[header.Height()]; ok {
+			// a (malicious) node may have re-published valid header to another da height (should never happen)
+			// we can already discard it, only the first one is valid
+			r.logger.Debug().Uint64("height", header.Height()).Uint64("da_height", daHeight).Msg("header blob already exists for height, discarding")
+			continue
+		}
+
+		r.pendingHeaders[header.Height()] = header
+	}
+
+	for _, data := range decodedData {
+		if _, ok := r.pendingData[data.Height()]; ok {
+			// a (malicious) node may have re-published valid data to another da height (should never happen)
+			// we can already discard it, only the first one is valid
+			r.logger.Debug().Uint64("height", data.Height()).Uint64("da_height", daHeight).Msg("data blob already exists for height, discarding")
+			continue
 		}
+
+		r.pendingData[data.Height()] = data
 	}
 
 	var events []common.DAHeightEvent
@@ -294,7 +309,7 @@ func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 	// Attempt to unmarshal as DAHeaderEnvelope and get the envelope signature
 	if envelopeSignature, err := header.UnmarshalDAEnvelope(bz); err != nil {
 		// If in strict mode, we REQUIRE an envelope.
-		if r.strictMode {
+		if r.strictMode.Load() {
 			r.logger.Warn().Err(err).Msg("strict mode is enabled, rejecting non-envelope blob")
 			return nil
 		}
@@ -328,14 +343,14 @@ func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 			isValidEnvelope = true
 		}
 	}
-	if r.strictMode && !isValidEnvelope {
+	if r.strictMode.Load() && !isValidEnvelope {
 		r.logger.Warn().Msg("strict mode: rejecting block that is not a fully valid envelope")
 		return nil
 	}
 	// Mode Switch Logic
-	if isValidEnvelope && !r.strictMode {
+	if isValidEnvelope && !r.strictMode.Load() {
 		r.logger.Info().Uint64("height", header.Height()).Msg("valid DA envelope detected, switching to STRICT MODE")
-		r.strictMode = true
+		r.strictMode.Store(true)
 	}
 
 	// Legacy blob support implies: strictMode == false AND (!isValidEnvelope).

block/internal/syncing/syncer.go

@@ -79,7 +79,6 @@ type Syncer struct {
 	p2pHandler    p2pHandler
 	raftRetriever *raftRetriever
 
-	// DA follower (replaces the old polling daWorkerLoop)
 	daFollower DAFollower
 
 	// Forced inclusion tracking
@@ -209,6 +208,8 @@ func (s *Syncer) Start(ctx context.Context) error {
 		DABlockTime:   s.config.DA.BlockTime.Duration,
 	})
 	if err := s.daFollower.Start(ctx); err != nil {
+		s.cancel()
+		s.wg.Wait()
 		return fmt.Errorf("failed to start DA follower: %w", err)
 	}
 
@@ -394,8 +395,6 @@ func (s *Syncer) HasReachedDAHead() bool {
 	return false
 }
 
-// fetchDAUntilCaughtUp was removed — the DAFollower handles this concern.
-
 // PendingCount returns the number of unprocessed height events in the pipeline.
 func (s *Syncer) PendingCount() int {
 	return len(s.heightInCh)

block/internal/syncing/syncer_backoff_test.go

@@ -255,7 +255,7 @@ func TestDAFollower_CatchupThenReachHead(t *testing.T) {
 }
 
 // TestDAFollower_InlineProcessing verifies the fast path: when the subscription
-// delivers blobs at the current localDAHeight, handleSubscriptionEvent processes
+// delivers blobs at the current localNextDAHeight, 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) {
@@ -285,7 +285,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 		daRetriever.On("ProcessBlobs", mock.Anything, blobs, uint64(10)).
 			Return(expectedEvents).Once()
 
-		// Simulate subscription event at the current localDAHeight
+		// Simulate subscription event at the current localNextDAHeight
 		follower.handleSubscriptionEvent(t.Context(), datypes.SubscriptionEvent{
 			Height: 10,
 			Blobs:  blobs,
@@ -294,7 +294,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 		// 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.Equal(t, uint64(11), follower.localNextDAHeight.Load(), "localNextDAHeight should advance past processed height")
 		assert.True(t, follower.HasReachedHead(), "should mark head as reached after inline processing")
 	})
 
@@ -324,7 +324,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 
 		// 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.localNextDAHeight.Load(), "localNextDAHeight should not change")
 		assert.Equal(t, uint64(15), follower.highestSeenDAHeight.Load(), "highestSeen should be updated")
 	})
 
@@ -350,7 +350,7 @@ func TestDAFollower_InlineProcessing(t *testing.T) {
 
 		// 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.localNextDAHeight.Load(), "localNextDAHeight should not change")
 		assert.Equal(t, uint64(10), follower.highestSeenDAHeight.Load(), "highestSeen should be updated")
 	})
 }