From b040fdd268b66cd7a3bfb2fbca65c84f735d0b0d Mon Sep 17 00:00:00 2001 From: vpetersson-bot Date: Tue, 20 Oct 2026 08:13:02 +0000 Subject: [PATCH] fix(sentry): thin a device's retry loop so it can't spend the quota The project's Sentry quota is being consumed faster than it replenishes, so errors stop being accepted partway through each billing period and every issue's lastSeen freezes on the day the allowance ran out. The cost of that is not the missing events, it is that "it has not recurred since" stops being evidence of anything -- a regression can land and leave no record. What spends the quota is depth, not breadth. A device stuck in a retry loop repeats one error thousands of times while the rest of the fleet contributes a handful each: in a fully recorded day one board produced 1,761 of an issue's 1,762 events (ANTHIAS-3H) and another 1,063 of 1,093 (ANTHIAS-2D). Two boards crowded out every other issue. Count occurrences per issue per process over a one-hour window and keep the 1st, 2nd, 4th, 8th, ..., dropping what falls between two powers of two. A loop costs O(log n) events instead of n -- 1,761 becomes 11 -- and the magnitude is not lost, because each surviving event carries its true running count in a throttle.occurrence tag. Every device still reports every distinct issue it hits, so nothing can be hidden by this; only repetition of something already reported is. The gate runs last in before_send on purpose, so the transient noise filtered above it never spends a key's budget. Issues are keyed by exception type plus the innermost Anthias frame, or for logging-integration events by logger plus the unformatted message template -- the template being what keeps "Timed out waiting for UP message from " and the same line for worker 27 one key rather than one per worker (ANTHIAS-1Q). An event matching neither shape is left unthrottled rather than given a coarse key that could merge unrelated issues. Co-Authored-By: Claude Opus 5 (1M context) Co-authored-by: multica-agent --- src/anthias_server/django_project/settings.py | 151 ++++++++- tests/test_sentry.py | 306 ++++++++++++++++++ 2 files changed, 455 insertions(+), 2 deletions(-) diff --git a/src/anthias_server/django_project/settings.py b/src/anthias_server/django_project/settings.py index fe7939796..ee98fb2a6 100644 --- a/src/anthias_server/django_project/settings.py +++ b/src/anthias_server/django_project/settings.py @@ -16,6 +16,8 @@ import secrets import socket import sys +import threading +import time import zoneinfo from collections.abc import Iterator from os import getenv @@ -159,6 +161,141 @@ def _is_late_soft_time_limit(exc: BaseException) -> bool: ) +# Per-issue event throttle. +# +# The project's Sentry quota is finite and is being spent faster than +# it replenishes, which leaves the project blind for most of each +# billing period: errors stop being accepted, and an issue's +# ``lastSeen`` freezes at the day the allowance ran out. "It has not +# recurred since" then means nothing, which is the part that hurts — +# a regression lands and nothing records it. +# +# What spends the quota is not breadth but depth. A device stuck in a +# retry loop repeats one error thousands of times while the rest of +# the fleet contributes a handful each: the worst offender observed +# in a fully recorded day was one board producing 1,761 of an issue's +# 1,762 events, and the second was another board producing 1,063 of +# 1,093. The first copy of that error is worth paying for and so is +# knowing how many devices hit it; the thousandth copy from the same +# board is what crowds out every other issue on the fleet. +# +# So each process counts occurrences per issue over a one-hour window +# and keeps the 1st, 2nd, 4th, 8th, 16th, ..., dropping what falls +# between two powers of two. A loop costs O(log n) events instead of +# n — 1,761 becomes 11 — and the magnitude survives the thinning, +# because each kept event carries its true running count in the +# ``throttle.occurrence`` tag. Every device still reports every +# distinct issue it hits, so no issue can be hidden by this; only +# repetition of one already reported is. +_SENTRY_THROTTLE_WINDOW_SECONDS = 3600 +# Bounds the bookkeeping on a long-lived process. A process producing +# more than this many *distinct* issues within one window has larger +# problems than its quota, and the overflow path only forgets windows, +# which can let more events through but never fewer. +_SENTRY_THROTTLE_MAX_KEYS = 512 + +_sentry_throttle_lock = threading.Lock() +# key -> (window started at, occurrences counted in that window) +_sentry_throttle_state: dict[tuple[str, ...], tuple[float, int]] = {} + + +def _throttle_key(event: Event, hint: Hint) -> tuple[str, ...] | None: + """A stand-in for the Sentry issue this event will group into. + + Sentry fingerprints server-side and the SDK cannot ask it, so this + approximates the grouping with what grouping itself keys on: the + exception type plus the innermost Anthias frame (falling back to + the innermost frame of any kind when our code appears nowhere), or + for a logging-integration event the logger and the *unformatted* + message template. Using the template is what keeps "Timed out + waiting for UP message from " and the same line + for worker 27 a single key rather than one per worker. + + Returns ``None`` when neither shape fits, which leaves the event + unthrottled: a key coarse enough to merge unrelated issues could + drop a report that has no duplicate anywhere, and this filter must + never be the reason a real bug goes unseen. + """ + exc_info = hint.get('exc_info') + if exc_info: + exc_type, _, tb = exc_info + innermost = '' + anthias_frame = '' + while tb is not None: + module = tb.tb_frame.f_globals.get('__name__', '') + innermost = f'{module}:{tb.tb_frame.f_code.co_name}' + # Prefer the innermost frame of *ours*: the third-party + # frames below it (django's cursor plumbing, redis' socket + # read) are shared by otherwise unrelated failures. + if module.startswith('anthias_'): + anthias_frame = innermost + tb = tb.tb_next + return ( + 'exc', + exc_type.__module__, + exc_type.__qualname__, + anthias_frame or innermost, + ) + record = hint.get('log_record') + if record is not None: + return ('log', record.name, str(record.msg)[:200]) + return None + + +def _evict_throttle_keys(now: float) -> None: + """Expired windows first, then the oldest, until back under the cap. + + The caller holds ``_sentry_throttle_lock``. + """ + for key, (window_start, _) in list(_sentry_throttle_state.items()): + if now - window_start >= _SENTRY_THROTTLE_WINDOW_SECONDS: + del _sentry_throttle_state[key] + excess = len(_sentry_throttle_state) - _SENTRY_THROTTLE_MAX_KEYS + if excess <= 0: + return + by_age = sorted( + _sentry_throttle_state, + key=lambda key: _sentry_throttle_state[key][0], + ) + for key in by_age[:excess]: + del _sentry_throttle_state[key] + + +def _throttled(event: Event, hint: Hint) -> bool: + """Whether this event is a suppressed repeat; counts it either way. + + Occurrences 1, 2, 4, 8, ... within the window survive; the ones in + between are dropped. A survivor past the first is tagged with the + running count, so the repetition the throttle removed is still + legible on the event that did get through. + """ + key = _throttle_key(event, hint) + if key is None: + return False + # Monotonic: a device that has never had a clock sync can step its + # wall clock by years the moment NTP lands, which would otherwise + # expire or freeze every window at once. + now = time.monotonic() + with _sentry_throttle_lock: + window_start, count = _sentry_throttle_state.get(key, (now, 0)) + if now - window_start >= _SENTRY_THROTTLE_WINDOW_SECONDS: + window_start, count = now, 0 + count += 1 + _sentry_throttle_state[key] = (window_start, count) + if len(_sentry_throttle_state) > _SENTRY_THROTTLE_MAX_KEYS: + _evict_throttle_keys(now) + # ``n & (n - 1)`` is zero exactly on the powers of two. + if count & (count - 1): + return True + if count > 1: + tags = event.setdefault('tags', {}) + # The protocol also allows a list of pairs; the SDK only ever + # builds the mapping, so skip the tag rather than guess. + if isinstance(tags, dict): + tags['throttle.occurrence'] = str(count) + return False + + def _sentry_before_send(event: Event, hint: Hint) -> Event | None: """Drop events that report expected transient states, not bugs. @@ -235,6 +372,12 @@ def _sentry_before_send(event: Event, hint: Hint) -> Event | None: import celery — the viewer imports these settings and its image ships no celery. This one rule is applied to the head exception only rather than down the chain; see the comment at the check. + + Anything that survives all of the above is a real report, and is + then passed through the repetition throttle (``_throttled``), + which thins one device's retry loop down to the powers of two so + a single looping board cannot spend the fleet's whole monthly + quota. That gate runs last, on log-only events too. """ # Imported lazily — this runs only when an event is about to send, # well after Django is configured, and avoids an import cycle at @@ -244,7 +387,7 @@ def _sentry_before_send(event: Event, hint: Hint) -> Event | None: exc_info = hint.get('exc_info') if not exc_info: - return event + return None if _throttled(event, hint) else event transient_redis = ( redis.exceptions.ConnectionError, redis.exceptions.TimeoutError, @@ -277,7 +420,11 @@ def _sentry_before_send(event: Event, hint: Hint) -> Event | None: or exc_cls.__module__.startswith('yt_dlp.') ): return None - return event + # The repetition throttle is the last gate deliberately: noise + # dropped above is not a report at all and must not spend a key's + # budget, or a loop of filtered redis blips would throttle the one + # real error raised alongside it. + return None if _throttled(event, hint) else event # celery's consumer, the embedded beat scheduler, and the redis diff --git a/tests/test_sentry.py b/tests/test_sentry.py index 451b4bdde..e9dc1173a 100644 --- a/tests/test_sentry.py +++ b/tests/test_sentry.py @@ -9,6 +9,10 @@ and capture calls are dropped. """ +import json +import logging +import time +from collections.abc import Iterator from pathlib import Path from unittest import mock @@ -76,6 +80,308 @@ def test_sentry_does_not_send_under_pytest() -> None: assert sentry_sdk.capture_message('must not be sent') is None +@pytest.fixture(autouse=True) +def _reset_sentry_throttle() -> Iterator[None]: + """Clear the before_send repetition throttle around every test. + + Its counters live at module scope in settings.py, so without this + a key counted by one test is still counted in the next, and an + "this event is kept" assertion starts failing on whichever test + happens to raise that exception type for the third time. + """ + from anthias_server.django_project.settings import ( + _sentry_throttle_state, + ) + + _sentry_throttle_state.clear() + yield + _sentry_throttle_state.clear() + + +class TestBeforeSendRepetitionThrottle: + """One looping device must not spend the whole Sentry quota. + + What exhausts the quota is depth rather than breadth: a single + board accounted for 1,761 of ANTHIAS-3H's 1,762 events, and + another for 1,063 of ANTHIAS-2D's 1,093, crowding out every other + issue for the rest of the billing period. before_send keeps the + powers of two within an hour-long window per issue and drops what + falls between them, so a loop costs O(log n) events. + """ + + @staticmethod + def _hint_for(exc: BaseException) -> dict[str, object]: + return {'exc_info': (type(exc), exc, None)} + + @staticmethod + def _log_hint(template: str, *args: object) -> dict[str, object]: + return { + 'log_record': logging.LogRecord( + name='celery.worker', + level=logging.ERROR, + pathname='worker.py', + lineno=1, + msg=template, + args=args, + exc_info=None, + ) + } + + def test_keeps_the_powers_of_two_and_drops_the_rest(self) -> None: + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + kept = [ + occurrence + for occurrence in range(1, 33) + if _sentry_before_send( + {'event_id': str(occurrence)}, + self._hint_for(ValueError('the same loop')), + ) + is not None + ] + assert kept == [1, 2, 4, 8, 16, 32] + + def test_first_event_of_a_window_is_untagged(self) -> None: + # Nothing has been thinned yet, so an occurrence tag would be + # noise on the overwhelmingly common single-event case. + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + event: Event = {'event_id': 'x'} + assert _sentry_before_send(event, self._hint_for(ValueError())) is ( + event + ) + assert 'tags' not in event + + def test_survivors_carry_the_true_occurrence_count(self) -> None: + # The thinning is lossy about *count*, so the count has to + # ride along: triage must be able to tell the 4th occurrence + # from the 4th distinct event. + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + tagged: list[str | None] = [] + for occurrence in range(1, 9): + event: Event = {'event_id': str(occurrence)} + if ( + _sentry_before_send(event, self._hint_for(ValueError())) + is not None + ): + tagged.append(event.get('tags', {}).get('throttle.occurrence')) + assert tagged == [None, '2', '4', '8'] + + def test_preserves_an_existing_tag_mapping(self) -> None: + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + # Occurrences 1, 2 and 4 are kept and 3 is dropped, so the + # last event here is the 4th — a tagged survivor. + for _ in range(4): + event: Event = {'event_id': 'x', 'tags': {'device_type': 'pi5'}} + _sentry_before_send(event, self._hint_for(ValueError())) + assert event['tags'] == { + 'device_type': 'pi5', + 'throttle.occurrence': '4', + } + + def test_distinct_issues_do_not_share_a_budget(self) -> None: + # The quota problem is depth, not breadth. A board looping on + # one error must never mask a different error beside it. + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + for _ in range(50): + _sentry_before_send( + {'event_id': 'x'}, self._hint_for(ValueError()) + ) + event: Event = {'event_id': 'other'} + assert _sentry_before_send(event, self._hint_for(TypeError())) is ( + event + ) + + def test_noise_dropped_earlier_does_not_spend_a_budget(self) -> None: + # The throttle runs last on purpose: a redis outage filtered + # out above is not a report, and must not throttle the real + # error raised alongside it. + import redis.exceptions + + from anthias_server.django_project.settings import ( + _sentry_before_send, + _sentry_throttle_state, + ) + + for _ in range(100): + _sentry_before_send( + {'event_id': 'x'}, + self._hint_for(redis.exceptions.ConnectionError()), + ) + assert _sentry_throttle_state == {} + + def test_one_log_template_is_one_key_across_its_arguments(self) -> None: + # "Timed out waiting for UP message from " + # and the same line for worker 27 are one Sentry issue, so + # they have to share one budget — keying on the formatted + # message would give every worker its own and throttle + # nothing (Sentry ANTHIAS-1Q). + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + kept = [ + worker + for worker in range(1, 17) + if _sentry_before_send( + {'event_id': str(worker)}, + self._log_hint( + 'Timed out waiting for UP message from %s', + f'ForkPoolWorker-{worker}', + ), + ) + is not None + ] + assert kept == [1, 2, 4, 8, 16] + + def test_different_log_templates_are_different_keys(self) -> None: + from anthias_server.django_project.settings import ( + _sentry_before_send, + ) + + for _ in range(20): + _sentry_before_send( + {'event_id': 'x'}, self._log_hint('first template %s', 'a') + ) + event: Event = {'event_id': 'y'} + assert _sentry_before_send( + event, self._log_hint('second template %s', 'a') + ) == (event) + + def test_an_event_it_cannot_key_is_never_throttled(self) -> None: + # A key coarse enough to merge unrelated issues could drop a + # report with no duplicate anywhere, so an unkeyable event is + # left alone rather than lumped together. + from anthias_server.django_project.settings import ( + _sentry_before_send, + _sentry_throttle_state, + ) + + for _ in range(20): + event: Event = {'event_id': 'x'} + assert _sentry_before_send(event, {}) == event + assert _sentry_throttle_state == {} + + def test_a_stale_window_starts_the_count_again(self) -> None: + from anthias_server.django_project.settings import ( + _SENTRY_THROTTLE_WINDOW_SECONDS, + _sentry_before_send, + _sentry_throttle_state, + _throttle_key, + ) + + hint = self._hint_for(ValueError()) + key = _throttle_key({'event_id': 'x'}, hint) + assert key is not None + # Rewind the window rather than patch the clock: the same + # state the throttle would be in an hour later, with no + # global time.monotonic monkeypatch for the rest of the suite. + _sentry_throttle_state[key] = ( + time.monotonic() - _SENTRY_THROTTLE_WINDOW_SECONDS - 1, + 9999, + ) + event: Event = {'event_id': 'x'} + assert _sentry_before_send(event, hint) == event + assert 'tags' not in event + assert _sentry_throttle_state[key][1] == 1 + + def test_expired_windows_are_evicted_before_live_ones(self) -> None: + from anthias_server.django_project.settings import ( + _SENTRY_THROTTLE_MAX_KEYS, + _SENTRY_THROTTLE_WINDOW_SECONDS, + _sentry_before_send, + _sentry_throttle_state, + ) + + stale = time.monotonic() - _SENTRY_THROTTLE_WINDOW_SECONDS - 1 + for index in range(_SENTRY_THROTTLE_MAX_KEYS + 100): + _sentry_throttle_state[('exc', 'm', f'E{index}', '')] = (stale, 1) + _sentry_before_send({'event_id': 'x'}, self._hint_for(ValueError())) + assert len(_sentry_throttle_state) == 1 + + def test_live_windows_are_capped_by_dropping_the_oldest(self) -> None: + from anthias_server.django_project.settings import ( + _SENTRY_THROTTLE_MAX_KEYS, + _sentry_before_send, + _sentry_throttle_state, + ) + + now = time.monotonic() + for index in range(_SENTRY_THROTTLE_MAX_KEYS + 100): + _sentry_throttle_state[('exc', 'm', f'E{index}', '')] = ( + now + index, + 1, + ) + _sentry_before_send({'event_id': 'x'}, self._hint_for(ValueError())) + assert len(_sentry_throttle_state) == _SENTRY_THROTTLE_MAX_KEYS + # The newest live windows are the ones still counting. + assert ('exc', 'm', 'E0', '') not in _sentry_throttle_state + assert ( + 'exc', + 'm', + f'E{_SENTRY_THROTTLE_MAX_KEYS + 99}', + '', + ) in _sentry_throttle_state + + +class TestThrottleKey: + """What the throttle treats as "the same issue".""" + + def test_prefers_the_innermost_anthias_frame(self) -> None: + from anthias_server.django_project.settings import _throttle_key + from anthias_server.lib import cec_client + + with mock.patch.object( + cec_client, 'connect_to_redis', side_effect=ValueError('boom') + ): + try: + cec_client.available() + except ValueError as exc: + key = _throttle_key( + {'event_id': 'x'}, + {'exc_info': (type(exc), exc, exc.__traceback__)}, + ) + assert key is not None + assert key[0] == 'exc' + assert key[3].startswith('anthias_server.lib.cec_client:') + + def test_falls_back_to_the_innermost_frame_without_our_code( + self, + ) -> None: + # No Anthias frame anywhere: the raise site is still the most + # specific thing available, so the key uses it rather than + # giving up and leaving a third-party loop unthrottled. + from anthias_server.django_project.settings import _throttle_key + + try: + json.loads('{nope') + except ValueError as exc: + key = _throttle_key( + {'event_id': 'x'}, + {'exc_info': (type(exc), exc, exc.__traceback__)}, + ) + assert key is not None + assert key[3].startswith('json') + + def test_has_no_key_for_an_event_with_neither_shape(self) -> None: + from anthias_server.django_project.settings import _throttle_key + + assert _throttle_key({'event_id': 'x'}, {}) is None + + class TestBeforeSendTransientNoise: """The before_send hook must drop expected transient states (redis restarting, an HTTP client hanging up) and nothing else.