$jdZddlZddlZddlZddlZddlZddlZddlmZddl m Z m Z m Z ddl mZmZddlmZmZddlmZd d lmZmZd d lmZmZer dd lmZdd lmZej e!Z"dZ#dZ$dZ%ddde&fdZ'eGddZ(eGddZ)eGddZ*GddZ+dS)a Worker poller for distributed task execution. Polls the database for pending tasks and executes them using FOR UPDATE SKIP LOCKED for safe concurrent claiming. Backend-agnostic: works with any DatabaseBackend implementation (PostgreSQL via asyncpg, Oracle via oracledb, etc.). N)Counter) AwaitableCallableIterable) dataclassfield) TYPE_CHECKINGAny)fq_table_explicit)DeferOperation RetryTaskAt) StageHolder bind_holder)DatabaseBackend)TenantExtensioni,i`Tsiblingsz Iterable[Any]returncg}|D]B}|ddkr|dpd}|r||C|sdSt|dd\}}|S)aPick a representative error message for a parent whose children failed. Used when a batch_retain parent transitions to 'failed' because at least one child sub-batch failed. Without this, the parent gets a generic "One or more sub-batches failed" string and any consumer that reasons about errors via error_message (dashboards, alert filters, log aggregators) loses the actual cause -- a class of failures that all share the same root reason at the child level becomes indistinguishable at the parent level. Strategy: pick the most common non-empty error_message among failed siblings. If they all failed for the same reason (the common case), the parent inherits that reason verbatim. If they vary, the most-common one is still a useful representative. Falls back to the legacy generic string when no failed sibling carries an error_message at all. statusfailed error_messagezOne or more sub-batches failedr r)stripappendr most_common)r failed_errorssmsgr_counts c/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/hindsight_api/worker/poller.py_summarise_child_error_messagesr$)s" "M && X;( " " !'R..00  &   % % % 0//!-00< 38n9 KKKHHr%r?cFeZdZdZ dCdddedeeeefgedfd e d edzd d d e deee fdzfdZ de edzfdZ de edzde edzfdZdefdZdDdedefdZde efdZd edzdeee fde de efdZd edzdeee fde de efdZded edzfdZdeded edzfd Zd!ed edzddfd"Zded#d$ded edzfd%Zded&d$d'ed edzfd(Zd)efd*Zded+efd,ZdEd)ed-edzfd.Z de fd/Z!d edzde fd0Z"d1Z#dFdefd3Z$d4Z%defd5Z&defd6Z'd7eeeee ffddfd8Z(d9eee)fd:eddfd;Z*deddfd?Z+dGd@Z,e-defdAZ.e-defdBZ/dS)H WorkerPollera< Polls the database for pending tasks and executes them. Uses FOR UPDATE SKIP LOCKED for safe distributed claiming, allowing multiple workers to process tasks without conflicts. Supports dynamic multi-tenant discovery via tenant_extension. Backend-agnostic via DatabaseBackend abstraction. N backendr worker_idexecutorpoll_interval_msr*tenant_extensionzTenantExtension | None max_slotsslot_reservationsc ||_||_||_||_||_|ddlm} |rd|ini} | | }||_||_||nddi|_ ddl m } | |j|_ tj|_t!|_d|_tj|_d |_d|_i|_i|_d|_dS) a Initialize the worker poller. Args: backend: Database backend (PostgreSQL, Oracle, etc.) worker_id: Unique identifier for this worker executor: Async function to execute tasks (typically MemoryEngine.execute_task) poll_interval_ms: Interval between polls when no tasks found (milliseconds) schema: Database schema for single-tenant support (deprecated, use tenant_extension) tenant_extension: Extension for dynamic multi-tenant discovery. If None, creates a DefaultTenantExtension with the configured schema. max_slots: Maximum concurrent tasks per worker slot_reservations: Per-operation-type reserved slot counts (e.g. {"consolidation": 2, "retain": 3}). Reserved slots guarantee capacity for that operation type. Remaining slots (max_slots - sum of reservations) form a shared pool usable by any operation type. Defaults to {"consolidation": 2} if None. Nr )DefaultTenantExtensionr*)config consolidation)OptionalRoutinesr)_backend _worker_id _executor_poll_interval_ms_schemaextensions.builtin.tenantrN_tenant_extension _max_slots_slot_reservationsengine.db.optional_routinesrQ_optional_routinesasyncioEvent _shutdownset_current_tasks_in_flight_countLock_in_flight_lock_last_progress_log_tasks_completed_since_log _active_tasks_in_flight_by_type_next_schema_idx) selfrFrGrHrIr*rJrKrLrNrOrQs r#__init__zWorkerPoller.__init__~s8  #!!1  # J J J J J J,29h''rF55VDDD !1#!2!>  _VWDX  CBBBBB"2"24="A"A 14 !&|~~"%*+'8:24&'r%rcnKddlm|jd{V}fd|DS)zKGet list of schemas to poll. Returns [None] for default schema (no prefix).r )DEFAULT_DATABASE_SCHEMANc8g|]}|jkr|jndSNr*).0trns r# z-WorkerPoller._get_schemas..s,[[[aAH(???T[[[r%)rOrnrY list_tenants)rktenantsrns @r# _get_schemaszWorkerPoller._get_schemassZ444444.;;========[[[[SZ[[[[r%schemascK|j4d{V}|j|dd{Vr9|dd{V}d|Dcdddd{VSt }|D]Y}t d|} |d|dd{V}|r||J#t$rYVwxYw|cdddd{VS#1d{VswxYwYdS)aFind which schemas have pending work. Prefers a server-side PL/pgSQL routine (single DB round-trip, ~200ms for 1400+ schemas) when ``public.schemas_with_pending_work()`` is installed. The presence check goes through ``OptionalRoutines.is_installed`` which probes ``pg_proc`` once and caches the result, so we don't generate a server-side error on every poll cycle when the routine isn't installed. Falls back to per-schema Python EXISTS queries (~4ms each) on non-PostgreSQL backends or when the routine isn't installed. See ``hindsight_api.engine.db.optional_routines`` for the canonical install SQL. Nschemas_with_pending_workz0SELECT * FROM public.schemas_with_pending_work()ch|] }|d Srr8)rrrs r# z4WorkerPoller._scan_active_schemas..s+++!+++r%async_operationszSELECT EXISTS(SELECT 1 FROM z? WHERE status = 'pending' AND task_payload IS NOT NULL LIMIT 1)) rSacquirer] is_installedfetchrafq_tablefetchvaladd Exception)rkrxconnrowsactiver*tablehas_works r#_scan_active_schemasz!WorkerPoller._scan_active_schemass:=((**       d,99$@[\\\\\\\\ ,!ZZ(Z[[[[[[[[++d+++               '*eeF!   !3V<<%)]]ZuZZZ&&      H + 6*** D%                              s<AC;="C; 6CC; C$!C;#C$$C;; DDcLK|j4d{V|j}t|j}dddd{Vn#1d{VswxYwYi}d}|jD]D\}}||d}td||z ||<|t||z }Et|j }td|j |z } td||z } td| | z } t|| S)u Calculate available slots for claiming tasks. Each operation type can have reserved slots (via ``slot_reservations``). Reserved slots guarantee capacity for that type — they cannot be used by other types. The remaining slots (``max_slots - sum(reservations)``) form a shared pool usable by any operation type on a first-come basis. When an operation type's in-flight count exceeds its reservation, the excess tasks are considered to be using shared pool slots. Nr)r@rA) rercr=rir[itemsgetmaxminsumvaluesrZr?) rktotal_in_flightin_flight_snapshotreserved_availabletasks_in_reservedr(r@ in_flightsum_reservationsshared_pool_sizetasks_in_sharedshared_availables r#_get_available_slotsz!WorkerPoller._get_available_slotss' ? ? ? ? ? ? ? ?"3O!%d&=!>!>  ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? .0!%!8!>!>!@!@ : : GX*..w::I*-aI1E*F*F w ' Xy!9!9 9  t6==??@@q$/4D"DEEa3D!DEEq"2_"DEE);DTUUUUs> A A$@timeoutcKtj} |j4d{V|jdkr dddd{VdS dddd{Vn#1d{VswxYwYtj|z }||krdStjdd{V)ax Wait for all active background tasks to complete (test helper). This is a test-only utility that allows tests to synchronize with fire-and-forget background tasks without using sleep(). Args: timeout: Maximum time to wait in seconds Returns: True if all tasks completed, False if timeout was reached TNrFg{Gz?)r^get_event_looptimerercsleep)rkr start_timeelapseds r#wait_for_active_tasksz"WorkerPoller.wait_for_active_tasks s+--2244  &+        (A--              -                           ,..3355 BG'!!u-%% % % % % % % % &s A)) A36A3ctK|d{V}td|jDr |jdkrgS|d{V}|sgS||d{Vs!|jdzt|z|_gStt|}fd|D}|jt|zfd|Dfd|Dz}g}t|j|jd}g}dtffd } d ttddffd } |D]\} } | snd D} dkrtdnd}|| | |d{V}| ||r| }|| | f||| rs|rq|D]n\} } | sn^|| d Dd{V}| ||r| }||o||dzt|z|_ndzt|z|_|S)a Claim pending tasks atomically across all tenant schemas, respecting per-operation-type slot reservations and shared pool limits. Uses FOR UPDATE SKIP LOCKED to ensure no conflicts with other workers. Schema iteration is round-robin to prevent one busy tenant from starving others. Each poll starts at ``self._next_schema_idx`` and wraps around the full list. First pass caps at 1 claim per pool per schema so every tenant with pending work gets a fair chance; a second pass backfills remaining slots from any schema when there's spare capacity. After the call, the offset advances past the last schema we serviced (or by 1 if nothing was claimed) so the next poll starts at a different position. Returns: List of ClaimedTask objects containing operation_id, task_dict, and schema Nc3"K|] }|dkV dSrNr8rrvs r# z+WorkerPoller.claim_batch..9s&>>!qAv>>>>>>r%rr c&g|] \}}|v ||fSr8r8)rrir active_schemass r#rtz,WorkerPoller.claim_batch..Ls+PPPTQA.Ps">>>! 1 r%c,g|]}|dk|Sr|r8rs r#rtz,WorkerPoller.claim_batch..Ps+AlAlAl_`ab_cfk_k_k!_k_k_kr%rcftdDpdkS)Nc3"K|] }|dkV dSrr8rs r#rzBWorkerPoller.claim_batch.._has_capacity..Ys&BBq1uBBBBBBr%r)anyr)remaining_reservedremaining_sharedsr# _has_capacityz/WorkerPoller.claim_batch.._has_capacityXs9BB&8&?&?&A&ABBBBBZFVYZFZ Zr%tasksc|D]C}|jdd}|vr|dkr|xxdzcc<>dzDdS)Noperation_typeunknownrr )r<r)rtaskr(rrs r#_account_tasksz0WorkerPoller.claim_batch.._account_tasks[s~ * *.,,-=yII0005G5PST5T5T&w///14////$)$$  * *r%cBi|]\}}|dk |td|S)rr )rrrrsrs r# z,WorkerPoller.claim_batch..js2ZZZdaTUXYTYTYQAq TYTYTYr%c&i|]\}}|dk ||Sr|r8rs r#rz,WorkerPoller.claim_batch..~s$RRRdaAPQEEQEEEr%)rallr@rrArwrrjlenlist enumerater=boolr:rr_claim_batch_for_schemarextend)rk availabilityrx all_indexedactive_indexedrotated all_taskslast_serviced_idxschemas_with_workrrorig_idxr* fair_reserved fair_sharedrrrrrs @@@@r# claim_batchzWorkerPoller.claim_batch#s("6688888888 >>|4;;==>>> > > #g,,$ND !I9W--.. PPPP[PPP%G 4>>>>n>>>AlAlAlAl^AlAlAll') !,"788'.(,:< [t [ [ [ [ [ [ [ *${"3 * * * * * * * *!( $ $ Hf =?? ZZ6H6N6N6P6PZZZM6F6J6J#a!1222PQK66v}kZZZZZZZZE N5 ! ! ! =$,!!(((F);<<<   U # # # # =?? (0 ($5 ( ( &$}E"::RR.@.F.F.H.HRRRTdu%%%1(0%  ''''  (%6%:c'll$JD ! !%*QY#g,,$>D !r%reserved_limits shared_limitc K ||||d{VS#t$rK}|rd|dnt|}td|jd|d|gcYd}~Sd}~wwxYw)zNClaim tasks from a specific schema respecting per-type and shared slot limits.N"Worker z" failed to claim tasks for schema : )_claim_batch_for_schema_innerrr4loggerwarningrT)rkr*rreschema_displays r#rz$WorkerPoller._claim_batch_for_schemas ;;FOUabbbbbbbb b   .4E]]]]]#f++N NNmT_mmXfmmjkmm n n nIIIIII  s! A6AA1+A61A6c LKtd|}|j4d{V}|4d{V|jj|||j||d{V}|s&gcdddd{Vcdddd{VSg}|D]}|d} t| trtj | n| } |d| d<t|d| d<|d} | r| | d<| tt|d| | |cdddd{Vcdddd{VS#1d{VswxYwY dddd{VdS#1d{VswxYwYdS) zInner implementation for claiming tasks from a specific schema. Delegates the SQL claiming logic to backend.ops.claim_tasks() which handles backend-specific differences (e.g. Oracle's ORA-02014 workaround). rN task_payload retry_count _retry_countr; _operation_idr)r;r<r*) rrSr transactionops claim_tasksrT isinstancer4jsonloadsrr:) rkr*rrrrall_rowsresultrowpayloadr< db_op_types r#rz*WorkerPoller._claim_batch_for_schema_inners+V44=((**       d''))        !%!2!>!>O# ""                            #C!.1G7A'37O7O \ 7 3 3 3U\I03M0BIn-14S5H1I1IIo."%%5!6J!A6@ "23MM#),S-@)A)A&/#)?                                                                        sBF 2E.?F$B%E. F. E8 8F;E8 <F F Fr;cKtd|}|j4d{V}|d|d|d{Vdddd{VdS#1d{VswxYwYdS)zMark a task as completed.rN UPDATE z SET status = 'completed', completed_at = now(), updated_at = now() WHERE operation_id = $1 )rrSrexecute)rkr;r*rrs r#_mark_completedzWorkerPoller._mark_completeds5+V44=((**       d,,                                       s!A&& A03A0rcKtd|}t|dkr |ddn|}|j4d{V}|4d{V|d|d||d{V||||d{Vdddd{Vn#1d{VswxYwYdddd{VdS#1d{VswxYwYdS)zQMark a task as failed with error message, then propagate to parent if applicable.rN UPDATE z SET status = 'failed', error_message = $2, completed_at = now(), updated_at = now() WHERE operation_id = $1 )rrrSrrr_maybe_update_parent_operation)rkr;rr*rrs r# _mark_failedzWorkerPoller._mark_faileds+V4403M0B0BT0I0I ete,,} =((** V V V V V V Vd'')) V V V V V V V Vll! !!99,PTUUUUUUUUU V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V V Vs6C!,?B=+C!= C C! C C!! C+.C+child_operation_idc0Kddl}ddl}td|} |d|d||d{V}|sdS|dpi}t |t r|j|}|d} | sdS|d} |d |d || | d{V} | sdS| d |d | |j d| id{V} | rtd | DsdStd| D} | rB| d|d|| t| d{Vn3| d|d|| d{Vtd| d| rdndddS#t"$r+}td|d|Yd}~dSd}~wwxYw)a?If this operation is a child of a batch_retain, update the parent status when all siblings are done. Must be called within an active transaction that has already updated the child's status. The memory engine has an equivalent method that runs inside task execution transactions. This poller-level version handles the case where a task fails via an unhandled exception that bypasses the memory engine's own failure path (e.g. a DB constraint violation that rolls back the engine's transaction before it can update the parent). rNrz%SELECT result_metadata, bank_id FROM z WHERE operation_id = $1result_metadataparent_operation_idr)zSELECT operation_id FROM z4 WHERE operation_id = $1 AND bank_id = $2 FOR UPDATEz3 SELECT status, error_message FROM zn WHERE bank_id = $1 AND result_metadata::jsonb @> $2::jsonb c3*K|]}|ddvVdS)r) completedrNr8rrr s r#rz>WorkerPoller._maybe_update_parent_operation..s-&`&`RSq{6M'M&`&`&`&`&`&`r%c3.K|]}|ddkVdS)rrNr8rs r#rz>WorkerPoller._maybe_update_parent_operation.."s+GGQx[H4GGGGGGr%rz SET status = 'failed', error_message = $2, updated_at = now() WHERE operation_id = $1 z SET status = 'completed', updated_at = now(), completed_at = now() WHERE operation_id = $1 z Poller updated parent operation z to rrz (all siblings done)z,Failed to update parent operation for child r)ruuidrfetchrowUUIDrr4rrrdumpsrrrr$rinforerror)rkrr*rrrrrrrr) parent_rowr any_failedrs r#rz+WorkerPoller._maybe_update_parent_operations   +V44G c WWWW ,--C !"34:O/3// >",$*_"="="1"5"56K"L"L & )nG $}}gEggg -..  J  "ZZ38  13FGHHH 3&`&`W_&`&`&`#`#` GGhGGGGGJ ll! II1223H==ll! II122  KKQ3FQQ):88{QQQ      c c c LLaHZaa^_aa b b b b b b b b b  cs25G AG >G A G )B5G H* HHretry_atr cxKtd|}t|dkr |ddn|}|j4d{V}|d|d|||d{Vdddd{Vn#1d{VswxYwYt d|d|d|dS) z4Reset task to pending with a future retry timestamp.rrNrz SET status = 'pending', next_retry_at = $2, worker_id = NULL, claimed_at = NULL, retry_count = retry_count + 1, error_message = $3, updated_at = now() WHERE operation_id = $1 Task z scheduled for retry at r)rrrSrrrr)rkr;r rr*rrs r#_schedule_retryzWorkerPoller._schedule_retry@s+V4403M0B0BT0I0I ete,,} =((**       d,,                                      `|``X``Q^``aaaaas#B BB exec_datereasonc8Ktd|}|j4d{V}|d|d||d{Vdddd{Vn#1d{VswxYwYtd|d|d|dS)uReset task to pending for re-pickup at exec_date without counting as a retry. Unlike `_schedule_retry`, this does not bump `retry_count` and does not populate `error_message` — defer is intentional backpressure, not a failure. rNrz SET status = 'pending', next_retry_at = $2, worker_id = NULL, claimed_at = NULL, updated_at = now() WHERE operation_id = $1 r z deferred until r)rrSrrrr)rkr;rrr*rrs r#_defer_operationzWorkerPoller._defer_operationRs^ +V44=((**       d,,                                       OLOO)OOvOOPPPPPs"A&& A03A0rc Kjdd}jddjdd}td|}tj|}j4d{Vt|j|tj ||j j <xj d z c_ jd d zj<dddd{Vn#1d{VswxYwY|fd dS) zz+WorkerPoller.execute_task..s'G,?@R@RSWSdft@u@u,v,vr%)r<rrr^r_execute_task_innerrer'r*r monotonicrhr;rcriadd_done_callback)rkrr/r)holderr+rs`` @r# execute_taskzWorkerPoller.execute_taskfsN&&vy99 ++,&>tV&L&LMM' i i i i i i i i4B&{>++##555D t0 1  ! !Q & ! !6:6M6Q6QR`bc6d6dgh6hD #N 3 i i i i i i i i i i i i i i i i i i i i i i i i i i i !!"v"v"v"v"v"vwwwwwsA0D!! D+.D+rcfK|j4d{V||jvrr|j|d|xjdzc_|j|d}|dkr&|dz |j|<|j|dkr|j|= dddd{VdS#1d{VswxYwYdS)z+Remove task from tracking after completion.Nr r)rerhpoprcrir)rkr;rcounts r#rzWorkerPoller._cleanup_tasks' D D D D D D D Dt111"&&|T:::%%*%%/33NAFF199>CaiD+N;.~>!CC 3NCC D D D D D D D D D D D D D D D D D D D D D D D D D D D D D DsA>'batch_id' IS NULL rrz$ failed to recover tasks for schema rz recovered z stale tasks from previous run)rwr_recover_batch_operationsrSrrrTr7splitrr4rrr) rkrx total_countr*r batch_countrrr"rrs r#recover_own_taskszWorkerPoller.recover_own_taskss))++++++++  u uF u !3V<<%)$B$B6$J$JJJJJJJ {*  =0022d#'<< %  $$F4:@FLLNN2.///qu$  u u u28!IVc&kksss^lsspqsstttttttt u ?? KKi$/iikiii j j js<AC.4&B,C., B6 6C.9B6 :3C.. E8?D==Ec DKtd|} |j4d{V}|d|dd{V}dddd{Vn#1d{VswxYwY|sdSd}|D]Q}t |d}|d}|d} t | trt j| } | d } | d d } t d |d | d| t |trt j|} n|} |j4d{V}| d|d|d{Vdddd{Vn#1d{VswxYwY|dz }t d|dS|S#t$rB} |rd|dnt |}t d|d| Yd} ~ dSd} ~ wwxYw)aM Recover batch API operations that were in-flight when worker crashed. Finds operations with batch_id in metadata and re-submits them as tasks so polling can resume. Args: schema: Database schema to recover from Returns: Number of batch operations recovered rNza SELECT operation_id, task_payload, result_metadata FROM z WHERE status = 'processing' AND result_metadata ? 'batch_id' AND task_payload IS NOT NULL rr;rrbatch_idbatch_provideropenaiz)Recovering batch operation: operation_id=z , batch_id=z , provider=r+z SET status = 'pending', worker_id = NULL, claimed_at = NULL, updated_at = now() WHERE operation_id = $1 r zBatch operation z# reset to pending for re-processingrz.Failed to recover batch operations for schema r)rrSrrr4rrrrrrrrr)rkr*rrr recoveredrr;rrr3r4r<rrs r#r-z&WorkerPoller._recover_batch_operationss+V44; },,..       $!ZZ                            qI# b# b"3~#677 ">2 "%&7"8os33B&*j&A&AO*..z::!0!4!45Ex!P!P ~ ~~Ya~~n|~~ lC00- $ < 8 8II ,I =0022d,, % % Q  `|```aaaa    .4E]]]]]#f++N LL_.__\]__ ` ` `11111 seG A%G% A//G2A/3G:C.G(!F G F% %G(F% ))G H7HHc K|d{V|jr6dd|jDnd}t d|jt |jz }t d|j d|jd|d |d |j s} | d{V}|r.=s0MMTQjjQjjMMMMMMr%nonerrz" starting polling loop (max_slots=z, reservations=[z], shared_pool=r$rrrr rPc3*K|]\}}|d|VdS):Nr8r;s r#rz#WorkerPoller.run..Ws0)T)TAQ****)T)T)T)T)T)Tr%c3"K|] }|r|ndV dSdefaultNr8rs r#rz#WorkerPoller.run..Ys++X+Xa,AAA +X+X+X+X+X+Xr%z claimed z tasks (z consolidation): z (schemas: i)rz polling loop cancelledz error in polling loop: z polling loop stopped)!r1r[joinrrrZrrrrrTr`is_setrrar<rrr*rrr^wait_forwaitrV TimeoutError_log_progress_if_dueCancelledErrorrrr'r(r) rkreservations_str shared_poolr task_types schemas_seenconsolidation_countrrsr( types_str schemas_strrs r#runzWorkerPoller.run3s$$&&&&&&&&&RVQh tDIIMMT-D-J-J-L-LMMM M M Mnt !T_s43J3Q3Q3S3S/T/TTUU   kdo k k/ k k;K k k\g k k k   .''))5 '4 '"..0000000013J47EEL*+' %55 N..vyAA"&."4"45Ey"Q"Q(2q!(<(;I:9J*:J  J* J  J**2M M'AL>>M>@cKtd|jd|jt j}t j|z |kr |j4d{V|j }d|j D}dddd{Vn#1d{VswxYwY|dkr%td|jddStd|jd|d|r+t j |d t j d{V\}}nt jd d{Vt j|z |k td|jd |d |j4d{Vt!|j D]7\}}|js|j8 dddd{VdS#1d{VswxYwYdS) z Signal shutdown and wait for current tasks to complete. Args: timeout: Maximum time to wait for in-flight tasks (seconds) rz initiating graceful shutdownNcg|] }|j Sr8)r+rrrs r#rtz2WorkerPoller.shutdown_graceful..s&\&\&\t|&\&\&\r%rz graceful shutdown completez waiting for z in-flight tasksg?)r return_whenz shutdown timeout after zs, cancelling remaining tasks)rrrTr`rar^rrrercrhrrFFIRST_COMPLETEDrrrrr+donecancel) rkrrractive_task_objectsrXrr;rs r#shutdown_gracefulzWorkerPoller.shutdown_graceful~s  LdoLLLMMM +--2244 $&&++-- :WDD+ ] ] ] ] ] ] ] ] 1 &\&\@R@Y@Y@[@[&\&\&\# ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ]A~~ RdoRRRSSS KK[$/[[ [[[ \ \ \# ) ' -@#[b[r s s sssssssaamC((((((((($&&++-- :WDD" ppp'pppqqq' * * * * * * * *&*4+=+C+C+E+E&F&F * *" d|((***L''))) * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *s%+C C%(C%$AI I!$I!c  Ktj}||jz tkrdS||_ |j4d{V|j}t |j}t |j}dddd{Vn#1d{VswxYwYd}g}|j D]b\}}| |d} td|| z } |t|| z }| |d| d|d| dct|j} td|j| z } td||z } td| | z }|rd|nd}i}|D],}|j|jf}| |dd z||<-d | D}|rd|dd nd}t)|d kr|d t)|d z d z }|d{V}t)|}||d{V}d|D}||z}d}i}i}|j4d{V}|D]7}t3d|} |d|dd{V} n#t6$rg} YnwxYw| D]}!|!dpd}||ddddd}"|"dxx|!dz cc<|"dxx|!dz cc<|"dxx|!dz cc<|"dxx|!dz cc<||!dz } |d|dd{V}#n#t6$rg}#YnwxYw|#D].}$|$dpd}%| |%d|$dz||%</9 dddd{Vn#1d{VswxYwYg}&| D]*\}%}'|%|jkr|& |%d|'+|&rd|&nd}(|})|}*t)|}+tAd|D},d|,dd}-t)|,dkr|-d t)|,dz d z }-tB"d |jd!|d|jd"|d#| d| d|d$|d%|+d|d&|-d'|(d(|)d)|*d*||dkr|#||$|tj%+|&d{VdS#t6$r(}.tB'd,|.Yd}.~.dSd}.~.wwxYw)-aLog progress stats every PROGRESS_LOG_INTERVAL seconds. Emits four kinds of lines: * [WORKER_STATS] - aggregate slots / pool / global pending counts * [WORKER_TASK] - one line per in-flight task with age + stage * [STUCK_STACK] - async stack trace for tasks past stuck thresholds * [DB_WAITS] - any non-idle hindsight session waiting on a lock Nrr:/z(avail=r$r8r=r c0g|]\\}}}|d|d|dS)r?(r$r8)rropbankcnts r#rtz5WorkerPoller._log_progress_if_due..s9```"dS"44t44c444```r%rEz +z morech|] }|j Sr8rqrUs r#r~z4WorkerPoller._log_progress_if_due..s(W(W(W(W(W(Wr%ra SELECT operation_type, COUNT(*) AS total, SUM(CASE WHEN task_payload IS NULL THEN 1 ELSE 0 END) AS payload_null, SUM(CASE WHEN next_retry_at IS NOT NULL AND next_retry_at > now() THEN 1 ELSE 0 END) AS retry_blocked, SUM(CASE WHEN worker_id IS NOT NULL THEN 1 ELSE 0 END) AS assigned FROM z WHERE status = 'pending' GROUP BY operation_type rr)total payload_null retry_blockedassignedrdrerfrgzb SELECT worker_id, COUNT(*) as count FROM z WHERE status = 'processing' GROUP BY worker_id rGr"r?c3"K|] }|r|ndV dSrAr8rs r#rz4WorkerPoller._log_progress_if_due..+s+!R!RAq"7!!i!R!R!R!R!R!Rr%z[WORKER_STATS] worker=z slots=z | reserved: [z ] | shared=z) | global: pending=z (queried=z schemas: z ) | others: z | pool: z | proc: z | my_active: )nowzFailed to log progress stats: )(rrfPROGRESS_LOG_INTERVALrercr=rirhr[rrrrrrrrZrCr(r)rrwrrSrrrr setdefaultrT_format_pool_stats_format_proc_statssortedrr_log_pending_breakdown_log_per_task_linesr _log_db_waitsr&)/rkrjrin_flight_by_type active_tasksrreserved_partsr(r@type_in_flighttype_availablerrrr reserved_str task_groupsrkeyprocessing_infoprocessing_strrxtotal_schema_countschemas_with_pendingschemas_with_active_tasksschemas_to_queryglobal_pendingall_worker_countspending_breakdownrr*rbreakdown_rowsbrbucket worker_rowswrwid other_workersrb others_strpool_strproc_str queried_count queried_listrPrs/ r#rHz!WorkerPoller._log_progress_if_dues ikk ( (+@ @ @ F"%Y ?+ 8 8 8 8 8 8 8 8 1 $()@$A$A!#D$677  8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 !" N%)%<%B%B%D%D h h!!2!6!6w!B!B!$Q>(A!B!B!S>%B%BB!%%&f&f>&f&fH&f&fUc&f&f&fgggg"4#:#A#A#C#CDD "1do8H&HII !!Y1B%BCCO"1&6&HII 8FR499^444FL79K$++-- ? ?|T\2#.??3#:#:Q#> C  ``KL]L]L_L_```O@O[TYYss';<<>>  $ $\t~7G7G $ H H H$$&& & & & & & & & & & ? ? ? LL=!== > > > > > > > > > ?s W 0B0W  B  W B H:W P!LP L PLBP!OP O PO6PW  P##W &P#'F$W W?W::W?c ddl}ddl}||j}|j}|jdkr|dz}|dz }d|dS#t $r(}td|Yd}~d Sd}~wwxYw) zVRender lightweight process memory stats. Returns 'unavailable' if introspection fails.rNdarwiniizrss_mb=.0fzProcess stats unavailable: unavailable) resourcesys getrusage RUSAGE_SELF ru_maxrssplatformrrr&)rkrrusagerssrss_mbrs r#rnzWorkerPoller._format_proc_statsMs ! OOO JJJ&&x';< 0`` while workers report free slots, the bug is somewhere else (lock contention, tenant discovery, etc.) - this line narrows the search. Nrdrerfrgz: total=z claimable=z payload_null=z retry_blocked=z assigned=z[PENDING_BREAKDOWN] z | )rorrrrC)rkrrr(b claimables r#rpz#WorkerPoller._log_pending_breakdowns  Fi((  G'"A' Q~%66?9KKaPZm[I LL,,AgJ,,9,, !. 1,,BCOBT,,jM,,      >5::e+<+<>>?????r%rtrjc|sdSt|fdd}|D]\}}|jz }|j}||jnd}| |jz nd} |t krdnd} |jr d |jnd} t d | d |d |j d |j d|j | d|dd|d| dd| |||dS)zEmit one [WORKER_TASK] line per in-flight task and dump stuck stacks. Sorted by age desc so the oldest (most likely stuck) tasks appear first. Nc4|djz |dfS)Nr r)r,)kvrjs r#rz2WorkerPoller._log_per_task_lines..sC"Q%"22BqE:r%T)rzreverserrRz [STUCK?] rz schema=z[WORKER_TASK] zop= type=z op_type= bank= age=rs stage=z stage_age=r )rorr,r-rr%STUCK_STACK_INITIAL_THRESHOLD_Sr*rrr/r(r)_maybe_dump_stuck_stack) rkrtrjorderedop_idrage_srr stage_age_s stuck_marker schema_parts ` r#rqz WorkerPoller._log_per_task_liness   F    ::::    # = =KE4$/)E&F$*$6FLLIE7=7I3!222sK*/3R*R*R;;XZL6:kI2T[222rK KKOOO%OOt~OO<OO/3|O=HOONOO*/OO'>'">'>'DcN+Yt_<= >'  >' d >'3>'>' S>D0>'>'>'>'@\Dt$4\\\\!$sTz2B!s3QU:!!!!FV,<VVVV@&&5&D&&&&4l4 #4llll\ Dj 37S> QT  k     *Dj*37S>*QT* k ****X # sTz    VsV3VPSVZPZVVVV&UcsUcTWZ^T^UcimUcUcUcUcnb#bbWZbdgjndnbbbb$Q3Q5QRUQ_bei_iQQQQ(x{xxxxB D DS D D D D!L!Lk!L;QUCU!L!L!L!LF/////bJcDjJSJJJJXIFIFIFV#*#*u#*#*#*#*Jh?h?h?T!C!!!!$!!C!!!!!!!!F@S$sCx.5H0I@d@@@@:=S.5H0I=PU=Z^====< FS F Fu FY] F F F FD0000d3X'T'''X'''r%rC),r3r^rrloggingrr' collectionsrcollections.abcrrr dataclassesrrtypingr r engine.schemar r exceptionsrrrrrhindsight_api.engine.db.baserhindsight_api.extensions.tenantr getLoggerr0rrkrrr4r$r'r:r?rCr8r%r#rsC  9999999999((((((((%%%%%%%%99999933333333++++++++@<<<<<<??????  8 $ $#&&o#<  &    I I I I I I I  If'f'f'f'f'f'f'f'f'f'r%