From afddb65839ec377934d31f5ec5648383550a103e Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Thu, 30 Jul 2026 16:39:05 +0200 Subject: [PATCH 1/3] fix(flags): prevent stale definition publication --- .sampo/changesets/ardent-baroness-tapio.md | 5 + posthog/client.py | 192 +++++++++++------- posthog/test/test_client_fork.py | 20 ++ posthog/test/test_feature_flags.py | 220 +++++++++++++++++++++ 4 files changed, 369 insertions(+), 68 deletions(-) create mode 100644 .sampo/changesets/ardent-baroness-tapio.md diff --git a/.sampo/changesets/ardent-baroness-tapio.md b/.sampo/changesets/ardent-baroness-tapio.md new file mode 100644 index 00000000..da4c7e56 --- /dev/null +++ b/.sampo/changesets/ardent-baroness-tapio.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Prevent stale feature flag definition publication diff --git a/posthog/client.py b/posthog/client.py index 9469407b..9f96315a 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -627,6 +627,11 @@ def __init__( self.flag_cache = self._initialize_flag_cache(flag_fallback_cache_url) self.flag_definition_version = 0 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_cache_write_lock = threading.RLock() self._flag_definition_cache_provider = flag_definition_cache_provider self._flag_definition_cache_provider_async_runner: Optional[ _BackgroundEventLoopRunner @@ -1864,6 +1869,11 @@ def _reinit_after_fork(self): self._flag_definition_cache_provider_async_runner = None self._flag_definition_cache_provider_async_runner_lock = threading.Lock() + # A parent thread may have been publishing or caching flag definitions at + # fork time. + self._flag_definition_publication_lock = threading.Lock() + self._flag_definition_cache_write_lock = threading.RLock() + # Metrics locks may have been held by a parent thread at fork time; replace # them (never acquire them) so the child can't deadlock on a vanished holder. self._metrics_lock = threading.Lock() @@ -2216,91 +2226,137 @@ def _fetch_feature_flags_from_api(self): ) return - try: - # Store old flags to detect changes - old_flags_by_key: dict[str, dict] = self.feature_flags_by_key or {} + with self._flag_definition_publication_lock: + self._flag_definition_fetch_generation += 1 + fetch_generation = self._flag_definition_fetch_generation + request_etag = self._flags_etag + cache_data_to_store: Optional[FlagDefinitionCacheData] = None + try: response = get( personal_api_key, f"/flags/definitions?token={self.api_key}&send_cohorts", self.host, timeout=10, - etag=self._flags_etag, + etag=request_etag, ) - # Update stored ETag (clear if server stops sending one) - self._flags_etag = response.etag - - # If 304 Not Modified, flags haven't changed - skip processing - if response.not_modified: - self.log.debug( - "[FEATURE FLAGS] Flags not modified (304), using cached data" - ) - self._last_feature_flag_poll = datetime.now(tz=timezone.utc) - return - - if response.data is None: - self.log.error( - "[FEATURE FLAGS] Unexpected empty response data in non-304 response" - ) - return - - self._update_flag_state(response.data, old_flags_by_key=old_flags_by_key) + with self._flag_definition_publication_lock: + if fetch_generation <= self._flag_definition_published_generation: + self.log.debug( + "[FEATURE FLAGS] Ignoring stale flag definition response" + ) + self._last_feature_flag_poll = datetime.now(tz=timezone.utc) + return - # Store in external cache if provider is configured - if self._flag_definition_cache_provider: - try: - self._resolve_flag_definition_cache_provider_result( - self._flag_definition_cache_provider.on_flag_definitions_received( - { - "flags": self.feature_flags or [], - "group_type_mapping": self.group_type_mapping or {}, - "cohorts": self.cohorts or {}, - "minimal_flag_called_events": self._minimal_flag_called_events, - } + # A 304 is valid only for the ETag used by this request. Another + # overlapping response may already have installed newer definitions. + if response.not_modified: + if self._flags_etag != request_etag: + self.log.debug( + "[FEATURE FLAGS] Ignoring stale 304 flag definition response" ) - ) - except Exception as e: - self.log.error(f"[FEATURE FLAGS] Cache provider store error: {e}") - # Flags are already in memory, so continue normally + self._last_feature_flag_poll = datetime.now(tz=timezone.utc) + return - except APIError as e: - if e.status == 401: - detail = ( - f"Error loading feature flags: {e.message}. " - "Please verify both your project_api_key and secret_key. " - "More information: https://posthog.com/docs/api/overview" - ) - self.log.error("[FEATURE FLAGS] %s", detail) - self.feature_flags = [] - self.group_type_mapping = {} - self.cohorts = {} + self.log.debug( + "[FEATURE FLAGS] Flags not modified (304), using cached data" + ) + self._last_feature_flag_poll = datetime.now(tz=timezone.utc) + return - if self.flag_cache: - self.flag_cache.clear() + if response.data is None: + self.log.error( + "[FEATURE FLAGS] Unexpected empty response data in non-304 response" + ) + return - if self.debug: - raise APIError(status=401, message=detail) - elif e.status == 402: - self.log.warning( - "[FEATURE FLAGS] PostHog feature flags quota limited, resetting feature flag data. Learn more about billing limits at https://posthog.com/docs/billing/limits-alerts" + 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 ) - # Reset all feature flag data when quota limited - self.feature_flags = [] - self.group_type_mapping = {} - self.cohorts = {} - # Clear flag cache when quota limited - if self.flag_cache: - self.flag_cache.clear() + if self._flag_definition_cache_provider: + cache_data_to_store = { + "flags": self.feature_flags or [], + "group_type_mapping": self.group_type_mapping or {}, + "cohorts": self.cohorts or {}, + "minimal_flag_called_events": self._minimal_flag_called_events, + } + + # Publish the ETag only after its matching flag state is installed. + self._flags_etag = response.etag + self._flag_definition_published_generation = fetch_generation + self._flag_definition_cache_generation = fetch_generation + + if cache_data_to_store and self._flag_definition_cache_provider: + # Keep provider I/O out of the publication lock. The separate lock + # preserves cache write order without delaying newer API fetches or + # in-memory publication. + with self._flag_definition_cache_write_lock: + with self._flag_definition_publication_lock: + should_store = ( + fetch_generation == self._flag_definition_cache_generation + ) + if should_store: + try: + self._resolve_flag_definition_cache_provider_result( + self._flag_definition_cache_provider.on_flag_definitions_received( + cache_data_to_store + ) + ) + except Exception as e: + self.log.error( + f"[FEATURE FLAGS] Cache provider store error: {e}" + ) + # Flags are already in memory, so continue normally - if self.debug: - raise APIError( - status=402, - message="PostHog feature flags quota limited", + except APIError as e: + with self._flag_definition_publication_lock: + if fetch_generation <= self._flag_definition_published_generation: + self.log.debug("[FEATURE FLAGS] Ignoring stale API error response") + elif e.status == 401: + detail = ( + f"Error loading feature flags: {e.message}. " + "Please verify both your project_api_key and secret_key. " + "More information: https://posthog.com/docs/api/overview" ) - else: - self.log.error(f"[FEATURE FLAGS] Error loading feature flags: {e}") + self.log.error("[FEATURE FLAGS] %s", detail) + self.feature_flags = [] + self.group_type_mapping = {} + self.cohorts = {} + 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() + + if self.debug: + raise APIError(status=401, message=detail) + elif e.status == 402: + self.log.warning( + "[FEATURE FLAGS] PostHog feature flags quota limited, resetting feature flag data. Learn more about billing limits at https://posthog.com/docs/billing/limits-alerts" + ) + # Reset all feature flag data when quota limited + self.feature_flags = [] + self.group_type_mapping = {} + self.cohorts = {} + self._flags_etag = None + self._flag_definition_published_generation = fetch_generation + self._flag_definition_cache_generation = fetch_generation + + # Clear flag cache when quota limited + if self.flag_cache: + self.flag_cache.clear() + + if self.debug: + raise APIError( + status=402, + message="PostHog feature flags quota limited", + ) + else: + self.log.error(f"[FEATURE FLAGS] Error loading feature flags: {e}") except Exception as e: self.log.warning( "[FEATURE FLAGS] Fetching feature flags failed with following error. We will retry in %s seconds." diff --git a/posthog/test/test_client_fork.py b/posthog/test/test_client_fork.py index 294dcffb..ece9d53f 100644 --- a/posthog/test/test_client_fork.py +++ b/posthog/test/test_client_fork.py @@ -92,6 +92,26 @@ def test_reinit_after_fork_clears_poller_when_local_evaluation_disabled(self): self.assertIsNone(client.poller) + def test_reinit_after_fork_replaces_flag_definition_locks(self): + client = Client(FAKE_TEST_API_KEY, send=False) + old_publication_lock = client._flag_definition_publication_lock + old_cache_write_lock = client._flag_definition_cache_write_lock + old_publication_lock.acquire() + old_cache_write_lock.acquire() + + try: + client._reinit_after_fork() + for new_lock, old_lock in ( + (client._flag_definition_publication_lock, old_publication_lock), + (client._flag_definition_cache_write_lock, old_cache_write_lock), + ): + self.assertIsNot(new_lock, old_lock) + self.assertTrue(new_lock.acquire(blocking=False)) + new_lock.release() + finally: + old_cache_write_lock.release() + old_publication_lock.release() + @mock.patch("posthog.client.reset_sessions") def test_reinit_after_fork_resets_sessions(self, mock_reset_sessions): client = Client(FAKE_TEST_API_KEY, send=False) diff --git a/posthog/test/test_feature_flags.py b/posthog/test/test_feature_flags.py index 497069ef..14773a85 100644 --- a/posthog/test/test_feature_flags.py +++ b/posthog/test/test_feature_flags.py @@ -1,4 +1,5 @@ import datetime +import threading import unittest from unittest import mock @@ -3233,6 +3234,190 @@ def test_load_feature_flags_etag_updated_on_new_response( self.assertEqual(client._flags_etag, '"etag-v2"') self.assertEqual(client.feature_flags[0]["key"], "flag-v2") + @mock.patch("posthog.client.get") + def test_load_feature_flags_ignores_older_response_that_finishes_last( + self, patch_get + ): + first_request_started = threading.Event() + release_first_request = threading.Event() + second_load_finished = threading.Event() + call_count_lock = threading.Lock() + call_count = 0 + + old_response = GetResponse( + data={ + "flags": [{"id": 1, "key": "old-flag", "active": True}], + "group_type_mapping": {"0": "old-group"}, + "cohorts": {"old": {"properties": []}}, + "minimal_flag_called_events": False, + }, + etag='"old-etag"', + ) + new_response = GetResponse( + data={ + "flags": [{"id": 2, "key": "new-flag", "active": True}], + "group_type_mapping": {"0": "new-group"}, + "cohorts": {"new": {"properties": []}}, + "minimal_flag_called_events": True, + }, + etag='"new-etag"', + ) + + def get_with_reversed_completion(*args, **kwargs): + nonlocal call_count + with call_count_lock: + request_number = call_count + call_count += 1 + + if request_number == 0: + first_request_started.set() + release_first_request.wait(timeout=5) + return old_response + + return new_response + + patch_get.side_effect = get_with_reversed_completion + client = Client(FAKE_TEST_API_KEY, secret_key="test", send=False) + + def load_second_response(): + client._load_feature_flags() + second_load_finished.set() + + first_thread = threading.Thread(target=client._load_feature_flags) + second_thread = threading.Thread(target=load_second_response) + + first_thread.start() + self.assertTrue(first_request_started.wait(timeout=5)) + second_thread.start() + try: + self.assertTrue(second_load_finished.wait(timeout=5)) + finally: + release_first_request.set() + + first_thread.join(timeout=5) + second_thread.join(timeout=5) + self.assertFalse(first_thread.is_alive()) + self.assertFalse(second_thread.is_alive()) + self.assertEqual(client.feature_flags[0]["key"], "new-flag") + self.assertEqual(client.group_type_mapping, {"0": "new-group"}) + self.assertEqual(client.cohorts, {"new": {"properties": []}}) + self.assertTrue(client._minimal_flag_called_events) + self.assertEqual(client._flags_etag, '"new-etag"') + + @mock.patch("posthog.client.get") + def test_load_feature_flags_304_does_not_suppress_in_flight_update(self, patch_get): + first_request_started = threading.Event() + second_load_finished = threading.Event() + call_count_lock = threading.Lock() + call_count = 0 + + changed_response = GetResponse( + data={ + "flags": [{"id": 2, "key": "new-flag", "active": True}], + "group_type_mapping": {}, + "cohorts": {}, + }, + etag='"etag-b"', + ) + not_modified_response = GetResponse( + data=None, + etag='"etag-a"', + not_modified=True, + ) + + def get_304_before_changed_response(*args, **kwargs): + nonlocal call_count + with call_count_lock: + request_number = call_count + call_count += 1 + + if request_number == 0: + first_request_started.set() + second_load_finished.wait(timeout=5) + return changed_response + return not_modified_response + + patch_get.side_effect = get_304_before_changed_response + client = Client(FAKE_TEST_API_KEY, secret_key="test", send=False) + client._flags_etag = '"etag-a"' + + def load_not_modified_response(): + client._load_feature_flags() + second_load_finished.set() + + first_thread = threading.Thread(target=client._load_feature_flags) + second_thread = threading.Thread(target=load_not_modified_response) + first_thread.start() + self.assertTrue(first_request_started.wait(timeout=5)) + second_thread.start() + first_thread.join(timeout=5) + second_thread.join(timeout=5) + + self.assertFalse(first_thread.is_alive()) + self.assertFalse(second_thread.is_alive()) + self.assertEqual(client.feature_flags[0]["key"], "new-flag") + self.assertEqual(client._flags_etag, '"etag-b"') + + @mock.patch("posthog.client.get") + def test_load_feature_flags_ignores_304_for_superseded_etag(self, patch_get): + first_request_started = threading.Event() + second_request_started = threading.Event() + first_load_finished = threading.Event() + call_count_lock = threading.Lock() + call_count = 0 + + changed_response = GetResponse( + data={ + "flags": [{"id": 2, "key": "new-flag", "active": True}], + "group_type_mapping": {"0": "new-group"}, + "cohorts": {}, + }, + etag='"etag-b"', + ) + stale_not_modified_response = GetResponse( + data=None, + etag='"etag-a"', + not_modified=True, + ) + + def get_changed_then_delayed_304(*args, **kwargs): + nonlocal call_count + with call_count_lock: + request_number = call_count + call_count += 1 + + if request_number == 0: + first_request_started.set() + second_request_started.wait(timeout=5) + return changed_response + + second_request_started.set() + first_load_finished.wait(timeout=5) + return stale_not_modified_response + + patch_get.side_effect = get_changed_then_delayed_304 + client = Client(FAKE_TEST_API_KEY, secret_key="test", send=False) + client._flags_etag = '"etag-a"' + + def load_changed_response(): + client._load_feature_flags() + first_load_finished.set() + + first_thread = threading.Thread(target=load_changed_response) + second_thread = threading.Thread(target=client._load_feature_flags) + + first_thread.start() + self.assertTrue(first_request_started.wait(timeout=5)) + second_thread.start() + first_thread.join(timeout=5) + second_thread.join(timeout=5) + + self.assertFalse(first_thread.is_alive()) + self.assertFalse(second_thread.is_alive()) + self.assertEqual(client.feature_flags[0]["key"], "new-flag") + self.assertEqual(client.group_type_mapping, {"0": "new-group"}) + self.assertEqual(client._flags_etag, '"etag-b"') + @mock.patch("posthog.client.Poller") @mock.patch("posthog.client.get") def test_load_feature_flags_clears_etag_when_server_stops_sending( @@ -3266,6 +3451,41 @@ def test_load_feature_flags_clears_etag_when_server_stops_sending( self.assertIsNone(client._flags_etag) self.assertEqual(client.feature_flags[0]["key"], "flag-v2") + @parameterized.expand([401, 402]) + @mock.patch("posthog.client.get") + def test_load_feature_flags_refetches_after_state_reset(self, status, patch_get): + patch_get.side_effect = [ + GetResponse( + data={ + "flags": [{"id": 1, "key": "old-flag", "active": True}], + "group_type_mapping": {}, + "cohorts": {}, + }, + etag='"old-etag"', + ), + APIError(status, "reset flag definitions"), + GetResponse( + data={ + "flags": [{"id": 2, "key": "new-flag", "active": True}], + "group_type_mapping": {}, + "cohorts": {}, + }, + etag='"new-etag"', + ), + ] + client = Client(FAKE_TEST_API_KEY, secret_key="test", send=False) + + client._load_feature_flags() + client._load_feature_flags() + self.assertEqual(client.feature_flags, []) + self.assertIsNone(client._flags_etag) + + client._load_feature_flags() + + self.assertIsNone(patch_get.call_args_list[2].kwargs["etag"]) + self.assertEqual(client.feature_flags[0]["key"], "new-flag") + self.assertEqual(client._flags_etag, '"new-etag"') + @mock.patch("posthog.client.Poller") @mock.patch("posthog.client.get") def test_load_feature_flags_wrong_key(self, patch_get, _patch_poll): From 5e6963e23e7bfec46c9a3c1731655bed8685d0c5 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Fri, 31 Jul 2026 10:33:45 +0200 Subject: [PATCH 2/3] fix(flags): fence older responses after accepted 304 --- posthog/client.py | 2 ++ posthog/test/test_feature_flags.py | 20 ++++++++++++++++---- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/posthog/client.py b/posthog/client.py index 9f96315a..1ef51bcc 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -2259,6 +2259,8 @@ def _fetch_feature_flags_from_api(self): self._last_feature_flag_poll = datetime.now(tz=timezone.utc) return + self._flag_definition_published_generation = fetch_generation + self._flag_definition_cache_generation = fetch_generation self.log.debug( "[FEATURE FLAGS] Flags not modified (304), using cached data" ) diff --git a/posthog/test/test_feature_flags.py b/posthog/test/test_feature_flags.py index 14773a85..3a87b883 100644 --- a/posthog/test/test_feature_flags.py +++ b/posthog/test/test_feature_flags.py @@ -3305,12 +3305,22 @@ def load_second_response(): self.assertEqual(client._flags_etag, '"new-etag"') @mock.patch("posthog.client.get") - def test_load_feature_flags_304_does_not_suppress_in_flight_update(self, patch_get): + def test_load_feature_flags_accepted_304_suppresses_older_in_flight_update( + self, patch_get + ): first_request_started = threading.Event() second_load_finished = threading.Event() call_count_lock = threading.Lock() call_count = 0 + initial_response = GetResponse( + data={ + "flags": [{"id": 1, "key": "old-flag", "active": True}], + "group_type_mapping": {}, + "cohorts": {}, + }, + etag='"etag-a"', + ) changed_response = GetResponse( data={ "flags": [{"id": 2, "key": "new-flag", "active": True}], @@ -3332,6 +3342,8 @@ def get_304_before_changed_response(*args, **kwargs): call_count += 1 if request_number == 0: + return initial_response + if request_number == 1: first_request_started.set() second_load_finished.wait(timeout=5) return changed_response @@ -3339,7 +3351,7 @@ def get_304_before_changed_response(*args, **kwargs): patch_get.side_effect = get_304_before_changed_response client = Client(FAKE_TEST_API_KEY, secret_key="test", send=False) - client._flags_etag = '"etag-a"' + client._load_feature_flags() def load_not_modified_response(): client._load_feature_flags() @@ -3355,8 +3367,8 @@ def load_not_modified_response(): self.assertFalse(first_thread.is_alive()) self.assertFalse(second_thread.is_alive()) - self.assertEqual(client.feature_flags[0]["key"], "new-flag") - self.assertEqual(client._flags_etag, '"etag-b"') + self.assertEqual(client.feature_flags[0]["key"], "old-flag") + self.assertEqual(client._flags_etag, '"etag-a"') @mock.patch("posthog.client.get") def test_load_feature_flags_ignores_304_for_superseded_etag(self, patch_get): From 789041a402c7b2a572c0144faf98a8afb83dac7c Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Fri, 31 Jul 2026 13:09:52 +0200 Subject: [PATCH 3/3] fix(flags): preserve ETag updates from 304 responses --- posthog/client.py | 1 + posthog/test/test_feature_flags.py | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/posthog/client.py b/posthog/client.py index 1ef51bcc..87425426 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -2259,6 +2259,7 @@ def _fetch_feature_flags_from_api(self): self._last_feature_flag_poll = datetime.now(tz=timezone.utc) return + self._flags_etag = response.etag self._flag_definition_published_generation = fetch_generation self._flag_definition_cache_generation = fetch_generation self.log.debug( diff --git a/posthog/test/test_feature_flags.py b/posthog/test/test_feature_flags.py index 3a87b883..14640562 100644 --- a/posthog/test/test_feature_flags.py +++ b/posthog/test/test_feature_flags.py @@ -3180,7 +3180,7 @@ def test_load_feature_flags_304_not_modified(self, patch_get, patch_poll): # Second response is 304 Not Modified not_modified_response = GetResponse( data=None, - etag='"test-etag"', + etag='"updated-etag"', not_modified=True, ) patch_get.side_effect = [initial_response, not_modified_response] @@ -3200,6 +3200,7 @@ def test_load_feature_flags_304_not_modified(self, patch_get, patch_poll): self.assertEqual(len(client.feature_flags), 1) self.assertEqual(client.feature_flags[0]["key"], "beta-feature") self.assertEqual(client.group_type_mapping, {"0": "company"}) + self.assertEqual(client._flags_etag, '"updated-etag"') @mock.patch("posthog.client.Poller") @mock.patch("posthog.client.get")