fix: Do not propagate persistent-store errors from the sync FDv2 warm-start check

Author: jsonbailey

ldclient/client.py

@@ -402,14 +402,31 @@ def evaluate():
         hook_result = self.__evaluate_with_hooks(key=key, context=context, default_value=default_stage.value, method="migration_variation", block=evaluate)
         return hook_result.results['default_stage'], hook_result.results['tracker']
 
+    def _data_availability(self) -> DataAvailability:
+        """Reads the current data availability, degrading to ``DEFAULTS`` on error.
+
+        During the warm-start window (before a data source initializes) the
+        availability gate may query a persistent store, such as Redis. A store
+        I/O error must not propagate out of an evaluation, so treat it as "no
+        data available"; the caller then returns the default value with
+        ``CLIENT_NOT_READY`` instead of raising.
+        """
+        try:
+            return self._data_system.data_availability
+        except Exception as e:
+            log.error("Error checking data availability; treating data as unavailable: %s" % repr(e))
+            log.debug(traceback.format_exc())
+            return DataAvailability.DEFAULTS
+
     def _evaluate_internal(self, key: str, context: Context, default: Any, event_factory) -> Tuple[EvaluationDetail, Optional[FeatureFlag]]:
         default = self._config.get_default(key, default)
 
         if self._config.offline:
             return EvaluationDetail(default, None, error_reason('CLIENT_NOT_READY')), None
 
-        if self._data_system.data_availability != DataAvailability.REFRESHED:
-            if self._data_system.data_availability == DataAvailability.CACHED:
+        availability = self._data_availability()
+        if availability != DataAvailability.REFRESHED:
+            if availability == DataAvailability.CACHED:
                 log.warning("Feature Flag evaluation attempted before client has initialized - using last known values from feature store for feature key: " + key)
             else:
                 log.warning("Feature Flag evaluation attempted before client has initialized! Feature store unavailable - returning default: " + str(default) + " for feature key: " + key)
@@ -479,8 +496,9 @@ def all_flags_state(self, context: Context, **kwargs) -> FeatureFlagsState:
             log.warning("all_flags_state() called, but client is in offline mode. Returning empty state")
             return FeatureFlagsState(False)
 
-        if self._data_system.data_availability != DataAvailability.REFRESHED:
-            if self._data_system.data_availability == DataAvailability.CACHED:
+        availability = self._data_availability()
+        if availability != DataAvailability.REFRESHED:
+            if availability == DataAvailability.CACHED:
                 log.warning("all_flags_state() called before client has finished initializing! Using last known values from feature store")
             else:
                 log.warning("all_flags_state() called before client has finished initializing! Feature store unavailable - returning empty state")

ldclient/impl/datasystem/fdv2.py

@@ -392,7 +392,7 @@ def _consume_synchronizer_results(
         """
         Consume results from a synchronizer until a condition is met or it fails.
 
-        :return: Tuple of (should_remove_sync, fallback_to_fdv1, directive)
+        :return: the ConditionDirective describing how to proceed
         """
         action_queue: Queue = Queue()
         timer = RepeatingTask(

ldclient/testing/impl/datasystem/test_fdv2_persistence.py

@@ -3,11 +3,15 @@
 from threading import Event
 from typing import Any, Callable, Dict, List, Mapping, Optional
 
+import pytest
+
+from ldclient.client import Context
 from ldclient.config import Config, DataSystemConfig
 from ldclient.impl.datasystem import DataAvailability
 from ldclient.impl.datasystem.fdv2 import FDv2
 from ldclient.integrations.test_datav2 import TestDataV2
 from ldclient.interfaces import DataStoreMode, FeatureStore, FlagChange
+from ldclient.testing.test_ldclient import make_client
 from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind
 
 
@@ -782,3 +786,65 @@ def init(self, all_data):
     assert err is not None, "Commit should return error from persistent store"
     assert isinstance(err, RuntimeError)
     assert str(err) == "Simulated persistent store failure"
+
+
+class ThrowingInitializedStore(StubFeatureStore):
+    """A persistent store whose ``initialized`` check raises, to simulate a
+    store I/O error (for example a Redis connection failure) during the
+    warm-start availability gate."""
+
+    @property
+    def initialized(self) -> bool:
+        raise RuntimeError("persistent store I/O error")
+
+
+def test_variation_does_not_throw_when_persistent_store_errors_during_warm_start(caplog):
+    """A persistent-store error at the warm-start availability gate must not
+    propagate out of the client.
+
+    While a synchronizer is configured but has not yet supplied a basis, the
+    availability gate consults the persistent store's initialized state. If that
+    query raises, evaluation must degrade to the default value with
+    ``CLIENT_NOT_READY`` and ``all_flags_state()`` must return an invalid state,
+    rather than raising. This is the sync counterpart to the async fix in #486.
+    """
+    persistent_store = ThrowingInitializedStore()
+
+    # A synchronizer is configured but the data system is never started, so no
+    # basis arrives. This is the warm-start window in which the gate reads the
+    # persistent store's initialized state.
+    data_system_config = DataSystemConfig(
+        data_store_mode=DataStoreMode.READ_ONLY,
+        data_store=persistent_store,
+        initializers=None,
+        synchronizers=[TestDataV2.data_source().builder],
+    )
+    fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config)
+
+    # The gate itself raises: this is the unguarded root the fix addresses.
+    with pytest.raises(RuntimeError):
+        _ = fdv2.data_availability
+
+    # Drive the same failing gate through the client and confirm it degrades
+    # instead of propagating the error.
+    client = make_client()
+    try:
+        client._data_system = fdv2
+        context = Context.from_dict({"key": "user", "kind": "user"})
+
+        assert client.variation("flag-key", context, default="default-value") == "default-value"
+
+        detail = client.variation_detail("flag-key", context, default="default-value")
+        assert detail.value == "default-value"
+        assert detail.reason == {"kind": "ERROR", "errorKind": "CLIENT_NOT_READY"}
+        assert detail.is_default_value() is True
+
+        assert client.all_flags_state(context).valid is False
+    finally:
+        client.close()
+
+    assert any(
+        "Error checking data availability" in record.message
+        for record in caplog.records
+        if record.levelname == "ERROR"
+    )