From 8e214104f3d6d05ca423d4d1855dc298e8a98683 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 23 Jul 2026 16:33:54 +0800 Subject: [PATCH] perf(ecstore): add guarded encrypted multipart range seeks (#5145) * perf(ecstore): seek encrypted multipart Range GETs to the covering part boundary on the Legacy rio v1 backend Phase A of https://github.com/rustfs/backlog/issues/1316. Encrypted Range GETs on the default (non rio-v2) backend previously planned storage_offset=0, storage_length=oi.size and discarded the decrypted prefix, so a small Range on a large SSE object read, erasure-decoded, and decrypted the whole object. Eligible multipart objects now seek to the covering part boundary using only the per-part size/actual_size metadata facts; the multipart decrypt reader already handles streams starting at any part boundary, so rio and the on-disk format are untouched. Single-part objects, compressed payloads, defective parts tables, and zero-length ranges keep the previous full read, and RUSTFS_ENCRYPTED_RANGE_SEEK=false restores it globally. A new histogram rustfs_get_encrypted_range_read_amplification plus a full|part_seek path counter record the physical/plaintext amplification at the ReadPlan decision point. * fix(ecstore): guard encrypted multipart range seeks * fix(io-metrics): align encrypted range metric names with the rustfs_io_ prefix Every other metric in rustfs-io-metrics uses the rustfs_io_ prefix; the two encrypted-range-seek metrics were the only exception. Rename before first release so dashboards never see the unprefixed names. --- crates/ecstore/src/object_api/mod.rs | 31 + crates/ecstore/src/object_api/readers.rs | 1140 ++++++++++++++++- .../src/set_disk/core/io_primitives.rs | 73 +- crates/ecstore/src/set_disk/ops/multipart.rs | 1124 +++++++++++++++- crates/io-metrics/src/lib.rs | 13 + docs/architecture/compat-cleanup-register.md | 1 + 6 files changed, 2262 insertions(+), 120 deletions(-) diff --git a/crates/ecstore/src/object_api/mod.rs b/crates/ecstore/src/object_api/mod.rs index 4c5807630..5c57b69b2 100644 --- a/crates/ecstore/src/object_api/mod.rs +++ b/crates/ecstore/src/object_api/mod.rs @@ -51,6 +51,37 @@ use uuid::Uuid; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M +pub(crate) const ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX: &str = "encrypted-part-layout-quorum-candidate-v1"; +pub(crate) const ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX: &str = "encrypted-part-layout-quorum-v1"; +pub(crate) const ENV_RUSTFS_ENCRYPTED_RANGE_SEEK: &str = "RUSTFS_ENCRYPTED_RANGE_SEEK"; +pub(crate) const DEFAULT_RUSTFS_ENCRYPTED_RANGE_SEEK: bool = false; + +pub(crate) fn has_encrypted_part_layout_marker(metadata: &HashMap, suffix: &str, expected: &str) -> bool { + let mut value = None; + for (key, candidate) in metadata { + if !rustfs_utils::http::has_internal_suffix(key, suffix) { + continue; + } + if candidate.is_empty() || value.is_some_and(|current| current != candidate) { + return false; + } + value = Some(candidate); + } + value.is_some_and(|value| value == expected) +} + +pub(crate) fn legacy_encrypted_range_seek_enabled() -> bool { + // RUSTFS_COMPAT_TODO(backlog-1316): mixed-version MPUs need opt-in. Remove after all servers use candidate markers and uploadId locks. + #[cfg(test)] + { + rustfs_utils::get_env_bool(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, DEFAULT_RUSTFS_ENCRYPTED_RANGE_SEEK) + } + #[cfg(not(test))] + { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| rustfs_utils::get_env_bool(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, DEFAULT_RUSTFS_ENCRYPTED_RANGE_SEEK)) + } +} mod body_cache_hook; mod hook_slot; diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index 586ce0fa6..367e88a72 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -234,6 +234,211 @@ fn get_encrypted_offsets(oi: &ObjectInfo, offset: i64) -> Result<(i64, usize, us ))) } +/// Metric path label for a Legacy encrypted Range GET that kept the conservative full-object read. +const ENCRYPTED_RANGE_READ_PATH_FULL: &str = "full"; +/// Metric path label for a Legacy encrypted Range GET served by the part-boundary seek. +const ENCRYPTED_RANGE_READ_PATH_PART_SEEK: &str = "part_seek"; + +/// True when a Legacy (rio v1) encrypted Range GET may seek to the covering part +/// boundary instead of streaming the whole ciphertext. +/// +/// Every condition is required for correctness, not merely for benefit: +/// - `is_multipart`: single-part rio v1 streams have variable-length encrypted blocks +/// with no closed-form plaintext-to-physical mapping, so only part boundaries +/// (recorded in metadata) are safe seek targets. +/// - `!is_compressed`: under compression the requested offset addresses the +/// decompressed stream, not per-part plaintext, so stay on the full read. +/// - non-empty parts with `actual_size > 0` for every part: a part with +/// `actual_size <= 0` would make `part_plaintext_size` fall back to the physical +/// size and poison the plaintext cumulative sums. +/// - physical part sizes must add up to `oi.size`: guards against inconsistent +/// metadata scheduling reads past the object end in the erasure layer. +/// - plaintext part sizes must add up to the recorded decrypted object size. +/// - `requested_length > 0`: degenerate empty ranges keep their existing full-read +/// behavior (and its error surface) unchanged. +fn legacy_encrypted_seek_eligible( + oi: &ObjectInfo, + is_multipart: bool, + is_compressed: bool, + requested_length: i64, + recorded_plaintext_size: Option, +) -> bool { + if !legacy_encrypted_range_seek_enabled() || !is_multipart || is_compressed || requested_length <= 0 || oi.parts.is_empty() { + return false; + } + let Some(recorded_plaintext_size) = recorded_plaintext_size else { + return false; + }; + let Some(data_dir) = oi.data_dir.filter(|data_dir| !data_dir.is_nil()) else { + return false; + }; + let mut layout_token_buf = [0_u8; 36]; + let layout_token = data_dir.hyphenated().encode_lower(&mut layout_token_buf); + if !has_encrypted_part_layout_marker(&oi.user_defined, ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, layout_token) { + return false; + } + + let Some((physical_size, plaintext_size)) = legacy_part_layout_totals(&oi.parts) else { + return false; + }; + physical_size == oi.size && recorded_plaintext_size == plaintext_size +} + +fn legacy_part_layout_totals(parts: &[ObjectPartInfo]) -> Option<(i64, i64)> { + if parts.is_empty() { + return None; + } + // Complete votes on these four boundary fields in + // `resolve_read_part_from_responses`; final object reads additionally require + // the full `FileInfo` identity (including parts) to reach metadata quorum. + parts + .iter() + .try_fold((0_i64, 0_i64), |(physical_size, plaintext_size), part| { + let part_physical_size = i64::try_from(part.size).ok()?; + let part_plaintext_size = (part.actual_size > 0).then_some(part.actual_size)?; + Some(( + physical_size.checked_add(part_physical_size)?, + plaintext_size.checked_add(part_plaintext_size)?, + )) + }) +} + +/// Part-boundary seek offsets for a Legacy (rio v1) encrypted multipart object. +/// +/// Uses only the two per-part metadata facts — `size` (physical encrypted bytes on +/// disk) and `actual_size` (plaintext bytes) — and makes zero assumptions about the +/// encrypted block layout inside a part. Returns `(physical_offset, physical_length, +/// plaintext_skip_in_part, part_start_index, remaining_plaintext)` where +/// `physical_offset` is the ciphertext offset of the first part covering `offset`, +/// `physical_length` extends to the physical end of the last part covering +/// `offset + length - 1`, `plaintext_skip_in_part` is the decrypted plaintext to +/// discard before the range starts, and `remaining_plaintext` is the total plaintext +/// from the covering part through the end of the object (the same contract +/// `get_encrypted_offsets` hands to `into_reader`). +/// +/// Returns `None` when the range does not fall inside the parts table; the caller +/// must then keep the conservative full-object read. With the +/// `legacy_encrypted_seek_eligible` gate satisfied this cannot happen, because the +/// range spec already clamps `offset + length` to the summed part plaintext. +fn get_legacy_encrypted_offsets(oi: &ObjectInfo, offset: i64, length: i64) -> Option<(i64, i64, usize, usize, i64)> { + if offset < 0 || length <= 0 { + return None; + } + let end_plaintext = offset.checked_add(length)?.checked_sub(1)?; + let mut cumulative_plaintext_size = 0_i64; + let mut cumulative_encrypted_size = 0_i64; + + for (part_index, part) in oi.parts.iter().enumerate() { + let current_part_plaintext_size = part.actual_size; + let part_plaintext_end = cumulative_plaintext_size.checked_add(current_part_plaintext_size)?; + if offset < part_plaintext_end { + let mut covered_plaintext_end = cumulative_plaintext_size; + let mut physical_end = cumulative_encrypted_size; + let mut remaining_plaintext = 0_i64; + let mut covered = false; + for tail_part in &oi.parts[part_index..] { + remaining_plaintext = remaining_plaintext.checked_add(tail_part.actual_size)?; + if !covered { + physical_end = physical_end.checked_add(i64::try_from(tail_part.size).ok()?)?; + covered_plaintext_end = covered_plaintext_end.checked_add(tail_part.actual_size)?; + if end_plaintext < covered_plaintext_end { + covered = true; + } + } + } + if !covered { + return None; + } + let plaintext_skip_in_part = usize::try_from(offset - cumulative_plaintext_size).ok()?; + return Some(( + cumulative_encrypted_size, + physical_end.checked_sub(cumulative_encrypted_size)?, + plaintext_skip_in_part, + part_index, + remaining_plaintext, + )); + } + + cumulative_plaintext_size = part_plaintext_end; + cumulative_encrypted_size = cumulative_encrypted_size.checked_add(i64::try_from(part.size).ok()?)?; + } + + None +} + +/// Encrypted-range plan tuple shared by both Legacy (rio v1) build arms: +/// `(storage_offset, storage_length, decrypt_skip, plaintext_offset, +/// plaintext_length, total_plaintext_size, sequence_number, part_numbers)`. +type EncryptedRangePlan = (usize, i64, usize, usize, i64, usize, u32, Vec); + +/// Range plan for a Legacy (rio v1) encrypted object +/// (https://github.com/rustfs/backlog/issues/1316 Phase A). +/// +/// Eligible multipart objects seek to the covering part boundary: per-part physical +/// sizes are exact metadata facts, and the rio v1 multipart decrypt reader already +/// decrypts a stream that starts at any part boundary once it is handed the part +/// numbers from that part onward. Everything else — single-part objects, compressed +/// payloads, defective parts tables, the kill switch — keeps the conservative +/// full-object read: rio v1 encrypted blocks are variable-length, so existing +/// objects have no safe sub-part seek. +fn legacy_encrypted_range_plan( + oi: &ObjectInfo, + is_multipart: bool, + is_compressed: bool, + requested_offset: usize, + requested_length: i64, + full_plaintext_size: usize, + recorded_plaintext_size: Option, +) -> Result { + if legacy_encrypted_seek_eligible(oi, is_multipart, is_compressed, requested_length, recorded_plaintext_size) + && let Ok(requested_offset) = i64::try_from(requested_offset) + && let Some((physical_offset, physical_length, plaintext_skip_in_part, part_start_index, remaining_plaintext)) = + get_legacy_encrypted_offsets(oi, requested_offset, requested_length) + { + let storage_offset = usize::try_from(physical_offset) + .map_err(|_| Error::other(format!("invalid legacy encrypted offset {physical_offset}")))?; + let total_plaintext_size = usize::try_from(remaining_plaintext) + .map_err(|_| Error::other(format!("invalid legacy remaining decrypted size {remaining_plaintext}")))?; + record_encrypted_range_read_amplification(ENCRYPTED_RANGE_READ_PATH_PART_SEEK, physical_length, requested_length); + return Ok(( + storage_offset, + physical_length, + 0, + plaintext_skip_in_part, + requested_length, + total_plaintext_size, + 0, + multipart_part_numbers(&oi.parts[part_start_index..]), + )); + } + + // Skip the metric for compressed payloads: their requested length addresses + // the decompressed stream, so the physical/plaintext ratio would mix + // coordinate systems and pollute the histogram. + if !is_compressed { + record_encrypted_range_read_amplification(ENCRYPTED_RANGE_READ_PATH_FULL, oi.size, requested_length); + } + Ok(( + 0, + oi.size, + 0, + requested_offset, + requested_length, + full_plaintext_size, + 0, + multipart_part_numbers(&oi.parts), + )) +} + +/// Record path and physical/plaintext read amplification for one Legacy encrypted +/// Range GET at the ReadPlan decision point. +fn record_encrypted_range_read_amplification(path: &'static str, physical_length: i64, requested_length: i64) { + if requested_length <= 0 || physical_length < 0 { + return; + } + rustfs_io_metrics::record_get_encrypted_range_read_amplification(path, physical_length as f64 / requested_length as f64); +} + pub struct PutObjReader { pub stream: HashReader, } @@ -448,7 +653,8 @@ impl ReadPlan { #[cfg(feature = "rio-v2")] let encryption_backend = material.reader_backend; let is_multipart = is_multipart_encrypted_object(&oi.parts, oi.etag.as_deref()); - let plaintext_size = encrypted_plaintext_size(oi, is_multipart, is_compressed)?; + let recorded_plaintext_size = oi.encryption_original_size()?; + let plaintext_size = encrypted_plaintext_size(oi, is_multipart, is_compressed, recorded_plaintext_size)?; let full_plaintext_size = usize::try_from(plaintext_size).map_err(|_| Error::other(format!("invalid decrypted size {plaintext_size}")))?; let ( @@ -465,16 +671,15 @@ impl ReadPlan { #[cfg(feature = "rio-v2")] { if encryption_backend == crate::io_support::rio::ReadEncryptionBackend::Legacy { - ( - 0, - oi.size, - 0, + legacy_encrypted_range_plan( + oi, + is_multipart, + is_compressed, requested_offset, requested_length, full_plaintext_size, - 0, - multipart_part_numbers(&oi.parts), - ) + recorded_plaintext_size, + )? } else if is_compressed { let (physical_off, decompressed_skip, first_part_idx, decrypt_skip, seq_num) = get_compressed_offsets(oi, requested_offset as i64); @@ -512,16 +717,15 @@ impl ReadPlan { } #[cfg(not(feature = "rio-v2"))] { - ( - 0, - oi.size, - 0, + legacy_encrypted_range_plan( + oi, + is_multipart, + is_compressed, requested_offset, requested_length, full_plaintext_size, - 0, - multipart_part_numbers(&oi.parts), - ) + recorded_plaintext_size, + )? } } else { ( @@ -1084,16 +1288,23 @@ impl Drop for StreamConsumer { } } -fn encrypted_plaintext_size(oi: &ObjectInfo, is_multipart: bool, is_compressed: bool) -> Result { +fn encrypted_plaintext_size( + oi: &ObjectInfo, + is_multipart: bool, + is_compressed: bool, + recorded_plaintext_size: Option, +) -> Result { if is_compressed { return oi.get_actual_size().map_err(Into::into); } - if is_multipart { - return Ok(multipart_plaintext_size(&oi.parts, oi.decrypted_size()?)); + if is_multipart && recorded_plaintext_size.is_none() { + return Ok(legacy_part_layout_totals(&oi.parts) + .map(|(_, plaintext_size)| plaintext_size) + .unwrap_or(oi.size)); } - oi.decrypted_size().map_err(Into::into) + Ok(recorded_plaintext_size.unwrap_or(oi.size)) } fn is_multipart_encrypted_object(parts: &[ObjectPartInfo], etag: Option<&str>) -> bool { @@ -1104,12 +1315,6 @@ fn is_multipart_encrypted_object(parts: &[ObjectPartInfo], etag: Option<&str>) - etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false) } -fn multipart_plaintext_size(parts: &[ObjectPartInfo], fallback: i64) -> i64 { - let total: i64 = parts.iter().map(part_plaintext_size).sum(); - - if total > 0 { total } else { fallback } -} - fn multipart_part_numbers(parts: &[ObjectPartInfo]) -> Vec { parts.iter().map(|part| part.number).collect() } @@ -3166,6 +3371,891 @@ mod tests { assert_eq!(actual, b"56789ab"); } + /// Counts every byte pulled from the underlying ciphertext stream, standing in + /// for the physical bytes the erasure layer would decode for this read. + struct SpyReader { + inner: R, + bytes_read: Arc, + } + + impl SpyReader { + fn new(inner: R) -> (Self, Arc) { + let bytes_read = Arc::new(std::sync::atomic::AtomicU64::new(0)); + ( + Self { + inner, + bytes_read: bytes_read.clone(), + }, + bytes_read, + ) + } + } + + impl AsyncRead for SpyReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let before = buf.filled().len(); + let poll = Pin::new(&mut self.inner).poll_read(cx, buf); + if let Poll::Ready(Ok(())) = &poll { + let filled = (buf.filled().len() - before) as u64; + self.bytes_read.fetch_add(filled, std::sync::atomic::Ordering::Relaxed); + } + poll + } + } + + struct LegacyMultipartFixture { + plaintext: Vec, + ciphertext: Vec, + object_info: ObjectInfo, + part_physical_sizes: Vec, + } + + impl LegacyMultipartFixture { + /// Physical ciphertext offset where the given zero-based part starts. + fn physical_part_start(&self, part_index: usize) -> usize { + self.part_physical_sizes[..part_index].iter().sum() + } + } + + const LEGACY_FIXTURE_BASE_NONCE: [u8; 12] = [0x5A; 12]; + + fn legacy_fixture_part_plaintext(part_number: usize, size: usize) -> Vec { + (0..size).map(|i| ((i * 31 + part_number * 7) % 251) as u8).collect() + } + + /// Encrypts every part as its own rio v1 segment with the per-part nonce, + /// exactly like `WritePlan::apply` does for multipart uploads, and records the + /// physical/plaintext part sizes the way `PutObjectPart` persists them. + async fn build_legacy_multipart_fixture( + bucket: &str, + object: &str, + key_bytes: [u8; 32], + part_plain_sizes: &[usize], + mut user_defined: HashMap, + ) -> LegacyMultipartFixture { + let mut plaintext = Vec::new(); + let mut ciphertext = Vec::new(); + let mut parts = Vec::new(); + let mut part_physical_sizes = Vec::new(); + + for (part_index, &part_plain_size) in part_plain_sizes.iter().enumerate() { + let part_number = part_index + 1; + let part_plain = legacy_fixture_part_plaintext(part_number, part_plain_size); + let mut part_cipher = Vec::new(); + rustfs_rio::EncryptReader::new_multipart( + Cursor::new(part_plain.clone()), + key_bytes, + LEGACY_FIXTURE_BASE_NONCE, + part_number, + ) + .read_to_end(&mut part_cipher) + .await + .expect("encrypt multipart fixture part"); + + parts.push(ObjectPartInfo { + number: part_number, + size: part_cipher.len(), + actual_size: part_plain.len() as i64, + ..Default::default() + }); + part_physical_sizes.push(part_cipher.len()); + plaintext.extend_from_slice(&part_plain); + ciphertext.extend_from_slice(&part_cipher); + } + let data_dir = Uuid::from_u128(1); + rustfs_utils::http::insert_str(&mut user_defined, ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, data_dir.to_string()); + + let object_info = ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + size: ciphertext.len() as i64, + data_dir: Some(data_dir), + etag: Some(format!("d41d8cd98f00b204e9800998ecf8427e-{}", parts.len())), + parts: Arc::new(parts), + user_defined: Arc::new(user_defined), + ..Default::default() + }; + + LegacyMultipartFixture { + plaintext, + ciphertext, + object_info, + part_physical_sizes, + } + } + + fn legacy_ssec_multipart_metadata(key_bytes: [u8; 32], total_plaintext: usize) -> HashMap { + HashMap::from([ + ("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string()), + ( + "x-amz-server-side-encryption-customer-key-md5".to_string(), + BASE64_STANDARD.encode(md5_bytes(key_bytes)), + ), + ( + "x-amz-server-side-encryption-customer-original-size".to_string(), + total_plaintext.to_string(), + ), + ( + INTERNAL_ENCRYPTION_IV_HEADER.to_string(), + BASE64_STANDARD.encode(LEGACY_FIXTURE_BASE_NONCE), + ), + ]) + } + + async fn build_legacy_ssec_multipart_fixture(key_bytes: [u8; 32], part_plain_sizes: &[usize]) -> LegacyMultipartFixture { + let total_plaintext: usize = part_plain_sizes.iter().sum(); + build_legacy_multipart_fixture( + "bucket", + "legacy-multipart", + key_bytes, + part_plain_sizes, + legacy_ssec_multipart_metadata(key_bytes, total_plaintext), + ) + .await + } + + /// Serves exactly the ciphertext window the plan schedules — the contract the + /// erasure layer honors for `(storage_offset, storage_length)` — then decodes + /// the body end to end. Returns the body, the plan offsets, and the reported + /// object size. + async fn read_via_seek_window( + fixture: &LegacyMultipartFixture, + rs: Option, + opts: &ObjectOptions, + headers: &HeaderMap, + ) -> (Vec, usize, i64, i64) { + let plan = ReadPlan::build(rs.clone(), &fixture.object_info, opts, headers) + .await + .expect("plan should build for seek window read"); + let start = plan.storage_offset; + let window_len = usize::try_from(plan.storage_length).expect("storage length fits usize"); + assert!( + start + window_len <= fixture.ciphertext.len(), + "plan window [{start}, {}) exceeds ciphertext length {}", + start + window_len, + fixture.ciphertext.len() + ); + let window = fixture.ciphertext[start..start + window_len].to_vec(); + + let (mut reader, offset, length) = + GetObjectReader::new(Box::new(Cursor::new(window)), rs, &fixture.object_info, opts, headers) + .await + .expect("seek window read should build a reader"); + assert_eq!(offset, start, "reader offsets must match the probed plan"); + assert_eq!(length, plan.storage_length, "reader length must match the probed plan"); + + let mut body = Vec::new(); + reader.read_to_end(&mut body).await.expect("decode seek window body"); + (body, offset, length, reader.object_info.size) + } + + fn range(start: i64, end: i64) -> HTTPRangeSpec { + HTTPRangeSpec { + is_suffix_length: false, + start, + end, + } + } + + #[test] + fn test_get_legacy_encrypted_offsets_math() { + let parts = vec![ + ObjectPartInfo { + number: 1, + size: 1_000, + actual_size: 800, + ..Default::default() + }, + ObjectPartInfo { + number: 2, + size: 700, + actual_size: 500, + ..Default::default() + }, + ObjectPartInfo { + number: 3, + size: 400, + actual_size: 300, + ..Default::default() + }, + ]; + let oi = ObjectInfo { + size: 2_100, + parts: Arc::new(parts), + ..Default::default() + }; + + // Head range inside part 1 covers only part 1 but exposes the full tail plaintext. + assert_eq!(get_legacy_encrypted_offsets(&oi, 0, 10), Some((0, 1_000, 0, 0, 1_600))); + // Range ending exactly on the part 1 plaintext boundary stays within part 1. + assert_eq!(get_legacy_encrypted_offsets(&oi, 700, 100), Some((0, 1_000, 700, 0, 1_600))); + // One byte further crosses into part 2. + assert_eq!(get_legacy_encrypted_offsets(&oi, 700, 101), Some((0, 1_700, 700, 0, 1_600))); + // Range starting exactly at the part 2 boundary skips part 1 physically. + assert_eq!(get_legacy_encrypted_offsets(&oi, 800, 10), Some((1_000, 700, 0, 1, 800))); + // Last byte of the object covers only part 3. + assert_eq!(get_legacy_encrypted_offsets(&oi, 1_599, 1), Some((1_700, 400, 299, 2, 300))); + // Range spanning all parts covers the whole physical object. + assert_eq!(get_legacy_encrypted_offsets(&oi, 10, 1_590), Some((0, 2_100, 10, 0, 1_600))); + // Out-of-bounds offset yields no seek plan. + assert_eq!(get_legacy_encrypted_offsets(&oi, 1_600, 1), None); + assert_eq!(get_legacy_encrypted_offsets(&oi, 1_599, 2), None); + assert_eq!(get_legacy_encrypted_offsets(&oi, -1, 1), None); + assert_eq!(get_legacy_encrypted_offsets(&oi, 0, 0), None); + assert_eq!(get_legacy_encrypted_offsets(&oi, i64::MAX, 2), None); + } + + #[test] + fn test_legacy_part_layout_validation_rejects_invalid_totals() { + let assert_falls_back_to_object_size = |parts: Vec| { + let object_info = ObjectInfo { + size: 123, + parts: Arc::new(parts), + ..Default::default() + }; + assert_eq!( + encrypted_plaintext_size(&object_info, true, false, None).expect("invalid legacy layout should fall back"), + 123 + ); + }; + let valid = vec![ + ObjectPartInfo { + size: 7, + actual_size: 5, + ..Default::default() + }, + ObjectPartInfo { + size: 11, + actual_size: 9, + ..Default::default() + }, + ]; + assert_eq!(legacy_part_layout_totals(&valid), Some((18, 14))); + let object_info_without_recorded_size = ObjectInfo { + size: 18, + parts: Arc::new(valid.clone()), + ..Default::default() + }; + assert_eq!( + encrypted_plaintext_size(&object_info_without_recorded_size, true, false, None).expect("legacy total should resolve"), + 14 + ); + assert_eq!( + encrypted_plaintext_size(&object_info_without_recorded_size, true, false, Some(13)) + .expect("recorded total should win"), + 13 + ); + + for actual_size in [0, -1] { + let mut invalid = valid.clone(); + invalid[0].actual_size = actual_size; + assert_eq!(legacy_part_layout_totals(&invalid), None); + assert_falls_back_to_object_size(invalid); + } + assert_falls_back_to_object_size(Vec::new()); + + let mut plaintext_overflow = valid.clone(); + plaintext_overflow[0].actual_size = i64::MAX; + assert_eq!(legacy_part_layout_totals(&plaintext_overflow), None); + assert_falls_back_to_object_size(plaintext_overflow.clone()); + let plaintext_overflow_info = ObjectInfo { + parts: Arc::new(plaintext_overflow), + ..Default::default() + }; + assert_eq!(get_legacy_encrypted_offsets(&plaintext_overflow_info, 0, 1), None); + + let outer_plaintext_overflow = ObjectInfo { + parts: Arc::new(vec![ + ObjectPartInfo { + size: 1, + actual_size: i64::MAX - 1, + ..Default::default() + }, + ObjectPartInfo { + size: 1, + actual_size: 10, + ..Default::default() + }, + ]), + ..Default::default() + }; + assert_eq!(get_legacy_encrypted_offsets(&outer_plaintext_overflow, i64::MAX - 1, 1), None); + + #[cfg(target_pointer_width = "64")] + { + let mut physical_overflow = valid; + physical_overflow[0].size = usize::try_from(i64::MAX).expect("i64::MAX fits usize on 64-bit targets"); + assert_eq!(legacy_part_layout_totals(&physical_overflow), None); + let physical_overflow_info = ObjectInfo { + parts: Arc::new(physical_overflow), + ..Default::default() + }; + assert_eq!(get_legacy_encrypted_offsets(&physical_overflow_info, 5, 1), None); + + let conversion_overflow_size = usize::try_from(i64::MAX).expect("i64::MAX fits usize on 64-bit targets") + 1; + let conversion_overflow = vec![ObjectPartInfo { + size: conversion_overflow_size, + actual_size: 1, + ..Default::default() + }]; + assert_eq!(legacy_part_layout_totals(&conversion_overflow), None); + let conversion_overflow_info = ObjectInfo { + parts: Arc::new(conversion_overflow), + ..Default::default() + }; + assert_eq!(get_legacy_encrypted_offsets(&conversion_overflow_info, 0, 1), None); + + let outer_physical_overflow = ObjectInfo { + parts: Arc::new(vec![ + ObjectPartInfo { + size: usize::try_from(i64::MAX).expect("i64::MAX fits usize on 64-bit targets"), + actual_size: 1, + ..Default::default() + }, + ObjectPartInfo { + size: 1, + actual_size: 1, + ..Default::default() + }, + ObjectPartInfo { + size: 1, + actual_size: 1, + ..Default::default() + }, + ]), + ..Default::default() + }; + assert_eq!(get_legacy_encrypted_offsets(&outer_physical_overflow, 2, 1), None); + } + } + + #[tokio::test] + async fn test_legacy_ssec_multipart_range_seek_byte_exact_matrix() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x71; 32]; + // Part plaintexts exercise multi-block (2 x 8192 + tail), block + tail, + // and short single-block segments. + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions::default(); + let total_plaintext = fixture.plaintext.len() as i64; + let phys = &fixture.part_physical_sizes; + + // (range, expected physical offset, expected physical length) + let cases: Vec<(HTTPRangeSpec, usize, i64, &str)> = vec![ + (range(0, 9), 0, phys[0] as i64, "head of part 1"), + (range(20_100, 20_199), fixture.physical_part_start(1), phys[1] as i64, "inside part 2"), + (range(33_900, 33_999), fixture.physical_part_start(2), phys[2] as i64, "tail of part 3"), + ( + HTTPRangeSpec { + is_suffix_length: true, + start: 100, + end: -1, + }, + fixture.physical_part_start(2), + phys[2] as i64, + "suffix range", + ), + (range(19_999, 20_000), 0, (phys[0] + phys[1]) as i64, "part 1/2 boundary straddle"), + ( + range(8_191, 8_193), + 0, + phys[0] as i64, + "8 KiB encrypted-block boundary straddle in part 1", + ), + (range(19_999, 19_999), 0, phys[0] as i64, "last byte of part 1"), + ( + range(20_000, 20_000), + fixture.physical_part_start(1), + phys[1] as i64, + "first byte of part 2", + ), + (range(100, 33_950), 0, fixture.ciphertext.len() as i64, "range spanning all parts"), + (range(0, total_plaintext - 1), 0, fixture.ciphertext.len() as i64, "full range"), + ( + range(29_000, total_plaintext - 1), + fixture.physical_part_start(2), + phys[2] as i64, + "aligned suffix from part 3 start", + ), + ( + range(20_000, 29_100), + fixture.physical_part_start(1), + (phys[1] + phys[2]) as i64, + "part 2 into part 3", + ), + ]; + + for (rs, expected_offset, expected_length, label) in cases { + let (start, len) = rs.get_offset_length(total_plaintext).expect("valid case range"); + let expected_body = + &fixture.plaintext[start..start + usize::try_from(len).expect("valid range length fits usize")]; + + let (body, offset, length, reported_size) = read_via_seek_window(&fixture, Some(rs), &opts, &headers).await; + + assert_eq!(offset, expected_offset, "{label}: physical offset"); + assert_eq!(length, expected_length, "{label}: physical length"); + assert_eq!(reported_size, len, "{label}: reported plaintext size"); + assert_eq!(body.len(), expected_body.len(), "{label}: body length"); + assert_eq!(body, expected_body, "{label}: body bytes"); + } + }) + .await; + } + + /// The amplification-inversion guard: a small Range inside part 2 must schedule + /// only part 2 and must not pull more than part 2's ciphertext. The stream is + /// served from the planned offset all the way to the object end WITHOUT an end + /// truncation, so the covering-part bound is a real measurement of what the + /// decode chain pulls, not an artifact of window slicing. Reverting the Legacy + /// plan to the conservative `(0, oi.size)` full read fails the offset/length + /// assertions, and the pull bound as well: skipping the 20_100-byte plaintext + /// prefix would force the chain to pull all of part 1 first. + #[tokio::test] + async fn test_legacy_ssec_multipart_mid_range_reads_only_covering_part() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x72; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions::default(); + // 100 plaintext bytes inside part 2 (global plaintext [20_100, 20_200)). + let rs = range(20_100, 20_199); + + let plan = ReadPlan::build(Some(rs.clone()), &fixture.object_info, &opts, &headers) + .await + .expect("plan should build for mid range"); + let covering_part_physical = fixture.part_physical_sizes[1] as u64; + let object_physical = fixture.ciphertext.len() as u64; + assert_eq!( + plan.storage_offset, + fixture.physical_part_start(1), + "must seek to the covering part boundary" + ); + assert_eq!(plan.storage_length as u64, covering_part_physical, "must schedule only the covering part"); + + // Serve everything from the planned offset to the object end; the chain + // itself decides how much to pull. + let (spy, spy_bytes) = SpyReader::new(Cursor::new(fixture.ciphertext[plan.storage_offset..].to_vec())); + let (mut reader, _, _) = GetObjectReader::new(Box::new(spy), Some(rs), &fixture.object_info, &opts, &headers) + .await + .expect("mid range read should build a reader"); + let mut body = Vec::new(); + reader.read_to_end(&mut body).await.expect("decode mid range body"); + let pulled = spy_bytes.load(std::sync::atomic::Ordering::Relaxed); + + assert_eq!(body, &fixture.plaintext[20_100..20_200], "mid range body must stay byte-exact"); + assert!( + pulled <= covering_part_physical, + "physical read {pulled} exceeds covering part {covering_part_physical}" + ); + assert!( + pulled < object_physical, + "physical read {pulled} must stay below the whole object {object_physical}" + ); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_ssec_multipart_part_number_get_seeks_to_part() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x73; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions { + part_number: Some(2), + ..Default::default() + }; + + let (body, offset, length, reported_size) = read_via_seek_window(&fixture, None, &opts, &headers).await; + + assert_eq!(offset, fixture.physical_part_start(1), "partNumber GET must seek to part 2"); + assert_eq!(length, fixture.part_physical_sizes[1] as i64, "partNumber GET must cover only part 2"); + assert_eq!(reported_size, 9_000); + assert_eq!(body, &fixture.plaintext[20_000..29_000], "partNumber body must stay byte-exact"); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_managed_multipart_range_seek_byte_exact() { + async_with_vars( + [ + ("__RUSTFS_SSE_SIMPLE_CMK", Some(BASE64_STANDARD.encode([0u8; 32]))), + (ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true".to_string())), + ], + async { + let data_key = [0x74; 32]; + let encrypted_dek = encrypt_managed_dek_for_test(data_key, [0u8; 32]); + let total_plaintext: usize = 20_000 + 9_000 + 5_000; + let metadata = HashMap::from([ + ( + INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), + BASE64_STANDARD.encode(encrypted_dek.as_bytes()), + ), + (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "default".to_string()), + ( + INTERNAL_ENCRYPTION_IV_HEADER.to_string(), + BASE64_STANDARD.encode(LEGACY_FIXTURE_BASE_NONCE), + ), + (INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER.to_string(), total_plaintext.to_string()), + ]); + let fixture = + build_legacy_multipart_fixture("bucket", "managed-multipart", data_key, &[20_000, 9_000, 5_000], metadata) + .await; + let headers = HeaderMap::new(); + let opts = ObjectOptions::default(); + + for (rs, expected_offset, expected_length, label) in [ + ( + range(33_900, 33_999), + fixture.physical_part_start(2), + fixture.part_physical_sizes[2] as i64, + "managed tail range", + ), + ( + range(19_990, 20_010), + 0, + (fixture.part_physical_sizes[0] + fixture.part_physical_sizes[1]) as i64, + "managed boundary straddle", + ), + ] { + let (start, len) = rs.get_offset_length(total_plaintext as i64).expect("valid managed range"); + let expected_body = + &fixture.plaintext[start..start + usize::try_from(len).expect("valid managed range length fits usize")]; + + let (body, offset, length, reported_size) = read_via_seek_window(&fixture, Some(rs), &opts, &headers).await; + + assert_eq!(offset, expected_offset, "{label}: physical offset"); + assert_eq!(length, expected_length, "{label}: physical length"); + assert_eq!(reported_size, len, "{label}: reported plaintext size"); + assert_eq!(body, expected_body, "{label}: body bytes"); + } + }, + ) + .await; + } + + #[tokio::test] + async fn test_legacy_single_part_multipart_object_keeps_full_read_shape() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x75; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions::default(); + let rs = range(5_000, 5_099); + + let (body, offset, length, reported_size) = read_via_seek_window(&fixture, Some(rs), &opts, &headers).await; + + // A single-part object degenerates to the previous full-object plan. + assert_eq!(offset, 0); + assert_eq!(length, fixture.ciphertext.len() as i64); + assert_eq!(reported_size, 100); + assert_eq!(body, &fixture.plaintext[5_000..5_100]); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_kill_switch_restores_full_read() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("false"))], async { + let key_bytes = [0x76; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions::default(); + let rs = range(33_900, 33_999); + + let (body, offset, length, reported_size) = read_via_seek_window(&fixture, Some(rs), &opts, &headers).await; + + assert_eq!(offset, 0, "kill switch must restore the conservative full read"); + assert_eq!(length, fixture.ciphertext.len() as i64); + assert_eq!(reported_size, 100); + assert_eq!(body, &fixture.plaintext[33_900..34_000], "kill switch path must stay byte-exact"); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_defaults_disabled() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, None::<&str>)], async { + let key_bytes = [0x7D; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + + let plan = ReadPlan::build( + Some(range(33_900, 33_999)), + &fixture.object_info, + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await + .expect("default-disabled read plan should build"); + + assert_eq!(plan.storage_offset, 0); + assert_eq!(plan.storage_length, fixture.ciphertext.len() as i64); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_marker_validation() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let fixture = build_legacy_ssec_multipart_fixture([0x7F; 32], &[20_000, 9_000, 5_000]).await; + let recorded_size = fixture + .object_info + .encryption_original_size() + .expect("original size metadata should parse"); + assert!(!legacy_encrypted_seek_eligible(&fixture.object_info, true, false, 100, None)); + let layout_token = fixture + .object_info + .data_dir + .expect("fixture data dir should exist") + .to_string(); + let rustfs_key = format!("x-rustfs-internal-{ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX}"); + let minio_key = format!("x-minio-internal-{ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX}"); + let cases = [ + ( + "wrong", + HashMap::from([ + (rustfs_key.clone(), "wrong-token".to_string()), + (minio_key.clone(), "wrong-token".to_string()), + ]), + false, + ), + ( + "empty", + HashMap::from([(rustfs_key.clone(), String::new()), (minio_key.clone(), String::new())]), + false, + ), + ( + "conflicting", + HashMap::from([ + (rustfs_key.clone(), layout_token.clone()), + (minio_key.clone(), "wrong-token".to_string()), + ]), + false, + ), + ("rustfs only", HashMap::from([(rustfs_key, layout_token.clone())]), true), + ("minio only", HashMap::from([(minio_key, layout_token)]), true), + ]; + + for (case, marker_metadata, expected) in cases { + let mut object_info = fixture.object_info.clone(); + let metadata = Arc::make_mut(&mut object_info.user_defined); + rustfs_utils::http::remove_str(metadata, ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX); + metadata.extend(marker_metadata); + assert_eq!( + legacy_encrypted_seek_eligible(&object_info, true, false, 100, recorded_size), + expected, + "{case}" + ); + } + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_keeps_full_read_without_quorum_marker() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x7E; 32]; + let mut fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + rustfs_utils::http::remove_str( + Arc::make_mut(&mut fixture.object_info.user_defined), + ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + ); + + let (body, offset, length, reported_size) = read_via_seek_window( + &fixture, + Some(range(33_900, 33_999)), + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await; + + assert_eq!(offset, 0); + assert_eq!(length, fixture.ciphertext.len() as i64); + assert_eq!(reported_size, 100); + assert_eq!(body, &fixture.plaintext[33_900..34_000]); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_rejects_plaintext_total_mismatch() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x7C; 32]; + let mut fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let parts = Arc::make_mut(&mut fixture.object_info.parts); + parts[0].actual_size = 19_000; + + let (body, offset, length, reported_size) = read_via_seek_window( + &fixture, + Some(range(33_900, 33_999)), + &ObjectOptions::default(), + &ssec_headers_from_key(key_bytes), + ) + .await; + + assert_eq!(offset, 0, "a mismatched plaintext total must not control a physical seek"); + assert_eq!(length, fixture.ciphertext.len() as i64); + assert_eq!(reported_size, 100); + assert_eq!(body, &fixture.plaintext[33_900..34_000]); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_gate_rejects_zero_actual_size_part() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x77; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let mut object_info = fixture.object_info.clone(); + { + let parts = Arc::make_mut(&mut object_info.parts); + parts[1].actual_size = 0; + } + let headers = ssec_headers_from_key(key_bytes); + + let plan = ReadPlan::build(Some(range(0, 9)), &object_info, &ObjectOptions::default(), &headers) + .await + .expect("plan should build for zero actual_size sample"); + + assert_eq!(plan.storage_offset, 0, "a poisoned parts table must fall back to the full read"); + assert_eq!(plan.storage_length, object_info.size); + }) + .await; + } + + /// The physical part sizes must add up to `oi.size` for a seek to be safe; + /// inconsistent metadata must fall back to the previous full-object read + /// instead of scheduling an erasure read past the object end. + #[tokio::test] + async fn test_legacy_range_seek_gate_rejects_inconsistent_physical_sum() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x7B; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let mut object_info = fixture.object_info.clone(); + object_info.size -= 3; + let headers = ssec_headers_from_key(key_bytes); + + let plan = ReadPlan::build(Some(range(33_900, 33_999)), &object_info, &ObjectOptions::default(), &headers) + .await + .expect("plan should build for inconsistent physical sum sample"); + + assert_eq!(plan.storage_offset, 0, "an inconsistent physical sum must fall back to the full read"); + assert_eq!(plan.storage_length, object_info.size); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_range_seek_gate_rejects_compressed_encrypted_object() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x78; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let mut user_defined = fixture.object_info.user_defined.as_ref().clone(); + // "lz4" parses in both the default and the rio-v2 build (the MinIO + // "klauspost/compress/s2" token is only recognized under rio-v2). + user_defined.insert("x-minio-internal-compression".to_string(), "lz4".to_string()); + user_defined.insert("x-minio-internal-actual-size".to_string(), "34000".to_string()); + let mut object_info = fixture.object_info.clone(); + object_info.user_defined = Arc::new(user_defined); + let headers = ssec_headers_from_key(key_bytes); + + let plan = ReadPlan::build(Some(range(33_900, 33_999)), &object_info, &ObjectOptions::default(), &headers) + .await + .expect("plan should build for compressed encrypted sample"); + + assert_eq!(plan.storage_offset, 0, "compressed + encrypted must keep the conservative full read"); + assert_eq!(plan.storage_length, object_info.size); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_ssec_multipart_range_rejects_wrong_or_missing_key() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x79; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let rs = range(33_900, 33_999); + + let missing = match GetObjectReader::new( + Box::new(Cursor::new(fixture.ciphertext.clone())), + Some(rs.clone()), + &fixture.object_info, + &ObjectOptions::default(), + &HeaderMap::new(), + ) + .await + { + Ok(_) => panic!("missing SSE-C key must fail before any body is produced"), + Err(err) => err, + }; + assert!( + missing.to_string().contains("SSE-C"), + "missing-key failure must come from SSE-C validation: {missing}" + ); + + let wrong = match GetObjectReader::new( + Box::new(Cursor::new(fixture.ciphertext.clone())), + Some(rs), + &fixture.object_info, + &ObjectOptions::default(), + &ssec_headers_from_key([0x00; 32]), + ) + .await + { + Ok(_) => panic!("wrong SSE-C key must fail before any body is produced"), + Err(err) => err, + }; + assert!( + wrong.to_string().contains("SSE-C key does not match object metadata"), + "wrong-key failure must come from the stored key check: {wrong}" + ); + }) + .await; + } + + #[tokio::test] + async fn test_legacy_ssec_multipart_seek_tamper_fails_hard_with_no_plaintext() { + async_with_vars([(ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let key_bytes = [0x7A; 32]; + let fixture = build_legacy_ssec_multipart_fixture(key_bytes, &[20_000, 9_000, 5_000]).await; + let headers = ssec_headers_from_key(key_bytes); + let opts = ObjectOptions::default(); + let rs = range(33_900, 33_999); + + let plan = ReadPlan::build(Some(rs.clone()), &fixture.object_info, &opts, &headers) + .await + .expect("plan should build for tamper test"); + let start = plan.storage_offset; + let window_len = usize::try_from(plan.storage_length).expect("tamper window length fits usize"); + assert_eq!(start, fixture.physical_part_start(2), "tamper test must exercise the seek path"); + + let mut window = fixture.ciphertext[start..start + window_len].to_vec(); + // Flip one ciphertext byte inside the first encrypted block the seek + // lands on (past the 8-byte header and the length prefix). + window[64] ^= 0xFF; + + let (mut reader, _, _) = + GetObjectReader::new(Box::new(Cursor::new(window)), Some(rs), &fixture.object_info, &opts, &headers) + .await + .expect("reader builds before the tampered block is decrypted"); + + let mut body = Vec::new(); + let err = reader + .read_to_end(&mut body) + .await + .expect_err("tampered ciphertext must fail the read"); + assert!(body.is_empty(), "no unauthenticated plaintext may be emitted, got {} bytes", body.len()); + let message = err.to_string(); + assert!( + message.contains("decrypt") || message.contains("CRC32") || message.contains("Invalid encrypted block"), + "unexpected tamper error: {message}" + ); + }) + .await; + } + #[cfg(feature = "rio-v2")] #[tokio::test] async fn test_get_object_reader_uses_dare_package_offset_for_large_ssec_ranges() { diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 9d86328bf..abf567f49 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -602,8 +602,7 @@ pub(in crate::set_disk) fn resolve_read_part_from_responses( responses: &[Option>], read_quorum: usize, ) -> disk::error::Result { - let mut etag_quorum = HashMap::new(); - let mut part_infos = Vec::new(); + let mut part_quorum: HashMap<(&str, usize, usize, i64), (usize, &ObjectPartInfo)> = HashMap::new(); let mut present_count = 0usize; let mut missing_count = 0usize; let mut transient_error_count = 0usize; @@ -621,8 +620,10 @@ pub(in crate::set_disk) fn resolve_read_part_from_responses( if !parts[part_idx].etag.is_empty() { present_count += 1; - *etag_quorum.entry(parts[part_idx].etag.clone()).or_insert(0) += 1; - part_infos.push(parts[part_idx].clone()); + let part = &parts[part_idx]; + let key = (part.etag.as_str(), part.number, part.size, part.actual_size); + let (count, _) = part_quorum.entry(key).or_insert((0, part)); + *count += 1; continue; } @@ -633,31 +634,12 @@ pub(in crate::set_disk) fn resolve_read_part_from_responses( } } - let mut max_etag_quorum = 0; - let mut max_etag = None; - for (etag, quorum) in etag_quorum.iter() { - if quorum > &max_etag_quorum { - max_etag_quorum = *quorum; - max_etag = Some(etag); - } - } - let max_quorum = max_etag_quorum.max(missing_count); - - let mut found = None; - for info in part_infos.iter() { - if let Some(etag) = max_etag - && info.etag == *etag - { - found = Some(info.clone()); - break; - } - } - - if let (Some(found), Some(max_etag)) = (found, max_etag) - && !found.etag.is_empty() - && etag_quorum.get(max_etag).unwrap_or(&0) >= &read_quorum + let max_part = part_quorum.values().max_by_key(|(count, _)| count); + let max_quorum = max_part.map_or(0, |(count, _)| *count).max(missing_count); + if let Some((count, part)) = max_part + && *count >= read_quorum { - return Ok(found); + return Ok((*part).clone()); } if missing_count >= read_quorum { @@ -5134,17 +5116,34 @@ mod tests { } #[test] - fn resolve_read_part_returns_part_when_etag_reaches_quorum() { - let responses = vec![ - Some(vec![read_part_test_part(1, "winner")]), - Some(vec![read_part_test_part(1, "loser")]), - Some(vec![read_part_test_part(1, "winner")]), - ]; + fn resolve_read_part_requires_layout_fields_to_reach_quorum() { + let mut valid = read_part_test_part(1, "winner"); + valid.size = 100; + valid.actual_size = 90; + let mut wrong_etag = valid.clone(); + wrong_etag.etag = "loser".to_string(); + let mut wrong_number = valid.clone(); + wrong_number.number = 2; + let mut wrong_size = valid.clone(); + wrong_size.size = 50; + let mut wrong_actual_size = valid.clone(); + wrong_actual_size.actual_size = 40; - let part = resolve_read_part_from_responses("bucket", "upload/part.1.meta", 1, 0, 1, &responses, 2) - .expect("etag quorum should resolve the present part"); + for (field, corrupted) in [ + ("etag", wrong_etag), + ("number", wrong_number), + ("size", wrong_size), + ("actual_size", wrong_actual_size), + ] { + let responses = vec![Some(vec![corrupted]), Some(vec![valid.clone()]), Some(vec![valid.clone()])]; + let part = resolve_read_part_from_responses("bucket", "upload/part.1.meta", 1, 0, 1, &responses, 2) + .unwrap_or_else(|err| panic!("{field} mismatch must not defeat layout quorum: {err}")); - assert_eq!(part.etag, "winner"); + assert_eq!(part.etag, "winner", "{field}"); + assert_eq!(part.number, 1, "{field}"); + assert_eq!(part.size, 100, "{field}"); + assert_eq!(part.actual_size, 90, "{field}"); + } } #[test] diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 0a84df2be..1d922e09f 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -26,6 +26,93 @@ use std::future::Future; use std::time::Duration; use tokio::task::JoinSet; +#[cfg(test)] +#[derive(Clone, Copy, PartialEq, Eq)] +enum MultipartCommitPause { + PutPartBeforeLockLost, + PutPartAfterRename, + BeforeLockLost, + AfterRename, +} + +#[cfg(test)] +struct MultipartCommitBarrierState { + bucket: String, + object: String, + pause: MultipartCommitPause, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(test)] +struct MultipartCommitBarrier { + state: Arc, +} + +#[cfg(test)] +static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock>>> = + std::sync::OnceLock::new(); + +#[cfg(test)] +impl MultipartCommitBarrier { + fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self { + let state = Arc::new(MultipartCommitBarrierState { + bucket: bucket.to_string(), + object: object.to_string(), + pause, + arrived: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let mut slot = MULTIPART_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("multipart commit barrier mutex should not poison"); + assert!(slot.is_none(), "multipart commit barrier must be installed by one test at a time"); + *slot = Some(Arc::clone(&state)); + drop(slot); + Self { state } + } + + async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("multipart completion should reach the deterministic commit barrier"); + } + + fn release(&self) { + self.state.release.notify_one(); + } +} + +#[cfg(test)] +impl Drop for MultipartCommitBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = MULTIPART_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("multipart commit barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(test)] +async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) { + let barrier = MULTIPART_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("multipart commit barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + fn map_upload_id_metadata_error(bucket: &str, object: &str, upload_id: &str, err: DiskError) -> Error { if err == DiskError::FileNotFound { return StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()); @@ -533,9 +620,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // shards from two generations, each individually bitrot-valid, that only // surface as silent corruption at read time (backlog#853). A write lock // scoped to the uploadId namespace makes each commit atomic across disks, - // so the last committer wins consistently. Mirrors MinIO's per-uploadID - // NS lock; distinct from the object lock held by complete_multipart_upload - // (disjoint namespaces, no lock-ordering cycle). + // so the last committer wins consistently. A guarded completion takes + // the object lock before this upload lock to preserve global ordering. let _upload_commit_guard = if opts.no_lock { None } else { @@ -544,6 +630,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { .await?, ) }; + self.check_upload_id_exists(bucket, object, upload_id, false).await?; + #[cfg(test)] + pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await; + if _upload_commit_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "put_object_part_commit", + bucket: RUSTFS_META_MULTIPART_BUCKET.to_string(), + object: upload_id_path.clone(), + required: 1, + achieved: 0, + }); + } let _ = self .rename_part( @@ -564,6 +662,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ) .await?; + #[cfg(test)] + pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await; drop(_upload_commit_guard); let ret: PartInfo = PartInfo { @@ -866,6 +966,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let disks = disks.clone(); let mut user_defined = opts.user_defined.clone(); + rustfs_utils::http::metadata_compat::remove_str( + &mut user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + ); + rustfs_utils::http::metadata_compat::remove_str( + &mut user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + ); if let Some(ref etag) = opts.preserve_etag { user_defined.insert("etag".to_owned(), etag.clone()); @@ -902,7 +1010,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { fi.version_id = Some(Uuid::new_v4()); } - fi.data_dir = Some(Uuid::new_v4()); + let data_dir = Uuid::new_v4(); + fi.data_dir = Some(data_dir); + if crate::object_api::legacy_encrypted_range_seek_enabled() + && !opts.no_lock + && should_persist_encryption_original_size(&user_defined) + { + insert_str( + &mut user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + data_dir.to_string(), + ); + } if let Some(cssum) = get_header_map(&user_defined, SUFFIX_REPLICATION_SSEC_CRC) && !cssum.is_empty() @@ -935,7 +1054,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { user_defined.insert(RUSTFS_MULTIPART_OBJECT_KEY.to_string(), object.to_string()); let (shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi); - let mod_time = opts.mod_time.unwrap_or_else(OffsetDateTime::now_utc); for f in parts_metadatas.iter_mut() { @@ -944,10 +1062,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { f.fresh = true; } - // fi.mod_time = Some(now); - let upload_uuid = format!("{}x{}", Uuid::new_v4(), mod_time.unix_timestamp_nanos()); + // fi.mod_time = Some(now); + let upload_id = runtime_sources::deployment_upload_id(&upload_uuid); let upload_path = Self::get_upload_id_dir(bucket, object, upload_uuid.as_str()); @@ -1023,6 +1141,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { crate::hp_guard!("SetDisks::complete_multipart_upload"); self.invalidate_get_object_metadata_cache(bucket, object).await; + let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); + let range_seek_rollout_enabled = crate::object_api::legacy_encrypted_range_seek_enabled() && !opts.no_lock; let mut object_lock_guard = None; if opts.http_preconditions.is_some() { @@ -1039,7 +1159,51 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?; - let (mut fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; + let (mut fi, mut files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; + let has_layout_candidate = range_seek_rollout_enabled + && fi + .data_dir + .filter(|data_dir| !data_dir.is_nil()) + .map(|data_dir| data_dir.to_string()) + .is_some_and(|token| { + crate::object_api::has_encrypted_part_layout_marker( + &fi.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + &token, + ) + }); + let upload_guard = if has_layout_candidate { + if object_lock_guard.is_none() { + object_lock_guard = Some( + self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object) + .await?, + ); + } + let guard = self + .acquire_multipart_upload_write_lock("complete_multipart_upload_commit", bucket, object, upload_id, opts) + .await?; + (fi, files_metas) = self.check_upload_id_exists(bucket, object, upload_id, true).await?; + guard + } else { + None + }; + let quorum_validated_layout_token = upload_guard.as_ref().and_then(|_| { + fi.data_dir + .filter(|data_dir| !data_dir.is_nil()) + .map(|data_dir| data_dir.to_string()) + .filter(|token| { + crate::object_api::has_encrypted_part_layout_marker( + &fi.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + token, + ) + }) + }); + rustfs_utils::http::metadata_compat::remove_str( + &mut fi.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + ); + rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX); if expected_restore_operation_id.is_some() { rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_RESTORE_OPERATION_ID); } @@ -1052,8 +1216,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } else { fi.version_id = None; } - let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id); - let write_quorum = fi.write_quorum(self.default_write_quorum()); let read_quorum = fi.read_quorum(self.default_read_quorum()); @@ -1357,6 +1519,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { fi.metadata.insert("etag".to_owned(), etag); let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata); + if persist_encryption_original_size && let Some(layout_token) = quorum_validated_layout_token { + insert_str(&mut fi.metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, layout_token); + } else { + rustfs_utils::http::metadata_compat::remove_str( + &mut fi.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + ); + } if opts.replication_request { if let Some(actual_size) = get_str(&opts.user_defined, SUFFIX_ACTUAL_OBJECT_SIZE_CAP) { @@ -1418,13 +1588,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { .await?, ); } - // Phase 2 (backlog#899): fence the commit on lock loss before any destructive // step. If the refresh heartbeat has observed a refresh-quorum loss, another // writer may have re-acquired this object's lock; proceeding would race a // double-write. Abort with a retryable error *before* cleanup_multipart_path // (which removes the source parts) and rename_data (which commits the object), // so a lost lock leaves the upload intact and retryable. + #[cfg(test)] + pause_multipart_commit(bucket, object, MultipartCommitPause::BeforeLockLost).await; if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { return Err(StorageError::NamespaceLockQuorumUnavailable { mode: "complete_multipart_upload_commit", @@ -1434,6 +1605,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { achieved: 0, }); } + if upload_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode: "complete_multipart_upload_commit", + bucket: RUSTFS_META_MULTIPART_BUCKET.to_string(), + object: upload_id_path.clone(), + required: 1, + achieved: 0, + }); + } self.require_current_restore_operation_id( bucket, @@ -1508,6 +1688,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ); } + #[cfg(test)] + pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await; + drop(upload_guard); drop(object_lock_guard); // drop object lock guard to release the lock // backlog#1321: enqueue heal only when the committed replicas actually @@ -1583,15 +1766,225 @@ mod tests { use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::DiskAPI as _; use crate::disk::{endpoint::Endpoint, format::FormatV3}; + use crate::layout::endpoints::SetupType; use crate::set_disk::ops::object::hermetic_set_disks_support::{ - hermetic_set_disks, hermetic_set_disks_for_pool_with_default_parity, + hermetic_set_disks, hermetic_set_disks_for_pool_with_default_parity, hermetic_set_disks_with_lockers, }; use crate::storage_api_contracts::namespace::NamespaceLocking as _; + use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use rustfs_config::server_config::KVS; use rustfs_lock::{LockClient, client::local::LocalClient}; use serial_test::serial; + use std::sync::atomic::{AtomicUsize, Ordering}; use tempfile::TempDir; - use tokio::sync::RwLock; + use tokio::sync::{Notify, RwLock}; + + struct SetupTypeGuard { + previous: SetupType, + } + + impl SetupTypeGuard { + async fn switch_to(next: SetupType) -> Self { + let previous = crate::runtime::sources::current_setup_type().await; + crate::runtime::sources::set_setup_type(next).await; + Self { previous } + } + } + + impl Drop for SetupTypeGuard { + fn drop(&mut self) { + let previous = self.previous.clone(); + let handle = tokio::runtime::Handle::current(); + std::thread::spawn(move || { + handle.block_on(async { + crate::runtime::sources::set_setup_type(previous).await; + }); + }) + .join() + .expect("setup type restore thread should not panic"); + } + } + + #[derive(Debug)] + struct SignalingLockClient { + inner: Arc, + target: std::sync::RwLock>, + observed: std::sync::Mutex>, + attempts: AtomicUsize, + changed: Notify, + } + + impl SignalingLockClient { + fn new(inner: Arc) -> Self { + Self { + inner, + target: std::sync::RwLock::new(None), + observed: std::sync::Mutex::new(Vec::new()), + attempts: AtomicUsize::new(0), + changed: Notify::new(), + } + } + + fn set_target(&self, target: rustfs_lock::ObjectKey) { + *self.target.write().expect("signaling lock target should be writable") = Some(target); + } + + fn clear_observed(&self) { + self.observed + .lock() + .expect("observed lock resources should be writable") + .clear(); + } + + fn observed(&self) -> Vec { + self.observed + .lock() + .expect("observed lock resources should be readable") + .clone() + } + + async fn wait_for_attempts(&self, expected: usize) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let changed = self.changed.notified(); + if self.attempts.load(Ordering::Acquire) >= expected { + return; + } + changed.await; + } + }) + .await + .expect("target lock attempt should arrive"); + } + } + + #[async_trait::async_trait] + impl LockClient for SignalingLockClient { + async fn acquire_lock(&self, request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + self.observed + .lock() + .expect("observed lock resources should be writable") + .push(request.resource.clone()); + if self.target.read().expect("signaling lock target should be readable").as_ref() == Some(&request.resource) { + self.attempts.fetch_add(1, Ordering::Release); + self.changed.notify_waiters(); + } + self.inner.acquire_lock(request).await + } + + async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + self.inner.release(lock_id).await + } + + async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + self.inner.refresh(lock_id).await + } + + async fn force_release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + self.inner.force_release(lock_id).await + } + + async fn check_status(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + self.inner.check_status(lock_id).await + } + + async fn get_stats(&self) -> rustfs_lock::Result { + self.inner.get_stats().await + } + + async fn close(&self) -> rustfs_lock::Result<()> { + self.inner.close().await + } + + async fn is_online(&self) -> bool { + self.inner.is_online().await + } + + async fn is_local(&self) -> bool { + self.inner.is_local().await + } + } + + #[derive(Debug)] + struct SelectiveLockLossClient { + target: Arc>>, + refresh_calls: Arc, + active: tokio::sync::Mutex>, + } + + impl SelectiveLockLossClient { + fn new(target: Arc>>, refresh_calls: Arc) -> Self { + Self { + target, + refresh_calls, + active: tokio::sync::Mutex::new(HashMap::new()), + } + } + } + + #[async_trait::async_trait] + impl LockClient for SelectiveLockLossClient { + async fn acquire_lock(&self, request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + self.active + .lock() + .await + .insert(request.lock_id.clone(), request.resource.clone()); + Ok(rustfs_lock::LockResponse::success( + rustfs_lock::LockInfo { + id: request.lock_id.clone(), + resource: request.resource.clone(), + lock_type: request.lock_type, + status: rustfs_lock::LockStatus::Acquired, + owner: request.owner.clone(), + acquired_at: std::time::SystemTime::now(), + expires_at: std::time::SystemTime::now() + request.ttl, + last_refreshed: std::time::SystemTime::now(), + metadata: request.metadata.clone(), + priority: request.priority, + wait_start_time: None, + }, + Duration::ZERO, + )) + } + + async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(self.active.lock().await.remove(lock_id).is_some()) + } + + async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + let resource = self.active.lock().await.get(lock_id).cloned(); + let target = self.target.read().expect("lock-loss target should be readable").clone(); + if resource == target { + self.refresh_calls.fetch_add(1, Ordering::Release); + return Ok(false); + } + Ok(resource.is_some()) + } + + async fn force_release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + self.release(lock_id).await + } + + async fn check_status(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + Ok(None) + } + + async fn get_stats(&self) -> rustfs_lock::Result { + Ok(rustfs_lock::LockStats::default()) + } + + async fn close(&self) -> rustfs_lock::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + true + } + + async fn is_local(&self) -> bool { + false + } + } async fn non_trash_tmp_entries(temp_dirs: &[TempDir]) -> Vec { let mut leftovers = Vec::new(); @@ -1612,6 +2005,48 @@ mod tests { leftovers } + async fn make_bucket_on_all(disks: &[DiskStore], bucket: &str) { + for disk in disks { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + } + + async fn stage_upload_with_create_opts( + set_disks: &Arc, + bucket: &str, + object: &str, + content: &[u8], + create_opts: &ObjectOptions, + ) -> (String, Vec) { + let upload = set_disks + .new_multipart_upload(bucket, object, create_opts) + .await + .expect("multipart upload should be created"); + let mut reader = PutObjReader::new( + HashReader::from_stream( + Cursor::new(content.to_vec()), + content.len() as i64, + content.len() as i64, + None, + None, + false, + ) + .expect("hash reader should be constructed"), + ); + let part = set_disks + .put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + .expect("uploading the part should succeed"); + ( + upload.upload_id, + vec![CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }], + ) + } + async fn make_multipart_lock_test_set_disks() -> Arc { let endpoints = vec![ Endpoint::try_from("http://127.0.0.1:9000/data").expect("first endpoint should parse"), @@ -1918,6 +2353,624 @@ mod tests { ); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn put_object_part_rechecks_upload_after_commit_lock() { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager)))); + let lockers: Vec> = vec![signaling.clone()]; + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-post-lock-recheck-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + let upload_id = upload.upload_id; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone())); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let holder = set_disks + .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path) + .await + .expect("upload namespace lock should be created") + .get_write_lock(Duration::from_secs(5)) + .await + .expect("test should hold the upload commit lock"); + signaling.wait_for_attempts(1).await; + + let mut reader = PutObjReader::from_vec(vec![0x5a; 4096]); + let put_store = set_disks.clone(); + let put_upload_id = upload_id.clone(); + let put = tokio::spawn(async move { + put_store + .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + }); + + signaling.wait_for_attempts(2).await; + set_disks + .delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path) + .await + .expect("test should remove the upload after the part waits on its commit lock"); + drop(holder); + + let err = tokio::time::timeout(Duration::from_secs(10), put) + .await + .expect("part upload should finish") + .expect("part task should not panic") + .expect_err("part upload must recheck the upload after acquiring its commit lock"); + assert!(matches!(err, StorageError::InvalidUploadID(..))); + assert!(matches!( + set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await, + Err(StorageError::InvalidUploadID(..)) + )); + for temp_dir in temp_dirs { + assert!( + !temp_dir + .path() + .join(RUSTFS_META_MULTIPART_BUCKET) + .join(&upload_id_path) + .exists(), + "late UploadPart must not recreate multipart staging" + ); + } + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn put_object_part_holds_upload_lock_through_rename() { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager)))); + let lockers: Vec> = vec![signaling.clone()]; + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-put-part-lock-lifetime-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + let upload_id = upload.upload_id; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path)); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::PutPartAfterRename); + + let put_store = set_disks.clone(); + let put_upload_id = upload_id.clone(); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![0x45; 4096]); + put_store + .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + }); + barrier.wait_until_paused().await; + + let abort_store = set_disks.clone(); + let abort_upload_id = upload_id.clone(); + let abort = tokio::spawn(async move { + abort_store + .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default()) + .await + }); + signaling.wait_for_attempts(2).await; + tokio::task::yield_now().await; + assert!(!abort.is_finished(), "abort must wait until UploadPart releases the upload lock"); + + barrier.release(); + put.await + .expect("UploadPart task should not panic") + .expect("UploadPart should return after releasing the barrier"); + abort + .await + .expect("abort task should not panic") + .expect("abort should delete the upload after UploadPart releases the lock"); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn put_object_part_fences_upload_lock_loss_before_rename() { + let target = Arc::new(std::sync::RwLock::new(None)); + let refresh_calls = Arc::new(AtomicUsize::new(0)); + let lockers: Vec> = (0..4) + .map(|_| { + Arc::new(SelectiveLockLossClient::new(Arc::clone(&target), Arc::clone(&refresh_calls))) as Arc + }) + .collect(); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-put-part-lock-loss-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + let upload_id = upload.upload_id; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + *target.write().expect("lock-loss target should be writable") = + Some(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone())); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::PutPartBeforeLockLost); + + let put_store = set_disks.clone(); + let put_upload_id = upload_id.clone(); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![0x46; 4096]); + put_store + .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + }); + barrier.wait_until_paused().await; + tokio::time::advance(Duration::from_secs(11)).await; + tokio::task::yield_now().await; + assert!( + refresh_calls.load(Ordering::Acquire) > 0, + "upload lock heartbeat should reach the test client" + ); + barrier.release(); + + let err = put + .await + .expect("UploadPart task should not panic") + .expect_err("UploadPart must fail after losing the upload lock"); + match err { + StorageError::NamespaceLockQuorumUnavailable { + bucket: lock_bucket, + object: lock_object, + .. + } => { + assert_eq!(lock_bucket, RUSTFS_META_MULTIPART_BUCKET); + assert_eq!(lock_object, upload_id_path); + } + other => panic!("unexpected lock-loss error: {other:?}"), + } + let listed = set_disks + .list_object_parts(bucket, object, &upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default()) + .await + .expect("lock loss before rename must leave the upload readable"); + assert!(listed.parts.is_empty(), "lock loss before rename must not publish the part"); + } + + #[tokio::test] + async fn complete_encrypted_multipart_marks_quorum_validated_layout() { + temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + assert_eq!( + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + "encrypted-part-layout-quorum-candidate-v1" + ); + assert_eq!(crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, "encrypted-part-layout-quorum-v1"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-layout-quorum-marker-bucket"; + make_bucket_on_all(&disk_stores, bucket).await; + let encrypted_opts = ObjectOptions { + user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]), + ..Default::default() + }; + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, "encrypted", &[0x31; 4096], &encrypted_opts).await; + let (staged, _) = set_disks + .check_upload_id_exists(bucket, "encrypted", &upload_id, false) + .await + .expect("encrypted multipart staging metadata should be readable"); + let staged_layout_token = staged.data_dir.expect("staging data dir should exist").to_string(); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &staged.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX + ), + Some(staged_layout_token.as_str()) + ); + assert_eq!( + rustfs_utils::http::get_consistent_str(&staged.metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX), + None + ); + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, "encrypted", &upload_id, parts, &ObjectOptions::default()) + .await + .expect("encrypted multipart upload should complete"); + let reread = set_disks + .get_object_info(bucket, "encrypted", &ObjectOptions::default()) + .await + .expect("completed object metadata should be readable"); + let completed_layout_token = completed.data_dir.expect("completed data dir should exist").to_string(); + assert_eq!(reread.data_dir, completed.data_dir); + for metadata in [&completed.user_defined, &reread.user_defined] { + for prefix in [ + rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::MINIO_INTERNAL_PREFIX, + ] { + let key = format!("{prefix}{}", crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX); + assert_eq!(metadata.get(&key).map(String::as_str), Some(completed_layout_token.as_str())); + } + assert_eq!( + rustfs_utils::http::get_consistent_str(metadata, crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX), + None + ); + } + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, "unencrypted", &[0x32; 4096], &ObjectOptions::default()).await; + let unencrypted = set_disks + .clone() + .complete_multipart_upload(bucket, "unencrypted", &upload_id, parts, &ObjectOptions::default()) + .await + .expect("unencrypted multipart upload should complete"); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &unencrypted.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX + ), + None + ); + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, "unlocked", &[0x33; 4096], &encrypted_opts).await; + let unlocked = set_disks + .clone() + .complete_multipart_upload( + bucket, + "unlocked", + &upload_id, + parts, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("unlocked encrypted multipart upload should complete"); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &unlocked.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX + ), + None + ); + + let unlocked_create_opts = ObjectOptions { + no_lock: true, + user_defined: encrypted_opts.user_defined.clone(), + ..Default::default() + }; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, "unlocked-create", &[0x34; 4096], &unlocked_create_opts).await; + let unlocked_create = set_disks + .clone() + .complete_multipart_upload(bucket, "unlocked-create", &upload_id, parts, &ObjectOptions::default()) + .await + .expect("multipart upload created without locking should complete"); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &unlocked_create.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX + ), + None + ); + + let mut pre_upgrade_opts = encrypted_opts.clone(); + insert_str( + &mut pre_upgrade_opts.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + "forged".to_string(), + ); + insert_str( + &mut pre_upgrade_opts.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + "forged-final".to_string(), + ); + let (upload_id, parts) = temp_env::async_with_vars( + [(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("false"))], + stage_upload_with_create_opts(&set_disks, bucket, "pre-upgrade", &[0x35; 4096], &pre_upgrade_opts), + ) + .await; + let (staged, _) = set_disks + .check_upload_id_exists(bucket, "pre-upgrade", &upload_id, false) + .await + .expect("pre-upgrade multipart staging metadata should be readable"); + for suffix in [ + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + ] { + assert_eq!(rustfs_utils::http::get_consistent_str(&staged.metadata, suffix), None); + } + let pre_upgrade = set_disks + .clone() + .complete_multipart_upload(bucket, "pre-upgrade", &upload_id, parts, &ObjectOptions::default()) + .await + .expect("pre-upgrade multipart upload should complete"); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &pre_upgrade.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX + ), + None + ); + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, "disabled-complete", &[0x36; 4096], &encrypted_opts).await; + let disabled_complete = temp_env::async_with_vars( + [(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("false"))], + set_disks.clone().complete_multipart_upload( + bucket, + "disabled-complete", + &upload_id, + parts, + &ObjectOptions::default(), + ), + ) + .await + .expect("multipart completion with rollout disabled should succeed"); + for suffix in [ + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + ] { + assert_eq!(rustfs_utils::http::get_consistent_str(&disabled_complete.user_defined, suffix), None); + } + }) + .await; + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn complete_revalidates_layout_candidate_after_upload_lock() { + temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager)))); + let lockers: Vec> = vec![signaling.clone()]; + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-layout-recheck-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let create_opts = ObjectOptions { + user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]), + ..Default::default() + }; + let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x42; 4096], &create_opts).await; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone())); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let holder = set_disks + .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path) + .await + .expect("upload namespace lock should be created") + .get_write_lock(Duration::from_secs(5)) + .await + .expect("test should hold the upload commit lock"); + signaling.wait_for_attempts(1).await; + + let complete_store = set_disks.clone(); + let complete_upload_id = upload_id.clone(); + let complete = tokio::spawn(async move { + complete_store + .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default()) + .await + }); + signaling.wait_for_attempts(2).await; + + let disks = set_disks.disks.read().await.clone(); + let (fi, mut files_metas) = set_disks + .check_upload_id_exists(bucket, object, &upload_id, true) + .await + .expect("staged upload metadata should be readable"); + for meta in &mut files_metas { + rustfs_utils::http::metadata_compat::remove_str( + &mut meta.metadata, + crate::object_api::ENCRYPTED_PART_LAYOUT_CANDIDATE_SUFFIX, + ); + } + SetDisks::write_unique_file_info( + &disks, + bucket, + RUSTFS_META_MULTIPART_BUCKET, + &upload_id_path, + &files_metas, + fi.write_quorum(set_disks.default_write_quorum()), + ) + .await + .expect("staged candidate should be removed while completion waits"); + drop(holder); + + let completed = tokio::time::timeout(Duration::from_secs(10), complete) + .await + .expect("completion should finish") + .expect("completion task should not panic") + .expect("completion should safely fall back without a candidate"); + assert_eq!( + rustfs_utils::http::get_consistent_str( + &completed.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX + ), + None + ); + }) + .await; + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn complete_holds_object_then_upload_lock_through_commit() { + temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager)))); + let lockers: Vec> = vec![signaling.clone()]; + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-layout-lock-order-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let create_opts = ObjectOptions { + user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]), + ..Default::default() + }; + let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x43; 4096], &create_opts).await; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + let upload_resource = rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path); + let object_resource = rustfs_lock::ObjectKey::new(bucket, object); + signaling.set_target(upload_resource.clone()); + signaling.clear_observed(); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename); + + let complete_store = set_disks.clone(); + let complete_upload_id = upload_id.clone(); + let complete = tokio::spawn(async move { + complete_store + .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default()) + .await + }); + barrier.wait_until_paused().await; + + let observed = signaling.observed(); + let object_position = observed + .iter() + .position(|resource| resource == &object_resource) + .expect("completion should acquire the object lock"); + let upload_position = observed + .iter() + .position(|resource| resource == &upload_resource) + .expect("completion should acquire the upload lock"); + assert!(object_position < upload_position, "completion must acquire object before upload"); + + let abort_store = set_disks.clone(); + let abort_upload_id = upload_id.clone(); + let abort = tokio::spawn(async move { + abort_store + .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default()) + .await + }); + signaling.wait_for_attempts(2).await; + tokio::task::yield_now().await; + assert!(!abort.is_finished(), "abort must wait until completion releases the upload lock"); + + barrier.release(); + complete + .await + .expect("completion task should not panic") + .expect("completion should commit after the barrier is released"); + let abort_err = abort + .await + .expect("abort task should not panic") + .expect_err("the committed upload should no longer exist when abort acquires the lock"); + assert!(matches!(abort_err, StorageError::InvalidUploadID(..))); + }) + .await; + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn complete_fences_upload_lock_loss_before_commit() { + temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async { + let target = Arc::new(std::sync::RwLock::new(None)); + let refresh_calls = Arc::new(AtomicUsize::new(0)); + let lockers: Vec> = (0..4) + .map(|_| { + Arc::new(SelectiveLockLossClient::new(Arc::clone(&target), Arc::clone(&refresh_calls))) as Arc + }) + .collect(); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "multipart-layout-lock-loss-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let create_opts = ObjectOptions { + user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]), + ..Default::default() + }; + let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x44; 4096], &create_opts).await; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + *target.write().expect("lock-loss target should be writable") = + Some(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone())); + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeLockLost); + + let complete_store = set_disks.clone(); + let complete_upload_id = upload_id.clone(); + let complete = tokio::spawn(async move { + complete_store + .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default()) + .await + }); + barrier.wait_until_paused().await; + tokio::time::advance(Duration::from_secs(11)).await; + tokio::task::yield_now().await; + assert!( + refresh_calls.load(Ordering::Acquire) > 0, + "upload lock heartbeat should reach the test client" + ); + barrier.release(); + + let err = complete + .await + .expect("completion task should not panic") + .expect_err("completion must fail after losing the upload lock"); + match err { + StorageError::NamespaceLockQuorumUnavailable { + bucket: lock_bucket, + object: lock_object, + .. + } => { + assert_eq!(lock_bucket, RUSTFS_META_MULTIPART_BUCKET); + assert_eq!(lock_object, upload_id_path); + } + other => panic!("unexpected lock-loss error: {other:?}"), + } + set_disks + .check_upload_id_exists(bucket, object, &upload_id, true) + .await + .expect("lock loss before commit must leave the staged upload retryable"); + }) + .await; + } + + #[tokio::test] + async fn complete_encrypted_multipart_reuses_precondition_object_lock() { + temp_env::async_with_vars( + [ + (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")), + (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1")), + ], + async { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-layout-precondition-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let create_opts = ObjectOptions { + user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]), + ..Default::default() + }; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &[0x41; 4096], &create_opts).await; + + let completed = set_disks + .clone() + .complete_multipart_upload( + bucket, + object, + &upload_id, + parts, + &ObjectOptions { + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .expect("successful precondition should reuse the held object lock"); + let layout_token = completed.data_dir.expect("completed data dir should exist").to_string(); + assert!(crate::object_api::has_encrypted_part_layout_marker( + &completed.user_defined, + crate::object_api::ENCRYPTED_PART_LAYOUT_QUORUM_SUFFIX, + &layout_token, + )); + }, + ) + .await; + } + #[tokio::test] async fn second_pool_multipart_uses_its_own_layout_and_round_trips() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_for_pool_with_default_parity(2, 1, 2).await; @@ -2359,51 +3412,6 @@ mod tests { vec![fill; PART_SIZE] } - async fn make_bucket_on_all(disks: &[DiskStore], bucket: &str) { - for disk in disks { - disk.make_volume(bucket).await.expect("bucket volume should be created"); - } - } - - /// Stage a single-part multipart upload without completing it. A single - /// part is the last part, so the 5 MiB minimum-part-size gate does not - /// apply. Returns the upload id and the `CompletePart` list. - async fn stage_upload_with_create_opts( - set_disks: &Arc, - bucket: &str, - object: &str, - content: &[u8], - create_opts: &ObjectOptions, - ) -> (String, Vec) { - let upload = set_disks - .new_multipart_upload(bucket, object, create_opts) - .await - .expect("multipart upload should be created"); - let mut reader = PutObjReader::new( - HashReader::from_stream( - Cursor::new(content.to_vec()), - content.len() as i64, - content.len() as i64, - None, - None, - false, - ) - .expect("hash reader should be constructed"), - ); - let part = set_disks - .put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) - .await - .expect("uploading the part should succeed"); - ( - upload.upload_id, - vec![CompletePart { - part_num: part.part_num, - etag: part.etag, - ..Default::default() - }], - ) - } - async fn stage_upload( set_disks: &Arc, bucket: &str, diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index eae6138dc..9fd1dbe71 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -878,6 +878,19 @@ pub fn record_get_object_codec_streaming_fallback(reason: &'static str) { counter!("rustfs_io_get_object_codec_streaming_fallback_total", "reason" => reason).increment(1); } +/// Record the read path chosen for one encrypted Range GET on the Legacy (rio v1) backend +/// together with its read amplification — physical ciphertext bytes scheduled for the +/// erasure layer divided by the plaintext bytes the client requested. Observed at the +/// ReadPlan decision point (https://github.com/rustfs/backlog/issues/1316 Phase A). +#[inline(always)] +pub fn record_get_encrypted_range_read_amplification(path: &'static str, amplification: f64) { + if !get_stage_metrics_enabled() { + return; + } + counter!("rustfs_io_get_encrypted_range_read_path_total", "path" => path).increment(1); + histogram!("rustfs_io_get_encrypted_range_read_amplification", "path" => path).record(amplification); +} + /// Record the final codec-streaming rollout decision for a GET request. #[inline(always)] pub fn record_get_object_codec_streaming_decision(outcome: &'static str, object_class: &'static str, reason: &'static str) { diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 11da713d5..93a39aaa4 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -15,6 +15,7 @@ for later deletion. - `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability. - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request. - `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus. +- `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently. ## Review Checklist