fix(flags): stop losing queued feature flag reload callbacks

.changeset/queued-flags-reload-callbacks.md

@@ -0,0 +1,5 @@
+---
+"posthog": patch
+---
+
+Fix `onFeatureFlags` not running when the internal flags-loaded callback throws

posthog/src/main/java/com/posthog/internal/PostHogRemoteConfig.kt

@@ -39,7 +39,8 @@ public class PostHogRemoteConfig(
         PostHogFeatureFlagCalledProvider { _, _ -> },
     private val onRemoteConfigLoaded: PostHogOnRemoteConfigLoaded? = null,
 ) : PostHogFeatureFlagsInterface {
-    private var isLoadingFeatureFlags = AtomicBoolean(false)
+    // guarded by pendingFeatureFlagsLock
+    private var isLoadingFeatureFlags = false
     private var isLoadingRemoteConfig = AtomicBoolean(false)
 
     // True once the live remote config has been resolved for the current identity, via either the
@@ -50,16 +51,15 @@ public class PostHogRemoteConfig(
 
     // Track if an additional reload was requested while a request was in flight
     // This prevents dropping reload requests (e.g., from identify()) when preload is in progress
-    private var pendingFeatureFlagsReload = AtomicBoolean(false)
     private val pendingFeatureFlagsLock = Any()
 
     // Stores the parameters for the pending feature flags reload
     private data class PendingFeatureFlagsRequest(
         val distinctId: String,
         val anonymousId: String?,
         val groups: Map<String, String>?,
-        val internalOnFeatureFlags: PostHogOnFeatureFlags?,
-        val onFeatureFlags: PostHogOnFeatureFlags?,
+        val internalOnFeatureFlags: List<PostHogOnFeatureFlags>,
+        val onFeatureFlags: List<PostHogOnFeatureFlags>,
     )
 
     private var pendingFeatureFlagsRequest: PendingFeatureFlagsRequest? = null
@@ -232,18 +232,33 @@ public class PostHogRemoteConfig(
         return recordingActive
     }
 
+    private fun List<PostHogOnFeatureFlags>.asSingleCallback(): PostHogOnFeatureFlags? =
+        when (size) {
+            0 -> null
+            1 -> first()
+            else -> PostHogOnFeatureFlags { forEach { it.runSafely() } }
+        }
+
+    private fun PostHogOnFeatureFlags.runSafely() {
+        try {
+            loaded()
+        } catch (e: Throwable) {
+            try {
+                config.logger.log("Executing the feature flags callback failed: $e")
+            } catch (ignored: Throwable) {
+                // a logger that throws must not escape and strand the in-flight claim below
+            }
+        }
+    }
+
     private fun runOnFeatureFlagsCallbacks(
         internalOnFeatureFlags: PostHogOnFeatureFlags?,
         onFeatureFlags: PostHogOnFeatureFlags?,
     ) {
         // if we don't load the feature flags (because there are none), we need to call the callback
         // because the app might be waiting for it.
-        try {
-            internalOnFeatureFlags?.loaded()
-            onFeatureFlags?.loaded()
-        } catch (e: Throwable) {
-            config.logger.log("Executing the feature flags callback failed: $e")
-        }
+        internalOnFeatureFlags?.runSafely()
+        onFeatureFlags?.runSafely()
     }
 
     // Notifies that a remote config resolution attempt finished. Callers set
@@ -699,21 +714,35 @@ public class PostHogRemoteConfig(
             return
         }
 
-        if (isLoadingFeatureFlags.getAndSet(true)) {
-            config.logger.log("Feature flags are being loaded already, queuing reload.")
-            // Queue the reload request instead of dropping it
-            // This ensures that requests with $anon_distinct_id (from identify()) are not lost
-            synchronized(pendingFeatureFlagsLock) {
-                pendingFeatureFlagsReload.set(true)
+        // Claiming the in-flight slot and queuing behind it must happen under one lock, or a reload can
+        // observe "already loading" after the in-flight request has drained, and strand itself.
+        val queued: Boolean
+        synchronized(pendingFeatureFlagsLock) {
+            queued = isLoadingFeatureFlags
+            isLoadingFeatureFlags = true
+            if (queued) {
+                // Newest parameters win; displaced callers' callbacks are carried over so none is
+                // lost. The internal callback can repeat here, so it must stay idempotent.
+                val displaced = pendingFeatureFlagsRequest
                 pendingFeatureFlagsRequest =
                     PendingFeatureFlagsRequest(
                         distinctId = distinctId,
                         anonymousId = anonymousId,
                         groups = groups,
-                        internalOnFeatureFlags = internalOnFeatureFlags,
-                        onFeatureFlags = onFeatureFlags,
+                        // distinct(): the SDK's own listeners and config.onFeatureFlags are singleton
+                        // instances reused across reloads, so appending each queued reload would fire
+                        // them once per coalesced reload instead of once per response.
+                        internalOnFeatureFlags =
+                            (displaced?.internalOnFeatureFlags.orEmpty() + listOfNotNull(internalOnFeatureFlags))
+                                .distinct(),
+                        onFeatureFlags =
+                            (displaced?.onFeatureFlags.orEmpty() + listOfNotNull(onFeatureFlags)).distinct(),
                     )
             }
+        }
+        if (queued) {
+            // Queuing rather than dropping keeps identify()'s $anon_distinct_id request alive
+            config.logger.log("Feature flags are being loaded already, queuing reload.")
             return
         }
 
@@ -892,31 +921,31 @@ public class PostHogRemoteConfig(
                 notifyRemoteConfigResolved()
             }
         } finally {
-            runOnFeatureFlagsCallbacks(
-                internalOnFeatureFlags = internalOnFeatureFlags,
-                onFeatureFlags = onFeatureFlags,
-            )
+            try {
+                runOnFeatureFlagsCallbacks(
+                    internalOnFeatureFlags = internalOnFeatureFlags,
+                    onFeatureFlags = onFeatureFlags,
+                )
+            } catch (ignored: Throwable) {
+                // never skip the release below: leaving the claim set stops all future flag loads
+            }
 
             // Check if there's a pending reload request and execute it
             val pendingRequest: PendingFeatureFlagsRequest?
             synchronized(pendingFeatureFlagsLock) {
-                if (pendingFeatureFlagsReload.getAndSet(false)) {
-                    pendingRequest = pendingFeatureFlagsRequest
-                    pendingFeatureFlagsRequest = null
-                } else {
-                    pendingRequest = null
-                }
+                pendingRequest = pendingFeatureFlagsRequest
+                pendingFeatureFlagsRequest = null
+                isLoadingFeatureFlags = false
             }
-            isLoadingFeatureFlags.set(false)
 
             pendingRequest?.let { request ->
                 config.logger.log("Executing pending feature flags reload.")
                 executeFeatureFlags(
                     distinctId = request.distinctId,
                     anonymousId = request.anonymousId,
                     groups = request.groups,
-                    internalOnFeatureFlags = request.internalOnFeatureFlags,
-                    onFeatureFlags = request.onFeatureFlags,
+                    internalOnFeatureFlags = request.internalOnFeatureFlags.asSingleCallback(),
+                    onFeatureFlags = request.onFeatureFlags.asSingleCallback(),
                 )
             }
         }

posthog/src/test/java/com/posthog/internal/PostHogRemoteConfigTest.kt

@@ -2744,4 +2744,155 @@ internal class PostHogRemoteConfigTest {
         sut.clear()
         http.shutdown()
     }
+
+    // The queuing branch is unreachable on the single-threaded executor every production path uses,
+    // so forcing overlap is the only way to exercise the contract that machinery claims.
+    @Test
+    fun `queued reloads displaced from the pending slot still run their callbacks`() {
+        val http = mockHttp(response = MockResponse().setBody(responseFlagsApi))
+        repeat(9) {
+            http.enqueue(MockResponse().setBody(responseFlagsApi).setBodyDelay(300, TimeUnit.MILLISECONDS))
+        }
+        val overlapping = Executors.newFixedThreadPool(4, PostHogThreadFactory("TestOverlap"))
+        val config = PostHogConfig(API_KEY, http.url("/").toString()).apply { cachePreferences = preferences }
+        val sut =
+            PostHogRemoteConfig(
+                config,
+                PostHogApi(config),
+                executor = overlapping,
+                defaultPersonPropertiesProvider = { emptyMap() },
+                onRemoteConfigLoaded = null,
+            )
+
+        val reloads = 10
+        val fired = CountDownLatch(reloads)
+        val counts = List(reloads) { AtomicInteger(0) }
+        repeat(reloads) { index ->
+            sut.loadFeatureFlags(
+                "my_identify",
+                anonymousId = "anonId",
+                emptyMap(),
+                onFeatureFlags =
+                    PostHogOnFeatureFlags {
+                        counts[index].incrementAndGet()
+                        fired.countDown()
+                    },
+            )
+        }
+
+        assertTrue(fired.await(10, TimeUnit.SECONDS), "every queued reload must run its callback, ${fired.count} never did")
+        assertTrue(http.requestCount < reloads, "reloads must have coalesced, otherwise this never reached the queuing branch")
+        counts.forEachIndexed { index, count -> assertEquals(1, count.get(), "callback $index ran ${count.get()} times") }
+
+        sut.clear()
+        overlapping.shutdownAndAwaitTermination()
+        http.shutdown()
+    }
+
+    @Test
+    fun `a listener reused across queued reloads is notified once per response`() {
+        val http = mockHttp(response = MockResponse().setBody(responseFlagsApi))
+        repeat(9) {
+            http.enqueue(MockResponse().setBody(responseFlagsApi).setBodyDelay(300, TimeUnit.MILLISECONDS))
+        }
+        val overlapping = Executors.newFixedThreadPool(4, PostHogThreadFactory("TestShared"))
+        val config = PostHogConfig(API_KEY, http.url("/").toString()).apply { cachePreferences = preferences }
+        val sut =
+            PostHogRemoteConfig(
+                config,
+                PostHogApi(config),
+                executor = overlapping,
+                defaultPersonPropertiesProvider = { emptyMap() },
+                onRemoteConfigLoaded = null,
+            )
+
+        // the SDK passes the same listener instance on every automatic reload
+        val invocations = AtomicInteger(0)
+        val shared = PostHogOnFeatureFlags { invocations.incrementAndGet() }
+        repeat(10) {
+            sut.loadFeatureFlags("my_identify", anonymousId = "anonId", emptyMap(), onFeatureFlags = shared)
+        }
+
+        Thread.sleep(3000)
+        assertTrue(
+            invocations.get() <= http.requestCount,
+            "shared listener fired ${invocations.get()} times for ${http.requestCount} responses",
+        )
+
+        sut.clear()
+        overlapping.shutdownAndAwaitTermination()
+        http.shutdown()
+    }
+
+    @Test
+    fun `a throwing logger in a throwing callback does not stop later reloads`() {
+        val http = mockHttp(response = MockResponse().setBody(responseFlagsApi))
+        repeat(3) { http.enqueue(MockResponse().setBody(responseFlagsApi)) }
+        val config =
+            PostHogConfig(API_KEY, http.url("/").toString()).apply {
+                cachePreferences = preferences
+                logger =
+                    object : PostHogLogger {
+                        override fun log(message: String) = throw IllegalStateException("logger blew up")
+
+                        override fun isEnabled(): Boolean = true
+                    }
+            }
+        val sut =
+            PostHogRemoteConfig(
+                config,
+                PostHogApi(config),
+                executor = executor,
+                defaultPersonPropertiesProvider = { emptyMap() },
+                onRemoteConfigLoaded = null,
+            )
+
+        val first = CountDownLatch(1)
+        sut.loadFeatureFlags(
+            "my_identify",
+            anonymousId = "anonId",
+            emptyMap(),
+            onFeatureFlags =
+                PostHogOnFeatureFlags {
+                    first.countDown()
+                    throw IllegalStateException("callback blew up")
+                },
+        )
+        first.await(5, TimeUnit.SECONDS)
+        val afterFirst = http.requestCount
+
+        val second = CountDownLatch(1)
+        sut.loadFeatureFlags(
+            "my_identify",
+            anonymousId = "anonId",
+            emptyMap(),
+            onFeatureFlags = PostHogOnFeatureFlags { second.countDown() },
+        )
+
+        assertTrue(second.await(5, TimeUnit.SECONDS), "a later reload never ran - the in-flight claim was stranded")
+        assertTrue(http.requestCount > afterFirst, "a later reload never reached the network")
+
+        sut.clear()
+        http.shutdown()
+    }
+
+    @Test
+    fun `a throwing internal callback does not suppress the caller's callback`() {
+        val http = mockHttp(response = MockResponse().setBody(responseFlagsApi))
+        val sut = getSut(host = http.url("/").toString())
+
+        val userCallbackRan = CountDownLatch(1)
+        sut.loadFeatureFlags(
+            "my_identify",
+            anonymousId = "anonId",
+            emptyMap(),
+            internalOnFeatureFlags = PostHogOnFeatureFlags { throw IllegalStateException("internal blew up") },
+            onFeatureFlags = PostHogOnFeatureFlags { userCallbackRan.countDown() },
+        )
+
+        assertTrue(userCallbackRan.await(5, TimeUnit.SECONDS), "the caller's callback was suppressed by the internal one")
+
+        sut.clear()
+        http.shutdown()
+    }
 }