diff --git a/.sampo/changesets/versioned-property-matching.md b/.sampo/changesets/versioned-property-matching.md new file mode 100644 index 000000000..9abebdb7d --- /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. 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 59ccb9dca..07c931d4f 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 @@ -130,6 +131,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 +885,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 @@ -891,11 +898,17 @@ 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 + # 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 + ) self._flags_etag: Optional[str] = None 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 +2247,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 @@ -2243,10 +2256,20 @@ 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): 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_definition_fingerprint + ) reset_sessions() @@ -2868,28 +2891,63 @@ 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.""" - 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 - ) + """Publish definitions and their result-cache identity together.""" + fingerprint = _fingerprint or self._hash_flag_definitions(data) + with self._flag_definition_publication_lock: + 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 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 + # 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 @@ -2962,6 +3020,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( @@ -2997,7 +3062,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: @@ -3006,6 +3073,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,12 +3117,18 @@ def _fetch_feature_flags_from_api(self): self.feature_flags = [] 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 if self.flag_cache: - self.flag_cache.clear() + self.flag_definition_version += 1 + self.flag_cache._advance_generation( + self.flag_definition_version, + self._flag_definition_fingerprint, + ) if self.debug: raise APIError(status=401, message=detail) @@ -3066,13 +3140,19 @@ def _fetch_feature_flags_from_api(self): self.feature_flags = [] 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 - # 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, + self._flag_definition_fingerprint, + ) if self.debug: raise APIError( @@ -3103,14 +3183,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 +3219,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 +3243,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 +3278,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 +3296,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 +3429,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, @@ -3365,8 +3459,10 @@ def _get_feature_flag_result( key, flag_value, self._compute_payload_locally(key, flag_value) ) 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, self.flag_definition_version + distinct_id, key, cached_flag_result, local_definition_version ) elif only_evaluate_locally: if self.feature_flags is None: @@ -3403,10 +3499,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( @@ -3592,17 +3689,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 +3707,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 +3718,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 +4413,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 +4434,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_cache_publication.py b/posthog/test/test_flag_cache_publication.py new file mode 100644 index 000000000..52b61829e --- /dev/null +++ b/posthog/test/test_flag_cache_publication.py @@ -0,0 +1,248 @@ +"""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", client.flag_definition_version - 1) + 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_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) + client._update_flag_state(definitions(2)) + 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", new_version) == "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): + # 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 + 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_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..5af52fd27 --- /dev/null +++ b/posthog/test/test_property_matching_version.py @@ -0,0 +1,386 @@ +"""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 + ) + 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 ( + 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/posthog/test/test_redis_flag_cache_snapshot.py b/posthog/test/test_redis_flag_cache_snapshot.py new file mode 100644 index 000000000..bac587221 --- /dev/null +++ b/posthog/test/test_redis_flag_cache_snapshot.py @@ -0,0 +1,353 @@ +"""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 + # Without verified definitions, a new Client cannot reuse this provenance. + assert fallback(workers()) is None + + +@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 + + +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 diff --git a/posthog/test/test_utils.py b/posthog/test/test_utils.py index e075ff4a5..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" @@ -705,6 +727,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 @@ -919,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( @@ -946,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 e02037af5..5070afcf8 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 @@ -216,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 @@ -232,6 +234,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, fingerprint=None): + # 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 +256,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 +278,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 +288,28 @@ 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() + + # 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. + 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,11 +372,31 @@ 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, 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 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() @@ -364,6 +408,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 @@ -381,11 +427,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 @@ -397,10 +445,17 @@ 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 - ): - return entry.flag_result + 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 + ) + if valid: + return entry.flag_result return None except Exception: @@ -417,7 +472,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_entry_current(entry) + and entry.is_stale_but_usable(time.time(), max_stale_age) + ): return entry.flag_result return None @@ -430,15 +489,27 @@ 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 + 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 + flag_result, flag_definition_version, fingerprint=fingerprint ) - # 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 + # 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) except Exception: # Redis error - silently fail, don't break flag evaluation 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