feat: add thread-safe token bucket rate limiter utility

[bnusunny]

Summary

Adds a thread-safe rate limiter to samcli.lib.utils using the token bucket algorithm. This is useful for throttling API calls to AWS services to stay within service quotas.

Changes

• samcli/lib/utils/rate_limiter.py — RateLimiter class with:

  • try_acquire() — non-blocking token acquisition
  • acquire(timeout=) — blocking with optional timeout
  • Thread-safe via threading.Lock
  • Automatic token refill based on elapsed time

• tests/unit/lib/utils/test_rate_limiter.py — 8 unit tests covering:

  • Token acquisition and exhaustion
  • Refill over time
  • Timeout behavior
  • Input validation
  • Thread safety under contention

Usage

from samcli.lib.utils.rate_limiter import RateLimiter

limiter = RateLimiter(rate=10, burst=10)  # 10 req/s, burst of 10
if limiter.try_acquire():
    call_api()

[cortex-assistant]

Review: feat: add thread-safe token bucket rate limiter utility

Thanks for this addition — the token bucket algorithm is a solid choice and the overall structure is clean. However, there are a few issues that need to be addressed before merging, including a missing input validation that can silently corrupt state, a flawed thread-safety test, and some convention alignment.

Summary of findings

Severity	Count
🔴 Critical	1
🟠 Major	2
🟡 Minor	3
💅 Nit	1

See inline comments for details.

[cortex-assistant]

🔴 Critical: Negative tokens values corrupt the rate limiter state.

Neither try_acquire() nor acquire() validates the tokens parameter. Calling try_acquire(tokens=-1) will always succeed (since self._tokens >= -1 is always true) and then adds a token to the bucket (self._tokens -= -1), silently inflating the token count beyond burst.

Fix: Add validation at the top of both methods:

if tokens <= 0:
    raise ValueError("tokens must be positive")

[cortex-assistant]

🟠 Major: try_acquire(tokens=N) where N > burst will always return False silently.

If a caller requests more tokens than the bucket can ever hold, this is almost certainly a programming error — but the caller gets no feedback, just an eternal False (or an infinite spin in acquire() without timeout).

Fix: Validate in both methods:

if tokens > self._burst:
    raise ValueError(f"tokens ({tokens}) exceeds burst capacity ({self._burst})")

[cortex-assistant]

🟡 Minor: timeout: float = None should be timeout: Optional[float] = None.

The rest of the samcli codebase (e.g., boto_utils.py, async_utils.py) consistently uses Optional from typing. This also makes the type hint technically correct since None is not a float.

[cortex-assistant]

🟡 Minor: Missing module-level logger.

Every other non-trivial utility in samcli/lib/utils/ defines LOG = logging.getLogger(__name__). The acquire() method's busy-wait loop is a great candidate for LOG.debug() calls to aid diagnostics (e.g., logging when waiting for token refill).

[cortex-assistant]

💅 Nit: burst is typed as int but the <= 0 guard doesn't reject float values like 1.5. A caller passing burst=1.5 silently gets a float bucket capacity. Consider adding if not isinstance(burst, int): raise TypeError(...) to enforce the type hint.

[cortex-assistant]

🟠 Major: Thread-safety test assertion is flawed — will produce false negatives.

With rate=1000 tokens/sec, the bucket refills significantly during test execution (10 threads × 20 calls takes measurable wall time). The total acquired tokens can legitimately exceed the initial burst of 100, causing the assertion sum(results) <= 100 to fail sporadically.

Fix: Use a near-zero refill rate so the bucket doesn't meaningfully refill during the test:

limiter = RateLimiter(rate=0.001, burst=100)

This makes the assertion reliable while still testing thread safety of concurrent try_acquire() calls.

[cortex-assistant]

🟡 Minor: Missing test coverage for edge cases.

Consider adding tests for:

• try_acquire(tokens=-1) / try_acquire(tokens=0) — should raise ValueError after the fix above
• try_acquire(tokens=N) where N > burst — should raise ValueError
• acquire() with tokens > burst — verify it doesn't spin forever
• Concurrent acquire() calls (only try_acquire() is tested under concurrency)

[cortex-assistant]

I've completed the review of PR #55. Here's a summary of what I found:

Verdict: Request Changes — 7 inline comments submitted.

Key findings:

1. 🔴 Critical — Negative tokens corrupt state: try_acquire(tokens=-1) silently adds tokens to the bucket, bypassing the rate limit entirely. Both try_acquire() and acquire() need input validation.

2. 🟠 Major — tokens > burst is a silent trap: Requesting more tokens than the bucket capacity will always fail (or spin forever in acquire() without timeout) with no feedback to the caller.

3. 🟠 Major — Flawed thread-safety test: With rate=1000, the bucket refills significantly during test execution, making the sum(results) <= 100 assertion unreliable. Needs a near-zero refill rate.

4. 🟡 Minor items: Missing Optional[float] type hint, missing module-level logger (project convention), no type enforcement on burst parameter, and missing edge-case test coverage.