From b8df85f33095b08db6962f42c58d66711cb04db5 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 09:18:33 -0400
Subject: [PATCH 01/14] add lock to resuming to prevent race

---
 st2common/st2common/services/workflows.py | 139 ++++++++++++++++++----
 1 file changed, 116 insertions(+), 23 deletions(-)

diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index c99fb896b5..d8cb07aa2f 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -386,38 +386,131 @@ def request_resume(ac_ex_db):
 
     wf_ex_db = wf_ex_dbs[0]
 
-    if wf_ex_db.status in statuses.COMPLETED_STATUSES:
-        raise wf_exc.WorkflowExecutionIsCompletedException(str(wf_ex_db.id))
+    # Serialize with handle_action_execution_completion (which also takes this
+    # lock). Without it, a resume request can race with an in-flight completion
+    # and either double-write conductor state or silently no-op below.
+    with coord_svc.get_coordinator(start_heart=True).get_lock(
+        str(wf_ex_db.id).encode()
+    ):
+        # Re-read under the lock — status may have changed since the query.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+
+        LOG.debug(
+            "[%s] DEBUG: WorkflowExecution found - ID: %s, DB status: %s",
+            wf_ac_ex_id,
+            str(wf_ex_db.id),
+            wf_ex_db.status,
+        )
+        LOG.debug(
+            "[%s] DEBUG: WorkflowExecution state status: %s",
+            wf_ac_ex_id,
+            wf_ex_db.state.get("status") if wf_ex_db.state else "N/A",
+        )
+        LOG.debug(
+            "[%s] DEBUG: RUNNING_STATUSES: %s",
+            wf_ac_ex_id,
+            statuses.RUNNING_STATUSES,
+        )
 
-    if wf_ex_db.status in statuses.RUNNING_STATUSES:
-        msg = (
-            '[%s] Workflow execution "%s" is not resumed because it is already active.'
+        if wf_ex_db.status in statuses.COMPLETED_STATUSES:
+            raise wf_exc.WorkflowExecutionIsCompletedException(str(wf_ex_db.id))
+
+        # RESUMING is intentionally treated as "still resumable". Previously
+        # this branch silently returned on RESUMING because RESUMING is in
+        # RUNNING_STATUSES, which made the state self-trapping: if the first
+        # resume attempt failed to cascade (crash, orphaned message, etc.),
+        # every subsequent resume was a no-op. Now RESUMING falls through and
+        # we re-drive the transition below.
+        active_but_not_resuming = [
+            s for s in statuses.RUNNING_STATUSES if s != statuses.RESUMING
+        ]
+
+        LOG.debug(
+            "[%s] DEBUG: Checking if wf_ex_db.status (%s) is in RUNNING_STATUSES: %s",
+            wf_ac_ex_id,
+            wf_ex_db.status,
+            wf_ex_db.status in statuses.RUNNING_STATUSES,
         )
-        LOG.info(msg, wf_ac_ex_id, str(wf_ex_db.id))
-        return
 
-    conductor = deserialize_conductor(wf_ex_db)
+        if wf_ex_db.status in active_but_not_resuming:
+            msg = (
+                '[%s] Workflow execution "%s" is not resumed because it is already active. '
+                "(DB status check: %s is in RUNNING_STATUSES)"
+            )
+            LOG.info(msg, wf_ac_ex_id, str(wf_ex_db.id), wf_ex_db.status)
+            return
 
-    if conductor.get_workflow_status() in statuses.COMPLETED_STATUSES:
-        raise wf_exc.WorkflowExecutionIsCompletedException(str(wf_ex_db.id))
+        LOG.debug("[%s] DEBUG: Deserializing conductor...", wf_ac_ex_id)
+        conductor = deserialize_conductor(wf_ex_db)
+        conductor_status = conductor.get_workflow_status()
 
-    if conductor.get_workflow_status() in statuses.RUNNING_STATUSES:
-        msg = (
-            '[%s] Workflow execution "%s" is not resumed because it is already active.'
+        LOG.debug(
+            "[%s] DEBUG: Conductor deserialized - conductor.get_workflow_status(): %s",
+            wf_ac_ex_id,
+            conductor_status,
         )
-        LOG.info(msg, wf_ac_ex_id, str(wf_ex_db.id))
-        return
 
-    conductor.request_workflow_status(statuses.RESUMING)
+        if conductor.get_workflow_status() in statuses.COMPLETED_STATUSES:
+            raise wf_exc.WorkflowExecutionIsCompletedException(str(wf_ex_db.id))
 
-    # Write the updated workflow status and task flow to the database.
-    wf_ex_db.status = conductor.get_workflow_status()
-    wf_ex_db.state = conductor.workflow_state.serialize()
-    wf_db_access.WorkflowExecution.update(wf_ex_db, publish=False)
-    wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+        LOG.debug(
+            "[%s] DEBUG: Checking if conductor status (%s) is in RUNNING_STATUSES: %s",
+            wf_ac_ex_id,
+            conductor_status,
+            conductor_status in statuses.RUNNING_STATUSES,
+        )
 
-    # Publish status change.
-    wf_db_access.WorkflowExecution.publish_status(wf_ex_db)
+        if conductor.get_workflow_status() in active_but_not_resuming:
+            msg = (
+                '[%s] Workflow execution "%s" is not resumed because it is already active. '
+                "(Conductor status check: %s is in RUNNING_STATUSES)"
+            )
+            LOG.info(msg, wf_ac_ex_id, str(wf_ex_db.id), conductor_status)
+            return
+
+        # If we're re-driving a stuck RESUMING, roll the conductor back to
+        # PAUSED first. Orquesta dedupes redundant transitions, so requesting
+        # RESUMING while already in RESUMING is a no-op inside the state
+        # machine — the transition must actually fire this time.
+        if conductor.get_workflow_status() == statuses.RESUMING:
+            LOG.warning(
+                '[%s] Workflow execution "%s" is already in RESUMING. Rolling '
+                "conductor back to PAUSED and re-issuing resume to break out "
+                "of a stuck resume.",
+                wf_ac_ex_id,
+                str(wf_ex_db.id),
+            )
+            conductor.request_workflow_status(statuses.PAUSED)
+
+        LOG.debug(
+            "[%s] DEBUG: Requesting workflow status change to RESUMING",
+            wf_ac_ex_id,
+        )
+        conductor.request_workflow_status(statuses.RESUMING)
+
+        LOG.debug(
+            "[%s] DEBUG: After requesting RESUMING - conductor status: %s",
+            wf_ac_ex_id,
+            conductor.get_workflow_status(),
+        )
+
+        # Write the updated workflow status and task flow to the database.
+        wf_ex_db.status = conductor.get_workflow_status()
+        wf_ex_db.state = conductor.workflow_state.serialize()
+        LOG.debug(
+            "[%s] DEBUG: Updating WorkflowExecution in database with status: %s",
+            wf_ac_ex_id,
+            wf_ex_db.status,
+        )
+        wf_db_access.WorkflowExecution.update(wf_ex_db, publish=False)
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+
+        # Publish status change.
+        LOG.debug(
+            "[%s] DEBUG: Publishing workflow status change",
+            wf_ac_ex_id,
+        )
+        wf_db_access.WorkflowExecution.publish_status(wf_ex_db)
 
     LOG.info("[%s] Completed processing resume request for workflow.", wf_ac_ex_id)
 

From 0167b8cf94cf3de30b6131a1a9aa20fed9c35f0b Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 09:29:51 -0400
Subject: [PATCH 02/14] pass through resuming to set running

---
 st2common/st2common/services/workflows.py | 15 +++++++++++++--
 1 file changed, 13 insertions(+), 2 deletions(-)

diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index d8cb07aa2f..3c44e9cfc1 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -1179,8 +1179,19 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
     # Refresh records.
     conductor, wf_ex_db = refresh_conductor(str(wf_ex_db.id))
 
-    # If workflow is in requested status, set it to running.
-    if conductor.get_workflow_status() in [statuses.REQUESTED, statuses.SCHEDULED]:
+    # If workflow is in requested, scheduled, or resuming, set it to running.
+    # RESUMING is included so the engine can self-drive a resumed workflow
+    # forward when the normal child-cascade path (handle_action_execution_resume)
+    # never fires — e.g. inquiry responses, or a resume where the child status
+    # change failed to publish. refresh_conductor above pulls the latest state,
+    # so if a child cascade already transitioned to RUNNING this block is a
+    # no-op. The cascade's task-level update_task_state work is orthogonal and
+    # still runs when the cascade eventually fires.
+    if conductor.get_workflow_status() in [
+        statuses.REQUESTED,
+        statuses.SCHEDULED,
+        statuses.RESUMING,
+    ]:
         update_progress(
             wf_ex_db, "Requesting conductor to start running workflow execution."
         )

From 11d780d4ca4e5230d98b77d3e356553e0f65ef7f Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 09:33:33 -0400
Subject: [PATCH 03/14] action execution resume lock

---
 st2common/st2common/services/workflows.py | 36 ++++++++++++++---------
 1 file changed, 22 insertions(+), 14 deletions(-)

diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index 3c44e9cfc1..83ee680981 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -984,24 +984,32 @@ def handle_action_execution_resume(ac_ex_db):
     wf_ex_id = ac_ex_db.context["orquesta"]["workflow_execution_id"]
     task_ex_id = ac_ex_db.context["orquesta"]["task_execution_id"]
 
-    # Get execution records for logging purposes.
-    wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id)
-    task_ex_db = wf_db_access.TaskExecution.get_by_id(task_ex_id)
+    # Serialize with handle_action_execution_completion / request_resume, which
+    # also take this lock. Without it, resume_workflow_execution can race with
+    # a concurrent completion and both write conductor state with stale
+    # revisions — producing thrash under contention and, worst case, leaving
+    # the workflow in an inconsistent RESUMING/RUNNING split.
+    with coord_svc.get_coordinator(start_heart=True).get_lock(str(wf_ex_id).encode()):
+        # Get execution records for logging purposes.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id)
+        task_ex_db = wf_db_access.TaskExecution.get_by_id(task_ex_id)
 
-    msg = 'Handling resume of action execution "%s" for task "%s", route "%s".'
-    update_progress(
-        wf_ex_db,
-        msg % (str(ac_ex_db.id), task_ex_db.task_id, str(task_ex_db.task_route)),
-    )
+        msg = 'Handling resume of action execution "%s" for task "%s", route "%s".'
+        update_progress(
+            wf_ex_db,
+            msg % (str(ac_ex_db.id), task_ex_db.task_id, str(task_ex_db.task_route)),
+        )
 
-    # Updat task execution to running.
-    resume_task_execution(task_ex_id)
+        # Updat task execution to running.
+        resume_task_execution(task_ex_id)
 
-    # Update workflow execution to running.
-    resume_workflow_execution(wf_ex_id, task_ex_id)
+        # Update workflow execution to running.
+        resume_workflow_execution(wf_ex_id, task_ex_id)
 
-    # If action execution has a parent, cascade status change upstream and do not publish
-    # the status change because we do not want to trigger resume of other peer subworkflows.
+    # Cascade upstream OUTSIDE the current lock. The recursive call acquires
+    # the parent's own per-workflow lock; releasing ours first avoids holding
+    # a chain of N locks in deeply nested subworkflow scenarios. Direction is
+    # child → parent only, so no deadlock cycle.
     if "parent" in ac_ex_db.context:
         parent_ac_ex_id = ac_ex_db.context["parent"]["execution_id"]
         parent_ac_ex_db = ex_db_access.ActionExecution.get_by_id(parent_ac_ex_id)

From dcbdb0f0f8b82d4299f2bcac8e6aff8048079a69 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 09:49:55 -0400
Subject: [PATCH 04/14] time based request_next_tasks death

---
 st2common/st2common/config.py              | 10 ++++++++
 st2common/st2common/exceptions/workflow.py | 14 ++++++++++
 st2common/st2common/services/workflows.py  | 30 ++++++++++++++++++++++
 3 files changed, 54 insertions(+)

diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py
index 28b0c062ec..381e846d4e 100644
--- a/st2common/st2common/config.py
+++ b/st2common/st2common/config.py
@@ -931,6 +931,16 @@ def register_opts(ignore_errors=False):
             default=2,
             help="Time interval between subsequent queries to check executions handled by WFE.",
         ),
+        cfg.IntOpt(
+            "request_next_tasks_deadline_sec",
+            default=120,
+            help="Maximum wall-clock seconds a single request_next_tasks() call "
+            "may run before the workflow is failed to release the per-workflow "
+            "coordination lock. Guards against runaway conductor state or a "
+            "pathological chain of no-op tasks holding the lock indefinitely. "
+            "Typical calls complete in milliseconds; the default is intentionally "
+            "generous.",
+        ),
     ]
 
     do_register_opts(
diff --git a/st2common/st2common/exceptions/workflow.py b/st2common/st2common/exceptions/workflow.py
index 370030c6ec..b429a189f7 100644
--- a/st2common/st2common/exceptions/workflow.py
+++ b/st2common/st2common/exceptions/workflow.py
@@ -93,3 +93,17 @@ def __init__(self, wf_ex_id):
 class WorkflowExecutionRerunException(st2_exc.StackStormBaseException):
     def __init__(self, msg):
         Exception.__init__(self, msg)
+
+
+class WorkflowExecutionRequestNextTasksException(st2_exc.StackStormBaseException):
+    def __init__(self, wf_ex_id, deadline_sec, iterations):
+        Exception.__init__(
+            self,
+            'Workflow execution "%s" did not converge within %s seconds '
+            "(after %s iterations) of request_next_tasks. Failing to release "
+            "the per-workflow coordination lock."
+            % (wf_ex_id, deadline_sec, iterations),
+        )
+        self.wf_ex_id = wf_ex_id
+        self.deadline_sec = deadline_sec
+        self.iterations = iterations
diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index 83ee680981..016a04e8e4 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -20,6 +20,7 @@
 import retrying
 import six
 import sys
+import time
 import traceback
 
 from orquesta import conducting
@@ -1183,6 +1184,8 @@ def update_task_state(
 )
 def request_next_tasks(wf_ex_db, task_ex_id=None):
     iteration = 0
+    deadline_sec = cfg.CONF.workflow_engine.request_next_tasks_deadline_sec
+    deadline = time.time() + deadline_sec
 
     # Refresh records.
     conductor, wf_ex_db = refresh_conductor(str(wf_ex_db.id))
@@ -1237,6 +1240,33 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
     # task with no action execution defined, the task execution will complete
     # immediately with a new set of tasks available.
     while next_tasks:
+        # Deadline guard. request_next_tasks holds the per-workflow coord lock
+        # via handle_action_execution_completion, so a runaway loop here blocks
+        # every other message for this workflow until the lock is released.
+        # Fail loudly instead of holding forever. Typical calls take
+        # milliseconds; the default deadline is generous.
+        if time.time() > deadline:
+            msg = (
+                'request_next_tasks for workflow "%s" exceeded deadline of %ss '
+                "at iteration %s. Failing the workflow to release the "
+                "coordination lock. Undispatched task set: %s"
+            )
+            tasks_list = ", ".join(
+                ["%s (route %s)" % (t["id"], str(t["route"])) for t in next_tasks]
+            )
+            update_progress(
+                wf_ex_db,
+                msg % (str(wf_ex_db.id), deadline_sec, iteration, tasks_list),
+                severity="error",
+            )
+            fail_workflow_execution(
+                str(wf_ex_db.id),
+                wf_exc.WorkflowExecutionRequestNextTasksException(
+                    str(wf_ex_db.id), deadline_sec, iteration
+                ),
+            )
+            return
+
         msg = "Identified the following set of tasks to execute next: %s"
         tasks_list = ", ".join(
             ["%s (route %s)" % (t["id"], str(t["route"])) for t in next_tasks]

From 10e0afdfb3e23bb925ad5333211af10d92b61c04 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 10:00:26 -0400
Subject: [PATCH 05/14] handle_workflow_execution takes a coordinator lock

---
 st2actions/st2actions/workflows/workflows.py | 20 +++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/st2actions/st2actions/workflows/workflows.py b/st2actions/st2actions/workflows/workflows.py
index 6672069e6f..1e7c372e18 100644
--- a/st2actions/st2actions/workflows/workflows.py
+++ b/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.

From 1c9431b751e442278e90af9074cc1cf726ea9f0a Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 20:03:03 -0400
Subject: [PATCH 06/14] mongo retry max time

---
 st2common/st2common/services/workflows.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index 016a04e8e4..4f066255b6 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -319,6 +319,7 @@ def request(wf_def, ac_ex_db, st2_ctx, notify_cfg=None):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -364,6 +365,7 @@ def request_pause(ac_ex_db):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -520,6 +522,7 @@ def request_resume(ac_ex_db):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -577,6 +580,7 @@ def request_cancellation(ac_ex_db):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -807,6 +811,7 @@ def eval_action_execution_delay(task_ex_req, ac_ex_req, itemized=False):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1114,6 +1119,7 @@ def refresh_conductor(wf_ex_id):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1173,6 +1179,7 @@ def update_task_state(
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1365,6 +1372,7 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1452,6 +1460,7 @@ def update_task_execution(task_ex_id, ac_ex_status, ac_ex_result=None, ac_ex_ctx
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1478,6 +1487,7 @@ def resume_task_execution(task_ex_id):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1500,6 +1510,7 @@ def update_workflow_execution(wf_ex_id):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )
@@ -1525,6 +1536,7 @@ def resume_workflow_execution(wf_ex_id, task_ex_id):
 
 @retrying.retry(
     retry_on_exception=wf_exc.retry_on_transient_db_errors,
+    stop_max_delay=cfg.CONF.workflow_engine.retry_stop_max_msec,
     wait_fixed=cfg.CONF.workflow_engine.retry_wait_fixed_msec,
     wait_jitter_max=cfg.CONF.workflow_engine.retry_max_jitter_msec,
 )

From 1511b076bd07b6013e29335db3e1bbf2bc5bc7eb Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Tue, 25 Aug 2026 20:18:50 -0400
Subject: [PATCH 07/14] add mongo errors to retry and timeout logic

---
 st2common/st2common/exceptions/workflow.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/st2common/st2common/exceptions/workflow.py b/st2common/st2common/exceptions/workflow.py
index b429a189f7..f3c68a0345 100644
--- a/st2common/st2common/exceptions/workflow.py
+++ b/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:

From e33fd53bd241c77c252d784151d1b3b4842a371f Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Wed, 26 Aug 2026 11:08:49 -0400
Subject: [PATCH 08/14] workflow engine race condition and pool-wedge fixes

per-workflow coord lock on the resume paths and on
handle_workflow_execution, RESUMING self-transition
in request_next_tasks, wall-clock deadline on request_next_tasks,
stop_max_delay on all retry_on_transient_db_errors decorators, and
widened retry_on_connection_errors to cover pymongo.ConnectionFailure.
---
 CHANGELOG.rst | 119 ++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 119 insertions(+)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 5719ee2d16..5b1b05b3e3 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -3,6 +3,125 @@ Changelog
 
 in development
 --------------
+* Workflow engine race condition and pool-wedge fixes
+  Contributed by @guzzijones12.
+
+  **Symptom.** The workflow engine's ``BufferedDispatcher`` would log
+  ``BufferedDispatcher pool "<id>" has been busy with no free threads for
+  more than 60 seconds`` repeatedly, and the engine's 50-thread green
+  pool would remain saturated. The trigger was one workflow — often one
+  paused at an inquiry — whose in-flight action-execution completions
+  raced with concurrent workflow-level status messages, causing every
+  affected worker to spin on Mongo write conflicts under the per-workflow
+  coordination lock.
+
+  **Root cause.** Under high load the ``retry_on_transient_db_errors``
+  decorators (which cover ``StackStormDBObjectWriteConflictError``) had
+  no ``stop_max_delay`` — so once two engines started colliding on a hot
+  workflow's document they retried each other forever, holding coord
+  locks the entire time. Eventlet scheduler pressure from all those
+  spinning greenthreads then starved the tooz heartbeat greenthread,
+  which in turn caused the Valkey/Redis lock TTL to expire while the
+  holder was still executing — a second engine could then acquire the
+  supposedly-held lock, producing more write conflicts, and the feedback
+  loop escalated until the pool was fully wedged.
+
+  **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`` — wall-clock deadline. New config
+    ``workflow_engine.request_next_tasks_deadline_sec`` (default 120).
+    When exceeded, the workflow is failed via ``fail_workflow_execution``
+    and the coord lock releases through the normal error-return path.
+    Catches: infinite no-op-task chains, non-converging conductor state,
+    orquesta+DB divergence caused by a stuck resume.
+
+  * ``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.
+
+  **Operator note — the "BufferedDispatcher pool busy" log.** Prior to
+  these fixes, the pool-busy message was almost always the visible
+  symptom of the deadlock class described above. After these fixes
+  worker dwell time is bounded (60 s retry ceilings on Mongo/tooz plus a
+  120 s ``request_next_tasks_deadline_sec`` guard), so a stuck workflow
+  can no longer camp a worker slot forever. If the log fires again,
+  treat it as **real load** — 50+ workflows genuinely doing work
+  concurrently for more than a minute — and scale the workflow-engine
+  replicas out or raise the dispatcher pool size.
+
 * implemented zstandard compression for parameters and results. #5995
   contributed by @guzzijones12
 

From 241a7c122ce6d8c38026987ec6f2093e8539196a Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Wed, 26 Aug 2026 15:17:57 -0400
Subject: [PATCH 09/14] update configgen

---
 conf/st2.conf.sample | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample
index 27a2eb0a86..a70a541e05 100644
--- a/conf/st2.conf.sample
+++ b/conf/st2.conf.sample
@@ -394,6 +394,8 @@ exit_still_active_check = 300
 gc_max_idle_sec = 0
 # Location of the logging configuration file.
 logging = /etc/st2/logging.workflowengine.conf
+# Maximum wall-clock seconds a single request_next_tasks() call may run before the workflow is failed to release the per-workflow coordination lock. Guards against runaway conductor state or a pathological chain of no-op tasks holding the lock indefinitely. Typical calls complete in milliseconds; the default is intentionally generous.
+request_next_tasks_deadline_sec = 120
 # Max jitter interval to smooth out retries.
 retry_max_jitter_msec = 1000
 # Max time to stop retrying.

From 98b1985451e6b4a7dcf3eeb9eb77ffd55422263e Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Wed, 26 Aug 2026 17:11:15 -0400
Subject: [PATCH 10/14] fix deadlock on inquiry with redis

---
 st2common/st2common/services/workflows.py | 20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index 4f066255b6..d706d9735b 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -508,12 +508,20 @@ def request_resume(ac_ex_db):
         wf_db_access.WorkflowExecution.update(wf_ex_db, publish=False)
         wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
 
-        # Publish status change.
-        LOG.debug(
-            "[%s] DEBUG: Publishing workflow status change",
-            wf_ac_ex_id,
-        )
-        wf_db_access.WorkflowExecution.publish_status(wf_ex_db)
+    # Publish the status change OUTSIDE the per-workflow lock. Publishing a
+    # WorkflowExecutionDB re-dispatches into handle_workflow_execution, which
+    # takes this same per-workflow lock. In the engine that is a separate
+    # message/greenthread, but the unit-test transport (MockWorkflowExecution-
+    # Publisher) invokes the handler synchronously on this thread — so a publish
+    # while still holding the lock would re-enter and self-deadlock on a real
+    # (non-reentrant) tooz backend. Releasing first keeps the critical section
+    # to the DB write and lets the resulting RESUMING message acquire the lock
+    # cleanly.
+    LOG.debug(
+        "[%s] DEBUG: Publishing workflow status change",
+        wf_ac_ex_id,
+    )
+    wf_db_access.WorkflowExecution.publish_status(wf_ex_db)
 
     LOG.info("[%s] Completed processing resume request for workflow.", wf_ac_ex_id)
 

From d82297c2a37904c3f9796e03b666b9ecc867d3cb Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Thu, 27 Aug 2026 12:56:23 -0400
Subject: [PATCH 11/14] cap concurrency

---
 CHANGELOG.rst                                 | 59 ++++++--------
 conf/st2.conf.sample                          |  4 +-
 .../tests/unit/test_with_items.py             | 80 +++++++++++++++++++
 st2common/st2common/config.py                 | 17 ++--
 st2common/st2common/exceptions/workflow.py    | 14 ----
 st2common/st2common/services/workflows.py     | 71 +++++++++-------
 6 files changed, 154 insertions(+), 91 deletions(-)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 5b1b05b3e3..e02e7b03fe 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -3,28 +3,17 @@ Changelog
 
 in development
 --------------
-* Workflow engine race condition and pool-wedge fixes
+* Workflow engine race condition fixes
   Contributed by @guzzijones12.
 
-  **Symptom.** The workflow engine's ``BufferedDispatcher`` would log
-  ``BufferedDispatcher pool "<id>" has been busy with no free threads for
-  more than 60 seconds`` repeatedly, and the engine's 50-thread green
-  pool would remain saturated. The trigger was one workflow — often one
-  paused at an inquiry — whose in-flight action-execution completions
-  raced with concurrent workflow-level status messages, causing every
-  affected worker to spin on Mongo write conflicts under the per-workflow
-  coordination lock.
-
-  **Root cause.** Under high load the ``retry_on_transient_db_errors``
-  decorators (which cover ``StackStormDBObjectWriteConflictError``) had
-  no ``stop_max_delay`` — so once two engines started colliding on a hot
-  workflow's document they retried each other forever, holding coord
-  locks the entire time. Eventlet scheduler pressure from all those
-  spinning greenthreads then starved the tooz heartbeat greenthread,
-  which in turn caused the Valkey/Redis lock TTL to expire while the
-  holder was still executing — a second engine could then acquire the
-  supposedly-held lock, producing more write conflicts, and the feedback
-  loop escalated until the pool was fully wedged.
+  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):
 
@@ -58,12 +47,20 @@ in development
     Prevents holding a chain of N locks up the parent tree in deeply
     nested subworkflow scenarios.
 
-  * ``request_next_tasks`` — wall-clock deadline. New config
-    ``workflow_engine.request_next_tasks_deadline_sec`` (default 120).
-    When exceeded, the workflow is failed via ``fail_workflow_execution``
-    and the coord lock releases through the normal error-return path.
-    Catches: infinite no-op-task chains, non-converging conductor state,
-    orquesta+DB divergence caused by a stuck resume.
+  * ``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.max_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
@@ -112,16 +109,6 @@ in development
     network hiccup) are absorbed by the per-call retry envelope. Longer
     outages produce failed workflows, not a crashed engine.
 
-  **Operator note — the "BufferedDispatcher pool busy" log.** Prior to
-  these fixes, the pool-busy message was almost always the visible
-  symptom of the deadlock class described above. After these fixes
-  worker dwell time is bounded (60 s retry ceilings on Mongo/tooz plus a
-  120 s ``request_next_tasks_deadline_sec`` guard), so a stuck workflow
-  can no longer camp a worker slot forever. If the log fires again,
-  treat it as **real load** — 50+ workflows genuinely doing work
-  concurrently for more than a minute — and scale the workflow-engine
-  replicas out or raise the dispatcher pool size.
-
 * implemented zstandard compression for parameters and results. #5995
   contributed by @guzzijones12
 
diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample
index a70a541e05..cef8967a2d 100644
--- a/conf/st2.conf.sample
+++ b/conf/st2.conf.sample
@@ -394,8 +394,8 @@ exit_still_active_check = 300
 gc_max_idle_sec = 0
 # Location of the logging configuration file.
 logging = /etc/st2/logging.workflowengine.conf
-# Maximum wall-clock seconds a single request_next_tasks() call may run before the workflow is failed to release the per-workflow coordination lock. Guards against runaway conductor state or a pathological chain of no-op tasks holding the lock indefinitely. Typical calls complete in milliseconds; the default is intentionally generous.
-request_next_tasks_deadline_sec = 120
+# 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.
+max_with_items_concurrency = 0
 # Max jitter interval to smooth out retries.
 retry_max_jitter_msec = 1000
 # Max time to stop retrying.
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 072a9bdfae..87583b8125 100644
--- a/contrib/runners/orquesta_runner/tests/unit/test_with_items.py
+++ b/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 max_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(
+            "max_with_items_concurrency", concurrency, group="workflow_engine"
+        )
+        self.addCleanup(
+            cfg.CONF.clear_override,
+            "max_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",
diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py
index 381e846d4e..25b246473d 100644
--- a/st2common/st2common/config.py
+++ b/st2common/st2common/config.py
@@ -932,14 +932,15 @@ def register_opts(ignore_errors=False):
             help="Time interval between subsequent queries to check executions handled by WFE.",
         ),
         cfg.IntOpt(
-            "request_next_tasks_deadline_sec",
-            default=120,
-            help="Maximum wall-clock seconds a single request_next_tasks() call "
-            "may run before the workflow is failed to release the per-workflow "
-            "coordination lock. Guards against runaway conductor state or a "
-            "pathological chain of no-op tasks holding the lock indefinitely. "
-            "Typical calls complete in milliseconds; the default is intentionally "
-            "generous.",
+            "max_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.",
         ),
     ]
 
diff --git a/st2common/st2common/exceptions/workflow.py b/st2common/st2common/exceptions/workflow.py
index f3c68a0345..aa89251d59 100644
--- a/st2common/st2common/exceptions/workflow.py
+++ b/st2common/st2common/exceptions/workflow.py
@@ -96,17 +96,3 @@ def __init__(self, wf_ex_id):
 class WorkflowExecutionRerunException(st2_exc.StackStormBaseException):
     def __init__(self, msg):
         Exception.__init__(self, msg)
-
-
-class WorkflowExecutionRequestNextTasksException(st2_exc.StackStormBaseException):
-    def __init__(self, wf_ex_id, deadline_sec, iterations):
-        Exception.__init__(
-            self,
-            'Workflow execution "%s" did not converge within %s seconds '
-            "(after %s iterations) of request_next_tasks. Failing to release "
-            "the per-workflow coordination lock."
-            % (wf_ex_id, deadline_sec, iterations),
-        )
-        self.wf_ex_id = wf_ex_id
-        self.deadline_sec = deadline_sec
-        self.iterations = iterations
diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index d706d9735b..2e82940e0b 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -20,7 +20,6 @@
 import retrying
 import six
 import sys
-import time
 import traceback
 
 from orquesta import conducting
@@ -1104,6 +1103,43 @@ def handle_action_execution_completion(ac_ex_db):
         update_workflow_execution(wf_ex_id)
 
 
+def _apply_default_with_items_concurrency(conductor):
+    # Inject a default concurrency into with-items tasks that don't specify one,
+    # so orquesta's conductor dispatches items in bounded batches (via its native
+    # availability = concurrency - active_items math) instead of handing back every
+    # item at once. Without this, an unbounded with-items dispatches all N item
+    # action executions in a single request_next_tasks pass while holding the
+    # per-workflow coordination lock. This mutation is in-memory only —
+    # update_execution_records never writes wf_ex_db.spec, so the stored workflow
+    # definition stays pristine and this is re-applied fresh on each deserialize.
+    cap = cfg.CONF.workflow_engine.max_with_items_concurrency
+
+    if not cap or cap <= 0:
+        return conductor
+
+    try:
+        task_specs = conductor.spec.tasks
+    except AttributeError:
+        return conductor
+
+    for _, task_spec in task_specs.items():
+        if not task_spec.has_items():
+            continue
+
+        items_spec = task_spec.get_items_spec()
+
+        # Read concurrency from the underlying spec dict (orquesta's __getattr__
+        # resolves the scalar from there) and, when absent, write the default back
+        # into that same dict. The conductor evaluates concurrency off a .copy() of
+        # the task spec (conducting.py get_task), and copy() round-trips through
+        # serialize()/deserialize() using the raw spec dict — a plain Python
+        # attribute would be dropped, so the dict is the only thing that survives.
+        if items_spec is not None and items_spec.spec.get("concurrency") is None:
+            items_spec.spec["concurrency"] = cap
+
+    return conductor
+
+
 def deserialize_conductor(wf_ex_db):
     data = {
         "spec": wf_ex_db.spec,
@@ -1115,7 +1151,9 @@ def deserialize_conductor(wf_ex_db):
         "errors": wf_ex_db.errors,
     }
 
-    return conducting.WorkflowConductor.deserialize(data)
+    conductor = conducting.WorkflowConductor.deserialize(data)
+
+    return _apply_default_with_items_concurrency(conductor)
 
 
 def refresh_conductor(wf_ex_id):
@@ -1199,8 +1237,6 @@ def update_task_state(
 )
 def request_next_tasks(wf_ex_db, task_ex_id=None):
     iteration = 0
-    deadline_sec = cfg.CONF.workflow_engine.request_next_tasks_deadline_sec
-    deadline = time.time() + deadline_sec
 
     # Refresh records.
     conductor, wf_ex_db = refresh_conductor(str(wf_ex_db.id))
@@ -1255,33 +1291,6 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
     # task with no action execution defined, the task execution will complete
     # immediately with a new set of tasks available.
     while next_tasks:
-        # Deadline guard. request_next_tasks holds the per-workflow coord lock
-        # via handle_action_execution_completion, so a runaway loop here blocks
-        # every other message for this workflow until the lock is released.
-        # Fail loudly instead of holding forever. Typical calls take
-        # milliseconds; the default deadline is generous.
-        if time.time() > deadline:
-            msg = (
-                'request_next_tasks for workflow "%s" exceeded deadline of %ss '
-                "at iteration %s. Failing the workflow to release the "
-                "coordination lock. Undispatched task set: %s"
-            )
-            tasks_list = ", ".join(
-                ["%s (route %s)" % (t["id"], str(t["route"])) for t in next_tasks]
-            )
-            update_progress(
-                wf_ex_db,
-                msg % (str(wf_ex_db.id), deadline_sec, iteration, tasks_list),
-                severity="error",
-            )
-            fail_workflow_execution(
-                str(wf_ex_db.id),
-                wf_exc.WorkflowExecutionRequestNextTasksException(
-                    str(wf_ex_db.id), deadline_sec, iteration
-                ),
-            )
-            return
-
         msg = "Identified the following set of tasks to execute next: %s"
         tasks_list = ", ".join(
             ["%s (route %s)" % (t["id"], str(t["route"])) for t in next_tasks]

From 054f95731ca5d4e46ced8c7125e0794c3f515873 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Thu, 27 Aug 2026 15:59:21 -0400
Subject: [PATCH 12/14] fix orquesta hash

---
 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 cc1f861589b75b0500e4a931797422aea464b1b4 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Thu, 27 Aug 2026 16:18:39 -0400
Subject: [PATCH 13/14] default concurrency name change

---
 CHANGELOG.rst                                               | 2 +-
 conf/st2.conf.sample                                        | 2 +-
 .../runners/orquesta_runner/tests/unit/test_with_items.py   | 6 +++---
 st2common/st2common/config.py                               | 2 +-
 st2common/st2common/services/workflows.py                   | 2 +-
 5 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index e02e7b03fe..5dde3f3c6e 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -52,7 +52,7 @@ in development
     ``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.max_with_items_concurrency`` (default 0 = disabled)
+    ``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
diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample
index cef8967a2d..de8b6b868a 100644
--- a/conf/st2.conf.sample
+++ b/conf/st2.conf.sample
@@ -395,7 +395,7 @@ gc_max_idle_sec = 0
 # Location of the logging configuration file.
 logging = /etc/st2/logging.workflowengine.conf
 # 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.
-max_with_items_concurrency = 0
+default_with_items_concurrency = 0
 # Max jitter interval to smooth out retries.
 retry_max_jitter_msec = 1000
 # Max time to stop retrying.
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 87583b8125..386b22d6ee 100644
--- a/contrib/runners/orquesta_runner/tests/unit/test_with_items.py
+++ b/contrib/runners/orquesta_runner/tests/unit/test_with_items.py
@@ -334,17 +334,17 @@ def test_with_items_concurrency(self):
 
     def test_with_items_default_concurrency(self):
         # The workflow definition sets no concurrency on the with items task. The
-        # configured max_with_items_concurrency is injected as the default so 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(
-            "max_with_items_concurrency", concurrency, group="workflow_engine"
+            "default_with_items_concurrency", concurrency, group="workflow_engine"
         )
         self.addCleanup(
             cfg.CONF.clear_override,
-            "max_with_items_concurrency",
+            "default_with_items_concurrency",
             group="workflow_engine",
         )
 
diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py
index 25b246473d..2f81b2edd0 100644
--- a/st2common/st2common/config.py
+++ b/st2common/st2common/config.py
@@ -932,7 +932,7 @@ def register_opts(ignore_errors=False):
             help="Time interval between subsequent queries to check executions handled by WFE.",
         ),
         cfg.IntOpt(
-            "max_with_items_concurrency",
+            "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 "
diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py
index 2e82940e0b..476ecfff9b 100644
--- a/st2common/st2common/services/workflows.py
+++ b/st2common/st2common/services/workflows.py
@@ -1112,7 +1112,7 @@ def _apply_default_with_items_concurrency(conductor):
     # per-workflow coordination lock. This mutation is in-memory only —
     # update_execution_records never writes wf_ex_db.spec, so the stored workflow
     # definition stays pristine and this is re-applied fresh on each deserialize.
-    cap = cfg.CONF.workflow_engine.max_with_items_concurrency
+    cap = cfg.CONF.workflow_engine.default_with_items_concurrency
 
     if not cap or cap <= 0:
         return conductor

From 8ba750416f384f2165f11972e71d6b01dc949d88 Mon Sep 17 00:00:00 2001
From: guzzijones12 <guzzijones12@gmail.com>
Date: Thu, 27 Aug 2026 16:53:34 -0400
Subject: [PATCH 14/14] run make configgen

---
 conf/st2.conf.sample | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample
index de8b6b868a..19503b5f17 100644
--- a/conf/st2.conf.sample
+++ b/conf/st2.conf.sample
@@ -388,14 +388,14 @@ 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.
 gc_max_idle_sec = 0
 # Location of the logging configuration file.
 logging = /etc/st2/logging.workflowengine.conf
-# 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
 # Max jitter interval to smooth out retries.
 retry_max_jitter_msec = 1000
 # Max time to stop retrying.