Automatic task cleanup (without resumability)

docs/config.md

@@ -1579,10 +1579,32 @@ Read the {ref}`trace-report` page to learn more about the execution report that
 There are additional variables that can be defined within a configuration file that do not have a dedicated scope.
 
 `cleanup`
-: If `true`, on a successful completion of a run all files in *work* directory are automatically deleted.
+: :::{versionchanged} 24.04.0
+  Added `'lazy'`, `'eager'`, and `'aggressive'` strategies.
+  :::
+: Automatically delete task directories using one of the following strategies:
+
+  `false` (default)
+  : Disable automatic cleanup.
+
+  `true`
+  : Equivalent to `'lazy'`.
+
+  `'lazy'`
+  : If a workflow completes successfully, delete all task directories.
+  : Note that deleting all work directories at once can take a lot of time, especially when using a shared file system or remote cloud storage.
 
-  :::{warning}
-  The use of the `cleanup` option will prevent the use of the *resume* feature on subsequent executions of that pipeline run. Also, be aware that deleting all scratch files can take a lot of time, especially when using a shared file system or remote cloud storage.
+  `'eager'`
+  : Delete each task directory as soon as it is no longer needed by downstream tasks.
+  : A task can be deleted once all of the tasks that use any of its output files have completed.
+  : Output files that are published via symlink will be invalidated when the original task directory is deleted. Avoid using the following publish modes: `copyNoFollow`, `rellink`, `symlink`.
+
+  `'aggressive'`
+  : Equivalent to `'eager'`, but also deletes individual output files as soon as they are no longer needed.
+  : An output file can be deleted once the tasks that use it have completed. In some cases, an output file can be deleted sooner than its originating task.
+
+: :::{note}
+  Resumability is not currently supported, but will be supported in a future release for the `'lazy'` and `'eager'` cleanup strategies.
   :::
 
 `dumpHashes`

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -372,6 +372,9 @@ class Session implements ISession {
         // -- file porter config
         this.filePorter = new FilePorter(this)
 
+        // -- normalize cleanup config
+        if( config.cleanup == true )
+            config.cleanup = 'lazy'
     }
 
     protected Path cloudCachePath(Map cloudcache, Path workDir) {
@@ -964,6 +967,17 @@ class Session implements ISession {
         }
     }
 
+    void notifyProcessClose(TaskProcessor process) {
+        observers.each { observer ->
+            try {
+                observer.onProcessClose(process)
+            }
+            catch( Exception e ) {
+                log.debug(e.getMessage(), e)
+            }
+        }
+    }
+
     void notifyProcessTerminate(TaskProcessor process) {
         for( int i=0; i<observers.size(); i++ ) {
             final observer = observers.get(i)
@@ -1149,37 +1163,6 @@ class Session implements ISession {
         errorAction = action
     }
 
-    /**
-     * Delete the workflow work directory from tasks temporary files
-     */
-    void cleanup() {
-        if( !workDir || !config.cleanup )
-            return
-
-        if( aborted || cancelled || error )
-            return
-
-        CacheDB db = null
-        try {
-            log.trace "Cleaning-up workdir"
-            db = CacheFactory.create(uniqueId, runName).openForRead()
-            db.eachRecord { HashCode hash, TraceRecord record ->
-                def deleted = db.removeTaskEntry(hash)
-                if( deleted ) {
-                    // delete folder
-                    FileHelper.deletePath(FileHelper.asPath(record.workDir))
-                }
-            }
-            log.trace "Clean workdir complete"
-        }
-        catch( Exception e ) {
-            log.warn("Failed to cleanup work dir: ${workDir.toUriString()}")
-        }
-        finally {
-            db.close()
-        }
-    }
-
     @Memoized
     CondaConfig getCondaConfig() {
         final cfg = config.conda as Map ?: Collections.emptyMap()

modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy

@@ -183,7 +183,7 @@ class PublishDir {
 
         def result = new PublishDir()
         if( params.path )
-            result.path = params.path
+        result.path = params.path
 
         if( params.mode )
             result.mode = params.mode
@@ -217,6 +217,18 @@ class PublishDir {
         return result
     }
 
+    boolean canPublish(Path source, TaskRun task) {
+        if( !sourceDir ) {
+            this.sourceDir = task.targetDir
+            this.sourceFileSystem = sourceDir.fileSystem
+            this.stageInMode = task.config.stageInMode
+            this.task = task
+            validatePublishMode()
+        }
+
+        return getPublishTarget(source) != null
+    }
+
     protected void apply0(Set<Path> files) {
         assert path
 
@@ -301,13 +313,8 @@ class PublishDir {
 
     protected void apply1(Path source, boolean inProcess ) {
 
-        def target = sourceDir ? sourceDir.relativize(source) : source.getFileName()
-        if( matcher && !matcher.matches(target) ) {
-            // skip not matching file
-            return
-        }
-
-        if( saveAs && !(target=saveAs.call(target.toString()))) {
+        final target = getPublishTarget(source)
+        if( !target ) {
             // skip this file
             return
         }
@@ -339,6 +346,17 @@ class PublishDir {
 
     }
 
+    protected def getPublishTarget(Path source) {
+        def target = sourceDir ? sourceDir.relativize(source) : source.getFileName()
+        if( matcher && !matcher.matches(target) )
+            return null
+
+        if( saveAs && !(target=saveAs.call(target.toString())))
+            return null
+
+        return target
+    }
+
     protected Path resolveDestination(target) {
 
         if( target instanceof Path ) {

modules/nextflow/src/main/groovy/nextflow/processor/TaskProcessor.groovy

@@ -2341,6 +2341,10 @@ class TaskProcessor {
         state.update { StateObj it -> it.incCompleted() }
     }
 
+    protected void closeProcess() {
+        session.notifyProcessClose(this)
+    }
+
     protected void terminateProcess() {
         log.trace "<${name}> Sending poison pills and terminating process"
         sendPoisonPill()
@@ -2493,6 +2497,7 @@ class TaskProcessor {
             // apparently auto if-guard instrumented by @Slf4j is not honoured in inner classes - add it explicitly
             if( log.isTraceEnabled() )
                 log.trace "<${name}> After stop"
+            closeProcess()
         }
 
         /**

modules/nextflow/src/main/groovy/nextflow/script/ScriptRunner.groovy

@@ -257,7 +257,6 @@ class ScriptRunner {
 
     protected shutdown() {
         session.destroy()
-        session.cleanup()
         Global.cleanUp()
         log.debug "> Execution complete -- Goodbye"
     }

modules/nextflow/src/main/groovy/nextflow/trace/CleanupStrategy.groovy

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2013-2023, 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.
+ */
+
+package nextflow.trace
+
+import groovy.transform.CompileStatic
+
+/**
+ * Strategies to automatically cleanup task directories.
+ *
+ * @author Ben Sherman <[email protected]>
+ */
+@CompileStatic
+enum CleanupStrategy {
+    LAZY(1),
+    EAGER(2),
+    AGGRESSIVE(3)
+
+    final int level
+
+    CleanupStrategy(int level) {
+        this.level = level
+    }
+
+    static boolean isValid(CharSequence name) {
+        if( !name )
+            return false
+        try {
+            valueOf(name.toString().toUpperCase())
+            return true
+        }
+        catch( IllegalArgumentException e ) {
+            return false
+        }
+    }
+}

modules/nextflow/src/main/groovy/nextflow/trace/DefaultObserverFactory.groovy

@@ -25,6 +25,7 @@ class DefaultObserverFactory implements TraceObserverFactory {
         createTimelineObserver(result)
         createDagObserver(result)
         createAnsiLogObserver(result)
+        createTaskCleanupObserver(result)
         return result
     }
 
@@ -101,4 +102,12 @@ class DefaultObserverFactory implements TraceObserverFactory {
         result << observer
     }
 
+    protected void createTaskCleanupObserver(Collection<TraceObserver> result) {
+        final strategy = session.config.cleanup
+        if( strategy instanceof CharSequence && !CleanupStrategy.isValid(strategy) )
+            throw new IllegalArgumentException("Invalid cleanup strategy '${strategy}' -- available strategies are ${CleanupStrategy.values().join(',').toLowerCase()}")
+        if( strategy )
+            result << new TaskCleanupObserver(strategy.toString().toUpperCase() as CleanupStrategy)
+    }
+
 }