conductor-oss / conductor-oss/conductor
subworkflowChanged flag race condition with new WorkflowSweeper background threads
- Dominant language
- Java
- Stars
- 32.2k
- Forks
- 1k
- Avg merge
- 2d 3h
- Merged PRs (30d)
- 37
Description
**PR that fixes this:** #880
## What broke and why
`SubWorkflowSpec > Test retrying a subworkflow where parent workflow timed out due to workflowTimeout` was failing intermittently (the test checks `tasks[2].subworkflowChanged == true` after a subworkflow completes following a retry, but the flag was already `false`).
## Why only now? (Dec 2025 / Jan 2026)
The race condition was latent in the code for years but harmless — the old sweeper polled the decider queue slowly and rarely won the race. **The new multi-threaded `WorkflowSweeper` (landed Dec 28, 2025, commit `99ecc5f`, further stabilized in late Jan 2026) is what makes it fire.** It starts N background threads that hammer the decider queue in a tight loop, dramatically increasing the probability that a sweeper thread resets `subworkflowChanged=false` between the retry and the subworkflow completing. Before that sweeper landed, nobody would have seen this problem.
## History
**2021 (commit 50844095e)** — Aravindan Ramkumar introduced the `subworkflowChanged` flag and `adjustStateIfSubWorkflowChanged`. The idea: when a subworkflow is retried, `updateAndPushParents` sets `subworkflowChanged=true` on the parent's SUB_WORKFLOW task and pushes the parent to the decider queue. When `decide()` later runs, `adjustStateIfSubWorkflowChanged` finds the flag, resets it, and re-evaluates any CANCELED/FAILED JOIN tasks back to IN_PROGRESS so the fork/join can complete.
The design assumed the flag would stay set until `decide()` ran. That was safe because the old sweeper was fairly passive — it polled the decider queue but didn't hammer it.
**2024 (commit 6ea3a461e)** — Viren refactored `WorkflowExecutor` into an interface + `WorkflowExecutorOps`. Mechanical copy, no functional change. `updateParentWorkflowTask` (called from `completeWorkflow` and `terminateWorkflow`) was carried over as-is: it updated the parent's SUB_WORKFLOW task to COMPLETED but never set `subworkflowChanged=true`. That was fine at the time because `completeWorkflow` also called `expediteLazyWorkflowEvaluation` → `decide()` synchronously, so the flag wasn't needed on the completion path.
**Dec 2025 / Jan 2026 (commits by Manan Bhatt)** — The new `WorkflowSweeper` in `org.conductoross` went live. Unlike the old sweeper, it starts N background threads in its constructor that continuously poll `DECIDER_QUEUE` in a tight loop. Each thread calls `sweep()` → `decide()` → `adjustStateIfSubWorkflowChanged`.
This opened a race:
1. Subworkflow is retried → `updateAndPushParents` sets `subworkflowChanged=true`, parent pushed to decider queue
2. Background sweeper thread picks it up, calls `adjustStateIfSubWorkflowChanged`, resets `subworkflowChanged=false`, re-queues the JOIN tasks
3. Subworkflow eventually completes → `completeWorkflow` → `updateParentWorkflowTask` sets SUB_WORKFLOW task to COMPLETED but does **not** set `subworkflowChanged=true` (this was never needed before)
4. Parent is pushed to decider queue again, `decide()` runs, `adjustStateIfSubWorkflowChanged` finds no changed task → no-op
5. The JOIN tasks are IN_PROGRESS but not getting evaluated because the normal decide path handles this fine... except in the workflow-timeout retry scenario where the timing is tighter
The test was asserting the flag was still `true` after step 3, which was true before the new sweeper existed but now fails whenever a background thread wins the race between steps 1 and 3.
## The fix
`updateParentWorkflowTask` needs to set `subworkflowChanged=true` before saving. Semantically correct: the subworkflow just transitioned to a terminal state, which is exactly the change the flag was designed to signal. The flag is idempotent — if `adjustStateIfSubWorkflowChanged` already ran and the JOINs are IN_PROGRESS, it just resets the flag and moves on.
Two assertions in `NestedForkJoinSubWorkflowSpec` that checked `!tasks[2].subworkflowChanged` immediately after subworkflow completion (not after a sweep) were also updated, since that internal state now differs.
Contributor guide
Assessment
This issue has not been assessed yet.