fix(sftp): preserve OPEN-time client attrs as object metadata (#2913)

Co-authored-by: houseme <housemecn@gmail.com>

crates/e2e_test/src/protocols/sftp_compliance.rs

@@ -39,7 +39,7 @@
 use crate::protocols::sftp_compliance_tests::{
     cmptst_01, cmptst_02, cmptst_03, cmptst_04, cmptst_05, cmptst_06, cmptst_07, cmptst_08, cmptst_09, cmptst_10, cmptst_11,
     cmptst_12, cmptst_13, cmptst_14, cmptst_15, cmptst_16, cmptst_17, cmptst_18, cmptst_19, cmptst_20, cmptst_21, cmptst_22,
-    cmptst_23, cmptst_27, cmptst_28, cmptst_29, cmptst_32, cmptst_33, spawn_compliance_rustfs,
+    cmptst_23, cmptst_27, cmptst_28, cmptst_29, cmptst_32, cmptst_33, cmptst_34, spawn_compliance_rustfs,
 };
 #[cfg(target_os = "linux")]
 use crate::protocols::sftp_compliance_tests::{cmptst_24, cmptst_25, cmptst_26};
@@ -96,6 +96,15 @@ pub async fn test_sftp_compliance_suite() -> Result<()> {
         cmptst_13::run_implicit_dir_round_trip(&sftp).await?;
         cmptst_14::run_winscp_setstat_shape_on_handle(&sftp).await?;
 
+        // CMPTST-34 cross-checks the SFTP streaming-multipart write
+        // path against the S3 layer. The OPEN-time FileAttributes must
+        // reach the finalised object as x-amz-meta-* user metadata
+        // through the CreateMultipartUpload input field. The S3 client
+        // connects to the same rustfs process this suite already drives.
+        let s3 = build_test_s3_client(&format!("http://{COMPLIANCE_RW_S3_ADDRESS}"));
+        wait_for_s3_ready(&s3, 30).await?;
+        cmptst_34::run_open_attrs_round_trip_multipart(&sftp, &s3).await?;
+
         drop(sftp);
         session.disconnect(russh::Disconnect::ByApplication, "", "en").await?;
         info!("SFTP compliance suite passed");

crates/e2e_test/src/protocols/sftp_compliance_tests.rs

@@ -104,6 +104,11 @@
 //!   byte-exact with the production cache window.
 //! - CMPTST-33: read-cache disabled regression, 8 MiB download
 //!   byte-exact with RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES=0.
+//! - CMPTST-34: OPEN with non-default FileAttributes followed by a
+//!   payload that crosses the 5 MiB multipart boundary preserves the
+//!   client-supplied mtime and permissions through the streaming
+//!   CreateMultipartUpload path. HeadObject through aws-sdk-s3
+//!   confirms the metadata reached the finalised S3 object.
 
 use crate::common::rustfs_binary_path_with_features;
 use crate::protocols::sftp_helpers::{
@@ -3276,6 +3281,80 @@ pub(crate) mod cmptst_33 {
     }
 }
 
+// CMPTST-34: OPEN-time client attrs preservation across the streaming
+// multipart write path. The payload crosses the 5 MiB part-size
+// boundary so the driver transitions Buffering -> Streaming and
+// finalises via CompleteMultipartUpload. The OPEN-supplied mtime and
+// permissions must reach the resulting object as x-amz-meta-mtime and
+// x-amz-meta-mode. The S3 client connects to the same rustfs process
+// the shared-server suite already drives.
+pub(crate) mod cmptst_34 {
+    use super::*;
+
+    const COMPLIANCE_TEST_OUTPUT_ID: &str = "CMPTST-34";
+    const REQUESTED_MTIME: u32 = 1_715_000_010;
+    const REQUESTED_MODE: u32 = 0o600;
+
+    pub(crate) async fn run_open_attrs_round_trip_multipart(sftp: &SftpSession, s3: &S3Client) -> Result<()> {
+        info!("{COMPLIANCE_TEST_OUTPUT_ID}: OPEN with mtime + mode, multi-part payload, streaming path");
+        let bucket = "complopenattrsmpbucket";
+        let bucket_path = format!("/{bucket}");
+        sftp.create_dir(&bucket_path).await?;
+
+        let path = format!("/{bucket}/attr-mp.bin");
+        // 6 MiB exceeds the 5 MiB part-size boundary so the streaming
+        // path runs at least one full UploadPart before the CLOSE-time
+        // CompleteMultipartUpload finalises the object.
+        let payload = vec![0xA5u8; 6 * 1024 * 1024];
+
+        let client_attrs = FileAttributes {
+            mtime: Some(REQUESTED_MTIME),
+            atime: Some(REQUESTED_MTIME),
+            permissions: Some(REQUESTED_MODE),
+            ..FileAttributes::default()
+        };
+
+        let mut writer = sftp
+            .open_with_flags_and_attributes(&path, OpenFlags::CREATE | OpenFlags::TRUNCATE | OpenFlags::WRITE, client_attrs)
+            .await?;
+        writer.write_all(&payload).await?;
+        writer.flush().await?;
+        writer.shutdown().await?;
+
+        let head = s3
+            .head_object()
+            .bucket(bucket)
+            .key("attr-mp.bin")
+            .send()
+            .await
+            .map_err(|e| anyhow!("S3 HeadObject failed: {e:?}"))?;
+        let content_length = head.content_length().unwrap_or(0);
+        if content_length != payload.len() as i64 {
+            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} unexpected size: got {content_length} bytes"));
+        }
+        let metadata = head
+            .metadata()
+            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} HeadObject returned no metadata map"))?;
+        let mtime_value = metadata
+            .get("mtime")
+            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mtime key missing on the object"))?;
+        if mtime_value != &REQUESTED_MTIME.to_string() {
+            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mtime mismatch: got {mtime_value}"));
+        }
+        let mode_value = metadata
+            .get("mode")
+            .ok_or_else(|| anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mode key missing on the object"))?;
+        if mode_value != &REQUESTED_MODE.to_string() {
+            return Err(anyhow!("{COMPLIANCE_TEST_OUTPUT_ID} mode mismatch: got {mode_value}"));
+        }
+
+        sftp.remove_file(&path).await?;
+        sftp.remove_dir(&bucket_path).await?;
+        info!("PASS {COMPLIANCE_TEST_OUTPUT_ID}: multipart upload preserved mtime + mode end to end");
+        Ok(())
+    }
+}
+
 // Shared parameters for CMPTST-32 (cache enabled) and CMPTST-33 (cache
 // disabled). Both cases seed the same fixture and download it
 // end-to-end, then assert byte-count and SHA256 against the

rustfs/src/app/multipart_usecase.rs

@@ -21,7 +21,7 @@ use crate::error::ApiError;
 use crate::storage::access::has_bypass_governance_header;
 use crate::storage::helper::OperationHelper;
 use crate::storage::options::{
-    copy_src_opts, extract_metadata, get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts,
+    copy_src_opts, extract_metadata_from_mime, get_complete_multipart_upload_opts, get_content_sha256_with_query, get_opts,
     parse_copy_source_range, put_opts, validate_archive_content_encoding,
 };
 use crate::storage::s3_api::multipart::{
@@ -496,6 +496,7 @@ impl DefaultMultipartUsecase {
             object_lock_legal_hold_status,
             object_lock_mode,
             object_lock_retain_until_date,
+            metadata: input_metadata,
             ..
         } = req.input.clone();
 
@@ -524,7 +525,8 @@ impl DefaultMultipartUsecase {
             req.headers.get("content-encoding").and_then(|value| value.to_str().ok()),
         )?;
 
-        let mut metadata = extract_metadata(&req.headers);
+        let mut metadata = input_metadata.unwrap_or_default();
+        extract_metadata_from_mime(&req.headers, &mut metadata);
 
         if let Some(tags) = tagging {
             metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags);