diff --git a/.changeset/queued-flags-reload-callbacks.md b/.changeset/queued-flags-reload-callbacks.md new file mode 100644 index 000000000..71b0f9747 --- /dev/null +++ b/.changeset/queued-flags-reload-callbacks.md @@ -0,0 +1,5 @@ +--- +"posthog": patch +--- + +Fix `onFeatureFlags` not running when the internal flags-loaded callback throws diff --git a/posthog/src/main/java/com/posthog/internal/PostHogRemoteConfig.kt b/posthog/src/main/java/com/posthog/internal/PostHogRemoteConfig.kt index f8b2cc33c..12f3c79b2 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogRemoteConfig.kt +++ b/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,7 +51,6 @@ 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 @@ -58,8 +58,8 @@ public class PostHogRemoteConfig( val distinctId: String, val anonymousId: String?, val groups: Map?, - val internalOnFeatureFlags: PostHogOnFeatureFlags?, - val onFeatureFlags: PostHogOnFeatureFlags?, + val internalOnFeatureFlags: List, + val onFeatureFlags: List, ) private var pendingFeatureFlagsRequest: PendingFeatureFlagsRequest? = null @@ -232,18 +232,33 @@ public class PostHogRemoteConfig( return recordingActive } + private fun List.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,22 +921,22 @@ 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.") @@ -915,8 +944,8 @@ public class PostHogRemoteConfig( distinctId = request.distinctId, anonymousId = request.anonymousId, groups = request.groups, - internalOnFeatureFlags = request.internalOnFeatureFlags, - onFeatureFlags = request.onFeatureFlags, + internalOnFeatureFlags = request.internalOnFeatureFlags.asSingleCallback(), + onFeatureFlags = request.onFeatureFlags.asSingleCallback(), ) } } diff --git a/posthog/src/test/java/com/posthog/internal/PostHogRemoteConfigTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogRemoteConfigTest.kt index 7f952925f..016101f06 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogRemoteConfigTest.kt +++ b/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() + } }