From 8751ec8862258878e85a11272872c9796c77fb7f Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 11:30:55 -0400 Subject: [PATCH 1/8] boostrap resume exclude current engine --- st2actions/st2actions/workflows/workflows.py | 20 ++++++++- st2actions/tests/unit/test_workflow_engine.py | 41 +++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/st2actions/st2actions/workflows/workflows.py b/st2actions/st2actions/workflows/workflows.py index 6672069e6f..7a96973fca 100644 --- a/st2actions/st2actions/workflows/workflows.py +++ b/st2actions/st2actions/workflows/workflows.py @@ -129,8 +129,26 @@ def shutdown(self): except GroupNotCreated: pass + # Determine whether any *other* workflow engine is still a member of + # the coordination group. We must exclude our own member id: during a + # graceful shutdown this engine may not have deregistered itself yet, + # so the raw member list can still contain our own id. Keying off the + # raw list is wrong in both directions: + # * If we are the last engine, our own id is still present, so + # "not member_ids" is False and nothing is paused -- leaving the + # running workflows stuck with no engine to drive them. + # * If a peer engine ("engine B") is still up while this engine + # ("engine A") goes down, that peer will keep processing the + # workflows, so we must not pause them. + # Excluding our own id makes the decision correct in both cases: + # pause only when no other engine remains to take over. + our_member_id = coordination.get_member_id() + other_member_ids = [ + member_id for member_id in member_ids if member_id != our_member_id + ] + # Check if there are other WFEs in service registry - if cfg.CONF.coordination.service_registry and not member_ids: + if cfg.CONF.coordination.service_registry and not other_member_ids: ac_ex_dbs = self._get_running_workflows() for ac_ex_db in ac_ex_dbs: lv_ac = action_utils.get_liveaction_by_id(ac_ex_db.liveaction_id) diff --git a/st2actions/tests/unit/test_workflow_engine.py b/st2actions/tests/unit/test_workflow_engine.py index ed6440bd57..1b29abafbb 100644 --- a/st2actions/tests/unit/test_workflow_engine.py +++ b/st2actions/tests/unit/test_workflow_engine.py @@ -409,6 +409,47 @@ def test_workflow_engine_shutdown_with_multiple_members(self): 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) + @mock.patch.object( + RedisDriver, + "get_members", + mock.MagicMock( + return_value=coordination_service.NoOpAsyncResult( + (coordination_service.get_member_id(),) + ) + ), + ) + def test_workflow_engine_shutdown_last_engine_still_registered_pauses(self): + # Regression test for the last-engine shutdown case: this is the only + # workflow engine, but it has not yet deregistered itself from the + # coordination group, so the group still contains our own member id. + # The shutdown path must exclude our own id, recognize that no *other* + # engine remains to take over, and pause the running workflow. Keying + # off the raw member list (which is non-empty because it holds our own + # id) would wrongly skip the pause and strand the workflow in RUNNING. + self.reset_config(service_registry=True) + + wf_meta = self.get_wf_fixture_meta_data(TEST_PACK_PATH, "sequential.yaml") + lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta["name"]) + lv_ac_db, ac_ex_db = action_service.request(lv_ac_db) + + # Assert action 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) + wf_ex_db = wf_db_access.WorkflowExecution.query( + action_execution=str(ac_ex_db.id) + )[0] + self.assertEqual(wf_ex_db.status, action_constants.LIVEACTION_STATUS_RUNNING) + workflow_engine = workflows.get_engine() + + eventlet.spawn(workflow_engine.shutdown) + + # Sleep for few seconds to ensure shutdown sequence completes. + eventlet.sleep(5) + + # No other engine is left, so this engine must pause the workflow. + 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_PAUSING) + def test_workflow_engine_shutdown_with_service_registry_disabled(self): self.reset_config(service_registry=False) From 8d3bf9dba537a34af93c2f3ee1281d1ad1cf0c3a Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 11:45:40 -0400 Subject: [PATCH 2/8] conf update; disable bootstrap by default --- conf/st2.conf.sample | 2 ++ st2actions/st2actions/workflows/workflows.py | 7 ++++++- st2actions/tests/unit/test_workflow_engine.py | 12 +++++++++++- st2common/st2common/config.py | 8 ++++++++ 4 files changed, 27 insertions(+), 2 deletions(-) diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 27a2eb0a86..47d0d52d3c 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -388,6 +388,8 @@ logging = /etc/st2/logging.timersengine.conf webui_base_url = https://localhost [workflow_engine] +# On engine startup, resume workflows that were paused by a prior engine shutdown. Off by default; enable in clustered/rolling-restart environments where a shutdown can leave shutdown-paused workflows behind that need to be resumed. +bootstrap_enabled = False # How long to wait for process (in seconds) to exit after receiving shutdown signal. exit_still_active_check = 300 # Max seconds to allow workflow execution be idled before it is identified as orphaned and cancelled by the garbage collector. A value of zero means the feature is disabled. This is disabled by default. diff --git a/st2actions/st2actions/workflows/workflows.py b/st2actions/st2actions/workflows/workflows.py index 7a96973fca..b2a675eae4 100644 --- a/st2actions/st2actions/workflows/workflows.py +++ b/st2actions/st2actions/workflows/workflows.py @@ -107,7 +107,12 @@ def process(self, message): self._active_messages -= 1 def start(self, wait): - spawn_after(self._delay, self._resume_workflows_paused_during_shutdown) + # Resuming workflows paused by a prior engine shutdown is opt-in + # (off by default). Enable it via workflow_engine.bootstrap_enabled in + # environments where rolling restarts can leave shutdown-paused + # workflows behind. + if cfg.CONF.workflow_engine.bootstrap_enabled: + spawn_after(self._delay, self._resume_workflows_paused_during_shutdown) super(WorkflowExecutionHandler, self).start(wait=wait) def shutdown(self): diff --git a/st2actions/tests/unit/test_workflow_engine.py b/st2actions/tests/unit/test_workflow_engine.py index 1b29abafbb..1ae3e4d23a 100644 --- a/st2actions/tests/unit/test_workflow_engine.py +++ b/st2actions/tests/unit/test_workflow_engine.py @@ -99,6 +99,7 @@ def reset_config( exit_still_active_check=None, # default is 300 (st2common.config) still_active_check_interval=None, # default is 2 (st2common.config) service_registry=None, # default is False (st2common.config) + bootstrap_enabled=None, # default is False (st2common.config) ): tests_config.reset() tests_config.parse_args() @@ -124,6 +125,12 @@ def reset_config( cfg.CONF.set_override( name="service_registry", override=service_registry, group="coordination" ) + if bootstrap_enabled is not None: + cfg.CONF.set_override( + name="bootstrap_enabled", + override=bootstrap_enabled, + group="workflow_engine", + ) def test_process(self): self.reset_config() @@ -306,6 +313,7 @@ def test_workflow_engine_shutdown(self): exit_still_active_check=4, still_active_check_interval=1, service_registry=True, + bootstrap_enabled=True, ) wf_meta = self.get_wf_fixture_meta_data(TEST_PACK_PATH, "sequential.yaml") @@ -481,7 +489,9 @@ def test_workflow_engine_shutdown_with_service_registry_disabled(self): mock.MagicMock(return_value=coordination_service.NoOpLock(name="noop")), ) def test_workflow_engine_shutdown_first_then_start(self): - self.reset_config(service_registry=True, exit_still_active_check=0) + self.reset_config( + service_registry=True, exit_still_active_check=0, bootstrap_enabled=True + ) wf_meta = self.get_wf_fixture_meta_data(TEST_PACK_PATH, "sequential.yaml") lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta["name"]) diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index 28b0c062ec..d6e81a5e84 100644 --- a/st2common/st2common/config.py +++ b/st2common/st2common/config.py @@ -931,6 +931,14 @@ def register_opts(ignore_errors=False): default=2, help="Time interval between subsequent queries to check executions handled by WFE.", ), + cfg.BoolOpt( + "bootstrap_enabled", + default=False, + help="On engine startup, resume workflows that were paused by a " + "prior engine shutdown. Off by default; enable in " + "clustered/rolling-restart environments where a shutdown can " + "leave shutdown-paused workflows behind that need to be resumed.", + ), ] do_register_opts( From 30ef7ee318180bc7acbd1361b252bb6c48acdd2e Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Thu, 27 Aug 2026 15:59:21 -0400 Subject: [PATCH 3/8] fix orquesta hash (cherry picked from commit 054f95731ca5d4e46ced8c7125e0794c3f515873) --- lockfiles/st2.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lockfiles/st2.lock b/lockfiles/st2.lock index 546f12698e..7a9f94672c 100644 --- a/lockfiles/st2.lock +++ b/lockfiles/st2.lock @@ -3085,7 +3085,7 @@ "artifacts": [ { "algorithm": "sha256", - "hash": "491767e81c1bb11a54fb68d1a24119bdeede593a2beccca5bc09bfed36fdb35c", + "hash": "b9feb1769b48102061fe4fc59b2f5ad600bc2ac0b55cf12ef5fe49464ac0d230", "url": "git+https://github.com/StackStorm/orquesta.git" } ], From 38be6a81e629a74e6cf264d12fceeb3138591e16 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 12:05:24 -0400 Subject: [PATCH 4/8] add lookback for bootstrap --- conf/st2.conf.sample | 2 ++ st2actions/st2actions/workflows/workflows.py | 9 +++++++++ st2common/st2common/config.py | 7 +++++++ 3 files changed, 18 insertions(+) diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 47d0d52d3c..877441739f 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -390,6 +390,8 @@ webui_base_url = https://localhost [workflow_engine] # On engine startup, resume workflows that were paused by a prior engine shutdown. Off by default; enable in clustered/rolling-restart environments where a shutdown can leave shutdown-paused workflows behind that need to be resumed. bootstrap_enabled = False +# When bootstrap_enabled is set, only resume workflows whose LiveAction.start_timestamp is within this many days. Prevents accidentally resuming ancient paused workflows on engine startup. +bootstrap_lookback_days = 1 # How long to wait for process (in seconds) to exit after receiving shutdown signal. exit_still_active_check = 300 # Max seconds to allow workflow execution be idled before it is identified as orphaned and cancelled by the garbage collector. A value of zero means the feature is disabled. This is disabled by default. diff --git a/st2actions/st2actions/workflows/workflows.py b/st2actions/st2actions/workflows/workflows.py index b2a675eae4..845c9fd5ac 100644 --- a/st2actions/st2actions/workflows/workflows.py +++ b/st2actions/st2actions/workflows/workflows.py @@ -14,6 +14,9 @@ # limitations under the License. from __future__ import absolute_import + +import datetime + from oslo_config import cfg from orquesta import statuses @@ -37,6 +40,7 @@ from st2common.transport import utils as txpt_utils from st2common.util import concurrency from st2common.util import action_db as action_utils +from st2common.util import date as date_utils LOG = logging.getLogger(__name__) @@ -167,9 +171,14 @@ def _get_running_workflows(self): return ex_db_access.ActionExecution.query(**query_filters) def _get_workflows_paused_during_shutdown(self): + lookback_days = cfg.CONF.workflow_engine.bootstrap_lookback_days + start_timestamp_gte = date_utils.get_datetime_utc_now() - datetime.timedelta( + days=lookback_days + ) query_filters = { "status": ac_const.LIVEACTION_STATUS_PAUSED, "context__paused_by": WORKFLOW_ENGINE_START_STOP_SEQ, + "start_timestamp__gte": start_timestamp_gte, } return lv_db_access.LiveAction.query(**query_filters) diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index d6e81a5e84..247a968826 100644 --- a/st2common/st2common/config.py +++ b/st2common/st2common/config.py @@ -939,6 +939,13 @@ def register_opts(ignore_errors=False): "clustered/rolling-restart environments where a shutdown can " "leave shutdown-paused workflows behind that need to be resumed.", ), + cfg.IntOpt( + "bootstrap_lookback_days", + default=1, + help="When bootstrap_enabled is set, only resume workflows whose " + "LiveAction.start_timestamp is within this many days. Prevents " + "accidentally resuming ancient paused workflows on engine startup.", + ), ] do_register_opts( From 4131b9c4e4343815b9e348cffe6bb2555dd85d71 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 12:08:08 -0400 Subject: [PATCH 5/8] add changelog --- CHANGELOG.rst | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 5719ee2d16..244ee6db3a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -24,6 +24,10 @@ 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) +* Fix workflow engine shutdown stranding running workflows in ``running`` state. On shutdown the + engine now excludes its own coordination member id when deciding whether another engine is still + available, so it pauses running workflows only when no other engine remains to take them over + (last-engine shutdown), and leaves them running when a peer engine is still up. (by @guzzijones12@gmail.com) Changed ~~~~~~~ @@ -32,10 +36,17 @@ Changed * Replaced deprecated `pkg_resources` module with `importlib-metadata` and `importlib-resources`. * Replaced abandoned `flex` module by `openapi-spec-validator` * Replaced Stackstorm/logshipper (stops working with Python 3.12) and eventlet in the `linux.file_watch_sensor` with threading. (by @skiedude) + * Bumped the pinned ``orquesta`` git revision in ``lockfiles/st2.lock`` to pick up the workflow engine race-condition fix. (by @guzzijones12@gmail.com) Added ~~~~~ * added raw_string type to allow template strings to pass through variable processing (by @guzzijones12@gmail.com) #6351 +* added ``[workflow_engine].bootstrap_enabled`` option (default ``False``) to make resuming + shutdown-paused workflows on engine startup opt-in. Enable it in clustered/rolling-restart + environments where a shutdown can leave shutdown-paused workflows behind. (by @guzzijones12@gmail.com) +* added ``[workflow_engine].bootstrap_lookback_days`` option (default ``1``) to bound resume-on-startup + to workflows whose ``start_timestamp`` is within the given number of days, preventing accidental + resumption of ancient paused workflows. (by @guzzijones12@gmail.com) 3.9.0 - October 10, 2025 ------------------------ From ab31b01cac61fc85abf8312b8e35a30587e1c2da Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 12:37:00 -0400 Subject: [PATCH 6/8] st2 bootstrap recconcile workflow --- st2common/bin/st2-bootstrap-workflow | 22 ++ st2common/setup.py | 1 + st2common/st2common/cmd/bootstrap_workflow.py | 137 ++++++++++ st2common/st2common/services/workflows.py | 235 ++++++++++++++++++ .../tests/unit/test_bootstrap_workflow_cmd.py | 142 +++++++++++ 5 files changed, 537 insertions(+) create mode 100755 st2common/bin/st2-bootstrap-workflow create mode 100644 st2common/st2common/cmd/bootstrap_workflow.py create mode 100644 st2common/tests/unit/test_bootstrap_workflow_cmd.py diff --git a/st2common/bin/st2-bootstrap-workflow b/st2common/bin/st2-bootstrap-workflow new file mode 100755 index 0000000000..8d345ef48d --- /dev/null +++ b/st2common/bin/st2-bootstrap-workflow @@ -0,0 +1,22 @@ +#!/usr/bin/env python3 +# Licensed to the StackStorm, Inc ('StackStorm') under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. + + +import sys +from st2common.cmd.bootstrap_workflow import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/st2common/setup.py b/st2common/setup.py index 5e2764286d..ea5fb3f0d3 100644 --- a/st2common/setup.py +++ b/st2common/setup.py @@ -49,6 +49,7 @@ packages=find_packages(exclude=["setuptools", "tests"]), scripts=[ "bin/st2-bootstrap-rmq", + "bin/st2-bootstrap-workflow", "bin/st2-cleanup-db", "bin/st2-register-content", "bin/st2-purge-executions", diff --git a/st2common/st2common/cmd/bootstrap_workflow.py b/st2common/st2common/cmd/bootstrap_workflow.py new file mode 100644 index 0000000000..679182edab --- /dev/null +++ b/st2common/st2common/cmd/bootstrap_workflow.py @@ -0,0 +1,137 @@ +# 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. + + +""" +Manually bootstrap-resume a single workflow execution that was paused by a +prior workflow engine shutdown. Bypasses coordination first-member election +and the automatic-bootstrap lookback window; the operator is asserting they +want this specific execution resumed now. + +With --reconcile-running, instead re-drives a workflow that is stuck in RUNNING +(rather than PAUSED). This is the safety valve for the case where an engine was +hard-killed (e.g. OOM) after acking but before processing a message, leaving the +workflow RUNNING with no message left to advance it. The operator is asserting +this specific execution is stuck; see wf_svc.reconcile_running_execution. +""" + +from __future__ import absolute_import + +from oslo_config import cfg + +from st2common import config +from st2common import log as logging +from st2common.config import do_register_cli_opts +from st2common.constants import action as ac_const +from st2common.constants.exit_codes import FAILURE_EXIT_CODE +from st2common.constants.exit_codes import SUCCESS_EXIT_CODE +from st2common.persistence import execution as ex_db_access +from st2common.persistence import liveaction as lv_db_access +from st2common.script_setup import setup as common_setup +from st2common.script_setup import teardown as common_teardown +from st2common.services import workflows as wf_svc + + +__all__ = ["main"] + +LOG = logging.getLogger(__name__) + + +def _register_cli_opts(): + cli_opts = [ + cfg.StrOpt( + "execution-id", + default=None, + help="ActionExecution id of the shutdown-paused workflow to resume.", + ), + cfg.BoolOpt( + "reconcile-running", + default=False, + help=( + "Re-drive a workflow stuck in RUNNING (not PAUSED) whose driving " + "message was lost with a hard-killed engine. Only use this on an " + "execution you have determined is stuck; it is unsafe against a " + "workflow still being actively processed." + ), + ), + ] + do_register_cli_opts(cli_opts) + + +def _reconcile_running_one(execution_id): + ac_ex_db = ex_db_access.ActionExecution.get_by_id(execution_id) + lv_ac_db = lv_db_access.LiveAction.get_by_id(str(ac_ex_db.liveaction_id)) + + if lv_ac_db.status != ac_const.LIVEACTION_STATUS_RUNNING: + LOG.error( + "Execution %s is in status %r, not %r. Refusing to reconcile. " + "Use bootstrap-resume (without --reconcile-running) for paused workflows.", + execution_id, + lv_ac_db.status, + ac_const.LIVEACTION_STATUS_RUNNING, + ) + return FAILURE_EXIT_CODE + + wf_svc.reconcile_running_execution(lv_ac_db) + LOG.info("Reconciled stuck-running execution %s.", execution_id) + return SUCCESS_EXIT_CODE + + +def _bootstrap_one(execution_id): + ac_ex_db = ex_db_access.ActionExecution.get_by_id(execution_id) + lv_ac_db = lv_db_access.LiveAction.get_by_id(str(ac_ex_db.liveaction_id)) + + if lv_ac_db.status != ac_const.LIVEACTION_STATUS_PAUSED: + LOG.error( + "Execution %s is in status %r, not %r. Refusing to bootstrap-resume.", + execution_id, + lv_ac_db.status, + ac_const.LIVEACTION_STATUS_PAUSED, + ) + return FAILURE_EXIT_CODE + + paused_by = lv_ac_db.context.get("paused_by") + if paused_by != wf_svc.WORKFLOW_ENGINE_START_STOP_SEQ: + LOG.error( + "Execution %s was not paused by an engine shutdown " + "(paused_by=%r). Use `st2 execution resume` for user-paused workflows.", + execution_id, + paused_by, + ) + return FAILURE_EXIT_CODE + + wf_svc.bootstrap_resume_execution(lv_ac_db) + LOG.info("Bootstrap-resumed execution %s.", execution_id) + return SUCCESS_EXIT_CODE + + +def main(): + _register_cli_opts() + common_setup(config=config, setup_db=True, register_mq_exchanges=True) + + execution_id = cfg.CONF.execution_id + if not execution_id: + LOG.error("--execution-id is required. Aborting.") + common_teardown() + return FAILURE_EXIT_CODE + + try: + if cfg.CONF.reconcile_running: + return _reconcile_running_one(execution_id) + return _bootstrap_one(execution_id) + except Exception as e: + LOG.exception("Failed to bootstrap-resume execution %s: %s", execution_id, e) + return FAILURE_EXIT_CODE + finally: + common_teardown() diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py index c99fb896b5..85cb438057 100644 --- a/st2common/st2common/services/workflows.py +++ b/st2common/st2common/services/workflows.py @@ -55,6 +55,11 @@ LOG = logging.getLogger(__name__) +# Marker written into LiveAction.context.paused_by when the workflow engine +# pauses running workflows during its own shutdown. The manual +# st2-bootstrap-workflow CLI uses this marker to identify eligible workflows. +WORKFLOW_ENGINE_START_STOP_SEQ = "workflow_engine_start_stop_seq" + LOG_FUNCTIONS = { "audit": LOG.audit, "debug": LOG.debug, @@ -1575,3 +1580,233 @@ def identify_orphaned_workflows(): continue return orphaned + + +def sync_completed_tasks_to_conductor(wf_ex_id): + """ + Synchronize task executions from database to conductor state. + + Two scenarios are handled: + 1. Completed tasks: sync their completion into the conductor state so it + stops thinking they are still running. + 2. Running tasks: re-stage them so get_next_tasks() finds them. + + This is required after a workflow was paused during engine shutdown but + tasks continued to complete or transitioned to running before the pause was + fully processed. + """ + LOG.debug("Starting task synchronization for workflow execution %s", wf_ex_id) + + wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id) + conductor = deserialize_conductor(wf_ex_db) + + task_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=wf_ex_id) + LOG.debug("Found %d task execution(s) for workflow %s", len(task_ex_dbs), wf_ex_id) + + updated = False + restaged_count = 0 + + for task_ex_db in task_ex_dbs: + if task_ex_db.status in statuses.COMPLETED_STATUSES: + task_state = conductor.get_task_state_entry( + task_ex_db.task_id, task_ex_db.task_route + ) + if ( + task_state + and task_state.get("status") not in statuses.COMPLETED_STATUSES + ): + ac_ex_event = events.ActionExecutionEvent( + task_ex_db.status, result=task_ex_db.result + ) + conductor.update_task_state( + task_ex_db.task_id, task_ex_db.task_route, ac_ex_event + ) + updated = True + LOG.debug( + 'Synchronized completed task "%s" (status: %s) to conductor state', + task_ex_db.task_id, + task_ex_db.status, + ) + + elif task_ex_db.status == statuses.RUNNING: + staged_task = conductor.workflow_state.get_staged_task( + task_ex_db.task_id, task_ex_db.task_route + ) + + if not staged_task: + task_state = conductor.get_task_state_entry( + task_ex_db.task_id, task_ex_db.task_route + ) + + if task_state: + ctxs_in = task_state.get("ctxs", {}).get("in", [0]) + conductor.workflow_state.add_staged_task( + task_ex_db.task_id, + task_ex_db.task_route, + ctxs=ctxs_in, + prev=task_state.get("prev", {}), + ready=True, + ) + updated = True + restaged_count += 1 + LOG.debug( + 'Re-staged running task "%s" (route: %s) to conductor', + task_ex_db.task_id, + task_ex_db.task_route, + ) + else: + LOG.warning( + 'Cannot re-stage task "%s" - no task state entry found', + task_ex_db.task_id, + ) + + if updated: + wf_ex_db.state = conductor.workflow_state.serialize() + wf_db_access.WorkflowExecution.update(wf_ex_db, publish=False) + + completed_count = len( + [t for t in task_ex_dbs if t.status in statuses.COMPLETED_STATUSES] + ) + if completed_count > 0: + LOG.info( + 'Synchronized %d completed task(s) to conductor for workflow "%s"', + completed_count, + wf_ex_id, + ) + if restaged_count > 0: + LOG.info( + 'Re-staged %d running task(s) to conductor for workflow "%s"', + restaged_count, + wf_ex_id, + ) + else: + LOG.debug( + "No tasks needed synchronization for workflow %s (all tasks already in sync)", + wf_ex_id, + ) + + +def bootstrap_resume_execution(lv_ac_db): + """ + Resume a single LiveAction that was paused during a prior engine shutdown. + + Clears the paused_by marker, syncs any tasks that changed state while the + workflow was paused, then calls request_resume. Raises on failure so the + caller (manual CLI) can log/report per-execution. + """ + LOG.debug( + "[%s] Bootstrap-resume starting; LiveAction status: %s", + str(lv_ac_db.id), + lv_ac_db.status, + ) + + if "paused_by" in lv_ac_db.context: + del lv_ac_db.context["paused_by"] + lv_ac_db = lv_db_access.LiveAction.add_or_update(lv_ac_db, publish=False) + + ac_ex_db = ex_db_access.ActionExecution.get(liveaction_id=str(lv_ac_db.id)) + wf_ex_id = ac_ex_db.context.get("workflow_execution") + + if wf_ex_id: + sync_completed_tasks_to_conductor(wf_ex_id) + else: + LOG.warning( + "[%s] No workflow_execution ID in context; skipping task sync.", + str(ac_ex_db.id), + ) + + request_resume(ac_ex_db) + LOG.info('Bootstrap-resumed workflow execution "%s".', str(ac_ex_db.id)) + + +def reconcile_running_execution(lv_ac_db): + """ + Re-drive a workflow stuck in RUNNING because a message that would have + advanced it was lost (e.g. an engine was OOM-killed after acking a message + but before processing it -- see the ack-on-dispatch behavior in + st2common.transport.consumers). + + At the moment the message is lost, the child *action execution* has already + completed and been persisted, but its TaskExecution and the conductor were + never advanced. So the durable ground truth is the child action execution + status, not the TaskExecution status -- which is why this does NOT rely on + sync_completed_tasks_to_conductor (that keys off TaskExecution status). + + Recovery has two phases: + + 1. Replay lost task-completion messages. For any task that is not yet + completed but whose child action execution has finished, re-run + handle_action_execution_completion -- exactly what the lost message + would have done (advance the conductor, request the next tasks). Each + child action execution is handled at most once, and tasks whose action + is still running are left untouched, so a workflow a live engine is + still driving is not disturbed. + 2. Re-request next tasks, in case the lost message was the *request* to + start the next task (conductor advanced but no TaskExecution created). + conductor.get_next_tasks() will not return tasks it already tracks, so + this is a no-op when nothing is missing. + + This is intended for operator-initiated recovery of a workflow the operator + has already determined is stuck; it is deliberately not run automatically. + + Note: for itemized ("with items") tasks a task has multiple child action + executions; each completed child is replayed once, which matches normal + per-item completion handling. + """ + ac_ex_db = ex_db_access.ActionExecution.get(liveaction_id=str(lv_ac_db.id)) + wf_ex_id = ac_ex_db.context.get("workflow_execution") + + if not wf_ex_id: + LOG.warning( + "[%s] No workflow_execution ID in context; cannot reconcile.", + str(ac_ex_db.id), + ) + return + + # Phase 1: replay completed-but-unprocessed child action executions. + handled_child_ids = set() + replayed = 0 + + while True: + progressed = False + task_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=wf_ex_id) + + for task_ex_db in task_ex_dbs: + if task_ex_db.status in statuses.COMPLETED_STATUSES: + continue + + child_ac_ex_dbs = ex_db_access.ActionExecution.query( + task_execution=str(task_ex_db.id) + ) + for child_ac_ex_db in child_ac_ex_dbs: + if str(child_ac_ex_db.id) in handled_child_ids: + continue + if child_ac_ex_db.status not in ac_const.LIVEACTION_COMPLETED_STATES: + continue + + LOG.info( + '[%s] Replaying lost completion of action execution "%s" ' + 'for task "%s".', + str(ac_ex_db.id), + str(child_ac_ex_db.id), + task_ex_db.task_id, + ) + handle_action_execution_completion(child_ac_ex_db) + handled_child_ids.add(str(child_ac_ex_db.id)) + replayed += 1 + progressed = True + + if not progressed: + break + + # Phase 2: re-request next tasks if the workflow is still running, to cover + # the case where the lost message was the next-task request itself. + wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id) + if wf_ex_db.status in statuses.RUNNING_STATUSES: + request_next_tasks(wf_ex_db) + + LOG.info( + 'Reconciled stuck-running workflow execution "%s" (replayed %d completion(s)).', + str(ac_ex_db.id), + replayed, + ) diff --git a/st2common/tests/unit/test_bootstrap_workflow_cmd.py b/st2common/tests/unit/test_bootstrap_workflow_cmd.py new file mode 100644 index 0000000000..a26bb69cd1 --- /dev/null +++ b/st2common/tests/unit/test_bootstrap_workflow_cmd.py @@ -0,0 +1,142 @@ +# 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. + +from __future__ import absolute_import + +from st2common.util.monkey_patch import monkey_patch + +monkey_patch() + +import bson +import mock + +from st2common.cmd import bootstrap_workflow +from st2common.constants import action as action_constants +from st2common.constants.exit_codes import FAILURE_EXIT_CODE +from st2common.constants.exit_codes import SUCCESS_EXIT_CODE +from st2common.models.db.execution import ActionExecutionDB +from st2common.models.db.liveaction import LiveActionDB +from st2common.persistence.execution import ActionExecution +from st2common.persistence.liveaction import LiveAction +from st2common.services import workflows as wf_svc +from st2tests.base import CleanDbTestCase + + +class TestBootstrapWorkflowCLI(CleanDbTestCase): + def _make_paused_execution(self, paused_by): + lv_ac_db = LiveActionDB( + action="core.local", + status=action_constants.LIVEACTION_STATUS_PAUSED, + context={"paused_by": paused_by} if paused_by is not None else {}, + ) + lv_ac_db = LiveAction.add_or_update(lv_ac_db, publish=False) + ac_ex_db = ActionExecutionDB( + liveaction_id=str(lv_ac_db.id), + action={"ref": "core.local"}, + runner={"name": "local-shell-cmd"}, + status=action_constants.LIVEACTION_STATUS_PAUSED, + context={}, + ) + ac_ex_db = ActionExecution.add_or_update(ac_ex_db, publish=False) + return lv_ac_db, ac_ex_db + + def test_missing_execution_id_returns_not_found(self): + # No such execution exists. + bogus = str(bson.ObjectId()) + with mock.patch.object(wf_svc, "bootstrap_resume_execution") as mock_resume: + # Expect a database miss to raise, main catches → FAILURE. + # _bootstrap_one raises via get_by_id; main wraps it. + with self.assertRaises(Exception): + bootstrap_workflow._bootstrap_one(bogus) + mock_resume.assert_not_called() + + def test_rejects_execution_not_paused(self): + lv_ac_db = LiveActionDB( + action="core.local", + status=action_constants.LIVEACTION_STATUS_SUCCEEDED, + context={"paused_by": wf_svc.WORKFLOW_ENGINE_START_STOP_SEQ}, + ) + lv_ac_db = LiveAction.add_or_update(lv_ac_db, publish=False) + ac_ex_db = ActionExecutionDB( + liveaction_id=str(lv_ac_db.id), + action={"ref": "core.local"}, + runner={"name": "local-shell-cmd"}, + status=action_constants.LIVEACTION_STATUS_SUCCEEDED, + context={}, + ) + ac_ex_db = ActionExecution.add_or_update(ac_ex_db, publish=False) + + with mock.patch.object(wf_svc, "bootstrap_resume_execution") as mock_resume: + rc = bootstrap_workflow._bootstrap_one(str(ac_ex_db.id)) + self.assertEqual(rc, FAILURE_EXIT_CODE) + mock_resume.assert_not_called() + + def test_rejects_paused_by_other_actor(self): + _lv_ac, ac_ex_db = self._make_paused_execution(paused_by="some_user@stackstorm") + + with mock.patch.object(wf_svc, "bootstrap_resume_execution") as mock_resume: + rc = bootstrap_workflow._bootstrap_one(str(ac_ex_db.id)) + self.assertEqual(rc, FAILURE_EXIT_CODE) + mock_resume.assert_not_called() + + def test_happy_path_calls_service(self): + lv_ac_db, ac_ex_db = self._make_paused_execution( + paused_by=wf_svc.WORKFLOW_ENGINE_START_STOP_SEQ + ) + + with mock.patch.object(wf_svc, "bootstrap_resume_execution") as mock_resume: + rc = bootstrap_workflow._bootstrap_one(str(ac_ex_db.id)) + self.assertEqual(rc, SUCCESS_EXIT_CODE) + mock_resume.assert_called_once() + # Called with the LiveActionDB matching our record. + args, _ = mock_resume.call_args + self.assertEqual(str(args[0].id), str(lv_ac_db.id)) + + def _make_running_execution(self): + lv_ac_db = LiveActionDB( + action="core.local", + status=action_constants.LIVEACTION_STATUS_RUNNING, + context={}, + ) + lv_ac_db = LiveAction.add_or_update(lv_ac_db, publish=False) + ac_ex_db = ActionExecutionDB( + liveaction_id=str(lv_ac_db.id), + action={"ref": "core.local"}, + runner={"name": "orquesta"}, + status=action_constants.LIVEACTION_STATUS_RUNNING, + context={}, + ) + ac_ex_db = ActionExecution.add_or_update(ac_ex_db, publish=False) + return lv_ac_db, ac_ex_db + + def test_reconcile_rejects_execution_not_running(self): + # A paused (not running) execution must be rejected by the reconcile path. + _lv_ac, ac_ex_db = self._make_paused_execution( + paused_by=wf_svc.WORKFLOW_ENGINE_START_STOP_SEQ + ) + + with mock.patch.object(wf_svc, "reconcile_running_execution") as mock_reconcile: + rc = bootstrap_workflow._reconcile_running_one(str(ac_ex_db.id)) + self.assertEqual(rc, FAILURE_EXIT_CODE) + mock_reconcile.assert_not_called() + + def test_reconcile_happy_path_calls_service(self): + lv_ac_db, ac_ex_db = self._make_running_execution() + + with mock.patch.object(wf_svc, "reconcile_running_execution") as mock_reconcile: + rc = bootstrap_workflow._reconcile_running_one(str(ac_ex_db.id)) + self.assertEqual(rc, SUCCESS_EXIT_CODE) + mock_reconcile.assert_called_once() + args, _ = mock_reconcile.call_args + self.assertEqual(str(args[0].id), str(lv_ac_db.id)) From 478cdc102528d6b8e1adf1ecb3ed9da95e09890b Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 17:13:03 -0400 Subject: [PATCH 7/8] correct wording --- CHANGELOG.rst | 2 +- st2common/st2common/config.py | 6 ++---- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 244ee6db3a..1200d28348 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -42,7 +42,7 @@ Added ~~~~~ * added raw_string type to allow template strings to pass through variable processing (by @guzzijones12@gmail.com) #6351 * added ``[workflow_engine].bootstrap_enabled`` option (default ``False``) to make resuming - shutdown-paused workflows on engine startup opt-in. Enable it in clustered/rolling-restart + shutdown-paused workflows on engine startup opt-in. environments where a shutdown can leave shutdown-paused workflows behind. (by @guzzijones12@gmail.com) * added ``[workflow_engine].bootstrap_lookback_days`` option (default ``1``) to bound resume-on-startup to workflows whose ``start_timestamp`` is within the given number of days, preventing accidental diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index 247a968826..e17d8f7e3b 100644 --- a/st2common/st2common/config.py +++ b/st2common/st2common/config.py @@ -934,10 +934,8 @@ def register_opts(ignore_errors=False): cfg.BoolOpt( "bootstrap_enabled", default=False, - help="On engine startup, resume workflows that were paused by a " - "prior engine shutdown. Off by default; enable in " - "clustered/rolling-restart environments where a shutdown can " - "leave shutdown-paused workflows behind that need to be resumed.", + help="On leader engine startup, resume workflows that were paused by " + "prior all engines complete shutdown. Off by default.", ), cfg.IntOpt( "bootstrap_lookback_days", From c7222a4ced478ce15b5e79af11ff8d787f411fb5 Mon Sep 17 00:00:00 2001 From: guzzijones12 Date: Fri, 28 Aug 2026 18:26:28 -0400 Subject: [PATCH 8/8] make 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 877441739f..0e5bf0f320 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -388,7 +388,7 @@ logging = /etc/st2/logging.timersengine.conf webui_base_url = https://localhost [workflow_engine] -# On engine startup, resume workflows that were paused by a prior engine shutdown. Off by default; enable in clustered/rolling-restart environments where a shutdown can leave shutdown-paused workflows behind that need to be resumed. +# On leader engine startup, resume workflows that were paused by prior all engines complete shutdown. Off by default. bootstrap_enabled = False # When bootstrap_enabled is set, only resume workflows whose LiveAction.start_timestamp is within this many days. Prevents accidentally resuming ancient paused workflows on engine startup. bootstrap_lookback_days = 1