Qj.%SrSSKJr SSKrSSKrSSKrSSKrSSKrSSKJ r SSK J r J r J r JrJrJrJrJr SSKJr SSKJr \ "\5R1S5r\R4"5S \R6"53rS rS rS rS r Sr!Sr"S\#S'1Skr$Sr%SSjr&SSjr'S Sjr(S!Sjr)S"Sjr*S#Sjr+S$Sjr,"SS5r-S%Sjr.g)&aYJob records in Postgres. An upload job and a batch run each live in the memory of the task running them, as they always have; this module is the durable copy. The running task writes every stage transition, every finished batch row, and a heartbeat every few seconds. Any task can then answer a status poll or serve a run's rows from the tables, and a task that dies leaves a record whose heartbeat stops: a job with no heartbeat for ``STALE_SECONDS`` is marked failed with that reason the next time anything reads it, instead of sitting at its last stage forever. The tables are created by ``ensure_tables`` at startup from ``jobs.sql``, which is idempotent, so a running database gains them without a migration step. Nothing here is imported by the batch runner; the runner takes a store object with the same method names, and tests run it without one. ) annotationsN)Path)AnyCallableDictIterableList LiteralStringOptionalcast)logger)dbzjobs.sql:Z )donefailedz!the task running this job stopped)job_idfilenametitlefolder filing_typeyearperiodversiontrackreplace replace_id replace_keydoc_key content_hashdoc_id source_uristagestageserror error_infolog scan_pagesscan_pages_reasonztuple[str, ...]_UPLOAD_COLUMNS>r)r&r(i=ocH[[[R55n[R "5oR 5 URS[45 URU5 SSS5 SSS5 g!,(df  N=f!,(df  g=f)Nz SELECT pg_advisory_xact_lock(%s)) r r JOBS_SQL read_textrconnect transactionexecute _TABLES_LOCK)sqlconns 3/home/mande/repo/quber/src/quber/playground/jobs.py ensure_tablesr7Ns] }h002 3C //1 7,I S 211s#B)B1B B B B!c\/n[HlnURU5nUS:Xa[U=(d /5[*SnU[;aUb[ R "U5OSnURU5 Mn SR[5nSRS/[[5-5nSRS[SS55nSU-S-U-S -U-S -n[R"5nURU/UQ[P75 SSS5 g!,(df  g=f) zGWrite the job as it stands, log tail included, and stamp the heartbeat.r)N, z%sc3.# UH oSU3v M g7f)z = EXCLUDED.N).0cs r6 save_upload..ds&ZFYL' ALL(%s) AND (heartbeat IS NULL OR heartbeat < now() - make_interval(secs => %s)) RETURNING job_idNzupload job {} failed: {}) rr0r2STOPPED_REASONrBTERMINAL_STAGES STALE_SECONDSfetchallr warningrI)r5rowsrs r6mark_stale_uploadsre|ss || # T/2M B   (*   16>J t9 s 9B  Bc[5 SR[5n[R"5nUR SU-S-U45R 5nSSS5 W(a [U5$S$!,(df  N"=f)Nr9SELECT z2 FROM ade_playground.upload_jobs WHERE job_id = %s)rerHr,rr0r2fetchoner\)rrOr5r[s r6 load_uploadrisp!YY7G ll  "V V I  (*   %( C 1T1 s (A99 BcH[5 SR[5n[R"5nUR SU-S-[ [545R5nSSS5 WVs/sHn[U5PM sn$!,(df  N*=fs snf)z+Every upload job still running on any task.r9rgz7 FROM ade_playground.upload_jobs WHERE stage <> ALL(%s)N) rerHr,rr0r2rBr`rbr\)rOr5rdrs r6 live_uploadsrls!YY7G ||  "[ [ / " $  (*   *. .A Q  ..  /s5B6B Bc[U5nU(dg[R"5nURSU45 SSS5 g!,(df  g=f)NzNUPDATE ade_playground.upload_jobs SET heartbeat = now() WHERE job_id = ANY(%s)rBrr0r2)job_idsidsr5s r6heartbeat_uploadsrqs: w-C   \ F   A Acv\rSrSrSrS SjrS SjrSSjrSSjrSSjr SSjr SS jr S r g ) BatchStorezHThe batch runner's durable side. Method names are the runner's contract.c [R"5nURSXU[R"U5[ 45 SSS5 g!,(df  g=f)NzINSERT INTO ade_playground.batch_runs (job_id, doc_id, want, questions, owner, heartbeat) VALUES (%s, %s, %s, %s, %s, now()) ON CONFLICT (job_id) DO NOTHING)rr0r2rErFrJ)selfrr#want questionsr5s r6save_runBatchStore.save_runs? ZZ\T LL6tzz)' %s))z=SELECT index FROM ade_playground.batch_rows WHERE job_id = %srzINSERT INTO ade_playground.batch_rows (job_id, index, question, error) VALUES (%s, %s, %s, %s) ON CONFLICT DO NOTHINGzpUPDATE ade_playground.batch_runs SET running = false, failed = failed + %s, updated_at = now() WHERE job_id = %sz'batch run {} closed: {} ({} unanswered)N)rr0r2rarbrXrYrErZrangerIr_r rc) rwr5stalerryrkrimissingr}s r6mark_stale_runsBatchStore.mark_stale_runssOZZ\TLLm hj  &+!5? 35O5ODJJy1U^ "\\WZ`Ybhj!!aD! ',C N&;M&;}1&;M$ELLM (8.I%  G\6*=v~WZ[bWc'&+ \ N\s1A=E E"E : EE A-E  E  EcfUR5 [R"5nURSU45R 5nUc SSS5 gURSU45R 5nSSS5 [ WS[5(a[R"US5OUSnUUSUSUUSUSWVs/sHGnUSUSUS[ US[5(a[R"US5OUSS .PMI snS .$!,(df  N=fs snf) z,The run and its rows as plain data, or None.z`SELECT doc_id, want, questions, failed, running FROM ade_playground.batch_runs WHERE job_id = %sNzgSELECT index, question, error, response FROM ade_playground.batch_rows WHERE job_id = %s ORDER BY indexrr@)r}r~r'r)rr#rxryrrunningrd) rrr0r2rhrbrXrYrErZ)rwrr5runrdryrks r6load_runBatchStore.load_runs6  ZZ\T,,r hj { \<<y hj +5SVS*A*ADJJs1v&s1v !fF"!f1vA qT !!qT4>qtS4I4I 1Q4 0qQRt    \&s&D!D AD. D+cUR5 [R"5nURSU45R 5nSSS5 USL$!,(df  WSL$=f)NzMSELECT 1 FROM ade_playground.batch_runs WHERE doc_id = %s AND running LIMIT 1)rrr0r2rh)rwr#r5r[s r6 running_forBatchStore.running_fors_  ZZ\T,,_bhajhj $ \$s "A A%c[U5nU(dg[R"5nURSU45 SSS5 g!,(df  g=f)NzMUPDATE ade_playground.batch_runs SET heartbeat = now() WHERE job_id = ANY(%s)rn)rwrorpr5s r6 heartbeatBatchStore.heartbeats;7m  ZZ\T LL_beag \\rrr;N) rrYr#intrxrYryz List[str]returnNone) rrYr}rr~rYr'z Optional[str]rOptional[Dict[str, Any]]rr)rrYrrrrrrYrr)r#rrboolroz Iterable[str]rr) __name__ __module__ __qualname____firstlineno____doc__rzrrrrrr__static_attributes__r;r]r6rtrtsZR"%14=JVn  @ Br]rtcl^^^SUUU4Sjjn[R"USSS9nUR5 U$)zBeat, every ``HEARTBEAT_SECONDS``, for every job this process is still running. The callables name them at each beat, so a job that finished between beats is simply not beaten again.c>[R"[5 [T"55 TR T"55 MC![ a!n[ R"SU5 SnAN(SnAff=f)Nzheartbeat failed: {})timesleepHEARTBEAT_SECONDSrqr Exceptionr rc)excrun_idsstore upload_idss r6beatstart_heartbeat..beat3sY JJ( ) <!*,/ *   <5s;; rs"#  UUU > # #J /     "))+/  $4$20  .4  2 / ssr+ (  r]