j#dZddlmZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z d*d Zd+d Z d,d d ddd-dZd,d.dZd,d/dZd0d1dZd2dZ d3d4d#Zd5d6d&Zd7d8d)ZdS)9a4SQLite-backed durable task queue. This module replaces in-process ``threading.Thread`` dispatching for heavy workloads such as training, inference, and face library rebuilds. Tasks survive Web-process restarts and can be consumed by a dedicated ``worker.py`` process. The queue is stored in the same SQLite database used by the rest of the app (``SQLITE_DB_PATH``). Producer example:: from shared.task_queue import submit_task task_id = submit_task("train", payload={"dataset_id": "...", ...}) Consumer example:: from shared.task_queue import claim_task, complete_task, fail_task row = claim_task("train") if row: try: do_work(row["payload"]) complete_task(row["id"], result={...}) except Exception as e: fail_task(row["id"], error=str(e)) ) annotationsN)Any)uuid4)SQLITE_DB_PATHloggerreturnsqlite3.Connectionctjt}|rtj|dt jtd}t j|_| d|S)NT)exist_ok)timeoutzPRAGMA journal_mode=WAL) ospathdirnamermakedirssqlite3connectRow row_factoryexecute)parentconns 3/home/longshao/multi-rider-rag/shared/task_queue.py_connectr-sh W__^ , ,F + FT**** ?>2 6 6 6D{DLL*+++ KrNonec|d|d|ddS)zrCreate the ``task_queue`` table if it does not exist. Called once from :func:`shared.db.sqlite.init_db`. at CREATE TABLE IF NOT EXISTS task_queue ( id TEXT PRIMARY KEY, task_type TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', payload TEXT NOT NULL DEFAULT '{}', result TEXT NOT NULL DEFAULT '{}', error TEXT NOT NULL DEFAULT '', owner_key TEXT NOT NULL DEFAULT '', owner_ip TEXT NOT NULL DEFAULT '', created_ts INTEGER NOT NULL DEFAULT 0, claimed_ts INTEGER, finished_ts INTEGER, retries INTEGER NOT NULL DEFAULT 0 ) zVCREATE INDEX IF NOT EXISTS idx_task_queue_status_type ON task_queue(status, task_type)zPCREATE INDEX IF NOT EXISTS idx_task_queue_created ON task_queue(created_ts DESC)N)r)rs rinit_task_queue_tabler;s_  LL $ LL` LLZr) owner_keyowner_iptask_id task_typestrpayloaddict[str, Any] | Noner r!r" str | Nonec t|ptj}ttj}t 5}|d||t j|pid|||f|dddn #1swxYwYtj d|||S)z*Enqueue a new task and return its task id.z INSERT INTO task_queue (id, task_type, status, payload, owner_key, owner_ip, created_ts) VALUES (?, ?, 'pending', ?, ?, ?, ?) F ensure_asciiNztask submitted: %s type=%s) rhexinttimerrjsondumpscommitrinfo)r#r%r r!r"nowrs r submit_taskr3^s$G dikk  C t    7=bu===      K,gyAAA NsABBBcZttj}t5}|d |r*|d|f}n'|d}|"| ddddS|d||df}|jdkr"| ddddS|n##t$r|wxYw dddn #1swxYwY tj |d}n#t$ri}YnwxYw|d|d ||d |d |d d S)a Atomically claim the oldest pending task. When *task_type* is provided, only tasks of that type are considered. The ``BEGIN IMMEDIATE`` transaction prevents multiple worker processes from selecting the same pending row before it is marked as running. zBEGIN IMMEDIATEz SELECT * FROM task_queue WHERE status='pending' AND task_type=? ORDER BY created_ts ASC, id ASC LIMIT 1 z SELECT * FROM task_queue WHERE status='pending' ORDER BY created_ts ASC, id ASC LIMIT 1 NzTUPDATE task_queue SET status='running', claimed_ts=? WHERE id=? AND status='pending'idr%r#r r! created_ts)r5r#r%r r!r7) r,r-rrfetchoner0rowcountrollback Exceptionr.loads)r#r2rrowcurr%s r claim_taskr?sQ dikk  C %t &'''#  ll L(**ll(**{ 3%%%%%%%%6,,fc$i C|q   C%%%%%%%%D KKMMMM    MMOOO   E%%%%%%%%%%%%%%%N*S^,, $i%% O,'   sHE A)D==DDE  D==E  EEE44 FFresultcBttj}t5}|dt j|pid||f|dddn #1swxYwYtjd|dS)zcounts rreset_stale_runningrcs    /F tll q I    LE O >vNNN Ls,A**A.1A.r dayscttj|dzz }t5}|d|f}|dddn #1swxYwY|jS)z0Delete completed/failed tasks older than *days*.iQzRDELETE FROM task_queue WHERE status IN ('completed', 'failed') AND finished_ts < ?N)r,r-rrr0r9)rdrarr>s rcleanup_old_tasksrfs   u ,F tll ` I    <s,A--A14A1)rr )rr rr)N) r#r$r%r&r r$r!r$r"r'rr$)r#r'rr&)r"r$r@r&rr)r)r"r$rCr$rr)r"r$rr&)NNrP)r#r'rQr'rRr,rrS)r^)r_r,rr,)r )rdr,rr,)__doc__ __future__rr.rrr-typingruuidrshared.config.configrrrrr3r?rBrGrOr]rcrfrrrms8#"""""  77777777J&* F;;;;;| / / / / / F F F F F     !B            r