Track remote input file in FileHolder

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

@@ -1946,7 +1946,7 @@ class TaskProcessor {
             if( item instanceof Path || coerceToPath ) {
                 final path = resolvePath(item)
                 final target = executor.isForeignFile(path) ? foreignFiles.addToForeign(path) : path
-                final holder = new FileHolder(target)
+                final holder = new FileHolder(path, target)
                 files << holder
             }
             else {

modules/nf-commons/src/main/nextflow/file/FileHolder.groovy

@@ -20,39 +20,39 @@ package nextflow.file
 import java.nio.file.FileSystems
 import java.nio.file.Path
 
+import com.google.common.hash.Hasher
 import groovy.transform.CompileStatic
 import groovy.transform.EqualsAndHashCode
 import groovy.transform.PackageScope
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
+import nextflow.util.CacheFunnel
+import nextflow.util.CacheHelper
+import nextflow.util.CacheHelper.HashMode
 /**
  * Implements a special {@code Path} used to stage files in the work area
  */
 @Slf4j
 @ToString(includePackage = false, includeNames = true)
 @EqualsAndHashCode
 @CompileStatic
-class FileHolder  {
+class FileHolder implements CacheFunnel {
 
     final def sourceObj
 
     final Path storePath
 
     final String stageName
 
-    FileHolder( Path inputFile ) {
-        assert inputFile
-        this.sourceObj = inputFile
-        this.storePath = real(inputFile)
-        this.stageName = norm(inputFile.getFileName())
+    FileHolder( Path path ) {
+        this.sourceObj = path
+        this.storePath = real(path)
+        this.stageName = norm(path.getFileName())
     }
 
     FileHolder( def origin, Path path ) {
-        assert origin != null
-        assert path != null
-
         this.sourceObj = origin
-        this.storePath = path
+        this.storePath = real(path)
         this.stageName = norm(path.getFileName())
     }
 
@@ -66,10 +66,19 @@ class FileHolder  {
         new FileHolder( this.sourceObj, this.storePath, stageName )
     }
 
+    Path getSourcePath() {
+        sourceObj instanceof Path ? sourceObj : null
+    }
+
     Path getStorePath() { storePath }
 
     String getStageName() { stageName }
 
+    @Override
+    Hasher funnel(Hasher hasher, HashMode mode) {
+        return CacheHelper.hasher(hasher, sourceObj, mode)
+    }
+
     @PackageScope
     static FileHolder get( def path, def name = null ) {
         Path storePath = path as Path

modules/nf-commons/src/main/nextflow/util/HashBuilder.java

@@ -44,7 +44,6 @@
 import nextflow.ISession;
 import nextflow.extension.Bolts;
 import nextflow.extension.FilesEx;
-import nextflow.file.FileHolder;
 import nextflow.io.SerializableMarker;
 import nextflow.script.types.Bag;
 import org.slf4j.Logger;
@@ -163,9 +162,6 @@ else if( value instanceof Collection)
             for( Object item : ((Collection)value) )
                 with(item);
 
-        else if( value instanceof FileHolder )
-            with(((FileHolder) value).getSourceObj());
-
         else if( value instanceof Path )
             hashFile(hasher, (Path)value, mode, basePath);