From fe573b8e509ef0782a30489b029c75fc87797f43 Mon Sep 17 00:00:00 2001 From: Jayden Mapasure Date: Fri, 9 Oct 2026 07:31:38 +0200 Subject: [PATCH 1/5] fix: atomically enforce distributed rate limits using Redis WATCH --- app/services/rate_limiter.py | 85 +++++++++++++++++++----------------- 1 file changed, 45 insertions(+), 40 deletions(-) diff --git a/app/services/rate_limiter.py b/app/services/rate_limiter.py index d1802b5..bbd2e36 100644 --- a/app/services/rate_limiter.py +++ b/app/services/rate_limiter.py @@ -1,11 +1,11 @@ """Redis-backed distributed sliding-window rate limiter.""" -import random import time +import uuid from dataclasses import dataclass from redis.asyncio import Redis -from redis.exceptions import RedisError +from redis.exceptions import RedisError, WatchError from app.core.config import get_settings from app.core.exceptions import RateLimitExceededError @@ -61,45 +61,50 @@ async def check_rate_limit( clear_before = now - window_seconds redis_key = f"ratelimit:{key_identifier}" - try: - pipe = self.redis.pipeline(transaction=True) - pipe.zremrangebyscore(redis_key, "-inf", clear_before) - pipe.zcard(redis_key) - pipe.zrange(redis_key, 0, 0, withscores=True) - results = await pipe.execute() - - current_requests = int(results[1]) - oldest_entries = results[2] - - if current_requests < limit: - member = f"{now}:{random.randint(10000, 99999)}" - add_pipe = self.redis.pipeline(transaction=True) - add_pipe.zadd(redis_key, {member: now}) - add_pipe.expire(redis_key, int(window_seconds) + 1) - await add_pipe.execute() - - remaining = limit - current_requests - 1 - reset_epoch = int(now + window_seconds) - return RateLimitResult( - allowed=True, - limit=limit, - remaining=remaining, - reset_epoch=reset_epoch, - degraded=False, - ) - else: - reset_epoch = int(now + window_seconds) - if oldest_entries and len(oldest_entries) > 0: - oldest_score = float(oldest_entries[0][1]) - reset_epoch = int(oldest_score + window_seconds) + if limit < 1 or window_seconds < 1: + raise ValueError("rate limit and window must be positive") - return RateLimitResult( - allowed=False, - limit=limit, - remaining=0, - reset_epoch=reset_epoch, - degraded=False, - ) + try: + # WATCH makes the quota check and insertion a single optimistic + # transaction. Concurrent callers retry instead of oversubscribing. + for _ in range(20): + async with self.redis.pipeline(transaction=True) as pipe: + try: + await pipe.watch(redis_key) + # Do not mutate the watched key before MULTI. + entries = await pipe.zrangebyscore( + redis_key, clear_before, "+inf", withscores=True + ) + current_requests = len(entries) + reset_epoch = ( + int(float(entries[0][1]) + window_seconds + 1) + if entries else int(now + window_seconds) + ) + if current_requests >= limit: + await pipe.unwatch() + return RateLimitResult( + allowed=False, limit=limit, remaining=0, + reset_epoch=reset_epoch, + ) + + member = uuid.uuid4().hex + pipe.multi() + pipe.zremrangebyscore(redis_key, "-inf", clear_before) + pipe.zadd(redis_key, {member: now}) + pipe.expire(redis_key, window_seconds + 1) + await pipe.execute() + return RateLimitResult( + allowed=True, limit=limit, + remaining=limit - current_requests - 1, + reset_epoch=reset_epoch, + ) + except WatchError: + # Another request consumed quota while we were checking. + # Refresh the clock before retrying. + now = time.time() + clear_before = now - window_seconds + continue + raise RedisError("rate limiter contention retry budget exhausted") except RedisError as exc: logger.error( From 8a92c6559c13aba2a842ddee2d31089bad89465a Mon Sep 17 00:00:00 2001 From: Jayden Mapasure Date: Fri, 9 Oct 2026 07:31:43 +0200 Subject: [PATCH 2/5] perf: count active rate-limit events without loading entire window --- app/services/rate_limiter.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/app/services/rate_limiter.py b/app/services/rate_limiter.py index bbd2e36..4b434d0 100644 --- a/app/services/rate_limiter.py +++ b/app/services/rate_limiter.py @@ -72,10 +72,13 @@ async def check_rate_limit( try: await pipe.watch(redis_key) # Do not mutate the watched key before MULTI. + current_requests = await pipe.zcount( + redis_key, f"({clear_before}", "+inf" + ) entries = await pipe.zrangebyscore( - redis_key, clear_before, "+inf", withscores=True + redis_key, f"({clear_before}", "+inf", start=0, num=1, + withscores=True, ) - current_requests = len(entries) reset_epoch = ( int(float(entries[0][1]) + window_seconds + 1) if entries else int(now + window_seconds) From e4ba203fc0ad2fb95454173bbceb5be88783a521 Mon Sep 17 00:00:00 2001 From: Jayden Mapasure Date: Fri, 9 Oct 2026 07:31:48 +0200 Subject: [PATCH 3/5] test: cover concurrent quota enforcement and invalid limits --- tests/unit/test_rate_limiter.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/unit/test_rate_limiter.py b/tests/unit/test_rate_limiter.py index 4e49d25..8a5f8ee 100644 --- a/tests/unit/test_rate_limiter.py +++ b/tests/unit/test_rate_limiter.py @@ -119,3 +119,24 @@ def pipeline(self, *args, **kwargs): with pytest.raises(RateLimitExceededError) as exc_info: await limiter.check_rate_limit("client_broken", limit_override=10) assert "fail-closed" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_concurrent_requests_never_exceed_quota(test_redis: FakeRedis) -> None: + """Multiple callers must not consume the same remaining quota slot.""" + limiter = RateLimiter(redis_client=test_redis) + results = await asyncio.gather( + *(limiter.check_rate_limit("concurrent-key", limit_override=5) for _ in range(12)) + ) + assert sum(result.allowed for result in results) == 5 + assert sum(not result.allowed for result in results) == 7 + assert await test_redis.zcard("ratelimit:concurrent-key") == 5 + + +@pytest.mark.asyncio +async def test_invalid_rate_limit_configuration(test_redis: FakeRedis) -> None: + limiter = RateLimiter(redis_client=test_redis) + with pytest.raises(ValueError, match="positive"): + await limiter.check_rate_limit("invalid-limit", limit_override=0) + with pytest.raises(ValueError, match="positive"): + await limiter.check_rate_limit("invalid-window", window_seconds=0) From 2e6bbc28e85315ad2ef5a035d98db64f143e429c Mon Sep 17 00:00:00 2001 From: Jayden Mapasure Date: Fri, 9 Oct 2026 07:31:55 +0200 Subject: [PATCH 4/5] fix: do not retry permanent 4xx webhook failures --- app/services/webhook.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/app/services/webhook.py b/app/services/webhook.py index 097ab76..c9bb450 100644 --- a/app/services/webhook.py +++ b/app/services/webhook.py @@ -10,7 +10,7 @@ from tenacity import ( AsyncRetrying, RetryCallState, - retry_if_exception_type, + retry_if_exception, stop_after_attempt, wait_exponential, ) @@ -24,6 +24,12 @@ logger = get_logger("services.webhook") +def _retryable_webhook_error(exc: BaseException) -> bool: + if isinstance(exc, httpx.HTTPStatusError): + return exc.response.status_code == 429 or exc.response.status_code >= 500 + return isinstance(exc, httpx.RequestError) + + class WebhookService: """Delivers HMAC-signed webhook events with retry handling and DB audit persistence.""" @@ -106,7 +112,7 @@ def log_retry(retry_state: RetryCallState) -> None: reraise=True, stop=stop_after_attempt(max_retries), wait=wait_exponential(multiplier=1.0, min=0.5, max=5), - retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)), + retry=retry_if_exception(_retryable_webhook_error), before_sleep=log_retry, ): with attempt: From e0a4dfb8ea93099011d0b6f705b8f2ab185b11b4 Mon Sep 17 00:00:00 2001 From: Jayden Mapasure Date: Fri, 9 Oct 2026 07:32:00 +0200 Subject: [PATCH 5/5] test: assert webhook 400 errors are not retried --- tests/unit/test_webhooks.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/unit/test_webhooks.py b/tests/unit/test_webhooks.py index 07f59a2..fe8d575 100644 --- a/tests/unit/test_webhooks.py +++ b/tests/unit/test_webhooks.py @@ -85,3 +85,24 @@ async def mock_handler(_req: httpx.Request) -> httpx.Response: assert success is True assert attempts == 2 + + +@pytest.mark.asyncio +async def test_webhook_does_not_retry_permanent_client_error() -> None: + attempts = 0 + + async def handler(_request: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + return httpx.Response(400, text="Invalid payload") + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + service = WebhookService(http_client=client) + delivered = await service.dispatch( + webhook_url="https://client.example.com/events", + request_id="req_bad_webhook", + event_type="generation.completed", + payload={"result": "done"}, + ) + assert delivered is False + assert attempts == 1