Gracefully shut down st2 engines (workflow_engine, scheduler, rulesengine, actionrunner) on RabbitMQ failure to prevent stuck running tasks

.gitignore

@@ -3,6 +3,11 @@
 *.log
 *.orig
 .stamp*
+.agents/
+.clinerules/
+.agents
+.clinerules
+
 
 # C extensions
 *.so

CHANGELOG.rst

@@ -23,6 +23,7 @@ Fixed
 ~~~~~
 * Fix ``TypeError`` when displaying help for actions whose parameters have no ``description`` key. #6375
 * Fix utf-8 encode before checking paramter max size #6352
+* Fix stuck running workflow tasks #6398 (by @guzzijones12@gmail.com)
 
 Changed
 ~~~~~~~

conf/st2.conf.sample

@@ -229,6 +229,14 @@ cluster_urls =  # comma separated list allowed here.
 compression = None
 # How many times should we retry connection before failing.
 connection_retries = 10
+# Maximum retry interval in seconds for broker connection attempts.
+connection_retry_interval_max = 30
+# Starting retry interval in seconds for broker connection attempts.
+connection_retry_interval_start = 1
+# Increment for retry interval after each attempt (seconds).
+connection_retry_interval_step = 1
+# Maximum number of retry attempts for broker connection and reconnection. This prevents infinite retry loops when the broker is unavailable. Applies to both initial connection and reconnection during message publishing. Set to 0 to retry indefinitely (not recommended).
+connection_retry_max_attempts = 10
 # How long should we wait between connection retries.
 connection_retry_wait = 10000
 # Login method to use (AMQPLAIN, PLAIN, EXTERNAL, etc.).

contrib/runners/local_runner/tests/integration/BUILD

@@ -5,4 +5,11 @@ __defaults__(
 
 python_tests(
     name="tests",
-)
+    overrides={
+        "test_localrunner.py": dict(
+            stevedore_namespaces=[
+                "st2common.metrics.driver",
+            ],
+        ),
+    },
+)

contrib/runners/orquesta_runner/tests/unit/test_error_handling.py

@@ -268,10 +268,16 @@ def test_fail_start_task_action(self):
         lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta["name"])
         lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
 
-        # Assert action execution for task is not started and workflow failed.
+        # Manually trigger workflow execution to start tasks (simulates async workflow engine).
         wf_ex_db = wf_db_access.WorkflowExecution.query(
             action_execution=str(ac_ex_db.id)
         )[0]
+        wf_svc.request_next_tasks(wf_ex_db)
+
+        # Refresh workflow execution after task processing.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+
+        # Assert action execution for task is not started and workflow failed.
         tk_ex_dbs = wf_db_access.TaskExecution.query(
             workflow_execution=str(wf_ex_db.id)
         )
@@ -311,10 +317,16 @@ def test_fail_start_task_input_expr_eval(self):
         lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta["name"])
         lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
 
-        # Assert action execution for task is not started and workflow failed.
+        # Manually trigger workflow execution to start tasks (simulates async workflow engine).
         wf_ex_db = wf_db_access.WorkflowExecution.query(
             action_execution=str(ac_ex_db.id)
         )[0]
+        wf_svc.request_next_tasks(wf_ex_db)
+
+        # Refresh workflow execution after task processing.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+
+        # Assert action execution for task is not started and workflow failed.
         tk_ex_dbs = wf_db_access.TaskExecution.query(
             workflow_execution=str(wf_ex_db.id)
         )
@@ -351,10 +363,16 @@ def test_fail_start_task_input_value_type(self):
         )
         lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
 
-        # Assert workflow and task executions failed.
+        # Manually trigger workflow execution to start tasks (simulates async workflow engine).
         wf_ex_db = wf_db_access.WorkflowExecution.query(
             action_execution=str(ac_ex_db.id)
         )[0]
+        wf_svc.request_next_tasks(wf_ex_db)
+
+        # Refresh workflow execution after task processing.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+
+        # Assert workflow and task executions failed.
         self.assertEqual(wf_ex_db.status, wf_statuses.FAILED)
         self.assertListEqual(
             self.sort_workflow_errors(wf_ex_db.errors), expected_errors

contrib/runners/orquesta_runner/tests/unit/test_pause_and_resume.py

@@ -506,6 +506,14 @@ def test_resume(self):
 
         # Resume the workflow.
         lv_ac_db, ac_ex_db = ac_svc.request_resume(lv_ac_db, cfg.CONF.system_user.user)
+
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(wf_ex_dbs[0])
+
+        # Refresh to get updated status after workflow processing.
         lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
         self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
         wf_ex_dbs = wf_db_access.WorkflowExecution.query(
@@ -593,9 +601,24 @@ def test_resume_cascade_to_subworkflow(self):
 
         # Resume the main workflow and assert it is running.
         lv_ac_db, ac_ex_db = ac_svc.request_resume(lv_ac_db, cfg.CONF.system_user.user)
+
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(wf_ex_dbs[0])
+
+        # Refresh to get updated status after workflow processing.
         lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
         self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
 
+        # Resume cascades to subworkflow, so we need to trigger its processing too.
+        tk_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(tk_ac_ex_db.id))
+        sub_wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(tk_ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(sub_wf_ex_dbs[0])
+
         # Assert the subworkflow is running.
         tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk_lv_ac_db.id))
         self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
@@ -713,6 +736,14 @@ def test_resume_from_each_subworkflow_when_parent_is_paused(self):
         t1_lv_ac_db, t1_ac_ex_db = ac_svc.request_resume(
             t1_lv_ac_db, cfg.CONF.system_user.user
         )
+
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        t1_wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(t1_ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(t1_wf_ex_dbs[0])
+
+        # Refresh to get updated status after workflow processing.
         t1_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(t1_lv_ac_db.id))
         self.assertEqual(t1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
 
@@ -863,6 +894,14 @@ def test_resume_from_subworkflow_when_parent_is_paused(self):
         t1_lv_ac_db, t1_ac_ex_db = ac_svc.request_resume(
             t1_lv_ac_db, cfg.CONF.system_user.user
         )
+
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        t1_wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(t1_ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(t1_wf_ex_dbs[0])
+
+        # Refresh to get updated status after workflow processing.
         t1_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(t1_lv_ac_db.id))
         self.assertEqual(t1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
 
@@ -993,6 +1032,14 @@ def test_resume_from_subworkflow_when_parent_is_running(self):
         t1_lv_ac_db, t1_ac_ex_db = ac_svc.request_resume(
             t1_lv_ac_db, cfg.CONF.system_user.user
         )
+
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        t1_wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(t1_ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(t1_wf_ex_dbs[0])
+
+        # Refresh to get updated status after workflow processing.
         t1_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(t1_lv_ac_db.id))
         self.assertEqual(t1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
 

contrib/runners/orquesta_runner/tests/unit/test_with_items.py

@@ -223,6 +223,16 @@ def test_with_items_empty_list(self):
         )
         lv_ac_db, ac_ex_db = action_service.request(lv_ac_db)
 
+        # Manually trigger workflow execution processing for empty items case.
+        # With empty items, the workflow needs explicit processing to complete.
+        from st2common.services import workflows as wf_svc
+
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(ac_ex_db.id)
+        )
+        if wf_ex_dbs:
+            wf_svc.request_next_tasks(wf_ex_dbs[0])
+
         # Wait for the liveaction to complete.
         lv_ac_db = self._wait_on_status(
             lv_ac_db, action_constants.LIVEACTION_STATUS_SUCCEEDED
@@ -627,6 +637,14 @@ def test_with_items_concurrency_pause_and_resume(self):
         lv_ac_db, ac_ex_db = action_service.request_resume(lv_ac_db, requester)
         self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_RESUMING)
 
+        # Manually trigger workflow execution processing (simulates async workflow engine).
+        from st2common.services import workflows as wf_svc
+
+        wf_ex_dbs = wf_db_access.WorkflowExecution.query(
+            action_execution=str(ac_ex_db.id)
+        )
+        wf_svc.request_next_tasks(wf_ex_dbs[0])
+
         # Check that the workflow execution is running.
         lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
         self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_RUNNING)

st2actions/st2actions/cmd/scheduler.py

@@ -91,12 +91,58 @@ def _run_scheduler():
             "(PID=%s) Scheduler unable to populate action_execution_id.", os.getpid()
         )
 
+    # Bootstrap missing scheduling queue entries for requested LiveActions.
+    # This handles recovery from RabbitMQ failures where messages were never consumed.
+    handler._bootstrap_missing_scheduling_queue_items()
+
     try:
         handler.start()
         entrypoint.start()
 
-        # Wait on handler first since entrypoint is more durable.
-        handler.wait() or entrypoint.wait()
+        # Wait on both handler and entrypoint. If either fails, we want to shut down gracefully.
+        # Poll the threads to detect when any of them fails
+        import eventlet
+
+        threads_to_monitor = [
+            (handler._main_thread, "handler_main"),
+            (handler._cleanup_thread, "handler_cleanup"),
+            (entrypoint._consumer_thread, "entrypoint_consumer"),
+        ]
+
+        try:
+            # Poll threads in a loop - check if any has died/failed
+            while True:
+                dead_threads = [
+                    (thread, name) for thread, name in threads_to_monitor if thread.dead
+                ]
+
+                if dead_threads:
+                    # If any dead thread raised an exception, propagate it. We must
+                    # check *all* dead threads (not just the first one observed) because
+                    # a failing sibling thread can trigger a linked shutdown that causes
+                    # other threads to exit cleanly in the same scheduling tick. Returning
+                    # success based on the first-seen clean exit would swallow the real
+                    # failure.
+                    for thread, name in dead_threads:
+                        try:
+                            thread.wait()  # Raises if the thread raised.
+                        except Exception as e:
+                            LOG.error("Thread %s failed: %s", name, e)
+                            # Re-raise to let outer exception handler deal with shutdown
+                            raise
+
+                    # No exceptions - all dead threads exited cleanly (shouldn't
+                    # happen in normal operation).
+                    for _, name in dead_threads:
+                        LOG.info("Thread %s completed", name)
+                    return 0
+
+                # Sleep briefly to avoid tight loop and allow other greenlets to run
+                eventlet.sleep(0.1)
+        except Exception as e:
+            # If we caught an exception, it's already been logged and components shut down
+            # Re-raise it so tests and monitoring can detect the failure
+            raise e
     except (KeyboardInterrupt, SystemExit):
         LOG.info("(PID=%s) Scheduler stopped.", os.getpid())
 
@@ -121,7 +167,7 @@ def _run_scheduler():
         except:
             LOG.exception("Unable to shutdown scheduler.")
 
-        return 1
+        raise
 
     return 0
 

st2actions/st2actions/cmd/workflow_engine.py

@@ -73,10 +73,12 @@ def run_server():
         LOG.info("(PID=%s) Workflow engine stopped.", os.getpid())
         deregister_service(service=workflows.WORKFLOW_ENGINE)
         engine.shutdown()
+        return 0
     except:
         LOG.exception("(PID=%s) Workflow engine unexpectedly stopped.", os.getpid())
+        deregister_service(service=workflows.WORKFLOW_ENGINE)
+        engine.shutdown()
         return 1
-    return 0
 
 
 def teardown():