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 90b55036..669b479d 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 95253624..ecb25c2f 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,69 @@ 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.DataLinkFinishMultiPartUploadRequest; +import io.seqera.tower.model.DataLinkMultiPartUploadRequest; +import io.seqera.tower.model.DataLinkMultiPartUploadResponse; +import io.seqera.tower.model.UploadEtag; + 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.Collections; +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 +94,239 @@ 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. + */ + protected Map refreshUrls(String uploadId, long contentLength, List partNumbers) throws ApiException { + DataLinkMultiPartUploadRequest request = new DataLinkMultiPartUploadRequest(); + request.setUploadId(uploadId); + request.setFileName(relativeKey); + request.setContentLength(contentLength); + request.setPartNumbers(partNumbers); + + DataLinkMultiPartUploadResponse response; + try { + response = outputDir != null + ? dataLinksApi.generateDataLinkUploadUrlWithPath(id, outputDir, request, credId, wspId, null) + : dataLinksApi.generateDataLinkUploadUrl(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; + } + + // A Platform that predates re-signing ignores the uploadId/partNumbers fields and instead initiates a + // brand-new multi-part upload, returning a different uploadId. Detect that by the echoed uploadId and + // fail clearly rather than mixing URLs from a different upload into the in-progress one. + if (!uploadId.equals(response.getUploadId())) { + abandonUpload(response.getUploadId()); + throw new TowerRuntimeException("Token refresh is not supported for this Platform version."); + } + + List urls = response.getUploadUrls(); + Map map = new HashMap<>(); + if (urls != null) { + if (urls.size() != partNumbers.size()) { + throw new TowerRuntimeException("Platform returned " + urls.size() + + " refreshed upload URLs but " + partNumbers.size() + " were requested"); + } + for (int i = 0; i < partNumbers.size(); i++) { + map.put(partNumbers.get(i), urls.get(i)); + } + } + return map; + } + + /** + * Finalizes a multi-part upload on the Platform. With {@code withError} the upload is aborted instead of + * committed, which is also how an unwanted upload is cleaned up. + */ + protected void finishUpload(String uploadId, boolean withError, List tags) throws ApiException { + DataLinkFinishMultiPartUploadRequest request = new DataLinkFinishMultiPartUploadRequest(); + request.setFileName(relativeKey); + request.setUploadId(uploadId); + request.setWithError(withError); + request.setTags(tags); + + if (outputDir != null) { + dataLinksApi.finishDataLinkUploadWithPath(id, outputDir, request, credId, wspId); + } else { + dataLinksApi.finishDataLinkUpload(id, request, credId, wspId); + } + } + + private void abandonUpload(String uploadId) { + if (uploadId == null) { + return; + } + try { + finishUpload(uploadId, true, Collections.emptyList()); + } catch (Exception e) { + // ignore — cleanup is best-effort against a Platform that may not support it + } + } + + /** + * 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 / 429 / 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 + case "RequestTimeout": // S3 + case "ServerBusy": // Azure throttling + case "OperationTimedOut": // Azure + return UploadErrorType.TRANSIENT; + default: + // fall through to status-based classification + } + } + // 429 is how GCS (and some fronting proxies) signal throttling, without an XML error body. + if (statusCode == 429 || 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 7819d3d5..cfe11662 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,100 +20,67 @@ 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); - - HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create(url)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) - .build(); + for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + byte[] chunk = getChunk(file, partNumber - 1); - 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; throw new TowerRuntimeException("Failed to upload file: " + e.getMessage(), e); } finally { - finalizeUpload(urlResponse, withError, tags); - } - } - - private void finalizeUpload(DataLinkMultiPartUploadResponse urlResponse, boolean withError, List tags) throws ApiException { - // Finalize the upload - DataLinkFinishMultiPartUploadRequest finishMultiPartUploadRequest = new DataLinkFinishMultiPartUploadRequest(); - finishMultiPartUploadRequest.setFileName(relativeKey); - finishMultiPartUploadRequest.setUploadId(urlResponse.getUploadId()); - finishMultiPartUploadRequest.setWithError(withError); - finishMultiPartUploadRequest.setTags(tags); - - if (outputDir != null) { - dataLinksApi.finishDataLinkUploadWithPath(id, outputDir, finishMultiPartUploadRequest, credId, wspId); - } else { - dataLinksApi.finishDataLinkUpload(id, finishMultiPartUploadRequest, credId, wspId); + finishUpload(urlResponse.getUploadId(), withError, tags); } } @Override public void abortUpload(DataLinkMultiPartUploadResponse urlResponse) throws ApiException { - finalizeUpload(urlResponse, true, Collections.emptyList()); + finishUpload(urlResponse.getUploadId(), 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 1df8b897..818fa467 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 4664ad3f..b0d5b75b 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 36695def..df3cc94f 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 13e15376..28ba98a1 100644 --- a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java +++ b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java @@ -35,6 +35,7 @@ import org.junit.jupiter.params.provider.EnumSource; import io.seqera.tower.model.DataLinkItemType; import org.mockserver.client.MockServerClient; +import org.mockserver.matchers.MatchType; import org.mockserver.model.Header; import org.mockserver.model.MediaType; import org.mockserver.verify.VerificationTimes; @@ -891,7 +892,322 @@ 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.") + ); + + // Re-signing re-calls the upload endpoint with uploadId + partNumbers set; the response URLs are + // positional, matching the requested part numbers (here: a fresh URL for part 1). + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\"uploadId\": \"upload-123\", \"partNumbers\": [1]}", MatchType.ONLY_MATCHING_FIELDS)), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"upload-123\",\n \"uploadUrls\": [\"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 upload endpoint was re-invoked in re-sign mode (uploadId + partNumbers) exactly once + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload") + .withBody(json("{\"uploadId\": \"upload-123\", \"partNumbers\": [1]}", MatchType.ONLY_MATCHING_FIELDS)), 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()); + + // Re-signing (upload endpoint with uploadId + partNumbers) must NOT be triggered for a transient error + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload") + .withBody(json("{\"uploadId\": \"upload-123\", \"partNumbers\": [1]}", MatchType.ONLY_MATCHING_FIELDS)), 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 testUploadRetriesOnThrottlingAndSucceeds(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 is throttled with a bare 429 (how GCS and fronting proxies signal it, no XML error body) + mock.when( + request().withMethod("PUT").withPath("/upload"), exactly(1) + ).respond( + response().withStatusCode(429) + ); + + // 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()); + + // Throttling is not a credential problem, so re-signing must not be triggered + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload") + .withBody(json("{\"uploadId\": \"upload-123\", \"partNumbers\": [1]}", MatchType.ONLY_MATCHING_FIELDS)), 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: it predates re-signing, so it ignores the uploadId/partNumbers fields and instead + // initiates a brand-new upload, returning a *different* uploadId. The CLI must detect this mismatch. + mock.when( + request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload").withQueryStringParameter("workspaceId", "75887156211589").withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\"uploadId\": \"upload-123\", \"partNumbers\": [1]}", MatchType.ONLY_MATCHING_FIELDS)), exactly(1) + ).respond( + response().withStatusCode(200).withBody("{\n \"uploadId\": \"a-different-upload-999\",\n \"uploadUrls\": [\"http://localhost:" + mock.getPort() + "/upload\"]\n}").withContentType(MediaType.APPLICATION_JSON) + ); + + // 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) + ); + + // ...and the upload the old Platform started behind our back is aborted too, so it is not left open + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp") + .withBody(json("{\n\"uploadId\":\"a-different-upload-999\",\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)); + + // The upload the old Platform created in response to the refresh request is aborted as well + mock.verify(request().withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withBody(json("{\n\"uploadId\":\"a-different-upload-999\",\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);