fix: change workflow solution to avoid missing last job trigger for analysis full update

Author: rebecamurillo

src/sum_impact_assessment/services/full_impact_refresh_service.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import asyncio
+import threading
+import time
 from datetime import datetime, timedelta
 from typing import Any, Dict, List
 
@@ -69,14 +71,11 @@ def build_initial_dispatch_state(plan: List[Dict[str, Any]]) -> List[Dict[str, A
 
 def dispatch_full_refresh_sync(parent_run_id: str, plan: List[Dict[str, Any]]) -> None:
     """
-    Run the async dispatcher from a sync entrypoint compatible with BackgroundTasks.
-    """
-    asyncio.run(dispatch_full_refresh(parent_run_id, plan))
-
+    Create child job runs for the configured plan and launch each execution in a dedicated thread.
 
-async def dispatch_full_refresh(parent_run_id: str, plan: List[Dict[str, Any]]) -> None:
-    """
-    Create child job runs for the configured plan and launch each execution asynchronously.
+    Threads are independent of the event loop lifetime, ensuring every dispatched job is started —
+    including the last one in the sequence (which was previously vulnerable to asyncio task
+    cancellation when the loop closed before the task had a turn to run).
     """
     dispatch_failures = 0
     dispatch_count = len(plan)
@@ -122,14 +121,20 @@ async def dispatch_full_refresh(parent_run_id: str, plan: List[Dict[str, Any]])
                         },
                     )
 
-                asyncio.create_task(
-                    asyncio.to_thread(
-                        execute_job_in_background,
-                        job_name,
-                        child_job_run_id,
-                        params,
-                    )
+                logger.info(
+                    "Child job thread spawned",
+                    extra={
+                        "parent_run_id": parent_run_id,
+                        "child_run_id": child_job_run_id,
+                        "job_name": plan_item["job_name"],
+                        "sequence": sequence,
+                    },
                 )
+                threading.Thread(
+                    target=execute_job_in_background,
+                    args=(job_name, child_job_run_id, params),
+                    daemon=False,
+                ).start()
             except Exception as error:
                 dispatch_failures += 1
                 logger.error(
@@ -154,7 +159,7 @@ async def dispatch_full_refresh(parent_run_id: str, plan: List[Dict[str, Any]])
                     )
 
             if index < dispatch_count - 1:
-                await asyncio.sleep(settings.REFRESH_DISPATCH_INTERVAL_SECONDS)
+                time.sleep(settings.REFRESH_DISPATCH_INTERVAL_SECONDS)
 
         with get_db_session() as db:
             job_repository = JobRepository(db)
@@ -187,6 +192,13 @@ async def dispatch_full_refresh(parent_run_id: str, plan: List[Dict[str, Any]])
             )
 
 
+async def dispatch_full_refresh(parent_run_id: str, plan: List[Dict[str, Any]]) -> None:
+    """
+    Async wrapper around dispatch_full_refresh_sync for use from async contexts.
+    """
+    await asyncio.to_thread(dispatch_full_refresh_sync, parent_run_id, plan)
+
+
 def build_trigger_response(plan: List[Dict[str, Any]], parent_run_id: str, started_at: datetime) -> FullImpactRefreshTriggerResponse:
     """
     Build the trigger response for the newly accepted refresh run.

tests/test_services/test_full_impact_refresh_service.py

@@ -41,18 +41,15 @@ def test_build_initial_dispatch_state_sets_pending_status():
     assert all(item["job_run_id"] is None for item in dispatch_state)
 
 
-@pytest.mark.anyio
-@patch("src.sum_impact_assessment.services.full_impact_refresh_service.asyncio.create_task")
-@patch("src.sum_impact_assessment.services.full_impact_refresh_service.asyncio.to_thread", new_callable=Mock)
-@patch("src.sum_impact_assessment.services.full_impact_refresh_service.asyncio.sleep")
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.threading.Thread")
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.time.sleep")
 @patch("src.sum_impact_assessment.services.full_impact_refresh_service.JobRepository")
 @patch("src.sum_impact_assessment.services.full_impact_refresh_service.get_db_session")
-async def test_dispatch_full_refresh_dispatches_all_children(
+def test_dispatch_full_refresh_sync_dispatches_all_children(
     mock_get_db_session,
     mock_job_repository_class,
     mock_sleep,
-    mock_to_thread,
-    mock_create_task,
+    mock_thread_class,
 ):
     """Dispatching the full refresh should create all child job runs and mark the parent successful."""
     started_at = datetime(2026, 5, 12, 10, 0, 0)
@@ -66,14 +63,13 @@ async def test_dispatch_full_refresh_dispatches_all_children(
     repo_instance = Mock()
     repo_instance.create_job_run.side_effect = child_job_runs
     mock_job_repository_class.return_value = repo_instance
-    mock_to_thread.return_value = Mock()
 
     db_context = Mock()
     db_context.__enter__ = Mock(return_value=Mock())
     db_context.__exit__ = Mock(return_value=None)
     mock_get_db_session.return_value = db_context
 
-    await dispatch_full_refresh("parent-1", plan)
+    dispatch_full_refresh_sync("parent-1", plan)
 
     assert repo_instance.create_job_run.call_count == 7
     assert repo_instance.update_dispatched_job.call_count == 7
@@ -84,27 +80,69 @@ async def test_dispatch_full_refresh_dispatches_all_children(
     )
     final_call = repo_instance.update_job_status.call_args_list[-1]
     assert final_call.kwargs["status"] == JobStatusEnum.SUCCESS
-    assert mock_create_task.call_count == 7
-    assert mock_sleep.await_count == 6
-    first_dispatch = mock_to_thread.call_args_list[0]
-    assert first_dispatch.args[2] == "child-0"
+    assert mock_thread_class.call_count == 7
+    assert mock_thread_class.return_value.start.call_count == 7
+    assert mock_sleep.call_count == 6  # no trailing sleep after the last job
+    first_thread_call = mock_thread_class.call_args_list[0]
+    assert first_thread_call.kwargs["args"][1] == "child-0"
     first_update = repo_instance.update_dispatched_job.call_args_list[0]
     assert first_update.args[2]["job_run_id"] == "child-0"
     assert first_update.args[2]["scheduled_at"] == "2026-05-12T10:00:00"
 
 
-@patch("src.sum_impact_assessment.services.full_impact_refresh_service.asyncio.run")
-@patch("src.sum_impact_assessment.services.full_impact_refresh_service.dispatch_full_refresh", new_callable=Mock)
-def test_dispatch_full_refresh_sync_uses_asyncio_run(
-    mock_dispatch_full_refresh,
-    mock_asyncio_run,
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.threading.Thread")
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.time.sleep")
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.JobRepository")
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.get_db_session")
+def test_dispatch_full_refresh_sync_all_seven_children_are_started(
+    mock_get_db_session,
+    mock_job_repository_class,
+    mock_sleep,
+    mock_thread_class,
 ):
-    """The sync wrapper should delegate the coroutine through asyncio.run."""
-    plan = [{"job_name": "kpi_measures_analysis"}]
-    mock_coroutine = object()
-    mock_dispatch_full_refresh.return_value = mock_coroutine
+    """Regression: all 7 configured jobs must be passed to a started Thread — including the last one.
 
-    dispatch_full_refresh_sync("parent-1", plan)
+    Previously the last child (mcda_analysis_qualitative_nsm_providers) was never executed because
+    asyncio.create_task was used without awaiting, and asyncio.run() closed the loop before the
+    task had a chance to run in Docker environments with CPU contention.
+    """
+    started_at = datetime(2026, 5, 12, 10, 0, 0)
+    plan = build_dispatch_plan(started_at)
+
+    child_job_runs = [
+        Mock(id=f"child-{i}", created_at=datetime(2026, 5, 12, 10, 0, i))
+        for i in range(len(plan))
+    ]
+
+    repo_instance = Mock()
+    repo_instance.create_job_run.side_effect = child_job_runs
+    mock_job_repository_class.return_value = repo_instance
+
+    db_context = Mock()
+    db_context.__enter__ = Mock(return_value=Mock())
+    db_context.__exit__ = Mock(return_value=None)
+    mock_get_db_session.return_value = db_context
+
+    dispatch_full_refresh_sync("parent-reg", plan)
+
+    thread_child_ids = [
+        call.kwargs["args"][1]
+        for call in mock_thread_class.call_args_list
+    ]
+    expected_child_ids = [f"child-{i}" for i in range(7)]
+    assert thread_child_ids == expected_child_ids, (
+        f"Expected all 7 children to be started, got: {thread_child_ids}"
+    )
+
+
+@pytest.mark.anyio
+@patch("src.sum_impact_assessment.services.full_impact_refresh_service.dispatch_full_refresh_sync")
+async def test_dispatch_full_refresh_delegates_to_sync(
+    mock_dispatch_full_refresh_sync,
+):
+    """The async wrapper should delegate to dispatch_full_refresh_sync."""
+    plan = build_dispatch_plan(datetime(2026, 5, 12, 10, 0, 0))
+
+    await dispatch_full_refresh("parent-1", plan)
 
-    mock_dispatch_full_refresh.assert_called_once_with("parent-1", plan)
-    mock_asyncio_run.assert_called_once_with(mock_coroutine)
+    mock_dispatch_full_refresh_sync.assert_called_once_with("parent-1", plan)