chore: Add FDv2 Data System (#196)

<!-- CURSOR_SUMMARY -->
> [!NOTE]
> Adds the FDv2 data system with a new transactional update API,
composite source orchestration, write-through store, and integrated
streaming/polling with selector support, plus extensive tests.
> 
> - **FDv2 Data System (core)**:
> - Introduces `IDataSourceUpdatesV2` (transactional updates API) and
adapts legacy sources via `DataSourceUpdatesV2ToV1Adapter`.
> - Adds `FDv2DataSystem`, `WriteThroughStore` (memory+persistent,
selector-aware), and `SelectorSourceFacade`.
> - Wires `LdClient` to use FDv2 when `Configuration.DataSystem` is set.
> - **Composite source orchestration**:
> - Refactors to `CompositeSource` using `IDataSourceObserver`,
`ObservableDataSourceUpdates`, and `DataSourceUpdatesSanitizer` (V2).
> - Replaces fan-out/`IActionApplier` with observer pattern and
disableable sinks; updates `SourceFactory`/`ActionApplierFactory`
signatures.
> - **FDv2 data sources**:
> - Builds FDv2 streaming/polling with selector support and
transactional applies.
> - `FDv2DataSource` composes initializers/synchronizers with
fallback/blacklist logic and timed recovery.
> - Streaming: ignore unknown events (debug log), exposes handled event
list.
> - **Tests & support**:
> - Extensive new/updated unit tests for composite, FDv2
streaming/polling, and data flow; contract-test suppressions added.
>   - Adds test visibility in `AssemblyInfo`.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
dd9ec86. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Todd Anderson <[email protected]>
Co-authored-by: Todd Anderson <[email protected]>

Author: kinyoklion

pkgs/sdk/server/contract-tests/test-supressions-fdv2.txt

@@ -1,10 +1,10 @@
-streaming/
-evaluation/
-hooks/
-migrations/
-secure mode hash/
-context type/
-tags/
-big segments/
-events/
-service endpoints/
+streaming/retry behavior/do not retry after unrecoverable HTTP error on initial connect/error 401
+streaming/retry behavior/do not retry after unrecoverable HTTP error on initial connect/error 403
+streaming/retry behavior/do not retry after unrecoverable HTTP error on initial connect/error 405
+streaming/retry behavior/do not retry after unrecoverable HTTP error on reconnect/error 401
+streaming/retry behavior/do not retry after unrecoverable HTTP error on reconnect/error 403
+streaming/retry behavior/do not retry after unrecoverable HTTP error on reconnect/error 405
+streaming/fdv2/reconnection state management/initializes from 2 polling initializers
+streaming/fdv2/fallback to FDv1 handling
+streaming/fdv2/disconnects on goodbye
+streaming/fdv2/reconnection state management/initializes from polling initializer

pkgs/sdk/server/src/Integrations/FDv2PollingDataSourceBuilder.cs

@@ -87,7 +87,7 @@ public IDataSource Build(LdClientContext context)
                 context.DataSourceUpdates,
                 requestor,
                 _pollInterval,
-                () => Selector.Empty // TODO: Implement
+                () => context.SelectorSource?.Selector ?? Selector.Empty
             );
         }
 

pkgs/sdk/server/src/Integrations/FDv2StreamingDataSourceBuilder.cs

@@ -84,7 +84,7 @@ public IDataSource Build(LdClientContext context)
                 context.DataSourceUpdates,
                 configuredBaseUri,
                 _initialReconnectDelay,
-                () => Selector.Empty // TODO: Implement.
+                () => context.SelectorSource?.Selector ?? Selector.Empty
             );
         }
 

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

@@ -3,10 +3,10 @@
 namespace LaunchDarkly.Sdk.Server.Internal.DataSources
 {
     /// <summary>
-    /// A function type that creates a new <see cref="IActionApplier"/> instance.
+    /// A function type that creates a new <see cref="IDataSourceObserver"/> instance that is capable of applying actions to a composite data source.
     /// </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);
+    /// <returns>a new <see cref="IDataSourceObserver"/> instance</returns>
+    internal delegate IDataSourceObserver ActionApplierFactory(ICompositeSourceActionable actionable);
 }
 

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

@@ -4,6 +4,8 @@
 using LaunchDarkly.Sdk.Server.Interfaces;
 using LaunchDarkly.Sdk.Server.Subsystems;
 
+using static LaunchDarkly.Sdk.Server.Subsystems.DataStoreTypes;
+
 namespace LaunchDarkly.Sdk.Server.Internal.DataSources
 {
     /// <summary>
@@ -16,15 +18,15 @@ internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable
     {
         // All mutable state and the internal action queue are protected by this lock.
         // We also use a small, non-recursive action queue so that any re-entrant calls
-        // from IActionApplier logic are serialized and processed iteratively instead
+        // from action applier logic is serialized and processed iteratively instead
         // of recursively, avoiding the risk of stack overflows.
         private readonly object _lock = new object();
         private readonly Queue<Action> _pendingActions = new Queue<Action>();
         private bool _isProcessingActions;
         private bool _disposed;
 
-        private readonly IDataSourceUpdates _originalUpdateSink;
-        private readonly IDataSourceUpdates _sanitizedUpdateSink;
+        private readonly IDataSourceUpdatesV2 _originalUpdateSink;
+        private readonly IDataSourceUpdatesV2 _sanitizedUpdateSink;
         private readonly SourcesList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory)> _sourcesList;
         private readonly DisableableDataSourceUpdatesTracker _disableableTracker;
 
@@ -41,7 +43,7 @@ internal sealed class CompositeSource : IDataSource, ICompositeSourceActionable
         /// <param name="factoryTuples">the ordered list of source factories and their associated action applier factories</param>
         /// <param name="circular">whether to loop off the end of the list back to the start when fallback occurs</param>
         public CompositeSource(
-            IDataSourceUpdates updatesSink,
+            IDataSourceUpdatesV2 updatesSink,
             IList<(SourceFactory Factory, ActionApplierFactory ActionApplierFactory)> factoryTuples,
             bool circular = true)
         {
@@ -206,19 +208,25 @@ private void TryFindNextUnderLock()
                 InternalDispose(errorInfo);
                 return;
             }
-
-            // Build the list of update sinks conditionally based on whether we have an action applier factory
-            var updateSinks = new List<IDataSourceUpdates>();
+            
+            IReadOnlyList<IDataSourceObserver> observers;
+            // Conditionally add the action applier if we have one
             if (entry.ActionApplierFactory != null)
             {
+                var observersList = new List<IDataSourceObserver>();
                 var actionApplier = entry.ActionApplierFactory(this);
-                updateSinks.Add(actionApplier);
+                observersList.Add(actionApplier);
+                observers = observersList;
+            }
+            else
+            {
+                observers = Array.Empty<IDataSourceObserver>();
             }
-            updateSinks.Add(_sanitizedUpdateSink);
 
-            // here we make a fanout so that we can trigger actions as well as forward calls to the sanitized sink (order matters here)
-            var fanout = new FanOutDataSourceUpdates(updateSinks);
-            var disableableUpdates = _disableableTracker.WrapAndTrack(fanout);
+            // here we wrap the sink in observability so that we can trigger actions, the sanitized sink is
+            // invoked before the observers to ensure actions don't trigger before data can propagate
+            var observableUpdates = new ObservableDataSourceUpdates(_sanitizedUpdateSink, observers);
+            var disableableUpdates = _disableableTracker.WrapAndTrack(observableUpdates);
 
             _currentEntry = entry;
             _currentDataSource = entry.Factory(disableableUpdates);
@@ -395,5 +403,3 @@ public void BlacklistCurrent()
         #endregion
     }
 }
-
-

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

@@ -8,8 +8,8 @@
 namespace LaunchDarkly.Sdk.Server.Internal.DataSources
 {
     /// <summary>
-    /// An <see cref="IDataSourceUpdates"/> implementation that decorates an underlying
-    /// <see cref="IDataSourceUpdates"/> and sanitizes status updates.
+    /// An <see cref="IDataSourceUpdatesV2"/> implementation that decorates an underlying
+    /// <see cref="IDataSourceUpdatesV2"/> and sanitizes status updates.
     /// </summary>
     /// <remarks>
     /// This wrapper ensures the following:
@@ -19,9 +19,9 @@ namespace LaunchDarkly.Sdk.Server.Internal.DataSources
     /// <item><description>Does not report the same combination of state and error twice in a row.</description></item>
     /// </list>
     /// </remarks>
-    internal sealed class DataSourceUpdatesSanitizer : IDataSourceUpdates
+    internal sealed class DataSourceUpdatesSanitizer : IDataSourceUpdatesV2
     {
-        private readonly IDataSourceUpdates _inner;
+        private readonly IDataSourceUpdatesV2 _inner;
         private readonly object _lock = new object();
 
         private bool _alreadyReportedInitializing;
@@ -32,7 +32,7 @@ internal sealed class DataSourceUpdatesSanitizer : IDataSourceUpdates
         /// Creates a new <see cref="DataSourceUpdatesSanitizer"/>.
         /// </summary>
         /// <param name="inner">the underlying updates sink to delegate to</param>
-        public DataSourceUpdatesSanitizer(IDataSourceUpdates inner)
+        public DataSourceUpdatesSanitizer(IDataSourceUpdatesV2 inner)
         {
             _inner = inner ?? throw new ArgumentNullException(nameof(inner));
         }
@@ -41,17 +41,7 @@ public DataSourceUpdatesSanitizer(IDataSourceUpdates inner)
         /// The underlying <see cref="IDataStoreStatusProvider"/> from the decorated sink.
         /// </summary>
         public IDataStoreStatusProvider DataStoreStatusProvider => _inner.DataStoreStatusProvider;
-
-        public bool Init(FullDataSet<ItemDescriptor> allData)
-        {
-            return _inner.Init(allData);
-        }
-
-        public bool Upsert(DataKind kind, string key, ItemDescriptor item)
-        {
-            return _inner.Upsert(kind, key, item);
-        }
-
+        
         public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError)
         {
             lock (_lock)
@@ -87,6 +77,11 @@ public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? n
                 _inner.UpdateStatus(sanitized, newError);
             }
         }
+
+        public bool Apply(ChangeSet<ItemDescriptor> changeSet)
+        {
+            return _inner.Apply(changeSet);
+        }
     }
 }
 

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

@@ -21,11 +21,11 @@ public DisableableDataSourceUpdatesTracker()
         }
 
         /// <summary>
-        /// Wraps the provided <see cref="IDataSourceUpdates"/> in a new instance that can be disabled by
+        /// Wraps the provided <see cref="IDataSourceUpdatesV2"/> in a new instance that can be disabled by
         /// calling <see cref="DisablePreviouslyTracked"/>.
         /// </summary>
         /// <returns>a new proxy instance</returns>
-        public IDataSourceUpdates WrapAndTrack(IDataSourceUpdates dsUpdates)
+        public IDataSourceUpdatesV2 WrapAndTrack(IDataSourceUpdatesV2 dsUpdates)
         {
             var dis = new DisableableIDataSourceUpdates(dsUpdates);
             lock (_lock)
@@ -60,15 +60,15 @@ public void DisablePreviouslyTracked()
         }
 
         /// <summary>
-        /// A proxy for <see cref="IDataSourceUpdates"/> that can be disabled. When disabled,
+        /// A proxy for <see cref="IDataSourceUpdatesV2"/> that can be disabled. When disabled,
         /// all calls to the proxy will be ignored.
         /// </summary>
-        private sealed class DisableableIDataSourceUpdates : IDataSourceUpdates, IDataSourceUpdatesHeaders
+        private sealed class DisableableIDataSourceUpdates : IDataSourceUpdatesV2
         {
-            private readonly IDataSourceUpdates _updatesSink;
+            private readonly IDataSourceUpdatesV2 _updatesSink;
             private volatile bool _disabled;
 
-            public DisableableIDataSourceUpdates(IDataSourceUpdates updatesSink)
+            public DisableableIDataSourceUpdates(IDataSourceUpdatesV2 updatesSink)
             {
                 _updatesSink = updatesSink ?? throw new ArgumentNullException(nameof(updatesSink));
             }
@@ -84,27 +84,7 @@ public void Disable()
             private bool IsDisabled => _disabled;
 
             public IDataStoreStatusProvider DataStoreStatusProvider => _updatesSink.DataStoreStatusProvider;
-
-            public bool Init(FullDataSet<ItemDescriptor> allData)
-            {
-                if (IsDisabled)
-                {
-                    return false;
-                }
-
-                return _updatesSink.Init(allData);
-            }
-
-            public bool Upsert(DataKind kind, string key, ItemDescriptor item)
-            {
-                if (IsDisabled)
-                {
-                    return false;
-                }
-
-                return _updatesSink.Upsert(kind, key, item);
-            }
-
+            
             public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? newError)
             {
                 if (IsDisabled)
@@ -115,22 +95,9 @@ public void UpdateStatus(DataSourceState newState, DataSourceStatus.ErrorInfo? n
                 _updatesSink.UpdateStatus(newState, newError);
             }
 
-            public bool InitWithHeaders(
-                FullDataSet<ItemDescriptor> allData,
-                IEnumerable<KeyValuePair<string, IEnumerable<string>>> headers
-                )
+            public bool Apply(ChangeSet<ItemDescriptor> changeSet)
             {
-                if (IsDisabled)
-                {
-                    return false;
-                }
-
-                if (_updatesSink is IDataSourceUpdatesHeaders headersSink)
-                {
-                    return headersSink.InitWithHeaders(allData, headers);
-                }
-
-                return _updatesSink.Init(allData);
+                return !IsDisabled && _updatesSink.Apply(changeSet);
             }
         }
     }

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

@@ -3,7 +3,7 @@
 namespace LaunchDarkly.Sdk.Server.Internal.DataSources
 {
     /// <summary>
-    /// Defines operations that can be performed by <see cref="IActionApplier"/>s
+    /// Defines operations that can be performed by <see cref="IDataSourceObserver"/>s
     /// on a composite data source.
     /// </summary>
     internal interface ICompositeSourceActionable