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

@@ -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,

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<Option<String>, Box<dyn Error + Send + Sync>> {
+    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(())
+}