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

## 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 Guava 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 `DisableableCache` capability
interface and a `disableCache()` method on `PersistentDataStoreWrapper`.
The method sets a `volatile boolean cacheDisabled` flag and then
invalidates all three Guava caches. Every cache touch site
(`isInitialized`, `init`, `get`, `getAll`, `upsert`, `getCacheStats`,
`pollAvailabilityAfterOutage`) checks the flag and short-circuits to the
core path when set. The flag-check is required because the caches are
`LoadingCache` instances and plain `invalidateAll()` would
auto-repopulate via the registered `CacheLoader`s on the next `get()`.

`WriteThroughStore.maybeSwitchStore()` probes for the interface via
`instanceof` and invokes `disableCache()` inside the existing
`synchronized (activeStoreLock)` block, immediately after the active
read store is flipped to the in-memory store. The probe is unconditional
with respect to `DataStoreMode`: reads bypass the persistent store in
both `READ_ONLY` and `READ_WRITE` modes after the flip, so the cache is
dead weight in either mode.

The `PersistentDataStoreBuilder` cache-config setters (`noCaching`,
`cacheTime`, `cacheMillis`, `cacheSeconds`, `cacheForever`,
`staleValuesPolicy`, `recordCacheStats`) remain functional during the
bootstrap window for backward compatibility. Class-level javadoc on the
builder explains that under FDv2 these only govern the window before the
in-memory store has received its first payload. No `@Deprecated`
annotation; the SDK source has no `@Deprecated` precedent and Matthew
Keeler's pattern across Python, Ruby, and Go uses doc-only deprecation.

Mirrors the same change shipped for Python
(launchdarkly/python-server-sdk#426), Ruby
(launchdarkly/ruby-server-sdk#384), Go (launchdarkly/go-server-sdk#373),
and .NET (launchdarkly/dotnet-core#274).

**Naming note:** the ticket text says `CacheClearable` / `clearCache()`.
This change uses `DisableableCache` / `disableCache()` instead so the
verb conveys that the cache is off going forward (a state change), not
just emptied (a one-time imperative). This aligns with Python's
`disable_cache` and Ruby's `disable_cache`, and matches the `.NET`
sibling PR which made the same naming choice.

**Concurrency note:** an in-flight reader that passed the
`cacheDisabled` check before `disableCache()` ran can still complete its
cache operation and, on a miss, fire the `LoadingCache` loader. The
fresh value gets stored back in the cache after our `invalidateAll()`.
Those leftover entries are unreachable from any subsequent read (every
future caller bypasses) and hold fresh, correct values, so they don't
cause stale reads or torn observations. They simply persist in the cache
instance until GC reclaims it. See dotnet-core#274 for the full analysis
of this trade-off vs. a Go-style `AtomicReference` swap +
read-once-per-call refactor.

Tests follow the indirect-observation pattern: prime the cache, call
`disableCache()`, mutate the core directly, assert the next read sees
the new value. A new `MockDisableableCachePersistentStore` spy
(extending `MockTransactionalPersistentStore`) drives the
`WriteThroughStore` probe-and-invoke coverage, including the FDv1 `init`
path, the FDv2 `apply` path, the delta-does-not-redrop case, the
`READ_ONLY`-mode-still-drops case, and the non-disableable-store no-op
case. No `Thread.sleep`.

`./gradlew test` -- 1906 tests, 0 failures.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes the server SDK persistent-store read/write path at the FDv2
handoff; behavior is guarded by tests and matches other language SDKs,
but incorrect timing could affect bootstrap reads.
> 
> **Overview**
> Adds an internal **`DisableableCache`** / **`disableCache()`** path so
the persistent data store wrapper stops using its Guava caches after
FDv2 hands reads to the in-memory store.
> 
> **`PersistentDataStoreWrapper`** sets a **`cacheDisabled`** flag,
invalidates item/all/init caches, and skips cache on **`get`**,
**`getAll`**, **`init`**, **`upsert`**, **`isInitialized`**,
**`getCacheStats`**, and outage recovery so **`LoadingCache`** loaders
are not repopulated.
> 
> **`WriteThroughStore`** calls **`disableCache()`** once inside
**`maybeSwitchStore()`** when the first initializing payload flips the
active read store to memory (including **`READ_ONLY`** and legacy
**`init`**).
> 
> **`PersistentDataStoreBuilder`** javadoc notes cache options only
matter during the bootstrap window before that switch. New unit tests
cover bypass behavior and the one-time probe.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
6ca393f. 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

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/DisableableCache.java

@@ -0,0 +1,18 @@
+package com.launchdarkly.sdk.server;
+
+/**
+ * Optional interface for data stores that can disable their internal cache.
+ * <p>
+ * This is currently for internal implementations only.
+ */
+interface DisableableCache {
+  /**
+   * Disables the internal cache. After this call, the cache is no longer
+   * consulted on reads and no longer populated by writes.
+   * <p>
+   * Implementations should release the cache contents so the memory can be
+   * reclaimed. The call must be idempotent: subsequent invocations should be
+   * safe and have no further effect.
+   */
+  void disableCache();
+}

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PersistentDataStoreWrapper.java

@@ -43,7 +43,7 @@
  * <p>
  * This class is only constructed by {@link PersistentDataStoreBuilder}.
  */
-final class PersistentDataStoreWrapper implements DataStore, SettableCache {
+final class PersistentDataStoreWrapper implements DataStore, SettableCache, DisableableCache {
   private final PersistentDataStore core;
   private final LoadingCache<CacheKey, Optional<ItemDescriptor>> itemCache;
   private final LoadingCache<DataKind, KeyedItems<ItemDescriptor>> allCache;
@@ -54,9 +54,15 @@ final class PersistentDataStoreWrapper implements DataStore, SettableCache {
   private final AtomicBoolean inited = new AtomicBoolean(false);
   private final ListeningExecutorService cacheExecutor;
   private final LDLogger logger;
-  
+
   private final Object externalStoreLock = new Object();
   private volatile CacheExporter externalCache;
+
+  // Once true, the cache is bypassed on reads and writes; entries already in
+  // the cache have been invalidated by disableCache(). The cache instances
+  // themselves remain alive until GC reclaims them; the LoadingCache loaders
+  // are short-circuited because every touch site checks this flag first.
+  private volatile boolean cacheDisabled;
 
   PersistentDataStoreWrapper(
       final PersistentDataStore core,
@@ -151,14 +157,26 @@ public void close() throws IOException {
     core.close();
   }
 
+  @Override
+  public void disableCache() {
+    if (cacheDisabled) return;
+    // Volatile write publishes the bypass flag before clearing cache contents.
+    // Future readers observe cacheDisabled == true and skip the cache call
+    // sites.
+    cacheDisabled = true;
+    if (itemCache != null) itemCache.invalidateAll();
+    if (allCache != null) allCache.invalidateAll();
+    if (initCache != null) initCache.invalidateAll();
+  }
+
   @Override
   public boolean isInitialized() {
     if (inited.get()) {
       return true;
     }
     boolean result;
     try {
-      if (initCache != null) {
+      if (initCache != null && !cacheDisabled) {
         result = initCache.get("");
       } else {
         result = core.isInitialized();
@@ -187,7 +205,7 @@ public void init(FullDataSet<ItemDescriptor> allData) {
       allBuilder.add(new AbstractMap.SimpleEntry<>(kind, items));
     }
     RuntimeException failure = initCore(new FullDataSet<>(allBuilder.build(), allData.shouldPersist()));
-    if (itemCache != null && allCache != null) {
+    if (itemCache != null && allCache != null && !cacheDisabled) {
       itemCache.invalidateAll();
       allCache.invalidateAll();
       if (failure != null && !cacheIndefinitely) {
@@ -228,7 +246,7 @@ private RuntimeException initCore(FullDataSet<SerializedItemDescriptor> allData)
   @Override
   public ItemDescriptor get(DataKind kind, String key) {
     try {
-      ItemDescriptor ret = itemCache != null ? itemCache.get(CacheKey.forItem(kind, key)).orNull() :
+      ItemDescriptor ret = (itemCache != null && !cacheDisabled) ? itemCache.get(CacheKey.forItem(kind, key)).orNull() :
         getAndDeserializeItem(kind, key);
       processError(null);
       return ret;
@@ -242,7 +260,7 @@ public ItemDescriptor get(DataKind kind, String key) {
   public KeyedItems<ItemDescriptor> getAll(DataKind kind) {
     try {
       KeyedItems<ItemDescriptor> ret;
-      ret = allCache != null ? allCache.get(kind) : getAllAndDeserialize(kind);
+      ret = (allCache != null && !cacheDisabled) ? allCache.get(kind) : getAllAndDeserialize(kind);
       processError(null);
       return ret;
     } catch (Exception e) {
@@ -281,7 +299,7 @@ public boolean upsert(DataKind kind, String key, ItemDescriptor item) {
       }
       failure = e;
     }
-    if (itemCache != null) {
+    if (itemCache != null && !cacheDisabled) {
       CacheKey cacheKey = CacheKey.forItem(kind, key);
       if (failure == null) {
         if (updated) {
@@ -297,7 +315,7 @@ public boolean 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
       // update the item within the existing "all items" entry (since we want things to still work
@@ -340,7 +358,7 @@ public void setCacheExporter(CacheExporter externalDataSource) {
 
   @Override
   public CacheStats getCacheStats() {
-    if (itemCache == null || allCache == null) {
+    if (itemCache == null || allCache == null || cacheDisabled) {
       return null;
     }
     com.google.common.cache.CacheStats itemStats = itemCache.stats();
@@ -443,8 +461,9 @@ private boolean 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 externalCache 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
       // write the contents of the cache to the underlying data store.

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/WriteThroughStore.java

@@ -149,6 +149,9 @@ private void maybeSwitchStore() {
     }
     synchronized (activeStoreLock) {
       activeReadStore = memoryStore;
+      if (persistentStore instanceof DisableableCache) {
+        ((DisableableCache) persistentStore).disableCache();
+      }
     }
   }
 

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/PersistentDataStoreBuilder.java

@@ -32,10 +32,19 @@
  * </code></pre>
  * 
  * In this example, {@code .url()} is an option specifically for the Redis integration, whereas
- * {@code cacheSeconds()} is an option that can be used for any persistent data store. 
+ * {@code cacheSeconds()} is an option that can be used for any persistent data store.
  * <p>
  * Note that this class is abstract; the actual implementation is created by calling
  * {@link Components#persistentDataStore(ComponentConfigurer)}.
+ * <p>
+ * Under the FDv2 data system, the cache options configured here ({@link #cacheTime(Duration)},
+ * {@link #cacheSeconds(long)}, {@link #cacheMillis(long)}, {@link #cacheForever()},
+ * {@link #noCaching()}, {@link #staleValuesPolicy(StaleValuesPolicy)},
+ * {@link #recordCacheStats(boolean)}) 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.
  * @since 4.12.0
  */
 public abstract class PersistentDataStoreBuilder implements ComponentConfigurer<DataStore> {

lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PersistentDataStoreWrapperTest.java

@@ -35,6 +35,7 @@
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeThat;
@@ -713,6 +714,115 @@ public void statusRemainsUnavailableIfStoreSaysItIsAvailableButInitFails() throw
     assertThat(core.initedCount.get(), greaterThan(initedCount));
   }
 
+  @Test
+  public void disableCacheIsIdempotent() {
+    assumeThat(testMode.isCached(), is(true));
+    wrapper.disableCache();
+    wrapper.disableCache(); // must not throw
+  }
+
+  @Test
+  public void disableCacheIsSafeOnUncachedWrapper() {
+    assumeThat(testMode.isCached(), is(false));
+    wrapper.disableCache(); // must not throw
+  }
+
+  @Test
+  public void getAfterDisableCacheReturnsCurrentCoreState() {
+    assumeThat(testMode.isCached(), is(true));
+    TestItem item1v1 = new TestItem("key", 1);
+    TestItem item1v2 = new TestItem("key", 2);
+
+    core.forceSet(TEST_ITEMS, item1v1);
+    // Prime the cache.
+    assertThat(wrapper.get(TEST_ITEMS, item1v1.key), equalTo(item1v1.toItemDescriptor()));
+
+    wrapper.disableCache();
+
+    // Mutate the core behind the wrapper's back; if the cache were still
+    // serving reads we would see the stale v1.
+    core.forceSet(TEST_ITEMS, item1v2);
+    assertThat(wrapper.get(TEST_ITEMS, item1v2.key), equalTo(item1v2.toItemDescriptor()));
+  }
+
+  @Test
+  public void getAllAfterDisableCacheReturnsCurrentCoreState() {
+    assumeThat(testMode.isCached(), is(true));
+    TestItem item1 = new TestItem("keyA", 1);
+    TestItem item2 = new TestItem("keyB", 1);
+
+    core.forceSet(TEST_ITEMS, item1);
+    // Prime the cache.
+    Map<String, ItemDescriptor> primed = toItemsMap(wrapper.getAll(TEST_ITEMS));
+    assertThat(primed.size(), is(1));
+
+    wrapper.disableCache();
+
+    core.forceSet(TEST_ITEMS, item2);
+    Map<String, ItemDescriptor> afterDrop = toItemsMap(wrapper.getAll(TEST_ITEMS));
+    assertThat(afterDrop.size(), is(2));
+  }
+
+  @Test
+  public void upsertAfterDisableCacheWritesThroughToCoreOnly() {
+    assumeThat(testMode.isCached(), is(true));
+    TestItem item = new TestItem("key", 1);
+
+    wrapper.disableCache();
+
+    assertThat(wrapper.upsert(TEST_ITEMS, item.key, item.toItemDescriptor()), is(true));
+    // The write must have landed in the core.
+    assertThat(core.data.get(TEST_ITEMS).get(item.key), equalTo(item.toSerializedItemDescriptor()));
+    // And subsequent reads must reach the core (no repopulated cache).
+    assertThat(wrapper.get(TEST_ITEMS, item.key), equalTo(item.toItemDescriptor()));
+  }
+
+  @Test
+  public void initAfterDisableCacheWritesThroughToCoreWithoutRepopulatingCache() {
+    assumeThat(testMode.isCached(), is(true));
+    TestItem itemA = new TestItem("keyA", 1);
+    TestItem itemB = new TestItem("keyA", 2);
+
+    wrapper.disableCache();
+
+    wrapper.init(new DataBuilder().add(TEST_ITEMS, itemA).build());
+
+    assertThat(core.data.get(TEST_ITEMS).get(itemA.key), equalTo(itemA.toSerializedItemDescriptor()));
+
+    // Mutate the core behind the wrapper's back; if the cache had repopulated
+    // we would still see itemA on the next read.
+    core.forceSet(TEST_ITEMS, itemB);
+    assertThat(wrapper.get(TEST_ITEMS, itemB.key), equalTo(itemB.toItemDescriptor()));
+  }
+
+  @Test
+  public void getCacheStatsAfterDisableCacheReturnsNull() {
+    assumeThat(testMode.isCached(), is(true));
+    // Build a wrapper with stats recording enabled so getCacheStats is non-null pre-disable.
+    PersistentDataStoreWrapper w = new PersistentDataStoreWrapper(
+        new MockPersistentDataStore(),
+        testMode.getCacheTtl(),
+        PersistentDataStoreBuilder.StaleValuesPolicy.EVICT,
+        true,
+        this::updateStatus,
+        sharedExecutor,
+        testLogger);
+    try {
+      assertNotNull(w.getCacheStats());
+      w.disableCache();
+      assertNull(w.getCacheStats());
+    } finally {
+      try { w.close(); } catch (IOException e) { /* ignore */ }
+    }
+  }
+
+  @Test
+  public void closeAfterDisableCacheDoesNotThrow() throws IOException {
+    assumeThat(testMode.isCached(), is(true));
+    wrapper.disableCache();
+    wrapper.close(); // safety belt; tearDown will also call close, which must be safe
+  }
+
   private void causeStoreError(MockPersistentDataStore core, PersistentDataStoreWrapper w) {
     core.unavailable = true;
     core.fakeError = new RuntimeException(FAKE_ERROR.getMessage());