tPjEUdZddlZddlmZddlmZddlmZmZddlm Z ddl m Z m Z m Z mZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZmZmZddlmZddlm Z ddl!m"Z"e r ddl#m$Z$ddl%m&Z&e ejNjPejRjPdZ*e ee+e+fe,d<e dddZ-e ee+e+fe,d<dedefdZ.GddZ/y)am Coalesced reseed of spend counters from the authoritative DB. When a Redis spend counter expires (or is missing on a fresh pod), enforcement must read the current spend from somewhere. The in-process management cache (`user_api_key_cache.team_membership.spend`, etc.) is per-pod and lags DB writes from other pods, so trusting it allows budget bypass in multi-pod deployments. This module reseeds from the authoritative DB instead. A per-counter singleflight lock collapses concurrent reseeds on the same pod to one DB query per cold-cache window. The lock dict is bounded LRU to cap memory in long-lived deployments. N) OrderedDict)Mapping)datetimetimezone)MappingProxyType) TYPE_CHECKINGClassVarFinalOptional)verbose_proxy_logger)#SPEND_COUNTER_RESEED_LOCKS_MAX_SIZE)duration_in_seconds)Litellm_EntityType)OrganizationRepository)BudgetWindowSpendRepositorySpendLogsRepositoryTeamMembershipRepository)TeamRepository)UserRepository)VerificationTokenRepository) DualCache) PrismaClient)KeyTeam_WINDOW_SPEND_ENTITY_TYPESapi_keyteam_id_WINDOW_SPEND_LOG_FIELDSvaluereturnc^|j|S|jtjS)N)tzinfo)r"replacerutc)rs j/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy/db/spend_counter_reseed.py_as_utcr&:s%LL,5T%--x||-2TTceZdZUdZeZeded<dZee jdzed<e de de jfdZ e d ed de dedzfd Ze de defd Ze dd ed d dde dededzf dZe d ed de de de dededzf dZe d ed de de de dzdededzf dZe d ed de de dededzf dZe d ed d dde de de de dzdededzfdZy)SpendCounterReseeda Reseeds spend counters from the authoritative DB and warms the cache, coalesced via per-counter singleflight locks. Counter key prefixes map to DB tables: spend:key:{token} -> LiteLLM_VerificationToken.spend spend:team:{team_id} -> LiteLLM_TeamTable.spend spend:team_member:{uid}:{tid} -> LiteLLM_TeamMembership.spend spend:user:{user_id} -> LiteLLM_UserTable.spend spend:org:{org_id} -> LiteLLM_OrganizationTable.spend End-user and tag spend counters intentionally do not reseed here. Their auth paths already load the corresponding objects via get_end_user_object() and get_tag_objects_batch(); callers pass those values as fallback_spend. zOrderedDict[str, asyncio.Lock]_locksN_registry_lock counter_keyr cfKtjtjt_tj4d{tjj |}|2tjj ||cdddd{Stj}|tj|<ttjtkDr tjjd|cdddd{S777 #1d{7swYyxYww)NF)last) r)r+asyncioLockr*get move_to_endlenr popitem)r,locks r% _get_lockzSpendCounterReseed._get_lockRs  , , 407   -%44  %,,00=D"))55kB    <<>D59  % %k 2%,,-0SS"))11u1=        shAD1DD1AD D1DD1A(D D1DD1D1D1D."D% #D.*D1 prisma_clientrcvK|ytj|ry |jdr?|tdd}t |j j d|id{}n~|jdr\|tdd}d|vry|jdd\}}t|j j d||d id{}n|jd r>|td d}t|j j d |id{}n|jd r>|td d}t|j j d |id{}ns|jds|jdry|jdr>|tdd}t|j j d|id{}ny|ytt!|ddxsdS778777-#t$rtjd|YywxYww)al Read the authoritative spend for a counter from the DB. Returns the spend value (including 0.0) when the DB is reachable and the row exists. Returns None when prisma is unavailable, the row is missing, the key format is unrecognized, or the query raises. Callers use None to fall back to a caller-supplied source. N spend:key:tokenwherezspend:team_member::user_id_team_id)user_idr spend:team:rz spend:user:r@zspend:end_user:z spend:tag:z spend:org:organization_idz)SpendCounterReseed.from_db: failed for %sspend)r)_is_key_or_team_window_counter startswithr3rtable find_uniquersplitrrrr Exceptionr exceptionfloatgetattr)r7r,r:rowsuffixr@rorg_ids r%from_dbzSpendCounterReseed.from_dbasL    < <[ I %%l3*3|+<+>?7 FLLXX`gin_oXpp''(<= +C0D,E,G Hf$#)==a#8 4]CIIUU,'g.VWV'' 6%c-&8&:;*=9??KKS\^eRfKgg'' 6%c-&8&:;*=9??KKS\^eRfKgg''(9:k>T>TUa>b'' 5 +C ,=,? @2=AGGSS[lntZuSvv ;WS'3/63777q  hh w  * *+VXc d sH9A H&H')HH9AHH AH#H$AH2H3'HH9A H$H%H*H9H HHHHH63H95H66H9cdD]<}|j|s|jd\}}}|r|sy t|yy#t$rYywxYw)N)r9rAz:window:FT)rF rpartitionrrJ)r,prefix_ separatordurations r%rEz1SpendCounterReseed._is_key_or_team_window_countersl3 F))&1%0%;%;J%G "Ay(H #H-   s A AAspend_counter_cacherrequire_cache_warmcXKtj|d{}|4d{d}|jE |jj|d{}|t |cdddd{Sd}|s:|j j|}|t |cdddd{Stj||d{}| dddd{yt |} |j|jj||dd{} | r t |}n<|jj|d{} | t | n t |}|j j||n|j||dd{|cdddd{S777n7T#t $rY[wxYw7)7 7777?#t $rtjd||rYawxYw7W#1d{7swYyxYww) ao Reseed a cold spend counter from the DB and warm the cache, coalesced via a per-counter lock so concurrent callers (read path + write path) collapse to one DB query per cold-cache window. Returns the spend value (including 0.0 from a fresh budget reset) when the DB read succeeds, or None when the DB is unavailable. NFkeyTr\rnxr\r)r\r refresh_ttlz7SpendCounterReseed.coalesced: failed to warm counter %s)r)r6 redis_cacheasync_get_cacherLrJin_memory_cache get_cacherQasync_set_cache set_cacheasync_increment_cacher rK) r7rXr,rYr5redis_clean_missvaldb_spend current_valueseededcacheds r% coalescedzSpendCounterReseed.coalesceds^/88EE4 !4 ! % "..: 3 ? ? O OT_ O ``C$Sz4 !4 !4 !(,$$)99CC CT? :4 !4 !4 !"%7$>$>}k$ZZH'4 !4 !4 !4$)?M &22>*=*I*I*Y*Y'&+Z+%F (-h .A.M.M.].]bm.].n(n9?9Kf QVW_Q` '77AA'+B .CC [cquCvvv!i4 !4 !4 !F4 ! a 4 !!4 !"[#4 !:%)ow $..M&&  [4 !4 !4 !4 !sPH*G H*GH*HG GG * H*6G 7H*<G >*H( H*4G5H*:HG H H*&G#'H*, H8-G+%G%&1G+G'AG+)G)*G+.H/ H*;H<H*H*G  H* GHGHH* H#H*%G+'G+)G++"H HHHH*H'H H'#H* entity_type entity_idwindow_durationexpected_window_startcvK|ytj|}|y t|jj d|||did{}|yt|jt|kryt|jxsdS7B#t $rt jd|||YywxYww)a Read the maintained per-window spend row by primary key. Returns the row's spend only when the row belongs to the window the caller is enforcing, i.e. ``row.window_start >= expected_window_start``. A row at or past the expected start was rolled by a pod whose reset_at was at least as fresh as this caller's, so it is trusted; an older row means the window boundary was crossed and nothing has rolled the row yet, so its spend belongs to a previous window. Returns None for a missing, stale or unreadable row so the caller falls back to the spend-logs aggregate. ``entity_type`` is the counter-facing label ("Key"/"Team"); anything else has no row and returns None. N%entity_type_entity_id_window_duration)rorprqr;z@SpendCounterReseed.window_from_table: failed for %s=%s window=%srD) rr1rrGrHrJr rKr& window_startrLrC)r7rorprqrrrow_entity_typerNs r%window_from_tablez$SpendCounterReseed.window_from_tables,  !;!?!? !L  " :=IOO[[;'6%.+:> \ C$ ; 3## $w/D'E ESYY%#&&-  * *R    s:B9.B BB>B9B!B63B95B66B9rucK|&tj|||||d{}||Stj||||d{S7)7w)aP Authoritative window spend: the maintained row first, falling back to the spend-logs aggregate only when no current row exists. The aggregate range-scans an unindexed table, so it must stay a transitional path (window configured before the row existed) rather than a steady-state read. N)r7rorprqrr)r7rorpru)r)rwwindow_from_spend_logs)r7rorprqru from_tables r%window_from_dbz!SpendCounterReseed.window_from_dbsz  &&8&J&J+'# /&2 'K'!J%!!'>>'#% ?   ! s! AA $AAAAcK|ytj|}|y||dd|ii} t|jj |g|ddid{}|sy|d}t|tr|jd n t|d d}t|tr|jdn t|dd} t| xsdS7w#t $rt jd||YywxYww) N startTimegterCT)byr<sumz;SpendCounterReseed.window_from_spend_logs: failed for %s=%srDr_sum) rr1rrGgroup_byrJr rK isinstancedictrMrL) r7rorpru group_fieldr<response first_rowsum_rowrCs r%ryz)SpendCounterReseed.window_from_spend_logs=s  599+F    %.  $7 $F$L$L$U$U=dO%V%H#A; 2SpendCounterReseed.coalesced_window: failed to warm counter %s)r)r6rarbrLrJrcrdr{reasync_incrementrfrgr rK)r7rXr,rorprqrur5rhri window_spendrlrkcurrent_cached_values r%coalesced_windowz#SpendCounterReseed.coalesced_windowds/88EE6 6 $ "..: 3 ? ? O OT_ O ``C$Sz 6 6 6 (,$$)99CC CT? :6 6 6 );(I(I+'# /) )J)#L#/6 6 6 0 &22>*=*I*I*Y*Y'*+Z+%F (4 5H5T5T5d5dit5d5u/u,/72E2Q2Q2a2a$/&23b3-M -22F,GM'77AA'+B .CC [gChhh m6 6 6 F6 a 6 !6 #6 4%0v-i $..T  a6 6 6 6 sPH3G  H3GH3HGGG* H36G7H3<G>*H( H34G'5H3:HG*H H3*G-+H31-G8G0(G8G2&G8.G4/AG83G64G88H9 H3HH3H3GH3 G$ H#G$$H'H3*H-H30G82G84G86G88!HHH3H0$H' %H0,H3)F)__name__ __module__ __qualname____doc__rr*r __annotations__r+r/r0 staticmethodstrr6r rLrQboolrErnrrwr{ryrr'r%r)r)>s{ :EFH5 6F48NHW\\D018 S W\\  .8Xn%=.8C.8TY\`T`.8.8` C D   $) C!/C!(C!C!! C!  C!C!J2'/2'2'2' 2' ( 2'  2'2'h /   t       @$#/$#$#$# $#  $#$#L@ /@ (@ @  @  @ t @ @  @ @ r'r))0rr/ collectionsrcollections.abcrrrtypesrtypingrr r r litellm._loggingr litellm.constantsr *litellm.litellm_core_utils.duration_parserrlitellm.proxy._typesr,litellm.repositories.organization_repositoryr'litellm.repositories.table_repositoriesrrr$litellm.repositories.team_repositoryr$litellm.repositories.user_repositoryr2litellm.repositories.verification_token_repositoryrlitellm.caching.dual_cacherlitellm.proxy.utilsrKEYrTEAMrrrrr&r)rr'r%rs ##'";;1AJ3O @?408H!%%++"''--8E'#s("346F6%S 12U8UUg g r'