Fix cache invalidation for module binaries

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

@@ -96,6 +96,15 @@ class TaskHasher {
             keys.addAll(binEntries)
         }
 
+        // add the fingerprint of the module resources bundle (e.g. module binaries)
+        // since these files are not staged individually like the project bin dir
+        final moduleBundle = session.enableModuleBinaries() ? processor.getModuleBundle() : null
+        if( moduleBundle && moduleBundle.hasEntries() ) {
+            final fingerprint = moduleBundle.fingerprint()
+            log.trace "Task: ${task.processor.name} > Adding module bundle fingerprint: ${fingerprint}"
+            keys.add(fingerprint)
+        }
+
         // add environment modules (`module` directive)
         final modules = task.getConfig().getModule()
         if( modules ) {

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

@@ -20,6 +20,7 @@ import java.nio.file.Path
 
 import nextflow.Session
 import nextflow.script.ProcessConfig
+import nextflow.script.bundle.ResourcesBundle
 import spock.lang.Specification
 /**
  *
@@ -94,6 +95,44 @@ class TaskHasherTest extends Specification {
         result.contains(Path.of('/some/path/bar.sh'))
     }
 
+    def 'should include module bundle fingerprint in the task hash' () {
+
+        given: 'two module bundles with the same fingerprint, then one with a different fingerprint'
+        def bundleA1 = Mock(ResourcesBundle) { asBoolean() >> true; hasEntries() >> true; fingerprint() >> 'aaaaaaaa' }
+        def bundleA2 = Mock(ResourcesBundle) { asBoolean() >> true; hasEntries() >> true; fingerprint() >> 'aaaaaaaa' }
+        def bundleB = Mock(ResourcesBundle) { asBoolean() >> true; hasEntries() >> true; fingerprint() >> 'bbbbbbbb' }
+        and: 'a task whose process resolves those bundles across successive hash computations'
+        def session = Mock(Session) {
+            getUniqueId() >> UUID.fromString('b69b6eeb-b332-4d2c-9957-c291b15f498c')
+            getBinEntries() >> [:]
+            enableModuleBinaries() >> true
+        }
+        def processor = Mock(TaskProcessor) {
+            getName() >> 'hello'
+            getSession() >> session
+            getConfig() >> Mock(ProcessConfig)
+            getModuleBundle() >>> [bundleA1, bundleA2, bundleB]
+        }
+        def task = Mock(TaskRun) {
+            getSource() >> 'hello world'
+            isContainerEnabled() >> false
+            getConfig() >> Mock(TaskConfig)
+            getProcessor() >> processor
+        }
+        def hasher = Spy(new TaskHasher(task))
+        hasher.getTaskGlobalVars() >> [:]
+
+        when:
+        def hashA1 = hasher.compute()
+        def hashA2 = hasher.compute()
+        def hashB = hasher.compute()
+
+        then: 'the same bundle fingerprint yields the same hash'
+        hashA1 == hashA2
+        and: 'a different bundle fingerprint invalidates the cache'
+        hashA1 != hashB
+    }
+
     def 'should get task directive vars' () {
         given:
         def processor = Spy(TaskProcessor) {