Merge pull request #598 from ably/ECO-5305/fix-rest-retry

[ECO-5305] fix: rest retry logic

Author: ttypic

ably/http/http.py

@@ -188,6 +188,11 @@ async def make_request(self, method, path, version=None, headers=None, body=None
 
         hosts = self.get_rest_hosts()
         for retry_count, host in enumerate(hosts):
+            def should_stop_retrying():
+                time_passed = time.time() - requested_at
+                # if it's the last try or cumulative timeout is done, we stop retrying
+                return retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration
+
             base_url = "%s://%s:%d" % (self.preferred_scheme,
                                        host,
                                        self.preferred_port)
@@ -204,28 +209,33 @@ async def make_request(self, method, path, version=None, headers=None, body=None
             try:
                 response = await self.__client.send(request)
             except Exception as e:
-                # if last try or cumulative timeout is done, throw exception up
-                time_passed = time.time() - requested_at
-                if retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration:
+                if should_stop_retrying():
                     raise e
             else:
+                # RSC15l4
+                cloud_front_error = (response.headers.get('Server', '').lower() == 'cloudfront'
+                                     and response.status_code >= 400)
+                # RSC15l3
+                retryable_server_error = response.status_code >= 500 and response.status_code <= 504
+                # Resending requests that have failed for other failure conditions will not fix the problem
+                # and will simply increase the load on other datacenters unnecessarily
+                should_fallback = cloud_front_error or retryable_server_error
+
                 try:
                     if raise_on_error:
                         AblyException.raise_for_response(response)
 
+                    if should_fallback and not should_stop_retrying():
+                        continue
+
                     # Keep fallback host for later (RSC15f)
                     if retry_count > 0 and host != self.options.get_rest_host():
                         self.__host = host
                         self.__host_expires = time.time() + (self.options.fallback_retry_timeout / 1000.0)
 
                     return Response(response)
                 except AblyException as e:
-                    if not e.is_server_error:
-                        raise e
-
-                    # if last try or cumulative timeout is done, throw exception up
-                    time_passed = time.time() - requested_at
-                    if retry_count == len(hosts) - 1 or time_passed > http_max_retry_duration:
+                    if should_stop_retrying() or not should_fallback:
                         raise e
 
     async def delete(self, url, headers=None, skip_auth=False, timeout=None):

test/ably/rest/resthttp_test.py

@@ -151,6 +151,7 @@ async def test_no_retry_if_not_500_to_599_http_code(self):
 
         await ably.close()
 
+    @respx.mock
     async def test_500_errors(self):
         """
         Raise error if all the servers reply with a 5xx error.
@@ -159,16 +160,13 @@ async def test_500_errors(self):
 
         ably = AblyRest(token="foo")
 
-        def raise_ably_exception(*args, **kwargs):
-            raise AblyException(message="", status_code=500, code=50000)
+        mock_request = respx.route().mock(return_value=httpx.Response(500, text="Internal Server Error"))
 
-        with mock.patch('httpx.Request', wraps=httpx.Request):
-            with mock.patch('ably.util.exceptions.AblyException.raise_for_response',
-                            side_effect=raise_ably_exception) as send_mock:
-                with pytest.raises(AblyException):
-                    await ably.http.make_request('GET', '/', skip_auth=True)
+        with pytest.raises(AblyException):
+            await ably.http.make_request('GET', '/', skip_auth=True)
+
+        assert mock_request.call_count == 3
 
-                assert send_mock.call_count == 3
         await ably.close()
 
     def test_custom_http_timeouts(self):

test/ably/rest/restrequest_test.py

@@ -126,6 +126,101 @@ async def test_timeout(self):
             await ably.request('GET', '/time', version=Defaults.protocol_version)
         await ably.close()
 
+    # RSC15l3
+    @dont_vary_protocol
+    async def test_503_status_fallback(self):
+        default_endpoint = 'https://sandbox-rest.ably.io/time'
+        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        fallback_endpoint = f'https://{fallback_host}/time'
+        ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
+        with respx.mock:
+            default_route = respx.get(default_endpoint)
+            fallback_route = respx.get(fallback_endpoint)
+            headers = {
+                "Content-Type": "application/json"
+            }
+            default_route.return_value = httpx.Response(503, headers=headers)
+            fallback_route.return_value = httpx.Response(200, headers=headers, text='[123]')
+            result = await ably.request('GET', '/time', version=Defaults.protocol_version)
+            assert default_route.called
+            assert result.status_code == 200
+            assert result.items[0] == 123
+        await ably.close()
+
+    # RSC15l2
+    @dont_vary_protocol
+    async def test_httpx_timeout_fallback(self):
+        default_endpoint = 'https://sandbox-rest.ably.io/time'
+        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        fallback_endpoint = f'https://{fallback_host}/time'
+        ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
+        with respx.mock:
+            default_route = respx.get(default_endpoint)
+            fallback_route = respx.get(fallback_endpoint)
+            headers = {
+                "Content-Type": "application/json"
+            }
+            default_route.side_effect = httpx.ReadTimeout
+            fallback_route.return_value = httpx.Response(200, headers=headers, text='[123]')
+            result = await ably.request('GET', '/time', version=Defaults.protocol_version)
+            assert default_route.called
+            assert result.status_code == 200
+            assert result.items[0] == 123
+        await ably.close()
+
+    # RSC15l3
+    @dont_vary_protocol
+    async def test_503_status_fallback_on_publish(self):
+        default_endpoint = 'https://sandbox-rest.ably.io/channels/test/messages'
+        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        fallback_endpoint = f'https://{fallback_host}/channels/test/messages'
+
+        fallback_response_text = (
+            '{"id": "unique_id:0", "channel": "test", "name": "test", "data": "data", '
+            '"clientId": null, "connectionId": "connection_id", "timestamp": 1696944145000, '
+            '"encoding": null}'
+        )
+
+        ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
+        with respx.mock:
+            default_route = respx.post(default_endpoint)
+            fallback_route = respx.post(fallback_endpoint)
+            headers = {
+                "Content-Type": "application/json"
+            }
+            default_route.return_value = httpx.Response(503, headers=headers)
+            fallback_route.return_value = httpx.Response(
+                200,
+                headers=headers,
+                text=fallback_response_text,
+            )
+            message_response = await ably.channels['test'].publish('test', 'data')
+            assert default_route.called
+            assert message_response.to_native()['data'] == 'data'
+        await ably.close()
+
+    # RSC15l4
+    @dont_vary_protocol
+    async def test_400_cloudfront_fallback(self):
+        default_endpoint = 'https://sandbox-rest.ably.io/time'
+        fallback_host = 'sandbox-a-fallback.ably-realtime.com'
+        fallback_endpoint = f'https://{fallback_host}/time'
+        ably = await TestApp.get_ably_rest(fallback_hosts=[fallback_host])
+        with respx.mock:
+            default_route = respx.get(default_endpoint)
+            fallback_route = respx.get(fallback_endpoint)
+            headers = {
+                "Server": "CloudFront",
+                "Content-Type": "application/json",
+            }
+            default_route.return_value = httpx.Response(400, headers=headers, text='[456]')
+            fallback_route.return_value = httpx.Response(200, headers=headers, text='[123]')
+            result = await ably.request('GET', '/time', version=Defaults.protocol_version)
+            assert default_route.called
+            assert result.status_code == 200
+            assert result.items[0] == 123
+        await ably.close()
+
     async def test_version(self):
         version = "150"  # chosen arbitrarily
         result = await self.ably.request('GET', '/time', "150")