Use fusion symlinks file to resolve links

Signed-off-by: Ben Sherman <[email protected]>

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

@@ -29,6 +29,7 @@ import java.util.concurrent.ExecutorService
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.transform.EqualsAndHashCode
+import groovy.transform.Memoized
 import groovy.transform.PackageScope
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
@@ -287,12 +288,40 @@ class PublishDir {
         // resolve Fusion symlinks
         if( FusionHelper.isFusionEnabled(Global.session as Session) ) {
             final inputFiles = task.getInputFilesMap()
-            files = files.collect { inputFiles.getOrDefault(it.name, it) } as Set<Path>
+            files = files.collect { file ->
+                file.name in inputFiles
+                    ? resolveFusionLink(inputFiles[file.name])
+                    : file
+            } as Set<Path>
         }
 
         apply0(files)
     }
 
+    /**
+     * Resolve a Fusion symlink by following the .fusion.symlinks
+     * file in the task directory until the original file is reached.
+     *
+     * @param file
+     */
+    @CompileStatic
+    protected Path resolveFusionLink(Path file) {
+        while( file.name in getFusionLinks(file.parent) )
+            file = file.text.replace('/fusion/s3/', 's3://') as Path
+        return file
+    }
+
+    @CompileStatic
+    @Memoized
+    protected List<String> getFusionLinks(Path workDir) {
+        try {
+            final file = workDir.resolve('.fusion.symlinks')
+            return file.text.tokenize('\n')
+        }
+        catch( NoSuchFileException ) {
+            return []
+        }
+    }
 
     @CompileStatic
     protected void apply1(Path source, boolean inProcess ) {