From 66008c1f4d169d9c587e212590aaf06dd26633e1 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 11 May 2026 14:59:47 -0400 Subject: [PATCH 01/24] rollback on rabbit message failure update base and workflows do not catch base exception when closing channel. if there is a channel it should close parse user keys correctly in api model unused variable fix abandoned test fix orquesta testing race fix fix a couple more tests that needed to mock the workfow service request.next_tasks --- .gitignore | 5 + .../tests/unit/test_error_handling.py | 24 +- .../tests/unit/test_pause_and_resume.py | 47 ++++ .../tests/unit/test_with_items.py | 18 ++ st2actions/st2actions/container/base.py | 106 ++++++--- st2actions/st2actions/scheduler/handler.py | 8 + st2actions/st2actions/worker.py | 39 ++- .../tests/unit/policies/test_concurrency.py | 12 +- .../unit/policies/test_concurrency_by_attr.py | 12 +- .../unit/test_kombu_error_propagation.py | 225 ++++++++++++++++++ st2actions/tests/unit/test_worker.py | 141 ++++++----- st2api/st2api/controllers/v1/keyvalue.py | 2 - st2common/st2common/models/api/keyvalue.py | 18 +- st2common/st2common/persistence/base.py | 141 +++++++---- st2common/st2common/services/action.py | 2 +- .../transport/connection_retry_wrapper.py | 127 +++++----- .../unit/test_connection_retry_wrapper.py | 33 +-- .../tests/unit/test_persistence_rollback.py | 124 ++++++++++ 18 files changed, 837 insertions(+), 247 deletions(-) create mode 100644 st2actions/tests/unit/test_kombu_error_propagation.py create mode 100644 st2common/tests/unit/test_persistence_rollback.py diff --git a/.gitignore b/.gitignore index dc1b6aec20..94f3ade95f 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,11 @@ *.log *.orig .stamp* +.agents/ +.clinerules/ +.agents +.clinerules + # C extensions *.so diff --git a/contrib/runners/orquesta_runner/tests/unit/test_error_handling.py b/contrib/runners/orquesta_runner/tests/unit/test_error_handling.py index f6fcf8977b..77a73d9afb 100644 --- a/contrib/runners/orquesta_runner/tests/unit/test_error_handling.py +++ b/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 diff --git a/contrib/runners/orquesta_runner/tests/unit/test_pause_and_resume.py b/contrib/runners/orquesta_runner/tests/unit/test_pause_and_resume.py index bef405cb1c..49be139e69 100644 --- a/contrib/runners/orquesta_runner/tests/unit/test_pause_and_resume.py +++ b/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) diff --git a/contrib/runners/orquesta_runner/tests/unit/test_with_items.py b/contrib/runners/orquesta_runner/tests/unit/test_with_items.py index de9b0bea07..072a9bdfae 100644 --- a/contrib/runners/orquesta_runner/tests/unit/test_with_items.py +++ b/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) diff --git a/st2actions/st2actions/container/base.py b/st2actions/st2actions/container/base.py index 77317fa9e6..c36dcbb2e4 100644 --- a/st2actions/st2actions/container/base.py +++ b/st2actions/st2actions/container/base.py @@ -344,56 +344,90 @@ def _update_live_action_db(self, liveaction_id, status, result, context): return (liveaction_db, state_changed) def _update_status(self, liveaction_id, status, result, context): - with Timer(key="action.executions.update_liveaction_db"): + # NOTE: The next two operations take a very long time in master with large executions + # (long standing issue), but because start_timestamp and end_timestamp measure how long + # it took for the runner to run the action, it doesn't include the time it took to + # actually write results / persist execution into the database - that's a problem + # because we have no good direct visibility into that. + # + # The UX user would experience is - they would run an action which produces large + # result, CLI / API would show execution as running for a long time (until it's + # persisted in the database), but when it will finally be written to the database, + # duration will be shown as a short time, because it's measured based on start and + # end timestamp. + # + # This mean we can have, for example, Python runner action which returns a lot of data + # and takes only 0.5 second to finish, but next two database operations can easily take + # 10 seconds each. + # + # To work around that and provide some additional visibility into that to the operators + # and users, we update "end_timestamp" on each object again after both of them have + # already been written. That atomic single field update is very fast and adds no + # additional overhead. + + # Get the current liveaction from DB + liveaction_db = get_liveaction_by_id(liveaction_id) + + # Determine if state changed (for publishing decision) + state_changed = ( + liveaction_db.status != status + and liveaction_db.status not in action_constants.LIVEACTION_COMPLETED_STATES + ) + + # Prepare end_timestamp if action is completing + if status in action_constants.LIVEACTION_COMPLETED_STATES: + end_timestamp = date_utils.get_datetime_utc_now() + else: + end_timestamp = None + + # Update liveaction object in memory (not persisted yet) + liveaction_db.status = status + liveaction_db.result = result + if context: + liveaction_db.context.update(context) + if end_timestamp: + liveaction_db.end_timestamp = end_timestamp + + # FIRST: Update ActionExecution DB + publish to RabbitMQ + # If this fails with KombuError, exception propagates before LiveAction is persisted + # This prevents inconsistent state where liveaction shows succeeded but execution wasn't updated + with Timer(key="action.executions.update_execution_db"): try: - # NOTE: The next two operations take a very long time in master with large executions - # (long standing issue), but because start_timestamp and end_timestamp measure how long - # it took for the runner to run the action, it doesn't include the time it took to - # actually write results / persist execution into the database - that's a problem - # because we have no good direct visibility into that. - # - # The UX user would experience is - they would run an action which produces large - # result, CLI / API would show execution as running for a long time (until it's - # persisted in the database), but when it will finally be written to the database, - # duration will be shown as a short time, because it's measured based on start and - # end timestamp. - # - # This mean we can have, for example, Python runner action which returns a lot of data - # and takes only 0.5 second to finish, but next two database operations can easily take - # 10 seconds each. - # - # To work around that and provide some additional visibility into that to the operators - # and users, we update "end_timestamp" on each object again after both of them have - # already been written. That atomic single field update is very fast and adds no - # additional overhead. - LOG.debug( - "Setting status: %s for liveaction: %s", status, liveaction_id - ) - liveaction_db, state_changed = self._update_live_action_db( - liveaction_id, status, result, context + executions.update_execution( + liveaction_db, + publish=state_changed, + set_result_size=True, ) + extra = {"liveaction_db": liveaction_db} + LOG.debug("Updated action execution", extra=extra) except Exception as e: LOG.exception( - "Cannot update liveaction " - "(id: %s, status: %s, result: %s)." + "Cannot update action execution for liveaction " + "(id: %s, status: %s, result: %s). " + "LiveAction will not be updated to prevent inconsistent state." % (liveaction_id, status, result) ) raise e # live_action_written_to_db_dt = date_utils.get_datetime_utc_now() - with Timer(key="action.executions.update_execution_db"): + # SECOND: Only if execution update succeeded, persist LiveAction to DB + # We only update if state actually changed to avoid unnecessary writes + with Timer(key="action.executions.update_liveaction_db"): try: - executions.update_execution( - liveaction_db, - publish=state_changed, - set_result_size=True, + LOG.debug( + "Setting status: %s for liveaction: %s", status, liveaction_id + ) + liveaction_db = update_liveaction_status( + status=status if state_changed else liveaction_db.status, + result=result, + context=context, + end_timestamp=end_timestamp, + liveaction_db=liveaction_db, ) - extra = {"liveaction_db": liveaction_db} - LOG.debug("Updated liveaction after run", extra=extra) except Exception as e: LOG.exception( - "Cannot update action execution for liveaction " + "Cannot update liveaction " "(id: %s, status: %s, result: %s)." % (liveaction_id, status, result) ) diff --git a/st2actions/st2actions/scheduler/handler.py b/st2actions/st2actions/scheduler/handler.py index abe10d91f3..5d080fbafa 100644 --- a/st2actions/st2actions/scheduler/handler.py +++ b/st2actions/st2actions/scheduler/handler.py @@ -374,6 +374,14 @@ def _regulate_and_schedule(self, liveaction_db, execution_queue_item_db): return + # Complete cancellation transition: CANCELING → CANCELED + if liveaction_db.status == action_constants.LIVEACTION_STATUS_CANCELING: + liveaction_db = action_service.update_status( + liveaction_db, + action_constants.LIVEACTION_STATUS_CANCELED, + publish=True, + ) + if ( liveaction_db.status in action_constants.LIVEACTION_COMPLETED_STATES or liveaction_db.status in action_constants.LIVEACTION_CANCEL_STATES diff --git a/st2actions/st2actions/worker.py b/st2actions/st2actions/worker.py index b1d3fc790e..23c3c66ca2 100644 --- a/st2actions/st2actions/worker.py +++ b/st2actions/st2actions/worker.py @@ -17,6 +17,8 @@ import sys import traceback +from amqp import exceptions as amqp_exceptions +from kombu import exceptions as kombu_exceptions from tooz.coordination import GroupNotCreated from oslo_config import cfg @@ -27,6 +29,7 @@ from st2common.exceptions.db import StackStormDBObjectNotFoundError from st2common.models.db.liveaction import LiveActionDB from st2common.persistence.execution import ActionExecution +from st2common.persistence.liveaction import LiveAction from st2common.services import coordination from st2common.services import executions from st2common.services import workflows as wf_svc @@ -176,16 +179,46 @@ def _run_action(self, liveaction_db): # stamp liveaction with process_info runner_info = system_info.get_process_info() - # Update liveaction status to "running" + # Capture the previous status for potential rollback + previous_status = liveaction_db.status + + # Update liveaction status to "running" first (without publish to queue) + # This prevents the job from being re-dispatched if ActionExecution update fails liveaction_db = action_utils.update_liveaction_status( status=action_constants.LIVEACTION_STATUS_RUNNING, runner_info=runner_info, liveaction_id=liveaction_db.id, + publish=False, # Don't publish yet - wait until ActionExecution succeeds ) - self._running_liveactions.add(liveaction_db.id) - action_execution_db = executions.update_execution(liveaction_db) + try: + # Update ActionExecution to match LiveAction + # If this fails with KombuError or AMQPError, the persistence layer will handle + # ActionExecution rollback, but we also need to rollback LiveAction + action_execution_db = executions.update_execution(liveaction_db) + + # Both updates succeeded - now publish LiveAction status to the queue + # This is the final step that makes the status change visible to the system + LiveAction.publish_status(liveaction_db) + + except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + # KombuError or AMQPError during ActionExecution update or LiveAction publish + # Rollback LiveAction to prevent orphaned "running" status + LOG.warning( + "AMQP/Kombu error occurred during execution update for liveaction %s. " + "Rolling back LiveAction status from 'running' to '%s'.", + liveaction_db.id, + previous_status, + ) + # Restore previous status without publishing (to avoid another KombuError) + action_utils.update_liveaction_status( + status=previous_status, + liveaction_id=liveaction_db.id, + publish=False, + ) + # Re-raise to trigger process exit for K8s restart + raise # Launch action extra = { diff --git a/st2actions/tests/unit/policies/test_concurrency.py b/st2actions/tests/unit/policies/test_concurrency.py index 7d92edbdbb..57647b238a 100644 --- a/st2actions/tests/unit/policies/test_concurrency.py +++ b/st2actions/tests/unit/policies/test_concurrency.py @@ -367,7 +367,17 @@ def test_on_cancellation(self): # Cancel execution. action_service.request_cancellation(scheduled[0], "stanley") - expected_num_pubs += 2 # Tally the canceling and canceled states. + + # Verify the action was actually cancelled. + cancelled_action = LiveAction.get_by_id(str(scheduled[0].id)) + self.assertEqual( + cancelled_action.status, action_constants.LIVEACTION_STATUS_CANCELED + ) + + # Since the action has no parent workflow context and is in RUNNING state, + # request_cancellation transitions directly to CANCELED (skipping CANCELING state). + # This results in only 1 state publication instead of 2. + expected_num_pubs += 1 # Tally the canceled state. self.assertEqual( expected_num_pubs, LiveActionPublisher.publish_state.call_count ) diff --git a/st2actions/tests/unit/policies/test_concurrency_by_attr.py b/st2actions/tests/unit/policies/test_concurrency_by_attr.py index 2edcdb4af7..cdccbb2e22 100644 --- a/st2actions/tests/unit/policies/test_concurrency_by_attr.py +++ b/st2actions/tests/unit/policies/test_concurrency_by_attr.py @@ -396,7 +396,17 @@ def test_on_cancellation(self): # Cancel execution. action_service.request_cancellation(scheduled[0], "stanley") - expected_num_pubs += 2 # Tally the canceling and canceled states. + + # Verify the action was actually cancelled. + cancelled_action = LiveAction.get_by_id(str(scheduled[0].id)) + self.assertEqual( + cancelled_action.status, action_constants.LIVEACTION_STATUS_CANCELED + ) + + # Since the action has no parent workflow context and is in RUNNING state, + # request_cancellation transitions directly to CANCELED (skipping CANCELING state). + # This results in only 1 state publication instead of 2. + expected_num_pubs += 1 # Tally the canceled state. self.assertEqual( expected_num_pubs, LiveActionPublisher.publish_state.call_count ) diff --git a/st2actions/tests/unit/test_kombu_error_propagation.py b/st2actions/tests/unit/test_kombu_error_propagation.py new file mode 100644 index 0000000000..45e2c590ee --- /dev/null +++ b/st2actions/tests/unit/test_kombu_error_propagation.py @@ -0,0 +1,225 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Test that verifies KombuError exceptions from persistence layer propagate +correctly through the action runner, causing process exit for K8s restart. +""" + +from __future__ import absolute_import + +import mock +from oslo_config import cfg +from kombu import exceptions as kombu_exceptions + +from st2tests.base import DbTestCase +import st2tests.config as tests_config +from st2common.constants import action as action_constants +from st2common.models.db.liveaction import LiveActionDB +from st2common.models.system.common import ResourceReference +from st2common.persistence.execution import ActionExecution +from st2common.persistence.liveaction import LiveAction +from st2common.services import executions +from st2common.util import date as date_utils +from st2common.bootstrap import runnersregistrar as runners_registrar +from st2tests.fixtures.generic.fixture import PACK_NAME as FIXTURES_PACK +from st2tests.fixturesloader import FixturesLoader +import st2actions.worker as actions_worker + + +TEST_FIXTURES = {"actions": ["local.yaml"]} + + +class KombuErrorPropagationTestCase(DbTestCase): + """ + Test case to verify that KombuError exceptions from the persistence layer + propagate correctly to cause action runner process exit. + """ + + fixtures_loader = FixturesLoader() + + @classmethod + def setUpClass(cls): + super(KombuErrorPropagationTestCase, cls).setUpClass() + runners_registrar.register_runners() + + models = cls.fixtures_loader.save_fixtures_to_db( + fixtures_pack=FIXTURES_PACK, fixtures_dict=TEST_FIXTURES + ) + cls.local_action_db = models["actions"]["local.yaml"] + + def setUp(self): + super(KombuErrorPropagationTestCase, self).setUp() + tests_config.reset() + tests_config.parse_args() + + def _get_liveaction_model(self, action_db, params): + """Helper to create a LiveAction model for testing.""" + status = action_constants.LIVEACTION_STATUS_REQUESTED + start_timestamp = date_utils.get_datetime_utc_now() + action_ref = ResourceReference(name=action_db.name, pack=action_db.pack).ref + parameters = params + context = {"user": cfg.CONF.system_user.user} + liveaction_db = LiveActionDB( + status=status, + start_timestamp=start_timestamp, + action=action_ref, + parameters=parameters, + context=context, + ) + return liveaction_db + + def test_kombu_error_in_execution_update_propagates(self): + """ + Test that when ActionExecution.update() raises KombuError during + execution update, the exception propagates up through the worker to cause + process exit. + + This test also verifies that the persistence layer's built-in rollback + mechanism (in base.py) properly restores the ActionExecution to its + previous state when KombuError occurs during publish/dispatch. + + This ensures K8s can detect the failure and restart the action runner + to reconnect to RabbitMQ without leaving orphaned "running" records. + """ + action_worker = actions_worker.get_worker() + + # Create a liveaction + params = {"cmd": "echo 'test'"} + liveaction_db = self._get_liveaction_model(self.local_action_db, params) + liveaction_db = LiveAction.add_or_update(liveaction_db) + + # Create initial execution object (this will succeed) + executions.create_execution_object(liveaction_db) + + # Mock ActionExecution.update to raise KombuError on first call only + # This simulates the scenario where: + # 1. LiveAction update to "running" succeeds (in database) + # 2. ActionExecution.update fails with KombuError + # 3. Worker attempts rollback of LiveAction + # We need to ensure only the ActionExecution.update fails, not the rollback + original_update = ActionExecution.update + first_call = [True] + + def mock_update_first_call_only(model_object, **kwargs): + if first_call[0]: + first_call[0] = False + raise kombu_exceptions.KombuError("RabbitMQ connection failed") + return original_update(model_object, **kwargs) + + with mock.patch.object( + ActionExecution, + "update", + side_effect=mock_update_first_call_only, + ): + # Attempt to run the action - this should raise KombuError + with self.assertRaises(kombu_exceptions.KombuError) as cm: + action_worker._run_action(liveaction_db) + + # Verify the exception message + self.assertIn("RabbitMQ connection failed", str(cm.exception)) + + # Verify that both ActionExecution and LiveAction were rolled back + # The worker now implements a transaction-like pattern where: + # 1. LiveAction is updated to "running" without publish + # 2. ActionExecution is updated (with built-in rollback in base.py) + # 3. If step 2 fails with KombuError, LiveAction is also rolled back + # This prevents orphaned "running" LiveActions that could be re-dispatched + updated_liveaction = LiveAction.get_by_id(liveaction_db.id) + self.assertEqual( + updated_liveaction.status, action_constants.LIVEACTION_STATUS_REQUESTED + ) + + def test_kombu_connection_error_propagates(self): + """ + Test that ConnectionError (a subclass of KombuError) also propagates correctly + and that the persistence layer's built-in rollback works for this exception type. + """ + action_worker = actions_worker.get_worker() + + params = {"cmd": "echo 'test'"} + liveaction_db = self._get_liveaction_model(self.local_action_db, params) + liveaction_db = LiveAction.add_or_update(liveaction_db) + + # Create initial execution object + executions.create_execution_object(liveaction_db) + + # Mock to raise ConnectionError on first call only + original_update = ActionExecution.update + first_call = [True] + + def mock_update_first_call_only(model_object, **kwargs): + if first_call[0]: + first_call[0] = False + raise kombu_exceptions.ConnectionError("Connection lost") + return original_update(model_object, **kwargs) + + with mock.patch.object( + ActionExecution, + "update", + side_effect=mock_update_first_call_only, + ): + # Verify the exception propagates + with self.assertRaises(kombu_exceptions.ConnectionError) as cm: + action_worker._run_action(liveaction_db) + + self.assertIn("Connection lost", str(cm.exception)) + + # Verify that the transaction-like rollback worked + updated_liveaction = LiveAction.get_by_id(liveaction_db.id) + self.assertEqual( + updated_liveaction.status, action_constants.LIVEACTION_STATUS_REQUESTED + ) + + def test_kombu_operational_error_propagates(self): + """ + Test that OperationalError (another subclass of KombuError) also propagates + and that the persistence layer's built-in rollback works for this exception type. + """ + action_worker = actions_worker.get_worker() + + params = {"cmd": "echo 'test'"} + liveaction_db = self._get_liveaction_model(self.local_action_db, params) + liveaction_db = LiveAction.add_or_update(liveaction_db) + + # Create initial execution object + executions.create_execution_object(liveaction_db) + + # Mock to raise OperationalError on first call only + original_update = ActionExecution.update + first_call = [True] + + def mock_update_first_call_only(model_object, **kwargs): + if first_call[0]: + first_call[0] = False + raise kombu_exceptions.OperationalError("Channel error") + return original_update(model_object, **kwargs) + + with mock.patch.object( + ActionExecution, + "update", + side_effect=mock_update_first_call_only, + ): + # Verify the exception propagates + with self.assertRaises(kombu_exceptions.OperationalError) as cm: + action_worker._run_action(liveaction_db) + + self.assertIn("Channel error", str(cm.exception)) + + # Verify that the transaction-like rollback worked + updated_liveaction = LiveAction.get_by_id(liveaction_db.id) + self.assertEqual( + updated_liveaction.status, action_constants.LIVEACTION_STATUS_REQUESTED + ) diff --git a/st2actions/tests/unit/test_worker.py b/st2actions/tests/unit/test_worker.py index 917d0683e1..271d225a25 100644 --- a/st2actions/tests/unit/test_worker.py +++ b/st2actions/tests/unit/test_worker.py @@ -270,20 +270,23 @@ def test_worker_graceful_shutdown_with_multiple_runners(self): def test_worker_graceful_shutdown_with_single_runner(self): self.reset_config( - exit_still_active_check=10, - still_active_check_interval=1, + exit_still_active_check=2, + still_active_check_interval=0.2, service_registry=True, ) action_worker = actions_worker.get_worker() temp_file = None - # Create a temporary file that is deleted when the file is closed and then set up an - # action to wait for this file to be deleted. This allows this test to run the action - # over a separate thread, run the shutdown sequence on the main thread, and then let - # the local runner to exit gracefully and allow _run_action to finish execution. - with tempfile.NamedTemporaryFile() as fp: - temp_file = fp.name + # Create a temporary file that is NOT automatically deleted. This ensures the action + # stays running during shutdown/abandonment verification, preventing a race condition + # where the action completes and marks itself as "succeeded" before the shutdown + # abandonment logic runs. + fp = tempfile.NamedTemporaryFile(delete=False) + temp_file = fp.name + fp.close() + + try: self.assertIsNotNone(temp_file) self.assertTrue(os.path.isfile(temp_file)) @@ -296,9 +299,9 @@ def test_worker_graceful_shutdown_with_single_runner(self): executions.create_execution_object(liveaction_db) runner_thread = eventlet.spawn(action_worker._run_action, liveaction_db) - # Wait for the worker up to 10s to add the liveaction to _running_liveactions. - for i in range(0, int(10 / 0.1)): - eventlet.sleep(0.1) + # Wait for the worker up to 3s to add the liveaction to _running_liveactions. + for i in range(0, int(3 / 0.05)): + eventlet.sleep(0.05) if len(action_worker._running_liveactions) > 0: break @@ -307,31 +310,34 @@ def test_worker_graceful_shutdown_with_single_runner(self): # Shutdown the worker to trigger the abandon process. shutdown_thread = eventlet.spawn(action_worker.shutdown) # Wait for action runner shutdown sequence to complete - eventlet.sleep(5) + eventlet.sleep(0.5) - # Make sure the temporary file has been deleted. - self.assertFalse(os.path.isfile(temp_file)) + # Wait for the worker up to 3s to remove the liveaction from _running_liveactions. + for i in range(0, int(3 / 0.05)): + eventlet.sleep(0.05) + if len(action_worker._running_liveactions) < 1: + break + liveaction_db = LiveAction.get_by_id(liveaction_db.id) - # Wait for the worker up to 10s to remove the liveaction from _running_liveactions. - for i in range(0, int(10 / 0.1)): - eventlet.sleep(0.1) - if len(action_worker._running_liveactions) < 1: - break - liveaction_db = LiveAction.get_by_id(liveaction_db.id) + # Verify that _running_liveactions is empty and the liveaction is abandoned. + self.assertEqual(len(action_worker._running_liveactions), 0) + self.assertEqual( + liveaction_db.status, + action_constants.LIVEACTION_STATUS_ABANDONED, + str(liveaction_db), + ) - # Verify that _running_liveactions is empty and the liveaction is abandoned. - self.assertEqual(len(action_worker._running_liveactions), 0) - self.assertEqual( - liveaction_db.status, - action_constants.LIVEACTION_STATUS_ABANDONED, - str(liveaction_db), - ) + finally: + # Clean up: delete the temporary file to allow subprocess to exit + # This must happen before waiting for threads to prevent deadlock + if temp_file and os.path.exists(temp_file): + os.unlink(temp_file) - # Wait for the local runner to complete. This will activate the finally block in - # _run_action but will not result in KeyError because the discard method is used to - # to remove the liveaction from _running_liveactions. - runner_thread.wait() - shutdown_thread.kill() + # Wait for the local runner to complete. This will activate the finally block in + # _run_action but will not result in KeyError because the discard method is used to + # to remove the liveaction from _running_liveactions. + runner_thread.wait() + shutdown_thread.kill() @mock.patch.object( RedisDriver, @@ -339,22 +345,26 @@ def test_worker_graceful_shutdown_with_single_runner(self): mock.MagicMock(return_value=coordination.NoOpAsyncResult(("member-1",))), ) def test_worker_graceful_shutdown_exit_timeout(self): - self.reset_config(exit_still_active_check=5) + self.reset_config(exit_still_active_check=2) action_worker = actions_worker.get_worker() temp_file = None - # Create a temporary file that is deleted when the file is closed and then set up an - # action to wait for this file to be deleted. This allows this test to run the action - # over a separate thread, run the shutdown sequence on the main thread, and then let - # the local runner to exit gracefully and allow _run_action to finish execution. - with tempfile.NamedTemporaryFile() as fp: - temp_file = fp.name + # Create a temporary file that is NOT automatically deleted. This ensures the action + # stays running during shutdown/abandonment verification, preventing a race condition + # where the action completes and marks itself as "succeeded" before the shutdown + # abandonment logic runs. + fp = tempfile.NamedTemporaryFile(delete=False) + temp_file = fp.name + fp.close() + + try: self.assertIsNotNone(temp_file) self.assertTrue(os.path.isfile(temp_file)) # Launch the action execution in a separate thread. - params = {"cmd": "while [ -e '%s' ]; do sleep 0.1; done" % temp_file} + # Use longer sleep to ensure action runs past the timeout + params = {"cmd": "while [ -e '%s' ]; do sleep 5; done" % temp_file} liveaction_db = self._get_liveaction_model( WorkerTestCase.local_action_db, params ) @@ -372,29 +382,34 @@ def test_worker_graceful_shutdown_exit_timeout(self): # Shutdown the worker to trigger the abandon process. shutdown_thread = eventlet.spawn(action_worker.shutdown) - # Continue the excution for 5+ seconds to ensure timeout occurs. - eventlet.sleep(6) - - # Make sure the temporary file has been deleted. - self.assertFalse(os.path.isfile(temp_file)) + # Continue the execution for 2+ seconds to ensure timeout occurs. + # The action sleeps for 5 seconds, so it will still be running + # when the 2 second timeout expires. + eventlet.sleep(3) - # Wait for the worker up to 10s to remove the liveaction from _running_liveactions. - for i in range(0, int(10 / 0.1)): - eventlet.sleep(0.1) - if len(action_worker._running_liveactions) < 1: - break - liveaction_db = LiveAction.get_by_id(liveaction_db.id) + # Wait for the worker up to 10s to remove the liveaction from _running_liveactions. + for i in range(0, int(10 / 0.1)): + eventlet.sleep(0.1) + if len(action_worker._running_liveactions) < 1: + break + liveaction_db = LiveAction.get_by_id(liveaction_db.id) - # Verify that _running_liveactions is empty and the liveaction is abandoned. - self.assertEqual(len(action_worker._running_liveactions), 0) - self.assertEqual( - liveaction_db.status, - action_constants.LIVEACTION_STATUS_ABANDONED, - str(liveaction_db), - ) + # Verify that _running_liveactions is empty and the liveaction is abandoned. + self.assertEqual(len(action_worker._running_liveactions), 0) + self.assertEqual( + liveaction_db.status, + action_constants.LIVEACTION_STATUS_ABANDONED, + str(liveaction_db), + ) - # Wait for the local runner to complete. This will activate the finally block in - # _run_action but will not result in KeyError because the discard method is used to - # to remove the liveaction from _running_liveactions. - runner_thread.wait() - shutdown_thread.kill() + finally: + # Clean up: delete the temporary file to allow subprocess to exit + # This must happen before waiting for threads to prevent deadlock + if temp_file and os.path.exists(temp_file): + os.unlink(temp_file) + + # Wait for the local runner to complete. This will activate the finally block in + # _run_action but will not result in KeyError because the discard method is used to + # to remove the liveaction from _running_liveactions. + runner_thread.wait() + shutdown_thread.kill() diff --git a/st2api/st2api/controllers/v1/keyvalue.py b/st2api/st2api/controllers/v1/keyvalue.py index 3e6163e78f..d7c74b2ab3 100644 --- a/st2api/st2api/controllers/v1/keyvalue.py +++ b/st2api/st2api/controllers/v1/keyvalue.py @@ -178,7 +178,6 @@ def get_all( user = user or requester_user.name rbac_utils = get_rbac_backend().get_utils_class() - # Validate that the authenticated user is admin if user query param is provided rbac_utils.assert_user_is_admin_if_user_query_param_is_provided( user_db=requester_user, user=user, require_rbac=True @@ -451,7 +450,6 @@ def delete(self, name, requester_user, scope=None, user=None): scope=scope, name=key_ref, ) - # Check that user has permission to the key value pair. # If RBAC is enabled, this check will verify if user has system role with all access. # If RBAC is enabled, this check guards against a user accessing another user's kvp. diff --git a/st2common/st2common/models/api/keyvalue.py b/st2common/st2common/models/api/keyvalue.py index baf9d15c31..2b21cdc0fa 100644 --- a/st2common/st2common/models/api/keyvalue.py +++ b/st2common/st2common/models/api/keyvalue.py @@ -25,6 +25,7 @@ FULL_SYSTEM_SCOPE, FULL_USER_SCOPE, ALLOWED_SCOPES, + USER_SEPARATOR, ) from st2common.constants.keyvalue import SYSTEM_SCOPE, USER_SCOPE from st2common.exceptions.keyvalue import ( @@ -132,9 +133,20 @@ def from_model(cls, model, mask_secrets=True): key = doc.get("name", None) if (scope == USER_SCOPE or scope == FULL_USER_SCOPE) and key: - doc["user"] = UserKeyReference.get_user(key) - doc["name"] = UserKeyReference.get_name(key) - + # Check if name is in full "user:keyname" format + if USER_SEPARATOR in key: # USER_SEPARATOR + # Parse the full reference + doc["user"] = UserKeyReference.get_user(key) + doc["name"] = UserKeyReference.get_name(key) + else: + # Name is already clean, extract user from UID + # UID format: key_value_pair:st2kv.user:: + uid = doc.get("uid") + if uid: + parts = uid.split(USER_SEPARATOR) + if len(parts) >= 4: + doc["user"] = parts[2] # The username + # name stays as-is (already clean) doc["encrypted"] = encrypted attrs = {attr: value for attr, value in six.iteritems(doc) if value is not None} return cls(**attrs) diff --git a/st2common/st2common/persistence/base.py b/st2common/st2common/persistence/base.py index 1409571e48..3a7d6fe2a3 100644 --- a/st2common/st2common/persistence/base.py +++ b/st2common/st2common/persistence/base.py @@ -22,8 +22,12 @@ import six +from amqp import exceptions as amqp_exceptions from st2common import log as logging -from st2common.exceptions.db import StackStormDBObjectConflictError +from st2common.exceptions.db import ( + StackStormDBObjectConflictError, + StackStormDBObjectNotFoundError, +) from st2common.models.system.common import ResourceReference @@ -132,6 +136,7 @@ def insert( # Late import to avoid very expensive in-direct import (~1 second) when this function # is not called / used from mongoengine import NotUniqueError + from kombu import exceptions as kombu_exceptions if model_object.id: raise ValueError("id for object %s was unexpected." % model_object) @@ -151,21 +156,25 @@ def insert( message=message, conflict_id=conflict_id, model_object=model_object ) - # Publish internal event on the message bus - if publish: - try: + try: + # Publish internal event on the message bus + if publish: cls.publish_create(model_object) - except: - LOG.exception("Publish failed.") - # Dispatch trigger - if dispatch_trigger: - try: + # Dispatch trigger + if dispatch_trigger: cls.dispatch_create_trigger(model_object) - except: - LOG.exception("Trigger dispatch failed.") - return model_object + return model_object + except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + # RabbitMQ connection error - rollback the database insert + LOG.warning( + "RabbitMQ publish failed for object %s, rolling back database insert", + model_object.id, + ) + # Delete the newly inserted object + cls._get_impl().delete(model_object) + raise @classmethod def add_or_update( @@ -179,8 +188,18 @@ def add_or_update( # Late import to avoid very expensive in-direct import (~1 second) when this function # is not called / used from mongoengine import NotUniqueError + from kombu import exceptions as kombu_exceptions pre_persist_id = model_object.id + + # For updates, save the original state for potential rollback + original_object = None + if pre_persist_id and (publish or dispatch_trigger): + try: + original_object = cls.get_by_id(pre_persist_id) + except StackStormDBObjectNotFoundError: + pass + try: model_object = cls._get_impl().add_or_update(model_object, validate=True) except NotUniqueError as e: @@ -199,27 +218,36 @@ def add_or_update( is_update = str(pre_persist_id) == str(model_object.id) - # Publish internal event on the message bus - if publish: - try: + try: + # Publish internal event on the message bus + if publish: if is_update: cls.publish_update(model_object) else: cls.publish_create(model_object) - except: - LOG.exception("Publish failed.") - # Dispatch trigger - if dispatch_trigger: - try: + # Dispatch trigger + if dispatch_trigger: if is_update: cls.dispatch_update_trigger(model_object) else: cls.dispatch_create_trigger(model_object) - except: - LOG.exception("Trigger dispatch failed.") - return model_object + return model_object + except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + # RabbitMQ connection error - rollback the database operation + LOG.warning( + "RabbitMQ publish failed for object %s, rolling back database operation", + model_object.id, + ) + + if is_update and original_object: + # Restore the original state for updates + cls._get_impl().add_or_update(original_object, validate=False) + else: + # Delete the newly created object for inserts + cls._get_impl().delete(model_object) + raise @classmethod def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs): @@ -227,28 +255,51 @@ def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs): Use this method when - * upsert=False is desired * special operators like push, push_all are to be used. + + NOTE: If publish fails due to RabbitMQ connection errors, the database update + will be rolled back by restoring the original object state. """ + from kombu import exceptions as kombu_exceptions + + # Save the original state before update for potential rollback + original_object = cls.get_by_id(model_object.id) + + # Perform the database update cls._get_impl().update(model_object, **kwargs) # update does not return the object but a flag; likely success/fail but docs # are not very good on this one so ignoring. Explicitly get the object from - # DB abd return. - model_object = cls.get_by_id(model_object.id) + # DB and return. + updated_object = cls.get_by_id(model_object.id) - # Publish internal event on the message bus - if publish: - try: - cls.publish_update(model_object) - except: - LOG.exception("Publish failed.") - - # Dispatch trigger - if dispatch_trigger: - try: - cls.dispatch_update_trigger(model_object) - except: - LOG.exception("Trigger dispatch failed.") - - return model_object + try: + # Publish internal event on the message bus + if publish: + cls.publish_update(updated_object) + + # Dispatch trigger + if dispatch_trigger: + cls.dispatch_update_trigger(updated_object) + + return updated_object + except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + # RabbitMQ connection error - rollback the database update + if original_object: + LOG.warning( + "RabbitMQ publish failed for object %s, rolling back database update", + model_object.id, + ) + # Build rollback kwargs from the original state + rollback_kwargs = {} + for key, value in kwargs.items(): + if key.startswith("set__"): + field_name = key[5:] # Remove 'set__' prefix + original_value = getattr(original_object, field_name, None) + rollback_kwargs[key] = original_value + # For other operators, we'd need to handle them appropriately + # For now, only handling set__ which is the most common case + + cls._get_impl().update(model_object, **rollback_kwargs) + raise @classmethod def delete(cls, model_object, publish=True, dispatch_trigger=True): @@ -256,17 +307,11 @@ def delete(cls, model_object, publish=True, dispatch_trigger=True): # Publish internal event on the message bus if publish: - try: - cls.publish_delete(model_object) - except Exception: - LOG.exception("Publish failed.") + cls.publish_delete(model_object) # Dispatch trigger if dispatch_trigger: - try: - cls.dispatch_delete_trigger(model_object) - except Exception: - LOG.exception("Trigger dispatch failed.") + cls.dispatch_delete_trigger(model_object) return persisted_object diff --git a/st2common/st2common/services/action.py b/st2common/st2common/services/action.py index 5750db26df..d5f35eb204 100644 --- a/st2common/st2common/services/action.py +++ b/st2common/st2common/services/action.py @@ -305,7 +305,7 @@ def request_cancellation(liveaction, requester): # if the liveaction is operating under a workflow. if ( "parent" in liveaction.context - or liveaction.status in action_constants.LIVEACTION_STATUS_RUNNING + or liveaction.status == action_constants.LIVEACTION_STATUS_RUNNING ): status = action_constants.LIVEACTION_STATUS_CANCELING else: diff --git a/st2common/st2common/transport/connection_retry_wrapper.py b/st2common/st2common/transport/connection_retry_wrapper.py index 492aa24f32..6416b18f43 100644 --- a/st2common/st2common/transport/connection_retry_wrapper.py +++ b/st2common/st2common/transport/connection_retry_wrapper.py @@ -16,46 +16,50 @@ from __future__ import absolute_import import six +from kombu import exceptions as kombu_exceptions from st2common.util import concurrency __all__ = ["ConnectionRetryWrapper", "ClusterRetryContext"] +# Higher-level exception tuple that covers all connection-related errors + class ClusterRetryContext(object): """ - Stores retry context for cluster retries. It makes certain assumptions - on how cluster_size and retry should be determined. + Stores retry context for cluster retries. """ - def __init__(self, cluster_size): - # No of nodes in a cluster + def __init__(self, cluster_size, max_retries=2, wait_between_retry=10): self.cluster_size = cluster_size - # No of times to retry in a cluster - self.cluster_retry = 2 - # time to wait between retry in a cluster - self.wait_between_cluster = 10 - - # No of nodes attempted. Starts at 1 since the - self._nodes_attempted = 1 - - def test_should_stop(self, e=None): - # Special workaround for "(504) CHANNEL_ERROR - second 'channel.open' seen" which happens - # during tests on Travis and block and slown down the tests - # NOTE: This error is not fatal during tests and we can simply switch to a next connection - # without sleeping. + self.max_retries = max_retries + self.wait_between_retry = wait_between_retry + self._attempt_count = 0 + self._max_attempts = cluster_size * (max_retries + 1) + + def should_stop(self, e=None): + """ + Determine if retry should stop and how long to wait before next attempt. + + Returns: + tuple: (should_stop, wait_seconds) + """ + self._attempt_count += 1 + + # Special workaround for non-fatal test errors if "second 'channel.open' seen" in six.text_type(e): - return False, -1 + return False, 0 + + if self._attempt_count >= self._max_attempts: + return True, 0 - should_stop = True - if self._nodes_attempted > self.cluster_size * self.cluster_retry: - return should_stop, -1 - wait = 0 - should_stop = False - if self._nodes_attempted % self.cluster_size == 0: - wait = self.wait_between_cluster - self._nodes_attempted += 1 - return should_stop, wait + # Wait before retrying after cycling through all cluster nodes + wait = ( + self.wait_between_retry + if self._attempt_count % self.cluster_size == 0 + else 0 + ) + return False, wait class ConnectionRetryWrapper(object): @@ -103,11 +107,11 @@ def wrapped_callback(connection, channel): """ - def __init__(self, cluster_size, logger, ensure_max_retries=3): - self._retry_context = ClusterRetryContext(cluster_size=cluster_size) + def __init__(self, cluster_size, logger, max_retries=2, ensure_max_retries=3): + self._retry_context = ClusterRetryContext( + cluster_size=cluster_size, max_retries=max_retries + ) self._logger = logger - # How many times to try to retrying establishing a connection in a place where we are - # calling connection.ensure_connection self._ensure_max_retries = ensure_max_retries def errback(self, exc, interval): @@ -124,38 +128,32 @@ def run(self, connection, wrapped_callback): method. Expected signature of callback - ``def func(connection, channel)`` """ - should_stop = False channel = None - while not should_stop: + while True: try: channel = connection.channel() wrapped_callback(connection=connection, channel=channel) - should_stop = True - except connection.connection_errors + connection.channel_errors as e: - should_stop, wait = self._retry_context.test_should_stop(e) - # reset channel to None to avoid any channel closing errors. At this point - # in case of an exception there should be no channel but that is better to - # guarantee. - channel = None - # All attempts to re-establish connections have failed. This error needs to - # be notified so raise. + break # Success - exit the retry loop + except kombu_exceptions.KombuError as e: + channel = None # Reset channel to avoid closing errors + should_stop, wait = self._retry_context.should_stop(e) + if should_stop: + self._logger.error( + "Failed to execute operation after exhausting all retry attempts" + ) raise - # -1, 0 and 1+ are handled properly by eventlet.sleep - self._logger.debug( - "Received RabbitMQ server error, sleeping for %s seconds " - "before retrying: %s" % (wait, six.text_type(e)) - ) - concurrency.sleep(wait) + if wait > 0: + self._logger.debug( + "Received RabbitMQ server error, sleeping for %s seconds " + "before retrying: %s" % (wait, six.text_type(e)) + ) + concurrency.sleep(wait) connection.close() - # ensure_connection will automatically switch to an alternate. Other connections - # in the pool will be fixed independently. It would be nice to cut-over the - # entire ConnectionPool simultaneously but that would require writing our own - # ConnectionPool. If a server recovers it could happen that the same process - # ends up talking to separate nodes in a cluster. + # ensure_connection will automatically switch to an alternate node def log_error_on_conn_failure(exc, interval): self._logger.debug( "Failed to re-establish connection to RabbitMQ server, " @@ -163,31 +161,16 @@ def log_error_on_conn_failure(exc, interval): ) try: - # NOTE: This function blocks and tries to restablish a connection for - # indefinetly if "max_retries" argument is not specified connection.ensure_connection( max_retries=self._ensure_max_retries, errback=log_error_on_conn_failure, ) - except Exception: - self._logger.exception( - "Connections to RabbitMQ cannot be re-established: %s", - six.text_type(e), - ) + except kombu_exceptions.KombuError: + self._logger.error("Failed to re-establish connection to RabbitMQ") raise - except Exception as e: - self._logger.exception( - "Connections to RabbitMQ cannot be re-established: %s", - six.text_type(e), - ) - # Not being able to publish a message could be a significant issue for an app. - raise finally: - if should_stop and channel: - try: - channel.close() - except Exception: - self._logger.warning("Error closing channel.", exc_info=True) + if channel: + channel.close() def ensured(self, connection, obj, to_ensure_func, **kwargs): """ diff --git a/st2common/tests/unit/test_connection_retry_wrapper.py b/st2common/tests/unit/test_connection_retry_wrapper.py index 831ac8c22e..9fcc45158e 100644 --- a/st2common/tests/unit/test_connection_retry_wrapper.py +++ b/st2common/tests/unit/test_connection_retry_wrapper.py @@ -23,42 +23,47 @@ class TestClusterRetryContext(unittest.TestCase): def test_single_node_cluster_retry(self): retry_context = ClusterRetryContext(cluster_size=1) - should_stop, wait = retry_context.test_should_stop() + should_stop, wait = retry_context.should_stop() self.assertFalse(should_stop, "Not done trying.") self.assertEqual(wait, 10) - should_stop, wait = retry_context.test_should_stop() + should_stop, wait = retry_context.should_stop() self.assertFalse(should_stop, "Not done trying.") self.assertEqual(wait, 10) - should_stop, wait = retry_context.test_should_stop() + should_stop, wait = retry_context.should_stop() self.assertTrue(should_stop, "Done trying.") - self.assertEqual(wait, -1) + self.assertEqual(wait, 0) def test_should_stop_second_channel_open_error_should_be_non_fatal(self): retry_context = ClusterRetryContext(cluster_size=1) e = Exception("(504) CHANNEL_ERROR - second 'channel.open' seen") - should_stop, wait = retry_context.test_should_stop(e=e) + should_stop, wait = retry_context.should_stop(e=e) self.assertFalse(should_stop) - self.assertEqual(wait, -1) + self.assertEqual(wait, 0) e = Exception("CHANNEL_ERROR - second 'channel.open' seen") - should_stop, wait = retry_context.test_should_stop(e=e) + should_stop, wait = retry_context.should_stop(e=e) self.assertFalse(should_stop) - self.assertEqual(wait, -1) + self.assertEqual(wait, 0) def test_multiple_node_cluster_retry(self): cluster_size = 3 - last_index = cluster_size * 2 + max_retries = 2 + # _max_attempts = cluster_size * (max_retries + 1) = 3 * 3 = 9 + # First attempt doesn't count as retry, so we have 9 total attempts (indices 0-8) + last_index = (cluster_size * (max_retries + 1)) - 1 - retry_context = ClusterRetryContext(cluster_size=cluster_size) + retry_context = ClusterRetryContext( + cluster_size=cluster_size, max_retries=max_retries + ) for i in range(last_index + 1): - should_stop, wait = retry_context.test_should_stop() + should_stop, wait = retry_context.should_stop() if i == last_index: self.assertTrue(should_stop, "Done trying.") - self.assertEqual(wait, -1) + self.assertEqual(wait, 0) else: self.assertFalse(should_stop, "Not done trying.") # on cluster boundaries the wait is longer. Short wait when switching @@ -70,6 +75,6 @@ def test_multiple_node_cluster_retry(self): def test_zero_node_cluster_retry(self): retry_context = ClusterRetryContext(cluster_size=0) - should_stop, wait = retry_context.test_should_stop() + should_stop, wait = retry_context.should_stop() self.assertTrue(should_stop, "Done trying.") - self.assertEqual(wait, -1) + self.assertEqual(wait, 0) diff --git a/st2common/tests/unit/test_persistence_rollback.py b/st2common/tests/unit/test_persistence_rollback.py new file mode 100644 index 0000000000..947bbf1597 --- /dev/null +++ b/st2common/tests/unit/test_persistence_rollback.py @@ -0,0 +1,124 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Tests for rollback behavior in persistence layer when RabbitMQ publishing fails. +""" + +from __future__ import absolute_import +import uuid +from unittest import mock + +from kombu import exceptions as kombu_exceptions + +from st2tests import DbTestCase +from tests.unit.base import FakeModel, FakeModelDB + + +class TestPersistenceRollback(DbTestCase): + """Test rollback behavior when publishing fails in update() method""" + + @classmethod + def setUpClass(cls): + super(TestPersistenceRollback, cls).setUpClass() + cls.access = FakeModel() + + def tearDown(self): + FakeModelDB.drop_collection() + super(TestPersistenceRollback, self).tearDown() + + def test_update_rollback_on_kombu_error(self): + """Test that update() rolls back DB changes when KombuError occurs""" + # Create initial object + obj = FakeModelDB(name=uuid.uuid4().hex, context={"value": "original"}) + obj = self.access.add_or_update(obj, publish=False) + original_name = obj.name + # Mock publish_update at class level to raise KombuError + with mock.patch.object( + FakeModel, + "publish_update", + side_effect=kombu_exceptions.KombuError("Connection failed"), + ): + # Try to update with a new name + new_name = uuid.uuid4().hex + obj.name = new_name + + # Update should raise the exception + with self.assertRaises(kombu_exceptions.KombuError): + self.access.update(obj, publish=True, set__name=new_name) + + # Verify the database was rolled back to original state + retrieved = self.access.get_by_id(str(obj.id)) + self.assertEqual(retrieved.name, original_name) + self.assertNotEqual(retrieved.name, new_name) + + def test_update_no_rollback_on_other_exceptions(self): + """Test that update() does NOT rollback on non-RabbitMQ exceptions""" + # Create initial object + obj = FakeModelDB(name=uuid.uuid4().hex, context={"value": "original"}) + obj = self.access.add_or_update(obj, publish=False) + + # Mock publish_update at class level to raise a generic exception + with mock.patch.object( + FakeModel, "publish_update", side_effect=ValueError("Some other error") + ): + # Try to update with a new name + new_name = uuid.uuid4().hex + obj.name = new_name + + # Update should propagate the non-RabbitMQ exception + with self.assertRaises(ValueError): + self.access.update(obj, publish=True, set__name=new_name) + + # Since ValueError is not a KombuError, no rollback occurs + # The DB change remains + retrieved = self.access.get_by_id(str(obj.id)) + self.assertEqual(retrieved.name, new_name) + + def test_update_success_no_rollback(self): + """Test that successful update() with publish does not trigger rollback""" + # Create initial object + obj = FakeModelDB(name=uuid.uuid4().hex, context={"value": "original"}) + obj = self.access.add_or_update(obj, publish=False) + + # Update with a new name and publish=True (mocked to succeed) + new_name = uuid.uuid4().hex + obj.name = new_name + + with mock.patch.object(FakeModel, "publish_update", return_value=None): + result = self.access.update(obj, publish=True, set__name=new_name) + + # Verify the update succeeded + self.assertEqual(result.name, new_name) + retrieved = self.access.get_by_id(str(obj.id)) + self.assertEqual(retrieved.name, new_name) + + def test_update_without_publish_no_rollback_needed(self): + """Test that update() without publish=True doesn't save original state""" + # Create initial object + obj = FakeModelDB(name=uuid.uuid4().hex, context={"value": "original"}) + obj = self.access.add_or_update(obj, publish=False) + + # Update with publish=False + new_name = uuid.uuid4().hex + obj.name = new_name + result = self.access.update( + obj, publish=False, dispatch_trigger=False, set__name=new_name + ) + + # Verify the update succeeded + self.assertEqual(result.name, new_name) + retrieved = self.access.get_by_id(str(obj.id)) + self.assertEqual(retrieved.name, new_name) From 5cc1c7d08f5c6686402571fa5fa7bf20009a7c62 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 12:17:51 -0400 Subject: [PATCH 02/24] catch base exception to just catch everything including none type has not attribute --- st2common/st2common/persistence/base.py | 3 +- .../transport/connection_retry_wrapper.py | 2 +- .../unit/test_connection_retry_wrapper.py | 238 +++++++++++++++++- 3 files changed, 240 insertions(+), 3 deletions(-) diff --git a/st2common/st2common/persistence/base.py b/st2common/st2common/persistence/base.py index 3a7d6fe2a3..02aa50e718 100644 --- a/st2common/st2common/persistence/base.py +++ b/st2common/st2common/persistence/base.py @@ -23,6 +23,8 @@ import six from amqp import exceptions as amqp_exceptions +from kombu import exceptions as kombu_exceptions + from st2common import log as logging from st2common.exceptions.db import ( StackStormDBObjectConflictError, @@ -136,7 +138,6 @@ def insert( # Late import to avoid very expensive in-direct import (~1 second) when this function # is not called / used from mongoengine import NotUniqueError - from kombu import exceptions as kombu_exceptions if model_object.id: raise ValueError("id for object %s was unexpected." % model_object) diff --git a/st2common/st2common/transport/connection_retry_wrapper.py b/st2common/st2common/transport/connection_retry_wrapper.py index 6416b18f43..34ec00b140 100644 --- a/st2common/st2common/transport/connection_retry_wrapper.py +++ b/st2common/st2common/transport/connection_retry_wrapper.py @@ -134,7 +134,7 @@ def run(self, connection, wrapped_callback): channel = connection.channel() wrapped_callback(connection=connection, channel=channel) break # Success - exit the retry loop - except kombu_exceptions.KombuError as e: + except Exception as e: channel = None # Reset channel to avoid closing errors should_stop, wait = self._retry_context.should_stop(e) diff --git a/st2common/tests/unit/test_connection_retry_wrapper.py b/st2common/tests/unit/test_connection_retry_wrapper.py index 9fcc45158e..3066574f46 100644 --- a/st2common/tests/unit/test_connection_retry_wrapper.py +++ b/st2common/tests/unit/test_connection_retry_wrapper.py @@ -15,8 +15,12 @@ from __future__ import absolute_import import unittest +from unittest.mock import Mock -from st2common.transport.connection_retry_wrapper import ClusterRetryContext +from st2common.transport.connection_retry_wrapper import ( + ClusterRetryContext, + ConnectionRetryWrapper, +) from six.moves import range @@ -78,3 +82,235 @@ def test_zero_node_cluster_retry(self): should_stop, wait = retry_context.should_stop() self.assertTrue(should_stop, "Done trying.") self.assertEqual(wait, 0) + + +class TestConnectionRetryWrapper(unittest.TestCase): + """Test cases for ConnectionRetryWrapper class""" + + def test_connection_channel_attribute_error_with_none_connection(self): + """ + Test that ConnectionRetryWrapper handles AttributeError when connection.channel() + is called on a NoneType object (when connection.connection is None). + + This reproduces the error: + AttributeError: 'NoneType' object has no attribute 'channel' + + The retry wrapper should attempt retries and eventually raise the error + after exhausting all retry attempts. + """ + # Setup mock logger + mock_logger = Mock() + + # Create ConnectionRetryWrapper with single node cluster + # This will allow 3 attempts total: initial + 2 retries (max_retries=2) + wrapper = ConnectionRetryWrapper( + cluster_size=1, logger=mock_logger, max_retries=2 + ) + + # Create mock connection that raises AttributeError when channel() is called + mock_connection = Mock() + mock_connection.channel.side_effect = AttributeError( + "'NoneType' object has no attribute 'channel'" + ) + mock_connection.close = Mock() + mock_connection.ensure_connection = Mock() + + # Create a simple callback + callback = Mock() + + # Execute and expect AttributeError to be raised after retries exhausted + with self.assertRaises(AttributeError) as context: + wrapper.run(connection=mock_connection, wrapped_callback=callback) + + # Verify the error message + self.assertIn( + "'NoneType' object has no attribute 'channel'", str(context.exception) + ) + + # Verify that channel() was called multiple times (initial + retries) + # cluster_size=1, max_retries=2 means 3 total attempts + self.assertEqual(mock_connection.channel.call_count, 3) + + # Verify connection.close() was called on each retry attempt (not on final failure) + self.assertEqual(mock_connection.close.call_count, 2) + + # Verify ensure_connection was called on each retry + self.assertEqual(mock_connection.ensure_connection.call_count, 2) + + # Verify callback was never called since channel() always failed + callback.assert_not_called() + + # Verify error logging occurred + mock_logger.error.assert_called() + error_calls = [call for call in mock_logger.error.call_args_list] + self.assertTrue( + any( + "Failed to execute operation after exhausting all retry attempts" + in str(call) + for call in error_calls + ), + "Expected error message about exhausted retries", + ) + + def test_connection_retry_wrapper_successful_after_initial_failure(self): + """ + Test that ConnectionRetryWrapper successfully retries and completes + when an initial AttributeError occurs but subsequent attempts succeed. + """ + mock_logger = Mock() + wrapper = ConnectionRetryWrapper( + cluster_size=1, logger=mock_logger, max_retries=2 + ) + + # Create mock connection that fails first, then succeeds + mock_connection = Mock() + mock_channel = Mock() + + # First call raises AttributeError, second call succeeds + mock_connection.channel.side_effect = [ + AttributeError("'NoneType' object has no attribute 'channel'"), + mock_channel, + ] + mock_connection.close = Mock() + mock_connection.ensure_connection = Mock() + + # Create callback that should be called when channel is available + callback = Mock() + + # Execute - should succeed on second attempt + wrapper.run(connection=mock_connection, wrapped_callback=callback) + + # Verify channel() was called twice (failed once, succeeded once) + self.assertEqual(mock_connection.channel.call_count, 2) + + # Verify callback was called once with successful channel + callback.assert_called_once_with( + connection=mock_connection, channel=mock_channel + ) + + # Verify connection was closed after first failure + self.assertEqual(mock_connection.close.call_count, 1) + + # Verify ensure_connection was called after first failure + self.assertEqual(mock_connection.ensure_connection.call_count, 1) + + # Verify channel was properly closed + mock_channel.close.assert_called_once() + + def test_connection_retry_wrapper_handles_generic_exception(self): + """ + Test that ConnectionRetryWrapper handles other exceptions properly + and still attempts retries. + """ + mock_logger = Mock() + wrapper = ConnectionRetryWrapper( + cluster_size=1, logger=mock_logger, max_retries=1 + ) + + mock_connection = Mock() + mock_connection.channel.side_effect = RuntimeError("Connection failed") + mock_connection.close = Mock() + mock_connection.ensure_connection = Mock() + + callback = Mock() + + # Execute and expect RuntimeError after retries exhausted + with self.assertRaises(RuntimeError) as context: + wrapper.run(connection=mock_connection, wrapped_callback=callback) + + self.assertIn("Connection failed", str(context.exception)) + + # Verify retries occurred (initial + 1 retry = 2 attempts) + self.assertEqual(mock_connection.channel.call_count, 2) + self.assertEqual(mock_connection.close.call_count, 1) + self.assertEqual(mock_connection.ensure_connection.call_count, 1) + + def test_connection_refused_error_during_ensure_connection(self): + """ + Test that ConnectionRetryWrapper handles ConnectionRefusedError that occurs + during ensure_connection (when RabbitMQ is down or unreachable). + + This reproduces the error: + ConnectionRefusedError: [Errno 111] ECONNREFUSED + + The wrapper should attempt retries and eventually raise the error after + exhausting retry attempts, rather than retrying indefinitely. + """ + from kombu import exceptions as kombu_exceptions + + mock_logger = Mock() + wrapper = ConnectionRetryWrapper( + cluster_size=1, logger=mock_logger, max_retries=2, ensure_max_retries=3 + ) + + mock_connection = Mock() + # First call to channel() fails, triggering ensure_connection + mock_connection.channel.side_effect = OSError("Connection failed") + mock_connection.close = Mock() + + # ensure_connection raises KombuError wrapping ConnectionRefusedError + mock_connection.ensure_connection.side_effect = kombu_exceptions.KombuError( + "ConnectionRefusedError: [Errno 111] ECONNREFUSED" + ) + + callback = Mock() + + # Execute and expect KombuError to be raised after retries exhausted + with self.assertRaises(kombu_exceptions.KombuError) as context: + wrapper.run(connection=mock_connection, wrapped_callback=callback) + + # Verify the error message contains connection refused info + self.assertIn("ECONNREFUSED", str(context.exception)) + + # Verify channel() was called once (initial attempt that failed) + self.assertEqual(mock_connection.channel.call_count, 1) + + # Verify connection.close() was called before trying to re-establish + self.assertEqual(mock_connection.close.call_count, 1) + + # Verify ensure_connection was called once (and it failed with KombuError) + self.assertEqual(mock_connection.ensure_connection.call_count, 1) + + # Verify callback was never called since connection failed + callback.assert_not_called() + + # Verify error logging occurred + mock_logger.error.assert_called() + error_calls = [call for call in mock_logger.error.call_args_list] + self.assertTrue( + any( + "Failed to re-establish connection to RabbitMQ" in str(call) + for call in error_calls + ), + "Expected error message about failed connection re-establishment", + ) + + def test_connection_refused_during_channel_creation(self): + """ + Test ConnectionRefusedError raised directly during channel creation. + """ + mock_logger = Mock() + wrapper = ConnectionRetryWrapper( + cluster_size=1, logger=mock_logger, max_retries=1 + ) + + mock_connection = Mock() + # Simulate ConnectionRefusedError during channel creation + mock_connection.channel.side_effect = ConnectionRefusedError( + 111, "ECONNREFUSED" + ) + mock_connection.close = Mock() + mock_connection.ensure_connection = Mock() + + callback = Mock() + + # Execute and expect ConnectionRefusedError after retries exhausted + with self.assertRaises(ConnectionRefusedError) as context: + wrapper.run(connection=mock_connection, wrapped_callback=callback) + + self.assertIn("ECONNREFUSED", str(context.exception)) + + # Verify retries occurred (initial + 1 retry = 2 attempts) + self.assertEqual(mock_connection.channel.call_count, 2) + self.assertEqual(mock_connection.close.call_count, 1) + self.assertEqual(mock_connection.ensure_connection.call_count, 1) From 624bbd385a76c3292646eb44d0fe991c9a30449d Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 12:51:05 -0400 Subject: [PATCH 03/24] proper max connections --- st2actions/st2actions/worker.py | 11 ++ st2common/st2common/config.py | 22 +++ st2common/st2common/transport/utils.py | 13 ++ st2common/tests/unit/test_transport_utils.py | 144 +++++++++++++++++++ 4 files changed, 190 insertions(+) create mode 100644 st2common/tests/unit/test_transport_utils.py diff --git a/st2actions/st2actions/worker.py b/st2actions/st2actions/worker.py index 23c3c66ca2..97d0538a7f 100644 --- a/st2actions/st2actions/worker.py +++ b/st2actions/st2actions/worker.py @@ -367,5 +367,16 @@ def _resume_action(self, liveaction_db): def get_worker(): + """ + Create and return an ActionExecutionDispatcher worker. + + The worker connects to the messaging broker using connection retry settings + from the configuration. If the broker is unavailable and max retry attempts + are exhausted, the connection will raise an exception causing the process + to exit. This allows process supervisors (systemd, K8s) to restart the service. + + :return: ActionExecutionDispatcher instance + :rtype: ActionExecutionDispatcher + """ with transport_utils.get_connection() as conn: return ActionExecutionDispatcher(conn, ACTIONRUNNER_QUEUES) diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index ff9e1b4efd..e37a74c572 100644 --- a/st2common/st2common/config.py +++ b/st2common/st2common/config.py @@ -372,6 +372,28 @@ def register_opts(ignore_errors=False): default=10000, help="How long should we wait between connection retries.", ), + cfg.IntOpt( + "connection_retry_max_attempts", + default=10, + help="Maximum number of retry attempts for initial broker connection. " + "This prevents infinite retry loops when the broker is unavailable. " + "Set to 0 to retry indefinitely (not recommended).", + ), + cfg.IntOpt( + "connection_retry_interval_start", + default=1, + help="Starting retry interval in seconds for broker connection attempts.", + ), + cfg.IntOpt( + "connection_retry_interval_step", + default=1, + help="Increment for retry interval after each attempt (seconds).", + ), + cfg.IntOpt( + "connection_retry_interval_max", + default=30, + help="Maximum retry interval in seconds for broker connection attempts.", + ), cfg.BoolOpt( "ssl", default=False, diff --git a/st2common/st2common/transport/utils.py b/st2common/st2common/transport/utils.py index e479713ddc..6e628eb641 100644 --- a/st2common/st2common/transport/utils.py +++ b/st2common/st2common/transport/utils.py @@ -53,6 +53,15 @@ def get_connection(urls=None, connection_kwargs=None): kwargs = {} + # Transport options for connection retry behavior + # These options control the retry behavior during initial connection establishment + transport_options = { + "max_retries": cfg.CONF.messaging.connection_retry_max_attempts, + "interval_start": cfg.CONF.messaging.connection_retry_interval_start, + "interval_step": cfg.CONF.messaging.connection_retry_interval_step, + "interval_max": cfg.CONF.messaging.connection_retry_interval_max, + } + ssl_kwargs = _get_ssl_kwargs( ssl=cfg.CONF.messaging.ssl, ssl_keyfile=cfg.CONF.messaging.ssl_keyfile, @@ -70,11 +79,15 @@ def get_connection(urls=None, connection_kwargs=None): kwargs.update({"ssl": ssl_kwargs}) kwargs["login_method"] = cfg.CONF.messaging.login_method + kwargs["transport_options"] = transport_options kwargs.update(connection_kwargs) # NOTE: This line contains no secret values so it's OK to log it LOG.debug("Using SSL context for RabbitMQ connection: %s" % (ssl_kwargs)) + LOG.debug( + "Using transport options for RabbitMQ connection: %s" % (transport_options) + ) connection = Connection(urls, **kwargs) return connection diff --git a/st2common/tests/unit/test_transport_utils.py b/st2common/tests/unit/test_transport_utils.py new file mode 100644 index 0000000000..91272fb79b --- /dev/null +++ b/st2common/tests/unit/test_transport_utils.py @@ -0,0 +1,144 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import +import unittest +from unittest.mock import patch + +from oslo_config import cfg + +from st2common.transport import utils as transport_utils + + +class TestTransportUtils(unittest.TestCase): + """Test cases for transport utils module""" + + def setUp(self): + """Reset config before each test""" + super(TestTransportUtils, self).setUp() + # Clear any config overrides from previous tests + try: + cfg.CONF.clear_override("connection_retry_max_attempts", group="messaging") + except: + pass + try: + cfg.CONF.clear_override( + "connection_retry_interval_start", group="messaging" + ) + except: + pass + try: + cfg.CONF.clear_override("connection_retry_interval_step", group="messaging") + except: + pass + try: + cfg.CONF.clear_override("connection_retry_interval_max", group="messaging") + except: + pass + + @patch("st2common.transport.utils.Connection") + def test_get_connection_includes_transport_options(self, mock_connection): + """Test that get_connection passes transport_options with retry settings""" + # Setup config values + cfg.CONF.set_override("connection_retry_max_attempts", 15, group="messaging") + cfg.CONF.set_override("connection_retry_interval_start", 2, group="messaging") + cfg.CONF.set_override("connection_retry_interval_step", 2, group="messaging") + cfg.CONF.set_override("connection_retry_interval_max", 60, group="messaging") + + # Call get_connection + transport_utils.get_connection() + + # Verify Connection was called + self.assertTrue(mock_connection.called) + + # Get the kwargs passed to Connection + call_kwargs = mock_connection.call_args[1] + + # Verify transport_options are present + self.assertIn("transport_options", call_kwargs) + transport_options = call_kwargs["transport_options"] + + # Verify the retry settings + self.assertEqual(transport_options["max_retries"], 15) + self.assertEqual(transport_options["interval_start"], 2) + self.assertEqual(transport_options["interval_step"], 2) + self.assertEqual(transport_options["interval_max"], 60) + + @patch("st2common.transport.utils.Connection") + def test_get_connection_uses_default_transport_options(self, mock_connection): + """Test that get_connection uses default values from config""" + # Don't override config, use defaults + + # Call get_connection + transport_utils.get_connection() + + # Verify Connection was called + self.assertTrue(mock_connection.called) + + # Get the kwargs passed to Connection + call_kwargs = mock_connection.call_args[1] + + # Verify transport_options are present with defaults + self.assertIn("transport_options", call_kwargs) + transport_options = call_kwargs["transport_options"] + + # Verify default values (from config.py) + self.assertEqual(transport_options["max_retries"], 10) + self.assertEqual(transport_options["interval_start"], 1) + self.assertEqual(transport_options["interval_step"], 1) + self.assertEqual(transport_options["interval_max"], 30) + + @patch("st2common.transport.utils.Connection") + def test_get_connection_with_custom_connection_kwargs(self, mock_connection): + """Test that custom connection_kwargs don't override transport_options""" + cfg.CONF.set_override("connection_retry_max_attempts", 5, group="messaging") + + custom_kwargs = {"heartbeat": 60, "custom_param": "value"} + + # Call get_connection with custom kwargs + transport_utils.get_connection(connection_kwargs=custom_kwargs) + + # Verify Connection was called + self.assertTrue(mock_connection.called) + + # Get the kwargs passed to Connection + call_kwargs = mock_connection.call_args[1] + + # Verify transport_options are still present + self.assertIn("transport_options", call_kwargs) + self.assertEqual(call_kwargs["transport_options"]["max_retries"], 5) + + # Verify custom kwargs were also passed + self.assertEqual(call_kwargs["heartbeat"], 60) + self.assertEqual(call_kwargs["custom_param"], "value") + + @patch("st2common.transport.utils.Connection") + def test_get_connection_zero_max_retries_for_infinite(self, mock_connection): + """Test that setting max_retries to 0 enables infinite retries""" + # Set max_retries to 0 for infinite retries + cfg.CONF.set_override("connection_retry_max_attempts", 0, group="messaging") + + # Call get_connection + transport_utils.get_connection() + + # Verify Connection was called + self.assertTrue(mock_connection.called) + + # Get the kwargs passed to Connection + call_kwargs = mock_connection.call_args[1] + + # Verify transport_options has max_retries set to 0 + self.assertIn("transport_options", call_kwargs) + self.assertEqual(call_kwargs["transport_options"]["max_retries"], 0) From 7bf0091e8be128f675a528cdaace1104b528e202 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 12:55:23 -0400 Subject: [PATCH 04/24] recreate configgen --- conf/st2.conf.sample | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 40e8e7ef42..2fc025ed65 100644 --- a/conf/st2.conf.sample +++ b/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 initial broker connection. This prevents infinite retry loops when the broker is unavailable. 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.). From baad6336aa16a6343bd5ed549a6aa2c9c4a777a9 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 16:57:33 -0400 Subject: [PATCH 05/24] retry --- st2common/st2common/config.py | 3 ++- st2common/st2common/transport/publishers.py | 5 ++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index e37a74c572..28b0c062ec 100644 --- a/st2common/st2common/config.py +++ b/st2common/st2common/config.py @@ -375,8 +375,9 @@ def register_opts(ignore_errors=False): cfg.IntOpt( "connection_retry_max_attempts", default=10, - help="Maximum number of retry attempts for initial broker connection. " + help="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).", ), cfg.IntOpt( diff --git a/st2common/st2common/transport/publishers.py b/st2common/st2common/transport/publishers.py index 62484d4c46..af8558c10e 100644 --- a/st2common/st2common/transport/publishers.py +++ b/st2common/st2common/transport/publishers.py @@ -62,8 +62,11 @@ def publish(self, payload, exchange, routing_key="", compression=None): with Timer(key="amqp.pool_publisher.publish_with_retries." + exchange.name): with self.pool.acquire(block=True) as connection: + # Use the same retry settings from config for reconnection attempts retry_wrapper = ConnectionRetryWrapper( - cluster_size=self.cluster_size, logger=LOG + cluster_size=self.cluster_size, + logger=LOG, + ensure_max_retries=cfg.CONF.messaging.connection_retry_max_attempts, ) def do_publish(connection, channel): From 68844d83e7d4be28af64dce0b393eaab5ebe1561 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 17:34:37 -0400 Subject: [PATCH 06/24] consumer proper retry limit --- st2common/st2common/transport/consumers.py | 50 +++++++++ .../unit/test_cluster_retry_exhaustion.py | 70 ++++++++++++ .../tests/unit/test_consumer_retry_limits.py | 105 ++++++++++++++++++ 3 files changed, 225 insertions(+) create mode 100644 st2common/tests/unit/test_cluster_retry_exhaustion.py create mode 100644 st2common/tests/unit/test_consumer_retry_limits.py diff --git a/st2common/st2common/transport/consumers.py b/st2common/st2common/transport/consumers.py index 6f4cca7c87..49495281aa 100644 --- a/st2common/st2common/transport/consumers.py +++ b/st2common/st2common/transport/consumers.py @@ -42,6 +42,56 @@ def __init__(self, connection, queues, handler): self._queues = queues self._handler = handler + # Track connection retry attempts to enforce max_retries from config + self._connection_retry_count = 0 + self._max_connection_retries = cfg.CONF.messaging.connection_retry_max_attempts + + def on_connection_error(self, exc, interval): + """ + Override ConsumerMixin's connection error handler to enforce max retries. + + This prevents infinite retry loops when the broker is unavailable. + After max_retries attempts, we raise the exception to kill the consumer. + """ + self._connection_retry_count += 1 + + if ( + self._max_connection_retries > 0 + and self._connection_retry_count >= self._max_connection_retries + ): + LOG.error( + "Failed to connect to message broker after %d attempts. " + "Giving up. Error: %s", + self._connection_retry_count, + exc, + ) + # Raise the exception to stop the consumer + raise exc + + max_retries_display = ( + self._max_connection_retries if self._max_connection_retries > 0 else "∞" + ) + LOG.warning( + "Broker connection error (attempt %d/%s), " + "trying again in %.1f seconds: %s", + self._connection_retry_count, + max_retries_display, + interval, + exc, + ) + + def on_connection_revived(self): + """ + Reset retry counter when connection is successfully re-established. + """ + if self._connection_retry_count > 0: + LOG.info( + "Connection to message broker successfully re-established " + "after %d attempts", + self._connection_retry_count, + ) + self._connection_retry_count = 0 + def shutdown(self): self.should_stop = True self._dispatcher.shutdown() diff --git a/st2common/tests/unit/test_cluster_retry_exhaustion.py b/st2common/tests/unit/test_cluster_retry_exhaustion.py new file mode 100644 index 0000000000..50ec6f9f9a --- /dev/null +++ b/st2common/tests/unit/test_cluster_retry_exhaustion.py @@ -0,0 +1,70 @@ +# Copyright 2020 The StackStorm Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Test to verify ClusterRetryContext stops retrying after max attempts. +""" + +from __future__ import absolute_import +import unittest + +from st2common.transport.connection_retry_wrapper import ClusterRetryContext + + +class TestClusterRetryExhaustion(unittest.TestCase): + """Test that ClusterRetryContext respects max_retries""" + + def test_should_stop_returns_true_after_max_retries(self): + """Test that should_stop returns True after max_retries exhausted""" + context = ClusterRetryContext(cluster_size=2, max_retries=2) + + # Simulate failures on all nodes, cycling through the cluster + test_exc = Exception("Connection failed") + + # cluster_size=2, max_retries=2 means: 2 * (2+1) = 6 total attempts + # First cycle through cluster (2 nodes) + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop) # Node 1, attempt 1 + + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop) # Node 2, attempt 1 + + # Second cycle through cluster (2 nodes) + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop) # Node 1, attempt 2 + + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop) # Node 2, attempt 2 + + # Third cycle through cluster (2 nodes) + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop) # Node 1, attempt 3 + + should_stop, wait = context.should_stop(test_exc) + self.assertTrue(should_stop) # Node 2, attempt 3 - should stop here + + def test_should_stop_stops_at_exact_max_retries(self): + """Test that max_retries is respected exactly""" + context = ClusterRetryContext(cluster_size=3, max_retries=1) + + test_exc = Exception("Connection failed") + + # cluster_size=3, max_retries=1 means: 3 * (1+1) = 6 total attempts + for i in range(5): + should_stop, wait = context.should_stop(test_exc) + self.assertFalse(should_stop, f"Should not stop at attempt {i+1}") + + # 6th attempt should stop + should_stop, wait = context.should_stop(test_exc) + self.assertTrue(should_stop, "Should stop after 6 attempts") diff --git a/st2common/tests/unit/test_consumer_retry_limits.py b/st2common/tests/unit/test_consumer_retry_limits.py new file mode 100644 index 0000000000..61f69225a2 --- /dev/null +++ b/st2common/tests/unit/test_consumer_retry_limits.py @@ -0,0 +1,105 @@ +# Copyright 2020 The StackStorm Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Tests for QueueConsumer connection retry behavior. +""" + +from __future__ import absolute_import +import unittest +from unittest.mock import Mock + +from oslo_config import cfg + +from st2common.transport.consumers import QueueConsumer + + +class TestConsumerRetryLimits(unittest.TestCase): + """Test QueueConsumer respects connection retry limits""" + + def setUp(self): + """Reset config before each test""" + super(TestConsumerRetryLimits, self).setUp() + # Clear any config overrides from previous tests + try: + cfg.CONF.clear_override("connection_retry_max_attempts", group="messaging") + except: + pass + + def test_on_connection_error_raises_after_max_retries(self): + """Test that on_connection_error raises exception after max retries""" + cfg.CONF.set_override("connection_retry_max_attempts", 3, group="messaging") + + mock_connection = Mock() + mock_queues = [] + mock_handler = Mock() + + consumer = QueueConsumer(mock_connection, mock_queues, mock_handler) + + test_exc = ConnectionRefusedError(111, "ECONNREFUSED") + + # First 2 attempts should not raise + consumer.on_connection_error(test_exc, 1.0) + self.assertEqual(consumer._connection_retry_count, 1) + + consumer.on_connection_error(test_exc, 2.0) + self.assertEqual(consumer._connection_retry_count, 2) + + # 3rd attempt should raise + with self.assertRaises(ConnectionRefusedError): + consumer.on_connection_error(test_exc, 4.0) + + self.assertEqual(consumer._connection_retry_count, 3) + + def test_on_connection_revived_resets_counter(self): + """Test that on_connection_revived resets the retry counter""" + cfg.CONF.set_override("connection_retry_max_attempts", 5, group="messaging") + + mock_connection = Mock() + mock_queues = [] + mock_handler = Mock() + + consumer = QueueConsumer(mock_connection, mock_queues, mock_handler) + + test_exc = ConnectionRefusedError(111, "ECONNREFUSED") + + # Fail twice + consumer.on_connection_error(test_exc, 1.0) + consumer.on_connection_error(test_exc, 2.0) + self.assertEqual(consumer._connection_retry_count, 2) + + # Connection revived + consumer.on_connection_revived() + self.assertEqual(consumer._connection_retry_count, 0) + + # Can retry again from 0 + consumer.on_connection_error(test_exc, 1.0) + self.assertEqual(consumer._connection_retry_count, 1) + + def test_zero_max_retries_allows_infinite_retries(self): + """Test that setting max_retries to 0 allows infinite retries""" + cfg.CONF.set_override("connection_retry_max_attempts", 0, group="messaging") + + mock_connection = Mock() + mock_queues = [] + mock_handler = Mock() + + consumer = QueueConsumer(mock_connection, mock_queues, mock_handler) + + test_exc = ConnectionRefusedError(111, "ECONNREFUSED") + + # Should be able to retry many times without raising + for i in range(100): + consumer.on_connection_error(test_exc, 1.0) + self.assertEqual(consumer._connection_retry_count, i + 1) From d83d6be64d1d9d8a7119a715e439c1a37fa3d34c Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 18:19:39 -0400 Subject: [PATCH 07/24] configgen --- conf/st2.conf.sample | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 2fc025ed65..27a2eb0a86 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -235,7 +235,7 @@ connection_retry_interval_max = 30 connection_retry_interval_start = 1 # Increment for retry interval after each attempt (seconds). connection_retry_interval_step = 1 -# Maximum number of retry attempts for initial broker connection. This prevents infinite retry loops when the broker is unavailable. Set to 0 to retry indefinitely (not recommended). +# 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 From 430d2dc179b31a2cc609a16d5f8b4fdab671378e Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 21:31:00 -0400 Subject: [PATCH 08/24] just catch base and raise; add unit test for consumer --- st2common/st2common/persistence/base.py | 11 +++-------- .../st2common/transport/connection_retry_wrapper.py | 3 +-- st2common/st2common/transport/consumers.py | 6 ++---- 3 files changed, 6 insertions(+), 14 deletions(-) diff --git a/st2common/st2common/persistence/base.py b/st2common/st2common/persistence/base.py index 02aa50e718..8a789dae06 100644 --- a/st2common/st2common/persistence/base.py +++ b/st2common/st2common/persistence/base.py @@ -22,9 +22,6 @@ import six -from amqp import exceptions as amqp_exceptions -from kombu import exceptions as kombu_exceptions - from st2common import log as logging from st2common.exceptions.db import ( StackStormDBObjectConflictError, @@ -167,7 +164,7 @@ def insert( cls.dispatch_create_trigger(model_object) return model_object - except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + except Exception: # RabbitMQ connection error - rollback the database insert LOG.warning( "RabbitMQ publish failed for object %s, rolling back database insert", @@ -189,7 +186,6 @@ def add_or_update( # Late import to avoid very expensive in-direct import (~1 second) when this function # is not called / used from mongoengine import NotUniqueError - from kombu import exceptions as kombu_exceptions pre_persist_id = model_object.id @@ -235,7 +231,7 @@ def add_or_update( cls.dispatch_create_trigger(model_object) return model_object - except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + except Exception: # RabbitMQ connection error - rollback the database operation LOG.warning( "RabbitMQ publish failed for object %s, rolling back database operation", @@ -260,7 +256,6 @@ def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs): NOTE: If publish fails due to RabbitMQ connection errors, the database update will be rolled back by restoring the original object state. """ - from kombu import exceptions as kombu_exceptions # Save the original state before update for potential rollback original_object = cls.get_by_id(model_object.id) @@ -282,7 +277,7 @@ def update(cls, model_object, publish=True, dispatch_trigger=True, **kwargs): cls.dispatch_update_trigger(updated_object) return updated_object - except (kombu_exceptions.KombuError, amqp_exceptions.AMQPError): + except Exception: # RabbitMQ connection error - rollback the database update if original_object: LOG.warning( diff --git a/st2common/st2common/transport/connection_retry_wrapper.py b/st2common/st2common/transport/connection_retry_wrapper.py index 34ec00b140..d291dfe657 100644 --- a/st2common/st2common/transport/connection_retry_wrapper.py +++ b/st2common/st2common/transport/connection_retry_wrapper.py @@ -16,7 +16,6 @@ from __future__ import absolute_import import six -from kombu import exceptions as kombu_exceptions from st2common.util import concurrency @@ -165,7 +164,7 @@ def log_error_on_conn_failure(exc, interval): max_retries=self._ensure_max_retries, errback=log_error_on_conn_failure, ) - except kombu_exceptions.KombuError: + except Exception: self._logger.error("Failed to re-establish connection to RabbitMQ") raise finally: diff --git a/st2common/st2common/transport/consumers.py b/st2common/st2common/transport/consumers.py index 49495281aa..8361a665fe 100644 --- a/st2common/st2common/transport/consumers.py +++ b/st2common/st2common/transport/consumers.py @@ -165,11 +165,9 @@ class ActionsQueueConsumer(QueueConsumer): """ def __init__(self, connection, queues, handler): - self.connection = connection - - self._queues = queues - self._handler = handler + super(ActionsQueueConsumer, self).__init__(connection, queues, handler) + # Override the single dispatcher with two specialized dispatchers workflows_pool_size = cfg.CONF.actionrunner.workflows_pool_size actions_pool_size = cfg.CONF.actionrunner.actions_pool_size self._workflows_dispatcher = BufferedDispatcher( From bef61bbbdce0a7c9b7b4620977ed923e857dde1c Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 22:41:51 -0400 Subject: [PATCH 09/24] scheduler retries fixed --- .../st2common/services/triggerwatcher.py | 28 ++- .../transport/connection_retry_mixin.py | 103 +++++++++++ st2common/st2common/transport/consumers.py | 54 +----- .../tests/unit/test_connection_retry_mixin.py | 167 ++++++++++++++++++ 4 files changed, 298 insertions(+), 54 deletions(-) create mode 100644 st2common/st2common/transport/connection_retry_mixin.py create mode 100644 st2common/tests/unit/test_connection_retry_mixin.py diff --git a/st2common/st2common/services/triggerwatcher.py b/st2common/st2common/services/triggerwatcher.py index b82a46043a..bbb75562b5 100644 --- a/st2common/st2common/services/triggerwatcher.py +++ b/st2common/st2common/services/triggerwatcher.py @@ -23,13 +23,14 @@ from st2common.persistence.trigger import Trigger from st2common.transport import reactor, publishers from st2common.transport import utils as transport_utils +from st2common.transport.connection_retry_mixin import ConnectionRetryMixin from st2common.util import concurrency import st2common.util.queues as queue_utils LOG = logging.getLogger(__name__) -class TriggerWatcher(ConsumerMixin): +class TriggerWatcher(ConnectionRetryMixin, ConsumerMixin): sleep_interval = 0 # sleep to co-operatively yield after processing each message @@ -73,6 +74,9 @@ def __init__( self._load_thread = None self._updates_thread = None + # Initialize connection retry tracking from mixin + self._init_connection_retry() + self._handlers = { publishers.CREATE_RK: create_handler, publishers.UPDATE_RK: update_handler, @@ -125,13 +129,29 @@ def process_task(self, body, message): concurrency.sleep(self.sleep_interval) def start(self): + """ + Start the TriggerWatcher and establish RabbitMQ connection. + + The connection retry logic is handled by the ConsumerMixin.run() method + which will call on_connection_error() (from ConnectionRetryMixin) when + connection failures occur. + + Raises: + Exception: If connection cannot be established during initialization + """ try: self.connection = transport_utils.get_connection() self._updates_thread = concurrency.spawn(self.run) self._load_thread = concurrency.spawn(self._load_triggers_from_db) - except: - LOG.exception("Failed to start watcher.") - self.connection.release() + except Exception as e: + LOG.exception("Failed to start watcher: %s", six.text_type(e)) + # Only release connection if it was successfully created + if self.connection is not None: + try: + self.connection.release() + except Exception: + LOG.exception("Failed to release connection during cleanup") + raise def stop(self): try: diff --git a/st2common/st2common/transport/connection_retry_mixin.py b/st2common/st2common/transport/connection_retry_mixin.py new file mode 100644 index 0000000000..289cb97293 --- /dev/null +++ b/st2common/st2common/transport/connection_retry_mixin.py @@ -0,0 +1,103 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Mixin class for adding connection retry logic to Kombu ConsumerMixin classes. +""" + +from __future__ import absolute_import + +from oslo_config import cfg + +from st2common import log as logging + +__all__ = ["ConnectionRetryMixin"] + +LOG = logging.getLogger(__name__) + + +class ConnectionRetryMixin(object): + """ + Mixin that adds connection retry logic with configurable max attempts. + + This mixin prevents infinite retry loops when the message broker is unavailable + by enforcing the max_retries configuration from messaging.connection_retry_max_attempts. + + Classes using this mixin should be combined with kombu.mixins.ConsumerMixin. + + The ConsumerMixin.run() method has built-in retry logic that calls on_connection_error() + when connection fails, but by default it retries infinitely. This mixin overrides + on_connection_error() to stop after max_retries attempts. + + Example: + class MyConsumer(ConsumerMixin, ConnectionRetryMixin): + def __init__(self, connection): + self.connection = connection + self._init_connection_retry() + """ + + def _init_connection_retry(self): + """Initialize connection retry tracking. Call this in your __init__ method.""" + self._connection_retry_count = 0 + self._max_connection_retries = cfg.CONF.messaging.connection_retry_max_attempts + + def on_connection_error(self, exc, interval): + """ + Override ConsumerMixin's connection error handler to enforce max retries. + + This prevents infinite retry loops when the broker is unavailable. + After max_retries attempts, we raise the exception to kill the consumer. + + :param exc: The connection exception that occurred + :param interval: Time in seconds before next retry attempt + """ + self._connection_retry_count += 1 + + if ( + self._max_connection_retries > 0 + and self._connection_retry_count >= self._max_connection_retries + ): + LOG.error( + "Failed to connect to message broker after %d attempts. " + "Giving up. Error: %s", + self._connection_retry_count, + exc, + ) + # Raise the exception to stop the consumer + raise exc + + max_retries_display = ( + self._max_connection_retries if self._max_connection_retries > 0 else "∞" + ) + LOG.warning( + "Broker connection error (attempt %d/%s), " + "trying again in %.1f seconds: %s", + self._connection_retry_count, + max_retries_display, + interval, + exc, + ) + + def on_connection_revived(self): + """ + Reset retry counter when connection is successfully re-established. + """ + if self._connection_retry_count > 0: + LOG.info( + "Connection to message broker successfully re-established " + "after %d attempts", + self._connection_retry_count, + ) + self._connection_retry_count = 0 diff --git a/st2common/st2common/transport/consumers.py b/st2common/st2common/transport/consumers.py index 8361a665fe..82361f1aa8 100644 --- a/st2common/st2common/transport/consumers.py +++ b/st2common/st2common/transport/consumers.py @@ -21,6 +21,7 @@ from oslo_config import cfg from st2common import log as logging +from st2common.transport.connection_retry_mixin import ConnectionRetryMixin from st2common.util.greenpooldispatch import BufferedDispatcher from st2common.util import concurrency @@ -35,62 +36,15 @@ LOG = logging.getLogger(__name__) -class QueueConsumer(ConsumerMixin): +class QueueConsumer(ConnectionRetryMixin, ConsumerMixin): def __init__(self, connection, queues, handler): self.connection = connection self._dispatcher = BufferedDispatcher() self._queues = queues self._handler = handler - # Track connection retry attempts to enforce max_retries from config - self._connection_retry_count = 0 - self._max_connection_retries = cfg.CONF.messaging.connection_retry_max_attempts - - def on_connection_error(self, exc, interval): - """ - Override ConsumerMixin's connection error handler to enforce max retries. - - This prevents infinite retry loops when the broker is unavailable. - After max_retries attempts, we raise the exception to kill the consumer. - """ - self._connection_retry_count += 1 - - if ( - self._max_connection_retries > 0 - and self._connection_retry_count >= self._max_connection_retries - ): - LOG.error( - "Failed to connect to message broker after %d attempts. " - "Giving up. Error: %s", - self._connection_retry_count, - exc, - ) - # Raise the exception to stop the consumer - raise exc - - max_retries_display = ( - self._max_connection_retries if self._max_connection_retries > 0 else "∞" - ) - LOG.warning( - "Broker connection error (attempt %d/%s), " - "trying again in %.1f seconds: %s", - self._connection_retry_count, - max_retries_display, - interval, - exc, - ) - - def on_connection_revived(self): - """ - Reset retry counter when connection is successfully re-established. - """ - if self._connection_retry_count > 0: - LOG.info( - "Connection to message broker successfully re-established " - "after %d attempts", - self._connection_retry_count, - ) - self._connection_retry_count = 0 + # Initialize connection retry tracking from mixin + self._init_connection_retry() def shutdown(self): self.should_stop = True diff --git a/st2common/tests/unit/test_connection_retry_mixin.py b/st2common/tests/unit/test_connection_retry_mixin.py new file mode 100644 index 0000000000..5f8a37a952 --- /dev/null +++ b/st2common/tests/unit/test_connection_retry_mixin.py @@ -0,0 +1,167 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import + +import unittest +import mock + +from oslo_config import cfg + +from st2common.transport.connection_retry_mixin import ConnectionRetryMixin +from st2tests.config import parse_args + +parse_args() + + +class MockConsumer(ConnectionRetryMixin): + """Mock consumer class for testing the mixin.""" + + def __init__(self): + self._init_connection_retry() + + +class ConnectionRetryMixinTestCase(unittest.TestCase): + def setUp(self): + # Store original config value + self._original_max_retries = cfg.CONF.messaging.connection_retry_max_attempts + + def tearDown(self): + # Restore original config value + cfg.CONF.set_override( + "connection_retry_max_attempts", + self._original_max_retries, + group="messaging", + ) + + def test_init_connection_retry(self): + """Test that initialization sets up retry tracking correctly.""" + consumer = MockConsumer() + self.assertEqual(consumer._connection_retry_count, 0) + self.assertEqual( + consumer._max_connection_retries, + cfg.CONF.messaging.connection_retry_max_attempts, + ) + + def test_on_connection_error_within_limit(self): + """Test that connection errors within retry limit are logged but don't raise.""" + cfg.CONF.set_override("connection_retry_max_attempts", 5, group="messaging") + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # Should not raise for first few attempts + for i in range(4): + consumer.on_connection_error(exc, 1.0) + self.assertEqual(consumer._connection_retry_count, i + 1) + + def test_on_connection_error_exceeds_limit(self): + """Test that connection errors exceeding retry limit raise exception.""" + cfg.CONF.set_override("connection_retry_max_attempts", 3, group="messaging") + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # Should not raise for attempts within limit + consumer.on_connection_error(exc, 1.0) + consumer.on_connection_error(exc, 1.0) + self.assertEqual(consumer._connection_retry_count, 2) + + # Should raise when exceeding limit + with self.assertRaises(Exception) as ctx: + consumer.on_connection_error(exc, 1.0) + + self.assertEqual(str(ctx.exception), "Connection failed") + self.assertEqual(consumer._connection_retry_count, 3) + + def test_on_connection_error_unlimited_retries(self): + """Test that setting max_retries to 0 allows unlimited retries.""" + cfg.CONF.set_override("connection_retry_max_attempts", 0, group="messaging") + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # Should not raise even after many attempts + for i in range(100): + consumer.on_connection_error(exc, 1.0) + self.assertEqual(consumer._connection_retry_count, i + 1) + + def test_on_connection_revived(self): + """Test that connection revival resets retry counter.""" + cfg.CONF.set_override("connection_retry_max_attempts", 5, group="messaging") + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # Simulate some failed attempts + consumer.on_connection_error(exc, 1.0) + consumer.on_connection_error(exc, 1.0) + consumer.on_connection_error(exc, 1.0) + self.assertEqual(consumer._connection_retry_count, 3) + + # Connection revived should reset counter + consumer.on_connection_revived() + self.assertEqual(consumer._connection_retry_count, 0) + + def test_on_connection_revived_no_previous_errors(self): + """Test that connection revival with no previous errors is safe.""" + consumer = MockConsumer() + self.assertEqual(consumer._connection_retry_count, 0) + + # Should not raise or cause issues + consumer.on_connection_revived() + self.assertEqual(consumer._connection_retry_count, 0) + + @mock.patch("st2common.transport.connection_retry_mixin.LOG") + def test_logging_on_error(self, mock_log): + """Test that appropriate log messages are generated on connection errors.""" + cfg.CONF.set_override("connection_retry_max_attempts", 3, group="messaging") + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # First error should log warning + consumer.on_connection_error(exc, 1.0) + self.assertTrue(mock_log.warning.called) + + # Reset mock + mock_log.reset_mock() + + # Error exceeding limit should log error + consumer.on_connection_error(exc, 1.0) + + # Third call should raise and log error + with self.assertRaises(Exception): + consumer.on_connection_error(exc, 1.0) + + self.assertTrue(mock_log.error.called) + + @mock.patch("st2common.transport.connection_retry_mixin.LOG") + def test_logging_on_revival(self, mock_log): + """Test that log message is generated when connection is revived.""" + consumer = MockConsumer() + + exc = Exception("Connection failed") + + # Simulate some failures + consumer.on_connection_error(exc, 1.0) + consumer.on_connection_error(exc, 1.0) + + # Reset mock to check revival logging + mock_log.reset_mock() + + # Connection revived should log info + consumer.on_connection_revived() + self.assertTrue(mock_log.info.called) From 69b4775c9c968b52ef9342a677f65b5ba12d3e5e Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 22:43:23 -0400 Subject: [PATCH 10/24] remove no rollback on other exceptions test --- .../tests/unit/test_persistence_rollback.py | 23 ------------------- 1 file changed, 23 deletions(-) diff --git a/st2common/tests/unit/test_persistence_rollback.py b/st2common/tests/unit/test_persistence_rollback.py index 947bbf1597..b3008147de 100644 --- a/st2common/tests/unit/test_persistence_rollback.py +++ b/st2common/tests/unit/test_persistence_rollback.py @@ -64,29 +64,6 @@ def test_update_rollback_on_kombu_error(self): self.assertEqual(retrieved.name, original_name) self.assertNotEqual(retrieved.name, new_name) - def test_update_no_rollback_on_other_exceptions(self): - """Test that update() does NOT rollback on non-RabbitMQ exceptions""" - # Create initial object - obj = FakeModelDB(name=uuid.uuid4().hex, context={"value": "original"}) - obj = self.access.add_or_update(obj, publish=False) - - # Mock publish_update at class level to raise a generic exception - with mock.patch.object( - FakeModel, "publish_update", side_effect=ValueError("Some other error") - ): - # Try to update with a new name - new_name = uuid.uuid4().hex - obj.name = new_name - - # Update should propagate the non-RabbitMQ exception - with self.assertRaises(ValueError): - self.access.update(obj, publish=True, set__name=new_name) - - # Since ValueError is not a KombuError, no rollback occurs - # The DB change remains - retrieved = self.access.get_by_id(str(obj.id)) - self.assertEqual(retrieved.name, new_name) - def test_update_success_no_rollback(self): """Test that successful update() with publish does not trigger rollback""" # Create initial object From 33d48f37ea5a852731b56213b746b85ecc365410 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Tue, 19 May 2026 08:58:32 -0400 Subject: [PATCH 11/24] remove exception catch --- st2common/st2common/services/triggerwatcher.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/st2common/st2common/services/triggerwatcher.py b/st2common/st2common/services/triggerwatcher.py index bbb75562b5..126cc142bc 100644 --- a/st2common/st2common/services/triggerwatcher.py +++ b/st2common/st2common/services/triggerwatcher.py @@ -147,10 +147,7 @@ def start(self): LOG.exception("Failed to start watcher: %s", six.text_type(e)) # Only release connection if it was successfully created if self.connection is not None: - try: - self.connection.release() - except Exception: - LOG.exception("Failed to release connection during cleanup") + self.connection.release() raise def stop(self): From c2a5c2ea92c69147d9c98ecbcf49bcae30216eee Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Tue, 19 May 2026 11:15:50 -0400 Subject: [PATCH 12/24] conflict resolution --- st2actions/st2actions/cmd/scheduler.py | 36 +++- .../tests/unit/test_scheduler_entrypoint.py | 21 +- ..._scheduler_shutdown_on_rabbitmq_failure.py | 198 ++++++++++++++++++ st2reactor/st2reactor/cmd/rulesengine.py | 28 ++- ...ulesengine_shutdown_on_rabbitmq_failure.py | 107 ++++++++++ 5 files changed, 377 insertions(+), 13 deletions(-) create mode 100644 st2actions/tests/unit/test_scheduler_shutdown_on_rabbitmq_failure.py create mode 100644 st2reactor/tests/unit/test_rulesengine_shutdown_on_rabbitmq_failure.py diff --git a/st2actions/st2actions/cmd/scheduler.py b/st2actions/st2actions/cmd/scheduler.py index 465d067f31..22ddc2bc3b 100644 --- a/st2actions/st2actions/cmd/scheduler.py +++ b/st2actions/st2actions/cmd/scheduler.py @@ -95,8 +95,38 @@ def _run_scheduler(): 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: + for thread, name in threads_to_monitor: + if thread.dead: + # Thread died - try to get the exception if it raised one + try: + thread.wait() # This will raise 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 + # Thread completed successfully (shouldn't happen in normal operation) + 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 +151,7 @@ def _run_scheduler(): except: LOG.exception("Unable to shutdown scheduler.") - return 1 + raise return 0 diff --git a/st2actions/tests/unit/test_scheduler_entrypoint.py b/st2actions/tests/unit/test_scheduler_entrypoint.py index 2862ba2b3c..7aa101a702 100644 --- a/st2actions/tests/unit/test_scheduler_entrypoint.py +++ b/st2actions/tests/unit/test_scheduler_entrypoint.py @@ -48,9 +48,12 @@ class SchedulerServiceEntryPointTestCase(CleanDbTestCase): @mock.patch("st2actions.cmd.scheduler.LOG") def test_service_exits_correctly_on_fatal_exception_in_handler_run(self, mock_log): run_thread = eventlet.spawn(_run_scheduler) - result = run_thread.wait() - self.assertEqual(result, 1) + # The scheduler now raises exceptions instead of returning 1 + with self.assertRaises(Exception) as cm: + run_thread.wait() + + self.assertIn("handler run exception", str(cm.exception)) mock_log_exception_call = mock_log.exception.call_args_list[0][0][0] self.assertIn("Scheduler unexpectedly stopped", mock_log_exception_call) @@ -63,9 +66,12 @@ def test_service_exits_correctly_on_fatal_exception_in_handler_cleanup( self, mock_log ): run_thread = eventlet.spawn(_run_scheduler) - result = run_thread.wait() - self.assertEqual(result, 1) + # The scheduler now raises exceptions instead of returning 1 + with self.assertRaises(Exception) as cm: + run_thread.wait() + + self.assertIn("handler clean exception", str(cm.exception)) mock_log_exception_call = mock_log.exception.call_args_list[0][0][0] self.assertIn("Scheduler unexpectedly stopped", mock_log_exception_call) @@ -76,9 +82,12 @@ def test_service_exits_correctly_on_fatal_exception_in_entrypoint_start( self, mock_log ): run_thread = eventlet.spawn(_run_scheduler) - result = run_thread.wait() - self.assertEqual(result, 1) + # The scheduler now raises exceptions instead of returning 1 + with self.assertRaises(Exception) as cm: + run_thread.wait() + + self.assertIn("entrypoint start exception", str(cm.exception)) mock_log_exception_call = mock_log.exception.call_args_list[0][0][0] self.assertIn("Scheduler unexpectedly stopped", mock_log_exception_call) diff --git a/st2actions/tests/unit/test_scheduler_shutdown_on_rabbitmq_failure.py b/st2actions/tests/unit/test_scheduler_shutdown_on_rabbitmq_failure.py new file mode 100644 index 0000000000..0ed3a84296 --- /dev/null +++ b/st2actions/tests/unit/test_scheduler_shutdown_on_rabbitmq_failure.py @@ -0,0 +1,198 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Test that verifies the scheduler shuts down completely when RabbitMQ +connection failures exhaust retry attempts. +""" + +from __future__ import absolute_import + +import eventlet +import mock +from kombu import exceptions as kombu_exceptions + +from st2tests.base import DbTestCase +import st2tests.config as tests_config +from st2actions.cmd import scheduler + + +class SchedulerShutdownOnRabbitMQFailureTestCase(DbTestCase): + """ + Test case to verify that when the scheduler's entrypoint consumer fails + due to RabbitMQ connection exhaustion, the entire scheduler process + shuts down cleanly. + """ + + def setUp(self): + super(SchedulerShutdownOnRabbitMQFailureTestCase, self).setUp() + tests_config.reset() + tests_config.parse_args() + + @mock.patch("st2actions.scheduler.entrypoint.get_scheduler_entrypoint") + @mock.patch("st2actions.scheduler.handler.get_handler") + def test_scheduler_shuts_down_when_entrypoint_consumer_fails( + self, mock_get_handler, mock_get_entrypoint + ): + """ + Test that when the entrypoint consumer thread fails with KombuError + after exhausting retry attempts, the scheduler: + 1. Detects the failure via eventlet.wait_all() + 2. Calls shutdown() on both handler and entrypoint + 3. Re-raises the exception to exit the process + """ + # Create mock handler with threads that would run forever + mock_handler = mock.MagicMock() + mock_handler_main_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_handler_cleanup_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_handler._main_thread = mock_handler_main_thread + mock_handler._cleanup_thread = mock_handler_cleanup_thread + mock_get_handler.return_value = mock_handler + + # Create mock entrypoint with a consumer thread that fails immediately + mock_entrypoint = mock.MagicMock() + + # Simulate consumer thread failing with OperationalError (RabbitMQ connection exhausted) + def failing_consumer(): + raise kombu_exceptions.OperationalError("[Errno 111] ECONNREFUSED") + + mock_entrypoint_consumer_thread = eventlet.spawn(failing_consumer) + mock_entrypoint._consumer_thread = mock_entrypoint_consumer_thread + mock_get_entrypoint.return_value = mock_entrypoint + + # Run the scheduler and expect it to raise the exception + with self.assertRaises(kombu_exceptions.OperationalError) as cm: + scheduler._run_scheduler() + + # Verify the exception message + self.assertIn("ECONNREFUSED", str(cm.exception)) + + # Verify that shutdown was called on both components + mock_handler.shutdown.assert_called_once() + mock_entrypoint.shutdown.assert_called_once() + + @mock.patch("st2actions.scheduler.entrypoint.get_scheduler_entrypoint") + @mock.patch("st2actions.scheduler.handler.get_handler") + def test_scheduler_shuts_down_when_handler_main_thread_fails( + self, mock_get_handler, mock_get_entrypoint + ): + """ + Test that when the handler's main thread fails, the scheduler + detects it and shuts down both components. + """ + # Create mock handler with main thread that fails + mock_handler = mock.MagicMock() + + def failing_main_thread(): + raise RuntimeError("Handler main thread failed") + + mock_handler_main_thread = eventlet.spawn(failing_main_thread) + mock_handler_cleanup_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_handler._main_thread = mock_handler_main_thread + mock_handler._cleanup_thread = mock_handler_cleanup_thread + mock_get_handler.return_value = mock_handler + + # Create mock entrypoint that would run forever + mock_entrypoint = mock.MagicMock() + mock_entrypoint_consumer_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_entrypoint._consumer_thread = mock_entrypoint_consumer_thread + mock_get_entrypoint.return_value = mock_entrypoint + + # Run the scheduler and expect it to raise the exception + with self.assertRaises(RuntimeError) as cm: + scheduler._run_scheduler() + + # Verify the exception message + self.assertIn("Handler main thread failed", str(cm.exception)) + + # Verify that shutdown was called on both components + mock_handler.shutdown.assert_called_once() + mock_entrypoint.shutdown.assert_called_once() + + @mock.patch("st2actions.scheduler.entrypoint.get_scheduler_entrypoint") + @mock.patch("st2actions.scheduler.handler.get_handler") + def test_scheduler_shuts_down_when_handler_cleanup_thread_fails( + self, mock_get_handler, mock_get_entrypoint + ): + """ + Test that when the handler's cleanup thread fails, the scheduler + detects it and shuts down both components. + """ + # Create mock handler with cleanup thread that fails + mock_handler = mock.MagicMock() + mock_handler_main_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + + def failing_cleanup_thread(): + raise RuntimeError("Handler cleanup thread failed") + + mock_handler_cleanup_thread = eventlet.spawn(failing_cleanup_thread) + mock_handler._main_thread = mock_handler_main_thread + mock_handler._cleanup_thread = mock_handler_cleanup_thread + mock_get_handler.return_value = mock_handler + + # Create mock entrypoint that would run forever + mock_entrypoint = mock.MagicMock() + mock_entrypoint_consumer_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_entrypoint._consumer_thread = mock_entrypoint_consumer_thread + mock_get_entrypoint.return_value = mock_entrypoint + + # Run the scheduler and expect it to raise the exception + with self.assertRaises(RuntimeError) as cm: + scheduler._run_scheduler() + + # Verify the exception message + self.assertIn("Handler cleanup thread failed", str(cm.exception)) + + # Verify that shutdown was called on both components + mock_handler.shutdown.assert_called_once() + mock_entrypoint.shutdown.assert_called_once() + + @mock.patch("st2actions.scheduler.entrypoint.get_scheduler_entrypoint") + @mock.patch("st2actions.scheduler.handler.get_handler") + def test_scheduler_connection_error_propagates( + self, mock_get_handler, mock_get_entrypoint + ): + """ + Test that ConnectionError (a subclass of OperationalError) also + triggers proper shutdown. + """ + # Create mock handler with threads that would run forever + mock_handler = mock.MagicMock() + mock_handler_main_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_handler_cleanup_thread = eventlet.spawn(lambda: eventlet.sleep(1000)) + mock_handler._main_thread = mock_handler_main_thread + mock_handler._cleanup_thread = mock_handler_cleanup_thread + mock_get_handler.return_value = mock_handler + + # Create mock entrypoint with consumer thread that fails with ConnectionError + mock_entrypoint = mock.MagicMock() + + def failing_consumer(): + raise kombu_exceptions.ConnectionError("Connection lost") + + mock_entrypoint_consumer_thread = eventlet.spawn(failing_consumer) + mock_entrypoint._consumer_thread = mock_entrypoint_consumer_thread + mock_get_entrypoint.return_value = mock_entrypoint + + # Run the scheduler and expect it to raise the exception + with self.assertRaises(kombu_exceptions.ConnectionError) as cm: + scheduler._run_scheduler() + + # Verify the exception message + self.assertIn("Connection lost", str(cm.exception)) + + # Verify that shutdown was called on both components + mock_handler.shutdown.assert_called_once() + mock_entrypoint.shutdown.assert_called_once() diff --git a/st2reactor/st2reactor/cmd/rulesengine.py b/st2reactor/st2reactor/cmd/rulesengine.py index 3629345324..bdce9457ba 100644 --- a/st2reactor/st2reactor/cmd/rulesengine.py +++ b/st2reactor/st2reactor/cmd/rulesengine.py @@ -62,14 +62,34 @@ def _run_worker(): try: rules_engine_worker.start() - return rules_engine_worker.wait() + + # Monitor the worker thread - if it dies/fails, we need to exit cleanly + import eventlet + + # Poll the worker thread to detect failures + while True: + if rules_engine_worker.thread and rules_engine_worker.thread.dead: + # Thread died - try to get the exception if it raised one + try: + rules_engine_worker.thread.wait() # This will raise if thread raised + except Exception as e: + LOG.error("RulesEngine worker thread failed: %s", e) + raise + # Thread completed successfully (shouldn't happen in normal operation) + LOG.info("RulesEngine worker thread completed") + return 0 + + # Sleep briefly to avoid tight loop + eventlet.sleep(0.1) except (KeyboardInterrupt, SystemExit): LOG.info("(PID=%s) RulesEngine stopped.", os.getpid()) deregister_service(RULESENGINE) rules_engine_worker.shutdown() + raise except: - LOG.exception("(PID:%s) RulesEngine quit due to exception.", os.getpid()) - return 1 + LOG.exception("(PID=%s) RulesEngine quit due to exception.", os.getpid()) + rules_engine_worker.shutdown() + raise return 0 @@ -82,6 +102,6 @@ def main(): sys.exit(exit_code) except: LOG.exception("(PID=%s) RulesEngine quit due to exception.", os.getpid()) - return 1 + raise finally: _teardown() diff --git a/st2reactor/tests/unit/test_rulesengine_shutdown_on_rabbitmq_failure.py b/st2reactor/tests/unit/test_rulesengine_shutdown_on_rabbitmq_failure.py new file mode 100644 index 0000000000..04e510e2c4 --- /dev/null +++ b/st2reactor/tests/unit/test_rulesengine_shutdown_on_rabbitmq_failure.py @@ -0,0 +1,107 @@ +# Copyright 2020 The StackStorm Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Tests to verify that st2reactor rulesengine properly shuts down when RabbitMQ +connection failures occur, rather than hanging indefinitely. +""" + +from __future__ import absolute_import + +import eventlet +import mock + +from kombu.exceptions import OperationalError as KombuOperationalError + +from st2reactor.cmd.rulesengine import _run_worker +from st2reactor.rules.worker import TriggerInstanceDispatcher +from st2tests.base import CleanDbTestCase + +__all__ = ["RulesEngineShutdownOnRabbitMQFailureTestCase"] + + +class RulesEngineShutdownOnRabbitMQFailureTestCase(CleanDbTestCase): + """ + Test cases to ensure the rulesengine service exits cleanly when RabbitMQ + connection issues occur, preventing infinite hangs. + """ + + @mock.patch("st2reactor.rules.worker.transport_utils.get_connection") + def test_rulesengine_connection_error_propagates(self, mock_get_connection): + """ + Test that connection errors during worker initialization propagate + and cause the service to exit. + """ + # Simulate connection failure during worker.get_worker() + mock_get_connection.side_effect = KombuOperationalError("Connection refused") + + # Run the worker in a greenthread + run_thread = eventlet.spawn(_run_worker) + + # The worker should raise the connection error + with self.assertRaises(KombuOperationalError) as cm: + run_thread.wait() + + self.assertIn("Connection refused", str(cm.exception)) + + @mock.patch.object(TriggerInstanceDispatcher, "start") + @mock.patch.object(TriggerInstanceDispatcher, "shutdown") + def test_rulesengine_shuts_down_when_worker_thread_fails( + self, mock_shutdown, mock_start + ): + """ + Test that when the worker thread fails with an exception, + the rulesengine detects it, shuts down cleanly, and raises the exception. + """ + + def mock_start_that_fails(): + # Simulate the worker thread starting but then failing + eventlet.sleep(0.1) + raise RuntimeError("Worker thread failed") + + mock_start.side_effect = mock_start_that_fails + + # Run the worker + run_thread = eventlet.spawn(_run_worker) + + # Should raise the RuntimeError from the worker thread + with self.assertRaises(RuntimeError) as cm: + run_thread.wait() + + self.assertIn("Worker thread failed", str(cm.exception)) + + # Shutdown should have been called + mock_shutdown.assert_called_once() + + @mock.patch("st2reactor.rules.worker.transport_utils.get_connection") + @mock.patch.object(TriggerInstanceDispatcher, "shutdown") + def test_rulesengine_handles_connection_retry_exhaustion( + self, mock_shutdown, mock_get_connection + ): + """ + Test that when connection retries are exhausted (after max attempts), + the rulesengine exits cleanly with an exception. + """ + # Mock connection to fail during worker initialization + mock_get_connection.side_effect = KombuOperationalError( + "Failed to connect after 10 attempts" + ) + + run_thread = eventlet.spawn(_run_worker) + + # Should raise the connection error + with self.assertRaises(KombuOperationalError) as cm: + run_thread.wait() + + self.assertIn("Failed to connect", str(cm.exception)) From f86b6ebfda84ea9fa5bdd2d80f0e1b548250d2bb Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Mon, 18 May 2026 21:32:17 -0400 Subject: [PATCH 13/24] conflict resolution --- .../tests/unit/test_action_runner_worker.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/st2actions/tests/unit/test_action_runner_worker.py b/st2actions/tests/unit/test_action_runner_worker.py index 8477281b97..bffa1c7522 100644 --- a/st2actions/tests/unit/test_action_runner_worker.py +++ b/st2actions/tests/unit/test_action_runner_worker.py @@ -17,6 +17,8 @@ from unittest import TestCase from mock import Mock +from oslo_config import cfg + from st2common.transport.consumers import ActionsQueueConsumer from st2common.models.db.liveaction import LiveActionDB @@ -31,6 +33,36 @@ def setUpClass(cls): super().setUpClass() tests_config.parse_args() + def test_connection_retry_attributes_initialized(self): + """Test that ActionsQueueConsumer properly inherits connection retry attributes""" + handler = Mock() + handler.message_type = LiveActionDB + consumer = ActionsQueueConsumer(connection=None, queues=None, handler=handler) + + # Verify inherited attributes from QueueConsumer are present + self.assertTrue(hasattr(consumer, "_connection_retry_count")) + self.assertTrue(hasattr(consumer, "_max_connection_retries")) + self.assertEqual(consumer._connection_retry_count, 0) + self.assertEqual( + consumer._max_connection_retries, + cfg.CONF.messaging.connection_retry_max_attempts, + ) + + def test_on_connection_revived_works(self): + """Test that on_connection_revived method works correctly for ActionsQueueConsumer""" + handler = Mock() + handler.message_type = LiveActionDB + consumer = ActionsQueueConsumer(connection=None, queues=None, handler=handler) + + # Simulate some failed connection attempts + consumer._connection_retry_count = 3 + + # Call inherited method + consumer.on_connection_revived() + + # Should reset counter to 0 + self.assertEqual(consumer._connection_retry_count, 0) + def test_process_right_dispatcher_is_used(self): handler = Mock() handler.message_type = LiveActionDB From 466d9c2a797b62b56f1eba617ab5dd2d8142c271 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Tue, 19 May 2026 15:54:18 -0400 Subject: [PATCH 14/24] fix conflicts --- st2actions/st2actions/cmd/workflow_engine.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/st2actions/st2actions/cmd/workflow_engine.py b/st2actions/st2actions/cmd/workflow_engine.py index ac6626afde..45ab4286f7 100644 --- a/st2actions/st2actions/cmd/workflow_engine.py +++ b/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(): From 286432e8adee9fca6e2db5baa709cf5110dc8b2c Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Wed, 20 May 2026 14:19:26 -0400 Subject: [PATCH 15/24] fix conflicts --- st2actions/st2actions/cmd/scheduler.py | 4 + st2actions/st2actions/scheduler/handler.py | 61 ++++ .../unit/test_scheduler_bootstrap_recovery.py | 300 ++++++++++++++++++ st2reactor/st2reactor/cmd/rulesengine.py | 7 +- 4 files changed, 370 insertions(+), 2 deletions(-) create mode 100644 st2actions/tests/unit/test_scheduler_bootstrap_recovery.py diff --git a/st2actions/st2actions/cmd/scheduler.py b/st2actions/st2actions/cmd/scheduler.py index 22ddc2bc3b..27db9f5425 100644 --- a/st2actions/st2actions/cmd/scheduler.py +++ b/st2actions/st2actions/cmd/scheduler.py @@ -91,6 +91,10 @@ 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() diff --git a/st2actions/st2actions/scheduler/handler.py b/st2actions/st2actions/scheduler/handler.py index 5d080fbafa..59cd730edc 100644 --- a/st2actions/st2actions/scheduler/handler.py +++ b/st2actions/st2actions/scheduler/handler.py @@ -30,6 +30,7 @@ from st2common.services import coordination as coordination_service from st2common.services import executions as execution_service from st2common.services import policies as policy_service +from st2common.models.db.execution_queue import ActionExecutionSchedulingQueueItemDB from st2common.persistence.execution import ActionExecution from st2common.persistence.liveaction import LiveAction from st2common.persistence.execution_queue import ActionExecutionSchedulingQueue @@ -37,6 +38,7 @@ from st2common.metrics import base as metrics from st2common.exceptions import db as db_exc + __all__ = ["ActionExecutionSchedulingQueueHandler", "get_handler"] @@ -146,6 +148,65 @@ def _fix_missing_action_execution_id(self): entry.action_execution_id = str(execution_db.id) ActionExecutionSchedulingQueue.add_or_update(entry, publish=False) + def _bootstrap_missing_scheduling_queue_items(self): + """ + Bootstrap ActionExecutionSchedulingQueue entries for LiveActions in 'requested' + status that don't have a corresponding queue entry. This handles recovery from + RabbitMQ failures where the SchedulerEntrypoint never received the message. + + Note: We only handle 'requested' status because: + - 'delayed' status already has queue entries (created at initial request time) + - Policy-delayed executions update existing queue entries + """ + requested_liveactions = ( + LiveAction.query(status=action_constants.LIVEACTION_STATUS_REQUESTED) or [] + ) + + for liveaction_db in requested_liveactions: + # Check if this liveaction already has a queue entry + ex_que_qry = {"liveaction_id": str(liveaction_db.id)} + existing_queue_items = ( + ActionExecutionSchedulingQueue.query(**ex_que_qry) or [] + ) + + if len(existing_queue_items) > 0: + # Queue entry already exists, skip + continue + + # Get the associated ActionExecution + execution_db = ActionExecution.get(liveaction_id=str(liveaction_db.id)) + + # Skip if no execution exists (orphaned liveaction) + if not execution_db: + LOG.warning( + 'Skipping LiveAction "%s" - no ActionExecution found', + str(liveaction_db.id), + ) + continue + + # Create the missing queue entry + execution_queue_item_db = ActionExecutionSchedulingQueueItemDB() + execution_queue_item_db.action_execution_id = str(execution_db.id) + execution_queue_item_db.liveaction_id = str(liveaction_db.id) + execution_queue_item_db.original_start_timestamp = ( + liveaction_db.start_timestamp + ) + execution_queue_item_db.scheduled_start_timestamp = ( + date.append_milliseconds_to_time( + liveaction_db.start_timestamp, liveaction_db.delay or 0 + ) + ) + execution_queue_item_db.delay = liveaction_db.delay + + ActionExecutionSchedulingQueue.add_or_update( + execution_queue_item_db, publish=False + ) + LOG.info( + '[%s] Bootstrapped missing scheduling queue entry for LiveAction "%s".', + str(execution_db.id), + str(liveaction_db.id), + ) + # TODO: Remove this function for cleanup policy-delayed in v3.2. # This is a temporary cleanup to remove executions in deprecated policy-delayed status. def _cleanup_policy_delayed(self): diff --git a/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py b/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py new file mode 100644 index 0000000000..f3b3f152d4 --- /dev/null +++ b/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py @@ -0,0 +1,300 @@ +# Copyright 2020 The StackStorm Authors. +# Copyright 2019 Extreme Networks, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Test that verifies the scheduler bootstrap recovery mechanism for handling +LiveActions stuck in 'requested' status due to RabbitMQ failures. +""" + +from __future__ import absolute_import + +from st2common.constants import action as action_constants +from st2common.models.db.liveaction import LiveActionDB +from st2common.persistence.liveaction import LiveAction +from st2common.persistence.execution_queue import ActionExecutionSchedulingQueue +from st2common.util import date as date_utils +from st2tests.base import DbTestCase +from st2tests.fixturesloader import FixturesLoader +import st2tests.config as tests_config +from st2actions.scheduler.handler import ActionExecutionSchedulingQueueHandler + + +FIXTURES_PACK = "generic" +TEST_FIXTURES = {"runners": ["run-local.yaml"], "actions": ["local.yaml"]} + + +class SchedulerBootstrapRecoveryTestCase(DbTestCase): + """ + Test case to verify that the scheduler's bootstrap recovery mechanism + can recover LiveActions stuck in 'requested' status due to RabbitMQ failures. + """ + + @classmethod + def setUpClass(cls): + super(SchedulerBootstrapRecoveryTestCase, cls).setUpClass() + tests_config.reset() + tests_config.parse_args() + loader = FixturesLoader() + loader.save_fixtures_to_db( + fixtures_pack=FIXTURES_PACK, fixtures_dict=TEST_FIXTURES + ) + + def setUp(self): + super(SchedulerBootstrapRecoveryTestCase, self).setUp() + + def test_bootstrap_recovers_requested_liveaction_without_queue_entry(self): + """ + Test that _bootstrap_missing_scheduling_queue_items creates a queue entry + for a LiveAction in 'requested' status that doesn't have one. + + This simulates the scenario where RabbitMQ was down when the action was + created, so the SchedulerEntrypoint never consumed the message. + """ + # Create a LiveAction in 'requested' status directly in the database + # (simulating what happens when publish_request fails due to RabbitMQ being down) + liveaction_db = LiveActionDB() + liveaction_db.status = action_constants.LIVEACTION_STATUS_REQUESTED + liveaction_db.action = "core.local" + liveaction_db.parameters = {"cmd": "echo 'test'"} + liveaction_db.start_timestamp = date_utils.get_datetime_utc_now() + + # Save directly to DB without publishing (simulating RabbitMQ failure) + liveaction_db = LiveAction.add_or_update(liveaction_db, publish=False) + + # Create the associated ActionExecution + from st2common.services import executions + from st2common.util import action_db as action_utils + + action_db = action_utils.get_action_by_ref("core.local") + runnertype_db = action_utils.get_runnertype_by_name( + action_db.runner_type["name"] + ) + execution_db = executions.create_execution_object( + liveaction=liveaction_db, + action_db=action_db, + runnertype_db=runnertype_db, + publish=False, + ) + + # Verify no queue entry exists + queue_items = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual(len(queue_items), 0, "Queue entry should not exist initially") + + # Run the bootstrap recovery + handler = ActionExecutionSchedulingQueueHandler() + handler._bootstrap_missing_scheduling_queue_items() + + # Verify queue entry was created + queue_items = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual( + len(queue_items), 1, "Queue entry should be created by bootstrap" + ) + + queue_item = queue_items[0] + self.assertEqual(queue_item.liveaction_id, str(liveaction_db.id)) + self.assertEqual(queue_item.action_execution_id, str(execution_db.id)) + self.assertIsNotNone(queue_item.scheduled_start_timestamp) + self.assertIsNotNone(queue_item.original_start_timestamp) + + def test_bootstrap_skips_liveaction_with_existing_queue_entry(self): + """ + Test that _bootstrap_missing_scheduling_queue_items doesn't create duplicate + queue entries for LiveActions that already have them. + """ + # Create a LiveAction with a queue entry (normal case) + liveaction_db = LiveActionDB() + liveaction_db.status = action_constants.LIVEACTION_STATUS_REQUESTED + liveaction_db.action = "core.local" + liveaction_db.parameters = {"cmd": "echo 'test2'"} + liveaction_db.start_timestamp = date_utils.get_datetime_utc_now() + liveaction_db = LiveAction.add_or_update(liveaction_db, publish=False) + + # Create execution and queue entry + from st2common.services import executions + from st2common.util import action_db as action_utils + from st2common.models.db.execution_queue import ( + ActionExecutionSchedulingQueueItemDB, + ) + + action_db = action_utils.get_action_by_ref("core.local") + runnertype_db = action_utils.get_runnertype_by_name( + action_db.runner_type["name"] + ) + execution_db = executions.create_execution_object( + liveaction=liveaction_db, + action_db=action_db, + runnertype_db=runnertype_db, + publish=False, + ) + + # Manually create queue entry + queue_item_db = ActionExecutionSchedulingQueueItemDB() + queue_item_db.action_execution_id = str(execution_db.id) + queue_item_db.liveaction_id = str(liveaction_db.id) + queue_item_db.original_start_timestamp = liveaction_db.start_timestamp + queue_item_db.scheduled_start_timestamp = liveaction_db.start_timestamp + ActionExecutionSchedulingQueue.add_or_update(queue_item_db, publish=False) + + # Verify one queue entry exists + queue_items_before = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual(len(queue_items_before), 1) + original_queue_item_id = str(queue_items_before[0].id) + + # Run the bootstrap recovery + handler = ActionExecutionSchedulingQueueHandler() + handler._bootstrap_missing_scheduling_queue_items() + + # Verify still only one queue entry exists (no duplicate created) + queue_items_after = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual( + len(queue_items_after), 1, "Should not create duplicate queue entry" + ) + self.assertEqual(str(queue_items_after[0].id), original_queue_item_id) + + def test_bootstrap_ignores_non_requested_status(self): + """ + Test that _bootstrap_missing_scheduling_queue_items only processes + LiveActions in 'requested' status, not 'delayed', 'scheduled', or other statuses. + """ + statuses_to_test = [ + action_constants.LIVEACTION_STATUS_DELAYED, + action_constants.LIVEACTION_STATUS_SCHEDULED, + action_constants.LIVEACTION_STATUS_RUNNING, + action_constants.LIVEACTION_STATUS_SUCCEEDED, + ] + + created_liveactions = [] + for status in statuses_to_test: + liveaction_db = LiveActionDB() + liveaction_db.status = status + liveaction_db.action = "core.local" + liveaction_db.parameters = {"cmd": f"echo '{status}'"} + liveaction_db.start_timestamp = date_utils.get_datetime_utc_now() + liveaction_db = LiveAction.add_or_update(liveaction_db, publish=False) + created_liveactions.append(liveaction_db) + + # Create execution but no queue entry + from st2common.services import executions + from st2common.util import action_db as action_utils + + action_db = action_utils.get_action_by_ref("core.local") + runnertype_db = action_utils.get_runnertype_by_name( + action_db.runner_type["name"] + ) + executions.create_execution_object( + liveaction=liveaction_db, + action_db=action_db, + runnertype_db=runnertype_db, + publish=False, + ) + + # Run the bootstrap recovery + handler = ActionExecutionSchedulingQueueHandler() + handler._bootstrap_missing_scheduling_queue_items() + + # Verify no queue entries were created for non-requested statuses + for liveaction_db in created_liveactions: + queue_items = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual( + len(queue_items), + 0, + f"No queue entry should be created for status '{liveaction_db.status}'", + ) + + def test_bootstrap_handles_liveaction_without_execution(self): + """ + Test that _bootstrap_missing_scheduling_queue_items gracefully handles + the case where a LiveAction exists but its ActionExecution doesn't. + """ + # Create a LiveAction without an ActionExecution (edge case) + liveaction_db = LiveActionDB() + liveaction_db.status = action_constants.LIVEACTION_STATUS_REQUESTED + liveaction_db.action = "core.local" + liveaction_db.parameters = {"cmd": "echo 'orphan'"} + liveaction_db.start_timestamp = date_utils.get_datetime_utc_now() + liveaction_db = LiveAction.add_or_update(liveaction_db, publish=False) + + # Don't create an ActionExecution - this is the edge case + + # Run the bootstrap recovery - should not crash + handler = ActionExecutionSchedulingQueueHandler() + handler._bootstrap_missing_scheduling_queue_items() + + # Verify no queue entry was created (since there's no execution) + queue_items = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual( + len(queue_items), 0, "No queue entry should be created without execution" + ) + + def test_bootstrap_preserves_delay_field(self): + """ + Test that _bootstrap_missing_scheduling_queue_items correctly handles + the delay field when creating queue entries. + """ + # Create a LiveAction with a delay + delay_ms = 5000 # 5 seconds + liveaction_db = LiveActionDB() + liveaction_db.status = action_constants.LIVEACTION_STATUS_REQUESTED + liveaction_db.action = "core.local" + liveaction_db.parameters = {"cmd": "echo 'delayed'"} + liveaction_db.start_timestamp = date_utils.get_datetime_utc_now() + liveaction_db.delay = delay_ms + liveaction_db = LiveAction.add_or_update(liveaction_db, publish=False) + + # Create execution + from st2common.services import executions + from st2common.util import action_db as action_utils + + action_db = action_utils.get_action_by_ref("core.local") + runnertype_db = action_utils.get_runnertype_by_name( + action_db.runner_type["name"] + ) + executions.create_execution_object( + liveaction=liveaction_db, + action_db=action_db, + runnertype_db=runnertype_db, + publish=False, + ) + + # Run the bootstrap recovery + handler = ActionExecutionSchedulingQueueHandler() + handler._bootstrap_missing_scheduling_queue_items() + + # Verify queue entry was created with correct delay + queue_items = ActionExecutionSchedulingQueue.query( + liveaction_id=str(liveaction_db.id) + ) + self.assertEqual(len(queue_items), 1) + + queue_item = queue_items[0] + self.assertEqual(queue_item.delay, delay_ms) + + # Verify scheduled_start_timestamp is offset by the delay + expected_scheduled_time = date_utils.append_milliseconds_to_time( + liveaction_db.start_timestamp, delay_ms + ) + self.assertEqual(queue_item.scheduled_start_timestamp, expected_scheduled_time) diff --git a/st2reactor/st2reactor/cmd/rulesengine.py b/st2reactor/st2reactor/cmd/rulesengine.py index bdce9457ba..cefe58f584 100644 --- a/st2reactor/st2reactor/cmd/rulesengine.py +++ b/st2reactor/st2reactor/cmd/rulesengine.py @@ -68,10 +68,13 @@ def _run_worker(): # Poll the worker thread to detect failures while True: - if rules_engine_worker.thread and rules_engine_worker.thread.dead: + if ( + rules_engine_worker._consumer_thread + and rules_engine_worker._consumer_thread.dead + ): # Thread died - try to get the exception if it raised one try: - rules_engine_worker.thread.wait() # This will raise if thread raised + rules_engine_worker._consumer_thread.wait() # This will raise if thread raised except Exception as e: LOG.error("RulesEngine worker thread failed: %s", e) raise From c6a8c17b49fbe457d88f075277f581efe1060f14 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 6 Aug 2026 12:38:07 -0400 Subject: [PATCH 16/24] fix lint --- st2common/tests/unit/test_cluster_retry_exhaustion.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/st2common/tests/unit/test_cluster_retry_exhaustion.py b/st2common/tests/unit/test_cluster_retry_exhaustion.py index 50ec6f9f9a..fcb148dcc1 100644 --- a/st2common/tests/unit/test_cluster_retry_exhaustion.py +++ b/st2common/tests/unit/test_cluster_retry_exhaustion.py @@ -63,7 +63,7 @@ def test_should_stop_stops_at_exact_max_retries(self): # cluster_size=3, max_retries=1 means: 3 * (1+1) = 6 total attempts for i in range(5): should_stop, wait = context.should_stop(test_exc) - self.assertFalse(should_stop, f"Should not stop at attempt {i+1}") + self.assertFalse(should_stop, f"Should not stop at attempt {i + 1}") # 6th attempt should stop should_stop, wait = context.should_stop(test_exc) From 1bf3341b5d8484ca4b73cff99f8d971926ac3da9 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 6 Aug 2026 12:39:46 -0400 Subject: [PATCH 17/24] add changelog --- CHANGELOG.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 3c90a1aaf0..7769a158f0 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -23,7 +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 ~~~~~~~ * Removed Python 3.8 and 3.9 from testing and CI/CD. From 28d5d84ac5550efd41294da3aaad2d47f11d771a Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 6 Aug 2026 14:32:25 -0400 Subject: [PATCH 18/24] fix pants test; missing metrics driver dep --- contrib/runners/local_runner/tests/integration/BUILD | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/contrib/runners/local_runner/tests/integration/BUILD b/contrib/runners/local_runner/tests/integration/BUILD index 2d782aaea0..c7e8704f31 100644 --- a/contrib/runners/local_runner/tests/integration/BUILD +++ b/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", + ], + ), + }, +) \ No newline at end of file From 97d816288e88e51d342062b1107a69a64cf70b32 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 6 Aug 2026 17:05:10 -0400 Subject: [PATCH 19/24] move pants stevedore_namespaces up so all tests have them --- st2actions/tests/unit/BUILD | 20 ++++---------------- st2reactor/tests/unit/BUILD | 3 +++ 2 files changed, 7 insertions(+), 16 deletions(-) diff --git a/st2actions/tests/unit/BUILD b/st2actions/tests/unit/BUILD index ae3f66d70a..3ee9d7568b 100644 --- a/st2actions/tests/unit/BUILD +++ b/st2actions/tests/unit/BUILD @@ -6,6 +6,10 @@ __defaults__( python_tests( name="tests", uses=["mongo"], + stevedore_namespaces=[ + "st2common.runners.runner", + "st2common.metrics.driver", + ], overrides={ ( "test_executions.py", @@ -21,21 +25,5 @@ python_tests( "test_runner_container.py": dict( uses=["mongo", "system_user"], ), - ( - "test_execution*.py", - "test_notifier.py", - "test_output_schema.py", - "test_policies.py", - "test_queue_consumers.py", - "test_runner_container.py", - "test_scheduler*.py", - "test_worker.py", - "test_workflow_engine.py", - ): dict( - stevedore_namespaces=[ - "st2common.runners.runner", - "st2common.metrics.driver", - ], - ), }, ) diff --git a/st2reactor/tests/unit/BUILD b/st2reactor/tests/unit/BUILD index abd3115705..61e1510f8f 100644 --- a/st2reactor/tests/unit/BUILD +++ b/st2reactor/tests/unit/BUILD @@ -6,6 +6,9 @@ __defaults__( python_tests( name="tests", uses=["mongo"], + stevedore_namespaces=[ + "st2common.metrics.driver", + ], overrides={ "test_enforce.py": dict( stevedore_namespaces=[ From 28d62bb81ee1faa7316a9dc9099f926963aa8177 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 7 Aug 2026 08:17:30 -0400 Subject: [PATCH 20/24] restart ci From 6684f0a620f5672cc332ea7cc97a0bb284cf0577 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 7 Aug 2026 08:53:59 -0400 Subject: [PATCH 21/24] import fixtures for pants testing --- st2actions/tests/unit/test_scheduler_bootstrap_recovery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py b/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py index f3b3f152d4..b32ec4ac8d 100644 --- a/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py +++ b/st2actions/tests/unit/test_scheduler_bootstrap_recovery.py @@ -28,10 +28,10 @@ from st2tests.base import DbTestCase from st2tests.fixturesloader import FixturesLoader import st2tests.config as tests_config +from st2tests.fixtures.generic.fixture import PACK_NAME as FIXTURES_PACK from st2actions.scheduler.handler import ActionExecutionSchedulingQueueHandler -FIXTURES_PACK = "generic" TEST_FIXTURES = {"runners": ["run-local.yaml"], "actions": ["local.yaml"]} From df52b7a7bc8ab442e71fb083a92d97944f95f354 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 7 Aug 2026 09:18:15 -0400 Subject: [PATCH 22/24] lint fix changelog --- CHANGELOG.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 7769a158f0..5719ee2d16 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -24,6 +24,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 ~~~~~~~ * Removed Python 3.8 and 3.9 from testing and CI/CD. From 3956b61e030ec2651c82892057da4f4ae13f5257 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 7 Aug 2026 09:27:39 -0400 Subject: [PATCH 23/24] import fixtures for pants testing --- st2actions/st2actions/cmd/scheduler.py | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/st2actions/st2actions/cmd/scheduler.py b/st2actions/st2actions/cmd/scheduler.py index 27db9f5425..a2a0487237 100644 --- a/st2actions/st2actions/cmd/scheduler.py +++ b/st2actions/st2actions/cmd/scheduler.py @@ -112,18 +112,30 @@ def _run_scheduler(): try: # Poll threads in a loop - check if any has died/failed while True: - for thread, name in threads_to_monitor: - if thread.dead: - # Thread died - try to get the exception if it raised one + 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() # This will raise if the thread raised + 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 - # Thread completed successfully (shouldn't happen in normal operation) + + # 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 + return 0 # Sleep briefly to avoid tight loop and allow other greenlets to run eventlet.sleep(0.1) From 1dcc86401e5d5f8e4bfbfe3684d635d8b5db508f Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 7 Aug 2026 09:49:20 -0400 Subject: [PATCH 24/24] fix action alias --- st2common/st2common/transport/__init__.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/st2common/st2common/transport/__init__.py b/st2common/st2common/transport/__init__.py index 632c08dc0e..5eab8d45f5 100644 --- a/st2common/st2common/transport/__init__.py +++ b/st2common/st2common/transport/__init__.py @@ -15,12 +15,19 @@ from __future__ import absolute_import -from st2common.transport import liveaction, actionexecutionstate, execution, workflow +from st2common.transport import ( + actionalias, + liveaction, + actionexecutionstate, + execution, + workflow, +) from st2common.transport import publishers, reactor, utils, connection_retry_wrapper # TODO(manas) : Exchanges, Queues and RoutingKey design discussion pending. __all__ = [ + "actionalias", "liveaction", "actionexecutionstate", "execution",