fix(proxy): take the closed-day cutoff for the global spend rollup from the database clock

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/spend_tracking/daily_global_spend_rollup.py

@@ -12,7 +12,7 @@ a large deployment the first backfill is minutes of work.
 
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from datetime import date, datetime, timedelta, timezone
+from datetime import date, timedelta
 from typing import TYPE_CHECKING, Final
 
 from pydantic import BaseModel, ConfigDict, ValidationError
@@ -73,7 +73,7 @@ def _reconcile_day_sql() -> str:
 
 
 RECONCILE_DAY_SQL: Final = _reconcile_day_sql()
-_DB_NOW_SQL: Final = "SELECT (NOW() AT TIME ZONE 'UTC')::text AS now"
+_DB_NOW_SQL: Final = "SELECT (NOW() AT TIME ZONE 'UTC')::text AS now, (NOW() AT TIME ZONE 'UTC')::date::text AS today"
 _ALL_CLOSED_DAYS_SQL: Final = 'SELECT DISTINCT "date" FROM "LiteLLM_DailyUserSpend" WHERE "date" <= $1 ORDER BY "date"'
 # Pod clocks drift from the database clock and from each other, so rows are picked up from a
 # little before the previous scan; rewriting a day twice is idempotent.
@@ -111,6 +111,7 @@ class _NowRow(BaseModel):
     model_config = ConfigDict(frozen=True, extra="ignore")
 
     now: str
+    today: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -160,18 +161,18 @@ async def _record_marker(prisma_client: "PrismaClient", marker: ReconciledThroug
     await invalidate_config_param(DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM)
 
 
-async def _db_now(prisma_client: "PrismaClient") -> str:
+async def _db_now(prisma_client: "PrismaClient") -> _NowRow:
     rows: Final = await prisma_client.db.query_raw(_DB_NOW_SQL)
-    return _NowRow.model_validate(rows[0]).now
+    return _NowRow.model_validate(rows[0])
 
 
-async def _scan_pending(prisma_client: "PrismaClient", today: date) -> _PendingScan:
-    """Every closed UTC day (strictly before today) still to roll up, oldest first: days past the
-    marker, plus any day with per-key rows written since the scan behind the marker. Before a
-    run has fully succeeded there is no such scan, so every closed day is rolled up."""
+async def _scan_pending(prisma_client: "PrismaClient") -> _PendingScan:
+    """Every closed UTC day (strictly before the database's today) still to roll up, oldest first:
+    days past the marker, plus any day with per-key rows written since the scan behind the marker.
+    Before a run has fully succeeded there is no such scan, so every closed day is rolled up."""
     marker: Final = await read_marker(prisma_client)
-    scanned_at: Final = await _db_now(prisma_client)
-    last_closed_day: Final = (today - timedelta(days=1)).isoformat()
+    db_now: Final = await _db_now(prisma_client)
+    last_closed_day: Final = (date.fromisoformat(db_now.today) - timedelta(days=1)).isoformat()
     rows: Final = (
         await prisma_client.db.query_raw(_ALL_CLOSED_DAYS_SQL, last_closed_day)
         if marker is None or marker.scanned_at is None
@@ -179,11 +180,11 @@ async def _scan_pending(prisma_client: "PrismaClient", today: date) -> _PendingS
             _PENDING_DAYS_SQL, last_closed_day, marker.reconciled_through, marker.scanned_at
         )
     )
-    return _PendingScan(marker, scanned_at, tuple(_DateRow.model_validate(row).date for row in rows))
+    return _PendingScan(marker, db_now.now, tuple(_DateRow.model_validate(row).date for row in rows))
 
 
-async def pending_days(prisma_client: "PrismaClient", today: date) -> tuple[str, ...]:
-    return (await _scan_pending(prisma_client, today)).days
+async def pending_days(prisma_client: "PrismaClient") -> tuple[str, ...]:
+    return (await _scan_pending(prisma_client)).days
 
 
 async def reconcile_day(prisma_client: "PrismaClient", day: str) -> None:
@@ -192,15 +193,11 @@ async def reconcile_day(prisma_client: "PrismaClient", day: str) -> None:
     await prisma_client.db.execute_raw(RECONCILE_DAY_SQL, day)
 
 
-async def run_daily_global_spend_reconcile(
-    prisma_client: "PrismaClient",
-    today: date | None = None,
-) -> ReconcileResult:
+async def run_daily_global_spend_reconcile(prisma_client: "PrismaClient") -> ReconcileResult:
     """Roll up every pending day, advancing the marker after each; a failing day stops the run
     with the marker on the last good day so the next run resumes there. The scan time is only
     recorded once every pending day is done, so late rows a failed run saw are found again."""
-    effective_today: Final = today or datetime.now(timezone.utc).date()
-    scan: Final = await _scan_pending(prisma_client, effective_today)
+    scan: Final = await _scan_pending(prisma_client)
     done: Final = await _reconcile_until_failure(prisma_client, scan)
     if len(done) < len(scan.days):
         marker: Final = await reconciled_through(prisma_client)
@@ -243,13 +240,12 @@ async def run_scheduled_daily_global_spend_reconcile(
     prisma_client: "PrismaClient",
     pod_lock_manager: "PodLockManager | None" = None,
     alert: Callable[[str], Awaitable[None]] | None = None,
-    today: date | None = None,
 ) -> ReconcileResult | None:
     """Run the reconcile under a cross-pod lock so one proxy does the work; the lock only saves
     effort (each day is an idempotent rewrite), so an unreachable Redis runs unguarded rather than skipping."""
     redis_cache: Final = None if pod_lock_manager is None else pod_lock_manager.redis_cache
     if pod_lock_manager is None or redis_cache is None:
-        return await _run_and_alert(prisma_client, alert=alert, today=today)
+        return await _run_and_alert(prisma_client, alert=alert)
 
     acquired: Final = await pod_lock_manager.acquire_lock(
         cronjob_id=DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID, ttl=DAILY_GLOBAL_SPEND_RECONCILE_LOCK_TTL_SECONDS
@@ -258,7 +254,7 @@ async def run_scheduled_daily_global_spend_reconcile(
         verbose_proxy_logger.info("Daily global spend reconcile: another pod holds the lock, skipping this run")
         return None
     try:
-        return await _run_and_alert(prisma_client, alert=alert, today=today)
+        return await _run_and_alert(prisma_client, alert=alert)
     finally:
         if acquired:
             await pod_lock_manager.release_lock(cronjob_id=DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID)
@@ -277,9 +273,8 @@ async def _run_and_alert(
     prisma_client: "PrismaClient",
     *,
     alert: Callable[[str], Awaitable[None]] | None,
-    today: date | None,
 ) -> ReconcileResult:
-    result: Final = await run_daily_global_spend_reconcile(prisma_client, today=today)
+    result: Final = await run_daily_global_spend_reconcile(prisma_client)
     if result.days_reconciled:
         verbose_proxy_logger.info(
             "Daily global spend reconcile: rolled up %d day(s), reconciled through %s",

tests/test_litellm/proxy/spend_tracking/test_daily_global_spend_rollup.py

@@ -43,7 +43,8 @@ class _FakeConfigTable:
 
 class _FakeDb:
     """Per-key rows are ``{date: updated_at}`` with a fake database clock that ticks per query,
-    so "rows written since the last scan" behaves like Postgres would."""
+    so "rows written since the last scan" behaves like Postgres would. The database's own
+    date decides which day is still open, never the pod's clock."""
 
     def __init__(self, prisma: "_FakePrisma") -> None:
         self._prisma = prisma
@@ -52,7 +53,7 @@ class _FakeDb:
     async def query_raw(self, sql: str, *params: str) -> list[dict[str, str]]:
         if sql.startswith("SELECT (NOW()"):
             self._prisma.clock += 1
-            return [{"now": f"clock-{self._prisma.clock:04d}"}]
+            return [{"now": f"clock-{self._prisma.clock:04d}", "today": self._prisma.today.isoformat()}]
         rows = self._prisma.user_rows
         if len(params) == 1:
             (last,) = params
@@ -73,8 +74,11 @@ class _FakeDb:
 class _FakePrisma:
     """Enough of PrismaClient for the reconcile: per-key dates, a config table, and execute_raw."""
 
-    def __init__(self, user_days: tuple[str, ...], failing_days: frozenset[str] = frozenset()) -> None:
+    def __init__(
+        self, user_days: tuple[str, ...], failing_days: frozenset[str] = frozenset(), today: date = TODAY
+    ) -> None:
         self.clock = 0
+        self.today = today
         self.user_rows: dict[str, str] = {d: "clock-0000" for d in user_days}
         self.failing_days = failing_days
         self.reconciled: list[str] = []
@@ -98,12 +102,14 @@ async def _fresh_marker_cache():
 
 
 @pytest.mark.asyncio
-async def test_first_run_rolls_up_every_closed_day_and_never_today():
+async def test_first_run_rolls_up_every_closed_day_and_never_the_database_s_today():
     """Before any marker exists every closed day with per-key rows is rolled up. Today is left
-    out: pods are still flushing it, so it is served live from the per-key table until it closes."""
+    out: pods are still flushing it, so it is served live from the per-key table until it closes.
+    The database clock says which day that is; a pod booting with its clock a day ahead must not
+    roll the open day up and mark it reconciled."""
     prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-03", "2026-09-14", "2026-09-15"))
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-01", "2026-09-03", "2026-09-14")
     assert result.failed_day is None
@@ -114,11 +120,12 @@ async def test_first_run_rolls_up_every_closed_day_and_never_today():
 
 @pytest.mark.asyncio
 async def test_later_run_rolls_up_only_new_days_when_nothing_old_changed():
-    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-12", "2026-09-13", "2026-09-14"))
-    await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-12", "2026-09-13", "2026-09-14"), today=date(2026, 9, 14))
+    await run_daily_global_spend_reconcile(prisma)
     prisma.reconciled.clear()
+    prisma.today = TODAY
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-14",)
     assert await reconciled_through(prisma) == "2026-09-14"
@@ -128,13 +135,14 @@ async def test_later_run_rolls_up_only_new_days_when_nothing_old_changed():
 async def test_spend_landing_on_an_old_rolled_up_day_is_folded_in_by_the_next_run():
     """Per-key rows carry the request start date, so a delayed flush or retry can add spend to a
     day far behind the marker. That day is rewritten, and the marker never moves back for it."""
-    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-05", "2026-09-13"))
-    await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-05", "2026-09-13"), today=date(2026, 9, 14))
+    await run_daily_global_spend_reconcile(prisma)
     prisma.reconciled.clear()
+    prisma.today = TODAY
     prisma.write_late_row("2026-09-01")
     prisma.write_late_row("2026-09-03")
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-01", "2026-09-03")
     assert "2026-09-05" not in prisma.reconciled
@@ -145,15 +153,16 @@ async def test_spend_landing_on_an_old_rolled_up_day_is_folded_in_by_the_next_ru
 async def test_a_late_row_seen_by_a_failed_run_is_seen_again_by_the_next_one():
     """The scan time only advances when every pending day was rewritten, otherwise a late row
     found by the failed run would be counted as handled."""
-    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-13"))
-    await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-13"), today=date(2026, 9, 14))
+    await run_daily_global_spend_reconcile(prisma)
+    prisma.today = TODAY
     prisma.write_late_row("2026-09-01")
     prisma.failing_days = frozenset({"2026-09-01"})
-    failed = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    failed = await run_daily_global_spend_reconcile(prisma)
     prisma.failing_days = frozenset()
     prisma.reconciled.clear()
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert failed.failed_day == "2026-09-01"
     assert failed.reconciled_through == "2026-09-13"
@@ -166,7 +175,7 @@ async def test_a_marker_without_a_scan_time_rolls_every_closed_day_up_again():
     prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-13"))
     prisma.db.litellm_config.rows[DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM] = '{"reconciled_through": "2026-09-13"}'
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-01", "2026-09-13")
     marker = await read_marker(prisma)
@@ -175,11 +184,11 @@ async def test_a_marker_without_a_scan_time_rolls_every_closed_day_up_again():
 
 @pytest.mark.asyncio
 async def test_a_run_with_no_new_closed_days_keeps_the_marker():
-    prisma = _FakePrisma(user_days=("2026-09-13",))
-    await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    prisma = _FakePrisma(user_days=("2026-09-13",), today=date(2026, 9, 14))
+    await run_daily_global_spend_reconcile(prisma)
     prisma.reconciled.clear()
 
-    result = await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ()
     assert result.reconciled_through == "2026-09-13"
@@ -191,7 +200,7 @@ async def test_a_failing_day_stops_the_run_and_leaves_the_marker_on_the_last_goo
     a global table missing that day's spend."""
     prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-02", "2026-09-03"), failing_days=frozenset({"2026-09-02"}))
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-01",)
     assert result.failed_day == "2026-09-02"
@@ -203,10 +212,10 @@ async def test_a_failing_day_stops_the_run_and_leaves_the_marker_on_the_last_goo
 @pytest.mark.asyncio
 async def test_the_next_run_resumes_from_the_failed_day():
     prisma = _FakePrisma(user_days=("2026-09-01", "2026-09-02", "2026-09-03"), failing_days=frozenset({"2026-09-02"}))
-    await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    await run_daily_global_spend_reconcile(prisma)
     prisma.failing_days = frozenset()
 
-    result = await run_daily_global_spend_reconcile(prisma, today=TODAY)
+    result = await run_daily_global_spend_reconcile(prisma)
 
     assert result.days_reconciled == ("2026-09-01", "2026-09-02", "2026-09-03")
     assert await reconciled_through(prisma) == "2026-09-03"
@@ -215,13 +224,14 @@ async def test_the_next_run_resumes_from_the_failed_day():
 @pytest.mark.asyncio
 async def test_a_failure_with_nothing_done_reports_the_previous_marker_and_alerts():
     """When the rewrite of a late day fails the marker must stay put and the operator must hear about it."""
-    prisma = _FakePrisma(user_days=("2026-09-13",))
-    await run_daily_global_spend_reconcile(prisma, today=date(2026, 9, 14))
+    prisma = _FakePrisma(user_days=("2026-09-13",), today=date(2026, 9, 14))
+    await run_daily_global_spend_reconcile(prisma)
+    prisma.today = TODAY
     prisma.write_late_row("2026-09-12")
     prisma.failing_days = frozenset({"2026-09-12"})
     alert = AsyncMock()
 
-    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert, today=TODAY)
+    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert)
 
     assert result is not None
     assert result.days_reconciled == ()
@@ -236,7 +246,7 @@ async def test_a_clean_run_does_not_alert():
     prisma = _FakePrisma(user_days=("2026-09-13",))
     alert = AsyncMock()
 
-    await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert, today=TODAY)
+    await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=None, alert=alert)
 
     alert.assert_not_awaited()
 
@@ -256,7 +266,7 @@ async def test_scheduled_run_skips_when_another_pod_holds_the_lock():
     prisma = _FakePrisma(user_days=("2026-09-13",))
     lock = _pod_lock(acquired=False)
 
-    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
+    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock)
 
     assert result is None
     assert prisma.reconciled == []
@@ -268,7 +278,7 @@ async def test_scheduled_run_runs_and_releases_the_lock_when_it_wins():
     prisma = _FakePrisma(user_days=("2026-09-13",))
     lock = _pod_lock(acquired=True)
 
-    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
+    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock)
 
     assert result is not None and result.days_reconciled == ("2026-09-13",)
     lock.release_lock.assert_awaited_once()
@@ -282,7 +292,7 @@ async def test_scheduled_run_proceeds_when_the_lock_cannot_be_acquired_or_read()
     lock = _pod_lock(acquired=False)
     lock.redis_cache.async_get_cache = AsyncMock(side_effect=ConnectionError("redis down"))
 
-    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock, today=TODAY)
+    result = await run_scheduled_daily_global_spend_reconcile(prisma, pod_lock_manager=lock)
 
     assert result is not None and result.days_reconciled == ("2026-09-13",)
     lock.release_lock.assert_not_awaited()