From fd3fb77ed591d58adec6e765bfdb1cec74de1ee2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Wed, 13 May 2026 15:44:09 +0800 Subject: [PATCH] fix(sftp): avoid metadata on multipart copy (#2935) Co-authored-by: houseme --- crates/protocols/src/common/dummy_storage.rs | 20 ++++++-- crates/protocols/src/sftp/test_support.rs | 4 +- crates/protocols/src/sftp/write.rs | 51 +++++++++++++++----- 3 files changed, 58 insertions(+), 17 deletions(-) diff --git a/crates/protocols/src/common/dummy_storage.rs b/crates/protocols/src/common/dummy_storage.rs index aa21bd2b1..2926d66e7 100644 --- a/crates/protocols/src/common/dummy_storage.rs +++ b/crates/protocols/src/common/dummy_storage.rs @@ -34,10 +34,10 @@ use bytes::Bytes; use futures_util::stream::{self, StreamExt}; use s3s::dto::{ AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput, - CopyObjectInput, CopyObjectOutput, CreateBucketOutput, CreateMultipartUploadInput, CreateMultipartUploadOutput, - DeleteBucketOutput, DeleteObjectOutput, ETag, GetObjectOutput, HeadBucketOutput, HeadObjectOutput, ListBucketsOutput, - ListObjectsV2Input, ListObjectsV2Output, PutObjectInput, PutObjectOutput, StreamingBlob, Timestamp, UploadPartCopyInput, - UploadPartCopyOutput, UploadPartInput, UploadPartOutput, + CopyObjectInput, CopyObjectOutput, CopyPartResult, CreateBucketOutput, CreateMultipartUploadInput, + CreateMultipartUploadOutput, DeleteBucketOutput, DeleteObjectOutput, ETag, GetObjectOutput, HeadBucketOutput, + HeadObjectOutput, ListBucketsOutput, ListObjectsV2Input, ListObjectsV2Output, PutObjectInput, PutObjectOutput, StreamingBlob, + Timestamp, UploadPartCopyInput, UploadPartCopyOutput, UploadPartInput, UploadPartOutput, }; use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; @@ -318,6 +318,18 @@ impl DummyBackend { self.inner.lock().expect("lock").upload_part.push_back(Err(err)); } + /// Queue an upload_part_copy Ok response carrying the given ETag. + pub fn queue_upload_part_copy_ok(&self, e_tag: impl Into) { + let out = UploadPartCopyOutput { + copy_part_result: Some(CopyPartResult { + e_tag: Some(ETag::Strong(e_tag.into())), + ..Default::default() + }), + ..Default::default() + }; + self.inner.lock().expect("lock").upload_part_copy.push_back(Ok(out)); + } + /// Queue a complete_multipart_upload Ok response. pub fn queue_complete_multipart_upload_ok(&self) { self.inner diff --git a/crates/protocols/src/sftp/test_support.rs b/crates/protocols/src/sftp/test_support.rs index 65f0176ae..e5f1ab469 100644 --- a/crates/protocols/src/sftp/test_support.rs +++ b/crates/protocols/src/sftp/test_support.rs @@ -65,13 +65,13 @@ pub(super) fn file_handle(bucket: &str, key: &str, size: u64, attrs: FileAttribu /// Build a HandleState::Write with the given bucket, key, and /// WritePhase, ready to be inserted directly into a driver handle -/// table without running open_write. Default FileAttributes are used. +/// table without running open_write. Default FSTAT attrs and empty OPEN attrs are used. pub(super) fn write_handle(bucket: &str, key: &str, phase: WritePhase) -> HandleState { HandleState::Write { bucket: bucket.to_string(), key: key.to_string(), attrs: FileAttributes::default(), - open_attrs: FileAttributes::default(), + open_attrs: FileAttributes::empty(), phase, } } diff --git a/crates/protocols/src/sftp/write.rs b/crates/protocols/src/sftp/write.rs index fcf8b78ac..e3504d51a 100644 --- a/crates/protocols/src/sftp/write.rs +++ b/crates/protocols/src/sftp/write.rs @@ -159,9 +159,7 @@ pub(super) fn fstat_reported_size(phase: &WritePhase, part_size: u64, cached_siz /// Drop runs the same AbortMultipartUpload for both. /// /// attrs is required by the HandleState::Write variant layout but -/// Drop does not read it. Callers pass a clone of the live attrs, or -/// FileAttributes::default() at the write_dispatch_begin_streaming -/// site where no live attrs is in scope. +/// Drop does not read it. Callers pass a clone of the live attrs. pub(super) fn build_write_tombstone( bucket: &str, key: &str, @@ -173,7 +171,7 @@ pub(super) fn build_write_tombstone( bucket: bucket.to_string(), key: key.to_string(), attrs: attrs.clone(), - open_attrs: FileAttributes::default(), + open_attrs: FileAttributes::empty(), phase: WritePhase::Failed { upload_id, abort_authorized, @@ -952,7 +950,7 @@ impl SftpDriver { // close_abort_or_skip can honour a Deny-Abort policy without a // second IAM probe per error path. let mp = self - .start_multipart_upload(dst_bucket, dst_key, &FileAttributes::default()) + .start_multipart_upload(dst_bucket, dst_key, &FileAttributes::empty()) .await?; let result: Result, SftpError> = async { @@ -1590,7 +1588,7 @@ mod tests { with_test_auth_override( |_, _, _| true, - driver.write_dispatch_begin_streaming(&handle_id, &mut phase, "b", "k", &FileAttributes::default()), + driver.write_dispatch_begin_streaming(&handle_id, &mut phase, "b", "k", &FileAttributes::empty()), ) .await .expect("begin_streaming must succeed on queued Create Ok"); @@ -1630,7 +1628,7 @@ mod tests { backend.queue_create_multipart_upload_ok("UP-ALLOW"); let driver = build_driver(backend, TEST_PART_SIZE); - let mp = with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::default())) + let mp = with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::empty())) .await .expect("start_multipart_upload must succeed on Allow"); assert_eq!(mp.upload_id, "UP-ALLOW"); @@ -1672,7 +1670,7 @@ mod tests { backend.queue_create_multipart_upload_ok("UP-NO-ATTRS"); let driver = build_driver(backend.clone(), TEST_PART_SIZE); - with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::default())) + with_test_auth_override(|_, _, _| true, driver.start_multipart_upload("b", "k", &FileAttributes::empty())) .await .expect("start_multipart_upload must succeed"); @@ -1691,7 +1689,7 @@ mod tests { // a WORM-shaped IAM policy a principal can meet in production. let mp = with_test_auth_override( |action, _bucket, _object| !matches!(action, S3Action::AbortMultipartUpload), - driver.start_multipart_upload("b", "k", &FileAttributes::default()), + driver.start_multipart_upload("b", "k", &FileAttributes::empty()), ) .await .expect("Create Allow must succeed even when Abort is Deny"); @@ -1713,7 +1711,7 @@ mod tests { let err = with_test_auth_override( |action, _, _| !matches!(action, S3Action::CreateMultipartUpload), - driver.start_multipart_upload("b", "k", &FileAttributes::default()), + driver.start_multipart_upload("b", "k", &FileAttributes::empty()), ) .await .expect_err("Deny on CreateMultipartUpload must fail fast"); @@ -1724,6 +1722,37 @@ mod tests { ); } + #[tokio::test] + async fn multipart_copy_starts_upload_without_sftp_open_metadata() { + let backend = Arc::new(DummyBackend::new()); + backend.queue_create_multipart_upload_ok("UP-COPY"); + backend.queue_upload_part_copy_ok("etag-copy-1"); + backend.queue_complete_multipart_upload_ok(); + let driver = build_driver(backend.clone(), TEST_PART_SIZE); + + with_test_auth_override( + |_, _, _| true, + driver.multipart_copy("src-bucket", "src-key", "dst-bucket", "dst-key", TEST_PART_SIZE), + ) + .await + .expect("multipart copy must complete"); + + let create_calls = backend.create_multipart_calls(); + assert_eq!(create_calls.len(), 1); + assert_eq!(create_calls[0].bucket, "dst-bucket"); + assert_eq!(create_calls[0].key, "dst-key"); + assert!( + create_calls[0].metadata.is_none(), + "server-side multipart copy is not an SFTP OPEN path and must not write OPEN metadata" + ); + + let complete_calls = backend.complete_multipart_calls(); + assert_eq!(complete_calls.len(), 1); + assert_eq!(complete_calls[0].upload_id, "UP-COPY"); + assert_eq!(complete_calls[0].part_count, 1); + assert!(backend.abort_multipart_calls().is_empty(), "successful multipart copy must not abort"); + } + // --- upload_multipart_bytes --- #[tokio::test] @@ -1912,7 +1941,7 @@ mod tests { let driver = build_driver(backend.clone(), TEST_PART_SIZE); driver - .commit_write("b", "k", &FileAttributes::default(), b"hello".to_vec()) + .commit_write("b", "k", &FileAttributes::empty(), b"hello".to_vec()) .await .expect("commit_write must succeed");