fix(da): fix polling fallback when ws not available

Author: julienrbrt

apps/evm/server/force_inclusion_test.go

@@ -85,6 +85,8 @@ func (m *mockDA) HasForcedInclusionNamespace() bool {
 	return true
 }
 
+func (m *mockDA) SupportsSubscribe() bool { return true }
+
 func (m *mockDA) GetLatestDAHeight(_ context.Context) (uint64, error) {
 	return 0, nil
 }

block/internal/da/async_block_retriever_test.go

@@ -53,6 +53,8 @@ func TestAsyncBlockRetriever_SubscriptionDrivenCaching(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
+	client.On("SupportsSubscribe").Return(true)
+
 	// Create a subscription channel that delivers one event then blocks.
 	subCh := make(chan datypes.SubscriptionEvent, 1)
 	subCh <- datypes.SubscriptionEvent{
@@ -104,6 +106,8 @@ func TestAsyncBlockRetriever_CatchupFillsGaps(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
+	client.On("SupportsSubscribe").Return(true)
+
 	// Subscription delivers height 105 (no blobs — just a signal).
 	subCh := make(chan datypes.SubscriptionEvent, 1)
 	subCh <- datypes.SubscriptionEvent{Height: 105}
@@ -153,6 +157,8 @@ func TestAsyncBlockRetriever_HeightFromFuture(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
+	client.On("SupportsSubscribe").Return(true)
+
 	// Subscription delivers height 100 with no blobs.
 	subCh := make(chan datypes.SubscriptionEvent)
 	client.On("Subscribe", mock.Anything, fiNs, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(subCh), nil).Once()
@@ -187,6 +193,8 @@ func TestAsyncBlockRetriever_StopGracefully(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
+	client.On("SupportsSubscribe").Return(true)
+
 	blockCh := make(chan datypes.SubscriptionEvent)
 	client.On("Subscribe", mock.Anything, fiNs, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(blockCh), nil).Maybe()
 	client.On("Retrieve", mock.Anything, mock.Anything, fiNs).Return(datypes.ResultRetrieve{
@@ -211,6 +219,8 @@ func TestAsyncBlockRetriever_ReconnectOnSubscriptionError(t *testing.T) {
 	client := &mocks.MockClient{}
 	fiNs := datypes.NamespaceFromString("test-fi-ns").Bytes()
 
+	client.On("SupportsSubscribe").Return(true)
+
 	// First subscription closes immediately (simulating error).
 	closedCh := make(chan datypes.SubscriptionEvent)
 	close(closedCh)

block/internal/da/client.go

@@ -38,6 +38,7 @@ type client struct {
 	dataNamespaceBz    []byte
 	forcedNamespaceBz  []byte
 	hasForcedNamespace bool
+	isWebSocket        bool
 	timestampCache     *blockTimestampCache
 }
 
@@ -137,6 +138,7 @@ func NewClient(cfg Config) FullClient {
 		dataNamespaceBz:    datypes.NamespaceFromString(cfg.DataNamespace).Bytes(),
 		forcedNamespaceBz:  forcedNamespaceBz,
 		hasForcedNamespace: hasForcedNamespace,
+		isWebSocket:        cfg.DA.IsWebSocket,
 		timestampCache:     newBlockTimestampCache(blockTimestampCacheWindow),
 	}
 }
@@ -485,6 +487,12 @@ func (c *client) HasForcedInclusionNamespace() bool {
 	return c.hasForcedNamespace
 }
 
+// SupportsSubscribe reports whether the underlying transport supports
+// channel-based subscriptions (WebSocket).
+func (c *client) SupportsSubscribe() bool {
+	return c.isWebSocket
+}
+
 // Subscribe subscribes to blobs in the given namespace via the celestia-node
 // Subscribe API. It returns a channel that emits a SubscriptionEvent for every
 // DA block containing a matching blob. The channel is closed when ctx is

block/internal/da/interface.go

@@ -31,6 +31,11 @@ type Client interface {
 	// GetLatestDAHeight returns the latest height available on the DA layer.
 	GetLatestDAHeight(ctx context.Context) (uint64, error)
 
+	// SupportsSubscribe reports whether the underlying transport supports
+	// channel-based subscriptions (WebSocket). When false, callers must use
+	// polling-based retrieval via Retrieve instead.
+	SupportsSubscribe() bool
+
 	// Namespace accessors.
 	GetHeaderNamespace() []byte
 	GetDataNamespace() []byte

block/internal/da/subscriber.go

@@ -122,10 +122,15 @@ func (s *Subscriber) Start(ctx context.Context) error {
 
 	ctx, cancel := context.WithCancel(ctx)
 	s.cancel = cancel
-	s.wg.Add(2)
 	s.lifecycleMu.Unlock()
 
-	go s.followLoop(ctx)
+	if s.client.SupportsSubscribe() {
+		s.wg.Add(2)
+		go s.followLoop(ctx)
+	} else {
+		s.wg.Add(2)
+		go s.pollLoop(ctx)
+	}
 	go s.catchupLoop(ctx)
 
 	return nil
@@ -167,6 +172,57 @@ func (s *Subscriber) signalCatchup() {
 	}
 }
 
+// pollLoop periodically queries the latest DA height and triggers
+// catchup when new heights are available. This is used when the
+// underlying transport does not support channel-based subscriptions (HTTP).
+func (s *Subscriber) pollLoop(ctx context.Context) {
+	defer s.wg.Done()
+
+	s.logger.Info().Msg("starting poll loop")
+	defer s.logger.Info().Msg("poll loop stopped")
+
+	// Do an immediate poll on startup so we don't wait for the first tick.
+	s.pollDAHeight(ctx)
+
+	ticker := time.NewTicker(s.daBlockTime)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.pollDAHeight(ctx)
+		}
+	}
+}
+
+// pollDAHeight queries GetLatestDAHeight and signals catchup when a new height
+// is observed.
+func (s *Subscriber) pollDAHeight(ctx context.Context) {
+	height, err := s.client.GetLatestDAHeight(ctx)
+	if err != nil {
+		if ctx.Err() != nil {
+			return
+		}
+		s.logger.Warn().Err(err).Msg("poll: failed to get latest DA height")
+		return
+	}
+
+	cur := s.highestSeenDAHeight.Load()
+	if height <= cur {
+		return
+	}
+
+	s.seenSubscriptionEvent.Store(true)
+	s.logger.Debug().
+		Uint64("new_da_height", height).
+		Uint64("current_highest_seen", cur).
+		Msg("poll: observed new DA height")
+
+	s.updateHighest(height)
+}
+
 // followLoop subscribes to DA blob events and keeps highestSeenDAHeight up to date.
 func (s *Subscriber) followLoop(ctx context.Context) {
 	defer s.wg.Done()

block/internal/da/tracing.go

@@ -165,6 +165,9 @@ func (t *tracedClient) GetForcedInclusionNamespace() []byte {
 func (t *tracedClient) HasForcedInclusionNamespace() bool {
 	return t.inner.HasForcedInclusionNamespace()
 }
+func (t *tracedClient) SupportsSubscribe() bool {
+	return t.inner.SupportsSubscribe()
+}
 func (t *tracedClient) Subscribe(ctx context.Context, namespace []byte, includeTimestamp bool) (<-chan datypes.SubscriptionEvent, error) {
 	return t.inner.Subscribe(ctx, namespace, includeTimestamp)
 }

block/internal/da/tracing_test.go

@@ -74,6 +74,7 @@ func (m *mockFullClient) GetHeaderNamespace() []byte                          {
 func (m *mockFullClient) GetDataNamespace() []byte                            { return []byte{0x02} }
 func (m *mockFullClient) GetForcedInclusionNamespace() []byte                 { return []byte{0x03} }
 func (m *mockFullClient) HasForcedInclusionNamespace() bool                   { return true }
+func (m *mockFullClient) SupportsSubscribe() bool                             { return true }
 
 // setup a tracer provider + span recorder
 func setupDATrace(t *testing.T, inner FullClient) (FullClient, *tracetest.SpanRecorder) {

block/internal/syncing/syncer_test.go

@@ -116,6 +116,7 @@ func makeSignedHeaderBytes(
 func setupMockDAClient(tb testing.TB) (da.Client, chan datypes.SubscriptionEvent) {
 	mockClient := testmocks.NewMockClient(tb)
 	eventCh := make(chan datypes.SubscriptionEvent, 1)
+	mockClient.EXPECT().SupportsSubscribe().Return(true).Maybe()
 	mockClient.EXPECT().Subscribe(mock.Anything, mock.Anything, mock.Anything).Return((<-chan datypes.SubscriptionEvent)(eventCh), nil).Maybe()
 	return mockClient, eventCh
 }

pkg/da/jsonrpc/client.go

@@ -13,9 +13,10 @@ import (
 
 // Client dials the celestia-node RPC "blob" and "header" namespaces.
 type Client struct {
-	Blob   BlobAPI
-	Header HeaderAPI
-	closer jsonrpc.ClientCloser
+	Blob        BlobAPI
+	Header      HeaderAPI
+	IsWebSocket bool
+	closer      jsonrpc.ClientCloser
 }
 
 // Close closes the underlying JSON-RPC connection.
@@ -78,9 +79,15 @@ func NewWSClient(ctx context.Context, logger zerolog.Logger, addr, token string,
 	client, err := NewClient(ctx, httpToWS(addr), token, authHeaderName)
 	if err != nil {
 		logger.Warn().Err(err).Msg("DA websocket connection failed, falling back to DA polling")
-		return NewClient(ctx, addr, token, authHeaderName)
+		client, err = NewClient(ctx, addr, token, authHeaderName)
+		if err != nil {
+			return nil, err
+		}
+		client.IsWebSocket = false
+		return client, nil
 	}
 
+	client.IsWebSocket = true
 	return client, nil
 }