[ECO-4327] Feature - realtime fallbacks

README.md

@@ -450,8 +450,6 @@ See [jwt auth issue](https://github.com/ably/ably-go/issues/569) for more detail
 - Inband reauthentication is not supported; expiring tokens will trigger a disconnection and resume of a realtime
   connection. See [server initiated auth](https://github.com/ably/ably-go/issues/228) for more details.
 
-- Realtime connection failure handling is partially implemented. See [host fallback](https://github.com/ably/ably-go/issues/225) for more details.
-
 - Channel suspended state is partially implemented. See [suspended channel state](https://github.com/ably/ably-go/issues/568).
 
 - Realtime Ping function is not implemented.

ably/error.go

@@ -147,6 +147,17 @@ func errFromUnprocessableBody(resp *http.Response) error {
 	return &ErrorInfo{Code: ErrBadRequest, StatusCode: resp.StatusCode, err: err}
 }
 
+func isTimeoutOrDnsErr(err error) bool {
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		if netErr.Timeout() { // RSC15l2
+			return true
+		}
+	}
+	var dnsErr *net.DNSError
+	return errors.As(err, &dnsErr) // RSC15l1
+}
+
 func checkValidHTTPResponse(resp *http.Response) error {
 	type errorBody struct {
 		Error errorInfo `json:"error,omitempty" codec:"error,omitempty"`

ably/export_test.go

@@ -36,10 +36,6 @@ func (opts *clientOptions) RestURL() string {
 	return opts.restURL()
 }
 
-func (opts *clientOptions) RealtimeURL() string {
-	return opts.realtimeURL()
-}
-
 func (c *REST) Post(ctx context.Context, path string, in, out interface{}) (*http.Response, error) {
 	return c.post(ctx, path, in, out)
 }
@@ -93,6 +89,10 @@ func (c *REST) GetCachedFallbackHost() string {
 	return c.hostCache.get()
 }
 
+func (c *REST) ActiveRealtimeHost() string {
+	return c.activeRealtimeHost
+}
+
 func (c *RealtimeChannel) GetChannelSerial() string {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
@@ -121,6 +121,10 @@ func (opts *clientOptions) GetFallbackRetryTimeout() time.Duration {
 	return opts.fallbackRetryTimeout()
 }
 
+func (opts *clientOptions) HasActiveInternetConnection() bool {
+	return opts.hasActiveInternetConnection()
+}
+
 func NewErrorInfo(code ErrorCode, err error) *ErrorInfo {
 	return newError(code, err)
 }
@@ -222,6 +226,10 @@ func (c *Connection) SetKey(key string) {
 	c.key = key
 }
 
+func (r *Realtime) Rest() *REST {
+	return r.rest
+}
+
 func (c *RealtimePresence) Members() map[string]*PresenceMessage {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()

ably/options.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"log"
 	"net"
 	"net/http"
@@ -28,6 +29,9 @@ const (
 	Port           = 80
 	TLSPort        = 443
 	maxMessageSize = 65536 // 64kb, default value TO3l8
+
+	internetCheckUrl = "https://internet-up.ably-realtime.com/is-the-internet-up.txt"
+	internetCheckOk  = "yes\n"
 )
 
 var defaultOptions = clientOptions{
@@ -482,10 +486,10 @@ func (opts *clientOptions) restURL() (restUrl string) {
 	return "https://" + baseUrl
 }
 
-func (opts *clientOptions) realtimeURL() (realtimeUrl string) {
-	baseUrl := opts.getRealtimeHost()
+func (opts *clientOptions) realtimeURL(realtimeHost string) (realtimeUrl string) {
+	baseUrl := realtimeHost
 	_, _, err := net.SplitHostPort(baseUrl)
-	if err != nil { // set port if not set in baseUrl
+	if err != nil { // set port if not set in provided realtimeHost
 		port, _ := opts.activePort()
 		baseUrl = net.JoinHostPort(baseUrl, strconv.Itoa(port))
 	}
@@ -595,6 +599,19 @@ func (opts *clientOptions) idempotentRESTPublishing() bool {
 	return opts.IdempotentRESTPublishing
 }
 
+func (opts *clientOptions) hasActiveInternetConnection() bool {
+	res, err := opts.httpclient().Get(internetCheckUrl)
+	if err != nil {
+		return false
+	}
+	data, err := io.ReadAll(res.Body)
+	res.Body.Close()
+	if err != nil {
+		return false
+	}
+	return string(data) == internetCheckOk
+}
+
 type ScopeParams struct {
 	Start time.Time
 	End   time.Time

ably/options_test.go

@@ -14,31 +14,32 @@ import (
 )
 
 func TestDefaultFallbacks_RSC15h(t *testing.T) {
-	t.Run("with env should return environment fallback hosts", func(t *testing.T) {
-		expectedFallBackHosts := []string{
-			"a.ably-realtime.com",
-			"b.ably-realtime.com",
-			"c.ably-realtime.com",
-			"d.ably-realtime.com",
-			"e.ably-realtime.com",
-		}
-		hosts := ably.DefaultFallbackHosts()
-		assert.Equal(t, expectedFallBackHosts, hosts)
-	})
+	expectedFallBackHosts := []string{
+		"a.ably-realtime.com",
+		"b.ably-realtime.com",
+		"c.ably-realtime.com",
+		"d.ably-realtime.com",
+		"e.ably-realtime.com",
+	}
+	hosts := ably.DefaultFallbackHosts()
+	assert.Equal(t, expectedFallBackHosts, hosts)
 }
 
 func TestEnvFallbackHosts_RSC15i(t *testing.T) {
-	t.Run("with env should return environment fallback hosts", func(t *testing.T) {
-		expectedFallBackHosts := []string{
-			"sandbox-a-fallback.ably-realtime.com",
-			"sandbox-b-fallback.ably-realtime.com",
-			"sandbox-c-fallback.ably-realtime.com",
-			"sandbox-d-fallback.ably-realtime.com",
-			"sandbox-e-fallback.ably-realtime.com",
-		}
-		hosts := ably.GetEnvFallbackHosts("sandbox")
-		assert.Equal(t, expectedFallBackHosts, hosts)
-	})
+	expectedFallBackHosts := []string{
+		"sandbox-a-fallback.ably-realtime.com",
+		"sandbox-b-fallback.ably-realtime.com",
+		"sandbox-c-fallback.ably-realtime.com",
+		"sandbox-d-fallback.ably-realtime.com",
+		"sandbox-e-fallback.ably-realtime.com",
+	}
+	hosts := ably.GetEnvFallbackHosts("sandbox")
+	assert.Equal(t, expectedFallBackHosts, hosts)
+}
+
+func TestInternetConnectionCheck_RTN17c(t *testing.T) {
+	clientOptions := ably.NewClientOptions()
+	assert.True(t, clientOptions.HasActiveInternetConnection())
 }
 
 func TestFallbackHosts_RSC15b(t *testing.T) {

ably/realtime_client_integration_test.go

@@ -5,6 +5,7 @@ package ably_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
@@ -15,9 +16,11 @@ import (
 	"time"
 
 	"github.com/ably/ably-go/ably"
+	"github.com/ably/ably-go/ably/internal/ablyutil"
 	"github.com/ably/ably-go/ablytest"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestRealtime_RealtimeHost(t *testing.T) {
@@ -132,6 +135,183 @@ func TestRealtime_RSC7_AblyAgent(t *testing.T) {
 	})
 }
 
+func TestRealtime_RTN17_HostFallback(t *testing.T) {
+	t.Parallel()
+
+	getDNSErr := func() *net.DNSError {
+		return &net.DNSError{
+			IsTimeout: false,
+		}
+	}
+
+	getTimeoutErr := func() error {
+		return &errTimeout{}
+	}
+
+	setUpWithError := func(customErr error, opts ...ably.ClientOption) (visitedHosts []string) {
+		client, err := ably.NewRealtime(append(opts, ably.WithAutoConnect(false), ably.WithKey("fake:key"),
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (ably.Conn, error) {
+				visitedHosts = append(visitedHosts, u.Hostname())
+				return nil, customErr
+			}))...)
+		require.NoError(t, err)
+		ablytest.Wait(ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventDisconnected), nil)
+		return
+	}
+
+	t.Run("RTN17a: First attempt should be made on default primary host", func(t *testing.T) {
+		visitedHosts := setUpWithError(errors.New("host url is wrong"))
+		assert.Equal(t, "realtime.ably.io", visitedHosts[0])
+	})
+
+	t.Run("RTN17b: Fallback behaviour", func(t *testing.T) {
+		t.Parallel()
+
+		t.Run("apply when default realtime endpoint is not overridden, port/tlsport not set", func(t *testing.T) {
+			visitedHosts := setUpWithError(getTimeoutErr())
+			expectedPrimaryHost := "realtime.ably.io"
+			expectedFallbackHosts := ably.DefaultFallbackHosts()
+
+			assert.Equal(t, 6, len(visitedHosts))
+			assert.Equal(t, expectedPrimaryHost, visitedHosts[0])
+			assert.ElementsMatch(t, expectedFallbackHosts, visitedHosts[1:])
+		})
+
+		t.Run("does not apply when the custom realtime endpoint is used", func(t *testing.T) {
+			visitedHosts := setUpWithError(getTimeoutErr(), ably.WithRealtimeHost("custom-realtime.ably.io"))
+			expectedHost := "custom-realtime.ably.io"
+
+			require.Equal(t, 1, len(visitedHosts))
+			assert.Equal(t, expectedHost, visitedHosts[0])
+		})
+
+		t.Run("apply when fallbacks are provided", func(t *testing.T) {
+			fallbacks := []string{"fallback0", "fallback1", "fallback2"}
+			visitedHosts := setUpWithError(getTimeoutErr(), ably.WithFallbackHosts(fallbacks))
+			expectedPrimaryHost := "realtime.ably.io"
+
+			assert.Equal(t, 4, len(visitedHosts))
+			assert.Equal(t, expectedPrimaryHost, visitedHosts[0])
+			assert.ElementsMatch(t, fallbacks, visitedHosts[1:])
+		})
+
+		t.Run("apply when fallbackHostUseDefault is true, even if env. or host is set", func(t *testing.T) {
+			visitedHosts := setUpWithError(
+				getTimeoutErr(),
+				ably.WithFallbackHostsUseDefault(true),
+				ably.WithEnvironment("custom"),
+				ably.WithRealtimeHost("custom-ably.realtime.com"))
+
+			expectedPrimaryHost := "custom-ably.realtime.com"
+			expectedFallbackHosts := ably.DefaultFallbackHosts()
+
+			assert.Equal(t, 6, len(visitedHosts))
+			assert.Equal(t, expectedPrimaryHost, visitedHosts[0])
+			assert.ElementsMatch(t, expectedFallbackHosts, visitedHosts[1:])
+		})
+	})
+
+	t.Run("RTN17c: Verifies internet connection is active in case of error necessitating use of an alternative host", func(t *testing.T) {
+		t.Parallel()
+		const internetCheckUrl = "https://internet-up.ably-realtime.com/is-the-internet-up.txt"
+		rec, optn := ablytest.NewHttpRecorder()
+		visitedHosts := setUpWithError(getDNSErr(), optn...)
+		assert.Equal(t, 6, len(visitedHosts)) // including primary host
+		assert.Equal(t, 5, len(rec.Requests()))
+		for _, request := range rec.Requests() {
+			assert.Equal(t, request.URL.String(), internetCheckUrl)
+		}
+	})
+
+	t.Run("RTN17d: Check for compatible errors before attempting to reconnect to a fallback host", func(t *testing.T) {
+		visitedHosts := setUpWithError(fmt.Errorf("host url is wrong")) // non-dns or non-timeout error
+		assert.Equal(t, 1, len(visitedHosts))
+		visitedHosts = setUpWithError(getDNSErr())
+		assert.Equal(t, 6, len(visitedHosts))
+		visitedHosts = setUpWithError(getTimeoutErr())
+		assert.Equal(t, 6, len(visitedHosts))
+	})
+
+	t.Run("RTN17e: Same fallback host should be used for REST as Realtime Fallback Host for a given active connection", func(t *testing.T) {
+		errCh := make(chan error, 1)
+		errCh <- getTimeoutErr()
+		realtimeMsgRecorder := NewMessageRecorder()         // websocket recorder
+		restMsgRecorder, optn := ablytest.NewHttpRecorder() // http recorder
+		_, client := ablytest.NewRealtime(ably.WithAutoConnect(false),
+			ably.WithDial(func(protocol string, u *url.URL, timeout time.Duration) (ably.Conn, error) {
+				err, ok := <-errCh
+				if ok {
+					close(errCh)
+					return nil, err // return timeout error for primary host
+				}
+				return realtimeMsgRecorder.Dial(protocol, u, timeout) // return dial for subsequent dials
+			}), optn[0])
+		defer client.Close()
+
+		err := ablytest.Wait(ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected), nil)
+		if err != nil {
+			t.Fatalf("Error connecting host with error %v", err)
+		}
+		realtimeSuccessHost := realtimeMsgRecorder.URLs()[0].Hostname()
+		fallbackHosts := ably.GetEnvFallbackHosts("sandbox")
+		if !ablyutil.SliceContains(fallbackHosts, realtimeSuccessHost) {
+			t.Fatalf("realtime host must be one of fallback hosts, received %v", realtimeSuccessHost)
+		}
+
+		client.Time(context.Background())                            // make a rest request
+		restSuccessHost := restMsgRecorder.Request(1).URL.Hostname() // second request is to get the time, first for internet connection
+		assert.Equal(t, realtimeSuccessHost, restSuccessHost)
+	})
+}
+
+func TestRealtime_RTN17_Integration_HostFallback_Internal_Server_Error(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+	serverURL, err := url.Parse(server.URL)
+	assert.NoError(t, err)
+
+	app, realtime := ablytest.NewRealtime(
+		ably.WithAutoConnect(false),
+		ably.WithTLS(false),
+		ably.WithUseTokenAuth(true),
+		ably.WithFallbackHosts([]string{"sandbox-a-fallback.ably-realtime.com"}),
+		ably.WithRealtimeHost(serverURL.Host))
+
+	defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+	err = ablytest.Wait(ablytest.ConnWaiter(realtime, realtime.Connect, ably.ConnectionEventConnected), nil)
+	assert.NoError(t, err)
+
+	assert.Equal(t, "sandbox-a-fallback.ably-realtime.com", realtime.Rest().ActiveRealtimeHost())
+}
+
+func TestRealtime_RTN17_Integration_HostFallback_Timeout(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(3 * time.Second)
+		w.WriteHeader(http.StatusSwitchingProtocols)
+	}))
+	defer server.Close()
+	serverURL, err := url.Parse(server.URL)
+	assert.NoError(t, err)
+
+	app, realtime := ablytest.NewRealtime(
+		ably.WithAutoConnect(false),
+		ably.WithTLS(false),
+		ably.WithUseTokenAuth(true),
+		ably.WithFallbackHosts([]string{"sandbox-a-fallback.ably-realtime.com"}),
+		ably.WithRealtimeRequestTimeout(2*time.Second),
+		ably.WithRealtimeHost(serverURL.Host))
+
+	defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+	err = ablytest.Wait(ablytest.ConnWaiter(realtime, realtime.Connect, ably.ConnectionEventConnected), nil)
+	assert.NoError(t, err)
+
+	assert.Equal(t, "sandbox-a-fallback.ably-realtime.com", realtime.Rest().ActiveRealtimeHost())
+}
+
 func checkUnique(ch chan string, typ string, n int) error {
 	close(ch)
 	uniq := make(map[string]struct{}, n)