From ac821eff3f336dc187ca029533cdb4073bdcdba7 Mon Sep 17 00:00:00 2001 From: "georgi.hristov" Date: Mon, 3 Aug 2026 18:14:38 +0100 Subject: [PATCH 1/4] feat: parallelize data-link chunk uploads --- .../cli/commands/data/links/DownloadCmd.java | 7 +- .../cli/commands/data/links/UploadCmd.java | 19 ++- .../upload/AbstractProviderUploader.java | 115 ++++++++++++++++-- .../data/links/upload/AwsUploader.java | 30 +++-- .../data/links/upload/AzureUploader.java | 16 +-- .../data/links/upload/GoogleUploader.java | 15 +-- .../cli/utils/progress/PartProgress.java | 53 ++++++++ .../utils/progress/ProgressInputStream.java | 10 +- .../cli/utils/progress/ProgressSink.java | 26 ++++ .../cli/utils/progress/ProgressTracker.java | 36 +++--- .../ProgressTrackingBodyPublisher.java | 10 +- .../tower/cli/data/DataLinksCmdTest.java | 112 +++++++++++++++++ .../utils/progress/ProgressTrackerTest.java | 103 ++++++++++++++++ 13 files changed, 478 insertions(+), 74 deletions(-) create mode 100644 src/main/java/io/seqera/tower/cli/utils/progress/PartProgress.java create mode 100644 src/main/java/io/seqera/tower/cli/utils/progress/ProgressSink.java create mode 100644 src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java diff --git a/src/main/java/io/seqera/tower/cli/commands/data/links/DownloadCmd.java b/src/main/java/io/seqera/tower/cli/commands/data/links/DownloadCmd.java index c6943a7ba..0572408cc 100644 --- a/src/main/java/io/seqera/tower/cli/commands/data/links/DownloadCmd.java +++ b/src/main/java/io/seqera/tower/cli/commands/data/links/DownloadCmd.java @@ -61,6 +61,9 @@ public class DownloadCmd extends AbstractDataLinksCmd { @CommandLine.Option(names = {"-o", "--output-dir"}, description = "Output directory for downloaded files") public String outputDir; + @CommandLine.Option(names = {"--silent"}, description = "Suppress download progress indicators. Useful for scripting or logging to files.") + public boolean silent; + @CommandLine.Parameters(arity = "1..*", description = "Paths to files or directories to download") private List paths; @@ -111,7 +114,7 @@ protected Response exec() throws ApiException, IOException, InterruptedException private void downloadFile(String path, String id, String credId, Long wspId, Path targetPath) throws ApiException, IOException, InterruptedException { DataLinkDownloadUrlResponse urlResponse = dataLinksApi().generateDownloadUrlDataLink(id, path, credId, wspId, false, null); - boolean showProgress = app().output != OutputType.json; + boolean showProgress = app().output != OutputType.json && !silent; if (showProgress) { app().getOut().println(" Downloading file: " + path); @@ -138,7 +141,7 @@ private void downloadFile(String path, String id, String credId, Long wspId, Pat .orElse(-1); ProgressTracker tracker = new ProgressTracker(app().getOut(), showProgress, contentLength); - try (InputStream in = new ProgressInputStream(response.body(), tracker); + try (InputStream in = new ProgressInputStream(response.body(), tracker.newPart()); OutputStream output = Files.newOutputStream(targetPath)) { byte[] buffer = new byte[8192]; 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 669b479dd..bfcba1216 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 @@ -62,11 +62,20 @@ public class UploadCmd extends AbstractDataLinksCmd { @CommandLine.Option(names = {"-o", "--output-dir"}, description = "Destination directory in the data link") public String outputDir; + @CommandLine.Option(names = {"--silent"}, description = "Suppress upload progress indicators. Useful for scripting or logging to files.") + public boolean silent; + + @CommandLine.Option(names = {"--concurrency"}, defaultValue = "4", description = "Number of file chunks to upload in parallel (default: ${DEFAULT-VALUE}). Each in-flight chunk buffers up to 250 MB in memory, so peak memory is roughly concurrency x 250 MB.") + public int concurrency; + @CommandLine.Parameters(arity = "1..*", description = "Paths to files or directories to upload") private List paths; @Override protected Response exec() throws ApiException, IOException, InterruptedException { + if (concurrency < 1) { + throw new TowerRuntimeException("--concurrency must be at least 1."); + } checkFilesValidForUpload(); Long wspId = workspaceId(workspace.workspace); @@ -122,7 +131,7 @@ private void uploadFile(File file, String relativeKey, String id, String credId, } long contentLength = file.length(); - boolean showProgress = app().output != OutputType.json; + boolean showProgress = app().output != OutputType.json && !silent; if (showProgress) { app().getOut().println("Uploading file: " + file.getPath()); } @@ -149,14 +158,14 @@ private void uploadFile(File file, String relativeKey, String id, String credId, private CloudProviderUploader createUploadStrategy(DataLinkProvider provider, String id, String credId, Long wspId, String outputDir, String relativeKey) throws ApiException { switch (provider) { case AWS: - return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); + return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi(), concurrency); case GOOGLE: - return new GoogleUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); + return new GoogleUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi(), concurrency); case AZURE: - return new AzureUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); + return new AzureUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi(), concurrency); case SEQERACOMPUTE: // Seqera Compute uses S3-compatible uploads, same as AWS - return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi()); + return new AwsUploader(id, credId, wspId, outputDir, relativeKey, dataLinksApi(), concurrency); default: throw new TowerRuntimeException("Unsupported data-link provider: " + provider); } 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 23efc7f92..b18cecc11 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 @@ -19,6 +19,7 @@ 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.PartProgress; import io.seqera.tower.cli.utils.progress.ProgressTracker; import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; import io.seqera.tower.model.DataLinkRefreshMultiPartUploadRequest; @@ -37,14 +38,46 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletionService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; 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 + static final int DEFAULT_PART_SIZE_IN_BYTES = 250 * 1024 * 1024; // 250 MB + + /** + * Overrides the multipart part size (in bytes). + */ + public static final String PART_SIZE_ENV = "TOWER_UPLOAD_SIZE_PART_BYTES"; + + /** Multipart upload part size in bytes; honors {@link #PART_SIZE_ENV}, defaulting to 250 MB. */ + static int partSizeBytes() { + String value = System.getenv(PART_SIZE_ENV); + if (value == null) { + value = System.getProperty(PART_SIZE_ENV); + } + if (value != null) { + try { + int parsed = Integer.parseInt(value.trim()); + if (parsed > 0) { + return parsed; + } + } catch (NumberFormatException ignored) { + // fall through to the default + } + } + return DEFAULT_PART_SIZE_IN_BYTES; + } /** Max attempts per part, covering both refresh-on-expiry and transient-error retries. */ static final int MAX_PART_ATTEMPTS = 5; @@ -65,22 +98,30 @@ public abstract class AbstractProviderUploader implements CloudProviderUploader protected final String outputDir; protected final String relativeKey; protected final DataLinksApi dataLinksApi; + protected final int concurrency; - protected AbstractProviderUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { + protected AbstractProviderUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi, int concurrency) { this.id = id; this.credId = credId; this.wspId = wspId; this.outputDir = outputDir; this.relativeKey = relativeKey; this.dataLinksApi = dataLinksApi; + this.concurrency = concurrency; } protected enum UploadErrorType { EXPIRY, TRANSIENT, HARD_FAIL } + /** A unit of parallel work: uploads one part (1-based part number) and returns its result. */ + @FunctionalInterface + protected interface PartTask { + R run(int partNumber) throws Exception; + } + protected byte[] getChunk(File file, int index) { try (RandomAccessFile raf = new RandomAccessFile(file, "r")) { - long start = (long) index * MULTI_UPLOAD_PART_SIZE_IN_BYTES; - long end = Math.min(start + MULTI_UPLOAD_PART_SIZE_IN_BYTES, file.length()); + long start = (long) index * partSizeBytes(); + long end = Math.min(start + partSizeBytes(), file.length()); int length = (int) (end - start); byte[] buffer = new byte[length]; @@ -97,7 +138,7 @@ protected int totalParts(long contentLength) { if (contentLength <= 0) { return 1; } - return (int) Math.ceil((double) contentLength / MULTI_UPLOAD_PART_SIZE_IN_BYTES); + return (int) Math.ceil((double) contentLength / partSizeBytes()); } /** @@ -121,7 +162,7 @@ protected HttpResponse uploadPartWithRetry(HttpClient client, Map uploadPartWithRetry(HttpClient client, Map response = sendWithRetryOnTransientError(client, tracker, baseline, + HttpResponse response = sendWithRetryOnTransientError(client, part, () -> HttpRequest.newBuilder() .uri(URI.create(partUrl)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) + .PUT(new ProgressTrackingBodyPublisher(chunk, part)) .build()); if (response.statusCode() == successStatus) { @@ -147,7 +188,7 @@ protected HttpResponse uploadPartWithRetry(HttpClient client, Map uploadPartWithRetry(HttpClient client, Map Map uploadPartsInParallel(int totalParts, PartTask task) { + int workers = Math.max(1, Math.min(concurrency, totalParts)); + ExecutorService pool = Executors.newFixedThreadPool(workers); + Map results = new ConcurrentHashMap<>(); + List> futures = new ArrayList<>(); + try { + CompletionService completion = new ExecutorCompletionService<>(pool); + for (int p = 1; p <= totalParts; p++) { + final int partNumber = p; + futures.add(completion.submit(() -> { + R r = task.run(partNumber); + if (r != null) { + results.put(partNumber, r); + } + return partNumber; + })); + } + for (int i = 0; i < totalParts; i++) { + completion.take().get(); + } + return results; + } catch (ExecutionException e) { + // First failure: cancel the rest and surface the cause to the caller. + futures.forEach(f -> f.cancel(true)); + Throwable cause = e.getCause(); + throw new TowerRuntimeException(cause != null ? cause.getMessage() : e.getMessage(), cause); + } catch (InterruptedException e) { + futures.forEach(f -> f.cancel(true)); + Thread.currentThread().interrupt(); + throw new TowerRuntimeException("Upload interrupted", e); + } finally { + pool.shutdownNow(); + try { + pool.awaitTermination(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + /** * 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 @@ -167,15 +257,16 @@ protected HttpResponse uploadPartWithRetry(HttpClient client, Map sendWithRetryOnTransientError(HttpClient client, ProgressTracker tracker, long baseline, + protected HttpResponse sendWithRetryOnTransientError(HttpClient client, PartProgress part, Supplier request) throws IOException, InterruptedException { IOException lastError = null; for (int attempt = 1; attempt <= MAX_PART_ATTEMPTS; attempt++) { - tracker.restore(baseline); + part.reset(); try { HttpResponse response = client.send(request.get(), HttpResponse.BodyHandlers.ofString()); if (classify(response.statusCode(), response.body()) == UploadErrorType.TRANSIENT && attempt < MAX_PART_ATTEMPTS) { 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 6b2a74bec..9255198d1 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 @@ -29,15 +29,15 @@ 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; +import java.util.concurrent.ConcurrentHashMap; public class AwsUploader extends AbstractProviderUploader { - public AwsUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi) { - super(id, credId, wspId, outputDir, relativeKey, dataLinksApi); + public AwsUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi, int concurrency) { + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi, concurrency); } @Override @@ -50,27 +50,31 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P // 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<>(); + Map partUrls = new ConcurrentHashMap<>(); 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 (int partNumber = 1; partNumber <= totalParts; partNumber++) { + // Upload all parts concurrently; each returns its ETag keyed by part number. + Map etags = uploadPartsInParallel(totalParts, partNumber -> { byte[] chunk = getChunk(file, partNumber - 1); - 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(partNumber); - tags.add(uploadEtag); - } else { + if (etag.isEmpty()) { throw new TowerRuntimeException("Failed to upload file: Possible CORS issue"); } + return etag.get(); + }); + + // S3 requires the completed-parts list in ascending part-number order, so build it from + // the keyed results after all parts finish (completion order is nondeterministic). + for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + UploadEtag uploadEtag = new UploadEtag(); + uploadEtag.eTag(etags.get(partNumber)); + uploadEtag.partNumber(partNumber); + tags.add(uploadEtag); } } catch (Exception e) { withError = true; 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 818fa4679..164f7e1a1 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 @@ -28,15 +28,15 @@ 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.ConcurrentHashMap; 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); + public AzureUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi, int concurrency) { + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi, concurrency); } @Override @@ -44,18 +44,18 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P long contentLength = file.length(); List initialUrls = urlResponse.getUploadUrls(); - Map partUrls = new HashMap<>(); + Map partUrls = new ConcurrentHashMap<>(); for (int i = 0; i < initialUrls.size(); i++) { partUrls.put(i + 1, initialUrls.get(i)); } int totalParts = initialUrls.size(); - HttpClient client = HttpClient.newHttpClient(); - try { - for (int partNumber = 1; partNumber <= totalParts; partNumber++) { + try (HttpClient client = HttpClient.newHttpClient()) { + uploadPartsInParallel(totalParts, partNumber -> { byte[] chunk = getChunk(file, partNumber - 1); uploadPartWithRetry(client, partUrls, partNumber, chunk, tracker, null, contentLength, 201, false); - } + return Boolean.TRUE; // result unused; must be non-null to mark the part done + }); // Finalize the upload by sending the ordered list of block IDs List orderedUrls = new ArrayList<>(); 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 b0d5b75b6..f8234d90a 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 @@ -18,6 +18,7 @@ import io.seqera.tower.api.DataLinksApi; import io.seqera.tower.cli.exceptions.TowerRuntimeException; +import io.seqera.tower.cli.utils.progress.PartProgress; import io.seqera.tower.cli.utils.progress.ProgressTracker; import io.seqera.tower.cli.utils.progress.ProgressTrackingBodyPublisher; import io.seqera.tower.model.DataLinkMultiPartUploadResponse; @@ -31,8 +32,8 @@ 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); + public GoogleUploader(String id, String credId, Long wspId, String outputDir, String relativeKey, DataLinksApi dataLinksApi, int concurrency) { + super(id, credId, wspId, outputDir, relativeKey, dataLinksApi, concurrency); } @Override @@ -44,16 +45,16 @@ 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 / partSizeBytes()); byte[] chunk = getChunk(file, partNumber); final long start = nextByteToRead; final long end = start + chunk.length; - long baseline = tracker.snapshot(); + PartProgress part = tracker.newPart(); - HttpResponse response = sendWithRetryOnTransientError(client, tracker, baseline, + HttpResponse response = sendWithRetryOnTransientError(client, part, () -> HttpRequest.newBuilder() .uri(URI.create(url)) - .PUT(new ProgressTrackingBodyPublisher(chunk, tracker)) + .PUT(new ProgressTrackingBodyPublisher(chunk, part)) .header("Content-Range", String.format("bytes %d-%d/%d", start, Math.max(0, end - 1), fileSize)) .build()); @@ -67,7 +68,7 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P } else if (response.statusCode() == 200) { break; // Upload completed successfully } else { - tracker.restore(baseline); + part.reset(); throw new IOException("Failed to upload file: HTTP " + response.statusCode()); } } diff --git a/src/main/java/io/seqera/tower/cli/utils/progress/PartProgress.java b/src/main/java/io/seqera/tower/cli/utils/progress/PartProgress.java new file mode 100644 index 000000000..2d55192f3 --- /dev/null +++ b/src/main/java/io/seqera/tower/cli/utils/progress/PartProgress.java @@ -0,0 +1,53 @@ +/* + * Copyright 2021-2026, Seqera. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.seqera.tower.cli.utils.progress; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * Progress accounting scoped to a single upload part, created once per part via + * {@link ProgressTracker#newPart()} and reused across all of that part's retry attempts. + * + *

{@code reported} is an {@link AtomicLong} because {@link #update(long)} runs on the HTTP + * client's internal executor thread while {@link #reset()} runs on the uploading worker thread.

+ */ +public class PartProgress implements ProgressSink { + + private final ProgressTracker parent; + private final AtomicLong reported = new AtomicLong(); + + PartProgress(ProgressTracker parent) { + this.parent = parent; + } + + @Override + public void update(long count) { + reported.addAndGet(count); + parent.update(count); + } + + /** + * Rolls back the bytes this part has reported so far (used before retrying a failed attempt). + * A no-op when nothing has been reported yet. + */ + public void reset() { + long toUndo = reported.getAndSet(0); + if (toUndo != 0) { + parent.update(-toUndo); + } + } +} diff --git a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressInputStream.java b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressInputStream.java index efbc7a865..7db6c91c6 100644 --- a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressInputStream.java +++ b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressInputStream.java @@ -21,24 +21,24 @@ public class ProgressInputStream extends InputStream { private final InputStream source; - private final ProgressTracker tracker; + private final ProgressSink sink; - public ProgressInputStream(InputStream source, ProgressTracker tracker) { + public ProgressInputStream(InputStream source, ProgressSink sink) { this.source = source; - this.tracker = tracker; + this.sink = sink; } @Override public int read() throws IOException { int b = source.read(); - if (b != -1) tracker.update(1); + if (b != -1) sink.update(1); return b; } @Override public int read(byte[] b, int off, int len) throws IOException { int count = source.read(b, off, len); - if (count > 0) tracker.update(count); + if (count > 0) sink.update(count); return count; } diff --git a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressSink.java b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressSink.java new file mode 100644 index 000000000..edd0388d1 --- /dev/null +++ b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressSink.java @@ -0,0 +1,26 @@ +/* + * Copyright 2021-2026, Seqera. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.seqera.tower.cli.utils.progress; + +/** + * A sink that byte-streaming components ({@link ProgressTrackingBodyPublisher}/{@link ProgressInputStream}) + * report uploaded byte counts to as a request body is transmitted. Implemented by {@link PartProgress} + * so the streaming code is decoupled from the shared {@link ProgressTracker}. + */ +public interface ProgressSink { + void update(long count); +} 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 df3cc94f7..77cf0d2ae 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 @@ -25,6 +25,7 @@ public class ProgressTracker { public final long totalBytes; private volatile long uploadedBytes = 0; private volatile int lastPercent = -1; + private boolean finished = false; private final int barWidth = 40; private final Instant startTime = Instant.now(); @@ -35,47 +36,48 @@ public ProgressTracker(PrintWriter out, boolean showProgress, long 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). + * Creates a progress accounting handle scoped to a single upload part. Each part reports its + * bytes through the returned {@link PartProgress}, which can roll back only its own bytes on a + * failed attempt — safe even when many parts upload concurrently. */ - public synchronized long snapshot() { - return uploadedBytes; + public PartProgress newPart() { + return new PartProgress(this); } - /** - * 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; + /** Current cumulative uploaded-bytes count. Package-private, for observability in tests. */ + synchronized long currentBytes() { + return uploadedBytes; } - public synchronized void update(long count) { + synchronized void update(long count) { uploadedBytes += count; int percent = (int) ((uploadedBytes * 100) / totalBytes); if (percent != lastPercent) { lastPercent = percent; long elapsedMillis = java.time.Duration.between(startTime, Instant.now()).toMillis(); + long elapsedSeconds = elapsedMillis / 1000; double speed = uploadedBytes / (elapsedMillis / 1000.0); - double eta = (totalBytes - uploadedBytes) / speed; + double eta = (elapsedMillis <= 0 || !Double.isFinite(speed) || speed <= 0) + ? 0.0 : (totalBytes - uploadedBytes) / speed; if (showProgress) { if (totalBytes > 1024) { - renderBar(percent, uploadedBytes / 1024, totalBytes / 1024, "KBs", eta); + renderBar(percent, uploadedBytes / 1024, totalBytes / 1024, "KBs", eta, elapsedSeconds); } else { - renderBar(percent, uploadedBytes, totalBytes, "bytes", eta); + renderBar(percent, uploadedBytes, totalBytes, "bytes", eta, elapsedSeconds); } } } - if (showProgress && percent == 100 ) { + if (showProgress && percent == 100 && !finished) { + finished = true; out.println(""); } } - private void renderBar(int percent, long current, long total, String sizeUnitLabel,double eta) { + private void renderBar(int percent, long current, long total, String sizeUnitLabel, double eta, long elapsedSeconds) { int filled = (int) ((percent / 100.0) * barWidth); String bar = "[" + "=".repeat(filled) + " ".repeat(barWidth - filled) + "]"; - out.printf("\r Progress: %s %3d%% (%d/%d %s, ETA: %.1fs)", bar, percent, current, total, sizeUnitLabel, eta); + out.printf("\r Progress: %s %3d%% (%d/%d %s, ETA: %.1fs, Elapsed: %ds)", bar, percent, current, total, sizeUnitLabel, eta, elapsedSeconds); } } \ No newline at end of file diff --git a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTrackingBodyPublisher.java b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTrackingBodyPublisher.java index 45641b417..5b47d184d 100644 --- a/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTrackingBodyPublisher.java +++ b/src/main/java/io/seqera/tower/cli/utils/progress/ProgressTrackingBodyPublisher.java @@ -25,11 +25,11 @@ public class ProgressTrackingBodyPublisher implements HttpRequest.BodyPublisher { private final byte[] data; - private final ProgressTracker tracker; + private final ProgressSink sink; - public ProgressTrackingBodyPublisher(byte[] data, ProgressTracker tracker) { + public ProgressTrackingBodyPublisher(byte[] data, ProgressSink sink) { this.data = data; - this.tracker = tracker; + this.sink = sink; } @Override @@ -40,14 +40,14 @@ public long contentLength() { @Override public void subscribe(Flow.Subscriber subscriber) { // Wrap byte array in InputStream and monitor progress - InputStream input = new ProgressInputStream(new ByteArrayInputStream(data), tracker); + InputStream input = new ProgressInputStream(new ByteArrayInputStream(data), sink); subscriber.onSubscribe(new InputStreamSubscription(input, subscriber)); } private static class InputStreamSubscription implements Flow.Subscription { private final InputStream input; private final Flow.Subscriber subscriber; - private final int bufferSize = 8192; + private final int bufferSize = 256 * 1024; private boolean completed = false; public InputStreamSubscription(InputStream input, Flow.Subscriber subscriber) { 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 086d9c71f..44df10884 100644 --- a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java +++ b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import io.seqera.tower.cli.BaseCmdTest; import io.seqera.tower.cli.commands.data.links.ListCmd; +import io.seqera.tower.cli.commands.data.links.upload.AbstractProviderUploader; import io.seqera.tower.cli.commands.enums.OutputType; import io.seqera.tower.cli.exceptions.TowerRuntimeException; import io.seqera.tower.cli.responses.data.DataLinkDeleted; @@ -35,6 +36,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; @@ -801,6 +803,116 @@ void testUploadSingleFile(OutputType format, MockServerClient mock) throws IOExc Files.deleteIfExists(testFile); } + @ParameterizedTest + @EnumSource(value = OutputType.class, names = {"json"}) + void testUploadMultipartFileInParallel(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) + ); + // 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) + ); + + // 12-byte file which, with a 5-byte part size, becomes 3 parts: [0,5) [5,10) [10,12) + Path testFile = tempDir().resolve("test.txt"); + Files.write(testFile, "0123456789AB".getBytes()); + + // generate: return three positional part URLs (index+1 == partNumber) + 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-1\"," + + "\"http://localhost:" + mock.getPort() + "/upload-2\"," + + "\"http://localhost:" + mock.getPort() + "/upload-3\"]\n" + + "}").withContentType(MediaType.APPLICATION_JSON) + ); + + // Part 1 is delayed so it completes LAST — this proves the completed-parts list is ordered by + // part number, not by (nondeterministic) completion order. + mock.when(request().withMethod("PUT").withPath("/upload-1"), exactly(1)).respond( + response().withStatusCode(200).withHeader(new Header("Etag", "etag-1")) + .withDelay(java.util.concurrent.TimeUnit.MILLISECONDS, 500) + ); + mock.when(request().withMethod("PUT").withPath("/upload-2"), exactly(1)).respond( + response().withStatusCode(200).withHeader(new Header("Etag", "etag-2")) + ); + mock.when(request().withMethod("PUT").withPath("/upload-3"), exactly(1)).respond( + response().withStatusCode(200).withHeader(new Header("Etag", "etag-3")) + ); + + // finish + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200) + ); + + // Shrink the part size so the tiny file uploads as 3 parts; run them 3-way concurrently. + String previous = System.setProperty(AbstractProviderUploader.PART_SIZE_ENV, "5"); + ExecOut out; + try { + out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", + "-c", "57Ic6reczFn78H1DTaaXkp", "--concurrency", "3", testFile.toString()); + } finally { + if (previous == null) { + System.clearProperty(AbstractProviderUploader.PART_SIZE_ENV); + } else { + System.setProperty(AbstractProviderUploader.PART_SIZE_ENV, previous); + } + } + + assertOutput(format, out, DataLinkFileTransferResult.uploaded(List.of( + new DataLinkFileTransferResult.SimplePathInfo(DataLinkItemType.FILE, testFile.toString(), 1) + ))); + assertEquals("", out.stdErr); + assertEquals(0, out.exitCode); + + // Every part was uploaded exactly once... + mock.verify(request().withMethod("PUT").withPath("/upload-1"), VerificationTimes.exactly(1)); + mock.verify(request().withMethod("PUT").withPath("/upload-2"), VerificationTimes.exactly(1)); + mock.verify(request().withMethod("PUT").withPath("/upload-3"), VerificationTimes.exactly(1)); + // ...and finish received the tags in ascending part-number order (STRICT enforces array order), + // each part paired with the ETag from its own PUT response, despite part 1 completing last. + 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\":[{\"partNumber\":1,\"eTag\":\"etag-1\"},{\"partNumber\":2,\"eTag\":\"etag-2\"},{\"partNumber\":3,\"eTag\":\"etag-3\"}],\n" + + " \"withError\":false\n" + + "}\n", MatchType.STRICT)), VerificationTimes.exactly(1)); + + Files.deleteIfExists(testFile); + } + @ParameterizedTest @EnumSource(value = OutputType.class, names = {"json"}) void testUploadSingleFileFailsButStillFinalizesUpload(OutputType format, MockServerClient mock) throws IOException { diff --git a/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java b/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java new file mode 100644 index 000000000..535538328 --- /dev/null +++ b/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java @@ -0,0 +1,103 @@ +/* + * Copyright 2021-2026, Seqera. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.seqera.tower.cli.utils.progress; + +import org.junit.jupiter.api.Test; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class ProgressTrackerTest { + + private static ProgressTracker tracker(long totalBytes, PrintWriter out) { + return new ProgressTracker(out, true, totalBytes); + } + + @Test + void resetRollsBackOnlyThisPartsBytes() { + ProgressTracker t = tracker(100, new PrintWriter(new StringWriter())); + PartProgress part = t.newPart(); + + part.update(50); // failed attempt reports 50 of 100 + part.reset(); // roll it back + assertEquals(0, t.currentBytes()); + + part.update(100); // successful retry reports the full part + assertEquals(100, t.currentBytes()); + } + + @Test + void terminatingNewlinePrintedExactlyOnce() { + StringWriter sw = new StringWriter(); + PrintWriter out = new PrintWriter(sw); + ProgressTracker t = tracker(100, out); + PartProgress part = t.newPart(); + + part.update(100); // reach 100% -> newline + part.reset(); // dip back below 100% + part.update(100); // reach 100% again -> latch must suppress a second newline + out.flush(); + + long newlines = sw.toString().chars().filter(c -> c == '\n').count(); + assertEquals(1, newlines); + } + + @Test + void concurrentPartsWithRollbacksSettleAtTotal() throws Exception { + int parts = 16; + long bytesPerPart = 4096; + long total = parts * bytesPerPart; + + ProgressTracker t = tracker(total, new PrintWriter(new StringWriter())); + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < parts; i++) { + futures.add(pool.submit(() -> { + PartProgress part = t.newPart(); + // Simulate two failed attempts that each report a partial amount and roll back, + // then a successful attempt that reports the full part in small increments. + for (int attempt = 0; attempt < 2; attempt++) { + part.update(bytesPerPart / 2); + part.reset(); + } + for (long sent = 0; sent < bytesPerPart; sent += 512) { + part.update(512); + } + })); + } + for (Future f : futures) { + f.get(); + } + } finally { + pool.shutdownNow(); + pool.awaitTermination(10, TimeUnit.SECONDS); + } + + // Every part's rollbacks only ever undid its own bytes, so the shared total lands exactly + // on the sum of all committed parts — no cross-part corruption. + assertEquals(total, t.currentBytes()); + } +} From 0c4fc0e117062b43c2bac88c5a42310c34735b99 Mon Sep 17 00:00:00 2001 From: "georgi.hristov" Date: Tue, 4 Aug 2026 09:49:17 +0100 Subject: [PATCH 2/4] feat: render progress bar at every second, rather than every percent --- .../cli/utils/progress/ProgressTracker.java | 43 ++++++---- .../utils/progress/ProgressTrackerTest.java | 85 +++++++++++++++++++ 2 files changed, 113 insertions(+), 15 deletions(-) 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 77cf0d2ae..ab1ab0f20 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 @@ -17,22 +17,33 @@ package io.seqera.tower.cli.utils.progress; import java.io.PrintWriter; -import java.time.Instant; +import java.util.function.LongSupplier; public class ProgressTracker { + + /** Minimum wall-clock gap between two repaints of the progress bar. */ + private static final long RENDER_INTERVAL_MILLIS = 1000; + private final PrintWriter out; private final boolean showProgress; public final long totalBytes; private volatile long uploadedBytes = 0; - private volatile int lastPercent = -1; + private long lastRenderMillis = -1; // -1 = never rendered yet private boolean finished = false; private final int barWidth = 40; - private final Instant startTime = Instant.now(); + private final LongSupplier nowMillis; + private final long startMillis; public ProgressTracker(PrintWriter out, boolean showProgress, long totalBytes) { + this(out, showProgress, totalBytes, System::currentTimeMillis); + } + + ProgressTracker(PrintWriter out, boolean showProgress, long totalBytes, LongSupplier nowMillis) { this.out = out; this.showProgress = showProgress; this.totalBytes = totalBytes; + this.nowMillis = nowMillis; + this.startMillis = nowMillis.getAsLong(); } /** @@ -51,25 +62,27 @@ synchronized long currentBytes() { synchronized void update(long count) { uploadedBytes += count; - int percent = (int) ((uploadedBytes * 100) / totalBytes); - if (percent != lastPercent) { - lastPercent = percent; - long elapsedMillis = java.time.Duration.between(startTime, Instant.now()).toMillis(); + long elapsedMillis = nowMillis.getAsLong() - startMillis; + boolean complete = uploadedBytes >= totalBytes; + boolean due = lastRenderMillis < 0 || (elapsedMillis - lastRenderMillis) >= RENDER_INTERVAL_MILLIS; + + if (showProgress && !finished && (due || complete)) { + lastRenderMillis = elapsedMillis; + + int percent = (int) ((uploadedBytes * 100) / totalBytes); long elapsedSeconds = elapsedMillis / 1000; double speed = uploadedBytes / (elapsedMillis / 1000.0); double eta = (elapsedMillis <= 0 || !Double.isFinite(speed) || speed <= 0) ? 0.0 : (totalBytes - uploadedBytes) / speed; - if (showProgress) { - if (totalBytes > 1024) { - renderBar(percent, uploadedBytes / 1024, totalBytes / 1024, "KBs", eta, elapsedSeconds); - } - else { - renderBar(percent, uploadedBytes, totalBytes, "bytes", eta, elapsedSeconds); - } + if (totalBytes > 1024) { + renderBar(percent, uploadedBytes / 1024, totalBytes / 1024, "KBs", eta, elapsedSeconds); + } + else { + renderBar(percent, uploadedBytes, totalBytes, "bytes", eta, elapsedSeconds); } } - if (showProgress && percent == 100 && !finished) { + if (showProgress && complete && !finished) { finished = true; out.println(""); } diff --git a/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java b/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java index 535538328..40590e79f 100644 --- a/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java +++ b/src/test/java/io/seqera/tower/cli/utils/progress/ProgressTrackerTest.java @@ -26,15 +26,35 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; class ProgressTrackerTest { + /** A clock whose current time is set explicitly by the test. */ + private static final class FakeClock implements LongSupplier { + private final AtomicLong now = new AtomicLong(0); + void set(long millis) { now.set(millis); } + void advance(long millis) { now.addAndGet(millis); } + @Override public long getAsLong() { return now.get(); } + } + private static ProgressTracker tracker(long totalBytes, PrintWriter out) { return new ProgressTracker(out, true, totalBytes); } + private static long countRenders(StringWriter sw) { + Matcher m = Pattern.compile("Progress:").matcher(sw.toString()); + long count = 0; + while (m.find()) count++; + return count; + } + @Test void resetRollsBackOnlyThisPartsBytes() { ProgressTracker t = tracker(100, new PrintWriter(new StringWriter())); @@ -100,4 +120,69 @@ void concurrentPartsWithRollbacksSettleAtTotal() throws Exception { // on the sum of all committed parts — no cross-part corruption. assertEquals(total, t.currentBytes()); } + + @Test + void repaintsAtMostOncePerSecond() { + StringWriter sw = new StringWriter(); + FakeClock clock = new FakeClock(); + ProgressTracker t = new ProgressTracker(new PrintWriter(sw), true, 1000, clock); + PartProgress part = t.newPart(); + + // First update paints immediately (initial frame). + part.update(100); + assertEquals(1, countRenders(sw)); + + // More updates within the same second do not repaint. + clock.advance(500); + part.update(100); + clock.advance(499); + part.update(100); + assertEquals(1, countRenders(sw)); + + // Crossing the 1s boundary paints again. + clock.advance(1); // now 1000ms since the last paint + part.update(100); + assertEquals(2, countRenders(sw)); + } + + @Test + void finalFrameShows100PercentAndNewlineEvenWithinOneSecond() { + StringWriter sw = new StringWriter(); + FakeClock clock = new FakeClock(); + ProgressTracker t = new ProgressTracker(new PrintWriter(sw), true, 1000, clock); + PartProgress part = t.newPart(); + + // Complete the whole transfer inside a single sub-second window. + part.update(400); // first frame (immediate) + part.update(600); // reaches total within the same second -> forced final frame + + String out = sw.toString(); + assertTrue(out.contains("100%"), "final frame should show 100%"); + assertEquals(1, out.chars().filter(c -> c == '\n').count(), "exactly one terminating newline"); + } + + @Test + void elapsedSecondsAreNonDecreasingAcrossPaints() { + StringWriter sw = new StringWriter(); + FakeClock clock = new FakeClock(); + ProgressTracker t = new ProgressTracker(new PrintWriter(sw), true, 1000, clock); + PartProgress part = t.newPart(); + + part.update(100); // Elapsed: 0s + clock.advance(1000); + part.update(100); // Elapsed: 1s + clock.advance(2000); + part.update(100); // Elapsed: 3s + + Matcher m = Pattern.compile("Elapsed: (\\d+)s").matcher(sw.toString()); + long previous = -1; + int frames = 0; + while (m.find()) { + long elapsed = Long.parseLong(m.group(1)); + assertTrue(elapsed >= previous, "elapsed must be non-decreasing"); + previous = elapsed; + frames++; + } + assertEquals(3, frames); + } } From 35efb37e9a70b79a0c8e40590962c4235ad770de Mon Sep 17 00:00:00 2001 From: "georgi.hristov" Date: Thu, 13 Aug 2026 10:09:41 +0100 Subject: [PATCH 3/4] fix: make the partSize configurable only for tests + guard check against mismatch with platform on partSize --- .../upload/AbstractProviderUploader.java | 26 ++++-- .../data/links/upload/AwsUploader.java | 2 + .../data/links/upload/AzureUploader.java | 2 + .../tower/cli/data/DataLinksCmdTest.java | 91 ++++++++++++++++++- 4 files changed, 111 insertions(+), 10 deletions(-) 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 0e5ac2801..ee6d5fb9c 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 @@ -58,16 +58,14 @@ public abstract class AbstractProviderUploader implements CloudProviderUploader static final int DEFAULT_PART_SIZE_IN_BYTES = 250 * 1024 * 1024; // 250 MB /** - * Overrides the multipart part size (in bytes). + * Test-only override of the multipart part size, in bytes, so a test can drive a multi-part upload + * without producing a multi-hundred-megabyte fixture. */ - public static final String PART_SIZE_ENV = "TOWER_UPLOAD_SIZE_PART_BYTES"; + public static final String PART_SIZE_PROPERTY = "io.seqera.tower.cli.upload.partSizeBytes"; - /** Multipart upload part size in bytes; honors {@link #PART_SIZE_ENV}, defaulting to 250 MB. */ + /** Multipart upload part size in bytes; 250 MB unless overridden by {@link #PART_SIZE_PROPERTY} in tests. */ static int partSizeBytes() { - String value = System.getenv(PART_SIZE_ENV); - if (value == null) { - value = System.getProperty(PART_SIZE_ENV); - } + String value = System.getProperty(PART_SIZE_PROPERTY); if (value != null) { try { int parsed = Integer.parseInt(value.trim()); @@ -143,6 +141,20 @@ protected int totalParts(long contentLength) { return (int) Math.ceil((double) contentLength / partSizeBytes()); } + /** + * Guards against the CLI and Platform disagreeing on the part size. Part boundaries are computed + * locally from {@link #partSizeBytes()} while the number of parts comes from the presigned URLs + * Platform hands back; if the two disagree the file would be sliced into pieces that do not cover + * it and the upload would be finalized as a silently truncated object. + */ + protected void checkPartCount(long contentLength, int urlCount) { + int expected = totalParts(contentLength); + if (expected != urlCount) { + throw new TowerRuntimeException("Platform returned " + urlCount + " upload URLs but this file needs " + + expected + " parts of " + partSizeBytes() + " bytes; refusing to upload a truncated file."); + } + } + /** * Uploads a single part. * When {@code refreshable}, an expiry-class error causes the presigned URL (and a forward window of upcoming parts) 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 e80b6c99f..66cab95cc 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 @@ -56,6 +56,8 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P int totalParts = initialUrls.size(); try (HttpClient client = HttpClient.newHttpClient()) { + checkPartCount(contentLength, totalParts); + // Upload all parts concurrently; each returns its ETag keyed by part number. Map etags = uploadPartsInParallel(totalParts, partNumber -> { byte[] chunk = getChunk(file, partNumber - 1); 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 164f7e1a1..f0525dbf5 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 @@ -51,6 +51,8 @@ public void uploadFile(File file, DataLinkMultiPartUploadResponse urlResponse, P int totalParts = initialUrls.size(); try (HttpClient client = HttpClient.newHttpClient()) { + checkPartCount(contentLength, totalParts); + uploadPartsInParallel(totalParts, partNumber -> { byte[] chunk = getChunk(file, partNumber - 1); uploadPartWithRetry(client, partUrls, partNumber, chunk, tracker, null, contentLength, 201, false); 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 ed59e3c0a..ca4391b1b 100644 --- a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java +++ b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java @@ -875,16 +875,16 @@ void testUploadMultipartFileInParallel(OutputType format, MockServerClient mock) ); // Shrink the part size so the tiny file uploads as 3 parts; run them 3-way concurrently. - String previous = System.setProperty(AbstractProviderUploader.PART_SIZE_ENV, "5"); + String previous = System.setProperty(AbstractProviderUploader.PART_SIZE_PROPERTY, "5"); ExecOut out; try { out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", "-c", "57Ic6reczFn78H1DTaaXkp", "--concurrency", "3", testFile.toString()); } finally { if (previous == null) { - System.clearProperty(AbstractProviderUploader.PART_SIZE_ENV); + System.clearProperty(AbstractProviderUploader.PART_SIZE_PROPERTY); } else { - System.setProperty(AbstractProviderUploader.PART_SIZE_ENV, previous); + System.setProperty(AbstractProviderUploader.PART_SIZE_PROPERTY, previous); } } @@ -913,6 +913,91 @@ void testUploadMultipartFileInParallel(OutputType format, MockServerClient mock) Files.deleteIfExists(testFile); } + @ParameterizedTest + @EnumSource(value = OutputType.class, names = {"json"}) + void testUploadFailsWhenPartCountDisagreesWithPlatform(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) + ); + // 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) + ); + + // 12-byte file which, at the 5-byte part size below, needs 3 parts + Path testFile = tempDir().resolve("test.txt"); + Files.write(testFile, "0123456789AB".getBytes()); + + // ...but Platform sliced the file with a different part size and returns only 2 URLs. Uploading + // 2 x 5 bytes and finalizing would store a truncated object, so the CLI must refuse. + 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-1\"," + + "\"http://localhost:" + mock.getPort() + "/upload-2\"]\n" + + "}").withContentType(MediaType.APPLICATION_JSON) + ); + + // finish (abort) + mock.when(request() + .withMethod("POST").withPath("/data-links/v1-cloud-c2875f38a7b5c8fe34a5b382b5f9e0c4/upload/finish") + .withQueryStringParameter("workspaceId", "75887156211589") + .withQueryStringParameter("credentialsId", "57Ic6reczFn78H1DTaaXkp"), exactly(1) + ).respond( + response().withStatusCode(200) + ); + + String previous = System.setProperty(AbstractProviderUploader.PART_SIZE_PROPERTY, "5"); + ExecOut out; + try { + out = exec(format, mock, "data-links", "upload", "-w", "75887156211589", "-n", "a-test-bucket-eend-us-east-1", + "-c", "57Ic6reczFn78H1DTaaXkp", testFile.toString()); + } finally { + if (previous == null) { + System.clearProperty(AbstractProviderUploader.PART_SIZE_PROPERTY); + } else { + System.setProperty(AbstractProviderUploader.PART_SIZE_PROPERTY, previous); + } + } + + assertEquals(errorMessage(out.app, new TowerRuntimeException("Failed to upload file: Platform returned 2 upload URLs but this file needs 3 parts of 5 bytes; refusing to upload a truncated file.")), out.stdErr); + assertEquals("", out.stdOut); + assertEquals(1, out.exitCode); + + // Nothing was uploaded, and the multipart upload was aborted rather than committed + mock.verify(request().withMethod("PUT").withPath("/upload-1"), VerificationTimes.exactly(0)); + mock.verify(request().withMethod("PUT").withPath("/upload-2"), VerificationTimes.exactly(0)); + 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)); + + Files.deleteIfExists(testFile); + } + @ParameterizedTest @EnumSource(value = OutputType.class, names = {"json"}) void testUploadSingleFileFailsButStillFinalizesUpload(OutputType format, MockServerClient mock) throws IOException { From d674d089ab987f6a0f4239cb1c882efebed10ddc Mon Sep 17 00:00:00 2001 From: "georgi.hristov" Date: Fri, 14 Aug 2026 16:01:15 +0100 Subject: [PATCH 4/4] test: disable parallel multipart upload tests for binary tests; --- .../commands/data/links/upload/AbstractProviderUploader.java | 4 ++++ src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java | 5 +++++ 2 files changed, 9 insertions(+) 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 ee6d5fb9c..7c65c2ecc 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 @@ -60,6 +60,10 @@ public abstract class AbstractProviderUploader implements CloudProviderUploader /** * Test-only override of the multipart part size, in bytes, so a test can drive a multi-part upload * without producing a multi-hundred-megabyte fixture. + * + *

Only takes effect in-process: the binary test run executes the native image as a separate + * process, which does not inherit the test JVM's system properties, so tests relying on this must be + * disabled when {@code TOWER_CLI} is set. */ public static final String PART_SIZE_PROPERTY = "io.seqera.tower.cli.upload.partSizeBytes"; 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 ca4391b1b..d0c20df07 100644 --- a/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java +++ b/src/test/java/io/seqera/tower/cli/data/DataLinksCmdTest.java @@ -31,6 +31,7 @@ import io.seqera.tower.cli.utils.PaginationInfo; import io.seqera.tower.model.DataLinkDto; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.EnumSource; @@ -805,6 +806,8 @@ void testUploadSingleFile(OutputType format, MockServerClient mock) throws IOExc @ParameterizedTest @EnumSource(value = OutputType.class, names = {"json"}) + @DisabledIfEnvironmentVariable(named = "TOWER_CLI", matches = ".+", + disabledReason = "shrinks the part size through a system property, which cannot reach the native binary running as a separate process") void testUploadMultipartFileInParallel(OutputType format, MockServerClient mock) throws IOException { // credentials fetch mock.when( @@ -915,6 +918,8 @@ void testUploadMultipartFileInParallel(OutputType format, MockServerClient mock) @ParameterizedTest @EnumSource(value = OutputType.class, names = {"json"}) + @DisabledIfEnvironmentVariable(named = "TOWER_CLI", matches = ".+", + disabledReason = "shrinks the part size through a system property, which cannot reach the native binary running as a separate process") void testUploadFailsWhenPartCountDisagreesWithPlatform(OutputType format, MockServerClient mock) throws IOException { // credentials fetch mock.when(