diff --git a/litellm/repositories/performance_alert_repository.py b/litellm/repositories/performance_alert_repository.py index 125f5db8e74..2c85a4b90f7 100644 --- a/litellm/repositories/performance_alert_repository.py +++ b/litellm/repositories/performance_alert_repository.py @@ -151,7 +151,7 @@ _LIMITS: Final = f"""SELECT set_config('statement_timeout', '{_STATEMENT_TIMEOUT _CLAIM: Final = """INSERT INTO "LiteLLM_PerformanceAlertState" (id, owner, next_run_at, updated_at) VALUES (1, $1, now() + $2::interval, now()) ON CONFLICT (id) DO UPDATE SET owner = EXCLUDED.owner, next_run_at = EXCLUDED.next_run_at, updated_at = now() -WHERE "LiteLLM_PerformanceAlertState".next_run_at <= now() +WHERE "LiteLLM_PerformanceAlertState".next_run_at <= now() + interval '30 seconds' RETURNING day, scope, resume_after AS cursor, wrapped, checked, skipped, completed_at IS NOT NULL AS completed""" _SAVE: Final = """UPDATE "LiteLLM_PerformanceAlertState" SET day = $2, scope = $3, resume_after = $4, wrapped = $5, checked = $6, skipped = $7, @@ -179,7 +179,9 @@ class PerformanceAlertRepository: yield transaction async def claim(self, interval: timedelta) -> Claim | None: - """Take this deployment's next run, at most one per `interval` across every worker and pod.""" + """Take this deployment's next run, at most one per `interval` across every worker and pod. A worker whose + timer fires up to 30 seconds early still takes it, so the claim's own delay never pushes a run back a whole + interval.""" owner: Final = uuid.uuid4().hex async with self._transaction() as transaction: rows: Final = _STATE_ROWS.validate_python( diff --git a/tests/integration/database/test_performance_alerts.py b/tests/integration/database/test_performance_alerts.py index 0e5f8fbd31d..0b400634d7d 100644 --- a/tests/integration/database/test_performance_alerts.py +++ b/tests/integration/database/test_performance_alerts.py @@ -197,3 +197,15 @@ async def test_a_resource_that_cannot_be_read_is_skipped_without_blocking_the_re async def test_only_one_of_many_concurrent_workers_gets_each_run(fixture: Fixture) -> None: claims: Final = await asyncio.gather(*(fixture.repository.claim(timedelta(minutes=5)) for _ in range(8))) assert sum(claim is not None for claim in claims) == 1 + + +@pytest.mark.parametrize(("seconds_until_due", "claimed"), ((20, True), (60, False))) +@pytest.mark.asyncio +async def test_a_timer_firing_just_before_the_run_is_due_still_takes_it( + fixture: Fixture, seconds_until_due: int, claimed: bool +) -> None: + write_rows( + 'INSERT INTO "LiteLLM_PerformanceAlertState" (id, owner, next_run_at) VALUES (1, %s, now() + %s::interval)', + ("earlier-worker", f"{seconds_until_due} seconds"), + ) + assert (await fixture.repository.claim(timedelta(minutes=5)) is not None) is claimed