Update go reconnection logic

go/go.mod

@@ -3,8 +3,9 @@ module github.com/smartcontractkit/data-streams-sdk/go/v2
 go 1.24.0
 
 require (
+	github.com/coder/websocket v1.8.15
 	github.com/ethereum/go-ethereum v1.17.0
-	nhooyr.io/websocket v1.8.17
+	github.com/jpillora/backoff v1.0.0
 )
 
 require (

go/go.sum

@@ -1,5 +1,7 @@
 github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 h1:1zYrtlhrZ6/b6SAjLSfKzWtdgqK0U+HtH/VcBWh1BaU=
 github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6/go.mod h1:ioLG6R+5bUSO1oeGSDxOV3FADARuMoytZCSX6MEMQkI=
+github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
+github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/decred/dcrd/crypto/blake256 v1.0.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK09Y2A4Xv7EE0=
@@ -12,6 +14,8 @@ github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/holiman/uint256 v1.3.2 h1:a9EgMPSC1AAaj1SZL5zIQD3WbwTuHrMGOerLjGmM/TA=
 github.com/holiman/uint256 v1.3.2/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E=
+github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
+github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
@@ -20,5 +24,3 @@ golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
 golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
-nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=

go/stream.go

@@ -3,28 +3,30 @@ package streams
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
-	"math/rand"
 	"net/http"
 	"net/url"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/coder/websocket"
+	"github.com/jpillora/backoff"
 	"github.com/smartcontractkit/data-streams-sdk/go/v2/feed"
-	"nhooyr.io/websocket"
 )
 
 const (
-	defaultWSConnectTimeout      = time.Second * 5
-	minWSReconnectIntervalMillis = 1000
-	maxWSReconnectIntervalMIllis = 10000
-	maxWSReconnectAttempts       = 5
+	defaultWSConnectTimeout       = time.Second * 5
+	minWSReconnectIntervalSeconds = 1 * time.Second
+	maxWSReconnectIntervalSeconds = 20 * time.Second
+	maxWSReconnectAttempts        = 5
 )
 
 var (
-	ErrStreamClosed = fmt.Errorf("client: use of closed Stream")
+	ErrStreamClosed              = fmt.Errorf("client: use of closed Stream")
+	ErrStreamNoActiveConnections = fmt.Errorf("client: stream has no active connections, max reconnect attempts reached")
 )
 
 type message struct {
@@ -259,7 +261,9 @@ func (s *stream) monitorConn(conn *wsConn) {
 
 		re, err := s.newWSconnWithRetry(conn.origin)
 		if err != nil {
-			s.closeError.CompareAndSwap(nil, fmt.Errorf("stream has no active connections, last error: %w", err))
+			if errors.Is(err, ErrStreamNoActiveConnections) {
+				s.closeError.CompareAndSwap(nil, err)
+			}
 			s.Close()
 			return
 		}
@@ -279,15 +283,16 @@ func (s *stream) newWSconnWithRetry(origin string) (conn *wsConn, err error) {
 	// will try to reconnect until client is closed or
 	// we have no active connections and have exceeded maxWSReconnectAttempts
 	var attempts int
+	retryBackoff := &backoff.Backoff{Min: minWSReconnectIntervalSeconds, Max: maxWSReconnectIntervalSeconds, Jitter: true}
 	for {
 		if s.closed.Load() || s.streamCtx.Err() != nil {
-			return nil, fmt.Errorf("Retry cancelled, stream is closed")
+			return nil, ErrStreamClosed
 		}
 
 		// fail the stream if we are over the maxWSReconnectAttempts
 		// and there are no other active connection
 		if attempts >= s.config.WsMaxReconnect && s.stats.activeConnections.Load() == 0 {
-			return nil, err
+			return nil, fmt.Errorf("%w, last error: %w", ErrStreamNoActiveConnections, err)
 		}
 		attempts++
 
@@ -296,13 +301,19 @@ func (s *stream) newWSconnWithRetry(origin string) (conn *wsConn, err error) {
 		cancel()
 
 		if err != nil {
-			interval := time.Millisecond * time.Duration(
-				rand.Intn(maxWSReconnectIntervalMIllis-minWSReconnectIntervalMillis)+minWSReconnectIntervalMillis) //nolint:gosec
+			interval := retryBackoff.Duration()
 			s.config.logInfo(
 				"client: stream websocket %s: error reconnecting: %s, backing off: %s",
 				origin, err, interval.String(),
 			)
-			time.Sleep(interval)
+
+			timer := time.NewTimer(interval)
+			select {
+			case <-timer.C:
+			case <-s.streamCtx.Done():
+				timer.Stop()
+				return nil, ErrStreamClosed
+			}
 			continue
 		}
 		return conn, nil

go/stream_test.go

@@ -6,13 +6,14 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/coder/websocket"
 	"github.com/smartcontractkit/data-streams-sdk/go/v2/feed"
-	"nhooyr.io/websocket"
 )
 
 func TestClient_Subscribe(t *testing.T) {
@@ -836,6 +837,69 @@ func TestClient_StreamHA_AllOriginsFailInitialConnect(t *testing.T) {
 	time.Sleep(300 * time.Millisecond)
 }
 
+func TestClient_StreamReconnectExhaustedReturnsNoActiveConnectionsError(t *testing.T) {
+	connectAttempts := &atomic.Uint64{}
+
+	ms := newMockServer(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodHead {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+
+		if r.URL.Path != apiV2WS {
+			t.Errorf("expected path %s, got %s", apiV2WS, r.URL.Path)
+		}
+
+		attempt := connectAttempts.Add(1)
+		if attempt > 1 {
+			// After the initial successful subscribe, force reconnect attempts to fail.
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+
+		conn, err := websocket.Accept(
+			w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionContextTakeover},
+		)
+		if err != nil {
+			t.Fatalf("error accepting connection: %s", err)
+		}
+
+		// Close immediately so client enters reconnect flow.
+		_ = conn.CloseNow()
+	})
+	defer ms.Close()
+
+	streamsClient, err := ms.Client()
+	if err != nil {
+		t.Fatalf("error creating client %s", err)
+	}
+
+	cc := streamsClient.(*client)
+	cc.config.WsMaxReconnect = 1
+
+	sub, err := streamsClient.Stream(t.Context(), []feed.ID{feed1})
+	if err != nil {
+		t.Fatalf("error subscribing %s", err)
+	}
+	defer sub.Close()
+
+	readCtx, cancel := context.WithTimeout(t.Context(), maxWSReconnectIntervalSeconds)
+	defer cancel()
+
+	_, err = sub.Read(readCtx)
+	if err == nil {
+		t.Fatal("expected Read to fail after reconnect attempts are exhausted")
+	}
+
+	if !errors.Is(err, ErrStreamNoActiveConnections) {
+		t.Fatalf("expected errors.Is(err, ErrStreamNoActiveConnections) to be true, got err: %v", err)
+	}
+
+	if !strings.Contains(err.Error(), fmt.Sprintf("%d", http.StatusForbidden)) {
+		t.Fatalf("expected error to include underlying reconnect status code %d, got err: %v", http.StatusForbidden, err)
+	}
+}
+
 func TestClient_StreamOutOfOrder(t *testing.T) {
 	reports := []*ReportResponse{
 		{FeedID: feed1, ObservationsTimestamp: time.Unix(100, 0)},