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:
2026-07-07 21:04:23 +03:00
parent 0d09e24434
commit de023c7a31
17 changed files with 423 additions and 212 deletions

View File

@@ -42,6 +42,20 @@ def _make_dashboard(dash_id=1, title="Test Dash"):
return {"id": dash_id, "slug": f"slug-{dash_id}", "dashboard_title": title}
def _make_mock_superset_client():
"""Create a MagicMock that can be used as a SupersetClient (aclose is awaitable)."""
client = MagicMock()
client.aclose = AsyncMock()
return client
def _make_mock_mapping_service():
"""Create a MagicMock that can be used as IdMappingService (sync_environment is awaitable)."""
svc = MagicMock()
svc.sync_environment = AsyncMock()
return svc
class TestMigrationPluginProperties:
"""Verify static property values."""
@@ -115,10 +129,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip_content", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -128,8 +142,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -165,10 +178,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -178,8 +191,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -215,10 +227,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -228,8 +240,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -274,7 +285,7 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, []))
with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \
@@ -299,7 +310,7 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Other")]))
with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \
@@ -329,12 +340,12 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(side_effect=[
(b"zip_ok", "meta"), RuntimeError("Export failed for dash 2")
])
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -344,8 +355,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -374,10 +384,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -387,8 +397,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -423,11 +432,11 @@ class TestMigrationPluginExecute:
mock_task_manager.await_input = MagicMock()
mock_task_manager.wait_for_input = AsyncMock()
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
# First export succeeds, import fails with password error
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
# First import fails with password error, retry succeeds
mock_tgt_client.import_dashboard = AsyncMock(side_effect=[
RuntimeError("Must provide a password for the database 'PostgreSQL'"),
@@ -442,8 +451,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -474,10 +482,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -489,8 +497,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -516,10 +523,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -547,8 +554,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal', return_value=mock_db):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -580,7 +586,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC:
# Return None (falsy) for the source client
MockSC.side_effect = [None, MagicMock()]
MockSC.side_effect = [None, _make_mock_superset_client()]
with pytest.raises(ValueError, match="Clients not initialized"):
await plugin.execute({
@@ -602,10 +608,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -615,8 +621,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -647,7 +652,7 @@ class TestMigrationPluginExecute:
mock_task_manager.wait_for_resolution = AsyncMock()
# Use a single client mock that handles both get_dashboards and import_dashboard
mock_client = MagicMock()
mock_client = _make_mock_superset_client()
mock_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
mock_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_client.import_dashboard = AsyncMock()
@@ -675,8 +680,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal', return_value=mock_db):
mock_ctf.return_value.__enter__ = MagicMock(return_value="/tmp/test.zip")
@@ -715,10 +719,10 @@ class TestMigrationPluginExecute:
mock_task_manager.await_input = MagicMock()
mock_task_manager.wait_for_input = AsyncMock()
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, [_make_dashboard(1, "Dash")]))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock(side_effect=[
RuntimeError("Must provide a password for the database databases/PostgreSQL.yaml"),
None,
@@ -732,8 +736,7 @@ class TestMigrationPluginExecute:
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.dependencies.get_task_manager', return_value=mock_task_manager), \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=_make_mock_mapping_service()), \
patch('src.plugins.migration.SessionLocal'):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -763,10 +766,10 @@ class TestMigrationPluginExecute:
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = MagicMock()
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = MagicMock()
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
@@ -774,16 +777,13 @@ class TestMigrationPluginExecute:
mock_db_session = MagicMock()
mock_sync = MagicMock()
mock_sync.sync_environment = MagicMock(side_effect=RuntimeError("Sync failed"))
from src.core.mapping_service import IdMappingService
original_sync = IdMappingService.sync_environment
mock_sync.sync_environment = AsyncMock(side_effect=RuntimeError("Sync failed"))
with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService'), \
patch('src.plugins.migration.get_async_job_runner'), \
patch('src.plugins.migration.IdMappingService', return_value=mock_sync), \
patch('src.plugins.migration.SessionLocal', return_value=mock_db_session):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
@@ -797,5 +797,57 @@ class TestMigrationPluginExecute:
})
assert result["status"] == "SUCCESS"
assert mock_sync.sync_environment.called, "sync_environment should have been awaited"
# ── Regression: post-migration sync does not deadlock (direct await, not runner.run) ──
@pytest.mark.asyncio
async def test_execute_id_mapping_sync_completes_without_deadlock(self):
"""Regression: post-migration sync completes via direct await (not runner.run)."""
plugin = MigrationPlugin()
src_env = _make_env("env-1", "Source")
tgt_env = _make_env("env-2", "Target")
dashboards = [_make_dashboard(1, "Dash A"), _make_dashboard(2, "Dash B")]
mock_cm = MagicMock()
mock_cm.get_environments.return_value = [src_env, tgt_env]
mock_src_client = _make_mock_superset_client()
mock_src_client.get_dashboards = AsyncMock(return_value=(True, dashboards))
mock_src_client.export_dashboard = AsyncMock(return_value=(b"zip", "meta"))
mock_tgt_client = _make_mock_superset_client()
mock_tgt_client.import_dashboard = AsyncMock()
mock_engine = MagicMock()
mock_engine.transform_zip.return_value = True
mock_db_session = MagicMock()
mock_sync = MagicMock()
mock_sync.sync_environment = AsyncMock() # completes without error
with patch('src.plugins.migration.get_config_manager', return_value=mock_cm), \
patch('src.plugins.migration.SupersetClient') as MockSC, \
patch('src.plugins.migration.MigrationEngine', return_value=mock_engine), \
patch('src.plugins.migration.create_temp_file') as mock_ctf, \
patch('src.plugins.migration.IdMappingService', return_value=mock_sync), \
patch('src.plugins.migration.SessionLocal', return_value=mock_db_session):
MockSC.side_effect = [mock_src_client, mock_tgt_client]
mock_ctf.return_value.__enter__ = MagicMock(return_value="/tmp/test.zip")
result = await plugin.execute({
"source_env_id": "env-1",
"target_env_id": "env-2",
"selected_ids": [1, 2],
})
# Sync must complete and all dashboards must be migrated
assert result["status"] == "SUCCESS"
assert len(result["migrated_dashboards"]) == 2
assert result["failed_dashboards"] == []
# Verify sync was called with correct params
mock_sync.sync_environment.assert_called_once()
call_args = mock_sync.sync_environment.call_args
assert call_args[1]["incremental"] is True
# #endregion Test.MigrationPlugin