fix: Propagate expected exceptions from rate limit functions (#7736)

Stop reporting errors on expected transient exceptions during allocation
policy evaluation.

`rate_limit_start_request` and `rate_limit_finish_request` have broad
`except Exception` handlers that call `logger.exception()`, which
reports errors even for expected conditions like Redis timeouts and
unexpected pipeline result counts (`StopIteration`). However, the base
`AllocationPolicy` class can handle these at a higher level —
incrementing a `fail_open` metric and continuing without error logging.

This change re-raises `RedisTimeoutError` and `StopIteration` from the
low-level rate limit functions so they propagate to the allocation
policy base class, which handles them by failing open with a metric
(`reason` tagged by exception type) and avoiding noisy error reports.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: phacops

snuba/query/allocation_policies/__init__.py

@@ -483,13 +483,13 @@ def get_quota_allowance(
                     quota_unit=NO_UNITS,
                     suggestion=NO_SUGGESTION,
                 )
-            except RedisTimeoutError:
-                # Emit metric for timeout, but don't log since this is expected
-                # when Redis is slow. We fail open to avoid blocking requests.
+            except (RedisTimeoutError, StopIteration) as e:
+                # Expected transient errors (Redis timeouts, unexpected pipeline
+                # result counts). Fail open to avoid blocking requests.
                 self.metrics.increment(
                     "fail_open",
                     1,
-                    tags={"method": "get_quota_allowance", "reason": "redis_timeout"},
+                    tags={"method": "get_quota_allowance", "reason": type(e).__name__},
                 )
                 return DEFAULT_PASSTHROUGH_POLICY.get_quota_allowance(tenant_ids, query_id)
             except Exception:
@@ -553,11 +553,9 @@ def update_quota_balance(
         except InvalidTenantsForAllocationPolicy:
             # the policy did not do anything because the tenants were invalid, updating is also not necessary
             pass
-        except RedisTimeoutError:
-            # Emit metric for timeout, but don't log since this is expected
-            # when Redis is slow. We fail open to avoid blocking requests.
+        except (RedisTimeoutError, StopIteration) as e:
             self.metrics.increment(
-                "fail_open", 1, tags={"method": "update_quota_balance", "reason": "redis_timeout"}
+                "fail_open", 1, tags={"method": "update_quota_balance", "reason": type(e).__name__}
             )
         except Exception:
             self.metrics.increment("fail_open", 1, tags={"method": "update_quota_balance"})

snuba/state/rate_limit.py

@@ -11,6 +11,8 @@
 from typing import Any, Iterator, MutableMapping, Optional, Sequence, Type, cast
 from typing import ChainMap as TypingChainMap
 
+from redis.exceptions import TimeoutError as RedisTimeoutError
+
 from snuba import environment, state
 from snuba.redis import RedisClientKey, get_redis_client
 from snuba.state import get_configs, set_config
@@ -279,6 +281,14 @@ def rate_limit_start_request(
                 concurrent = sum(next(pipe_results) for _ in range(rate_limit_shard_factor))
             else:
                 concurrent = 0
+        except RedisTimeoutError:
+            raise
+        except StopIteration:
+            metrics.increment(
+                "rate_limit_fail_open",
+                tags={"reason": "StopIteration", "func": "rate_limit_start_request"},
+            )
+            return RateLimitStats(rate=-1, concurrent=-1)
         except Exception as ex:
             # if something goes wrong, we don't want to block the request,
             # set the values such that they pass under any limit
@@ -315,6 +325,8 @@ def rate_limit_finish_request(
                 pipe.zincrby(query_bucket, -float(max_query_duration_s), query_id)
                 pipe.expire(query_bucket, max_query_duration_s)
                 pipe.execute()
+    except RedisTimeoutError:
+        raise
     except Exception as ex:
         logger.exception(ex)