maxForks slot is leaked when TaskHandler.submit() throws

Bug report

If a task fails during TaskHandler.submit(), the process's forks counter is incremented but never decremented. Each such failure permanently consumes one maxForks slot for that process. Once maxForks failures have accumulated, canForkProcess() is false forever and every remaining task of that process is retained in the pending queue and never submitted.

In TaskPollingMonitor.submitPendingTasks() (TaskPollingMonitor.groovy) the counter is incremented before the submit is attempted:

count++
handler.incProcessForks()
submit(handler)
}
catch ( Throwable e ) {
    handleException(handler, e)
    notifyTaskComplete(handler)
}

but handleException() only returns the slot when the handler can be evicted:

final protected void handleException( TaskHandler handler, Throwable error ) {
    def fault = null
    try {
        if (evict(handler)) {
            handler.decProcessForks()
        }
        ...

and evict() removes from the running queue:

@Override
boolean evict(TaskHandler handler) {
    if( !handler ) return false
    if( remove(handler) ) {      // remove() == runningQueue.remove(handler)
        ...
        return true
    }
    return false
}

Precondition: this only bites when maxForks is set on the process. TaskProcessor initialises forksCount = maxForks ? new LongAdder() : null and both incProcessForks() and decProcessForks() use ?., so with the default maxForks = 0 there is no counter, no leak and no starvation. The bug is scoped to processes that set maxForks explicitly.

A handler whose submit() threw was never added to runningQueue — TaskPollingMonitor.submit() adds it only after handler.submit() returns:

// submit the job execution -- throws a ProcessException when submit operation fail
handler.prepareLauncher()
handler.submit()
// note: add the 'handler' into the polling queue *after* the submit operation,
// this guarantees that in the queue are only jobs successfully submitted
runningQueue.add(handler)

so evict() returns false, decProcessForks() is skipped, and the slot is gone.

The observable effect is that the process makes no further progress: its remaining tasks are retained in pendingQueue indefinitely and are reported every dumpInterval as %% executor <name> > tasks in the submission queue: N -- tasks to be submitted are shown below.

I originally wrote this up as also causing the run to hang, via pollLoop()'s requirement that pendingQueue be empty to break out on a normal ending. On closer tracing that attribution looks wrong, so I am leaving it out rather than sending maintainers to the wrong place: on a submit-failure path TaskProcessor's StateObj bookkeeping is not reconciled either (submitted is incremented by the dataflow beforeRun hook, while completed only advances via finalizeTask0(), which handleException() never calls), so processesBarrier would already be blocking a layer above pollLoop, and session.isTerminated() would never become true for that clause to matter. That is a more general problem, independent of maxForks and of this leak, and I have not characterised it well enough to file yet.

So: treat this issue as the counter leak and the resulting starvation of the affected process. Whether and how the overall run terminates is a separate question I am deliberately not making a claim about.

Expected behavior and actual behavior

Expected: a failed submit releases the maxForks slot it reserved, exactly as a successful submit followed by normal eviction does. Subsequent tasks of that process continue to be submitted.

Actual: the slot is leaked. After maxForks submit failures the process is permanently starved, and remaining tasks sit in the pending queue indefinitely.

(Applies only to processes with maxForks set — see the precondition above.)

Steps to reproduce the problem

Reproducer as a Spock spec. Drop into modules/nextflow/src/test/groovy/nextflow/processor/ and run with:

./gradlew :nextflow:test --tests "nextflow.processor.ForkLeakTest"

package nextflow.processor

import java.util.concurrent.atomic.LongAdder

import nextflow.Session
import nextflow.exception.ProcessUnrecoverableException
import nextflow.trace.TraceRecord
import nextflow.util.Duration
import spock.lang.Specification

class ForkLeakTest extends Specification {

    static class TestHandler extends TaskHandler {
        boolean failSubmit
        TestHandler(TaskRun task, boolean failSubmit=false) {
            super(task)
            this.failSubmit = failSubmit
        }
        @Override void submit() {
            if( failSubmit )
                throw new ProcessUnrecoverableException('cannot resolve the container')
        }
        @Override void prepareLauncher() {}
        @Override boolean checkIfRunning() { return false }
        @Override boolean checkIfCompleted() { return false }
        @Override protected void killTask() {}
        @Override boolean isReady() { return true }
        @Override TraceRecord getTraceRecord() { return null }
    }

    /** evict() uses the locks that start() would normally create */
    private TaskPollingMonitor makeMonitor(Session session) {
        def monitor = new TaskPollingMonitor(name: 'foo', session: session, pollInterval: Duration.of('1min'))
        def lock = new java.util.concurrent.locks.ReentrantLock()
        monitor.pendingLock = lock
        monitor.taskAvail = lock.newCondition()
        monitor.slotAvail = lock.newCondition()
        return monitor
    }

    private TestHandler makeHandler(LongAdder adder, int maxForks, boolean failSubmit) {
        def processor = Mock(TaskProcessor) {
            getForksCount() >> adder
            getMaxForks() >> maxForks
        }
        def task = Mock(TaskRun) { getProcessor() >> processor }
        return new TestHandler(task, failSubmit)
    }

    def 'should not leak the forks count when the task submit fails' () {
        given:
        def adder = new LongAdder()
        def session = Mock(Session) { canSubmitTasks() >> true }
        def monitor = makeMonitor(session)
        monitor.getPendingQueue().add( makeHandler(adder, 10, true) )

        when:
        monitor.submitPendingTasks()

        then: 'the handler is dropped from the pending queue'
        monitor.getPendingQueue().size() == 0
        and: 'it never reached the running queue'
        monitor.getRunningQueue().size() == 0
        and: 'the forks counter is released'
        adder.sum() == 0
    }

    def 'should release the forks count when the task submit succeeds and the task is evicted' () {
        given: 'the control case -- a successful submit followed by normal eviction'
        def adder = new LongAdder()
        def session = Mock(Session) { canSubmitTasks() >> true }
        def monitor = makeMonitor(session)
        def handler = makeHandler(adder, 10, false)
        monitor.getPendingQueue().add(handler)

        when:
        monitor.submitPendingTasks()
        then:
        monitor.getRunningQueue().size() == 1
        adder.sum() == 1

        when: 'the task completes and is evicted'
        monitor.evict(handler)
        handler.decProcessForks()
        then:
        adder.sum() == 0
    }

    def 'should keep submitting after maxForks submit failures' () {
        given:
        def MAX_FORKS = 2
        def adder = new LongAdder()
        def session = Mock(Session) { canSubmitTasks() >> true }
        def monitor = makeMonitor(session)

        when: 'maxForks tasks of the same process fail to submit'
        MAX_FORKS.times { monitor.getPendingQueue().add( makeHandler(adder, MAX_FORKS, true) ) }
        monitor.submitPendingTasks()

        and: 'a healthy task of that process is then scheduled'
        monitor.getPendingQueue().add( makeHandler(adder, MAX_FORKS, false) )
        monitor.submitPendingTasks()

        then: 'it is submitted rather than pinned in the pending queue forever'
        monitor.getRunningQueue().size() == 1
        monitor.getPendingQueue().size() == 0
    }
}

Program output

Against master at v26.07.0-edge-41-g13c56f411:

ForkLeakTest > should not leak the forks count when the task submit fails FAILED
    Condition not satisfied:

    adder.sum() == 0
    |     |     |
    1     1     false

ForkLeakTest > should release the forks count when the task submit succeeds and the task is evicted PASSED

ForkLeakTest > should keep submitting after maxForks submit failures FAILED
    Condition not satisfied:

    monitor.getRunningQueue().size() == 1
    |       |                 |      |
    |       []                0      false
    ... pendingQueue=[TaskHandler[id: null; name: null; status: NEW; ...]] runningQueue=[]

3 tests completed, 2 failed

The control case passing is the useful part: the accounting is correct on the normal path, so the failure is specific to the error path rather than an artefact of the test harness.

Environment

• Nextflow version: reproduced on master at v26.07.0-edge-41-g13c56f411 (re-verified at v26.07.0-edge-55-ga00130e25); the same code is present in 25.10.0
• Java version: OpenJDK 21.0.11
• Operating system: macOS (test is platform-independent)

Additional context

Suggested fix: decrement unconditionally when the handler was never added to the running queue, e.g.

final protected void handleException( TaskHandler handler, Throwable error ) {
    def fault = null
    try {
        // the handler may have failed before reaching the running queue,
        // in which case the forks slot still needs releasing
        if( evict(handler) || !runningQueue.contains(handler) ) {
            handler.decProcessForks()
        }

or restructure so the increment is committed only once the handler actually reaches the running queue. The naive version of that — moving incProcessForks() next to the runningQueue.add(handler) in submit() — is not sound as-is: ParallelPollingMonitor.submit() dispatches the real submission to a thread pool and returns immediately, so the increment would land after canForkProcess() had already been evaluated for subsequent pending tasks, allowing maxForks to be exceeded; and the TaskArrayRun branch adds N queue entries for a single incremented handler. The first option is probably the safer shape.

For history: the evict-conditional decrement in handleException() was introduced by 05ea0c868 (#2787, Apr 2022) for exceptions raised by running tasks, where the handler is in runningQueue and a successful evict() is the right ownership test. The submit-failure path reuses the same handleException() but can never satisfy that test, which is how the hole opened.

Note that ParallelPollingMonitor has the same exposure via its onFailure path, which additionally returns early when !session.success, so nothing is released in that case at all. (I have filed the post-abort submission behaviour of that class separately; it is a different defect.)

A note for anyone writing tests here, since it cost me a couple of false passes: a Spock Mock(TaskHandler) is not usable for this. It stubs the non-final canForkProcess() to false, so nothing is ever submitted and the pending queue never drains; and it intercepts the final incProcessForks() / decProcessForks() bodies, so the counter never moves and the leak assertion passes vacuously. A concrete subclass is required. Separately, evict() dereferences pendingLock, which is initialised only in start(), so a directly constructed monitor needs those lock fields set by hand.

Start with modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy, tracing submitPendingTasks(), submit(), evict(), and handleException(). Run the proposed ForkLeakTest with ./gradlew :nextflow:test --tests "nextflow.processor.ForkLeakTest". Done means failed submissions release the maxForks slot, the pending queue drains appropriately, and all three regression scenarios pass.