Fix cached task after abort

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<String> 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"
 

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
      */

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
     }

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

@@ -24,13 +24,16 @@ 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
 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()
+    }
 }

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([:])