fix(tests): Fix flakes from cross-worker FLUSHALL and recycled project IDs (#128317)

This fixes three flaky backend tests that fail on master and then pass
on retry. There are two root causes, and each one has its own commit.

**FLUSHALL from other xdist workers.** The xdist workers, and the
per-worker Snuba instances, each use their own DB on one Redis server.
`test_circuit_breaker2.py`, `utils/sentry_apps/test_webhooks.py` and
`test_artifact_bundles.py` called `flushall()` on the single-host
client, and FLUSHALL deletes every DB on the server. That wipes the
state of the tests that other workers are running at the same moment:
- The snowflake counters reset in the middle of a test. The worker then
generates IDs that it already used, and five collisions in a row raise
`MaxSnowflakeRetryError`. Tests that create a lot of orgs or projects
hit this most, for example `CursoredSchedulerTest`,
`test_organization_trace::test_with_missing_group` and
`test_backfill_group_action_log`.
- The reprocessing state is deleted while an event is reprocessed.
`test_minidump_full::test_reprocessing_objectstore` then gets the
unsymbolicated event back and its snapshot does not match.

These tests now use `flushdb()`. The test plugin also makes `flushall()`
on a single-host Redis client raise, so a new test cannot bring the
problem back. Prefixed redis-cluster clients still allow it, because
they already delete only their own worker's keys.

**Recycled snowflake project IDs.**
`test_replay_data_export_no_replay_data` expects Snuba to have no
replays, but it never reset the replays dataset. Redis is flushed after
each test, so when the test runs in the same second as the test before
it, `default_project` gets the same ID. The replay that the earlier test
stored then gets exported. The test now uses the `replay_store` fixture,
which drops the dataset.

Not covered: the `assert None is not None` failure in
`test_reprocessing_objectstore` and the Kafka `_MAX_POLL_EXCEEDED` in
`test_full_minidump_invalid_extra`. I could not reproduce either one
locally.

<!-- junior-request-attribution:start -->
via @joshuarli.
<!-- junior-request-attribution:end -->

<!-- junior-session-footer:start -->
<!-- junior-conversation-id:agent%3A4fe45e7cb5955bf5a36ba1ff0d1eb304 -->

--

[View Junior
Session](https://junior-prod.sentry.dev/conversations/agent%3A4fe45e7cb5955bf5a36ba1ff0d1eb304)
[[Sentry]](https://sentry.sentry.io/explore/conversations/agent%3A4fe45e7cb5955bf5a36ba1ff0d1eb304/?project=4510944073809921)

<!-- junior-session-footer:end -->

---------

Co-authored-by: sentry-junior[bot] <264270552+sentry-junior[bot]@users.noreply.github.com>
Co-authored-by: Joshua Li <14209781+joshuarli@users.noreply.github.com>

Author: sentry-junior[bot]

src/sentry/testutils/pytest/sentry.py

@@ -10,7 +10,7 @@
 from datetime import datetime
 from hashlib import sha256
 from pathlib import Path
-from typing import TypeVar
+from typing import Any, TypeVar
 from unittest import mock
 
 import pytest
@@ -149,6 +149,27 @@ def pytest_ignore_collect(collection_path: Path, config: pytest.Config) -> bool
     return True if rel not in _COLLECT_ALLOWED_FILES else None
 
 
+def _forbid_single_host_flushall() -> None:
+    # The xdist workers and the per-worker Snuba instances share one Redis server and
+    # each use their own DB on it. FLUSHALL deletes every DB, which resets the snowflake
+    # ID counters of the other workers in the middle of their tests. They then generate
+    # IDs that they already used and fail with MaxSnowflakeRetryError.
+    from redis.client import Redis
+    from rediscluster import RedisCluster
+
+    flushall = Redis.flushall
+
+    def guarded_flushall(self: Redis, *args: Any, **kwargs: Any) -> Any:
+        # Prefixed cluster clients only delete the keys of this worker.
+        if isinstance(self, RedisCluster):
+            return flushall(self, *args, **kwargs)
+        raise AssertionError(
+            "FLUSHALL deletes the Redis DBs of all xdist workers. Use flushdb() instead."
+        )
+
+    Redis.flushall = guarded_flushall  # type: ignore[method-assign]
+
+
 def pytest_configure(config: pytest.Config) -> None:
     import warnings
 
@@ -169,6 +190,8 @@ def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line("markers", "migrations: requires --migrations")
     config.addinivalue_line("markers", "symbolicator: test requires access to symbolicator")
 
+    _forbid_single_host_flushall()
+
     if sys.platform == "darwin" and shutil.which("colima"):
         # This is the only way other than pytest --basetemp to change
         # the temproot. We'd like to keep invocations to just "pytest".

tests/sentry/debug_files/test_artifact_bundles.py

@@ -105,7 +105,7 @@ def get_indexed_files(project, release_name="", dist_name="", distinct=False):
 class ArtifactLookupTest(TestCase):
     def clear_cache(self):
         redis_client = get_redis_cluster_for_artifact_bundles()
-        redis_client.flushall()
+        redis_client.flushdb()
 
     def test_indexing_artifacts(self) -> None:
         self.clear_cache()
@@ -215,7 +215,7 @@ def test_indexing_artifacts_with_cached_indexing_state(self) -> None:
 
 class GetCachedBundlesIndexingStateTest(TestCase):
     def setUp(self) -> None:
-        get_redis_cluster_for_artifact_bundles().flushall()
+        get_redis_cluster_for_artifact_bundles().flushdb()
 
     def create_bundles(self, count: int) -> None:
         for _ in range(count):

tests/sentry/replays/test_data_export.py

@@ -120,8 +120,10 @@ def test_replay_data_export_no_replay_projects(  # type: ignore[no-untyped-def]
 @pytest.mark.snuba
 @requires_snuba
 def test_replay_data_export_no_replay_data(  # type: ignore[no-untyped-def]
-    default_organization, default_project
+    default_organization, default_project, replay_store
 ) -> None:
+    # `replay_store` drops the replays written by earlier tests. `default_project` can
+    # get the same snowflake ID as the project of an earlier test.
     # Setting has_replays flag because the export will skip projects it assumes do not have
     # replays.
     default_project.flags.has_replays = True

tests/sentry/testutils/pytest/test_sentry.py

@@ -7,8 +7,10 @@
 from collections import Counter
 from pathlib import Path
 
+import pytest
+
 import sentry
-from sentry.utils import json
+from sentry.utils import json, redis
 
 SCRIPTS = Path(sentry.__file__).parents[2] / ".github" / "workflows" / "scripts"
 SHARDS = 4
@@ -96,3 +98,11 @@ def test_filtered_run_spreads_selected_tests_over_every_shard(tmp_path: Path) ->
         )
     assert len(unsharded) == 20
     assert sharded == unsharded
+
+
+def test_flushall_on_single_host_redis_is_rejected() -> None:
+    client = redis.redis_clusters.get("default")
+    with pytest.raises(AssertionError, match="flushdb"):
+        client.flushall()
+    with pytest.raises(AssertionError, match="flushdb"):
+        client.pipeline().flushall()

tests/sentry/utils/sentry_apps/test_webhooks.py

@@ -244,7 +244,7 @@ def setUp(self):
             organization=self.organization, slug=self.sentry_app.slug
         )
         client = redis.redis_clusters.get(settings.SENTRY_RATE_LIMIT_REDIS_CLUSTER)
-        client.flushall()
+        client.flushdb()
 
     def _make_event(self, install=None):
         return AppPlatformEvent(

tests/sentry/utils/test_circuit_breaker2.py

@@ -129,7 +129,7 @@ def setUp(self) -> None:
         )
 
         # Clear all existing keys from redis
-        self.breaker.redis_pipeline.flushall()
+        self.breaker.redis_pipeline.flushdb()
         self.breaker.redis_pipeline.execute()
 
     def test_sets_default_values(self) -> None:
@@ -261,7 +261,7 @@ def setUp(self) -> None:
         )
 
         # Clear all existing keys from redis
-        self.breaker.redis_pipeline.flushall()
+        self.breaker.redis_pipeline.flushdb()
         self.breaker.redis_pipeline.execute()
 
     def test_increments_error_count(self) -> None:
@@ -435,7 +435,7 @@ def setUp(self) -> None:
         )
 
         # Clear all existing keys from redis
-        self.breaker.redis_pipeline.flushall()
+        self.breaker.redis_pipeline.flushdb()
         self.breaker.redis_pipeline.execute()
 
     def test_allows_request_in_non_broken_state_with_quota_remaining(self) -> None:
@@ -516,7 +516,7 @@ def setUp(self) -> None:
         )
 
         # Clear all existing keys from redis
-        self.breaker.redis_pipeline.flushall()
+        self.breaker.redis_pipeline.flushdb()
         self.breaker.redis_pipeline.execute()
 
     def test_sets_rate_based_defaults(self) -> None:
@@ -689,7 +689,7 @@ def test_record_success_noop_when_not_rate_based(self) -> None:
         count_breaker = MockCircuitBreaker(
             "count_only", DEFAULT_CONFIG, CountBasedTripStrategy.from_config(DEFAULT_CONFIG)
         )
-        count_breaker.redis_pipeline.flushall()
+        count_breaker.redis_pipeline.flushdb()
         count_breaker.redis_pipeline.execute()
 
         assert isinstance(count_breaker.trip_strategy, CountBasedTripStrategy)