From 584a74c30f77e2b0e7cecbaf69cf6217a6d50f2e Mon Sep 17 00:00:00 2001 From: Alex Nalin Date: Mon, 29 Jun 2026 11:37:24 +0100 Subject: [PATCH] fix(datafeed): retry ClientPayloadError and reset _running so the loop can restart A truncated datafeed read raises aiohttp.ClientPayloadError, which was not classified as a transient error, so read_datafeed_retry re-raised it and the datafeed loop crashed. The loop then could not be restarted because AbstractDatafeedLoop._run_loop left self._running = True on an abnormal exit (only stop() reset it), so DatafeedLoopV2.start() raised "The datafeed service V2 is already started" on every restart. The combined effect turned a transient network blip into a permanent, silent event-loss outage. Changes: - strategy.is_client_timeout_error now treats ClientPayloadError as transient (also benefits datahose and auth-refresh paths). - AbstractDatafeedLoop._run_loop resets self._running = False in a finally, so the loop is restartable after any exit (stop, cancellation, or escaped exception). The concurrent-start guard in V1/V2 start() is unchanged. - Add regression tests for both behaviours. Co-Authored-By: Claude Opus 4.8 --- symphony/bdk/core/retry/strategy.py | 9 +++---- .../datafeed/abstract_datafeed_loop.py | 12 ++++++++-- tests/core/retry/strategy_test.py | 10 ++++++++ .../datafeed/abstract_datafeed_loop_test.py | 24 +++++++++++++++++++ 4 files changed, 49 insertions(+), 6 deletions(-) diff --git a/symphony/bdk/core/retry/strategy.py b/symphony/bdk/core/retry/strategy.py index 2fa56670..e753b5a5 100644 --- a/symphony/bdk/core/retry/strategy.py +++ b/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: diff --git a/symphony/bdk/core/service/datafeed/abstract_datafeed_loop.py b/symphony/bdk/core/service/datafeed/abstract_datafeed_loop.py index aa9d0d9c..38339335 100644 --- a/symphony/bdk/core/service/datafeed/abstract_datafeed_loop.py +++ b/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): diff --git a/tests/core/retry/strategy_test.py b/tests/core/retry/strategy_test.py index d2bc5449..7d7c6b04 100644 --- a/tests/core/retry/strategy_test.py +++ b/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 = [ diff --git a/tests/core/service/datafeed/abstract_datafeed_loop_test.py b/tests/core/service/datafeed/abstract_datafeed_loop_test.py index a6c86073..710d6e1a 100644 --- a/tests/core/service/datafeed/abstract_datafeed_loop_test.py +++ b/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)