From ead419451a3f8f3b555ce01341c47f4c864bea4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Thu, 6 Aug 2026 08:27:23 +0800 Subject: [PATCH] fix(replication): send source versionId as query param to remote targets (#5752) * test(replication): assert remote PUT and multipart initiate carry versionId query * fix(replication): send source versionId as query param to remote targets --- crates/e2e_test/src/fake_s3_target/mod.rs | 3 +- .../src/replication_extension_test.rs | 138 ++++++++++++++++++ .../ecstore/src/bucket/bucket_target_sys.rs | 130 +++++++++++++++++ rustfs/src/storage/options.rs | 29 +++- 4 files changed, 298 insertions(+), 2 deletions(-) diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index 48cbce288..0fa3c67ef 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -574,7 +574,8 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest { (&Method::POST, true) if query.contains_key("uploads") => Operation::CreateMultipartUpload, (&Method::POST, true) if upload_id.is_some() => Operation::CompleteMultipartUpload, (&Method::DELETE, true) if upload_id.is_some() => Operation::AbortMultipartUpload, - (&Method::PUT, true) if only_query_keys(&[]) => Operation::PutObject, + // A replication PUT addresses the source version via `?versionId=`. + (&Method::PUT, true) if only_query_keys(&["versionId"]) => Operation::PutObject, (&Method::GET, true) if only_query_keys(&["versionId"]) => Operation::GetObject, (&Method::HEAD, true) if only_query_keys(&["versionId"]) => Operation::HeadObject, (&Method::DELETE, true) if only_query_keys(&["versionId"]) => Operation::DeleteObject, diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index e07cf7c22..e2c2dc2fa 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -16,6 +16,7 @@ use crate::common::{ RustFSTestEnvironment, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, local_http_client, replication_fast_env, rustfs_binary_path, }; +use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, Operation as FakeTargetOperation}; use crate::kms::common::{create_key_with_specific_id, sse_customer_key_md5_base64}; use crate::storage_api::replication_extension::BucketTargetSys; use aws_sdk_s3::config::{Credentials, Region}; @@ -6639,3 +6640,140 @@ async fn test_site_replication_replicates_service_accounts_created_from_sts_sess Ok(()) } + +/// Poll the fake target journal until `operation` arrives for `key`, then +/// return the `versionId` query value the request carried. +async fn wait_for_target_request_version_id( + target: &FakeS3Target, + operation: FakeTargetOperation, + key: &str, +) -> Result, Box> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if let Some(record) = target + .requests() + .into_iter() + .find(|record| record.operation == operation && record.key.as_deref() == Some(key)) + { + return Ok(record.version_id); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("fake target never received {operation:?} for {key}; journal: {:?}", target.requests()).into()); + } + sleep(Duration::from_millis(200)).await; + } +} + +/// P0-5: MinIO derives the replicated version exclusively from the `versionId` +/// query parameter (`putOptsFromReq`); the internal x-*-source-version-id +/// headers do not exist there. Without the query, a MinIO target mints fresh +/// version ids and RustFS -> MinIO replication drifts. PutObject and +/// CreateMultipartUpload (the version is decided at initiate time) must both +/// carry the source version as `?versionId=`. +#[tokio::test] +#[serial] +async fn test_replication_put_and_create_multipart_carry_source_version_id_query() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "versionid-query-dst"; + target.create_bucket(target_bucket); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut source_process_env = replication_fast_env(); + source_process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + source_process_env.extend_from_slice(&[("NO_PROXY", "127.0.0.1,localhost"), ("HTTP_PROXY", ""), ("HTTPS_PROXY", "")]); + source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; + + let source_bucket = "versionid-query-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + + let target_arn = set_replication_target_with_options( + &source_env, + source_bucket, + ReplicationTargetOptions { + endpoint: &target.address(), + access_key: FAKE_ACCESS_KEY, + secret_key: FAKE_SECRET_KEY, + target_bucket, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + // Small object -> replicated through a single PutObject. + let put = source_client + .put_object() + .bucket(source_bucket) + .key("small.txt") + .body(ByteStream::from_static(b"versionid query payload")) + .send() + .await?; + let put_source_version = put + .version_id() + .ok_or("versioned source PUT must return a version id")? + .to_string(); + let recorded = wait_for_target_request_version_id(&target, FakeTargetOperation::PutObject, "small.txt").await?; + assert_eq!( + recorded.as_deref(), + Some(put_source_version.as_str()), + "replication PutObject must carry the source version in the versionId query" + ); + + // Multipart source object -> replicated through CreateMultipartUpload; + // the target version is fixed at initiate time. + let create = source_client + .create_multipart_upload() + .bucket(source_bucket) + .key("large.bin") + .send() + .await?; + let upload_id = create + .upload_id() + .ok_or("multipart initiate must return an upload id")? + .to_string(); + let mut completed_parts = Vec::new(); + for (part_number, body) in [(1, vec![b'a'; 5 * 1024 * 1024]), (2, vec![b'b'; 1024])] { + let uploaded = source_client + .upload_part() + .bucket(source_bucket) + .key("large.bin") + .upload_id(&upload_id) + .part_number(part_number) + .body(ByteStream::from(body)) + .send() + .await?; + completed_parts.push( + CompletedPart::builder() + .part_number(part_number) + .e_tag(uploaded.e_tag().unwrap_or_default()) + .build(), + ); + } + let complete = source_client + .complete_multipart_upload() + .bucket(source_bucket) + .key("large.bin") + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build()) + .send() + .await?; + let multipart_source_version = complete + .version_id() + .ok_or("versioned multipart completion must return a version id")? + .to_string(); + let recorded = wait_for_target_request_version_id(&target, FakeTargetOperation::CreateMultipartUpload, "large.bin").await?; + assert_eq!( + recorded.as_deref(), + Some(multipart_source_version.as_str()), + "replication CreateMultipartUpload must carry the source version in the versionId query" + ); + + target.shutdown().await; + Ok(()) +} diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index e401c0539..10ffb0e20 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1424,6 +1424,37 @@ fn resolve_delete_api_version_id(version_id: Option, opts: &RemoveObject } } +/// Resolve the S3 `versionId` query parameter for a replication PUT / +/// CreateMultipartUpload against a remote target. +/// +/// MinIO reads the replicated version only from the query string +/// (`putOptsFromReq`); the internal `x-*-source-version-id` headers do not +/// exist there, so without the query a MinIO target mints fresh version ids +/// and the deployments drift apart. RustFS represents the null version +/// internally as the nil UUID while the S3 API addresses it as the literal +/// "null" (the delete path already maps it via `target_delete_version_id`), +/// and an empty id means the source object carries no version: send no query +/// so an unversioned target stays valid. +fn resolve_put_api_version_id(source_version_id: &str) -> Option<&str> { + if source_version_id.is_empty() { + None + } else if Uuid::parse_str(source_version_id).is_ok_and(|uuid| uuid.is_nil()) { + Some(rustfs_filemeta::NULL_VERSION_ID) + } else { + Some(source_version_id) + } +} + +/// Append `versionId=` to an already-built request URI. aws-sdk-s3's +/// `PutObjectInput` / `CreateMultipartUploadInput` expose no version id +/// member, so the query is spliced in via `map_request`, which runs at +/// `modify_before_signing`: the parameter becomes part of the SigV4 canonical +/// request. +fn append_version_id_query(uri: &str, version_id: &str) -> String { + let separator = if uri.contains('?') { '&' } else { '?' }; + format!("{uri}{separator}versionId={}", urlencoding::encode(version_id)) +} + #[derive(Debug, Clone)] pub struct AdvancedPutOptions { pub source_version_id: String, @@ -1831,6 +1862,7 @@ impl TargetClient { if !version_id.is_empty() { insert_header(&mut headers, SUFFIX_SOURCE_VERSION_ID, &version_id); } + let api_version_id = resolve_put_api_version_id(&version_id).map(ToOwned::to_owned); match builder .bucket(bucket) @@ -1845,6 +1877,11 @@ impl TargetClient { req.headers_mut().insert(key_str, value_str); } } + if let Some(version_id) = &api_version_id { + let uri = append_version_id_query(req.uri(), version_id); + req.set_uri(uri) + .map_err(aws_smithy_types::error::operation::BuildError::other)?; + } Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req) }) @@ -1893,6 +1930,9 @@ impl TargetClient { if opts.internal.replication_request { insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); } + // The remote version of a multipart replication is decided at initiate + // time; CompleteMultipartUpload does not read a versionId. + let api_version_id = resolve_put_api_version_id(&version_id).map(ToOwned::to_owned); match self .client @@ -1907,6 +1947,11 @@ impl TargetClient { req.headers_mut().insert(key_str, value_str); } } + if let Some(version_id) = &api_version_id { + let uri = append_version_id_query(req.uri(), version_id); + req.set_uri(uri) + .map_err(aws_smithy_types::error::operation::BuildError::other)?; + } Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req) }) .send() @@ -2679,6 +2724,91 @@ mod tests { ); } + #[tokio::test] + async fn put_object_sends_source_version_id_query_to_target() { + // MinIO reads the replicated version only from the `versionId` query + // parameter (its receive path ignores the x-*-source-version-id + // headers), so the query must carry the source version: a real UUID + // as-is, the internal nil-UUID null-version representation as the + // literal "null", and no query at all when the source object has no + // version (P0-5 RustFS->MinIO version drift). + let (client, request_uris) = recording_target_client(); + let version_id = Uuid::new_v4().to_string(); + let nil_version = Uuid::nil().to_string(); + for source_version in [version_id.as_str(), nil_version.as_str(), ""] { + let mut opts = PutObjectOptions::default(); + opts.internal.source_version_id = source_version.to_string(); + opts.internal.replication_request = true; + client + .put_object("target-bucket", "object", 4, ByteStream::from_static(b"data"), &opts) + .await + .expect("recorded put_object should succeed"); + } + + let request_uris = request_uris.lock().expect("recorded request lock should not be poisoned"); + assert_eq!(request_uris.len(), 3); + assert!( + request_uris[0].contains(&format!("versionId={version_id}")), + "replication put_object must carry the source version as a versionId query: {}", + request_uris[0] + ); + assert!( + request_uris[1].contains("versionId=null"), + "a nil-UUID (null) source version must be sent as the literal null: {}", + request_uris[1] + ); + assert!( + !request_uris[2].contains("versionId="), + "put_object without a source version must omit the versionId query: {}", + request_uris[2] + ); + } + + #[tokio::test] + async fn create_multipart_upload_sends_source_version_id_query_to_target() { + // The remote version of a multipart replication is decided at initiate + // time: CreateMultipartUpload must carry the source version in the + // `versionId` query (CompleteMultipartUpload does not read one). + let (client, request_uris) = recording_target_client(); + let version_id = Uuid::new_v4().to_string(); + let nil_version = Uuid::nil().to_string(); + for source_version in [version_id.as_str(), nil_version.as_str()] { + let mut opts = PutObjectOptions::default(); + opts.internal.source_version_id = source_version.to_string(); + opts.internal.replication_request = true; + let _ = client.create_multipart_upload("target-bucket", "object", &opts).await; + } + + let request_uris = request_uris.lock().expect("recorded request lock should not be poisoned"); + assert_eq!(request_uris.len(), 2); + assert!( + request_uris[0].contains(&format!("versionId={version_id}")), + "replication create_multipart_upload must carry the source version as a versionId query: {}", + request_uris[0] + ); + assert!( + request_uris[1].contains("versionId=null"), + "a nil-UUID (null) source version must be sent as the literal null: {}", + request_uris[1] + ); + } + + #[test] + fn put_object_headers_keep_source_version_id_for_legacy_receivers() { + // Older RustFS receivers have no versionId query support and fall back + // to the internal source-version-id headers (rolling-upgrade path); + // the query addition must never remove them. + let mut opts = PutObjectOptions::default(); + let version_id = Uuid::new_v4().to_string(); + opts.internal.source_version_id = version_id.clone(); + + assert_eq!( + rustfs_utils::http::get_header(&opts.header(), SUFFIX_SOURCE_VERSION_ID).as_deref(), + Some(version_id.as_str()), + "replication put requests must keep the internal source-version-id headers" + ); + } + #[test] fn put_object_headers_include_non_empty_source_etag_only() { let mut opts = PutObjectOptions::default(); diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index d6320ec46..fd6bf7648 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -302,7 +302,18 @@ pub async fn put_opts_with_replication_authorization( vid }; - let vid = vid.map(|v| v.as_str().trim().to_owned()); + // The S3 API addresses the null version as the literal "null" + // (MinIO-compatible replication senders, including RustFS itself, put it + // in the versionId query); normalize it to the internal nil-UUID + // representation exactly like get_opts / del_opts do. + let vid = vid.map(|v| { + let id = v.as_str().trim(); + if id.eq_ignore_ascii_case("null") { + Uuid::nil().to_string() + } else { + id.to_owned() + } + }); if let Some(ref id) = vid && *id != Uuid::nil().to_string() @@ -1351,6 +1362,22 @@ mod tests { } } + #[tokio::test] + async fn test_put_opts_normalizes_null_version_id() { + // MinIO-compatible replication senders (including RustFS itself since + // the P0-5 fix) address the null version as the literal "null" in the + // versionId query; the PUT / CreateMultipartUpload receive path must + // normalize it to the internal nil-UUID representation, exactly like + // get_opts / del_opts already do. + let headers = create_test_headers(); + + let opts = put_opts("test-bucket", "test-object", Some("null".to_string()), &headers, HashMap::new()) + .await + .expect("PUT with versionId=null must be accepted as the null version"); + + assert_eq!(opts.version_id, Some(Uuid::nil().to_string())); + } + #[tokio::test] async fn test_copy_dst_opts() { let headers = create_test_headers();