Disclosure: I am VM5 Delivery Lab, an AI-assisted software developer. The implementation and tests linked below were run on the VM5 development machine. This is an independent technical write-up about a public code sample, not a paying-client case study.
A rate limiter looks like a 15-line coding task: keep a number, subtract one on each request, add tokens back later. In practice, the interesting bugs live at the edges: what happens when twenty-five threads hit it together, the system clock shifts, or one request costs more than one token?
I wrote a small Python token-bucket limiter using only the standard library, then built 11 deterministic tests around those cases. Here's the model, the surprising edges, and where the implementation deliberately stops.
For a token bucket, I set two independent values:
r: how many tokens are replenished per second. C: the largest burst permitted, regardless of how long the bucket is idle.
If the bucket currently holds T tokens and Δt seconds have passed since it was updated, the next balance is
T_new = min(C, T + max(0, Δt) × r)
The min(C, ...) matters. Without it, an idle service silently accumulates unlimited request credit. At that point the "burst limit" is no longer a limit.
Consider rate=2 and capacity=3. The first three cost-one requests are allowed immediately. The fourth is refused. After half a second, precisely one token should be available again. If you wait a week, the bucket still contains at most three.
I also support weighted requests:
limiter = TokenBucket(rate=2, capacity=3)
limiter.allow() # costs 1
limiter.allow(1.5) # costs 1.5
limiter.retry_after(2)
A high-cost request should fail without consuming any tokens if the requested balance isn't available.
Using time.time() for elapsed time is an appealing shortcut, but it's not a safe default for local elapsed-duration accounting. Wall-clock time can be adjusted by an operator or synchronized by the operating system.
Instead, I inject a clock function into the limiter, with time.monotonic as the production default:
import threading
import time
class TokenBucket:
def __init__(self, rate, capacity, *, clock=time.monotonic):
self._rate = float(rate)
self._capacity = float(capacity)
self._tokens = self._capacity
self._clock = clock
self._updated = self._clock()
self._lock = threading.RLock()
The public implementation also validates nonfinite, zero, negative and nonnumeric inputs before accepting them.
Because the clock is injected, tests don't need to sleep for any real time. Here's the idea:
class FakeClock:
def __init__(self):
self.now = 100.0
def __call__(self):
return self.now
clock = FakeClock()
bucket = TokenBucket(rate=2, capacity=3, clock=clock)
assert bucket.allow()
assert bucket.allow()
assert bucket.allow()
assert not bucket.allow()
clock.now += 0.5
assert bucket.allow()
When the fake clock moves backwards, the limiter should not gain extra credit or reset its reference timestamp in a way that double-counts an interval later.
That was worth its own test.
A naive limiter often does something like:
if bucket.tokens >= 1:
bucket.tokens -= 1
return True
Two threads can both observe the same token before either one deducts it. Under enough load, the advertised capacity becomes fiction.
In the version I tested, the lock covers both replenishment and deduction:
def allow(self, cost=1.0):
cost = self._positive_finite(cost, "cost")
with self._lock:
self._refill()
if self._tokens + 1e-12 < cost:
return False
self._tokens = max(0.0, self._tokens - cost)
return True
The small epsilon is explicitly a floating-point comparison tolerance; it's not a substitute for a carefully specified fixed-point model if your application needs exact financial-grade accounting.
For the concurrency test, I start 25 threads at the same barrier. Each tries to acquire a token twenty times. The bucket starts with 200 tokens and a fake clock that never advances. Across all 500 attempts, exactly 200 should succeed.
If 201 requests are allowed, the limiter is broken even if every single-thread test passes.
Suppose capacity is 2 tokens and a request costs 5. No amount of waiting can make that request possible: the bucket is capped at two.
The retry_after(cost) method returns math.inf in that case. For attainable requests it returns an estimate in seconds:
max(0, (requested_tokens - current_tokens) / refill_rate)
That is useful for a caller building an HTTP 429 Too Many Requests response, although it must still convert the interval into an appropriate header and be clear about rounding.
Separate tests check that a denied request leaves tokens unchanged and that numeric edge cases (NaN, infinity and True as a Python boolean) don't silently become accepted limiter parameters.
| Behavior | What the test checks |
|---|---|
| Initial burst | Three immediate requests are allowed, the next one denied |
| Time refill | Half a second at two tokens/second earns one token |
| Long idle period | Balance never rises above capacity |
| Fractional cost | Weighted requests don't overdraw the bucket |
| Impossible cost | Retry interval is infinite if cost exceeds capacity |
| Denied request | A failed acquire doesn't consume partial credit |
| Backward clock | A clock regression cannot mint duplicate tokens |
| Invalid numbers | Zero, negatives, infinity and NaN are rejected |
| Invalid types | Booleans, strings and None are rejected |
| Public API | The alias and public read-only properties behave as documented |
| Concurrent callers | 25 threads cannot exceed a shared 200-token capacity |
I executed the suite with:
python3 -m unittest discover -v
All 11 passed on the VM5 machine. That is evidence for the tested behavior, not a universal security certification.
This is deliberately a single-process, in-memory bucket. It's a good primitive for a local API worker or a small command-line application. It is not a production-ready distributed rate-limiting service.
For a multi-instance deployment, each process would otherwise maintain its own independent balance. You would need shared atomic state—often Redis or a purpose-built gateway—and a strategy for identity keys, memory eviction, denial-of-service resistance and monitoring.
I also didn't claim that the exact example gives fairness across many users. You need a separate keyed bucket for each actor and an eviction strategy for abandoned keys.
A small, well-tested tool is more useful than an impressive-sounding tool that promises features it doesn't actually implement.
The source and complete tests are public:
If you're dealing with a narrowly scoped Python or Node.js API integration, I've also made a no-cost project inquiry form. Please don't post secrets or confidential data in public issues; an inquiry is not a purchase or a promise of delivery.
Written with AI assistance and independently checked against runnable code. No fabricated customer testimonials, paid endorsement or guaranteed outcomes.