Fix read issue on concurrent open after truncate (#1781)

Fix race condition when 2 concurrent open requests occur after a
truncate and resulting in errors on read.

The issue is caused by a concurrent operation which ignores that the
inode is still completing an upload and tries to refresh its state by
performing a remote lookup to the bucket. By the time the remote lookup
completes, its result may be stale but still be used to overwrite the
result of the upload. This fix adds a check for a pending upload instead
of only relying on the `write_status` field.

### Does this change impact existing behavior?

No.

### Does this change need a changelog entry? Does it require a version
change?

Added changelog entries for a patch release.

---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

---------

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

Author: passaro

mountpoint-s3-fs/CHANGELOG.md

@@ -1,5 +1,7 @@
 ## Unreleased (v0.9.1)
 
+* Fix a race condition where concurrent operations after closing a truncated file could result in I/O errors on subsequent reads. The issue was introduced in `mountpoint-s3-fs` v0.9.0.
+  ([#1781](https://github.com/awslabs/mountpoint-s3/pull/1781))
 * Upgrade cargo dependencies.
 * Fix incorrect validation of default data cache limit which would cause Mountpoint to preserve less than 5% of available space ([#1779](https://github.com/awslabs/mountpoint-s3/pull/1779))
 

mountpoint-s3-fs/src/superblock.rs

@@ -280,11 +280,11 @@ impl<OC: ObjectClient + Send + Sync> Superblock<OC> {
 
         {
             let mut sync = inode.get_mut_inode_state()?;
-            let is_remote = sync.write_status == WriteStatus::Remote;
+            let has_local_state = sync.write_status != WriteStatus::Remote || sync.pending_upload_hook.is_some();
 
-            if !is_remote || !force_revalidate_if_remote {
-                // If the inode is local (open/unopened), extend its stat's validity before returning
-                if !is_remote {
+            if has_local_state || !force_revalidate_if_remote {
+                // If the inode is local (open/unopened) and/or has a pending upload, extend its stat's validity before returning
+                if has_local_state {
                     let validity = match inode.kind() {
                         InodeKind::File => self.inner.config.cache_config.file_ttl,
                         InodeKind::Directory => self.inner.config.cache_config.dir_ttl,
@@ -848,9 +848,9 @@ impl<OC: ObjectClient + Send + Sync + Clone> Metablock for Superblock<OC> {
             ));
         }
 
+        let inode = looked_up_inode.inode;
         let (pending_upload_hook, inode_lookup) = {
-            let inode = looked_up_inode.inode.clone();
-            let mut locked_inode = looked_up_inode.inode.get_mut_inode_state()?;
+            let mut locked_inode = inode.get_mut_inode_state()?;
 
             let pending_upload_hook = match mode {
                 ReadWriteMode::Read => self.start_reading(&mut locked_inode, inode.clone(), fh)?,
@@ -870,10 +870,19 @@ impl<OC: ObjectClient + Send + Sync + Clone> Metablock for Superblock<OC> {
             (pending_upload_hook, inode_lookup)
         };
 
-        let lookup = if let Some(upload_hook) = pending_upload_hook
-            && let Some(lookup_after_upload) = upload_hook.wait_for_completion().await?
-        {
-            lookup_after_upload
+        let lookup = if let Some(upload_hook) = pending_upload_hook {
+            if let Some(lookup_after_upload) = upload_hook.wait_for_completion().await? {
+                lookup_after_upload
+            } else {
+                // Return up-to-date lookup.
+                let locked_inode = inode.get_inode_state()?;
+                Lookup::new(
+                    inode.ino(),
+                    locked_inode.stat.clone(),
+                    inode.kind(),
+                    Some(S3Location::new(self.inner.s3_path.clone(), inode.valid_key().clone())),
+                )
+            }
         } else {
             inode_lookup
         };
@@ -1670,8 +1679,10 @@ impl<OC: ObjectClient + Send + Sync> SuperblockInner<OC> {
             (None, None) => Err(InodeError::FileDoesNotExist(name.to_owned(), parent.err())),
             (Some(remote), Some(existing_inode)) => {
                 let mut existing_state = existing_inode.get_mut_inode_state()?;
+                let existing_has_local_state =
+                    existing_state.write_status != WriteStatus::Remote || existing_state.pending_upload_hook.is_some();
                 if remote.kind == existing_inode.kind()
-                    && existing_state.write_status == WriteStatus::Remote
+                    && !existing_has_local_state
                     && existing_state.stat.etag == remote.stat.etag
                 {
                     trace!(parent=?existing_inode.parent(), name=?existing_inode.name(), ino=?existing_inode.ino(), "updating inode in place");
@@ -1758,11 +1769,12 @@ impl<OC: ObjectClient + Send + Sync> SuperblockInner<OC> {
                 // being consistent with our stated semantics about implicit directories and how
                 // directories shadow files.
                 let mut existing_state = existing_inode.get_mut_inode_state()?;
-                let existing_is_remote = existing_state.write_status == WriteStatus::Remote;
+                let existing_has_local_state =
+                    existing_state.write_status != WriteStatus::Remote || existing_state.pending_upload_hook.is_some();
 
                 // Remote files are always shadowed by existing local files/directories, so do
                 // nothing and return the existing inode.
-                if remote.kind == InodeKind::File && !existing_is_remote {
+                if remote.kind == InodeKind::File && existing_has_local_state {
                     return Ok(LookedUpInode {
                         inode: existing_inode.clone(),
                         stat: existing_state.stat.clone(),
@@ -1776,10 +1788,10 @@ impl<OC: ObjectClient + Send + Sync> SuperblockInner<OC> {
                 // updating the parent.
                 let same_kind = remote.kind == existing_inode.kind();
                 let same_etag = existing_state.stat.etag == remote.stat.etag;
-                if same_kind && same_etag && (existing_is_remote || remote.kind == InodeKind::Directory) {
+                if same_kind && same_etag && (!existing_has_local_state || remote.kind == InodeKind::Directory) {
                     trace!(parent=?existing_inode.parent(), name=?existing_inode.name(), ino=?existing_inode.ino(), "updating inode in place (slow path)");
                     existing_state.stat = remote.stat.clone();
-                    if remote.kind == InodeKind::Directory && !existing_is_remote {
+                    if remote.kind == InodeKind::Directory && existing_has_local_state {
                         trace!(parent=?existing_inode.parent(), name=?existing_inode.name(), ino=?existing_inode.ino(), "local directory has become remote");
                         existing_state.write_status = WriteStatus::Remote;
                         let InodeKindData::Directory { writing_children, .. } = &mut parent_state.kind_data else {
@@ -1799,7 +1811,7 @@ impl<OC: ObjectClient + Send + Sync> SuperblockInner<OC> {
                     ino=?existing_inode.ino(),
                     same_kind,
                     same_etag,
-                    existing_is_remote,
+                    existing_has_local_state,
                     remote_is_dir = remote.kind == InodeKind::Directory,
                     "inode could not be updated in place",
                 );

mountpoint-s3-fs/tests/fuse_tests/write_test.rs

@@ -1174,6 +1174,74 @@ fn overwrite_after_read_test_mock(prefix: &str) {
     overwrite_after_read_test(fuse::mock_session::new, prefix);
 }
 
+fn concurrent_open_after_overwrite_truncate_test(
+    creator_fn: impl TestSessionCreator,
+    prefix: &str,
+    rw_mode: ReadWriteMode,
+) {
+    let filesystem_config = S3FilesystemConfig {
+        allow_overwrite: true,
+        ..Default::default()
+    };
+    let test_config = TestSessionConfig {
+        filesystem_config,
+        ..Default::default()
+    };
+    let test_session = creator_fn(prefix, test_config);
+
+    // Make sure there's an existing directory and a file
+    test_session
+        .client()
+        .put_object("dir/hello.txt", b"hello world")
+        .unwrap();
+
+    let _subdir = test_session.mount_path().join("dir");
+    let path = test_session.mount_path().join("dir/hello.txt");
+
+    // File should be empty when opened with O_TRUNC even without any write
+    let write_fh = File::options()
+        .rw_mode(rw_mode)
+        .truncate(true)
+        .open(&path)
+        .expect("open should succeed");
+    drop(write_fh);
+
+    fn check_empty(path: impl AsRef<Path>) {
+        let mut read_fh = File::options().read(true).open(&path).unwrap();
+        let mut hello_contents = String::new();
+        read_fh.read_to_string(&mut hello_contents).unwrap();
+        assert!(hello_contents.is_empty());
+    }
+
+    std::thread::scope(|s| {
+        s.spawn(|| {
+            check_empty(&path);
+        });
+        s.spawn(|| {
+            check_empty(&path);
+        });
+    });
+}
+
+#[cfg(feature = "s3_tests")]
+#[test_matrix([WRITE_ONLY, READ_WRITE])]
+fn concurrent_open_after_overwrite_truncate_test_s3(rw_mode: ReadWriteMode) {
+    concurrent_open_after_overwrite_truncate_test(
+        fuse::s3_session::new,
+        "concurrent_open_after_overwrite_truncate_test",
+        rw_mode,
+    );
+}
+
+#[test_matrix([WRITE_ONLY, READ_WRITE])]
+fn concurrent_open_after_overwrite_truncate_test_mock(rw_mode: ReadWriteMode) {
+    concurrent_open_after_overwrite_truncate_test(
+        fuse::mock_session::new,
+        "concurrent_open_after_overwrite_truncate_test",
+        rw_mode,
+    );
+}
+
 fn write_handle_no_update_existing_empty_file(
     creator_fn: impl TestSessionCreator,
     prefix: &str,

mountpoint-s3/CHANGELOG.md

@@ -1,5 +1,7 @@
 ## Unreleased (v1.22.1)
 
+* Fix a race condition where concurrent operations after closing a truncated file could result in I/O errors on subsequent reads. The issue was introduced in v1.22.0.
+  ([#1781](https://github.com/awslabs/mountpoint-s3/pull/1781))
 * Upgrade cargo dependencies.
 * Fix incorrect validation of default data cache limit which would cause Mountpoint to preserve less than 5% of available space ([#1779](https://github.com/awslabs/mountpoint-s3/pull/1779))