Fix missing wave response (#5547) [ci fast]

Signed-off-by: Paolo Di Tommaso <[email protected]>
nextflow-io · Nov 27, 2024 · ee25217 · ee25217
1 parent 09ccd29
commit ee25217
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 14 deletions.
diff --git a/plugins/nf-wave/src/main/io/seqera/wave/plugin/WaveClient.groovy b/plugins/nf-wave/src/main/io/seqera/wave/plugin/WaveClient.groovy
@@ -27,6 +27,7 @@ import java.time.Duration
 import java.time.Instant
 import java.time.OffsetDateTime
 import java.time.temporal.ChronoUnit
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 import java.util.function.Predicate
@@ -104,7 +105,9 @@ class WaveClient {
 
     final private String endpoint
 
-    private Cache<String, Handle> cache
+    private Cache<String, SubmitContainerTokenResponse> cache
+
+    private Map<String,Handle> responses = new ConcurrentHashMap<>()
 
     private Session session
 
@@ -135,7 +138,7 @@ class WaveClient {
         this.packer = new Packer().withPreserveTimestamp(config.preserveFileTimestamp())
         this.waveRegistry = new URI(endpoint).getAuthority()
         // create cache
-        cache = CacheBuilder<String, Handle>
+        this.cache = CacheBuilder<String, Handle>
             .newBuilder()
             .expireAfterWrite(config.tokensCacheMaxDuration().toSeconds(), TimeUnit.SECONDS)
             .build()
@@ -572,8 +575,12 @@ class WaveClient {
             final key = assets.fingerprint()
             log.trace "Wave fingerprint: $key; assets: $assets"
             // get from cache or submit a new request
-            final handle = cache.get(key, () -> new Handle(sendRequest(assets),Instant.now()) )
-            return new ContainerInfo(assets.containerImage, handle.response.targetImage, key)
+            final resp = cache.get(key, () -> {
+                final ret = sendRequest(assets);
+                responses.put(key,new Handle(ret,Instant.now()));
+                return ret
+            })
+            return new ContainerInfo(assets.containerImage, resp.targetImage, key)
         }
         catch ( UncheckedExecutionException e ) {
             throw e.cause
@@ -633,7 +640,7 @@ class WaveClient {
     }
 
     boolean isContainerReady(String key) {
-        final handle = cache.getIfPresent(key)
+        final handle = responses.get(key)
         if( !handle )
             throw new IllegalStateException("Unable to find any container with key: $key")
         final resp = handle.response

diff --git a/plugins/nf-wave/src/test/io/seqera/wave/plugin/WaveClientTest.groovy b/plugins/nf-wave/src/test/io/seqera/wave/plugin/WaveClientTest.groovy
@@ -27,7 +27,6 @@ import java.nio.file.attribute.FileTime
 import java.time.Duration
 import java.time.Instant
 
-import com.google.common.cache.Cache
 import com.sun.net.httpserver.HttpExchange
 import com.sun.net.httpserver.HttpHandler
 import com.sun.net.httpserver.HttpServer
@@ -1303,18 +1302,18 @@ class WaveClientTest extends Specification {
     def 'should validate isContainerReady' () {
         given:
         def sess = Mock(Session) {getConfig() >> [wave: [build:[maxDuration: '500ms']]] }
-        def cache = Mock(Cache)
+        def cache = Mock(Map)
         and:
         def resp = Mock(SubmitContainerTokenResponse)
         def handle = new WaveClient.Handle(resp,Instant.now())
-        def wave = Spy(new WaveClient(session:sess, cache: cache))
+        def wave = Spy(new WaveClient(session:sess, responses: cache))
         boolean ready
 
         // container succeeded
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.requestId >> '12345'
         resp.succeeded >> true
@@ -1328,7 +1327,7 @@ class WaveClientTest extends Specification {
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.requestId >> '12345'
         resp.succeeded >> null
@@ -1342,7 +1341,7 @@ class WaveClientTest extends Specification {
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.requestId >> '12345'
         resp.succeeded >> false
@@ -1357,7 +1356,7 @@ class WaveClientTest extends Specification {
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.buildId >> 'bd-5678'
         resp.cached >> false
@@ -1371,7 +1370,7 @@ class WaveClientTest extends Specification {
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.requestId >> null
         resp.buildId >> 'bd-5678'
@@ -1386,7 +1385,7 @@ class WaveClientTest extends Specification {
         when:
         ready = wave.isContainerReady('xyz')
         then:
-        cache.getIfPresent('xyz') >> handle
+        cache.get('xyz') >> handle
         and:
         resp.requestId >> null
         resp.buildId >> 'bd-5678'