Add on_queue_full option for backpressure when the event queue is full#731
Add on_queue_full option for backpressure when the event queue is full#731emmayusufu wants to merge 3 commits into
Conversation
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.
|
Reviews (1): Last reviewed commit: "Add on_queue_full option for backpressur..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| 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") | ||
|
|
There was a problem hiding this comment.
The custom instructions for this repo say "we always prefer parameterised tests", and the file already uses @parameterized.expand extensively. The six new test methods are all independent cases for the same feature. At a minimum, test_on_queue_full_raise_propagates_in_debug and test_on_queue_full_raise_swallowed_without_debug are two cases of the same scenario (same setup, just debug=True vs debug=False), and test_on_queue_full_default_drops shares its structure with the timeout-drop case. Collapsing these into one or two @parameterized.expand tables would also remove the boilerplate duplication across test bodies.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| 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) |
There was a problem hiding this comment.
test_on_queue_full_default_drops duplicates test_overflow
test_overflow at line 2085 already verifies that the default queue-full policy drops events and returns None. This new test exercises exactly the same code path with fewer enqueues but doesn't add any new assertion. Per the simplicity rule "Has no superfluous parts", it could be removed in favour of the existing test, or folded into a parameterised suite as the baseline case.
| Category: | ||
| Initialization | ||
| """ | ||
| if on_queue_full not in ("drop", "block", "raise"): |
There was a problem hiding this comment.
queue_full_block_timeout not validated at init
A negative value is silently accepted here. When on_queue_full="block", Queue.put(block=True, timeout=-0.1) immediately raises ValueError: 'timeout' must be a non-negative number, which propagates out of _enqueue at call time rather than at construction — and in non-debug mode gets swallowed by @no_throw(). Adding a check at init (alongside the existing on_queue_full validation) would give a clearer error closer to the misconfiguration.
|
Addressed all three: negative |
dustinbyrne
left a comment
dustinbyrne
left a comment
There was a problem hiding this comment.
Hey @emmayusufu, thanks for putting this together. I've been thinking about this quite a bit. Most critically, I think data loss is our only real option here. Consider either of the following scenarios in which the buffer cannot be drained quickly enough:
- The client is producing events faster than they can be processed and flushed out of the buffer.
- An intermittent network issue or outage prevents captured events from reaching PostHog due to timeouts, server errors, etc.
Either scenario would catastrophic outcomes if the buffer becomes full and every subsequent capture becomes blocking. Because a full buffer already indicates that the consumer cannot keep up, blocking can amplify the problem by tying up application threads.
There is no lossless outcome once production remains above the drain rate, whatever the cause. We either drop events, allow unbounded memory growth, persist them elsewhere, or transfer back pressure into the application's worker pool. For our SDK, I don't believe the latter option is safe.
Again, I appreciate you taking the time to open this PR. I’d be interested to hear your thoughts, particularly if you had a workload in mind where blocking would be preferable to dropping.
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 <kimaswaemma36@gmail.com>
|
Hi @dustinbyrne, I do think you are right, for an analytics SDK, blocking the caller to save event data might not be best approach because If the app is already outrunning the flush or the network is down, tying up its threads makes it worse, so dropping events is defintely the safe approach |
2 participants
Implements the design from my comment on the issue: an
on_queue_fulloption controlling what_enqueuedoes when the internal queue is full."drop"stays the default, existing warn-and-drop behavior, nothing changes for current users."block"switches toput(block=True, timeout=queue_full_block_timeout).None(default) waits indefinitely; on timeout the event falls through the existing warn-and-drop branch so a bounded wait still terminates."raise"letsqueue.Fullpropagate. One caveat, documented on the param: public methods likecapture()swallow exceptions unlessdebug=True, so"raise"only surfaces in debug mode. I kept the wrapper untouched rather than special-casingqueue.Fullfor everyone.An invalid value raises
ValueErrorat init. The new params sit at the end of the signature so positional callers are unaffected.Six tests following the existing
test_overflowpattern (max_queue_size=1+client.join()): default drops, block waits for space, block times out and drops, raise propagates in debug, raise is swallowed without debug, invalid value rejected. Fulltest_client.pypasses (141 tests). Changeset included (minor).Closes #146