fix: deliver events under gevent monkey-patching with an SDK-owned lane queue

.sampo/changesets/gevent-queue-compat.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+fix: deliver events under gevent monkey-patching by giving lanes an SDK-owned queue (`LaneQueue`) with CPython `queue.Queue` semantics, including Python 3.13's `shutdown()`/`ShutDown` API; previously gevent's replacement `queue.Queue` lacked the private synchronization attributes the consumer and `flush()` rely on, so gevent gunicorn workers silently dropped every event. Note for integrators reaching into the backwards-compatible `Client.queue` property: the concrete type is now `posthog._queue.LaneQueue`, which matches `queue.Queue`'s full attribute surface but is deliberately not an instance of `queue.Queue` (inheriting would re-import the gevent bug).

LICENSE-PSF-2.0.txt

@@ -0,0 +1,54 @@
+This file applies to posthog/_queue.py, which is derived from CPython's
+Lib/queue.py. CPython is distributed under the Python Software Foundation
+License Version 2, reproduced below as required for derivative works.
+
+PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2
+--------------------------------------------
+
+1. This LICENSE AGREEMENT is between the Python Software Foundation
+("PSF"), and the Individual or Organization ("Licensee") accessing and
+otherwise using this software ("Python") in source or binary form and
+its associated documentation.
+
+2. Subject to the terms and conditions of this License Agreement, PSF hereby
+grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce,
+analyze, test, perform and/or display publicly, prepare derivative works,
+distribute, and otherwise use Python alone or in any derivative version,
+provided, however, that PSF's License Agreement and PSF's notice of copyright,
+i.e., "Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020, 2021, 2022, 2023 Python Software Foundation;
+All Rights Reserved" are retained in Python alone or in any derivative version
+prepared by Licensee.
+
+3. In the event Licensee prepares a derivative work that is based on
+or incorporates Python or any part thereof, and wants to make
+the derivative work available to others as provided herein, then
+Licensee hereby agrees to include in any such work a brief summary of
+the changes made to Python.
+
+4. PSF is making Python available to Licensee on an "AS IS"
+basis.  PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR
+IMPLIED.  BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND
+DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS
+FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT
+INFRINGE ANY THIRD PARTY RIGHTS.
+
+5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON
+FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS
+A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON,
+OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF.
+
+6. This License Agreement will automatically terminate upon a material
+breach of its terms and conditions.
+
+7. Nothing in this License Agreement shall be deemed to create any
+relationship of agency, partnership, or joint venture between PSF and
+Licensee.  This License Agreement does not grant permission to use PSF
+trademarks or trade name in a trademark sense to endorse or promote
+products or services of Licensee, or any third party.
+
+8. By copying, installing or otherwise using Python, Licensee
+agrees to be bound by the terms and conditions of this License
+Agreement.
+
+

posthog/_queue.py

@@ -0,0 +1,169 @@
+"""An SDK-owned FIFO queue with CPython ``queue.Queue`` semantics.
+
+The consumer and flush paths synchronize on attributes CPython's pure-Python
+``queue.Queue`` exposes but the ``queue`` module does not guarantee: ``mutex``,
+``not_empty``, ``not_full``, ``all_tasks_done``, ``unfinished_tasks``,
+``_qsize()``, and ``_get()``. Cooperative runtimes swap ``queue.Queue`` out for
+their own implementation without those attributes — gevent's
+``monkey.patch_all()`` installs ``gevent.queue.Queue``, on which the consumer
+thread dies with ``AttributeError: 'Queue' object has no attribute
+'not_empty'`` and ``flush()`` raises on ``all_tasks_done``, so a gevent
+gunicorn worker buffers every event forever and delivers none (#865).
+
+``LaneQueue`` is that pure-Python implementation carried by the SDK itself,
+derived from CPython's ``Lib/queue.py`` (distributed under the PSF-2.0
+license — see ``LICENSE-PSF-2.0.txt``), including Python 3.13's
+``shutdown()``/``ShutDown`` API. It builds only on ``threading`` primitives,
+which gevent patches compatibly, so it behaves identically on stock CPython
+and under monkey-patching — and its private surface can't be swapped out from
+under the SDK.
+
+Deliberate non-goal: ``LaneQueue`` does not inherit from ``queue.Queue``, so
+``isinstance(client.queue, queue.Queue)`` is ``False``. Inheriting would
+re-import the bug — under monkey-patching the base would resolve to gevent's
+incompatible class — and ``queue.Queue`` is not an ABC, so virtual
+registration is unavailable.
+"""
+
+import threading
+from collections import deque
+from queue import Empty, Full
+from time import monotonic
+
+try:
+    from queue import ShutDown
+except ImportError:
+    # Python < 3.13 has no queue.ShutDown; carry an equivalent.
+    class ShutDown(Exception):  # type: ignore[no-redef]
+        """Raised when put/get is called on a shut-down LaneQueue."""
+
+
+class LaneQueue:
+    """A FIFO queue with the full CPython ``queue.Queue`` interface.
+
+    ``maxsize`` bounds the queue; a ``maxsize`` of zero or less means the
+    queue is unbounded.
+    """
+
+    def __init__(self, maxsize: int = 0):
+        self.maxsize = maxsize
+        self.queue: deque = deque()  # named `queue` to match CPython's attribute
+        self.mutex = threading.Lock()
+        self.not_empty = threading.Condition(self.mutex)
+        self.not_full = threading.Condition(self.mutex)
+        self.all_tasks_done = threading.Condition(self.mutex)
+        self.unfinished_tasks = 0
+        self.is_shutdown = False
+
+    def task_done(self) -> None:
+        with self.all_tasks_done:
+            unfinished = self.unfinished_tasks - 1
+            if unfinished <= 0:
+                if unfinished < 0:
+                    raise ValueError("task_done() called too many times")
+                self.all_tasks_done.notify_all()
+            self.unfinished_tasks = unfinished
+
+    def join(self) -> None:
+        with self.all_tasks_done:
+            while self.unfinished_tasks:
+                self.all_tasks_done.wait()
+
+    def qsize(self) -> int:
+        with self.mutex:
+            return self._qsize()
+
+    def empty(self) -> bool:
+        with self.mutex:
+            return not self._qsize()
+
+    def full(self) -> bool:
+        with self.mutex:
+            return 0 < self.maxsize <= self._qsize()
+
+    def put(self, item, block: bool = True, timeout=None) -> None:
+        with self.not_full:
+            if self.is_shutdown:
+                raise ShutDown
+            if self.maxsize > 0:
+                if not block:
+                    if self._qsize() >= self.maxsize:
+                        raise Full
+                elif timeout is None:
+                    while self._qsize() >= self.maxsize:
+                        self.not_full.wait()
+                        if self.is_shutdown:
+                            raise ShutDown
+                elif timeout < 0:
+                    raise ValueError("'timeout' must be a non-negative number")
+                else:
+                    endtime = monotonic() + timeout
+                    while self._qsize() >= self.maxsize:
+                        remaining = endtime - monotonic()
+                        if remaining <= 0.0:
+                            raise Full
+                        self.not_full.wait(remaining)
+                        if self.is_shutdown:
+                            raise ShutDown
+            self._put(item)
+            self.unfinished_tasks += 1
+            self.not_empty.notify()
+
+    def get(self, block: bool = True, timeout=None):
+        with self.not_empty:
+            if self.is_shutdown and not self._qsize():
+                raise ShutDown
+            if not block:
+                if not self._qsize():
+                    raise Empty
+            elif timeout is None:
+                while not self._qsize():
+                    self.not_empty.wait()
+                    if self.is_shutdown and not self._qsize():
+                        raise ShutDown
+            elif timeout < 0:
+                raise ValueError("'timeout' must be a non-negative number")
+            else:
+                endtime = monotonic() + timeout
+                while not self._qsize():
+                    remaining = endtime - monotonic()
+                    if remaining <= 0.0:
+                        raise Empty
+                    self.not_empty.wait(remaining)
+                    if self.is_shutdown and not self._qsize():
+                        raise ShutDown
+            item = self._get()
+            self.not_full.notify()
+            return item
+
+    def put_nowait(self, item) -> None:
+        self.put(item, block=False)
+
+    def get_nowait(self):
+        return self.get(block=False)
+
+    def shutdown(self, immediate: bool = False) -> None:
+        """Shut down the queue: further put() raises ShutDown, get() drains.
+
+        With ``immediate=True``, pending items are discarded and blocked
+        ``get()``/``join()`` callers are released now.
+        """
+        with self.mutex:
+            self.is_shutdown = True
+            if immediate:
+                while self._qsize():
+                    self._get()
+                    if self.unfinished_tasks > 0:
+                        self.unfinished_tasks -= 1
+                self.all_tasks_done.notify_all()
+            self.not_empty.notify_all()
+            self.not_full.notify_all()
+
+    def _qsize(self) -> int:
+        return len(self.queue)
+
+    def _put(self, item) -> None:
+        self.queue.append(item)
+
+    def _get(self):
+        return self.queue.popleft()

posthog/client.py

@@ -9,13 +9,15 @@
 import warnings
 import weakref
 from contextvars import ContextVar
+from queue import Empty, Full
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, Dict, List, Mapping, Optional, Union
 from uuid import UUID, uuid4
 
 from typing_extensions import Unpack
 
 from posthog._async_utils import _BackgroundEventLoopRunner
+from ._queue import LaneQueue
 from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs
 from posthog.metrics_capture import PostHogMetrics
 from posthog.capture_compression import (
@@ -114,9 +116,6 @@
 from posthog.version import VERSION
 
 
-from queue import Empty, Full, Queue
-
-
 _configure_posthog_logging()
 
 MAX_DICT_SIZE = 50_000
@@ -327,7 +326,7 @@ def __init__(
         self._max_queue_size = max_queue_size
         self._thread_count = thread_count
         self._eager_start = eager_start
-        self.queue: Queue = Queue(max_queue_size)
+        self.queue: LaneQueue = LaneQueue(max_queue_size)
         self.consumers: List[Consumer] = []
         self._started = False
         self._closed = False
@@ -549,7 +548,7 @@ def rebuild_after_fork(self, *, closed: bool) -> None:
         the client's fork-visible lifecycle state. An eager open lane restarts
         immediately; a lazy lane returns to not-started and restarts on next use.
         """
-        self.queue = Queue(self._max_queue_size)
+        self.queue = LaneQueue(self._max_queue_size)
         self.reset_sync_send_state_after_fork()
         self._drain_signal = _DrainSignal(self.queue)
         self.consumers = []
@@ -981,7 +980,7 @@ def __init__(
         self._warn_if_duplicate_async_client()
 
     @property
-    def queue(self) -> Queue:
+    def queue(self) -> LaneQueue:
         """The analytics lane's queue (kept for backwards compatibility)."""
         return self._analytics_lane.queue
 

posthog/test/test_consumer.py

@@ -9,10 +9,9 @@
 from unittest import mock
 from parameterized import parameterized
 
-try:
-    from queue import Queue
-except ImportError:
-    from Queue import Queue
+# The consumer suite must exercise the queue class production delivery rides
+# on (posthog._queue.LaneQueue), not stdlib queue.Queue — see #865.
+from posthog._queue import LaneQueue as Queue
 
 from posthog.capture_compression import CaptureCompression
 from posthog.capture_mode import CaptureMode