inquiry race condition fix; default max concurrency, db max delay

CHANGELOG.rst

@@ -3,6 +3,112 @@ Changelog
 
 in development
 --------------
+* Workflow engine race condition fixes
+  Contributed by @guzzijones12.
+
+  A set of fixes for concurrency races in the orquesta workflow engine,
+  centered on the resume paths and per-workflow serialization. They
+  harden the engine against interleaved action-execution completions and
+  workflow-level status messages for the same workflow — the classic
+  trigger being an inquiry response, which publishes a workflow
+  ``RESUMING`` and an action-execution ``SUCCEEDED`` nearly
+  simultaneously — and bound how long the per-workflow coordination lock
+  is held when an unbounded ``with-items`` task fans out.
+
+  **Fixes** (in cherry-pick order):
+
+  * ``request_resume`` — wrap body in the per-workflow coord lock and
+    re-read the workflow under the lock. Remove ``RESUMING`` from the
+    "already active, silently skip" short-circuit so a workflow stuck
+    mid-resume (previous attempt crashed after writing
+    ``status=resuming`` but before completing) can be re-driven instead
+    of silently no-op'd. If the conductor is already in ``RESUMING`` when
+    a fresh resume arrives, transition it to ``PAUSED`` first and then
+    re-request ``RESUMING`` — orquesta dedupes redundant transitions and
+    would otherwise treat the second request as a no-op inside its state
+    machine.
+
+  * ``request_next_tasks`` — auto-transition ``RESUMING`` to ``RUNNING``
+    at the top of the function. Previously the engine relied entirely on
+    a child-cascade path to escape ``RESUMING``. For workflows paused at
+    an inquiry (inquiry LiveAction goes ``PENDING`` → ``SUCCEEDED``
+    without ever passing through ``RESUMING``) no child cascade ever
+    fires, so the workflow stayed in ``RESUMING`` indefinitely.
+    ``refresh_conductor`` above reads the latest state, so if a
+    legitimate child cascade reached ``RUNNING`` before us the
+    transition is a no-op.
+
+  * ``handle_action_execution_resume`` — coord-locked local work,
+    cascade released outside the lock. ``resume_task_execution`` and
+    ``resume_workflow_execution`` now run under the per-workflow lock;
+    the upstream ``handle_action_execution_resume(parent_ac_ex_db)``
+    cascade is called *after* the lock is released. Direction is
+    child → parent only, so no deadlock cycle from releasing early.
+    Prevents holding a chain of N locks up the parent tree in deeply
+    nested subworkflow scenarios.
+
+  * ``request_next_tasks`` / ``deserialize_conductor`` — bound
+    ``with-items`` fan-out. A ``with-items`` task that does not set
+    ``concurrency`` previously had every item's action execution
+    dispatched synchronously in a single ``request_next_tasks`` pass while
+    the per-workflow coord lock was held. New config
+    ``workflow_engine.default_with_items_concurrency`` (default 0 = disabled)
+    is injected as the default ``concurrency`` for such tasks when set, so
+    orquesta's existing concurrency machinery dispatches items in bounded
+    batches. Tasks that specify their own ``concurrency`` are left
+    untouched, and the injection is in-memory only (the stored workflow
+    spec is not modified). Replaces the earlier wall-clock
+    ``request_next_tasks_deadline_sec`` guard, which only bounded the
+    outer conductor-step loop (not the item fan-out) and failed otherwise
+    healthy workflows on expiry; it has been removed.
+
+  * ``WorkflowExecutionHandler.handle_workflow_execution`` — take the
+    per-workflow coord lock so it serializes against
+    ``handle_action_execution_completion`` (which already took the same
+    lock). Eliminates the inquiry-response race where a workflow-level
+    ``RESUMING`` message and an action-execution ``SUCCEEDED`` message
+    for the same workflow processed concurrently. Wraps the whole body:
+    re-reads the workflow under the lock (in case the queued message is
+    stale), then calls ``request_next_tasks``.
+
+  * ``stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec``
+    (60 s default) added to every ``retry_on_transient_db_errors``
+    decorator in ``st2common.services.workflows`` — 10 call sites
+    including ``request``, ``request_pause``, ``request_resume``,
+    ``request_cancellation``, ``update_task_state``,
+    ``update_task_execution``, ``resume_task_execution``,
+    ``update_workflow_execution``, ``resume_workflow_execution``,
+    ``fail_workflow_execution``. When the ceiling is hit, the last
+    ``StackStormDBObjectWriteConflictError`` propagates out to
+    ``WorkflowExecutionHandler.process``, which routes it to
+    ``fail_workflow_execution`` and releases the coordination lock —
+    freeing every other message waiting behind that workflow.
+
+  * Widened ``retry_on_connection_errors`` in
+    ``st2common.exceptions.workflow`` to also match
+    ``pymongo.errors.ConnectionFailure``. That covers ``AutoReconnect``,
+    ``NotPrimaryError``, ``ServerSelectionTimeoutError`` and
+    ``NetworkTimeout``. Mongo replica-set failovers now retry cleanly
+    under the same 60 s ceiling instead of failing the workflow on the
+    first failover-induced exception.
+
+  **Design note: Mongo vs. RabbitMQ failure handling.** The engine's
+  response to a lost dependency is intentionally asymmetric:
+
+  * **RabbitMQ:** the consumer connection has bounded retries built into
+    kombu. After the retry envelope gives up, the exception propagates
+    out of the consumer thread, the engine process exits, and the
+    container orchestrator (Kubernetes, systemd, …) restarts it. A
+    fresh process gives the cleanest recovery path when the broker
+    comes back.
+  * **Mongo:** the workflow-service retry decorators are bounded per-call
+    (60 s via ``retry_stop_max_msec``), but the outer consumer loop nacks
+    failed messages back to RabbitMQ for redelivery. Individual workflows
+    fail gracefully after 60 s of exhausted Mongo retries; the engine
+    process itself stays alive. Short blips (replica-set failover, brief
+    network hiccup) are absorbed by the per-call retry envelope. Longer
+    outages produce failed workflows, not a crashed engine.
+
 * implemented zstandard compression for parameters and results. #5995
   contributed by @guzzijones12
 

conf/st2.conf.sample

@@ -388,6 +388,8 @@ logging = /etc/st2/logging.timersengine.conf
 webui_base_url = https://localhost
 
 [workflow_engine]
+# Default concurrency applied to with-items tasks that do not specify their own concurrency. This bounds how many item action executions the engine dispatches per pass while holding the per-workflow coordination lock, instead of dispatching every item at once. Tasks that set concurrency in the workflow definition are left untouched. A value of zero disables this and preserves unbounded (spec-defined only) behavior. This is disabled by default.
+default_with_items_concurrency = 0
 # 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.

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

@@ -332,6 +332,86 @@ def test_with_items_concurrency(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_SUCCEEDED)
 
+    def test_with_items_default_concurrency(self):
+        # The workflow definition sets no concurrency on the with items task. The
+        # configured default_with_items_concurrency is injected as the default so the
+        # engine dispatches items in bounded batches instead of all at once.
+        num_items = 3
+        concurrency = 2
+
+        cfg.CONF.set_override(
+            "default_with_items_concurrency", concurrency, group="workflow_engine"
+        )
+        self.addCleanup(
+            cfg.CONF.clear_override,
+            "default_with_items_concurrency",
+            group="workflow_engine",
+        )
+
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, "with-items.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)
+
+        # Only the first batch (== configured default concurrency) is dispatched,
+        # not every item at once.
+        query_filters = {"workflow_execution": str(wf_ex_db.id), "task_id": "task1"}
+        t1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_ac_ex_dbs = ex_db_access.ActionExecution.query(
+            task_execution=str(t1_ex_db.id)
+        )
+
+        self.assertEqual(len(t1_ac_ex_dbs), concurrency)
+
+        status = [
+            ac_ex.status == action_constants.LIVEACTION_STATUS_SUCCEEDED
+            for ac_ex in t1_ac_ex_dbs
+        ]
+
+        self.assertTrue(all(status))
+
+        for t1_ac_ex_db in t1_ac_ex_dbs:
+            workflows.get_engine().process(t1_ac_ex_db)
+
+        t1_ex_db = wf_db_access.TaskExecution.get_by_id(t1_ex_db.id)
+        self.assertEqual(t1_ex_db.status, wf_statuses.RUNNING)
+
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        self.assertEqual(wf_ex_db.status, wf_statuses.RUNNING)
+
+        # The remaining items are dispatched once the first batch completes.
+        t1_ac_ex_dbs = ex_db_access.ActionExecution.query(
+            task_execution=str(t1_ex_db.id)
+        )
+
+        self.assertEqual(len(t1_ac_ex_dbs), num_items)
+
+        status = [
+            ac_ex.status == action_constants.LIVEACTION_STATUS_SUCCEEDED
+            for ac_ex in t1_ac_ex_dbs
+        ]
+
+        self.assertTrue(all(status))
+
+        for t1_ac_ex_db in t1_ac_ex_dbs[concurrency:]:
+            workflows.get_engine().process(t1_ac_ex_db)
+
+        t1_ex_db = wf_db_access.TaskExecution.get_by_id(t1_ex_db.id)
+        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
+
+        # Assert the main workflow is completed.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
+        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_SUCCEEDED)
+
     @mock.patch.object(
         local_shell_command_runner.LocalShellCommandRunner,
         "run",

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

@@ -197,9 +197,23 @@ def fail_workflow_execution(self, message, exception):
         wf_svc.fail_workflow_execution(wf_ex_id, exception, task=task)
 
     def handle_workflow_execution(self, wf_ex_db):
-        # Request the next set of tasks to execute.
-        wf_svc.update_progress(wf_ex_db, "Processing request for workflow execution.")
-        wf_svc.request_next_tasks(wf_ex_db)
+        # Serialize with handle_action_execution_completion (which also takes
+        # this per-workflow lock). Without it, a workflow-level message (e.g.
+        # RESUMING published by wf_svc.request_resume, or REQUESTED at start)
+        # can race with an in-flight ActionExecution completion for the same
+        # workflow — both call request_next_tasks and interleave conductor
+        # state writes. The classic trigger is an inquiry response, which
+        # publishes both a workflow RESUMING and an ac_ex SUCCEEDED nearly
+        # simultaneously.
+        wf_ex_id = str(wf_ex_db.id)
+        with coordination.get_coordinator(start_heart=True).get_lock(wf_ex_id.encode()):
+            # Re-read under the lock — the queued message may be stale.
+            wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id)
+            # Request the next set of tasks to execute.
+            wf_svc.update_progress(
+                wf_ex_db, "Processing request for workflow execution."
+            )
+            wf_svc.request_next_tasks(wf_ex_db)
 
     def handle_action_execution(self, ac_ex_db):
         # Exit if action execution is not executed under an orquesta workflow.

st2common/st2common/config.py

@@ -931,6 +931,17 @@ def register_opts(ignore_errors=False):
             default=2,
             help="Time interval between subsequent queries to check executions handled by WFE.",
         ),
+        cfg.IntOpt(
+            "default_with_items_concurrency",
+            default=0,
+            help="Default concurrency applied to with-items tasks that do not "
+            "specify their own concurrency. This bounds how many item action "
+            "executions the engine dispatches per pass while holding the "
+            "per-workflow coordination lock, instead of dispatching every item at "
+            "once. Tasks that set concurrency in the workflow definition are left "
+            "untouched. A value of zero disables this and preserves unbounded "
+            "(spec-defined only) behavior. This is disabled by default.",
+        ),
     ]
 
     do_register_opts(

st2common/st2common/exceptions/workflow.py

@@ -16,6 +16,7 @@
 from __future__ import absolute_import
 
 import mongoengine
+import pymongo
 import tooz
 
 from st2common import exceptions as st2_exc
@@ -29,8 +30,10 @@
 def retry_on_connection_errors(exc):
     LOG.warning("Determining if exception %s should be retried.", type(exc))
 
-    retrying = isinstance(exc, tooz.coordination.ToozConnectionError) or isinstance(
-        exc, mongoengine.connection.ConnectionFailure
+    retrying = (
+        isinstance(exc, tooz.coordination.ToozConnectionError)
+        or isinstance(exc, mongoengine.connection.ConnectionFailure)
+        or isinstance(exc, pymongo.errors.ConnectionFailure)
     )
 
     if retrying: