Prevent cache invalidation for task directives (#4339)

This commit prevents the cache invalidation then a process directive is modified
over a resume, as introduced by PR #2382, since they are configured hints for the process 
execution. When the task has been successfully completed, should not be re-executed if a 
directive value changes e.g. `cpus` or `memory`.  

The extension directive i.e. with the prefix `ext.` is an exception to this rule and will invalidate 
the cache in any case. 

The experimental variable `NXF_ENABLE_CACHE_INVALIDATION_ON_TASK_DIRECTIVE_CHANGE` 
has been removed in favour of the above logic.

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

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

@@ -131,12 +131,6 @@ class TaskProcessor {
 
     final private static Pattern QUESTION_MARK = ~/(\?+)/
 
-    @Memoized
-    static boolean getInvalidateCacheOnTaskDirectiveChange() {
-        final value = System.getenv("NXF_ENABLE_CACHE_INVALIDATION_ON_TASK_DIRECTIVE_CHANGE")
-        return value==null || value =='true'
-    }
-
     @TestOnly private static volatile TaskProcessor currentProcessor0
 
     @TestOnly static TaskProcessor currentProcessor() { currentProcessor0 }
@@ -2210,19 +2204,17 @@ class TaskProcessor {
 
     protected Map<String,Object> getTaskGlobalVars(TaskRun task) {
         final result = task.getGlobalVars(ownerScript.binding)
-        if( invalidateCacheOnTaskDirectiveChange ) {
-            final directives = getTaskDirectiveVars(task)
-            result.putAll(directives)
-        }
+        final directives = getTaskExtensionDirectiveVars(task)
+        result.putAll(directives)
         return result
     }
 
-    protected Map<String,Object> getTaskDirectiveVars(TaskRun task) {
+    protected Map<String,Object> getTaskExtensionDirectiveVars(TaskRun task) {
         final variableNames = task.getVariableNames()
         final result = new HashMap(variableNames.size())
         final taskConfig = task.config
         for( String key : variableNames ) {
-            if( !key.startsWith('task.') ) continue
+            if( !key.startsWith('task.ext.') ) continue
             final value = taskConfig.eval(key.substring(5))
             result.put(key, value)
         }

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

@@ -864,13 +864,12 @@ class TaskProcessorTest extends Specification {
         config.setContext( foo: 'DDDD', bar: 'OOOO' )
 
         when:
-        def result = processor.getTaskDirectiveVars(task)
+        def result = processor.getTaskExtensionDirectiveVars(task)
         then:
         1 * task.getVariableNames() >> {[ 'task.cpus', 'task.ext.alpha', 'task.ext.delta', 'task.ext.omega' ] as Set}
         1 * task.getConfig() >> config
         then:
         result == [
-                'task.cpus': 4,
                 'task.ext.alpha': 'AAAA',
                 'task.ext.delta': 'DDDD',
                 'task.ext.omega': 'OOOO',