feat: adding file data source as an intializer (#381)

**Requirements**

- [x] I have added test coverage for new or changed functionality
- [x] I have followed the repository's [pull request submission
guidelines](../blob/main/CONTRIBUTING.md#submitting-pull-requests)
- [x] I have validated my changes against all supported platform
versions

This PR will:
1. add the file data source to the custom fdv2 datasystem configuration
2. modify the initialization logic so it aligns with spec. Namely, we
will continue through the initializer list until we either run out of
intializers (in which case the synchronizers will try to do a valid
initialization) OR if an intializer fetched a proper basis with a
selector.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduces a file-based initializer and updates FDv2 to mark ready
only when an initializer provides a selector, with tests validating
initializer ordering and early completion.
> 
> - **Core**:
> - Add `file_ds_builder(paths)` returning `_FileDataSourceV2`
initializer in `ldclient/datasystem.py`; import `_FileDataSourceV2`.
> - **FDv2 Initialization Logic** (`ldclient/impl/datasystem/fdv2.py`):
> - After applying an initializer basis, set ready only if
`basis.change_set.selector` is defined; then return to proceed to
synchronizers.
> - **Tests**
(`ldclient/testing/impl/datasystem/test_fdv2_datasystem.py`):
> - Add tests ensuring initializers run until first success and stop on
first initializer that provides a selector; verify transition to
synchronizers.
> - Include file-based data source usage via `file_ds_builder` with temp
JSON files.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
04a2c53. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: joker23

ldclient/datasystem.py

@@ -16,6 +16,9 @@
     StreamingDataSource,
     StreamingDataSourceBuilder
 )
+from ldclient.impl.integrations.files.file_data_sourcev2 import (
+    _FileDataSourceV2
+)
 from ldclient.interfaces import (
     DataStoreMode,
     FeatureStore,
@@ -125,6 +128,13 @@ def builder(config: LDConfig) -> StreamingDataSource:
     return builder
 
 
+def file_ds_builder(paths: List[str]) -> Builder[Initializer]:
+    def builder(_: LDConfig) -> Initializer:
+        return _FileDataSourceV2(paths)
+
+    return builder
+
+
 def default() -> ConfigBuilder:
     """
     Default is LaunchDarkly's recommended flag data acquisition strategy.

ldclient/impl/datasystem/fdv2.py

@@ -409,9 +409,10 @@ def _run_initializers(self, set_on_ready: Event):
                 # Apply the basis to the store
                 self._store.apply(basis.change_set, basis.persist)
 
-                # Set ready event
-                if not set_on_ready.is_set():
+                # Set ready event if an only if a selector is defined for the changeset
+                if basis.change_set.selector is not None and basis.change_set.selector.is_defined():
                     set_on_ready.set()
+                    return
             except Exception as e:
                 log.error("Initializer failed with exception: %s", e)
 

ldclient/testing/impl/datasystem/test_fdv2_datasystem.py

@@ -1,11 +1,14 @@
 # pylint: disable=missing-docstring
 
+import os
+import tempfile
 from threading import Event
 from typing import List
 
 from mock import Mock
 
 from ldclient.config import Config, DataSystemConfig
+from ldclient.datasystem import file_ds_builder
 from ldclient.impl.datasystem import DataAvailability
 from ldclient.impl.datasystem.fdv2 import FDv2
 from ldclient.integrations.test_datav2 import TestDataV2
@@ -432,3 +435,113 @@ def test_fdv2_stays_on_fdv1_after_fallback():
     store = fdv2.store
     flag = store.get(FEATURES, "fdv1-flag", lambda x: x)
     assert flag is not None
+
+
+def test_fdv2_initializer_should_run_until_success():
+    """
+    Test that FDv2 initializers will run in order until a successful run. Then
+    the datasystem is expected to transition to run synchronizers.
+    """
+    initial_flag_data = '''
+{
+  "flags": {
+    "feature-flag": {
+      "key": "feature-flag",
+      "version": 0,
+      "on": false,
+      "fallthrough": {
+        "variation": 0
+      },
+      "variations": ["off", "on"]
+    }
+  }
+}
+'''
+    f, path = tempfile.mkstemp(suffix='.json')
+    try:
+        os.write(f, initial_flag_data.encode("utf-8"))
+        os.close(f)
+
+        td_initializer = TestDataV2.data_source()
+        td_initializer.update(td_initializer.flag("feature-flag").on(True))
+
+        # We actually do not care what this synchronizer does.
+        td_synchronizer = TestDataV2.data_source()
+
+        data_system_config = DataSystemConfig(
+            initializers=[file_ds_builder([path]), td_initializer.build_initializer],
+            primary_synchronizer=td_synchronizer.build_synchronizer,
+        )
+
+        set_on_ready = Event()
+        synchronizer_ran = Event()
+        fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config)
+        count = 0
+
+        def listener(_: FlagChange):
+            nonlocal count
+            count += 1
+            if count == 3:
+                synchronizer_ran.set()
+
+        fdv2.flag_tracker.add_listener(listener)
+
+        fdv2.start(set_on_ready)
+        assert set_on_ready.wait(1), "Data system did not become ready in time"
+        assert synchronizer_ran.wait(1), "Data system did not transition to synchronizer"
+    finally:
+        os.remove(path)
+
+
+def test_fdv2_should_finish_initialization_on_first_successful_initializer():
+    """
+    Test that when a FDv2 initializer returns a basis and selector that the rest
+    of the intializers will be skipped and the client starts synchronizing phase.
+    """
+    initial_flag_data = '''
+{
+  "flags": {
+    "feature-flag": {
+      "key": "feature-flag",
+      "version": 0,
+      "on": false,
+      "fallthrough": {
+        "variation": 0
+      },
+      "variations": ["off", "on"]
+    }
+  }
+}
+'''
+    f, path = tempfile.mkstemp(suffix='.json')
+    try:
+        os.write(f, initial_flag_data.encode("utf-8"))
+        os.close(f)
+
+        td_initializer = TestDataV2.data_source()
+        td_initializer.update(td_initializer.flag("feature-flag").on(True))
+
+        # We actually do not care what this synchronizer does.
+        td_synchronizer = TestDataV2.data_source()
+
+        data_system_config = DataSystemConfig(
+            initializers=[td_initializer.build_initializer, file_ds_builder([path])],
+            primary_synchronizer=None,
+        )
+
+        set_on_ready = Event()
+        fdv2 = FDv2(Config(sdk_key="dummy"), data_system_config)
+        count = 0
+
+        def listener(_: FlagChange):
+            nonlocal count
+            count += 1
+
+        fdv2.flag_tracker.add_listener(listener)
+
+        fdv2.start(set_on_ready)
+        assert set_on_ready.wait(1), "Data system did not become ready in time"
+        assert count == 1, "Invalid initializer process"
+        fdv2.stop()
+    finally:
+        os.remove(path)