feat: Drop persistent-store cache after FDv2 in-memory store init

Once the FDv2 in-memory store takes over as the active read source, the
persistent-store wrapper's three Guava caches are no longer consulted on reads
and roughly double the in-memory footprint of flag data (or worse, indefinitely,
in cacheForever() mode).

Add an internal DisableableCache capability interface implemented by
PersistentDataStoreWrapper. The wrapper's disableCache() sets a volatile
cacheDisabled flag and 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
when set; this is required because the caches are LoadingCache instances and
plain invalidation would auto-repopulate via the registered CacheLoaders on
the next get(). WriteThroughStore.maybeSwitchStore() probes for the interface
and invokes disableCache() inside the existing synchronized block, after the
active read store has been flipped to the in-memory store.

The PersistentDataStoreBuilder cache-config setters (cacheTime, cacheSeconds,
cacheMillis, cacheForever, noCaching, staleValuesPolicy, recordCacheStats)
remain functional during the bootstrap window for backward compatibility;
class-level javadoc explains that under FDv2 they only govern that window.

Naming note: the capability is named disableCache rather than the ticket's
clearCache so the verb conveys that the cache is off going forward, not just
emptied. This aligns with Python and Ruby's disable_cache and matches the
dotnet sibling PR.

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

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,31 @@ 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; an in-flight reader past the flag check may still complete its
+    // current cache operation and, on a miss, fire the LoadingCache loader
+    // which can repopulate. Those leftover entries hold fresh values and are
+    // unreachable from any subsequent read (every future caller bypasses), so
+    // they don't cause stale reads or torn observations; they simply persist
+    // until GC reclaims the cache. See dotnet-core#274 for the analysis.
+    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 +210,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 +251,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 +265,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 +304,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 +320,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 +363,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 +466,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());