fix: keep queue failures lane-local

Author: marandaneto

posthog/_disabled_lane_queue.py

@@ -3,7 +3,7 @@
 
 
 class _DisabledLaneQueue:
-    """Minimal empty queue used for lifecycle cleanup on a disabled client."""
+    """Minimal empty queue used for lifecycle cleanup on an unavailable lane."""
 
     def __init__(self, maxsize: int) -> None:
         self.maxsize = maxsize

posthog/client.py

@@ -142,12 +142,14 @@ def _supports_lane_synchronization(queue) -> bool:
 
 
 def _new_lane_queue(maxsize: int) -> Queue:
-    """Return a safe queue, disabling async capture instead of raising on failure."""
+    """Return a safe queue, disabling the lane instead of raising on failure."""
     log = logging.getLogger("posthog")
     try:
         queue: Queue = Queue(maxsize)
     except Exception:
-        log.exception("Failed to initialize queue.Queue; disabling the PostHog client")
+        log.exception(
+            "Failed to initialize queue.Queue; disabling asynchronous capture for the lane"
+        )
         return cast(Queue, _DisabledLaneQueue(maxsize))
 
     if _supports_lane_synchronization(queue):
@@ -157,15 +159,16 @@ def _new_lane_queue(maxsize: int) -> Queue:
     if monkey is None:
         log.error(
             "queue.Queue lacks the synchronization interface required by PostHog "
-            "and gevent.monkey is not loaded; disabling the PostHog client"
+            "and gevent.monkey is not loaded; disabling asynchronous capture for the lane"
         )
         return cast(Queue, _DisabledLaneQueue(maxsize))
 
     try:
         if not monkey.is_object_patched("queue", "Queue"):
             log.error(
                 "queue.Queue lacks the synchronization interface required by PostHog "
-                "but gevent does not report it as patched; disabling the PostHog client"
+                "but gevent does not report it as patched; disabling asynchronous "
+                "capture for the lane"
             )
             return cast(Queue, _DisabledLaneQueue(maxsize))
 
@@ -174,7 +177,7 @@ def _new_lane_queue(maxsize: int) -> Queue:
     except Exception:
         log.exception(
             "Failed to restore the original queue.Queue after gevent monkey-patching; "
-            "disabling the PostHog client"
+            "disabling asynchronous capture for the lane"
         )
         return cast(Queue, _DisabledLaneQueue(maxsize))
 
@@ -183,7 +186,7 @@ def _new_lane_queue(maxsize: int) -> Queue:
 
     log.error(
         "The queue.Queue restored after gevent monkey-patching lacks the synchronization "
-        "interface required by PostHog; disabling the PostHog client"
+        "interface required by PostHog; disabling asynchronous capture for the lane"
     )
     return cast(Queue, _DisabledLaneQueue(maxsize))
 
@@ -403,7 +406,7 @@ def __init__(
             self.start()
 
     def _start_locked(self) -> None:
-        if self._started or self._closed:
+        if self._started or self._closed or not self.available:
             return
         for _ in range(self._thread_count):
             consumer = Consumer(
@@ -441,7 +444,7 @@ def start(self):
     def enqueue(self, msg) -> bool:
         """Atomically admit and queue `msg`, starting the lane on its first event."""
         with self._start_lock:
-            if self._closed:
+            if self._closed or not self.available:
                 return False
             self._start_locked()
             try:
@@ -1036,8 +1039,6 @@ def __init__(
             eager_start=False,
         )
         self._lanes = [self._analytics_lane, self._ai_lane]
-        if not all(lane.available for lane in self._lanes):
-            self.disabled = True
 
         if hasattr(os, "register_at_fork"):
             weak_self = weakref.ref(self)
@@ -2137,8 +2138,6 @@ def _reinit_after_fork(self):
         )
         for lane in self._lanes:
             lane.rebuild_after_fork(closed=terminal_requested)
-        if not all(lane.available for lane in self._lanes):
-            self.disabled = True
 
         self._lifecycle_lock = threading.Lock()
         self._lifecycle_condition = threading.Condition(self._lifecycle_lock)
@@ -2319,7 +2318,14 @@ def send_sync() -> None:
             self.log.debug("enqueued %s.", msg["event"])
             return sent_uuid
 
-        if lane._closed:
+        if not lane.available:
+            self.log.warning(
+                "%s lane is unavailable because a compatible queue could not be "
+                "initialized, dropping event %s",
+                lane.name,
+                msg["event"],
+            )
+        elif lane._closed:
             self.log.warning(
                 "%s lane received event %s after shutdown, dropping it",
                 lane.name,

posthog/test/test_gevent_compat.py

@@ -35,7 +35,7 @@ def test_disables_capture_if_no_compatible_queue_is_available(self):
         self.assertEqual(queue.unfinished_tasks, 0)
         self.assertIsNone(queue.task_done())
         self.assertIn("gevent.monkey is not loaded", logs.output[0])
-        self.assertIn("disabling the PostHog client", logs.output[0])
+        self.assertIn("disabling asynchronous capture for the lane", logs.output[0])
 
     def test_logs_gevent_recovery_failure_before_disabling(self):
         incompatible_queue = mock.Mock(spec=[])
@@ -54,7 +54,7 @@ def test_logs_gevent_recovery_failure_before_disabling(self):
         self.assertIn("Failed to restore the original queue.Queue", logs.output[0])
         self.assertIn("broken gevent state", logs.output[0])
 
-    def test_disables_client_if_no_compatible_queue_is_available(self):
+    def test_disables_only_async_capture_if_no_compatible_queue_is_available(self):
         incompatible_queue = mock.Mock(spec=[])
 
         with self.assertLogs("posthog", level="ERROR"):
@@ -64,12 +64,38 @@ def test_disables_client_if_no_compatible_queue_is_available(self):
             ):
                 client = Client(FAKE_TEST_API_KEY)
 
-        self.assertTrue(client.disabled)
+        self.assertFalse(client.disabled)
+        self.assertFalse(client._analytics_lane.available)
+        self.assertFalse(client._ai_lane.available)
         self.assertEqual(client.consumers, [])
         self.assertIsNone(client.capture("disabled-queue", distinct_id="distinct_id"))
+        self.assertEqual(client.consumers, [])
         client.flush()
         client.shutdown()
 
+    def test_incompatible_queue_does_not_disable_queue_independent_capabilities(self):
+        incompatible_queue = mock.Mock(spec=[])
+
+        with self.assertLogs("posthog", level="ERROR"):
+            with (
+                mock.patch("posthog.client.Queue", return_value=incompatible_queue),
+                mock.patch.dict(sys.modules, {"gevent.monkey": None}),
+                mock.patch("posthog.client.batch_post") as mock_post,
+                mock.patch(
+                    "posthog.client.flags",
+                    return_value={"featureFlags": {"beta-feature": True}},
+                ) as mock_flags,
+            ):
+                client = Client(FAKE_TEST_API_KEY, sync_mode=True)
+                event_uuid = client.capture("sync-capture", distinct_id="distinct_id")
+                decision = client.get_flags_decision("distinct_id")
+
+        self.assertFalse(client.disabled)
+        self.assertIsNotNone(event_uuid)
+        mock_post.assert_called_once()
+        self.assertTrue(decision["flags"]["beta-feature"].enabled)
+        mock_flags.assert_called_once()
+
 
 @unittest.skipUnless(importlib.util.find_spec("gevent"), "gevent is not installed")
 class TestGeventCompatibility(unittest.TestCase):