diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index 24ca390ba..aa8792038 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -584,6 +584,10 @@ struct MultipartPart { body: Bytes, e_tag: String, digest: [u8; 16], + /// Plaintext length declared by an SSE-C passthrough sender + /// (`x-rustfs-replication-part-actual-size`); RustFS validates the 5 MiB + /// minimum against it rather than against the stored bytes. + actual_size: Option, } #[derive(Clone)] @@ -2624,6 +2628,11 @@ impl S3 for FakeBackend { async fn upload_part(&self, req: S3Request) -> S3Result> { let fault = request_fault(&req); + let declared_actual_size = req + .headers + .get("x-rustfs-replication-part-actual-size") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); let _body_permit = timeout(MAX_FAULT_DURATION, Arc::clone(&self.body_limit).acquire_owned()) .await .map_err(|_| s3s::s3_error!(RequestTimeout, "fake target body limiter wait exceeded 30 seconds"))? @@ -2665,6 +2674,7 @@ impl S3 for FakeBackend { body, e_tag: e_tag.clone(), digest, + actual_size: declared_actual_size, }, ); Ok(apply_response_fault( @@ -2734,7 +2744,9 @@ impl S3 for FakeBackend { if requested_etag != &stored.e_tag { return Err(s3s::s3_error!(InvalidPart, "part ETag does not match")); } - if index + 1 != requested_parts.len() && stored.body.len() < MIN_MULTIPART_PART_BYTES { + if index + 1 != requested_parts.len() + && stored.actual_size.unwrap_or(stored.body.len()) < MIN_MULTIPART_PART_BYTES + { return Err(s3s::s3_error!(EntityTooSmall, "non-final multipart part is smaller than 5 MiB")); } selected.push((*number, stored.clone())); diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index d6fc11e6a..799126792 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -10183,3 +10183,199 @@ async fn test_get_object_tagging_proxies_unreplicated_object_to_replication_targ target.shutdown().await; Ok(()) } + +// --------------------------------------------------------------------------- +// backlog#2363 +// --------------------------------------------------------------------------- + +/// Wait until the source reports a terminal replication status for `key`. +async fn wait_terminal_replication_status( + client: &Client, + bucket: &str, + key: &str, + ssec: bool, + timeout: Duration, +) -> Result> { + let customer_key = BASE64_STANDARD.encode_to_string(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let request = client.head_object().bucket(bucket).key(key); + let head = if ssec { + request + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await? + } else { + request.send().await? + }; + let status = head.replication_status().map(|status| status.as_str().to_string()); + if matches!(status.as_deref(), Some("COMPLETED") | Some("FAILED")) { + return Ok(status.unwrap_or_default()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("{bucket}/{key}: replication never reached a terminal status; last {status:?}").into()); + } + sleep(Duration::from_millis(250)).await; + } +} + +/// backlog#2363: SSE-C ciphertext passthrough of objects the source stored +/// compressed. The replica on a RustFS target must decrypt to the original +/// bytes for a single PUT and for a multipart upload. +#[tokio::test] +async fn test_bucket_replication_sse_c_compressed_passthrough() -> TestResult { + init_logging(); + const PART_SIZE: usize = 5 * 1024 * 1024; + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut target_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(FAST_SCANNER_ENV); + source_process_env.extend_from_slice(&[ + ("NO_PROXY", "127.0.0.1,localhost"), + ("HTTP_PROXY", ""), + ("HTTPS_PROXY", ""), + ("RUSTFS_COMPRESSION_ENABLED", "true"), + ("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true"), + ]); + source_env.start_rustfs_server_with_env(vec![], &source_process_env).await?; + target_env + .start_rustfs_server_without_cleanup_with_env(&[ + ("NO_PROXY", "127.0.0.1,localhost"), + ("HTTP_PROXY", ""), + ("HTTPS_PROXY", ""), + ]) + .await?; + + let source_bucket = "ssec-compressed-src"; + let target_bucket = "ssec-compressed-dst"; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + target_client.create_bucket().bucket(target_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + let customer_key = BASE64_STANDARD.encode_to_string(REPL17_SSEC_KEY); + let customer_key_md5 = sse_customer_key_md5_base64(REPL17_SSEC_KEY); + let text = |len: usize, seed: u32| -> Vec { + let mut out = Vec::with_capacity(len + 64); + let mut line = 0u64; + while out.len() < len { + out.extend_from_slice(format!("ssec compressed passthrough seed={seed} line={line} lorem ipsum dolor\n").as_bytes()); + line += 1; + } + out.truncate(len); + out + }; + + let single_key = "ssec-compressed-single.txt"; + let single_body = text(1024 * 1024 + 17, 1); + source_client + .put_object() + .bucket(source_bucket) + .key(single_key) + .content_type("text/plain") + .body(ByteStream::from(single_body.clone())) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + + let multipart_key = "ssec-compressed-multipart.txt"; + let multipart_parts = [text(PART_SIZE, 2), text(1024 * 1024 + 4096, 3)]; + let multipart_body: Vec = multipart_parts.concat(); + let created = source_client + .create_multipart_upload() + .bucket(source_bucket) + .key(multipart_key) + .content_type("text/plain") + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + let upload_id = created.upload_id().ok_or("missing multipart upload id")?.to_string(); + let mut completed = Vec::new(); + for (index, part) in multipart_parts.iter().enumerate() { + let part_number = i32::try_from(index + 1)?; + let uploaded = source_client + .upload_part() + .bucket(source_bucket) + .key(multipart_key) + .upload_id(&upload_id) + .part_number(part_number) + .body(ByteStream::from(part.clone())) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + completed.push( + CompletedPart::builder() + .part_number(part_number) + .set_e_tag(uploaded.e_tag().map(str::to_string)) + .build(), + ); + } + source_client + .complete_multipart_upload() + .bucket(source_bucket) + .key(multipart_key) + .upload_id(&upload_id) + .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build()) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await?; + + let mut failures = Vec::new(); + for (key, body) in [(single_key, &single_body), (multipart_key, &multipart_body)] { + let status = wait_terminal_replication_status(&source_client, source_bucket, key, true, Duration::from_secs(120)).await?; + if status != "COMPLETED" { + failures.push(format!("{key}: source reports {status}")); + continue; + } + let replica = target_client + .get_object() + .bucket(target_bucket) + .key(key) + .sse_customer_algorithm("AES256") + .sse_customer_key(&customer_key) + .sse_customer_key_md5(&customer_key_md5) + .send() + .await; + match replica { + Ok(replica) => { + let content_length = replica.content_length(); + match replica.body.collect().await { + Ok(collected) => { + let bytes = collected.into_bytes(); + if bytes.as_ref() != body.as_slice() { + failures.push(format!( + "{key}: replica bytes differ (content_length={content_length:?}, got {} bytes, want {})", + bytes.len(), + body.len() + )); + } + } + Err(err) => failures.push(format!("{key}: replica body read failed: {err}")), + } + } + Err(err) => failures.push(format!("{key}: replica GET failed: {err}")), + } + } + assert!( + failures.is_empty(), + "SSE-C compressed passthrough replicas must decrypt to the source bytes: {failures:?}" + ); + Ok(()) +} diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index 770f57b7f..390b528ce 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -1473,20 +1473,17 @@ fn layout_cases() -> Vec { true, ), // SSE-C passthrough replicates the stored ciphertext part by part; a - // compressible first part is stored well below 5 MiB and a standard - // target rejects it with EntityTooSmall. rc.5 fails the same way (see - // `rc5_baseline_replicates_multipart_layouts`), so the outcome is - // recorded rather than asserted here; tracked as rustfs/backlog#2363. - LayoutCase { - assert_replication: false, - ..case( - LAYOUT_PLAIN_BUCKET, - "plain/ssec-compressed-multipart-2.txt", - two.clone(), - layout_text(total(&two), 7), - true, - ) - }, + // compressible first part is stored well below 5 MiB, so the sender + // declares each part's plaintext length and the target validates the + // 5 MiB minimum against it (rustfs/backlog#2363). rc.5 as the sender + // still fails this layout (see `rc5_baseline_replicates_multipart_layouts`). + case( + LAYOUT_PLAIN_BUCKET, + "plain/ssec-compressed-multipart-2.txt", + two.clone(), + layout_text(total(&two), 7), + true, + ), case( LAYOUT_ENCRYPTED_BUCKET, "encrypted/single.bin", @@ -1838,27 +1835,20 @@ async fn direct_upgrade_from_rc5_preserves_multipart_layouts() -> TestResult { transport.uploaded_parts, expected, "{label}: stored parts must replicate as the same multipart layout" ); - // The current build can drive an existing object twice (two - // full CreateMultipartUpload/UploadPart/Complete rounds with - // distinct upload ids) while its status is still PENDING; the - // rc.5 baseline drives once. That is a scheduling difference, - // not a layout one, tracked as rustfs/backlog#2362. - assert!(transport.completes >= 1, "{label}: at least one CompleteMultipartUpload"); - if transport.completes > 1 { - tracing::warn!( - target: "e2e_test::upgrade_compatibility_test", - object = %label, - completes = transport.completes, - journal = ?transport.journal, - "existing-object replication drove the same object more than once (rustfs/backlog#2362)" - ); - } + // An object still PENDING when the next scanner cycle arrives is + // not driven a second time (rustfs/backlog#2362); the journal is + // logged so a duplicate round is visible if this ever regresses. + assert_eq!( + transport.completes, 1, + "{label}: exactly one CompleteMultipartUpload; journal {:?}", + transport.journal + ); assert_eq!( transport.single_puts, 0, "{label}: a multipart layout must not go out as a single PutObject" ); } else { - assert!(transport.single_puts >= 1, "{label}: a single PUT replicates as PutObject"); + assert_eq!(transport.single_puts, 1, "{label}: a single PUT replicates as exactly one PutObject"); assert!(transport.uploaded_parts.is_empty(), "{label}: a single PUT must not go out as multipart"); } if !case.ssec { @@ -1928,3 +1918,59 @@ async fn rc5_baseline_replicates_multipart_layouts() -> TestResult { tracing::info!(target: "e2e_test::upgrade_compatibility_test", ?summary, "rc.5 baseline replication outcomes"); Ok(()) } + +/// backlog#2362 under the same conditions that reproduced it with the rc.5 +/// writer, but with the workspace build on both sides so it runs in the +/// ordinary lane: every pre-existing layout is driven through exactly one +/// upload round even though the scanner re-scans it every second while the +/// first round is still in flight. +#[tokio::test] +async fn existing_object_replication_drives_each_layout_once() -> TestResult { + init_logging(); + let server_env = layout_server_env(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &server_env).await?; + let writer = env.create_s3_client(); + env.create_test_bucket(LAYOUT_PLAIN_BUCKET).await?; + env.create_test_bucket(LAYOUT_ENCRYPTED_BUCKET).await?; + enable_versioning(&writer, LAYOUT_PLAIN_BUCKET).await?; + enable_versioning(&writer, LAYOUT_ENCRYPTED_BUCKET).await?; + put_default_sse_s3_encryption(&writer, LAYOUT_ENCRYPTED_BUCKET).await?; + + let mut cases = layout_cases(); + for case in cases.iter_mut() { + layout_write(&writer, case).await?; + let head = layout_head(&writer, case).await?; + case.rc5_etag = head + .e_tag() + .ok_or_else(|| format!("{}: HEAD omitted the ETag", case.label()))? + .trim_matches('"') + .to_string(); + } + + // The objects come from an earlier process lifetime: the scanner starts + // cold and every object is a candidate at once. + env.restart_server_preserving_data(vec![], &server_env).await?; + let client = env.create_s3_client(); + let (_target, transports) = replicate_layouts(&env, &client, &cases).await?; + let mut duplicates = Vec::new(); + for (case, transport) in cases.iter().zip(&transports) { + assert_eq!( + transport.status, + "COMPLETED", + "{}: existing-object replication must complete", + case.label() + ); + let rounds = if case.is_multipart_layout() { + transport.completes + } else { + transport.single_puts + }; + if rounds != 1 { + duplicates.push(format!("{}: {rounds} upload rounds; journal {:?}", case.label(), transport.journal)); + } + } + assert!(duplicates.is_empty(), "each existing object must be driven exactly once: {duplicates:?}"); + Ok(()) +} diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index e20ad9dbe..d733a3134 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -75,6 +75,7 @@ use tracing::{debug, info, instrument, warn}; const EVENT_REPLICATION_WORKER_RESIZE_SKIPPED: &str = "replication_worker_resize_skipped"; const EVENT_REPLICATION_WORKER_RESIZED: &str = "replication_worker_resized"; const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure"; +const EVENT_REPLICATION_IN_FLIGHT_SKIPPED: &str = "replication_in_flight_skipped"; const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped"; const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered"; const EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE: &str = "replication_mrf_queue_unavailable"; @@ -1089,6 +1090,9 @@ pub struct ReplicationPool { workers: RwLock>>, lrg_workers: RwLock>>, + /// Object versions queued or being replicated right now (backlog#2362). + in_flight: Arc, + // MRF (Most Recent Failures) channels mrf_replica_tx: Sender, // Shared among N MRF workers; Arc allows spawning more than one worker. @@ -1147,6 +1151,7 @@ impl ReplicationPool { storage, workers: RwLock::new(Vec::new()), lrg_workers: RwLock::new(Vec::new()), + in_flight: Arc::new(ReplicationInFlight::default()), mrf_replica_tx, mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), mrf_save_tx, @@ -1202,12 +1207,13 @@ impl ReplicationPool { let active_counter = self.active_lrg_workers.clone(); let storage = self.storage.clone(); let stats = self.stats.clone(); + let in_flight = self.in_flight.clone(); let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), storage.clone()).await; + process_replication_operation(operation, stats.clone(), storage.clone(), in_flight.clone()).await; } }); @@ -1261,12 +1267,13 @@ impl ReplicationPool { let active_counter = self.active_workers.clone(); let stats = self.stats.clone(); let storage = self.storage.clone(); + let in_flight = self.in_flight.clone(); let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), storage.clone()).await; + process_replication_operation(operation, stats.clone(), storage.clone(), in_flight.clone()).await; } }); @@ -1305,6 +1312,7 @@ impl ReplicationPool { let active_counter = self.active_mrf_workers.clone(); let stats = self.stats.clone(); let storage = self.storage.clone(); + let in_flight = self.in_flight.clone(); let mrf_rx = Arc::clone(&self.mrf_replica_rx); let handle = tokio::spawn(async move { @@ -1324,7 +1332,7 @@ impl ReplicationPool { let Some(operation) = operation else { break }; let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), storage.clone()).await; + process_replication_operation(operation, stats.clone(), storage.clone(), in_flight.clone()).await; } }); self.task_handles.lock().await.push(handle); @@ -1454,6 +1462,24 @@ impl ReplicationPool { /// Queues a replica task pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission { + // A version that is already queued or being uploaded is not driven a + // second time: the scanner heal pass sees it as PENDING until the + // first upload lands and would otherwise re-queue it every cycle + // (backlog#2362). The key is released when the worker finishes, or + // below when no worker accepts the task. + if !self.in_flight.try_begin(&ri) { + debug!( + event = EVENT_REPLICATION_IN_FLIGHT_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket = %ri.bucket, + object = %ri.name, + version_id = ?ri.version_id, + op_type = ?ri.op_type, + "Replication task already in flight; not queued again" + ); + return ReplicationQueueAdmission::Skipped; + } let target_arns = ri.dsc.replicate_target_arns(); // If object is large, queue it to a static set of large workers if should_queue_large_object(ri.size) { @@ -1484,7 +1510,9 @@ impl ReplicationPool { let resize = large_worker_backpressure_resize(existing, self.active_lrg_workers(), max_l_workers); drop(lrg_workers); - // Queue to MRF if worker is busy. + // Queue to MRF if worker is busy. The MRF replay re-enters + // this function, so the version is no longer in flight. + self.in_flight.finish(&ri); let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "large_object").await; if let Some(resize) = resize { @@ -1493,6 +1521,7 @@ impl ReplicationPool { return admission; } } + self.in_flight.finish(&ri); return ReplicationQueueAdmission::Missed; } @@ -1501,6 +1530,7 @@ impl ReplicationPool { let ch = self.worker_queue_channel(&ri.op_type, &ri.bucket, &ri.name, ri.size).await; let Some(channel) = ch else { + self.in_flight.finish(&ri); return ReplicationQueueAdmission::Missed; }; @@ -1512,7 +1542,9 @@ impl ReplicationPool { self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type); self.stats.dec_target_q(&ri.bucket, &target_arns, ri.size); - // Queue to MRF if all workers are busy. + // Queue to MRF if all workers are busy. The MRF replay re-enters this + // function, so the version is no longer in flight. + self.in_flight.finish(&ri); let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "object").await; // Try to scale up workers based on priority @@ -1811,7 +1843,7 @@ impl ReplicationPool { ) { while let Some(operation) = rx.recv().await { let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), self.storage.clone()).await; + process_replication_operation(operation, stats.clone(), self.storage.clone(), self.in_flight.clone()).await; } } @@ -1829,7 +1861,7 @@ impl ReplicationPool { ) { while let Some(operation) = rx.recv().await { let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), storage.clone()).await; + process_replication_operation(operation, stats.clone(), storage.clone(), self.in_flight.clone()).await; } } @@ -1846,7 +1878,7 @@ impl ReplicationPool { ) { while let Some(operation) = rx.recv().await { let _active = ActiveWorkerGuard::new(active_counter.clone()); - process_replication_operation(operation, stats.clone(), self.storage.clone()).await; + process_replication_operation(operation, stats.clone(), self.storage.clone(), self.in_flight.clone()).await; } } @@ -2281,6 +2313,64 @@ impl ReplicationPool { } } +/// Object versions currently queued or being uploaded, keyed by bucket, +/// object name and version. `queue_replica_task` admits a version only once +/// while it is in flight; the scanner heal pass and MRF replays that arrive +/// in the meantime are `Skipped` instead of driving a second complete upload +/// (backlog#2362). Entries are removed when the worker finishes the task or +/// when no worker accepted it. +#[derive(Debug, Default)] +pub(crate) struct ReplicationInFlight { + keys: std::sync::Mutex)>>, +} + +impl ReplicationInFlight { + fn lock(&self) -> std::sync::MutexGuard<'_, std::collections::HashSet<(String, String, Option)>> { + self.keys.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Claim `ri`; `false` when the same version is already in flight. + fn try_begin(&self, ri: &ReplicateObjectInfo) -> bool { + self.lock().insert((ri.bucket.clone(), ri.name.clone(), ri.version_id)) + } + + fn finish(&self, ri: &ReplicateObjectInfo) { + self.lock().remove(&(ri.bucket.clone(), ri.name.clone(), ri.version_id)); + } + + #[cfg(test)] + fn len(&self) -> usize { + self.lock().len() + } +} + +/// Releases the in-flight claim when the worker is done with the task, +/// including when replication panics. +struct ReplicationInFlightGuard { + in_flight: Arc, + key: ReplicateObjectInfo, +} + +impl ReplicationInFlightGuard { + fn new(in_flight: Arc, ri: &ReplicateObjectInfo) -> Self { + Self { + in_flight, + key: ReplicateObjectInfo { + bucket: ri.bucket.clone(), + name: ri.name.clone(), + version_id: ri.version_id, + ..Default::default() + }, + } + } +} + +impl Drop for ReplicationInFlightGuard { + fn drop(&mut self) { + self.in_flight.finish(&self.key); + } +} + struct ActiveWorkerGuard { counter: Arc, } @@ -2342,10 +2432,12 @@ async fn process_replication_operation( operation: ReplicationOperation, stats: Arc, storage: Arc, + in_flight: Arc, ) { match operation { ReplicationOperation::Object(obj_info) => { let _backlog = ReplicationBacklogGuard::for_object(stats, obj_info.as_ref()); + let _in_flight = ReplicationInFlightGuard::new(in_flight, obj_info.as_ref()); replicate_object(*obj_info, storage).await; } ReplicationOperation::Delete(del_info) => { @@ -3707,6 +3799,7 @@ mod tests { stats: Arc::new(ReplicationStats::new()), workers: RwLock::new(Vec::new()), lrg_workers: RwLock::new(Vec::new()), + in_flight: Arc::new(ReplicationInFlight::default()), mrf_replica_tx, mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), mrf_save_tx, @@ -3773,6 +3866,90 @@ mod tests { assert_eq!(current_queue(&pool, "admission-bucket").await, (1, 4096)); } + #[tokio::test] + async fn queue_replica_task_admits_a_version_once_while_it_is_in_flight() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + let (tx, _rx) = mpsc::channel(4); + pool.workers.write().await.push(tx); + let ri = ReplicateObjectInfo { + bucket: "in-flight-bucket".to_string(), + name: "object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + size: 4096, + op_type: ReplicationType::Object, + ..Default::default() + }; + + assert_eq!(pool.queue_replica_task(ri.clone()).await, ReplicationQueueAdmission::Queued); + // backlog#2362: the scanner heal pass sees the version as PENDING + // until the worker lands it; a second request must not drive it again. + assert_eq!(pool.queue_replica_task(ri.clone()).await, ReplicationQueueAdmission::Skipped); + assert_eq!(current_queue(&pool, "in-flight-bucket").await, (1, 4096)); + + // Another version of the same key is independent work. + let newer = ReplicateObjectInfo { + version_id: Some(uuid::Uuid::new_v4()), + ..ri.clone() + }; + assert_eq!(pool.queue_replica_task(newer).await, ReplicationQueueAdmission::Queued); + assert_eq!(pool.in_flight.len(), 2); + + // Once the worker finishes, the same version may be queued again + // (for example after a FAILED status). + pool.in_flight.finish(&ri); + assert_eq!(pool.queue_replica_task(ri).await, ReplicationQueueAdmission::Queued); + assert_eq!(current_queue(&pool, "in-flight-bucket").await, (3, 3 * 4096)); + } + + #[tokio::test] + async fn queue_replica_task_releases_the_version_when_no_worker_accepts_it() { + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; + let ri = ReplicateObjectInfo { + bucket: "no-worker-bucket".to_string(), + name: "object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + size: 4096, + op_type: ReplicationType::Object, + ..Default::default() + }; + + // No worker channel: the task is missed and must not stay claimed. + assert_eq!(pool.queue_replica_task(ri.clone()).await, ReplicationQueueAdmission::Missed); + assert_eq!(pool.in_flight.len(), 0); + assert_eq!(pool.queue_replica_task(ri.clone()).await, ReplicationQueueAdmission::Missed); + + // A full worker channel hands the task to the MRF save path; the MRF + // replay re-enters the queue, so the claim is released here too. + let (tx, _rx) = mpsc::channel(1); + pool.workers.write().await.push(tx); + assert_eq!(pool.queue_replica_task(ri.clone()).await, ReplicationQueueAdmission::Queued); + let overflow = ReplicateObjectInfo { + version_id: Some(uuid::Uuid::new_v4()), + ..ri + }; + assert_eq!(pool.queue_replica_task(overflow).await, ReplicationQueueAdmission::Queued); + assert_eq!(pool.in_flight.len(), 1, "only the version held by the worker channel stays in flight"); + } + + #[test] + fn in_flight_guard_releases_the_version_on_drop() { + let in_flight = Arc::new(ReplicationInFlight::default()); + let ri = ReplicateObjectInfo { + bucket: "guard-bucket".to_string(), + name: "object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + ..Default::default() + }; + assert!(in_flight.try_begin(&ri)); + assert!(!in_flight.try_begin(&ri)); + { + let _guard = ReplicationInFlightGuard::new(in_flight.clone(), &ri); + assert_eq!(in_flight.len(), 1); + } + assert_eq!(in_flight.len(), 0); + assert!(in_flight.try_begin(&ri)); + } + #[tokio::test] async fn regular_worker_admission_counts_target_backlog_before_receive() { let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 40aa4c513..5d23fc6eb 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -4815,6 +4815,17 @@ async fn replicate_multipart_parts_and_complete( }; header_size = 0; + // Passthrough parts are the stored bytes; the replica learns each + // part's plaintext length from this header (backlog#2363). + let mut part_options = PutObjectPartOptions::default(); + if obj_opts.raw_data_movement_read && part_info.actual_size > 0 { + rustfs_utils::http::insert_header( + &mut part_options.custom_header, + rustfs_utils::http::SUFFIX_REPLICATION_PART_ACTUAL_SIZE, + part_info.actual_size.to_string(), + ); + } + let object_part = cli .put_object_part( dst_bucket, @@ -4823,7 +4834,7 @@ async fn replicate_multipart_parts_and_complete( part_plan.part_number, part_plan.part_size, byte_stream, - &PutObjectPartOptions { ..Default::default() }, + &part_options, ) .await .map_err(|e| std::io::Error::other(e.to_string()))?; @@ -6871,47 +6882,77 @@ mod tests { #[tokio::test] async fn multipart_transport_preserves_legacy_zero_actual_sizes() { - run_transport(4096, None).await; + run_transport(4096, None, false).await; } #[tokio::test] async fn multipart_transport_uploads_an_empty_last_part_without_reading_a_range() { - run_transport(0, None).await; + run_transport(0, None, false).await; } #[tokio::test] async fn multipart_transport_preserves_transformed_unknown_nonempty_parts() { for unknown_part in [(0, 0), (1, 0), (0, -1), (1, -1)] { - run_transport(4096, Some(unknown_part)).await; + run_transport(4096, Some(unknown_part), false).await; } } #[tokio::test] async fn multipart_transport_preserves_transformed_empty_tail() { - run_transport(0, Some((1, 0))).await; + run_transport(0, Some((1, 0)), false).await; } - async fn run_transport(tail_size: usize, unknown_part: Option<(usize, i64)>) { + /// SSE-C passthrough of a compressed object: the stored bytes go out + /// as-is and every UploadPart declares the part's plaintext length + /// (backlog#2363). + #[tokio::test] + async fn multipart_transport_declares_passthrough_part_lengths() { + run_transport(4096, None, true).await; + } + + async fn run_transport(tail_size: usize, unknown_part: Option<(usize, i64)>, passthrough: bool) { const FIRST_SIZE: usize = 5 * 1024 * 1024; + // The stored (compressed ciphertext) bytes of a passthrough part + // are shorter than the plaintext they represent. + const PASSTHROUGH_PLAINTEXT_FACTOR: usize = 4; let body = Bytes::from([vec![0x35; FIRST_SIZE], vec![0xa7; tail_size]].concat()); let etag = faster_hex::hex_string(rustfs_utils::hash::HashAlgorithm::Md5.hash_encode(&body).as_ref()); + let mut user_defined = if unknown_part.is_some() { + HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]) + } else { + HashMap::new() + }; + if passthrough { + user_defined.insert(rustfs_utils::http::SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string()); + rustfs_utils::http::insert_str( + &mut user_defined, + rustfs_utils::http::SUFFIX_COMPRESSION, + "klauspost/compress/s2".to_string(), + ); + } + let plaintext_len = |stored: usize| { + i64::try_from(if passthrough { + stored * PASSTHROUGH_PLAINTEXT_FACTOR + } else { + stored + }) + .expect("plaintext size") + }; let source = Arc::new(Source { info: ObjectInfo { size: i64::try_from(body.len() + if unknown_part.is_some() { 16 } else { 0 }).expect("stored size"), - actual_size: i64::try_from(body.len()).expect("body size"), + actual_size: plaintext_len(body.len()), etag: Some(etag.clone()), version_id: Some(Uuid::new_v4()), - user_defined: Arc::new(if unknown_part.is_some() { - HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]) - } else { - HashMap::new() - }), + user_defined: Arc::new(user_defined), parts: Arc::new(vec![ ObjectPartInfo { number: 1, size: FIRST_SIZE + if unknown_part.is_some() { 8 } else { 0 }, actual_size: if let Some((0, size)) = unknown_part { size + } else if passthrough { + plaintext_len(FIRST_SIZE) } else if unknown_part.is_some() || tail_size == 0 { i64::try_from(FIRST_SIZE).expect("first part size") } else { @@ -6924,6 +6965,8 @@ mod tests { size: tail_size + if unknown_part.is_some() { 8 } else { 0 }, actual_size: if let Some((1, size)) = unknown_part { size + } else if passthrough { + plaintext_len(tail_size) } else if unknown_part.is_some() { i64::try_from(tail_size).expect("tail logical size") } else { @@ -7002,6 +7045,7 @@ mod tests { let (put_opts, is_multipart) = replication_put_object_options("STANDARD", &source.info).expect("replication options"); let opts = ObjectOptions { version_id: source.info.version_id.map(|id| id.to_string()), + raw_data_movement_read: passthrough, ..Default::default() }; let reader = source @@ -7084,6 +7128,36 @@ mod tests { requests[index].headers.get("content-length").expect("part content length"), expected.len().to_string().as_str() ); + let declared = rustfs_utils::http::get_header( + &requests[index].headers, + rustfs_utils::http::SUFFIX_REPLICATION_PART_ACTUAL_SIZE, + ); + if passthrough { + assert_eq!( + declared.as_deref(), + Some(plaintext_len(expected.len()).to_string().as_str()), + "passthrough parts declare their plaintext length" + ); + } else { + assert!(declared.is_none(), "decrypted transport carries no passthrough part length"); + } + } + if passthrough { + let create = &requests[0]; + assert_eq!( + rustfs_utils::http::get_header(&create.headers, rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION) + .as_deref(), + Some("klauspost/compress/s2"), + "the session carries the source's compression scheme" + ); + assert_eq!( + rustfs_utils::http::get_header( + &create.headers, + rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE + ) + .as_deref(), + Some(plaintext_len(body.len()).to_string().as_str()) + ); } let complete = &requests[3]; assert_eq!(complete.method, http::Method::POST); diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 3f0f33369..387da2035 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -247,6 +247,23 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) meta.insert(key.to_string(), value.to_string()); } + // A compressed SSE-C object passes through as its stored bytes. The target + // cannot infer the compression layout from ciphertext, so the scheme and + // the plaintext size travel as transport headers; each UploadPart carries + // its own plaintext length (backlog#2363). + if is_ssec && let Some(scheme) = get_str(&object_info.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION) { + insert_header_map(&mut meta, rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION, scheme); + if let Ok(actual_size) = object_info.get_actual_size() + && actual_size >= 0 + { + insert_header_map( + &mut meta, + rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE, + actual_size.to_string(), + ); + } + } + // Managed SSE replicates as plaintext (the replication reader decrypts via // the object-encryption resolver) and re-encrypts on the target with the // target's own KMS. Send only the encryption intent — never the source @@ -626,6 +643,59 @@ mod tests { } } + #[test] + fn compressed_ssec_objects_declare_their_compression_layout_on_the_wire() { + use rustfs_utils::http::{ + SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_REPLICATION_COMPRESSION, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE, + insert_str, + }; + + let mut ssec_compressed = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); + insert_str(&mut ssec_compressed, SUFFIX_COMPRESSION, "klauspost/compress/s2".to_string()); + insert_str(&mut ssec_compressed, SUFFIX_ACTUAL_SIZE, "6295552".to_string()); + let object_info = ObjectInfo { + etag: Some("0123456789abcdef0123456789abcdef-2".to_string()), + size: 4321, + actual_size: 6295552, + user_defined: Arc::new(ssec_compressed), + ..Default::default() + }; + + // SSE-C passthrough sends stored bytes: the scheme and the plaintext + // size travel as transport headers, never as the internal key + // (backlog#2363). + let (options, _) = replication_put_object_options("STANDARD", &object_info).expect("ssec put options"); + assert_eq!( + get_header_map(&options.user_metadata, SUFFIX_REPLICATION_COMPRESSION).as_deref(), + Some("klauspost/compress/s2") + ); + assert_eq!( + get_header_map(&options.user_metadata, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE).as_deref(), + Some("6295552") + ); + assert!( + !options + .user_metadata + .keys() + .any(|key| rustfs_utils::http::is_internal_key(key)), + "internal metadata never leaves the source as plain metadata: {:?}", + options.user_metadata + ); + + // A compressed object that is not SSE-C is decompressed by the + // replication reader and travels as plaintext: no layout headers. + let mut plain_compressed = HashMap::new(); + insert_str(&mut plain_compressed, SUFFIX_COMPRESSION, "klauspost/compress/s2".to_string()); + insert_str(&mut plain_compressed, SUFFIX_ACTUAL_SIZE, "6295552".to_string()); + let plain = ObjectInfo { + user_defined: Arc::new(plain_compressed), + ..object_info + }; + let (options, _) = replication_put_object_options("STANDARD", &plain).expect("plain put options"); + assert!(get_header_map(&options.user_metadata, SUFFIX_REPLICATION_COMPRESSION).is_none()); + assert!(get_header_map(&options.user_metadata, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE).is_none()); + } + #[test] fn legacy_transformed_single_put_parts_keep_the_previous_replication_route() { let [_, (_, compressed), (_, encrypted), (_, ssec)] = replication_route_metadata(); diff --git a/crates/utils/src/http/header_compat.rs b/crates/utils/src/http/header_compat.rs index c36632d8e..4534161a3 100644 --- a/crates/utils/src/http/header_compat.rs +++ b/crates/utils/src/http/header_compat.rs @@ -43,6 +43,16 @@ pub const SUFFIX_FORCE_DELETE: &str = "force-delete"; pub const SUFFIX_INCLUDE_DELETED: &str = "include-deleted"; pub const SUFFIX_REPLICATION_RESET_STATUS: &str = "replication-reset-status"; pub const SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE: &str = "replication-actual-object-size"; +/// SSE-C ciphertext passthrough of an object the source stored compressed: +/// the stored compression scheme travels under this name so the replica +/// decompresses after decrypting (backlog#2363). +pub const SUFFIX_REPLICATION_COMPRESSION: &str = "replication-compression"; +/// Plaintext size of a compressed passthrough object (backlog#2363). +pub const SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE: &str = "replication-compression-actual-size"; +/// Plaintext length of one passthrough multipart part, sent on UploadPart so +/// the replica records the logical part size and checks the 5 MiB minimum +/// against it rather than against the stored bytes (backlog#2363). +pub const SUFFIX_REPLICATION_PART_ACTUAL_SIZE: &str = "replication-part-actual-size"; pub const SUFFIX_SOURCE_VERSION_ID: &str = "source-version-id"; pub const SUFFIX_SOURCE_MTIME: &str = "source-mtime"; pub const SUFFIX_SOURCE_ETAG: &str = "source-etag"; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 501fdf83a..00811cdad 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -1247,9 +1247,19 @@ impl DefaultMultipartUsecase { StreamReader::new(body_stream.map(|f| f.map_err(s3s_body_error_to_io))), ); - let is_disk_compressed = rustfs_utils::http::contains_key_str(&fi.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION); + // An SSE-C passthrough session stores ciphertext parts verbatim: a + // compression key restored on the session describes those stored + // bytes and must not add a second compression layer, and each part's + // plaintext length comes from the sender (backlog#2363). + let preserve_ciphertext = contains_key_str(&fi.user_defined, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT); + let is_disk_compressed = !preserve_ciphertext + && rustfs_utils::http::contains_key_str(&fi.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION); - let actual_size = size; + let actual_size = if preserve_ciphertext { + passthrough_part_actual_size(&req.headers).unwrap_or(size) + } else { + size + }; let mut md5hex = if let Some(base64_md5) = input.content_md5 { let md5 = base64_simd::STANDARD @@ -1286,7 +1296,6 @@ impl DefaultMultipartUsecase { // An SSE-C passthrough session stores ciphertext parts verbatim: no // material recovery, no validation against the (absent) customer key. - let preserve_ciphertext = contains_key_str(&fi.user_defined, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT); let has_ssec = !preserve_ciphertext && fi .user_defined @@ -1903,6 +1912,15 @@ impl DefaultMultipartUsecase { } } +/// Plaintext length of one SSE-C passthrough part, declared by the sender on +/// UploadPart (backlog#2363). +fn passthrough_part_actual_size(headers: &HeaderMap) -> Option { + get_header(headers, rustfs_utils::http::SUFFIX_REPLICATION_PART_ACTUAL_SIZE)? + .parse::() + .ok() + .filter(|size| *size > 0) +} + #[cfg(test)] mod tests { use super::*; diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index f5eb8d0f4..365e9af1c 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -1681,7 +1681,16 @@ impl DefaultObjectUsecase { }; rustfs_io_metrics::record_put_object_stage_duration_from("app_prelookup", prelookup_stage_start); - let actual_size = size; + // A compressed SSE-C passthrough body is the source's stored bytes: + // its logical length is the plaintext size restored from the + // transport headers (backlog#2363). The body itself is still read + // at its wire size. + let body_size = size; + let actual_size = if ciphertext_passthrough { + passthrough_compressed_actual_size(&opts.user_defined).unwrap_or(size) + } else { + size + }; if !ciphertext_passthrough && let Some(quota_check) = quota_check.as_ref() { ensure_object_size_within_quota( quota_check, @@ -1733,17 +1742,17 @@ impl DefaultObjectUsecase { } else { if use_zero_copy_eager_put_path { let zero_copy_start = std::time::Instant::now(); - let eager_body = read_zero_copy_put_body_exact(body, actual_size as usize).await?; - rustfs_io_metrics::record_zero_copy_write(actual_size as usize, zero_copy_start.elapsed().as_secs_f64() * 1000.0); + let eager_body = read_zero_copy_put_body_exact(body, body_size as usize).await?; + rustfs_io_metrics::record_zero_copy_write(body_size as usize, zero_copy_start.elapsed().as_secs_f64() * 1000.0); HashReader::from_stream(eager_body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? } else if use_empty_or_small_eager_put_path { - if (actual_size as usize) <= POOL_BYPASS_MAX_SIZE { + if (body_size as usize) <= POOL_BYPASS_MAX_SIZE { // Bypass BytesPool for very small objects to avoid Small-tier // Mutex contention under high concurrency. Direct allocation // for ≤4KiB is negligible cost. let eager_body = read_small_put_body_exact_direct( StreamReader::new(body.map(|f| f.map_err(s3s_body_error_to_io))), - actual_size as usize, + body_size as usize, ) .await?; HashReader::from_stream(eager_body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? @@ -1751,11 +1760,11 @@ impl DefaultObjectUsecase { let pool = get_concurrency_manager().bytes_pool(); let eager_body = read_small_put_body_exact_pooled( StreamReader::new(body.map(|f| f.map_err(s3s_body_error_to_io))), - actual_size as usize, + body_size as usize, pool.as_ref(), ) .await?; - let eager_reader = PooledBufferReader::new(eager_body, actual_size as usize); + let eager_reader = PooledBufferReader::new(eager_body, body_size as usize); HashReader::from_stream(eager_reader, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)? } } else { @@ -2105,6 +2114,18 @@ pub(super) fn previous_current_size_from_backfill(backfill: Option) -> Option { + if !rustfs_utils::http::contains_key_str(user_defined, SUFFIX_COMPRESSION) { + return None; + } + rustfs_utils::http::get_str(user_defined, SUFFIX_ACTUAL_SIZE)? + .parse::() + .ok() + .filter(|size| *size >= 0) +} + #[cfg(test)] mod tests { use super::*; diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index c723c4237..400b72a92 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -500,6 +500,25 @@ pub fn put_opts_from_headers_with_replication_authorization( if let Some(restored) = rustfs_utils::http::ssec_transport_to_stored_metadata(headers) { opts.user_defined.extend(restored); opts.preserve_ciphertext = true; + // A compressed passthrough object restores its compression + // layout as well, so the replica decompresses after decrypting + // (backlog#2363). The value is validated when the object is read. + if let Some(scheme) = get_header(headers, rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION) { + rustfs_utils::http::insert_str( + &mut opts.user_defined, + rustfs_utils::http::SUFFIX_COMPRESSION, + scheme.into_owned(), + ); + if let Some(actual_size) = get_header(headers, rustfs_utils::http::SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE) + && actual_size.parse::().is_ok_and(|size| size >= 0) + { + rustfs_utils::http::insert_str( + &mut opts.user_defined, + rustfs_utils::http::SUFFIX_ACTUAL_SIZE, + actual_size.into_owned(), + ); + } + } } if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned()); @@ -1588,6 +1607,59 @@ mod tests { assert!(!has_replication_retention_update(&missing_request, true)); } + #[test] + fn put_opts_from_headers_restores_the_compression_layout_only_for_ssec_passthrough() { + use rustfs_utils::http::object_encryption_keys::REPLICATION_SSEC_ALGORITHM_HEADER; + use rustfs_utils::http::{ + SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_REPLICATION_COMPRESSION, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE, + get_str, + }; + + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, SUFFIX_REPLICATION_COMPRESSION, "klauspost/compress/s2"); + insert_header(&mut headers, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE, "6295552"); + + // Without SSE-C transport headers the request is not a passthrough: + // the target compresses (or not) by its own policy and must not adopt + // a layout the body does not have. + let plain = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + assert!(!plain.preserve_ciphertext); + assert!(get_str(&plain.user_defined, SUFFIX_COMPRESSION).is_none()); + assert!(get_str(&plain.user_defined, SUFFIX_ACTUAL_SIZE).is_none()); + + headers.insert( + REPLICATION_SSEC_ALGORITHM_HEADER.parse::().unwrap(), + HeaderValue::from_static("AES256"), + ); + + // Unauthorized: inert, like the SSE-C transport itself. + let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created"); + assert!(!untrusted.preserve_ciphertext); + assert!(get_str(&untrusted.user_defined, SUFFIX_COMPRESSION).is_none()); + + // Authorized passthrough: the stored bytes are compressed ciphertext, + // so the replica records the scheme and the plaintext size + // (backlog#2363). + let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + assert!(trusted.preserve_ciphertext); + assert_eq!( + get_str(&trusted.user_defined, SUFFIX_COMPRESSION).as_deref(), + Some("klauspost/compress/s2") + ); + assert_eq!(get_str(&trusted.user_defined, SUFFIX_ACTUAL_SIZE).as_deref(), Some("6295552")); + + // A malformed plaintext size is dropped; the scheme alone still lets + // the read path derive the size from the parts. + insert_header(&mut headers, SUFFIX_REPLICATION_COMPRESSION_ACTUAL_SIZE, "-5"); + let malformed = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + assert!(get_str(&malformed.user_defined, SUFFIX_COMPRESSION).is_some()); + assert!(get_str(&malformed.user_defined, SUFFIX_ACTUAL_SIZE).is_none()); + } + #[test] fn test_put_opts_from_headers_gates_ssec_passthrough_on_authorization() { use rustfs_utils::http::object_encryption_keys::{