Fix Length::UpTo usage in FsBuilder so it does not exceed availab…

…le file length (#3797)

## Motivation and Context
awslabs/aws-sdk-rust#821

## Description
This PR addresses an issue with the
[Length::UpTo](https://docs.rs/aws-smithy-types/1.2.2/aws_smithy_types/byte_stream/enum.Length.html)
usage in `FsBuilder`. Previously, if a value specified for `UpTo`
exceeded the remaining file length, `FsBuilder` would incorrectly accept
the value. This discrepancy led to failures in subsequent request
dispatches, as the actual body size did not match the advertised
`Content-Length`, as explained
[here](awslabs/aws-sdk-rust#821 (comment))
(thank you @pablosichert for a self-contained reproducer and problem
analysis!).

## Testing
- Added a unit test for `FsBuilder` verifying the `Length::UpTo` usage
- Ran successfully a customer provided
[reproducer](awslabs/aws-sdk-rust#821 (comment))
with the code changes in this PR (with an added a call to
`complete_multipart_upload` at the end, it also succeeded in uploading
the object):
```
upload_id: cTDSngbubD25cOoFCNgjpG55o0hAMQNjO16dNFyNTKjg9PEtkcrKG5rTGzBns7CXoO8T.Qm9GpNj6jgwJTKcXDpsca95wSMWMDfPF0DBhmbk3OAGHuuGM1E70spk2suW
File created
part_number: 1, e_tag: "5648ddf58c7c90a788d7f16717a61b08"
part_number: 2, e_tag: "a6bdad6d65d18d842ef1d57ca4673bc3"
part_number: 3, e_tag: "f518f6b19b255ec49b61d511288554fc"
part_number: 4, e_tag: "1496524801eb1d0a7cfbe608eb037d9c"
part_number: 5, e_tag: "21340de04927ce1bed58ad0375c03e01"
```

## Checklist
- [x] For changes to the smithy-rs codegen or runtime crates, I have
created a changelog entry Markdown file in the `.changelog` directory,
specifying "client," "server," or both in the `applies_to` key.
- [x] For changes to the AWS SDK, generated SDK code, or SDK runtime
crates, I have created a changelog entry Markdown file in the
`.changelog` directory, specifying "aws-sdk-rust" in the `applies_to`
key.

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

.changelog/1724182177.md

@@ -0,0 +1,14 @@
+---
+applies_to:
+- aws-sdk-rust
+- client
+authors:
+- ysaito1001
+references:
+- aws-sdk-rust#821
+- smithy-rs#3797
+breaking: false
+new_feature: false
+bug_fix: true
+---
+Fix the [Length::UpTo](https://docs.rs/aws-smithy-types/1.2.2/aws_smithy_types/byte_stream/enum.Length.html) usage in [FsBuilder](https://docs.rs/aws-smithy-types/1.2.2/aws_smithy_types/byte_stream/struct.FsBuilder.html), ensuring that the specified length does not exceed the remaining file length.

rust-runtime/aws-smithy-types/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aws-smithy-types"
-version = "1.2.2"
+version = "1.2.3"
 authors = [
     "AWS Rust SDK Team <[email protected]>",
     "Russell Cohen <[email protected]>",

rust-runtime/aws-smithy-types/src/byte_stream/bytestream_util.rs

@@ -5,6 +5,7 @@
 
 use crate::body::SdkBody;
 use crate::byte_stream::{error::Error, error::ErrorKind, ByteStream};
+use std::cmp::min;
 use std::future::Future;
 use std::path::PathBuf;
 use std::pin::Pin;
@@ -190,15 +191,16 @@ impl FsBuilder {
             return Err(ErrorKind::OffsetLargerThanFileSize.into());
         }
 
+        let remaining_file_length = file_length - offset;
         let length = match self.length {
             Some(Length::Exact(length)) => {
-                if length > file_length - offset {
+                if length > remaining_file_length {
                     return Err(ErrorKind::LengthLargerThanFileSizeMinusReadOffset.into());
                 }
                 length
             }
-            Some(Length::UpTo(length)) => length,
-            None => file_length - offset,
+            Some(Length::UpTo(length)) => min(length, remaining_file_length),
+            None => remaining_file_length,
         };
 
         if let Some(path) = self.path {
@@ -247,3 +249,57 @@ enum State {
         bytes_left: u64,
     },
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use tempfile::NamedTempFile;
+
+    #[tokio::test]
+    async fn length_up_to_should_work() {
+        const FILE_LEN: usize = 1000;
+        // up to less than `FILE_LEN`
+        {
+            let mut file = NamedTempFile::new().unwrap();
+            file.write_all(vec![0; FILE_LEN].as_slice()).unwrap();
+            let byte_stream = FsBuilder::new()
+                .path(file.path())
+                .length(Length::UpTo((FILE_LEN / 2) as u64))
+                .build()
+                .await
+                .unwrap();
+            let (lower, upper) = byte_stream.size_hint();
+            assert_eq!(lower, upper.unwrap());
+            assert_eq!((FILE_LEN / 2) as u64, lower);
+        }
+        // up to equal to `FILE_LEN`
+        {
+            let mut file = NamedTempFile::new().unwrap();
+            file.write_all(vec![0; FILE_LEN].as_slice()).unwrap();
+            let byte_stream = FsBuilder::new()
+                .path(file.path())
+                .length(Length::UpTo(FILE_LEN as u64))
+                .build()
+                .await
+                .unwrap();
+            let (lower, upper) = byte_stream.size_hint();
+            assert_eq!(lower, upper.unwrap());
+            assert_eq!(FILE_LEN as u64, lower);
+        }
+        // up to greater than `FILE_LEN`
+        {
+            let mut file = NamedTempFile::new().unwrap();
+            file.write_all(vec![0; FILE_LEN].as_slice()).unwrap();
+            let byte_stream = FsBuilder::new()
+                .path(file.path())
+                .length(Length::UpTo((FILE_LEN * 2) as u64))
+                .build()
+                .await
+                .unwrap();
+            let (lower, upper) = byte_stream.size_hint();
+            assert_eq!(lower, upper.unwrap());
+            assert_eq!(FILE_LEN as u64, lower);
+        }
+    }
+}