feat. Async pause (#1008)

Author: TingDaoK

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  BUILDER_VERSION: v0.9.96
+  BUILDER_VERSION: latest
   BUILDER_SOURCE: releases
   BUILDER_HOST: https://d19elf31gohf1l.cloudfront.net
   PACKAGE_NAME: aws-crt-java

crt/aws-c-common

@@ -64,6 +64,17 @@ public ResumeToken build() {
     private long numPartsCompleted;
     private String uploadId;
 
+    /* Download specific fields, populated by native code. */
+    private String etag;
+    private String versionId;
+    private String s3ObjectLastModified;
+    private long objectSize;
+    private long objectRangeStart;
+    private long objectRangeEnd;
+    private long continuesDownloadedBytes;
+    private long totalDownloadedBytes;
+    private long fileLastModifiedEpochNs;
+
     public ResumeToken(PutResumeTokenBuilder builder) {
         this.nativeType = S3MetaRequestOptions.MetaRequestType.PUT_OBJECT.getNativeValue();
         this.partSize = builder.partSize;
@@ -122,4 +133,122 @@ public String getUploadId() {
 
         return uploadId;
     }
+
+    /******
+     * Download Specific fields.
+     ******/
+
+    private void validateDownloadToken(String field) {
+        if (getType() != S3MetaRequestOptions.MetaRequestType.GET_OBJECT) {
+            throw new IllegalArgumentException(
+                    "ResumeToken - " + field + " is only defined for Get Object Resume tokens");
+        }
+    }
+
+    /**
+     * ETag of the S3 object being downloaded, captured from the first response.
+     * May be null/empty if the download was paused before the first response arrived.
+     *
+     * @return etag of the object
+     */
+    public String getEtag() {
+        validateDownloadToken("etag");
+        return etag;
+    }
+
+    /**
+     * Version ID of the S3 object being downloaded.
+     * Optional: null/empty when the bucket is unversioned or the version id was not captured.
+     *
+     * @return version id of the object
+     */
+    public String getVersionId() {
+        validateDownloadToken("version id");
+        return versionId;
+    }
+
+    /**
+     * Last-Modified of the S3 object being downloaded, in HTTP-date format
+     * (RFC 9110 5.6.7, "Wed, 09 Oct 2024 22:28:00 GMT"), captured from the first
+     * response. The exact string in the response header.
+     * Optional: null/empty if the value was not captured before the pause.
+     *
+     * @return Last-Modified of the object as an HTTP-date string
+     */
+    public String getS3ObjectLastModified() {
+        validateDownloadToken("s3 object last modified");
+        return s3ObjectLastModified;
+    }
+
+    /**
+     * Total size of the S3 object being downloaded, regardless of any Range header
+     * on the request (for a ranged download this is larger than the range being
+     * fetched). 0 if the download was paused before the object size was discovered.
+     *
+     * @return total object size in bytes
+     */
+    public long getObjectSize() {
+        validateDownloadToken("object size");
+        return objectSize;
+    }
+
+    /**
+     * Absolute byte offset in the object where the download's range starts.
+     * 0 for a download without a Range header.
+     *
+     * @return range start offset in bytes
+     */
+    public long getObjectRangeStart() {
+        validateDownloadToken("object range start");
+        return objectRangeStart;
+    }
+
+    /**
+     * Absolute byte offset in the object where the download's range ends (inclusive).
+     * For a download without a Range header this is object size - 1.
+     * 0 if the download was paused before the object size was discovered.
+     *
+     * @return range end offset in bytes (inclusive)
+     */
+    public long getObjectRangeEnd() {
+        validateDownloadToken("object range end");
+        return objectRangeEnd;
+    }
+
+    /**
+     * Number of bytes downloaded continuously from the start of the range, with no
+     * gaps. Everything before this offset (relative to the object range start) has
+     * been downloaded.
+     *
+     * @return continuously downloaded bytes
+     */
+    public long getContinuesDownloadedBytes() {
+        validateDownloadToken("continues downloaded bytes");
+        return continuesDownloadedBytes;
+    }
+
+    /**
+     * Total number of bytes downloaded before the pause. May be greater than
+     * {@link #getContinuesDownloadedBytes()} when parts completed out of order,
+     * leaving gaps. Equals it when delivery was strictly in order.
+     *
+     * @return total downloaded bytes
+     */
+    public long getTotalDownloadedBytes() {
+        validateDownloadToken("total downloaded bytes");
+        return totalDownloadedBytes;
+    }
+
+    /**
+     * Last-modified time of the local receive file (nanoseconds since the Unix
+     * epoch), captured after the file handle was closed during the pause.
+     * Only set when the download was writing to a file; 0 for downloads that
+     * deliver via body callback or when the timestamp could not be queried.
+     *
+     * @return local receive file last-modified time in epoch nanoseconds
+     */
+    public long getFileLastModifiedEpochNs() {
+        validateDownloadToken("file last modified epoch ns");
+        return fileLastModifiedEpochNs;
+    }
 }

crt/aws-c-mqtt

@@ -5,7 +5,9 @@
 package software.amazon.awssdk.crt.s3;
 
 import java.util.concurrent.CompletableFuture;
+import software.amazon.awssdk.crt.CRT;
 import software.amazon.awssdk.crt.CrtResource;
+import software.amazon.awssdk.crt.CrtRuntimeException;
 
 public class S3MetaRequest extends CrtResource {
 
@@ -21,6 +23,23 @@ private void onShutdownComplete() {
         this.shutdownComplete.complete(null);
     }
 
+    /**
+     * Called from native when an async pause completes. The resume token is null when no
+     * resumable state was captured, which is not an error.
+     *
+     * @param future the future to complete with the pause result
+     * @param errorCode 0 on success, otherwise the CRT error code that made the pause fail
+     * @param resumeToken the resume token, or null if no resumable state was captured
+     */
+    private static void onPauseComplete(
+            CompletableFuture<ResumeToken> future, int errorCode, ResumeToken resumeToken) {
+        if (errorCode != CRT.AWS_CRT_SUCCESS) {
+            future.completeExceptionally(new CrtRuntimeException(errorCode));
+            return;
+        }
+        future.complete(resumeToken);
+    }
+
     /**
      * Determines whether a resource releases its dependencies at the same time the
      * native handle is released or if it waits. Resources that wait are responsible
@@ -75,6 +94,38 @@ public ResumeToken pause() {
         return s3MetaRequestPause(getNativeHandle());
     }
 
+    /**
+     * Asynchronously pause the meta request. Works for both uploads (PUT) and downloads (GET).
+     * The returned future completes once all in-flight work has finished (in-flight parts for
+     * uploads, file writes for downloads) and the resume token is ready.
+     * <p>
+     * For PutObject resume, input stream should always start at the beginning,
+     * already uploaded parts will be skipped, but checksums on those will be verified if
+     * the request specified a checksum algorithm.
+     * <p>
+     * Note: consuming a download (GET) resume token to resume via meta request options is not
+     * supported yet. To resume a download, issue a new ranged GET starting at
+     * {@link ResumeToken#getContinuesDownloadedBytes()} (offset from
+     * {@link ResumeToken#getObjectRangeStart()}) through the end of the original download.
+     *
+     * @return future completed with the resume token once the pause completes. The token may be
+     *         null if the request had not progressed far enough to produce one (equivalent to
+     *         restarting the transfer). Completed exceptionally with a CrtRuntimeException if
+     *         the pause failed.
+     */
+    public CompletableFuture<ResumeToken> pauseAsync() {
+        if (isNull()) {
+            throw new IllegalStateException("S3MetaRequest has been closed.");
+        }
+        CompletableFuture<ResumeToken> future = new CompletableFuture<>();
+        try {
+            s3MetaRequestPauseAsync(getNativeHandle(), future);
+        } catch (Exception e) {
+            future.completeExceptionally(e);
+        }
+        return future;
+    }
+
     /**
      * Increment the flow-control window, so that response data continues downloading.
      * <p>
@@ -114,5 +165,7 @@ public void incrementReadWindow(long bytes) {
 
     private static native ResumeToken s3MetaRequestPause(long s3MetaRequest);
 
+    private static native void s3MetaRequestPauseAsync(long s3MetaRequest, CompletableFuture<ResumeToken> future);
+
     private static native void s3MetaRequestIncrementReadWindow(long s3MetaRequest, long bytes);
 }

crt/aws-c-s3

@@ -79,4 +79,25 @@ default void onProgress(final S3MetaRequestProgress progress) {
      */
     default void onTelemetry(S3RequestMetrics requestMetrics) {
     }
+
+    /**
+     * Invoked with a resume token when the meta request fails unexpectedly.
+     * Allows persisting state for later resume without re-transferring completed parts.
+     * Supported for both upload (PUT) and download (GET) meta requests.
+     * Always invoked exactly once on unexpected failure; not invoked on success or
+     * explicit pause (use {@link S3MetaRequest#pauseAsync} for that).
+     * The token is null when no resumable state was captured.
+     * <p>
+     * WARNING: for a file download with
+     * {@link S3MetaRequestOptions#withResponseFileDeleteOnFailure} set true, the deletion
+     * is respected — the partial file is deleted on error, leaving nothing to resume on,
+     * and this callback fires with a null token. Do not set responseFileDeleteOnFailure
+     * if you intend to resume from this callback's token.
+     *
+     * @param errorCode the CRT error code that caused the meta request to fail
+     * @param resumeToken resumable state captured at failure time, null if no
+     *        resumable state was captured
+     */
+    default void onErrorResumeToken(final int errorCode, final ResumeToken resumeToken) {
+    }
 }

crt/aws-c-sdkutils

@@ -36,4 +36,8 @@ void onProgress(final S3MetaRequestProgress progress) {
     void onTelemetry(final S3RequestMetrics requestMetrics) {
         responseHandler.onTelemetry(requestMetrics);
     }
+
+    void onErrorResumeToken(final int errorCode, final ResumeToken resumeToken) {
+        responseHandler.onErrorResumeToken(errorCode, resumeToken);
+    }
 }

src/main/java/software/amazon/awssdk/crt/s3/ResumeToken.java

@@ -2124,20 +2124,47 @@
   {
     "name": "software.amazon.awssdk.crt.s3.ResumeToken",
     "fields": [
+      {
+        "name": "continuesDownloadedBytes"
+      },
+      {
+        "name": "etag"
+      },
+      {
+        "name": "fileLastModifiedEpochNs"
+      },
       {
         "name": "nativeType"
       },
       {
         "name": "numPartsCompleted"
       },
+      {
+        "name": "objectRangeEnd"
+      },
+      {
+        "name": "objectRangeStart"
+      },
+      {
+        "name": "objectSize"
+      },
       {
         "name": "partSize"
       },
+      {
+        "name": "s3ObjectLastModified"
+      },
+      {
+        "name": "totalDownloadedBytes"
+      },
       {
         "name": "totalNumParts"
       },
       {
         "name": "uploadId"
+      },
+      {
+        "name": "versionId"
       }
     ],
     "methods": [
@@ -2204,6 +2231,14 @@
   {
     "name": "software.amazon.awssdk.crt.s3.S3MetaRequest",
     "methods": [
+      {
+        "name": "onPauseComplete",
+        "parameterTypes": [
+          "java.util.concurrent.CompletableFuture",
+          "int",
+          "software.amazon.awssdk.crt.s3.ResumeToken"
+        ]
+      },
       {
         "name": "onShutdownComplete",
         "parameterTypes": []
@@ -2364,6 +2399,13 @@
   {
     "name": "software.amazon.awssdk.crt.s3.S3MetaRequestResponseHandlerNativeAdapter",
     "methods": [
+      {
+        "name": "onErrorResumeToken",
+        "parameterTypes": [
+          "int",
+          "software.amazon.awssdk.crt.s3.ResumeToken"
+        ]
+      },
       {
         "name": "onFinished",
         "parameterTypes": [