Fix Issue: Ensure proper locking in WorkflowSweeper to prevent race conditions

[rq-dbrady]

Pull Request type

• Bugfix
• Feature
• Refactoring (no functional changes, no api changes)
• Build related changes
• WHOSUSING.md
• Other (please describe):

NOTE: Please remember to run ./gradlew spotlessApply to fix any format violations.

Changes in this PR

Issue:#213
Correct locking mechanism in WorkflowSweeper to prevent race conditions

Previously, the WorkflowSweeper.class had a potential race condition due to the order of operations in the sweep method. The workflow was being fetched from the executionDaoFacade before acquiring the lock, followed by a verifyAndRepair operation that could mutate the state. This sequence allowed for a small window (~50µ to 100µ seconds) where a workflow could be in two different states across different threads, causing inconsistencies and failures in workflow listeners or completion checks.

Changes made:

• Removed the decideWithLock method.
• Moved the locking logic directly into the sweep method to ensure the workflow is locked before any operations are performed on it.
• Ensured the workflow lock is only released after it is removed from the queue to prevent race conditions.

Observed issues:

• Race conditions were observed at large scale (30 replicas in Kubernetes, Redis cluster with Redis lock, ~75-90 workflows/sec).
• Workflows could be in a "Running" state even after triggering the finish, leading to listener and completion check failures.

New implementation in sweep method:

• Acquire lock before fetching the workflow and performing any operations.
• Handle verify and repair within the lock.
• Ensure the workflow is locked throughout the decision process.
• Release the lock only after removing the workflow from the queue.

This fix ensures atomicity in operations on workflows, preventing the race conditions previously observed.

[manan164]

Hi @rq-dbrady , Instead of using stopwatch, Can we try to acquire a lock here and if successful, then we do repair or come out of sweep logic?

[rq-dbrady]

So currently it follows the logic:

1. Create stopwatch.
2. Set the workflow context.
3. AcquireLock (if fail return from sweep)
4. Gets workflow from store.
5. Run repair
6. Start Stop Watch.
7. Decide
8. Remove From DeciderQueue
9. Release lock and Stop StopWatch.

Are you proposing we Acquire the lock at the method start and then start the stop watch ? so something like

 public void sweep(String workflowId) {
WorkflowModel workflow = null;
StopWatch watch = new StopWatch();
 if (!executionLockService.acquireLock(workflowId)) {
watch.start();
  //Do repair logic 
// Decide etc ...
}

[manan164]

Hi @rq-dbrady , Yes, the first thing we should try is to acquire a lock. then start a stop watch, run decide, stop stopwatch. release a lock. Optionally we dont need stop watch to measure execution time. We can use System.currentTimeMillis()

[rq-dbrady]

@manan164 , Applied changes in the logic order suggested !

1. Acquire lock
2. Run repair
3. Start timer
4. Decide
5. End timer
6. Release lock