Fix .command.stage file write when using Google Cloud

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

@@ -117,6 +117,14 @@ class BashWrapperBuilder {
 
     private BashTemplateEngine engine = new BashTemplateEngine()
 
+    // Flag to enable .command.stage file. False by default and enabled for Grid executors
+    // see https://github.com/nextflow-io/nextflow/issues/4279  and https://github.com/nextflow-io/nextflow/issues/5888
+    private Boolean stageFileEnabled = false
+
+    void enableStageFile(){
+        stageFileEnabled = true
+    }
+
     BashWrapperBuilder( TaskRun task ) {
         this(new TaskBean(task))
     }
@@ -272,9 +280,7 @@ class BashWrapperBuilder {
             return null
 
         final header = "# stage input files\n"
-        // enable only when the stage uses the default file system, i.e. it's not a remote object storage file
-        // see https://github.com/nextflow-io/nextflow/issues/4279
-        if( stageFile.fileSystem == FileSystems.default && stagingScript.size() >= stageFileThreshold.bytes ) {
+        if( stageFileEnabled && stagingScript.size() >= stageFileThreshold.bytes ) {
             stageScript = stagingScript
             return header + "bash ${stageFile}"
         }

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

@@ -103,7 +103,10 @@ class GridTaskHandler extends TaskHandler implements FusionAwareTask {
     @Override
     void prepareLauncher() {
         // -- create the wrapper script
-        createTaskWrapper(task).build()
+        final builder = createTaskWrapper(task)
+        // Enable stageFile to avoid problems with submission script limitations
+        builder.enableStageFile()
+        builder.build()
     }
 
     protected ProcessBuilder createProcessBuilder() {

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

@@ -473,7 +473,7 @@ class BashWrapperBuilderTest extends Specification {
         binding.stage_inputs == stageScript
     }
 
-    def 'should stage inputs to external file' () {
+    def 'should stage inputs to external file when enabled' () {
         given:
         SysEnv.push([NXF_WRAPPER_STAGE_FILE_THRESHOLD: '100'])
         and:
@@ -494,6 +494,7 @@ class BashWrapperBuilderTest extends Specification {
                 workDir: folder,
                 targetDir: folder,
                 inputFiles: inputs ])
+        builder.enableStageFile()
 
         when:
         def binding = builder.makeBinding()
@@ -510,6 +511,43 @@ class BashWrapperBuilderTest extends Specification {
         folder?.deleteDir()
     }
 
+    def 'should not stage inputs to external file when not enabled' () {
+        given:
+        SysEnv.push([NXF_WRAPPER_STAGE_FILE_THRESHOLD: '100'])
+        and:
+        def folder = Files.createTempDirectory('test')
+        and:
+        def inputs = [
+            'sample_1.fq': Paths.get('/some/data/sample_1.fq'),
+            'sample_2.fq': Paths.get('/some/data/sample_2.fq'),
+        ]
+        def stageScript = '''\
+                rm -f sample_1.fq
+                rm -f sample_2.fq
+                ln -s /some/data/sample_1.fq sample_1.fq
+                ln -s /some/data/sample_2.fq sample_2.fq
+                '''.stripIndent().rightTrim()
+        and:
+        def builder = newBashWrapperBuilder([
+            workDir: folder,
+            targetDir: folder,
+            inputFiles: inputs ])
+
+        when:
+        def binding = builder.makeBinding()
+        then:
+        binding.stage_inputs == "# stage input files\n"+stageScript
+
+        when:
+        builder.build()
+        then:
+        folder.resolve('.command.stage').exists() == false
+
+        cleanup:
+        SysEnv.pop()
+        folder?.deleteDir()
+    }
+
     def 'should include sync command' () {
         given:
         SysEnv.push([NXF_ENABLE_FS_SYNC: 'true'])