From aaf888a4b84957e140c1f8bad1386af084737ef0 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 14 Aug 2026 08:15:56 +0800 Subject: [PATCH] chore(ecstore): fold the ListObjects forwarders into the ECStore impl (#6079) --- crates/config/src/constants/object.rs | 25 + crates/ecstore/src/erasure/coding/erasure.rs | 234 ++++- crates/ecstore/src/object_api/readers.rs | 85 +- .../ecstore/src/services/notification_sys.rs | 32 + crates/ecstore/src/set_disk/mod.rs | 140 +++ crates/ecstore/src/set_disk/ops/heal.rs | 27 + crates/ecstore/src/set_disk/ops/multipart.rs | 391 ++++++- crates/ecstore/src/set_disk/ops/object.rs | 992 +++++++++++++++++- crates/ecstore/src/set_disk/read.rs | 45 +- crates/ecstore/src/set_disk/replication.rs | 19 + crates/ecstore/src/store/list.rs | 81 -- crates/ecstore/src/store/list_objects.rs | 2 +- crates/ecstore/src/store/mod.rs | 7 +- crates/filemeta/src/fileinfo.rs | 61 +- crates/utils/src/http/metadata_compat.rs | 1 + rustfs/src/app/multipart_usecase.rs | 152 +-- rustfs/src/app/object_usecase.rs | 312 ++++-- rustfs/src/app/storage_api.rs | 4 +- scripts/check_logging_guardrails.sh | 4 +- 19 files changed, 2255 insertions(+), 359 deletions(-) delete mode 100644 crates/ecstore/src/store/list.rs diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 279789719..081308754 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -137,6 +137,22 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false; const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE); const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED); +/// Request the object-transaction fencing contract used by storage-owned +/// cleanup receipts and lock-window optimizations. +/// +/// This is fail-closed: enabling the writer without a live fleet proof rejects +/// the commit rather than silently using a legacy-safe path. +pub const ENV_OBJECT_TRANSACTION_FENCING_WRITE: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE"; +pub const DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE: bool = false; + +/// Operator-attested confirmation that every serving node understands the +/// object transaction fencing contract. +pub const ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED"; +pub const DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: bool = false; + +const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE); +const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED); + /// Request preserving legacy per-part checksum metadata during data movement. /// /// This remains ineffective until @@ -673,4 +689,13 @@ mod remote_version_state_tests { "RUSTFS_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED" ); } + + #[test] + fn object_transaction_fencing_gate_uses_stable_environment_names() { + assert_eq!(super::ENV_OBJECT_TRANSACTION_FENCING_WRITE, "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE"); + assert_eq!( + super::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED" + ); + } } diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index 8b7ff1c10..c9e7d73cb 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -71,10 +71,16 @@ impl EncodedBlock { const MODERN_MAX_TOTAL_SHARDS: usize = ::ORDER; const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64; +const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16; +// Vec growth may retain twice the requested logical length. Keeping the logical +// workspace at half the budget bounds each cached workspace's shard allocation to 1 MiB. +const LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE: usize = 512 * 1024; type ModernReedSolomonCache = RwLock>>; +type LegacyReedSolomonCache = RwLock>>; static MODERN_REED_SOLOMON_CACHE: OnceLock = OnceLock::new(); +static LEGACY_REED_SOLOMON_CACHE: OnceLock = OnceLock::new(); /// Errors returned when constructing an [`Erasure`] codec. #[derive(Debug, thiserror::Error)] @@ -141,43 +147,61 @@ pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize { struct LegacyReedSolomonEncoder { data_shards: usize, parity_shards: usize, - encoder_cache: std::sync::RwLock>, - decoder_cache: std::sync::RwLock>, -} - -impl Clone for LegacyReedSolomonEncoder { - fn clone(&self) -> Self { - Self { - data_shards: self.data_shards, - parity_shards: self.parity_shards, - encoder_cache: std::sync::RwLock::new(None), - decoder_cache: std::sync::RwLock::new(None), - } - } + cache_workspaces: bool, + encoder_cache: RwLock>, + decoder_cache: RwLock>, } impl LegacyReedSolomonEncoder { - fn new(_data_shards: usize, _parity_shards: usize) -> io::Result { + fn new(data_shards: usize, parity_shards: usize) -> io::Result { + Self::with_workspace_cache(data_shards, parity_shards, false) + } + + fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result { Ok(Self { - data_shards: _data_shards, - parity_shards: _parity_shards, - encoder_cache: std::sync::RwLock::new(None), - decoder_cache: std::sync::RwLock::new(None), + data_shards, + parity_shards, + cache_workspaces, + encoder_cache: RwLock::new(None), + decoder_cache: RwLock::new(None), }) } + fn logical_shard_bytes_upper_bound(&self, shard_len: usize) -> Option { + let aligned_shard_len = shard_len.checked_add(63)?.checked_div(64)?.checked_mul(64)?; + let high_rate_decoder_work_count = self + .parity_shards + .checked_next_power_of_two()? + .checked_add(self.data_shards)? + .checked_next_power_of_two()?; + let low_rate_decoder_work_count = self + .data_shards + .checked_next_power_of_two()? + .checked_add(self.parity_shards)? + .checked_next_power_of_two()?; + aligned_shard_len.checked_mul(high_rate_decoder_work_count.max(low_rate_decoder_work_count)) + } + + fn should_cache_workspace(&self, shard_len: usize) -> bool { + self.cache_workspaces + && self + .logical_shard_bytes_upper_bound(shard_len) + .is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE) + } + fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> { let mut shards_vec: Vec<&mut [u8]> = shards.into_vec(); if shards_vec.is_empty() { return Ok(()); } let shard_len = shards_vec[0].len(); + let cached_encoder = self + .encoder_cache + .write() + .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))? + .take(); let mut encoder = { - let mut cache_guard = self - .encoder_cache - .write() - .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?; - match cache_guard.take() { + match cached_encoder { Some(mut cached) => { if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() { reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len) @@ -204,10 +228,15 @@ impl LegacyReedSolomonEncoder { } } drop(result); - *self - .encoder_cache - .write() - .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder); + if self.should_cache_workspace(shard_len) { + let mut cache = self + .encoder_cache + .write() + .map_err(|_| io::Error::other("Failed to return encoder to cache"))?; + if cache.is_none() { + *cache = Some(encoder); + } + } Ok(()) } @@ -221,13 +250,13 @@ impl LegacyReedSolomonEncoder { .find_map(|s| s.as_ref().map(|v| v.len())) .ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?; + let cached_decoder = self + .decoder_cache + .write() + .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))? + .take(); let mut decoder = { - let mut cache_guard = self - .decoder_cache - .write() - .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?; - - match cache_guard.take() { + match cached_decoder { Some(mut cached_decoder) => { if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) { warn!("Failed to reset SIMD decoder: {:?}, creating new one", e); @@ -274,10 +303,15 @@ impl LegacyReedSolomonEncoder { drop(result); - *self - .decoder_cache - .write() - .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder); + if self.should_cache_workspace(shard_len) { + let mut cache = self + .decoder_cache + .write() + .map_err(|_| io::Error::other("Failed to return decoder to cache"))?; + if cache.is_none() { + *cache = Some(decoder); + } + } Ok(()) } @@ -435,6 +469,39 @@ fn cached_modern_reed_solomon(data_shards: usize, parity_shards: usize) -> Resul Ok(encoder) } +fn cached_legacy_reed_solomon(data_shards: usize, parity_shards: usize) -> io::Result> { + let cache = LEGACY_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new())); + cached_legacy_reed_solomon_in(cache, data_shards, parity_shards) +} + +fn cached_legacy_reed_solomon_in( + cache: &LegacyReedSolomonCache, + data_shards: usize, + parity_shards: usize, +) -> io::Result> { + let key = (data_shards, parity_shards); + if let Some(encoder) = cache + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&key) + .cloned() + { + return Ok(encoder); + } + + let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(existing) = cache.get(&key) { + return Ok(Arc::clone(existing)); + } + if cache.len() < LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES { + let encoder = Arc::new(LegacyReedSolomonEncoder::with_workspace_cache(data_shards, parity_shards, true)?); + cache.insert(key, Arc::clone(&encoder)); + return Ok(encoder); + } + drop(cache); + Ok(Arc::new(LegacyReedSolomonEncoder::new(data_shards, parity_shards)?)) +} + fn encode_parity_shards(shards: &mut [Option>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()> where F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>, @@ -551,7 +618,7 @@ pub struct Erasure { pub data_shards: usize, pub parity_shards: usize, encoder: Option, - legacy_encoder: Option, + legacy_encoder: Option>, pub block_size: usize, uses_legacy: bool, _id: Uuid, @@ -687,7 +754,7 @@ impl Erasure { let legacy_encoder = if uses_legacy && parity_shards > 0 { Some( - LegacyReedSolomonEncoder::new(data_shards, parity_shards) + cached_legacy_reed_solomon(data_shards, parity_shards) .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?, ) } else { @@ -1405,7 +1472,7 @@ mod tests { assert_eq!(cloned.block_size, legacy.block_size); assert!(cloned.uses_legacy); - let data = b"legacy clone should keep independent SIMD caches"; + let data = b"legacy clone should preserve SIMD codec behavior"; let encoded = cloned.encode_data(data).expect("legacy clone should encode"); let mut shards = optional_shards(&encoded); shards[0] = None; @@ -1413,6 +1480,93 @@ mod tests { assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data); } + #[test] + fn legacy_codecs_share_process_cache_across_erasure_instances() { + let first = Erasure::new_with_options(6, 3, 64, true) + .legacy_encoder + .expect("legacy codec should be initialized"); + let second = Erasure::new_with_options(6, 3, 128, true) + .legacy_encoder + .expect("same legacy shard layout should be initialized"); + + assert!(Arc::ptr_eq(&first, &second)); + } + + #[test] + fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() { + let four_plus_two = Erasure::new_with_options(4, 2, 64, true) + .legacy_encoder + .expect("legacy codec should be initialized"); + let four_plus_one = Erasure::new_with_options(4, 1, 64, true) + .legacy_encoder + .expect("distinct parity layout should be initialized"); + let three_plus_two = Erasure::new_with_options(3, 2, 64, true) + .legacy_encoder + .expect("distinct data layout should be initialized"); + + assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one)); + assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two)); + assert_eq!(four_plus_two.logical_shard_bytes_upper_bound(64 * 1024), Some(512 * 1024)); + assert!(four_plus_two.should_cache_workspace(64 * 1024)); + assert!(!four_plus_two.should_cache_workspace(64 * 1024 + 1)); + + let nine_plus_seven = + LegacyReedSolomonEncoder::with_workspace_cache(9, 7, true).expect("9+7 legacy codec should construct"); + assert_eq!(nine_plus_seven.logical_shard_bytes_upper_bound(16 * 1024), Some(512 * 1024)); + assert!(nine_plus_seven.should_cache_workspace(16 * 1024)); + assert!(!nine_plus_seven.should_cache_workspace(16 * 1024 + 1)); + + let uncached = LegacyReedSolomonEncoder::new(4, 2).expect("uncached legacy codec should construct"); + assert!(!uncached.should_cache_workspace(64)); + } + + #[test] + fn saturated_legacy_codec_cache_does_not_retain_more_workspaces() { + let cache = RwLock::new(HashMap::new()); + for parity_shards in 1..=LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES { + let cached = + cached_legacy_reed_solomon_in(&cache, 32, parity_shards).expect("cacheable legacy codec should construct"); + assert!(cached.cache_workspaces); + } + + let uncached = + cached_legacy_reed_solomon_in(&cache, 31, 1).expect("uncached legacy codec should construct after saturation"); + assert!(!uncached.cache_workspaces); + assert_eq!( + cache.read().expect("cache lock should remain healthy").len(), + LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES + ); + } + + #[test] + fn concurrent_legacy_codecs_preserve_byte_exact_results() { + let barrier = Arc::new(std::sync::Barrier::new(2)); + let payloads = [vec![0x35; 257], vec![0xca; 1025]]; + + std::thread::scope(|scope| { + let handles = payloads.each_ref().map(|payload| { + let barrier = Arc::clone(&barrier); + scope.spawn(move || { + let erasure = Erasure::new_with_options(6, 3, 2048, true); + barrier.wait(); + let encoded = erasure.encode_data(payload).expect("concurrent legacy encode should succeed"); + barrier.wait(); + + let mut shards = optional_shards(&encoded); + shards[0] = None; + erasure + .decode_data(&mut shards) + .expect("concurrent legacy decode should reconstruct the missing shard"); + recover_data(&shards, erasure.data_shards, payload.len()) + }) + }); + + for (handle, payload) in handles.into_iter().zip(payloads.iter()) { + assert_eq!(handle.join().expect("concurrent legacy codec worker should not panic"), *payload); + } + }); + } + #[test] fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() { let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct"); diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index 35aefe053..27178c17d 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -15,6 +15,7 @@ use super::*; use crate::io_support::rio::Index; +use std::mem::MaybeUninit; #[cfg(feature = "rio-v2")] const DARE_PAYLOAD_SIZE: i64 = 64 * 1024; @@ -922,7 +923,7 @@ struct SkipReader { inner: R, bytes_to_skip: usize, bytes_skipped: usize, - scratch: Vec, + scratch: Box<[MaybeUninit]>, } impl SkipReader { @@ -931,7 +932,7 @@ impl SkipReader { inner, bytes_to_skip, bytes_skipped: 0, - scratch: vec![0u8; 8192], + scratch: Box::<[u8]>::new_uninit_slice(8192), } } } @@ -943,7 +944,7 @@ impl AsyncRead for SkipReader { while this.bytes_skipped < this.bytes_to_skip { let remaining = this.bytes_to_skip - this.bytes_skipped; let scratch_len = remaining.min(this.scratch.len()); - let mut scratch_buf = ReadBuf::new(&mut this.scratch[..scratch_len]); + let mut scratch_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]); match Pin::new(&mut this.inner).poll_read(cx, &mut scratch_buf) { Poll::Pending => return Poll::Pending, Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), @@ -974,7 +975,7 @@ pub struct RangedDecompressReader target_length: usize, current_offset: usize, bytes_returned: usize, - scratch: Vec, + scratch: Box<[MaybeUninit]>, drain_on_done: bool, drain_task: Option>, } @@ -1012,7 +1013,7 @@ impl RangedDecompressReader { target_length: actual_length, current_offset: 0, bytes_returned: 0, - scratch: vec![0u8; 8192], + scratch: Box::<[u8]>::new_uninit_slice(8192), drain_on_done, drain_task: None, }) @@ -1062,7 +1063,7 @@ impl AsyncRead for RangedDecompres } let scratch_len = std::cmp::min(this.scratch.len(), std::cmp::max(buf_capacity, 1)); - let mut temp_read_buf = ReadBuf::new(&mut this.scratch[..scratch_len]); + let mut temp_read_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]); let Some(inner) = this.inner.as_mut() else { return Poll::Ready(Ok(())); @@ -1114,7 +1115,8 @@ impl AsyncRead for RangedDecompres ); if bytes_to_return > 0 { - let data_slice = &this.scratch[data_start_in_buffer..data_start_in_buffer + bytes_to_return]; + let data_slice = + &temp_read_buf.filled()[data_start_in_buffer..data_start_in_buffer + bytes_to_return]; buf.put_slice(data_slice); this.bytes_returned += bytes_to_return; @@ -1133,7 +1135,7 @@ impl AsyncRead for RangedDecompres std::cmp::min(n, std::cmp::min(buf.remaining(), this.target_length - this.bytes_returned)); if bytes_to_return > 0 { - buf.put_slice(&this.scratch[..bytes_to_return]); + buf.put_slice(&temp_read_buf.filled()[..bytes_to_return]); this.bytes_returned += bytes_to_return; tracing::trace!("Returned {} bytes at offset {}", bytes_to_return, old_offset); @@ -1263,6 +1265,43 @@ mod tests { use temp_env::async_with_vars; use tokio::io::AsyncReadExt; + #[derive(Debug)] + struct PendingPartialReader { + data: &'static [u8], + position: usize, + pending: bool, + } + + impl PendingPartialReader { + fn new(data: &'static [u8]) -> Self { + Self { + data, + position: 0, + pending: true, + } + } + } + + impl AsyncRead for PendingPartialReader { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + if self.pending { + self.pending = false; + cx.waker().wake_by_ref(); + return Poll::Pending; + } + if self.position == self.data.len() { + return Poll::Ready(Ok(())); + } + + let length = buf.remaining().min(3).min(self.data.len() - self.position); + let end = self.position + length; + buf.put_slice(&self.data[self.position..end]); + self.position = end; + self.pending = true; + Poll::Ready(Ok(())) + } + } + const TEST_DIRECT_KEY_HEADER: &str = "x-rustfs-test-direct-key"; const TEST_OBJECT_KEY_HEADER: &str = "x-rustfs-test-object-key"; const TEST_NONCE_HEADER: &str = "x-rustfs-test-nonce"; @@ -1400,6 +1439,36 @@ mod tests { assert_eq!(result, b"World"); } + #[tokio::test] + async fn uninitialized_scratch_preserves_partial_pending_and_eof_reads() { + let mut skipped = SkipReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5); + let mut skipped_output = Vec::new(); + skipped + .read_to_end(&mut skipped_output) + .await + .expect("skip reader should survive partial pending reads through EOF"); + assert_eq!(skipped_output, b"56789abcdef"); + + let mut ranged = RangedDecompressReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5, 7, 16) + .expect("valid range should construct"); + let mut ranged_output = Vec::new(); + ranged + .read_to_end(&mut ranged_output) + .await + .expect("range reader should survive partial pending reads through EOF"); + assert_eq!(ranged_output, b"56789ab"); + } + + #[tokio::test] + async fn uninitialized_skip_scratch_reports_early_eof() { + let mut reader = SkipReader::new(PendingPartialReader::new(b"short"), 6); + let error = reader + .read_to_end(&mut Vec::new()) + .await + .expect_err("EOF before the skip boundary must remain visible"); + assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof); + } + #[tokio::test] async fn test_ranged_decompress_reader_from_start() { let original_data = b"Hello, World! This is a test."; diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index b08211853..c63dbc401 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -206,6 +206,38 @@ pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStat }) } +#[cfg(test)] +pub(crate) struct RemoteVersionStateFleetProofGuard; + +#[cfg(test)] +impl Drop for RemoteVersionStateFleetProofGuard { + fn drop(&mut self) { + replace_remote_version_state_fleet_proof(None); + } +} + +#[cfg(test)] +pub(crate) fn install_remote_version_state_fleet_proof_for_test(topology_fingerprint: &str) -> RemoteVersionStateFleetProofGuard { + match REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.to_string()) { + Ok(()) => {} + Err(_) + if REMOTE_VERSION_STATE_PROBE_TOPOLOGY + .get() + .is_some_and(|current| current == topology_fingerprint) => {} + Err(_) => panic!("remote version state test topology is already bound to another fingerprint"), + } + let peer_epochs = BTreeMap::new(); + if let Some(err) = publish_remote_version_state_probe_result( + remote_version_state_fleet_proof_slot(), + topology_fingerprint, + Ok(peer_epochs), + Instant::now(), + ) { + panic!("test proof installation must not fail: {err}"); + } + RemoteVersionStateFleetProofGuard +} + fn remote_version_state_fleet_proof_valid_at( proof: Option<&RemoteVersionStateFleetProof>, expected_topology: &str, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4abbfe749..27e382a3e 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -2792,6 +2792,9 @@ pub struct SetDisks { get_object_metadata_cache: moka::future::Cache>, get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState, get_object_metadata_cache_generations: Arc<[AtomicU64]>, + /// GET codecs keyed by every persisted layout dimension that affects + /// decoding. Clones of a set share the memoized shells. + erasure_cache: Arc, pub lockers: Vec>, shared_lockers: Arc<[Arc]>, local_lock_manager: Arc, @@ -2814,6 +2817,137 @@ pub struct SetDisks { storage_class_config_override: Arc>>>, } +const ERASURE_CACHE_MAX_ENTRIES: usize = 32; + +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +struct ErasureCacheKey { + data_shards: usize, + parity_shards: usize, + block_size: usize, + uses_legacy: bool, +} + +struct ErasureCache { + entries: parking_lot::RwLock>>, +} + +impl std::fmt::Debug for ErasureCache { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ErasureCache") + .field("entries", &self.entries.read().len()) + .finish() + } +} + +impl ErasureCache { + fn new() -> Self { + Self { + entries: parking_lot::RwLock::new(HashMap::new()), + } + } + + fn get_or_try_insert( + &self, + key: ErasureCacheKey, + ) -> std::result::Result, coding::ErasureConstructionError> { + if let Some(erasure) = self.entries.read().get(&key) { + return Ok(Arc::clone(erasure)); + } + + // Serialize first construction for a key so concurrent cold GETs still + // create exactly one shell. Codec construction never awaits. + let mut entries = self.entries.write(); + if let Some(erasure) = entries.get(&key) { + return Ok(Arc::clone(erasure)); + } + let erasure = Arc::new(coding::Erasure::try_new_with_options( + key.data_shards, + key.parity_shards, + key.block_size, + key.uses_legacy, + )?); + if entries.len() < ERASURE_CACHE_MAX_ENTRIES { + entries.insert(key, Arc::clone(&erasure)); + } + Ok(erasure) + } + + fn get_for_file_info(&self, fi: &FileInfo) -> Result> { + self.get_or_try_insert(ErasureCacheKey { + data_shards: fi.erasure.data_blocks, + parity_shards: fi.erasure.parity_blocks, + block_size: fi.erasure.block_size, + uses_legacy: fi.uses_legacy_checksum, + }) + .map_err(Error::from) + } +} + +#[cfg(test)] +mod erasure_cache_tests { + use super::*; + + #[test] + fn reuses_shells_and_keeps_every_layout_dimension_in_the_key() { + let cache = ErasureCache::new(); + let base = ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size: 1_048_576, + uses_legacy: false, + }; + let first = cache.get_or_try_insert(base).expect("modern shell should construct"); + let reused = cache.get_or_try_insert(base).expect("same modern shell should be cached"); + assert!(Arc::ptr_eq(&first, &reused)); + + for distinct in [ + ErasureCacheKey { data_shards: 3, ..base }, + ErasureCacheKey { + parity_shards: 1, + ..base + }, + ErasureCacheKey { + block_size: 524_288, + ..base + }, + ErasureCacheKey { + uses_legacy: true, + ..base + }, + ] { + let shell = cache.get_or_try_insert(distinct).expect("distinct shell should construct"); + assert!(!Arc::ptr_eq(&first, &shell)); + } + assert_eq!(cache.entries.read().len(), 5); + } + + #[test] + fn does_not_cache_invalid_layouts_or_grow_past_the_bound() { + let cache = ErasureCache::new(); + let invalid = ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size: 0, + uses_legacy: false, + }; + assert!(cache.get_or_try_insert(invalid).is_err()); + assert!(cache.entries.read().is_empty()); + + for block_size in 1..=(ERASURE_CACHE_MAX_ENTRIES + 1) { + cache + .get_or_try_insert(ErasureCacheKey { + data_shards: 4, + parity_shards: 2, + block_size, + uses_legacy: false, + }) + .expect("bounded cache fixture should construct"); + } + assert_eq!(cache.entries.read().len(), ERASURE_CACHE_MAX_ENTRIES); + } +} + #[derive(Clone, Debug, Eq, PartialEq)] struct GetObjectMetadataCacheKey { bucket: Arc, @@ -3212,6 +3346,7 @@ impl SetDisks { .map(|_| AtomicU64::new(0)) .collect::>(), ), + erasure_cache: Arc::new(ErasureCache::new()), lockers, shared_lockers, // Sourced from the instance context so each instance owns its lock @@ -9816,6 +9951,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), &fi, &disk_files, &disks, @@ -9879,6 +10015,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), &fi, &disk_files, &vec![Some(disk); erasure.total_shard_count()], @@ -9959,6 +10096,7 @@ mod tests { let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -10044,6 +10182,7 @@ mod tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), range_offset, range_length as i64, &mut writer, @@ -10155,6 +10294,7 @@ mod tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, total_size as i64, &mut writer, diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index e5ca4fbec..089515ec6 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -1425,6 +1425,33 @@ impl SetDisks { /// post-heal tail — reclaim identically. Never fails the heal: delete errors /// are logged and swallowed. Callers must gate this on `!opts.dry_run`. async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) { + match self.reconcile_old_data_cleanup_receipts(bucket, object).await { + Ok(removed) if removed > 0 => { + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + removed, + state = "old_data_cleanup_receipt_reconciled", + "Set disk old-data cleanup receipts reconciled" + ); + } + Ok(_) => {} + Err(e) => { + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + error = %e, + state = "old_data_cleanup_receipt_reconcile_failed", + "Set disk old-data cleanup receipt reconcile failed" + ); + } + } match self.reclaim_orphan_data_dirs(bucket, object).await { Ok(removed) if removed > 0 => { debug!( diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 93770ca2e..7ecde7910 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -22,6 +22,11 @@ use super::super::*; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; +use super::object::{ + assign_object_transaction_epoch, object_transaction_fencing_fleet_proof, object_transaction_fencing_fleet_proof_matches, + object_transaction_fencing_requested, old_data_cleanup_receipt_path, read_object_transaction_epoch_fence, + verify_object_transaction_epoch_fence, +}; use crate::crash_inject::{self, CrashPoint}; use crate::multipart_listing::paginate_multipart_listing; use futures::{StreamExt, stream}; @@ -63,6 +68,7 @@ pub(crate) enum MultipartCommitPause { PutPartBeforeLockLost, PutPartAfterRename, BeforeLockLost, + BeforeTransactionEpochVerify, AfterRename, } @@ -153,13 +159,24 @@ impl Drop for MultipartCommitBarrier { #[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(); + let barrier = { + 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(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) + { + if pause == MultipartCommitPause::BeforeTransactionEpochVerify { + slot.take() + } else { + slot.clone() + } + } else { + None + } + }; if let Some(barrier) = barrier && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { (current < barrier.expected_arrivals).then_some(current + 1) @@ -2296,6 +2313,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?; + let transaction_fencing_proof = object_transaction_fencing_fleet_proof(); + if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() { + return Err(Error::other("object transaction fencing requires a live fleet capability proof")); + } + let transaction_epoch_fence = if transaction_fencing_proof.is_some() { + Some(read_object_transaction_epoch_fence(self.as_ref(), bucket, object).await?) + } else { + None + }; + let transaction_epoch = + transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas)); + let commit_set = self.clone(); let commit_bucket = bucket.to_owned(); let commit_object = object.to_owned(); @@ -2323,6 +2352,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // The trailing `_` drops the rename_data old-size backfill // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. + if let Some(proof) = transaction_fencing_proof.as_ref() + && !object_transaction_fencing_fleet_proof_matches(proof) + { + return Err(Error::other( + "object transaction fencing fleet capability changed during complete_multipart_upload", + )); + } + if let Some(expected) = transaction_epoch_fence { + #[cfg(test)] + pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeTransactionEpochVerify).await; + verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?; + } let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data( &shuffle_disks, RUSTFS_META_MULTIPART_BUCKET, @@ -2354,6 +2395,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { }); } + if let Some(old_dir) = op_old_dir { + commit_set + .persist_old_data_cleanup_receipts( + &cleanup_disks, + &commit_bucket, + &commit_object, + old_dir, + fi.data_dir, + transaction_epoch, + ) + .await; + } + // Crash-consistency injection: hard power loss after the authoritative // rename_data commit succeeded but before the stale part.N.meta cleanup. // The new version is durably committed and visible, so a crash here must @@ -2469,9 +2523,10 @@ fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart]) mod tests { use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; - use crate::disk::DiskAPI as _; + use crate::disk::{DiskAPI as _, ReadOptions}; use crate::disk::{endpoint::Endpoint, format::FormatV3}; use crate::layout::endpoints::SetupType; + use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test; // No-locker helpers resolve to the isolated-context variants (see // `hermetic_set_disks_isolated`); the guard-based tests build through // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context @@ -2882,6 +2937,208 @@ mod tests { } } + async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec> { + let mut epochs = Vec::with_capacity(disks.len()); + for (disk_index, disk) in disks.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}")); + epochs.push( + file_info + .object_transaction_epoch() + .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")), + ); + } + epochs + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_requires_live_fleet_proof_before_multipart_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-transaction-fencing-no-proof"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let (upload_id, parts) = stage_upload_with_create_opts( + &set_disks, + bucket, + object, + b"must-not-complete-without-proof", + &ObjectOptions::default(), + ) + .await; + + let err = temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default()) + .await + }, + ) + .await + .expect_err("multipart completion must fail closed without a live fleet proof"); + + assert!( + err.to_string() + .contains("object transaction fencing requires a live fleet capability proof"), + "unexpected error: {err:?}" + ); + set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect_err("failed fenced completion must not publish object metadata"); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_persists_epoch_on_multipart_commit() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-object-transaction-epoch"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default()) + .await + .expect("fenced multipart completion should commit with a live proof"); + }, + ) + .await; + + let epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + let first = epochs[0].expect("fenced multipart completion should persist an epoch"); + assert!(!first.is_nil()); + assert!(epochs.into_iter().all(|epoch| epoch == Some(first))); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_rejects_stale_multipart_epoch() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-object-transaction-stale-epoch"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut initial_reader = PutObjReader::from_vec(b"initial fenced object".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial fenced PUT should commit"); + let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("initial fenced PUT should persist an epoch"); + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, b"stale multipart body", &ObjectOptions::default()) + .await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeTransactionEpochVerify); + let stale_set = Arc::clone(&set_disks); + let stale = tokio::spawn(async move { + stale_set + .clone() + .complete_multipart_upload( + bucket, + object, + &upload_id, + parts, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + }); + barrier.wait_until_paused().await; + + let mut winner_reader = PutObjReader::from_vec(b"winning put body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut winner_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("concurrent fenced PUT should advance the epoch"); + let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("winning fenced PUT should persist an epoch"); + assert_ne!(winning_epoch, initial_epoch); + + barrier.release(); + let err = stale + .await + .expect("stale multipart task should not panic") + .expect_err("stale epoch multipart completion must be rejected"); + assert_eq!(err, StorageError::PreconditionFailed); + + let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch))); + let mut reader = set_disks + .get_object_reader( + bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("winning object should remain readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("winning body should stream"); + assert_eq!(restored, b"winning put body"); + }, + ) + .await; + } + #[tokio::test] async fn complete_multipart_quota_rejection_preserves_destination_and_upload() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; @@ -6193,6 +6450,24 @@ mod tests { (body, etag) } + async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("current object metadata should read") + .data_dir + .expect("test object should be stored out-of-line") + } + + async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok() + } + + async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir)) + .await + .is_ok() + } + async fn upload_is_listed(set_disks: &Arc, bucket: &str, object: &str, upload_id: &str) -> bool { let page = set_disks .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None) @@ -6313,6 +6588,106 @@ mod tests { let (body_after, _) = read_object(&set_disks, bucket, object).await; assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object"); } + + #[tokio::test] + #[serial(storage_class_env)] + async fn post_commit_crash_receipt_reclaims_old_data_after_restart() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-crash-old-data-receipt"; + let object = "crash-old-data-object"; + make_bucket_on_all(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let old = payload(0x51); + let (u_old, parts_old) = stage_upload(&set_disks, bucket, object, &old).await; + complete(&set_disks, bucket, object, &u_old, parts_old) + .await + .expect("the old version should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let new = payload(0x52); + let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; + crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await; + assert!( + matches!(crashed, Err(StorageError::Unexpected)), + "the post-commit crash point must surface as unexpected, got {crashed:?}" + ); + crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + + let (body, _) = read_object(&set_disks, bucket, object).await; + assert_eq!(body, new, "the committed replacement must remain readable after the crash"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "post-commit crash must leave a durable old-data cleanup receipt" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "post-commit crash must leave old data for restart reconciliation" + ); + } + + let restarted_endpoints = temp_dirs + .iter() + .enumerate() + .map(|(disk_idx, dir)| { + let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")) + .expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + endpoint + }) + .collect::>(); + let mut reloaded = Vec::with_capacity(restarted_endpoints.len()); + for endpoint in &restarted_endpoints { + reloaded.push( + new_disk( + endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should restart"), + ); + } + let restarted_set = SetDisks::new_with_instance_ctx( + "restart-cleanup-receipt-test-owner".to_string(), + Arc::new(RwLock::new(reloaded.iter().cloned().map(Some).collect())), + 4, + 2, + 0, + 0, + restarted_endpoints, + set_disks.format.clone(), + Vec::new(), + Arc::new(crate::runtime::instance::InstanceContext::new()), + ) + .await; + let removed = restarted_set + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("restart receipt reconciliation should succeed"); + assert_eq!(removed, 4, "restart reconciler should delete all receipt targets"); + for disk in &reloaded { + assert!( + !data_dir_exists(disk, bucket, object, old_dir).await, + "restart reconciler must reclaim the old data dir" + ); + } + }, + ) + .await; + } } #[test] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 46b25a3c7..cd450dac3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -44,10 +44,11 @@ use crate::bucket::replication::{ DeleteReplicationConfigSnapshot, VersionPurgeStatusType, replication_state_to_filemeta, version_purge_status_to_filemeta, }; use crate::diagnostics::get::GetObjectFailureReason; -use crate::disk::OldCurrentSize; +use crate::disk::{DataDirDeleteStatus, OldCurrentSize}; use crate::error::is_err_invalid_upload_id; use crate::object_api::NamespaceLockFence; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; +use crate::services::notification_sys::RemoteVersionStateFleetProofToken; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; use crate::store::ECStore; use crate::store::utils::clean_metadata; @@ -55,6 +56,9 @@ use futures::FutureExt as _; use http::HeaderValue; use rustfs_utils::path::decode_dir_object; use std::future::Future; +use std::sync::OnceLock; + +const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.json"; #[inline] fn duration_millis_f64(duration: std::time::Duration) -> f64 { @@ -65,6 +69,133 @@ fn committed_response_metadata_slot(committed_disks: &[Option], fallback_s committed_disks.iter().position(Option::is_some).unwrap_or(fallback_slot) } +pub(in crate::set_disk::ops) fn assign_object_transaction_epoch( + shuffle_disks: &[Option], + parts_metadatas: &mut [FileInfo], +) -> Uuid { + let epoch = Uuid::new_v4(); + for (disk, file_info) in shuffle_disks.iter().zip(parts_metadatas.iter_mut()) { + if disk.is_some() { + file_info.set_object_transaction_epoch(epoch); + } + } + epoch +} + +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct OldDataCleanupReceiptRecord { + epoch: String, + old_data_dir: String, + committed_data_dir: Option, +} + +#[derive(Clone, Copy)] +struct OldDataCleanupReceipt { + epoch: Uuid, + old_data_dir: Uuid, + committed_data_dir: Option, +} + +impl OldDataCleanupReceipt { + fn new(epoch: Uuid, old_data_dir: Uuid, committed_data_dir: Option) -> Self { + Self { + epoch, + old_data_dir, + committed_data_dir, + } + } + + fn encode(self) -> disk::error::Result { + let record = OldDataCleanupReceiptRecord { + epoch: self.epoch.to_string(), + old_data_dir: self.old_data_dir.to_string(), + committed_data_dir: self.committed_data_dir.map(|dir| dir.to_string()), + }; + Ok(Bytes::from(serde_json::to_vec(&record)?)) + } + + fn decode(data: &[u8]) -> disk::error::Result { + let record: OldDataCleanupReceiptRecord = serde_json::from_slice(data)?; + let epoch = Uuid::parse_str(&record.epoch).map_err(DiskError::other)?; + let old_data_dir = Uuid::parse_str(&record.old_data_dir).map_err(DiskError::other)?; + let committed_data_dir = record + .committed_data_dir + .as_deref() + .map(Uuid::parse_str) + .transpose() + .map_err(DiskError::other)?; + if epoch.is_nil() || old_data_dir.is_nil() || committed_data_dir.is_some_and(|dir| dir.is_nil()) { + return Err(DiskError::FileCorrupt); + } + Ok(Self::new(epoch, old_data_dir, committed_data_dir)) + } +} + +pub(in crate::set_disk::ops) fn old_data_cleanup_receipt_path(object: &str, old_data_dir: Uuid) -> String { + path_join_buf(&[object, &old_data_dir.to_string(), OLD_DATA_CLEANUP_RECEIPT_FILE]) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(in crate::set_disk::ops) enum ObjectTransactionEpochFence { + Absent, + Present(Uuid), +} + +impl ObjectTransactionEpochFence { + fn from_file_info(file_info: &FileInfo) -> Result { + match file_info.object_transaction_epoch() { + Ok(Some(epoch)) => Ok(Self::Present(epoch)), + Ok(None) => Ok(Self::Absent), + Err(_) => Err(StorageError::FileCorrupt), + } + } +} + +pub(in crate::set_disk::ops) async fn read_object_transaction_epoch_fence( + set: &SetDisks, + bucket: &str, + object: &str, +) -> Result { + let current = set + .get_object_fileinfo( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + versioned: true, + ..Default::default() + }, + false, + false, + ) + .await; + match current { + Ok(snapshot) => ObjectTransactionEpochFence::from_file_info(snapshot.fi()), + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(ObjectTransactionEpochFence::Absent), + Err(err) => Err(err), + } +} + +pub(in crate::set_disk::ops) async fn verify_object_transaction_epoch_fence( + set: &SetDisks, + bucket: &str, + object: &str, + expected: ObjectTransactionEpochFence, +) -> Result<()> { + let current = read_object_transaction_epoch_fence(set, bucket, object).await?; + if current == expected { + Ok(()) + } else { + Err(StorageError::PreconditionFailed) + } +} + +fn old_data_cleanup_receipt_epoch_matches_current(receipt: OldDataCleanupReceipt, current: ObjectTransactionEpochFence) -> bool { + matches!(current, ObjectTransactionEpochFence::Present(epoch) if epoch == receipt.epoch) +} + #[cfg(test)] mod duration_metrics_tests { use super::duration_millis_f64; @@ -660,7 +791,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let erasure = erasure_from_file_info(fi, fi.uses_legacy_checksum)?; + let erasure = self.erasure_cache.get_for_file_info(fi)?; let read_length = erasure.shard_file_offset(0, object_size, object_size); let total_shards = data_shards + fi.erasure.parity_blocks; let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); @@ -829,6 +960,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), fi, files, disks, @@ -864,6 +996,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { Self::get_object_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), 0, object_info.size, &mut output, @@ -907,6 +1040,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { match Self::get_object_decode_reader_with_fileinfo( bucket, object, + Arc::clone(&self.erasure_cache), fi, files, disks, @@ -971,6 +1105,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let set_index = self.set_index; let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; + let erasure_cache = Arc::clone(&self.erasure_cache); let (fi, files, disks) = snapshot.into_owned(); tokio::spawn(async move { let _guard = read_lock_guard; @@ -982,6 +1117,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { if let Err(e) = Self::get_object_with_fileinfo( &bucket, &object, + erasure_cache, offset, length, &mut writer, @@ -1075,6 +1211,114 @@ fn delete_file_info_with_replication_transport_metadata(fi: &FileInfo) -> FileIn } impl SetDisks { + pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts( + &self, + disks: &[Option], + bucket: &str, + object: &str, + old_data_dir: Uuid, + committed_data_dir: Option, + epoch: Option, + ) { + let Some(epoch) = epoch else { return }; + if committed_data_dir == Some(old_data_dir) { + return; + } + let receipt = OldDataCleanupReceipt::new(epoch, old_data_dir, committed_data_dir); + let Ok(encoded) = receipt.encode() else { + return; + }; + let path = old_data_cleanup_receipt_path(object, old_data_dir); + let futures = disks.iter().filter_map(|disk| { + disk.as_ref().map(|disk| { + let disk = disk.clone(); + let encoded = encoded.clone(); + let bucket = bucket.to_owned(); + let path = path.clone(); + async move { disk.write_all(&bucket, &path, encoded).await } + }) + }); + for result in join_all(futures).await { + if let Err(err) = result { + debug!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + old_dir = %old_data_dir, + error = %err, + state = "cleanup_receipt_persist_failed", + "SetDisk old-data cleanup receipt persist failed" + ); + } + } + } + + pub(in crate::set_disk) async fn reconcile_old_data_cleanup_receipts( + &self, + bucket: &str, + object: &str, + ) -> disk::error::Result { + if object_transaction_fencing_fleet_proof().is_none() { + return Ok(0); + } + let current = read_object_transaction_epoch_fence(self, bucket, object) + .await + .map_err(DiskError::from)?; + let disks = self.get_disks_internal().await; + let mut removed = 0usize; + + for disk in disks.iter().flatten() { + let entries = match disk.list_dir("", bucket, object, 0).await { + Ok(entries) => entries, + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => continue, + Err(err) => return Err(err), + }; + for entry in entries { + let Some(name) = entry.strip_suffix(SLASH_SEPARATOR) else { continue }; + let Ok(data_dir) = Uuid::parse_str(name) else { continue }; + if data_dir.is_nil() { + continue; + } + + let receipt_path = old_data_cleanup_receipt_path(object, data_dir); + let receipt_bytes = match disk.read_all(bucket, &receipt_path).await { + Ok(bytes) => bytes, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => continue, + Err(err) => return Err(err), + }; + let receipt = OldDataCleanupReceipt::decode(&receipt_bytes)?; + if receipt.old_data_dir != data_dir + || receipt.committed_data_dir == Some(receipt.old_data_dir) + || !old_data_cleanup_receipt_epoch_matches_current(receipt, current) + { + continue; + } + + let old_path = format!("{object}/{}", receipt.old_data_dir); + match disk + .delete_data_dir( + bucket, + &old_path, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + { + Ok(DataDirDeleteStatus::Deleted) => removed += 1, + Ok(DataDirDeleteStatus::Deferred) => {} + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {} + Err(err) => return Err(err), + } + } + } + + Ok(removed) + } + async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid) -> Result<()> { let current = metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?; if current != expected { @@ -1676,6 +1920,18 @@ impl SetDisks { }); } + let transaction_fencing_proof = object_transaction_fencing_fleet_proof(); + if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() { + return Err(Error::other("object transaction fencing requires a live fleet capability proof")); + } + let transaction_epoch_fence = if transaction_fencing_proof.is_some() { + Some(read_object_transaction_epoch_fence(self, bucket, object).await?) + } else { + None + }; + let transaction_epoch = + transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas)); + let commit_set = self.clone(); let commit_bucket = bucket.to_owned(); let commit_object = object.to_owned(); @@ -1693,6 +1949,39 @@ impl SetDisks { let _object_lock_guard = commit_object_lock_guard; let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard; let rename_stage_start = Instant::now(); + let pre_rename_result: Result<()> = async { + if let Some(proof) = transaction_fencing_proof.as_ref() + && !object_transaction_fencing_fleet_proof_matches(proof) + { + return Err(Error::other("object transaction fencing fleet capability changed during put_object")); + } + if let Some(expected) = transaction_epoch_fence { + #[cfg(any(test, feature = "test-util"))] + pause_put_object_commit( + &commit_bucket, + &commit_object, + PutObjectCommitPause::BeforeTransactionEpochVerify, + ) + .await; + verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?; + } + Ok(()) + } + .await; + if let Err(err) = pre_rename_result { + if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await { + warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data"); + } else if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + bucket = %commit_bucket, + object = %commit_object, + tmp_dir = %commit_tmp_dir, + "issue3031_put_object_tmp_cleanup_done" + ); + } + return Err(err); + } let rename_result = SetDisks::rename_data( &shuffle_disks, RUSTFS_META_TMP_BUCKET, @@ -1741,6 +2030,19 @@ impl SetDisks { let rename_stage_elapsed = rename_stage_start.elapsed(); let rename_stage_ms = rename_stage_elapsed.as_millis() as u64; + if let Some(old_dir) = op_old_dir { + commit_set + .persist_old_data_cleanup_receipts( + &cleanup_disks, + &commit_bucket, + &commit_object, + old_dir, + committed_data_dir, + transaction_epoch, + ) + .await; + } + commit_set .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object) .await; @@ -2837,14 +3139,12 @@ fn remote_version_state_writer_enabled() -> bool { remote_version_state_writer_fleet_proof().is_some() } -fn remote_version_state_writer_fleet_proof() -> Option { - remote_version_state_writer_requested() - .then(crate::services::notification_sys::acquire_remote_version_state_fleet_proof) - .flatten() +fn remote_version_state_writer_fleet_proof() -> Option { + transaction_fencing_fleet_proof(remote_version_state_writer_requested()) } fn remote_version_state_writer_requested() -> bool { - remote_version_state_writer_enabled_for( + transaction_fencing_gate_requested_for( rustfs_utils::get_env_bool( rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE, rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE, @@ -2857,20 +3157,66 @@ fn remote_version_state_writer_requested() -> bool { ) } -fn remote_version_state_writer_fleet_proof_matches( - proof: &crate::services::notification_sys::RemoteVersionStateFleetProofToken, -) -> bool { - remote_version_state_writer_fleet_proof_matches_for( +fn remote_version_state_writer_fleet_proof_matches(proof: &RemoteVersionStateFleetProofToken) -> bool { + transaction_fencing_fleet_proof_matches_for( remote_version_state_writer_requested(), crate::services::notification_sys::remote_version_state_fleet_proof_matches(proof), ) } -fn remote_version_state_writer_fleet_proof_matches_for(requested: bool, fleet_proof_matches: bool) -> bool { +pub(in crate::set_disk::ops) fn object_transaction_fencing_fleet_proof() -> Option { + transaction_fencing_fleet_proof(object_transaction_fencing_requested()) +} + +pub(in crate::set_disk::ops) fn object_transaction_fencing_requested() -> bool { + object_transaction_fencing_requested_cached() +} + +#[cfg(not(test))] +fn object_transaction_fencing_requested_cached() -> bool { + static REQUESTED: OnceLock = OnceLock::new(); + *REQUESTED.get_or_init(load_object_transaction_fencing_requested) +} + +#[cfg(test)] +fn object_transaction_fencing_requested_cached() -> bool { + load_object_transaction_fencing_requested() +} + +fn load_object_transaction_fencing_requested() -> bool { + transaction_fencing_gate_requested_for( + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, + rustfs_config::DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE, + ), + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + rustfs_config::DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + ), + true, + ) +} + +pub(in crate::set_disk::ops) fn object_transaction_fencing_fleet_proof_matches( + proof: &RemoteVersionStateFleetProofToken, +) -> bool { + transaction_fencing_fleet_proof_matches_for( + object_transaction_fencing_requested(), + crate::services::notification_sys::remote_version_state_fleet_proof_matches(proof), + ) +} + +fn transaction_fencing_fleet_proof(requested: bool) -> Option { + requested + .then(crate::services::notification_sys::acquire_remote_version_state_fleet_proof) + .flatten() +} + +fn transaction_fencing_fleet_proof_matches_for(requested: bool, fleet_proof_matches: bool) -> bool { requested && fleet_proof_matches } -fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool, fleet_proof_valid: bool) -> bool { +fn transaction_fencing_gate_requested_for(requested: bool, fleet_confirmed: bool, fleet_proof_valid: bool) -> bool { requested && fleet_confirmed && fleet_proof_valid } @@ -2880,6 +3226,7 @@ pub enum PutObjectCommitPause { BeforeNamespace, AfterNamespace, BeforeMetadata, + BeforeTransactionEpochVerify, } #[cfg(any(test, feature = "test-util"))] @@ -2961,13 +3308,24 @@ impl Drop for PutObjectCommitBarrier { #[cfg(any(test, feature = "test-util"))] async fn pause_put_object_commit(bucket: &str, object: &str, pause: PutObjectCommitPause) { - let barrier = PUT_OBJECT_COMMIT_BARRIER - .get_or_init(|| std::sync::Mutex::new(Vec::new())) - .lock() - .expect("put object commit barrier mutex should not poison") - .iter() - .find(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) - .cloned(); + let barrier = { + let mut slot = PUT_OBJECT_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(Vec::new())) + .lock() + .expect("put object commit barrier mutex should not poison"); + if let Some(index) = slot + .iter() + .position(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) + { + if pause == PutObjectCommitPause::BeforeTransactionEpochVerify { + Some(slot.remove(index)) + } else { + Some(Arc::clone(&slot[index])) + } + } else { + None + } + }; if let Some(barrier) = barrier { barrier.arrived.notify_one(); barrier.release.notified().await; @@ -3327,7 +3685,7 @@ mod transition_upload_completion_tests { mod transition_version_id_tests { use super::{ TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate, - remote_version_state_writer_enabled_for, remote_version_state_writer_fleet_proof_matches_for, + transaction_fencing_fleet_proof_matches_for, transaction_fencing_gate_requested_for, }; use rustfs_filemeta::TransitionVersionState; use uuid::Uuid; @@ -3380,7 +3738,24 @@ mod transition_version_id_tests { ("fully upgraded fleet", true, true, true, true), ] { assert_eq!( - remote_version_state_writer_enabled_for(requested, fleet_confirmed, fleet_proof_valid), + transaction_fencing_gate_requested_for(requested, fleet_confirmed, fleet_proof_valid), + expected, + "{case}" + ); + } + } + + #[test] + fn object_transaction_fencing_gate_requires_request_confirmation_and_live_proof() { + for (case, requested, fleet_confirmed, fleet_proof_valid, expected) in [ + ("old defaults", false, false, false, false), + ("missing fleet confirmation", true, false, true, false), + ("missing local opt-in", false, true, true, false), + ("missing fleet proof", true, true, false, false), + ("fully upgraded fleet", true, true, true, true), + ] { + assert_eq!( + transaction_fencing_gate_requested_for(requested, fleet_confirmed, fleet_proof_valid), expected, "{case}" ); @@ -3395,7 +3770,7 @@ mod transition_version_id_tests { ("current authorization", true, true, true), ] { assert_eq!( - remote_version_state_writer_fleet_proof_matches_for(requested, fleet_proof_matches), + transaction_fencing_fleet_proof_matches_for(requested, fleet_proof_matches), expected, "{case}" ); @@ -5009,11 +5384,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(cloned_fi.size); + let erasure_cache = Arc::clone(&self.erasure_cache); let producer = async move { let mut writer = TransitionUploadWriter::new(pw); Self::get_object_with_fileinfo( &cloned_bucket, &cloned_object, + erasure_cache, 0, cloned_fi.size, &mut writer, @@ -5926,6 +6303,7 @@ mod inline_put_commit_path_tests { use crate::disk::{DiskAPI as _, ReadOptions}; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use rustfs_config::server_config::KVS; + use serial_test::serial; use tokio::io::AsyncReadExt; async fn make_bucket(disks: &[DiskStore], bucket: &str) { @@ -5989,6 +6367,37 @@ mod inline_put_commit_path_tests { assert_eq!(restored, payload); } + #[tokio::test] + async fn repeated_gets_reuse_the_set_erasure_shell() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "get-erasure-shell-cache"; + let object = "object.bin"; + let payload = vec![0x4d; 1024 * 1024]; + make_bucket(&disk_stores, bucket).await; + + let mut reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("non-inline object should commit"); + assert!(set_disks.erasure_cache.entries.read().is_empty()); + + for _ in 0..2 { + let mut object_reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("cached-shell GET should succeed"); + let mut restored = Vec::new(); + object_reader + .stream + .read_to_end(&mut restored) + .await + .expect("cached-shell GET should stream"); + assert_eq!(restored, payload); + assert_eq!(set_disks.erasure_cache.entries.read().len(), 1); + } + } + #[tokio::test] async fn ec_8_4_default_budget_keeps_large_inline_candidate_out_of_xl_meta() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await; @@ -7800,6 +8209,139 @@ mod transition_commit_failure_tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn no_lock_restore_finalize_requires_live_namespace_fence() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-finalize-fence-bucket"; + let object = "object.bin"; + let payload = b"restore finalize no_lock fence must fail closed".repeat(1024); + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let operation_id = Uuid::new_v4(); + let mut reader = PutObjReader::from_vec(payload); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(restore_metadata(operation_id, true)), + ..Default::default() + }, + ) + .await + .expect("restore metadata should be installed"); + let restoring = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should be readable"); + + let err = set_disks + .finalize_restore_metadata( + bucket, + object, + &restoring, + &ObjectOptions { + no_lock: true, + namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()), + user_defined: restore_operation_id_metadata(operation_id), + ..Default::default() + }, + ) + .await + .expect_err("lost outer namespace fence must reject no_lock restore finalization"); + assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. })); + + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should remain readable"); + let restore_status = parse_restore_obj_status( + current + .user_defined + .get(s3s::header::X_AMZ_RESTORE.as_str()) + .expect("restore header must remain pending"), + ) + .expect("restore header should remain parseable"); + assert!( + restore_status.on_going(), + "lost no_lock finalization must not publish restored completion metadata" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn no_lock_restore_cleanup_requires_live_namespace_fence() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-cleanup-fence-bucket"; + let object = "object.bin"; + let payload = b"restore cleanup no_lock fence must fail closed".repeat(1024); + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let operation_id = Uuid::new_v4(); + let mut reader = PutObjReader::from_vec(payload); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(restore_metadata(operation_id, true)), + ..Default::default() + }, + ) + .await + .expect("restore metadata should be installed"); + let restoring = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should be readable"); + + let err = set_disks + .update_restore_metadata( + bucket, + object, + &restoring, + &ObjectOptions { + no_lock: true, + namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()), + user_defined: restore_operation_id_metadata(operation_id), + ..Default::default() + }, + ) + .await + .expect_err("lost outer namespace fence must reject no_lock restore cleanup"); + assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. })); + + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should remain readable"); + assert!( + current.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()), + "lost no_lock cleanup must not remove the restore header" + ); + assert_eq!( + rustfs_utils::http::metadata_compat::get_consistent_str( + current.user_defined.as_ref(), + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ), + Some(operation_id.to_string().as_str()), + "lost no_lock cleanup must not remove the restore operation id" + ); + } + #[tokio::test] #[serial_test::serial] async fn restore_worker_propagates_operation_id_to_final_put_commit() { @@ -8327,6 +8869,48 @@ mod transition_commit_failure_tests { .expect("the transitioned object body should drain"); assert_eq!(restored, payload); } + + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn object_transaction_fencing_requires_live_fleet_proof_before_put_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transaction-fencing-no-proof"; + let object = "object.bin"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let err = temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut reader = PutObjReader::from_vec(b"must-not-commit-without-proof".to_vec()); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + }, + ) + .await + .expect_err("object fencing must fail closed without a live fleet proof"); + + assert!( + err.to_string() + .contains("object transaction fencing requires a live fleet capability proof"), + "unexpected error: {err:?}" + ); + for disk in &disk_stores { + let missing = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect_err("failed fenced PUT must not publish object metadata"); + assert!( + matches!(missing, DiskError::FileNotFound | DiskError::FileVersionNotFound), + "failed fenced PUT left unexpected disk state: {missing:?}" + ); + } + } } #[cfg(all(test, feature = "test-util"))] @@ -9793,13 +10377,62 @@ mod transition_source_identity_matrix_tests { #[cfg(test)] mod heterogeneous_pool_put_tests { - use super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity; + use super::hermetic_set_disks_support::{ + hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity, + hermetic_set_disks_isolated as hermetic_set_disks, + }; use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::{DiskAPI as _, ReadOptions}; + use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test; use rustfs_config::server_config::KVS; + use serial_test::serial; use tokio::io::AsyncReadExt; + async fn make_bucket(disks: &[DiskStore], bucket: &str) { + for disk in disks { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + } + + async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec> { + let mut epochs = Vec::with_capacity(disks.len()); + for (disk_index, disk) in disks.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}")); + epochs.push( + file_info + .object_transaction_epoch() + .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")), + ); + } + epochs + } + + async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("current object metadata should read") + .data_dir + .expect("test object should be stored out-of-line") + } + + async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok() + } + + async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir)) + .await + .is_ok() + } + + fn large_payload(fill: u8) -> Vec { + vec![fill; 1024 * 1024] + } + #[tokio::test] async fn second_pool_regular_put_uses_its_own_layout_and_round_trips() { // Deliberately inject the first pool's invalid scalar fallback. The @@ -9844,6 +10477,317 @@ mod heterogeneous_pool_put_tests { .expect("second-pool regular PUT should stream"); assert_eq!(restored, payload); } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_persists_epoch_only_when_gate_is_enabled() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-object-transaction-epoch"; + make_bucket(&disk_stores, bucket).await; + + let mut default_reader = PutObjReader::from_vec(b"default gate stays epoch-free".to_vec()); + set_disks + .put_object(bucket, "default.bin", &mut default_reader, &ObjectOptions::default()) + .await + .expect("default PUT should commit"); + assert_eq!( + object_transaction_epochs(&disk_stores, bucket, "default.bin").await, + vec![None, None, None, None], + "live proof alone must not write epoch metadata while the opt-in gate is disabled" + ); + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut fenced_reader = PutObjReader::from_vec(b"fenced epoch commit".to_vec()); + set_disks + .put_object(bucket, "fenced.bin", &mut fenced_reader, &ObjectOptions::default()) + .await + .expect("fenced PUT should commit with a live proof"); + }, + ) + .await; + + let epochs = object_transaction_epochs(&disk_stores, bucket, "fenced.bin").await; + let first = epochs[0].expect("fenced PUT should persist an epoch"); + assert!(!first.is_nil()); + assert!(epochs.into_iter().all(|epoch| epoch == Some(first))); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_reconciles_failed_put_cleanup_idempotently() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-reconcile"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut first_reader = PutObjReader::from_vec(large_payload(0x11)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("first fenced PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut overwrite_reader = PutObjReader::from_vec(large_payload(0x22)); + set_disks + .put_object(bucket, object, &mut overwrite_reader, &ObjectOptions::default()) + .await + .expect("overwrite should commit even when old-data cleanup fails"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "fenced failed cleanup should leave a durable receipt" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "injected cleanup failure should leave the old data dir for reconciliation" + ); + } + drop(fault); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("receipt reconciliation should succeed"); + assert_eq!(removed, 4, "all receipt-targeted old dirs should be reclaimed"); + let repeated = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("receipt reconciliation should be idempotent"); + assert_eq!(repeated, 0, "a drained receipt must not be counted again"); + for disk in &disk_stores { + assert!( + !data_dir_exists(disk, bucket, object, old_dir).await, + "reconciled old data dir must be gone" + ); + } + }, + ) + .await; + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_noops_after_epoch_mismatch() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-epoch-mismatch"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut first_reader = PutObjReader::from_vec(large_payload(0x31)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("first fenced PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut second_reader = PutObjReader::from_vec(large_payload(0x32)); + set_disks + .put_object(bucket, object, &mut second_reader, &ObjectOptions::default()) + .await + .expect("second fenced PUT should commit and leave a receipt"); + drop(fault); + + let mut third_reader = PutObjReader::from_vec(large_payload(0x33)); + set_disks + .put_object(bucket, object, &mut third_reader, &ObjectOptions::default()) + .await + .expect("third fenced PUT should advance the current epoch"); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("stale receipt reconciliation should succeed"); + assert_eq!(removed, 0, "stale receipt epoch must not reclaim after a newer commit"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "stale receipt should remain as no-op evidence" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "epoch mismatch must preserve the old receipt target" + ); + } + }, + ) + .await; + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_is_not_persisted_without_epoch_gate() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-gate-off"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + let mut first_reader = PutObjReader::from_vec(large_payload(0x41)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("default-gate first PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let _fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut second_reader = PutObjReader::from_vec(large_payload(0x42)); + set_disks + .put_object(bucket, object, &mut second_reader, &ObjectOptions::default()) + .await + .expect("default-gate overwrite should commit"); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("gate-off receipt reconciliation should be a no-op"); + assert_eq!(removed, 0, "mixed-version/default gate path must not consume epoch-dependent receipts"); + for disk in &disk_stores { + assert!( + !cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "mixed-version/default gate path must not write epoch-dependent receipts" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "cleanup fault should leave old dir without receipt" + ); + } + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_rejects_stale_no_lock_put_epoch() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-object-transaction-stale-epoch"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut initial_reader = PutObjReader::from_vec(b"initial fenced body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial fenced PUT should commit"); + let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("initial fenced PUT should persist an epoch"); + + let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeTransactionEpochVerify); + let stale_set = Arc::clone(&set_disks); + let stale = tokio::spawn(async move { + let mut stale_reader = PutObjReader::from_vec(b"stale writer body".to_vec()); + stale_set + .put_object( + bucket, + object, + &mut stale_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + }); + barrier.wait_until_paused().await; + + let mut winner_reader = PutObjReader::from_vec(b"winning writer body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut winner_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("concurrent fenced PUT should advance the epoch"); + let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("winning fenced PUT should persist an epoch"); + assert_ne!(winning_epoch, initial_epoch); + + barrier.release(); + let err = stale + .await + .expect("stale PUT task should not panic") + .expect_err("stale epoch PUT must be rejected"); + assert_eq!(err, StorageError::PreconditionFailed); + + let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch))); + let mut reader = set_disks + .get_object_reader( + bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("winning object should remain readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("winning body should stream"); + assert_eq!(restored, b"winning writer body"); + }, + ) + .await; + } } #[cfg(test)] diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 3fd19b151..d0011cd11 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -482,6 +482,7 @@ impl SetDisks { pub(super) async fn try_get_object_direct_data_shards_with_fileinfo( bucket: &str, object: &str, + erasure_cache: Arc, fi: &FileInfo, files: &[FileInfo], disks: &[Option], @@ -502,13 +503,7 @@ impl SetDisks { return Ok(None); } - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; + let erasure = erasure_cache.get_for_file_info(fi)?; let checksum_info = fi.erasure.get_checksum_info(part.number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { @@ -636,6 +631,7 @@ impl SetDisks { // &self, bucket: &str, object: &str, + erasure_cache: Arc, offset: usize, length: i64, writer: &mut W, @@ -730,13 +726,7 @@ impl SetDisks { object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds" ); - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; + let erasure = erasure_cache.get_for_file_info(&fi)?; let part_indices: Vec = (part_index..=last_part_index).collect(); debug!(bucket, object, ?part_indices, "Multipart part indices to stream"); @@ -1170,6 +1160,7 @@ impl SetDisks { pub(super) async fn get_object_decode_reader_with_fileinfo( bucket: &str, object: &str, + erasure_cache: Arc, fi: &FileInfo, files: &[FileInfo], disks: &[Option], @@ -1180,14 +1171,7 @@ impl SetDisks { metrics_size_bucket: &'static str, prefer_data_blocks_first_reader_setup: bool, ) -> Result { - let erasure = coding::Erasure::try_new_with_options( - fi.erasure.data_blocks, - fi.erasure.parity_blocks, - fi.erasure.block_size, - fi.uses_legacy_checksum, - ) - .map_err(Error::from)?; - + let erasure = erasure_cache.get_for_file_info(fi)?; let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); if fi.parts.len() == 1 { @@ -1574,7 +1558,7 @@ struct LazyCodecPartContext { fi: FileInfo, files: Vec, disks: Vec>, - erasure: coding::Erasure, + erasure: Arc, skip_verify_bitrot: bool, metrics_object_class: &'static str, metrics_size_bucket: &'static str, @@ -2058,6 +2042,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( "bucket", "object", + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -2088,6 +2073,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 2, 1, &mut output, @@ -2111,6 +2097,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), usize::MAX, 1, &mut output, @@ -2132,6 +2119,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 1, 1, &mut output, @@ -2155,6 +2143,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -2192,6 +2181,7 @@ mod metadata_cache_tests { SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 0, &mut output, @@ -2224,6 +2214,7 @@ mod metadata_cache_tests { let err = SetDisks::get_object_with_fileinfo( bucket, object, + Arc::new(ErasureCache::new()), 0, 1, &mut output, @@ -4128,6 +4119,7 @@ mod tests { let result = SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &[], &[], @@ -4150,6 +4142,7 @@ mod tests { let invalid_size = SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &single_part, &[], &[], @@ -4170,6 +4163,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4194,6 +4188,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4222,6 +4217,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &multipart, &[], &[], @@ -4275,6 +4271,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -4328,6 +4325,7 @@ mod tests { SetDisks::get_object_decode_reader_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), &fi, &files, &disks, @@ -4372,6 +4370,7 @@ mod tests { SetDisks::get_object_with_fileinfo( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, + Arc::new(ErasureCache::new()), 0, part_data.len() as i64, &mut output, diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index d968cd487..86126c535 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -48,6 +48,23 @@ impl RestoreCleanupIdentity { } } +fn ensure_restore_metadata_lock_held(bucket: &str, object: &str, opts: &ObjectOptions, mode: &'static str) -> Result<()> { + if opts + .namespace_lock_fence + .as_ref() + .is_some_and(NamespaceLockFence::is_lock_lost) + { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode, + bucket: bucket.to_string(), + object: object.to_string(), + required: 1, + achieved: 0, + }); + } + Ok(()) +} + impl SetDisks { pub(super) async fn finalize_restore_metadata( &self, @@ -88,6 +105,7 @@ impl SetDisks { if !expected.matches_file_info(&fi, &expected_etag) { return Err(Error::other("restored object changed before restore metadata finalization")); } + ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?; let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1)); fi.metadata.insert( @@ -159,6 +177,7 @@ impl SetDisks { if !expected.matches_file_info(&fi, &expected_etag) { return Ok(()); } + ensure_restore_metadata_lock_held(bucket, object, opts, "restore_cleanup_metadata")?; fi.metadata.remove(X_AMZ_RESTORE.as_str()); fi.metadata.remove(AMZ_RESTORE_EXPIRY_DAYS); fi.metadata.remove(AMZ_RESTORE_REQUEST_DATE); diff --git a/crates/ecstore/src/store/list.rs b/crates/ecstore/src/store/list.rs deleted file mode 100644 index 5f7f480b5..000000000 --- a/crates/ecstore/src/store/list.rs +++ /dev/null @@ -1,81 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use super::*; - -impl ECStore { - #[instrument(level = "trace", skip(self))] - #[allow(clippy::too_many_arguments)] - pub(super) async fn handle_list_objects_v2( - self: Arc, - bucket: &str, - prefix: &str, - continuation_token: Option, - delimiter: Option, - max_keys: i32, - fetch_owner: bool, - start_after: Option, - incl_deleted: bool, - ) -> Result { - self.inner_list_objects_v2( - bucket, - prefix, - continuation_token, - delimiter, - max_keys, - fetch_owner, - start_after, - incl_deleted, - ) - .await - } - - #[instrument(skip(self))] - pub(super) async fn handle_list_object_versions( - self: Arc, - bucket: &str, - prefix: &str, - marker: Option, - version_marker: Option, - delimiter: Option, - max_keys: i32, - ) -> Result { - self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) - .await - } - - pub(crate) async fn list_object_versions_for_lifecycle( - self: Arc, - bucket: &str, - prefix: &str, - marker: Option, - version_marker: Option, - delimiter: Option, - max_keys: i32, - ) -> Result { - self.inner_list_object_versions_for_lifecycle(bucket, prefix, marker, version_marker, delimiter, max_keys) - .await - } - - pub(super) async fn handle_walk( - self: Arc, - rx: CancellationToken, - bucket: &str, - prefix: &str, - result: tokio::sync::mpsc::Sender, - opts: WalkOptions, - ) -> Result<()> { - self.walk_internal(rx, bucket, prefix, result, opts).await - } -} diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index df4cb3980..9adadbb13 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -3844,7 +3844,7 @@ impl ECStore { .await } - pub(crate) async fn inner_list_object_versions_for_lifecycle( + pub(crate) async fn list_object_versions_for_lifecycle( self: Arc, bucket: &str, prefix: &str, diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 30a76bb69..f51fa6df5 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -148,7 +148,6 @@ mod heal_walk; pub use heal_walk::HealWalkVersion; mod init; pub(crate) mod init_format; -mod list; pub(crate) mod list_objects; mod multipart; mod object; @@ -601,7 +600,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore { start_after: Option, incl_deleted: bool, ) -> Result { - self.handle_list_objects_v2( + self.inner_list_objects_v2( bucket, prefix, continuation_token, @@ -624,7 +623,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore { delimiter: Option, max_keys: i32, ) -> Result { - self.handle_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) + self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) .await } @@ -636,7 +635,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore { result: tokio::sync::mpsc::Sender, opts: WalkOptions, ) -> Result<()> { - self.handle_walk(rx, bucket, prefix, result, opts).await + self.walk_internal(rx, bucket, prefix, result, opts).await } } diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index 0f799a40e..393975c18 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -18,8 +18,9 @@ use rmp_serde::Serializer; use rustfs_utils::HashAlgorithm; use rustfs_utils::http::{ AMZ_OBJECT_TAGGING, SUFFIX_COMPRESSION, SUFFIX_DATA_MOVED, SUFFIX_DATA_MOVED_TAGS, SUFFIX_FREE_VERSION, SUFFIX_HEALING, - SUFFIX_INLINE_DATA, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, contains_key_str, get_str, - has_internal_suffix, insert_str, is_encryption_metadata_key, starts_with_ignore_ascii_case, + SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, + contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key, + starts_with_ignore_ascii_case, }; use s3s::dto::{RestoreStatus, Timestamp}; use s3s::header::X_AMZ_RESTORE; @@ -1172,6 +1173,22 @@ impl FileInfo { insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, String::new()); } + pub fn set_object_transaction_epoch(&mut self, epoch: Uuid) { + insert_str(&mut self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH, epoch.to_string()); + } + + pub fn object_transaction_epoch(&self) -> Result> { + if !contains_key_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH) { + return Ok(None); + } + let value = get_consistent_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH).ok_or(Error::FileCorrupt)?; + let epoch = Uuid::parse_str(value).map_err(|_| Error::FileCorrupt)?; + if epoch.is_nil() { + return Err(Error::FileCorrupt); + } + Ok(Some(epoch)) + } + pub fn inline_data(&self) -> bool { contains_key_str(&self.metadata, SUFFIX_INLINE_DATA) && !self.is_remote() } @@ -1484,6 +1501,46 @@ mod tests { assert_eq!(ei.get_checksum_info(99).algorithm, HashAlgorithm::HighwayHash256S); } + #[test] + fn object_transaction_epoch_uses_consistent_dual_internal_metadata() { + let mut fi = validation_test_fileinfo(); + assert_eq!(fi.object_transaction_epoch().expect("absent epoch should decode"), None); + + let epoch = Uuid::new_v4(); + let epoch_text = epoch.to_string(); + fi.set_object_transaction_epoch(epoch); + assert_eq!(fi.object_transaction_epoch().expect("written epoch should decode"), Some(epoch)); + assert_eq!(fi.metadata.get("x-rustfs-internal-object-transaction-epoch"), Some(&epoch_text)); + assert_eq!(fi.metadata.get("x-minio-internal-object-transaction-epoch"), Some(&epoch_text)); + + let mut rustfs_only = validation_test_fileinfo(); + rustfs_only + .metadata + .insert("x-rustfs-internal-object-transaction-epoch".to_string(), epoch_text); + assert_eq!( + rustfs_only + .object_transaction_epoch() + .expect("single compatibility key should decode"), + Some(epoch) + ); + + let mut conflicting = fi.clone(); + conflicting + .metadata + .insert("x-minio-internal-object-transaction-epoch".to_string(), Uuid::new_v4().to_string()); + assert_eq!(conflicting.object_transaction_epoch(), Err(Error::FileCorrupt)); + + let mut malformed = validation_test_fileinfo(); + malformed + .metadata + .insert("x-rustfs-internal-object-transaction-epoch".to_string(), "not-a-uuid".to_string()); + assert_eq!(malformed.object_transaction_epoch(), Err(Error::FileCorrupt)); + + let mut nil = validation_test_fileinfo(); + nil.set_object_transaction_epoch(Uuid::nil()); + assert_eq!(nil.object_transaction_epoch(), Err(Error::FileCorrupt)); + } + // backlog#949: distribution range/permutation validation. #[test] fn is_valid_distribution_accepts_permutation() { diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index bc71f81ec..a3cc07b6d 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -59,6 +59,7 @@ pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destina pub const SUFFIX_TRANSITION_TRANSACTION_ID: &str = "transition-transaction-id"; pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id"; pub const SUFFIX_BUCKET_INCARNATION_ID: &str = "bucket-incarnation-id"; +pub const SUFFIX_OBJECT_TRANSACTION_EPOCH: &str = "object-transaction-epoch"; pub const SUFFIX_FREE_VERSION: &str = "free-version"; pub const SUFFIX_PURGESTATUS: &str = "purgestatus"; pub const SUFFIX_REPLICA_STATUS: &str = "replica-status"; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index cc4d0b798..8dd561da1 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -46,6 +46,7 @@ use super::storage_api::multipart_usecase::options::{ get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, put_opts_with_replication_authorization, validate_archive_content_encoding, }; +use super::storage_api::multipart_usecase::request_context::spawn_traced_join; use super::storage_api::multipart_usecase::s3_api::multipart::{ ListMultipartUploadsParams, build_list_multipart_uploads_output, build_list_parts_output, parse_list_multipart_uploads_params, parse_list_parts_params, parse_upload_part_number, @@ -588,56 +589,94 @@ impl DefaultMultipartUsecase { None => None, }; - let obj_info = store - .clone() - .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts) - .await - .map_err(ApiError::from)?; - let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await; - record_capacity_write(Some(capacity_scope_token)).await; - - if let Some(metadata_sys) = quota_metadata_sys.as_ref() { - if opts.replication_request { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); - match quota_checker - .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) + let complete_commit = spawn_traced_join({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + let key = key.clone(); + let upload_id = upload_id.clone(); + let opts = opts.clone(); + let quota_metadata_sys = quota_metadata_sys.clone(); + async move { + let obj_info = store + .clone() + .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts) .await - { - Ok(check_result) if !check_result.allowed => { - let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await; - let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; - return Err(S3Error::with_message( - S3ErrorCode::InvalidRequest, - format!( - "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", - check_result.current_usage.unwrap_or(0), - check_result.quota_limit.unwrap_or(0) - ), - )); + .map_err(ApiError::from)?; + let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await; + record_capacity_write(Some(capacity_scope_token)).await; + + if let Some(metadata_sys) = quota_metadata_sys.as_ref() { + if opts.replication_request { + let quota_checker = QuotaChecker::new(metadata_sys.clone()); + match quota_checker + .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) + .await + { + Ok(check_result) if !check_result.allowed => { + let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await; + let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!( + "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", + check_result.current_usage.unwrap_or(0), + check_result.quota_limit.unwrap_or(0) + ), + )); + } + Err(err) => { + warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + } + Ok(_) => {} + } } - Err(err) => { - warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + + let committed_size = if opts.replication_request { + obj_info.size.max(0) as u64 + } else { + quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())? + }; + if versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await; + } else { + record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await; } - Ok(_) => {} } + + enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; + + let mt2 = obj_info.user_defined.clone(); + let dsc = must_replicate_object( + &bucket, + &key, + &mt2, + "".to_string(), + opts.delete_marker_replication_status(), + opts.clone(), + ) + .await; + + if dsc.replicate_any() { + warn!("need multipart replication"); + schedule_object_replication(obj_info.clone(), store, dsc).await; + } + + rustfs_scanner::record_dirty_usage_bucket(&bucket); + Ok::<_, S3Error>(obj_info) } + }); + let obj_info = complete_commit.await.map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("complete multipart upload commit owner task failed: {err}"), + ) + })??; - let committed_size = if opts.replication_request { - obj_info.size.max(0) as u64 - } else { - quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())? - }; - if versioned { - record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await; - } else { - record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await; - } - } - - enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; - - let raw_mpu_version = obj_info.version_id.map(|v| v.to_string()); - let mpu_version = if versioned { raw_mpu_version.clone() } else { None }; + let mpu_version = if versioned { + obj_info.version_id.map(|v| v.to_string()) + } else { + None + }; let mpu_version_for_event = mpu_version.clone(); // checksum: stored (decrypted) values take precedence over the request input; // additional algorithms (XXHash3/64/128, SHA-512, MD5), which have no typed @@ -660,28 +699,18 @@ impl DefaultMultipartUsecase { bucket: Some(bucket.clone()), key: Some(key.clone()), e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)), - location: Some(location.clone()), + location: Some(location), server_side_encryption: server_side_encryption.clone(), ssekms_key_id: ssekms_key_id.clone(), - checksum_crc32: checksum_crc32.clone(), - checksum_crc32c: checksum_crc32c.clone(), - checksum_sha1: checksum_sha1.clone(), - checksum_sha256: checksum_sha256.clone(), - checksum_crc64nvme: checksum_crc64nvme.clone(), - checksum_type: checksum_type.clone(), + checksum_crc32, + checksum_crc32c, + checksum_sha1, + checksum_sha256, + checksum_crc64nvme, + checksum_type, version_id: mpu_version, ..Default::default() }; - let mt2 = obj_info.user_defined.clone(); - let dsc = - must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) - .await; - - if dsc.replicate_any() { - warn!("need multipart replication"); - schedule_object_replication(obj_info.clone(), store, dsc).await; - } - // Set object info for event notification helper = helper.object(obj_info); if let Some(version_id) = &mpu_version_for_event { @@ -712,7 +741,6 @@ impl DefaultMultipartUsecase { } let result = Ok(response); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); result } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 0e3bbb031..b48430d28 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -2989,6 +2989,11 @@ struct PutObjectChecksums { crc64nvme: Option, } +struct PutObjectCommitResult { + obj_info: ObjectInfo, + put_versioned: bool, +} + fn normalize_delete_objects_version_id( version_id: Option, ) -> std::result::Result<(Option, Option), String> { @@ -5932,7 +5937,7 @@ impl DefaultObjectUsecase { reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?; rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start); - let mut reader = PutObjReader::new(reader); + let reader = PutObjReader::new(reader); let mt2 = metadata.clone(); opts.user_defined.extend(metadata); @@ -6005,97 +6010,145 @@ impl DefaultObjectUsecase { } else { None }; - let object_traffic_progress = object_traffic_health - .as_deref() - .and_then(ObjectTrafficHealth::track_write_storage); - let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now); - let (obj_info, backfilled_old_current_size) = match store - .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) - .await - .map_err(ApiError::from) - { - Ok(obj_info) => { - store_put_watchdog.cancel(); - debug!( - target: "rustfs::app::object_usecase", - event = EVENT_PUT_OBJECT_STORE_RETURNED, - component = LOG_COMPONENT_APP, - subsystem = LOG_SUBSYSTEM_OBJECT, - request_id = %request_id, - bucket = %bucket, - key = %key, - put_path = put_path, - object_size = actual_size, - duration_ms = start_time.elapsed().as_millis() as u64, - result = "success", - "PutObject store write returned" - ); - obj_info + let put_commit = spawn_traced_join({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + let key = key.clone(); + let opts = opts.clone(); + let cache_adapter = cache_adapter.clone(); + let request_id = request_id.clone(); + let put_path = put_path.to_string(); + async move { + let object_traffic_progress = object_traffic_health + .as_deref() + .and_then(ObjectTrafficHealth::track_write_storage); + let mut reader = reader; + let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now); + let (obj_info, backfilled_old_current_size) = match store + .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) + .await + .map_err(ApiError::from) + { + Ok(obj_info) => { + store_put_watchdog.cancel(); + debug!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = %put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "success", + "PutObject store write returned" + ); + obj_info + } + Err(err) => { + store_put_watchdog.cancel(); + rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); + warn!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = %put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "error", + error = %err, + "PutObject store write returned" + ); + return Err(err.into()); + } + }; + rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); + drop(object_traffic_progress); + #[cfg(test)] + wait_for_put_post_store_test_hook(&bucket).await; + + let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now); + maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; + let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await; + + let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; + // Fast in-memory update for immediate quota and admin usage consistency. + // The previous current size comes from the prelookup when it ran, + // otherwise from the rename_data backfill (rustfs/backlog#1009); the + // backfill reproduces the lookup's observation bit for bit (latest + // version's ObjectInfo.size — 0 for a delete-marker latest — or + // not-found → None). + match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) + { + Some(previous_current_size) => { + if put_versioned { + record_bucket_object_version_write_memory( + &bucket, + previous_current_size, + obj_info.size.max(0) as u64, + ) + .await; + } else { + record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + } + } + None => { + // Neither source could determine the previous state (peers + // predating the backfill field during a rolling upgrade, or + // sub-quorum metadata divergence). Record the components that + // are correct regardless; the next authoritative scanner + // refresh replaces the in-memory numbers. + debug!( + target: "rustfs::app::object_usecase", + bucket = %bucket, + key = %key, + put_versioned, + "put_object old-size backfill unknown; recording degraded usage delta" + ); + record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned) + .await; + } + } + + if dsc.replicate_any() { + schedule_object_replication(obj_info.clone(), store, dsc).await; + } + + rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); + + let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now); + let manager = get_capacity_manager(); + manager.record_write_operation().await; + rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start); + + Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned }) + } + }); + let PutObjectCommitResult { obj_info, put_versioned } = match put_commit.await { + Ok(Ok(result)) => result, + Ok(Err(err)) => { + let result: S3Result> = Err(err); + put_request_guard.finish_err(); + let _ = helper.complete(&result); + return result; } Err(err) => { - store_put_watchdog.cancel(); - rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); - warn!( - target: "rustfs::app::object_usecase", - event = EVENT_PUT_OBJECT_STORE_RETURNED, - component = LOG_COMPONENT_APP, - subsystem = LOG_SUBSYSTEM_OBJECT, - request_id = %request_id, - bucket = %bucket, - key = %key, - put_path = put_path, - object_size = actual_size, - duration_ms = start_time.elapsed().as_millis() as u64, - result = "error", - error = %err, - "PutObject store write returned" - ); - let result: S3Result> = Err(err.into()); + let result: S3Result> = Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!("put object commit owner task failed: {err}"), + )); put_request_guard.finish_err(); let _ = helper.complete(&result); return result; } }; - rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); - drop(object_traffic_progress); - #[cfg(test)] - wait_for_put_post_store_test_hook(&bucket).await; - - let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now); - maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; - let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await; - - let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; - // Fast in-memory update for immediate quota and admin usage consistency. - // The previous current size comes from the prelookup when it ran, - // otherwise from the rename_data backfill (rustfs/backlog#1009); the - // backfill reproduces the lookup's observation bit for bit (latest - // version's ObjectInfo.size — 0 for a delete-marker latest — or - // not-found → None). - match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) { - Some(previous_current_size) => { - if put_versioned { - record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; - } else { - record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; - } - } - None => { - // Neither source could determine the previous state (peers - // predating the backfill field during a rolling upgrade, or - // sub-quorum metadata divergence). Record the components that - // are correct regardless; the next authoritative scanner - // refresh replaces the in-memory numbers. - debug!( - target: "rustfs::app::object_usecase", - bucket = %bucket, - key = %key, - put_versioned, - "put_object old-size backfill unknown; recording degraded usage delta" - ); - record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned).await; - } - } let raw_version = obj_info.version_id.map(|v| v.to_string()); @@ -6110,17 +6163,6 @@ impl DefaultObjectUsecase { let expiration = resolve_put_object_expiration(&bucket, &obj_info).await; - // Reuse the single replication decision computed before commit (see `dsc` - // above) so the pending metadata persisted with the object and the - // post-commit schedule always derive from the same immutable decision. - // Recomputing here would repeat the versioning/config/target traversal and, - // worse, allow a replication-config hot update between the two phases to - // produce a pending-without-schedule or schedule-without-pending divergence - // (https://github.com/rustfs/backlog/issues/1320). - if dsc.replicate_any() { - schedule_object_replication(obj_info.clone(), store, dsc).await; - } - let mut checksums = PutObjectChecksums { crc32: input.checksum_crc32, crc32c: input.checksum_crc32c, @@ -6159,14 +6201,6 @@ impl DefaultObjectUsecase { inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers); let result = Ok(response); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); - rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); - - // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead) - let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now); - let manager = get_capacity_manager(); - manager.record_write_operation().await; - rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start); // Record PutObject metrics via zero-copy-metrics { @@ -11507,6 +11541,76 @@ mod tests { assert!(!recovered.write_stalled); } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn cancelled_put_request_completes_post_commit_publication() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let (store, context) = real_cold_fill_test_context().await; + let bucket = format!("put-owner-tail-{}", Uuid::new_v4()); + let object = "object.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("PUT owner-tail bucket must be created"); + + let old_body = Bytes::from_static(b"old body that must be invalidated"); + let old_info = put_real_cold_fill_object(&store, &bucket, object, &old_body).await; + let adapter = context.object_data_cache(); + let old_plan = real_cold_fill_plan(&adapter, &bucket, object, &old_info); + + let post_store_entered = Arc::new(tokio::sync::Barrier::new(2)); + let post_store_resume = Arc::new(tokio::sync::Barrier::new(2)); + install_put_post_store_test_hook(bucket.clone(), Arc::clone(&post_store_entered), Arc::clone(&post_store_resume)); + + let payload = Bytes::from_static(b"published despite caller cancellation"); + let put_input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone())))) + .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64"))) + .build() + .expect("PUT input must build"); + let put_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let put = tokio::spawn(async move { + put_usecase + .execute_put_object(&FS::new(), build_request(put_input, Method::PUT)) + .await + }); + + tokio::time::timeout(Duration::from_secs(10), post_store_entered.wait()) + .await + .expect("PUT must reach the post-store owner-tail hook"); + assert_eq!( + adapter.fill_body(&old_plan, old_body.clone()).await, + rustfs_object_data_cache::ObjectDataCacheFillResult::Inserted, + "test must republish the old body while the owner tail is paused" + ); + put.abort(); + post_store_resume.wait().await; + let _ = put.await.expect_err("outer request task must be cancelled"); + + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if matches!( + adapter.lookup_body(&old_plan).await, + rustfs_object_data_cache::ObjectDataCacheLookup::Miss + ) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("post-commit owner tail must invalidate stale body cache after caller cancellation"); + + let recovered = store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("cancelled request's owned commit must still publish the object"); + assert_eq!(recovered.size, i64::try_from(payload.len()).expect("test payload length must fit i64")); + } + #[tokio::test] async fn object_progress_tracks_zero_byte_and_zero_copy_put_lock_waits() { use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c18e8c3b3..39d7d7685 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1150,7 +1150,9 @@ pub(crate) mod multipart_usecase { } } - pub(crate) use super::{access, bucket, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse}; + pub(crate) use super::{ + access, bucket, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse, + }; pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader}; } diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index c9b0a14eb..6db62ecc2 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -984,7 +984,9 @@ trace_hot_spans=( "crates/ecstore/src/store/object.rs:handle_get_object_info" "crates/ecstore/src/set_disk/ops/object.rs:get_object_info" "crates/ecstore/src/store/mod.rs:list_objects_v2" - "crates/ecstore/src/store/list.rs:handle_list_objects_v2" + # The ECStore handle_list_objects_v2 forwarder was folded into the trait impl + # above, so store/mod.rs now carries this hot path's TRACE requirement + # directly (backlog#1821). # The pool-level Sets::list_objects_v2 wrapper was removed with its duplicate # pagination pipeline (backlog#1821); the remaining ECStore and SetDisks # wrappers below still carry the TRACE requirement.