Do not submit queued tasks after the session has aborted

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

@@ -75,6 +75,14 @@ class ParallelPollingMonitor extends TaskPollingMonitor {
         // using an thread-pool via the executor service
         final wrapper = (Callable)new ThrottlingExecutor.Recoverable() {
             @Override protected Object invoke() {
+                // the session may have been aborted or cancelled while this
+                // submission was sitting in the submitter queue -- submitting now
+                // would create a job that the shutdown kill sweep has already
+                // missed and that the polling loop will never observe
+                if( !session.canSubmitTasks() ) {
+                    log.debug "Skipping task submission -- session no longer accepting submissions > $handler"
+                    return null
+                }
                 return submit0(handler)
             }
 

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

@@ -65,6 +65,9 @@ class ParallelPollingMonitorTest extends Specification {
 
         then:
         handler.submit() >> { println "c=$count"; if(count.getAndIncrement()<2) throw new IllegalArgumentException("Ooops!") }
+        // the session is healthy in this scenario -- the retries come from the
+        // submit operation failing, not from the session being aborted
+        _ * session.canSubmitTasks() >> true
 
         success.get() == 1
         retry.get() == 2
@@ -118,6 +121,51 @@ class ParallelPollingMonitorTest extends Specification {
         10          |   1    | false | false
     }
 
+    def 'should not submit a queued task once the session can no longer submit' () {
+        given:
+        def session = Mock(Session)
+        def handler = Mock(TaskHandler)
+        and:
+        def opts = new ThrottlingExecutor.Options().withRateLimit('100/sec')
+        def exec = ThrottlingExecutor.create(opts)
+        def mon = new ParallelPollingMonitor(exec, [session:session, name:'foo', pollInterval:'1sec'])
+
+        when:
+        mon.submit(handler)
+        exec.shutdown()
+        exec.awaitTermination(1, TimeUnit.MINUTES)
+
+        then:
+        // the session was aborted while the wrapper was sitting in the submitter queue
+        _ * session.canSubmitTasks() >> false
+        and:
+        0 * handler.prepareLauncher()
+        0 * handler.submit()
+        0 * session.notifyTaskSubmit(_)
+    }
+
+    def 'should submit a queued task while the session can still submit' () {
+        given:
+        def session = Mock(Session)
+        def handler = Mock(TaskHandler)
+        and:
+        def opts = new ThrottlingExecutor.Options().withRateLimit('100/sec')
+        def exec = ThrottlingExecutor.create(opts)
+        def mon = new ParallelPollingMonitor(exec, [session:session, name:'foo', pollInterval:'1sec'])
+
+        when:
+        mon.submit(handler)
+        exec.shutdown()
+        exec.awaitTermination(1, TimeUnit.MINUTES)
+
+        then:
+        _ * session.canSubmitTasks() >> true
+        and:
+        1 * handler.prepareLauncher()
+        1 * handler.submit()
+        1 * session.notifyTaskSubmit(handler)
+    }
+
     def 'should inherit array size validation from parent TaskPollingMonitor' () {
         given:
         def session = Mock(Session)