fix: Prevent a persistent-store outage from throwing in the sync FDv2 evaluation (#506)

Author: jsonbailey

ldclient/client.py

@@ -408,8 +408,9 @@ def _evaluate_internal(self, key: str, context: Context, default: Any, event_fac
         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_system.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 +480,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_system.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

@@ -380,7 +380,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/impl/datasystem/fdv2_common.py

@@ -398,10 +398,16 @@ def data_availability(self) -> DataAvailability:
         if self._store.selector().is_defined():
             return DataAvailability.REFRESHED
 
-        if not self._configured_with_data_sources or self._store.is_initialized():
+        if not self._configured_with_data_sources:
             return DataAvailability.CACHED
 
-        return DataAvailability.DEFAULTS
+        try:
+            store_initialized = self._store.is_initialized()
+        except Exception as e:
+            log.error("Error checking persistent store readiness; treating data as unavailable: %s", e)
+            return DataAvailability.DEFAULTS
+
+        return DataAvailability.CACHED if store_initialized else DataAvailability.DEFAULTS
 
     @property
     def target_availability(self) -> DataAvailability:

ldclient/impl/datasystem/store.py

@@ -521,7 +521,7 @@ def __mapping(data: Dict[str, ModelEntity]) -> Dict[str, Dict[str, Any]]:
                     return e
         return None
 
-    def close(self) -> Optional[Exception]:
+    def close(self) -> None:
         """Close the store and any persistent store if configured."""
         with self._lock:
             if self._persistent_store is not None:
@@ -532,8 +532,7 @@ def close(self) -> Optional[Exception]:
                     if callable(close):
                         close()
                 except Exception as e:
-                    return e
-        return None
+                    log.warning("Error closing the persistent store: %s", e)
 
     def get_data_store_status_provider(self) -> Optional[DataStoreStatusProvider]:
         """Get the data store status provider for the persistent store, if configured."""

ldclient/testing/impl/datasystem/test_fdv2_persistence.py

@@ -3,11 +3,13 @@
 from threading import Event
 from typing import Any, Callable, Dict, List, Mapping, Optional
 
+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 +784,86 @@ 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 must not raise: it degrades to DEFAULTS instead.
+    assert fdv2.data_availability == DataAvailability.DEFAULTS
+
+    # Drive the same 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 persistent store readiness" in record.message
+        for record in caplog.records
+        if record.levelname == "ERROR"
+    )
+
+
+def test_persistent_store_close_logs_and_swallows_error(caplog):
+    """A persistent-store close error is logged as a warning, not raised."""
+    from ldclient.impl.datasystem.store import Store
+    from ldclient.impl.listeners import Listeners
+
+    class ClosingFailsStore(StubFeatureStore):
+        def close(self):
+            raise RuntimeError("close boom")
+
+    store = Store(Listeners(), Listeners())
+    store.with_persistence(ClosingFailsStore(), True, None)
+
+    # close() must log the error rather than raise it.
+    store.close()
+
+    assert any(
+        "Error closing the persistent store" in record.message
+        for record in caplog.records
+        if record.levelname == "WARNING"
+    )