// WorkflowSweeper.sweep() — acquires lock here:
if (!executionLockService.acquireLock(workflowId)) { ... }

// Then calls decide() which ALSO calls acquireLock() internally:
workflow = workflowExecutor.decide(workflowId);  // always returns null with non-reentrant lock

// null is misread as contention -> re-queue with backoff -> infinite loop
if (workflow == null) {
    queueDAO.push(DECIDER_QUEUE, workflowId, 0, backoffSeconds);
    return;
}

/**
 * Reproduction for: sweep() acquires the execution lock, then calls decide(workflowId) which
 * also calls acquireLock() internally. With a non-reentrant lock (e.g. Postgres advisory lock)
 * the second acquisition always fails, decide() returns null, and the sweeper re-queues the
 * workflow with a backoff delay — leaving it permanently stuck even though the sub-workflow has
 * already completed.
 *
 * <p>Expected (correct) behaviour: decide() is called and the workflow advances.
 * Actual (buggy) behaviour: decide() returns null because the sweeper already holds the lock;
 * the workflow is pushed back to the decider queue with a 30-second delay indefinitely.
 */
@Test
public void sweepDoesNotStallWhenSubWorkflowCompletedAndLockIsNonReentrant() {
    TaskModel subWorkflowTask =
            newTask(
                    "sub-workflow-task",
                    TaskType.TASK_TYPE_SUB_WORKFLOW,
                    TaskModel.Status.IN_PROGRESS);
    subWorkflowTask.setSubWorkflowId("sub-workflow-id");
    WorkflowModel workflow = newWorkflow(List.of(subWorkflowTask));

    WorkflowModel subWorkflow = new WorkflowModel();
    subWorkflow.setStatus(WorkflowModel.Status.COMPLETED);
    subWorkflow.setOutput(Map.of("result", "ok"));

    // The sweeper acquires the lock successfully on the first call.
    // The second call simulates a non-reentrant lock: decide(workflowId) calls acquireLock()
    // internally and cannot acquire a lock that the sweeper already holds.
    when(executionLockService.acquireLock(WORKFLOW_ID))
            .thenReturn(true)   // sweeper's own pre-acquire
            .thenReturn(false); // decide()'s internal acquire — non-reentrant lock fails

    when(workflowExecutor.getWorkflow(WORKFLOW_ID, true)).thenReturn(workflow);
    // decide(workflowId) returns null when it cannot acquire the lock
    when(workflowExecutor.decide(WORKFLOW_ID)).thenReturn(null);

    workflowSweeper.sweep(WORKFLOW_ID);

    // Bug: the workflow is pushed back to the decider queue with a backoff delay,
    // causing it to stall permanently even though the sub-workflow is already done.
    verify(queueDAO, never()).push(anyString(), anyString(), anyInt(), anyLong());
}

// WorkflowSweeper.sweep() — REMOVE these lines:
if (!executionLockService.acquireLock(workflowId)) {
    log.error("Couldn't acquire lock to sweep workflow {}", workflowId);
    return;
}
// ...
} finally {
    executionLockService.releaseLock(workflowId);  // also remove
}