Ensure that file modification exceptions in FileAsyncRequestBody are propogated correctly.

Author: alextwoods

.changes/next-release/bugfix-AWSSDKforJavav2-4bab915.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Ensure that file modification exceptions in FileAsyncRequestBody are propogated correctly."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBody.java

@@ -71,6 +71,8 @@ public final class FileAsyncRequestBody implements AsyncRequestBody {
     private final int chunkSizeInBytes;
     private final long position;
     private final long numBytesToRead;
+    private final FileTime modifiedTimeAtStart;
+    private final long sizeAtStart;
 
     private FileAsyncRequestBody(DefaultBuilder builder) {
         this.path = builder.path;
@@ -79,6 +81,22 @@ private FileAsyncRequestBody(DefaultBuilder builder) {
         this.position = builder.position == null ? 0 : Validate.isNotNegative(builder.position, "position");
         this.numBytesToRead = builder.numBytesToRead == null ? fileLength - this.position :
                               Validate.isNotNegative(builder.numBytesToRead, "numBytesToRead");
+        try {
+            if (builder.modifiedTimeAtStart != null) {
+                this.modifiedTimeAtStart = builder.modifiedTimeAtStart;
+            } else {
+                this.modifiedTimeAtStart = Files.getLastModifiedTime(path);
+            }
+
+            if (builder.sizeAtStart != null) {
+                this.sizeAtStart = builder.sizeAtStart;
+            } else {
+                this.sizeAtStart = Files.size(path);
+            }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
     }
 
     @Override
@@ -112,6 +130,14 @@ public long numBytesToRead() {
         return numBytesToRead;
     }
 
+    public FileTime modifiedTimeAtStart() {
+        return modifiedTimeAtStart;
+    }
+
+    public long sizeAtStart() {
+        return sizeAtStart;
+    }
+
     @Override
     public Optional<Long> contentLength() {
         return Optional.of(numBytesToRead);
@@ -131,7 +157,7 @@ public void subscribe(Subscriber<? super ByteBuffer> s) {
             // We need to synchronize here because the subscriber could call
             // request() from within onSubscribe which would potentially
             // trigger onNext before onSubscribe is finished.
-            Subscription subscription = new FileSubscription(channel, s);
+            Subscription subscription = new FileSubscription(channel, s, modifiedTimeAtStart, sizeAtStart);
 
             synchronized (subscription) {
                 s.onSubscribe(subscription);
@@ -203,6 +229,20 @@ public interface Builder extends SdkBuilder<Builder, FileAsyncRequestBody> {
          * @return The builder for method chaining.
          */
         Builder numBytesToRead(Long numBytesToRead);
+
+        /**
+         * Optional - sets the file modified time at the start of the request.
+         * @param modifiedTimeAtStart initial file modification time
+         * @return The builder for method chaining.
+         */
+        Builder modifiedTimeAtStart(FileTime modifiedTimeAtStart);
+
+        /**
+         * Optional - sets the file size in bytes at the start of the request.
+         * @param sizeAtStart initial file size at start.
+         * @return The builder for method chaining.
+         */
+        Builder sizeAtStart(Long sizeAtStart);
     }
 
     private static final class DefaultBuilder implements Builder {
@@ -211,6 +251,8 @@ private static final class DefaultBuilder implements Builder {
         private Path path;
         private Integer chunkSizeInBytes;
         private Long numBytesToRead;
+        private FileTime modifiedTimeAtStart;
+        private Long sizeAtStart;
 
         @Override
         public Builder path(Path path) {
@@ -240,6 +282,18 @@ public Builder numBytesToRead(Long numBytesToRead) {
             return this;
         }
 
+        @Override
+        public Builder modifiedTimeAtStart(FileTime modifiedTimeAtStart) {
+            this.modifiedTimeAtStart = modifiedTimeAtStart;
+            return this;
+        }
+
+        @Override
+        public Builder sizeAtStart(Long sizeAtStart) {
+            this.sizeAtStart = sizeAtStart;
+            return this;
+        }
+
         public void setChunkSizeInBytes(Integer chunkSizeInBytes) {
             chunkSizeInBytes(chunkSizeInBytes);
         }
@@ -267,11 +321,12 @@ private final class FileSubscription implements Subscription {
         private final Object lock = new Object();
 
         private FileSubscription(AsynchronousFileChannel inputChannel,
-                                 Subscriber<? super ByteBuffer> subscriber) throws IOException {
+                                 Subscriber<? super ByteBuffer> subscriber,
+                                 FileTime modifiedTimeAtStart, long sizeAtStart) throws IOException {
             this.inputChannel = inputChannel;
             this.subscriber = subscriber;
-            this.sizeAtStart = inputChannel.size();
-            this.modifiedTimeAtStart = Files.getLastModifiedTime(path);
+            this.sizeAtStart = sizeAtStart;
+            this.modifiedTimeAtStart = modifiedTimeAtStart;
             this.remainingBytes = new AtomicLong(numBytesToRead);
             this.currentPosition = new AtomicLong(position);
         }
@@ -338,12 +393,19 @@ public void completed(Integer result, ByteBuffer attachment) {
 
                             int readBytes = attachment.remaining();
                             currentPosition.addAndGet(readBytes);
-                            remainingBytes.addAndGet(-readBytes);
+                            long remaining = remainingBytes.addAndGet(-readBytes);
+
+                            // we need to validate the file is unchanged before providing the last bytes to subscriber
+                            // the subscriber (eg: NettyRequestExecutor) may cancel subscription once all expected bytes have
+                            // been received.  Validating here ensures errors are correctly signaled.
+                            if (remaining == 0) {
+                                closeFile();
+                                validateFileUnchanged();
+                            }
 
                             signalOnNext(attachment);
 
-                            if (remainingBytes.get() == 0) {
-                                closeFile();
+                            if (remaining == 0) {
                                 signalOnComplete();
                             }
 
@@ -391,42 +453,50 @@ private void signalOnNext(ByteBuffer attachment) {
         }
 
         private void signalOnComplete() {
+            if (!validateFileUnchanged()) {
+                return;
+            }
+
+            synchronized (this) {
+                if (!done) {
+                    done = true;
+                    subscriber.onComplete();
+                }
+            }
+        }
+
+        private boolean validateFileUnchanged() {
             try {
                 long sizeAtEnd = Files.size(path);
                 if (sizeAtStart != sizeAtEnd) {
-                    signalOnError(new IOException("File size changed after reading started. Initial size: " + sizeAtStart + ". "
-                                                  + "Current size: " + sizeAtEnd));
-                    return;
+                    signalOnError(new RuntimeException("File size changed after reading started. Initial size: "
+                                                       + sizeAtStart + ". Current size: " + sizeAtEnd));
+                    return false;
                 }
 
                 if (remainingBytes.get() > 0) {
-                    signalOnError(new IOException("Fewer bytes were read than were expected, was the file modified after "
+                    signalOnError(new RuntimeException("Fewer bytes were read than were expected, was the file modified after "
                                                   + "reading started?"));
-                    return;
+                    return false;
                 }
 
                 FileTime modifiedTimeAtEnd = Files.getLastModifiedTime(path);
                 if (modifiedTimeAtStart.compareTo(modifiedTimeAtEnd) != 0) {
-                    signalOnError(new IOException("File last-modified time changed after reading started. Initial modification "
-                                                  + "time: " + modifiedTimeAtStart + ". Current modification time: " +
-                                                  modifiedTimeAtEnd));
-                    return;
+                    signalOnError(
+                        new RuntimeException("File last-modified time changed after reading started. Initial modification "
+                                             + "time: " + modifiedTimeAtStart + ". Current modification time: " +
+                                             modifiedTimeAtEnd));
+                    return false;
                 }
             } catch (NoSuchFileException e) {
                 signalOnError(new IOException("Unable to check file status after read. Was the file deleted or were its "
                                               + "permissions changed?", e));
-                return;
+                return false;
             } catch (IOException e) {
                 signalOnError(new IOException("Unable to check file status after read.", e));
-                return;
-            }
-
-            synchronized (this) {
-                if (!done) {
-                    done = true;
-                    subscriber.onComplete();
-                }
+                return false;
             }
+            return true;
         }
 
         private void signalOnError(Throwable t) {

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodySplitHelper.java

@@ -17,6 +17,7 @@
 
 import java.nio.ByteBuffer;
 import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -53,6 +54,8 @@ public final class FileAsyncRequestBodySplitHelper {
 
     private AtomicInteger numAsyncRequestBodiesInFlight = new AtomicInteger(0);
     private AtomicInteger chunkIndex = new AtomicInteger(0);
+    private final FileTime modifiedTimeAtStart;
+    private final long sizeAtStart;
 
     public FileAsyncRequestBodySplitHelper(FileAsyncRequestBody asyncRequestBody,
                                            AsyncRequestBodySplitConfiguration splitConfiguration) {
@@ -70,6 +73,8 @@ public FileAsyncRequestBodySplitHelper(FileAsyncRequestBody asyncRequestBody,
                                splitConfiguration.bufferSizeInBytes();
         this.bufferPerAsyncRequestBody = Math.min(asyncRequestBody.chunkSizeInBytes(),
                                                   NumericUtils.saturatedCast(totalBufferSize));
+        this.modifiedTimeAtStart = asyncRequestBody.modifiedTimeAtStart();
+        this.sizeAtStart = asyncRequestBody.sizeAtStart();
     }
 
     public SdkPublisher<AsyncRequestBody> split() {
@@ -134,6 +139,8 @@ private AsyncRequestBody newFileAsyncRequestBody(SimplePublisher<AsyncRequestBod
                                                                         .position(position)
                                                                         .numBytesToRead(numBytesToReadForThisChunk)
                                                                         .chunkSizeInBytes(bufferPerAsyncRequestBody)
+                                                                        .modifiedTimeAtStart(modifiedTimeAtStart)
+                                                                        .sizeAtStart(sizeAtStart)
                                                                         .build();
         return new FileAsyncRequestBodyWrapper(fileAsyncRequestBody, simplePublisher);
     }