feat: implement sync mode metrics and subscription functionality for DA client

Author: tac0turtle

block/internal/common/metrics.go

@@ -69,6 +69,11 @@ type Metrics struct {
 	// Forced inclusion metrics
 	ForcedInclusionTxsInGracePeriod metrics.Gauge   // Number of forced inclusion txs currently in grace period
 	ForcedInclusionTxsMalicious     metrics.Counter // Total number of forced inclusion txs marked as malicious
+
+	// Sync mode metrics
+	SyncMode        metrics.Gauge   // Current sync mode: 0=catchup, 1=follow
+	SubscribeErrors metrics.Counter // Number of subscription failures
+	ModeSwitches    metrics.Counter // Number of catchup<->follow mode transitions
 }
 
 // PrometheusMetrics returns Metrics built using Prometheus client library
@@ -201,6 +206,28 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 		Help:      "Total number of forced inclusion transactions marked as malicious (past grace boundary)",
 	}, labels).With(labelsAndValues...)
 
+	// Sync mode metrics
+	m.SyncMode = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: MetricsSubsystem,
+		Name:      "sync_mode",
+		Help:      "Current sync mode: 0=catchup (polling), 1=follow (subscription)",
+	}, labels).With(labelsAndValues...)
+
+	m.SubscribeErrors = prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: MetricsSubsystem,
+		Name:      "subscribe_errors_total",
+		Help:      "Total number of DA subscription failures",
+	}, labels).With(labelsAndValues...)
+
+	m.ModeSwitches = prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: MetricsSubsystem,
+		Name:      "mode_switches_total",
+		Help:      "Total number of sync mode transitions between catchup and follow",
+	}, labels).With(labelsAndValues...)
+
 	// DA Submitter metrics
 	m.DASubmitterPendingBlobs = prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 		Namespace: namespace,
@@ -269,6 +296,11 @@ func NopMetrics() *Metrics {
 		// Forced inclusion metrics
 		ForcedInclusionTxsInGracePeriod: discard.NewGauge(),
 		ForcedInclusionTxsMalicious:     discard.NewCounter(),
+
+		// Sync mode metrics
+		SyncMode:        discard.NewGauge(),
+		SubscribeErrors: discard.NewCounter(),
+		ModeSwitches:    discard.NewCounter(),
 	}
 
 	// Initialize maps with no-op metrics

block/internal/da/client.go

@@ -442,3 +442,27 @@ func (c *client) Validate(ctx context.Context, ids []datypes.ID, proofs []datype
 
 	return results, nil
 }
+
+// Subscribe subscribes to blobs in the specified namespace.
+// Returns a channel that receives subscription responses as new blobs are included.
+func (c *client) Subscribe(ctx context.Context, namespace []byte) (<-chan *blobrpc.SubscriptionResponse, error) {
+	ns, err := share.NewNamespaceFromBytes(namespace)
+	if err != nil {
+		return nil, fmt.Errorf("invalid namespace: %w", err)
+	}
+
+	return c.blobAPI.Subscribe(ctx, ns)
+}
+
+// LocalHead returns the height of the locally synced DA head.
+func (c *client) LocalHead(ctx context.Context) (uint64, error) {
+	headCtx, cancel := context.WithTimeout(ctx, c.defaultTimeout)
+	defer cancel()
+
+	header, err := c.headerAPI.LocalHead(headCtx)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get local head: %w", err)
+	}
+
+	return header.Height, nil
+}

block/internal/da/interface.go

@@ -3,6 +3,7 @@ package da
 import (
 	"context"
 
+	blobrpc "github.com/evstack/ev-node/pkg/da/jsonrpc"
 	datypes "github.com/evstack/ev-node/pkg/da/types"
 )
 
@@ -22,6 +23,15 @@ type Client interface {
 	GetDataNamespace() []byte
 	GetForcedInclusionNamespace() []byte
 	HasForcedInclusionNamespace() bool
+
+	// Subscribe subscribes to blobs in the specified namespace.
+	// Returns a channel that receives subscription responses as new blobs are included.
+	// Used for follow mode to receive real-time blob notifications.
+	Subscribe(ctx context.Context, namespace []byte) (<-chan *blobrpc.SubscriptionResponse, error)
+
+	// LocalHead returns the height of the locally synced DA head.
+	// Used to determine if the node is caught up with the DA layer.
+	LocalHead(ctx context.Context) (uint64, error)
 }
 
 // Verifier defines the interface for DA proof verification operations.

block/internal/syncing/syncer.go

@@ -22,12 +22,45 @@ import (
 	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/block/internal/da"
 	"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/pkg/genesis"
 	"github.com/evstack/ev-node/pkg/store"
 	"github.com/evstack/ev-node/types"
 )
 
+// SyncMode represents the current synchronization mode for the DA worker.
+type SyncMode int
+
+const (
+	// SyncModeCatchup indicates the node is behind the DA chain head and polling aggressively.
+	SyncModeCatchup SyncMode = iota
+	// SyncModeFollow indicates the node is caught up and using subscription for real-time updates.
+	SyncModeFollow
+)
+
+// String returns a human-readable representation of the sync mode.
+func (m SyncMode) String() string {
+	switch m {
+	case SyncModeCatchup:
+		return "catchup"
+	case SyncModeFollow:
+		return "follow"
+	default:
+		return "unknown"
+	}
+}
+
+const (
+	// catchupThreshold is the number of DA blocks behind local head
+	// before switching from follow to catchup mode.
+	catchupThreshold = 2
+
+	// followWatchdogMultiplier is the multiplier for BlockTime
+	// used as subscription watchdog timeout.
+	followWatchdogMultiplier = 3
+)
+
 // forcedInclusionGracePeriodConfig contains internal configuration for forced inclusion grace periods.
 type forcedInclusionGracePeriodConfig struct {
 	// basePeriod is the base number of additional epochs allowed for including forced inclusion transactions
@@ -118,6 +151,9 @@ type Syncer struct {
 
 	// P2P wait coordination
 	p2pWaitState atomic.Value // stores p2pWaitState
+
+	// Sync mode tracking
+	currentSyncMode atomic.Int32 // stores SyncMode as int32
 }
 
 // pendingForcedInclusionTx represents a forced inclusion transaction that hasn't been included yet
@@ -318,28 +354,213 @@ func (s *Syncer) daWorkerLoop() {
 	defer s.logger.Info().Msg("DA worker stopped")
 
 	for {
-		err := s.fetchDAUntilCaughtUp()
+		select {
+		case <-s.ctx.Done():
+			return
+		default:
+		}
 
-		var backoff time.Duration
-		if err == nil {
-			// No error, means we are caught up.
-			backoff = s.config.DA.BlockTime.Duration
-		} else {
-			// Error, back off for a shorter duration.
-			backoff = s.config.DA.BlockTime.Duration
-			if backoff <= 0 {
-				backoff = 2 * time.Second
-			}
+		mode := s.determineSyncMode()
+		previousMode := SyncMode(s.currentSyncMode.Load())
+
+		// Track mode switches
+		if mode != previousMode {
+			s.currentSyncMode.Store(int32(mode))
+			s.metrics.ModeSwitches.Add(1)
+			s.logger.Info().
+				Str("from", previousMode.String()).
+				Str("to", mode.String()).
+				Msg("sync mode changed")
+		}
+
+		switch mode {
+		case SyncModeCatchup:
+			s.runCatchupMode()
+		case SyncModeFollow:
+			s.runFollowMode()
 		}
+	}
+}
 
+// determineSyncMode checks the current DA sync status and returns the appropriate mode.
+func (s *Syncer) determineSyncMode() SyncMode {
+	// If DA client is nil (e.g., in tests), default to catchup mode
+	if s.daClient == nil {
+		return SyncModeCatchup
+	}
+
+	localHead, err := s.daClient.LocalHead(s.ctx)
+	if err != nil {
+		// Default to catchup on error - safer to poll than assume we're caught up
+		s.logger.Debug().Err(err).Msg("failed to get local DA head, defaulting to catchup mode")
+		return SyncModeCatchup
+	}
+
+	currentDAHeight := s.daRetrieverHeight.Load()
+
+	// Consider "caught up" if within catchupThreshold blocks of local head
+	if currentDAHeight+catchupThreshold >= localHead {
+		return SyncModeFollow
+	}
+	return SyncModeCatchup
+}
+
+// runCatchupMode runs the catchup sync mode - aggressive polling until caught up.
+func (s *Syncer) runCatchupMode() {
+	s.logger.Debug().Msg("running catchup mode")
+	s.metrics.SyncMode.Set(float64(SyncModeCatchup))
+
+	err := s.fetchDAUntilCaughtUp()
+	if errors.Is(err, context.Canceled) {
+		return
+	}
+
+	// Back off before next iteration:
+	// - On error: wait before retrying to avoid hammering a failing DA layer
+	// - On success (caught up): wait for new DA blocks to appear
+	backoff := s.config.DA.BlockTime.Duration
+	if backoff <= 0 {
+		backoff = 2 * time.Second
+	}
+
+	if err != nil {
+		s.logger.Debug().Err(err).Msg("catchup failed, backing off before retry")
+	} else {
+		s.logger.Debug().Msg("caught up with DA, backing off before next check")
+	}
+
+	s.sleepOrDone(backoff)
+}
+
+// runFollowMode runs the follow sync mode - subscription-based real-time updates.
+func (s *Syncer) runFollowMode() {
+	s.logger.Debug().Msg("running follow mode")
+	s.metrics.SyncMode.Set(float64(SyncModeFollow))
+
+	err := s.subscribeAndFollow()
+	if err != nil && !errors.Is(err, context.Canceled) {
+		s.metrics.SubscribeErrors.Add(1)
+		s.logger.Warn().Err(err).Msg("subscribe failed, falling back to catchup")
+		// Don't sleep - go straight to catchup mode to recover
+	}
+}
+
+// subscribeAndFollow uses the DA subscription API to receive real-time blob notifications.
+// It subscribes to both header and data namespaces and processes incoming blobs.
+// Returns when subscription fails, context is cancelled, or node falls behind.
+func (s *Syncer) subscribeAndFollow() error {
+	// Get namespaces
+	headerNS := s.daClient.GetHeaderNamespace()
+	dataNS := s.daClient.GetDataNamespace()
+
+	// Create subscription context with cancellation
+	subCtx, cancel := context.WithCancel(s.ctx)
+	defer cancel()
+
+	// Subscribe to header namespace
+	headerCh, err := s.daClient.Subscribe(subCtx, headerNS)
+	if err != nil {
+		return fmt.Errorf("failed to subscribe to header namespace: %w", err)
+	}
+
+	// Subscribe to data namespace (only if different from header namespace)
+	var dataCh <-chan *blobrpc.SubscriptionResponse
+	if !bytes.Equal(headerNS, dataNS) {
+		dataCh, err = s.daClient.Subscribe(subCtx, dataNS)
+		if err != nil {
+			return fmt.Errorf("failed to subscribe to data namespace: %w", err)
+		}
+	}
+
+	s.logger.Info().Msg("subscribed to DA namespaces for follow mode")
+
+	// Calculate watchdog timeout
+	watchdogTimeout := s.config.DA.BlockTime.Duration * followWatchdogMultiplier
+	if watchdogTimeout <= 0 {
+		watchdogTimeout = 30 * time.Second
+	}
+
+	// Process subscription events
+	for {
 		select {
 		case <-s.ctx.Done():
-			return
-		case <-time.After(backoff):
+			return s.ctx.Err()
+
+		case resp, ok := <-headerCh:
+			if !ok {
+				return errors.New("header subscription closed")
+			}
+			if err := s.processSubscriptionResponse(resp); err != nil {
+				s.logger.Error().Err(err).Uint64("height", resp.Height).Msg("failed to process header subscription")
+			}
+
+		case resp, ok := <-dataCh:
+			if dataCh == nil {
+				// Data channel not used (same namespace), continue
+				continue
+			}
+			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 <-time.After(watchdogTimeout):
+			// Watchdog: if no events for watchdogTimeout, recheck mode
+			// Might have fallen behind due to network issues
+			s.logger.Debug().Dur("timeout", watchdogTimeout).Msg("subscription watchdog triggered, checking sync mode")
+			if s.determineSyncMode() == SyncModeCatchup {
+				return errors.New("fell behind, switching to catchup")
+			}
 		}
 	}
 }
 
+// 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 {
+		return nil
+	}
+
+	s.logger.Debug().
+		Uint64("da_height", resp.Height).
+		Int("blobs", len(resp.Blobs)).
+		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()
+	}
+
+	// Process blobs using the DA retriever's processBlobs method
+	events := s.daRetriever.(*daRetriever).processBlobs(s.ctx, blobs, resp.Height)
+
+	// Send events to the processing channel
+	for _, event := range events {
+		select {
+		case s.heightInCh <- event:
+			s.logger.Debug().
+				Uint64("height", event.Header.Height()).
+				Uint64("da_height", event.DaHeight).
+				Msg("sent subscription event to processing")
+		default:
+			s.cache.SetPendingEvent(event.Header.Height(), &event)
+			s.logger.Debug().
+				Uint64("height", event.Header.Height()).
+				Msg("subscription event queued as pending")
+		}
+	}
+
+	// Update retriever height
+	if resp.Height >= s.daRetrieverHeight.Load() {
+		s.daRetrieverHeight.Store(resp.Height + 1)
+	}
+
+	return nil
+}
+
 func (s *Syncer) fetchDAUntilCaughtUp() error {
 	for {
 		select {