fix: stop FDv2DataSource.Conditions from leaking on healthy primary

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

@@ -11,9 +11,11 @@
 import com.launchdarkly.sdk.server.subsystems.DataSource;
 import com.launchdarkly.sdk.server.subsystems.DataSourceUpdateSinkV2;
 
+import java.lang.ref.WeakReference;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -591,21 +593,115 @@ private void maybeReportUnexpectedExhaustion(String message) {
 
     /**
      * Helper class to manage the lifecycle of conditions with automatic cleanup.
+     *
+     * <p>{@link #getFuture()} returns a <em>fresh</em> {@link CompletableFuture}
+     * per call rather than returning the same shared instance. This matters
+     * because the run loop calls {@code CompletableFuture.anyOf(getFuture(),
+     * synchronizerNext)} on every iteration: if {@code getFuture()} returned a
+     * shared instance, each {@code anyOf} call would permanently attach an
+     * {@code OrRelay} {@code Completion} to its {@code stack}. On a healthy
+     * primary synchronizer that streams ChangeSets without ever arming the
+     * fallback timer, the aggregate never completes, so those Completion nodes
+     * accumulate monotonically for the synchronizer's full tenure -- a real
+     * memory leak proportional to event rate.
+     *
+     * <p>The fix: a single permanent listener on the underlying aggregate fans
+     * out completion to every fresh future handed out by {@link #getFuture()}.
+     * Fresh futures are tracked via {@link WeakReference} on a pending list, so
+     * a fresh future whose only strong references were in the caller's loop
+     * iteration becomes garbage-collectable once that iteration ends. Pending
+     * entries whose referent has been collected are pruned opportunistically on
+     * subsequent {@code getFuture()} calls and on {@link #close()}.
+     *
+     * <p>Package-private (rather than private) so that direct unit tests can
+     * exercise the API surface and assert per-call distinctness.
      */
-    private static class Conditions implements AutoCloseable {
+    static class Conditions implements AutoCloseable {
         private final List<Condition> conditions;
-        private final CompletableFuture<Object> conditionsFuture;
+        private final CompletableFuture<Object> aggregate;
+        private final Object lock = new Object();
+
+        /**
+         * Holds the value the aggregate completed with, once it has completed.
+         * {@code volatile} so the fast path in {@link #getFuture()} avoids
+         * taking the lock. Set under {@code lock} together with clearing
+         * {@code pending} so the two stay consistent.
+         */
+        private volatile Object completedValue;
+
+        /**
+         * Tracks futures previously returned by {@link #getFuture()} that have
+         * not yet been completed. {@code null} once the aggregate has fired
+         * (and all pending entries have been drained). Mutated only under
+         * {@code lock}.
+         */
+        private List<WeakReference<CompletableFuture<Object>>> pending = new ArrayList<>();
 
         public Conditions(List<Condition> conditions) {
             this.conditions = conditions;
-            this.conditionsFuture = conditions.isEmpty()
+            this.aggregate = conditions.isEmpty()
                 ? new CompletableFuture<>() // Never completes if no conditions
                 : CompletableFuture.anyOf(
-                conditions.stream().map(Condition::execute).toArray(CompletableFuture[]::new));
+                    conditions.stream().map(Condition::execute).toArray(CompletableFuture[]::new));
+
+            // Single permanent listener. This is the only Completion node ever
+            // attached to aggregate.stack -- subsequent getFuture() calls do
+            // not touch the aggregate at all.
+            this.aggregate.whenComplete((result, throwable) -> {
+                List<WeakReference<CompletableFuture<Object>>> snapshot;
+                synchronized (lock) {
+                    completedValue = (throwable == null) ? result : null;
+                    snapshot = pending;
+                    pending = null;
+                }
+                if (snapshot == null) {
+                    return;
+                }
+                for (WeakReference<CompletableFuture<Object>> ref : snapshot) {
+                    CompletableFuture<Object> cf = ref.get();
+                    if (cf == null) {
+                        continue; // Already GC'd -- nothing to complete.
+                    }
+                    if (throwable != null) {
+                        cf.completeExceptionally(throwable);
+                    } else {
+                        cf.complete(result);
+                    }
+                }
+            });
         }
 
+        /**
+         * Returns a fresh future that will complete when the underlying
+         * aggregate condition fires, or an already-completed future if the
+         * aggregate has already fired by the time this method is called.
+         */
         public CompletableFuture<Object> getFuture() {
-            return conditionsFuture;
+            Object v = completedValue;
+            if (v != null) {
+                return CompletableFuture.completedFuture(v);
+            }
+
+            CompletableFuture<Object> fresh = new CompletableFuture<>();
+            synchronized (lock) {
+                if (pending == null) {
+                    // Raced with aggregate completion. completedValue is now
+                    // guaranteed populated (set under lock before pending was
+                    // nulled).
+                    return CompletableFuture.completedFuture(completedValue);
+                }
+                // Opportunistic prune of weak refs whose target has been
+                // collected. Keeps pending bounded even if the aggregate never
+                // fires.
+                Iterator<WeakReference<CompletableFuture<Object>>> it = pending.iterator();
+                while (it.hasNext()) {
+                    if (it.next().get() == null) {
+                        it.remove();
+                    }
+                }
+                pending.add(new WeakReference<>(fresh));
+            }
+            return fresh;
         }
 
         public void inform(FDv2SourceResult result) {
@@ -615,6 +711,11 @@ public void inform(FDv2SourceResult result) {
         @Override
         public void close() {
             conditions.forEach(Condition::close);
+            synchronized (lock) {
+                if (pending != null) {
+                    pending.clear();
+                }
+            }
         }
     }
 }

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

@@ -0,0 +1,148 @@
+package com.launchdarkly.sdk.server;
+
+import com.launchdarkly.sdk.server.FDv2DataSourceConditions.Condition;
+import com.launchdarkly.sdk.server.FDv2DataSourceConditions.FallbackCondition;
+import com.launchdarkly.sdk.server.FDv2DataSourceConditions.RecoveryCondition;
+import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
+import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Direct tests for {@link FDv2DataSource.Conditions}.
+ *
+ * <p>The Conditions class is the aggregator that races fallback/recovery
+ * condition futures against synchronizer.next() in the FDv2DataSource run
+ * loop. Each iteration of that loop calls getFuture() and passes the result to
+ * CompletableFuture.anyOf(...) -- so getFuture() must not return a shared
+ * instance, or every anyOf call permanently attaches a Completion node to the
+ * shared instance's stack, leaking memory proportional to event rate during
+ * the synchronizer's tenure on a healthy primary.
+ */
+public class FDv2DataSourceConditionsAggregateTest {
+    private ScheduledExecutorService executor;
+
+    @Before
+    public void setUp() {
+        executor = Executors.newScheduledThreadPool(1);
+    }
+
+    @After
+    public void tearDown() {
+        executor.shutdownNow();
+    }
+
+    /**
+     * Bug-proving test: getFuture() must return a fresh instance per call.
+     *
+     * <p>If it returns the same instance (as it did before the fix), the run
+     * loop's per-iteration {@code anyOf(getFuture(), syncNext)} attaches a new
+     * OrRelay Completion to the shared future's stack every iteration, with no
+     * deregister path -- a monotonic leak for a non-firing aggregate.
+     */
+    @Test
+    public void getFutureReturnsDistinctInstancesPerCall() {
+        Condition fallback = new FallbackCondition(executor, 60);
+        try (FDv2DataSource.Conditions conditions =
+                 new FDv2DataSource.Conditions(Collections.singletonList(fallback))) {
+            CompletableFuture<Object> f1 = conditions.getFuture();
+            CompletableFuture<Object> f2 = conditions.getFuture();
+            CompletableFuture<Object> f3 = conditions.getFuture();
+            assertThat(f1, not(sameInstance(f2)));
+            assertThat(f2, not(sameInstance(f3)));
+            assertThat(f1, not(sameInstance(f3)));
+        }
+    }
+
+    /**
+     * Even with no underlying conditions (a single-synchronizer configuration),
+     * getFuture() must return fresh instances. The aggregate never completes
+     * in this case, which is exactly the scenario where any per-iteration
+     * accumulation would be most damaging.
+     */
+    @Test
+    public void getFutureReturnsDistinctInstancesEvenWithNoConditions() {
+        try (FDv2DataSource.Conditions conditions =
+                 new FDv2DataSource.Conditions(Collections.emptyList())) {
+            CompletableFuture<Object> f1 = conditions.getFuture();
+            CompletableFuture<Object> f2 = conditions.getFuture();
+            assertThat(f1, not(sameInstance(f2)));
+        }
+    }
+
+    /**
+     * Every fresh future returned by getFuture() must complete when the
+     * underlying aggregate fires. The fan-out via the single permanent listener
+     * is what makes the fresh-per-call pattern work; verify it actually
+     * delivers.
+     */
+    @Test
+    public void allFreshFuturesCompleteWhenAggregateFires() throws Exception {
+        // 0-second timeout -> fires on first INTERRUPTED inform.
+        Condition fallback = new FallbackCondition(executor, 0);
+        try (FDv2DataSource.Conditions conditions =
+                 new FDv2DataSource.Conditions(Collections.singletonList(fallback))) {
+            CompletableFuture<Object> f1 = conditions.getFuture();
+            CompletableFuture<Object> f2 = conditions.getFuture();
+            CompletableFuture<Object> f3 = conditions.getFuture();
+
+            conditions.inform(makeInterruptedResult());
+
+            Object r1 = f1.get(2, TimeUnit.SECONDS);
+            Object r2 = f2.get(2, TimeUnit.SECONDS);
+            Object r3 = f3.get(2, TimeUnit.SECONDS);
+
+            assertNotNull(r1);
+            assertNotNull(r2);
+            assertNotNull(r3);
+            assertTrue(r1 instanceof Condition);
+            assertTrue(r2 instanceof Condition);
+            assertTrue(r3 instanceof Condition);
+        }
+    }
+
+    /**
+     * getFuture() called after the aggregate has already fired returns an
+     * already-completed future synchronously (the fast path).
+     */
+    @Test
+    public void getFutureAfterAggregateFiresReturnsCompletedFuture() throws Exception {
+        // RecoveryCondition arms its timer in the constructor and fires after
+        // the configured timeout. With timeout=0 it fires near-immediately.
+        Condition recovery = new RecoveryCondition(executor, 0);
+        try (FDv2DataSource.Conditions conditions =
+                 new FDv2DataSource.Conditions(Collections.singletonList(recovery))) {
+            // Drain a future to confirm the aggregate has fired.
+            conditions.getFuture().get(2, TimeUnit.SECONDS);
+
+            CompletableFuture<Object> postFire = conditions.getFuture();
+            assertTrue("post-fire getFuture() should be already complete", postFire.isDone());
+            assertNotNull(postFire.get(0, TimeUnit.SECONDS));
+        }
+    }
+
+    private static FDv2SourceResult makeInterruptedResult() {
+        return FDv2SourceResult.interrupted(
+            new DataSourceStatusProvider.ErrorInfo(
+                DataSourceStatusProvider.ErrorKind.NETWORK_ERROR,
+                0,
+                "simulated",
+                Instant.now()),
+            false);
+    }
+}