mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-07 20:46:11 +00:00
fix(replication): carry the compression layout through SSE-C passthrough (#7366)
* fix(scanner): drop the unused Digest import * fix(replication): carry the compression layout through SSE-C passthrough (#7372) * fix(replication): queue an in-flight version only once (#7376)
This commit is contained in:
Zhengchao An
@@ -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<usize>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -2624,6 +2628,11 @@ impl S3 for FakeBackend {
|
||||
|
||||
async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
|
||||
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::<usize>().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()));
|
||||
|
||||
@@ -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<String, Box<dyn Error + Send + Sync>> {
|
||||
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<u8> {
|
||||
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<u8> = 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(())
|
||||
}
|
||||
|
||||
@@ -1473,20 +1473,17 @@ fn layout_cases() -> Vec<LayoutCase> {
|
||||
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(())
|
||||
}
|
||||
|
||||
@@ -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<S: ReplicationStorage> {
|
||||
workers: RwLock<Vec<Sender<ReplicationOperation>>>,
|
||||
lrg_workers: RwLock<Vec<Sender<ReplicationOperation>>>,
|
||||
|
||||
/// Object versions queued or being replicated right now (backlog#2362).
|
||||
in_flight: Arc<ReplicationInFlight>,
|
||||
|
||||
// MRF (Most Recent Failures) channels
|
||||
mrf_replica_tx: Sender<ReplicationOperation>,
|
||||
// Shared among N MRF workers; Arc allows spawning more than one worker.
|
||||
@@ -1147,6 +1151,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
|
||||
/// 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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
return admission;
|
||||
}
|
||||
}
|
||||
self.in_flight.finish(&ri);
|
||||
return ReplicationQueueAdmission::Missed;
|
||||
}
|
||||
|
||||
@@ -1501,6 +1530,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
) {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
) {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
) {
|
||||
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<S: ReplicationStorage> ReplicationPool<S> {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<std::collections::HashSet<(String, String, Option<uuid::Uuid>)>>,
|
||||
}
|
||||
|
||||
impl ReplicationInFlight {
|
||||
fn lock(&self) -> std::sync::MutexGuard<'_, std::collections::HashSet<(String, String, Option<uuid::Uuid>)>> {
|
||||
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<ReplicationInFlight>,
|
||||
key: ReplicateObjectInfo,
|
||||
}
|
||||
|
||||
impl ReplicationInFlightGuard {
|
||||
fn new(in_flight: Arc<ReplicationInFlight>, 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<AtomicI32>,
|
||||
}
|
||||
@@ -2342,10 +2432,12 @@ async fn process_replication_operation<S: ReplicationStorage>(
|
||||
operation: ReplicationOperation,
|
||||
stats: Arc<ReplicationStats>,
|
||||
storage: Arc<S>,
|
||||
in_flight: Arc<ReplicationInFlight>,
|
||||
) {
|
||||
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;
|
||||
|
||||
@@ -4815,6 +4815,17 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
|
||||
};
|
||||
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<S: ReplicationObjectIO>(
|
||||
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);
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<i64> {
|
||||
get_header(headers, rustfs_utils::http::SUFFIX_REPLICATION_PART_ACTUAL_SIZE)?
|
||||
.parse::<i64>()
|
||||
.ok()
|
||||
.filter(|size| *size > 0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -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<OldCurrentSiz
|
||||
})
|
||||
}
|
||||
|
||||
/// Plaintext size of a compressed SSE-C passthrough body, restored from the
|
||||
/// replication transport headers into the object metadata (backlog#2363).
|
||||
fn passthrough_compressed_actual_size(user_defined: &HashMap<String, String>) -> Option<i64> {
|
||||
if !rustfs_utils::http::contains_key_str(user_defined, SUFFIX_COMPRESSION) {
|
||||
return None;
|
||||
}
|
||||
rustfs_utils::http::get_str(user_defined, SUFFIX_ACTUAL_SIZE)?
|
||||
.parse::<i64>()
|
||||
.ok()
|
||||
.filter(|size| *size >= 0)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -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::<i64>().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::<http::HeaderName>().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::{
|
||||
|
||||
Reference in New Issue
Block a user