fix(alerting): start the Slack periodic flush task once, not per config reload

update_values is re-invoked by the proxy's periodic deployment reconcile
(add_deployment_job -> _update_llm_router ->
_add_general_settings_from_db_config), which passes `alerting` whenever the
DB config carries that key. That branch created a periodic_flush task
unconditionally, and each one is a `while True` loop that never exits, so a
long-running proxy accumulates one task per reload interval for the lifetime
of the process. The sibling `alerting_args` branch was already guarded; this
routes every start site through one helper so the guard cannot be missed
again.

Also keeps a reference to the task: with the leak gone this is the only
flush task, and a bare create_task may be garbage collected mid-execution.

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -106,6 +106,7 @@ class SlackAlerting(CustomBatchLogger):
         self.default_webhook_url = default_webhook_url
         self.flush_lock = asyncio.Lock()
         self.periodic_started = False
+        self._periodic_flush_task: asyncio.Task | None = None
         self.hanging_request_check = AlertingHangingRequestCheck(
             slack_alerting_object=self,
         )
@@ -117,6 +118,20 @@ class SlackAlerting(CustomBatchLogger):
         self.digest_lock = asyncio.Lock()
         super().__init__(**kwargs, flush_lock=self.flush_lock)
 
+    def _ensure_periodic_flush_started(self) -> None:
+        """Start the periodic flush loop, at most once.
+
+        ``update_values`` is re-invoked on a timer by the proxy's deployment
+        refresh, so creating the task unconditionally leaks one ``while True``
+        task per call for the lifetime of the process.
+        """
+        if self.periodic_started:
+            return
+        # Keep a reference: a bare create_task may be garbage collected
+        # mid-execution, and this is now the only flush task.
+        self._periodic_flush_task = asyncio.create_task(self.periodic_flush())
+        self.periodic_started = True
+
     def update_values(
         self,
         alerting: list | None = None,
@@ -129,17 +144,14 @@ class SlackAlerting(CustomBatchLogger):
     ):
         if alerting is not None:
             self.alerting = alerting
-            asyncio.create_task(self.periodic_flush())
-            self.periodic_started = True
+            self._ensure_periodic_flush_started()
         if alerting_threshold is not None:
             self.alerting_threshold = alerting_threshold
         if alert_types is not None:
             self.alert_types = alert_types
         if alerting_args is not None:
             self.alerting_args = SlackAlertingArgs(**alerting_args)
-            if not self.periodic_started:
-                asyncio.create_task(self.periodic_flush())
-                self.periodic_started = True
+            self._ensure_periodic_flush_started()
         if alert_type_config is not None:
             for key, val in alert_type_config.items():
                 self.alert_type_config[key] = AlertTypeConfig(**val) if isinstance(val, dict) else val
@@ -1420,9 +1432,8 @@ Model Info:
             return
 
         # Start periodic flush if not already started
-        if not self.periodic_started and self.alerting is not None and len(self.alerting) > 0:
-            asyncio.create_task(self.periodic_flush())
-            self.periodic_started = True
+        if len(self.alerting) > 0:
+            self._ensure_periodic_flush_started()
 
         if "webhook" in self.alerting and alert_type == "budget_alerts" and user_info is not None:
             await self.send_webhook_alert(webhook_event=user_info)

tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py

@@ -164,6 +164,20 @@ class TestSlackAlerting(unittest.TestCase):
         self.slack_alerting.update_values(alerting_args={"slack_alerting": "True"})
         assert self.slack_alerting.periodic_started == True
 
+    # The proxy re-invokes update_values on a timer; it must not spawn a new
+    # periodic_flush task each time (each one is a `while True` loop that
+    # never exits, so they accumulate for the lifetime of the process).
+    @patch("asyncio.create_task")
+    def test_update_values_starts_periodic_task_only_once(self, mock_create_task):
+        mock_create_task.return_value = AsyncMock()
+
+        for _ in range(5):
+            self.slack_alerting.update_values(alerting=["slack"])
+            self.slack_alerting.update_values(alerting_args={"slack_alerting": "True"})
+
+        assert self.slack_alerting.periodic_started is True
+        assert mock_create_task.call_count == 1
+
     @patch("litellm.integrations.SlackAlerting.slack_alerting.datetime")
     def test_alert_type_in_formatted_message(self, mock_datetime):
         # Setup mocks