fix(datafeed): retry ClientPayloadError and reset _running so the loo…

symphony/bdk/core/retry/strategy.py

@@ -1,6 +1,6 @@
 from asyncio import TimeoutError
 
-from aiohttp import ClientConnectionError
+from aiohttp import ClientConnectionError, ClientPayloadError
 from tenacity import RetryCallState
 
 from symphony.bdk.core.auth.exception import AuthUnauthorizedError
@@ -16,12 +16,13 @@ def is_client_error(exception: Exception) -> bool:
 
 
 def is_client_timeout_error(exception: Exception):
-    """Checks if the exception is a client timeout error
+    """Checks if the exception is a client timeout or transient stream error
 
     :param exception: The exception to be checked
-    :return: True if checks the predicate, False otherwise
+    :return: True for connection errors, asyncio timeouts, and truncated/aborted
+             streaming payloads (ClientPayloadError), False otherwise
     """
-    return isinstance(exception, ClientConnectionError) or isinstance(exception, TimeoutError)
+    return isinstance(exception, (ClientConnectionError, ClientPayloadError, TimeoutError))
 
 
 def can_authentication_be_retried(exception: Exception) -> bool:

symphony/bdk/core/service/datafeed/abstract_datafeed_loop.py

@@ -137,8 +137,16 @@ def unsubscribe(self, listener: RealTimeEventListener):
 
     async def _run_loop(self):
         self._running = True
-        while self._running:
-            await self._run_loop_iteration()
+        try:
+            while self._running:
+                await self._run_loop_iteration()
+        finally:
+            # Allow the loop to be restarted after ANY exit (explicit stop,
+            # cancellation, or an exception escaping retry). Without this, _running
+            # stays True after an abnormal crash and a subsequent start() raises
+            # "already started" forever, turning a transient error into a permanent
+            # silent outage.
+            self._running = False
 
     @abstractmethod
     async def _run_loop_iteration(self):

tests/core/retry/strategy_test.py

@@ -144,6 +144,16 @@ async def test_unexpected_api_exception_is_raised(self):
         assert thing.call_count == 1
 
 
+def test_client_payload_error_is_classified_as_transient():
+    """A truncated/aborted streaming payload (ClientPayloadError) must be treated as a
+    transient error so the datafeed/datahose loops reconnect instead of crashing."""
+    err = aiohttp.ClientPayloadError("Response payload is not completed")
+
+    assert strategy.is_client_timeout_error(err) is True
+    assert strategy.is_network_or_minor_error(err) is True
+    assert strategy.is_network_or_minor_error_or_client(err) is True
+
+
 @pytest.mark.asyncio
 async def test_should_retry():
     strategies = [

tests/core/service/datafeed/abstract_datafeed_loop_test.py

@@ -145,6 +145,30 @@ async def test_unexpected_error_should_be_propagated_and_call_stop_tasks(bare_df
     bare_df_loop._stop_listener_tasks.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_running_flag_reset_after_crash_allows_restart(
+    bare_df_loop, run_iteration_side_effect
+):
+    # After an exception escapes the loop, _running must be reset to False so the loop
+    # is restartable. Previously it stayed True, so DatafeedLoopV2.start() would raise
+    # "already started" forever, turning a transient error into a permanent outage.
+    bare_df_loop._run_loop_iteration.side_effect = ValueError("boom")
+
+    with pytest.raises(ValueError):
+        await bare_df_loop.start()
+
+    assert bare_df_loop._running is False
+
+    # A subsequent start() must run again and stop cleanly.
+    bare_df_loop._run_loop_iteration.side_effect = run_iteration_side_effect
+    t = asyncio.create_task(bare_df_loop.start())
+    await asyncio.sleep(0.001)
+    await bare_df_loop.stop()
+    await t
+
+    assert bare_df_loop._running is False
+
+
 @pytest.mark.asyncio
 async def test_create_listener_tasks_none(df_loop, listener):
     tasks = await df_loop._create_listener_tasks(None)