Verify the file size does not change while AsyncRequestBody.fromFile's publisher is executing. (aws#2796)

millems · web-flow · commit 4f812748e8ae · 2021-10-26T11:03:43.000-07:00
This adds validation that the file size at the start of reading the file and
at the end of reading the file are equal, as well as that we actually read the amount of data
from the file that we expected to read.

This will hopefully reduce some instances of checksum mismatched exceptions by replacing it with
a more useful message.

In the future, we might consider adding functionality for the customer to opt-in to locking the
file for even more safety. That's not done as part of this PR, because it requires new public APIs
and there hasn't been any customer demand for such a thing.
diff --git a/.changes/next-release/feature-AWSSDKforJavav2-65c9fff.json b/.changes/next-release/feature-AWSSDKforJavav2-65c9fff.json
@@ -0,0 +1,6 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "contributor": "", 
+    "type": "feature", 
+    "description": "Provide an error message if a AsyncRequestBody.fromFile source file changes length or update time while the SDK is reading from the file."
+}
diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBody.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBody.java
@@ -23,8 +23,10 @@
 import java.nio.channels.AsynchronousFileChannel;
 import java.nio.channels.CompletionHandler;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.FileTime;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
 import org.reactivestreams.Subscriber;
@@ -90,13 +92,12 @@ 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.
-            //
-            // Note: size() can throw IOE here
-            Subscription subscription = new FileSubscription(channel, channel.size(), s, chunkSizeInBytes);
+            Subscription subscription = new FileSubscription(path, channel, s, chunkSizeInBytes);
+
             synchronized (subscription) {
                 s.onSubscribe(subscription);
             }
-        } catch (IOException e) {
+        } catch (IOException | RuntimeException e) {
             if (channel != null) {
                 runAndLogError(log.logger(), "Unable to close file channel", channel::close);
             }
@@ -176,25 +177,31 @@ public FileAsyncRequestBody build() {
      * Reads the file for one subscriber.
      */
     private static final class FileSubscription implements Subscription {
+        private final Path path;
         private final AsynchronousFileChannel inputChannel;
         private final Subscriber<? super ByteBuffer> subscriber;
         private final int chunkSize;
 
         private final AtomicLong position = new AtomicLong(0);
         private final AtomicLong remainingBytes = new AtomicLong(0);
+        private final long sizeAtStart;
+        private final FileTime modifiedTimeAtStart;
         private long outstandingDemand = 0;
         private boolean readInProgress = false;
         private volatile boolean done = false;
         private final Object lock = new Object();
 
-        private FileSubscription(AsynchronousFileChannel inputChannel,
-                                 long size,
+        private FileSubscription(Path path,
+                                 AsynchronousFileChannel inputChannel,
                                  Subscriber<? super ByteBuffer> subscriber,
-                                 int chunkSize) {
+                                 int chunkSize) throws IOException {
+            this.path = path;
             this.inputChannel = inputChannel;
             this.subscriber = subscriber;
             this.chunkSize = chunkSize;
-            this.remainingBytes.set(Validate.isNotNegative(size, "size"));
+            this.sizeAtStart = inputChannel.size();
+            this.modifiedTimeAtStart = Files.getLastModifiedTime(path);
+            this.remainingBytes.set(Validate.isNotNegative(sizeAtStart, "size"));
         }
 
         @Override
@@ -307,6 +314,36 @@ private void signalOnNext(ByteBuffer attachment) {
         }
 
         private void signalOnComplete() {
+            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;
+                }
+
+                if (remainingBytes.get() > 0) {
+                    signalOnError(new IOException("Fewer bytes were read than were expected, was the file modified after "
+                                                  + "reading started?"));
+                    return;
+                }
+
+                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;
+                }
+            } catch (NoSuchFileException e) {
+                signalOnError(new IOException("Unable to check file status after read. Was the file deleted or were its "
+                                              + "permissions changed?", e));
+                return;
+            } catch (IOException e) {
+                signalOnError(new IOException("Unable to check file status after read.", e));
+                return;
+            }
+
             synchronized (this) {
                 if (!done) {
                     done = true;
diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodyTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncRequestBodyTest.java
@@ -15,38 +15,52 @@
 
 package software.amazon.awssdk.core.internal.async;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertTrue;
+import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
+
+import java.io.ByteArrayOutputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousFileChannel;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.FileTime;
+import java.time.Instant;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.testutils.RandomTempFile;
+import software.amazon.awssdk.utils.BinaryUtils;
 
 public class FileAsyncRequestBodyTest {
     private static final long MiB = 1024 * 1024;
     private static final long TEST_FILE_SIZE = 10 * MiB;
     private static Path testFile;
 
-    @BeforeClass
-    public static void setup() throws IOException {
+    @Before
+    public void setup() throws IOException {
         testFile = new RandomTempFile(TEST_FILE_SIZE).toPath();
     }
 
-    @AfterClass
-    public static void teardown() throws IOException {
-        Files.delete(testFile);
+    @After
+    public void teardown() throws IOException {
+        try {
+            Files.delete(testFile);
+        } catch (NoSuchFileException e) {
+            // ignore
+        }
     }
 
     // If we issue just enough requests to read the file entirely but not more (to go past EOF), we should still receive
@@ -92,4 +106,154 @@ public void onComplete() {
 
         completed.get(5, TimeUnit.SECONDS);
     }
+
+    @Test
+    public void changingFile_fileGetsShorterThanAlreadyRead_failsBecauseTooShort() throws Exception {
+        AsyncRequestBody asyncRequestBody = FileAsyncRequestBody.builder()
+                                                                .path(testFile)
+                                                                .build();
+
+        ControllableSubscriber subscriber = new ControllableSubscriber();
+
+        // Start reading file
+        asyncRequestBody.subscribe(subscriber);
+        subscriber.sub.request(1);
+        assertTrue(subscriber.onNextSemaphore.tryAcquire(5, TimeUnit.SECONDS));
+
+        // Change the file to be shorter than the amount read so far
+        Files.write(testFile, "Hello".getBytes(StandardCharsets.UTF_8));
+
+        // Finishing reading the file
+        subscriber.sub.request(Long.MAX_VALUE);
+
+        assertThatThrownBy(() -> subscriber.completed.get(5, TimeUnit.SECONDS))
+            .hasCauseInstanceOf(IOException.class);
+    }
+
+    @Test
+    public void changingFile_fileGetsShorterThanExistingLength_failsBecauseTooShort() throws Exception {
+        AsyncRequestBody asyncRequestBody = FileAsyncRequestBody.builder()
+                                                                .path(testFile)
+                                                                .build();
+
+        ControllableSubscriber subscriber = new ControllableSubscriber();
+
+        // Start reading file
+        asyncRequestBody.subscribe(subscriber);
+        subscriber.sub.request(1);
+        assertTrue(subscriber.onNextSemaphore.tryAcquire(5, TimeUnit.SECONDS));
+
+        // Change the file to be 1 byte shorter than when we started
+        int currentSize = Math.toIntExact(Files.size(testFile));
+        byte[] slightlyShorterFileContent = new byte[currentSize - 1];
+        ThreadLocalRandom.current().nextBytes(slightlyShorterFileContent);
+        Files.write(testFile, slightlyShorterFileContent);
+
+        // Finishing reading the file
+        subscriber.sub.request(Long.MAX_VALUE);
+
+        assertThatThrownBy(() -> subscriber.completed.get(5, TimeUnit.SECONDS))
+            .hasCauseInstanceOf(IOException.class);
+    }
+
+    @Test
+    public void changingFile_fileGetsLongerThanExistingLength_failsBecauseTooLong() throws Exception {
+        AsyncRequestBody asyncRequestBody = FileAsyncRequestBody.builder()
+                                                                .path(testFile)
+                                                                .build();
+
+        ControllableSubscriber subscriber = new ControllableSubscriber();
+
+        // Start reading file
+        asyncRequestBody.subscribe(subscriber);
+        subscriber.sub.request(1);
+        assertTrue(subscriber.onNextSemaphore.tryAcquire(5, TimeUnit.SECONDS));
+
+        // Change the file to be 1 byte longer than when we started
+        int currentSize = Math.toIntExact(Files.size(testFile));
+        byte[] slightlyLongerFileContent = new byte[currentSize + 1];
+        ThreadLocalRandom.current().nextBytes(slightlyLongerFileContent);
+        Files.write(testFile, slightlyLongerFileContent);
+
+        // Finishing reading the file
+        subscriber.sub.request(Long.MAX_VALUE);
+
+        assertThatThrownBy(() -> subscriber.completed.get(5, TimeUnit.SECONDS))
+            .hasCauseInstanceOf(IOException.class);
+    }
+
+    @Test
+    public void changingFile_fileGetsTouched_failsBecauseUpdatedModificationTime() throws Exception {
+        AsyncRequestBody asyncRequestBody = FileAsyncRequestBody.builder()
+                                                                .path(testFile)
+                                                                .build();
+
+        ControllableSubscriber subscriber = new ControllableSubscriber();
+
+        // Start reading file
+        asyncRequestBody.subscribe(subscriber);
+        subscriber.sub.request(1);
+        assertTrue(subscriber.onNextSemaphore.tryAcquire(5, TimeUnit.SECONDS));
+
+        // Change the file to be updated
+        Thread.sleep(1_000); // Wait for 1 second so that we are definitely in a different second than when the file was created
+        Files.setLastModifiedTime(testFile, FileTime.from(Instant.now()));
+
+        // Finishing reading the file
+        subscriber.sub.request(Long.MAX_VALUE);
+
+        assertThatThrownBy(() -> subscriber.completed.get(5, TimeUnit.SECONDS))
+            .hasCauseInstanceOf(IOException.class);
+    }
+
+    @Test
+    public void changingFile_fileGetsDeleted_failsBecauseDeleted() throws Exception {
+        AsyncRequestBody asyncRequestBody = FileAsyncRequestBody.builder()
+                                                                .path(testFile)
+                                                                .build();
+
+        ControllableSubscriber subscriber = new ControllableSubscriber();
+
+        // Start reading file
+        asyncRequestBody.subscribe(subscriber);
+        subscriber.sub.request(1);
+        assertTrue(subscriber.onNextSemaphore.tryAcquire(5, TimeUnit.SECONDS));
+
+        // Delete the file
+        Files.delete(testFile);
+
+        // Finishing reading the file
+        subscriber.sub.request(Long.MAX_VALUE);
+
+        assertThatThrownBy(() -> subscriber.completed.get(5, TimeUnit.SECONDS))
+            .hasCauseInstanceOf(IOException.class);
+    }
+
+    private static class ControllableSubscriber implements Subscriber<ByteBuffer> {
+        private final ByteArrayOutputStream output = new ByteArrayOutputStream();
+        private final CompletableFuture<Void> completed = new CompletableFuture<>();
+        private final Semaphore onNextSemaphore = new Semaphore(0);
+        private Subscription sub;
+
+        @Override
+        public void onSubscribe(Subscription subscription) {
+            this.sub = subscription;
+        }
+
+        @Override
+        public void onNext(ByteBuffer byteBuffer) {
+            invokeSafely(() -> output.write(BinaryUtils.copyBytesFrom(byteBuffer)));
+            onNextSemaphore.release();
+        }
+
+        @Override
+        public void onError(Throwable throwable) {
+            completed.completeExceptionally(throwable);
+        }
+
+        @Override
+        public void onComplete() {
+            completed.complete(null);
+        }
+    }
 }
