From 407181b79b3a87c1c819dd0589971303f724a68e Mon Sep 17 00:00:00 2001 From: Emmanuel Yusufu Kimaswa Date: Mon, 6 Jul 2026 15:47:55 +0300 Subject: [PATCH 1/3] Add on_queue_full option for backpressure when the event queue is full The queue put always dropped on Full. on_queue_full keeps drop as the default, block waits for space with an optional queue_full_block_timeout and falls back to the warn-and-drop path on timeout, and raise propagates queue.Full. Raise only surfaces when debug=True because public methods swallow exceptions otherwise, documented on the param. --- .../changesets/on-queue-full-backpressure.md | 5 ++ posthog/client.py | 24 ++++++- posthog/test/test_client.py | 71 +++++++++++++++++++ 3 files changed, 99 insertions(+), 1 deletion(-) create mode 100644 .sampo/changesets/on-queue-full-backpressure.md diff --git a/.sampo/changesets/on-queue-full-backpressure.md b/.sampo/changesets/on-queue-full-backpressure.md new file mode 100644 index 00000000..72648eda --- /dev/null +++ b/.sampo/changesets/on-queue-full-backpressure.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Add on_queue_full option (drop, block, or raise) with queue_full_block_timeout so callers can apply backpressure instead of dropping events when the queue is full diff --git a/posthog/client.py b/posthog/client.py index 99ef4801..dec56a0b 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -260,6 +260,8 @@ def __init__( exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, + on_queue_full="drop", + queue_full_block_timeout=None, _dedicated_ai_endpoint=False, ): """ @@ -343,6 +345,16 @@ def __init__( interval for each exception type's bucket. exception_autocapture_refill_interval_seconds: Seconds between token refills for autocaptured exception rate limiting. + on_queue_full: What to do when the internal event queue is full. + ``"drop"`` (default) drops the event and logs a warning, + ``"block"`` waits for queue space, and ``"raise"`` raises + ``queue.Full``. Public methods such as ``capture()`` swallow + exceptions unless ``debug=True``, so ``"raise"`` only + propagates in debug mode. + queue_full_block_timeout: Maximum seconds to wait for queue space + when ``on_queue_full="block"``. ``None`` (default) waits + indefinitely. If the wait times out, the event is dropped + with a warning. Examples: ```python @@ -354,6 +366,11 @@ def __init__( Category: Initialization """ + if on_queue_full not in ("drop", "block", "raise"): + raise ValueError("on_queue_full must be one of 'drop', 'block', or 'raise'") + self.on_queue_full = on_queue_full + self.queue_full_block_timeout = queue_full_block_timeout + self._max_queue_size = max_queue_size self.queue: Queue = Queue(max_queue_size) @@ -1642,10 +1659,15 @@ def _enqueue(self, msg, disable_geoip): return sent_uuid try: - self.queue.put(msg, block=False) + if self.on_queue_full == "block": + self.queue.put(msg, block=True, timeout=self.queue_full_block_timeout) + else: + self.queue.put(msg, block=False) self.log.debug("enqueued %s.", msg["event"]) return sent_uuid except Full: + if self.on_queue_full == "raise": + raise self.log.warning("analytics-python queue is full") return None diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 98bb4914..45ffb826 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -1,8 +1,10 @@ import logging import asyncio +import threading import time import unittest from datetime import datetime +from queue import Full from uuid import UUID, uuid4 from unittest import mock @@ -2092,6 +2094,75 @@ def test_overflow(self): # Make sure we are informed that the queue is at capacity self.assertIsNone(msg_uuid) + def test_on_queue_full_default_drops(self): + client = Client(FAKE_TEST_API_KEY, max_queue_size=1) + client.join() + + first_uuid = client.capture("test event", distinct_id="distinct_id") + second_uuid = client.capture("test event", distinct_id="distinct_id") + + self.assertIsNotNone(first_uuid) + self.assertIsNone(second_uuid) + + def test_on_queue_full_block_waits_for_space(self): + client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="block") + client.join() + client.capture("first event", distinct_id="distinct_id") + + def drain_soon(): + time.sleep(0.2) + client.queue.get() + client.queue.task_done() + + drainer = threading.Thread(target=drain_soon) + drainer.start() + try: + msg_uuid = client.capture("second event", distinct_id="distinct_id") + finally: + drainer.join() + + self.assertIsNotNone(msg_uuid) + + def test_on_queue_full_block_timeout_drops(self): + client = Client( + FAKE_TEST_API_KEY, + max_queue_size=1, + on_queue_full="block", + queue_full_block_timeout=0.05, + ) + client.join() + client.capture("first event", distinct_id="distinct_id") + + start = time.monotonic() + msg_uuid = client.capture("second event", distinct_id="distinct_id") + waited = time.monotonic() - start + + self.assertIsNone(msg_uuid) + self.assertGreaterEqual(waited, 0.05) + + def test_on_queue_full_raise_propagates_in_debug(self): + client = Client( + FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="raise", debug=True + ) + client.join() + client.capture("first event", distinct_id="distinct_id") + + with self.assertRaises(Full): + client.capture("second event", distinct_id="distinct_id") + + def test_on_queue_full_raise_swallowed_without_debug(self): + client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="raise") + client.join() + client.capture("first event", distinct_id="distinct_id") + + msg_uuid = client.capture("second event", distinct_id="distinct_id") + + self.assertIsNone(msg_uuid) + + def test_on_queue_full_invalid_value(self): + with self.assertRaises(ValueError): + Client(FAKE_TEST_API_KEY, on_queue_full="explode") + def test_unicode(self): Client("unicode_key") From b2f680899a01a92b152c26d8de8800eff258cc5c Mon Sep 17 00:00:00 2001 From: Emmanuel Yusufu Kimaswa Date: Mon, 6 Jul 2026 15:54:13 +0300 Subject: [PATCH 2/3] Validate queue_full_block_timeout and parameterise the queue-full tests --- posthog/client.py | 4 ++ posthog/test/test_client.py | 83 +++++++++++++++++-------------------- 2 files changed, 42 insertions(+), 45 deletions(-) diff --git a/posthog/client.py b/posthog/client.py index dec56a0b..a2946497 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -368,6 +368,10 @@ def __init__( """ if on_queue_full not in ("drop", "block", "raise"): raise ValueError("on_queue_full must be one of 'drop', 'block', or 'raise'") + if queue_full_block_timeout is not None and queue_full_block_timeout < 0: + raise ValueError( + "queue_full_block_timeout must be a non-negative number of seconds" + ) self.on_queue_full = on_queue_full self.queue_full_block_timeout = queue_full_block_timeout diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 45ffb826..4ffa3d31 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -2094,15 +2094,35 @@ def test_overflow(self): # Make sure we are informed that the queue is at capacity self.assertIsNone(msg_uuid) - def test_on_queue_full_default_drops(self): - client = Client(FAKE_TEST_API_KEY, max_queue_size=1) + @parameterized.expand( + [ + ("drop", {"on_queue_full": "drop"}, False, 0), + ( + "block_with_timeout", + {"on_queue_full": "block", "queue_full_block_timeout": 0.05}, + False, + 0.05, + ), + ("raise_without_debug", {"on_queue_full": "raise"}, False, 0), + ("raise_with_debug", {"on_queue_full": "raise", "debug": True}, True, 0), + ] + ) + def test_on_queue_full_when_queue_stays_full( + self, _name, client_kwargs, expect_raises, min_wait + ): + client = Client(FAKE_TEST_API_KEY, max_queue_size=1, **client_kwargs) client.join() - - first_uuid = client.capture("test event", distinct_id="distinct_id") - second_uuid = client.capture("test event", distinct_id="distinct_id") - + first_uuid = client.capture("first event", distinct_id="distinct_id") self.assertIsNotNone(first_uuid) - self.assertIsNone(second_uuid) + + start = time.monotonic() + if expect_raises: + with self.assertRaises(Full): + client.capture("second event", distinct_id="distinct_id") + else: + msg_uuid = client.capture("second event", distinct_id="distinct_id") + self.assertIsNone(msg_uuid) + self.assertGreaterEqual(time.monotonic() - start, min_wait) def test_on_queue_full_block_waits_for_space(self): client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="block") @@ -2123,45 +2143,18 @@ def drain_soon(): self.assertIsNotNone(msg_uuid) - def test_on_queue_full_block_timeout_drops(self): - client = Client( - FAKE_TEST_API_KEY, - max_queue_size=1, - on_queue_full="block", - queue_full_block_timeout=0.05, - ) - client.join() - client.capture("first event", distinct_id="distinct_id") - - start = time.monotonic() - msg_uuid = client.capture("second event", distinct_id="distinct_id") - waited = time.monotonic() - start - - self.assertIsNone(msg_uuid) - self.assertGreaterEqual(waited, 0.05) - - def test_on_queue_full_raise_propagates_in_debug(self): - client = Client( - FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="raise", debug=True - ) - client.join() - client.capture("first event", distinct_id="distinct_id") - - with self.assertRaises(Full): - client.capture("second event", distinct_id="distinct_id") - - def test_on_queue_full_raise_swallowed_without_debug(self): - client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="raise") - client.join() - client.capture("first event", distinct_id="distinct_id") - - msg_uuid = client.capture("second event", distinct_id="distinct_id") - - self.assertIsNone(msg_uuid) - - def test_on_queue_full_invalid_value(self): + @parameterized.expand( + [ + ("invalid_strategy", {"on_queue_full": "explode"}), + ( + "negative_block_timeout", + {"on_queue_full": "block", "queue_full_block_timeout": -1}, + ), + ] + ) + def test_on_queue_full_invalid_config(self, _name, client_kwargs): with self.assertRaises(ValueError): - Client(FAKE_TEST_API_KEY, on_queue_full="explode") + Client(FAKE_TEST_API_KEY, **client_kwargs) def test_unicode(self): Client("unicode_key") From c701230a1958cfd0a302e209b443d6e16f739127 Mon Sep 17 00:00:00 2001 From: Emmanuel Yusufu Kimaswa Date: Sat, 11 Jul 2026 13:47:19 +0300 Subject: [PATCH 3/3] Replace on_queue_full modes with an on_drop callback Blocking or raising when the queue is full lets the SDK stall the application it measures, so dropping is the only safe behavior. Add an on_drop callback invoked with the event on drop, so callers can observe or count dropped events instead of applying backpressure. Signed-off-by: Emmanuel Yusufu Kimaswa --- .sampo/changesets/on-drop-callback.md | 5 ++ .../changesets/on-queue-full-backpressure.md | 5 -- posthog/client.py | 40 ++++------ posthog/test/test_client.py | 76 ++++++------------- 4 files changed, 42 insertions(+), 84 deletions(-) create mode 100644 .sampo/changesets/on-drop-callback.md delete mode 100644 .sampo/changesets/on-queue-full-backpressure.md diff --git a/.sampo/changesets/on-drop-callback.md b/.sampo/changesets/on-drop-callback.md new file mode 100644 index 00000000..c3d38d77 --- /dev/null +++ b/.sampo/changesets/on-drop-callback.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Add an on_drop callback invoked with the event when the internal queue is full, so callers can observe or count dropped events. The SDK always drops rather than blocking the caller. diff --git a/.sampo/changesets/on-queue-full-backpressure.md b/.sampo/changesets/on-queue-full-backpressure.md deleted file mode 100644 index 72648eda..00000000 --- a/.sampo/changesets/on-queue-full-backpressure.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -pypi/posthog: minor ---- - -Add on_queue_full option (drop, block, or raise) with queue_full_block_timeout so callers can apply backpressure instead of dropping events when the queue is full diff --git a/posthog/client.py b/posthog/client.py index a2946497..9fcb1520 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -260,8 +260,7 @@ def __init__( exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, - on_queue_full="drop", - queue_full_block_timeout=None, + on_drop=None, _dedicated_ai_endpoint=False, ): """ @@ -345,16 +344,10 @@ def __init__( interval for each exception type's bucket. exception_autocapture_refill_interval_seconds: Seconds between token refills for autocaptured exception rate limiting. - on_queue_full: What to do when the internal event queue is full. - ``"drop"`` (default) drops the event and logs a warning, - ``"block"`` waits for queue space, and ``"raise"`` raises - ``queue.Full``. Public methods such as ``capture()`` swallow - exceptions unless ``debug=True``, so ``"raise"`` only - propagates in debug mode. - queue_full_block_timeout: Maximum seconds to wait for queue space - when ``on_queue_full="block"``. ``None`` (default) waits - indefinitely. If the wait times out, the event is dropped - with a warning. + on_drop: Optional callback invoked with the dropped event when the + internal queue is full. The SDK always drops rather than blocking + the caller, so this is how you observe or count dropped events. + Runs on the calling thread, so keep it fast and non-blocking. Examples: ```python @@ -366,14 +359,9 @@ def __init__( Category: Initialization """ - if on_queue_full not in ("drop", "block", "raise"): - raise ValueError("on_queue_full must be one of 'drop', 'block', or 'raise'") - if queue_full_block_timeout is not None and queue_full_block_timeout < 0: - raise ValueError( - "queue_full_block_timeout must be a non-negative number of seconds" - ) - self.on_queue_full = on_queue_full - self.queue_full_block_timeout = queue_full_block_timeout + if on_drop is not None and not callable(on_drop): + raise ValueError("on_drop must be callable") + self.on_drop = on_drop self._max_queue_size = max_queue_size self.queue: Queue = Queue(max_queue_size) @@ -1663,16 +1651,16 @@ def _enqueue(self, msg, disable_geoip): return sent_uuid try: - if self.on_queue_full == "block": - self.queue.put(msg, block=True, timeout=self.queue_full_block_timeout) - else: - self.queue.put(msg, block=False) + self.queue.put(msg, block=False) self.log.debug("enqueued %s.", msg["event"]) return sent_uuid except Full: - if self.on_queue_full == "raise": - raise self.log.warning("analytics-python queue is full") + if self.on_drop is not None: + try: + self.on_drop(msg) + except Exception: + self.log.exception("on_drop callback raised") return None def flush(self, timeout_seconds: Optional[float] = 10) -> None: diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 4ffa3d31..a3ef2152 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -1,10 +1,8 @@ import logging import asyncio -import threading import time import unittest from datetime import datetime -from queue import Full from uuid import UUID, uuid4 from unittest import mock @@ -2094,67 +2092,39 @@ def test_overflow(self): # Make sure we are informed that the queue is at capacity self.assertIsNone(msg_uuid) - @parameterized.expand( - [ - ("drop", {"on_queue_full": "drop"}, False, 0), - ( - "block_with_timeout", - {"on_queue_full": "block", "queue_full_block_timeout": 0.05}, - False, - 0.05, - ), - ("raise_without_debug", {"on_queue_full": "raise"}, False, 0), - ("raise_with_debug", {"on_queue_full": "raise", "debug": True}, True, 0), - ] - ) - def test_on_queue_full_when_queue_stays_full( - self, _name, client_kwargs, expect_raises, min_wait - ): - client = Client(FAKE_TEST_API_KEY, max_queue_size=1, **client_kwargs) + def test_on_drop_called_when_queue_full(self): + dropped = [] + client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_drop=dropped.append) client.join() first_uuid = client.capture("first event", distinct_id="distinct_id") self.assertIsNotNone(first_uuid) - start = time.monotonic() - if expect_raises: - with self.assertRaises(Full): - client.capture("second event", distinct_id="distinct_id") - else: - msg_uuid = client.capture("second event", distinct_id="distinct_id") - self.assertIsNone(msg_uuid) - self.assertGreaterEqual(time.monotonic() - start, min_wait) + second_uuid = client.capture("second event", distinct_id="distinct_id") + self.assertIsNone(second_uuid) + self.assertEqual(len(dropped), 1) + self.assertEqual(dropped[0]["event"], "second event") - def test_on_queue_full_block_waits_for_space(self): - client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_queue_full="block") + def test_on_drop_not_called_when_space_available(self): + dropped = [] + client = Client(FAKE_TEST_API_KEY, max_queue_size=10, on_drop=dropped.append) client.join() - client.capture("first event", distinct_id="distinct_id") - - def drain_soon(): - time.sleep(0.2) - client.queue.get() - client.queue.task_done() + first_uuid = client.capture("first event", distinct_id="distinct_id") + self.assertIsNotNone(first_uuid) + self.assertEqual(dropped, []) - drainer = threading.Thread(target=drain_soon) - drainer.start() - try: - msg_uuid = client.capture("second event", distinct_id="distinct_id") - finally: - drainer.join() + def test_on_drop_callback_exception_is_swallowed(self): + def boom(_msg): + raise RuntimeError("boom") - self.assertIsNotNone(msg_uuid) + client = Client(FAKE_TEST_API_KEY, max_queue_size=1, on_drop=boom) + client.join() + client.capture("first event", distinct_id="distinct_id") + second_uuid = client.capture("second event", distinct_id="distinct_id") + self.assertIsNone(second_uuid) - @parameterized.expand( - [ - ("invalid_strategy", {"on_queue_full": "explode"}), - ( - "negative_block_timeout", - {"on_queue_full": "block", "queue_full_block_timeout": -1}, - ), - ] - ) - def test_on_queue_full_invalid_config(self, _name, client_kwargs): + def test_on_drop_invalid_config(self): with self.assertRaises(ValueError): - Client(FAKE_TEST_API_KEY, **client_kwargs) + Client(FAKE_TEST_API_KEY, on_drop="not callable") def test_unicode(self): Client("unicode_key")