fix: retry config fetch when CDN returns a config older than the SSE update

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

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