diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 5719ee2d16..5dde3f3c6e 100644 --- a/CHANGELOG.rst +++ b/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 diff --git a/conf/st2.conf.sample b/conf/st2.conf.sample index 27a2eb0a86..19503b5f17 100644 --- a/conf/st2.conf.sample +++ b/conf/st2.conf.sample @@ -388,6 +388,8 @@ logging = /etc/st2/logging.timersengine.conf webui_base_url = https://localhost [workflow_engine] +# 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. 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..386b22d6ee 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 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", 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" } ], 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. diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py index 28b0c062ec..2f81b2edd0 100644 --- a/st2common/st2common/config.py +++ b/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( diff --git a/st2common/st2common/exceptions/workflow.py b/st2common/st2common/exceptions/workflow.py index 370030c6ec..aa89251d59 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: diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py index c99fb896b5..476ecfff9b 100644 --- a/st2common/st2common/services/workflows.py +++ b/st2common/st2common/services/workflows.py @@ -318,6 +318,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, ) @@ -363,6 +364,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, ) @@ -386,37 +388,138 @@ 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. + 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 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) @@ -426,6 +529,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, ) @@ -483,6 +587,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, ) @@ -713,6 +818,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, ) @@ -891,24 +997,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) @@ -989,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.default_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, @@ -1000,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): @@ -1012,6 +1165,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, ) @@ -1071,6 +1225,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, ) @@ -1086,8 +1241,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." ) @@ -1223,6 +1389,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, ) @@ -1310,6 +1477,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, ) @@ -1336,6 +1504,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, ) @@ -1358,6 +1527,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, ) @@ -1383,6 +1553,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, )