Updates - WIP

Author: alpe

block/internal/da/async_block_retriever.go

@@ -88,8 +88,8 @@ func NewAsyncBlockRetriever(
 		DABlockTime:         daBlockTime,
 		Handler:             f,
 		FetchBlockTimestamp: true,
+		StartHeight:         daStartHeight,
 	})
-	f.subscriber.SetStartHeight(daStartHeight)
 
 	return f
 }
@@ -173,8 +173,12 @@ func (f *asyncBlockRetriever) GetCachedBlock(ctx context.Context, daHeight uint6
 
 // HandleEvent caches blobs from the subscription inline, even empty ones,
 // to record that the DA height was seen and has 0 blobs.
-func (f *asyncBlockRetriever) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent) {
+func (f *asyncBlockRetriever) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error {
 	f.cacheBlock(ctx, ev.Height, ev.Timestamp, ev.Blobs)
+	if isInline {
+		return errors.New("async block retriever relies on catchup state machine")
+	}
+	return nil
 }
 
 // HandleCatchup fetches a single height via Retrieve and caches it.

block/internal/da/subscriber.go

@@ -17,13 +17,15 @@ import (
 // SubscriberHandler is the callback interface for subscription consumers.
 // Implementations drive the consumer-specific logic (caching, piping events, etc.).
 type SubscriberHandler interface {
-	// HandleEvent processes a subscription event inline (fast path).
-	// Called on the followLoop goroutine for each subscription event.
-	HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent)
+	// HandleEvent processes a subscription event.
+	// isInline is true if the subscriber successfully claimed this height (via CAS).
+	// Returning an error when isInline is true instructs the Subscriber to roll back the localDAHeight.
+	HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error
 
 	// HandleCatchup is called for each height during sequential catchup.
-	// The subscriber advances localDAHeight only after this returns nil.
-	// Returning an error triggers a backoff retry.
+	// The subscriber advances localDAHeight only after this returns (true, nil).
+	// Returning (false, nil) rolls back localDAHeight without triggering a backoff.
+	// Returning an error rolls back localDAHeight and triggers a backoff retry.
 	HandleCatchup(ctx context.Context, height uint64) error
 }
 
@@ -36,6 +38,8 @@ type SubscriberConfig struct {
 	Handler     SubscriberHandler
 	// Deprecated: Remove with https://github.com/evstack/ev-node/issues/3142
 	FetchBlockTimestamp bool // the timestamp comes with an extra api call before Celestia v0.29.1-mocha.
+
+	StartHeight uint64 // initial localDAHeight
 }
 
 // Subscriber is a shared DA subscription primitive that encapsulates the
@@ -87,17 +91,15 @@ func NewSubscriber(cfg SubscriberConfig) *Subscriber {
 		daBlockTime:         cfg.DABlockTime,
 		fetchBlockTimestamp: cfg.FetchBlockTimestamp,
 	}
+	s.localDAHeight.Store(cfg.StartHeight)
+	s.catchupSignal <- struct{}{}
+
 	if len(s.namespaces) == 0 {
 		s.logger.Warn().Msg("no namespaces configured, subscriber will stay idle")
 	}
 	return s
 }
 
-// SetStartHeight sets the initial local DA height before Start is called.
-func (s *Subscriber) SetStartHeight(height uint64) {
-	s.localDAHeight.Store(height)
-}
-
 // Start begins the follow and catchup goroutines.
 func (s *Subscriber) Start(ctx context.Context) error {
 	if len(s.namespaces) == 0 {
@@ -108,6 +110,7 @@ func (s *Subscriber) Start(ctx context.Context) error {
 	s.wg.Add(2)
 	go s.followLoop(ctx)
 	go s.catchupLoop(ctx)
+
 	return nil
 }
 
@@ -134,29 +137,6 @@ func (s *Subscriber) HasReachedHead() bool {
 	return s.headReached.Load()
 }
 
-// CompareAndSwapLocalHeight attempts a CAS on localDAHeight.
-// Used by handlers that want to claim exclusive processing of a height.
-func (s *Subscriber) CompareAndSwapLocalHeight(old, new uint64) bool {
-	return s.localDAHeight.CompareAndSwap(old, new)
-}
-
-// SetLocalHeight stores a new localDAHeight value.
-func (s *Subscriber) SetLocalHeight(height uint64) {
-	s.localDAHeight.Store(height)
-}
-
-// UpdateHighestForTest directly sets the highest seen DA height and signals catchup.
-// Only for use in tests that bypass the subscription loop.
-func (s *Subscriber) UpdateHighestForTest(height uint64) {
-	s.updateHighest(height)
-}
-
-// RunCatchupForTest runs a single catchup pass. Only for use in tests that
-// bypass the catchup loop's signal-wait mechanism.
-func (s *Subscriber) RunCatchupForTest(ctx context.Context) {
-	s.runCatchup(ctx)
-}
-
 // ---------------------------------------------------------------------------
 // Follow loop
 // ---------------------------------------------------------------------------
@@ -215,7 +195,21 @@ func (s *Subscriber) runSubscription(ctx context.Context) error {
 				return errors.New("subscription channel closed")
 			}
 			s.updateHighest(ev.Height)
-			s.handler.HandleEvent(ctx, ev)
+
+			local := s.localDAHeight.Load()
+			isInline := ev.Height == local && s.localDAHeight.CompareAndSwap(local, local+1)
+
+			err = s.handler.HandleEvent(ctx, ev, isInline)
+			if isInline {
+				if err != nil {
+					s.localDAHeight.Store(local)
+					s.logger.Debug().Err(err).Uint64("da_height", ev.Height).
+						Msg("inline processing skipped/failed, rolling back")
+				} else {
+					s.headReached.Store(true)
+				}
+			}
+
 			watchdog.Reset(watchdogTimeout)
 		case <-watchdog.C:
 			return errors.New("subscription watchdog: no events received, reconnecting")
@@ -325,9 +319,7 @@ func (s *Subscriber) runCatchup(ctx context.Context) {
 		}
 
 		local := s.localDAHeight.Load()
-		highest := s.highestSeenDAHeight.Load()
-
-		if local > highest {
+		if local > s.highestSeenDAHeight.Load() {
 			s.headReached.Store(true)
 			return
 		}

block/internal/syncing/da_follower.go

@@ -68,8 +68,8 @@ func NewDAFollower(cfg DAFollowerConfig) DAFollower {
 		Namespaces:  [][]byte{cfg.Namespace, dataNs},
 		DABlockTime: cfg.DABlockTime,
 		Handler:     f,
+		StartHeight: cfg.StartDAHeight,
 	})
-	f.subscriber.SetStartHeight(cfg.StartDAHeight)
 
 	return f
 }
@@ -97,35 +97,27 @@ func (f *daFollower) HasReachedHead() bool {
 // caught up (ev.Height == localDAHeight) and blobs are available, it processes
 // them inline — avoiding a DA re-fetch round trip. Otherwise, it just lets
 // the catchup loop handle retrieval.
-//
-// Uses CAS on localDAHeight to claim exclusive access to processBlobs,
-// preventing concurrent map access with catchupLoop.
-func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent) {
-	// 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.subscriber.CompareAndSwapLocalHeight(ev.Height, ev.Height+1) {
-		events := f.retriever.ProcessBlobs(ctx, ev.Blobs, ev.Height)
-		for _, event := range events {
-			if err := f.eventSink.PipeEvent(ctx, event); err != nil {
-				// Roll back so catchupLoop can retry this height.
-				f.subscriber.SetLocalHeight(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.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.subscriber.SetLocalHeight(ev.Height)
+func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error {
+	if !isInline || len(ev.Blobs) == 0 {
+		return nil // skip
+	}
+
+	events := f.retriever.ProcessBlobs(ctx, ev.Blobs, ev.Height)
+	if len(events) == 0 {
+		return errors.New("skip inline: no complete events") // Split namespace, subscriber rolls back
+	}
+
+	for _, event := range events {
+		if err := f.eventSink.PipeEvent(ctx, event); err != nil {
+			f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
+				Msg("failed to pipe inline event, catchup will retry")
+			return err // Actual pipe failure, subscriber rolls back
 		}
-		return
 	}
 
-	// Slow path: behind, no blobs, or catchupLoop claimed this height.
+	f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
+		Msg("processed subscription blobs inline (fast path)")
+	return nil
 }
 
 // HandleCatchup retrieves events at a single DA height and pipes them
@@ -138,11 +130,7 @@ func (f *daFollower) HandleEvent(ctx context.Context, ev datypes.SubscriptionEve
 //     skipping the current daHeight or stopping catchup.
 func (f *daFollower) HandleCatchup(ctx context.Context, daHeight uint64) error {
 	// 1. Drain stale or future priority heights from P2P hints
-	for {
-		priorityHeight := f.popPriorityHeight()
-		if priorityHeight == 0 {
-			break
-		}
+	for priorityHeight := f.popPriorityHeight(); priorityHeight != 0; priorityHeight = f.popPriorityHeight() {
 		if priorityHeight < daHeight {
 			continue // skip stale hints without yielding back to the catchup loop
 		}
@@ -159,14 +147,9 @@ func (f *daFollower) HandleCatchup(ctx context.Context, daHeight uint64) error {
 				continue
 			}
 			// Roll back so daHeight is attempted again next cycle after backoff.
-			f.subscriber.SetLocalHeight(daHeight)
 			return err
 		}
-
-		// We successfully handled a priority height (we didn't actually process `daHeight`)
-		// Roll back so daHeight is attempted again next cycle.
-		f.subscriber.SetLocalHeight(daHeight)
-		return nil
+		break // continue with daHeight
 	}
 
 	// 2. Normal sequential fetch