From bf8a4a87a16b1d1fb29852c4d4db0c00eceffa8d Mon Sep 17 00:00:00 2001 From: Shintaro Inoue <6049156+sanamio@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:38:45 +0900 Subject: [PATCH] fix: retry config fetch when CDN returns a config older than the SSE update When a refetchConfig SSE message arrives before the CDN has the new config, the response is older than the message and is discarded. Nothing retried, so the SDK kept the old config until the next poll. Retry the fetch in the background with backoff (up to 5 attempts), similar to the Go server SDK. Co-Authored-By: Claude Opus 5.5 --- .../managers/config_manager.py | 51 +++++++++++-- test/managers/test_config_manager.py | 74 +++++++++++++++++++ 2 files changed, 120 insertions(+), 5 deletions(-) diff --git a/devcycle_python_sdk/managers/config_manager.py b/devcycle_python_sdk/managers/config_manager.py index cfa16f7..1ae5cc8 100755 --- a/devcycle_python_sdk/managers/config_manager.py +++ b/devcycle_python_sdk/managers/config_manager.py @@ -46,6 +46,12 @@ def __init__( self._max_reconnect_interval = 300.0 # Cap at 5 minutes self._last_reconnect_attempt_time: Optional[float] = None self._sse_reconnecting = False + # Retry when the CDN returns a config older than an SSE update + self._stale_refetch_max_attempts = 5 + self._stale_refetch_max_delay = 30.0 + self._stale_refetch_lock = threading.Lock() + self._stale_refetching = False + self._stale_refetch_last_modified: Optional[float] = None self._config_api_client = ConfigAPIClient(self._sdk_key, self._options) self._polling_enabled = True @@ -97,7 +103,8 @@ def _delayed_sse_reconnect(self, delay_seconds: float): finally: self._sse_reconnecting = False - def _get_config(self, last_modified: Optional[float] = None): + def _get_config(self, last_modified: Optional[float] = None) -> bool: + """Fetch the config. Returns False if the CDN returned a config older than last_modified.""" try: lm_header = self._config_lastmodified if last_modified is not None: @@ -110,16 +117,16 @@ def _get_config(self, last_modified: Optional[float] = None): # Abort early if the last modified is before the sent one. if new_config is None and new_etag is None and new_lastmodified is None: - return + return False if new_config is None and new_etag == self._config_etag: # api not returning data and the etag is the same # no change to the config since last request - return + return True elif new_config is None: logger.warning( "DevCycle: Config CDN fetch returned no data with a different etag" ) - return + return True trigger_on_client_initialized = self._config is None @@ -157,6 +164,38 @@ def _get_config(self, last_modified: Optional[float] = None): "DevCycle: Error initializing DevCycle: Invalid SDK key provided." ) self._polling_enabled = False + return True + + def _schedule_stale_refetch(self, last_modified: float): + """Retry the config fetch in the background until the CDN has the config from the SSE update.""" + with self._stale_refetch_lock: + if ( + self._stale_refetch_last_modified is None + or last_modified > self._stale_refetch_last_modified + ): + self._stale_refetch_last_modified = last_modified + if self._stale_refetching: + return + self._stale_refetching = True + threading.Thread(target=self._refetch_stale_config, daemon=True).start() + + def _refetch_stale_config(self): + try: + delay = 1.0 + for _ in range(self._stale_refetch_max_attempts): + time.sleep(delay) + if not self._polling_enabled: + return + if self._get_config(self._stale_refetch_last_modified): + return + delay = min(delay * 2, self._stale_refetch_max_delay) + logger.warning( + "DevCycle: Config CDN still returned an older config after retries" + ) + finally: + with self._stale_refetch_lock: + self._stale_refetching = False + self._stale_refetch_last_modified = None def get_config_metadata(self) -> Optional[ConfigMetadata]: return self._local_bucketing.get_config_metadata() @@ -189,7 +228,9 @@ def sse_message(self, message: ld_eventsource.actions.Event): or dvc_data.get("type") is None ): logger.info("DevCycle: Received refetchConfig message - updating config") - self._get_config(dvc_data["lastModified"] / 1000.0) + last_modified = dvc_data["lastModified"] / 1000.0 + if not self._get_config(last_modified): + self._schedule_stale_refetch(last_modified) # SSE connection healthy, reconnect attempts reset. if dvc_data.get("type") == "ping" or dvc_data.get("type") == "refetchConfig": self._sse_reconnect_attempts = 0 diff --git a/test/managers/test_config_manager.py b/test/managers/test_config_manager.py index 1e2fbaa..cf929b8 100644 --- a/test/managers/test_config_manager.py +++ b/test/managers/test_config_manager.py @@ -392,5 +392,79 @@ def test_successful_state_resets_reconnection_flags( config_manager.close() +class StaleConfigRefetchTest(unittest.TestCase): + def setUp(self) -> None: + self.sdk_key = "dvc_server_" + str(uuid.uuid4()) + self.test_local_bucketing = MagicMock() + self.test_options = DevCycleLocalOptions( + config_polling_interval_ms=500, disable_realtime_updates=True + ) + self.test_config_json = small_config_json() + + def _refetch_message(self): + return ld_eventsource.actions.Event( + event="message", + data='{"data": "{\\"type\\": \\"refetchConfig\\", \\"lastModified\\": 1234567890000}"}', + ) + + @patch("devcycle_python_sdk.api.config_client.ConfigAPIClient.get_config") + @patch("threading.Thread") + def test_stale_sse_fetch_schedules_refetch(self, mock_thread, mock_get_config): + mock_get_config.return_value = (None, None, None) + with patch.object(EnvironmentConfigManager, "start"): + config_manager = EnvironmentConfigManager( + self.sdk_key, self.test_options, self.test_local_bucketing + ) + + config_manager.sse_message(self._refetch_message()) + + mock_thread.assert_called_once() + self.assertEqual( + mock_thread.call_args[1]["target"], config_manager._refetch_stale_config + ) + self.assertEqual(config_manager._stale_refetch_last_modified, 1234567890.0) + + @patch("devcycle_python_sdk.api.config_client.ConfigAPIClient.get_config") + @patch("time.sleep") + def test_refetch_stops_when_cdn_returns_new_config( + self, mock_sleep, mock_get_config + ): + with patch.object(EnvironmentConfigManager, "start"): + config_manager = EnvironmentConfigManager( + self.sdk_key, self.test_options, self.test_local_bucketing + ) + mock_get_config.side_effect = [ + (None, None, None), + (self.test_config_json, "new-etag", "Wed, 21 Oct 2015 07:28:00 GMT"), + ] + config_manager._stale_refetching = True + config_manager._stale_refetch_last_modified = 1234567890.0 + + config_manager._refetch_stale_config() + + self.assertEqual(mock_get_config.call_count, 2) + self.assertEqual(config_manager._config, self.test_config_json) + self.assertFalse(config_manager._stale_refetching) + + @patch("devcycle_python_sdk.api.config_client.ConfigAPIClient.get_config") + @patch("time.sleep") + def test_refetch_gives_up_after_max_attempts(self, mock_sleep, mock_get_config): + with patch.object(EnvironmentConfigManager, "start"): + config_manager = EnvironmentConfigManager( + self.sdk_key, self.test_options, self.test_local_bucketing + ) + mock_get_config.return_value = (None, None, None) + config_manager._stale_refetching = True + config_manager._stale_refetch_last_modified = 1234567890.0 + + config_manager._refetch_stale_config() + + self.assertEqual( + mock_get_config.call_count, config_manager._stale_refetch_max_attempts + ) + self.assertIsNone(config_manager._config) + self.assertFalse(config_manager._stale_refetching) + + if __name__ == "__main__": unittest.main()