tPj1RUdZddlmZddlmZmZmZddlmZmZm Z ddl m Z ddl m Z ddlmZmZmZerddlmZd Zeed <d Zeed <d Zeed<dZeed<dZeed<dZeed<dZeed<edZeed<Gdde Zdddeded ed!eed"ed#zd$ed#zfd%Z d&ed$efd'Z!d(ed$e"eeeffd)Z#ddd*e"ed+fd$e$e"eeeffd,Z%ddd(ed-e$e"eeefd.ed$ef d/Z&d(ed$ed#zfd0Z'd(ed1ed2ed$e"eeeeeeeffd3Z(e fddd*eed.ed$d#fd4Z)dddeded5ed6ed$d#f d7Z*y#)8a Writer for LiteLLM_BudgetWindowSpend. The table holds one row per configured budget window whose window_start rolls forward in place, so budget enforcement can read a maintained running total instead of aggregating LiteLLM_SpendLogs every time a window counter goes cold (issue #35766). Raw SQL rather than the Prisma upsert helper because the conditional roll cannot be expressed through the query builder. Seeding a row that does not exist yet reads LiteLLM_SpendLogs once, excluding the requests whose increments are in the same batch so neither source counts them twice. One gap survives that exclusion: without the Redis transaction buffer every pod flushes its own increments, so a row seeded by one pod can include spend logs whose increments are still queued on another pod, and those increments are added again when that pod flushes. That is bounded by a single flush interval, happens at most once per window row, and only ever over-counts: the seed never omits spend, because every increment not yet in the row still reaches it on its own pod's next flush. A row therefore lags real spend by at most one flush interval of queued increments, the same lag the SpendLogs aggregate it replaces (and every other spend column) already has. )Sequence)datetime timedeltatimezone) TYPE_CHECKINGFinalProtocol)verbose_proxy_logger)Litellm_EntityType)WindowSpendTransaction to_naive_utcwindow_spend_group_key) PrismaClientzSELECT entity_type, entity_id, window_duration FROM "LiteLLM_BudgetWindowSpend" WHERE (entity_type, entity_id, window_duration) IN (SELECT * FROM unnest($1::text[], $2::text[], $3::text[]))_SELECT_EXISTING_ROWS_SQLaqINSERT INTO "LiteLLM_BudgetWindowSpend" (entity_type, entity_id, window_duration, window_start, spend, created_at, updated_at) VALUES ($1, $2, $3, ($4::timestamptz AT TIME ZONE 'UTC'), $5, ($7::timestamptz AT TIME ZONE 'UTC'), ($7::timestamptz AT TIME ZONE 'UTC')) ON CONFLICT (entity_type, entity_id, window_duration) DO UPDATE SET spend = CASE WHEN "LiteLLM_BudgetWindowSpend".window_start >= EXCLUDED.window_start THEN "LiteLLM_BudgetWindowSpend".spend + $6 ELSE EXCLUDED.spend END, window_start = GREATEST("LiteLLM_BudgetWindowSpend".window_start, EXCLUDED.window_start), updated_at = ($7::timestamptz AT TIME ZONE 'UTC')_UPSERT_WINDOW_SPEND_SQLaUPDATE "LiteLLM_BudgetWindowSpend" SET window_start = ($4::timestamptz AT TIME ZONE 'UTC'), spend = 0, updated_at = ($5::timestamptz AT TIME ZONE 'UTC') WHERE entity_type = $1 AND entity_id = $2 AND window_duration = $3 AND window_start < ($4::timestamptz AT TIME ZONE 'UTC')_ROLL_WINDOW_SPEND_SQLzSELECT COALESCE(SUM(spend), 0.0) AS total FROM "LiteLLM_SpendLogs" WHERE api_key = $1 AND "startTime" >= ($2::timestamptz AT TIME ZONE 'UTC') AND NOT (request_id = ANY($3::text[]) AND "startTime" >= ($4::timestamptz AT TIME ZONE 'UTC'))_SEED_FROM_SPEND_LOGS_KEY_SQLzSELECT COALESCE(SUM(spend), 0.0) AS total FROM "LiteLLM_SpendLogs" WHERE team_id = $1 AND "startTime" >= ($2::timestamptz AT TIME ZONE 'UTC') AND NOT (request_id = ANY($3::text[]) AND "startTime" >= ($4::timestamptz AT TIME ZONE 'UTC'))_SEED_FROM_SPEND_LOGS_TEAM_SQLzSELECT COALESCE(SUM(spend), 0.0) AS total FROM "LiteLLM_SpendLogs" WHERE api_key = $1 AND "startTime" >= ($2::timestamptz AT TIME ZONE 'UTC')'_SEED_FROM_SPEND_LOGS_KEY_UNBOUNDED_SQLzSELECT COALESCE(SUM(spend), 0.0) AS total FROM "LiteLLM_SpendLogs" WHERE team_id = $1 AND "startTime" >= ($2::timestamptz AT TIME ZONE 'UTC')(_SEED_FROM_SPEND_LOGS_TEAM_UNBOUNDED_SQL<)seconds_UPSERT_TRANSACTION_TIMEOUTcFeZdZdZdddedededeeded zd ed zfd Zy ) WindowSpendLogsAggregatezSums LiteLLM_SpendLogs for one entity since window_start, ignoring the requests whose ids are handed in. Injected so the flush can be exercised without a database and so the expensive aggregate stays swappable. prisma_clientr entity_type entity_id window_startexclude_request_idsexclude_started_atNreturnc Kyw)N)selfrrrrr r!s p/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy/db/budget_window_spend_writer.py__call__z!WindowSpendLogsAggregate.__call__fs s) __name__ __module__ __qualname____doc__strrrfloatr'r$r&rr^s\%    &c] %tO r.rrrrrrr r!Nr"c K|tjjk(r tt}}n+|tj jk(r t t}}ny||s%|jj|||d{n8|jj|||t|t|d{}|syt|djdxsdS7a7*w)aLiteLLM_SpendLogs spend for one entity since window_start, minus the requests already accounted for by the increments being flushed. The spend log writer drains its own queue on a ~2s poll whenever anything is queued, while window increments flush on the much slower batch tick, so by the time a window row is seeded its batch's log rows are normally already in the table. Counting them in the seed and again in the increment is what made a fresh row land at twice the true spend. The exclusion is bounded to rows that started at or after the batch's earliest request. request_id can be chosen by the client (x-litellm-call-id), so an unbounded exclusion would let a replayed old id erase a historical row from the seed while its increment still lands. Without a known start the batch's ids are not excluded at all: that can only over-count once, which enforcement tolerates, whereas under-counting is a budget bypass. Nrtotal)r KEYvaluerrTEAMrrdb query_rawtuple_exclusion_lower_boundr-get) rrrrr r! bounded_sql unbounded_sqlrowss r&spend_logs_total_excludingr=qs2(,,222%BDk] *//55 5%CEm]   %-@(( <PPP ##--    % & "#5 6      aW%, -- Q s$A:C"<C=8C"5C 6)C" C" started_atc8t|jdS)zLiteLLM_SpendLogs.startTime is TIMESTAMP(3); floor to the second so a millisecond rounding of the batch's own earliest row cannot slip under it.r) microsecond)r replace)r>s r&r8r8s  # + + + ::r. transactionc|d|d|dfS)Nrrwindow_durationr$)rBs r& _primary_keyrEs%M"K %& r. transactions.c K|jjttd|Dtd|Dtd|Dd{}t d|xsdDS7w)Nc3&K|] }|d yw)rNr$.0rBs r& z)_existing_primary_keys..s I[k-( Ic3&K|] }|d yw)rNr$rIs r&rKz)_existing_primary_keys..s G;k+& GrLc3&K|] }|d yw)rDNr$rIs r&rKz)_existing_primary_keys..s Mk+, MrLc38K|]}|d|d|dfyw)rrrDNr$)rJrows r&rKz)_existing_primary_keys..s(jX[c-(#k*:C@Q(A<)A>c|jd}|ytj|jtj S)Nr>rV)r9rrYrArrZ)rBr>s r&r[r[s<# 5J  ! !* - 5 5X\\ 5 JJr. seed_basenowctt|d}|d|d|dtj|d||z||fS)a$5 is what a brand new row starts at (pre-existing spend plus this increment); $6 is the increment alone, which is all an already-current row may add. They are equal for every row that already existed, so a row is never seeded twice when two pods flush the same new window.spendrrrDr)r-rrY)rBr_r` increments r&_upsert_paramsrdsW[12IM"K %&{>:;I  r.c K|sytt|t}t||d{}t|Dcgc]}t ||||d{c}}t t jtj}tjdt|t||jjt4d{}|j!4d{} t#||D](\}} | j$t&gt)|| |*dddd{dddd{y7"7cc}w77h7##1d{7swY3xYw7*#1d{7swYyxYww)aApply aggregated window increments to LiteLLM_BudgetWindowSpend. An increment at or behind the row's window_start adds into the row (this is how in-flight requests that raced a reset carry into the new window); an increment ahead of it rolls the window and starts from that increment. Statements are ordered by primary key so concurrent pods take row locks in the same order, with window_start breaking ties so an older window is applied before the roll that supersedes it. N)key)rrF)rrBrSrTzPSpend tracking - committing %d budget window spend upserts over %d existing rows)timeout)rBr_r`)r7sortedrrRr]r rr`rrZr debuglenr5txrbatch_zip execute_rawrrd) rrFrTorderedrSrB seed_basesr`db_transactionbatcherr_s r&commit_window_spend_updatesrss 6,4JKLG)?#*$ '  -+'&;%9     Jhll8<<89CZ G  !" $?@DR#*&)':&>  "K G  ( K9RUV   -$   "s.FE FEEEBFEF E95E6E998E"1 E9<E =E9 F E7 FEFE9 E9"E4 (E+)E4 0E97F9F ?F F FrDnew_window_startc K|jjt|||t|tt j t jd{y7w)aMove a row onto the window that just started and zero its spend. Conditional on the stored window_start still being behind the new one so a pod that already rolled the row (or increments that arrived under the new window) are not clobbered. N)r5rnrr rr`rrZ)rrrrDrts r&roll_window_spend_rowrv%sP    & &%&X\\(,,/0  sAA%A#A%)+r+collections.abcrrrrtypingrrr litellm._loggingr litellm.proxy._typesr ?litellm.proxy.db.db_transaction_queue.window_spend_update_queuer r rlitellm.proxy.utilsrr__annotations__rrrrrrrrr,r-r=r8r7rErQrRr]r[rdrsrvr$r.r&r~s,%221113 0D5 8%  >gugS( S)% &/r%:U:x&,.!,.,.,. ,. "# ,. !4 ,. T\,.^;x;H; 4sC}9M k! k.34 kuS#s]#$ k!'%U3S=%9:3   4K)?KHtOK'  3S(E5( :; 06P1!112131 1h!     r.