From 66965b7d8ad594eb15b4366d705b9fc01c9206f9 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 13:47:08 +0200 Subject: [PATCH 1/2] fix(server): honor versioned local property matching --- .changeset/quiet-booleans-match.md | 6 + .../PostHogFlagDefinitionCacheProvider.kt | 4 +- .../posthog/server/internal/FlagEvaluator.kt | 7 +- .../server/internal/PostHogFeatureFlags.kt | 72 ++-- .../server/internal/FlagEvaluatorTest.kt | 54 +++ .../internal/VersionedPropertyMatchingTest.kt | 344 ++++++++++++++++++ posthog/api/posthog.api | 8 +- .../internal/PostHogLocalEvaluationModels.kt | 16 +- 8 files changed, 473 insertions(+), 38 deletions(-) create mode 100644 .changeset/quiet-booleans-match.md create mode 100644 posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt diff --git a/.changeset/quiet-booleans-match.md b/.changeset/quiet-booleans-match.md new file mode 100644 index 000000000..e9fb81bb5 --- /dev/null +++ b/.changeset/quiet-booleans-match.md @@ -0,0 +1,6 @@ +--- +"posthog": patch +"posthog-server": patch +--- + +Retain `property_matching_version` in local-evaluation definitions and shared caches. Server-side local evaluation now uses explicit boolean matching for version 2, preserves legacy matching for missing/1, and keeps one definition snapshot through group, cohort, and dependency evaluation and version-only refreshes. diff --git a/posthog-server/src/main/java/com/posthog/server/PostHogFlagDefinitionCacheProvider.kt b/posthog-server/src/main/java/com/posthog/server/PostHogFlagDefinitionCacheProvider.kt index b02d0e027..30d2c385b 100644 --- a/posthog-server/src/main/java/com/posthog/server/PostHogFlagDefinitionCacheProvider.kt +++ b/posthog-server/src/main/java/com/posthog/server/PostHogFlagDefinitionCacheProvider.kt @@ -25,7 +25,9 @@ public interface PostHogFlagDefinitionCacheProvider { * Return cached flag definitions, or null when the cache is empty or unavailable. * * The data should use the shared local-evaluation definitions shape returned by PostHog's - * `/flags/definitions` endpoint: `flags`, `group_type_mapping`, and `cohorts`. + * `/flags/definitions` endpoint: `flags`, `group_type_mapping`, `cohorts`, and + * `property_matching_version`. Preserve the matching version with the definitions; + * older entries without it use legacy property matching. */ public fun getFlagDefinitions(): CompletionStage?> diff --git a/posthog-server/src/main/java/com/posthog/server/internal/FlagEvaluator.kt b/posthog-server/src/main/java/com/posthog/server/internal/FlagEvaluator.kt index 2ddbc0f71..c51c4433c 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/FlagEvaluator.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/FlagEvaluator.kt @@ -26,6 +26,7 @@ import java.util.regex.PatternSyntaxException */ internal class FlagEvaluator( private val config: PostHogConfig, + private val propertyMatchingVersion: Int? = null, ) { companion object { private const val LONG_SCALE = 0xFFFFFFFFFFFFFFF.toDouble() @@ -240,7 +241,11 @@ internal class FlagEvaluator( propertyValue: Any?, overrideValue: Any?, ): Boolean { - if (isTruthyOrFalsyPropertyValue(propertyValue)) { + // Empty filters retain recursive ALL truthiness in both matching versions. + if (propertyValue is List<*> && propertyValue.isEmpty()) { + return isTruthyPropertyValue(overrideValue) + } + if (propertyMatchingVersion != 2 && isTruthyOrFalsyPropertyValue(propertyValue)) { return isTruthyPropertyValue(propertyValue) == isTruthyPropertyValue(overrideValue) } diff --git a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt index 631c9f055..d3df06a9d 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt @@ -38,7 +38,8 @@ internal class PostHogFeatureFlags( private val missingFlagKeysMaxSize: Int = DEFAULT_MISSING_FLAG_KEYS_MAX_SIZE, private val missingFlagProbeWaitTimeoutMs: Long = MISSING_FLAG_PROBE_WAIT_TIMEOUT_MS, ) : PostHogFeatureFlagsInterface { - private val cache = + @Volatile + private var cache = PostHogFeatureFlagCache( maxSize = cacheMaxSize, maxAgeMs = cacheMaxAgeMs, @@ -51,19 +52,19 @@ internal class PostHogFeatureFlags( private var missingFlagKeysGeneration: Long = 0 private var remoteFlagEvidenceSequence: Long = 0 - @Volatile - private var featureFlags: List? = null - - @Volatile - private var flagDefinitions: Map? = null - - @Volatile - private var cohorts: Map? = null + // Publish definitions and their immutable matcher together; one evaluation pass retains this snapshot. + private data class DefinitionSnapshot( + val flagsByKey: Map?, + val groupTypeMapping: Map?, + val cohorts: Map?, + val evaluator: FlagEvaluator, + ) @Volatile - private var groupTypeMapping: Map? = null + private var definitionSnapshot: DefinitionSnapshot? = null - private val evaluator: FlagEvaluator = FlagEvaluator(config) + private val flagDefinitions: Map? + get() = definitionSnapshot?.flagsByKey @Volatile private var poller: LocalEvaluationPoller? = null @@ -188,7 +189,8 @@ internal class PostHogFeatureFlags( loadFeatureFlagDefinitions() } - val flagDef = flagDefinitions?.get(key) + val snapshot = definitionSnapshot + val flagDef = snapshot?.flagsByKey?.get(key) if (flagDef != null) { try { config.logger.log("Attempting local evaluation for flag '$key' for distinctId: $distinctId") @@ -196,6 +198,7 @@ internal class PostHogFeatureFlags( val result = computeFlagLocally( + snapshot = snapshot, key = key, distinctId = distinctId, personProperties = props, @@ -301,7 +304,8 @@ internal class PostHogFeatureFlags( loadFeatureFlagDefinitions() } - val currentFlagDefinitions = flagDefinitions + val snapshot = definitionSnapshot + val currentFlagDefinitions = snapshot?.flagsByKey if (currentFlagDefinitions == null) { return null } @@ -320,6 +324,7 @@ internal class PostHogFeatureFlags( try { val result = computeFlagLocally( + snapshot = snapshot, key = key, distinctId = distinctId, personProperties = props, @@ -395,6 +400,8 @@ internal class PostHogFeatureFlags( bypassCache: Boolean = false, onResponse: ((PostHogFlagsResponse) -> Unit)? = null, ): Map? { + // A refresh replaces the cache; an in-flight response may only populate its original cache. + val responseCache = cache val cacheKey = FeatureFlagCacheKey( distinctId = distinctId, @@ -406,7 +413,7 @@ internal class PostHogFeatureFlags( ) if (!bypassCache) { - val cachedFlags = cache.get(cacheKey) + val cachedFlags = responseCache.get(cacheKey) if (cachedFlags != null) { return cachedFlags } @@ -426,7 +433,7 @@ internal class PostHogFeatureFlags( disableGeoip = disableGeoip, ) val flags = response?.flags - cache.put( + responseCache.put( cacheKey, flags, response?.requestId, @@ -440,23 +447,23 @@ internal class PostHogFeatureFlags( flags } catch (e: SocketTimeoutException) { config.logger.log("Loading remote feature flags timed out: $e") - cache.put(cacheKey, null, error = FeatureFlagError.TIMEOUT) + responseCache.put(cacheKey, null, error = FeatureFlagError.TIMEOUT) null } catch (e: ConnectException) { config.logger.log("Loading remote feature flags connection failed: $e") - cache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) + responseCache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) null } catch (e: UnknownHostException) { config.logger.log("Loading remote feature flags DNS lookup failed: $e") - cache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) + responseCache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) null } catch (e: PostHogApiError) { config.logger.log("Loading remote feature flags API error: $e") - cache.put(cacheKey, null, error = FeatureFlagError.apiError(e.statusCode)) + responseCache.put(cacheKey, null, error = FeatureFlagError.apiError(e.statusCode)) null } catch (e: Throwable) { config.logger.log("Loading remote feature flags failed: $e") - cache.put(cacheKey, null, error = FeatureFlagError.UNKNOWN_ERROR) + responseCache.put(cacheKey, null, error = FeatureFlagError.UNKNOWN_ERROR) null } } @@ -599,6 +606,7 @@ internal class PostHogFeatureFlags( flags = apiResponse.flags, groupTypeMapping = apiResponse.groupTypeMapping, cohorts = apiResponse.cohorts, + propertyMatchingVersion = apiResponse.propertyMatchingVersion, ) config.logger.log("Loaded ${apiResponse.flags?.size ?: 0} feature flags for local evaluation") @@ -651,6 +659,7 @@ internal class PostHogFeatureFlags( flags = response.flags, groupTypeMapping = response.groupTypeMapping, cohorts = response.cohorts, + propertyMatchingVersion = response.propertyMatchingVersion, ) config.logger.log("Loaded ${response.flags?.size ?: 0} feature flags from flag definition cache") notifyFeatureFlagsLoaded() @@ -705,6 +714,7 @@ internal class PostHogFeatureFlags( "flags" to (response.flags ?: emptyList()), "group_type_mapping" to (response.groupTypeMapping ?: emptyMap()), "cohorts" to (response.cohorts ?: emptyMap()), + "property_matching_version" to response.propertyMatchingVersion, ) val writer = StringWriter() config.serializer.serialize(cacheData, writer) @@ -728,14 +738,19 @@ internal class PostHogFeatureFlags( flags: List?, groupTypeMapping: Map?, cohorts: Map?, + propertyMatchingVersion: Int?, ) { val invalidated = synchronized(missingFlagKeysLock) { synchronized(loadLock) { - featureFlags = flags - flagDefinitions = flags?.associateBy { it.key } - this.cohorts = cohorts - this.groupTypeMapping = groupTypeMapping + definitionSnapshot = + DefinitionSnapshot( + flags?.associateBy { it.key }, + groupTypeMapping, + cohorts, + FlagEvaluator(config, propertyMatchingVersion), + ) + cache = PostHogFeatureFlagCache(maxSize = cacheMaxSize, maxAgeMs = cacheMaxAgeMs) definitionsLoaded = true definitionsLoadedAt = System.currentTimeMillis() } @@ -855,13 +870,14 @@ internal class PostHogFeatureFlags( * Compute a flag locally using the evaluation engine */ private fun computeFlagLocally( + snapshot: DefinitionSnapshot, key: String, distinctId: String, groups: Map?, personProperties: Map?, groupProperties: Map>?, ): Any? { - val flags = this.flagDefinitions ?: return null + val flags = snapshot.flagsByKey ?: return null val flag = flags[key] ?: return null if (!flag.active) { @@ -874,7 +890,7 @@ internal class PostHogFeatureFlags( val (evaluationId, evaluationProperties) = if (aggregationGroupIndex != null) { // Group-based flag - evaluate at group level - val groupTypeName = groupTypeMapping?.get(aggregationGroupIndex.toString()) + val groupTypeName = snapshot.groupTypeMapping?.get(aggregationGroupIndex.toString()) if (groupTypeName == null) { config.logger.log("Unknown group type index $aggregationGroupIndex for flag '$key'") @@ -896,11 +912,11 @@ internal class PostHogFeatureFlags( } val evaluationCache = mutableMapOf() - return evaluator.matchFeatureFlagProperties( + return snapshot.evaluator.matchFeatureFlagProperties( flag = flag, distinctId = evaluationId, properties = evaluationProperties ?: EMPTY_PROPERTIES, - cohortProperties = cohorts ?: EMPTY_COHORT_PROPERTIES, + cohortProperties = snapshot.cohorts ?: EMPTY_COHORT_PROPERTIES, flagsByKey = flags, evaluationCache = evaluationCache, ) diff --git a/posthog-server/src/test/java/com/posthog/server/internal/FlagEvaluatorTest.kt b/posthog-server/src/test/java/com/posthog/server/internal/FlagEvaluatorTest.kt index 7a06d9af8..6d7a080a6 100644 --- a/posthog-server/src/test/java/com/posthog/server/internal/FlagEvaluatorTest.kt +++ b/posthog-server/src/test/java/com/posthog/server/internal/FlagEvaluatorTest.kt @@ -25,6 +25,60 @@ internal class FlagEvaluatorTest { evaluator = FlagEvaluator(config) } + @Test + internal fun testVersionedExactAndIsNotMatching() { + data class Row(val filter: Any?, val value: Any?, val legacy: Boolean, val explicit: Boolean) + + val rows = + listOf( + Row(false, "banana", true, false), + Row(false, 0, true, false), + Row(listOf("true", "false"), "true", false, true), + Row(listOf("true", "false"), "pro", true, false), + Row(emptyList(), true, true, true), + Row(emptyList(), emptyList(), true, true), + Row(true, listOf(true), true, false), + Row(false, "FALSE", true, true), + Row(false, null, true, false), + Row(false, "", true, false), + Row(emptyList(), listOf(true, listOf("TRUE", emptyList())), true, true), + Row(emptyList(), false, false, false), + Row(emptyList(), 1, false, false), + Row(emptyList(), "banana", false, false), + Row(emptyList(), null, false, false), + Row(listOf(true, "x"), "TRUE", true, true), + Row(listOf("FREE", "PRO"), "pro", true, true), + Row(listOf(null, "x"), null, true, true), + Row(listOf(listOf(true)), true, true, false), + Row("İ", "i\u0307", true, true), + ) + for (version in listOf(null, 1, 2, 3)) { + val matcher = FlagEvaluator(config, version) + for (row in rows) { + for (operator in listOf(PropertyOperator.EXACT, PropertyOperator.IS_NOT)) { + val expected = if (version == 2) row.explicit else row.legacy + val property = FlagProperty("value", row.filter, operator, PropertyType.PERSON, false, null) + assertEquals( + "version=$version operator=$operator row=$row", + if (operator == PropertyOperator.EXACT) expected else !expected, + matcher.matchProperty(property, mapOf("value" to row.value)), + ) + kotlin.test.assertFailsWith { + matcher.matchProperty(property, emptyMap()) + } + } + } + kotlin.test.assertFailsWith { + matcher.matchProperty( + FlagProperty("value", "1.5", PropertyOperator.EXACT, PropertyType.PERSON, false, null), + mapOf("value" to 1.5), + ) + } + } + // Existing callers that do not select a version retain service legacy behavior. + assertTrue(evaluator.matchProperty(FlagProperty("value", false, null, null, null, null), mapOf("value" to "banana"))) + } + @Test internal fun testHashConsistency() { // Test that hash function returns consistent values for same inputs diff --git a/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt b/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt new file mode 100644 index 000000000..de142f218 --- /dev/null +++ b/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt @@ -0,0 +1,344 @@ +package com.posthog.server.internal + +import com.posthog.PostHogConfig +import com.posthog.internal.PostHogApi +import com.posthog.internal.PostHogLogger +import com.posthog.server.CountingDispatcher +import com.posthog.server.PostHogBlockingFlagDefinitionCacheProvider +import com.posthog.server.PostHogFlagDefinitionCacheProvider +import com.posthog.server.createFlagsResponse +import com.posthog.server.createTestConfig +import com.posthog.server.jsonResponse +import com.posthog.server.shutdownAndAwaitTermination +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import java.nio.file.Files +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionStage +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +internal class VersionedPropertyMatchingTest { + private val keys = listOf("person", "group", "cohort", "dependency") + private val properties = mapOf("value" to "banana") + private val groups = mapOf("company" to "company-1") + private val groupProperties = mapOf("company" to properties) + + private fun definitions(version: Int?): String { + val leaf = """{"key": "value", "value": false, "operator": "exact", "type": "person"}""" + val cohort = """{"key": "id", "value": "outer", "type": "cohort"}""" + val dependency = + """ + {"key": "cohort", "value": true, "operator": "flag_evaluates_to", + "type": "flag", "dependency_chain": ["cohort"]} + """.trimIndent() + + fun flag( + key: String, + condition: String, + group: Boolean = false, + ): String = + """ + {"id": 1, "name": "$key", "key": "$key", "active": true, "version": 42, + "filters": {${if (group) "\"aggregation_group_type_index\": 0," else ""} + "groups": [{"properties": [$condition]}]}} + """.trimIndent() + + return """ + { + ${version?.let { "\"property_matching_version\": $it," } ?: ""} + "flags": [${flag("person", leaf)}, ${flag("group", leaf, true)}, + ${flag("cohort", cohort)}, ${flag("dependency", dependency)}], + "group_type_mapping": {"0": "company"}, + "cohorts": { + "outer": {"type": "AND", "values": [{"type": "OR", "values": [{"key": "id", "value": "inner", "type": "cohort"}]}]}, + "inner": {"type": "AND", "values": [$leaf]} + } + } + """.trimIndent() + } + + private fun createSut( + config: PostHogConfig, + provider: PostHogFlagDefinitionCacheProvider? = null, + ): PostHogFeatureFlags = + PostHogFeatureFlags( + config, + PostHogApi(config), + 60000, + 100, + localEvaluation = true, + personalApiKey = "personal", + pollerEnabled = false, + flagDefinitionCacheProvider = provider, + ) + + private fun assertLocalResults( + sut: PostHogFeatureFlags, + expected: Boolean, + ) { + for (key in keys) { + assertEquals( + expected, + sut.getFeatureFlag( + key, + distinctId = "user", + personProperties = properties, + groups = groups, + groupProperties = groupProperties, + ), + key, + ) + assertEquals(expected, sut.getFeatureFlagResult(key, "user", groups, properties, groupProperties)?.enabled, key) + } + val all = sut.getFeatureFlags("user", groups, properties, groupProperties) + assertEquals(keys.toSet(), all?.keys) + assertTrue(all!!.values.all { it.enabled == expected }) + // Selecting only the dependent flag must still use the full snapshot for recursive evaluation. + for (selection in listOf(null, listOf("dependency"))) { + val result = sut.evaluateFlags("user", groups, properties, groupProperties, selection, true, false) + assertEquals(selection?.toSet() ?: keys.toSet(), result.flags.keys) + assertTrue(result.flags.values.all { it.enabled == expected }) + assertTrue(result.locallyEvaluated.values.all { it }) + } + } + + @Test + fun `definitions response selects explicit boolean matching`() { + val http = MockWebServer() + http.enqueue(jsonResponse(definitions(2))) + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + try { + sut.loadFeatureFlagDefinitions() + assertLocalResults(sut, false) + assertEquals(1, http.requestCount) + } finally { + sut.clear() + sut.shutDown() + http.shutdown() + } + } + + @Test + fun `is_not complements person group cohort and dependency leaf results`() { + val http = MockWebServer() + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + try { + for (version in listOf(null, 1, 2)) { + http.enqueue(jsonResponse(definitions(version).replace("\"exact\"", "\"is_not\""))) + sut.loadFeatureFlagDefinitions() + assertLocalResults(sut, version == 2) + } + assertEquals(3, http.requestCount) + } finally { + sut.clear() + sut.shutDown() + http.shutdown() + } + } + + @Test + fun `version only refresh resets matching and preserves snapshot on 304 and failure`() { + val http = MockWebServer() + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + try { + for (version in listOf(null, 1, 2, 1, 2, null, 3, 2)) { + http.enqueue(jsonResponse(definitions(version)).setHeader("ETag", "version-$version")) + sut.loadFeatureFlagDefinitions() + assertLocalResults(sut, version != 2) + } + http.enqueue(MockResponse().setResponseCode(304)) + sut.loadFeatureFlagDefinitions() + assertLocalResults(sut, false) + http.enqueue(MockResponse().setResponseCode(500)) + sut.loadFeatureFlagDefinitions() + assertLocalResults(sut, false) + assertEquals(10, http.requestCount, "All requests must be definition loads, never remote evaluation") + repeat(10) { + assertTrue(http.takeRequest().path!!.startsWith("/api/feature_flag/local_evaluation/")) + } + } finally { + sut.clear() + sut.shutDown() + http.shutdown() + } + } + + @Test + fun `definition refresh invalidates cached remote results even when only version changes`() { + val http = MockWebServer() + http.enqueue(jsonResponse(definitions(1))) + http.enqueue(jsonResponse(createFlagsResponse("person", enabled = true))) + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + try { + sut.loadFeatureFlagDefinitions() + // Requesting an unknown flag populates the remote cache for this identity and property set. + sut.getFeatureFlag("unknown", distinctId = "user", personProperties = properties) + assertEquals(true, sut.getFeatureFlag("person", distinctId = "user", personProperties = properties)) + http.enqueue(jsonResponse(definitions(2))) + sut.loadFeatureFlagDefinitions() + assertEquals(false, sut.getFeatureFlag("person", distinctId = "user", personProperties = properties)) + http.enqueue(jsonResponse(definitions(1))) + sut.loadFeatureFlagDefinitions() + assertEquals(true, sut.getFeatureFlag("person", distinctId = "user", personProperties = properties)) + assertEquals(4, http.requestCount) + } finally { + sut.clear() + sut.shutDown() + http.shutdown() + } + } + + @Test + fun `version only refresh prevents an in flight remote response from repopulating the cache`() { + val responseStarted = CountDownLatch(1) + val releaseResponse = CountDownLatch(1) + val version = AtomicInteger(1) + val dispatcher = + CountingDispatcher( + { jsonResponse(definitions(version.get())) }, + { + responseStarted.countDown() + check(releaseResponse.await(5, TimeUnit.SECONDS)) + jsonResponse(createFlagsResponse("person", enabled = true)) + }, + ) + val http = MockWebServer() + http.dispatcher = dispatcher + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + val executor = Executors.newSingleThreadExecutor() + try { + sut.loadFeatureFlagDefinitions() + val pending = + executor.submit { + // An unknown-key fallback shares the same result-cache key as subsequent local reads. + sut.getFeatureFlag( + "unknown", + distinctId = "user", + groups = groups, + personProperties = properties, + groupProperties = groupProperties, + ) + } + assertTrue(responseStarted.await(5, TimeUnit.SECONDS)) + version.set(2) + sut.loadFeatureFlagDefinitions() + releaseResponse.countDown() + pending.get(5, TimeUnit.SECONDS) + + assertLocalResults(sut, false) + val result = sut.evaluateFlags("user", groups, properties, groupProperties, null, false, false) + assertEquals(keys.toSet(), result.flags.keys) + assertTrue(result.flags.values.all { !it.enabled }) + assertTrue(result.locallyEvaluated.values.all { it }) + assertEquals(2, dispatcher.localEvaluationCalls.get()) + assertEquals(1, dispatcher.flagsCalls.get(), "New-snapshot reads must not fall back remotely") + } finally { + releaseResponse.countDown() + executor.shutdownAndAwaitTermination() + sut.clear() + sut.shutDown() + http.shutdown() + } + } + + @Test + fun `disk cache round trip retains version and older entries reset to legacy`() { + val http = MockWebServer() + http.enqueue(jsonResponse(definitions(2))) + http.start() + val config = createTestConfig(host = http.url("/").toString()) + val file = Files.createTempFile("posthog-definitions", ".json").toFile() + var fetch = true + var cacheUnavailable = false + val provider = + object : PostHogBlockingFlagDefinitionCacheProvider() { + override fun shouldFetchFlagDefinitionsBlocking(): Boolean = fetch + + override fun getFlagDefinitionsBlocking(): Map? = + if (cacheUnavailable) null else file.reader().use { config.serializer.deserialize(it) } + + override fun onFlagDefinitionsReceivedBlocking(data: Map) { + file.writer().use { config.serializer.serialize(data, it) } + } + } + val writer = createSut(config, provider) + val asyncProvider = + object : PostHogFlagDefinitionCacheProvider by provider { + override fun getFlagDefinitions(): CompletionStage?> = + CompletableFuture.supplyAsync { provider.getFlagDefinitionsBlocking() } + } + val reader = createSut(config, asyncProvider) + try { + writer.loadFeatureFlagDefinitions() + assertEquals(2, (provider.getFlagDefinitionsBlocking()!!["property_matching_version"] as Number).toInt()) + fetch = false + reader.loadFeatureFlagDefinitions() + assertLocalResults(reader, false) + cacheUnavailable = true + reader.loadFeatureFlagDefinitions() + assertLocalResults(reader, false) + cacheUnavailable = false + file.writeText("invalid-json") + reader.loadFeatureFlagDefinitions() + assertLocalResults(reader, false) + for (version in listOf(1, 2, null)) { + file.writeText(definitions(version)) + reader.loadFeatureFlagDefinitions() + assertLocalResults(reader, version != 2) + } + assertEquals(1, http.requestCount, "Cache hydration must not fetch or fall back remotely") + } finally { + writer.clear() + writer.shutDown() + reader.clear() + reader.shutDown() + http.shutdown() + file.delete() + } + } + + @Test + fun `evaluation retains one snapshot when definitions refresh during a pass`() { + val http = MockWebServer() + http.enqueue(jsonResponse(definitions(1))) + http.enqueue(jsonResponse(definitions(2))) + http.start() + val config = createTestConfig(host = http.url("/").toString()) + val sut = createSut(config) + try { + sut.loadFeatureFlagDefinitions() + var refresh = true + config.logger = + object : PostHogLogger { + override fun isEnabled(): Boolean = true + + override fun log(message: String) { + if (refresh && message.startsWith("Attempting local evaluation for distinctId:")) { + refresh = false + sut.loadFeatureFlagDefinitions() + } + } + } + val result = sut.evaluateFlags("user", groups, properties, groupProperties, null, true, false) + assertEquals(keys.toSet(), result.flags.keys) + assertTrue(result.flags.values.all { it.enabled }, "The in-progress pass must retain legacy definitions and matcher") + assertLocalResults(sut, false) + assertEquals(2, http.requestCount) + } finally { + sut.clear() + sut.shutDown() + http.shutdown() + } + } +} diff --git a/posthog/api/posthog.api b/posthog/api/posthog.api index ec32a74d9..9d574a12c 100644 --- a/posthog/api/posthog.api +++ b/posthog/api/posthog.api @@ -716,15 +716,19 @@ public final class com/posthog/internal/LocalEvaluationApiResponse$Companion { public final class com/posthog/internal/LocalEvaluationResponse { public fun (Ljava/util/List;Ljava/util/Map;Ljava/util/Map;)V + public fun (Ljava/util/List;Ljava/util/Map;Ljava/util/Map;Ljava/lang/Integer;)V + public synthetic fun (Ljava/util/List;Ljava/util/Map;Ljava/util/Map;Ljava/lang/Integer;ILkotlin/jvm/internal/DefaultConstructorMarker;)V public final fun component1 ()Ljava/util/List; public final fun component2 ()Ljava/util/Map; public final fun component3 ()Ljava/util/Map; - public final fun copy (Ljava/util/List;Ljava/util/Map;Ljava/util/Map;)Lcom/posthog/internal/LocalEvaluationResponse; - public static synthetic fun copy$default (Lcom/posthog/internal/LocalEvaluationResponse;Ljava/util/List;Ljava/util/Map;Ljava/util/Map;ILjava/lang/Object;)Lcom/posthog/internal/LocalEvaluationResponse; + public final fun component4 ()Ljava/lang/Integer; + public final fun copy (Ljava/util/List;Ljava/util/Map;Ljava/util/Map;Ljava/lang/Integer;)Lcom/posthog/internal/LocalEvaluationResponse; + public static synthetic fun copy$default (Lcom/posthog/internal/LocalEvaluationResponse;Ljava/util/List;Ljava/util/Map;Ljava/util/Map;Ljava/lang/Integer;ILjava/lang/Object;)Lcom/posthog/internal/LocalEvaluationResponse; public fun equals (Ljava/lang/Object;)Z public final fun getCohorts ()Ljava/util/Map; public final fun getFlags ()Ljava/util/List; public final fun getGroupTypeMapping ()Ljava/util/Map; + public final fun getPropertyMatchingVersion ()Ljava/lang/Integer; public fun hashCode ()I public fun toString ()Ljava/lang/String; } diff --git a/posthog/src/main/java/com/posthog/internal/PostHogLocalEvaluationModels.kt b/posthog/src/main/java/com/posthog/internal/PostHogLocalEvaluationModels.kt index 0e129a144..719cd47e1 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogLocalEvaluationModels.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogLocalEvaluationModels.kt @@ -7,12 +7,16 @@ import com.posthog.PostHogInternal * Response from /api/feature_flag/local_evaluation/ */ @PostHogInternal -public data class LocalEvaluationResponse( - val flags: List?, - @SerializedName("group_type_mapping") - val groupTypeMapping: Map?, - val cohorts: Map?, -) +public data class LocalEvaluationResponse + @JvmOverloads + constructor( + val flags: List?, + @SerializedName("group_type_mapping") + val groupTypeMapping: Map?, + val cohorts: Map?, + @SerializedName("property_matching_version") + val propertyMatchingVersion: Int? = null, + ) /** * Complete feature flag definition for local evaluation From c6da08f6c85691016a791d0302d6f01a0d4e93d6 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 17:14:42 +0200 Subject: [PATCH 2/2] fix(server): retain remote response metadata across definition refreshes --- .changeset/quiet-booleans-match.md | 2 +- .../internal/PostHogFeatureFlagCache.kt | 3 +- .../server/internal/PostHogFeatureFlags.kt | 106 ++++++++---------- .../internal/VersionedPropertyMatchingTest.kt | 88 +++++++++++++++ 4 files changed, 138 insertions(+), 61 deletions(-) diff --git a/.changeset/quiet-booleans-match.md b/.changeset/quiet-booleans-match.md index e9fb81bb5..e6524c1a3 100644 --- a/.changeset/quiet-booleans-match.md +++ b/.changeset/quiet-booleans-match.md @@ -3,4 +3,4 @@ "posthog-server": patch --- -Retain `property_matching_version` in local-evaluation definitions and shared caches. Server-side local evaluation now uses explicit boolean matching for version 2, preserves legacy matching for missing/1, and keeps one definition snapshot through group, cohort, and dependency evaluation and version-only refreshes. +Retain `property_matching_version` in local-evaluation definitions and shared caches. Server-side local evaluation now uses explicit boolean matching for version 2, preserves legacy matching for missing/1, and keeps one definition snapshot through group, cohort, and dependency evaluation and version-only refreshes. In-flight remote evaluations retain their response errors and request metadata when definitions refresh, without repopulating the new result cache. diff --git a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlagCache.kt b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlagCache.kt index c0d38b28b..562d0911d 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlagCache.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlagCache.kt @@ -53,7 +53,7 @@ internal class PostHogFeatureFlagCache( requestId: String? = null, evaluatedAt: Long? = null, error: String? = null, - ) { + ): FeatureFlagCacheEntry { val currentTime = System.currentTimeMillis() val entry = FeatureFlagCacheEntry( @@ -66,6 +66,7 @@ internal class PostHogFeatureFlagCache( ) cache[key] = entry + return entry } /** diff --git a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt index d3df06a9d..6a5c75e63 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/PostHogFeatureFlags.kt @@ -227,7 +227,7 @@ internal class PostHogFeatureFlags( groups, personProperties, groupProperties, - )?.get(key) + ).flags?.get(key) } private fun getFeatureFlagsFromCache( @@ -399,7 +399,7 @@ internal class PostHogFeatureFlags( disableGeoip: Boolean = false, bypassCache: Boolean = false, onResponse: ((PostHogFlagsResponse) -> Unit)? = null, - ): Map? { + ): FeatureFlagCacheEntry { // A refresh replaces the cache; an in-flight response may only populate its original cache. val responseCache = cache val cacheKey = @@ -413,9 +413,9 @@ internal class PostHogFeatureFlags( ) if (!bypassCache) { - val cachedFlags = responseCache.get(cacheKey) - if (cachedFlags != null) { - return cachedFlags + val cached = responseCache.getEntry(cacheKey) + if (cached?.flags != null) { + return cached } } @@ -432,39 +432,35 @@ internal class PostHogFeatureFlags( flagKeys = flagKeys, disableGeoip = disableGeoip, ) - val flags = response?.flags - responseCache.put( - cacheKey, - flags, - response?.requestId, - response?.evaluatedAt, - computeResponseError(response), - ) + val entry = + responseCache.put( + cacheKey, + response?.flags, + response?.requestId, + response?.evaluatedAt, + computeResponseError(response), + ) if (response != null) { reconcileReturnedFlagEvidence(responseGeneration, response) onResponse?.invoke(response) } - flags + // Return this request's envelope even if a refresh replaced the cache while it was in flight. + entry } catch (e: SocketTimeoutException) { config.logger.log("Loading remote feature flags timed out: $e") responseCache.put(cacheKey, null, error = FeatureFlagError.TIMEOUT) - null } catch (e: ConnectException) { config.logger.log("Loading remote feature flags connection failed: $e") responseCache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) - null } catch (e: UnknownHostException) { config.logger.log("Loading remote feature flags DNS lookup failed: $e") responseCache.put(cacheKey, null, error = FeatureFlagError.CONNECTION_ERROR) - null } catch (e: PostHogApiError) { config.logger.log("Loading remote feature flags API error: $e") responseCache.put(cacheKey, null, error = FeatureFlagError.apiError(e.statusCode)) - null } catch (e: Throwable) { config.logger.log("Loading remote feature flags failed: $e") responseCache.put(cacheKey, null, error = FeatureFlagError.UNKNOWN_ERROR) - null } } @@ -512,7 +508,7 @@ internal class PostHogFeatureFlags( } // Finally, fall back to remote fetch - return getFeatureFlagsFromRemote(distinctId, groups, personProperties, groupProperties) + return getFeatureFlagsFromRemote(distinctId, groups, personProperties, groupProperties).flags } override fun clear() { @@ -1093,7 +1089,7 @@ internal class PostHogFeatureFlags( return EMPTY_EVALUATE_FLAGS_RESULT } - val (remoteFlags, entry) = + val entry = if (missingDefinitionKeys.isNotEmpty()) { evaluateMissingFlagsRemotely( cacheKey, @@ -1107,21 +1103,15 @@ internal class PostHogFeatureFlags( local?.needsRemote == true, ) } else { - var cached = cache.getEntry(cacheKey) - val flags = - if (cached != null) { - cached.flags - } else { - getFeatureFlagsFromRemote( - distinctId, - groups, - personProperties, - groupProperties, - flagKeys, - disableGeoip, - ).also { cached = cache.getEntry(cacheKey) } - } - flags to cached + cache.getEntry(cacheKey) + ?: getFeatureFlagsFromRemote( + distinctId, + groups, + personProperties, + groupProperties, + flagKeys, + disableGeoip, + ) } // Forward the caller's original scope to `/flags`; locally resolved values win below. @@ -1129,7 +1119,7 @@ internal class PostHogFeatureFlags( // Same precedence as posthog-python, which skips remote keys already in // `locally_evaluated_keys`. Note a group flag evaluated without `groups` resolves locally to // `false`, and that now beats the server's answer — pass `groups` when gating on one. - val merged = LinkedHashMap(remoteFlags ?: EMPTY_FLAGS).apply { putAll(localFlags) } + val merged = LinkedHashMap(entry?.flags ?: EMPTY_FLAGS).apply { putAll(localFlags) } return EvaluateFlagsResult( flags = merged, locallyEvaluated = merged.mapValues { it.key in localFlags }, @@ -1150,44 +1140,42 @@ internal class PostHogFeatureFlags( disableGeoip: Boolean, missingDefinitionKeys: Set, localNeedsRemote: Boolean, - ): Pair?, FeatureFlagCacheEntry?> { + ): FeatureFlagCacheEntry? { while (true) { val plan = planMissingFlagProbe(missingDefinitionKeys) if (plan.waiting.isNotEmpty()) { - if (!plan.waiting.all { it.await(missingFlagProbeWaitTimeoutMs) }) return null to null + if (!plan.waiting.all { it.await(missingFlagProbeWaitTimeoutMs) }) return null continue } val refreshMadeKeysLocal = plan.currentlyMissing.size < missingDefinitionKeys.size val needsRemote = localNeedsRemote || refreshMadeKeysLocal || plan.knownRemote.isNotEmpty() || plan.owned.isNotEmpty() - if (!needsRemote) return null to null + if (!needsRemote) return null var entry = cache.getEntry(cacheKey) val bypassCache = shouldBypassCacheFor(plan, refreshMadeKeysLocal, entry) if (bypassCache) entry = null var response: PostHogFlagsResponse? = null - val flags = - try { - if (entry != null) { - entry.flags - } else { - getFeatureFlagsFromRemote( - distinctId, - groups, - personProperties, - groupProperties, - flagKeys, - disableGeoip, - bypassCache = bypassCache, - onResponse = { response = it }, - ).also { entry = cache.getEntry(cacheKey) } - } - } finally { - completeMissingFlagProbe(plan, response) + return try { + if (entry != null) { + entry + } else { + getFeatureFlagsFromRemote( + distinctId, + groups, + personProperties, + groupProperties, + flagKeys, + disableGeoip, + bypassCache = bypassCache, + onResponse = { response = it }, + ) } - return flags to entry + } finally { + completeMissingFlagProbe(plan, response) + } } } diff --git a/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt b/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt index de142f218..05e7bcc10 100644 --- a/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt +++ b/posthog-server/src/test/java/com/posthog/server/internal/VersionedPropertyMatchingTest.kt @@ -7,6 +7,7 @@ import com.posthog.server.CountingDispatcher import com.posthog.server.PostHogBlockingFlagDefinitionCacheProvider import com.posthog.server.PostHogFlagDefinitionCacheProvider import com.posthog.server.createFlagsResponse +import com.posthog.server.createFlagsResponseWithErrors import com.posthog.server.createTestConfig import com.posthog.server.jsonResponse import com.posthog.server.shutdownAndAwaitTermination @@ -252,6 +253,93 @@ internal class VersionedPropertyMatchingTest { } } + @Test + fun `delayed missing flag response retains its envelope across refresh`() { + assertDelayedRemoteEnvelope("remote", fails = false) + } + + @Test + fun `delayed inconclusive flag response retains its envelope across refresh`() { + assertDelayedRemoteEnvelope("person", fails = false) + } + + @Test + fun `delayed missing flag failure retains its error across refresh`() { + assertDelayedRemoteEnvelope("remote", fails = true) + } + + @Test + fun `delayed inconclusive flag failure retains its error across refresh`() { + assertDelayedRemoteEnvelope("person", fails = true) + } + + private fun assertDelayedRemoteEnvelope( + key: String, + fails: Boolean, + ) { + val responseStarted = CountDownLatch(1) + val releaseResponse = CountDownLatch(1) + val version = AtomicInteger(1) + val requests = AtomicInteger() + val dispatcher = + CountingDispatcher( + { jsonResponse(definitions(version.get())) }, + { + val first = requests.incrementAndGet() == 1 + if (first) { + responseStarted.countDown() + check(releaseResponse.await(5, TimeUnit.SECONDS)) + } + if (first && fails) { + MockResponse().setResponseCode(500) + } else { + val requestId = if (first) "old-request" else "new-request" + jsonResponse( + createFlagsResponseWithErrors(key) + .replaceFirst("{", """{"requestId":"$requestId","evaluatedAt":123,"""), + ) + } + }, + ) + val http = MockWebServer() + http.dispatcher = dispatcher + http.start() + val sut = createSut(createTestConfig(host = http.url("/").toString())) + val executor = Executors.newSingleThreadExecutor() + try { + sut.loadFeatureFlagDefinitions() + val pending = + executor.submit { + // No person properties makes a known definition inconclusive. + sut.evaluateFlags("user", null, null, null, listOf(key), false, false) + } + assertTrue(responseStarted.await(5, TimeUnit.SECONDS)) + version.set(2) + sut.loadFeatureFlagDefinitions() + releaseResponse.countDown() + val result = pending.get(5, TimeUnit.SECONDS) + assertEquals(if (fails) null else true, result.flags[key]?.enabled) + assertEquals(if (fails) "api_error_500" else "errors_while_computing_flags", result.responseError) + assertEquals(if (fails) null else "old-request", result.requestId) + assertEquals(if (fails) null else 123L, result.evaluatedAt) + + val fresh = sut.evaluateFlags("user", null, null, null, listOf(key), false, false) + assertEquals(true, fresh.flags[key]?.enabled) + assertEquals("new-request", fresh.requestId) + assertEquals("errors_while_computing_flags", fresh.responseError) + assertEquals(123L, fresh.evaluatedAt) + assertEquals(fresh, sut.evaluateFlags("user", null, null, null, listOf(key), false, false)) + assertEquals(2, dispatcher.localEvaluationCalls.get()) + assertEquals(2, dispatcher.flagsCalls.get(), "The old response must not populate the refreshed cache") + } finally { + releaseResponse.countDown() + executor.shutdownAndAwaitTermination() + sut.clear() + sut.shutDown() + http.shutdown() + } + } + @Test fun `disk cache round trip retains version and older entries reset to legacy`() { val http = MockWebServer()