Fix Job arrays exceeding queue size limit (#5920) (#6345) [ci fast]

Author: pditommaso

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -197,6 +197,47 @@ class TaskPollingMonitor implements TaskMonitor {
     int getCapacity() { capacity }
 
 
+    /**
+     * Get the number of queue slots consumed by a task handler.
+     * Regular tasks consume 1 slot, job arrays consume slots equal to their array size.
+     *
+     * @param handler A {@link TaskHandler} for the task
+     * @return The number of queue slots consumed by this task
+     */
+    private int getTaskSlots(TaskHandler handler) {
+        return handler.task instanceof TaskArrayRun ?
+            ((TaskArrayRun) handler.task).getArraySize() : 1
+    }
+
+    /**
+     * Validates that a task handler can be submitted based on queue capacity constraints.
+     * Performs validation for job arrays that exceed queue capacity.
+     *
+     * @param handler A {@link TaskHandler} for the task to validate
+     * @return {@code true} if the task meets capacity constraints
+     * @throws IllegalArgumentException if job array size exceeds queue capacity
+     */
+    private boolean checkQueueCapacity(TaskHandler handler) {
+        // Calculate slots consumed by this task:
+        // - Regular tasks consume 1 slot
+        // - Job arrays consume slots equal to their array size  
+        int slots = getTaskSlots(handler)
+        
+        // Prevent job arrays larger than the entire queue capacity
+        // This catches configuration errors early with a clear error message
+        if (slots > capacity) {
+            throw new IllegalArgumentException(
+                "Process '${handler.task.name}' declares array size ($slots) " +
+                "which exceeds the executor queue size ($capacity)")
+        }
+        
+        // Check if there's enough remaining capacity for this task:
+        // - runningQueue.size() = currently running tasks
+        // - slots = slots needed for this new task
+        // - capacity = maximum allowed concurrent tasks
+        return runningQueue.size() + slots <= capacity
+    }
+
     /**
      * Defines the strategy determining if a task can be submitted for execution.
      *
@@ -207,7 +248,7 @@ class TaskPollingMonitor implements TaskMonitor {
      *      by the polling monitor
      */
     protected boolean canSubmit(TaskHandler handler) {
-        (capacity>0 ? runningQueue.size() < capacity : true) && handler.canForkProcess() && handler.isReady()
+        (capacity > 0 ? checkQueueCapacity(handler) : true) && handler.canForkProcess() && handler.isReady()
     }
 
     /**

modules/nextflow/src/test/groovy/nextflow/processor/ParallelPollingMonitorTest.groovy

@@ -118,4 +118,26 @@ class ParallelPollingMonitorTest extends Specification {
         10          |   1    | false | false
     }
 
+    def 'should inherit array size validation from parent TaskPollingMonitor' () {
+        given:
+        def session = Mock(Session)
+        def throttlingExecutor = Mock(ThrottlingExecutor)
+        def monitor = new ParallelPollingMonitor(throttlingExecutor, [name: 'foo', session: session, capacity: 5, pollInterval: Duration.of('1min')])
+        and:
+        def processor = Mock(TaskProcessor)
+        def arrayHandler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskArrayRun) {
+                getName() >> 'oversized_array'
+                getArraySize() >> 10
+                getProcessor() >> processor
+            }
+        }
+
+        when:
+        monitor.canSubmit(arrayHandler)
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains("Process 'oversized_array' declares array size (10) which exceeds the executor queue size (5)")
+    }
+
 }

modules/nextflow/src/test/groovy/nextflow/processor/TaskPollingMonitorTest.groovy

@@ -22,6 +22,7 @@ import nextflow.executor.ExecutorConfig
 import nextflow.util.Duration
 import nextflow.util.RateUnit
 import spock.lang.Specification
+import spock.lang.Unroll
 /**
  *
  * @author Paolo Di Tommaso <[email protected]>
@@ -150,4 +151,76 @@ class TaskPollingMonitorTest extends Specification {
         3 * session.notifyTaskSubmit(handler)
     }
 
+    @Unroll
+    def 'should throw error if job array size exceeds queue size [capacity: #CAPACITY, array: #ARRAY_SIZE]' () {
+        given:
+        def session = Mock(Session)
+        def monitor = Spy(new TaskPollingMonitor(name: 'foo', session: session, capacity: CAPACITY, pollInterval: Duration.of('1min')))
+        and:
+        def processor = Mock(TaskProcessor)
+        def arrayHandler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskArrayRun) {
+                getName() >> TASK_NAME
+                getArraySize() >> ARRAY_SIZE
+                getProcessor() >> processor
+            }
+        }
+
+        when:
+        monitor.canSubmit(arrayHandler)
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains("Process '$TASK_NAME' declares array size ($ARRAY_SIZE) which exceeds the executor queue size ($CAPACITY)")
+
+        where:
+        CAPACITY | ARRAY_SIZE | TASK_NAME
+        10       | 15         | 'test_array'
+        5        | 10         | 'large_array'  
+        1        | 2          | 'small_array'
+    }
+
+    @Unroll
+    def 'should validate array size accounting in queue capacity [capacity: #CAPACITY, running: #RUNNING_COUNT, array: #ARRAY_SIZE]' () {
+        given:
+        def session = Mock(Session)
+        def monitor = Spy(new TaskPollingMonitor(name: 'foo', session: session, capacity: CAPACITY, pollInterval: Duration.of('1min')))
+        and:
+        def processor = Mock(TaskProcessor)
+        def regularHandler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskRun) {
+                getProcessor() >> processor
+            }
+            canForkProcess() >> CAN_FORK
+            isReady() >> IS_READY
+        }
+        def arrayHandler = Mock(TaskHandler) {
+            getTask() >> Mock(TaskArrayRun) {
+                getArraySize() >> ARRAY_SIZE
+                getProcessor() >> processor
+            }
+            canForkProcess() >> CAN_FORK
+            isReady() >> IS_READY
+        }
+
+        and:
+        RUNNING_COUNT.times { monitor.runningQueue.add(regularHandler) }
+
+        expect:
+        monitor.runningQueue.size() == RUNNING_COUNT
+        monitor.canSubmit(regularHandler) == REGULAR_EXPECTED
+        monitor.canSubmit(arrayHandler) == ARRAY_EXPECTED
+
+        where:
+        CAPACITY | RUNNING_COUNT | ARRAY_SIZE | CAN_FORK | IS_READY | REGULAR_EXPECTED | ARRAY_EXPECTED
+        10       | 6             | 5          | true     | true     | true             | false     // Array too big (6+5>10)
+        10       | 5             | 5          | true     | true     | true             | true      // Array fits exactly (5+5=10)
+        10       | 9             | 1          | true     | true     | true             | true      // Both fit (9+1=10)
+        10       | 10            | 1          | true     | true     | false            | false     // Queue full (10+1>10)
+        5        | 4             | 1          | true     | true     | true             | true      // Both fit (4+1=5)
+        5        | 4             | 2          | true     | true     | true             | false     // Array too big (4+2>5)
+        0        | 5             | 10         | true     | true     | true             | true      // Unlimited capacity
+        10       | 5             | 3          | false    | true     | false            | false     // Cannot fork
+        10       | 5             | 3          | true     | false    | false            | false     // Not ready
+    }
+
 }