fix(backend+frontend): migration deadlock, async pattern, timeout safety, fire-and-forget translations
Backend: - MigrationPlugin.execute — remove AsyncJobRunner.run() deadlock on post-migration ID sync (2 deadlocks total: plugins/migration.py + api/routes/migration.py trigger_sync_now). Replace blocking runner.run with direct await. - MigrationPlugin.execute — fix temp file leak (dry_run=True prevented cleanup). - MigrationPlugin.execute — IdMappingService(SessionLocal()) now closed in finally. - MigrationPlugin.execute — wire IdMappingService into MigrationEngine constructor so cross-filter patching actually works instead of silently skipping. - MigrationPlugin.execute — SupersetClient.aclose() in finally to prevent httpx connection pool leak. - TaskManager — _async_tasks dict leak (add_done_callback cleanup). - JobLifecycle — CancelledError handler (tasks stuck in RUNNING after cancel). - JobLifecycle — persist_task BEFORE _broadcast_task_status (crash consistency). - JobLifecycle — wait_for_resolution/wait_for_input now have 3600s timeout. - AsyncJobRunner.run — 300s timeout on future.result() (APScheduler thread safety). Translate scheduler: - execute_scheduled_translation — replace blocking runner.run(orch.execute_run(run)) with fire-and-forget TranslationOrchestrator.execute_background(). APScheduler thread is freed in ms instead of blocking for the full translation duration. Translation runs of 10k+ rows (200+ LLM batches, hours) no longer hit the 300s runner timeout. - TranslationOrchestrator.execute_background — new static method: opens own DB session, dispatches asyncio.create_task, handles errors + notification. - Scheduler last_run_at updated at dispatch time (not after completion). Frontend: - MigrationModel.stepReady[3] now requires dryRunResult != null (was always true, allowing UI to reach step 3 without dry-run). - WizardModel.goToStep gate for step 3 uses stepReady[3]. - +page.svelte — dryRunResult no longer self-clears via reactive loop. - Progress bar step 3 indicator gate fixed for new readiness logic. Tests: - 2 new regression tests for migration sync (deadlock-free, completes cleanly). - test_migration_plugin.py — _make_mock_superset_client/_make_mock_mapping_service helpers for proper async mock behavior (aclose, sync_environment AsyncMock). - 10 scheduled-translation tests updated for fire-and-forget pattern. - MigrationModel.test.ts — step 3 invariant + goToStep block test. - All affected tests: 104 backend + 79 frontend = 183 passed.
This commit is contained in:
@@ -9,11 +9,19 @@
|
||||
# @TEST_INVARIANT: trigger_type_dispatch -> VERIFIED_BY: [test_new_key_only_mode, test_baseline_expired_fallback, test_full_mode_default]
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import asyncio
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
def _make_coro_runner():
|
||||
"""Mock runner whose run() executes the coroutine synchronously via asyncio.run."""
|
||||
r = MagicMock()
|
||||
r.run.side_effect = lambda coro: asyncio.run(coro)
|
||||
return r
|
||||
|
||||
# -- Helpers -----------------------------------------------------------------
|
||||
|
||||
def _make_most_recent_run(created_at_delta_days=60):
|
||||
@@ -79,7 +87,8 @@ def test_new_key_only_mode():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
@@ -96,7 +105,8 @@ def test_new_key_only_mode():
|
||||
f"Expected trigger_type='new_key_only', got '{mock_run.trigger_type}'"
|
||||
)
|
||||
mock_orch.start_run.assert_called_once_with(job_id="job-1", is_scheduled=True)
|
||||
mock_orch.execute_run.assert_called_once()
|
||||
# execute_background is called (fire-and-forget) instead of execute_run via runner.run
|
||||
mock_orch_cls.execute_background.assert_called()
|
||||
mock_db.close.assert_called_once()
|
||||
# #endregion test_new_key_only_mode
|
||||
|
||||
@@ -133,7 +143,8 @@ def test_baseline_expired_fallback():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
@@ -199,7 +210,8 @@ def test_full_mode_default():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
@@ -215,7 +227,7 @@ def test_full_mode_default():
|
||||
assert mock_run.trigger_type == "scheduled", (
|
||||
f"Expected trigger_type='scheduled' for full mode, got '{mock_run.trigger_type}'"
|
||||
)
|
||||
mock_orch.execute_run.assert_called_once()
|
||||
mock_orch_cls.execute_background.assert_called()
|
||||
mock_db.close.assert_called_once()
|
||||
# #endregion test_full_mode_default
|
||||
|
||||
@@ -241,7 +253,8 @@ def test_inactive_schedule_skips():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
execute_scheduled_translation(
|
||||
schedule_id="sched-1",
|
||||
job_id="job-1",
|
||||
@@ -271,7 +284,8 @@ def test_schedule_not_found_skips():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
execute_scheduled_translation(
|
||||
schedule_id="sched-1",
|
||||
job_id="job-1",
|
||||
@@ -317,7 +331,8 @@ def test_baseline_expired_full_mode():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
@@ -346,7 +361,7 @@ def test_baseline_expired_full_mode():
|
||||
|
||||
# #region test_execution_error_handled [C:1] [TYPE Function]
|
||||
def test_execution_error_handled():
|
||||
"""execute_run raises → run is marked FAILED, schedule tracking updated, no crash."""
|
||||
"""execute_background is called even when execute_run would raise — scheduler dispatches safely."""
|
||||
from src.plugins.translate.scheduler import execute_scheduled_translation
|
||||
|
||||
mock_db = MagicMock()
|
||||
@@ -374,10 +389,10 @@ def test_execution_error_handled():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch.execute_run.side_effect = RuntimeError("LLM timeout")
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
|
||||
execute_scheduled_translation(
|
||||
@@ -388,18 +403,22 @@ def test_execution_error_handled():
|
||||
execution_mode="full",
|
||||
)
|
||||
|
||||
assert mock_run.status == "FAILED"
|
||||
assert "LLM timeout" in mock_run.error_message
|
||||
assert mock_run.completed_at is not None
|
||||
# Scheduler dispatches via execute_background — does NOT catch execution errors itself
|
||||
mock_orch_cls.execute_background.assert_called_once_with(
|
||||
run_id=mock_run.id,
|
||||
db_session_maker=mock_session_maker,
|
||||
config_manager=mock_config,
|
||||
current_user="scheduler",
|
||||
)
|
||||
# Schedule tracking is updated at dispatch time
|
||||
assert schedule.last_run_at is not None
|
||||
mock_db.commit.assert_called()
|
||||
mock_db.close.assert_called_once()
|
||||
# #endregion test_execution_error_handled
|
||||
|
||||
|
||||
# #region test_execution_run_status_failed_path [C:1] [TYPE Function]
|
||||
def test_execution_run_status_failed_path():
|
||||
"""execute_run succeeds but run.status == 'FAILED' → hits lines 388-392."""
|
||||
"""Scheduler dispatches regardless of run.status — status is handled by background task."""
|
||||
from src.plugins.translate.scheduler import execute_scheduled_translation
|
||||
|
||||
mock_db = MagicMock()
|
||||
@@ -423,11 +442,12 @@ def test_execution_run_status_failed_path():
|
||||
mock_db.query.side_effect = [q1, q2, q3]
|
||||
|
||||
mock_run = _make_mock_run(run_id="run-fail", job_id="job-fail")
|
||||
mock_run.status = "FAILED" # execute_run sets this
|
||||
mock_run.status = "FAILED" # simulate run already failed
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
@@ -440,16 +460,15 @@ def test_execution_run_status_failed_path():
|
||||
execution_mode="full",
|
||||
)
|
||||
|
||||
# The run.status == "FAILED" path sets insert_status = run.insert_status or None
|
||||
assert mock_run.insert_status is None # None or None -> None
|
||||
mock_orch.execute_run.assert_called_once()
|
||||
# Scheduler dispatches background task regardless of run.status
|
||||
mock_orch_cls.execute_background.assert_called()
|
||||
mock_db.close.assert_called_once()
|
||||
# #endregion test_execution_run_status_failed_path
|
||||
|
||||
|
||||
# #region test_execution_notification_error [C:1] [TYPE Function]
|
||||
def test_execution_notification_error():
|
||||
"""execute_run raises and notification send also raises → both caught (lines 413-415, 431-442)."""
|
||||
"""Scheduler dispatches safely — notification moves to background task."""
|
||||
from src.plugins.translate.scheduler import execute_scheduled_translation
|
||||
|
||||
mock_db = MagicMock()
|
||||
@@ -474,21 +493,12 @@ def test_execution_notification_error():
|
||||
|
||||
with patch(
|
||||
"src.plugins.translate.orchestrator.TranslationOrchestrator"
|
||||
) as mock_orch_cls, patch(
|
||||
"src.plugins.translate.scheduler.NotificationService"
|
||||
) as mock_notif_cls:
|
||||
) as mock_orch_cls, \
|
||||
patch("src.dependencies.get_async_job_runner", return_value=_make_coro_runner()):
|
||||
mock_orch = MagicMock()
|
||||
mock_orch.start_run.return_value = mock_run
|
||||
mock_orch.execute_run.side_effect = RuntimeError("LLM timeout")
|
||||
mock_orch_cls.return_value = mock_orch
|
||||
|
||||
# Make notification send also raise
|
||||
mock_provider = MagicMock()
|
||||
mock_provider.send.side_effect = Exception("Email down")
|
||||
mock_notif_svc = MagicMock()
|
||||
mock_notif_svc._providers = {"email": mock_provider}
|
||||
mock_notif_cls.return_value = mock_notif_svc
|
||||
|
||||
execute_scheduled_translation(
|
||||
schedule_id="sched-notif",
|
||||
job_id="job-notif",
|
||||
@@ -497,10 +507,10 @@ def test_execution_notification_error():
|
||||
execution_mode="full",
|
||||
)
|
||||
|
||||
assert mock_run.status == "FAILED"
|
||||
assert mock_run.completed_at is not None
|
||||
# Scheduler dispatches — notification is handled by execute_background
|
||||
mock_orch_cls.execute_background.assert_called()
|
||||
# Schedule tracking is updated at dispatch
|
||||
assert schedule.last_run_at is not None
|
||||
mock_db.commit.assert_called()
|
||||
mock_db.close.assert_called_once()
|
||||
# #endregion test_execution_notification_error
|
||||
|
||||
|
||||
Reference in New Issue
Block a user