From 858db386559b891f9de1c09479f959cb24a7eb47 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 14:44:00 +0200 Subject: [PATCH 1/7] fix(flags): honor versioned local property matching --- .../changesets/versioned-property-matching.md | 5 + posthog/client.py | 129 ++++-- posthog/feature_flags.py | 30 +- posthog/flag_definition_cache.py | 3 + posthog/test/test_flag_definition_cache.py | 6 +- .../test/test_property_matching_version.py | 391 ++++++++++++++++++ references/public_api_snapshot.txt | 13 +- 7 files changed, 524 insertions(+), 53 deletions(-) create mode 100644 .sampo/changesets/versioned-property-matching.md create mode 100644 posthog/test/test_property_matching_version.py diff --git a/.sampo/changesets/versioned-property-matching.md b/.sampo/changesets/versioned-property-matching.md new file mode 100644 index 000000000..e5ce0057f --- /dev/null +++ b/.sampo/changesets/versioned-property-matching.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Honor the definitions snapshot's `property_matching_version` during local feature flag evaluation, including person, group, cohort, and flag dependency conditions. Version 2 uses explicit boolean equality; missing/1 retains legacy truthiness. Preserve the selector through definition caches and invalidate evaluated results on version-only refreshes. diff --git a/posthog/client.py b/posthog/client.py index 59ccb9dca..390c32fc5 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -130,6 +130,11 @@ _atexit_deadline_lock = threading.Lock() +class _LocalEvaluationSnapshot(FlagDefinitionCacheData): + flags_by_key: Dict[str, Any] + flag_definition_version: int + + def _supports_lane_synchronization(queue) -> bool: return all( hasattr(queue, attribute) @@ -879,6 +884,7 @@ def __init__( self.feature_flags_by_key: Optional[dict[str, Any]] = None self.group_type_mapping: Optional[dict[str, str]] = None self.cohorts: Optional[dict[str, Any]] = None + self._property_matching_version = 1 self.poll_interval = poll_interval self.feature_flags_request_timeout_seconds = ( feature_flags_request_timeout_seconds @@ -895,7 +901,7 @@ def __init__( self._flag_definition_fetch_generation = 0 self._flag_definition_published_generation = 0 self._flag_definition_cache_generation = 0 - self._flag_definition_publication_lock = threading.Lock() + self._flag_definition_publication_lock = threading.RLock() self._flag_definition_cache_write_lock = threading.RLock() self._flag_definition_cache_provider = flag_definition_cache_provider self._flag_definition_cache_provider_async_runner: Optional[ @@ -2234,7 +2240,7 @@ def _reinit_after_fork(self): # A parent thread may have been publishing or caching flag definitions at # fork time. - self._flag_definition_publication_lock = threading.Lock() + self._flag_definition_publication_lock = threading.RLock() self._flag_definition_cache_write_lock = threading.RLock() # Metrics locks may have been held by a parent thread at fork time; replace @@ -2872,24 +2878,40 @@ def _update_flag_state( self, data: FlagDefinitionCacheData, old_flags_by_key: Optional[dict] = None ) -> None: """Update internal flag state from cache data and invalidate evaluation cache if changed.""" - self.feature_flags = data["flags"] - self.group_type_mapping = data["group_type_mapping"] - self.cohorts = data["cohorts"] - # Server-controlled gate for minimal $feature_flag_called events; the - # local-evaluation payload carries it as a top-level key. Absent means False. - self._minimal_flag_called_events = ( - data.get("minimal_flag_called_events") is True - ) + with self._flag_definition_publication_lock: + old_matching_version = self._property_matching_version + self.feature_flags = data["flags"] + self.group_type_mapping = data["group_type_mapping"] + self.cohorts = data["cohorts"] + self._property_matching_version = data.get("property_matching_version", 1) + # Absent server-controlled metadata resets to its legacy default. + self._minimal_flag_called_events = ( + data.get("minimal_flag_called_events") is True + ) - # Invalidate evaluation cache if flag definitions changed - if ( - self.flag_cache - and old_flags_by_key is not None - and old_flags_by_key != (self.feature_flags_by_key or {}) - ): - old_version = self.flag_definition_version - self.flag_definition_version += 1 - self.flag_cache.invalidate_version(old_version) + if self.flag_cache and ( + old_matching_version != self._property_matching_version + or ( + old_flags_by_key is not None + and old_flags_by_key != (self.feature_flags_by_key or {}) + ) + ): + old_version = self.flag_definition_version + self.flag_definition_version += 1 + self.flag_cache.invalidate_version(old_version) + + def _local_evaluation_snapshot(self) -> _LocalEvaluationSnapshot: + # Capture references together. Publication replaces these collections, so + # recursive and multi-flag evaluations can finish on their original rules. + with self._flag_definition_publication_lock: + return { + "flags": self.feature_flags or [], + "flags_by_key": self.feature_flags_by_key or {}, + "group_type_mapping": self.group_type_mapping or {}, + "cohorts": self.cohorts or {}, + "property_matching_version": self._property_matching_version, + "flag_definition_version": self.flag_definition_version, + } def _load_feature_flags(self): should_fetch = True @@ -3006,6 +3028,7 @@ def _fetch_feature_flags_from_api(self): "group_type_mapping": self.group_type_mapping or {}, "cohorts": self.cohorts or {}, "minimal_flag_called_events": self._minimal_flag_called_events, + "property_matching_version": self._property_matching_version, } # Publish the ETag only after its matching flag state is installed. @@ -3049,6 +3072,7 @@ def _fetch_feature_flags_from_api(self): self.feature_flags = [] self.group_type_mapping = {} self.cohorts = {} + self._property_matching_version = 1 self._flags_etag = None self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation @@ -3066,6 +3090,7 @@ def _fetch_feature_flags_from_api(self): self.feature_flags = [] self.group_type_mapping = {} self.cohorts = {} + self._property_matching_version = 1 self._flags_etag = None self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation @@ -3103,14 +3128,18 @@ def load_feature_flags(self): Feature flags """ if self.disabled: - self.feature_flags = [] + with self._flag_definition_publication_lock: + self.feature_flags = [] + self._property_matching_version = 1 return if not self.personal_api_key: self.log.warning( "[FEATURE FLAGS] You have to specify a secret_key to use feature flags." ) - self.feature_flags = [] + with self._flag_definition_publication_lock: + self.feature_flags = [] + self._property_matching_version = 1 return self._load_feature_flags() @@ -3135,7 +3164,15 @@ def _compute_flag_locally( group_properties=None, warn_on_unknown_groups=True, device_id=None, + _definition_snapshot: Optional[_LocalEvaluationSnapshot] = None, ) -> FlagValue: + snapshot = ( + _definition_snapshot + if _definition_snapshot is not None + else self._local_evaluation_snapshot() + ) + flags_by_key = snapshot["flags_by_key"] + property_matching_version = snapshot.get("property_matching_version", 1) groups = groups or {} person_properties = person_properties or {} group_properties = group_properties or {} @@ -3151,7 +3188,7 @@ def _compute_flag_locally( flag_filters = feature_flag.get("filters") or {} aggregation_group_type_index = flag_filters.get("aggregation_group_type_index") - group_type_mapping = self.group_type_mapping or {} + group_type_mapping = snapshot["group_type_mapping"] if aggregation_group_type_index is not None: group_name = group_type_mapping.get(str(aggregation_group_type_index)) @@ -3186,8 +3223,9 @@ def _compute_flag_locally( feature_flag, group_key, focused_group_properties, - cohort_properties=self.cohorts, - flags_by_key=self.feature_flags_by_key, + cohort_properties=snapshot["cohorts"], + flags_by_key=flags_by_key, + property_matching_version=property_matching_version, evaluation_cache=evaluation_cache, device_id=device_id, bucketing_value=group_key, @@ -3203,8 +3241,9 @@ def _compute_flag_locally( feature_flag, distinct_id, person_properties, - cohort_properties=self.cohorts, - flags_by_key=self.feature_flags_by_key, + cohort_properties=snapshot["cohorts"], + flags_by_key=flags_by_key, + property_matching_version=property_matching_version, evaluation_cache=evaluation_cache, device_id=device_id, bucketing_value=bucketing_value, @@ -3335,7 +3374,7 @@ def _get_feature_flag_result( local_person_properties = self._person_properties_for_local_evaluation( distinct_id, person_properties ) - flag_value = self._locally_evaluate_flag( + flag_value, local_definition_version = self._locally_evaluate_flag( key, distinct_id, groups, @@ -3364,10 +3403,15 @@ def _get_feature_flag_result( cached_flag_result = FeatureFlagResult.from_value_and_payload( key, flag_value, self._compute_payload_locally(key, flag_value) ) - if self.flag_cache and cached_flag_result: - self.flag_cache.set_cached_flag( - distinct_id, key, cached_flag_result, self.flag_definition_version - ) + with self._flag_definition_publication_lock: + if ( + self.flag_cache + and cached_flag_result + and local_definition_version == self.flag_definition_version + ): + self.flag_cache.set_cached_flag( + distinct_id, key, cached_flag_result, local_definition_version + ) elif only_evaluate_locally: if self.feature_flags is None: self.log.warning( @@ -3592,17 +3636,15 @@ def _locally_evaluate_flag( person_properties: dict[str, str], group_properties: dict[str, dict[str, Any]], device_id: Optional[str] = None, - ) -> Optional[FlagValue]: + ) -> tuple[Optional[FlagValue], int]: + """Return the local value and the generation of the evaluated snapshot.""" if self.feature_flags is None and self.personal_api_key: self.load_feature_flags() response = None - if self.feature_flags: - assert self.feature_flags_by_key is not None, ( - "feature_flags_by_key should be initialized when feature_flags is set" - ) - # Local evaluation - flag = self.feature_flags_by_key.get(key) + snapshot = self._local_evaluation_snapshot() + if snapshot["flags"]: + flag = snapshot["flags_by_key"].get(key) if flag: try: response = self._compute_flag_locally( @@ -3612,6 +3654,7 @@ def _locally_evaluate_flag( person_properties=person_properties, group_properties=group_properties, device_id=device_id, + _definition_snapshot=snapshot, ) self.log.debug( f"Successfully computed flag locally: {key} -> {response}" @@ -3622,7 +3665,7 @@ def _locally_evaluate_flag( self.log.exception( f"[FEATURE FLAGS] Error while computing variant locally: {e}" ) - return response + return response, snapshot["flag_definition_version"] def get_feature_flag_payload( self, @@ -4317,14 +4360,15 @@ def _get_all_flags_and_payloads_locally( flags: dict[str, FlagValue] = {} payloads: dict[str, str] = {} fallback_to_flags = False + snapshot = self._local_evaluation_snapshot() # If loading in previous line failed - if self.feature_flags: + if snapshot["flags"]: # Filter flags based on flag_keys_to_evaluate if provided - flags_to_process = self.feature_flags + flags_to_process = snapshot["flags"] if flag_keys_to_evaluate: flag_keys_set = set(flag_keys_to_evaluate) flags_to_process = [ - flag for flag in self.feature_flags if flag["key"] in flag_keys_set + flag for flag in snapshot["flags"] if flag["key"] in flag_keys_set ] for flag in flags_to_process: @@ -4337,6 +4381,7 @@ def _get_all_flags_and_payloads_locally( group_properties=group_properties, warn_on_unknown_groups=warn_on_unknown_groups, device_id=device_id, + _definition_snapshot=snapshot, ) matched_payload = self._compute_payload_locally( flag["key"], flags[flag["key"]] diff --git a/posthog/feature_flags.py b/posthog/feature_flags.py index 808053e32..cb4daebb1 100644 --- a/posthog/feature_flags.py +++ b/posthog/feature_flags.py @@ -150,6 +150,7 @@ def evaluate_flag_dependency( properties, cohort_properties, device_id=None, + property_matching_version: int = 1, ): """ Evaluate a flag dependency condition under local evaluation. @@ -260,6 +261,7 @@ def evaluate_flag_dependency( flags_by_key=flags_by_key, evaluation_cache=evaluation_cache, device_id=device_id, + property_matching_version=property_matching_version, bucketing_value=dep_bucketing_value, ) evaluation_cache[dep_flag_key] = dep_result @@ -350,6 +352,7 @@ def match_feature_flag_properties( group_type_mapping=None, groups=None, group_properties=None, + property_matching_version: int = 1, ) -> FlagValue: if bucketing_value is None: warnings.warn( @@ -416,6 +419,7 @@ def match_feature_flag_properties( evaluation_cache, bucketing_value=effective_bucketing, device_id=device_id, + property_matching_version=property_matching_version, ) if match_result == ConditionMatch.MATCH: variant_override = condition.get("variant") @@ -463,6 +467,7 @@ def is_condition_match( *, bucketing_value, device_id=None, + property_matching_version: int = 1, ) -> ConditionMatch: rollout_percentage = condition.get("rollout_percentage") if len(condition.get("properties") or []) > 0: @@ -477,6 +482,7 @@ def is_condition_match( evaluation_cache, distinct_id, device_id=device_id, + property_matching_version=property_matching_version, ) elif property_type == "flag": matches = evaluate_flag_dependency( @@ -487,9 +493,10 @@ def is_condition_match( properties, cohort_properties, device_id=device_id, + property_matching_version=property_matching_version, ) else: - matches = match_property(prop, properties) + matches = match_property(prop, properties, property_matching_version) if not matches: return ConditionMatch.NO_MATCH @@ -593,7 +600,9 @@ def _is_truthy_property_value(value) -> bool: return False -def match_property(property, property_values) -> bool: +def match_property( + property, property_values, property_matching_version: int = 1 +) -> bool: # only looks for matches where key exists in override_property_values key = property.get("key") operator = property.get("operator") or "exact" @@ -623,7 +632,12 @@ def match_property(property, property_values) -> bool: def compute_exact_match(value, override_value): override_string = _value_to_string(override_value).lower() - if _is_truthy_or_falsy_property_value(value): + # Empty filters retain recursive truthiness in both matching modes. + if value == []: + return _is_truthy_property_value(override_value) + if property_matching_version != 2 and _is_truthy_or_falsy_property_value( + value + ): return _is_truthy_property_value(value) == _is_truthy_property_value( override_value ) @@ -814,6 +828,7 @@ def match_cohort( evaluation_cache=None, distinct_id=None, device_id=None, + property_matching_version: int = 1, ) -> bool: # Cohort properties are in the form of property groups like this: # { @@ -839,6 +854,7 @@ def match_cohort( evaluation_cache, distinct_id, device_id=device_id, + property_matching_version=property_matching_version, ) operator = property.get("operator") or "exact" @@ -857,6 +873,7 @@ def match_property_group( evaluation_cache=None, distinct_id=None, device_id=None, + property_matching_version: int = 1, ) -> bool: # The backend serializes its canonical empty PropertyGroup as {}. if property_group == {}: @@ -893,6 +910,7 @@ def match_property_group( evaluation_cache, distinct_id, device_id=device_id, + property_matching_version=property_matching_version, ) negation = False elif prop.get("type") == "cohort": @@ -904,6 +922,7 @@ def match_property_group( evaluation_cache, distinct_id, device_id=device_id, + property_matching_version=property_matching_version, ) negation = prop.get("negation", False) elif prop.get("type") == "flag": @@ -915,10 +934,13 @@ def match_property_group( property_values, cohort_properties, device_id=device_id, + property_matching_version=property_matching_version, ) negation = prop.get("negation", False) else: - matches = match_property(prop, property_values) + matches = match_property( + prop, property_values, property_matching_version + ) negation = prop.get("negation", False) effective_match = matches != bool(negation) diff --git a/posthog/flag_definition_cache.py b/posthog/flag_definition_cache.py index e1e6325c7..e87cfaa7f 100644 --- a/posthog/flag_definition_cache.py +++ b/posthog/flag_definition_cache.py @@ -40,6 +40,8 @@ class FlagDefinitionCacheData(TypedDict): flags: List of feature flag definition dictionaries from the API. group_type_mapping: Mapping of group type indices to group names. cohorts: Dictionary of cohort definitions for local evaluation. + property_matching_version: Exact/is_not matching selector. Missing means + legacy; only version 2 enables explicit matching. minimal_flag_called_events: Server-controlled gate for minimal ``$feature_flag_called`` events. Treated as False when absent. """ @@ -48,6 +50,7 @@ class FlagDefinitionCacheData(TypedDict): group_type_mapping: Required[Dict[str, str]] cohorts: Required[Dict[str, Any]] minimal_flag_called_events: NotRequired[bool] + property_matching_version: NotRequired[int] @runtime_checkable diff --git a/posthog/test/test_flag_definition_cache.py b/posthog/test/test_flag_definition_cache.py index 149ee332e..b66ee7c0f 100644 --- a/posthog/test/test_flag_definition_cache.py +++ b/posthog/test/test_flag_definition_cache.py @@ -462,7 +462,11 @@ def test_awaits_async_provider_when_fetching_from_api(self, mock_get): # the flag definitions so it survives cache round-trips. self.assertEqual( self.cache_provider.stored_data, - {**self.sample_flags_data, "minimal_flag_called_events": False}, + { + **self.sample_flags_data, + "minimal_flag_called_events": False, + "property_matching_version": 1, + }, ) self.assertEqual(len(set(self.cache_provider.loop_ids)), 1) diff --git a/posthog/test/test_property_matching_version.py b/posthog/test/test_property_matching_version.py new file mode 100644 index 000000000..fd1747cf6 --- /dev/null +++ b/posthog/test/test_property_matching_version.py @@ -0,0 +1,391 @@ +"""Versioned property matching and definitions snapshot regression tests.""" + +from copy import deepcopy +from unittest import mock + +import pytest + +from posthog.client import Client +from posthog.feature_flags import InconclusiveMatchError, match_property +from posthog.request import APIError, GetResponse +from posthog.test.test_flag_definition_cache import ( + AsyncMockCacheProvider, + MockCacheProvider, +) +from posthog.test.test_utils import FAKE_TEST_API_KEY + + +ROWS = [ + (False, "banana", True, False), + (False, 0, True, False), + (["true", "false"], "true", False, True), + (["true", "false"], "pro", True, False), + ([], True, True, True), + ([], [], True, True), + (True, [True], True, False), + (False, "FALSE", True, True), + (False, None, True, False), + (False, "", True, False), + ([], [True, ["TRUE", []]], True, True), + ([], [True, [False]], False, False), + ([], False, False, False), + ([], None, False, False), + ([], 0, False, False), + ([], 1, False, False), + ([], "banana", False, False), + ([False, "PRO"], "pro", True, True), + ([[True], "PRO"], [True], True, True), + ([None, "PRO"], "null", True, True), + ([1, "PRO"], "1", True, True), + ("οδος", "ΟΔΟΣ", True, True), + ("i\u0307", "İ", True, True), + ("ss", "ß", False, False), +] + + +@pytest.mark.parametrize("filter_value,property_value,legacy,explicit", ROWS) +@pytest.mark.parametrize("version", [None, 1, 2, 0, 3, "2"]) +@pytest.mark.parametrize("operator", ["exact", "is_not"]) +def test_match_property_version_rows( + filter_value, property_value, legacy, explicit, version, operator +): + kwargs = {} if version is None else {"property_matching_version": version} + expected = explicit if version == 2 else legacy + assert match_property( + {"key": "value", "value": filter_value, "operator": operator}, + {"value": property_value}, + **kwargs, + ) is (expected if operator == "exact" else not expected) + + +@pytest.mark.parametrize("version", [1, 2]) +@pytest.mark.parametrize("operator", ["exact", "is_not"]) +def test_match_property_version_missing_is_inconclusive(version, operator): + with pytest.raises(InconclusiveMatchError): + match_property( + {"key": "value", "value": False, "operator": operator}, + {}, + property_matching_version=version, + ) + + +def definitions(version=None): + prop = {"key": "value", "value": False, "operator": "exact"} + person = { + "key": "person", + "active": True, + "version": 2, # Individual flag versions must not select matching semantics. + "filters": {"groups": [{"properties": [prop]}]}, + } + group = deepcopy(person) + group["key"] = "group" + group["filters"]["aggregation_group_type_index"] = 0 + mixed = deepcopy(person) + mixed["key"] = "mixed" + mixed["filters"]["groups"][0]["aggregation_group_type_index"] = 0 + cohort = deepcopy(person) + cohort["key"] = "cohort" + cohort["filters"]["groups"][0]["properties"] = [{"type": "cohort", "value": 1}] + dependency = deepcopy(person) + dependency["key"] = "dependency" + dependency["filters"]["groups"][0]["properties"] = [ + { + "type": "flag", + "key": "person", + "value": True, + "operator": "flag_evaluates_to", + "dependency_chain": ["person"], + } + ] + data = { + "flags": [person, group, mixed, cohort, dependency], + "group_type_mapping": {"0": "company"}, + "cohorts": { + "1": { + "type": "AND", + "values": [{"type": "OR", "values": [{"type": "cohort", "value": 2}]}], + }, + "2": {"type": "AND", "values": [prop]}, + }, + } + if version is not None: + data["property_matching_version"] = version + return data + + +@pytest.fixture +def client(): + client = Client( + FAKE_TEST_API_KEY, + secret_key="test-secret", + enable_local_evaluation=False, + send=False, + flag_fallback_cache_url="memory://local/?ttl=300&size=100", + ) + with mock.patch( + "posthog.client.flags", side_effect=AssertionError("remote fallback") + ): + yield client + client.shutdown() + + +def evaluate(client, key="person"): + return client.get_feature_flag( + key, + "user", + person_properties={"value": "banana"}, + groups={"company": "company-id"}, + group_properties={"company": {"value": "banana"}}, + only_evaluate_locally=True, + send_feature_flag_events=False, + ) + + +@pytest.mark.parametrize( + "version,expected", [(None, True), (1, True), (2, False), (3, True)] +) +def test_client_version_person_group_cohort_dependency_and_full_api( + client, version, expected +): + client._update_flag_state(definitions(version)) + for key in ("person", "group", "mixed", "cohort", "dependency"): + assert evaluate(client, key) is expected + results = client.evaluate_flags( + "user", + person_properties={"value": "banana"}, + groups={"company": "company-id"}, + group_properties={"company": {"value": "banana"}}, + only_evaluate_locally=True, + ) + for key in ("person", "group", "mixed", "cohort", "dependency"): + assert results.get_flag(key) is expected + + +@pytest.mark.parametrize("provider_class", [MockCacheProvider, AsyncMockCacheProvider]) +def test_version_only_reload_invalidates_results_and_round_trips_provider( + client, provider_class +): + provider = provider_class() + client._flag_definition_cache_provider = provider + with mock.patch("posthog.client.get") as get: + for version, expected in [ + (1, True), + (2, False), + (1, True), + (2, False), + (None, True), + ]: + previous_generation = client.flag_definition_version + get.return_value = GetResponse( + data=definitions(version), etag=str(version), not_modified=False + ) + with mock.patch.object( + client.flag_cache, + "invalidate_version", + wraps=client.flag_cache.invalidate_version, + ) as invalidate: + client._load_feature_flags() + invalidate.assert_called_once_with(previous_generation) + assert client.flag_cache.get_stale_cached_flag("user", "person") is None + assert evaluate(client) is expected + assert ( + client.flag_cache.get_cached_flag( + "user", "person", client.flag_definition_version + ).get_value() + is expected + ) + assert provider.stored_data["property_matching_version"] == (version or 1) + + +@pytest.mark.parametrize( + "refresh", ["304", "503", "exception", "empty-provider", "failed-provider"] +) +def test_failed_or_not_modified_refresh_preserves_version(client, refresh): + with mock.patch("posthog.client.get") as get: + get.return_value = GetResponse( + data=definitions(2), etag="current", not_modified=False + ) + client._load_feature_flags() + generation = client.flag_definition_version + if refresh == "304": + get.return_value = GetResponse(data=None, etag="current", not_modified=True) + elif refresh == "503": + get.side_effect = APIError(503, "unavailable") + elif refresh == "exception": + get.side_effect = RuntimeError("offline") + else: + provider = MockCacheProvider() + client._flag_definition_cache_provider = provider + provider.should_fetch_return_value = False + if refresh == "failed-provider": + provider.get_error = RuntimeError("cache unavailable") + get.side_effect = RuntimeError("offline") + client._load_feature_flags() + assert client.flag_definition_version == generation + assert evaluate(client) is False + + +def test_in_flight_full_evaluation_keeps_matching_snapshot(client): + client._update_flag_state(definitions(1)) + from posthog import client as client_module + + original = client_module.match_feature_flag_properties + calls = 0 + + def reload_during_evaluation(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + client._update_flag_state(definitions(2), client.feature_flags_by_key) + return original(*args, **kwargs) + + with mock.patch( + "posthog.client.match_feature_flag_properties", + side_effect=reload_during_evaluation, + ): + result, fallback = client._get_all_flags_and_payloads_locally( + "user", + groups={"company": "company-id"}, + person_properties={"value": "banana"}, + group_properties={"company": {"value": "banana"}}, + ) + assert not fallback + assert all(result["featureFlags"].values()) + assert evaluate(client) is False + + +@pytest.mark.parametrize("provider_class", [MockCacheProvider, AsyncMockCacheProvider]) +def test_provider_hydration_version_only_changes_and_older_entries( + client, provider_class +): + provider = provider_class() + provider.should_fetch_return_value = False + client._flag_definition_cache_provider = provider + with mock.patch( + "posthog.client.get", side_effect=AssertionError("API fetch") + ) as get: + for version, expected in [ + (1, True), + (2, False), + (1, True), + (2, False), + (None, True), + ]: + provider.stored_data = definitions(version) + generation = client.flag_definition_version + client._load_feature_flags() + assert client.flag_definition_version == generation + 1 + assert client.flag_cache.get_stale_cached_flag("user", "person") is None + for key in ("person", "group", "mixed", "cohort", "dependency"): + assert evaluate(client, key) is expected + get.assert_not_called() + + +@pytest.mark.parametrize("provider_class", [MockCacheProvider, AsyncMockCacheProvider]) +def test_provider_round_trip_to_second_worker(client, provider_class): + provider = provider_class() + client._flag_definition_cache_provider = provider + with mock.patch( + "posthog.client.get", + return_value=GetResponse(data=definitions(2), etag="v2", not_modified=False), + ): + client._load_feature_flags() + provider.should_fetch_return_value = False + reader = Client( + FAKE_TEST_API_KEY, + secret_key="reader-secret", + enable_local_evaluation=False, + send=False, + flag_definition_cache_provider=provider, + ) + try: + with mock.patch( + "posthog.client.get", side_effect=AssertionError("API fetch") + ) as get: + reader._load_feature_flags() + for key in ("person", "group", "mixed", "cohort", "dependency"): + assert evaluate(reader, key) is False + get.assert_not_called() + finally: + reader.shutdown() + + +@pytest.mark.parametrize("status", [401, 402]) +def test_clearing_definitions_resets_matching_version(client, status): + client._update_flag_state(definitions(2)) + assert evaluate(client) is False + with mock.patch("posthog.client.get", side_effect=APIError(status, "reset")): + client._load_feature_flags() + assert client._local_evaluation_snapshot()["property_matching_version"] == 1 + assert not client.feature_flags + assert client.flag_cache.get_stale_cached_flag("user", "person") is None + client._update_flag_state(definitions()) + assert evaluate(client) is True + + +@pytest.mark.parametrize("version,expected", [(None, True), (1, True), (2, False)]) +@pytest.mark.parametrize( + "provider_class", [None, MockCacheProvider, AsyncMockCacheProvider] +) +def test_lazy_loading_caches_first_local_result_for_outage_fallback( + client, version, expected, provider_class +): + assert client.feature_flags is None + assert client.flag_definition_version == 0 + if provider_class is not None: + provider = provider_class() + provider.should_fetch_return_value = False + provider.stored_data = definitions(version) + client._flag_definition_cache_provider = provider + + with mock.patch( + "posthog.client.get", + return_value=GetResponse( + data=definitions(version), etag="initial", not_modified=False + ), + ) as get: + assert evaluate(client) is expected + if provider_class is None: + get.assert_called_once() + else: + get.assert_not_called() + assert provider.get_call_count == 1 + + assert client.flag_definition_version == 1 + cached = client.flag_cache.get_cached_flag( + "user", "person", client.flag_definition_version + ) + assert cached is not None + assert cached.get_value() is expected + + # Missing properties make local evaluation inconclusive; an API outage must + # still be able to fall back to the first successful evaluation. + with mock.patch.object( + client, + "_get_feature_flag_details_from_server", + side_effect=APIError(503, "offline"), + ) as remote: + result = client.get_feature_flag_result( + "person", "user", send_feature_flag_events=False + ) + remote.assert_called_once() + assert result is cached + + +def test_in_flight_result_is_not_cached_in_new_generation(client): + client._update_flag_state(definitions(1)) + from posthog import client as client_module + + original = client_module.match_feature_flag_properties + + def reload_during_evaluation(*args, **kwargs): + client._update_flag_state(definitions(2), client.feature_flags_by_key) + return original(*args, **kwargs) + + with mock.patch( + "posthog.client.match_feature_flag_properties", + side_effect=reload_during_evaluation, + ): + assert evaluate(client) is True + assert client.flag_cache.get_stale_cached_flag("user", "person") is None + assert evaluate(client) is False diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 06d7394ee..0fd421988 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -717,6 +717,7 @@ attribute posthog.flag_definition_cache.FlagDefinitionCacheData.cohorts: Require attribute posthog.flag_definition_cache.FlagDefinitionCacheData.flags: Required[List[Dict[str, Any]]] attribute posthog.flag_definition_cache.FlagDefinitionCacheData.group_type_mapping: Required[Dict[str, str]] attribute posthog.flag_definition_cache.FlagDefinitionCacheData.minimal_flag_called_events: NotRequired[bool] +attribute posthog.flag_definition_cache.FlagDefinitionCacheData.property_matching_version: NotRequired[int] attribute posthog.flag_definition_cache_provider = None attribute posthog.host = None attribute posthog.in_app_modules = None @@ -1144,13 +1145,13 @@ function posthog.exception_utils.try_attach_code_variables_to_frames(all_excepti function posthog.exception_utils.walk_exception_chain(exc_info) function posthog.feature_enabled(key: str, distinct_id: ID_TYPES, groups: Optional[Mapping[str, Union[str, int]]] = None, person_properties: Optional[Dict[str, Any]] = None, group_properties: Optional[Dict[str, Dict[str, Any]]] = None, only_evaluate_locally: bool = False, send_feature_flag_events: bool = True, disable_geoip: Optional[bool] = None, device_id: Optional[str] = None) -> Optional[bool] function posthog.feature_flag_definitions() -function posthog.feature_flags.evaluate_flag_dependency(property, flags_by_key, evaluation_cache, distinct_id, properties, cohort_properties, device_id=None) +function posthog.feature_flags.evaluate_flag_dependency(property, flags_by_key, evaluation_cache, distinct_id, properties, cohort_properties, device_id=None, property_matching_version: int = 1) function posthog.feature_flags.get_matching_variant(flag, bucketing_value) -function posthog.feature_flags.is_condition_match(feature_flag, distinct_id, condition, properties, cohort_properties, flags_by_key=None, evaluation_cache=None, *, bucketing_value, device_id=None) -> ConditionMatch -function posthog.feature_flags.match_cohort(property, property_values, cohort_properties, flags_by_key=None, evaluation_cache=None, distinct_id=None, device_id=None) -> bool -function posthog.feature_flags.match_feature_flag_properties(flag, distinct_id, properties, *, cohort_properties=None, flags_by_key=None, evaluation_cache=None, device_id=None, bucketing_value=None, group_type_mapping=None, groups=None, group_properties=None) -> FlagValue -function posthog.feature_flags.match_property(property, property_values) -> bool -function posthog.feature_flags.match_property_group(property_group, property_values, cohort_properties, flags_by_key=None, evaluation_cache=None, distinct_id=None, device_id=None) -> bool +function posthog.feature_flags.is_condition_match(feature_flag, distinct_id, condition, properties, cohort_properties, flags_by_key=None, evaluation_cache=None, *, bucketing_value, device_id=None, property_matching_version: int = 1) -> ConditionMatch +function posthog.feature_flags.match_cohort(property, property_values, cohort_properties, flags_by_key=None, evaluation_cache=None, distinct_id=None, device_id=None, property_matching_version: int = 1) -> bool +function posthog.feature_flags.match_feature_flag_properties(flag, distinct_id, properties, *, cohort_properties=None, flags_by_key=None, evaluation_cache=None, device_id=None, bucketing_value=None, group_type_mapping=None, groups=None, group_properties=None, property_matching_version: int = 1) -> FlagValue +function posthog.feature_flags.match_property(property, property_values, property_matching_version: int = 1) -> bool +function posthog.feature_flags.match_property_group(property_group, property_values, cohort_properties, flags_by_key=None, evaluation_cache=None, distinct_id=None, device_id=None, property_matching_version: int = 1) -> bool function posthog.feature_flags.matches_dependency_value(expected_value, actual_value) function posthog.feature_flags.parse_datetime(value: str) -> datetime.datetime function posthog.feature_flags.parse_semver(value: str) -> tuple From ef127fdbd2a378cf5bd084977b881cd8a3c0adc3 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 17:25:42 +0200 Subject: [PATCH 2/7] fix(flags): keep result cache I/O outside snapshot publication --- posthog/client.py | 35 +-- posthog/test/test_flag_cache_publication.py | 238 ++++++++++++++++++ .../test/test_property_matching_version.py | 9 +- posthog/test/test_utils.py | 1 + posthog/utils.py | 94 +++++-- 5 files changed, 332 insertions(+), 45 deletions(-) create mode 100644 posthog/test/test_flag_cache_publication.py diff --git a/posthog/client.py b/posthog/client.py index 390c32fc5..b5637a251 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -2253,6 +2253,9 @@ def _reinit_after_fork(self): # If using Memory cache, we keep it as-is to benefit from the inherited warm cache. if isinstance(self.flag_cache, RedisFlagCache): self.flag_cache = self._initialize_flag_cache(self.flag_fallback_cache_url) + if self.flag_cache: + self.flag_cache._write_lock = threading.Lock() + self.flag_cache._advance_generation(self.flag_definition_version) reset_sessions() @@ -2896,9 +2899,10 @@ def _update_flag_state( and old_flags_by_key != (self.feature_flags_by_key or {}) ) ): - old_version = self.flag_definition_version self.flag_definition_version += 1 - self.flag_cache.invalidate_version(old_version) + # Only advance an in-memory fence while publishing. Redis I/O + # must not delay snapshots, and late writes remain invalid. + self.flag_cache._advance_generation(self.flag_definition_version) def _local_evaluation_snapshot(self) -> _LocalEvaluationSnapshot: # Capture references together. Publication replaces these collections, so @@ -3078,7 +3082,10 @@ def _fetch_feature_flags_from_api(self): self._flag_definition_cache_generation = fetch_generation if self.flag_cache: - self.flag_cache.clear() + self.flag_definition_version += 1 + self.flag_cache._advance_generation( + self.flag_definition_version + ) if self.debug: raise APIError(status=401, message=detail) @@ -3095,9 +3102,12 @@ def _fetch_feature_flags_from_api(self): self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation - # Clear flag cache when quota limited + # Invalidate results without waiting for external cache I/O. if self.flag_cache: - self.flag_cache.clear() + self.flag_definition_version += 1 + self.flag_cache._advance_generation( + self.flag_definition_version + ) if self.debug: raise APIError( @@ -3403,15 +3413,12 @@ def _get_feature_flag_result( cached_flag_result = FeatureFlagResult.from_value_and_payload( key, flag_value, self._compute_payload_locally(key, flag_value) ) - with self._flag_definition_publication_lock: - if ( - self.flag_cache - and cached_flag_result - and local_definition_version == self.flag_definition_version - ): - self.flag_cache.set_cached_flag( - distinct_id, key, cached_flag_result, local_definition_version - ) + if self.flag_cache and cached_flag_result: + # The cache rejects invalidated generations, including writes + # already in flight when new definitions are published. + self.flag_cache.set_cached_flag( + distinct_id, key, cached_flag_result, local_definition_version + ) elif only_evaluate_locally: if self.feature_flags is None: self.log.warning( diff --git a/posthog/test/test_flag_cache_publication.py b/posthog/test/test_flag_cache_publication.py new file mode 100644 index 000000000..bc45269a6 --- /dev/null +++ b/posthog/test/test_flag_cache_publication.py @@ -0,0 +1,238 @@ +"""Result cache I/O must not block definition publication or local snapshots.""" + +import threading +from unittest import mock + +import pytest + +from posthog.client import Client +from posthog.request import APIError, GetResponse +from posthog.test.test_property_matching_version import definitions, evaluate +from posthog.test.test_utils import FAKE_TEST_API_KEY, FakeRedis +from posthog.utils import FlagCache, FlagCacheEntry, RedisFlagCache + + +@pytest.fixture(params=["memory", "redis"]) +def client(request): + client = Client( + FAKE_TEST_API_KEY, + secret_key="test-secret", + send=False, + enable_local_evaluation=False, + ) + client.flag_cache = ( + FlagCache() if request.param == "memory" else RedisFlagCache(FakeRedis()) + ) + client._update_flag_state(definitions(1)) + yield client + client.shutdown() + + +def test_bulk_and_refresh_do_not_wait_for_result_write(client): + entered = threading.Event() + release = threading.Event() + finished = threading.Event() + errors = [] + cache = client.flag_cache + if isinstance(cache, RedisFlagCache): + target, attribute, original = cache.redis, "setex", cache.redis.setex + else: + from posthog import utils + + target, attribute, original = utils, "FlagCacheEntry", FlagCacheEntry + + def pause_write(*args, **kwargs): + entered.set() + assert release.wait(5) + return original(*args, **kwargs) + + def write_old_result(): + try: + assert evaluate(client) is True + except BaseException as error: + errors.append(error) + + def refresh_and_evaluate(): + try: + # Neither bulk evaluation nor version-only publication may wait on + # the optional result cache, even while its write lock is held. + result = client.evaluate_flags( + "other-user", + person_properties={"value": "banana"}, + only_evaluate_locally=True, + ) + assert result.get_flag("person") is True + with mock.patch( + "posthog.client.get", + return_value=GetResponse( + data=definitions(2), etag="v2", not_modified=False + ), + ): + client._load_feature_flags() + result = client.evaluate_flags( + "other-user", + person_properties={"value": "banana"}, + only_evaluate_locally=True, + ) + assert result.get_flag("person") is False + except BaseException as error: + errors.append(error) + finally: + finished.set() + + writer = threading.Thread(target=write_old_result) + reader = threading.Thread(target=refresh_and_evaluate) + with mock.patch.object(target, attribute, side_effect=pause_write): + try: + writer.start() + assert entered.wait(2) + reader.start() + assert finished.wait(1), "local snapshot/refresh blocked by cache write" + finally: + release.set() + writer.join(5) + if reader.ident is not None: + reader.join(5) + assert not writer.is_alive() and not reader.is_alive() + assert not errors + # The old write completed AFTER refresh. Neither ordinary cache lookup nor + # API-outage fallback may expose it, even when requesting its old version. + assert cache.get_cached_flag("user", "person", 0) is None + assert cache.get_stale_cached_flag("user", "person") is None + with mock.patch.object( + client, + "_get_feature_flag_details_from_server", + side_effect=APIError(503, "offline"), + ): + assert ( + client.get_feature_flag_result( + "person", "user", send_feature_flag_events=False + ) + is None + ) + assert evaluate(client) is False + with mock.patch.object( + client, + "_get_feature_flag_details_from_server", + side_effect=APIError(503, "offline"), + ): + result = client.get_feature_flag_result( + "person", "user", send_feature_flag_events=False + ) + assert result is not None and result.get_value() is False + + +def test_new_generation_write_finishes_after_paused_old_write(client): + cache = client.flag_cache + entered = threading.Event() + release = threading.Event() + new_started = threading.Event() + errors = [] + if isinstance(cache, RedisFlagCache): + target, attribute, original = cache.redis, "setex", cache.redis.setex + else: + from posthog import utils + + target, attribute, original = utils, "FlagCacheEntry", FlagCacheEntry + + def pause_first_write(*args, **kwargs): + if not entered.is_set(): + entered.set() + assert release.wait(5) + return original(*args, **kwargs) + + def write(value, version): + try: + if value == "new": + new_started.set() + cache.set_cached_flag("user", "person", value, version) + except BaseException as error: + errors.append(error) + + old = threading.Thread(target=write, args=("old", 0)) + new = threading.Thread(target=write, args=("new", 1)) + with mock.patch.object(target, attribute, side_effect=pause_first_write): + try: + old.start() + assert entered.wait(2) + cache._advance_generation(1) + new.start() + assert new_started.wait(2) + finally: + release.set() + old.join(5) + if new.ident is not None: + new.join(5) + assert not old.is_alive() and not new.is_alive() + assert not errors + assert cache.get_cached_flag("user", "person", 1) == "new" + assert cache.get_stale_cached_flag("user", "person") == "new" + + +def test_queued_old_write_cannot_replace_new_generation(client): + cache = client.flag_cache + old_version = client.flag_definition_version + client._update_flag_state(definitions(2), client.feature_flags_by_key) + assert evaluate(client) is False + cache.set_cached_flag("user", "person", "old result", old_version) + assert cache.get_stale_cached_flag("user", "person").get_value() is False + + +@pytest.mark.parametrize("status", [401, 402]) +def test_reset_does_not_use_external_cache_cleanup(client, status): + assert evaluate(client) is True + with ( + mock.patch.object( + client.flag_cache, "clear", side_effect=AssertionError("cache I/O") + ), + mock.patch.object( + client.flag_cache, + "invalidate_version", + side_effect=AssertionError("cache I/O"), + ), + mock.patch("posthog.client.get", side_effect=APIError(status, "reset")), + ): + client._load_feature_flags() + assert not client.feature_flags + assert client.flag_cache.get_stale_cached_flag("user", "person") is None + client._update_flag_state(definitions(1)) + assert evaluate(client) is True + + +def test_memory_generation_churn_prunes_removed_flags(): + cache = FlagCache() + for version in range(100): + cache._advance_generation(version) + cache.set_cached_flag("reused-user", f"flag-{version}", True, version) + assert list(cache.cache["reused-user"]) == ["flag-99"] + + +def test_standalone_cache_can_reuse_invalidated_version(client): + cache = client.flag_cache + cache.set_cached_flag("user", "flag", True, 1) + cache.invalidate_version(1) + assert cache.get_stale_cached_flag("user", "flag") is None + cache.set_cached_flag("user", "flag", False, 1) + assert cache.get_stale_cached_flag("user", "flag") is False + cache.clear() + cache.set_cached_flag("user", "flag", True, 0) + assert cache.get_stale_cached_flag("user", "flag") is True + + +def test_fork_replaces_held_cache_write_lock_and_preserves_fence(client): + cache = client.flag_cache + client._update_flag_state(definitions(2), client.feature_flags_by_key) + lock = cache._write_lock + lock.acquire() + try: + with mock.patch.object( + client, "_initialize_flag_cache", return_value=RedisFlagCache(FakeRedis()) + ): + client._reinit_after_fork() + assert client.flag_cache._write_lock is not lock + assert client.flag_cache._minimum_version == client.flag_definition_version + assert client.flag_cache._write_lock.acquire(blocking=False) + client.flag_cache._write_lock.release() + finally: + lock.release() + assert evaluate(client) is False diff --git a/posthog/test/test_property_matching_version.py b/posthog/test/test_property_matching_version.py index fd1747cf6..5af52fd27 100644 --- a/posthog/test/test_property_matching_version.py +++ b/posthog/test/test_property_matching_version.py @@ -179,13 +179,8 @@ def test_version_only_reload_invalidates_results_and_round_trips_provider( get.return_value = GetResponse( data=definitions(version), etag=str(version), not_modified=False ) - with mock.patch.object( - client.flag_cache, - "invalidate_version", - wraps=client.flag_cache.invalidate_version, - ) as invalidate: - client._load_feature_flags() - invalidate.assert_called_once_with(previous_generation) + client._load_feature_flags() + assert client.flag_definition_version == previous_generation + 1 assert client.flag_cache.get_stale_cached_flag("user", "person") is None assert evaluate(client) is expected assert ( diff --git a/posthog/test/test_utils.py b/posthog/test/test_utils.py index e075ff4a5..5f994938a 100644 --- a/posthog/test/test_utils.py +++ b/posthog/test/test_utils.py @@ -705,6 +705,7 @@ def test_stale_cache_returns_none_when_entry_is_too_old(self): def test_stale_cache_passes_current_time_and_max_age(self): class StrictEntry: flag_result = "stale-result" + flag_definition_version = 1 def is_stale_but_usable(self, current_time, max_stale_age=3600): assert current_time == 1234 diff --git a/posthog/utils.py b/posthog/utils.py index e02037af5..8df14b0f9 100644 --- a/posthog/utils.py +++ b/posthog/utils.py @@ -2,6 +2,7 @@ import logging import numbers import re +import threading import time from collections import defaultdict from dataclasses import asdict, is_dataclass @@ -232,6 +233,16 @@ def __init__(self, max_size=CACHE_MAX_SIZE, default_ttl=CACHE_TTL): self.access_times = {} # distinct_id -> last_access_time self.max_size = max_size self.default_ttl = default_ttl + self._minimum_version = None + self._write_lock = threading.Lock() + + def _advance_generation(self, version): + # Client generations only move forward. Standalone cache invalidation + # retains its existing exact-version deletion semantics. + self._minimum_version = version + + def _is_version_current(self, version): + return self._minimum_version is None or version >= self._minimum_version def get_cached_flag(self, distinct_id, flag_key, current_flag_version): current_time = time.time() @@ -244,7 +255,9 @@ def get_cached_flag(self, distinct_id, flag_key, current_flag_version): return None entry = user_flags[flag_key] - if entry.is_valid(current_time, self.default_ttl, current_flag_version): + if self._is_version_current(entry.flag_definition_version) and entry.is_valid( + current_time, self.default_ttl, current_flag_version + ): self.access_times[distinct_id] = current_time return entry.flag_result @@ -264,7 +277,9 @@ def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=None): return None entry = user_flags[flag_key] - if entry.is_stale_but_usable(current_time, max_stale_age): + if self._is_version_current( + entry.flag_definition_version + ) and entry.is_stale_but_usable(current_time, max_stale_age): return entry.flag_result return None @@ -272,20 +287,30 @@ def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=None): def set_cached_flag( self, distinct_id, flag_key, flag_result, flag_definition_version ): - current_time = time.time() - - # Evict LRU users if we're at capacity - if distinct_id not in self.cache and len(self.cache) >= self.max_size: - self._evict_lru() - - # Initialize user cache if needed - if distinct_id not in self.cache: - self.cache[distinct_id] = {} - - # Store the flag result - entry = FlagCacheEntry(flag_result, flag_definition_version) - self.cache[distinct_id][flag_key] = entry - self.access_times[distinct_id] = current_time + with self._write_lock: + if not self._is_version_current(flag_definition_version): + return + current_time = time.time() + + # Evict LRU users if we're at capacity + if distinct_id not in self.cache and len(self.cache) >= self.max_size: + self._evict_lru() + + # Initialize user cache if needed + if distinct_id not in self.cache: + self.cache[distinct_id] = {} + + # Prune invalidated flags for reused users, not only on LRU eviction. + self.cache[distinct_id] = { + key: entry + for key, entry in self.cache[distinct_id].items() + if self._is_version_current(entry.flag_definition_version) + } + + # Store the flag result + entry = FlagCacheEntry(flag_result, flag_definition_version) + self.cache[distinct_id][flag_key] = entry + self.access_times[distinct_id] = current_time def invalidate_version(self, old_version): users_to_remove = [ @@ -348,6 +373,15 @@ def __init__( self.stale_ttl = stale_ttl self.key_prefix = key_prefix self.version_key = f"{key_prefix}version" + self._minimum_version = None + self._write_lock = threading.Lock() + + def _advance_generation(self, version): + # No Redis I/O or write lock here: publication must never wait for Redis. + self._minimum_version = version + + def _is_version_current(self, version): + return self._minimum_version is None or version >= self._minimum_version def _get_cache_key(self, distinct_id, flag_key): return f"{self.key_prefix}{distinct_id}:{flag_key}" @@ -397,8 +431,12 @@ def get_cached_flag(self, distinct_id, flag_key, current_flag_version): if data: entry = self._deserialize_entry(data) - if entry and entry.is_valid( - time.time(), self.default_ttl, current_flag_version + if ( + entry + and self._is_version_current(entry.flag_definition_version) + and entry.is_valid( + time.time(), self.default_ttl, current_flag_version + ) ): return entry.flag_result @@ -417,7 +455,11 @@ def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=None): if data: entry = self._deserialize_entry(data) - if entry and entry.is_stale_but_usable(time.time(), max_stale_age): + if ( + entry + and self._is_version_current(entry.flag_definition_version) + and entry.is_stale_but_usable(time.time(), max_stale_age) + ): return entry.flag_result return None @@ -434,11 +476,15 @@ def set_cached_flag( flag_result, flag_definition_version ) - # Set with TTL for automatic cleanup (use stale_ttl for total lifetime) - self.redis.setex(cache_key, self.stale_ttl, serialized_entry) - - # Update the current version - self.redis.set(self.version_key, flag_definition_version) + # Serialize writes so an old in-flight SETEX cannot overwrite a newer + # result. Publication advances the fence without taking this lock. + with self._write_lock: + if not self._is_version_current(flag_definition_version): + return + # Old entries (including writes finishing after publication) are + # rejected on reads and expire using the existing Redis TTL. + self.redis.setex(cache_key, self.stale_ttl, serialized_entry) + self.redis.set(self.version_key, flag_definition_version) except Exception: # Redis error - silently fail, don't break flag evaluation From 806f73588f0abfa947a69c4f77b3087a1118e116 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 17:56:56 +0200 Subject: [PATCH 3/7] fix(flags): bind Redis results to definition snapshots --- .../changesets/versioned-property-matching.md | 2 +- posthog/client.py | 80 +++-- posthog/test/test_flag_cache_publication.py | 22 +- .../test/test_redis_flag_cache_snapshot.py | 326 ++++++++++++++++++ posthog/utils.py | 57 ++- 5 files changed, 449 insertions(+), 38 deletions(-) create mode 100644 posthog/test/test_redis_flag_cache_snapshot.py diff --git a/.sampo/changesets/versioned-property-matching.md b/.sampo/changesets/versioned-property-matching.md index e5ce0057f..9abebdb7d 100644 --- a/.sampo/changesets/versioned-property-matching.md +++ b/.sampo/changesets/versioned-property-matching.md @@ -2,4 +2,4 @@ pypi/posthog: patch --- -Honor the definitions snapshot's `property_matching_version` during local feature flag evaluation, including person, group, cohort, and flag dependency conditions. Version 2 uses explicit boolean equality; missing/1 retains legacy truthiness. Preserve the selector through definition caches and invalidate evaluated results on version-only refreshes. +Honor the definitions snapshot's `property_matching_version` during local feature flag evaluation, including person, group, cohort, and flag dependency conditions. Version 2 uses explicit boolean equality; missing/1 retains legacy truthiness. Preserve the selector through definition caches and invalidate evaluated results on version-only refreshes. Bind Client-managed Redis results to their definitions snapshot so invalidated entries cannot revive after a worker restart. Older entries without snapshot metadata become cache misses for these clients. diff --git a/posthog/client.py b/posthog/client.py index b5637a251..d9a31bcb1 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -1,4 +1,5 @@ import atexit +import hashlib as _hashlib import inspect import json import logging @@ -897,6 +898,11 @@ def __init__( self.flag_fallback_cache_url = flag_fallback_cache_url self.flag_cache = self._initialize_flag_cache(flag_fallback_cache_url) self.flag_definition_version = 0 + self._flag_definition_fingerprint = "remote-only" + if self.flag_cache: + self.flag_cache._advance_generation( + self.flag_definition_version, self._flag_definition_fingerprint + ) self._flags_etag: Optional[str] = None self._flag_definition_fetch_generation = 0 self._flag_definition_published_generation = 0 @@ -2255,7 +2261,9 @@ def _reinit_after_fork(self): self.flag_cache = self._initialize_flag_cache(self.flag_fallback_cache_url) if self.flag_cache: self.flag_cache._write_lock = threading.Lock() - self.flag_cache._advance_generation(self.flag_definition_version) + self.flag_cache._advance_generation( + self.flag_definition_version, self._flag_definition_fingerprint + ) reset_sessions() @@ -2877,12 +2885,32 @@ def _shutdown_flag_definition_cache_provider(self): self._flag_definition_cache_provider_async_runner.close() self._flag_definition_cache_provider_async_runner = None + @staticmethod + def _hash_flag_definitions(data: FlagDefinitionCacheData) -> str: + # Hash only evaluation inputs, not transport metadata such as the ETag. + serialized = json.dumps( + { + "flags": data["flags"], + "cohorts": data["cohorts"], + "group_type_mapping": data["group_type_mapping"], + "property_matching_version": data.get("property_matching_version", 1), + }, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ) + return _hashlib.sha256(serialized.encode("utf-8")).hexdigest() + def _update_flag_state( - self, data: FlagDefinitionCacheData, old_flags_by_key: Optional[dict] = None + self, + data: FlagDefinitionCacheData, + old_flags_by_key: Optional[dict] = None, + *, + _fingerprint: Optional[str] = None, ) -> None: - """Update internal flag state from cache data and invalidate evaluation cache if changed.""" + """Publish definitions and their result-cache identity together.""" + fingerprint = _fingerprint or self._hash_flag_definitions(data) with self._flag_definition_publication_lock: - old_matching_version = self._property_matching_version self.feature_flags = data["flags"] self.group_type_mapping = data["group_type_mapping"] self.cohorts = data["cohorts"] @@ -2892,17 +2920,15 @@ def _update_flag_state( data.get("minimal_flag_called_events") is True ) - if self.flag_cache and ( - old_matching_version != self._property_matching_version - or ( - old_flags_by_key is not None - and old_flags_by_key != (self.feature_flags_by_key or {}) - ) - ): - self.flag_definition_version += 1 - # Only advance an in-memory fence while publishing. Redis I/O - # must not delay snapshots, and late writes remain invalid. - self.flag_cache._advance_generation(self.flag_definition_version) + if fingerprint != self._flag_definition_fingerprint: + self._flag_definition_fingerprint = fingerprint + if self.flag_cache: + self.flag_definition_version += 1 + # No Redis I/O under this lock. The fingerprint remains valid + # across processes; the counter fences this process's writes. + self.flag_cache._advance_generation( + self.flag_definition_version, fingerprint + ) def _local_evaluation_snapshot(self) -> _LocalEvaluationSnapshot: # Capture references together. Publication replaces these collections, so @@ -2988,6 +3014,13 @@ def _fetch_feature_flags_from_api(self): **self._request_identity_kwargs(), ) + # Canonical serialization can be expensive; do it before publication. + fingerprint = ( + self._hash_flag_definitions(response.data) + if response.data is not None and not response.not_modified + else None + ) + with self._flag_definition_publication_lock: if fetch_generation <= self._flag_definition_published_generation: self.log.debug( @@ -3023,7 +3056,9 @@ def _fetch_feature_flags_from_api(self): old_flags_by_key: dict[str, dict] = self.feature_flags_by_key or {} self._update_flag_state( - response.data, old_flags_by_key=old_flags_by_key + response.data, + old_flags_by_key=old_flags_by_key, + _fingerprint=fingerprint, ) if self._flag_definition_cache_provider: @@ -3077,6 +3112,7 @@ def _fetch_feature_flags_from_api(self): self.group_type_mapping = {} self.cohorts = {} self._property_matching_version = 1 + self._flag_definition_fingerprint = "" self._flags_etag = None self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation @@ -3084,7 +3120,8 @@ def _fetch_feature_flags_from_api(self): if self.flag_cache: self.flag_definition_version += 1 self.flag_cache._advance_generation( - self.flag_definition_version + self.flag_definition_version, + self._flag_definition_fingerprint, ) if self.debug: @@ -3098,6 +3135,7 @@ def _fetch_feature_flags_from_api(self): self.group_type_mapping = {} self.cohorts = {} self._property_matching_version = 1 + self._flag_definition_fingerprint = "" self._flags_etag = None self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation @@ -3106,7 +3144,8 @@ def _fetch_feature_flags_from_api(self): if self.flag_cache: self.flag_definition_version += 1 self.flag_cache._advance_generation( - self.flag_definition_version + self.flag_definition_version, + self._flag_definition_fingerprint, ) if self.debug: @@ -3454,10 +3493,11 @@ def _get_feature_flag_result( flag_details, override_match_value ) - # Cache successful remote evaluation + # The request-start generation is an invalidation boundary, not + # a claim about the server's definitions. Refresh rejects late writes. if self.flag_cache and flag_result: self.flag_cache.set_cached_flag( - distinct_id, key, flag_result, self.flag_definition_version + distinct_id, key, flag_result, local_definition_version ) self.log.debug( diff --git a/posthog/test/test_flag_cache_publication.py b/posthog/test/test_flag_cache_publication.py index bc45269a6..52b61829e 100644 --- a/posthog/test/test_flag_cache_publication.py +++ b/posthog/test/test_flag_cache_publication.py @@ -97,7 +97,10 @@ def refresh_and_evaluate(): assert not errors # The old write completed AFTER refresh. Neither ordinary cache lookup nor # API-outage fallback may expose it, even when requesting its old version. - assert cache.get_cached_flag("user", "person", 0) is None + assert ( + cache.get_cached_flag("user", "person", client.flag_definition_version - 1) + is None + ) assert cache.get_stale_cached_flag("user", "person") is None with mock.patch.object( client, @@ -149,13 +152,15 @@ def write(value, version): except BaseException as error: errors.append(error) - old = threading.Thread(target=write, args=("old", 0)) - new = threading.Thread(target=write, args=("new", 1)) + old_version = client.flag_definition_version + new_version = old_version + 1 + old = threading.Thread(target=write, args=("old", old_version)) + new = threading.Thread(target=write, args=("new", new_version)) with mock.patch.object(target, attribute, side_effect=pause_first_write): try: old.start() assert entered.wait(2) - cache._advance_generation(1) + client._update_flag_state(definitions(2)) new.start() assert new_started.wait(2) finally: @@ -165,7 +170,7 @@ def write(value, version): new.join(5) assert not old.is_alive() and not new.is_alive() assert not errors - assert cache.get_cached_flag("user", "person", 1) == "new" + assert cache.get_cached_flag("user", "person", new_version) == "new" assert cache.get_stale_cached_flag("user", "person") == "new" @@ -208,7 +213,12 @@ def test_memory_generation_churn_prunes_removed_flags(): def test_standalone_cache_can_reuse_invalidated_version(client): - cache = client.flag_cache + # Standalone instances have no Client snapshot binding. + cache = ( + RedisFlagCache(FakeRedis()) + if isinstance(client.flag_cache, RedisFlagCache) + else FlagCache() + ) cache.set_cached_flag("user", "flag", True, 1) cache.invalidate_version(1) assert cache.get_stale_cached_flag("user", "flag") is None diff --git a/posthog/test/test_redis_flag_cache_snapshot.py b/posthog/test/test_redis_flag_cache_snapshot.py new file mode 100644 index 000000000..130ee4665 --- /dev/null +++ b/posthog/test/test_redis_flag_cache_snapshot.py @@ -0,0 +1,326 @@ +"""Shared Redis results must retain the definitions that produced them.""" + +import json +import threading +from copy import deepcopy +from unittest import mock + +import pytest + +from posthog.client import Client +from posthog.request import APIError, GetResponse +from posthog.test.test_flag_definition_cache import ( + AsyncMockCacheProvider, + MockCacheProvider, +) +from posthog.test.test_property_matching_version import definitions +from posthog.test.test_utils import FAKE_TEST_API_KEY, FakeRedis +from posthog.types import FeatureFlag, FeatureFlagResult +from posthog.utils import RedisFlagCache + + +@pytest.fixture +def workers(): + redis = FakeRedis() + clients = [] + + def worker(): + with mock.patch.object( + Client, "_initialize_flag_cache", return_value=RedisFlagCache(redis) + ): + client = Client( + FAKE_TEST_API_KEY, + secret_key="test-secret", + send=False, + enable_local_evaluation=False, + ) + clients.append(client) + return client + + yield worker + for client in clients: + client.shutdown() + + +def load(client, data): + with mock.patch( + "posthog.client.get", + return_value=GetResponse(data=data, etag=None, not_modified=False), + ): + client._load_feature_flags() + + +def evaluate(client, key="person"): + return client.get_feature_flag_result( + key, + "user", + person_properties={"value": "banana"}, + groups={"company": "company-id"}, + group_properties={"company": {"value": "banana"}}, + only_evaluate_locally=True, + send_feature_flag_events=False, + ) + + +def fallback(client, key="person"): + with mock.patch.object( + client, + "_get_feature_flag_details_from_server", + side_effect=APIError(503, "offline"), + ) as remote: + result = client.get_feature_flag_result( + key, "user", send_feature_flag_events=False + ) + remote.assert_called_once() + return result + + +def changed_definitions(change): + # The shared fixture reuses property dicts across flags and cohorts. A wire + # round trip makes these independent so cohort-only changes really are isolated. + data = json.loads(json.dumps(definitions(1))) + if change == "version": + data["property_matching_version"] = 2 + elif change == "flags": + data["flags"][0]["filters"]["groups"][0]["properties"][0]["value"] = True + elif change == "cohorts": + data["cohorts"]["2"]["values"][0]["value"] = True + else: + data["group_type_mapping"] = {"0": "organization"} + return data + + +@pytest.mark.parametrize("change", ["flags", "version", "cohorts", "mapping"]) +@pytest.mark.parametrize( + "provider_class", [None, MockCacheProvider, AsyncMockCacheProvider] +) +def test_invalidated_result_does_not_revive_after_restart( + workers, change, provider_class +): + writer = workers() + load(writer, definitions(1)) + assert evaluate(writer).get_value() is True + updated = changed_definitions(change) + load(writer, updated) + assert writer.flag_cache.get_stale_cached_flag("user", "person") is None + writer.shutdown() + + reader = workers() + if provider_class is None: + load(reader, updated) + else: + provider = provider_class() + provider.should_fetch_return_value = False + provider.stored_data = updated + reader._flag_definition_cache_provider = provider + reader._load_feature_flags() + assert ( + reader.flag_cache.get_cached_flag( + "user", "person", reader.flag_definition_version + ) + is None + ) + assert fallback(reader) is None + + +@pytest.mark.parametrize("change", ["flags", "version", "cohorts", "mapping"]) +def test_old_worker_write_is_not_accepted_by_current_worker(workers, change): + old = workers() + current = workers() + load(old, definitions(1)) + load(current, changed_definitions(change)) + assert old.flag_definition_version == current.flag_definition_version + assert evaluate(old).get_value() is True + assert ( + current.flag_cache.get_cached_flag( + "user", "person", current.flag_definition_version + ) + is None + ) + assert fallback(current) is None + + +@pytest.mark.parametrize("change", ["cohorts", "mapping"]) +def test_refresh_during_evaluation_cannot_relabel_result(workers, change): + client = workers() + load(client, definitions(1)) + from posthog import client as client_module + + original = client_module.match_feature_flag_properties + + def refresh(*args, **kwargs): + load(client, changed_definitions(change)) + return original(*args, **kwargs) + + with mock.patch( + "posthog.client.match_feature_flag_properties", side_effect=refresh + ): + assert evaluate(client).get_value() is True + assert fallback(client) is None + + +def test_paused_setex_keeps_original_fingerprint_after_refresh_and_restart(workers): + writer = workers() + load(writer, definitions(1)) + entered = threading.Event() + release = threading.Event() + errors = [] + original = writer.flag_cache.redis.setex + + def pause(*args, **kwargs): + entered.set() + assert release.wait(5) + return original(*args, **kwargs) + + def write(): + try: + assert evaluate(writer).get_value() is True + except BaseException as error: + errors.append(error) + + thread = threading.Thread(target=write) + with mock.patch.object(writer.flag_cache.redis, "setex", side_effect=pause): + try: + thread.start() + assert entered.wait(2) + load(writer, definitions(2)) + finally: + release.set() + thread.join(5) + assert not thread.is_alive() and not errors + reader = workers() + load(reader, definitions(2)) + assert fallback(reader) is None + + +def test_legacy_missing_metadata_is_miss_but_standalone_cache_still_reads(workers): + client = workers() + load(client, definitions(1)) + cache = client.flag_cache + standalone = RedisFlagCache(cache.redis) + standalone.set_cached_flag("user", "person", True, client.flag_definition_version) + assert ( + standalone.get_cached_flag("user", "person", client.flag_definition_version) + is True + ) + assert standalone.get_stale_cached_flag("user", "person") is True + assert ( + cache.get_cached_flag("user", "person", client.flag_definition_version) is None + ) + assert fallback(client) is None + assert evaluate(client).get_value() is True + # Additive metadata remains readable without opting into snapshot validation. + assert standalone.get_stale_cached_flag("user", "person").get_value() is True + + +def test_matching_snapshot_reused_despite_different_counters_and_key_order(workers): + writer = workers() + load(writer, definitions(2)) + load(writer, definitions(1)) + assert evaluate(writer).get_value() is True + reader = workers() + reordered = json.loads(json.dumps(definitions(1), sort_keys=True)) + reordered.pop("property_matching_version") # Missing and 1 have the same default. + load(reader, reordered) + assert writer.flag_definition_version != reader.flag_definition_version + assert ( + reader.flag_cache.get_cached_flag( + "user", "person", reader.flag_definition_version + ).get_value() + is True + ) + assert fallback(reader).get_value() is True + generation = reader.flag_definition_version + load(reader, deepcopy(reordered)) + assert reader.flag_definition_version == generation + assert fallback(reader).get_value() is True + + +def test_fork_retains_snapshot_binding(workers): + client = workers() + load(client, definitions(2)) + assert evaluate(client).get_value() is False + redis = client.flag_cache.redis + with mock.patch.object( + client, "_initialize_flag_cache", return_value=RedisFlagCache(redis) + ): + client._reinit_after_fork() + assert fallback(client).get_value() is False + old = workers() + load(old, definitions(1)) + assert evaluate(old).get_value() is True + assert fallback(client) is None + + +def remote_success(client, during_request=None): + details = FeatureFlag.from_json({"key": "person", "enabled": True}) + + def request(*args, **kwargs): + if during_request: + during_request() + return details, None, None, False, False + + with mock.patch.object( + client, "_get_feature_flag_details_from_server", side_effect=request + ): + return client.get_feature_flag_result( + "person", "user", send_feature_flag_events=False + ) + + +def test_remote_only_success_remains_available_for_stale_fallback(workers): + writer = workers() + # No local definitions were available from the server. + with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): + assert remote_success(writer).get_value() is True + assert fallback(writer).get_value() is True + assert fallback(workers()).get_value() is True + + +@pytest.mark.parametrize("transition", ["hydrate", "empty-hydrate", 401, 402]) +def test_delayed_remote_response_cannot_cross_snapshot_transition(workers, transition): + client = workers() + if isinstance(transition, int): + load(client, definitions(1)) + + def change(): + if isinstance(transition, int): + with mock.patch( + "posthog.client.get", side_effect=APIError(transition, "reset") + ): + client._load_feature_flags() + else: + data = definitions(1) + if transition == "empty-hydrate": + data["flags"] = [] + load(client, data) + + with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): + assert remote_success(client, change).get_value() is True + assert fallback(client) is None + load(client, definitions(1)) + assert evaluate(client).get_value() is True + assert fallback(client).get_value() is True + + +@pytest.mark.parametrize("status", [401, 402]) +def test_reset_empty_fingerprint_cannot_cache_or_read_results(workers, status): + client = workers() + load(client, definitions(1)) + with mock.patch("posthog.client.get", side_effect=APIError(status, "reset")): + client._load_feature_flags() + assert remote_success(client).get_value() is True + cache = client.flag_cache + cache.redis.setex( + cache._get_cache_key("user", "person"), + cache.stale_ttl, + cache._serialize_entry( + FeatureFlagResult.from_value_and_payload("person", True, None), + client.flag_definition_version, + fingerprint="", + ), + ) + assert ( + cache.get_cached_flag("user", "person", client.flag_definition_version) is None + ) + assert fallback(client) is None diff --git a/posthog/utils.py b/posthog/utils.py index 8df14b0f9..70b595b69 100644 --- a/posthog/utils.py +++ b/posthog/utils.py @@ -217,6 +217,7 @@ def __init__(self, flag_result, flag_definition_version, timestamp=None): self.flag_result = flag_result self.flag_definition_version = flag_definition_version self.timestamp = timestamp or time.time() + self._snapshot_fingerprint: Optional[str] = None def is_valid(self, current_time, ttl, current_flag_version): time_valid = (current_time - self.timestamp) < ttl @@ -236,7 +237,7 @@ def __init__(self, max_size=CACHE_MAX_SIZE, default_ttl=CACHE_TTL): self._minimum_version = None self._write_lock = threading.Lock() - def _advance_generation(self, version): + def _advance_generation(self, version, fingerprint=None): # Client generations only move forward. Standalone cache invalidation # retains its existing exact-version deletion semantics. self._minimum_version = version @@ -373,12 +374,21 @@ def __init__( self.stale_ttl = stale_ttl self.key_prefix = key_prefix self.version_key = f"{key_prefix}version" + self._snapshot = None self._minimum_version = None self._write_lock = threading.Lock() - def _advance_generation(self, version): + def _advance_generation(self, version, fingerprint=None): # No Redis I/O or write lock here: publication must never wait for Redis. self._minimum_version = version + if fingerprint is not None: + self._snapshot = (version, fingerprint) + + def _is_entry_current(self, entry): + snapshot = self._snapshot + if snapshot is not None: + return bool(snapshot[1]) and entry._snapshot_fingerprint == snapshot[1] + return self._is_version_current(entry.flag_definition_version) def _is_version_current(self, version): return self._minimum_version is None or version >= self._minimum_version @@ -386,7 +396,9 @@ def _is_version_current(self, version): def _get_cache_key(self, distinct_id, flag_key): return f"{self.key_prefix}{distinct_id}:{flag_key}" - def _serialize_entry(self, flag_result, flag_definition_version, timestamp=None): + def _serialize_entry( + self, flag_result, flag_definition_version, timestamp=None, fingerprint=None + ): if timestamp is None: timestamp = time.time() @@ -398,6 +410,8 @@ def _serialize_entry(self, flag_result, flag_definition_version, timestamp=None) "flag_version": flag_definition_version, "timestamp": timestamp, } + if fingerprint is not None: + entry["snapshot_fingerprint"] = fingerprint if isinstance(flag_result, _FeatureFlagResult): # Additive metadata keeps the existing entry shape readable by older SDKs. entry["flag_result_type"] = _FEATURE_FLAG_RESULT_TYPE @@ -415,11 +429,13 @@ def _deserialize_entry(self, data): ): return None flag_result = _FeatureFlagResult(**flag_result) - return FlagCacheEntry( + result = FlagCacheEntry( flag_result=flag_result, flag_definition_version=entry["flag_version"], timestamp=entry["timestamp"], ) + result._snapshot_fingerprint = entry.get("snapshot_fingerprint") + return result except (json.JSONDecodeError, KeyError, TypeError, ValueError): # If deserialization fails, treat as cache miss return None @@ -433,9 +449,19 @@ def get_cached_flag(self, distinct_id, flag_key, current_flag_version): entry = self._deserialize_entry(data) if ( entry - and self._is_version_current(entry.flag_definition_version) - and entry.is_valid( - time.time(), self.default_ttl, current_flag_version + and self._is_entry_current(entry) + and ( + ( + self._snapshot is not None + and current_flag_version == self._snapshot[0] + and entry.is_stale_but_usable(time.time(), self.default_ttl) + ) + or ( + self._snapshot is None + and entry.is_valid( + time.time(), self.default_ttl, current_flag_version + ) + ) ) ): return entry.flag_result @@ -457,7 +483,7 @@ def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=None): entry = self._deserialize_entry(data) if ( entry - and self._is_version_current(entry.flag_definition_version) + and self._is_entry_current(entry) and entry.is_stale_but_usable(time.time(), max_stale_age) ): return entry.flag_result @@ -472,8 +498,17 @@ def set_cached_flag( ): try: cache_key = self._get_cache_key(distinct_id, flag_key) + # Capture provenance before any blocking work. Never stamp an old + # evaluation with the fingerprint installed by a concurrent refresh. + snapshot = self._snapshot + if snapshot is not None and ( + flag_definition_version != snapshot[0] or not snapshot[1] + ): + return serialized_entry = self._serialize_entry( - flag_result, flag_definition_version + flag_result, + flag_definition_version, + fingerprint=snapshot[1] if snapshot is not None else None, ) # Serialize writes so an old in-flight SETEX cannot overwrite a newer @@ -481,8 +516,8 @@ def set_cached_flag( with self._write_lock: if not self._is_version_current(flag_definition_version): return - # Old entries (including writes finishing after publication) are - # rejected on reads and expire using the existing Redis TTL. + # Late writes keep their original fingerprint. Other workers + # reject them too, even if their local generation counters differ. self.redis.setex(cache_key, self.stale_ttl, serialized_entry) self.redis.set(self.version_key, flag_definition_version) From 3dbe088604af0a5731a4649bab0ddfffe6e63af5 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 18:03:11 +0200 Subject: [PATCH 4/7] fix(flags): isolate unverifiable remote cache provenance --- posthog/client.py | 8 ++++- .../test/test_redis_flag_cache_snapshot.py | 29 ++++++++++++++++++- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/posthog/client.py b/posthog/client.py index d9a31bcb1..07c931d4f 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -898,7 +898,8 @@ def __init__( self.flag_fallback_cache_url = flag_fallback_cache_url self.flag_cache = self._initialize_flag_cache(flag_fallback_cache_url) self.flag_definition_version = 0 - self._flag_definition_fingerprint = "remote-only" + # Without definitions, remote results have only process-local provenance. + self._flag_definition_fingerprint = f"remote-only:{uuid4().hex}" if self.flag_cache: self.flag_cache._advance_generation( self.flag_definition_version, self._flag_definition_fingerprint @@ -2255,6 +2256,11 @@ def _reinit_after_fork(self): if self._metrics is not None: self._metrics._reinit_after_fork() + # Unknown definitions cannot establish shared provenance in a new worker. + if self._flag_definition_fingerprint.startswith("remote-only:"): + self._flag_definition_fingerprint = f"remote-only:{uuid4().hex}" + self.flag_definition_version += 1 + # If using Redis cache, we must reinitialize to get a fresh connection (fork-safe). # If using Memory cache, we keep it as-is to benefit from the inherited warm cache. if isinstance(self.flag_cache, RedisFlagCache): diff --git a/posthog/test/test_redis_flag_cache_snapshot.py b/posthog/test/test_redis_flag_cache_snapshot.py index 130ee4665..bac587221 100644 --- a/posthog/test/test_redis_flag_cache_snapshot.py +++ b/posthog/test/test_redis_flag_cache_snapshot.py @@ -274,7 +274,8 @@ def test_remote_only_success_remains_available_for_stale_fallback(workers): with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): assert remote_success(writer).get_value() is True assert fallback(writer).get_value() is True - assert fallback(workers()).get_value() is True + # Without verified definitions, a new Client cannot reuse this provenance. + assert fallback(workers()) is None @pytest.mark.parametrize("transition", ["hydrate", "empty-hydrate", 401, 402]) @@ -324,3 +325,29 @@ def test_reset_empty_fingerprint_cannot_cache_or_read_results(workers, status): cache.get_cached_flag("user", "person", client.flag_definition_version) is None ) assert fallback(client) is None + + +def test_invalidated_remote_only_result_does_not_revive_after_restart(workers): + writer = workers() + with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): + assert remote_success(writer).get_value() is True + assert fallback(writer).get_value() is True + load(writer, definitions(1)) + assert fallback(writer) is None + writer.shutdown() + with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): + assert fallback(workers()) is None + + +def test_fork_renews_remote_only_provenance(workers): + client = workers() + with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")): + assert remote_success(client).get_value() is True + redis = client.flag_cache.redis + with mock.patch.object( + client, "_initialize_flag_cache", return_value=RedisFlagCache(redis) + ): + client._reinit_after_fork() + assert fallback(client) is None + assert remote_success(client).get_value() is True + assert fallback(client).get_value() is True From a5a6dfb8e96c4b5cb8ad4e857ab2a18da3e6209e Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Sat, 5 Sep 2026 19:41:09 +0200 Subject: [PATCH 5/7] test(flags): cover cache provenance in targeted mutation suite --- posthog/test/test_utils.py | 121 +++++++++++++++++++++++++++++++++++++ posthog/utils.py | 48 ++++++--------- 2 files changed, 140 insertions(+), 29 deletions(-) diff --git a/posthog/test/test_utils.py b/posthog/test/test_utils.py index 5f994938a..dd39e17c6 100644 --- a/posthog/test/test_utils.py +++ b/posthog/test/test_utils.py @@ -586,6 +586,8 @@ def test_cache_entry_validity(self): self.flag_result, flag_definition_version=1, timestamp=100 ) + # Standalone entries have no definition snapshot provenance. + assert entry._snapshot_fingerprint is None assert entry.is_valid(current_time=109, ttl=10, current_flag_version=1) is True assert entry.is_valid(current_time=110, ttl=10, current_flag_version=1) is False assert entry.is_valid(current_time=111, ttl=10, current_flag_version=1) is False @@ -612,6 +614,26 @@ def test_cache_basic_operations(self): assert result is not None assert result.get_value() + def test_generation_fences_reads_and_writes_and_prunes_reused_users(self): + self.cache.set_cached_flag("user", "old", False, 1) + self.cache.set_cached_flag("user", "current", "variant", 2) + self.cache._advance_generation(2) + + assert self.cache.get_cached_flag("user", "old", 1) is None + assert self.cache.get_stale_cached_flag("user", "old") is None + assert self.cache.get_cached_flag("user", "current", 2) == "variant" + assert self.cache.get_stale_cached_flag("user", "current") == "variant" + access_times = self.cache.access_times.copy() + self.cache.set_cached_flag("missing-user", "late", True, 1) + self.cache.set_cached_flag("user", "current", False, 1) + assert "missing-user" not in self.cache.cache + assert self.cache.access_times == access_times + assert self.cache.get_cached_flag("user", "current", 2) == "variant" + + self.cache.set_cached_flag("user", "future", True, 3) + assert set(self.cache.cache["user"]) == {"current", "future"} + assert self.cache.get_cached_flag("user", "future", 3) is True + def test_cache_ttl_expiration(self): distinct_id = "user123" flag_key = "test-flag" @@ -920,6 +942,7 @@ def test_get_set_and_stale_cached_flags(self): assert self.cache.get_cached_flag("user123", "beta", 8) is None assert self.redis.store["test:flags:version"] == 7 assert self.redis.setex_calls[0][1] == 60 + assert "snapshot_fingerprint" not in json.loads(self.redis.setex_calls[0][2]) stale_key = self.cache._get_cache_key("user123", "old-beta") self.redis.store[stale_key] = self.cache._serialize_entry( @@ -947,6 +970,104 @@ def test_get_set_and_stale_cached_flags(self): ) assert self.cache.get_stale_cached_flag("user123", "boundary-stale") is None + def test_snapshot_round_trip_reuses_matching_worker_with_different_generation(self): + self.cache._advance_generation(7, "snapshot-a") + with mock.patch("posthog.utils.time.time", return_value=100): + self.cache.set_cached_flag("user", "beta", False, 7) + key, ttl, data = self.redis.setex_calls[-1] + assert key == "test:flags:user:beta" + assert ttl == 60 + assert json.loads(data) == { + "flag_result": False, + "flag_version": 7, + "timestamp": 100, + "snapshot_fingerprint": "snapshot-a", + } + assert self.redis.store[self.cache.version_key] == 7 + reader = utils.RedisFlagCache( + self.redis, default_ttl=10, stale_ttl=60, key_prefix="test:flags:" + ) + reader._advance_generation(2, "snapshot-a") + with mock.patch("posthog.utils.time.time", return_value=109): + assert reader.get_cached_flag("user", "beta", 2) is False + assert reader.get_cached_flag("user", "beta", 7) is None + assert reader.get_stale_cached_flag("user", "beta") is False + with mock.patch("posthog.utils.time.time", return_value=110): + assert reader.get_cached_flag("user", "beta", 2) is None + assert reader.get_stale_cached_flag("user", "beta") is False + with mock.patch("posthog.utils.time.time", return_value=160): + assert reader.get_stale_cached_flag("user", "beta") is None + + @parameterized.expand([(None,), ("snapshot-old",), ("",)]) + def test_snapshot_reads_reject_missing_or_mismatched_fingerprints( + self, fingerprint + ): + self.cache._advance_generation(7, "snapshot-current") + key = self.cache._get_cache_key("user", "beta") + self.redis.store[key] = self.cache._serialize_entry( + True, 7, fingerprint=fingerprint + ) + assert self.cache.get_cached_flag("user", "beta", 7) is None + assert self.cache.get_stale_cached_flag("user", "beta") is None + + @parameterized.expand([(6, "snapshot-a"), (8, "snapshot-a"), (7, "")]) + def test_snapshot_writes_reject_wrong_generation_or_disabled_snapshot( + self, version, fingerprint + ): + self.cache._advance_generation(7, fingerprint) + self.cache.set_cached_flag("user", "beta", True, version) + assert self.redis.store == {} + assert self.redis.setex_calls == [] + + def test_disabled_snapshot_rejects_matching_empty_fingerprint(self): + self.cache._advance_generation(7, "") + key = self.cache._get_cache_key("user", "beta") + self.redis.store[key] = self.cache._serialize_entry(True, 7, fingerprint="") + assert self.cache.get_cached_flag("user", "beta", 7) is None + assert self.cache.get_stale_cached_flag("user", "beta") is None + + def test_generation_without_snapshot_fences_standalone_cache(self): + self.cache.set_cached_flag("user", "old", True, 1) + self.cache._advance_generation(2) + assert self.cache.get_cached_flag("user", "old", 1) is None + assert self.cache.get_stale_cached_flag("user", "old") is None + before = self.redis.store.copy() + self.cache.set_cached_flag("user", "old", False, 1) + assert self.redis.store == before + for version in (2, 3): + self.cache.set_cached_flag("user", "current", "variant", version) + assert self.cache.get_cached_flag("user", "current", version) == "variant" + assert self.cache.get_stale_cached_flag("user", "current") == "variant" + + def test_generation_advance_during_serialization_discards_late_write(self): + self.cache._advance_generation(7, "snapshot-a") + serialize = self.cache._serialize_entry + + def refresh_then_serialize(*args, **kwargs): + self.cache._advance_generation(8, "snapshot-b") + return serialize(*args, **kwargs) + + with mock.patch.object( + self.cache, "_serialize_entry", side_effect=refresh_then_serialize + ): + self.cache.set_cached_flag("user", "beta", True, 7) + assert self.redis.store == {} + assert self.redis.setex_calls == [] + + def test_generation_advance_without_fingerprint_preserves_snapshot_binding(self): + self.cache._advance_generation(7, "snapshot-a") + self.cache._advance_generation(8) + assert self.cache._snapshot == (7, "snapshot-a") + assert not self.cache._is_version_current(7) + assert self.cache._is_version_current(8) + + @parameterized.expand([(None,), ("not json",)]) + def test_missing_or_corrupt_entry_is_a_cache_miss(self, data): + if data is not None: + self.redis.store[self.cache._get_cache_key("user", "beta")] = data + assert self.cache.get_cached_flag("user", "beta", 7) is None + assert self.cache.get_stale_cached_flag("user", "beta") is None + def test_redis_errors_fall_back_to_miss(self): failing_cache = utils.RedisFlagCache(FakeRedis(fail=True)) diff --git a/posthog/utils.py b/posthog/utils.py index 70b595b69..5070afcf8 100644 --- a/posthog/utils.py +++ b/posthog/utils.py @@ -293,12 +293,10 @@ def set_cached_flag( return current_time = time.time() - # Evict LRU users if we're at capacity - if distinct_id not in self.cache and len(self.cache) >= self.max_size: - self._evict_lru() - - # Initialize user cache if needed + # Initialize new users, evicting LRU users if we're at capacity. if distinct_id not in self.cache: + if len(self.cache) >= self.max_size: + self._evict_lru() self.cache[distinct_id] = {} # Prune invalidated flags for reused users, not only on LRU eviction. @@ -447,24 +445,17 @@ def get_cached_flag(self, distinct_id, flag_key, current_flag_version): if data: entry = self._deserialize_entry(data) - if ( - entry - and self._is_entry_current(entry) - and ( - ( - self._snapshot is not None - and current_flag_version == self._snapshot[0] - and entry.is_stale_but_usable(time.time(), self.default_ttl) - ) - or ( - self._snapshot is None - and entry.is_valid( - time.time(), self.default_ttl, current_flag_version - ) + if entry and self._is_entry_current(entry): + if self._snapshot is not None: + valid = current_flag_version == self._snapshot[ + 0 + ] and entry.is_stale_but_usable(time.time(), self.default_ttl) + else: + valid = entry.is_valid( + time.time(), self.default_ttl, current_flag_version ) - ) - ): - return entry.flag_result + if valid: + return entry.flag_result return None except Exception: @@ -501,14 +492,13 @@ def set_cached_flag( # Capture provenance before any blocking work. Never stamp an old # evaluation with the fingerprint installed by a concurrent refresh. snapshot = self._snapshot - if snapshot is not None and ( - flag_definition_version != snapshot[0] or not snapshot[1] - ): - return + fingerprint = None + if snapshot is not None: + if flag_definition_version != snapshot[0] or not snapshot[1]: + return + fingerprint = snapshot[1] serialized_entry = self._serialize_entry( - flag_result, - flag_definition_version, - fingerprint=snapshot[1] if snapshot is not None else None, + flag_result, flag_definition_version, fingerprint=fingerprint ) # Serialize writes so an old in-flight SETEX cannot overwrite a newer From 6ffbe455bf16b37ffcf4d2d6bdc049b0b0317d13 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Tue, 8 Sep 2026 09:35:25 +0200 Subject: [PATCH 6/7] test: enable local feature flag evaluation compliance --- .github/workflows/sdk-compliance.yml | 25 +- sdk_compliance_adapter/CONTRIBUTING.md | 13 +- sdk_compliance_adapter/README.md | 27 ++ sdk_compliance_adapter/adapter.py | 160 +++++++++-- sdk_compliance_adapter/docker-compose.yml | 2 +- sdk_compliance_adapter/test_adapter.py | 327 ++++++++++++++++++++++ 6 files changed, 517 insertions(+), 37 deletions(-) create mode 100644 sdk_compliance_adapter/test_adapter.py diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index 21d080595..1108654c6 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -12,20 +12,37 @@ on: - main jobs: + adapter-tests: + name: Compliance adapter protocol tests + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + - name: Install adapter and test dependencies + run: python -m pip install -e . -r sdk_compliance_adapter/requirements.txt pytest pytest-timeout pytest-asyncio + - name: Test adapter protocol + run: python -m pytest sdk_compliance_adapter/test_adapter.py --timeout=30 + compliance: name: PostHog SDK compliance tests (capture v0) - uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@03d972e49be84402c491324320b0a0f38c2ddc53 + uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@15011d6868ee73e11333860b6392d35298b92535 # 1.1.0 with: adapter-dockerfile: "sdk_compliance_adapter/Dockerfile" adapter-context: "." - test-harness-version: "0.10.0" + test-harness-version: "1.1.0" + continue-on-error: false report-name: "sdk-compliance-report-v0" compliance-v1: name: PostHog SDK compliance tests (capture v1) - uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@03d972e49be84402c491324320b0a0f38c2ddc53 + uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@15011d6868ee73e11333860b6392d35298b92535 # 1.1.0 with: adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1" adapter-context: "." - test-harness-version: "0.10.0" + test-harness-version: "1.1.0" + continue-on-error: false report-name: "sdk-compliance-report-v1" diff --git a/sdk_compliance_adapter/CONTRIBUTING.md b/sdk_compliance_adapter/CONTRIBUTING.md index a1e0fe163..f76cab02a 100644 --- a/sdk_compliance_adapter/CONTRIBUTING.md +++ b/sdk_compliance_adapter/CONTRIBUTING.md @@ -4,7 +4,16 @@ This package contains the PostHog Python SDK compliance adapter used with the Po ## Running tests -Tests run automatically in CI via GitHub Actions. +Tests run automatically in CI via GitHub Actions against harness **1.1.0**, for both capture protocols. Both jobs opt into the local-evaluation suite and fail on compliance regressions. + +Run adapter protocol tests from the repository root in an activated virtual environment: + +```bash +python -m pip install -e . -r sdk_compliance_adapter/requirements.txt pytest pytest-timeout pytest-asyncio +python -m pytest sdk_compliance_adapter/test_adapter.py --timeout=30 +``` + +These tests exercise the real SDK loader/evaluator with controlled transports, including failed/late reloads, local false versus inconclusive, and forced remote evaluation. ### Locally with Docker Compose @@ -35,7 +44,7 @@ docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-pyt docker run --rm \ --name test-harness \ --network test-network \ - ghcr.io/posthog/sdk-test-harness:0.10.0 \ + ghcr.io/posthog/sdk-test-harness:1.1.0 \ run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081 # Cleanup diff --git a/sdk_compliance_adapter/README.md b/sdk_compliance_adapter/README.md index 5cd730c3e..a204fc59e 100644 --- a/sdk_compliance_adapter/README.md +++ b/sdk_compliance_adapter/README.md @@ -24,6 +24,8 @@ The adapter implements the standard SDK adapter interface defined in the [test h - `POST /capture` - Capture an event - `POST /flush` - Flush pending events - `GET /state` - Return internal state +- `POST /get_feature_flag` - Evaluate a flag locally or remotely +- `POST /reload_feature_flag_definitions` - Fresh, bounded definitions readiness barrier - `POST /reset` - Reset SDK state ### Key Implementation Details @@ -34,6 +36,31 @@ The adapter implements the standard SDK adapter interface defined in the [test h **UUID Tracking**: Extracts and tracks UUIDs from batches to verify deduplication. +### Local feature flag evaluation + +Both capture adapters advertise `feature_flags_local_evaluation_v1` for harness +**1.1.0**. The capability versions the adapter protocol and tests both legacy and +explicit property matching; it does not change the SDK's default matching mode. + +- `/init` maps optional `personal_api_key` to the SDK's `secret_key`. Ordinary + capture/remote tests do not need it. Background polling is disabled in the + adapter; explicit reloads still use the real SDK definitions loader. +- `/reload_feature_flag_definitions` takes `timeout_ms` (default 5000, range + 1–30000). It waits for a fresh successful publication, not merely an existing + snapshot. Failed fetches and authorization/quota resets return `ready: false`. + A timeout returns HTTP 504; the SDK's in-flight request may finish later on its + original Client, and another reload on that Client is rejected while it runs. +- `/get_feature_flag` with `only_evaluate_locally: true` uses the SDK's local-only + result API without emitting flag-called events. A conclusive false has + `locally_evaluated: true`; an inconclusive result has `value: null`, + `success: false`, and `locally_evaluated: false`, never remote fallback. +- `force_remote: true` conflicts with local-only mode. When definitions are + enabled, forced remote calls use a separate definitions-free SDK Client so + they cannot accidentally resolve from local rules. Legacy remote responses + and flag-called events are preserved. Reset disposes both Clients. + +The adapter is sequential (it does not advertise parallel-test support). + ## Documentation For complete documentation on the test harness and how to implement adapters, see: diff --git a/sdk_compliance_adapter/adapter.py b/sdk_compliance_adapter/adapter.py index b9c0de33d..0d803ae52 100644 --- a/sdk_compliance_adapter/adapter.py +++ b/sdk_compliance_adapter/adapter.py @@ -78,25 +78,26 @@ def __init__(self): self.last_error: Optional[str] = None self.requests_made: List[RequestInfo] = [] self.client: Optional[Client] = None + self.remote_client: Client | None = None + self.reload_thread: threading.Thread | None = None self.retry_attempts: Dict[str, int] = {} # Track retry attempts by batch ID def reset(self): """Reset all state""" - client_to_shutdown = None with self.lock: - client_to_shutdown = self.client + clients_to_shutdown = (self.client, self.remote_client) self.client = None - - if client_to_shutdown: - # Flush and shutdown the existing client outside state.lock. - # The patched transport records successful flush requests through - # SDKState.record_request(), which also needs state.lock. Holding the - # lock while shutdown() waits for the queue to drain can deadlock when - # a pending background event is being flushed during test reset. - try: - client_to_shutdown.shutdown() - except Exception as e: - logger.warning(f"Error shutting down client: {e}") + self.remote_client = None + # A timed-out load only owns its old Client, never a replacement. + self.reload_thread = None + + for client in clients_to_shutdown: + if client: + # Flush outside state.lock: transport instrumentation needs it. + try: + client.shutdown() + except Exception as e: + logger.warning(f"Error shutting down client: {e}") with self.lock: self.pending_events = 0 @@ -309,6 +310,7 @@ def health(): if is_v1() else ["capture_v0", "capture_ai_v0", "encoding_gzip"] ) + capabilities.append("feature_flags_local_evaluation_v1") return jsonify( { "sdk_name": "posthog-python", @@ -352,21 +354,29 @@ def init(): # One adapter process speaks one capture protocol, selected by CAPTURE_MODE. capture_mode = "v1" if is_v1() else "v0" - # Create client - client = Client( - project_api_key=api_key, - host=host, - flush_at=flush_at, - flush_interval=flush_interval, - gzip=enable_compression, - max_retries=max_retries, - debug=False, - disable_geoip=disable_geoip, - historical_migration=historical_migration, - capture_mode=capture_mode, - ) - + # Explicit reloads exercise the real loader without background polling + # racing the harness's per-test definition snapshots. + client_options = { + "project_api_key": api_key, + "host": host, + "flush_at": flush_at, + "flush_interval": flush_interval, + "gzip": enable_compression, + "max_retries": max_retries, + "debug": False, + "disable_geoip": disable_geoip, + "historical_migration": historical_migration, + "capture_mode": capture_mode, + "enable_local_evaluation": False, + } + personal_api_key = data.get("personal_api_key") + client = Client(**client_options, secret_key=personal_api_key) state.client = client + if personal_api_key: + # The SDK has no force-remote switch once definitions are loaded. + # A definitions-free Client preserves the real remote API and its + # event side effects without mutating the local Client's snapshot. + state.remote_client = Client(**client_options) logger.info( f"Initialized SDK with api_key={api_key[:10]}..., host={host}, " @@ -562,6 +572,70 @@ def get_state(): return jsonify({"error": str(e)}), 500 +@app.route("/reload_feature_flag_definitions", methods=["POST"]) +def reload_feature_flag_definitions(): + """Bound a fresh SDK load and acknowledge only its successful publication.""" + data = request.json or {} + timeout_ms = data.get("timeout_ms", 5000) + if type(timeout_ms) is not int or not 1 <= timeout_ms <= 30000: + return jsonify(success=False, ready=False, error="Invalid timeout_ms"), 400 + + errors = [] + with state.lock: + client = state.client + if client is None or not client.personal_api_key: + return jsonify( + success=False, ready=False, error="A personal_api_key is required" + ), 400 + if state.reload_thread and state.reload_thread.is_alive(): + return jsonify( + success=False, + ready=False, + error="A definitions reload is still running", + ), 409 + previous_generation = client._flag_definition_published_generation + + def load(): + try: + client.load_feature_flags() + except Exception as error: + logger.exception("Error reloading feature flag definitions") + errors.append(str(error)) + + worker = threading.Thread(target=load, daemon=True) + state.reload_thread = worker + worker.start() + + # The SDK's definitions transport timeout is longer than the adapter's + # deadline. Do not block this endpoint on it or start overlapping reloads. + worker.join(timeout_ms / 1000) + if worker.is_alive(): + return jsonify( + success=False, ready=False, error="Definitions reload timed out" + ), 504 + if errors: + return jsonify(success=False, ready=False, error=errors[0]), 502 + with state.lock: + if state.client is not client: + return jsonify( + success=False, ready=False, error="SDK reset during reload" + ), 409 + # load_feature_flags returns None even on failure. Its publication generation + # advances on successful GET/304 and auth/quota resets; the latter clear the + # fingerprint. Checking both avoids acknowledging stale or reset definitions. + with client._flag_definition_publication_lock: + ready = ( + client._flag_definition_published_generation > previous_generation + and bool(client._flag_definition_fingerprint) + and client.feature_flags is not None + ) + if not ready: + return jsonify( + success=False, ready=False, error="Fresh definitions were not loaded" + ), 502 + return jsonify(success=True, ready=True) + + @app.route("/get_feature_flag", methods=["POST"]) def get_feature_flag(): """Evaluate a feature flag""" @@ -577,14 +651,40 @@ def get_feature_flag(): groups = data.get("groups") group_properties = data.get("group_properties") disable_geoip = data.get("disable_geoip") - force_remote = data.get("force_remote", True) + only_evaluate_locally = data.get("only_evaluate_locally", False) + force_remote = data.get("force_remote", not only_evaluate_locally) + if only_evaluate_locally and force_remote: + return jsonify( + {"error": "only_evaluate_locally conflicts with force_remote"} + ), 400 if not key: return jsonify({"error": "key is required"}), 400 if not distinct_id: return jsonify({"error": "distinct_id is required"}), 400 - value = state.client.get_feature_flag( + if only_evaluate_locally: + result = state.client.get_feature_flag_result( + key, + distinct_id, + person_properties=person_properties, + groups=groups, + group_properties=group_properties, + disable_geoip=disable_geoip, + only_evaluate_locally=True, + send_feature_flag_events=False, + ) + # The real local-only SDK API returns None for inconclusive results; + # a conclusive false is a FeatureFlagResult, not a cache miss. + conclusive = result is not None + return jsonify( + success=conclusive, + value=result.get_value() if result is not None else None, + locally_evaluated=conclusive, + ) + + client = (state.remote_client or state.client) if force_remote else state.client + value = client.get_feature_flag( key, distinct_id, person_properties=person_properties, @@ -598,7 +698,7 @@ def get_feature_flag(): # the adapter action returns. Otherwise the harness may reset mock-server # state for the next test while the background consumer is still flushing, # leaking the previous test's event into the next test. - state.client.flush() + client.flush() logger.info(f"Feature flag {key} for {distinct_id}: {value}") diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index e6519de56..2f9ded30e 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -23,7 +23,7 @@ services: # Test harness test-harness: - image: ghcr.io/posthog/sdk-test-harness:0.10.0 + image: ghcr.io/posthog/sdk-test-harness:1.1.0 command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"] networks: - test-network diff --git a/sdk_compliance_adapter/test_adapter.py b/sdk_compliance_adapter/test_adapter.py new file mode 100644 index 000000000..8acc78386 --- /dev/null +++ b/sdk_compliance_adapter/test_adapter.py @@ -0,0 +1,327 @@ +"""Adapter protocol tests, run separately from the SDK's optional-dependency suite.""" + +import importlib.util +import threading +import time +from pathlib import Path +from unittest.mock import Mock + +import pytest + +import posthog.capture_v1 +import posthog.client +import posthog.consumer +import posthog.request +from posthog.request import APIError, GetResponse + + +@pytest.fixture +def adapter(monkeypatch): + # Importing the adapter installs transport instrumentation. Restore it after + # every test so collecting these tests alongside SDK tests is safe. + for module, name in [ + (posthog.request, "batch_post"), + (posthog.consumer, "batch_post"), + (posthog.capture_v1, "_post_v1"), + ]: + monkeypatch.setattr(module, name, getattr(module, name)) + spec = importlib.util.spec_from_file_location( + "compliance_adapter_test", Path(__file__).with_name("adapter.py") + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + module.app.config["TESTING"] = True + yield module + module.state.reset() + + +def definitions(version=1): + return { + "flags": [ + { + "id": 1, + "key": "flag", + "active": True, + "filters": { + "groups": [ + { + "rollout_percentage": 100, + "properties": [ + { + "key": "plan", + "value": False, + "operator": "exact", + "type": "person", + } + ], + } + ] + }, + } + ], + "cohorts": {}, + "group_type_mapping": {}, + "property_matching_version": version, + } + + +def initialize(adapter, **overrides): + config = { + "api_key": "phc_test_key", + "host": "http://127.0.0.1:1", + "personal_api_key": "phx_test_key", + } + config.update(overrides) + response = adapter.app.test_client().post("/init", json=config) + assert response.status_code == 200 + assert response.json["success"] is True + return adapter.app.test_client() + + +@pytest.mark.parametrize("mode,capability", [("", "capture_v0"), ("v1", "capture_v1")]) +def test_health_opts_into_local_evaluation_without_losing_capture( + adapter, monkeypatch, mode, capability +): + monkeypatch.setattr(adapter, "CAPTURE_MODE", mode) + capabilities = adapter.app.test_client().get("/health").json["capabilities"] + assert "feature_flags_local_evaluation_v1" in capabilities + assert capability in capabilities + assert "capture_ai_v0" in capabilities + + +def test_init_enables_explicit_definitions_loading_without_polling(adapter): + initialize(adapter) + assert adapter.state.client.personal_api_key == "phx_test_key" + assert adapter.state.client.enable_local_evaluation is False + assert adapter.state.client.poller is None + assert adapter.state.remote_client.personal_api_key is None + + +def test_remote_only_init_does_not_require_a_privileged_key(adapter): + initialize(adapter, personal_api_key=None) + assert adapter.state.client.personal_api_key is None + assert adapter.state.remote_client is None + + +@pytest.mark.parametrize("version,expected", [(1, True), (2, False)]) +def test_reload_and_conclusive_local_result(adapter, monkeypatch, version, expected): + client = initialize(adapter) + get = Mock(return_value=GetResponse(data=definitions(version))) + monkeypatch.setattr(posthog.client, "get", get) + remote = Mock(side_effect=AssertionError("Local-only must never request /flags")) + monkeypatch.setattr(adapter.state.client, "_get_flags_decision", remote) + assert client.post("/reload_feature_flag_definitions", json={}).json == { + "success": True, + "ready": True, + } + response = client.post( + "/get_feature_flag", + json={ + "key": "flag", + "distinct_id": "user", + "person_properties": {"plan": "banana"}, + "only_evaluate_locally": True, + }, + ) + assert response.json == { + "success": True, + "value": expected, + "locally_evaluated": True, + } + get.assert_called_once() + assert get.call_args.args[0] == "phx_test_key" + assert get.call_args.args[1].startswith("/flags/definitions?token=phc_test_key") + assert adapter.state.client.poller is None + remote.assert_not_called() + + +def test_inconclusive_local_result_is_not_reported_as_false(adapter, monkeypatch): + client = initialize(adapter) + monkeypatch.setattr( + posthog.client, "get", Mock(return_value=GetResponse(data=definitions())) + ) + client.post("/reload_feature_flag_definitions", json={}) + remote = Mock(side_effect=AssertionError("Unexpected /flags fallback")) + monkeypatch.setattr(adapter.state.client, "_get_flags_decision", remote) + response = client.post( + "/get_feature_flag", + json={"key": "flag", "distinct_id": "user", "only_evaluate_locally": True}, + ) + assert response.json["success"] is False + assert response.json["value"] is None + assert response.json["locally_evaluated"] is False + remote.assert_not_called() + + +def test_force_remote_bypasses_loaded_local_definitions(adapter, monkeypatch): + client = initialize(adapter) + monkeypatch.setattr( + posthog.client, "get", Mock(return_value=GetResponse(data=definitions())) + ) + client.post("/reload_feature_flag_definitions", json={}) + remote = Mock( + return_value=posthog.client.normalize_flags_response( + {"featureFlags": {"flag": False}} + ) + ) + monkeypatch.setattr(adapter.state.remote_client, "_get_flags_decision", remote) + monkeypatch.setattr(adapter.state.remote_client, "capture", Mock()) + response = client.post( + "/get_feature_flag", + json={ + "key": "flag", + "distinct_id": "user", + "person_properties": {"plan": "banana"}, + "force_remote": True, + }, + ) + assert response.json == {"success": True, "value": False} + remote.assert_called_once() + + +def test_rejects_conflicting_evaluation_modes(adapter): + client = initialize(adapter) + response = client.post( + "/get_feature_flag", + json={ + "key": "flag", + "distinct_id": "user", + "only_evaluate_locally": True, + "force_remote": True, + }, + ) + assert response.status_code == 400 + + +@pytest.mark.parametrize("timeout", [0, -1, 30001, True, "100", None]) +def test_reload_validates_deadline(adapter, timeout): + client = initialize(adapter) + assert ( + client.post( + "/reload_feature_flag_definitions", json={"timeout_ms": timeout} + ).status_code + == 400 + ) + + +def test_reload_requires_client_and_privileged_key(adapter): + client = adapter.app.test_client() + assert client.post("/reload_feature_flag_definitions", json={}).status_code == 400 + initialize(adapter, personal_api_key=None) + assert client.post("/reload_feature_flag_definitions", json={}).status_code == 400 + + +@pytest.mark.parametrize("status", [401, 402, 500]) +def test_failed_reload_does_not_report_previous_snapshot_ready( + adapter, monkeypatch, status +): + client = initialize(adapter) + get = Mock(return_value=GetResponse(data=definitions())) + monkeypatch.setattr(posthog.client, "get", get) + assert ( + client.post("/reload_feature_flag_definitions", json={}).json["ready"] is True + ) + get.side_effect = APIError(status, "definitions unavailable") + response = client.post("/reload_feature_flag_definitions", json={}) + assert response.status_code == 502 + assert response.json["ready"] is False + assert response.json["success"] is False + assert get.call_count == 2 + + +def test_reload_is_bounded_and_does_not_overlap_requests(adapter, monkeypatch): + client = initialize(adapter) + release = threading.Event() + get = Mock( + side_effect=lambda *args, **kwargs: ( + release.wait(2), + GetResponse(data=definitions()), + )[1] + ) + monkeypatch.setattr(posthog.client, "get", get) + try: + start = time.monotonic() + response = client.post( + "/reload_feature_flag_definitions", json={"timeout_ms": 10} + ) + assert response.status_code == 504 + assert response.json["ready"] is False + assert time.monotonic() - start < 1 + assert ( + client.post("/reload_feature_flag_definitions", json={}).status_code == 409 + ) + get.assert_called_once() + finally: + release.set() + thread = getattr(adapter.state, "reload_thread", None) + if thread: + thread.join(timeout=3) + + +def test_reset_disposes_both_clients_and_clears_reload_state(adapter, monkeypatch): + client = initialize(adapter) + local = adapter.state.client + remote = adapter.state.remote_client + local_shutdown = Mock(wraps=local.shutdown) + remote_shutdown = Mock(wraps=remote.shutdown) + monkeypatch.setattr(local, "shutdown", local_shutdown) + monkeypatch.setattr(remote, "shutdown", remote_shutdown) + assert client.post("/reset").json == {"success": True} + local_shutdown.assert_called_once() + remote_shutdown.assert_called_once() + assert adapter.state.client is None + assert adapter.state.remote_client is None + assert adapter.state.reload_thread is None + + +def test_reload_refreshes_even_when_definitions_are_empty(adapter, monkeypatch): + client = initialize(adapter) + empty = definitions() + empty["flags"] = [] + get = Mock(return_value=GetResponse(data=empty)) + monkeypatch.setattr(posthog.client, "get", get) + for _ in range(2): + assert client.post("/reload_feature_flag_definitions", json={}).json == { + "success": True, + "ready": True, + } + assert get.call_count == 2 + + +def test_timed_out_reload_cannot_publish_into_replacement_client(adapter, monkeypatch): + client = initialize(adapter) + old_client = adapter.state.client + release = threading.Event() + monkeypatch.setattr( + posthog.client, + "get", + Mock( + side_effect=lambda *args, **kwargs: ( + release.wait(2), + GetResponse(data=definitions(2)), + )[1] + ), + ) + thread = None + try: + assert ( + client.post( + "/reload_feature_flag_definitions", json={"timeout_ms": 10} + ).status_code + == 504 + ) + thread = adapter.state.reload_thread + initialize(adapter) + new_client = adapter.state.client + release.set() + thread.join(timeout=3) + assert not thread.is_alive() + assert old_client.feature_flags is not None + assert new_client is not old_client + assert new_client.feature_flags is None + assert new_client.poller is None + finally: + release.set() + if thread: + thread.join(timeout=3) From 8939d3fa4de04e984ebb3e870db391809716c5f2 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Wed, 9 Sep 2026 08:53:02 +0200 Subject: [PATCH 7/7] test: upgrade SDK compliance harness to 1.1.1 --- .github/workflows/sdk-compliance.yml | 8 ++++---- sdk_compliance_adapter/CONTRIBUTING.md | 4 ++-- sdk_compliance_adapter/README.md | 2 +- sdk_compliance_adapter/docker-compose.yml | 2 +- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index 1108654c6..89b2eb51c 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -29,20 +29,20 @@ jobs: compliance: name: PostHog SDK compliance tests (capture v0) - uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@15011d6868ee73e11333860b6392d35298b92535 # 1.1.0 + uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@1d6b197ef46758f577f470535bb3d70e37a15fa2 # 1.1.1 with: adapter-dockerfile: "sdk_compliance_adapter/Dockerfile" adapter-context: "." - test-harness-version: "1.1.0" + test-harness-version: "1.1.1" continue-on-error: false report-name: "sdk-compliance-report-v0" compliance-v1: name: PostHog SDK compliance tests (capture v1) - uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@15011d6868ee73e11333860b6392d35298b92535 # 1.1.0 + uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@1d6b197ef46758f577f470535bb3d70e37a15fa2 # 1.1.1 with: adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1" adapter-context: "." - test-harness-version: "1.1.0" + test-harness-version: "1.1.1" continue-on-error: false report-name: "sdk-compliance-report-v1" diff --git a/sdk_compliance_adapter/CONTRIBUTING.md b/sdk_compliance_adapter/CONTRIBUTING.md index f76cab02a..3033609b3 100644 --- a/sdk_compliance_adapter/CONTRIBUTING.md +++ b/sdk_compliance_adapter/CONTRIBUTING.md @@ -4,7 +4,7 @@ This package contains the PostHog Python SDK compliance adapter used with the Po ## Running tests -Tests run automatically in CI via GitHub Actions against harness **1.1.0**, for both capture protocols. Both jobs opt into the local-evaluation suite and fail on compliance regressions. +Tests run automatically in CI via GitHub Actions against harness **1.1.1**, for both capture protocols. Both jobs opt into the local-evaluation suite and fail on compliance regressions. Run adapter protocol tests from the repository root in an activated virtual environment: @@ -44,7 +44,7 @@ docker run -d --name sdk-adapter --network test-network -p 8080:8080 posthog-pyt docker run --rm \ --name test-harness \ --network test-network \ - ghcr.io/posthog/sdk-test-harness:1.1.0 \ + ghcr.io/posthog/sdk-test-harness:1.1.1 \ run --adapter-url http://sdk-adapter:8080 --mock-url http://test-harness:8081 # Cleanup diff --git a/sdk_compliance_adapter/README.md b/sdk_compliance_adapter/README.md index a204fc59e..eae25916b 100644 --- a/sdk_compliance_adapter/README.md +++ b/sdk_compliance_adapter/README.md @@ -39,7 +39,7 @@ The adapter implements the standard SDK adapter interface defined in the [test h ### Local feature flag evaluation Both capture adapters advertise `feature_flags_local_evaluation_v1` for harness -**1.1.0**. The capability versions the adapter protocol and tests both legacy and +**1.1.1**. The capability versions the adapter protocol and tests both legacy and explicit property matching; it does not change the SDK's default matching mode. - `/init` maps optional `personal_api_key` to the SDK's `secret_key`. Ordinary diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index 2f9ded30e..c679f5c4f 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -23,7 +23,7 @@ services: # Test harness test-harness: - image: ghcr.io/posthog/sdk-test-harness:1.1.0 + image: ghcr.io/posthog/sdk-test-harness:1.1.1 command: ["run", "--adapter-url", "http://sdk-adapter:8080", "--mock-url", "http://test-harness:8081"] networks: - test-network