Fix concurrent modification with job arrays

modules/nextflow/src/main/groovy/nextflow/executor/BashWrapperBuilder.groovy

@@ -504,7 +504,9 @@ class BashWrapperBuilder {
         final lines = new StringBuilder()
         lines << '### ---\n'
         lines << "### name: '${bean.name}'\n"
-        if( bean.arrayIndexName ) {
+        // the array metadata block only applies to the array dispatcher task (which carries the
+        // child work-dirs); array children only carry the index name to expose it in the container
+        if( bean.arrayIndexName && bean.arrayWorkDirs ) {
             lines << '### array:\n'
             lines << "###   index-name: ${bean.arrayIndexName}\n"
             lines << "###   index-start: ${bean.arrayIndexStart}\n"
@@ -742,6 +744,11 @@ class BashWrapperBuilder {
             builder.addEnv(var)
         }
 
+        // for job arrays, propagate the array index variable (e.g. SLURM_ARRAY_TASK_ID)
+        // into the container environment
+        if( arrayIndexName )
+            builder.addEnv(arrayIndexName)
+
         // when secret are not managed by the execution platform natively
         // the secret names are added to the container env var white list
         if( !isSecretNative() && secretNames )  {

modules/nextflow/src/main/groovy/nextflow/executor/GridTaskHandler.groovy

@@ -209,9 +209,15 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
     }
 
     protected BashWrapperBuilder createTaskWrapper(TaskRun task) {
-        return fusionEnabled()
-            ? fusionLauncher()
-            : executor.createBashWrapperBuilder(task)
+        if( fusionEnabled() )
+            return fusionLauncher()
+        final builder = executor.createBashWrapperBuilder(task)
+        // for a containerised array child, the container is built from the child task bean
+        // (which does not carry the array index variable); expose the scheduler array-index
+        // variable (e.g. SLURM_ARRAY_TASK_ID) inside the task container
+        if( isArrayChild && executor instanceof TaskArrayExecutor )
+            builder.arrayIndexName = ((TaskArrayExecutor) executor).getArrayIndexName()
+        return builder
     }
 
     protected String stdinLauncherScript() {

modules/nextflow/src/main/groovy/nextflow/fusion/FusionHelper.groovy

@@ -63,6 +63,11 @@ class FusionHelper {
         for( String env : containerConfig.getEnvWhitelist())
             containerBuilder.addEnv(env)
 
+        // for job arrays, propagate the array index variable (e.g. SLURM_ARRAY_TASK_ID)
+        // into the container environment
+        if( launcher.arrayIndexName )
+            containerBuilder.addEnv(launcher.arrayIndexName)
+
         // patch the cmd wrapping the last item in quotes
         final patchCmd = new ArrayList(runCmd)
         patchCmd[-1] = "'${patchCmd[-1]}'"

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

@@ -18,8 +18,6 @@ package nextflow.processor
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
-import nextflow.container.ContainerConfig
-import nextflow.executor.TaskArrayExecutor
 
 /**
  * Models a task array, which submits a collection of independent
@@ -37,17 +35,6 @@ class TaskArrayRun extends TaskRun {
         children.size()
     }
 
-    @Override
-    ContainerConfig getContainerConfig() {
-        final config = super.getContainerConfig()
-        final envWhitelist = config.getEnvWhitelist()
-        if( envWhitelist != null ) {
-            final executor = (TaskArrayExecutor)processor.getExecutor()
-            envWhitelist.add(executor.getArrayIndexName())
-        }
-        return config
-    }
-
     @Override
     boolean isContainerEnabled() {
         return false

modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy

@@ -227,6 +227,35 @@ class BashWrapperBuilderTest extends Specification {
         builder.mounts == [ Paths.get('/my/bin') ]
     }
 
+    def 'should add the array index variable to the container env' () {
+        given:
+        def bash = Spy(new BashWrapperBuilder(Mock(TaskBean)))
+        and:
+        bash.getEnvironment() >> [:]
+        bash.getBinDirs() >> []
+        bash.getWorkDir() >> Paths.get('/my/work/dir')
+        bash.isStatsEnabled() >> false
+        bash.getStageInMode() >> 'symlink'
+        bash.getInputFiles() >> [:]
+        bash.getContainerConfig() >> new SingularityConfig(envWhitelist: 'FOO,BAR')
+        bash.getContainerImage() >> 'foo/bar'
+        bash.getContainerMount() >> null
+        bash.getContainerMemory() >> null
+        bash.getContainerCpus() >> null
+        bash.getContainerCpuset() >> null
+        bash.getContainerOptions() >> null
+        bash.getContainerPlatform() >> 'amd64'
+        bash.isSecretNative() >> false
+        bash.getSecretNames() >> []
+        and: 'the task is an array child, exposing the array index variable'
+        bash.getArrayIndexName() >> 'SLURM_ARRAY_TASK_ID'
+
+        when:
+        def builder = bash.createContainerBuilder(null)
+        then: 'the array index variable is whitelisted without mutating the container config'
+        builder.env == ['NXF_TASK_WORKDIR', 'FOO', 'BAR', 'SLURM_ARRAY_TASK_ID']
+    }
+
     def 'should add resolved inputs'() {
         given:
         def bash = Spy(new BashWrapperBuilder(Mock(TaskBean)))

modules/nextflow/src/test/groovy/nextflow/executor/GridTaskHandlerTest.groovy

@@ -148,4 +148,49 @@ class GridTaskHandlerTest extends Specification {
                 LAUNCH_COMMAND_EOF
                 '''.stripIndent()
     }
+
+    def 'should propagate the array index variable to a containerised array child launcher' () {
+        given: 'a plain child builder built from a non-array TaskRun (arrayIndexName not set - the problematic state)'
+        def childBuilder = new BashWrapperBuilder(new TaskBean(name: 'foo', workDir: Paths.get('/work/dir'), inputFiles: [:]))
+        and:
+        def task = Mock(TaskRun) { getWorkDir() >> Paths.get('/work/dir') }
+        def exec = Mock(SlurmExecutor) {
+            getConfig() >> new ExecutorConfig([:])
+            getName() >> 'slurm'
+            getArrayIndexName() >> 'SLURM_ARRAY_TASK_ID'
+            createBashWrapperBuilder(task) >> childBuilder
+        }
+        def handler = Spy(new GridTaskHandler(task, exec))
+        handler.withArrayChild(true)
+        handler.fusionEnabled() >> false
+
+        expect: 'the child bean does not carry the array index variable on its own'
+        childBuilder.arrayIndexName == null
+
+        when:
+        def builder = handler.createTaskWrapper(task)
+        then: 'the handler injects the executor array index name so the container env can expose it'
+        builder.arrayIndexName == 'SLURM_ARRAY_TASK_ID'
+    }
+
+    def 'should not add the array index variable for a non-array task' () {
+        given:
+        def childBuilder = new BashWrapperBuilder(new TaskBean(name: 'foo', workDir: Paths.get('/work/dir'), inputFiles: [:]))
+        and:
+        def task = Mock(TaskRun) { getWorkDir() >> Paths.get('/work/dir') }
+        def exec = Mock(SlurmExecutor) {
+            getConfig() >> new ExecutorConfig([:])
+            getName() >> 'slurm'
+            createBashWrapperBuilder(task) >> childBuilder
+        }
+        def handler = Spy(new GridTaskHandler(task, exec))   // isArrayChild == false
+        handler.fusionEnabled() >> false
+
+        when:
+        def builder = handler.createTaskWrapper(task)
+        then:
+        builder.arrayIndexName == null
+        and: 'the executor array index name is never requested for a non-array task'
+        0 * exec.getArrayIndexName()
+    }
 }

modules/nextflow/src/test/groovy/nextflow/fusion/FusionHelperTest.groovy

@@ -60,6 +60,20 @@ class FusionHelperTest extends Specification {
         result == "docker run -i --platform linux/amd64 --rm --privileged image:1 echo 'hello'"
     }
 
+    def 'should propagate the array index variable to the fusion container' () {
+        given:
+        def launcher = Mock(FusionScriptLauncher) {
+            fusionEnv() >> [:]
+            getArrayIndexName() >> 'SLURM_ARRAY_TASK_ID'
+        }
+        def config = new DockerConfig([:])
+
+        when:
+        def result = FusionHelper.runWithContainer(launcher, config, 'image:1', null, ['echo', 'hello'])
+        then:
+        result.contains('-e "SLURM_ARRAY_TASK_ID"')
+    }
+
     def 'should return fusion container command' () {
         given:
         def launcher = Mock(FusionScriptLauncher) {

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

@@ -132,7 +132,8 @@ class TaskArrayCollectorTest extends Specification {
         taskArray.script == 'the-task-array-script'
         and:
         taskArray.getArraySize() == 3
-        taskArray.getContainerConfig().getEnvWhitelist() == [ 'ARRAY_JOB_INDEX' ]
+        and: 'the array index variable is no longer injected into the shared container config; it is added to the container env by the launcher (BashWrapperBuilder/FusionHelper)'
+        taskArray.getContainerConfig().getEnvWhitelist() == []
         taskArray.isContainerEnabled() == false
     }