Catch generic exception to have last chance to exit clearly.

Improve giveup handler.
Apply backoff policy to generic websocket errors.

Author: Francesco Faraone

connect/eaas/exceptions.py

@@ -13,3 +13,7 @@ class MaintenanceError(EaaSError):
 
 class CommunicationError(EaaSError):
     pass
+
+
+class StopBackoffError(EaaSError):
+    pass

connect/eaas/worker.py

@@ -34,6 +34,7 @@
 from connect.eaas.exceptions import (
     CommunicationError,
     MaintenanceError,
+    StopBackoffError,
 )
 from connect.eaas.handler import ExtensionHandler
 from connect.eaas.managers import (
@@ -168,7 +169,7 @@ def get_capabilities(self):
             ),
         )
 
-    async def communicate(self):
+    async def communicate(self):  # noqa: CCR001
         @backoff.on_exception(
             backoff.expo,
             CommunicationError,
@@ -177,7 +178,7 @@ async def communicate(self):
             max_value=_get_max_retry_delay_time,
             jitter=backoff.random_jitter,
             on_backoff=_on_communication_backoff,
-            giveup=lambda _: not self.run_event.is_set(),
+            giveup=self._backoff_shutdown,
         )
         @backoff.on_exception(
             backoff.expo,
@@ -187,7 +188,7 @@ async def communicate(self):
             max_value=_get_max_retry_delay_time,
             jitter=backoff.random_jitter,
             on_backoff=_on_communication_backoff,
-            giveup=lambda _: not self.run_event.is_set(),
+            giveup=self._backoff_shutdown,
         )
         async def _do_communicate():
             try:
@@ -209,6 +210,12 @@ async def _do_communicate():
                 else:
                     logger.warning(f'InvalidStatusCode {ic.status_code} raised.')
                     raise CommunicationError()
+            except WebSocketException as wse:
+                logger.warning(f'Unexpected websocket exception {wse}.')
+                raise CommunicationError()
+            except Exception as e:
+                logger.warning(f'Unexpected error in communicate: {e}.')
+                raise CommunicationError()
 
         await _do_communicate()
 
@@ -225,9 +232,6 @@ async def run(self):  # noqa: CCR001
                 await self.communicate()
             except ConnectionClosedOK:
                 self.run_event.clear()
-            except WebSocketException:
-                logger.exception('Unexpected websocket exception. Retrying in 2 seconds.')
-                await asyncio.sleep(2)
             except CommunicationError:
                 logger.error(
                     f'Max retries exceeded after {MAX_RETRY_TIME_GENERIC_SECONDS} seconds',
@@ -239,6 +243,8 @@ async def run(self):  # noqa: CCR001
                     'seconds',
                 )
                 self.run_event.clear()
+            except StopBackoffError:
+                pass
         if self.ws:
             await self.ws.close()
 
@@ -370,3 +376,8 @@ def stop(self):
         """
         logger.info('Stopping control worker...')
         self.run_event.clear()
+
+    def _backoff_shutdown(self, _):
+        if not self.run_event.is_set():
+            logger.info('Worker exiting, stop backoff loop')
+            raise StopBackoffError()

tests/test_worker.py

@@ -1083,3 +1083,41 @@ def test__on_communication_backoff(caplog, tries, ordinal):
     with caplog.at_level(logging.INFO):
         _on_communication_backoff(details)
     assert expected in caplog.records[0].message
+
+
+@pytest.mark.asyncio
+async def test_connection_unexpected_error(mocker, ws_server, unused_port, caplog):
+    mocker.patch('connect.eaas.worker.MAX_RETRY_TIME_MAINTENANCE_SECONDS', 1)
+    mocker.patch('connect.eaas.worker.MAX_RETRY_DELAY_TIME_SECONDS', 1)
+    mocker.patch('connect.eaas.handler.get_extension_class')
+    mocker.patch('connect.eaas.handler.get_extension_type')
+    mocker.patch(
+        'connect.eaas.config.get_environment',
+        return_value={
+            'ws_address': f'127.0.0.1:{unused_port}',
+            'api_address': f'127.0.0.1:{unused_port}',
+            'api_key': 'SU-000:XXXX',
+            'environment_id': 'ENV-000-0001',
+            'instance_id': 'INS-000-0002',
+            'background_task_max_execution_time': 300,
+            'interactive_task_max_execution_time': 120,
+            'scheduled_task_max_execution_time': 43200,
+        },
+    )
+    handler = WSHandler(
+        '/public/v1/devops/ws/ENV-000-0001/INS-000-0002?running_tasks=0&running_scheduled_tasks=0',
+        None,
+        [],
+    )
+
+    async with ws_server(handler):
+        worker = Worker(secure=False)
+        worker.send = mocker.AsyncMock(side_effect=RuntimeError('generic error'))
+        with caplog.at_level(logging.INFO):
+            task = asyncio.create_task(worker.start())
+            await asyncio.sleep(.5)
+            worker.stop()
+            await task
+
+    assert 'Unexpected error in communicate: generic error' in caplog.text
+    assert 'Backing off ' in caplog.text