Fix listFiles() and listDirectory() to follow symbolic links (#6808)

Signed-off-by: adamrtalbot <12817534+adamrtalbot@users.noreply.github.com>

Author: adamrtalbot

modules/nextflow/src/main/groovy/nextflow/extension/PublishOp.groovy

@@ -66,18 +66,38 @@ class PublishOp {
         if( opts.pathResolver instanceof Closure )
             this.pathResolver = opts.pathResolver as Closure
         if( opts.index )
-            this.indexOpts = new IndexOpts(session.outputDir, opts.index as Map)
+            this.indexOpts = new IndexOpts(opts.index as Map)
     }
 
     DataflowVariable apply() {
         final events = new HashMap(2)
-        events.onNext = this.&onNext
-        events.onComplete = this.&onComplete
+        events.onNext = { value ->
+            safeExecute { onNext(value) }
+        }
+        events.onComplete = {
+            safeExecute { onComplete() }
+        }
         DataflowHelper.subscribeImpl(source, events)
         this.target = new DataflowVariable()
         return target
     }
 
+    /**
+     * Perform an action. If an exception is raised, bind the
+     * excpetion to the target and don't perform any more actions.
+     *
+     * @param action
+     */
+    private void safeExecute(Runnable action) {
+        if( target.isError() )
+            return
+        try {
+            action.run()
+        } catch( Throwable e ) {
+            target.bindError(e)
+        }
+    }
+
     /**
      * For each incoming value, perform the following:
      *
@@ -151,7 +171,8 @@ class PublishOp {
         if( resolvedPath instanceof CharSequence )
             return outputDir.resolve(resolvedPath.toString())
 
-        throw new ScriptRuntimeException("Invalid output `path` directive -- it should either return a string or use the `>>` operator to publish files")
+        final invalid = mapping ?: resolvedPath
+        throw new ScriptRuntimeException("Invalid `path` directive for workflow output '${name}' -- expected a string or publish statements, but received: ${invalid} [${invalid.class.simpleName}]")
     }
 
     private class PublishDsl {
@@ -165,12 +186,12 @@ class PublishOp {
             }
             else if( source instanceof Collection<Path> ) {
                 if( !target.endsWith('/') )
-                    throw new ScriptRuntimeException("Invalid publish target '${target}' -- should be a directory (end with a `/`) when publishing a collection of files")
+                    throw new ScriptRuntimeException("Invalid publish target '${target}' for workflow output '${name}' -- should be a directory (end with a `/`) when publishing a collection of files")
                 for( final path : source )
                     publish0(path, target)
             }
             else {
-                throw new ScriptRuntimeException("Publish source should be a file or collection of files, but received a ${source.class.name}")
+                throw new ScriptRuntimeException("Invalid publish source for workflow output '${name}' -- expected a file or collection of files, but received: ${source} [${source.class.simpleName}]")
             }
         }
 
@@ -196,14 +217,16 @@ class PublishOp {
      * Once all channel values have been published, publish the final
      * workflow output and index file (if enabled).
      */
-    protected void onComplete(nope) {
+    protected void onComplete() {
         // publish individual record if source is a value channel
         final outputValue = CH.isValue(source)
             ? publishedValues.first()
             : publishedValues
 
         // publish workflow output
-        final indexPath = indexOpts ? indexOpts.path : null
+        final indexPath = indexOpts
+            ? session.outputDir.resolve(indexOpts.path)
+            : null
         session.notifyWorkflowOutput(new WorkflowOutputEvent(name, outputValue, indexPath))
 
         // write value to index file
@@ -220,7 +243,7 @@ class PublishOp {
                 indexPath.text = DumpHelper.prettyPrintYaml(outputValue)
             }
             else {
-                log.warn "Invalid extension '${ext}' for index file '${indexPath}' -- should be CSV, JSON, or YAML"
+                throw new ScriptRuntimeException("Invalid extension '${ext}' for index file '${indexOpts.path}' -- should be CSV, JSON, or YAML")
             }
             session.notifyFilePublish(new FilePublishEvent(null, indexPath, publishOpts.labels as List))
         }
@@ -292,7 +315,7 @@ class PublishOp {
             }
         }
 
-        throw new IllegalArgumentException("Index file record must be a list, map, or file: ${value} [${value.class.simpleName}]")
+        throw new ScriptRuntimeException("Invalid value for workflow output '${name}' -- expected a list, map, or file, but received: ${value} [${value.class.simpleName}]")
     }
 
     /**
@@ -359,13 +382,12 @@ class PublishOp {
     }
 
     static class IndexOpts {
-        Path path
+        String path
         def /* boolean | List<String> */ header = false
         String sep = ','
 
-        IndexOpts(Path targetDir, Map opts) {
-            this.path = targetDir.resolve(opts.path as String)
-
+        IndexOpts(Map opts) {
+            this.path = opts.path as String
             if( opts.header != null )
                 this.header = opts.header
             if( opts.sep )

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

@@ -416,24 +416,14 @@ class AnsiLogObserver implements TraceObserverV2 {
     private final static Pattern TAG_REGEX = ~/ \((.+)\)( *)$/
     private final static Pattern LBL_REPLACE = ~/ \(.+\) *$/
 
-    // OSC 8 hyperlink escape sequences
-    // Format: ESC ] 8 ; ; URL ST TEXT ESC ] 8 ; ; ST
-    // Using BEL (\007) as String Terminator for better compatibility
+    // OSC 8 hyperlink escape sequences (using BEL as String Terminator)
     private final static String HYPERLINK_START = '\033]8;;'
     private final static String HYPERLINK_SEP = '\007'
     private final static String HYPERLINK_END = '\033]8;;\007'
 
-    /**
-     * Wraps text in an OSC 8 terminal hyperlink escape sequence.
-     *
-     * @param text The visible text to display
-     * @param url The URL to link to
-     * @return The text wrapped in hyperlink escape sequences
-     */
     protected static String hyperlink(String text, String url) {
         if( !url )
             return text
-        // Add file:// prefix for local paths (toUriString returns raw path for local files)
         final href = url.startsWith('/') ? 'file://' + url : url
         return HYPERLINK_START + href + HYPERLINK_SEP + text + HYPERLINK_END
     }
@@ -464,8 +454,7 @@ class AnsiLogObserver implements TraceObserverV2 {
         // eg. 1 of 1
         final numbs = " ${(int)com} of ${(int)tot}".toString()
 
-        // Task hash, eg: [fa/71091a]
-        // If workDir is available and cleanup is not enabled, make the hash a clickable hyperlink
+        // Task hash - make clickable hyperlink to work dir when available (and cleanup not enabled)
         final hashDisplay = (stats.workDir && !session.config.cleanup) ? hyperlink(hh, stats.workDir) : hh
         term.a(Attribute.INTENSITY_FAINT).a('[').reset()
         term.fg(Color.BLUE).a(hashDisplay).reset()

modules/nextflow/src/test/groovy/nextflow/script/OutputDslTest.groovy

@@ -1,10 +1,12 @@
 package nextflow.script
 
 import java.nio.file.Files
+import java.nio.file.Path
 
 import nextflow.Channel
 import nextflow.Session
 import nextflow.SysEnv
+import nextflow.exception.ScriptRuntimeException
 import nextflow.trace.event.FilePublishEvent
 import nextflow.trace.event.WorkflowOutputEvent
 import nextflow.trace.event.WorkflowPublishEvent
@@ -208,4 +210,104 @@ class OutputDslTest extends Specification {
         ]
     }
 
+    def 'should report error for invalid path directive' () {
+        when:
+        def session = new Session(outputDir: Path.of('results'))
+
+        session.outputs.put('foo', Channel.of(1, 2, 3))
+
+        def dsl = new OutputDsl()
+        dsl.declare('foo') {
+            path { v -> 42 }
+        }
+        dsl.apply(session)
+        session.fireDataflowNetwork()
+        dsl.getOutput()
+
+        then:
+        def e = thrown(ScriptRuntimeException)
+        e.message.contains "Invalid `path` directive for workflow output 'foo'"
+        e.message.contains "expected a string or publish statements, but received: 42 [Integer]"
+    }
+
+    def 'should report error for invalid publish target' () {
+        when:
+        def session = new Session(outputDir: Path.of('results'))
+        def file = Path.of('output.txt')
+
+        session.outputs.put('foo', Channel.of([file, file, file]))
+
+        def dsl = new OutputDsl()
+        dsl.declare('foo') {
+            path { files -> publish(files, 'foo.txt') }
+        }
+        dsl.apply(session)
+        session.fireDataflowNetwork()
+        dsl.getOutput()
+
+        then:
+        def e = thrown(ScriptRuntimeException)
+        e.message.contains "Invalid publish target 'foo.txt' for workflow output 'foo'"
+    }
+
+    def 'should report error for invalid publish source' () {
+        when:
+        def session = new Session(outputDir: Path.of('results'))
+
+        session.outputs.put('foo', Channel.of(42))
+
+        def dsl = new OutputDsl()
+        dsl.declare('foo') {
+            path { v -> publish(v, 'foo') }
+        }
+        dsl.apply(session)
+        session.fireDataflowNetwork()
+        dsl.getOutput()
+
+        then:
+        def e = thrown(ScriptRuntimeException)
+        e.message.contains "Invalid publish source for workflow output 'foo'"
+        e.message.contains "expected a file or collection of files, but received: 42 [Integer]"
+    }
+
+    def 'should report error for invalid index file extension' () {
+        when:
+        def session = new Session(outputDir: Path.of('results'))
+
+        session.outputs.put('foo', Channel.empty())
+
+        def dsl = new OutputDsl()
+        dsl.declare('foo') {
+            index {
+                path 'index.txt'
+            }
+        }
+        dsl.apply(session)
+        session.fireDataflowNetwork()
+        dsl.getOutput()
+
+        then:
+        def e = thrown(ScriptRuntimeException)
+        e.message.contains "Invalid extension 'txt' for index file 'index.txt'"
+    }
+
+    def 'should report error for invalid published value' () {
+        when:
+        def session = new Session(outputDir: Path.of('results'))
+
+        session.outputs.put('foo', Channel.of(42))
+
+        def dsl = new OutputDsl()
+        dsl.declare('foo') {
+        }
+        dsl.apply(session)
+        session.fireDataflowNetwork()
+        dsl.getOutput()
+
+        then:
+        def e = thrown(ScriptRuntimeException)
+        e.message.contains "Invalid value for workflow output 'foo'"
+        e.message.contains "expected a list, map, or file, but received: 42 [Integer]"
+    }
+
 }

modules/nextflow/src/test/groovy/nextflow/script/parser/v2/ScriptLoaderV2Test.groovy

@@ -349,4 +349,36 @@ class ScriptLoaderV2Test extends Dsl2Spec {
         parser.getResult().toString() == 'TUESDAY'
     }
 
+    def 'should report error for invalid publish statements in output block' () {
+        given:
+        def session = new Session()
+        def parser = new ScriptLoaderV2(session)
+
+        def TEXT = '''
+            workflow {
+                main:
+                ch = channel.empty()
+
+                publish:
+                samples = ch
+            }
+
+            output {
+                samples {
+                    path { v ->
+                        if( true ) return 42
+                        v >> 'foo'
+                    }
+                }
+            }
+            '''
+
+        when:
+        parser.parse(TEXT)
+        parser.runScript()
+        then:
+        def e = thrown(ScriptCompilationException)
+        e.cause.message.contains 'Publish statements cannot be mixed with other statements in a dynamic publish path'
+    }
+
 }

modules/nf-commons/src/main/nextflow/extension/FilesEx.groovy

@@ -22,6 +22,7 @@ import java.nio.ByteBuffer
 import java.nio.channels.SeekableByteChannel
 import java.nio.file.FileAlreadyExistsException
 import java.nio.file.FileSystemException
+import java.nio.file.FileVisitOption
 import java.nio.file.FileVisitResult
 import java.nio.file.Files
 import java.nio.file.LinkOption
@@ -658,8 +659,9 @@ class FilesEx {
             return null
 
         final result = []
+        final opts = EnumSet.of(FileVisitOption.FOLLOW_LINKS)
 
-        Files.walkFileTree(self, new SimpleFileVisitor<Path>() {
+        Files.walkFileTree(self, opts, Integer.MAX_VALUE, new SimpleFileVisitor<Path>() {
 
             FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
                 if( filter == null || invokeFilter(filter, file, attrs) )

modules/nf-commons/src/test/nextflow/extension/FilesExTest.groovy

@@ -603,6 +603,28 @@ class FilesExTest extends Specification {
         sourceFolder.deleteDir()
     }
 
+    def 'should list files in symlinked directory' () {
+
+        given:
+        def folder = Files.createTempDirectory('test')
+        def realDir = folder.resolve('realDir')
+        Files.createDirectories(realDir)
+        realDir.resolve('file1.txt').text = 'Hello1'
+        realDir.resolve('file2.txt').text = 'Hello2'
+        def linkDir = folder.resolve('linkDir')
+        Files.createSymbolicLink(linkDir, realDir)
+
+        when:
+        def paths = linkDir.listFiles()
+
+        then:
+        paths.size() == 2
+        paths.collect { it.name }.sort() == ['file1.txt', 'file2.txt']
+
+        cleanup:
+        folder?.deleteDir()
+    }
+
 
     def testSetReadonly() {