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

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

* feedback

* fix race

* update mocks

* fix ctx

Author: julienrbrt

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## v1.1.4
+
+### Fixed
+
+- DA client falls back to HTTP polling with `Retrieve` when the WebSocket connection fails, instead of trying to use the WS-only `Subscribe` over HTTP. A background goroutine retries WS every 30s so transient outages don't force a permanent downgrade [#3361](https://github.com/evstack/ev-node/pull/3361)
+
 ## v1.1.3
 
 ### Fixed

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

@@ -32,6 +32,7 @@ type Config struct {
 type client struct {
 	blobAPI            *blobrpc.BlobAPI
 	headerAPI          *blobrpc.HeaderAPI
+	da                 *blobrpc.Client // kept to read live IsWebSocket after transport upgrades
 	logger             zerolog.Logger
 	defaultTimeout     time.Duration
 	namespaceBz        []byte
@@ -131,6 +132,7 @@ func NewClient(cfg Config) FullClient {
 	return &client{
 		blobAPI:            &cfg.DA.Blob,
 		headerAPI:          &cfg.DA.Header,
+		da:                 cfg.DA,
 		logger:             cfg.Logger.With().Str("component", "da_client").Logger(),
 		defaultTimeout:     cfg.DefaultTimeout,
 		namespaceBz:        datypes.NamespaceFromString(cfg.Namespace).Bytes(),
@@ -485,6 +487,13 @@ func (c *client) HasForcedInclusionNamespace() bool {
 	return c.hasForcedNamespace
 }
 
+// SupportsSubscribe reports whether the underlying transport supports
+// channel-based subscriptions (WebSocket). Reads the live IsWebSocket flag
+// from the jsonrpc client so transport upgrades are visible immediately.
+func (c *client) SupportsSubscribe() bool {
+	return c.da != nil && c.da.IsWebSocket.Load()
+}
+
 // 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,14 @@ 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)
+	s.wg.Add(2)
+	if s.client.SupportsSubscribe() {
+		go s.followLoop(ctx)
+	} else {
+		go s.pollLoop(ctx)
+	}
 	go s.catchupLoop(ctx)
 
 	return nil
@@ -167,6 +171,72 @@ func (s *Subscriber) signalCatchup() {
 	}
 }
 
+// pollLoop periodically queries the latest DA height and triggers
+// catchup when new heights are available. The catchup loop fetches blobs
+// via Retrieve (which uses GetAll) so each height is fetched exactly once.
+// Periodically checks whether the underlying transport has been upgraded
+// to WebSocket and switches to followLoop when that happens.
+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)
+
+	interval := s.daBlockTime
+	if interval <= 0 {
+		interval = 2 * time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	for {
+		// If the transport has been upgraded to WS in the background,
+		// switch to the subscription-based follow loop.
+		if s.client.SupportsSubscribe() {
+			s.logger.Info().Msg("WebSocket available, switching from poll to follow loop")
+			s.wg.Add(1)
+			go s.followLoop(ctx)
+			return
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.pollDAHeight(ctx)
+		}
+	}
+}
+
+// pollDAHeight queries GetLatestDAHeight and signals catchup when a new
+// height is observed. The actual blob retrieval is done by catchupLoop.
+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

@@ -5,6 +5,9 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
 
 	libshare "github.com/celestiaorg/go-square/v3/share"
 	"github.com/filecoin-project/go-jsonrpc"
@@ -13,15 +16,30 @@ 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 atomic.Bool
+
+	mu          sync.Mutex
+	closer      jsonrpc.ClientCloser
+	retryCancel context.CancelFunc // stops the background WS retry loop
 }
 
-// Close closes the underlying JSON-RPC connection.
+// Close closes the underlying JSON-RPC connection and stops any
+// background WebSocket retry loop.
 func (c *Client) Close() {
-	if c != nil && c.closer != nil {
-		c.closer()
+	if c == nil {
+		return
+	}
+	c.mu.Lock()
+	if c.retryCancel != nil {
+		c.retryCancel()
+		c.retryCancel = nil
+	}
+	closer := c.closer
+	c.mu.Unlock()
+	if closer != nil {
+		closer()
 	}
 }
 
@@ -72,18 +90,88 @@ func NewClient(ctx context.Context, addr, token string, authHeaderName string) (
 // NewWSClient connects to the DA RPC endpoint over WebSocket.
 // Automatically converts http:// to ws:// (and https:// to wss://).
 // Supports channel-based subscriptions (e.g. Subscribe).
-// Note: WebSocket connections are eager — they connect at creation time
-// if the initial WS dial fails, falls back to HTTP polling for the entire session.
+// WebSocket connections are eager — they connect at creation time.
+// If the initial WS dial fails, it falls back to HTTP polling and spawns a
+// background goroutine that periodically retries the WS connection. When
+// the WS endpoint becomes reachable, the transport is transparently upgraded.
 func NewWSClient(ctx context.Context, logger zerolog.Logger, addr, token string, authHeaderName string) (*Client, error) {
 	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.Store(false)
+
+		// Retry WS in the background so transient outages don't force a permanent downgrade.
+		retryCtx, retryCancel := context.WithCancel(ctx)
+		client.retryCancel = retryCancel
+		go client.retryWSLoop(retryCtx, logger, addr, token, authHeaderName)
+
+		return client, nil
 	}
 
+	client.IsWebSocket.Store(true)
 	return client, nil
 }
 
+const wsRetryInterval = 30 * time.Second
+
+// retryWSLoop periodically attempts to re-establish a WebSocket connection.
+// When successful, it swaps the transport in-place and exits.
+func (c *Client) retryWSLoop(ctx context.Context, logger zerolog.Logger, addr, token, authHeaderName string) {
+	ticker := time.NewTicker(wsRetryInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if c.tryUpgradeWS(ctx, logger, addr, token, authHeaderName) {
+				return
+			}
+		}
+	}
+}
+
+// tryUpgradeWS attempts to open a WS connection and, if successful, swaps
+// the transport internals so subsequent calls use WebSocket. Returns true
+// when the upgrade succeeds (or the client is already on WS).
+func (c *Client) tryUpgradeWS(ctx context.Context, logger zerolog.Logger, addr, token, authHeaderName string) bool {
+	wsClient, err := NewClient(ctx, httpToWS(addr), token, authHeaderName)
+	if err != nil {
+		return false
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// Another goroutine may have already upgraded.
+	if c.IsWebSocket.Load() {
+		wsClient.Close()
+		return true
+	}
+
+	// Swap function pointers from the new WS client into the active client.
+	c.Blob.Internal = wsClient.Blob.Internal
+	c.Header.Internal = wsClient.Header.Internal
+
+	// Close the old HTTP connections and wire the new closer.
+	oldCloser := c.closer
+	c.closer = func() {
+		wsClient.closer()
+		if oldCloser != nil {
+			oldCloser()
+		}
+	}
+
+	c.IsWebSocket.Store(true)
+	logger.Info().Msg("DA websocket connection restored, switching back from HTTP polling")
+	return true
+}
+
 // BlobAPI mirrors celestia-node's blob module (nodebuilder/blob/blob.go).
 // jsonrpc.NewClient wires Internal.* to RPC stubs.
 type BlobAPI struct {