Why is re-acquiring with delay expected to be successful?

[a1d4r]

Hi, I tried to use this library to avoid exceeding rate limits to an external API. I tested the rate limiter locally on one user and everything worked like a charm. But when I deployed the rate limiter in production on 100 users, I started catching exceptions in Sentry:

Re-acquiring with delay expected to be successful, if it failed then either clock or bucket is probably unstable

and

  File "/opt/pysetup/.venv/lib/python3.12/site-packages/pyrate_limiter/limiter.py", line 322, in _handle_async_result
    result = await result
                   └ <coroutine object Limiter.handle_bucket_put.<locals>._put_async at 0xed5962020930>
  File "/opt/pysetup/.venv/lib/python3.12/site-packages/pyrate_limiter/limiter.py", line 257, in _put_async
    result = await result
                   └ <coroutine object Limiter.delay_or_raise.<locals>._handle_async at 0xed59621c8540>
  File "/opt/pysetup/.venv/lib/python3.12/site-packages/pyrate_limiter/limiter.py", line 180, in _handle_async
    delay = await delay
                  └ <coroutine object AbstractBucket.waiting.<locals>._calc_waiting_async at 0xed5962096500>
  File "/opt/pysetup/.venv/lib/python3.12/site-packages/pyrate_limiter/abstracts/bucket.py", line 99, in _calc_waiting_async
    return _calc_waiting(bound_item)
           │             └ <pyrate_limiter.abstracts.rate.RateItem object at 0xed596209ddc0>
           └ <function AbstractBucket.waiting.<locals>._calc_waiting at 0xed59621eb6a0>
  File "/opt/pysetup/.venv/lib/python3.12/site-packages/pyrate_limiter/abstracts/bucket.py", line 83, in _calc_waiting
    assert self.failing_rate is not None  # NOTE: silence mypy
           │    └ None
           └ <pyrate_limiter.buckets.redis_bucket.RedisBucket object at 0xed5962fd87a0>

AssertionError: assert self.failing_rate is not None  # NOTE: silence mypy

I don't quite get why re-acquiring has to be successful. If there are multiple concurrent workers sending requests to API, many of them might exhaust rate limits and go to asyncio.sleep. After the delay, the worker might exceed the limit again, if other workers made requests. I was thinking about implementing the queue with exactly one consumer which will send API requests, but in this case I need the response back. Implementing RPCs is not quite trivial and requires message broker.

[vutran1710]

the mechanism behind re-acquiring is, we try to measure the time until the next slot for the item to become available - then we sleep and wait. So technically after such time has passed, the slot should be free and the bucket should be able to accept the waiting item to come it.

[a1d4r]

Thank you for the answer.

So technically after such time has passed, the slot should be free

In single-threaded applications it should be fine, but when there are multiple threads or coroutines, the race condition might occur. Is there any workaround for such scenarios?

[vutran1710]

Thank you for the answer.

So technically after such time has passed, the slot should be free

In single-threaded applications it should be fine, but when there are multiple threads or coroutines, the race condition might occur. Is there any workaround for such scenarios?

Must admit I havent spent enough time testing it out on the multi-thread environments. Therefore im not aware of any workaround yet

[RaccoonWang]

Shouldn't this be a bigger issue? I came across this library via a suggestion by Google Gemini, I am using multiple threads/clients to query a single endpoint and I need to implement rate limits (which is why I am using this library).

I ran into the same error here as the OP post.

It seems like a pretty significant error which should be acknowledged at least in the README?

Edit:

For me it looks like i may have a workaround. Semaphores can be used as a sort of crude in-application queue for single-process applications

`How asyncio.Semaphore Works:

Concurrency Limit, Not Rate: The semaphore value (e.g., 50) means only 50 tasks can pass the semaphore at the same time. It's a limit on concurrency, not speed.
Efficient Waiting: When a task tries to acquire the semaphore (async with semaphore:) and the limit (50) is already reached, the task doesn't "retry" or "spin". The asyncio event loop pauses that task and puts it into a waiting queue for the semaphore. While paused, the task consumes virtually no CPU. It yields control, allowing other tasks (or the event loop itself) to run.
Waking Up: When one of the 50 tasks finishes its work and releases the semaphore, the event loop wakes up one of the waiting tasks, which then acquires the semaphore and proceeds.
What Happens in Your Scenario (1000 Requests, Semaphore(50), Rate Limit 200/min):

Initial Burst: Your 1000 tasks start.
Semaphore Hit: All 1000 try to enter the async with semaphore: block.
First 50 Pass: The first 50 tasks successfully acquire the semaphore and move on towards your PyrateLimiter call.
950 Wait (Efficiently): The other 950 tasks are paused by the event loop in the semaphore's queue. They are not retrying; they are simply waiting for their turn without using CPU.
Rate Limiter Hit: The 50 tasks that passed the semaphore now hit PyrateLimiter. Your rate limit is 200/minute, which is about 3-4 requests per second.
Rate Limit Processing:
The first 3-4 tasks will pass the rate limit, do their work, and then release the semaphore.
The remaining ~46 tasks (out of the initial 50) will likely hit the rate limit and be put to sleep by PyrateLimiter (using asyncio.sleep), also waiting efficiently.
Semaphore Releases: As the first 3-4 tasks finish and release their semaphore, the event loop wakes up 3-4 tasks from the 950 waiting ones.
New Batch Hits Limiter: These 3-4 newly awakened tasks now proceed to PyrateLimiter. They will likely also hit the limit and go to sleep (because the limit is still being consumed).
The Flow: You establish a flow where:
A maximum of 50 tasks are "active" (either hitting the limiter, sleeping due to the limiter, or doing work).
The 950+ tasks wait efficiently at the semaphore.
The actual processing rate is dictated by your 200/minute limit. It will still take 5 minutes (1000 / 200) to process everything.
Why This is Good:

No CPU Waste: The 800 (or 950) tasks are not continuously retrying; they are waiting efficiently.
Reduced Contention: You avoid the "thundering herd" of 1000 tasks hitting PyrateLimiter and its Redis backend simultaneously. Only a manageable 50 are active, making the PyrateLimiter (and issue #185) much less likely to cause problems.
Stability: The system becomes more stable and predictable, processing requests at the rate you've defined, without overwhelming the limiter or Redis.
So, using asyncio.Semaphore prevents the "continuous retries" problem you're worried about. It introduces an orderly, efficient waiting mechanism to manage high concurrency before it hits your rate limiter.`

[vutran1710]

@RaccoonWang Tks for the suggestion. I'll see what I can do once I got some spare time.

[paultiq]

This should be solved by #208, using an asyncio lock in front of the tasks.

See asyncio_ratelimit.py