Merge 502c7d2953 into add7dea9aa

litellm/proxy/common_utils/scheduled_job_stagger.py

@@ -33,9 +33,9 @@ import hashlib
 import os
 import socket
 from collections.abc import Callable, Mapping, Sequence
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from types import MappingProxyType
-from typing import Final, Protocol
+from typing import Final, Protocol, overload
 
 from apscheduler.events import EVENT_JOB_SUBMITTED
 from apscheduler.triggers.base import BaseTrigger
@@ -108,6 +108,19 @@ class JobSubmission(Protocol):
     def scheduled_run_times(self) -> Sequence[datetime]: ...
 
 
+@overload
+def _shift(dt: None, delta: timedelta) -> None: ...
+@overload
+def _shift(dt: datetime, delta: timedelta) -> datetime: ...
+def _shift(dt: datetime | None, delta: timedelta) -> datetime | None:
+    if dt is None:
+        return None
+    if dt.tzinfo is not None:
+        # Perform arithmetic in UTC to preserve monotonic timeline across DST transitions (#48286)
+        return (dt.astimezone(timezone.utc) + delta).astimezone(dt.tzinfo)
+    return dt + delta
+
+
 class _OffsetTrigger:
     """
     Delegates to ``base`` on a clock rolled back by ``offset``, then rolls the answer
@@ -125,9 +138,9 @@ class _OffsetTrigger:
         self.offset = offset
 
     def get_next_fire_time(self, previous_fire_time: datetime | None, now: datetime) -> datetime | None:
-        shifted_previous: Final = None if previous_fire_time is None else previous_fire_time - self.offset
-        next_fire_time: Final = self.base.get_next_fire_time(shifted_previous, now - self.offset)
-        return None if next_fire_time is None else next_fire_time + self.offset
+        shifted_previous: Final = _shift(previous_fire_time, -self.offset)
+        next_fire_time: Final = self.base.get_next_fire_time(shifted_previous, _shift(now, -self.offset))
+        return _shift(next_fire_time, self.offset)
 
     def __str__(self) -> str:
         return f"{self.base}[+{int(self.offset.total_seconds())}s]"

litellm/proxy/proxy_server.py

@@ -1916,7 +1916,7 @@ async def _proxy_lifespan(app: FastAPI, resources: AsyncExitStack) -> AsyncGener
         "disable_model_info_refresh" in general_settings and general_settings["disable_model_info_refresh"] is True
     )
     model_info_scheduler: Final = (
-        None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler()
+        None if model_info_refresh_disabled else scheduler if scheduler is not None else AsyncIOScheduler(timezone="UTC")
     )
     if model_info_scheduler is not None:
         model_info_scheduler.add_job(
@@ -11587,8 +11587,8 @@ class ProxyStartupEvent:
             executors={
                 "default": scheduler_executor,
             },
-            # Disable timezone awareness to reduce computation
-            timezone=None,
+            # Explicitly use UTC timezone to prevent infinite loops and unbounded memory growth during DST transitions (#48285)
+            timezone="UTC",
         )
 
         # Use fixed intervals with small random offset instead of jitter

tests/unit/proxy/common_utils/test_scheduled_job_stagger.py

@@ -3,6 +3,7 @@ import logging
 import os
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
+from zoneinfo import ZoneInfo
 
 import pytest
 from apscheduler.executors.asyncio import AsyncIOExecutor
@@ -14,6 +15,7 @@ from apscheduler.triggers.interval import IntervalTrigger
 from litellm.constants import PTU_ROLLUP_JOB_ID, PTU_ROLLUP_LOCK_TTL_SECONDS
 from litellm.proxy._types import ScheduledJobStaggerSettings
 from litellm.proxy.common_utils.scheduled_job_stagger import (
+    _OffsetTrigger,
     apply_scheduled_job_stagger,
     attach_job_timing_logger,
     offset_seconds,
@@ -309,3 +311,35 @@ def test_job_timing_is_logged_with_scheduled_and_actual_start(caplog):
     assert f"scheduled_run_time={scheduled.isoformat()}" in message
     assert "actual_start_time=" in message
     assert "delay=2." in message
+
+
+def test_offset_trigger_dst_fallback_does_not_loop():
+    """
+    Regression test for #48285 / #48286.
+    During DST fall-back (repeated hour, fold=1), naive timedelta arithmetic drops fold
+    and sends shifted_previous 1 hour back in time, causing Job._get_run_times to spin
+    infinitely and exhaust memory. UTC-safe shifting ensures monotonic progress.
+    """
+    tz = ZoneInfo("Europe/Paris")
+    base = IntervalTrigger(seconds=60, timezone=tz)
+    staggered = _OffsetTrigger(base, timedelta(seconds=30))
+
+    def count_advances(trigger, now: datetime, cap: int = 100) -> int:
+        nrt = datetime.fromtimestamp(now.timestamp() - 90, tz)
+        n = 0
+        while nrt and nrt.astimezone(timezone.utc) <= now.astimezone(timezone.utc):
+            n += 1
+            if n >= cap:
+                return cap
+            nrt = trigger.get_next_fire_time(nrt, now)
+        return n
+
+    for label, now in [
+        ("01:30 CEST", datetime(2026, 10, 25, 1, 30, tzinfo=tz)),
+        ("02:30 CET (repeated hour)", datetime(2026, 10, 25, 2, 30, fold=1, tzinfo=tz)),
+        ("03:30 CET", datetime(2026, 10, 25, 3, 30, tzinfo=tz)),
+    ]:
+        plain_count = count_advances(base, now)
+        staggered_count = count_advances(staggered, now)
+        assert staggered_count == plain_count == 2, f"Failed for {label}: staggered={staggered_count}, plain={plain_count}"
+