fix: dispatcher crash on invalid failure_stage + per-task error isolation
Two fixes: 1. db_layer.update_task_status: map invalid failure_stage values (e.g. 'process' from Edge) to 'callback' before DB write, preventing MariaDB ENUM DataError (1265 "Data truncated") 2. dispatcher._poll_once: wrap each task dispatch in try/except so one task's failure doesn't skip remaining tasks in the batch Root cause: Edge returns failure_stage='process' but DB ENUM only allows download/extract/vlm_visual/vlm_fusion/callback. The DataError crashed _poll_once(), causing all subsequent PENDING tasks to be skipped.
This commit is contained in:
@@ -92,6 +92,9 @@ def get_tasks_by_status(status: str, limit=10) -> List[Dict]:
|
|||||||
def update_task_status(task_id: int, status: str, error_message: str = None,
|
def update_task_status(task_id: int, status: str, error_message: str = None,
|
||||||
failure_stage: str = None):
|
failure_stage: str = None):
|
||||||
"""更新任务状态"""
|
"""更新任务状态"""
|
||||||
|
valid_stages = {'download', 'extract', 'vlm_visual', 'vlm_fusion', 'callback'}
|
||||||
|
if failure_stage and failure_stage not in valid_stages:
|
||||||
|
failure_stage = 'callback'
|
||||||
conn = get_conn()
|
conn = get_conn()
|
||||||
try:
|
try:
|
||||||
cursor = conn.cursor()
|
cursor = conn.cursor()
|
||||||
|
|||||||
@@ -165,7 +165,10 @@ class Dispatcher:
|
|||||||
tasks = db_layer.get_pending_tasks(limit=10)
|
tasks = db_layer.get_pending_tasks(limit=10)
|
||||||
for task in tasks:
|
for task in tasks:
|
||||||
if self._should_retry(task):
|
if self._should_retry(task):
|
||||||
|
try:
|
||||||
self._dispatch_one(task)
|
self._dispatch_one(task)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"[task_id={task['task_id']}] dispatch 异常: {e}", exc_info=True)
|
||||||
|
|
||||||
def _run(self):
|
def _run(self):
|
||||||
"""线程主循环"""
|
"""线程主循环"""
|
||||||
|
|||||||
Reference in New Issue
Block a user