diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy index 3c46f95287..f4787e7b17 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy @@ -15,8 +15,6 @@ */ package nextflow.processor -import static nextflow.processor.ErrorStrategy.* - import java.nio.file.FileSystems import java.nio.file.Path import java.util.concurrent.atomic.AtomicBoolean @@ -799,32 +797,48 @@ class TaskProcessor { @CompileStatic final protected void checkCachedOrLaunchTask( TaskRun task, HashCode hash, boolean shouldTryCache ) { - int tries = task.failCount +1 + int tries = task.previousTryCount + task.failCount + 1 while( true ) { + // increment the task hash based on the "try count" + // in order to produce a deterministic sequence of + // task hashes for subsequent attempts hash = HashBuilder.defaultHasher().putBytes(hash.asBytes()).putInt(tries).hash() Path resumeDir = null boolean exists = false try { + // check for a cache entry and task directory for the given task hash final entry = session.cache.getTaskEntry(hash, this) resumeDir = entry ? FileHelper.asPath(entry.trace.getWorkDir()) : null if( resumeDir ) exists = resumeDir.exists() + // resume the task if it completed successfully and outputs are present log.trace "[${safeTaskName(task)}] Cacheable folder=${resumeDir?.toUriString()} -- exists=$exists; try=$tries; shouldTryCache=$shouldTryCache; entry=$entry" final cached = shouldTryCache && exists && entry.trace.isCompleted() && checkCachedOutput(task.clone(), resumeDir, hash, entry) if( cached ) break + + // if the task execution failed or was aborted, increment the + // try count so that any subsequent successful execution + // can be recovered + if( entry?.trace?.isFailed() || entry?.trace?.isAborted() ) + task.previousTryCount += 1 } catch (Throwable t) { log.warn1("[${safeTaskName(task)}] Unable to resume cached task -- See log file for details", causedBy: t) } + // if the task directory exists but was not resumed, it is a + // failed or invalid execution -- increment the try count and + // check the next task hash if( exists ) { tries++ continue } + // if the task directory does not exist, execute the task + // with this task hash final lock = lockManager.acquire(hash) final workDir = task.getWorkDirFor(hash) try { @@ -1022,22 +1036,21 @@ class TaskProcessor { return true } - return fault == TERMINATE || fault == FINISH + return fault == ErrorStrategy.TERMINATE || fault == ErrorStrategy.FINISH } /** * @param task The {@code TaskRun} instance that raised an error * @param error The error object * @return - * Either a value of value of {@link ErrorStrategy} representing the error strategy chosen - * or an instance of {@TaskFault} representing the cause of the error (that implicitly means - * a {@link ErrorStrategy#TERMINATE}) + * Either an {@link ErrorStrategy} representing the selected error strategy, + * or a {@link TaskFault} representing the cause of the error (implies {@link ErrorStrategy#TERMINATE}) */ @PackageScope - final synchronized resumeOrDie( TaskRun task, Throwable error, TraceRecord traceRecord = null) { + final synchronized Object resumeOrDie( TaskRun task, Throwable error, TraceRecord traceRecord = null) { log.debug "Handling unexpected condition for\n task: name=${safeTaskName(task)}; work-dir=${task?.workDirStr}\n error [${error?.class?.name}]: ${error?.getMessage()?:error}" - ErrorStrategy errorStrategy = TERMINATE + ErrorStrategy errorStrategy = ErrorStrategy.TERMINATE final List message = [] try { // -- do not recoverable error, just re-throw it @@ -1049,7 +1062,7 @@ class TaskProcessor { log.info "[$task.hashLog] NOTE: ${error.message} -- Execution is retried" else log.info "[$task.hashLog] NOTE: ${error.message} -- Cause: ${error.cause.message} -- Execution is retried" - task.failCount+=1 + task.failCount += 1 final taskCopy = task.makeCopy() session.getExecService().submit { try { @@ -1062,8 +1075,8 @@ class TaskProcessor { } } task.failed = true - task.errorAction = RETRY - return RETRY + task.errorAction = ErrorStrategy.RETRY + return ErrorStrategy.RETRY } final submitTimeout = error.cause instanceof ProcessSubmitTimeoutException @@ -1087,9 +1100,9 @@ class TaskProcessor { errorStrategy = checkErrorStrategy(task, error, taskErrCount, procErrCount, submitRetries) if( errorStrategy.soft ) { def msg = "[$task.hashLog] NOTE: ${submitTimeout ? submitErrMsg : error.message}" - if( errorStrategy == IGNORE ) + if( errorStrategy == ErrorStrategy.IGNORE ) msg += " -- Error is ignored" - else if( errorStrategy == RETRY ) + else if( errorStrategy == ErrorStrategy.RETRY ) msg += " -- Execution is retried (${submitTimeout ? submitRetries : taskErrCount})" log.info msg task.failed = true @@ -1156,16 +1169,16 @@ class TaskProcessor { // retry is not allowed when the script cannot be compiled or similar errors if( error instanceof ProcessUnrecoverableException || error.cause instanceof ProcessUnrecoverableException ) { - return !action.soft ? action : TERMINATE + return !action.soft ? action : ErrorStrategy.TERMINATE } // IGNORE strategy -- just continue - if( action == IGNORE ) { - return IGNORE + if( action == ErrorStrategy.IGNORE ) { + return ErrorStrategy.IGNORE } // RETRY strategy -- check that process do not exceed 'maxError' and the task do not exceed 'maxRetries' - if( action == RETRY ) { + if( action == ErrorStrategy.RETRY ) { final int maxErrors = task.config.getMaxErrors() final int maxRetries = task.config.getMaxRetries() @@ -1184,10 +1197,10 @@ class TaskProcessor { session.abort(e) } } as Runnable) - return RETRY + return ErrorStrategy.RETRY } - return TERMINATE + return ErrorStrategy.TERMINATE } return action @@ -1931,9 +1944,13 @@ class TaskProcessor { * and binding output values accordingly * * @param task The {@code TaskRun} instance to finalize + * @return + * Either an {@link ErrorStrategy} representing the selected error strategy, + * a {@link TaskFault} representing the cause of the error (implies {@link ErrorStrategy#TERMINATE}), + * or null if the task completed successfully. */ @PackageScope - final finalizeTask( TaskHandler handler) { + final Object finalizeTask( TaskHandler handler) { def task = handler.task log.trace "finalizing process > ${safeTaskName(task)} -- $task" diff --git a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy index 4537cf623c..65e53de852 100644 --- a/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/processor/TaskRun.groovy @@ -325,6 +325,13 @@ class TaskRun implements Cloneable { */ volatile int failCount + /** + * The number of unsuccessful attempts in the previous run. Used to + * recover the hash of a successful execution if it occurred after + * one or more failed attempts. + */ + volatile int previousTryCount + /** * The number of times the submit of the task has been retried */ diff --git a/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy b/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy index fdc5344d8f..d74be38e66 100644 --- a/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy +++ b/modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy @@ -600,6 +600,14 @@ class TraceRecord implements Serializable { store.status == 'COMPLETED' } + boolean isAborted() { + store.status == 'ABORTED' + } + + boolean isFailed() { + store.status == 'FAILED' + } + String getExecutorName() { return executorName } diff --git a/modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy b/modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy index a4f6c7ef2d..fc1c14b8dd 100644 --- a/modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/processor/TaskProcessorTest.groovy @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutorService import com.google.common.hash.HashCode import groovyx.gpars.agent.Agent import nextflow.Session +import nextflow.cache.CacheDB import nextflow.exception.IllegalArityException import nextflow.exception.ProcessException import nextflow.exception.ProcessUnrecoverableException @@ -31,6 +32,8 @@ import nextflow.executor.Executor import nextflow.executor.NopeExecutor import nextflow.file.FileHolder import nextflow.file.FilePorter +import nextflow.trace.TraceRecord +import nextflow.util.HashBuilder import nextflow.script.BaseScript import nextflow.script.BodyDef import nextflow.script.ProcessConfig @@ -53,6 +56,13 @@ class TaskProcessorTest extends Specification { return new DummyProcessor(name, session, Mock(BaseScript), new ProcessConfig([:])) } + private static TraceRecord traceRecord(String status, String workdir) { + def rec = new TraceRecord() + rec.put('status', status) + rec.put('workdir', workdir) + return rec + } + static class DummyProcessor extends TaskProcessor { DummyProcessor(String name, Session session, BaseScript script, ProcessConfig config) { @@ -618,4 +628,110 @@ class TaskProcessorTest extends Specification { 0 * collector.collect(task) 1 * exec.submit(task) } + + def 'should increment previousTryCount when cache entry is failed'() { + given: + def baseDir = Files.createTempDirectory('test-cached') + def resumeDir = Files.createDirectory(baseDir.resolve('resume-dir')) + def cache = Mock(CacheDB) + def session = Mock(Session) { getCache() >> cache } + def executor = Mock(Executor) { getWorkDir() >> baseDir } + def processor = new TaskProcessor(session: session, executor: executor, name: 'foo') + and: + def task = new TaskRun(config: new TaskConfig(), name: 'foo', processor: processor) + + when: + processor.checkCachedOrLaunchTask(task, HashCode.fromString('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), true) + + then: + // first lookup: failed entry with an existing resumeDir → forces next hash + 1 * cache.getTaskEntry(_, processor) >> new TaskEntry(trace: traceRecord('FAILED', resumeDir.toString())) + // second lookup: no entry → task is submitted + 1 * cache.getTaskEntry(_, processor) >> null + 1 * executor.submit(task) + and: + task.previousTryCount == 1 + + cleanup: + baseDir.toFile().deleteDir() + } + + def 'should increment previousTryCount when cache entry is aborted'() { + given: + def baseDir = Files.createTempDirectory('test-cached') + def resumeDir = Files.createDirectory(baseDir.resolve('resume-dir')) + def cache = Mock(CacheDB) + def session = Mock(Session) { getCache() >> cache } + def executor = Mock(Executor) { getWorkDir() >> baseDir } + def processor = new TaskProcessor(session: session, executor: executor, name: 'foo') + and: + def task = new TaskRun(config: new TaskConfig(), name: 'foo', processor: processor) + + when: + processor.checkCachedOrLaunchTask(task, HashCode.fromString('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'), true) + + then: + 1 * cache.getTaskEntry(_, processor) >> new TaskEntry(trace: traceRecord('ABORTED', resumeDir.toString())) + 1 * cache.getTaskEntry(_, processor) >> null + 1 * executor.submit(task) + and: + task.previousTryCount == 1 + + cleanup: + baseDir.toFile().deleteDir() + } + + def 'should not increment previousTryCount when no cache entry exists'() { + given: + def baseDir = Files.createTempDirectory('test-cached') + def cache = Mock(CacheDB) + def session = Mock(Session) { getCache() >> cache } + def executor = Mock(Executor) { getWorkDir() >> baseDir } + def processor = new TaskProcessor(session: session, executor: executor, name: 'foo') + and: + def task = new TaskRun(config: new TaskConfig(), name: 'foo', processor: processor) + + when: + processor.checkCachedOrLaunchTask(task, HashCode.fromString('cccccccccccccccccccccccccccccccc'), false) + + then: + 1 * cache.getTaskEntry(_, processor) >> null + 1 * executor.submit(task) + and: + task.previousTryCount == 0 + + cleanup: + baseDir.toFile().deleteDir() + } + + def 'should seed initial tries from previousTryCount and failCount'() { + given: + def baseDir = Files.createTempDirectory('test-cached') + def cache = Mock(CacheDB) + def session = Mock(Session) { getCache() >> cache } + def executor = Mock(Executor) { getWorkDir() >> baseDir } + def processor = new TaskProcessor(session: session, executor: executor, name: 'foo') + and: + def task = new TaskRun(config: new TaskConfig(), name: 'foo', processor: processor) + task.previousTryCount = 2 + task.failCount = 1 + def inputHash = HashCode.fromString('dddddddddddddddddddddddddddddddd') + + when: + processor.checkCachedOrLaunchTask(task, inputHash, false) + + then: + // no cached entry exists so the task is submitted immediately with the seeded hash + 1 * cache.getTaskEntry(_, processor) >> null + 1 * executor.submit(task) + and: + // the seeded hash corresponds to tries = previousTryCount + failCount + 1 = 4 + task.hash == HashBuilder.defaultHasher().putBytes(inputHash.asBytes()).putInt(4).hash() + and: + // no failed/aborted entry encountered, so counter is unchanged + task.previousTryCount == 2 + + cleanup: + baseDir.toFile().deleteDir() + } } diff --git a/modules/nextflow/src/test/groovy/nextflow/trace/TraceRecordTest.groovy b/modules/nextflow/src/test/groovy/nextflow/trace/TraceRecordTest.groovy index deabb2253d..d6d4fc8273 100644 --- a/modules/nextflow/src/test/groovy/nextflow/trace/TraceRecordTest.groovy +++ b/modules/nextflow/src/test/groovy/nextflow/trace/TraceRecordTest.groovy @@ -340,6 +340,40 @@ class TraceRecordTest extends Specification { 'COMPLETED' | true } + @Unroll + def 'should validate aborted status' () { + given: + def rec = new TraceRecord([status:STATUS]) + + expect: + rec.isAborted() == EXPECTED + + where: + STATUS | EXPECTED + null | false + 'NEW' | false + 'COMPLETED' | false + 'FAILED' | false + 'ABORTED' | true + } + + @Unroll + def 'should validate failed status' () { + given: + def rec = new TraceRecord([status:STATUS]) + + expect: + rec.isFailed() == EXPECTED + + where: + STATUS | EXPECTED + null | false + 'NEW' | false + 'COMPLETED' | false + 'ABORTED' | false + 'FAILED' | true + } + def 'should throw file not found exception' () { given: def rec = new TraceRecord([:]) diff --git a/tests/checks/resume-retried-with-abort.nf/.checks b/tests/checks/resume-retried-with-abort.nf/.checks new file mode 100644 index 0000000000..4aea9df918 --- /dev/null +++ b/tests/checks/resume-retried-with-abort.nf/.checks @@ -0,0 +1,35 @@ +set -e + +# +# run with abort +# +echo '' +timeout --signal=INT 6 $NXF_RUN | tee stdout || true + +[[ `< .nextflow.log grep -c 'Submitted process > SMALL_SLEEP_RETRY'` == 1 ]] || false + +# +# RESUME mode with fail and abort +# +echo '' +timeout --signal=INT 12 $NXF_RUN -resume | tee stdout || true + +[[ `< .nextflow.log grep -c 'Submitted process > SMALL_SLEEP_RETRY'` == 1 ]] || false +[[ `< .nextflow.log grep -c 'Re-submitted process > SMALL_SLEEP_RETRY'` == 1 ]] || false + +# +# RESUME mode with completed and abort +# +echo '' +timeout --signal=INT 20 $NXF_RUN -resume | tee stdout || true + +[[ `< .nextflow.log grep -c 'Submitted process > SMALL_SLEEP_RETRY'` == 1 ]] || false +[[ `< .nextflow.log grep -c 'Re-submitted process > SMALL_SLEEP_RETRY'` == 1 ]] || false + +# +# RESUME mode with cached and finish +# +echo '' +$NXF_RUN -resume | tee stdout + +[[ `< .nextflow.log grep -c 'Cached process > SMALL_SLEEP_RETRY'` == 1 ]] || false diff --git a/tests/resume-retried-task.nf b/tests/resume-retried-task.nf index d7ecb85998..b3320392c5 100644 --- a/tests/resume-retried-task.nf +++ b/tests/resume-retried-task.nf @@ -20,16 +20,18 @@ process foo { maxRetries 3 script: - if( task.attempt < 3 ) - """ - exit 1 - """ - else - """ - echo ciao - """ + if( task.attempt < 3 ) { + """ + exit 1 + """ + } + else { + """ + echo ciao + """ + } } workflow { - foo() + foo() } diff --git a/tests/resume-retried-with-abort.nf b/tests/resume-retried-with-abort.nf new file mode 100644 index 0000000000..cdff53f4ce --- /dev/null +++ b/tests/resume-retried-with-abort.nf @@ -0,0 +1,58 @@ +#!/usr/bin/env nextflow +/* + * Copyright 2013-2026, Seqera Labs + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +process LONG_SLEEP { + tag "long-sleep" + + output: + stdout + + script: + ''' + echo "LONG_SLEEP start" + sleep 30 + echo "LONG_SLEEP done" + ''' +} + +process SMALL_SLEEP_RETRY { + tag "small-sleep-retry" + + errorStrategy 'retry' + maxRetries 1 + + output: + stdout + + script: + """ + echo "SMALL_SLEEP_RETRY attempt: ${task.attempt}" + sleep 7 + + if [[ ${task.attempt} -eq 1 ]]; then + echo "Failing first attempt on purpose" + exit 1 + fi + + echo "Second attempt succeeded" + """ +} + +workflow { + LONG_SLEEP() + SMALL_SLEEP_RETRY() +}