Bootstrap resume workflows config options

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.
+  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
 ------------------------

conf/st2.conf.sample

@@ -388,6 +388,10 @@ logging = /etc/st2/logging.timersengine.conf
 webui_base_url = https://localhost
 
 [workflow_engine]
+# 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
 # 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.

lockfiles/st2.lock

@@ -3085,7 +3085,7 @@
           "artifacts": [
             {
               "algorithm": "sha256",
-              "hash": "491767e81c1bb11a54fb68d1a24119bdeede593a2beccca5bc09bfed36fdb35c",
+              "hash": "b9feb1769b48102061fe4fc59b2f5ad600bc2ac0b55cf12ef5fe49464ac0d230",
               "url": "git+https://github.com/StackStorm/orquesta.git"
             }
           ],

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__)
 
@@ -107,7 +111,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):
@@ -129,8 +138,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)
@@ -144,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)
 

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")
@@ -409,6 +417,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)
 
@@ -440,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"])

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())

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",

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()