Fix cache invalidation for module binaries with Wave

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

@@ -16,6 +16,7 @@
 package nextflow.processor
 
 import java.nio.file.Path
+import java.nio.file.Files
 
 import com.google.common.hash.HashCode
 import groovy.json.JsonOutput
@@ -215,13 +216,28 @@ class TaskHasher {
         final tokenizer = new StringTokenizer(script, " \t\n\r\f()[]{};&|<>`")
         while( tokenizer.hasMoreTokens() ) {
             final token = tokenizer.nextToken()
-            final path = session.binEntries.get(token)
+            final path = getBinEntry(token)
             if( path )
                 result.add(path)
         }
         return result
     }
 
+    @Memoized
+    protected Path getBinEntry(String name) {
+        final fromProjectBin = session.binEntries.get(name)
+        if( fromProjectBin )
+            return fromProjectBin
+
+        for( final dir : processor.getBinDirs() ) {
+            final candidate = dir.resolve(name)
+            if( Files.isExecutable(candidate) )
+                return candidate
+        }
+
+        return null
+    }
+
     private String safeTaskName(TaskRun task) {
         return task != null ? task.lazyName() : task.processor.name
     }

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

@@ -16,17 +16,22 @@
 
 package nextflow.processor
 
+import java.nio.file.Files
 import java.nio.file.Path
 
 import nextflow.Session
 import nextflow.script.ProcessConfig
 import spock.lang.Specification
+import spock.lang.TempDir
 /**
  *
  * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
 class TaskHasherTest extends Specification {
 
+    @TempDir
+    Path tempDir
+
     def 'should compute unique task hash' () {
 
         given:
@@ -94,6 +99,55 @@ class TaskHasherTest extends Specification {
         result.contains(Path.of('/some/path/bar.sh'))
     }
 
+    def 'should include referenced module bin files in the task hash' () {
+
+        given:
+        def moduleBin = tempDir.resolve('resources/usr/bin')
+        Files.createDirectories(moduleBin)
+        def script = moduleBin.resolve('foo.sh')
+        Files.writeString(script, 'echo first version\n')
+        script.toFile().setExecutable(true)
+        def session = Mock(Session) {
+            getUniqueId() >> UUID.fromString('b69b6eeb-b332-4d2c-9957-c291b15f498c')
+            getBinEntries() >> [:]
+        }
+        def config = Mock(ProcessConfig) {
+            getHashMode() >> null
+        }
+        def processor = Mock(TaskProcessor) {
+            getName() >> 'hello'
+            getSession() >> session
+            getConfig() >> config
+            getBinDirs() >> [moduleBin]
+        }
+        def task = Mock(TaskRun) {
+            getSource() >> 'foo.sh --version'
+            isContainerEnabled() >> false
+            getInputs() >> [:]
+            getOutputEvals() >> null
+            getConfig() >> Mock(TaskConfig) {
+                getModule() >> null
+                getStubBlock() >> null
+                getArchitecture() >> null
+            }
+            getCondaEnv() >> null
+            getSpackEnv() >> null
+            getProcessor() >> processor
+        }
+
+        when:
+        def firstHash = Spy(TaskHasher, constructorArgs: [task]) {
+            getTaskGlobalVars() >> [:]
+        }.compute()
+        Files.writeString(script, 'echo second version\n')
+        def secondHash = Spy(TaskHasher, constructorArgs: [task]) {
+            getTaskGlobalVars() >> [:]
+        }.compute()
+
+        then:
+        firstHash != secondHash
+    }
+
     def 'should get task directive vars' () {
         given:
         def processor = Spy(TaskProcessor) {