fix(translate): handle row lock timeout in cancel, add flush fallback, migrate progress to store
- Extract _fallback_cancel_request for row lock timeout recovery in orchestrator_cancel - Add flush lock timeout fallback in cancel_run - Set error_message when all records fail in executor - Track insert_failed state in scheduler and frontend store - Migrate TranslationRunProgress from local polling to centralized store - Fix nginx resolver for Docker variable-based proxy_pass - Add FRONTEND_HOST_PORT env var to docker-compose
This commit is contained in:
@@ -223,6 +223,38 @@ class TestCancelRunOrthogonal:
|
||||
orch.cancel_run("run-ghost")
|
||||
# endregion test_cancel_run_lock_timeout_nonexistent_after_flag
|
||||
|
||||
# region test_cancel_run_flush_lock_timeout_fallback [C:2] [TYPE Function]
|
||||
# @BRIEF If row lock happens during flush, fallback sets CANCEL_REQUESTED and does not raise 500.
|
||||
def test_cancel_run_flush_lock_timeout_fallback(self) -> None:
|
||||
from sqlalchemy.exc import OperationalError
|
||||
|
||||
db = MagicMock()
|
||||
config_manager = MagicMock()
|
||||
|
||||
locked_run = MagicMock(spec=TranslationRun)
|
||||
locked_run.id = "run-flush-lock-1"
|
||||
locked_run.job_id = "job-1"
|
||||
locked_run.status = "RUNNING"
|
||||
locked_run.error_message = None
|
||||
|
||||
flagged_run = MagicMock(spec=TranslationRun)
|
||||
flagged_run.id = "run-flush-lock-1"
|
||||
flagged_run.job_id = "job-1"
|
||||
flagged_run.status = "RUNNING"
|
||||
flagged_run.error_message = "CANCEL_REQUESTED"
|
||||
|
||||
db.query.return_value.filter.return_value.first.side_effect = [locked_run, flagged_run]
|
||||
db.flush.side_effect = OperationalError("lock timeout", {}, None)
|
||||
|
||||
orch = TranslationOrchestrator(db, config_manager, "test-user")
|
||||
result = orch.cancel_run("run-flush-lock-1")
|
||||
|
||||
db.rollback.assert_called_once()
|
||||
assert db.execute.call_count >= 2, "Expected SET LOCAL and fallback UPDATE to run"
|
||||
assert result.error_message == "CANCEL_REQUESTED"
|
||||
assert result.status == "RUNNING"
|
||||
# endregion test_cancel_run_flush_lock_timeout_fallback
|
||||
|
||||
# region test_cancel_run_failed_status_rejected [C:2] [TYPE Function]
|
||||
# @BRIEF FAILED runs cannot be cancelled (only PENDING/RUNNING).
|
||||
def test_cancel_run_failed_status_rejected(self) -> None:
|
||||
|
||||
@@ -174,6 +174,7 @@ class TranslationExecutor:
|
||||
run.status = "COMPLETED"
|
||||
elif s == 0:
|
||||
run.status = "FAILED"
|
||||
run.error_message = f"All {f} record(s) failed — see individual record errors for details"
|
||||
else:
|
||||
run.status = "COMPLETED"
|
||||
run.completed_at = datetime.now(UTC)
|
||||
|
||||
@@ -18,6 +18,24 @@ from .events import TranslationEventLog
|
||||
from .orchestrator_sql import SQLInsertService
|
||||
|
||||
|
||||
def _fallback_cancel_request(db: Session, run_id: str) -> TranslationRun:
|
||||
"""Set cancellation flag when row lock prevents direct status update."""
|
||||
try:
|
||||
db.rollback()
|
||||
except Exception:
|
||||
pass
|
||||
logger.explore("Row lock timeout — setting cancellation flag via direct SQL", {"run_id": run_id})
|
||||
db.execute(
|
||||
text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"),
|
||||
{"run_id": run_id},
|
||||
)
|
||||
db.commit()
|
||||
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
return run
|
||||
|
||||
|
||||
# #region cancel_run [C:3] [TYPE Function] [SEMANTICS translate, cancel, run]
|
||||
# @BRIEF Cancel a running translation run.
|
||||
# @SIDE_EFFECT DB writes; event log.
|
||||
@@ -35,16 +53,7 @@ def cancel_run(
|
||||
db.execute(text("SET LOCAL lock_timeout = '3s'"))
|
||||
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
except Exception:
|
||||
logger.explore("Row lock timeout — setting cancellation flag via direct SQL", {"run_id": run_id})
|
||||
db.execute(
|
||||
text("UPDATE translation_runs SET error_message = 'CANCEL_REQUESTED' WHERE id = :run_id"),
|
||||
{"run_id": run_id},
|
||||
)
|
||||
db.commit()
|
||||
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
return run
|
||||
return _fallback_cancel_request(db, run_id)
|
||||
|
||||
if not run:
|
||||
raise ValueError(f"Run '{run_id}' not found")
|
||||
@@ -56,7 +65,10 @@ def cancel_run(
|
||||
|
||||
run.status = "CANCELLED"
|
||||
run.completed_at = datetime.now(UTC)
|
||||
db.flush()
|
||||
try:
|
||||
db.flush()
|
||||
except Exception:
|
||||
return _fallback_cancel_request(db, run_id)
|
||||
event_log.log_event(
|
||||
job_id=run.job_id, run_id=run.id, event_type="RUN_CANCELLED",
|
||||
payload={}, created_by=current_user,
|
||||
|
||||
@@ -381,7 +381,14 @@ def execute_scheduled_translation(
|
||||
# Execute in same thread (APScheduler runs in background thread pool)
|
||||
try:
|
||||
orch.execute_run(run)
|
||||
run.insert_status = run.insert_status or "succeeded"
|
||||
if run.status == "FAILED":
|
||||
logger.explore("Scheduled translation completed with all records failed", {
|
||||
"run_id": run.id,
|
||||
"error": run.error_message,
|
||||
})
|
||||
run.insert_status = run.insert_status or None
|
||||
else:
|
||||
run.insert_status = run.insert_status or "succeeded"
|
||||
except Exception as exec_err:
|
||||
logger.explore("Scheduled translation execution failed", {
|
||||
"run_id": run.id,
|
||||
|
||||
Reference in New Issue
Block a user