feat: Drop persistent-store cache after FDv2 in-memory store init (#274)

## Summary

With FDv2, the in-memory store retains every flag and segment once it
has received a full payload. The persistent-store wrapper's three
internal caches (item, all-items, init) become dead weight at that
point, roughly doubling the in-memory footprint of flag data (or worse,
indefinitely, in `CacheForever()` mode).

This change introduces an internal `IDisableableCache` capability
interface and a `DisableCache()` method on `PersistentStoreWrapper`. The
method atomically swaps each cache field to null via
`Interlocked.Exchange`, then clears and disposes the captured instances.
`WriteThroughStore.MaybeSwitchStore()` probes for the interface and
invokes the method inside the existing `_activeStoreLock`, immediately
after the active read store is flipped to the in-memory store. Existing
null-guards at every cache touch site already short-circuit when a cache
field is null (the "caching disabled" config path nulls them at
construction), so no read/write call sites needed to be rewritten.

The drop is mode-independent: reads bypass the persistent store in both
`ReadOnly` and `ReadWrite` modes after the flip, so the cache is unused
in either case. `Dispose(bool)` is refactored to call `DisableCache()`
in place of disposing the three caches individually, making the cache
lifecycle idempotent.

The `PersistentDataStoreBuilder` cache-config setters (`NoCaching`,
`CacheTime`, `CacheMillis`, `CacheSeconds`, `CacheMaximumEntries`,
`CacheForever`) remain functional during the bootstrap window for
backward compatibility. XML documentation on the class explains that
under FDv2 these only govern the window before the in-memory store has
received its first payload.

Mirrors the same change Matthew Keeler shipped for Python
(launchdarkly/python-server-sdk#426), Ruby
(launchdarkly/ruby-server-sdk#384), and Go
(launchdarkly/go-server-sdk#373). Naming differs from the ticket text:
the method is `DisableCache` rather than `DiscardCache` so the verb
conveys that the cache is off from this point forward (matching Python
and Ruby's `disable_cache`); the interface is `IDisableableCache` to
mirror the existing `ISettableCache` precedent in the same `Subsystems/`
namespace.

Tests follow the indirect-observation pattern used in the reference
SDKs: prime the cache, drop, mutate the core directly, assert the next
read sees the new value. A new spy `MockDisableableCachePersistentStore`
(with a `DisableCacheCallCount` counter) drives the `WriteThroughStore`
probe-and-invoke coverage, including the FDv1 `Init` path, the FDv2
`Apply` path, the delta-does-not-redrop case, the
`ReadOnly`-mode-still-drops case, and the non-disabler-store no-op case.
No `Thread.Sleep`.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches core data-store read/write paths and the FDv2 store handoff;
behavior change is scoped to post-init cache bypass with broad test
coverage, but incorrect timing could cause stale reads if the switch ran
too early.
> 
> **Overview**
> Under **FDv2**, once the in-memory store has taken the first full
payload, the persistent wrapper’s item/all/init caches are redundant and
can roughly **double flag memory** (worse with `CacheForever()`). This
PR adds internal **`IDisableableCache`** / **`DisableCache()`** on
**`PersistentStoreWrapper`**, which sets a volatile bypass flag, clears
caches, and routes **`Get`**, **`GetAll`**, **`Init`**, **`Upsert`**,
and **`Initialized`** straight to the core when disabled.
> 
> **`WriteThroughStore.MaybeSwitchStore()`** invokes
**`DisableCache()`** once (inside **`_activeStoreLock`**) when reads
flip to the memory store—on first **`Init`** or **`Apply`**, including
**ReadOnly** mode; stores without the interface are unchanged.
> 
> **`PersistentDataStoreBuilder`** docs now state cache options only
matter during the **bootstrap** window before that handoff; options
remain for backward compatibility.
> 
> Tests cover idempotent disable, read-through after disable, and
**`WriteThroughStore`** calling disable exactly once.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
81f1b82. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: tanderson-ld

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

@@ -25,6 +25,17 @@ namespace LaunchDarkly.Sdk.Server.Integrations
     ///         .DataStore(myStore)
     ///         .Build();
     /// </code>
+    /// <para>
+    /// Note: under the FDv2 data system, the cache settings configured here
+    /// (<see cref="CacheTime(TimeSpan)"/>, <see cref="CacheSeconds(int)"/>,
+    /// <see cref="CacheMillis(int)"/>, <see cref="CacheMaximumEntries(int?)"/>,
+    /// <see cref="CacheForever"/>, <see cref="NoCaching"/>) only govern the
+    /// brief bootstrap window before the in-memory store has received its
+    /// first full payload. Once the in-memory store takes over as the active
+    /// read source, the persistent-store cache is released and these settings
+    /// have no further effect. These options are kept for backward compatibility
+    /// and may be deprecated in a future major version.
+    /// </para>
     /// </remarks>
     public class PersistentDataStoreBuilder : IComponentConfigurer<IDataStore>, IDiagnosticDescription
     {

pkgs/sdk/server/src/Internal/DataStores/PersistentStoreWrapper.cs

@@ -22,7 +22,7 @@ namespace LaunchDarkly.Sdk.Server.Internal.DataStores
     /// class adds the caching behavior that we normally want for any persistent data store.
     /// </para>
     /// </remarks>
-    internal sealed class PersistentStoreWrapper : IDataStore, ISettableCache
+    internal sealed class PersistentStoreWrapper : IDataStore, ISettableCache, IDisableableCache
     {
         private readonly IPersistentDataStore _core;
         private readonly DataStoreCacheConfig _caching;
@@ -39,6 +39,13 @@ internal sealed class PersistentStoreWrapper : IDataStore, ISettableCache
         private ICacheExporter _externalCache;
 
         private volatile bool _inited;
+
+        // Once true, the cache is bypassed on reads and writes; entries already in
+        // the cache have been Clear()-ed by DisableCache(). The cache objects
+        // themselves remain alive until Dispose() so that any reader currently
+        // holding a reference does not observe a disposed cache. Volatile so the
+        // bypass becomes visible to other threads without an explicit lock.
+        private volatile bool _cacheDisabled;
 
         internal PersistentStoreWrapper(
             IPersistentDataStoreAsync coreAsync,
@@ -112,7 +119,7 @@ public bool Initialized()
                 return true;
             }
             bool result;
-            if (_initCache != null)
+            if (_initCache != null && !_cacheDisabled)
             {
                 result = _initCache.Get();
             }
@@ -163,7 +170,7 @@ public void Init(FullDataSet<ItemDescriptor> items)
                 kindAndItems => PersistentDataStoreConverter.SerializeAll(kindAndItems.Key, kindAndItems.Value.Items)
             );
             Exception failure = InitCore(new FullDataSet<SerializedItemDescriptor>(serializedItems));
-            if (_itemCache != null && _allCache != null)
+            if (_itemCache != null && _allCache != null && !_cacheDisabled)
             {
                 _itemCache.Clear();
                 _allCache.Clear();
@@ -199,7 +206,7 @@ public void Init(FullDataSet<ItemDescriptor> items)
         {
             try
             {
-                var ret = _itemCache is null ? GetAndDeserializeItem(kind, key) :
+                var ret = (_itemCache is null || _cacheDisabled) ? GetAndDeserializeItem(kind, key) :
                     _itemCache.Get(new CacheKey(kind, key));
                 ProcessError(null);
                 return ret;
@@ -215,7 +222,7 @@ public KeyedItems<ItemDescriptor> GetAll(DataKind kind)
         {
             try
             {
-                var ret = new KeyedItems<ItemDescriptor>(_allCache is null ?
+                var ret = new KeyedItems<ItemDescriptor>((_allCache is null || _cacheDisabled) ?
                     GetAllAndDeserialize(kind) : _allCache.Get(kind));
                 ProcessError(null);
                 return ret;
@@ -250,7 +257,7 @@ public bool Upsert(DataKind kind, string key, ItemDescriptor item)
                 }
                 failure = e;
             }
-            if (_itemCache != null)
+            if (_itemCache != null && !_cacheDisabled)
             {
                 var cacheKey = new CacheKey(kind, key);
                 if (failure is null)
@@ -284,7 +291,7 @@ public bool Upsert(DataKind kind, string key, ItemDescriptor item)
                     }
                 }
             }
-            if (_allCache != null)
+            if (_allCache != null && !_cacheDisabled)
             {
                 // If the cache has a finite TTL, then we should remove the "all items" cache entry to force
                 // a reread the next time All is called. However, if it's an infinite TTL, we need to just
@@ -315,6 +322,24 @@ public bool Upsert(DataKind kind, string key, ItemDescriptor item)
             return updated;
         }
 
+        /// <summary>
+        /// Disables the internal cache. After this call, reads and writes go straight
+        /// to the underlying persistent store; subsequent invocations are no-ops.
+        /// </summary>
+        /// <remarks>
+        /// At the time of writing, it is likely that the cache was disabled when 
+        /// another store took over as the active store and so reads may not even get
+        /// to this layer anymore.  If they do, they do go straight to persistence if
+        /// this layer's DisableCache method was called.
+        /// </remarks>
+        public void DisableCache()
+        {
+            if (_cacheDisabled) return;
+            _cacheDisabled = true;
+            _itemCache?.Clear();
+            _allCache?.Clear();
+        }
+
         public void Dispose()
         {
             Dispose(true);
@@ -437,8 +462,9 @@ private bool PollAvailabilityAfterOutage()
             }
 
             // Fall back to cache-based recovery if external store is not available/initialized
-            // and we're in infinite cache mode
-            if (_cacheIndefinitely && _allCache != null)
+            // and we're in infinite cache mode. Under FDv2 this branch is dead once
+            // DisableCache has run: the ICacheExporter path above supersedes it.
+            if (_cacheIndefinitely && _allCache != null && !_cacheDisabled)
             {
                 // If we're in infinite cache mode, then we can assume the cache has a full set of current
                 // flag data (since presumably the data source has still been running) and we can just

pkgs/sdk/server/src/Internal/DataSystem/WriteThroughStore.cs

@@ -127,6 +127,10 @@ private void MaybeSwitchStore()
             lock (_activeStoreLock)
             {
                 _activeReadStore = _memoryStore;
+                if (_persistentStore is IDisableableCache disableable)
+                {
+                    disableable.DisableCache();
+                }
             }
         }
 

pkgs/sdk/server/src/Subsystems/IDisableableCache.cs

@@ -0,0 +1,22 @@
+namespace LaunchDarkly.Sdk.Server.Subsystems
+{
+    /// <summary>
+    /// Optional interface for data stores that can disable their internal cache.
+    /// </summary>
+    /// <remarks>
+    /// This is currently for internal implementations only.
+    /// </remarks>
+    internal interface IDisableableCache
+    {
+        /// <summary>
+        /// Disables the internal cache. After this call, the cache is no longer
+        /// consulted on reads and no longer populated by writes.
+        /// </summary>
+        /// <remarks>
+        /// Implementations should clear, dispose, and dereference cache instances
+        /// so the memory can be reclaimed. The call must be idempotent: subsequent
+        /// invocations should be safe and have no further effect.
+        /// </remarks>
+        void DisableCache();
+    }
+}

pkgs/sdk/server/test/Internal/DataStores/PersistentStoreWrapperTestBase.cs

@@ -640,6 +640,109 @@ public void StatusRemainsUnavailableIfStoreSaysItIsAvailableButInitFails()
             }
         }
 
+        [Fact]
+        public void DisableCacheIsIdempotent()
+        {
+            var wrapper = MakeWrapper(Cached);
+            wrapper.DisableCache();
+            wrapper.DisableCache(); // second call must not throw
+        }
+
+        [Fact]
+        public void DisableCacheIsSafeOnUncachedWrapper()
+        {
+            var wrapper = MakeWrapper(Uncached);
+            wrapper.DisableCache(); // no caches to release, must not throw
+        }
+
+        [Fact]
+        public void GetAfterDisableCacheReturnsCurrentCoreState()
+        {
+            var wrapper = MakeWrapper(Cached);
+            var key = "flag";
+            var itemv1 = new TestItem("itemv1");
+            var itemv2 = new TestItem("itemv2");
+
+            _core.ForceSet(TestDataKind, key, 1, itemv1);
+            // Prime the cache.
+            Assert.Equal(itemv1.WithVersion(1), wrapper.Get(TestDataKind, key));
+
+            wrapper.DisableCache();
+
+            // Mutate the core behind the wrapper's back. If the cache is still
+            // serving reads we would see the stale itemv1.
+            _core.ForceSet(TestDataKind, key, 2, itemv2);
+            Assert.Equal(itemv2.WithVersion(2), wrapper.Get(TestDataKind, key));
+        }
+
+        [Fact]
+        public void GetAllAfterDisableCacheReturnsCurrentCoreState()
+        {
+            var wrapper = MakeWrapper(Cached);
+            var itemA = new TestItem("itemA");
+            var itemB = new TestItem("itemB");
+
+            _core.ForceSet(TestDataKind, "keyA", 1, itemA);
+            // Prime the cache.
+            var primed = wrapper.GetAll(TestDataKind);
+            Assert.Single(primed.Items);
+
+            wrapper.DisableCache();
+
+            _core.ForceSet(TestDataKind, "keyB", 1, itemB);
+            var afterDrop = wrapper.GetAll(TestDataKind);
+            Assert.Equal(2, afterDrop.Items.Count());
+        }
+
+        [Fact]
+        public void UpsertAfterDisableCacheWritesThroughToCoreOnly()
+        {
+            var wrapper = MakeWrapper(Cached);
+            var key = "flag";
+            var itemv1 = new TestItem("itemv1");
+
+            wrapper.DisableCache();
+
+            Assert.True(wrapper.Upsert(TestDataKind, key, itemv1.WithVersion(1)));
+            // The write must have landed in the core, and a subsequent Get
+            // must reach the core (not a repopulated cache).
+            Assert.True(_core.Data[TestDataKind].ContainsKey(key));
+            Assert.Equal(itemv1.WithVersion(1), wrapper.Get(TestDataKind, key));
+        }
+
+        [Fact]
+        public void InitAfterDisableCacheWritesThroughToCoreWithoutRepopulatingCache()
+        {
+            var wrapper = MakeWrapper(Cached);
+            var itemA = new TestItem("itemA");
+            var itemB = new TestItem("itemB");
+
+            wrapper.DisableCache();
+
+            var allData = new TestDataBuilder()
+                .Add(TestDataKind, "keyA", 1, itemA)
+                .Build();
+            wrapper.Init(allData);
+
+            // The init must have written to the core.
+            Assert.True(_core.Data[TestDataKind].ContainsKey("keyA"));
+
+            // If the cache had repopulated, the next ForceRemove behind the
+            // wrapper's back would still yield itemA on Get. With cache
+            // disabled, the new core state is what we observe.
+            _core.ForceRemove(TestDataKind, "keyA");
+            _core.ForceSet(TestDataKind, "keyA", 2, itemB);
+            Assert.Equal(itemB.WithVersion(2), wrapper.Get(TestDataKind, "keyA"));
+        }
+
+        [Fact]
+        public void DisposeIsSafeAfterDisableCache()
+        {
+            var wrapper = MakeWrapper(Cached);
+            wrapper.DisableCache();
+            wrapper.Dispose(); // must not throw, even though caches are already gone
+        }
+
         private PersistentStoreWrapper MakeWrapper(CacheMode cacheMode) =>
             MakeWrapper(new TestParams { CacheMode = cacheMode, PersistMode = PersistWithMetadata });