Improve resilience to delayed shared file system visibility

modules/nextflow/src/main/resources/nextflow/executor/command-run.txt

@@ -125,10 +125,16 @@ on_exit() {
     local exit_status=${nxf_main_ret:=0}
     [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0}
     [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err}
-    printf -- $exit_status {{exit_file}}
     set +u
-    {{cleanup_cmd}}
+    ## Flush the pending writes *before* publishing the exit status: the `.exitcode`
+    ## file is the completion marker polled by the executors, therefore it must not
+    ## become visible to the driver before everything else the task has written into
+    ## the work directory. On a shared file system with delayed write-back the driver
+    ## may otherwise observe a task as completed while its outputs are still missing
+    ## or truncated.
     {{sync_cmd}}
+    printf -- $exit_status {{exit_file}}
+    {{cleanup_cmd}}
     exit $exit_status
 }
 

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

@@ -524,6 +524,23 @@ class BashWrapperBuilderTest extends Specification {
         SysEnv.pop()
     }
 
+    def 'should flush the file system before writing the exit file' () {
+        given:
+        SysEnv.push([NXF_ENABLE_FS_SYNC: 'true'])
+
+        when:
+        def wrapper = newBashWrapperBuilder(workDir: Paths.get('/work/dir')).buildNew0()
+        then:
+        // the `.exitcode` file is the completion marker polled by the executors, therefore
+        // it must be written only after all other task writes have been flushed, otherwise
+        // a shared file system may expose the marker before the task outputs
+        wrapper.indexOf('sync || true') > 0
+        wrapper.indexOf('sync || true') < wrapper.indexOf('printf -- $exit_status > /work/dir/.exitcode')
+
+        cleanup:
+        SysEnv.pop()
+    }
+
     def 'should unstage outputs' () {
         given:
         def folder = Paths.get('/work/dir')

modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt

@@ -281,8 +281,8 @@ on_exit() {
     local exit_status=${nxf_main_ret:=0}
     [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0}
     [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err}
-    printf -- $exit_status > {{folder}}/.exitcode
     set +u
+    printf -- $exit_status > {{folder}}/.exitcode
     exit $exit_status
 }
 

modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt

@@ -89,8 +89,8 @@ on_exit() {
     local exit_status=${nxf_main_ret:=0}
     [[ ${exit_status} -eq 0 && ${nxf_unstage_ret:=0} -ne 0 ]] && exit_status=${nxf_unstage_ret:=0}
     [[ ${exit_status} -eq 0 && ${last_err} -ne 0 ]] && exit_status=${last_err}
-    printf -- $exit_status > {{folder}}/.exitcode
     set +u
+    printf -- $exit_status > {{folder}}/.exitcode
     exit $exit_status
 }
 

modules/nf-commons/src/main/nextflow/file/FileHelper.groovy

@@ -472,19 +472,36 @@ class FileHelper {
     }
 
     /**
-     * Check if the specified path is a NFS mount
+     * The file system types that are known to be network shared file systems, and
+     * therefore may expose a delayed view of the writes made by another client.
+     * These are the values reported by {@code stat -f -c %T}.
+     */
+    private static final Set<String> SHARED_FS_TYPES = Set.of('nfs', 'lustre', 'ceph', 'beegfs', 'gpfs')
+
+    /**
+     * Check if the given file system type, as reported by {@code stat -f -c %T},
+     * denotes a network shared file system
+     *
+     * @param type The file system type name
+     * @return {@code true} when the type is a shared file system, {@code false} otherwise
+     */
+    static boolean isSharedFsType(String type) {
+        return type != null && type in SHARED_FS_TYPES
+    }
+
+    /**
+     * Check if the specified path is a shared file system mount
      *
      * @param path The path to verify
-     * @return The {@code true} when the path is a NFS mount {@code false} otherwise
+     * @return The {@code true} when the path is a shared file system mount {@code false} otherwise
      */
     @Memoized
     static boolean isPathSharedFS(Path path) {
         assert path
         if( path.getFileSystem() != FileSystems.getDefault() )
             return false
 
-        final type = getPathFsType(path)
-        final result = type == 'nfs' || type == 'lustre'
+        final result = isSharedFsType(getPathFsType(path))
         log.debug "FS path type ($result): $path"
         return result
     }

modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy

@@ -1048,6 +1048,25 @@ class FileHelperTest extends Specification {
         null                 | '1234://xyz.com/abc'
     }
 
+    @Unroll
+    def 'should detect shared file system type'() {
+        expect:
+        FileHelper.isSharedFsType(TYPE) == EXPECTED
+
+        where:
+        TYPE        | EXPECTED
+        'nfs'       | true
+        'lustre'    | true
+        'ceph'      | true
+        'beegfs'    | true
+        'gpfs'      | true
+        'ext2/ext3' | false
+        'xfs'       | false
+        'overlayfs' | false
+        'tmpfs'     | false
+        null        | false
+    }
+
     def 'should check symlink status'() {
         given:
         def folder = Files.createTempDirectory('test')

plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy

@@ -32,6 +32,7 @@ import nextflow.exception.NodeTerminationException
 import nextflow.k8s.client.PodUnschedulableException
 import nextflow.exception.ProcessSubmitException
 import nextflow.executor.BashWrapperBuilder
+import nextflow.file.FileHelper
 import nextflow.fusion.FusionAwareTask
 import nextflow.k8s.client.K8sClient
 import nextflow.k8s.client.K8sResponseException
@@ -78,12 +79,14 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask {
 
     private Path errorFile
 
-    private Path exitFile
+    protected Path exitFile
 
     private Map state
 
     private long timestamp
 
+    private long terminatedTimestamp
+
     private K8sExecutor executor
 
     private String runsOnNode = null
@@ -439,6 +442,15 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask {
                 // https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#containerstateterminated-v1-core
                 log.trace("[k8s] Container Terminated state ${state.terminated}")
                 final k8sExitCode = (state.terminated as Map)?.exitCode as Integer
+                // the K8s API can report the pod as terminated before the writes made by the task
+                // are visible on the node running the driver. The `.exitcode` file is the last file
+                // written by the task wrapper, therefore waiting for it also guarantees the
+                // visibility of the task outputs, `.command.env` and `.command.out`.
+                // The wait is skipped when the API reports a failure, because a task killed by the
+                // system never writes the exit file and the API status is authoritative anyway
+                // See https://github.com/nextflow-io/nextflow/issues/7247
+                if( (k8sExitCode == null || k8sExitCode == 0) && !isExitFileVisible() )
+                    return false
                 task.exitStatus = k8sExitCode != null ? k8sExitCode : readExitFile()
                 task.stdout = outputFile
                 task.stderr = errorFile
@@ -489,6 +501,55 @@ class K8sTaskHandler extends TaskHandler implements FusionAwareTask {
         }
     }
 
+    /**
+     * @return
+     *      {@code true} when the task work directory is hosted by a shared file system
+     *      that may expose a delayed view of the writes made by another node.
+     *      Fusion mounts are excluded because task outputs are uploaded to the object
+     *      storage before the task terminates.
+     */
+    protected boolean isWorkDirSharedFS() {
+        return !fusionEnabled() && FileHelper.workDirIsSharedFS
+    }
+
+    protected long getExitReadTimeoutMillis() {
+        return executor.config.getExitReadTimeout(executor.name).toMillis()
+    }
+
+    /**
+     * Check that the task `.exitcode` file is visible on the node running the driver.
+     *
+     * The file is written by the task wrapper only after all other task writes have been
+     * flushed, therefore it acts as a barrier for the whole task work directory. On a
+     * shared file system with delayed write-back the pod can be reported as terminated
+     * while the file is still missing or empty; in that case the task completion is
+     * deferred to a subsequent polling cycle, up to `executor.exitReadTimeout`.
+     *
+     * @return
+     *      {@code true} when the exit file is visible, or the wait timed out,
+     *      {@code false} to keep waiting
+     */
+    protected boolean isExitFileVisible() {
+        if( !isWorkDirSharedFS() )
+            return true
+
+        final attrs = FileHelper.readAttributes(exitFile)
+        if( attrs && attrs.size() > 0 )
+            return true
+
+        if( !terminatedTimestamp )
+            terminatedTimestamp = System.currentTimeMillis()
+
+        final delta = System.currentTimeMillis() - terminatedTimestamp
+        if( delta < getExitReadTimeoutMillis() ) {
+            log.trace "[K8s] Waiting for exit file to become visible for task: `$task.name` -- delta: $delta ms"
+            return false
+        }
+
+        log.warn "[K8s] Exit file is still not available after $delta ms for task: `$task.name` -- work dir: ${task.workDir?.toUriString()}"
+        return true
+    }
+
     /**
      * Terminates the current task execution
      */

plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy

@@ -489,6 +489,7 @@ class K8sTaskHandlerTest extends Specification {
         and:
         def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME, outputFile: OUT_FILE, errorFile: ERR_FILE))
         handler.getClient() >> client
+        handler.isExitFileVisible() >> true
 
         when:
         def result = handler.checkIfCompleted()
@@ -537,6 +538,109 @@ class K8sTaskHandlerTest extends Specification {
 
     }
 
+    def 'should not complete the task until the exit file is visible' () {
+        given:
+        def POD_NAME = 'pod-xyz'
+        def termState = [ reason: "Completed",
+                          startedAt: "2018-01-13T10:09:36Z",
+                          finishedAt: "2018-01-13T10:19:36Z",
+                          exitCode: 0 ]
+        def task = new TaskRun()
+        def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME))
+
+        when:
+        def result = handler.checkIfCompleted()
+        then:
+        1 * handler.getState() >> [terminated: termState]
+        1 * handler.isExitFileVisible() >> false
+        0 * handler.updateTimestamps(_)
+        0 * handler.deleteJobIfSuccessful(_)
+        handler.status != TaskStatus.COMPLETED
+        result == false
+    }
+
+    def 'should not wait for the exit file when the pod reports a failure' () {
+        given:
+        def POD_NAME = 'pod-xyz'
+        def client = Mock(K8sClient)
+        // a task killed by the system (e.g. OOMKilled) never writes the exit file,
+        // therefore waiting for it would only delay the error report
+        def termState = [ reason: "OOMKilled",
+                          startedAt: "2018-01-13T10:09:36Z",
+                          finishedAt: "2018-01-13T10:19:36Z",
+                          exitCode: 137 ]
+        def task = new TaskRun()
+        def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME))
+        handler.getClient() >> client
+
+        when:
+        def result = handler.checkIfCompleted()
+        then:
+        1 * handler.getState() >> [terminated: termState]
+        0 * handler.isExitFileVisible()
+        1 * handler.deleteJobIfSuccessful(task) >> null
+        1 * handler.saveJobLogOnError(task) >> null
+        handler.task.exitStatus == 137
+        handler.status == TaskStatus.COMPLETED
+        result == true
+    }
+
+    def 'should wait for the exit file on a shared file system' () {
+        given:
+        def folder = Files.createTempDirectory('test')
+        def exitFile = folder.resolve('.exitcode')
+        def task = new TaskRun(name: 'foo', workDir: folder)
+        def handler = Spy(new K8sTaskHandler(task: task, exitFile: exitFile))
+        handler.isWorkDirSharedFS() >> true
+        handler.getExitReadTimeoutMillis() >> 10_000
+
+        expect: 'the exit file does not exist yet'
+        !handler.isExitFileVisible()
+
+        when: 'the exit file exists but is still empty'
+        Files.createFile(exitFile)
+        then:
+        !handler.isExitFileVisible()
+
+        when: 'the exit file holds the exit status'
+        exitFile.text = '0'
+        then:
+        handler.isExitFileVisible()
+
+        cleanup:
+        folder?.deleteDir()
+    }
+
+    def 'should give up waiting for the exit file after the read timeout' () {
+        given:
+        def folder = Files.createTempDirectory('test')
+        def exitFile = folder.resolve('.exitcode')
+        def task = new TaskRun(name: 'foo', workDir: folder)
+        def handler = Spy(new K8sTaskHandler(task: task, exitFile: exitFile))
+        handler.isWorkDirSharedFS() >> true
+        handler.getExitReadTimeoutMillis() >> 0
+
+        expect:
+        handler.isExitFileVisible()
+
+        cleanup:
+        folder?.deleteDir()
+    }
+
+    def 'should not wait for the exit file when the work dir is not a shared file system' () {
+        given:
+        def folder = Files.createTempDirectory('test')
+        def task = new TaskRun(name: 'foo', workDir: folder)
+        def handler = Spy(new K8sTaskHandler(task: task, exitFile: folder.resolve('.exitcode')))
+        handler.isWorkDirSharedFS() >> false
+
+        expect:
+        handler.isExitFileVisible()
+
+        cleanup:
+        folder?.deleteDir()
+    }
+
     def 'should use K8s exit code when available' () {
         given:
         def ERR_FILE = Paths.get('err.file')
@@ -550,6 +654,7 @@ class K8sTaskHandlerTest extends Specification {
         def task = new TaskRun()
         def handler = Spy(new K8sTaskHandler(task: task, podName: POD_NAME, outputFile: OUT_FILE, errorFile: ERR_FILE))
         handler.getClient() >> client
+        handler.isExitFileVisible() >> true
 
         when:
         def result = handler.checkIfCompleted()