diff --git a/conf/reflect-config.json b/conf/reflect-config.json index f7e062552..6b26a38dd 100644 --- a/conf/reflect-config.json +++ b/conf/reflect-config.json @@ -3672,6 +3672,18 @@ "queryAllDeclaredConstructors":true, "methods":[{"name":"$values","parameterTypes":[] }, {"name":"","parameterTypes":[] }, {"name":"","parameterTypes":["java.lang.String","int","java.lang.String"] }, {"name":"fromValue","parameterTypes":["java.lang.String"] }, {"name":"getValue","parameterTypes":[] }, {"name":"toString","parameterTypes":[] }, {"name":"valueOf","parameterTypes":["java.lang.String"] }, {"name":"values","parameterTypes":[] }] }, +{ + "name":"io.seqera.tower.model.DataLinkRefreshMultiPartUploadRequest", + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[{"name":"","parameterTypes":[] }, {"name":"addPartNumbersItem","parameterTypes":["java.lang.Integer"] }, {"name":"contentLength","parameterTypes":["java.lang.Long"] }, {"name":"contentType","parameterTypes":["java.lang.String"] }, {"name":"dirPath","parameterTypes":["java.lang.String"] }, {"name":"equals","parameterTypes":["java.lang.Object"] }, {"name":"fileName","parameterTypes":["java.lang.String"] }, {"name":"getContentLength","parameterTypes":[] }, {"name":"getContentType","parameterTypes":[] }, {"name":"getDirPath","parameterTypes":[] }, {"name":"getFileName","parameterTypes":[] }, {"name":"getPartNumbers","parameterTypes":[] }, {"name":"getUploadId","parameterTypes":[] }, {"name":"hashCode","parameterTypes":[] }, {"name":"partNumbers","parameterTypes":["java.util.List"] }, {"name":"setContentLength","parameterTypes":["java.lang.Long"] }, {"name":"setContentType","parameterTypes":["java.lang.String"] }, {"name":"setDirPath","parameterTypes":["java.lang.String"] }, {"name":"setFileName","parameterTypes":["java.lang.String"] }, {"name":"setPartNumbers","parameterTypes":["java.util.List"] }, {"name":"setUploadId","parameterTypes":["java.lang.String"] }, {"name":"toIndentedString","parameterTypes":["java.lang.Object"] }, {"name":"toString","parameterTypes":[] }, {"name":"uploadId","parameterTypes":["java.lang.String"] }] +}, +{ + "name":"io.seqera.tower.model.DataLinkRefreshMultiPartUploadResponse", + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[{"name":"","parameterTypes":[] }, {"name":"addUploadUrlsItem","parameterTypes":["io.seqera.tower.model.PartUploadUrl"] }, {"name":"equals","parameterTypes":["java.lang.Object"] }, {"name":"getUploadId","parameterTypes":[] }, {"name":"getUploadUrls","parameterTypes":[] }, {"name":"hashCode","parameterTypes":[] }, {"name":"setUploadId","parameterTypes":["java.lang.String"] }, {"name":"setUploadUrls","parameterTypes":["java.util.List"] }, {"name":"toIndentedString","parameterTypes":["java.lang.Object"] }, {"name":"toString","parameterTypes":[] }, {"name":"uploadId","parameterTypes":["java.lang.String"] }, {"name":"uploadUrls","parameterTypes":["java.util.List"] }] +}, { "name":"io.seqera.tower.model.DataLinkResponse", "allDeclaredFields":true, @@ -5029,6 +5041,12 @@ "queryAllDeclaredConstructors":true, "methods":[{"name":"","parameterTypes":[] }, {"name":"equals","parameterTypes":["java.lang.Object"] }, {"name":"getMaxCustomRolesPerOrg","parameterTypes":[] }, {"name":"getMaxDataStudiosRunning","parameterTypes":[] }, {"name":"getMaxDatasetsPerWorkspace","parameterTypes":[] }, {"name":"getMaxFusionThroughputBytes","parameterTypes":[] }, {"name":"getMaxLabelsPerWorkspace","parameterTypes":[] }, {"name":"getMaxMembers","parameterTypes":[] }, {"name":"getMaxParticipantsPerWorkspace","parameterTypes":[] }, {"name":"getMaxPipelinesPerWorkspace","parameterTypes":[] }, {"name":"getMaxRunHistory","parameterTypes":[] }, {"name":"getMaxRuns","parameterTypes":[] }, {"name":"getMaxSeqeraComputeComputeEnvs","parameterTypes":[] }, {"name":"getMaxTeams","parameterTypes":[] }, {"name":"getMaxVersionsPerDataset","parameterTypes":[] }, {"name":"getMaxWorkspaces","parameterTypes":[] }, {"name":"hashCode","parameterTypes":[] }, {"name":"maxCustomRolesPerOrg","parameterTypes":["java.lang.Long"] }, {"name":"maxDataStudiosRunning","parameterTypes":["java.lang.Integer"] }, {"name":"maxDataStudiosRunning","parameterTypes":["java.lang.Long"] }, {"name":"maxDatasetsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"maxDatasetsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"maxFusionThroughputBytes","parameterTypes":["java.lang.Long"] }, {"name":"maxLabelsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"maxLabelsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"maxMembers","parameterTypes":["java.lang.Integer"] }, {"name":"maxMembers","parameterTypes":["java.lang.Long"] }, {"name":"maxParticipantsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"maxParticipantsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"maxPipelinesPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"maxPipelinesPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"maxRunHistory","parameterTypes":["java.lang.Integer"] }, {"name":"maxRunHistory","parameterTypes":["java.lang.Long"] }, {"name":"maxRuns","parameterTypes":["java.lang.Integer"] }, {"name":"maxRuns","parameterTypes":["java.lang.Long"] }, {"name":"maxSeqeraComputeComputeEnvs","parameterTypes":["java.lang.Integer"] }, {"name":"maxSeqeraComputeComputeEnvs","parameterTypes":["java.lang.Long"] }, {"name":"maxTeams","parameterTypes":["java.lang.Integer"] }, {"name":"maxTeams","parameterTypes":["java.lang.Long"] }, {"name":"maxVersionsPerDataset","parameterTypes":["java.lang.Integer"] }, {"name":"maxVersionsPerDataset","parameterTypes":["java.lang.Long"] }, {"name":"maxWorkspaces","parameterTypes":["java.lang.Integer"] }, {"name":"maxWorkspaces","parameterTypes":["java.lang.Long"] }, {"name":"setMaxCustomRolesPerOrg","parameterTypes":["java.lang.Long"] }, {"name":"setMaxDataStudiosRunning","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxDataStudiosRunning","parameterTypes":["java.lang.Long"] }, {"name":"setMaxDatasetsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxDatasetsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"setMaxFusionThroughputBytes","parameterTypes":["java.lang.Long"] }, {"name":"setMaxLabelsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxLabelsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"setMaxMembers","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxMembers","parameterTypes":["java.lang.Long"] }, {"name":"setMaxParticipantsPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxParticipantsPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"setMaxPipelinesPerWorkspace","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxPipelinesPerWorkspace","parameterTypes":["java.lang.Long"] }, {"name":"setMaxRunHistory","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxRunHistory","parameterTypes":["java.lang.Long"] }, {"name":"setMaxRuns","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxRuns","parameterTypes":["java.lang.Long"] }, {"name":"setMaxSeqeraComputeComputeEnvs","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxSeqeraComputeComputeEnvs","parameterTypes":["java.lang.Long"] }, {"name":"setMaxTeams","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxTeams","parameterTypes":["java.lang.Long"] }, {"name":"setMaxVersionsPerDataset","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxVersionsPerDataset","parameterTypes":["java.lang.Long"] }, {"name":"setMaxWorkspaces","parameterTypes":["java.lang.Integer"] }, {"name":"setMaxWorkspaces","parameterTypes":["java.lang.Long"] }, {"name":"toIndentedString","parameterTypes":["java.lang.Object"] }, {"name":"toString","parameterTypes":[] }] }, +{ + "name":"io.seqera.tower.model.PartUploadUrl", + "queryAllDeclaredMethods":true, + "queryAllDeclaredConstructors":true, + "methods":[{"name":"","parameterTypes":[] }, {"name":"equals","parameterTypes":["java.lang.Object"] }, {"name":"getPartNumber","parameterTypes":[] }, {"name":"getUrl","parameterTypes":[] }, {"name":"hashCode","parameterTypes":[] }, {"name":"partNumber","parameterTypes":["java.lang.Integer"] }, {"name":"setPartNumber","parameterTypes":["java.lang.Integer"] }, {"name":"setUrl","parameterTypes":["java.lang.String"] }, {"name":"toIndentedString","parameterTypes":["java.lang.Object"] }, {"name":"toString","parameterTypes":[] }, {"name":"url","parameterTypes":["java.lang.String"] }] +}, { "name":"io.seqera.tower.model.ParticipantImage", "queryAllDeclaredMethods":true, diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 2dcbab980..589167037 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -13,7 +13,7 @@ mockserverVersion = "5.15.0" picocliVersion = "4.6.3" shadowVersion = "9.4.1" slf4jVersion = "2.0.17" -towerJavaSdkVersion = "1.190.0" +towerJavaSdkVersion = "1.192.0" xzVersion = "1.10" [libraries] diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/UploadCmd.java b/src/main/java/io/seqera/tower/cli/commands/data/links/UploadCmd.java index 90b55036b..669b479dd 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/UploadCmd.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/UploadCmd.java @@ -151,9 +151,9 @@ private CloudProviderUploader createUploadStrategy(DataLinkProvider provider, St case AWS: return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); case GOOGLE: - return new GoogleUploader(); + return new GoogleUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); case AZURE: - return new AzureUploader(); + return new AzureUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); case SEQERACOMPUTE: // Seqera Compute uses S3-compatible uploads, same as AWS return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AbstractProviderUploader.java b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AbstractProviderUploader.java index 952536244..23efc7f92 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AbstractProviderUploader.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AbstractProviderUploader.java @@ -16,15 +16,67 @@ package io.seqera.tower.cli.commands.data.links.upload; +import io.seqera.tower.ApiException; +import io.seqera.tower.api.DataLinksApi; +import io.seqera.tower.cli.exceptions.TowerRuntimeException; +import io.seqera.tower.cli.utils.progress.ProgressTracker; +import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; +import io.seqera.tower.model.DataLinkRefreshMultiPartUploadRequest; +import io.seqera.tower.model.DataLinkRefreshMultiPartUploadResponse; +import io.seqera.tower.model.PartUploadUrl; + import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; import java.io.UncheckedIOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; public abstract class AbstractProviderUploader implements CloudProviderUploader { static final Integer MULTI_UPLOAD_PART_SIZE_IN_BYTES = 250 * 1024 * 1024; // 250 MB + /** Max attempts per part, covering both refresh-on-expiry and transient-error retries. */ + static final int MAX_PART_ATTEMPTS = 5; + + /** Number of upcoming parts whose URLs are refreshed in a single call when a credential expires. */ + static final int REFRESH_WINDOW = 100; + + private static final long BACKOFF_BASE_MILLIS = 500L; + private static final long BACKOFF_MAX_MILLIS = 10_000L; + + // Provider error codes (from the S3/Azure XML ... body) that mean the + // signing credentials have expired and the URL must be refreshed before retrying. + private static final Pattern ERROR_CODE = Pattern.compile("(.*?)", Pattern.DOTALL); + + protected final String id; + protected final String credId; + protected final Long wspId; + protected final String outputDir; + protected final String relativeKey; + protected final DataLinksApi dataLinksApi; + + protected AbstractProviderUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { + this.id = id; + this.credId = credId; + this.wspId = wspId; + this.outputDir = outputDir; + this.relativeKey = relativeKey; + this.dataLinksApi = dataLinksApi; + } + + protected enum UploadErrorType { EXPIRY, TRANSIENT, HARD_FAIL } + protected byte[] getChunk(File file, int index) { try (RandomAccessFile raf = new RandomAccessFile(file, "r")) { long start = (long) index * MULTI_UPLOAD_PART_SIZE_IN_BYTES; @@ -40,4 +92,194 @@ protected byte[] getChunk(File file, int index) { throw new UncheckedIOException(e); } } -} \ No newline at end of file + + protected int totalParts(long contentLength) { + if (contentLength <= 0) { + return 1; + } + return (int) Math.ceil((double) contentLength / MULTI_UPLOAD_PART_SIZE_IN_BYTES); + } + + /** + * Uploads a single part. + * When {@code refreshable}, an expiry-class error causes the presigned URL (and a forward window of upcoming parts) + * to be refreshed via the Platform and the part retried; + * otherwise an expiry is terminal (providers such as Azure/GCS cannot re-mint URLs for an in-progress upload). + * + * In both modes a transient error retries the same URL with exponential backoff. + * On a hard failure, or once the attempt budget is exhausted, the error is propagated so the + * caller can finalize/abort the upload. + * + * @param partUrls mutable part-number -> URL map, seeded from the initial upload response and + * updated in place as URLs are refreshed + * @param uploadId the in-progress multi-part upload id (may be {@code null}, e.g. for Azure) + * @param successStatus the HTTP status that indicates a successful part upload (200 for S3, 201 for Azure) + * @param refreshable whether the provider supports refreshing URLs on expiry (S3 only) + * @return the successful HTTP response (headers/body available to the caller, e.g. for the S3 ETag) + */ + protected HttpResponse uploadPartWithRetry(HttpClient client, Map partUrls, int partNumber, + byte[] chunk, ProgressTracker tracker, String uploadId, long contentLength, int successStatus, boolean refreshable) + throws ApiException, IOException, InterruptedException { + + long baseline = tracker.snapshot(); + for (int attempt = 1; attempt <= MAX_PART_ATTEMPTS; attempt++) { + String url = partUrls.get(partNumber); + if (url == null) { + if (!refreshable) { + throw new TowerRuntimeException("Failed to obtain an upload URL for part " + partNumber); + } + partUrls.putAll(refreshUrls(uploadId, contentLength, refreshWindow(partNumber, contentLength))); + url = partUrls.get(partNumber); + if (url == null) { + throw new TowerRuntimeException("Failed to obtain an upload URL for part " + partNumber); + } + } + + final String partUrl = url; + HttpResponse response = sendWithRetryOnTransientError(client, tracker, baseline, + () -> HttpRequest.newBuilder() + .uri(URI.create(partUrl)) + .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) + .build()); + + if (response.statusCode() == successStatus) { + return response; + } + + // Non-success: discard the bytes this attempt reported before deciding what to do. + tracker.restore(baseline); + UploadErrorType type = classify(response.statusCode(), response.body()); + if (type == UploadErrorType.EXPIRY && refreshable && attempt < MAX_PART_ATTEMPTS) { + // Re-sign this part and a forward window of upcoming parts in a single call, then retry. + partUrls.putAll(refreshUrls(uploadId, contentLength, refreshWindow(partNumber, contentLength))); + continue; + } + throw new IOException("Failed to upload part " + partNumber + ": HTTP " + response.statusCode() + + (isNotEmpty(response.body()) ? ", Message: " + response.body() : "")); + } + throw new IOException("Failed to upload part " + partNumber + " after " + MAX_PART_ATTEMPTS + " attempts"); + } + + /** + * Sends a request with transient-failure recovery shared by all providers: network errors and transient + * HTTP responses (5xx / throttling, per {@link #classify}) are retried by re-sending the same request + * with exponential backoff, up to {@link #MAX_PART_ATTEMPTS}. Returns the first response that is not a + * transient failure — the caller decides whether that means success, expiry, resume, or hard failure. + * Throws the last network error if the budget is exhausted by network failures. + * + * @param baseline the tracker snapshot taken before the first attempt (see {@link ProgressTracker#snapshot()}) + * @param request factory invoked once per attempt to build a fresh request (and body publisher) + */ + protected HttpResponse sendWithRetryOnTransientError(HttpClient client, ProgressTracker tracker, long baseline, + Supplier request) throws IOException, InterruptedException { + + IOException lastError = null; + for (int attempt = 1; attempt <= MAX_PART_ATTEMPTS; attempt++) { + tracker.restore(baseline); + try { + HttpResponse response = client.send(request.get(), HttpResponse.BodyHandlers.ofString()); + if (classify(response.statusCode(), response.body()) == UploadErrorType.TRANSIENT && attempt < MAX_PART_ATTEMPTS) { + backoff(attempt); + continue; + } + return response; + } catch (IOException e) { + // Network-level failure (connection reset, socket timeout, ...) — treat as transient. + lastError = e; + if (attempt == MAX_PART_ATTEMPTS) { + break; + } + backoff(attempt); + } + } + throw lastError != null ? lastError : new IOException("Request failed after " + MAX_PART_ATTEMPTS + " attempts"); + } + + private List refreshWindow(int partNumber, long contentLength) { + int total = totalParts(contentLength); + List parts = new ArrayList<>(); + for (int p = partNumber; p < partNumber + REFRESH_WINDOW && p <= total; p++) { + parts.add(p); + } + return parts; + } + + /** + * Requests freshly-signed upload URLs for the given part numbers from the Platform. + * A 404 means the Platform predates the refresh endpoint — surfaced as a clear, actionable message. + */ + protected Map refreshUrls(String uploadId, long contentLength, List partNumbers) throws ApiException { + DataLinkRefreshMultiPartUploadRequest request = new DataLinkRefreshMultiPartUploadRequest(); + request.setUploadId(uploadId); + request.setFileName(relativeKey); + request.setContentLength(contentLength); + request.setPartNumbers(partNumbers); + + DataLinkRefreshMultiPartUploadResponse response; + try { + response = outputDir != null + ? dataLinksApi.refreshDataLinkUploadUrlWithPath(id, outputDir, request, credId, wspId, null) + : dataLinksApi.refreshDataLinkUploadUrl(id, request, credId, wspId, null); + } catch (ApiException e) { + if (e.getCode() == 404) { + throw new TowerRuntimeException("Token refresh is not supported for this Platform version."); + } + throw e; + } + + Map map = new HashMap<>(); + if (response.getUploadUrls() != null) { + for (PartUploadUrl part : response.getUploadUrls()) { + map.put(part.getPartNumber(), part.getUrl()); + } + } + return map; + } + + /** + * Classifies a failed part upload from its HTTP status and provider error body: + *
    + *
  • EXPIRY — the signing credentials expired; the URL must be refreshed before retrying
    • + *
  • TRANSIENT — a temporary error (5xx / throttling / network); retry the same URL with backoff
    • + *
  • HARD_FAIL — anything else; do not retry
    • + *
+ */ + protected UploadErrorType classify(int statusCode, String body) { + String code = extractErrorCode(body); + if (code != null) { + switch (code) { + case "ExpiredToken": // S3 + case "SignatureDoesNotMatch": // S3 + case "RequestTimeTooSkewed": // S3 + return UploadErrorType.EXPIRY; + case "InternalError": // S3 + case "SlowDown": // S3 throttling + return UploadErrorType.TRANSIENT; + default: + // fall through to status-based classification + } + } + if (statusCode == 500 || statusCode == 502 || statusCode == 503 || statusCode == 504) { + return UploadErrorType.TRANSIENT; + } + return UploadErrorType.HARD_FAIL; + } + + protected static String extractErrorCode(String body) { + if (!isNotEmpty(body)) { + return null; + } + Matcher m = ERROR_CODE.matcher(body); + return m.find() ? m.group(1).trim() : null; + } + + protected void backoff(int attempt) throws InterruptedException { + long base = BACKOFF_BASE_MILLIS * (1L << (attempt - 1)); + long jitter = ThreadLocalRandom.current().nextLong(BACKOFF_BASE_MILLIS / 2); + Thread.sleep(Math.min(base + jitter, BACKOFF_MAX_MILLIS)); + } + + private static boolean isNotEmpty(String s) { + return s != null && !s.isEmpty(); + } +} diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AwsUploader.java b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AwsUploader.java index 7819d3d52..6b2a74bec 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AwsUploader.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AwsUploader.java @@ -20,74 +20,57 @@ import io.seqera.tower.api.DataLinksApi; import io.seqera.tower.cli.exceptions.TowerRuntimeException; import io.seqera.tower.cli.utils.progress.ProgressTracker; -import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; import io.seqera.tower.model.DataLinkFinishMultiPartUploadRequest; import io.seqera.tower.model.DataLinkMultiPartUploadResponse; import io.seqera.tower.model.UploadEtag; import java.io.File; -import java.io.IOException; -import java.net.URI; import java.net.http.HttpClient; -import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; public class AwsUploader extends AbstractProviderUploader { - private final String id; - private final String credId; - private final Long wspId; - private final String outputDir; - private final String relativeKey; - private final DataLinksApi dataLinksApi; - public AwsUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { - this.id = id; - this.credId = credId; - this.wspId = wspId; - this.outputDir = outputDir; - this.relativeKey = relativeKey; - this.dataLinksApi = dataLinksApi; + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi); } @Override public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, ProgressTracker tracker) throws ApiException { - int index = 0; boolean withError = false; List tags = new ArrayList<>(); + String uploadId = urlResponse.getUploadId(); + long contentLength = file.length(); + + // Seed the part-number -> URL map from the initially generated URLs (positional: index+1 == partNumber). + // The map is updated in place by uploadPartWithRetry whenever URLs are refreshed on expiry. + List initialUrls = urlResponse.getUploadUrls(); + Map partUrls = new HashMap<>(); + for (int i = 0; i < initialUrls.size(); i++) { + partUrls.put(i + 1, initialUrls.get(i)); + } + int totalParts = initialUrls.size(); try (HttpClient client = HttpClient.newHttpClient()) { - for (String url : urlResponse.getUploadUrls()) { - byte[] chunk = getChunk(file, index); + for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + byte[] chunk = getChunk(file, partNumber - 1); - HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create(url)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) - .build(); - - HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); - - if (response.statusCode() != 200) { - withError = true; - throw new IOException("Failed to upload file: HTTP " + response.statusCode() +", Message: " + response.body()); - } + HttpResponse response = uploadPartWithRetry(client, partUrls, partNumber, chunk, tracker, uploadId, contentLength, 200, true); Optional etag = response.headers().firstValue("ETag"); - if (etag.isPresent()) { UploadEtag uploadEtag = new UploadEtag(); uploadEtag.eTag(etag.get()); - uploadEtag.partNumber(index+1); + uploadEtag.partNumber(partNumber); tags.add(uploadEtag); - } - else { + } else { throw new TowerRuntimeException("Failed to upload file: Possible CORS issue"); } - index++; } } catch (Exception e) { withError = true; @@ -114,6 +97,6 @@ private void finalizeUpload(DataLinkMultiPartUploadResponse urlResponse, boolean @Override public void abortUpload(DataLinkMultiPartUploadResponse urlResponse) throws ApiException { - finalizeUpload(urlResponse, true, Collections.emptyList()); + finalizeUpload(urlResponse, true, Collections.emptyList()); } -} \ No newline at end of file +} diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AzureUploader.java b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AzureUploader.java index 1df8b897e..818fa4679 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AzureUploader.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/AzureUploader.java @@ -16,9 +16,9 @@ package io.seqera.tower.cli.commands.data.links.upload; +import io.seqera.tower.api.DataLinksApi; import io.seqera.tower.cli.exceptions.TowerRuntimeException; import io.seqera.tower.cli.utils.progress.ProgressTracker; -import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; import io.seqera.tower.model.DataLinkMultiPartUploadResponse; import java.io.File; @@ -27,37 +27,42 @@ import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; public class AzureUploader extends AbstractProviderUploader { + public AzureUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi); + } + @Override public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, ProgressTracker tracker) { - List urls = urlResponse.getUploadUrls(); + long contentLength = file.length(); + + List initialUrls = urlResponse.getUploadUrls(); + Map partUrls = new HashMap<>(); + for (int i = 0; i < initialUrls.size(); i++) { + partUrls.put(i + 1, initialUrls.get(i)); + } + int totalParts = initialUrls.size(); HttpClient client = HttpClient.newHttpClient(); try { - // Upload chunks - for (int i = 0; i < urls.size(); i++) { - String url = urls.get(i); - byte[] chunk = getChunk(file, i); - - HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create(url)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) - .build(); - - HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); - - if (response.statusCode() != 201) { - // Abort the upload before throwing the exception - throw new IOException("Failed to upload chunk: HTTP " + response.statusCode()); - } + for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + byte[] chunk = getChunk(file, partNumber - 1); + uploadPartWithRetry(client, partUrls, partNumber, chunk, tracker, null, contentLength, 201, false); } - // Finalize the upload by sending list of block IDs - finalizeUpload(urls, client); + // Finalize the upload by sending the ordered list of block IDs + List orderedUrls = new ArrayList<>(); + for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + orderedUrls.add(partUrls.get(partNumber)); + } + finalizeUpload(orderedUrls, client); } catch (Exception e) { abortUpload(urlResponse); @@ -74,7 +79,7 @@ public void abortUpload(DataLinkMultiPartUploadResponse urlResponse) { // Send an empty block list to abort the upload // Per Azure documentation, any Uncommitted blocks not part of the final BlockList are garbage collected String emptyBlockList = ""; - + HttpRequest abortRequest = HttpRequest.newBuilder() .uri(URI.create(abortUrl)) .PUT(HttpRequest.BodyPublishers.ofString(emptyBlockList)) @@ -126,4 +131,4 @@ private String buildBlockList(List blockIds) { xml.append(""); return xml.toString(); } -} \ No newline at end of file +} diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/GoogleUploader.java b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/GoogleUploader.java index 4664ad3f4..b0d5b75b6 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/upload/GoogleUploader.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/upload/GoogleUploader.java @@ -16,7 +16,7 @@ package io.seqera.tower.cli.commands.data.links.upload; -import io.seqera.tower.ApiException; +import io.seqera.tower.api.DataLinksApi; import io.seqera.tower.cli.exceptions.TowerRuntimeException; import io.seqera.tower.cli.utils.progress.ProgressTracker; import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; @@ -31,6 +31,10 @@ public class GoogleUploader extends AbstractProviderUploader { + public GoogleUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi); + } + @Override public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, ProgressTracker tracker) { String url = urlResponse.getUploadUrls().get(0); @@ -40,17 +44,18 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P HttpClient client = HttpClient.newHttpClient(); try { while (nextByteToRead < fileSize) { - int partNumber = (int)(nextByteToRead / MULTI_UPLOAD_PART_SIZE_IN_BYTES); + int partNumber = (int) (nextByteToRead / MULTI_UPLOAD_PART_SIZE_IN_BYTES); byte[] chunk = getChunk(file, partNumber); - long end = nextByteToRead + chunk.length; + final long start = nextByteToRead; + final long end = start + chunk.length; + long baseline = tracker.snapshot(); - HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create(url)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) - .header("Content-Range", String.format("bytes %d-%d/%d", nextByteToRead, Math.max(0, end - 1), fileSize)) - .build(); - - HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + HttpResponse response = sendWithRetryOnTransientError(client, tracker, baseline, + () -> HttpRequest.newBuilder() + .uri(URI.create(url)) + .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) + .header("Content-Range", String.format("bytes %d-%d/%d", start, Math.max(0, end - 1), fileSize)) + .build()); if (response.statusCode() == 308) { // Resume upload from the last byte received by the server @@ -59,10 +64,11 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P long lastByte = Long.parseLong(range.substring(range.lastIndexOf('-') + 1)); nextByteToRead = lastByte + 1; } - } else if (response.statusCode() != 200) { - throw new IOException("Failed to upload file: HTTP " + response.statusCode()); - } else { + } else if (response.statusCode() == 200) { break; // Upload completed successfully + } else { + tracker.restore(baseline); + throw new IOException("Failed to upload file: HTTP " + response.statusCode()); } } } catch (Exception e) { @@ -88,4 +94,4 @@ public void abortUpload(DataLinkMultiPartUploadResponse urlResponse) { throw new TowerRuntimeException("Failed to upload file and encountered error while attempting to cancel upload " + e.getMessage(), e); } } -} \ No newline at end of file +} diff --git a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTracker.java b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTracker.java index 36695def6..df3cc94f7 100644 --- a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTracker.java +++ b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTracker.java @@ -34,6 +34,22 @@ public ProgressTracker(PrintWriter out, boolean showProgress, long totalBytes) { this.totalBytes = totalBytes; } + /** + * Returns the current cumulative uploaded-bytes count, so a caller can roll back to it if an + * in-flight part fails and has to be retried (avoids double-counting the re-sent bytes). + */ + public synchronized long snapshot() { + return uploadedBytes; + } + + /** + * Restores the cumulative uploaded-bytes count to a previously taken {@link #snapshot()} value, + * used to discard the progress reported by a failed part before it is retried. + */ + public synchronized void restore(long bytes) { + uploadedBytes = bytes; + } + public synchronized void update(long count) { uploadedBytes += count; int percent = (int) ((uploadedBytes * 100) / totalBytes); diff --git a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java index 13e15376b..086d9c71f 100644 --- a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java +++ b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java @@ -891,7 +891,229 @@ void testUploadSingleFileFailsButStillFinalizesUpload(OutputType format, MockSer "}\n")) , VerificationTimes.exactly(1)); - assertEquals(errorMessage(out.app, new TowerRuntimeException("Failed to upload file: Failed to upload file: HTTP 404, Message: not found")), out.stdErr); + assertEquals(errorMessage(out.app, new TowerRuntimeException("Failed to upload file: Failed to upload part 1: HTTP 404, Message: not found")), out.stdErr); + assertEquals("", out.stdOut); + assertEquals(1, out.exitCode); + + Files.deleteIfExists(testFile); + } + + @ParameterizedTest + @EnumSource(value = OutputType.class, names = {"json"}) + void testUploadRefreshesUrlOnExpiredTokenAndSucceeds(OutputType format, MockServerClient mock) throws IOException { + // credentials fetch + mock.when( + request().withMethod("GET").withPath("/credentials").withQueryStringParameter("workspaceId", "75887156211589"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\"credentials\":[{\"id\":\"57Ic6reczFn78H1DTaaXkp\",\"name\":\"aws\",\"description\":null,\"discriminator\":\"aws\",\"baseUrl\":null,\"category\":null,\"deleted\":null,\"lastUsed\":\"2021-09-09T07:20:53Z\",\"dateCreated\":\"2021-09-08T05:48:51Z\",\"lastUpdated\":\"2021-09-08T05:48:51Z\"}]}").withContentType(MediaType.APPLICATION_JSON) + ); + + // status check + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("offset", "0").withQueryStringParameter("max", "1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + // mock fetch data links list + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("search", "a-test-bucket-eend-us-east-1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + + Path testFile = tempDir().resolve("test.txt"); + Files.write(testFile, "test content".getBytes()); + + // Mock multipart upload request + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"upload-123\",\n \"uploadUrls\": [\"http://localhost:" + mock.getPort() + "/upload\"]\n}").withContentType(MediaType.APPLICATION_JSON) + ); + + // First PUT fails with an expired-token error + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(400).withBody("ExpiredTokenThe provided token has expired.") + ); + + // Refresh endpoint returns a fresh URL for part 1 + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/refresh").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"upload-123\",\n \"uploadUrls\": [{\"partNumber\": 1, \"url\": \"http://localhost:" + mock.getPort() + "/upload\"}]\n}").withContentType(MediaType.APPLICATION_JSON) + ); + + // Second PUT (after refresh) succeeds + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(200).withHeader(new Header("Etag", "etag-123")) + ); + + // Finish upload (success) + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\n\"uploadId\":\"upload-123\",\n\"fileName\":\"test.txt\",\n\"tags\":[{\"partNumber\":1,\"eTag\":\"etag-123\"}],\n\"withError\":false\n}")), exactly(1) + ).respond( + response().withStatusCode(200) + ); + + ExecOut out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", + "-c", "57Ic6reczFn78H1DTaaXkp", testFile.toString()); + + // The refresh endpoint was invoked exactly once + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/refresh"), VerificationTimes.exactly(1)); + + assertOutput(format, out, DataLinkFileTransferResult.uploaded(List.of( + new DataLinkFileTransferResult.SimplePathInfo(DataLinkItemType.FILE, testFile.toString(), 1) + ))); + assertEquals("", out.stdErr); + assertEquals(0, out.exitCode); + + Files.deleteIfExists(testFile); + } + + @ParameterizedTest + @EnumSource(value = OutputType.class, names = {"json"}) + void testUploadRetriesOnTransientErrorAndSucceeds(OutputType format, MockServerClient mock) throws IOException { + // credentials fetch + mock.when( + request().withMethod("GET").withPath("/credentials").withQueryStringParameter("workspaceId", "75887156211589"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\"credentials\":[{\"id\":\"57Ic6reczFn78H1DTaaXkp\",\"name\":\"aws\",\"description\":null,\"discriminator\":\"aws\",\"baseUrl\":null,\"category\":null,\"deleted\":null,\"lastUsed\":\"2021-09-09T07:20:53Z\",\"dateCreated\":\"2021-09-08T05:48:51Z\",\"lastUpdated\":\"2021-09-08T05:48:51Z\"}]}").withContentType(MediaType.APPLICATION_JSON) + ); + + // status check + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("offset", "0").withQueryStringParameter("max", "1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + // mock fetch data links list + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("search", "a-test-bucket-eend-us-east-1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + + Path testFile = tempDir().resolve("test.txt"); + Files.write(testFile, "test content".getBytes()); + + // Mock multipart upload request + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"upload-123\",\n \"uploadUrls\": [\"http://localhost:" + mock.getPort() + "/upload\"]\n}").withContentType(MediaType.APPLICATION_JSON) + ); + + // First PUT fails with a transient 503 (no provider error code -> classified transient by status) + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(503) + ); + + // Second PUT (retry of the same URL) succeeds + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(200).withHeader(new Header("Etag", "etag-123")) + ); + + // Finish upload (success) + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\n\"uploadId\":\"upload-123\",\n\"fileName\":\"test.txt\",\n\"tags\":[{\"partNumber\":1,\"eTag\":\"etag-123\"}],\n\"withError\":false\n}")), exactly(1) + ).respond( + response().withStatusCode(200) + ); + + ExecOut out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", + "-c", "57Ic6reczFn78H1DTaaXkp", testFile.toString()); + + // The refresh endpoint must NOT be called for a transient error + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/refresh"), VerificationTimes.exactly(0)); + + assertOutput(format, out, DataLinkFileTransferResult.uploaded(List.of( + new DataLinkFileTransferResult.SimplePathInfo(DataLinkItemType.FILE, testFile.toString(), 1) + ))); + assertEquals("", out.stdErr); + assertEquals(0, out.exitCode); + + Files.deleteIfExists(testFile); + } + + @ParameterizedTest + @EnumSource(value = OutputType.class, names = {"json"}) + void testUploadFailsWithClearMessageWhenRefreshNotSupported(OutputType format, MockServerClient mock) throws IOException { + // credentials fetch + mock.when( + request().withMethod("GET").withPath("/credentials").withQueryStringParameter("workspaceId", "75887156211589"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\"credentials\":[{\"id\":\"57Ic6reczFn78H1DTaaXkp\",\"name\":\"aws\",\"description\":null,\"discriminator\":\"aws\",\"baseUrl\":null,\"category\":null,\"deleted\":null,\"lastUsed\":\"2021-09-09T07:20:53Z\",\"dateCreated\":\"2021-09-08T05:48:51Z\",\"lastUpdated\":\"2021-09-08T05:48:51Z\"}]}").withContentType(MediaType.APPLICATION_JSON) + ); + + // status check + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("offset", "0").withQueryStringParameter("max", "1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + // mock fetch data links list + mock.when( + request().withMethod("GET").withPath("/data-links").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("search", "a-test-bucket-eend-us-east-1"), exactly(1) + ).respond( + response().withStatusCode(200).withBody(loadResource("data/links/datalinks_list")).withContentType(MediaType.APPLICATION_JSON) + ); + + Path testFile = tempDir().resolve("test.txt"); + Files.write(testFile, "test content".getBytes()); + + // Mock multipart upload request + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"upload-123\",\n \"uploadUrls\": [\"http://localhost:" + mock.getPort() + "/upload\"]\n}").withContentType(MediaType.APPLICATION_JSON) + ); + + // PUT fails with an expired-token error -> triggers a refresh attempt + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(400).withBody("ExpiredTokenThe provided token has expired.") + ); + + // Old Platform: the refresh endpoint does not exist -> 404 + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/refresh"), exactly(1) + ).respond( + response().withStatusCode(404) + ); + + // Finish upload is still called with withError=true to abort the multipart upload + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\n\"uploadId\":\"upload-123\",\n\"fileName\":\"test.txt\",\n\"tags\":[],\n\"withError\":true\n}")), exactly(1) + ).respond( + response().withStatusCode(200) + ); + + ExecOut out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", + "-c", "57Ic6reczFn78H1DTaaXkp", testFile.toString()); + + // The upload is still finalized with withError=true + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withBody(json("{\n\"uploadId\":\"upload-123\",\n\"fileName\":\"test.txt\",\n\"tags\":[],\n\"withError\":true\n}")), VerificationTimes.exactly(1)); + + assertEquals(errorMessage(out.app, new TowerRuntimeException("Failed to upload file: Token refresh is not supported for this Platform version.")), out.stdErr); assertEquals("", out.stdOut); assertEquals(1, out.exitCode);