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()