Release the maxForks slot when task submission fails

[Mohit-Ak]

Problem

maxForks slots are leaked whenever TaskHandler.submit() throws. In
TaskPollingMonitor.submitPendingTasks() the process forks counter is
incremented before the submit is attempted:

handler.incProcessForks()
submit(handler)

but the slot is only returned through handleException(), which is
conditional on eviction:

if( evict(handler) ) {
    handler.decProcessForks()
}

and evict() removes from the running queue. A handler whose submit()
threw was never added to that queue — TaskPollingMonitor.submit() adds it only
after handler.submit() returns — so evict() returns false, the decrement is
skipped, and the slot is gone for good.

After maxForks such failures canForkProcess() is permanently false and the
process is starved: its remaining tasks sit in the pending queue indefinitely and
are re-reported every dumpInterval.

This only bites processes that set maxForks explicitly. With the default
maxForks = 0, TaskProcessor leaves forksCount null and the ?. in
incProcessForks()/decProcessForks() makes the whole thing a no-op, so there
is no counter and no leak.

The evict-conditional decrement came in with 05ea0c8 (#2787) to handle
exceptions from running tasks, where the handler really is in runningQueue
and a successful evict() is the correct ownership test. The submit-failure path
reuses the same handleException() but can never satisfy that test.

Fix

Release the slot at the submit-failure site rather than widening the condition
inside handleException(). The new releaseSubmitSlot() helper decrements only
when the handler is not in the running queue, which is precisely the "took a
slot but never made it into the queue" case:

protected void releaseSubmitSlot(TaskHandler handler) {
    if( !handler || runningQueue.contains(handler) )
        return
    handler.decProcessForks()
}

Loosening handleException() instead would double-release: checkTaskStatus()
already calls decProcessForks() before its own evict(), so tasks that fail
after a successful submit would give the slot back twice and let the process
exceed maxForks.

I deliberately did not move incProcessForks() down next to
runningQueue.add(handler). ParallelPollingMonitor.submit() hands the real
submission to a thread pool and returns immediately, so the increment would land
after canForkProcess() had already been evaluated for later pending tasks —
which would let maxForks be exceeded — and the TaskArrayRun branch adds N
queue entries for a single incremented handler.

ParallelPollingMonitor routes submit failures through its own
Recoverable.onFailure() rather than the try/catch in submitPendingTasks(),
so it has the identical leak and gets the same release. Its early return on
!session.success is left alone — that is separate behaviour and not this bug.

Testing

Two Spock specs, one per affected call path.

Per the note in the issue, Mock(TaskHandler) is unusable here: it stubs the
non-final canForkProcess() to false so nothing is ever submitted, and it
intercepts the final incProcessForks()/decProcessForks() bodies so the
counter never moves and the assertion passes vacuously. Both tests use
Spy(TaskHandler) over a real LongAdder. TaskPollingMonitorTest also needed
pendingLock/taskAvail/slotAvail set by hand, since evict() dereferences
them and they are only created in start().

Reverting both source changes and re-running the new tests:

FAILED: should release the forks slot when the task submit throws
  Condition not satisfied: adder.intValue() == 0
                           |                   |
                           1                   false
FAILED: should release the forks slot when the parallel submit fails
  Condition not satisfied: adder.intValue() == 0
                           |                   |
                           1                   false
TOTALS tests=36 failures=2 errors=0

The other 34 tests in those two classes pass with the fix reverted, so the two
failures are specific to the leak and not to the harness.

With the fix applied:

./gradlew :nextflow:test --tests "nextflow.processor.*"
  tests=468 failures=0 errors=0 skipped=1

./gradlew :nextflow:test
  tests=3416 failures=0 errors=0 skipped=89

Fixes #7447

[netlify]

✅ Deploy Preview for nextflow-docs canceled.

Name	Link
🔨 Latest commit	9576252
🔍 Latest deploy log	https://app.netlify.com/projects/nextflow-docs/deploys/6a79dae74547870008d8011b