fix(reprocessing): Cull parallel reprocessing tasks

src/sentry/conf/server.py

@@ -2900,6 +2900,9 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 # How long reprocessing counters are kept in Redis before they expire.
 SENTRY_REPROCESSING_SYNC_TTL = 30 * 24 * 3600  # 30 days
 
+# How long the reprocessing page claims are kept in Redis before they expire.
+SENTRY_REPROCESSING_PAGE_CLAIM_TTL = 24 * 3600  # 1 day
+
 # How many events to query for at once while paginating through an entire
 # issue. Note that this needs to be kept in sync with the time-limits on
 # `sentry.tasks.reprocessing2.reprocess_group`. That task is responsible for

src/sentry/services/eventstore/reprocessing/base.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from datetime import datetime
 from typing import Any, TypedDict
 
@@ -24,6 +25,7 @@ class ReprocessingStore(Service):
         "start_reprocessing",
         "get_pending",
         "get_progress",
+        "try_claim_page",
     )
 
     def __init__(self, **options: Any) -> None:
@@ -81,3 +83,13 @@ def get_pending(self, group_id: int) -> Any:
 
     def get_progress(self, group_id: int) -> ReprocessingInfo | None:
         raise NotImplementedError()
+
+    def try_claim_page(
+        self,
+        project_id: int,
+        group_id: int,
+        new_group_id: int,
+        state: Mapping[str, str] | None,
+        claimant: str,
+    ) -> bool:
+        raise NotImplementedError()

src/sentry/services/eventstore/reprocessing/redis.py

@@ -1,4 +1,5 @@
 import uuid
+from collections.abc import Mapping
 from datetime import datetime
 from typing import Any
 
@@ -28,6 +29,12 @@ def _get_remaining_key(project_id: int, group_id: int) -> str:
     return f"re2:remaining:{{{project_id}:{group_id}}}"
 
 
+def _get_page_claim_key(
+    project_id: int, group_id: int, new_group_id: int, timestamp: str, event_id: str
+) -> str:
+    return f"re2:pageclaim:{{{project_id}:{group_id}}}:{new_group_id}:{timestamp}:{event_id}"
+
+
 class RedisReprocessingStore(ReprocessingStore):
     def __init__(self, **options: dict[str, Any]) -> None:
         cluster = options.pop("cluster", "default")
@@ -180,3 +187,18 @@ def get_progress(self, group_id: int) -> ReprocessingInfo | None:
         if info is None:
             return None
         return orjson.loads(info)
+
+    def try_claim_page(
+        self,
+        project_id: int,
+        group_id: int,
+        new_group_id: int,
+        state: Mapping[str, str] | None,
+        claimant: str,
+    ) -> bool:
+        timestamp = state["timestamp"] if state is not None else "start"
+        event_id = state["event_id"] if state is not None else "start"
+        key = _get_page_claim_key(project_id, group_id, new_group_id, timestamp, event_id)
+        if self.redis.set(key, claimant, nx=True, ex=settings.SENTRY_REPROCESSING_PAGE_CLAIM_TTL):
+            return True
+        return self.redis.get(key) == claimant

src/sentry/tasks/reprocessing2.py

@@ -20,6 +20,7 @@
 from sentry.search.eap.occurrences.query_utils import build_group_id_in_filter
 from sentry.services import eventstore
 from sentry.services.eventstore.models import Event
+from sentry.services.eventstore.reprocessing import reprocessing_store
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.tasks.process_buffer import buffer_incr
@@ -109,6 +110,33 @@ def reprocess_group(
 
     assert new_group_id is not None
 
+    # To the best of our knowledge we still have quite some `reprocess_group` tasks running in parallel, this logic
+    # is intended to cull all but one. This is temporary to recover from a bad state and should be dead code after that.
+    if activation_id:
+        try:
+            page_owned = reprocessing_store.try_claim_page(
+                project_id=project_id,
+                group_id=group_id,
+                new_group_id=new_group_id,
+                state=query_state,
+                claimant=activation_id,
+            )
+        except Exception:
+            logger.warning("reprocessing2.page_claim.error", exc_info=True)
+            page_owned = True
+        if not page_owned:
+            logger.info(
+                "reprocessing2.page_claim.culled",
+                extra={
+                    "project_id": project_id,
+                    "group_id": group_id,
+                    "new_group_id": new_group_id,
+                    "query_state": query_state,
+                    "activation_id": activation_id,
+                },
+            )
+            return
+
     query_state, events = task_run_batch_query(
         filter=eventstore.Filter(project_ids=[project_id], group_ids=[group_id]),
         batch_size=settings.SENTRY_REPROCESSING_PAGE_SIZE,

tests/sentry/services/eventstore/processing/test_redis_cluster.py

@@ -20,3 +20,22 @@ def test_mark_event_reprocessed() -> None:
     assert progress is not None
     assert progress.get("syncCount") == 10
     assert progress.get("totalEvents") == 20
+
+
+@use_redis_cluster()
+def test_try_claim_page() -> None:
+    store = RedisReprocessingStore()
+    project_id = 1
+    group_id = 2
+    new_group_id = 3
+    state = {"timestamp": "2026-08-04T06:10:59+00:00", "event_id": "42"}
+
+    # First claim is ok, and reclaiming from same claimant is a NOOP.
+    assert store.try_claim_page(project_id, group_id, new_group_id, state, claimant="A")
+    assert store.try_claim_page(project_id, group_id, new_group_id, state, claimant="A")
+
+    # Claiming from another claimant should not work.
+    assert not store.try_claim_page(project_id, group_id, new_group_id, state, claimant="B")
+
+    # Different reprocessing run but same state should be unaffected.
+    assert store.try_claim_page(project_id, 4, 5, state, claimant="B")