fix: restart TM resume of serialized part tokens

.changes/next-release/bugfix-S3TransferManager-0ea8067.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "S3 Transfer Manager",
+    "contributor": "",
+    "description": "Fixed `S3TransferManager#resumeDownloadFile` so that a download paused with the Java-based multipart S3 async client and resumed from a serialized `ResumableFileDownload` restarts from the beginning if any parts had completed."
+}

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/CrtS3TransferManager.java

@@ -21,8 +21,8 @@
 import static software.amazon.awssdk.services.s3.internal.crt.DefaultS3CrtAsyncClient.RESPONSE_FILE_OPTION;
 import static software.amazon.awssdk.services.s3.internal.crt.DefaultS3CrtAsyncClient.RESPONSE_FILE_PATH;
 import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.CRT_PAUSE_RESUME_TOKEN;
-import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.canResumeDownload;
 import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.hasUnresumableRange;
+import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.isObjectOrFileModified;
 import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.toCrtDownloadFileRequest;
 
 import java.util.concurrent.CompletableFuture;
@@ -195,7 +195,7 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
         CompletableFutureUtils.forwardExceptionTo(returnFuture, headFuture);
 
         headFuture.thenAccept(headObjectResponse -> {
-            boolean restartFromBeginning = !canResumeDownload(resumableFileDownload, headObjectResponse)
+            boolean restartFromBeginning = isObjectOrFileModified(resumableFileDownload, headObjectResponse)
                                            || hasCompletedParts(resumableFileDownload)
                                            || hasUnresumableRange(originalDownloadRequest,
                                                                   resumableFileDownload.bytesTransferred(),
@@ -236,15 +236,17 @@ FileDownload doResumeDownloadFile(ResumableFileDownload resumableFileDownload) {
     /**
      * A download that was paused while fetching individual parts can leave gaps in the destination file, because parts are
      * written at their own offsets rather than in order. The CRT-based client only ever appends to the end of the file, so
-     * such a download cannot be continued and has to start over. This only arises when a {@link ResumableFileDownload}
-     * produced by the Java-based transfer manager is resumed with the CRT-based one.
+     * such a download cannot be continued and has to start over. Completed parts come from the Java-based transfer manager,
+     * either directly or carried over by a CRT pause that has not yet written to the file. The Java-based transfer manager
+     * restarts in the same way when the token no longer has the state needed to continue the parts, for example after the
+     * token was serialized.
      */
     private static boolean hasCompletedParts(ResumableFileDownload resumableFileDownload) {
         if (resumableFileDownload.completedParts().isEmpty()) {
             return false;
         }
-        log.debug(() -> "The paused download had completed individual parts, which the CRT-based S3 client cannot continue "
-                        + "from. The SDK will download the S3 object from the beginning.");
+        log.debug(() -> "The paused download had completed individual parts, so the destination file may have gaps between "
+                        + "them. The SDK will download the S3 object from the beginning.");
         return true;
     }
 

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/model/CrtFileDownload.java

@@ -18,6 +18,8 @@
 import java.io.File;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -30,6 +32,7 @@
 import software.amazon.awssdk.crt.s3.S3MetaRequestOptions;
 import software.amazon.awssdk.services.s3.internal.crt.S3MetaRequestPauseObservable;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter;
 import software.amazon.awssdk.transfer.s3.model.CompletedFileDownload;
 import software.amazon.awssdk.transfer.s3.model.DownloadFileRequest;
 import software.amazon.awssdk.transfer.s3.model.FileDownload;
@@ -150,7 +153,8 @@ private ResumableFileDownload resumableFileDownload(ResumeToken token) {
         ResumableFileDownload.Builder builder =
             ResumableFileDownload.builder()
                                  .downloadFileRequest(request)
-                                 .fileLastModified(Instant.ofEpochMilli(destination.lastModified()));
+                                 .fileLastModified(Instant.ofEpochMilli(destination.lastModified()))
+                                 .completedParts(completedPartsToSaveOnPause(destination));
 
         if (isDownloadResumeToken(token)) {
             // bytesTransferred must equal the total bytes on disk.
@@ -201,6 +205,19 @@ private static boolean isDownloadResumeToken(ResumeToken token) {
         return token != null && token.getType() == S3MetaRequestOptions.MetaRequestType.GET_OBJECT;
     }
 
+    /**
+     * The completed parts to save in the pause token. CRT never records completed parts itself, but the token being resumed
+     * can have them: a download paused with the Java-based multipart client, whose parts were written at their own offsets,
+     * so the file may have gaps between them. If this CRT download has not written to the file yet, those parts still
+     * describe the file on disk and are kept, so that the next resume restarts from the beginning instead of continuing
+     * from the end of a file with gaps. Once CRT has written to the file, it holds only what CRT wrote, in order, and no
+     * parts are kept.
+     */
+    private List<Integer> completedPartsToSaveOnPause(File destination) {
+        return ResumableRequestConverter.completedPartsToSaveOnPause(Collections.emptyList(), resumedDownload,
+                                                                     destination.toPath());
+    }
+
     private static Instant s3ObjectLastModified(ResumeToken token) {
         String lastModified = emptyToNull(token.getS3ObjectLastModified());
         if (lastModified == null) {

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/model/DefaultFileDownload.java

@@ -15,6 +15,8 @@
 
 package software.amazon.awssdk.transfer.s3.internal.model;
 
+import static software.amazon.awssdk.transfer.s3.internal.utils.ResumableRequestConverter.completedPartsToSaveOnPause;
+
 import java.io.File;
 import java.time.Instant;
 import java.util.List;
@@ -86,7 +88,9 @@ private ResumableFileDownload doPause() {
         File destination = request.destination().toFile();
         long length = destination.length();
         Instant fileLastModified = Instant.ofEpochMilli(destination.lastModified());
-        List<Integer> completedParts = MultipartDownloadUtils.completedParts(request.getObjectRequest());
+        List<Integer> completedParts =
+            completedPartsToSaveOnPause(MultipartDownloadUtils.completedParts(request.getObjectRequest()), resumedDownload,
+                                        destination.toPath());
         return ResumableFileDownload.builder()
                                     .downloadFileRequest(request)
                                     .s3ObjectLastModified(s3objectLastModified)

services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/utils/ResumableRequestConverter.java

@@ -20,7 +20,9 @@
 import static software.amazon.awssdk.services.s3.multipart.S3MultipartExecutionAttribute.MULTIPART_DOWNLOAD_RESUME_CONTEXT;
 import static software.amazon.awssdk.transfer.s3.internal.utils.FileUtils.fileNotModified;
 
+import java.nio.file.Path;
 import java.time.Instant;
+import java.util.List;
 import java.util.Optional;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
@@ -48,7 +50,8 @@ private ResumableRequestConverter() {
      * Converts a {@link ResumableFileDownload} to {@link DownloadFileRequest} and {@link AsyncResponseTransformer} pair.
      * <p>
      * If before resuming the download the file on disk was modified, or the s3 object was modified, we need to restart the
-     * download from the beginning.
+     * download from the beginning. The same applies if individual parts had completed but the multipart download state that
+     * records them is not available, for example because the {@link ResumableFileDownload} was serialized.
      * <p>
      * If the original requests has some individual parts downloaded, we need to make a multipart GET for the remaining parts.
      * <p>
@@ -62,10 +65,7 @@ private ResumableRequestConverter() {
         GetObjectRequest getObjectRequest = originalDownloadRequest.getObjectRequest();
         DownloadFileRequest newDownloadFileRequest;
 
-        // partNumber and Range cannot coexist, S3 rejects with 400 InvalidRequest.
-        // When partNumber is set, restart the part download from the beginning with CREATE_OR_REPLACE
-        if (getObjectRequest.partNumber() != null
-            || !canResumeDownload(resumableFileDownload, headObjectResponse)) {
+        if (!canResumeDownload(resumableFileDownload, headObjectResponse, getObjectRequest)) {
             newDownloadFileRequest = newDownloadFileRequest(originalDownloadRequest, getObjectRequest, headObjectResponse);
 
             AsyncResponseTransformer<GetObjectResponse, GetObjectResponse> responseTransformer =
@@ -85,11 +85,26 @@ private ResumableRequestConverter() {
             return Pair.of(attachIfMatch(originalDownloadRequest, resumableFileDownload), responseTransformer);
         }
 
+        return resumeWithRangeGetIfPossible(resumableFileDownload, headObjectResponse, originalDownloadRequest);
+    }
+
+    /**
+     * Resumes the paused download by fetching the remaining bytes with one range GET, appended to the destination file. That
+     * is only correct when the paused download wrote the file in order: a single GET (a non-multipart client, or a user
+     * range), or a multipart download paused before its first part completed. Tokens with completed parts are handled
+     * before this is called. If a user-specified range cannot be parsed or has already been fully downloaded, the download
+     * restarts from the beginning instead.
+     */
+    private static Pair<DownloadFileRequest, AsyncResponseTransformer<GetObjectResponse, GetObjectResponse>>
+        resumeWithRangeGetIfPossible(ResumableFileDownload resumableFileDownload,
+                           HeadObjectResponse headObjectResponse,
+                           DownloadFileRequest originalDownloadRequest) {
+        GetObjectRequest getObjectRequest = originalDownloadRequest.getObjectRequest();
         log.debug(() -> "The paused download was performed with range GET, now resuming download of remaining bytes.");
-        newDownloadFileRequest = resumedDownloadFileRequest(resumableFileDownload,
-                                                            originalDownloadRequest,
-                                                            getObjectRequest,
-                                                            headObjectResponse);
+        DownloadFileRequest newDownloadFileRequest = resumedDownloadFileRequest(resumableFileDownload,
+                                                                                originalDownloadRequest,
+                                                                                getObjectRequest,
+                                                                                headObjectResponse);
         if (newDownloadFileRequest == null) {
             log.debug(() -> "Could not compute resumed range for the original range '"
                             + getObjectRequest.range()
@@ -107,13 +122,26 @@ private ResumableRequestConverter() {
 
     /**
      * Determines whether a paused download can continue from where it left off, or whether it has to be restarted from the
-     * beginning because either the S3 object or the local file was modified while the download was paused. Logs the reason at
-     * debug level when a restart is required.
+     * beginning.
+     */
+    private static boolean canResumeDownload(ResumableFileDownload resumableFileDownload,
+                                             HeadObjectResponse headObjectResponse,
+                                             GetObjectRequest getObjectRequest) {
+        // partNumber and Range cannot coexist, S3 rejects with 400 InvalidRequest, so a part download is restarted with
+        // CREATE_OR_REPLACE.
+        return getObjectRequest.partNumber() == null
+               && !isObjectOrFileModified(resumableFileDownload, headObjectResponse)
+               && !completedPartsMissingFromResumeContext(resumableFileDownload, getObjectRequest);
+    }
+
+    /**
+     * Determines whether the S3 object or the local file was modified while the download was paused, in which case the
+     * download has to be restarted from the beginning. Logs the reason at debug level.
      *
-     * @return true if the remaining bytes can be fetched, false if the whole object must be downloaded again
+     * @return true if the whole object must be downloaded again, false if the remaining bytes can be fetched
      */
-    public static boolean canResumeDownload(ResumableFileDownload resumableFileDownload,
-                                            HeadObjectResponse headObjectResponse) {
+    public static boolean isObjectOrFileModified(ResumableFileDownload resumableFileDownload,
+                                                 HeadObjectResponse headObjectResponse) {
         DownloadFileRequest downloadRequest = resumableFileDownload.downloadFileRequest();
         Instant lastModified = resumableFileDownload.s3ObjectLastModified().orElse(null);
         String resumableFileDownloadEtag = resumableFileDownload.s3ObjectEtag().orElse(null);
@@ -129,9 +157,9 @@ public static boolean canResumeDownload(ResumableFileDownload resumableFileDownl
 
         if (fileModified || s3ObjectModified || etagModified) {
             logIfNeeded(downloadRequest, downloadRequest.getObjectRequest(), fileModified, s3ObjectModified, etagModified);
-            return false;
+            return true;
         }
-        return true;
+        return false;
     }
 
     /**
@@ -181,6 +209,48 @@ private static DownloadFileRequest attachIfMatch(DownloadFileRequest downloadReq
                               .build();
     }
 
+    /**
+     * Returns true if the {@link ResumableFileDownload} lists completed parts but the {@link MultipartDownloadResumeContext}
+     * on its request does not record them, because the token was serialized or a restarted download was paused before
+     * completing a part. The paused download fetched parts that are written at their own offsets, so the file may have gaps,
+     * and the download restarts from the beginning.
+     */
+    private static boolean completedPartsMissingFromResumeContext(ResumableFileDownload resumableFileDownload,
+                                                                  GetObjectRequest getObjectRequest) {
+        boolean tokenHasCompletedParts = !resumableFileDownload.completedParts().isEmpty();
+        boolean resumeContextHasCompletedParts =
+            MultipartDownloadUtils.multipartDownloadResumeContext(getObjectRequest)
+                                  .map(ctx -> !ctx.completedParts().isEmpty())
+                                  .orElse(false);
+        if (!tokenHasCompletedParts || resumeContextHasCompletedParts) {
+            return false;
+        }
+        log.debug(() -> "The paused download had completed individual parts, but the resume context that records them is "
+                        + "not available, for example because the ResumableFileDownload was serialized. The SDK will download "
+                        + "the S3 object from the beginning.");
+        return true;
+    }
+
+    /**
+     * The completed parts to save in the token when a resumed download is paused again. If the resumed download has not
+     * recorded any parts yet and the destination file is unchanged since {@code resumedDownload}, the file still holds what
+     * that token described, including any gaps between its completed parts, so its completed parts are kept. Without them,
+     * the next resume would continue from the end of the file and skip the gaps; with them,
+     * {@link #completedPartsMissingFromResumeContext} turns the next resume into a restart.
+     *
+     * @param recordedCompletedParts the parts the paused download itself recorded
+     * @param resumedDownload the token this download was resumed from, or null if it was not a resumed download
+     */
+    public static List<Integer> completedPartsToSaveOnPause(List<Integer> recordedCompletedParts,
+                                                            ResumableFileDownload resumedDownload,
+                                                            Path destination) {
+        if (!recordedCompletedParts.isEmpty() || resumedDownload == null
+            || !fileNotModified(resumedDownload.bytesTransferred(), resumedDownload.fileLastModified(), destination)) {
+            return recordedCompletedParts;
+        }
+        return resumedDownload.completedParts();
+    }
+
     private static boolean hasRemainingParts(GetObjectRequest getObjectRequest) {
         Optional<MultipartDownloadResumeContext> optCtx = MultipartDownloadUtils.multipartDownloadResumeContext(getObjectRequest);
         if (!optCtx.isPresent()) {
@@ -264,20 +334,12 @@ private static DownloadFileRequest resumedDownloadFileRequest(ResumableFileDownl
     }
 
     /**
-     * Returns {@code true} if the original request has a range that cannot be safely resumed.
+     * Returns {@code true} if the original request has a range that cannot be safely resumed, because it cannot be parsed or
+     * has already been fully downloaded, so the download has to restart from the beginning.
      */
     public static boolean hasUnresumableRange(DownloadFileRequest downloadRequest, long bytesTransferred,
                                               long contentLength) {
-        String range = downloadRequest.getObjectRequest().range();
-        if (range == null) {
-            return false;
-        }
-        long[] parsed = parseRange(range, contentLength);
-        if (parsed == null) {
-            return true;
-        }
-        long resumedStart = parsed[0] + bytesTransferred;
-        return resumedStart > parsed[1];
+        return computeResumedRange(downloadRequest.getObjectRequest().range(), bytesTransferred, contentLength) == null;
     }
 
     /**