diff --git a/go/go.mod b/go/go.mod index f820158..5a58b19 100644 --- a/go/go.mod +++ b/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 ( diff --git a/go/go.sum b/go/go.sum index e4dd149..2a640bb 100644 --- a/go/go.sum +++ b/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= diff --git a/go/stream.go b/go/stream.go index 7a4389a..f0f391c 100644 --- a/go/stream.go +++ b/go/stream.go @@ -3,8 +3,8 @@ package streams import ( "context" "encoding/json" + "errors" "fmt" - "math/rand" "net/http" "net/url" "strings" @@ -12,19 +12,21 @@ import ( "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 diff --git a/go/stream_test.go b/go/stream_test.go index 20c93bc..5dcc219 100644 --- a/go/stream_test.go +++ b/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)},