fix(flags): restore cold-worker remote cache fallback

Author: marandaneto

posthog/client.py

@@ -898,7 +898,7 @@ 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.
+        # Until definitions load, only explicitly remote Redis results are shared.
         self._flag_definition_fingerprint = f"remote-only:{uuid4().hex}"
         if self.flag_cache:
             self.flag_cache._advance_generation(
@@ -2910,7 +2910,6 @@ def _hash_flag_definitions(data: FlagDefinitionCacheData) -> str:
     def _update_flag_state(
         self,
         data: FlagDefinitionCacheData,
-        old_flags_by_key: Optional[dict] = None,
         *,
         _fingerprint: Optional[str] = None,
     ) -> None:
@@ -2973,9 +2972,7 @@ def _load_feature_flags(self):
                     self.log.debug(
                         "[FEATURE FLAGS] Using cached flag definitions from external cache"
                     )
-                    self._update_flag_state(
-                        cached_data, old_flags_by_key=self.feature_flags_by_key or {}
-                    )
+                    self._update_flag_state(cached_data)
                     self._last_feature_flag_poll = datetime.now(tz=timezone.utc)
                     return
                 else:
@@ -3060,12 +3057,7 @@ def _fetch_feature_flags_from_api(self):
                     )
                     return
 
-                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,
-                    _fingerprint=fingerprint,
-                )
+                self._update_flag_state(response.data, _fingerprint=fingerprint)
 
                 if self._flag_definition_cache_provider:
                     cache_data_to_store = {
@@ -3509,7 +3501,7 @@ def _get_feature_flag_result(
                 # 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(
+                    self.flag_cache._set_cached_remote_flag(
                         distinct_id, key, flag_result, local_definition_version
                     )
 

posthog/test/test_flag_cache_publication.py

@@ -177,7 +177,7 @@ def write(value, version):
 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)
+    client._update_flag_state(definitions(2))
     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
@@ -231,7 +231,7 @@ def test_standalone_cache_can_reuse_invalidated_version(client):
 
 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)
+    client._update_flag_state(definitions(2))
     lock = cache._write_lock
     lock.acquire()
     try:

posthog/test/test_property_matching_version.py

@@ -231,7 +231,7 @@ def reload_during_evaluation(*args, **kwargs):
         nonlocal calls
         calls += 1
         if calls == 1:
-            client._update_flag_state(definitions(2), client.feature_flags_by_key)
+            client._update_flag_state(definitions(2))
         return original(*args, **kwargs)
 
     with mock.patch(
@@ -481,7 +481,7 @@ def test_in_flight_result_is_not_cached_in_new_generation(client):
     original = client_module.match_feature_flag_properties
 
     def reload_during_evaluation(*args, **kwargs):
-        client._update_flag_state(definitions(2), client.feature_flags_by_key)
+        client._update_flag_state(definitions(2))
         return original(*args, **kwargs)
 
     with mock.patch(

posthog/test/test_redis_flag_cache_snapshot.py

@@ -24,13 +24,13 @@ def workers():
     redis = FakeRedis()
     clients = []
 
-    def worker():
+    def worker(secret_key="test-secret"):
         with mock.patch.object(
             Client, "_initialize_flag_cache", return_value=RedisFlagCache(redis)
         ):
             client = Client(
                 FAKE_TEST_API_KEY,
-                secret_key="test-secret",
+                secret_key=secret_key,
                 send=False,
                 enable_local_evaluation=False,
             )
@@ -253,7 +253,13 @@ def test_fork_retains_snapshot_binding(workers):
 
 
 def remote_success(client, during_request=None):
-    details = FeatureFlag.from_json({"key": "person", "enabled": True})
+    details = FeatureFlag.from_json(
+        {
+            "key": "person",
+            "enabled": True,
+            "metadata": {"payload": '{"source":"remote"}'},
+        }
+    )
 
     def request(*args, **kwargs):
         if during_request:
@@ -274,8 +280,8 @@ def test_remote_only_success_remains_available_for_stale_fallback(workers):
     with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
         assert remote_success(writer).get_value() is True
         assert fallback(writer).get_value() is True
-        # Without verified definitions, a new Client cannot reuse this provenance.
-        assert fallback(workers()) is None
+        # Remote results remain usable by a worker starting during the outage.
+        assert fallback(workers()).get_value() is True
 
 
 @pytest.mark.parametrize("transition", ["hydrate", "empty-hydrate", 401, 402])
@@ -299,6 +305,7 @@ def change():
     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
+        assert fallback(workers()) is None
     load(client, definitions(1))
     assert evaluate(client).get_value() is True
     assert fallback(client).get_value() is True
@@ -327,7 +334,7 @@ def test_reset_empty_fingerprint_cannot_cache_or_read_results(workers, status):
     assert fallback(client) is None
 
 
-def test_invalidated_remote_only_result_does_not_revive_after_restart(workers):
+def test_remote_result_requires_matching_snapshot_after_hydration(workers):
     writer = workers()
     with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
         assert remote_success(writer).get_value() is True
@@ -336,10 +343,10 @@ def test_invalidated_remote_only_result_does_not_revive_after_restart(workers):
     assert fallback(writer) is None
     writer.shutdown()
     with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
-        assert fallback(workers()) is None
+        assert fallback(workers()).get_value() is True
 
 
-def test_fork_renews_remote_only_provenance(workers):
+def test_fork_retains_remote_fallback(workers):
     client = workers()
     with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
         assert remote_success(client).get_value() is True
@@ -348,6 +355,68 @@ def test_fork_renews_remote_only_provenance(workers):
             client, "_initialize_flag_cache", return_value=RedisFlagCache(redis)
         ):
             client._reinit_after_fork()
-        assert fallback(client) is None
+        assert fallback(client).get_value() is True
         assert remote_success(client).get_value() is True
         assert fallback(client).get_value() is True
+
+
+@pytest.mark.parametrize("secret_key", [None, "test-secret"])
+@pytest.mark.parametrize("writer_has_definitions", [False, True])
+def test_cold_worker_uses_shared_remote_result_during_outage(
+    workers, secret_key, writer_has_definitions
+):
+    writer = workers(secret_key)
+    if writer_has_definitions:
+        load(writer, definitions(1))
+    with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
+        assert remote_success(writer).get_value() is True
+        reader = workers(secret_key)
+        # Reproduce startup when neither endpoint is available.
+        reader._load_feature_flags()
+        assert reader.feature_flags is None
+        result = fallback(reader)
+    assert result is not None
+    assert result.get_value() is True
+    assert result.payload == {"source": "remote"}
+
+
+def test_cold_worker_rejects_shared_local_result_during_outage(workers):
+    writer = workers()
+    load(writer, definitions(1))
+    assert evaluate(writer).get_value() is True
+    with mock.patch("posthog.client.get", side_effect=APIError(503, "offline")):
+        assert fallback(workers()) is None
+
+
+def test_loaded_worker_rejects_remote_result_from_different_snapshot(workers):
+    writer = workers()
+    load(writer, definitions(1))
+    assert remote_success(writer).get_value() is True
+    reader = workers()
+    load(reader, definitions(2))
+    assert fallback(reader) is None
+
+
+@pytest.mark.parametrize("status", [401, 402])
+def test_reset_worker_rejects_shared_remote_result(workers, status):
+    writer = workers()
+    load(writer, definitions(1))
+    assert remote_success(writer).get_value() is True
+    reader = workers()
+    with mock.patch("posthog.client.get", side_effect=APIError(status, "reset")):
+        reader._load_feature_flags()
+        assert fallback(reader) is None
+
+
+def test_shared_remote_fallback_respects_stale_ttl(workers):
+    writer = workers()
+    with (
+        mock.patch("posthog.client.get", side_effect=APIError(503, "offline")),
+        mock.patch("posthog.utils.time.time", return_value=100) as clock,
+    ):
+        assert remote_success(writer).get_value() is True
+        reader = workers()
+        clock.return_value = 100 + writer.flag_cache.stale_ttl - 1
+        assert fallback(reader).get_value() is True
+        clock.return_value += 1
+        assert fallback(reader) is None

posthog/utils.py

@@ -218,6 +218,7 @@ def __init__(self, flag_result, flag_definition_version, timestamp=None):
         self.flag_definition_version = flag_definition_version
         self.timestamp = timestamp or time.time()
         self._snapshot_fingerprint: Optional[str] = None
+        self._is_remote = False
 
     def is_valid(self, current_time, ttl, current_flag_version):
         time_valid = (current_time - self.timestamp) < ttl
@@ -237,6 +238,13 @@ def __init__(self, max_size=CACHE_MAX_SIZE, default_ttl=CACHE_TTL):
         self._minimum_version = None
         self._write_lock = threading.Lock()
 
+    def _set_cached_remote_flag(
+        self, distinct_id, flag_key, flag_result, flag_definition_version
+    ):
+        self.set_cached_flag(
+            distinct_id, flag_key, flag_result, flag_definition_version
+        )
+
     def _advance_generation(self, version, fingerprint=None):
         # Client generations only move forward. Standalone cache invalidation
         # retains its existing exact-version deletion semantics.
@@ -385,6 +393,10 @@ def _advance_generation(self, version, fingerprint=None):
     def _is_entry_current(self, entry):
         snapshot = self._snapshot
         if snapshot is not None:
+            # A worker starting during an outage has no local snapshot to verify.
+            # Only explicitly remote results may cross that provenance boundary.
+            if snapshot[1].startswith("remote-only:") and entry._is_remote:
+                return True
             return bool(snapshot[1]) and entry._snapshot_fingerprint == snapshot[1]
         return self._is_version_current(entry.flag_definition_version)
 
@@ -395,7 +407,12 @@ 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, fingerprint=None
+        self,
+        flag_result,
+        flag_definition_version,
+        timestamp=None,
+        fingerprint=None,
+        is_remote=False,
     ):
         if timestamp is None:
             timestamp = time.time()
@@ -410,6 +427,8 @@ def _serialize_entry(
         }
         if fingerprint is not None:
             entry["snapshot_fingerprint"] = fingerprint
+        if is_remote:
+            entry["evaluation_source"] = "remote"
         if isinstance(flag_result, _FeatureFlagResult):
             # Additive metadata keeps the existing entry shape readable by older SDKs.
             entry["flag_result_type"] = _FEATURE_FLAG_RESULT_TYPE
@@ -433,6 +452,7 @@ def _deserialize_entry(self, data):
                 timestamp=entry["timestamp"],
             )
             result._snapshot_fingerprint = entry.get("snapshot_fingerprint")
+            result._is_remote = entry.get("evaluation_source") == "remote"
             return result
         except (json.JSONDecodeError, KeyError, TypeError, ValueError):
             # If deserialization fails, treat as cache miss
@@ -486,6 +506,25 @@ 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
+    ):
+        self._set_cached_flag(
+            distinct_id, flag_key, flag_result, flag_definition_version
+        )
+
+    def _set_cached_remote_flag(
+        self, distinct_id, flag_key, flag_result, flag_definition_version
+    ):
+        self._set_cached_flag(
+            distinct_id, flag_key, flag_result, flag_definition_version, is_remote=True
+        )
+
+    def _set_cached_flag(
+        self,
+        distinct_id,
+        flag_key,
+        flag_result,
+        flag_definition_version,
+        is_remote=False,
     ):
         try:
             cache_key = self._get_cache_key(distinct_id, flag_key)
@@ -498,16 +537,19 @@ def set_cached_flag(
                     return
                 fingerprint = snapshot[1]
             serialized_entry = self._serialize_entry(
-                flag_result, flag_definition_version, fingerprint=fingerprint
+                flag_result,
+                flag_definition_version,
+                fingerprint=fingerprint,
+                is_remote=is_remote,
             )
 
             # 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.
+                # Late writes keep their original fingerprint, so workers with
+                # different loaded definitions cannot accept them.
                 self.redis.setex(cache_key, self.stale_ttl, serialized_entry)
                 self.redis.set(self.version_key, flag_definition_version)