From 015d3e70b3c702cbfb87d434bf7d6f1da0eee0c3 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 16 Aug 2026 10:20:21 +0200 Subject: [PATCH 1/3] Recognise ceph, beegfs and gpfs as shared file systems `FileHelper.isPathSharedFS` only matched `nfs` and `lustre`, so every delayed-visibility mitigation gated on `workDirIsSharedFS` was silently disabled on CephFS (and BeeGFS/GPFS) work directories. Extract the type classification into `isSharedFsType` so it can be tested without depending on the file system the tests run on. Ref https://github.com/nextflow-io/nextflow/issues/7247 Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso --- .../src/main/nextflow/file/FileHelper.groovy | 25 ++++++++++++++++--- .../test/nextflow/file/FileHelperTest.groovy | 19 ++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy b/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy index 2ded21331e..174ee9606f 100644 --- a/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy +++ b/modules/nf-commons/src/main/nextflow/file/FileHelper.groovy @@ -472,10 +472,28 @@ 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 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) { @@ -483,8 +501,7 @@ class FileHelper { 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 } diff --git a/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy b/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy index c95513f1b8..7260473798 100644 --- a/modules/nf-commons/src/test/nextflow/file/FileHelperTest.groovy +++ b/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') From e6256b35abe418b102ee21f753e9b5ee79e10ae9 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 16 Aug 2026 10:30:45 +0200 Subject: [PATCH 2/3] Flush the file system before writing the task exit file The `.exitcode` file is the completion marker polled by the executors, but `on_exit` wrote it *before* running the optional `sync` command. As a result `NXF_ENABLE_FS_SYNC=true` provided no write ordering guarantee: on a shared file system with delayed write-back the driver could observe the marker (or, for API driven executors such as K8s, the terminated state that follows it) while the task outputs were still missing or truncated. Move the sync command ahead of the exit file write so that everything the task wrote into the work directory is flushed before completion is published. Ref https://github.com/nextflow-io/nextflow/issues/7247 Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso --- .../resources/nextflow/executor/command-run.txt | 10 ++++++++-- .../executor/BashWrapperBuilderTest.groovy | 17 +++++++++++++++++ .../executor/test-bash-wrapper-with-trace.txt | 2 +- .../nextflow/executor/test-bash-wrapper.txt | 2 +- 4 files changed, 27 insertions(+), 4 deletions(-) diff --git a/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt b/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt index 5aefcaa57b..a23cdb4073 100644 --- a/modules/nextflow/src/main/resources/nextflow/executor/command-run.txt +++ b/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 } diff --git a/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy b/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy index 75e1c1c99f..506101fb14 100644 --- a/modules/nextflow/src/test/groovy/nextflow/executor/BashWrapperBuilderTest.groovy +++ b/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') diff --git a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt index b766c16f00..fcef9adde4 100644 --- a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper-with-trace.txt +++ b/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 } diff --git a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt b/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt index 28ff124443..bc56898c26 100644 --- a/modules/nextflow/src/test/resources/nextflow/executor/test-bash-wrapper.txt +++ b/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 } From 87890e1df26024adbbe1cd6f1df808ac4af71991 Mon Sep 17 00:00:00 2001 From: Paolo Di Tommaso Date: Sun, 16 Aug 2026 11:15:16 +0200 Subject: [PATCH 3/3] Wait for the exit file before completing a K8s task on a shared FS The K8s executor detects task completion from the pod terminated state returned by the API, then immediately reads the task work directory. On a shared file system with delayed write-back the driver can observe the pod as terminated while the task writes are not yet visible on its own mount, which surfaces as an empty `.exitcode` ("terminated for an unknown reason"), a truncated `.command.env` ("Missing environment variable: nxf_out_eval_N") or missing output files. Defer the task completion until the `.exitcode` file is visible and non empty, up to `executor.exitReadTimeout`, mirroring the logic already used by the grid executors. Since the exit file is the last file written by the task wrapper, it acts as a barrier for the whole work directory. The wait is skipped when the work dir is not a shared file system, when Fusion is enabled, and when the API reports a non-zero exit code -- a task killed by the system (e.g. OOMKilled) never writes the exit file, so waiting for it would only delay the error report. Ref https://github.com/nextflow-io/nextflow/issues/7247 Assisted-by: Claude Code (Opus 5) Signed-off-by: Paolo Di Tommaso --- .../main/nextflow/k8s/K8sTaskHandler.groovy | 63 ++++++++++- .../nextflow/k8s/K8sTaskHandlerTest.groovy | 105 ++++++++++++++++++ 2 files changed, 167 insertions(+), 1 deletion(-) diff --git a/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy b/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy index 270e20dad6..0a6b179e63 100644 --- a/plugins/nf-k8s/src/main/nextflow/k8s/K8sTaskHandler.groovy +++ b/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 */ diff --git a/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy b/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy index 1ecf193bec..bd80685345 100644 --- a/plugins/nf-k8s/src/test/nextflow/k8s/K8sTaskHandlerTest.groovy +++ b/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()