Add design document for parallel step execution in workflow engine - #1444
Add design document for parallel step execution in workflow engine#1444pboers1988 wants to merge 3 commits into
Conversation
Merging this PR will not alter performance
Comparing 
Footnotes
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #1444 +/- ##
==========================================
- Coverage 89.76% 89.50% -0.27%
==========================================
Files 276 277 +1
Lines 14404 14726 +322
Branches 1393 1431 +38
==========================================
+ Hits 12930 13180 +250
- Misses 1190 1251 +61
- Partials 284 295 +11
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
| # Validate no inputsteps in branches | ||
| for branch_idx, branch in enumerate(branches): | ||
| for s in branch: | ||
| if s.form is not None: | ||
| raise ValueError( | ||
| f"Parallel branches must not contain inputsteps. " | ||
| f"Found inputstep '{s.name}' in branch {branch_idx}." | ||
| ) |
There was a problem hiding this comment.
Should this also disallow callback steps?
There was a problem hiding this comment.
Need to check if checking for a form is an implicit check for callback steps.
| def _exec_parallel_branches( | ||
| branches: list[StepList], | ||
| initial_state: State, | ||
| dblogstep: StepLogFuncInternal, |
There was a problem hiding this comment.
dblogstep is not actually called. Currently, run_branch seems to default to _noop_dblogstep which no doubt runs just fine :)
I'm not sure if the logstep should be done within each branch as this shares the same sqlalchemy session between them. If one branch then does a rollback, the other branches cannot commit anything.
I'd probably make it so that the point where the parallelization happens, is in charge of writing to the database; with either an aggregated result of all successfully executed steps, or a partial result with N successfully executed steps and M failed steps. This way the algorithm can be made capable of retrying a parallel step which has been partially completed (without having to assume that the parallelized step is idempotent)
There was a problem hiding this comment.
Good point. Probably should define a test-case for this :)
| 1. **Static parallel** — a fixed set of independent branches known at definition time (e.g., always provision port A and port B together). | ||
| 2. **Dynamic parallel** — a variable number of branches driven by a runtime iterable (e.g., provision one port per item in `state["ports"]`). |
There was a problem hiding this comment.
I think that the subworkflow execution is worth describing here; or generalizing the concept of "parallel branches" to "parallel steplists".
For example instead of "always provisioning port A and port B", one could design a workflow that "always triggers subworkflows X and Y". The "parent" workflow remains static so long as none of the subworkflows use dynamic parallelism.
Extending this to dynamic parallelism; instead of "provision one port per item in state["ports"]" you could have a workflow that "always triggers subworkflow X per item in state["ports"]".
There was a problem hiding this comment.
Drawn the different scenarios that this would make possible.
Parallel with n=1
Parallel with n=m
Nested parallel
|
|
||
| - Enable parallel execution of independent steps within a workflow | ||
| - Extend the workflow DSL with fork/join semantics for both static and dynamic branching | ||
| - Remain **fully backwards compatible** — existing workflows, APIs, and database schema unchanged |
There was a problem hiding this comment.
DB schema will be different; the state for parallel branches will be stored in rows, and relationships between steps will be maintained in a separate table, supporting nested relations (similar to subscription_instance_relations)
| - **Dict items**: each item dict is merged into the branch's initial state (`initial_state | item`) | ||
| - **Scalar items**: injected as `{"item": <value>, "item_index": <int>}` | ||
| - **Seed keys are stripped from output** before the join — they are input-only and cannot cause key-conflict errors | ||
| - Branches must still write to **distinct output keys**; use the item's natural identifier in the key name |
There was a problem hiding this comment.
A similar pattern to the subscription_instance_relations table. Use an association proxy.
|
|
||
| ### 4.3 Execution Strategy | ||
|
|
||
| Parallel branches execute using `concurrent.futures.ThreadPoolExecutor` (already available in the codebase via the thread-based workflow execution). Each branch runs in its own thread, but all branches share the same database session scope (the parallel group acts as a single transaction boundary). |
There was a problem hiding this comment.
As discussed yesterday we'll execute/schedule parallel branches (steplists) to run on workers, but as workflows (instead of per step) to reduce the scope of this change somewhat. But that means most of what's written here needs to be updated
(for trivial branching, e.g. when the parallel steplists are n=1, we could still consider the threadpool approach)
Another note: how much of the parallel step functionality do we want to support when running the orchestrator in threadpool mode?
| **Operator precedence note:** Python's `>>` binds tighter than `|`, so: | ||
| ```python | ||
| begin >> step_a | begin >> step_b | ||
| # parses as: | ||
| (begin >> step_a) | (begin >> step_b) | ||
| ``` | ||
| This is exactly the grouping we want — each branch is fully built with `>>` before `|` combines them. |
There was a problem hiding this comment.
While it's cool to do this with bitwise operators, I think this a more confusing syntax.
But we could implement it for the first experimental version, and take a survey to see what people think.
|
|
||
| - Enable parallel execution of independent steps within a workflow | ||
| - Extend the workflow DSL with fork/join semantics for both static and dynamic branching | ||
| - Remain **fully backwards compatible** — existing workflows, APIs, and database schema unchanged |
There was a problem hiding this comment.
Introduce an association proxy table so we can link multiple parallel_steps to the same workflow. They should contain a reference to the step that branches and an order_id or other mechanism to map the step to the correct parallel steplist.
| - **Dict items**: each item dict is merged into the branch's initial state (`initial_state | item`) | ||
| - **Scalar items**: injected as `{"item": <value>, "item_index": <int>}` | ||
| - **Seed keys are stripped from output** before the join — they are input-only and cannot cause key-conflict errors | ||
| - Branches must still write to **distinct output keys**; use the item's natural identifier in the key name |
There was a problem hiding this comment.
A similar pattern to the subscription_instance_relations table. Use an association proxy.
| - All branches run to completion — no early termination on first failure | ||
| - After all futures resolve, `_join_results` merges results in the parent thread | ||
|
|
||
| ### Phase 4: Celery per-step re-queue (future sketch) |
There was a problem hiding this comment.
This is too complex. We want to parallelize the running of steplists. Not single steps. In other words, when the engine encouters a branch. Create as many steplist celery tasks as branches that are needed. Perhaps to scale correctly we need a different queue. To get this correct.
| 4. **Branch-level timeout**: Per-branch timeout with cancellation | ||
| 5. **Dynamic branching**: Generate branches from state (e.g., one branch per subscription) | ||
| 6. **Branch-scoped transactions**: Each branch in its own DB transaction for independent rollback | ||
| 7. **inputstep support**: Allow user interaction within parallel branches (serialize/deserialize branch state) |
| 5. **Dynamic branching**: Generate branches from state (e.g., one branch per subscription) | ||
| 6. **Branch-scoped transactions**: Each branch in its own DB transaction for independent rollback | ||
| 7. **inputstep support**: Allow user interaction within parallel branches (serialize/deserialize branch state) | ||
| 8. **Progress tracking**: Report per-branch progress to UI via WebSocket |
There was a problem hiding this comment.
Use the current broadcast mechanism.
a5649bd to
f497785
Compare
| # Different routes/queues so we can assign them priorities | ||
| from orchestrator.settings import app_settings | ||
|
|
||
| parallel_queue = app_settings.PARALLEL_BRANCH_QUEUE or "new_tasks" |
There was a problem hiding this comment.
What about workflows? Add a new config option and similar pattern for workflows
| local_logger.info("Resume workflow", process_id=process_id) | ||
| return resume_process(process_id, user=user) | ||
|
|
||
| @celery_task(name=EXECUTE_PARALLEL_BRANCH) # type: ignore |
There was a problem hiding this comment.
Same as other comment about Parallel execution
| if isinstance(other, ParallelStepList): | ||
| name = other.name or _auto_parallel_name(other) | ||
| par_step = _make_parallel_step(name, other.branches) | ||
| return StepList([*self, par_step]) | ||
|
|
||
| # Named parallel via dict: ... >> {"Name": (begin >> a) | (begin >> b)} >> ... | ||
| if isinstance(other, dict): | ||
| if len(other) != 1: | ||
| raise ValueError("Parallel dict must have exactly one key (the name)") | ||
| name, value = next(iter(other.items())) | ||
| if not isinstance(name, str): |
There was a problem hiding this comment.
Case match instead of instance checks?
46a0cec to
04f73c1
Compare
0e43013 to
0ce7898
Compare
Squashed from 21 commits on parallel-step-design branch: - Add parallel step execution support to workflow engine - Add scaffold for parallel stress integration tests - Add nested parallel stress tests - Add mixed parallel/foreach_parallel stress tests - Add asymmetric branch stress tests - Add scale and concurrency stress tests - Add error propagation stress tests for parallel workflow compositions - Add edge case stress tests for parallel workflow execution - Propagate ContextVars to threadpool branches via copy_context - Add recursive _find_parallel_step and process_stat_var setup for Celery workers - Update stress test assertions for nested fork step persistence - Simplify: parametrize find tests, use accumulate, remove duplicate test and WHAT comments - Fix foreach_parallel WORKER dispatch bug and add integration test - Add integration test: parallel branch failure terminates without retry - Add integration test: foreach_parallel single branch failure with worker - Add failing integration test: resume after parallel skips remaining steps - Fix _join_and_resume to update main Waiting process step - Filter fork and branch steps in load_process to fix step count inflation - Fix resume test to simulate production Celery ordering
0ce7898 to
939b4c4
Compare
When the parallel step returns Success with __replace_last_state=True, _get_current_step_to_update found "last_db_step" via completed_at DESC. After a parallel block, that row is the last-finishing branch — not the parallel fork step — so the parent workflow's state would silently clobber the branch's recorded result. Fix: exclude rows that are children of a fork step (via process_step_relations) from the last_db_step query. Also preserve the fork step's original started_at/completed_at on update so it remains before its children in the UI's completed_at ordering. Regression test exercises this end-to-end with 2, 3, and 5 branches.
2 participants
Introduces fork/join semantics using the | operator and dict naming syntax, with a parallel() function for advanced use cases. Includes detailed design for state isolation, error handling, join logic, and comprehensive test plan.
Relates to: #870