fix(sweeper): re-queue on lock miss to eliminate multi-minute workflow pauses

[NicholasDCole]

Problem

Reported on Conductor 3.30.2 (Spanner persistence, Redis queue + locks): a workflow with many JOIN boundaries stalls for an intermittent ~10 minutes per run after upgrading from 3.21.4.

Root cause

The active sweeper (org.conductoross.conductor.core.execution.WorkflowSweeper, default via conductor.app.sweeper.enabled) bare-returned when it couldn't acquire the workflow lock — no decide(), no re-queue:

if (!executionLockService.acquireLock(workflowId)) {
    log.error("Couldn't acquire lock to sweep workflow {}", workflowId);
    return;   // <-- workflow parks
}

A lock-contended workflow (e.g. the post-completion decide() at a FORK/JOIN boundary) then had no near-term wake-up. Its only remaining decider-queue entry is postponed by ExecutionService.adjustDeciderQueuePostpone() to responseTimeoutSeconds * 1000 when a worker polls a task. With the affected task defs carrying responseTimeoutSeconds=600, that fallback is 600 s ≈ 10 minutes — the observed pause. It's intermittent ("sometimes") because it only surfaces when the event-driven decide is missed.

Fix

On a lock miss, re-queue the workflow to the decider queue with a bounded backoff (lockLeaseTime/2, capped by maxPostponeDurationSeconds) so a contended workflow is retried in seconds instead of parking:

if (!executionLockService.acquireLock(workflowId)) {
    long backoffSeconds = lockContentionBackoffSeconds();
    log.error("Couldn't acquire lock to sweep workflow {}", workflowId);
    queueDAO.push(DECIDER_QUEUE, workflowId, 0, backoffSeconds);
    return;
}

• The identical backoff math previously inlined in the decide()==null path is factored into a shared lockContentionBackoffSeconds() helper.
• The "Couldn't acquire lock to sweep workflow" error log is kept verbatim so existing log-based diagnostics/alerts still fire.
• This changes only recovery latency; it does not alter the (correct) responseTimeout-based postpone for genuinely long-running tasks.

Tests

Test	Purpose
WorkflowSweeperTest.sweepReQueuesOnLockMissWithBoundedBackoffInsteadOfParking	Regression — fails before fix (zero pushes), passes after (30 s re-queue)
WorkflowSweeperTest.sweepReQueuesWith30sBackoffWhenDecideCannotAcquireLock	Pins the decide()==null backoff at lockLeaseTime/2
ExecutionServiceTest.testPollSetsDeciderQueuePostponeToResponseTimeout_reproducesTenMinutePause	Documents the 600 s decider postpone that is the pause duration's source
TestWorkflowSweeper (legacy path)	Legacy-sweeper reproduction of the same responseTimeout-driven fallback, retained as a diagnostic

All green; spotless clean.

Note for affected deployments

As an immediate config mitigation before adopting a build with this fix, lowering responseTimeoutSeconds on the task defs (e.g. 600 → 60) shrinks the fallback pause proportionally.

Test evidence (validated with / without the fix)

All tests were run both with the fix in place and with the lock-miss re-queue reverted (call site A in WorkflowSweeper.sweep() restored to the original bare return, i.e. delete queueDAO.push(DECIDER_QUEUE, workflowId, 0, backoffSeconds)), on the same commit otherwise. The two tests that assert the fix flip from pass → fail when the fix is removed, confirming they are genuine regression gates (not vacuous). Environment: local_only execution lock; the e2e spec runs against the real Redis-backed decider queue the harness provisions via Testcontainers (redis:6.2-alpine).

✅ With the fix (current branch)

$ ./gradlew :conductor-core:test \
    --tests "org.conductoross.conductor.core.execution.WorkflowSweeperTest" \
    --tests "com.netflix.conductor.service.ExecutionServiceTest" \
    --tests "com.netflix.conductor.core.reconciliation.TestWorkflowSweeper"
WorkflowSweeperTest      tests=6  failures=0  skipped=0
ExecutionServiceTest     tests=13 failures=0  skipped=0
TestWorkflowSweeper      tests=20 failures=0  skipped=0

$ ./gradlew :conductor-test-harness:test \
    --tests "com.netflix.conductor.test.integration.DynamicForkJoinLockContentionSpec"
DynamicForkJoinLockContentionSpec  tests=2  failures=0  skipped=0
  ✓ sweeper re-queues a lock-contended workflow instead of leaving it parked at the response timeout (17.3s)
  ✓ control: with no lock contention the same JOIN schedules the next task immediately (0.46s)
BUILD SUCCESSFUL

❌ With the fix reverted (bare return on lock miss)

Unit regression — WorkflowSweeperTest.sweepReQueuesOnLockMissWithBoundedBackoffInsteadOfParking (tests=6, failures=1):

Wanted but not invoked:
queueDAO.push(
    "_deciderQueue",
    "workflow-id",
    0,
    30L
);
-> at WorkflowSweeperTest.sweepReQueuesOnLockMissWithBoundedBackoffInsteadOfParking(WorkflowSweeperTest.java:221)
Actually, there were zero interactions with this mock.

The contended sweep() never re-queued the workflow — exactly the bug (the 30s = lockLeaseTime/2 backoff push is absent).

E2E — DynamicForkJoinLockContentionSpec › "sweeper re-queues a lock-contended workflow…" (tests=2, failures=1; the control still passes):

Condition not satisfied:

queueDAO.containsMessage(Utils.DECIDER_QUEUE, workflowInstanceId)
|        |               |     |              |
|        false           |     _deciderQueue  bfd6fe7a-9aa6-4cfe-b714-06ea94c9c65a
|                        class com.netflix.conductor.core.utils.Utils
<io.orkes.conductor.mq.dao.RedisQueueDAO@... queueNamespace=integtest.test ...>

After the JOIN completes under a held workflow lock (post-JOIN decide() returns null, integration_task_4 never scheduled) and the decider entry is cleared, a contended sweep() leaves nothing in _deciderQueue for the workflow — it is parked with no near-term wake-up (in production, until the responseTimeoutSeconds=600 postpone fires). The control case (no contention) schedules integration_task_4 immediately in both variants, isolating the lock as the sole cause.

Restore check

Restoring the fix returns WorkflowSweeper.java to this PR's committed content (git diff empty) and both suites go green again.

🤖 Generated with Claude Code