chore: Separate status check from synchronizer functionality

In the previous setup, we would only check the fallback or recovery
conditions once the synchronizer returned an update. If the synchronizer
was stuck, or nothing was changing in the environment, we would never
check the conditions.

This configuration also exposed an interesting behavior. If the
synchronizer cannot connect, it will emit error updates. Each time we
receive an error, we check if we have failed to initialize for the last
10 seconds. If so, we re-create the primary synchronizer.

When it continues to fail, the first update will trigger the condition
check. And since it has still failed for 10 seconds, it will immediately
error out. With this change, we can be assured a synchronizer is given
at least 10 seconds to try before the condition is evaluated.

Author: keelerm84

ldclient/impl/datasourcev2/status.py

@@ -19,7 +19,7 @@
 class DataSourceStatusProviderImpl(DataSourceStatusProvider):
     def __init__(self, listeners: Listeners):
         self.__listeners = listeners
-        self.__status = DataSourceStatus(DataSourceState.INITIALIZING, 0, None)
+        self.__status = DataSourceStatus(DataSourceState.INITIALIZING, time.time(), None)
         self.__lock = ReadWriteLock()
 
     @property

ldclient/impl/datasourcev2/streaming.py

@@ -405,13 +405,6 @@ def _handle_error(self, error: Exception, envid: Optional[str]) -> Tuple[Optiona
 
         return (update, True)
 
-    # magic methods for "with" statement (used in testing)
-    def __enter__(self):
-        return self
-
-    def __exit__(self, type, value, traceback):
-        self.stop()
-
 
 class StreamingDataSourceBuilder:  # disable: pylint: disable=too-few-public-methods
     """

ldclient/impl/datasystem/fdv2.py

@@ -2,6 +2,7 @@
 import time
 from threading import Event, Thread
 from typing import Any, Callable, Dict, List, Mapping, Optional
+from queue import Queue, Empty
 
 from ldclient.config import Builder, Config, DataSystemConfig
 from ldclient.feature_store import _FeatureStoreDataSetSorter
@@ -367,11 +368,12 @@ def synchronizer_loop(self: 'FDv2'):
                         else:
                             log.info("Fallback condition met")
 
-                        if self._secondary_synchronizer_builder is None:
-                            continue
                         if self._stop_event.is_set():
                             break
 
+                        if self._secondary_synchronizer_builder is None:
+                            continue
+
                         self._lock.lock()
                         secondary_sync = self._secondary_synchronizer_builder(self._config)
                         if isinstance(secondary_sync, DiagnosticSource) and self._diagnostic_accumulator is not None:
@@ -433,8 +435,45 @@ def _consume_synchronizer_results(
 
         :return: Tuple of (should_remove_sync, fallback_to_fdv1)
         """
+        action_queue = Queue()
+        timer = RepeatingTask(
+            label="FDv2-sync-cond-timer",
+            interval=10,
+            initial_delay=10,
+            callable=lambda: action_queue.put("check")
+        )
+
+        def reader(self: 'FDv2'):
+            try:
+                for update in synchronizer.sync(self._store):
+                    action_queue.put(update)
+            finally:
+                action_queue.put("quit")
+
+        sync_reader = Thread(
+            target=reader,
+            name="FDv2-sync-reader",
+            args=(self,),
+            daemon=True
+        )
+
         try:
-            for update in synchronizer.sync(self._store):
+            timer.start()
+            sync_reader.start()
+
+            while True:
+                update = action_queue.get(True)
+                if isinstance(update, str):
+                    if update == "quit":
+                        break
+
+                    if update == "check":
+                        # Check condition periodically
+                        current_status = self._data_source_status_provider.status
+                        if condition_func(current_status):
+                            return False, False
+                    continue
+
                 log.info("Synchronizer %s update: %s", synchronizer.name, update.state)
                 if self._stop_event.is_set():
                     return False, False
@@ -457,17 +496,14 @@ def _consume_synchronizer_results(
                 # Check for OFF state indicating permanent failure
                 if update.state == DataSourceState.OFF:
                     return True, False
-
-                # Check condition periodically
-                current_status = self._data_source_status_provider.status
-                if condition_func(current_status):
-                    return False, False
-
         except Exception as e:
             log.error("Error consuming synchronizer results: %s", e)
             return True, False
         finally:
             synchronizer.stop()
+            timer.stop()
+
+            sync_reader.join(0.5)
 
         return True, False