chore: adds FDv2 source using composite datasource (#191)

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

- [ ] I have validated my changes against all supported platform
versions
Not trivially possible yet, still need to hook up more code to what
@kinyoklion has written.


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduces FDv2 data source built from composable sources; refactors
CompositeSource to delegate-based factories with async action processing
and exhaustion/error reporting; adds extensive tests and test helpers.
> 
> - **FDv2 Data Source**:
> - Add `FDv2DataSource` with initializer/synchronizer composition using
`CompositeSource`.
> - Implement action appliers: fast fallback for initializers, timed
fallback/recovery for synchronizers, and blacklist-on-success/off.
> - **CompositeSource Refactor**:
> - Replace `ISourceFactory`/`IActionApplierFactory` with delegate types
`SourceFactory` and `ActionApplierFactory`.
> - Process enqueued actions on a background thread; allow disposal
while actions are queued.
> - Report `Off` directly with optional `ErrorInfo`; emit exhaustion
error when no sources remain.
> - Build sinks via `FanOutDataSourceUpdates` and disableable tracker;
maintain current entry for blacklist/remove.
> - **APIs**:
> - Add delegates `SourceFactory` and `ActionApplierFactory`; remove
corresponding interfaces.
> - **Tests**:
> - Expand `CompositeSourceTest` for fallback, blacklist, disposal
semantics, exhaustion, and disabled-source isolation.
> - Add comprehensive `FDv2DataSourceTest` covering
initializer/synchronizer flows, failures, and edge cases.
>   - Enhance `SourcesListTest` with non-circular end-of-list behavior.
> - **Test Utilities**:
> - Extend `CapturingDataSourceUpdates` to record and wait for ordered
status updates.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
370e94e. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: tanderson-ld

pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/ActionApplierFactory.cs

@@ -0,0 +1,12 @@
+using LaunchDarkly.Sdk.Server.Subsystems;
+
+namespace LaunchDarkly.Sdk.Server.Internal.DataSources
+{
+    /// <summary>
+    /// A function type that creates a new <see cref="IActionApplier"/> instance.
+    /// </summary>
+    /// <param name="actionable">the <see cref="ICompositeSourceActionable"/> is the entity that actions should be applied to</param>
+    /// <returns>a new <see cref="IActionApplier"/> instance</returns>
+    internal delegate IActionApplier ActionApplierFactory(ICompositeSourceActionable actionable);
+}
+

pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/CompositeSource.cs

@@ -8,8 +8,8 @@ namespace LaunchDarkly.Sdk.Server.Internal.DataSources
 {
     /// <summary>
     /// A composite source is a source that can dynamically switch between sources with the
-    /// help of a list of <see cref="ISourceFactory"/> instances and <see cref="IActionApplierFactory"/> instances.
-    /// The ISourceFactory instances are used to create the data sources, and the IActionApplierFactory creates the action appliers that are used
+    /// help of a list of <see cref="SourceFactory"/> delegates and <see cref="ActionApplierFactory"/> delegates.
+    /// The SourceFactory delegates are used to create the data sources, and the ActionApplierFactory creates the action appliers that are used
     ///  to apply actions to the composite source as updates are received from the data sources.
     /// </summary>
     internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable
@@ -25,13 +25,13 @@ internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable
 
         private readonly IDataSourceUpdates _originalUpdateSink;
         private readonly IDataSourceUpdates _sanitizedUpdateSink;
-        private readonly SourcesList<(ISourceFactory Factory, IActionApplierFactory ActionApplierFactory)> _sourcesList;
+        private readonly SourcesList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory)> _sourcesList;
         private readonly DisableableDataSourceUpdatesTracker _disableableTracker;
 
         // Tracks the entry from the sources list that was used to create the current
         // data source instance. This allows operations such as blacklist to remove
         // the correct factory/action-applier-factory tuple from the list.
-        private (ISourceFactory Factory, IActionApplierFactory ActionApplierFactory) _currentEntry;
+        private (SourceFactory Factory, ActionApplierFactory ActionApplierFactory) _currentEntry;
         private IDataSource _currentDataSource;
 
         /// <summary>
@@ -42,7 +42,7 @@ internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable
         /// <param name="circular">whether to loop off the end of the list back to the start when fallback occurs</param>
         public CompositeSource(
             IDataSourceUpdates updatesSink,
-            IList<(ISourceFactory Factory, IActionApplierFactory ActionApplierFactory)> factoryTuples,
+            IList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory)> factoryTuples,
             bool circular = true)
         {
             if (updatesSink is null)
@@ -60,7 +60,7 @@ public CompositeSource(
             // this tracker is used to disconnect the current source from the updates sink when it is no longer needed.
             _disableableTracker = new DisableableDataSourceUpdatesTracker();
 
-            _sourcesList = new SourcesList<(ISourceFactory SourceFactory, IActionApplierFactory ActionApplierFactory)>(
+            _sourcesList = new SourcesList<(SourceFactory SourceFactory, ActionApplierFactory ActionApplierFactory)>(
                 circular: circular,
                 initialList: factoryTuples
             );
@@ -86,6 +86,11 @@ public Task<bool> Start()
         /// Disposes of the composite data source.
         /// </summary>
         public void Dispose()
+        {
+            InternalDispose();
+        }
+
+        private void InternalDispose(DataSourceStatus.ErrorInfo? error = null)
         {
             // When disposing the whole composite, we bypass the action queue and tear
             // down the current data source immediately while still honoring the same
@@ -104,23 +109,23 @@ public void Dispose()
                 // clear any queued actions and reset processing state
                 _pendingActions.Clear();
                 _isProcessingActions = false;
-                _sourcesList.Reset();
                 _currentEntry = default;
-                
                 _disposed = true;
             }
 
             // report state Off directly to the original sink, bypassing the sanitizer
             // which would map Off to Interrupted (that mapping is only for underlying sources)
-            _originalUpdateSink.UpdateStatus(DataSourceState.Off, null);
+            _originalUpdateSink.UpdateStatus(DataSourceState.Off, error);
         }
 
         /// <summary>
         /// Enqueue a state-changing operation to be executed under the shared lock.
-        /// If no other operation is currently running, this will synchronously process
-        /// the queue in a simple loop on the current thread. Any re-entrant calls from
-        /// within the operations will only enqueue more work; they will not trigger
-        /// another processing loop, so the call stack does not grow with the queue length.
+        /// If no other operation is currently running, this will asynchronously process
+        /// the queue on a background thread. Any re-entrant calls from within the operations
+        /// will only enqueue more work; they will not trigger another processing loop, so
+        /// the call stack does not grow with the queue length. Processing actions on a
+        /// background thread prevents blocking the calling thread and allows operations like
+        /// disposal to proceed even when actions are continuously being enqueued.
         /// </summary>
         private void EnqueueAction(Action action)
         {
@@ -137,12 +142,14 @@ private void EnqueueAction(Action action)
 
             if (shouldProcess)
             {
-                ProcessQueuedActions();
+                // Process actions on a background thread to prevent blocking the caller
+                // and allow Start() to return even when actions are continuously enqueued
+                _ = Task.Run(() => ProcessQueuedActions());
             }
         }
 
         /// <summary>
-        /// Processes the queued actions.
+        /// Processes the queued actions on a background thread.
         /// </summary>
         private void ProcessQueuedActions()
         {
@@ -151,7 +158,8 @@ private void ProcessQueuedActions()
                 Action action;
                 lock (_lock)
                 {
-                    if (_pendingActions.Count == 0)
+                    // Check if disposed to allow disposal to interrupt action processing
+                    if (_disposed || _pendingActions.Count == 0)
                     {
                         _isProcessingActions = false;
                         return;
@@ -177,9 +185,9 @@ private void ProcessQueuedActions()
             }
         }
 
+        // This method must only be called while holding _lock.
         private void TryFindNextUnderLock()
         {
-            // This method must only be called while holding _lock.
             if (_currentDataSource != null)
             {
                 return;
@@ -188,14 +196,22 @@ private void TryFindNextUnderLock()
             var entry = _sourcesList.Next();
             if (entry.Factory == null)
             {
+                // Failed to find a next source, report error and shut down the composite source
+                var errorInfo = new DataSourceStatus.ErrorInfo
+                {
+                    Kind = DataSourceStatus.ErrorKind.Unknown,
+                    Message = "CompositeDataSource has exhausted all available sources.",
+                    Time = DateTime.Now
+                };
+                InternalDispose(errorInfo);
                 return;
             }
 
             // Build the list of update sinks conditionally based on whether we have an action applier factory
             var updateSinks = new List<IDataSourceUpdates>();
             if (entry.ActionApplierFactory != null)
             {
-                var actionApplier = entry.ActionApplierFactory.CreateActionApplier(this);
+                var actionApplier = entry.ActionApplierFactory(this);
                 updateSinks.Add(actionApplier);
             }
             updateSinks.Add(_sanitizedUpdateSink);
@@ -205,7 +221,7 @@ private void TryFindNextUnderLock()
             var disableableUpdates = _disableableTracker.WrapAndTrack(fanout);
 
             _currentEntry = entry;
-            _currentDataSource = entry.Factory.CreateSource(disableableUpdates);
+            _currentDataSource = entry.Factory(disableableUpdates);
         }
 
         #region ICompositeSourceActionable

pkgs/sdk/server/src/Internal/DataSources/CompositeDataSource/IActionApplierFactory.cs

@@ -0,0 +1,12 @@
+using LaunchDarkly.Sdk.Server.Subsystems;
+
+namespace LaunchDarkly.Sdk.Server.Internal.DataSources
+{
+    /// <summary>
+    /// A function type that creates a new <see cref="IDataSource"/> instance.
+    /// </summary>
+    /// <param name="updatesSink">the updates sink for the new source</param>
+    /// <returns>a new <see cref="IDataSource"/> instance</returns>
+    internal delegate IDataSource SourceFactory(IDataSourceUpdates updatesSink);
+}
+