R3i&UddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlmZmZmZddlmZddlmZmZmZmZmZmZmZddlZe j8dkr ddlmZdd lm Z ndd lm Z dd l!m"Z"dd l#m$Z$m%Z%m&Z&d d l'm(Z(m)Z)d dl*m+Z+ddl,m-Z-ddl.m/Z/m0Z0m1Z1ddl2m3Z3d dl4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@mAZAmBZBmCZCd dlDmEZEmFZFmGZGmHZHmIZId dlJmKZKmLZLmMZMd dlNmOZOmPZPmQZQmRZRmSZSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\m]Z]edZ^dZ_d Z`eaZbdecd<GddeZde7e>d d'dZeejegZhd ecd<e"jegZjd!ecd"<Gd#d$eZkGd%d&Zly)() annotationsN)AsyncExitStackcontextmanager)datetime timedeltatimezone) TracebackType)Any GeneratorMappingProtocol TypeAlias TypedDictcast) )ExceptionGroup) TaskGroup)trace)Status StatusCodeTracer)CANCEL_MSG_CLEANUP cancel_task)suppress_instrumentation)Redis)ConnectionError LockError ResponseError)Self)AdmissionBlockedCompletionHandlerCurrentExecution DependencyFailedDependencyFailureHandler PerpetualRuntime SharedContext TaskLogger TaskOutcomeformat_durationget_single_dependency_of_type'get_single_dependency_parameter_of_typeresolved_dependencies)Docket Execution RedisMessageRedisMessageIDRedisReadGroupResponse) TaskFunctioncompact_signature get_signature) QUEUE_DEPTHREDIS_DISRUPTIONSSCHEDULE_DEPTH TASK_DURATIONTASK_PUNCTUALITYTASKS_COMPLETED TASKS_FAILEDTASKS_REDELIVERED TASKS_RUNNING TASKS_STARTEDTASKS_STRICKENTASKS_SUCCEEDEDTASKS_SUPERSEDEDhealthcheck_servermetrics_serverd milliseconds rTaskKeyc:eZdZUdZded<ded<ded<ded<y ) PubSubMessagez9Message received from Redis pub/sub pattern subscription.strtypebytespatternchannelz bytes | strdataN)__name__ __module__ __qualname____doc____annotations__W/mnt/e/genesis-system/.venvs/voice-bridge/lib/python3.12/site-packages/docket/worker.pyrNrNesC I N N r[rN) executionloggerr^cDK|jd|jyw)zd%Zd>d&Zd'ed(<d?d)Zd@dAd*Zd@dBd+ZdCd,Z dDd-Zd>d.ZdEd/ZdFd0Zd>d1ZedGd2ZdHd3ZdId4Zd>d5Zd>d6ZdJd7Z y)KWorkera A Worker executes tasks on a Docket. You may run as many workers as you like to work a single Docket. Example: ```python async with Docket() as docket: async with Worker(docket) as worker: await worker.run_forever() ``` r1docketrOnameint concurrencyrredelivery_timeoutreconnection_delayminimum_check_intervalscheduling_resolutionboolschedule_automatic_tasksenable_internal_instrumentationr6 fallback_taskNrK)minutessecondsrITFc ||_|xs+tjdtj|_||_||_||_||_ ||_ ||_ | |_ | xst|_y)N#)rqsocket gethostnameosgetpidrrrtrurvrwrxrzr{rdr|) rjrqrrrtrurvrwrxrzr{r|s r\__init__zWorker.__init__sx Cv1134Abiik]C &"4"4&<#%:"(@%/N,*C.Cr[c#tK|jst5ddddydy#1swYyxYww)a Suppress OTel auto-instrumentation for internal Redis operations. When enable_internal_instrumentation is False (default), this context manager suppresses OpenTelemetry auto-instrumentation spans for internal Redis polling operations like XREADGROUP, XAUTOCLAIM, and Lua script evaluations. This prevents thousands of noisy spans per minute from overwhelming trace storage. Task execution spans and user-facing operations (schedule, cancel, etc.) are NOT suppressed. N)r{rrjs r\_maybe_suppress_instrumentationz&Worker._maybe_suppress_instrumentations833)+      s8,858cvKt_jjd{tj_jj fdtj_jj fdjjtj_ jj fdi_ jj fdi_ jj fdtjjjjd_jj fdjj#t$j t&t)j_jj fd jj-j*d{S77 w) NctdS)N_worker_stoppingdelattrrsr\z#Worker.__aenter__..sWT3E%Fr[ctdS)N _worker_donerrsr\rz#Worker.__aenter__..sWT>%Br[ctdS)N_cancellation_readyrrsr\rz#Worker.__aenter__..sWT3H%Ir[ctdS)N_execution_countsrrsr\rz#Worker.__aenter__..sWT3F%Gr[ctdS)N _tasks_by_keyrrsr\rz#Worker.__aenter__..sWT?%Cr[z - heartbeatrrctdS)N_heartbeat_taskrrsr\rz#Worker.__aenter__..WT3D%Er[ctdS)N_shared_contextrrsr\rz#Worker.__aenter__..rr[)r_stack __aenter__asyncioEventrcallbackrsetrrr create_task _heartbeatrqrrrpush_async_callbackrrr*renter_async_contextrs`r\rzWorker.__aenter__s$& kk$$&&&!(  FG#MMO BC #*==?  IJ13 GH@B CD&22 OO t{{'7'7&8 $E  EF '' --/A -T[[$? EFkk--d.B.BCCC ; '6 Ds"-H9H4G.handle_shutdownHs2 JH$HMMO$OO--<8r[SIGTERMcdS)NrrZrsr\rzWorker.run..Qs OI4Nr[cdS)NSIGINTrZrsr\rzWorker.run..Ts ?83Lr[z - workerr)rrOrnNone) rpartition importlib import_modulegetattrrFrGr1register_collectionrprget_running_loophasattrsignaladd_signal_handlerrrrrun_until_finished run_foreverCancelledErrorremove_signal_handler)cls docket_namerrrrtrurvrwrxrzr{until_finishedhealthcheck_port metrics_porttasksr|resolved_fallback_task module_name_ member_namemodulerq task_pathworkerlooprrs @@r\runz Worker.runs87; *7*B*B3*G 'KK,,[9F%,V[%A " $4 5@ F  -@ F 0O< F< F!&:I..y9:%!$/+=+=/E.C1I8W&< 4F4F #335D:>H.vy1//"NN,N//"MM+LF)'.':': & 9 9 ;(3}I%>(H (/':': & 2 2 4(3}I%>(H'#695 66v~~F 66v}}Ei4F4F< F< F @ F@ F@ F< F4F\'"11#695 66v~~F 66v}}E6e4F4F4F4F< F< F< F< F @ F@ F@ F@ Fs AJ1 J%J+G6 ,J/2J!G9 "J%A3I- AG>1G< 2G>6AI-  JI+ J J J!J%J%- J16J9J<G>>H HH HAI( (I- +J-I? 3I64I? ;JJJ J  J JJ" J%%J.*J1cBK|jdd{S7w)z8Run the worker until there are no more tasks to process.FforeverN_runrs r\rzWorker.run_until_finishedjsYYuY---- cBK|jdd{S7w)zRun the worker indefinitely.TrNrrs r\rzWorker.run_forevernsYYtY,,,,rzdict[str, int]rcKDcic]}|dc}_dfd }jjj| j d{jjj |i_ycc}w76#jjj |i_wxYww)al Run the worker until there are no more tasks to process, but limit specified task keys to a maximum number of iterations. This is particularly useful for testing self-perpetuating tasks that would otherwise run indefinitely. Args: iterations_by_key: Maps task keys to their maximum allowed executions rcT|j}|vryj||k\ryy)NFT)keyr)r]riterations_by_keyrjs r\has_reached_max_iterationsz6Worker.run_at_most..has_reached_max_iterationss7--C++%%c*.?.DDr[N)r]r2rnry)rrq strike_list add_conditionrremove_condition)rjrrrs`` r\ run_at_mostzWorker.run_at_mostts5F!FS#q&!F  --.HI ())+ + + KK # # 4 45O P%'D "%"G , KK # # 4 45O P%'D "s9C B 4CBBB 2CB.CCcK|j |jj4d{}|j||d{cdddd{S7077 #1d{7swYnxYwn#t$rt j d|jtjd|jdtj|jjd{7YnwxYww)NTrrz,Error connecting to redis, retrying in %s...exc_info) _startup_logrqredis _worker_looprr:addrr^r`rvrsleep total_seconds)rjrrs r\rz Worker._runs  M;;,,.KK%!%!2!25'!2!JJKKKJKKKKK" M!%%a7B++! mmD$;$;$I$I$KLLL M sD BA$BA*A&A* BA(B#D $B&A*(B*A<0A3 1A<8B?D A?D?DDD DD c Kjjjjjjiijj dfd }dfd dfd  d dfd }dfd }dfd  t 4d{}|jjjjdjjd{jrjd{|jjjjd |jjjjd d }jj }|s|sr|s|d{jt#z d kr6t%j&j(j+d{ofD]E} | d{D])\} } | d k(} | D]\} }|s || || d{+d ksEn|ss|d{}|s|sr|sωjj-dddd{r,t%j4dd id{|d{jj-y7]777A77777a#1d{7swYqxYw#t$j.$r&r!t0j3dt#YwxYw77#r.t%j4dd id{7|d{7jj-wxYww)NcKtjdj4d{}|jjj |j jj|jd{}|d}|d}|dkDxs|dkDcdddd{S77-7 #1d{7swYyxYww)NzChecking for workextrarr) r^debugpipelinexlenrqrzcardrexecute)rresults stream_len queue_len log_contextrrjs r\check_for_workz+Worker._worker_loop..check_for_works LL,KL @~~' 7 78 dkk445t{{445+3+;+;+=%=$QZ #AJ !A~6Q  7 7 7&> 7 7 7 7sW,CB=CACB?C+ C7C8C?CCC C CCc 4Ktjd j5|jjj jj jtjjdzdd{^}}}ddddfgS7#1swYxYw#t$rL}dt|vr9jjd{7|d{7cYd}~Sd}~wwxYww)NzGetting redeliveriesrz0-0)rr groupname consumername min_idle_timestart_idcountNOGROUP__redelivery__)r^rr xautoclaimrqrworker_group_namerrrsrurr rO_ensure_stream_and_group)rr redeliverieseavailable_slotsget_redeliveriesrrjs r\rz-Worker._worker_loop..get_redeliveriess LL/{L C 99; 050@0@![[33"&++"?"?%)YY&) 33AACdJ'"'-1A 1 +'A|a  ' 56 6 +  ! A&++>>@@@!1%!8888  s|DCA1B4B2B4%C-D2B44B=9C D *D3C64DDD D DDDDc Ktjdjjj d} j 5|j jjjjjdi|rdn%tjjdzd{}ddd|r7s5t!j"jjd{S7G#1swYFxYw#t$rL}dt|vr9jjd{7|d{7cYd}~Sd}~wwxYw7lw) NzGetting new deliveriesrz memory://>rr )r r streamsblockrr)r^rrqr startswithr xreadgrouprrrrrsrwrr rOrrr)r is_memoryresultrrget_new_deliveriesrrjs r\r"z/Worker._worker_loop..get_new_deliveriess? LL1L E 22;?I 99; #(#3#3"&++"?"?%)YY!%!7!7 =$  !>@@@!3E!::::   Qs=E7DA6DDD D4E7E5 E7DDD E2&*E-EE-!E$"E-&E2'E7,E--E22E7c^Ktjj||jd{}t j j |jjd|j}||<||<|j|j<dzy7vw)N) redeliveredr|z - task:rr) r2 from_messagerqr|rr_executerrrr) message_idmessage is_redeliveryr]task active_tasksrrjtask_executionss r\ start_taskz'Worker._worker_loop..start_tasks (44 )"00 I&& i( (())--AD",L $-OD !04D  y}} - q O!s1B-B+A7B-cxKDchc]}|js|}}|D]f}j|} j|} jj|jd |d{|d{hycc}w77#t$r}t j d|jjtjtjtz|j_ |jj|d{7Yd}~ d}~wwxYww)Nu7🔒 Task %s blocked by admission control, reschedulingr)reschedule_message)rpoprrr"r^rr]rnowrutcADMISSION_BLOCKED_RETRY_DELAYwhenschedule) r*completed_tasksr'r]r ack_messager+rrrjr,s r\process_completed_tasksz4Worker._worker_loop..process_completed_taskss 0<L tLOL' N)--d3 +//5 ""&&y}}d; NJJ%eZ888  NM 8' NLLQ )! ! X\\25RRKK$++..*.MMM NspD:BBAD:2B7B8BB B D:BB D7"BD2&D)'D2,D:2D77D:cKtjd|j4d{}|jjj jj ||jjj ||jd{dddd{y777 #1d{7swYyxYww)NzAcknowledging messager) r^rrxackrqrrxdelr)rr'rrrjs r\r7z(Worker._worker_loop..ack_messages LL0 L D~~' ) )8 KK**KK11  KK**&&((( ) ) )) ) ) ) )sW,CB>CA5C(C)C- C8C9CCCC C CCz - cancellation listenerrz - schedulerz - lease renewalTrrz0Shutdown requested, finishing %d active tasks...rreturn_exceptions)rnry)rrrnr5F)r'r4r(r3r)ryrnrrnr)rrr'r4rnr)rclearrrrtrrr_cancellation_listenerrqrrrrz'_schedule_all_automatic_perpetual_tasks_scheduler_loop _renew_leasesis_setlenrrrwrrrr^rgather)rjrrr r-r8infrahas_workstoppingsourcermessagesr)r'r(r7r+rr"rrr,s`` @@@@@@@r\rzWorker._worker_loops ##% !   &&(AC ?A**'')  7 7 7*  :#( !& !! !  !  ! !. N N( )? $ {/ ,/ ,e!!//1 KK,,--EF"..3355500FFHHH!!((/ KK,,-\:"!!&&ul; KK,,--=>" "&0077(lHJ1333&*&6&6\9J&JO&!+%mmD,G,G,U,U,WXXX #35G"H ":@-4GU0J,6:K,KM7?U 3 G'.$,&0Wm&T T T UU+a/! "#<)7)9#9-(lHJ2%%))+_/ ,/ ,tnnlKdKKK-///    ! ! #}/ , 6I4 Y5H !U $:Y/ ,/ ,/ ,/ ,b%%  F %% L/nnlKdKKK-///    ! ! #sGB-O9ML$M AL:!L'"#L:L*B&L:,L--AL:>L0?L:L2'L:>L4 ? L: L:L6L:1L: ML8MO5N 6ON O$M'L:*L:-L:0L:2L:4L:6L:8M:M M M M6NNNN O OO*N-+O:N=; OOcKtt|jd}|j}|jj sO t jd||j5||jj|jjgtjtjj!|jj"gd{\}}ddddkDrCt jd||jj|jj| t)j*|jj-|j.j1 d{yyy7#1swYxYw#t$$rt j'dd | YwxYw7<#t(j2$rYnwxYw|jj sow) z7Loop that moves due tasks from the queue to the stream.a local total_work = redis.call('ZCARD', KEYS[1]) local due_work = 0 if total_work > 0 then local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1]) for i, key in ipairs(tasks) do local hash_key = ARGV[2] .. ":" .. key local task_data = redis.call('HGETALL', hash_key) if #task_data > 0 then local task = {} for j = 1, #task_data, 2 do task[task_data[j]] = task_data[j+1] end redis.call('XADD', KEYS[2], '*', 'key', task['key'], 'when', task['when'], 'function', task['function'], 'args', task['args'], 'kwargs', task['kwargs'], 'attempt', task['attempt'], 'generation', task['generation'] or '0' ) redis.call('DEL', hash_key) -- Set run state to queued local run_key = ARGV[2] .. ":runs:" .. task['key'] redis.call('HSET', run_key, 'state', 'queued') -- Publish state change event to pub/sub local channel = ARGV[2] .. ":state:" .. task['key'] local payload = '{"type":"state","key":"' .. task['key'] .. '","state":"queued","when":"' .. task['when'] .. '"}' redis.call('PUBLISH', channel, payload) due_work = due_work + 1 end end end if due_work > 0 then redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, ARGV[1]) end return {total_work, due_work} zScheduling due tasksr)rkrbNrz#Moved %d/%d due tasks from %s to %szError in scheduler loopTrrtimeout)rrgregister_scriptrrrDr^rrrqrrrr1rr2 timestampprefix Exception exceptionrwait_forrrxr TimeoutError)rjrstream_due_tasksr total_workdue_works r\rBzWorker._scheduler_loop`s/3   ! ! / 6 9/ v'') ''..0  3; G99;1A"kk33T[[5K5KL$LL6@@B KK..2,(Ja<LL= " -- ..) !  &&))..0 66DDFE1,$   -!%!  ''  G''..0sA H 'F+5A=F2F3F:AF+ A GGGH FF($F++!G H GH GG*'H )G**H c K|jjdz }|jjs: t j |jj |d{yyy7#tj$rYnwxYwt|j}|s |j5|j|jj|jj|jd|dd{7dddn #1swYnxYwn&#t $rt"j%ddYnwxYw|jjs.w) zPeriodically renew leases on stream messages. Calls XCLAIM with idle=0 to reset the message's idle time, preventing XAUTOCLAIM from reclaiming it while we're still processing. rNNr)rrr r r message_idsidlezFailed to renew leasesTr)rurrrDrrUrrVlistvaluesrxclaimrqrrrrrSr^r`)rjractive_messagesrenewal_intervalr\s r\rCzWorker._renew_leasessN 22@@BQF''..0 &&))..0:J1 ''  5578K H99;,,![[33"&++"?"?%)YY&'$/ ' H7$G H5''..0s8E)2A7-A5.A72E)5A77B  E) B  E)-D&=A D D  D D&D"D&%E)& E E)E  E)c K|jjd{|jj4d{} |j|jj dt d4d{|jj jD]c}t|t}||js#|j}|jj||j|d{edddd{dddd{y7777.7#1d{7swY.xYw#t$rYdddd{7ywxYw7E#1d{7swYyxYww)Nzperpetual:lockF)rOblocking)r4r)rqwait_for_strikes_loadedrlockr(AUTOMATIC_PERPETUAL_LOCK_TIMEOUT_SECONDSrr_r/r( automaticrUr initial_whenr)rjr task_function perpetualrs r\rAz.Worker._schedule_all_automatic_perpetual_taskssskk11333;;$$&  %  ::KKOO$45D"&  *.):):)A)A)C  $K)9% %0Y5H5H"/"8"8C#$++// -I4J4JPS#2##           4        %   " #    sFD."FD1FE0 5E>D3?E:D9= D9 FcKtjd|j|jj |j }|j |ddd{|jj|j }|jj|j }|j||d{y7i7w)NzDeleting known taskrknown stream_id) r^rrrqruns_keyrhdelknown_task_key stream_id_keydelete)rjrr]rorqrrs r\_delete_known_taskzWorker._delete_known_task s *$2C2C2E F;;'' 6jj7K88833IMMB 11)--@ ll>=999 9 :s%A!C#C $A#CCCCc xKi|j|j}i|j|j}|j }|j j j|r|j j4d{}|j||d{dddd{|jd{tjd||tjd|ddizy|j|j d{s4tj#d||t%jd|ddizy|j&|j(vr!|j(|j&xxdz cc<t+j*}||j,j/z }i|d|i}d}t1jd||j2rt5jd|t7jd|t9j:|||j<dkDrd nd } tj#d | t?|||i} t@jC|jDtFjHjJi|j|jd |jDi|jM 5} tO||4d{} | jQD cic]\} } tS| tTr| | }} } |jWD](}tS|jXtZs|jX|rRt]ddj_|jaz|jWDcgc]}|jXc}i|jb| }te| tf}|r1|ji||jj|jl|d{}n$|jj|jli|d{}t+j*|z x}|d<tojd|| jqtsttjvte| tx}t{t}||}|r5|j||d{r|jdd{nd}||j jrtj|}tj|jd}|j&}t|j jj}|j jj|d|i|d{|j|d{tj#dt?|||dddd{t7jd|tjd|tj:|| dddy777#1d{7swYxYw77@7qcc} } wcc}w7k7I77777#1d{7swYxYw#tZ$rtj$rt+j*|z x}|d<| jqtsttjv|jd{7tj#dt?|||YDt$r}t+j*|z x}|d<tjd|| j|| jqtsttjt|t{t}||}te| t}|r|j||d{7rnLte| tx}|r|j||d{7rn"tjdt?|||d}|j jrtj|}tj|jd}|j&}t|j jj}|j jj|d|i|d{7t|jdt|}|j||d{7Yd}~Zd}~wwxYw#t7jd|tjd|tj:||wxYw#1swYyxYww)Nu🗙 %srrz docket.whereru↬ %s (superseded) punctualitygu↬u↪z %s [%s] %szcode.function.name)kind attributeslinksz1Failed to resolve dependencies for parameter(s): z, durationr)rzr!) result_keyasciirT)ttlu ↩ [%s] %su✗ [%s] %s (cancelled))rzrTz: )Zrspecific_labelsrgeneral_labels call_reprrqr is_strickenrrtmark_as_cancelledr^r`rCrclaimrrrrErrtimer4rQrBr$r@rAr=recordattemptr-restart_as_current_spanrarSpanKindCONSUMERincoming_span_linksr0items isinstancer&r_errorr"rjoinrkrcr.r)rfunctionrbrD set_statusrrOKr#r,r on_completemark_as_completed execution_ttl cloudpickledumpsbase64 b64encodedecodersrresult_storageputrrrSr?record_exceptionERRORrOr'handle_failurerTrPrUmark_as_failedr>r<)rjr]rcounter_labelscallrstartrvrzarrow dependenciesspankvdependency_failuresfailure dependency final_kwargsruntimer!completion_handleroutcomer{pickled_resultencoded_result ttl_secondsrfailure_handlerpickled_exceptionencoded_exception error_msgs r\r&zWorker._executes L**,L 0I0I0KL HDKKMHY-E-E-GH""$ ;; " " . .y 9{{((* @ @e--eY??? @ @--/ / / NN9d+N >   q.NH3M"M N __TYY/// KK-t;K G  Nnh5O$O P  ==D22 2  " "9== 1Q 6 1 inn6688 AAm[A !^,   ! !!^ 4!^, ^<"**Q.E %!=t;  /1  ) )  # #((++-++-%i&=&= //1*  h ?^ ?0yAQQ\%1$6$6$8+ Aq%a)9:1+'+$7#=#=#?0%gmm5EF")--/0+, S"&)),?,D,D,F"G!H 3F2L2L2N$.!+ 0 0  $Hi&6&6#G,#GLu9LLH{:6#''>:OOF:==$9:*G$&7*&*!*8!<%G*4F4R4R!75//(99T9JJJ&* !-$++2K2K-8->->v-FN-3-=-=n-M-T-T '.N*3J*-dkk.G.G.U.U.W*XK"&++"<"<"@"@ *V^,D+#A#(99Z9PPP )+H5 "- $YQQv!!"n5##A~6$$X~>Qh ?h ?G @? @ @ @ @ 0 0JQ+$"""/KQWQQQQd$ )) 59YY[55HH;z2z}} 5611333 -#H-%  9 U59YY[55HH;z2  N3%%a(z'7'7Q @A%&x8 #@ .##_-K-Kw.(( *G$&7*&*4F4R4R!75//(()+H5 "- )"&J{{00,7,=,=a,@),2,<,<=N,O,V,V#-)&/]] &)$++*C*C*Q*Q*S&T "kk88<<&1B(C= $(7#3#3"4Bs1vh ?I#2292TTTs9 Uv!!"n5##A~6$$X~>Qh ?h ?sBf:Yf:Y4Y5Y9 f:Yf:Y0Af:3Y34F?f:3f.5Z+Y6Z+ ZY9 :0Z+AZ2Y?AZZ$Z>Z?BZZ Z5Z 6B>Z4Z5ZZ&Z5 Z+ZZ+Af. f:Yf:Y- Y# !Y-( f:3f:6Z+9 ZZ Z ZZZZ+Z( ZZ( $Z++A3e$\!'e$e' e$B3e`.e4`75Ced>eeee'e$$e''Af++f..f73f:c tjd|j|jjj D].\}}tjd|t t|0y)Nz,Starting worker %r with the following tasks:z* %s(%s))r^rrrrqrrr7r8)rj task_namer*s r\rzWorker._startup_logsX BDIIN#{{00668 WOIt KK I/@tAT/U V Wr[c.|jjSri)rq workers_setrs r\rzWorker.workers_sets{{&&&r[c8|jj|Sri)rqworker_tasks_set)rj worker_names r\rzWorker.worker_tasks_sets{{++K88r[c8|jj|Sri)rqtask_workers_set)rjrs r\rzWorker.task_workers_sets{{++I66r[c  K tjtjj }|j j |j jz}||jz }t|j j}|j j4d{}|j5|j4d{}|j|jd||j!|j|j"|i|D]D}|j%|}|j|d||j!||j"|iF|j&|j)|j"g||j+|j)|j"t-|t/t0|j3d{dddd{|j4d{}|j5|j j6|j9|j j:d||j9|j j:|d|j3d{} dddd{ddd d} | d} | d} t=j>| | z|j jAtCj>| |j jAdddd{tEjV|j j jd{~7777#1d{7swYxYw777#1d{7swYxYw#1swY xYw7#1d{7swYxYw#tDjF$rYytH$rNtKjLd|jAtNjQdd|jSYtT$r*tNjQdd|jSY@wxYw7 w) NTrrz+infrzError sending worker heartbeatrM),rr1rr2rQrqheartbeat_intervalmissed_heartbeatsrr^rrrrzremrangebyscorerzaddrrrsaddrexpiremaxrMINIMUM_TTL_SECONDSrrrzcountrr9rrr;rrrr:rr^rTrrSr) rjr1 maximum_ageoldest task_namesrrrrr stream_depth overdue_depthschedule_depths r\rzWorker._heartbeats: ll8<<0::<KK22T[[5R5RR{88::!$++"3"34 ;;,,.!M!M!==?J#$::<558$55d6F6F6R$MM$*:*:TYYC@P QR *HMM$*?*? *JXZX$OO $ 5 5dii @ #$/CV1W!"#+"2"2"444#55&$%::<JJ8$MM$++*@*@A$OODKK,A,A1cJ$OODKK,A,A3O7?7G7G7I1IG JJ)J6$+1:L$+AJM%,QZNOOL=$@$++BTBTBVW"&&~t{{7I7I7KLC!M!Mf-- > > L L NO O O{!M5"5#5555&J 2J JJJJ)JJ!M!M!M!MF)) " !%%a7  4!++-!    4!++-!  Ps_RB.O&3N4O&7OON O!DN2N 3N7 ON ON$ OBN,&N' 'N,+ O6N*7O;A6O1 O&<O=O&:R;R<RO&ON ON! NN!  O'N,*O,N? 2N53N? :OO OO&O#O O#O&&R9R;AR R.R>RRRcK|jjd}|j}|jj s |jj 4d{}|j |d{|jj|jj sY|jddd{}|!|ddk(r|j|d{|jj sYdddd{|jj syy777s7R7)#1d{7swY9xYw#t$rx|jj rYytjd|jtj!d | t#j$dd{7Yt&$rV|jj rYytj)d d| t#j$dd{7YwxYww) z:Listen for cancellation signals and cancel matching tasks.zcancel:*NTg?)ignore_subscribe_messagesrOrPpmessagerz@Redis connection error in cancellation listener, reconnecting...rzError in cancellation listenerrM)rqrrrrD_pubsub psubscriberr get_message_handle_cancellationrr:rrr^r`rrrSrT)rjcancel_patternrpubsubr(s r\r@zWorker._cancellation_listener7s4'') ''..0 ';;..0 E EF ++N;;;,,002"33::<(.(:(:6:C);)##.76?j3P"&";";G"DDD #33::< E E''..0 E;#E E E E E# '((//1!%%a7V%mmA&&& '((//1  4!%! mmA&&& 'sAH? E'D?(E+E EAE E"E 2E3E  EEE"H?=H??EE E E E EE EE#H<H?AH<GH<H?"H<H?0H<2H53H<8H?;H<<H?cK|d}t|tr|jn|}|jj |x}r7t j d||j|jyyw)z>Handle a cancellation message by cancelling the matching task.rTzCancelling running task %rrN) rrQrrgetr^rrr)rjr(rTrr*s r\rzWorker._handle_cancellation[stv(24(?t{{}T%%))#. .4 . KK,'')   KKM /sA=A?)rqr1rr str | Nonertrsrurrvrrwrrxrrzryr{ryr|zTaskFunction | Nonernr)rnzGenerator[None, None, None])rnr!)rztype[BaseException] | NonerzBaseException | NonerzTracebackType | Nonernr)rnzMapping[str, str]) rrOrrOrrrrtrsrurrvrrwrrxrrzryr{ryrryr int | Nonerrrrmr|rrnrr>)rzMapping[str, int]rnrr=)rryrnr)rrrry)rrrnr)rrraz(dict[asyncio.Task[None], RedisMessageID]rnr)rrr]r2rnr)r]r2rnr)rnrO)rrOrnrO)rrOrnrO)r(rNrnr)!rUrVrWrXrYrrrrrrrr classmethodrrrrrrrBrCrArtr&rpropertyrrrrr@rrZr[r\rprps  N I!!!!%%$$""%))  (1!(<(1!(<,53,G+4#+F)-05-1DDD D & D & D!*D )D#'D*.D+D D0"B,((      $-(1!(<(1!(<,53,G+4#+F)-05$'+#'9:$(!aFaFaF aF  aF & aF&aF!*aF )aF#'aF*.aFaF%aF!aFaF "!aF" #aFaFF.-&%(>M {$zdL(H(HB(H  (HT2 :R?hW ''97>P@"'H r[rp) rbr r]r2r^z%logging.LoggerAdapter[logging.Logger]rcr rnr)m __future__rrrrloggingrrrsysr contextlibrrrrrtypesr typingr r r r rrrr version_infoexceptiongroupr taskgroupr opentelemetryropentelemetry.tracerrr _cancellationrr _telemetryr redis.asynciorredis.exceptionsrrr typing_extensionsr!rr"r#r$r%r&r'r(r)r*r+r,r-r.r/r0rqr1r2r3r4r5r]r6r7r8instrumentationr9r:r;r<r=r>r?r@rArBrCrDrErFrGr3rgrrOrLrYrNrd getLoggerrUr^ get_tracerrergrprZr[r\rsu"  522g-#!:::0FF""FE*!*s ;,.(I,-4>L     2    +**844!!!(++ ``r[