Repository navigation
Backpressure when queue is full #146
Description
Activity
Looking to take this. Played with the customer's config (
flush_interval=2, flush_at=100, max_queue_size=1000) against_enqueuearound line 1330 —self.queue.put(msg, block=False)always drops when full.What I'd add: an
on_queue_fullparam with three values."drop"stays default, no break."block"switches toput(block=True, timeout=...), withqueue_full_block_timeout(defaultNone) for callers who want a bounded wait. If the timeout expires, falls back through the existing warn-and-drop branch so it still terminates."raise"letsqueue.Fullpropagate.
One snag worth raising before opening the PR — the
capturewrapper at line 142 swallows exceptions unlessdebug=True. So"raise"only raises when debug is on. Two ways to handle: document it on the param, or carvequeue.Fullout of the wrapper as a special case. I'd document. The wrapper carve-out changes contract for everyone using the lib, just to support one strategy.Tests are written locally, six of them, all green.
We uncovered a case where a key customer was losing events due to the way this library is configured/built.
With the
flush_interval=2(default is0.5),flush_at=100andmax_queue_size=1000(default 10k), their queue got full during peak hours and they started dropping events.Their recommended solution is very reasonable:
We should implement some way to block when the queue is full, or some way to let the user know the queue size.