diff --git a/crates/ecstore/src/erasure/codec/workspace.rs b/crates/ecstore/src/erasure/codec/workspace.rs index 0ed30b245..9cd1a54cc 100644 --- a/crates/ecstore/src/erasure/codec/workspace.rs +++ b/crates/ecstore/src/erasure/codec/workspace.rs @@ -76,6 +76,13 @@ impl ShardBufferPool { self.buffers[index] = Some(buf); } + #[cfg(test)] + pub(crate) fn stored_allocation(&self, index: usize) -> Option<(*const u8, usize)> { + self.buffers + .get(index) + .and_then(|buf| buf.as_ref().map(|buf| (buf.as_ptr(), buf.capacity()))) + } + #[cfg(test)] fn stored_capacity(&self, index: usize) -> Option { self.buffers.get(index).and_then(|buf| buf.as_ref().map(Vec::capacity)) diff --git a/crates/ecstore/src/erasure/coding/bitrot.rs b/crates/ecstore/src/erasure/coding/bitrot.rs index e1cfe3971..7b3166fbc 100644 --- a/crates/ecstore/src/erasure/coding/bitrot.rs +++ b/crates/ecstore/src/erasure/coding/bitrot.rs @@ -125,6 +125,11 @@ where self.last_verify_duration } + #[cfg(test)] + pub(crate) fn inner_ref(&self) -> &R { + &self.inner + } + /// Read a single (hash+data) block, verify hash, and copy `out.len()` bytes /// into `out`. Returns an error if the shard is short, the hash mismatches, /// or `out` is larger than one shard. On error `out`'s contents are diff --git a/crates/ecstore/src/erasure/coding/decode.rs b/crates/ecstore/src/erasure/coding/decode.rs index 5eca7df1d..f20bbce2e 100644 --- a/crates/ecstore/src/erasure/coding/decode.rs +++ b/crates/ecstore/src/erasure/coding/decode.rs @@ -25,7 +25,9 @@ use crate::disk::error_reduce::reduce_errs; use crate::erasure::codec::workspace::ShardBufferPool; use crate::erasure::coding::{BitrotReader, Erasure}; use crate::io_support::bitrot::DeferredReaderStripeHandle; -use crate::set_disk::shard_source::{ShardReadCost, ShardStripeSource, StripeReadState}; +use crate::set_disk::shard_source::{ + INLINE_SHARD_SLOTS, ShardBuffers, ShardErrors, ShardReadCost, ShardStripeSource, StripeReadState, +}; use futures::FutureExt; use futures::stream::{FuturesUnordered, StreamExt}; use pin_project_lite::pin_project; @@ -41,9 +43,6 @@ use tracing::{debug, error, warn}; type ShardReadFuture<'a> = Pin, Error>, bool)> + Send + 'a>>; -const INLINE_SHARD_SLOTS: usize = 32; -type ShardBuffers = SmallVec<[Option>; INLINE_SHARD_SLOTS]>; -type ShardErrors = SmallVec<[Option; INLINE_SHARD_SLOTS]>; type ShardIndexes = SmallVec<[usize; INLINE_SHARD_SLOTS]>; type ActiveReaders = SmallVec<[bool; INLINE_SHARD_SLOTS]>; @@ -392,6 +391,7 @@ pub(crate) struct ParallelReader { // Request-scoped shard buffers keyed by shard index. Keeping ownership in // `ParallelReader` avoids dropping unused parity/backup slot buffers between stripes. buffers: ShardBufferPool, + stripe_state: Option>, // Lockstep-path state (verify_reconstruction == true). `engaged[i]` marks // readers that participate in each stripe read: all data slots from the // start, parity slots only once a data shard is missing/dead. Unengaged @@ -596,6 +596,7 @@ where verify_reconstruction, locality_preference_enabled: get_shard_locality_preference_enabled(), buffers: ShardBufferPool::new(e.data_shards + e.parity_shards), + stripe_state: None, engaged, deferred_handles: Vec::new(), stripe_index: 0, @@ -700,6 +701,12 @@ where { #[hotpath::measure(impl_type = "ParallelReader")] pub async fn read(&mut self) -> StripeReadOutput { + let mut state = StripeReadState::with_slot_count(self.readers.len(), self.data_shards); + self.read_into_state(&mut state).await; + state.into_parts() + } + + async fn read_into_state(&mut self, state: &mut StripeReadState) { // On the reconstruction-verifying GET path, read every live shard reader // in lockstep so all readers advance one block per stripe and stay // mutually aligned. The adaptive data-first path below only reads @@ -709,12 +716,14 @@ where // than the data shards, producing "inconsistent read source shards" and // truncating large-object GETs under concurrency (backlog#832). if self.verify_reconstruction { - return self.read_lockstep().await; + self.read_lockstep(state).await; + return; } // if self.readers.len() != self.total_shards { // return Err(io::Error::new(ErrorKind::InvalidInput, "Invalid number of readers")); // } let num_readers = self.readers.len(); + state.reset(num_readers, self.data_shards); let shard_size = if self.offset + self.shard_size > self.shard_file_size { self.shard_file_size - self.offset @@ -723,7 +732,7 @@ where }; if shard_size == 0 { - return (smallvec![None; num_readers], smallvec![None; num_readers]); + return; } // Advance to the next stripe so the following read() computes the correct @@ -734,8 +743,7 @@ where // is only read above to derive `shard_size`, so advancing here is safe. self.offset += shard_size; - let mut shards: ShardBuffers = smallvec![None; num_readers]; - let mut errs: ShardErrors = smallvec![None; num_readers]; + let (shards, errs) = state.parts_mut(); let read_costs = self.read_costs.as_slice(); let locality_preference_enabled = self.locality_preference_enabled; let low_cost_available = self @@ -882,8 +890,8 @@ where } let result_is_err = record_shard_read_result( - &mut shards, - &mut errs, + shards, + errs, &mut retire_readers, &mut success, &mut successful_costs, @@ -944,8 +952,8 @@ where active_readers[i] = false; completed += 1; if record_shard_read_result( - &mut shards, - &mut errs, + shards, + errs, &mut retire_readers, &mut success, &mut successful_costs, @@ -957,7 +965,7 @@ where failed += 1; } } - retire_abandoned_readers(&mut errs, &mut retire_readers, &active_readers); + retire_abandoned_readers(errs, &mut retire_readers, &active_readers); } if let Some(path) = self.metrics_path { @@ -1001,8 +1009,6 @@ where for i in retire_readers { self.readers[i] = None; } - - (shards, errs) } /// Lockstep stripe read for the reconstruction-verifying GET path. @@ -1030,18 +1036,18 @@ where /// stripe would reintroduce the desync. A parity reader that cannot be /// realigned (no pending deferred handle) is likewise retired instead of /// being read out of position. - async fn read_lockstep(&mut self) -> StripeReadOutput { + async fn read_lockstep(&mut self, state: &mut StripeReadState) { let num_readers = self.readers.len(); + state.reset(num_readers, self.data_shards); let shard_size = if self.offset + self.shard_size > self.shard_file_size { self.shard_file_size - self.offset } else { self.shard_size }; - let mut shards: ShardBuffers = smallvec![None; num_readers]; - let mut errs: ShardErrors = smallvec![None; num_readers]; + let (shards, errs) = state.parts_mut(); if shard_size == 0 { - return (shards, errs); + return; } // Advance to the next stripe (see the matching note in `read`); the @@ -1279,8 +1285,6 @@ where for i in retire_readers { self.readers[i] = None; } - - (shards, errs) } /// Attempt to bring an as-yet-unread parity reader into the lockstep read @@ -1337,10 +1341,20 @@ impl ShardStripeSource for ParallelReader where R: crate::erasure::coding::ShardSource, { - async fn read_next_stripe(&mut self) -> StripeReadState { - let read_quorum = self.data_shards; - let (shards, errors) = ParallelReader::read(self).await; - StripeReadState::from_parts_with_read_costs(shards, errors, &self.read_costs, read_quorum) + async fn read_next_stripe(&mut self) -> Box { + let mut state = self + .stripe_state + .take() + .unwrap_or_else(|| Box::new(StripeReadState::with_slot_count(self.readers.len(), self.data_shards))); + self.read_into_state(&mut state).await; + state + } + + fn recycle_stripe(&mut self, mut state: Box) { + self.recycle_shards(state.shards_mut()); + state.reset(0, self.data_shards); + debug_assert!(self.stripe_state.is_none(), "a stripe cannot be recycled twice"); + self.stripe_state = Some(state); } } @@ -1972,13 +1986,18 @@ mod tests { type BoxedShardReader = crate::io_support::bitrot::ShardReader; #[test] - fn shard_scratch_stays_inline_through_the_common_limit_and_spills_safely() { - let inline: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS]; - assert!(!inline.spilled(), "the common shard-count boundary must not allocate"); - - let spilled: ShardBuffers = smallvec![None; INLINE_SHARD_SLOTS + 1]; - assert!(spilled.spilled(), "larger supported shard counts must fall back to the heap"); - assert_eq!(spilled.len(), INLINE_SHARD_SLOTS + 1); + fn parallel_reader_keeps_stripe_scratch_out_of_line() { + eprintln!( + "parallel_reader={} stripe_state={} cached_state={}", + std::mem::size_of::>>>(), + std::mem::size_of::(), + std::mem::size_of::>>() + ); + assert_eq!( + std::mem::size_of::>>(), + std::mem::size_of::(), + "the request-scoped cache must remain pointer-sized", + ); } #[tokio::test] @@ -1997,6 +2016,62 @@ mod tests { assert_eq!(errors.len(), TOTAL_SHARDS); } + #[tokio::test] + async fn codec_reader_reuses_inline_and_spilled_stripe_scratch_between_reads() { + for total_shards in [INLINE_SHARD_SLOTS, INLINE_SHARD_SLOTS + 1] { + let data_shards = total_shards - 1; + let readers = std::iter::repeat_with(|| None).take(total_shards).collect(); + let erasure = Erasure::new(data_shards, 1, data_shards * 2); + let mut reader: ParallelReader>> = ParallelReader::new(readers, erasure, 0, data_shards * 2); + + let first = ShardStripeSource::read_next_stripe(&mut reader).await; + let first_state = (&*first) as *const StripeReadState; + let first_storage = first.scratch_storage(); + assert_eq!(first_storage.2, total_shards > INLINE_SHARD_SLOTS); + assert_eq!(first_storage.3, total_shards > INLINE_SHARD_SLOTS); + ShardStripeSource::recycle_stripe(&mut reader, first); + + let second = ShardStripeSource::read_next_stripe(&mut reader).await; + let second_storage = second.scratch_storage(); + + assert_eq!( + (&*second) as *const StripeReadState, + first_state, + "the request-scoped state must be reused" + ); + assert_eq!(second_storage.0, first_storage.0, "shard slots must reuse their allocation"); + assert_eq!(second_storage.1, first_storage.1, "error slots must reuse their allocation"); + assert_eq!(second.into_parts().0.len(), total_shards); + } + } + + #[tokio::test] + async fn codec_reader_returns_shard_allocations_to_the_request_pool() { + const SHARD_SIZE: usize = 16; + let hash_algo = HashAlgorithm::None; + let readers = vec![Some(create_reader(SHARD_SIZE, 2, 0x5a, &hash_algo, false).await)]; + let erasure = Erasure::new(1, 0, SHARD_SIZE); + let mut reader = ParallelReader::new(readers, erasure, 0, SHARD_SIZE * 2); + + let first = ShardStripeSource::read_next_stripe(&mut reader).await; + let first_allocation = first + .shard_allocation(0) + .expect("the first stripe should own its shard allocation"); + ShardStripeSource::recycle_stripe(&mut reader, first); + assert_eq!( + reader.buffers.stored_allocation(0), + Some(first_allocation), + "recycling a stripe must return its shard allocation to the request pool" + ); + + let second = ShardStripeSource::read_next_stripe(&mut reader).await; + assert_eq!( + second.shard_allocation(0), + Some(first_allocation), + "the next stripe must reuse the pooled shard allocation" + ); + } + /// Counts the raw bytes pulled from a shard stream, to prove which shards /// a decode path actually touches (backlog#923 call-count evidence). struct CountingShardReader { diff --git a/crates/ecstore/src/erasure/coding/decode_reader.rs b/crates/ecstore/src/erasure/coding/decode_reader.rs index 91d4c6479..8fb7a7cfc 100644 --- a/crates/ecstore/src/erasure/coding/decode_reader.rs +++ b/crates/ecstore/src/erasure/coding/decode_reader.rs @@ -65,7 +65,7 @@ enum FillPolicy { } impl FillPolicy { - fn from_env() -> Self { + fn load() -> Self { match rustfs_utils::get_env_usize( ENV_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MAX_INFLIGHT, @@ -75,6 +75,22 @@ impl FillPolicy { } } + fn from_env() -> Self { + #[cfg(test)] + { + Self::load() + } + #[cfg(not(test))] + { + Self::cached_core(Self::load) + } + } + + fn cached_core(load: impl FnOnce() -> Self) -> Self { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(load) + } + const fn max_inflight(self) -> usize { match self { Self::SingleInFlight => 1, @@ -479,22 +495,30 @@ where let mut deferred_error = None; let fill_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let state = source.read_next_stripe().await; + let mut state = source.read_next_stripe().await; record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start); let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let mut output_buf = reusable_buffers.pop().unwrap_or_default(); - let result = - match decode_stripe_into(metrics_path, stage_metrics_enabled, engine, workspace, state, remaining, &mut output_buf) { - Ok(true) => Ok(Some(output_buf)), - Ok(false) => { - reusable_buffers.push(output_buf); - Ok(None) - } - Err(err) => { - reusable_buffers.push(output_buf); - Err(err) - } - }; + let result = match decode_stripe_into( + metrics_path, + stage_metrics_enabled, + engine, + workspace, + &mut state, + remaining, + &mut output_buf, + ) { + Ok(true) => Ok(Some(output_buf)), + Ok(false) => { + reusable_buffers.push(output_buf); + Ok(None) + } + Err(err) => { + reusable_buffers.push(output_buf); + Err(err) + } + }; + source.recycle_stripe(state); record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start); if let Ok(Some(first_buf)) = result.as_ref() { let mut remaining_after_first = remaining.saturating_sub(first_buf.len()); @@ -503,7 +527,7 @@ where break; } let stripe_read_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let state = source.read_next_stripe().await; + let mut state = source.read_next_stripe().await; record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_STRIPE_READ, stripe_read_stage_start); let decode_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let mut queued_buf = reusable_buffers.pop().unwrap_or_default(); @@ -512,10 +536,11 @@ where stage_metrics_enabled, engine, workspace, - state, + &mut state, remaining_after_first, &mut queued_buf, ); + source.recycle_stripe(state); record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_DECODE, decode_stage_start); match queued_result { Ok(true) => { @@ -717,7 +742,7 @@ fn decode_stripe_into( stage_metrics_enabled: bool, engine: &E, workspace: &mut E::Workspace, - state: StripeReadState, + state: &mut StripeReadState, remaining: usize, output: &mut Vec, ) -> io::Result @@ -725,7 +750,7 @@ where E: ErasureDecodeEngine, { output.clear(); - if state.slots().is_empty() { + if state.is_empty() { return Ok(false); } if !state.can_decode() { @@ -741,13 +766,12 @@ where ); record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start); let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); - emit_data_shards_into(&state, engine.data_shards(), engine.block_size(), remaining, output)?; + emit_data_shards_into(state, engine.data_shards(), engine.block_size(), remaining, output)?; record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_EMIT, emit_stage_start); return Ok(true); } - let (mut shards, _errs) = state.into_parts(); - let reconstruct_outcome = match engine.reconstruct_into(&mut shards, workspace) { + let reconstruct_outcome = match engine.reconstruct_into(state.shards_mut(), workspace) { Ok(outcome) => outcome, Err(err) => { record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start); @@ -757,7 +781,7 @@ where rustfs_io_metrics::record_get_object_reconstruct_outcome(metrics_path, engine.engine_name(), reconstruct_outcome); record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_RECONSTRUCT, reconstruct_stage_start); - if shards.len() < engine.data_shards() { + if state.shards_mut().len() < engine.data_shards() { return Err(io::Error::new( ErrorKind::UnexpectedEof, "decoded stripe has fewer shards than data shard count", @@ -766,7 +790,7 @@ where let emit_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); reserve_output_capacity(output, engine.block_size().min(remaining)); - for shard in shards.iter().take(engine.data_shards()) { + for shard in state.shards_mut().iter().take(engine.data_shards()) { if output.len() >= remaining { break; } @@ -806,10 +830,7 @@ fn emit_data_shards_into( if output.len() >= remaining { break; } - let Some(slot) = state.slot_by_index(index) else { - return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard")); - }; - let Some(shard) = slot.data_bytes() else { + let Some(shard) = state.data_bytes(index) else { return Err(io::Error::new(ErrorKind::UnexpectedEof, "decoded stripe is missing a data shard")); }; let copy_len = shard.len().min(remaining - output.len()); @@ -826,7 +847,7 @@ mod tests { }; use crate::erasure::coding::decode::ParallelReader; use crate::erasure::coding::{BitrotReader, BitrotWriter, Erasure}; - use crate::set_disk::shard_source::{ShardSlot, StripeReadState}; + use crate::set_disk::shard_source::StripeReadState; use rustfs_utils::HashAlgorithm; use std::collections::VecDeque; use std::future::{pending, poll_fn}; @@ -845,6 +866,13 @@ mod tests { read_count: Option>, } + struct RecordingStripeSource { + stripes: VecDeque, + read_quorum: usize, + reads: usize, + recycles: usize, + } + struct BlockingSource { started: Arc, dropped: Arc, @@ -899,25 +927,43 @@ mod tests { #[async_trait::async_trait] impl ShardStripeSource for VecStripeSource { - async fn read_next_stripe(&mut self) -> StripeReadState { + async fn read_next_stripe(&mut self) -> Box { if let Some(read_count) = &self.read_count { read_count.fetch_add(1, Ordering::SeqCst); } - self.stripes - .pop_front() - .unwrap_or_else(|| StripeReadState::new(Vec::new(), self.read_quorum)) + Box::new( + self.stripes + .pop_front() + .unwrap_or_else(|| StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum)), + ) + } + } + + #[async_trait::async_trait] + impl ShardStripeSource for RecordingStripeSource { + async fn read_next_stripe(&mut self) -> Box { + self.reads += 1; + Box::new( + self.stripes + .pop_front() + .unwrap_or_else(|| StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum)), + ) + } + + fn recycle_stripe(&mut self, _state: Box) { + self.recycles += 1; } } #[async_trait::async_trait] impl ShardStripeSource for BlockingSource { - async fn read_next_stripe(&mut self) -> StripeReadState { + async fn read_next_stripe(&mut self) -> Box { let _guard = BlockingSourceDropGuard { dropped: Arc::clone(&self.dropped), }; self.started.notify_one(); pending::<()>().await; - StripeReadState::new(Vec::new(), self.read_quorum) + Box::new(StripeReadState::from_parts(Vec::new(), Vec::new(), self.read_quorum)) } } @@ -1090,6 +1136,23 @@ mod tests { }); } + #[test] + fn fill_policy_production_cache_loads_once() { + use std::cell::Cell; + + let loads = Cell::new(0); + for _ in 0..3 { + assert_eq!( + FillPolicy::cached_core(|| { + loads.set(loads.get() + 1); + FillPolicy::DualInFlight + }), + FillPolicy::DualInFlight + ); + } + assert_eq!(loads.get(), 1, "the production fill policy must not re-read the environment per reader"); + } + #[test] fn erasure_decode_reader_rejects_invalid_engine_shape() { let source = VecStripeSource { @@ -1689,7 +1752,10 @@ mod tests { .pop_front() .expect("first stripe should exist"); let mut source = VecStripeSource { - stripes: VecDeque::from([first_state, StripeReadState::new(Vec::new(), erasure.data_shards)]), + stripes: VecDeque::from([ + first_state, + StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards), + ]), read_quorum: erasure.data_shards, read_count: None, }; @@ -1724,13 +1790,14 @@ mod tests { .stripes .pop_front() .expect("first stripe should exist"); - let mut source = VecStripeSource { + let mut source = RecordingStripeSource { stripes: VecDeque::from([ first_state, - StripeReadState::new(vec![ShardSlot::data(0, vec![1])], erasure.data_shards), + StripeReadState::from_parts(vec![Some(vec![1])], Vec::new(), erasure.data_shards), ]), read_quorum: erasure.data_shards, - read_count: None, + reads: 0, + recycles: 0, }; let engine = LegacyEcDecodeEngine::new(erasure); let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared"); @@ -1756,6 +1823,8 @@ mod tests { .kind(), ErrorKind::Other ); + assert_eq!(source.reads, 2, "the fill must read the primary and queued stripe"); + assert_eq!(source.recycles, source.reads, "every completed stripe read must be recycled"); } #[tokio::test] @@ -1768,7 +1837,7 @@ mod tests { .stripes .pop_front() .expect("first stripe should exist"), - StripeReadState::new(Vec::new(), erasure.data_shards), + StripeReadState::from_parts(Vec::new(), Vec::new(), erasure.data_shards), ]), read_quorum: erasure.data_shards, read_count: None, @@ -2028,17 +2097,11 @@ mod tests { } #[test] - fn emit_data_shards_preserves_output_order_for_out_of_order_slots() { - let state = StripeReadState::new( - vec![ - ShardSlot::data(1, b"cd".to_vec()), - ShardSlot::data(0, b"ab".to_vec()), - ShardSlot::data(2, b"ef".to_vec()), - ], - 2, - ); + fn emit_data_shards_preserves_output_order() { + let state = + StripeReadState::from_parts(vec![Some(b"ab".to_vec()), Some(b"cd".to_vec()), Some(b"ef".to_vec())], Vec::new(), 2); - let output = emit_data_shards(&state, 3, 6, 5).expect("out-of-order data slots should emit by shard index"); + let output = emit_data_shards(&state, 3, 6, 5).expect("data slots should emit by shard index"); assert_eq!(output, b"abcde"); } @@ -2051,27 +2114,27 @@ mod tests { }; let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared"); let mut output = Vec::with_capacity(1); - let short_state = StripeReadState::new(vec![ShardSlot::data(0, vec![1, 2, 3, 4])], 1); + let mut short_state = StripeReadState::from_parts(vec![Some(vec![1, 2, 3, 4])], Vec::new(), 1); let err = decode_stripe_into( GET_OBJECT_PATH_CODEC_STREAMING, false, &engine, &mut workspace, - short_state, + &mut short_state, 8, &mut output, ) .expect_err("decoded stripe shorter than data shard count must fail"); assert_eq!(err.kind(), ErrorKind::UnexpectedEof); - let missing_state = StripeReadState::from_parts(vec![None, Some(vec![5, 6, 7, 8])], Vec::new(), 1); + let mut missing_state = StripeReadState::from_parts(vec![None, Some(vec![5, 6, 7, 8])], Vec::new(), 1); let err = decode_stripe_into( GET_OBJECT_PATH_CODEC_STREAMING, false, &engine, &mut workspace, - missing_state, + &mut missing_state, 8, &mut output, ) @@ -2082,6 +2145,35 @@ mod tests { assert!(output.capacity() >= 32); } + #[test] + fn decode_stripe_reconstructs_in_place_without_replacing_slot_storage() { + let erasure = Erasure::new(2, 1, 8); + let engine = LegacyEcDecodeEngine::new(erasure.clone()); + let mut workspace = engine.prepare_workspace(4).expect("workspace should be prepared"); + let encoded = erasure.encode_data(b"abcdefgh").expect("test stripe should encode"); + let mut shards = encoded.into_iter().map(|shard| Some(shard.to_vec())).collect::>(); + shards[0] = None; + let mut state = StripeReadState::from_parts(shards, vec![Some(DiskError::FileCorrupt)], 2); + let before = state.scratch_storage(); + let mut output = Vec::new(); + + let decoded = decode_stripe_into( + GET_OBJECT_PATH_CODEC_STREAMING, + false, + &engine, + &mut workspace, + &mut state, + 8, + &mut output, + ) + .expect("degraded stripe should reconstruct"); + + assert!(decoded); + assert_eq!(output, b"abcdefgh"); + assert_eq!(state.scratch_storage().0, before.0, "reconstruction must retain shard slot storage"); + assert_eq!(state.scratch_storage().1, before.1, "unused error storage must not be rebuilt"); + } + #[tokio::test] async fn erasure_decode_reader_reports_short_source() { let erasure = Erasure::new(4, 2, 32); diff --git a/crates/ecstore/src/erasure/coding/encode.rs b/crates/ecstore/src/erasure/coding/encode.rs index 39b792485..100cb3a1d 100644 --- a/crates/ecstore/src/erasure/coding/encode.rs +++ b/crates/ecstore/src/erasure/coding/encode.rs @@ -586,9 +586,9 @@ impl Erasure { )); } - let shards = self.encode_data_owned(buf)?; + let block = self.encode_data_owned_block(buf)?; let mut mw = MultiWriter::new(writers, quorum); - mw.write(shards).await?; + mw.write_block(&block).await?; mw.shutdown().await?; Ok((reader, total)) } @@ -613,13 +613,13 @@ impl Erasure { return Ok((reader, 0, Vec::new())); } - let shards = self.encode_data_owned(buf)?; - let mut inline_shards = Vec::with_capacity(shards.len()); - for shard in shards { - let hash = HashAlgorithm::HighwayHash256S.hash_encode(&shard); + let block = self.encode_data_owned_block(buf)?; + let mut inline_shards = Vec::with_capacity(block.shards().len()); + for shard in block.shards() { + let hash = HashAlgorithm::HighwayHash256S.hash_encode(shard); let mut encoded = BytesMut::with_capacity(hash.as_ref().len() + shard.len()); encoded.extend_from_slice(hash.as_ref()); - encoded.extend_from_slice(&shard); + encoded.extend_from_slice(shard); inline_shards.push(encoded.freeze()); } @@ -2164,6 +2164,39 @@ mod tests { ); } + #[tokio::test] + async fn cancelling_inline_small_drops_stalled_write() { + const BLOCK_SIZE: usize = 16; + + let (writer_entered_tx, writer_entered) = oneshot::channel(); + let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut writers = vec![Some(bitrot_writer_plain( + StallOnWriteWithSignal { + entered: Some(writer_entered_tx), + writes: writes.clone(), + }, + BLOCK_SIZE, + ))]; + let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE)); + let reader = tokio::io::BufReader::new(Cursor::new(vec![0xA5; BLOCK_SIZE - 1])); + let encode = tokio::spawn(async move { erasure.encode_inline_small(reader, &mut writers, 1).await }); + + tokio::time::timeout(Duration::from_secs(1), writer_entered) + .await + .expect("inline writer should enter before cancellation") + .expect("stalling writer should signal entry"); + encode.abort(); + assert!( + matches!(encode.await, Err(err) if err.is_cancelled()), + "inline encode task should be cancelled" + ); + assert_eq!( + writes.load(std::sync::atomic::Ordering::SeqCst), + 1, + "cancellation must drop the stalled write instead of polling it again" + ); + } + #[tokio::test] async fn encode_returns_unexpected_eof_for_truncated_limited_reader() { let committed = Arc::new(Mutex::new(Vec::new())); @@ -2395,27 +2428,33 @@ mod tests { const DATA_SHARDS: usize = 2; const PARITY_SHARDS: usize = 2; const BLOCK_SIZE: usize = 64; - let payload = b"inline commit payload".to_vec(); let checksum_algo = HashAlgorithm::HighwayHash256S; - let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); - let reader = tokio::io::BufReader::new(Cursor::new(payload.clone())); + for uses_legacy in [false, true] { + let erasure = Arc::new(Erasure::new_with_options(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE, uses_legacy)); + for payload in [Vec::new(), vec![0xA5], vec![0x5A; BLOCK_SIZE - 1], vec![0xC3; BLOCK_SIZE]] { + let reader = tokio::io::BufReader::new(Cursor::new(payload.clone())); + let (_reader, total, inline_shards) = erasure + .clone() + .encode_inline_shards_with_size_hint(reader, payload.len()) + .await + .expect("inline shards should encode"); - let (_reader, total, inline_shards) = erasure - .clone() - .encode_inline_shards_with_size_hint(reader, payload.len()) - .await - .expect("inline shards should encode"); - let raw_shards = erasure - .encode_data_owned(payload.clone()) - .expect("reference shards should encode"); + assert_eq!(total, payload.len()); + if payload.is_empty() { + assert!(inline_shards.is_empty()); + continue; + } - assert_eq!(total, payload.len()); - assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_SHARDS); - for (inline, raw) in inline_shards.iter().zip(raw_shards) { - let mut writer = BitrotWriterWrapper::new(CustomWriter::new_inline_buffer(), raw.len(), checksum_algo.clone()); - writer.write(&raw).await.expect("reference writer should accept shard"); - writer.shutdown().await.expect("reference writer should shutdown"); - assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes")); + let raw_shards = erasure.encode_data(&payload).expect("reference shards should encode"); + assert_eq!(inline_shards.len(), DATA_SHARDS + PARITY_SHARDS); + for (inline, raw) in inline_shards.iter().zip(raw_shards) { + let mut writer = + BitrotWriterWrapper::new(CustomWriter::new_inline_buffer(), raw.len(), checksum_algo.clone()); + writer.write(&raw).await.expect("reference writer should accept shard"); + writer.shutdown().await.expect("reference writer should shutdown"); + assert_eq!(inline.as_ref(), writer.into_inline_data().expect("reference writer should retain bytes")); + } + } } } diff --git a/crates/ecstore/src/erasure/coding/erasure.rs b/crates/ecstore/src/erasure/coding/erasure.rs index 23cac93c7..8b7ff1c10 100644 --- a/crates/ecstore/src/erasure/coding/erasure.rs +++ b/crates/ecstore/src/erasure/coding/erasure.rs @@ -726,101 +726,37 @@ impl Erasure { } fn encode_data_block_inner(&self, data: &[u8]) -> io::Result { - let shard_size_fn = if self.uses_legacy { - calc_shard_size_legacy - } else { - calc_shard_size - }; - let per_shard_size = shard_size_fn(data.len(), self.data_shards); - if per_shard_size == 0 { - return Ok(EncodedBlock::empty()); - } - let need_total_size = per_shard_size * self.total_shard_count(); - - let mut data_buffer = BytesMut::with_capacity(need_total_size); + let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data.len())); data_buffer.extend_from_slice(data); - data_buffer.resize(need_total_size, 0u8); - - { - let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect(); - - if self.parity_shards > 0 { - if self.uses_legacy { - if let Some(encoder) = self.legacy_encoder.as_ref() { - encoder.encode(data_slices)?; - } else { - warn!("parity_shards > 0, uses_legacy but legacy_encoder is None"); - } - } else if let Some(encoder) = self.encoder.as_ref() { - encoder.encode(data_slices)?; - } else { - warn!("parity_shards > 0, but encoder is None"); - } - } - } - - Ok(EncodedBlock { - data: data_buffer.freeze(), - shard_size: per_shard_size, - }) + self.encode_buffer(data_buffer, data.len()) } /// Encode owned data, avoiding a copy when the caller already has a heap buffer. /// Falls back to copying into a new buffer if zero-copy conversion fails. #[hotpath::measure(impl_type = "Erasure")] pub fn encode_data_owned(&self, data: Vec) -> io::Result> { - let shard_size_fn = if self.uses_legacy { - calc_shard_size_legacy - } else { - calc_shard_size - }; - let per_shard_size = shard_size_fn(data.len(), self.data_shards); - if per_shard_size == 0 { - return Ok(vec![Bytes::new(); self.total_shard_count()]); - } - let need_total_size = per_shard_size * self.total_shard_count(); + self.encode_data_owned_block_inner(data) + .map(|block| block.into_shards(self.total_shard_count())) + } + #[hotpath::measure(label = "Erasure::encode_data_owned", impl_type = "Erasure")] + pub(crate) fn encode_data_owned_block(&self, data: Vec) -> io::Result { + self.encode_data_owned_block_inner(data) + } + + fn encode_data_owned_block_inner(&self, data: Vec) -> io::Result { + let data_len = data.len(); // Try zero-copy: Vec -> Bytes -> BytesMut (succeeds when refcount == 1) - let mut data_buffer = match Bytes::from(data).try_into_mut() { - Ok(mut bm) => { - bm.resize(need_total_size, 0u8); - bm - } + let data_buffer = match Bytes::from(data).try_into_mut() { + Ok(data_buffer) => data_buffer, Err(b) => { // Rare path: refcount != 1, fall back to copy - let mut bm = BytesMut::with_capacity(need_total_size); - bm.extend_from_slice(&b); - bm.resize(need_total_size, 0u8); - bm + let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data_len)); + data_buffer.extend_from_slice(&b); + data_buffer } }; - - { - let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect(); - - if self.parity_shards > 0 { - if self.uses_legacy { - if let Some(encoder) = self.legacy_encoder.as_ref() { - encoder.encode(data_slices)?; - } else { - warn!("parity_shards > 0, uses_legacy but legacy_encoder is None"); - } - } else if let Some(encoder) = self.encoder.as_ref() { - encoder.encode(data_slices)?; - } else { - warn!("parity_shards > 0, but encoder is None"); - } - } - } - - let mut data_buffer = data_buffer.freeze(); - let mut shards = Vec::with_capacity(self.total_shard_count()); - for _ in 0..self.total_shard_count() { - let shard = data_buffer.split_to(per_shard_size); - shards.push(shard); - } - - Ok(shards) + self.encode_buffer(data_buffer, data_len) } /// Encode data from an owned `BytesMut` buffer, avoiding the initial copy @@ -833,16 +769,16 @@ impl Erasure { /// `data_len` — so this function never reallocates the buffer. #[hotpath::measure(impl_type = "Erasure")] pub fn encode_data_bytes_mut(&self, data_buffer: BytesMut, data_len: usize) -> io::Result> { - self.encode_data_bytes_mut_block_inner(data_buffer, data_len) + self.encode_buffer(data_buffer, data_len) .map(|block| block.into_shards(self.total_shard_count())) } #[hotpath::measure(label = "Erasure::encode_data_bytes_mut", impl_type = "Erasure")] pub(crate) fn encode_data_bytes_mut_block(&self, data_buffer: BytesMut, data_len: usize) -> io::Result { - self.encode_data_bytes_mut_block_inner(data_buffer, data_len) + self.encode_buffer(data_buffer, data_len) } - fn encode_data_bytes_mut_block_inner(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result { + fn encode_buffer(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result { let shard_size_fn = if self.uses_legacy { calc_shard_size_legacy } else { @@ -1550,10 +1486,16 @@ mod tests { fn encode_data_owned_matches_borrowed_path() { for uses_legacy in [false, true] { let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy); - - assert_owned_encode_matches_borrowed(&erasure, Vec::new()); - assert_owned_encode_matches_borrowed(&erasure, b"small payload".to_vec()); - assert_owned_encode_matches_borrowed(&erasure, (0_u8..37).collect()); + for data in [ + Vec::new(), + vec![0xA5; 1], + b"small payload".to_vec(), + (0_u8..37).collect(), + vec![0xA5; erasure.block_size - 1], + vec![0x5A; erasure.block_size], + ] { + assert_owned_encode_matches_borrowed(&erasure, data); + } } } @@ -1601,26 +1543,41 @@ mod tests { #[test] fn streaming_encoded_block_uses_one_contiguous_backing_buffer() { - let erasure = Erasure::new(8, 8, 64); + for uses_legacy in [false, true] { + let erasure = Erasure::new_with_options(8, 8, 64, uses_legacy); - for data_len in [1, 63, 64] { - let data = (0..data_len).map(|i| i as u8).collect::>(); - let expected = erasure.encode_data(&data).expect("public encode should succeed"); - let borrowed = erasure - .encode_data_block(&data) - .expect("borrowed streaming encode should succeed"); - let owned = erasure - .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len()) - .expect("BytesMut streaming encode should succeed"); + for data_len in [0, 1, 63, 64] { + let data = (0..data_len).map(|i| i as u8).collect::>(); + let expected = erasure.encode_data(&data).expect("public encode should succeed"); + let borrowed = erasure + .encode_data_block(&data) + .expect("borrowed streaming encode should succeed"); + let owned = erasure + .encode_data_owned_block(data.clone()) + .expect("owned streaming encode should succeed"); + let bytes_mut = erasure + .encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len()) + .expect("BytesMut streaming encode should succeed"); - assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref))); - assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref))); - assert_eq!(borrowed.shards().len(), 16); - assert_eq!(borrowed.queued_bytes(), owned.queued_bytes()); + assert_eq!(borrowed.queued_bytes(), owned.queued_bytes()); + assert_eq!(borrowed.queued_bytes(), bytes_mut.queued_bytes()); - let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr(); - for (index, shard) in borrowed.shards().enumerate() { - assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len())); + if data_len == 0 { + assert!(expected.iter().all(Bytes::is_empty)); + assert!(borrowed.is_empty()); + assert!(owned.is_empty()); + assert!(bytes_mut.is_empty()); + continue; + } + + assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref))); + assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref))); + assert!(bytes_mut.shards().eq(expected.iter().map(Bytes::as_ref))); + assert_eq!(borrowed.shards().len(), 16); + let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr(); + for (index, shard) in borrowed.shards().enumerate() { + assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len())); + } } } assert_eq!( diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index 5a3120760..9a4d6dfab 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -56,6 +56,16 @@ pub enum ShardReader { Stream(Box), } +#[cfg(test)] +impl ShardReader { + pub(crate) fn inline_bytes(&self) -> Option<&Bytes> { + match self { + Self::InMemory(cursor) => Some(cursor.get_ref()), + Self::Chunked(_) | Self::Stream(_) => None, + } + } +} + impl AsyncRead for ShardReader { fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut tokio::io::ReadBuf<'_>) -> Poll> { match self.get_mut() { @@ -646,7 +656,7 @@ pub async fn create_bitrot_reader_from_bytes( } #[allow(clippy::too_many_arguments)] -async fn create_bitrot_reader_from_bytes_with_stage_metrics( +pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics( inline_data: Option, disk: Option<&DiskStore>, bucket: &str, diff --git a/crates/ecstore/src/layout/disks_layout.rs b/crates/ecstore/src/layout/disks_layout.rs index f880bfeac..75ca09714 100644 --- a/crates/ecstore/src/layout/disks_layout.rs +++ b/crates/ecstore/src/layout/disks_layout.rs @@ -21,7 +21,8 @@ use tracing::debug; /// Supported set sizes this is used to find the optimal /// single set size. -const SET_SIZES: [usize; 15] = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]; +pub(crate) const MAX_ERASURE_SET_DRIVE_COUNT: usize = 16; +const SET_SIZES: [usize; 15] = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, MAX_ERASURE_SET_DRIVE_COUNT]; const ENV_RUSTFS_ERASURE_SET_DRIVE_COUNT: &str = "RUSTFS_ERASURE_SET_DRIVE_COUNT"; #[derive(Deserialize, Debug, Default)] @@ -327,7 +328,7 @@ fn possible_set_counts(set_size: usize) -> Vec { /// checks whether given count is a valid set size for erasure coding. fn is_valid_set_size(count: usize) -> bool { - count >= SET_SIZES[0] && count <= SET_SIZES[SET_SIZES.len() - 1] + count >= SET_SIZES[0] && count <= MAX_ERASURE_SET_DRIVE_COUNT } /// Final set size with all the symmetry accounted for. diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 29ada8dfd..944c3858e 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -54,8 +54,9 @@ use crate::disk::{ use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::ShardReader; use crate::io_support::bitrot::{ - BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params, create_bitrot_reader_with_stage_metrics, - create_deferred_bitrot_reader_with_stripe_handle, object_mmap_read_enabled, object_mmap_read_max_length, + BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params, + create_bitrot_reader_from_bytes_with_stage_metrics, create_deferred_bitrot_reader_with_stripe_handle, + object_mmap_read_enabled, object_mmap_read_max_length, }; use crate::set_disk::shard_source::ShardReadCost; use futures::FutureExt as _; @@ -1262,13 +1263,13 @@ pub(in crate::set_disk) fn schedule_bitrot_reader_task<'a>( return; } - let inline_data = files[idx].data.as_deref(); + let inline_data = files[idx].data.clone(); let data_dir = files[idx].data_dir.unwrap_or_default(); let disk = disks[idx].as_ref(); let path = format!("{object}/{data_dir}/part.{part_number}"); reader_tasks.push(Box::pin(async move { - let result = create_bitrot_reader_with_stage_metrics( + let result = create_bitrot_reader_from_bytes_with_stage_metrics( inline_data, disk, bucket, @@ -1560,14 +1561,14 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards( let schedule_stage_start = stage_metrics.map(|_| Instant::now()); for (idx, disk_op) in disks.iter().enumerate() { setup.mark_scheduled(idx); - let inline_data = files[idx].data.as_deref(); + let inline_data = files[idx].data.clone(); let data_dir = files[idx].data_dir.unwrap_or_default(); let disk = disk_op.as_ref(); let path = format!("{object}/{data_dir}/part.{part_number}"); let checksum_algo = checksum_algo.clone(); reader_tasks.push(async move { - let result = create_bitrot_reader_with_stage_metrics( + let result = create_bitrot_reader_from_bytes_with_stage_metrics( inline_data, disk, bucket, diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 3b37b64a6..20e481c2d 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -247,8 +247,8 @@ impl SetDisks { no_lock: true, ..Default::default() }; - let (current, _, _) = self.get_object_fileinfo(bucket, object, &read_opts, true, false).await?; - restore_operation_id_from_metadata(¤t.metadata)? + let current = self.get_object_fileinfo(bucket, object, &read_opts, true, false).await?; + restore_operation_id_from_metadata(¤t.fi().metadata)? .filter(|actual| *actual == expected) .ok_or_else(|| Error::other(format!("restore operation id changed before {mode}: expected {expected}")))?; Ok(()) @@ -736,41 +736,91 @@ mod transition_matrix_tests; pub use ops::heal_walk::HealWalkVersion; -pub(in crate::set_disk) enum GetObjectMetadata { - Owned(T), - Shared(Arc), +pub(in crate::set_disk) struct GetObjectFileInfo { + owned: Option, + shared: Option>, } -impl std::ops::Deref for GetObjectMetadata { - type Target = T; +struct OwnedGetObjectFileInfo { + fi: FileInfo, + parts_metadata: Vec, + online_disks: Vec>, +} - fn deref(&self) -> &Self::Target { - match self { - Self::Owned(value) => value, - Self::Shared(value) => value, +impl GetObjectFileInfo { + fn owned(fi: FileInfo, parts_metadata: Vec, online_disks: Vec>) -> Self { + Self { + owned: Some(OwnedGetObjectFileInfo { + fi, + parts_metadata, + online_disks, + }), + shared: None, } } -} -impl GetObjectMetadata { - fn into_owned(self) -> T { - match self { - Self::Owned(value) => value, - Self::Shared(value) => Arc::try_unwrap(value).unwrap_or_else(|value| (*value).clone()), + fn shared(entry: Arc) -> Self { + Self { + owned: None, + shared: Some(entry), } } -} -type GetObjectFileInfo = ( - GetObjectMetadata, - GetObjectMetadata>, - GetObjectMetadata>>, -); + fn fi(&self) -> &FileInfo { + match (&self.owned, &self.shared) { + (Some(snapshot), None) => &snapshot.fi, + (None, Some(entry)) => &entry.fi, + _ => unreachable!("GET metadata snapshot representation must be exclusive"), + } + } + + fn parts_metadata(&self) -> &[FileInfo] { + match (&self.owned, &self.shared) { + (Some(snapshot), None) => &snapshot.parts_metadata, + (None, Some(entry)) => &entry.parts_metadata, + _ => unreachable!("GET metadata snapshot representation must be exclusive"), + } + } + + fn online_disks(&self) -> &[Option] { + match (&self.owned, &self.shared) { + (Some(snapshot), None) => &snapshot.online_disks, + (None, Some(entry)) => &entry.online_disks, + _ => unreachable!("GET metadata snapshot representation must be exclusive"), + } + } + + fn into_owned(self) -> (FileInfo, Vec, Vec>) { + match (self.owned, self.shared) { + (Some(snapshot), None) => { + let OwnedGetObjectFileInfo { + fi, + parts_metadata, + online_disks, + } = snapshot; + (fi, parts_metadata, online_disks) + } + (None, Some(entry)) => match Arc::try_unwrap(entry) { + Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks), + Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone()), + }, + _ => unreachable!("GET metadata snapshot representation must be exclusive"), + } + } + + #[cfg(test)] + fn has_valid_representation(&self) -> bool { + self.owned.is_some() ^ self.shared.is_some() + } + + #[cfg(test)] + fn shared_entry(&self) -> Option<&Arc> { + self.shared.as_ref() + } +} pub(crate) struct PreparedGetObjectMetadata { - fi: GetObjectMetadata, - files: GetObjectMetadata>, - disks: GetObjectMetadata>>, + snapshot: GetObjectFileInfo, object_info: Option, } @@ -788,7 +838,7 @@ impl PreparedGetObjectMetadata { } pub(crate) fn read_semantics_identity(&self) -> [u8; 32] { - SetDisks::file_info_quorum_hash(&self.fi) + SetDisks::file_info_quorum_hash(self.snapshot.fi()) } } @@ -838,10 +888,11 @@ mod prepared_get_object_metadata_tests { #[tokio::test] async fn prepared_metadata_is_consumed_exactly_once() { + let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new()); + assert!(snapshot.has_valid_representation()); + assert!(snapshot.shared_entry().is_none()); let metadata = PreparedGetObjectMetadata { - fi: GetObjectMetadata::Owned(FileInfo::default()), - files: GetObjectMetadata::Owned(Vec::new()), - disks: GetObjectMetadata::Owned(Vec::new()), + snapshot, object_info: None, }; @@ -853,6 +904,44 @@ mod prepared_get_object_metadata_tests { assert!(take_prepared_get_object_metadata().is_none()); } + #[test] + fn cache_hit_consumers_release_snapshot_at_legacy_ownership() { + let fi = FileInfo { + name: "object".to_owned(), + ..Default::default() + }; + let cached = Arc::new(GetObjectMetadataCacheEntry { + created_at: Instant::now(), + parts_metadata: vec![fi.clone()], + fi, + online_disks: vec![None], + read_quorum: 0, + }); + let snapshot = GetObjectFileInfo::shared(Arc::clone(&cached)); + + assert!(snapshot.has_valid_representation()); + assert!(std::mem::size_of::() >= std::mem::size_of::()); + assert!( + std::mem::size_of::() + <= std::mem::size_of::() + 2 * std::mem::size_of::() + ); + assert_eq!(Arc::strong_count(&cached), 2, "a cache hit must add one snapshot reference"); + assert_eq!(snapshot.fi().name, "object"); + assert_eq!(snapshot.parts_metadata().len(), 1); + assert_eq!(snapshot.online_disks().len(), 1); + assert_eq!(Arc::strong_count(&cached), 2, "borrowing consumers must not clone the snapshot"); + + let (owned_fi, owned_parts, disks) = snapshot.into_owned(); + assert_eq!(owned_fi.name, "object"); + assert_eq!(owned_parts.len(), 1); + assert_eq!(disks.len(), 1); + assert_eq!( + Arc::strong_count(&cached), + 1, + "legacy ownership must release the cache snapshot after cloning its owned inputs" + ); + } + #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn prepared_reader_reuses_metadata_fanout_exactly_once() { @@ -1016,12 +1105,10 @@ impl SetDisks { object: &str, opts: &ObjectOptions, ) -> Result { - let (fi, files, disks) = self.get_object_fileinfo(bucket, object, opts, true, true).await?; - let object_info = build_get_object_info(&fi, bucket, object, opts.versioned || opts.version_suspended); + let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?; + let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended); Ok(PreparedGetObjectMetadata { - fi, - files, - disks, + snapshot, object_info: Some(object_info), }) } @@ -1137,24 +1224,6 @@ pub fn is_deadlock_detection_enabled() -> bool { // require process restart to take effect. // ============================================================================ -/// Check if codec streaming is enabled (base flag). -/// -/// **Note**: Cached via `OnceLock` — env var changes require process restart. -/// In test mode, bypasses cache to allow per-test env var overrides. -fn is_get_codec_streaming_enabled() -> bool { - #[cfg(test)] - { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE) - } - #[cfg(not(test))] - { - static CACHED: OnceLock = OnceLock::new(); - *CACHED.get_or_init(|| { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE) - }) - } -} - /// Check if multipart codec streaming is enabled. /// /// When enabled, multipart objects use per-part codec streaming @@ -1309,22 +1378,6 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool { } } -// --- Rollout Percentage Functions --- - -fn get_codec_streaming_rollout_pct() -> u32 { - #[cfg(test)] - { - rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT) - } - #[cfg(not(test))] - { - static CACHED: OnceLock = OnceLock::new(); - *CACHED.get_or_init(|| { - rustfs_utils::get_env_u32(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT) - }) - } -} - fn get_metadata_early_stop_rollout_pct() -> u32 { static CACHED: OnceLock = OnceLock::new(); *CACHED.get_or_init(|| { @@ -1359,10 +1412,8 @@ fn is_optimization_enabled_for_request(base_enabled: bool, rollout_pct: u32, buc (hash as u32) < rollout_pct } /// Should this specific request use codec streaming? -pub fn should_use_codec_streaming(bucket: &str, object: &str) -> bool { - let base = is_get_codec_streaming_enabled(); - let pct = get_codec_streaming_rollout_pct(); - is_optimization_enabled_for_request(base, pct, bucket, object) +fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, object: &str) -> bool { + is_optimization_enabled_for_request(config.enabled, config.rollout_pct, bucket, object) } /// Should this specific request use metadata early-stop? @@ -1372,20 +1423,6 @@ pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool { is_optimization_enabled_for_request(base, pct, bucket, object) } -fn get_codec_streaming_min_size() -> usize { - if std::env::var_os(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE).is_some() { - return rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE); - } - - match get_codec_streaming_engine() { - GetCodecStreamingEngine::Rustfs => rustfs_utils::get_env_usize( - ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, - DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, - ), - GetCodecStreamingEngine::Legacy => DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, - } -} - fn is_get_codec_streaming_data_blocks_first_enabled() -> bool { #[cfg(test)] { @@ -1488,8 +1525,18 @@ enum GetCodecStreamingEngine { Rustfs, } -fn get_codec_streaming_engine() -> GetCodecStreamingEngine { - let engine = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE); +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct GetCodecStreamingConfig { + enabled: bool, + rollout: GetCodecStreamingRollout, + rollout_pct: u32, + body_compat_confirmed: bool, + header_compat_confirmed: bool, + engine: GetCodecStreamingEngine, + min_size: usize, +} + +fn parse_get_codec_streaming_engine(engine: &str) -> GetCodecStreamingEngine { match engine.trim() { value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_RUSTFS) => GetCodecStreamingEngine::Rustfs, value if value.eq_ignore_ascii_case(GET_CODEC_STREAMING_ENGINE_LEGACY) => GetCodecStreamingEngine::Legacy, @@ -1497,8 +1544,7 @@ fn get_codec_streaming_engine() -> GetCodecStreamingEngine { } } -fn get_codec_streaming_rollout() -> GetCodecStreamingRollout { - let rollout = rustfs_utils::get_env_str(ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT); +fn parse_get_codec_streaming_rollout(rollout: &str) -> GetCodecStreamingRollout { match rollout.trim() { // Clean production token. `internal`/`benchmark` remain accepted aliases // for backward compatibility; all three opt the fast path in. @@ -1515,20 +1561,58 @@ fn get_codec_streaming_rollout() -> GetCodecStreamingRollout { } } -/// Emergency kill-switch (defaults to `true`). Set -/// `RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED=false` to force the fast path -/// off. Body compatibility is confirmed by the parity e2e net + bench (backlog#1183), -/// so this no longer gates enablement — the `..._ROLLOUT` switch does. -fn is_get_codec_streaming_body_compat_confirmed() -> bool { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, true) +fn load_get_codec_streaming_config() -> GetCodecStreamingConfig { + let engine = parse_get_codec_streaming_engine(&rustfs_utils::get_env_str( + ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, + DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENGINE, + )); + let min_size = if std::env::var_os(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE).is_some() { + rustfs_utils::get_env_usize(ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE) + } else { + match engine { + GetCodecStreamingEngine::Rustfs => rustfs_utils::get_env_usize( + ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, + DEFAULT_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, + ), + GetCodecStreamingEngine::Legacy => DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, + } + }; + + GetCodecStreamingConfig { + enabled: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, DEFAULT_RUSTFS_GET_CODEC_STREAMING_ENABLE), + rollout: parse_get_codec_streaming_rollout(&rustfs_utils::get_env_str( + ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, + DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, + )), + rollout_pct: rustfs_utils::get_env_u32( + ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, + DEFAULT_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, + ), + body_compat_confirmed: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, true), + header_compat_confirmed: rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, true), + engine, + min_size, + } } -/// Emergency kill-switch (defaults to `true`). Set -/// `RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED=false` to force the fast path -/// off. Header compatibility is confirmed by the parity e2e net + bench (backlog#1183), -/// so this no longer gates enablement — the `..._ROLLOUT` switch does. -fn is_get_codec_streaming_header_compat_confirmed() -> bool { - rustfs_utils::get_env_bool(ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, true) +fn get_codec_streaming_config_cached_core(load: impl FnOnce() -> GetCodecStreamingConfig) -> GetCodecStreamingConfig { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(load) +} + +fn get_codec_streaming_config() -> GetCodecStreamingConfig { + #[cfg(test)] + { + load_get_codec_streaming_config() + } + #[cfg(not(test))] + { + get_codec_streaming_config_cached_core(load_get_codec_streaming_config) + } +} + +fn get_codec_streaming_engine() -> GetCodecStreamingEngine { + get_codec_streaming_config().engine } fn build_get_codec_streaming_decode_engine(erasure: coding::Erasure) -> std::io::Result { @@ -1897,43 +1981,43 @@ fn should_prefer_codec_streaming_data_blocks_first_reader_setup( fn get_codec_streaming_reader_gate( bucket: &str, object: &str, - range: &Option, part_number: Option, + object_class: GetCodecStreamingObjectClass, object_info: &ObjectInfo, fi: &FileInfo, lock_optimization_enabled: bool, ) -> GetCodecStreamingGate { - let object_class = classify_get_codec_streaming_object_class(range, object_info, fi); + let config = get_codec_streaming_config(); - if !is_get_codec_streaming_enabled() { + if !config.enabled { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Disabled), prefer_data_blocks_first_reader_setup: false, }; } - if !get_codec_streaming_rollout().is_opted_in() { + if !config.rollout.is_opted_in() { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutNotOptedIn), prefer_data_blocks_first_reader_setup: false, }; } - if !should_use_codec_streaming(bucket, object) { + if !should_use_codec_streaming(config, bucket, object) { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::RolloutPctNotSelected), prefer_data_blocks_first_reader_setup: false, }; } - if !is_get_codec_streaming_body_compat_confirmed() { + if !config.body_compat_confirmed { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BodyCompatibilityUnconfirmed), prefer_data_blocks_first_reader_setup: false, }; } - if !is_get_codec_streaming_header_compat_confirmed() { + if !config.header_compat_confirmed { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::HeaderCompatibilityUnconfirmed), @@ -2006,7 +2090,7 @@ fn get_codec_streaming_reader_gate( }; } } - let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else { + let Ok(min_size) = i64::try_from(config.min_size) else { return GetCodecStreamingGate { object_class, decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize), @@ -2376,6 +2460,7 @@ pub struct SetDisks { get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState, get_object_metadata_cache_generations: Arc<[AtomicU64]>, pub lockers: Vec>, + shared_lockers: Arc<[Arc]>, local_lock_manager: Arc, /// Per-instance runtime context (Phase 5, backlog#939). /// @@ -2503,9 +2588,9 @@ impl Hash for GetObjectMetadataCacheKey { struct GetObjectMetadataCacheEntry { #[allow(dead_code)] // Kept for debugging; moka handles TTL internally created_at: Instant, - fi: Arc, - parts_metadata: Arc>, - online_disks: Arc>>, + fi: FileInfo, + parts_metadata: Vec, + online_disks: Vec>, read_quorum: usize, } @@ -2772,6 +2857,7 @@ impl SetDisks { ) -> Arc { let ctx = instance_ctx; let set_lock_namespace: Arc = format!("set-{pool_index}-{set_index}").into(); + let shared_lockers = Arc::from(lockers.to_vec()); Arc::new(SetDisks { locker_owner, disks, @@ -2794,6 +2880,7 @@ impl SetDisks { .collect::>(), ), lockers, + shared_lockers, // Sourced from the instance context so each instance owns its lock // namespace (Phase 5 Slice 3). Single-instance: ctx aliases the // process lock-manager singleton, so this is unchanged. @@ -4999,6 +5086,89 @@ mod tests { assert_eq!(Arc::strong_count(&set.set_lock_namespace), before); } + #[tokio::test] + async fn new_ns_lock_shares_clients_without_changing_quorum() { + let healthy: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let failing: Arc = Arc::new(FailingClient); + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::DistErasure).await; + let set = make_test_set_disks_with_ctx(vec![healthy.clone(), failing.clone()], ctx).await; + + assert!(Arc::ptr_eq(&set.lockers[0], &healthy)); + assert!(Arc::ptr_eq(&set.lockers[1], &failing)); + let clients_before = Arc::strong_count(&set.shared_lockers); + let healthy_before = Arc::strong_count(&healthy); + let failing_before = Arc::strong_count(&failing); + let write_lock = set + .new_ns_lock("bucket", "write-object") + .await + .expect("namespace lock should be created"); + + assert_eq!( + Arc::strong_count(&set.shared_lockers), + clients_before + 1, + "each object lock should share one client slice allocation" + ); + assert_eq!( + Arc::strong_count(&healthy), + healthy_before, + "constructing an object lock must not clone each client Arc" + ); + assert_eq!( + Arc::strong_count(&failing), + failing_before, + "constructing an object lock must not clone each client Arc" + ); + + let write_error = write_lock + .get_write_lock(Duration::from_millis(500)) + .await + .expect_err("one healthy client must not satisfy the two-client write quorum"); + assert!( + matches!( + write_error, + LockError::QuorumNotReached { + required: 2, + achieved: 1 + } + ), + "the shared client representation must preserve the exact write quorum result: {write_error}" + ); + let read_lock = set + .new_ns_lock("bucket", "read-object") + .await + .expect("second namespace lock should be created"); + assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2); + let read_guard = read_lock + .get_read_lock(Duration::from_millis(500)) + .await + .expect("one healthy client should satisfy the two-client read quorum"); + assert!(matches!(read_guard, NamespaceLockGuard::Standard(_))); + } + + #[tokio::test] + async fn new_ns_lock_uses_the_current_public_client_domain() { + let stale_a: Arc = Arc::new(FailingClient); + let stale_b: Arc = Arc::new(FailingClient); + let healthy_a: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let healthy_b: Arc = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new()))); + let ctx = Arc::new(InstanceContext::new()); + ctx.update_erasure_type(SetupType::DistErasure).await; + let set = make_test_set_disks_with_ctx(vec![stale_a, stale_b], ctx).await; + let mut set = (*set).clone(); + set.lockers = vec![healthy_a, healthy_b]; + + let lock = set + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should use the current public clients"); + let guard = lock + .get_write_lock(Duration::from_millis(500)) + .await + .expect("the current healthy clients should satisfy the two-client quorum"); + assert!(matches!(guard, NamespaceLockGuard::Standard(_))); + } + struct SetupTypeGuard { previous: SetupType, } diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 718050533..e5ca4fbec 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -3171,10 +3171,11 @@ mod heal_result_report_tests { .await .expect("object should be written"); - let (fi, _, _) = set + let snapshot = set .get_object_fileinfo(bucket, object, &opts, true, false) .await .expect("object metadata should resolve"); + let fi = snapshot.fi(); assert_eq!(fi.erasure.parity_blocks, 0); let data_dir = fi.data_dir.expect("non-inline object should have a data directory"); let part_path = dir.path().join(bucket).join(object).join(data_dir.to_string()).join("part.1"); diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index e90772b7a..d975d69cd 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -36,10 +36,21 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { // test's transient DistErasure window) would push this set's namespace // locking onto its own — possibly empty — dist locker list. let set_lock = if self.ctx.is_dist_erasure().await { - // Calculate quorum based on lockers count (majority) - let lockers_count = self.lockers.len(); + let lockers = if self.lockers.len() == self.shared_lockers.len() + && self + .lockers + .iter() + .zip(self.shared_lockers.iter()) + .all(|(current, shared)| Arc::ptr_eq(current, shared)) + { + self.shared_lockers.clone() + } else { + Arc::from(self.lockers.clone()) + }; + // Calculate quorum from the exact client domain used by this lock. + let lockers_count = lockers.len(); let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 }; - NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), self.lockers.clone(), write_quorum) + NamespaceLock::with_clients_and_quorum_shared(self.set_lock_namespace.clone(), lockers, write_quorum) } else { NamespaceLock::with_local_manager_shared(self.set_lock_namespace.clone(), self.local_lock_manager.clone()) }; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index a38ec657b..10dda4d81 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -379,7 +379,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { type GetObjectReader = GetObjectReader; type PutObjectReader = PutObjReader; - #[tracing::instrument(level = "debug", skip(self))] + #[tracing::instrument(level = "debug", skip(self, h))] async fn get_object_reader( &self, bucket: &str, @@ -428,11 +428,11 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { }; let metadata_stage_start = Instant::now(); - let (fi, files, disks, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() { - (prepared.fi, prepared.files, prepared.disks, prepared.object_info) + let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() { + (prepared.snapshot, prepared.object_info) } else { match self.get_object_fileinfo(bucket, object, opts, true, true).await { - Ok((fi, files, disks)) => (fi, files, disks, None), + Ok(snapshot) => (snapshot, None), Err(err) => { rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64()); let failure_path = if is_meta_bucketname(bucket) { @@ -445,10 +445,13 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } }; + let fi = snapshot.fi(); + let files = snapshot.parts_metadata(); + let disks = snapshot.online_disks(); let object_info_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let object_info = prepared_object_info - .unwrap_or_else(|| build_get_object_info(&fi, bucket, object, opts.versioned || opts.version_suspended)); - let object_class = classify_get_codec_streaming_object_class(&range, &object_info, &fi); + .unwrap_or_else(|| build_get_object_info(fi, bucket, object, opts.versioned || opts.version_suspended)); + let object_class = classify_get_codec_streaming_object_class(&range, &object_info, fi); let size_bucket = rustfs_io_metrics::get_object_size_bucket(object_info.size); record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_OBJECT_INFO, object_info_stage_start); let metadata_elapsed = metadata_stage_start.elapsed().as_secs_f64(); @@ -496,7 +499,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // Uses the shared predicate from ObjectInfo; additionally checks that // inline data is actually present and neither range nor partNumber is // in flight. - if should_use_inline_fast_path(&range, &object_info, &fi, opts) { + if should_use_inline_fast_path(&range, &object_info, fi, opts) { let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let data_shards = fi.erasure.data_blocks; @@ -512,7 +515,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { }; if can_try_inline_data_shards_direct(object_size, fi.erasure.block_size) - && let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(&files, &fi, data_shards, |index| { + && let Some(data_files) = collect_inline_data_shard_fileinfos_by_index(files, fi, data_shards, |index| { disks.get(index).is_some_and(Option::is_some) }) { @@ -580,10 +583,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let erasure = erasure_from_file_info(&fi, fi.uses_legacy_checksum)?; + let erasure = erasure_from_file_info(fi, fi.uses_legacy_checksum)?; 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); + let (_disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi); // Check if we have enough inline data shards let inline_count = files @@ -662,10 +665,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let codec_streaming_gate = get_codec_streaming_reader_gate( bucket, object, - &range, opts.part_number, + object_class, &object_info, - &fi, + fi, lock_optimization_enabled, ); record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start); @@ -743,15 +746,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } - let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, &fi, opts); + let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, fi, opts); record_get_direct_memory_decision(object_class, direct_memory_decision, size_bucket); if let GetDirectMemoryDecision::Use { object_size } = direct_memory_decision { if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo( bucket, object, - &fi, - &files, - &disks, + fi, + files, + disks, opts.skip_verify_bitrot, object_class.as_str(), size_bucket, @@ -780,14 +783,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } let mut output = Vec::with_capacity(object_size); + let (fi, files, disks) = snapshot.into_owned(); Self::get_object_with_fileinfo( bucket, object, 0, object_info.size, &mut output, - fi.into_owned(), - files.into_owned(), + fi, + files, &disks, self.set_index, self.pool_index, @@ -826,9 +830,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { match Self::get_object_decode_reader_with_fileinfo( bucket, object, - &fi, - &files, - &disks, + fi, + files, + disks, self.set_index, self.pool_index, opts.skip_verify_bitrot, @@ -890,6 +894,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 (fi, files, disks) = snapshot.into_owned(); tokio::spawn(async move { let _guard = read_lock_guard; let mut writer = GetObjectDownstreamWriter::new(wd); @@ -903,8 +908,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { offset, length, &mut writer, - fi.into_owned(), - files.into_owned(), + fi, + files, &disks, set_index, pool_index, @@ -3294,10 +3299,10 @@ impl SetDisks { // quorum, failing write quorum on update_object_meta (backlog#872). let mut read_opts = opts.clone(); read_opts.include_part_checksums = true; - let (fi, _, disks) = self + let (mut fi, _, disks) = self .get_object_fileinfo_gated(bucket, object, &read_opts, false, false) - .await?; - let mut fi = fi.into_owned(); + .await? + .into_owned(); fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned()); if let Some(eval_metadata) = &opts.eval_metadata { @@ -4575,12 +4580,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // Use the same full xl.meta read path as GetObject metadata resolution. // This avoids HEAD/GetObject metadata visibility skew immediately after // PutObject/CompleteMultipartUpload. - let (fi, _, _) = self + let snapshot = self .get_object_fileinfo(bucket, object, opts, true, false) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; - let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); + let oi = ObjectInfo::from_file_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended); Ok(oi) } @@ -4748,10 +4753,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let mut transition_read_opts = opts.clone(); transition_read_opts.include_part_checksums = true; - let (fi, meta_arr, online_disks) = self + let (mut fi, meta_arr, online_disks) = self .get_object_fileinfo(bucket, object, &transition_read_opts, true, false) - .await?; - let mut fi = fi.into_owned(); + .await? + .into_owned(); /*if err != nil { return Err(to_object_err(err, vec![bucket, object])); }*/ @@ -4866,7 +4871,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { cloned_fi.size, &mut writer, cloned_fi, - meta_arr.into_owned(), + meta_arr, &online_disks, set_index, pool_index, @@ -4991,7 +4996,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { }; self.invalidate_get_object_metadata_cache(bucket, object).await; let current = self.get_object_fileinfo(bucket, object, &commit_opts, true, false).await; - let (current_fi, _, _) = match current { + let current = match current { Ok(current) => current, Err(err) => { drop(transition_lock_guard); @@ -5002,7 +5007,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { return Err(err); } }; - let mut current_fi = current_fi.into_owned(); + let (mut current_fi, _, _) = current.into_owned(); let source_matches = current_fi.version_id == fi.version_id && current_fi.data_dir == fi.data_dir && current_fi.mod_time == fi.mod_time @@ -5233,9 +5238,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if let Err(err) = fi { return set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await; } - let (actual_fi, _, _) = fi?; + let actual = fi?; + let actual_fi = actual.fi(); - oi = ObjectInfo::from_file_info(&actual_fi, bucket, object, opts.versioned || opts.version_suspended); + oi = ObjectInfo::from_file_info(actual_fi, bucket, object, opts.versioned || opts.version_suspended); let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?; if let Some(expected_operation_id) = expected_operation_id { require_restore_operation_id(oi.user_defined.as_ref(), expected_operation_id)?; @@ -5513,8 +5519,40 @@ mod object_encryption_resolver_wiring_tests { use super::*; use crate::object_api::{EncryptionResolutionError, ObjectEncryptionResolver, ReadEncryptionMaterial, ReadEncryptionRequest}; use std::io::Cursor; + use std::sync::Mutex; use std::sync::atomic::{AtomicUsize, Ordering}; + #[derive(Clone, Default)] + struct CapturedLogs(Arc>>); + + struct CapturedLogWriter(Arc>>); + + impl CapturedLogs { + fn contents(&self) -> String { + String::from_utf8(self.0.lock().expect("captured logs mutex should not poison").clone()) + .expect("captured logs should be valid UTF-8") + } + } + + impl std::io::Write for CapturedLogWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let mut captured = self.0.lock().expect("captured logs mutex should not poison"); + std::io::Write::write(&mut *captured, buf) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLogs { + type Writer = CapturedLogWriter; + + fn make_writer(&'writer self) -> Self::Writer { + CapturedLogWriter(Arc::clone(&self.0)) + } + } + struct CountingResolver { calls: AtomicUsize, } @@ -5561,6 +5599,34 @@ mod object_encryption_resolver_wiring_tests { assert!(result.is_err(), "resolver returning no material must fail closed"); assert_eq!(resolver.calls.load(Ordering::Relaxed), 1); } + + #[tokio::test(flavor = "current_thread")] + async fn get_object_reader_span_never_records_transport_headers() { + use super::hermetic_set_disks_support::hermetic_set_disks_isolated; + use crate::storage_api_contracts::object::ObjectIO as _; + use rustfs_utils::http::headers::SSEC_KEY_HEADER; + + let logs = CapturedLogs::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::DEBUG) + .with_writer(logs.clone()) + .with_ansi(false) + .without_time() + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + let (_temp_dirs, _disks, set_disks) = hermetic_set_disks_isolated(4).await; + let mut headers = HeaderMap::new(); + headers.insert(http::header::AUTHORIZATION, HeaderValue::from_static("credential-must-not-be-logged")); + headers.insert(SSEC_KEY_HEADER, HeaderValue::from_static("customer-key-must-not-be-logged")); + + let _ = set_disks + .get_object_reader("missing-bucket", "missing-object", None, headers, &ObjectOptions::default()) + .await; + + let captured = logs.contents(); + assert!(!captured.contains("credential-must-not-be-logged")); + assert!(!captured.contains("customer-key-must-not-be-logged")); + } } #[cfg(test)] @@ -6200,9 +6266,9 @@ mod metadata_mutation_generation_tests { false, ) .await - .expect("object metadata should be readable before adding the checksum sidecar"); - let mut fi = fi.into_owned(); - let disks = disks.into_owned(); + .expect("object metadata should be readable before adding the checksum sidecar") + .into_owned(); + let mut fi = fi; rustfs_utils::http::insert_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, value.to_string()); set_disks .update_object_meta(bucket, object, fi, &disks) @@ -6359,9 +6425,9 @@ mod metadata_mutation_generation_tests { false, ) .await - .expect("conflicting object metadata should be readable before corruption is injected"); - let mut fi = fi.into_owned(); - let disks = disks.into_owned(); + .expect("conflicting object metadata should be readable before corruption is injected") + .into_owned(); + let mut fi = fi; let rustfs_key = format!( "{}{}", rustfs_utils::http::RUSTFS_INTERNAL_PREFIX, @@ -6524,9 +6590,9 @@ mod transition_commit_failure_tests { false, ) .await - .expect("source metadata should be readable before adding the checksum sidecar"); - let mut source_fi = source_fi.into_owned(); - let online_disks = online_disks.into_owned(); + .expect("source metadata should be readable before adding the checksum sidecar") + .into_owned(); + let mut source_fi = source_fi; rustfs_utils::http::insert_str( &mut source_fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, @@ -6800,10 +6866,11 @@ mod transition_commit_failure_tests { .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); - let (fi, parts_metadata, online_disks) = set_disks + let snapshot = set_disks .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) .await .expect("source metadata should resolve"); + let (fi, parts_metadata, online_disks) = snapshot.into_owned(); let generation = set_disks .get_object_metadata_cache_generation(bucket, object) .expect("metadata cache generation should be active"); @@ -6814,9 +6881,9 @@ mod transition_commit_failure_tests { cache_key.clone(), Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), - fi: Arc::new((*fi).clone()), - parts_metadata: Arc::new(parts_metadata.into_owned()), - online_disks: Arc::new(online_disks.into_owned()), + fi, + parts_metadata, + online_disks, read_quorum: 2, }), ) @@ -6943,7 +7010,7 @@ mod transition_commit_failure_tests { let result = transition.await.expect("transition task should not panic"); result.expect_err("partial local commit must fail when only two of four disks are writable"); - let fi = set_disks + let snapshot = set_disks .get_object_fileinfo( bucket, object, @@ -6955,10 +7022,10 @@ mod transition_commit_failure_tests { false, ) .await - .expect("rollback should keep the source metadata readable on applied disks") - .0; + .expect("rollback should keep the source metadata readable on applied disks"); assert_ne!( - fi.transition_status, TRANSITION_COMPLETE, + snapshot.fi().transition_status, + TRANSITION_COMPLETE, "rollback must not leave the applied disks marked as transitioned" ); let mut restored = Vec::new(); @@ -8093,7 +8160,7 @@ mod transition_upload_integrity_tests { .await .expect("local source should drain after failed transition"); assert_eq!(restored, payload); - let (fi, _, _) = set_disks + let snapshot = set_disks .get_object_fileinfo( bucket, object, @@ -8107,7 +8174,7 @@ mod transition_upload_integrity_tests { ) .await .expect("local source metadata should remain available"); - assert_ne!(fi.transition_status, TRANSITION_COMPLETE); + assert_ne!(snapshot.fi().transition_status, TRANSITION_COMPLETE); } async fn write_source( @@ -8204,7 +8271,8 @@ mod transition_upload_integrity_tests { false, ) .await - .expect("the existing target should remain readable"); + .expect("the existing target should remain readable") + .into_owned(); assert_ne!(stored.transition_status, TRANSITION_COMPLETE); assert!(stored.transition_version.is_none()); } @@ -8276,9 +8344,9 @@ mod transition_upload_integrity_tests { false, ) .await - .expect("source metadata should be readable"); - let mut source_fi = source_fi.into_owned(); - let online_disks = online_disks.into_owned(); + .expect("source metadata should be readable") + .into_owned(); + let mut source_fi = source_fi; rustfs_utils::http::insert_str( &mut source_fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, @@ -8305,7 +8373,7 @@ mod transition_upload_integrity_tests { remote_object.starts_with(crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_PREFIX), "remote object should be transaction-scoped: {remote_object}" ); - let (fi, _, _) = set_disks + let snapshot = set_disks .get_object_fileinfo( bucket, object, @@ -8320,6 +8388,7 @@ mod transition_upload_integrity_tests { ) .await .expect("committed transition metadata should be readable"); + let fi = snapshot.fi(); assert_eq!(fi.transition_status, TRANSITION_COMPLETE); assert_eq!(fi.transitioned_objname, *remote_object); assert_eq!( @@ -8754,7 +8823,7 @@ mod transition_upload_integrity_tests { let object = format!("{}-corrupt.bin", position.label()); let payload = vec![0x41; 2 * 1024 * 1024]; let original = write_source(&set_disks, &disk_stores, &bucket, &object, &payload).await; - let (source, _, _) = set_disks + let source = set_disks .get_object_fileinfo( &bucket, &object, @@ -8768,6 +8837,7 @@ mod transition_upload_integrity_tests { ) .await .expect("source metadata should be available before shard corruption"); + let source = source.fi(); let data_dir = source.data_dir.expect("source object should have a data directory"); corrupt_beyond_read_quorum(&temp_dirs, &bucket, &object, data_dir, source.erasure.parity_blocks, position).await; @@ -8787,7 +8857,7 @@ mod transition_upload_integrity_tests { ), "{position:?}: unexpected transition producer error: {error:?}" ); - let (after, _, _) = set_disks + let after = set_disks .get_object_fileinfo( &bucket, &object, @@ -8801,6 +8871,7 @@ mod transition_upload_integrity_tests { ) .await .expect("failed transition must leave metadata readable"); + let after = after.fi(); assert_eq!( after.data_dir, Some(data_dir), @@ -8847,7 +8918,7 @@ mod transition_upload_integrity_tests { .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect("an unversioned remote version must commit"); - let (fi, _, _) = set_disks + let snapshot = set_disks .get_object_fileinfo( bucket, object, @@ -8861,6 +8932,7 @@ mod transition_upload_integrity_tests { ) .await .expect("committed unversioned transition metadata should be readable"); + let fi = snapshot.fi(); assert_eq!(fi.transition_version_id, None); assert_eq!(fi.transition_version, None); assert_eq!(fi.transition_version_state, rustfs_filemeta::TransitionVersionState::KnownDisabled); @@ -9068,7 +9140,7 @@ mod transition_upload_integrity_tests { matches!(error, StorageError::NamespaceLockQuorumUnavailable { .. }), "unexpected tagging lock-lost error: {error:?}" ); - let (fi, _, _) = set_disks + let snapshot = set_disks .get_object_fileinfo( bucket, object, @@ -9083,7 +9155,7 @@ mod transition_upload_integrity_tests { .await .expect("source metadata should remain readable"); assert!( - !fi.metadata.contains_key(AMZ_OBJECT_TAGGING), + !snapshot.fi().metadata.contains_key(AMZ_OBJECT_TAGGING), "a stale tagging writer must not write metadata after refresh-quorum loss" ); } @@ -9279,13 +9351,14 @@ mod transition_source_identity_matrix_tests { .put_object(bucket, &object, &mut reader, &source_opts) .await .expect("source object should be written"); - let (source, _, _) = set_disks + let source = set_disks .get_object_fileinfo(bucket, &object, &source_opts, true, false) .await .expect("source metadata should resolve"); + let source = source.fi(); assert_eq!(source.version_id, Some(source_version_id)); assert_eq!( - transition_source_identity(bucket, &object, &source, &source_opts, &get_raw_etag(&source.metadata)) + transition_source_identity(bucket, &object, source, &source_opts, &get_raw_etag(&source.metadata)) .expect("persisted versioned source identity should build") .version_mode, TransitionSourceVersionMode::Versioned @@ -9333,10 +9406,11 @@ mod transition_source_identity_matrix_tests { versioned: true, ..Default::default() }; - let (persisted, _, _) = set_disks + let persisted = set_disks .get_object_fileinfo(bucket, &object, &persisted_opts, true, false) .await .expect("drifted source metadata should resolve"); + let persisted = persisted.fi(); put_barrier.release(); let result = transition.await.expect("transition task should not panic"); @@ -10356,9 +10430,9 @@ mod put_object_tags_early_stop_regression_tests { false, ) .await - .expect("object metadata should be readable before adding the checksum sidecar"); - let mut fi = fi.into_owned(); - let disks = disks.into_owned(); + .expect("object metadata should be readable before adding the checksum sidecar") + .into_owned(); + let mut fi = fi; rustfs_utils::http::insert_str( &mut fi.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 119d6599d..3fd19b151 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -180,9 +180,9 @@ impl SetDisks { let key = GetObjectMetadataCacheKey::new(bucket, object, generation); let entry = Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), - fi: Arc::new(fi.clone()), - parts_metadata: Arc::new(parts_metadata.to_vec()), - online_disks: Arc::new(online_disks.to_vec()), + fi: fi.clone(), + parts_metadata: parts_metadata.to_vec(), + online_disks: online_disks.to_vec(), read_quorum, }); self.insert_get_object_metadata_cache_entry_after_insert(key, generation, entry, || {}) @@ -300,11 +300,7 @@ impl SetDisks { GET_STAGE_METADATA_CACHE_LOOKUP, metadata_cache_lookup_start, ); - return Ok(( - GetObjectMetadata::Shared(Arc::clone(&cached.fi)), - GetObjectMetadata::Shared(Arc::clone(&cached.parts_metadata)), - GetObjectMetadata::Shared(Arc::clone(&cached.online_disks)), - )); + return Ok(GetObjectFileInfo::shared(cached)); } MetadataCacheLookup::Miss => { rustfs_io_metrics::record_get_object_metadata_cache_decision( @@ -436,11 +432,7 @@ impl SetDisks { // let online_disks: Vec> = op_online_disks.iter().filter(|v| v.is_some()).cloned().collect(); - Ok(( - GetObjectMetadata::Owned(fi), - GetObjectMetadata::Owned(parts_metadata), - GetObjectMetadata::Owned(op_online_disks), - )) + Ok(GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks)) } #[hotpath::measure(impl_type = "SetDisks")] @@ -450,14 +442,15 @@ impl SetDisks { object: &str, opts: &ObjectOptions, ) -> (ObjectInfo, usize, Option) { - let fi = match self.get_object_fileinfo(bucket, object, opts, false, false).await { - Ok((fi, _, _)) => fi, + let snapshot = match self.get_object_fileinfo(bucket, object, opts, false, false).await { + Ok(snapshot) => snapshot, Err(e) => return (ObjectInfo::default(), 0, Some(e)), }; + let fi = snapshot.fi(); let write_quorum = fi.write_quorum(self.default_write_quorum()); - let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); + let oi = ObjectInfo::from_file_info(fi, bucket, object, opts.versioned || opts.version_suspended); if !fi.version_purge_status().is_empty() && opts.version_id.is_some() { return ( @@ -2723,22 +2716,14 @@ mod metadata_cache_tests { .await .expect("fresh cache entry should be returned"); - let (returned_fi, returned_parts_metadata, returned_online_disks) = set + let returned = set .get_object_fileinfo("bucket", "object", &ObjectOptions::default(), true, false) .await .expect("cache-backed metadata lookup should succeed"); assert!( - matches!(returned_fi, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.fi)), - "cache hits must share FileInfo ownership" - ); - assert!( - matches!(returned_parts_metadata, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.parts_metadata)), - "cache hits must share the metadata vector" - ); - assert!( - matches!(returned_online_disks, GetObjectMetadata::Shared(ref value) if Arc::ptr_eq(value, &cached.online_disks)), - "cache hits must share the online-disk vector" + returned.shared_entry().is_some_and(|value| Arc::ptr_eq(value, &cached)), + "cache hits must share the complete metadata snapshot" ); } @@ -2781,9 +2766,9 @@ mod metadata_cache_tests { ), Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), - fi: Arc::new(fi.clone()), - parts_metadata: Arc::new(vec![fi]), - online_disks: Arc::new(vec![None]), + fi: fi.clone(), + parts_metadata: vec![fi], + online_disks: vec![None], read_quorum: 1, }), ) @@ -2877,13 +2862,12 @@ mod metadata_cache_tests { barrier.wait_until_paused().await; set.invalidate_get_object_metadata_cache(bucket, object).await; barrier.release(); - let (fi, parts_metadata, online_disks) = read + let snapshot = read .await .expect("metadata read task should not panic") .expect("metadata fanout should still return its selected FileInfo"); - assert!(matches!(fi, GetObjectMetadata::Owned(_))); - assert!(matches!(parts_metadata, GetObjectMetadata::Owned(_))); - assert!(matches!(online_disks, GetObjectMetadata::Owned(_))); + assert!(snapshot.owned.is_some()); + assert!(snapshot.has_valid_representation()); assert!( set.get_object_metadata_cache @@ -2930,9 +2914,9 @@ mod metadata_cache_tests { let key = GetObjectMetadataCacheKey::new("bucket", "object", generation); let entry = Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), - fi: Arc::new(fi.clone()), - parts_metadata: Arc::new(vec![fi]), - online_disks: Arc::new(Vec::new()), + fi: fi.clone(), + parts_metadata: vec![fi], + online_disks: Vec::new(), read_quorum: 0, }); @@ -3040,9 +3024,9 @@ mod metadata_cache_tests { let entry = |fi: FileInfo| { Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), - parts_metadata: Arc::new(vec![fi.clone()]), - fi: Arc::new(fi), - online_disks: Arc::new(Vec::new()), + parts_metadata: vec![fi.clone()], + fi, + online_disks: Vec::new(), read_quorum: 0, }) }; @@ -4105,8 +4089,8 @@ mod tests { get_codec_streaming_reader_gate( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, - range, None, + classify_get_codec_streaming_object_class(range, object_info, fi), object_info, fi, lock_optimization_enabled, @@ -4123,8 +4107,8 @@ mod tests { get_codec_streaming_reader_gate( CODEC_STREAMING_TEST_BUCKET, CODEC_STREAMING_TEST_OBJECT, - range, part_number, + classify_get_codec_streaming_object_class(range, object_info, fi), object_info, fi, lock_optimization_enabled, @@ -4850,6 +4834,114 @@ mod tests { .await } + async fn encoded_inline_blocks(blocks: &[&[u8]], shard_size: usize, hash_algo: HashAlgorithm) -> Bytes { + let mut writer = BitrotWriter::new(Cursor::new(Vec::new()), shard_size, hash_algo); + for block in blocks { + writer.write(block).await.expect("test block should be encoded"); + } + Bytes::from(writer.into_inner().into_inner()) + } + + fn assert_reader_shares_inline_allocation(reader: &ObjectBitrotReader, source: &Bytes) { + let reader_bytes = reader + .inner_ref() + .inline_bytes() + .expect("inline scheduler should retain an in-memory Bytes source"); + assert_eq!( + reader_bytes.as_ptr(), + source.as_ptr(), + "the scheduler must clone Bytes ownership instead of copying the inline shard payload" + ); + } + + #[tokio::test] + async fn inline_range_scheduler_shares_bytes_and_rejects_bitrot_mismatch() { + const SHARD_SIZE: usize = 16; + let hash_algo = HashAlgorithm::HighwayHash256S; + let first = [b'a'; SHARD_SIZE]; + let second = [b'b'; SHARD_SIZE]; + let mut source = encoded_inline_blocks(&[&first, &second], SHARD_SIZE, hash_algo.clone()).await; + let second_payload = hash_algo.size() * 2 + SHARD_SIZE; + source = { + let mut corrupt = source.to_vec(); + corrupt[second_payload] ^= 0xff; + Bytes::from(corrupt) + }; + let files = vec![encoded_reader_setup_fileinfo(Some(source.to_vec()))]; + let source = files[0].data.clone().expect("inline shard should exist"); + let disks = vec![None]; + + let mut setup = create_bitrot_readers_until_quorum_with_preference( + &files, + &disks, + "bucket", + "object", + 1, + SHARD_SIZE, + SHARD_SIZE, + SHARD_SIZE, + hash_algo, + false, + false, + 1, + 0, + BitrotReaderSetupMode::ReadQuorum, + true, + None, + None, + ) + .await; + let mut reader = setup.readers[0].take().expect("range reader should be ready"); + assert_reader_shares_inline_allocation(&reader, &source); + + let err = reader + .read(&mut [0; SHARD_SIZE]) + .await + .expect_err("corrupt ranged inline block must fail bitrot verification"); + assert_eq!(err.kind(), ErrorKind::InvalidData); + } + + #[tokio::test] + async fn inline_part_scheduler_shares_bytes_and_rejects_bitrot_mismatch() { + const SHARD_SIZE: usize = 16; + let hash_algo = HashAlgorithm::HighwayHash256S; + let block = [b'p'; SHARD_SIZE]; + let encoded = encoded_inline_blocks(&[&block], SHARD_SIZE, hash_algo.clone()).await; + let mut corrupt = encoded.to_vec(); + corrupt[hash_algo.size()] ^= 0xff; + let files = vec![encoded_reader_setup_fileinfo(Some(corrupt))]; + let source = files[0].data.clone().expect("inline shard should exist"); + let disks = vec![None]; + + let mut setup = create_bitrot_readers_until_quorum_all_shards( + &files, + &disks, + "bucket", + "object", + 7, + 0, + SHARD_SIZE, + SHARD_SIZE, + hash_algo, + false, + false, + 1, + 0, + BitrotReaderSetupMode::VerifyReconstruction, + None, + None, + ) + .await; + let mut reader = setup.readers[0].take().expect("part reader should be ready"); + assert_reader_shares_inline_allocation(&reader, &source); + + let err = reader + .read(&mut [0; SHARD_SIZE]) + .await + .expect_err("corrupt inline part must fail bitrot verification"); + assert_eq!(err.kind(), ErrorKind::InvalidData); + } + async fn decode_codec_data_blocks_first_setup( erasure: coding::Erasure, data: &[u8], @@ -5550,6 +5642,63 @@ mod tests { }); } + #[test] + fn codec_streaming_config_cache_loads_once() { + use std::cell::Cell; + + let loads = Cell::new(0); + let expected = GetCodecStreamingConfig { + enabled: true, + rollout: GetCodecStreamingRollout::Off, + rollout_pct: 100, + body_compat_confirmed: true, + header_compat_confirmed: true, + engine: GetCodecStreamingEngine::Legacy, + min_size: DEFAULT_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, + }; + + for _ in 0..3 { + assert_eq!( + get_codec_streaming_config_cached_core(|| { + loads.set(loads.get() + 1); + expected + }), + expected + ); + } + assert_eq!(loads.get(), 1, "production config cache must not reload env per GET"); + } + + #[test] + fn codec_streaming_config_loader_preserves_all_gate_env_overrides() { + temp_env::with_vars( + [ + (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("production")), + (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT_PCT, Some("37")), + (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("false")), + (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, None::<&str>), + (ENV_RUSTFS_GET_CODEC_STREAMING_RUSTFS_MIN_SIZE, Some("262144")), + ], + || { + assert_eq!( + load_get_codec_streaming_config(), + GetCodecStreamingConfig { + enabled: false, + rollout: GetCodecStreamingRollout::On, + rollout_pct: 37, + body_compat_confirmed: false, + header_compat_confirmed: false, + engine: GetCodecStreamingEngine::Rustfs, + min_size: 262144, + } + ); + }, + ); + } + #[test] fn codec_streaming_default_min_size_meets_direct_memory_ceiling() { for engine in [None, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS)] { diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index 523053f9d..d968cd487 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -78,10 +78,10 @@ impl SetDisks { include_part_checksums: true, ..Default::default() }; - let (fi, _, disks) = self + let (mut fi, _, disks) = self .get_object_fileinfo_gated(bucket, object, &read_opts, false, false) - .await?; - let mut fi = fi.into_owned(); + .await? + .into_owned(); if let Some(expected_operation_id) = expected_operation_id { require_restore_operation_id(&fi.metadata, expected_operation_id)?; } @@ -146,10 +146,10 @@ impl SetDisks { include_part_checksums: true, ..Default::default() }; - let (fi, _, disks) = self + let (mut fi, _, disks) = self .get_object_fileinfo_gated(bucket, object, &read_opts, false, false) - .await?; - let mut fi = fi.into_owned(); + .await? + .into_owned(); if let Some(expected_operation_id) = expected_operation_id { match restore_operation_id_from_metadata(&fi.metadata)? { Some(actual_operation_id) if actual_operation_id == expected_operation_id => {} diff --git a/crates/ecstore/src/set_disk/shard_source.rs b/crates/ecstore/src/set_disk/shard_source.rs index d265759aa..5dae946e4 100644 --- a/crates/ecstore/src/set_disk/shard_source.rs +++ b/crates/ecstore/src/set_disk/shard_source.rs @@ -16,6 +16,14 @@ use crate::diagnostics::get::{ GET_SHARD_READ_COST_LOCAL, GET_SHARD_READ_COST_REMOTE, GET_SHARD_READ_COST_SAME_NODE, GET_SHARD_READ_COST_UNKNOWN, }; use crate::disk::error::Error; +use crate::layout::disks_layout::MAX_ERASURE_SET_DRIVE_COUNT; +use smallvec::SmallVec; + +/// Generic codec callers may exceed the production set limit; `SmallVec` then +/// spills without changing slot semantics. +pub(crate) const INLINE_SHARD_SLOTS: usize = MAX_ERASURE_SET_DRIVE_COUNT; +pub(crate) type ShardBuffers = SmallVec<[Option>; INLINE_SHARD_SLOTS]>; +pub(crate) type ShardErrors = SmallVec<[Option; INLINE_SHARD_SLOTS]>; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ShardReadCost { @@ -43,202 +51,137 @@ impl ShardReadCost { } } -#[derive(Debug, Clone, PartialEq, Eq)] -pub(crate) struct ShardSlot { - index: usize, - read_cost: ShardReadCost, - data: Option>, - error: Option, -} - -impl ShardSlot { - pub(crate) fn new(index: usize, data: Option>, error: Option) -> Self { - Self::with_read_cost(index, ShardReadCost::Unknown, data, error) - } - - pub(crate) fn with_read_cost(index: usize, read_cost: ShardReadCost, data: Option>, error: Option) -> Self { - Self { - index, - read_cost, - data, - error, - } - } - - pub(crate) fn data(index: usize, data: Vec) -> Self { - Self::new(index, Some(data), None) - } - - pub(crate) fn data_with_read_cost(index: usize, read_cost: ShardReadCost, data: Vec) -> Self { - Self::with_read_cost(index, read_cost, Some(data), None) - } - - pub(crate) fn missing(index: usize, error: Error) -> Self { - Self::new(index, None, Some(error)) - } - - pub(crate) fn missing_with_read_cost(index: usize, read_cost: ShardReadCost, error: Error) -> Self { - Self::with_read_cost(index, read_cost, None, Some(error)) - } - - pub(crate) fn index(&self) -> usize { - self.index - } - - pub(crate) fn read_cost(&self) -> ShardReadCost { - self.read_cost - } - - pub(crate) fn has_data(&self) -> bool { - self.data.is_some() - } - - pub(crate) fn data_bytes(&self) -> Option<&[u8]> { - self.data.as_deref() - } - - pub(crate) fn error(&self) -> Option<&Error> { - self.error.as_ref() - } -} - #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct StripeReadState { - slots: Vec, + shards: ShardBuffers, + errors: ShardErrors, read_quorum: usize, } impl StripeReadState { - pub(crate) fn new(slots: Vec, read_quorum: usize) -> Self { - Self { slots, read_quorum } - } - + #[cfg(test)] pub(crate) fn from_parts(shards: Vec>>, errors: Vec>, read_quorum: usize) -> Self { - Self::from_parts_with_read_costs(shards, errors, &[], read_quorum) + let mut shards = SmallVec::from_vec(shards); + let mut errors = SmallVec::from_vec(errors); + let slot_count = shards.len().max(errors.len()); + shards.resize_with(slot_count, || None); + errors.resize_with(slot_count, || None); + Self { + shards, + errors, + read_quorum, + } } - pub(crate) fn from_parts_with_read_costs(shards: S, errors: E, read_costs: &[ShardReadCost], read_quorum: usize) -> Self - where - S: IntoIterator>>, - S::IntoIter: ExactSizeIterator, - E: IntoIterator>, - E::IntoIter: ExactSizeIterator, - { - let mut shards = shards.into_iter(); - let mut errors = errors.into_iter(); - let slot_count = shards.len().max(errors.len()); - let mut slots = Vec::with_capacity(slot_count); - for index in 0..slot_count { - let read_cost = read_costs.get(index).copied().unwrap_or(ShardReadCost::Unknown); - slots.push(ShardSlot::with_read_cost( - index, - read_cost, - shards.next().flatten(), - errors.next().flatten(), - )); - } - Self::new(slots, read_quorum) + pub(crate) fn with_slot_count(slot_count: usize, read_quorum: usize) -> Self { + let mut state = Self { + shards: SmallVec::new(), + errors: SmallVec::new(), + read_quorum, + }; + state.reset(slot_count, read_quorum); + state + } + + pub(crate) fn reset(&mut self, slot_count: usize, read_quorum: usize) { + self.shards.clear(); + self.shards.resize_with(slot_count, || None); + self.errors.clear(); + self.errors.resize_with(slot_count, || None); + self.read_quorum = read_quorum; } pub(crate) fn available_shards(&self) -> usize { - self.slots.iter().filter(|slot| slot.has_data()).count() + self.shards.iter().filter(|shard| shard.is_some()).count() } pub(crate) fn can_decode(&self) -> bool { self.available_shards() >= self.read_quorum } - pub(crate) fn slots(&self) -> &[ShardSlot] { - &self.slots + pub(crate) fn is_empty(&self) -> bool { + self.shards.is_empty() } - pub(crate) fn slot_by_index(&self, index: usize) -> Option<&ShardSlot> { - if let Some(slot) = self.slots.get(index) - && slot.index == index - { - return Some(slot); - } - self.slots.iter().find(|slot| slot.index == index) + pub(crate) fn data_bytes(&self, index: usize) -> Option<&[u8]> { + self.shards.get(index).and_then(Option::as_deref) + } + + #[cfg(test)] + pub(crate) fn error(&self, index: usize) -> Option<&Error> { + self.errors.get(index).and_then(Option::as_ref) } pub(crate) fn data_shards_complete(&self, data_shards: usize) -> bool { - (0..data_shards).all(|index| self.slot_by_index(index).is_some_and(ShardSlot::has_data)) + self.shards.len() >= data_shards && self.shards.iter().take(data_shards).all(Option::is_some) } - pub(crate) fn into_parts(self) -> (Vec>>, Vec>) { - let part_count = self.slots.iter().map(|slot| slot.index).max().map_or(0, |index| index + 1); - let mut shards = Vec::with_capacity(part_count); - shards.resize_with(part_count, || None); - let mut errors = Vec::with_capacity(part_count); - errors.resize_with(part_count, || None); - for slot in self.slots { - shards[slot.index] = slot.data; - errors[slot.index] = slot.error; - } - (shards, errors) + pub(crate) fn parts_mut(&mut self) -> (&mut ShardBuffers, &mut ShardErrors) { + (&mut self.shards, &mut self.errors) + } + + pub(crate) fn shards_mut(&mut self) -> &mut ShardBuffers { + &mut self.shards + } + + pub(crate) fn into_parts(self) -> (ShardBuffers, ShardErrors) { + (self.shards, self.errors) + } + + #[cfg(test)] + pub(crate) fn scratch_storage(&self) -> (*const Option>, *const Option, bool, bool) { + (self.shards.as_ptr(), self.errors.as_ptr(), self.shards.spilled(), self.errors.spilled()) + } + + #[cfg(test)] + pub(crate) fn shard_allocation(&self, index: usize) -> Option<(*const u8, usize)> { + self.shards + .get(index) + .and_then(|shard| shard.as_ref().map(|shard| (shard.as_ptr(), shard.capacity()))) } } #[async_trait::async_trait] pub(crate) trait ShardStripeSource: Send { - async fn read_next_stripe(&mut self) -> StripeReadState; + async fn read_next_stripe(&mut self) -> Box; + + fn recycle_stripe(&mut self, _state: Box) {} } #[cfg(test)] mod tests { use super::*; + use std::mem::size_of; #[test] - fn stripe_read_state_tracks_decode_quorum() { - let state = StripeReadState::new( - vec![ - ShardSlot::data_with_read_cost(0, ShardReadCost::Local, vec![1]), - ShardSlot::missing_with_read_cost(1, ShardReadCost::Remote, Error::FileNotFound), - ShardSlot::data_with_read_cost(2, ShardReadCost::SameNode, vec![2]), - ], - 2, - ); + fn stripe_scratch_capacity_matches_the_production_set_limit() { + type OversizedShardBuffers = SmallVec<[Option>; 32]>; + type OversizedShardErrors = SmallVec<[Option; 32]>; + + assert_eq!(INLINE_SHARD_SLOTS, MAX_ERASURE_SET_DRIVE_COUNT); + assert!(size_of::() < size_of::()); + assert!(size_of::() < size_of::()); + } + + #[test] + fn stripe_read_state_tracks_decode_quorum_and_slot_access() { + let state = + StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![2])], vec![None, Some(Error::FileNotFound), None], 2); assert_eq!(state.available_shards(), 2); assert!(state.can_decode()); - assert_eq!(state.slots()[1].index(), 1); - assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local); - assert!(state.slots()[2].read_cost().is_low_cost()); + assert_eq!(state.data_bytes(0), Some(&[1][..])); + assert_eq!(state.error(1), Some(&Error::FileNotFound)); } #[test] fn stripe_read_state_preserves_shards_and_errors() { - let state = StripeReadState::new(vec![ShardSlot::missing(1, Error::FileCorrupt), ShardSlot::data(0, vec![1, 2, 3])], 2); + let state = StripeReadState::from_parts(vec![Some(vec![1, 2, 3]), None], vec![None, Some(Error::FileCorrupt)], 2); assert!(!state.can_decode()); let (shards, errors) = state.into_parts(); - assert_eq!(shards, vec![Some(vec![1, 2, 3]), None]); - assert_eq!(errors, vec![None, Some(Error::FileCorrupt)]); - } - - #[test] - fn stripe_read_state_builds_slots_from_parallel_reader_parts() { - let state = - StripeReadState::from_parts(vec![Some(vec![1]), None, Some(vec![3])], vec![None, Some(Error::FileNotFound)], 2); - - assert!(state.can_decode()); - assert_eq!(state.slots()[1].index(), 1); - assert_eq!(state.slots()[1].error(), Some(&Error::FileNotFound)); - } - - #[test] - fn stripe_read_state_preserves_read_cost_hints() { - let state = StripeReadState::from_parts_with_read_costs( - vec![Some(vec![1]), None, Some(vec![3])], - vec![None, Some(Error::FileNotFound)], - &[ShardReadCost::Local, ShardReadCost::Remote, ShardReadCost::Unknown], - 2, - ); - - assert_eq!(state.slots()[0].read_cost(), ShardReadCost::Local); - assert_eq!(state.slots()[1].read_cost(), ShardReadCost::Remote); - assert_eq!(state.slots()[2].read_cost(), ShardReadCost::Unknown); - assert_eq!(ShardReadCost::SameNode.as_str(), GET_SHARD_READ_COST_SAME_NODE); + assert_eq!(shards.as_slice(), &[Some(vec![1, 2, 3]), None]); + assert_eq!(errors.as_slice(), &[None, Some(Error::FileCorrupt)]); } #[test] @@ -258,8 +201,8 @@ mod tests { let state = StripeReadState::from_parts(vec![Some(vec![1]), Some(vec![2]), None], Vec::new(), 2); assert!(state.data_shards_complete(2)); - assert_eq!(state.slots()[0].data_bytes(), Some(&[1][..])); - assert_eq!(state.slot_by_index(1).and_then(ShardSlot::data_bytes), Some(&[2][..])); + assert_eq!(state.data_bytes(0), Some(&[1][..])); + assert_eq!(state.data_bytes(1), Some(&[2][..])); } #[test] @@ -268,12 +211,4 @@ mod tests { assert!(!state.data_shards_complete(2)); } - - #[test] - fn stripe_read_state_finds_out_of_order_slots_by_index() { - let state = StripeReadState::new(vec![ShardSlot::data(2, vec![3]), ShardSlot::data(0, vec![1])], 2); - - assert_eq!(state.slot_by_index(0).and_then(ShardSlot::data_bytes), Some(&[1][..])); - assert!(state.slot_by_index(1).is_none()); - } } diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 8fcf6b065..30a76bb69 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -389,7 +389,7 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore { type GetObjectReader = GetObjectReader; type PutObjectReader = PutObjReader; - #[instrument(level = "debug", skip(self))] + #[instrument(level = "debug", skip(self, h))] async fn get_object_reader( &self, bucket: &str, diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index df915af7f..ac6b520f0 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -667,18 +667,6 @@ impl SelectObjectSnapshotLockLossWake { } } -fn select_object_ssec_headers(headers: &HeaderMap) -> HeaderMap { - use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; - - let mut selected = HeaderMap::new(); - for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] { - if let Some(value) = headers.get(name) { - selected.insert(name, value.clone()); - } - } - selected -} - // LockRegistry clones its canonical client Arc for each endpoint host, so an // exact Arc set identifies one distributed namespace-lock quorum domain. fn same_distributed_lock_domain(left: &[Arc], right: &[Arc]) -> bool { @@ -1296,7 +1284,7 @@ impl ECStore { pool, bucket: bucket.to_owned(), object, - headers: select_object_ssec_headers(headers), + headers: rustfs_utils::http::project_ssec_transport_headers(headers), opts, object_info, logical_size, @@ -1769,7 +1757,7 @@ impl ECStore { Self::resolve_decommission_tiered_object_result(result, bucket, &object) } - #[instrument(level = "debug", skip(self))] + #[instrument(level = "debug", skip(self, h))] #[hotpath::measure(impl_type = "ECStore")] pub(super) async fn handle_get_object_reader( &self, @@ -3360,25 +3348,6 @@ mod tests { assert!(second_signal.is_lost()); } - #[test] - fn select_snapshot_retains_only_ssec_headers() { - use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; - - let mut headers = HeaderMap::new(); - headers.insert(SSEC_ALGORITHM_HEADER, "AES256".parse().expect("valid SSE-C algorithm header")); - headers.insert(SSEC_KEY_HEADER, "secret-key".parse().expect("valid SSE-C key header")); - headers.insert(SSEC_KEY_MD5_HEADER, "key-md5".parse().expect("valid SSE-C key digest header")); - headers.insert("authorization", "credential".parse().expect("valid authorization header")); - - let selected = select_object_ssec_headers(&headers); - - assert_eq!(selected.len(), 3); - assert_eq!(selected.get(SSEC_ALGORITHM_HEADER), headers.get(SSEC_ALGORITHM_HEADER)); - assert_eq!(selected.get(SSEC_KEY_HEADER), headers.get(SSEC_KEY_HEADER)); - assert_eq!(selected.get(SSEC_KEY_MD5_HEADER), headers.get(SSEC_KEY_MD5_HEADER)); - assert!(selected.get("authorization").is_none()); - } - #[test] fn tier_delete_entry_is_prepared_and_bound_to_source_generation() { let identity = [9_u8; 32]; @@ -4608,6 +4577,8 @@ mod tests { assert_eq!(snapshot.headers.get(SSEC_KEY_HEADER), request_headers.get(SSEC_KEY_HEADER)); assert_eq!(snapshot.headers.get(SSEC_KEY_MD5_HEADER), request_headers.get(SSEC_KEY_MD5_HEADER)); assert!(snapshot.headers.get("authorization").is_none()); + assert!(snapshot.headers.values().all(http::HeaderValue::is_sensitive)); + assert!(!format!("{:?}", snapshot.headers).contains("secret-key")); assert_eq!( snapshot.logical_size(), u64::try_from(payload.len()).expect("test payload length should fit in u64") diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index d1cbadf75..04d6aa804 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -477,7 +477,7 @@ impl Drop for DistributedLockGuard { #[derive(Debug)] pub struct DistributedLock { /// Lock clients for this namespace - clients: Vec>, + clients: Arc<[Arc]>, /// Namespace identifier namespace: Arc, /// Quorum size for exclusive/write operations @@ -496,11 +496,11 @@ struct LockAcquireQuorumResult { impl DistributedLock { /// Create new distributed lock pub fn new(namespace: String, clients: Vec>, quorum: usize) -> Self { - Self::new_shared(namespace.into(), clients, quorum) + Self::new_shared(namespace.into(), clients.into(), quorum) } - /// Create a distributed lock that shares an existing namespace allocation. - pub(crate) fn new_shared(namespace: Arc, clients: Vec>, quorum: usize) -> Self { + /// Create a distributed lock that shares existing namespace and client allocations. + pub(crate) fn new_shared(namespace: Arc, clients: Arc<[Arc]>, quorum: usize) -> Self { let q = if clients.len() <= 1 { 1 } else { @@ -777,7 +777,7 @@ impl DistributedLock { fn spawn_pending_cleanup( mut pending: JoinSet, - clients: Vec>, + clients: Arc<[Arc]>, fallback_lock_id: LockId, context: &'static str, ) { diff --git a/crates/lock/src/namespace/mod.rs b/crates/lock/src/namespace/mod.rs index 4a8854185..07039d6d4 100644 --- a/crates/lock/src/namespace/mod.rs +++ b/crates/lock/src/namespace/mod.rs @@ -200,9 +200,13 @@ impl NamespaceLock { Self::Distributed(DistributedLock::new(namespace, clients, quorum)) } - /// Create a namespace lock that shares an existing namespace allocation. - pub fn with_clients_and_quorum_shared(namespace: Arc, clients: Vec>, quorum: usize) -> Self { - Self::Distributed(DistributedLock::new_shared(namespace, clients, quorum)) + /// Create a namespace lock that shares existing namespace and client allocations. + pub fn with_clients_and_quorum_shared( + namespace: Arc, + clients: impl Into]>>, + quorum: usize, + ) -> Self { + Self::Distributed(DistributedLock::new_shared(namespace, clients.into(), quorum)) } /// Get namespace identifier diff --git a/crates/utils/src/http/object_encryption_keys.rs b/crates/utils/src/http/object_encryption_keys.rs index 2a623e8cc..7b71faeae 100644 --- a/crates/utils/src/http/object_encryption_keys.rs +++ b/crates/utils/src/http/object_encryption_keys.rs @@ -25,7 +25,7 @@ // The lowercase stored forms, matching exactly what encryption_material_to_metadata // persists. The read-path SSE-C check is case-sensitive, so restoring under any // other casing would classify the replica as managed-SSE and reject SSE-C GETs. -use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_MD5_HEADER}; +use super::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; pub const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id"; pub const INTERNAL_ENCRYPTION_KEY_HEADER: &str = "x-rustfs-encryption-key"; @@ -85,6 +85,20 @@ pub const SSEC_REPLICATION_TRANSPORT_HEADERS: &[(&str, &str)] = &[ (MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER, REPLICATION_ENCRYPTED_MULTIPART_HEADER), ]; +/// Retains only the SSE-C headers consumed by object readers and marks their +/// values sensitive so instrumented storage calls cannot expose key material. +pub fn project_ssec_transport_headers(headers: &http::HeaderMap) -> http::HeaderMap { + let mut projected = http::HeaderMap::new(); + for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] { + if let Some(value) = headers.get(name) { + let mut value = value.clone(); + value.set_sensitive(true); + projected.insert(name, value); + } + } + projected +} + /// Prefixes of replication SSE transport keys whose values carry encryption /// material and must never reach logs. Consumed by `rustfs_filemeta` redaction. pub const REPLICATION_SSE_TRANSPORT_PREFIXES: &[&str] = &[ @@ -155,6 +169,22 @@ pub fn is_replication_stripped_encryption_key(key: &str) -> bool { mod tests { use super::*; + #[test] + fn ssec_transport_projection_retains_only_redacted_reader_headers() { + let mut headers = http::HeaderMap::new(); + headers.insert(SSEC_ALGORITHM_HEADER, http::HeaderValue::from_static("AES256")); + headers.insert(SSEC_KEY_HEADER, http::HeaderValue::from_static("secret-key")); + headers.insert(SSEC_KEY_MD5_HEADER, http::HeaderValue::from_static("key-md5")); + headers.insert(http::header::AUTHORIZATION, http::HeaderValue::from_static("credential")); + + let projected = project_ssec_transport_headers(&headers); + + assert_eq!(projected.len(), 3); + assert!(projected.values().all(http::HeaderValue::is_sensitive)); + assert!(projected.get(http::header::AUTHORIZATION).is_none()); + assert!(!format!("{projected:?}").contains("secret-key")); + } + #[test] fn transport_metadata_roundtrip_restores_stored_keys() { let mut headers = http::HeaderMap::new(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index d531b0fcd..5c4e1e4c6 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -134,6 +134,8 @@ use rustfs_s3_ops::{S3Operation, delete_event_name_for_marker, put_event_name_fo use rustfs_s3select_api::object_store::bytes_stream; use rustfs_targets::{EventName, get_request_host, get_request_port, get_request_user_agent}; use rustfs_utils::CompressionAlgorithm; +#[cfg(test)] +use rustfs_utils::http::headers::{SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_TYPE, AMZ_WEBSITE_REDIRECT_LOCATION, CONTENT_TYPE, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_REPLICA_STATUS, SUFFIX_REPLICA_TIMESTAMP, @@ -147,9 +149,8 @@ use rustfs_utils::http::{ AMZ_RUSTFS_SNOWBALL_IGNORE_ERRORS, AMZ_RUSTFS_SNOWBALL_PREFIX, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, AMZ_SNOWBALL_EXTRACT, AMZ_SNOWBALL_IGNORE_DIRS, AMZ_SNOWBALL_IGNORE_ERRORS, AMZ_SNOWBALL_PREFIX, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, - SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER, }, - insert_str, remove_str, + insert_str, project_ssec_transport_headers, remove_str, }; use rustfs_utils::path::{encode_dir_object, is_dir_object, path_join_buf}; use rustfs_utils::retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, RetryTimer}; @@ -2045,18 +2046,6 @@ struct GetObjectResumeContext { identity: GetObjectResumeIdentity, } -fn get_object_store_headers(request_headers: &HeaderMap) -> HeaderMap { - let mut headers = HeaderMap::new(); - for name in [SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER] { - if let Some(value) = request_headers.get(name) { - let mut value = value.clone(); - value.set_sensitive(true); - headers.insert(name, value); - } - } - headers -} - impl GetObjectResumeContext { #[allow(clippy::too_many_arguments)] fn new( @@ -2076,7 +2065,7 @@ impl GetObjectResumeContext { } // Store spans record their header argument at debug level. Retain only // the SSE-C inputs needed to reopen the reader and keep them redacted. - let ssec_headers = get_object_store_headers(request_headers); + let ssec_headers = project_ssec_transport_headers(request_headers); Self { store, bucket: bucket.to_string(), @@ -4469,7 +4458,7 @@ impl DefaultObjectUsecase { ) -> S3Result { let read_start = std::time::Instant::now(); let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start); - let store_headers = get_object_store_headers(&req.headers); + let store_headers = project_ssec_transport_headers(&req.headers); let cache_adapter = self.object_data_cache(); if cache_adapter.is_disabled() || !cache_adapter.materialize_fill_enabled() { let io_planning = Self::acquire_get_object_io_planning( @@ -4484,7 +4473,7 @@ impl DefaultObjectUsecase { .await?; let reader = track_object_read_setup( object_traffic_health.as_deref(), - store.get_object_reader(bucket, key, rs.clone(), store_headers.clone(), opts), + store.get_object_reader(bucket, key, rs.clone(), store_headers, opts), ) .await .map_err(map_get_object_reader_error)?; @@ -4773,12 +4762,10 @@ impl DefaultObjectUsecase { let io_planning = metadata_admission .take() .ok_or_else(|| s3_error!(InternalError, "prepared metadata admission is unavailable"))?; - let reader = track_object_read_setup( - object_traffic_health.as_deref(), - prepared.with_headers(store_headers.clone()).into_reader(), - ) - .await - .map_err(map_get_object_reader_error)?; + let reader = + track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader()) + .await + .map_err(map_get_object_reader_error)?; (io_planning, reader) } else { let io_planning = Self::acquire_get_object_io_planning( @@ -4798,12 +4785,9 @@ impl DefaultObjectUsecase { ) .await .map_err(map_get_object_reader_error)?; - track_object_read_setup( - object_traffic_health.as_deref(), - prepared.with_headers(store_headers.clone()).into_reader(), - ) - .await - .map_err(map_get_object_reader_error)? + track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader()) + .await + .map_err(map_get_object_reader_error)? } else { track_object_read_setup( object_traffic_health.as_deref(), @@ -13823,11 +13807,12 @@ mod tests { request_headers.insert(SSEC_KEY_MD5_HEADER, HeaderValue::from_static("bWQ1")); request_headers.insert(http::header::AUTHORIZATION, HeaderValue::from_static("AWS4-HMAC-SHA256 Credential=test")); request_headers.insert("x-amz-security-token", HeaderValue::from_static("session-token")); - let store_headers = get_object_store_headers(&request_headers); + let store_headers = project_ssec_transport_headers(&request_headers); assert_eq!(store_headers.len(), 3, "only store-consumed SSE-C headers are forwarded"); assert!(store_headers.values().all(HeaderValue::is_sensitive)); assert!(store_headers.get(http::header::AUTHORIZATION).is_none()); assert!(store_headers.get("x-amz-security-token").is_none()); + assert!(!format!("{store_headers:?}").contains("dGVzdC1rZXk=")); let plain_info = ObjectInfo { size: 11, ..Default::default() diff --git a/rustfs/src/storage/minio_generated_read_test.rs b/rustfs/src/storage/minio_generated_read_test.rs index 36285c565..35b2d8eed 100644 --- a/rustfs/src/storage/minio_generated_read_test.rs +++ b/rustfs/src/storage/minio_generated_read_test.rs @@ -103,6 +103,7 @@ fn load_file_info(case_dir: &Path, manifest: &ManifestRecord) -> FileInfo { FileInfoOpts { data: true, include_free_versions: true, + include_part_checksums: false, }, ) .unwrap_or_else(|err| panic!("decode {}: {err}", xl_meta_path.display())) diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index 1a418ec01..66dd2095a 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -1718,20 +1718,23 @@ pub(crate) fn build_ssec_read_headers( let mut headers = HeaderMap::new(); if let Some(algorithm) = algorithm - && let Ok(value) = HeaderValue::from_str(algorithm.as_str()) + && let Ok(mut value) = HeaderValue::from_str(algorithm.as_str()) { + value.set_sensitive(true); headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, value); } if let Some(key) = key - && let Ok(value) = HeaderValue::from_str(key.as_str()) + && let Ok(mut value) = HeaderValue::from_str(key.as_str()) { + value.set_sensitive(true); headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, value); } if let Some(key_md5) = key_md5 - && let Ok(value) = HeaderValue::from_str(key_md5.as_str()) + && let Ok(mut value) = HeaderValue::from_str(key_md5.as_str()) { + value.set_sensitive(true); headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, value); } @@ -3481,6 +3484,21 @@ mod tests { }; use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER; + #[test] + fn ssec_read_headers_are_sensitive() { + let headers = super::build_ssec_read_headers( + Some(&SSECustomerAlgorithm::from("AES256".to_string())), + Some(&SSECustomerKey::from("dHJhbnNwb3J0LXNlY3JldA==".to_string())), + Some(&SSECustomerKeyMD5::from("bWQ1LXNlY3JldA==".to_string())), + ); + + assert_eq!(headers.len(), 3); + assert!(headers.values().all(HeaderValue::is_sensitive)); + let debug = format!("{headers:?}"); + assert!(!debug.contains("dHJhbnNwb3J0LXNlY3JldA==")); + assert!(!debug.contains("bWQ1LXNlY3JldA==")); + } + #[test] fn anonymous_s3_request_builds_kms_principal() { let mut request = s3s::S3Request {