feat(lw-deletions): Use PartMutation metric for backpressure signal

Switch the lw-deletions consumer's backpressure signal from "number of
incomplete mutations" (system.mutations WHERE is_done=0) to the
PartMutation metric from system.metrics. This gives a real-time
measurement of how many parts are actively being mutated, which
correlates directly with CPU usage.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: onewland

snuba/lw_deletions/strategy.py

@@ -42,7 +42,7 @@
 from snuba.web.delete_query import (
     TooManyOngoingMutationsError,
     _execute_query,
-    _num_ongoing_mutations,
+    _num_parts_currently_mutating,
 )
 
 TPayload = TypeVar("TPayload")
@@ -322,23 +322,23 @@ def _execute_single_delete(
             time.sleep(delay_ms / 1000.0)
 
     def _check_ongoing_mutations(self, skip_throttle: bool = False) -> None:
-        max_ongoing_mutations = typing.cast(
+        max_parts_mutating = typing.cast(
             int,
             get_int_config(
-                "max_ongoing_mutations_for_delete",
-                default=settings.MAX_ONGOING_MUTATIONS_FOR_DELETE,
+                "max_parts_mutating_for_delete",
+                default=settings.MAX_PARTS_MUTATING_FOR_DELETE,
             ),
         )
 
         # Fast path: local counter already exceeds limit, no need to query CH
-        if self.__local_inflight_count > max_ongoing_mutations:
+        if self.__local_inflight_count > max_parts_mutating:
             now = time.time()
             if (
                 self.__last_ongoing_mutations_check is not None
                 and now - self.__last_ongoing_mutations_check < 1.0
             ):
                 raise TooManyOngoingMutationsError(
-                    f"local inflight count {self.__local_inflight_count} exceeds max {max_ongoing_mutations}"
+                    f"local inflight count {self.__local_inflight_count} exceeds max {max_parts_mutating}"
                 )
             # Fall through to reconcile with CH
 
@@ -352,14 +352,14 @@ def _check_ongoing_mutations(self, skip_throttle: bool = False) -> None:
                 "ongoing mutations check is throttled to once per second"
             )
         start = time.time()
-        ongoing_mutations = _num_ongoing_mutations(self.__storage.get_cluster(), self.__tables)
+        parts_mutating = _num_parts_currently_mutating(self.__storage.get_cluster())
         self.__last_ongoing_mutations_check = time.time()
         # Reconcile: trust CH as source of truth for completions
-        self.__local_inflight_count = ongoing_mutations
+        self.__local_inflight_count = parts_mutating
         self.__metrics.timing("ongoing_mutations_query_ms", (time.time() - start) * 1000)
-        if ongoing_mutations > max_ongoing_mutations:
+        if parts_mutating > max_parts_mutating:
             raise TooManyOngoingMutationsError(
-                f"{ongoing_mutations} mutations for {self.__tables} table(s) is above max ongoing mutations: {max_ongoing_mutations} "
+                f"{parts_mutating} parts mutating is above max: {max_parts_mutating} "
             )
 
     def close(self) -> None:

snuba/settings/__init__.py

@@ -469,6 +469,7 @@ class RedisClusters(TypedDict):
 VALIDATE_DATASET_YAMLS_ON_STARTUP = False
 
 MAX_ONGOING_MUTATIONS_FOR_DELETE = 5
+MAX_PARTS_MUTATING_FOR_DELETE = 20
 LW_DELETES_PARTITION_TRACKING_TTL = 3600
 LW_DELETES_PER_SUBMIT_BUDGET = 5
 SNQL_DISABLED_DATASETS: set[str] = set([])

snuba/web/delete_query.py

@@ -203,6 +203,25 @@ def _num_ongoing_mutations(cluster: ClickhouseCluster, tables: Sequence[str]) ->
     )
 
 
+def _num_parts_currently_mutating(cluster: ClickhouseCluster) -> int:
+    """
+    Returns the number of parts currently being mutated across the cluster.
+    Uses the PartMutation metric from system.metrics, which directly correlates
+    with CPU usage from ongoing mutations.
+    """
+    if cluster.is_single_node():
+        query = "SELECT value FROM system.metrics WHERE metric = 'PartMutation'"
+    else:
+        query = f"""
+SELECT max(value)
+FROM clusterAllReplicas('{cluster.get_clickhouse_cluster_name()}', 'system', metrics)
+WHERE metric = 'PartMutation'
+"""
+    return int(
+        cluster.get_query_connection(ClickhouseClientSettings.QUERY).execute(query).results[0][0]
+    )
+
+
 def deletes_are_enabled() -> bool:
     return bool(get_config("storage_deletes_enabled", 1))
 

tests/lw_deletions/test_lw_deletions.py

@@ -59,7 +59,7 @@ def generate_message() -> Iterator[Message[KafkaPayload]]:
         i += 1
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_multiple_batches_strategies(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -85,7 +85,7 @@ def test_multiple_batches_strategies(mock_execute: Mock, mock_num_mutations: Moc
     assert commit_step.submit.call_count == 2
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_clickhouse_settings(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -125,7 +125,7 @@ def test_clickhouse_settings(mock_execute: Mock, mock_num_mutations: Mock) -> No
     assert commit_step.submit.call_count == 2
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_single_batch(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -160,16 +160,16 @@ def test_single_batch(mock_execute: Mock, mock_num_mutations: Mock) -> None:
     assert commit_step.submit.call_count == 1
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=10)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=100)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_too_many_mutations(mock_execute: Mock, mock_num_mutations: Mock) -> None:
     """
     Before we execute the DELETE FROM query, we check to see how many
-    ongoing mutations there are.If there are more ongoing mutations than
-    the max allows, we raise MessageRejected and back pressure is applied.
+    parts are currently mutating. If there are more than the max allows,
+    we raise MessageRejected and back pressure is applied.
 
-    The max is 5 (the default) and our mocked ongoing mutations is 10.
+    The max is 20 (the default) and our mocked parts mutating is 100.
     """
     commit_step = Mock()
     metrics = Mock()
@@ -221,7 +221,7 @@ def _make_single_message(
     )
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @patch.object(
     FormatQuery,
@@ -275,7 +275,7 @@ def test_split_by_partition_enabled(mock_execute: Mock, mock_num_mutations: Mock
     assert len(increment_calls) == 3
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_split_by_partition_disabled(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -304,7 +304,7 @@ def test_split_by_partition_disabled(mock_execute: Mock, mock_num_mutations: Moc
     assert commit_step.submit.call_count == 1
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_split_by_partition_redis_tracking(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -394,7 +394,7 @@ def test_split_by_partition_redis_tracking(mock_execute: Mock, mock_num_mutation
     assert commit_step.submit.call_count == 1
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_split_by_partition_fallback(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -436,7 +436,7 @@ def test_split_by_partition_fallback(mock_execute: Mock, mock_num_mutations: Moc
     assert commit_step.submit.call_count == 1
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_partition_date_filtering(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -500,7 +500,7 @@ def test_local_inflight_counter_reconciles_with_ch(mock_execute: Mock) -> None:
     metrics = Mock()
     storage = get_writable_storage(StorageKey("search_issues"))
 
-    state.set_config("max_ongoing_mutations_for_delete", 3)
+    state.set_config("max_parts_mutating_for_delete", 3)
     state.set_config("lw_deletes_split_by_partition_search_issues", 1)
     state.set_config("lw_deletes_per_submit_budget", 100)
 
@@ -510,7 +510,7 @@ def test_local_inflight_counter_reconciles_with_ch(mock_execute: Mock) -> None:
 
     # CH always returns 0 (stale) — all 3 partitions should execute
     with (
-        patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=0),
+        patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=0),
         patch.object(
             format_query,
             "_FormatQuery__partition_column",
@@ -548,11 +548,11 @@ def test_local_counter_increments_after_each_delete(mock_execute: Mock) -> None:
     metrics = Mock()
     storage = get_writable_storage(StorageKey("search_issues"))
 
-    state.set_config("max_ongoing_mutations_for_delete", 5)
+    state.set_config("max_parts_mutating_for_delete", 5)
 
     format_query = FormatQuery(Mock(), storage, SearchIssuesFormatter(), metrics)
 
-    with patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=0):
+    with patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=0):
         # Initial CH check: reconciles local counter to 0
         format_query._check_ongoing_mutations()
         assert format_query._FormatQuery__local_inflight_count == 0  # type: ignore[attr-defined]
@@ -570,7 +570,7 @@ def test_local_counter_increments_after_each_delete(mock_execute: Mock) -> None:
         assert format_query._FormatQuery__local_inflight_count == 0  # type: ignore[attr-defined]
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @pytest.mark.redis_db
 def test_per_submit_budget_exhaustion(mock_execute: Mock, mock_num_mutations: Mock) -> None:
@@ -655,7 +655,7 @@ def test_per_submit_budget_exhaustion(mock_execute: Mock, mock_num_mutations: Mo
         assert len(members) == 5
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @patch("snuba.lw_deletions.strategy.time.sleep")
 @pytest.mark.redis_db
@@ -684,7 +684,7 @@ def test_inter_delete_delay(mock_sleep: Mock, mock_execute: Mock, mock_num_mutat
     mock_sleep.assert_called_once_with(0.2)
 
 
-@patch("snuba.lw_deletions.strategy._num_ongoing_mutations", return_value=1)
+@patch("snuba.lw_deletions.strategy._num_parts_currently_mutating", return_value=1)
 @patch("snuba.lw_deletions.strategy._execute_query")
 @patch("snuba.lw_deletions.strategy.time.sleep")
 @pytest.mark.redis_db