fix: surface transfer errors when caching blobs (COMP-2142)

src/main/groovy/io/seqera/wave/proxy/ProxyClient.groovy

@@ -386,6 +386,16 @@ class ProxyClient {
         final result = new ArrayList(20)
         result.add('curl')
         result.add('-s')
+        // fail with a non-zero exit status on HTTP errors, instead of writing
+        // the error body (or nothing at all) to stdout and exiting zero
+        result.add('-f')
+        // retry transient failures; note '--retry-all-errors' is deliberately not added because
+        // it would retry a response whose body was already partially piped to the upload command,
+        // duplicating those bytes in the uploaded object
+        result.add('--retry'); result.add(String.valueOf(httpConfig.retryAttempts))
+        // bound the connection setup only; an upstream that connects and then stalls mid-body is
+        // still capped by the transfer job timeout, not by this option
+        result.add('--connect-timeout'); result.add(String.valueOf(httpConfig.connectTimeout.toSeconds()))
         result.add('-X'); result.add('GET')
         //  copy headers
         for( Map.Entry<String,String> entry : headers )  {

src/main/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImpl.groovy

@@ -42,6 +42,7 @@ import jakarta.inject.Inject
 import jakarta.inject.Named
 import jakarta.inject.Singleton
 import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.DeleteObjectRequest
 import software.amazon.awssdk.services.s3.model.HeadObjectRequest
 import software.amazon.awssdk.services.s3.model.S3Exception
 /**
@@ -113,6 +114,16 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler<BlobEntry> {
     }
 
     protected boolean blobExists(String blobLocation) {
+        return blobSize(blobLocation) != null
+    }
+
+    /**
+     * Retrieve the size of the object stored in the object storage cache
+     *
+     * @param blobLocation The object storage path e.g. {@code s3://bucket-name/some/path}
+     * @return The size in bytes of the stored object or {@code null} if the object does not exist
+     */
+    protected Long blobSize(String blobLocation) {
         try {
             final object = BucketTokenizer.from(blobLocation)
             final request = HeadObjectRequest
@@ -121,14 +132,42 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler<BlobEntry> {
                     .key(object.key)
                     .build() as HeadObjectRequest
             // Execute the request
-            s3Client.headObject(request)
-            return true
+            return s3Client.headObject(request).contentLength()
         }
         catch (S3Exception e) {
             if (e.statusCode() != 404) {
                 log.error "Unexpected response=${e.statusCode()} checking existence for object=${blobLocation} - cause: ${e.message}"
             }
-            return false
+            return null
+        }
+        catch (Exception e) {
+            log.error "Unexpected error checking existence for object=${blobLocation} - cause: ${e.message}", e
+            return null
+        }
+    }
+
+    /**
+     * Delete the object stored in the object storage cache. This is used to remove an invalid
+     * object left behind by a failed transfer, so that it is not reported as a valid cache entry
+     * by {@link #blobExists(java.lang.String)} on a subsequent request.
+     *
+     * @param blobLocation The object storage path e.g. {@code s3://bucket-name/some/path}
+     */
+    protected void deleteBlob(String blobLocation) {
+        try {
+            final object = BucketTokenizer.from(blobLocation)
+            final request = DeleteObjectRequest
+                    .builder()
+                    .bucket(object.bucket)
+                    .key(object.key)
+                    .build() as DeleteObjectRequest
+            s3Client.deleteObject(request)
+            log.debug "Deleted invalid blob cache object=${blobLocation}"
+        }
+        catch (Exception e) {
+            // the entry is marked as errored regardless, therefore a failure to clean up the
+            // object must not propagate out of the job completion handling
+            log.error "Unable to delete invalid blob cache object=${blobLocation} - cause: ${e.message}", e
         }
     }
 
@@ -172,10 +211,16 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler<BlobEntry> {
         final curl = proxyService.curl(route, info.headers)
         final s5cmd = s5cmd(route, info)
 
+        // 'set -o pipefail' makes the pipeline fail when the curl command fails; without it
+        // the exit status is the one of the last command i.e. s5cmd, which happily exits zero
+        // after uploading an empty stream when curl failed to download the blob.
+        // Note 'bash' is required instead of 'sh': in the s5cmd image /bin/sh is 'dash', which
+        // does not implement 'pipefail' and aborts the whole script with "Illegal option -o
+        // pipefail" before the pipeline is even executed
         final command = List.of(
-                'sh',
+                'bash',
                 '-c',
-                Escape.cli(curl) + ' | ' + Escape.cli(s5cmd) )
+                'set -o pipefail; ' + Escape.cli(curl) + ' | ' + Escape.cli(s5cmd) )
 
         log.trace "== Blob cache transfer command: ${command.join(' ')}"
         return command
@@ -274,12 +319,50 @@ class BlobCacheServiceImpl implements BlobCacheService, JobHandler<BlobEntry> {
         blobStore.getBlob(job.entryKey)
     }
 
+    /**
+     * Verify the object uploaded in the object storage cache matches the expected size, to prevent
+     * an empty or truncated transfer from being cached and served as if it were valid. Note the
+     * transfer job and the Wave proxy issue two separate requests to the upstream registry, therefore
+     * a job can report success even when no byte was actually downloaded.
+     *
+     * @param entry The {@link BlobEntry} associated with the completed transfer
+     * @return An error message describing the mismatch or {@code null} when the transfer is valid
+     */
+    protected String checkTransferredSize(BlobEntry entry) {
+        final uploaded = blobSize(entry.objectUri)
+        if( uploaded == null )
+            return "Blob cache object '${entry.objectUri}' was not uploaded to the object storage".toString()
+        // the upstream content length is not always provided, in that case only an
+        // empty object can be detected as invalid
+        final expected = entry.contentLength
+        if( expected == null ) {
+            return uploaded == 0
+                    ? "Blob cache object '${entry.objectUri}' is empty".toString()
+                    : null
+        }
+        if( uploaded != expected )
+            return "Blob cache object '${entry.objectUri}' size does not match the expected content length - uploaded: ${uploaded}; expected: ${expected}".toString()
+        return null
+    }
+
     @Override
     void onJobCompletion(JobSpec job, BlobEntry entry, JobState state) {
+        // the transfer job exit status only tells the command pipeline exited zero, it does not
+        // guarantee the blob bytes made it to the object storage - validate before completing
+        final error = state.succeeded()
+                ? checkTransferredSize(entry)
+                : state.stdout
+        if( error ) {
+            log.warn "== Blob cache transfer invalid for object '${entry.objectUri}'; operation=${job.operationName} - cause: ${error}"
+            // remove whatever the failed transfer left behind: a failing pipeline can still have
+            // uploaded an empty or truncated object, and blobExists() would then report it as a
+            // valid cache entry on the next request, making the bad transfer sticky
+            deleteBlob(entry.objectUri)
+        }
         // update the entry status
-        final result = state.succeeded()
+        final result = !error
                 ? entry.completed(state.exitCode, state.stdout)
-                : entry.errored(state.stdout)
+                : entry.errored(error)
         blobStore.storeBlob(entry.getKey(), result)
         log.debug "== Blob cache completed for object '${entry.objectUri}'; operation=${job.operationName}; status=${result.exitStatus}; duration=${result.duration()}"
     }

src/test/groovy/io/seqera/wave/proxy/ProxyClientTest.groovy

@@ -289,12 +289,17 @@ class ProxyClientTest extends Specification {
         then:
         cli[0] == 'curl'
         cli[1] == '-s'
-        cli[2] == '-X'
-        cli[3] == 'GET'
-        cli[4] == '-H'
-        cli[5] == 'content-type: application/octet-stream'
-        cli[6] == '-H'
-        cli[7] =~ /Authorization: Bearer.*/
-        cli[8] == 'https://registry-1.docker.io/v2/library/hello-world/manifests/sha256:aa0cc8055b82dc2509bed2e19b275c8f463506616377219d9642221ab53cf9fe'
+        cli[2] == '-f'
+        cli[3] == '--retry'
+        cli[4] == String.valueOf(httpConfig.retryAttempts)
+        cli[5] == '--connect-timeout'
+        cli[6] == String.valueOf(httpConfig.connectTimeout.toSeconds())
+        cli[7] == '-X'
+        cli[8] == 'GET'
+        cli[9] == '-H'
+        cli[10] == 'content-type: application/octet-stream'
+        cli[11] == '-H'
+        cli[12] =~ /Authorization: Bearer.*/
+        cli[13] == 'https://registry-1.docker.io/v2/library/hello-world/manifests/sha256:aa0cc8055b82dc2509bed2e19b275c8f463506616377219d9642221ab53cf9fe'
     }
 }

src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest.groovy

@@ -19,11 +19,17 @@ package io.seqera.wave.service.blob.impl
 
 import spock.lang.Specification
 
+import java.time.Duration
+import java.time.Instant
+
 import io.seqera.wave.configuration.BlobCacheConfig
 import io.seqera.wave.core.RegistryProxyService
 import io.seqera.wave.core.RoutePath
 import io.seqera.wave.model.ContainerCoordinates
 import io.seqera.wave.service.blob.BlobEntry
+import io.seqera.wave.service.blob.BlobStateStore
+import io.seqera.wave.service.job.JobSpec
+import io.seqera.wave.service.job.JobState
 import io.seqera.wave.test.AwsS3TestContainer
 
 /**
@@ -77,10 +83,97 @@ class BlobCacheServiceImplTest extends Specification implements AwsS3TestContain
         proxyService.curl(route, [foo:'one']) >> ['curl', '-X', 'GET', 'http://foo']
         and:
         result == [
-                'sh',
+                'bash',
                 '-c',
-                "curl -X GET 'http://foo' | s5cmd --json pipe --content-type something 's3://store/blobs/docker.io/v2/library/ubuntu/manifests/sha256:aabbcc'"
+                "set -o pipefail; curl -X GET 'http://foo' | s5cmd --json pipe --content-type something 's3://store/blobs/docker.io/v2/library/ubuntu/manifests/sha256:aabbcc'"
         ]
     }
 
+    private static BlobEntry blobEntry(String objectUri, Long contentLength) {
+        final response = contentLength!=null
+                ? ['Content-Length': [String.valueOf(contentLength)]]
+                : [:] as Map<String,List<String>>
+        return BlobEntry.create('http://foo', objectUri, [:], response)
+    }
+
+    def 'should validate the transferred blob size' () {
+        given:
+        def OBJECT = 's3://store/blobs/foo'
+        def service = Spy(BlobCacheServiceImpl)
+        def entry = blobEntry(OBJECT, LENGTH)
+
+        when:
+        def result = service.checkTransferredSize(entry)
+        then:
+        1 * service.blobSize(OBJECT) >> UPLOADED
+        and:
+        (result != null) == ERROR
+
+        where:
+        LENGTH  | UPLOADED  | ERROR
+        100L    | 100L      | false
+        100L    | 0L        | true
+        100L    | 50L       | true
+        100L    | null      | true
+        null    | 100L      | false
+        null    | 0L        | true
+        null    | null      | true
+    }
+
+    def 'should error the blob entry when the uploaded object is empty' () {
+        given:
+        def OBJECT = 's3://store/blobs/foo'
+        def blobStore = Mock(BlobStateStore)
+        def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore))
+        def entry = blobEntry(OBJECT, 100L)
+        def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1))
+
+        when:
+        service.onJobCompletion(job, entry, JobState.succeeded('some logs'))
+        then:
+        1 * service.blobSize(OBJECT) >> 0L
+        and:
+        // the invalid object is removed so it is not served as a valid cache entry
+        1 * service.deleteBlob(OBJECT) >> null
+        and:
+        1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.ERRORED && !it.succeeded() })
+    }
+
+    def 'should complete the blob entry when the uploaded object matches the content length' () {
+        given:
+        def OBJECT = 's3://store/blobs/foo'
+        def blobStore = Mock(BlobStateStore)
+        def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore))
+        def entry = blobEntry(OBJECT, 100L)
+        def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1))
+
+        when:
+        service.onJobCompletion(job, entry, JobState.succeeded('some logs'))
+        then:
+        1 * service.blobSize(OBJECT) >> 100L
+        and:
+        0 * service.deleteBlob(_) >> null
+        and:
+        1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.COMPLETED && it.succeeded() })
+    }
+
+    def 'should not validate the object when the transfer job failed' () {
+        given:
+        def OBJECT = 's3://store/blobs/foo'
+        def blobStore = Mock(BlobStateStore)
+        def service = Spy(new BlobCacheServiceImpl(blobStore: blobStore))
+        def entry = blobEntry(OBJECT, 100L)
+        def job = JobSpec.transfer('1', 'operation-1', Instant.now(), Duration.ofMinutes(1))
+
+        when:
+        service.onJobCompletion(job, entry, JobState.failed(1, 'curl failed'))
+        then:
+        0 * service.blobSize(_) >> null
+        and:
+        // a failed pipeline can still have uploaded a partial object, remove it
+        1 * service.deleteBlob(OBJECT) >> null
+        and:
+        1 * blobStore.storeBlob(OBJECT, { BlobEntry it -> it.state==BlobEntry.State.ERRORED && it.logs=='curl failed' })
+    }
+
 }

src/test/groovy/io/seqera/wave/service/blob/impl/BlobCacheServiceImplTest2.groovy

@@ -132,19 +132,25 @@ class BlobCacheServiceImplTest2 extends Specification implements AwsS3TestContai
         def store = Mock(BlobStateStore)
         def blob = BlobEntry.create('http://some/blob','s3://some/blob', [:], [:])
         def config = new BlobCacheConfig(statusDelay: Duration.ofSeconds(2))
-        def service = new BlobCacheServiceImpl(blobStore: store, blobConfig: config)
+        def service = Spy(new BlobCacheServiceImpl(blobStore: store, blobConfig: config))
         def job =  JobSpec.transfer('job-id', 'foo', Instant.now(), Duration.ofMinutes(1))
         def failed = new JobState(JobState.Status.FAILED, 1, 'Oops')
         def ok = new JobState(JobState.Status.SUCCEEDED, 0, 'done')
 
         when:
         service.onJobCompletion(job, blob, failed)
         then:
+        // a failed pipeline can still have uploaded a partial object, remove it
+        1 * service.deleteBlob('s3://some/blob') >> null
+        and:
         1 * store.storeBlob(blob.getKey(), _ as BlobEntry) >> { id, BlobEntry info -> info.state==BlobEntry.State.ERRORED }
 
         when:
         service.onJobCompletion(job, blob, ok)
         then:
+        // the transferred object is validated before the entry is marked as completed
+        1 * service.blobSize('s3://some/blob') >> 100L
+        and:
         1 * store.storeBlob(blob.getKey(), _ as BlobEntry) >> { id, BlobEntry info -> info.state==BlobEntry.State.COMPLETED }
 
     }