fix(intelligent-eval): atomic CAS in assign_task and complete_task (resolves §6.1)
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
Replace read-check-write in task_queue.assign_task with UPDATE...WHERE status='pending' and decide on rowcount so two concurrent workers cannot both claim the same task. Also harden complete_task with the same CAS pattern (status='assigned') so a worker + stuck-handler double-complete leaves the DB in one state. The xfail guard in test_worker_task_resilience now passes (4/4).
This commit is contained in:
parent
3b14eb1bda
commit
38e3817433
@ -10,7 +10,7 @@ OpenClaw workers to pick up. Tasks are prioritized by:
|
||||
from datetime import timedelta
|
||||
from typing import Optional
|
||||
|
||||
from sqlmodel import Session, select
|
||||
from sqlmodel import Session, select, update
|
||||
|
||||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||||
from agenteval.storage.db import (
|
||||
@ -113,39 +113,50 @@ def get_next_task(session: Session) -> Optional[IntelligentEvalTaskQueueDB]:
|
||||
|
||||
|
||||
def assign_task(task_id: str, cron_id: str, session: Session) -> bool:
|
||||
"""Assign a task to a cron.
|
||||
"""Atomically assign a pending task to a cron (CAS on status).
|
||||
|
||||
Returns:
|
||||
True if assigned successfully, False if task not found or already assigned
|
||||
P1 真问题修复(§6.1): use ``UPDATE ... WHERE status='pending'`` and decide
|
||||
on ``rowcount`` so two concurrent workers cannot both claim the same task.
|
||||
The previous read-check-write left a race because SQLite + two sessions
|
||||
could each read ``status=pending`` and each commit.
|
||||
"""
|
||||
task = session.get(IntelligentEvalTaskQueueDB, task_id)
|
||||
if task is None or task.status != "pending":
|
||||
return False
|
||||
|
||||
task.status = "assigned"
|
||||
task.assigned_cron_id = cron_id
|
||||
task.assigned_at = utc_now()
|
||||
task.updated_at = utc_now()
|
||||
stmt = (
|
||||
update(IntelligentEvalTaskQueueDB)
|
||||
.where(IntelligentEvalTaskQueueDB.id == task_id)
|
||||
.where(IntelligentEvalTaskQueueDB.status == "pending")
|
||||
.values(
|
||||
status="assigned",
|
||||
assigned_cron_id=cron_id,
|
||||
assigned_at=utc_now(),
|
||||
updated_at=utc_now(),
|
||||
)
|
||||
)
|
||||
result = session.exec(stmt)
|
||||
session.commit()
|
||||
return True
|
||||
return result.rowcount > 0
|
||||
|
||||
|
||||
def complete_task(task_id: str, success: bool, error: Optional[str], session: Session) -> bool:
|
||||
"""Mark a task as completed or failed.
|
||||
"""Atomically mark a task as completed/failed (CAS on status).
|
||||
|
||||
Returns:
|
||||
True if completed successfully, False if task not found
|
||||
P1 真问题修复(§6.1 审计): guard with ``status='assigned'`` so a
|
||||
double-complete from worker + stuck-handler leaves the DB in one state.
|
||||
"""
|
||||
task = session.get(IntelligentEvalTaskQueueDB, task_id)
|
||||
if task is None:
|
||||
return False
|
||||
|
||||
task.status = "completed" if success else "failed"
|
||||
task.completed_at = utc_now()
|
||||
task.error = error
|
||||
task.updated_at = utc_now()
|
||||
terminal = "completed" if success else "failed"
|
||||
stmt = (
|
||||
update(IntelligentEvalTaskQueueDB)
|
||||
.where(IntelligentEvalTaskQueueDB.id == task_id)
|
||||
.where(IntelligentEvalTaskQueueDB.status == "assigned")
|
||||
.values(
|
||||
status=terminal,
|
||||
completed_at=utc_now(),
|
||||
error=error,
|
||||
updated_at=utc_now(),
|
||||
)
|
||||
)
|
||||
result = session.exec(stmt)
|
||||
session.commit()
|
||||
return True
|
||||
return result.rowcount > 0
|
||||
|
||||
def requeue_stuck_task(eval_id: str, cron_id: str, session: Session) -> bool:
|
||||
"""Mark the cron-stuck task as failed and enqueue a retry task.
|
||||
|
||||
@ -41,14 +41,6 @@ def test_concurrent_assign_only_one_succeeds(db_session):
|
||||
assert task.assigned_cron_id == "cron-A"
|
||||
|
||||
|
||||
@pytest.mark.xfail(
|
||||
reason=(
|
||||
"Known race: assign_task lacks atomic CAS — two sessions can both read "
|
||||
"status=pending and both commit. Tracked in .scratch/v111-architecture-scan.md "
|
||||
"discoveries. Fix requires atomic UPDATE ... WHERE status=pending in assign_task."
|
||||
),
|
||||
strict=False,
|
||||
)
|
||||
def test_concurrent_assign_via_two_sessions(tmp_db_path, db_session):
|
||||
task = IntelligentEvalTaskQueueDB(
|
||||
eval_id="eval1", status="pending", priority=1, reason="slot_due"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user