// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. //! `ObjectIO` and `ObjectOperations` storage-api contract impls for `SetDisks` //! — the core object read/write hot path (P6 of the God-Object split, tracking //! backlog#815, issue backlog#821). Relocated verbatim from set_disk/mod.rs; //! the contracts stay implemented `for SetDisks`, so their associated-type //! bounds are unchanged, and the impls reach shared primitives through the //! SetDisks core (io_primitives) via inherent calls. use super::super::*; use crate::bucket::lifecycle::{ tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, tier_sweeper::{Jentry, RemoteTierDeleteOutcome, delete_object_from_remote_tier_with_lease_idempotent}, transition_transaction::{ TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record, save_transition_transaction_record, }, }; use crate::disk::OldCurrentSize; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; use crate::store::ECStore; use futures::FutureExt as _; use std::future::Future; fn erasure_from_file_info(fi: &FileInfo, uses_legacy: bool) -> Result { coding::Erasure::try_new_with_options(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size, uses_legacy) .map_err(Error::from) } /// Length of the full plaintext body when — and only when — this read's output /// is exactly the object's complete plaintext, so the app-layer body cache may /// serve it in place of the erasure read. /// /// A hook hit bypasses `ReadPlan`/`ReadTransform` entirely, so it is sound only /// where the normal read path would produce that same plaintext byte-for-byte /// AND expose the same `object_info.size`. This is a fail-closed allow-list, not /// a deny-list: every read whose `ReadPlan` applies some other transform returns /// `None`, so a newly added `ReadPlan` branch bypasses the cache by default /// instead of silently serving bytes in the wrong representation. /// /// Refused reads, each mapping to a `ReadPlan::build` branch: /// - ranged / part-number reads — the cache only holds whole objects; /// - `raw_data_movement_read` / `data_movement` — yields the STORED /// representation, e.g. compressed bytes (backlog#1108); /// - restore reads — `restore_request_active` forces the `Plain` branch, so a /// compressed object yields STORED bytes under its compressed `size`; /// - encrypted objects — `ReadTransform::Encrypted` rewrites size and the cache /// must never hold their plaintext; /// - remote (transitioned) objects — served from the warm tier. /// /// Compressed objects ARE eligible: `ReadTransform::Compressed` returns the full /// plaintext and rewrites `object_info.size` to the decompressed length, which /// the caller must replicate with the returned length (backlog#1109). /// /// The fail-closed shape here is enforced by `scripts/check_body_cache_whitelist.sh` /// (backlog#1146): the guard requires every exclusion predicate and a `return /// None` to precede the first `Some(..)`, so a refactor to a deny-list fails CI. /// If you rename or move this function, update that script. fn full_object_plaintext_len(range: &Option, opts: &ObjectOptions, object_info: &ObjectInfo) -> Option { if range.is_some() || opts.part_number.is_some() || opts.raw_data_movement_read || opts.data_movement || crate::object_api::restore_request_active(opts) || object_info.is_encrypted() || object_info.is_remote() || object_info.delete_marker || object_info.size == 0 || object_info.version_only || object_info.metadata_only || object_info.is_inline_fast_path_eligible() { return None; } if object_info.is_compressed() { return object_info.get_actual_size().ok(); } Some(object_info.size) } pub(crate) fn body_cache_plaintext_len( range: &Option, opts: &ObjectOptions, object_info: &ObjectInfo, ) -> Option { full_object_plaintext_len(range, opts, object_info) } #[async_trait::async_trait] impl crate::storage_api_contracts::object::ObjectIO for SetDisks { type Error = Error; type RangeSpec = HTTPRangeSpec; type HeaderMap = HeaderMap; type ObjectOptions = ObjectOptions; type ObjectInfo = ObjectInfo; type GetObjectReader = GetObjectReader; type PutObjectReader = PutObjReader; #[tracing::instrument(level = "debug", skip(self))] async fn get_object_reader( &self, bucket: &str, object: &str, range: Option, h: HeaderMap, opts: &ObjectOptions, ) -> Result { crate::hp_guard!("SetDisks::get_object_reader"); let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); // Check if lock optimization is enabled for reads that are fully materialized in memory. let lock_optimization_enabled = is_lock_optimization_enabled(); // Acquire a shared read-lock early to protect read consistency let mut read_lock_guard = if !opts.no_lock { let acquire_start = stage_metrics_enabled.then(Instant::now); let lock_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); // Record lock wait for deadlock detection if is_deadlock_detection_enabled() { debug!( lock_id = format!("{}:{}", bucket, object), lock_type = "read", resource = format!("{}/{}", bucket, object), "Waiting for read lock" ); } let guard = self.acquire_read_lock_diag("get_object", bucket, object).await?; // Record lock acquisition for deadlock detection let _lock_id = record_lock_acquire(bucket, object, "read"); // Record lock statistics only when GET stage metrics are enabled, // matching the adjacent stage timer. Avoids a per-GET clock read and // two global-recorder lookups when observability/stage metrics are off. if let Some(acquire_start) = acquire_start { metrics::counter!("rustfs.lock.acquire.total", "type" => "read").increment(1); metrics::histogram!("rustfs.lock.acquire.duration.seconds").record(acquire_start.elapsed().as_secs_f64()); } record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_LOCK_ACQUIRE, lock_stage_start); Some(guard) } else { None }; 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) } else { match self.get_object_fileinfo(bucket, object, opts, true, true).await { Ok((fi, files, disks)) => (fi, files, disks, None), Err(err) => { rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64()); record_get_object_pipeline_failure(GET_STAGE_METADATA, classify_storage_error(&err)); return Err(to_object_err(err, vec![bucket, object])); } } }; 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); 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(); rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_elapsed); rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_SET_DISK, GET_STAGE_METADATA, object_class.as_str(), size_bucket, metadata_elapsed, ); if object_info.delete_marker { if opts.version_id.is_none() { return Err(to_object_err(Error::FileNotFound, vec![bucket, object])); } return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object])); } // if object_info.size == 0 { // let empty_rd: Box = Box::new(Bytes::new()); // return Ok(GetObjectReader { // stream: empty_rd, // object_info, // }); // } if object_info.size == 0 { record_get_object_reader_path_observation(GET_OBJECT_PATH_EMPTY, object_class, size_bucket); // if let Some(rs) = range { // let _ = rs.get_offset_length(object_info.size)?; // } let reader = GetObjectReader { stream: Box::new(Cursor::new(Vec::new())), object_info, buffered_body: Some(Bytes::new()), body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } // Inline data fast path: skip duplex pipe for small inline objects. // 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) { let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let data_shards = fi.erasure.data_blocks; let object_size = usize::try_from(fi.size) .map_err(|_| to_object_err(Error::other("inline fast path object size is invalid"), vec![bucket, object]))?; let checksum_info = fi.erasure.get_checksum_info(fi.parts[0].number); let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S { rustfs_utils::HashAlgorithm::HighwayHash256SLegacy } else { checksum_info.algorithm }; 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| { disks.get(index).is_some_and(Option::is_some) }) { let read_length = inline_erasure_shard_file_offset( 0, object_size, object_size, fi.erasure.block_size, data_shards, fi.uses_legacy_checksum, ); let shard_size = inline_erasure_shard_size(fi.erasure.block_size, data_shards, fi.uses_legacy_checksum); if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_INLINE_PREPARE, object_class.as_str(), size_bucket, inline_prepare_stage_start.elapsed().as_secs_f64(), ); } let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); let mut readers = build_inline_bitrot_readers_from_refs( &data_files, bucket, object, read_length, shard_size, &checksum_algo, opts.skip_verify_bitrot, ) .await?; if let Some(reader_setup_stage_start) = reader_setup_stage_start { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_READER_SETUP, object_class.as_str(), size_bucket, reader_setup_stage_start.elapsed().as_secs_f64(), ); } // Decode directly let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); if let Some(body) = try_read_inline_data_shards_direct(&mut readers, data_shards, read_length, object_size).await { if let Some(decode_stage_start) = decode_stage_start { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_DECODE, object_class.as_str(), size_bucket, decode_stage_start.elapsed().as_secs_f64(), ); } record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket); let reader = GetObjectReader { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } } 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); // Check if we have enough inline data shards let inline_count = files .iter() .take(data_shards) .filter(|f| f.data.as_ref().is_some_and(|d| !d.is_empty())) .count(); if inline_count >= data_shards { if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_INLINE_PREPARE, object_class.as_str(), size_bucket, inline_prepare_stage_start.elapsed().as_secs_f64(), ); } let reader_setup_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); let readers = build_inline_bitrot_readers( &files, total_shards, bucket, object, read_length, erasure.shard_size(), &checksum_algo, opts.skip_verify_bitrot, ) .await?; if let Some(reader_setup_stage_start) = reader_setup_stage_start { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_READER_SETUP, object_class.as_str(), size_bucket, reader_setup_stage_start.elapsed().as_secs_f64(), ); } let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now); let mut output = Cursor::new(Vec::with_capacity(object_size)); let (written, err) = erasure.decode(&mut output, readers, 0, object_size, object_size).await; if let Some(e) = err { return Err(to_object_err(e.into(), vec![bucket, object])); } if written == 0 && fi.size > 0 { return Err(to_object_err( Error::other("inline fast path: erasure decode returned 0 bytes"), vec![bucket, object], )); } let body = Bytes::from(output.into_inner()); if let Some(decode_stage_start) = decode_stage_start { rustfs_io_metrics::record_get_object_stage_duration_by_size( GET_OBJECT_PATH_INLINE_DIRECT, GET_STAGE_DECODE, object_class.as_str(), size_bucket, decode_stage_start.elapsed().as_secs_f64(), ); } record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket); let reader = GetObjectReader { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), body_source: GetObjectBodySource::Unprobed, }; return Ok(reader); } } let path_decision_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let codec_streaming_gate = get_codec_streaming_reader_gate( bucket, object, &range, opts.part_number, &object_info, &fi, lock_optimization_enabled, ); record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_PATH_DECISION, path_decision_stage_start); if object_info.is_remote() { if let GetCodecStreamingDecision::Fallback(reason) = codec_streaming_gate.decision { record_get_codec_streaming_gate_decision( codec_streaming_gate.object_class, codec_streaming_gate.decision, size_bucket, ); rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); } record_get_object_reader_path_observation(GET_OBJECT_PATH_REMOTE_TRANSITION, object_class, size_bucket); let mut opts = opts.clone(); if object_info.parts.len() == 1 { opts.part_number = Some(1); } let gr = get_transitioned_object_reader_with_tier_manager( bucket, object, &range, &h, &object_info, &opts, &self.ctx.tier_config_mgr(), ) .await?; return Ok(finish_set_disk_read_lock( gr, read_lock_guard.take(), lock_optimization_enabled, bucket, object, )); } // App-layer object data cache probe: metadata (etag/size) is resolved // but no data shards have been read yet, so a hit skips the erasure // read, bitrot verify and decode entirely. The hook validates object // identity and rejects anything it cannot serve byte-identically. // // `plaintext_len` carries the size the normal `ReadPlan` would have // published, which the reader below must reproduce: for a compressed // object `ReadTransform::Compressed` sets `object_info.size` to the // decompressed length, and consumers such as UploadPartCopy read the // copy length straight off that field (backlog#1109). // Records whether the app-layer cache probe ran for this read, so the // app layer does not repeat the lookup it already performed after fresh // metadata resolution (backlog#1121 / ODC-16). It stays `Unprobed` when // the read is ineligible under the allow-list or the hook is not // registered; the direct-memory and streaming readers built below carry // it forward. let mut body_source = GetObjectBodySource::Unprobed; if let Some(plaintext_len) = full_object_plaintext_len(&range, opts, &object_info) && !get_object_body_cache_hook_suppressed() && let Some(hook) = get_object_body_cache_hook() { match hook.lookup(bucket, object, &object_info).await { Some(body) if i64::try_from(body.len()).is_ok_and(|len| len == plaintext_len) => { record_get_object_reader_path_observation(GET_OBJECT_PATH_BODY_CACHE, object_class, size_bucket); let mut object_info = object_info; object_info.size = plaintext_len; let reader = GetObjectReader { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), body_source: GetObjectBodySource::HookServed, }; if lock_optimization_enabled { release_materialized_read_lock(bucket, object, read_lock_guard.take()); } return Ok(reader); } // Probed after fresh metadata resolution but no usable body: a // genuine miss, or a length-defensive rejection. The miss is // authoritative, so the app layer must not look up again. _ => { body_source = GetObjectBodySource::HookMissed; } } } 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, opts.skip_verify_bitrot, object_class.as_str(), size_bucket, ) .await? { if body.len() != object_size { return Err(to_object_err( Error::other("direct-memory GET decoded length mismatch"), vec![bucket, object], )); } record_get_object_reader_path_observation(GET_OBJECT_PATH_DIRECT_MEMORY, object_class, size_bucket); let reader = GetObjectReader { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), body_source, }; if lock_optimization_enabled { release_materialized_read_lock(bucket, object, read_lock_guard.take()); debug!(bucket, object, "Lock optimization: released read lock after direct-memory read"); } return Ok(reader); } let mut output = Vec::with_capacity(object_size); Self::get_object_with_fileinfo( bucket, object, 0, object_info.size, &mut output, fi, files, &disks, self.set_index, self.pool_index, opts.skip_verify_bitrot, true, GET_OBJECT_PATH_DIRECT_MEMORY, object_class.as_str(), size_bucket, ) .await?; if output.len() != object_size { return Err(to_object_err( Error::other("direct-memory GET decoded length mismatch"), vec![bucket, object], )); } record_get_object_reader_path_observation(GET_OBJECT_PATH_DIRECT_MEMORY, object_class, size_bucket); let body = Bytes::from(output); let reader = GetObjectReader { stream: Box::new(Cursor::new(body.clone())), object_info, buffered_body: Some(body), body_source, }; if lock_optimization_enabled { release_materialized_read_lock(bucket, object, read_lock_guard.take()); debug!(bucket, object, "Lock optimization: released read lock after direct-memory read"); } return Ok(reader); } match codec_streaming_gate.decision { GetCodecStreamingDecision::Use => { match Self::get_object_decode_reader_with_fileinfo( bucket, object, &fi, &files, &disks, self.set_index, self.pool_index, opts.skip_verify_bitrot, object_class.as_str(), size_bucket, codec_streaming_gate.prefer_data_blocks_first_reader_setup, ) .await? { core::io_primitives::GetCodecStreamingReaderBuildOutcome::Reader(stream) => { record_get_codec_streaming_gate_decision( codec_streaming_gate.object_class, GetCodecStreamingDecision::Use, size_bucket, ); record_get_object_reader_path_observation(GET_OBJECT_PATH_CODEC_STREAMING, object_class, size_bucket); let (mut reader, _offset, _length) = GetObjectReader::new(stream, range, &object_info, opts, &h).await?; // Carry the hook probe result so the app layer skips its // now-redundant lookup on the streaming miss path (ODC-16). reader.body_source = body_source; return Ok(finish_set_disk_read_lock( reader, read_lock_guard.take(), lock_optimization_enabled, bucket, object, )); } core::io_primitives::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => { record_get_codec_streaming_gate_decision( codec_streaming_gate.object_class, GetCodecStreamingDecision::Fallback(reason), size_bucket, ); rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); } } } GetCodecStreamingDecision::Fallback(reason) => { record_get_codec_streaming_gate_decision( codec_streaming_gate.object_class, codec_streaming_gate.decision, size_bucket, ); rustfs_io_metrics::record_get_object_codec_streaming_fallback(reason.as_str()); } } record_get_object_reader_path_observation(GET_OBJECT_PATH_LEGACY_DUPLEX, object_class, size_bucket); let duplex_buffer_size = adaptive_duplex_buffer_size(object_info.size); let (rd, wd) = tokio::io::duplex(duplex_buffer_size); debug!(bucket, object, duplex_buffer_size, "Created duplex pipe for object data transfer"); let (mut reader, offset, length) = GetObjectReader::new(Box::new(rd), range, &object_info, opts, &h).await?; // Carry the hook probe result so the app layer skips its now-redundant // lookup on the streaming miss path (ODC-16). reader.body_source = body_source; // let disks = disks.clone(); let bucket = bucket.to_owned(); let object = object.to_owned(); let set_index = self.set_index; let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; if lock_optimization_enabled { release_materialized_read_lock(&bucket, &object, read_lock_guard.take()); debug!(bucket, object, "Lock optimization: released read lock before streaming read"); } // When lock optimization is disabled, keep the read-lock guard in the // task so it lives for the duration of the streaming read. tokio::spawn(async move { let _guard = read_lock_guard; let mut writer = wd; // Do not wrap the entire read+write pipeline in `disk_read_timeout`. // `get_object_with_fileinfo` also waits on `writer`, so an outer timeout // would incorrectly treat downstream backpressure as disk-read latency. // Disk read timeouts must be enforced at the actual disk I/O operations. if let Err(e) = Self::get_object_with_fileinfo( &bucket, &object, offset, length, &mut writer, fi, files, &disks, set_index, pool_index, skip_verify, false, GET_OBJECT_PATH_LEGACY_DUPLEX, object_class.as_str(), size_bucket, ) .await { let reason = classify_storage_error(&e); record_get_object_pipeline_failure(GET_STAGE_EMIT, reason); error!( event = EVENT_SET_DISK_WRITE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket, object, pool_index, set_index, offset, requested_length = length, skip_verify_bitrot = skip_verify, state = "read_pipeline_failed", stage = GET_STAGE_EMIT, reason = reason.as_str(), error = ?e, "Set disk object read pipeline failed" ); }; }); Ok(reader) } async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { self.put_object_with_old_current_size(bucket, object, data, opts) .await .map(|(object_info, _)| object_info) } } impl SetDisks { /// `put_object` plus the destination key's previous current-version size, /// quorum-reduced from the dst `xl.meta` copies `rename_data` reads while /// committing (rustfs/backlog#1009). `None` means unknown (mixed-version /// peers, unparsable metadata, or sub-quorum divergence) — callers must /// fall back to degraded accounting, never assume "absent". The extra /// value is deliberately *not* part of `ObjectInfo`, which feeds S3 /// responses, event payloads, replication, and ILM verbatim. #[tracing::instrument(skip(self, data,))] pub async fn put_object_with_old_current_size( &self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result<(ObjectInfo, Option)> { crate::hp_guard!("SetDisks::put_object"); let storage_class_config = self.storage_class_config_snapshot(); self.invalidate_get_object_metadata_cache(bucket, object).await; let disks = self.get_disks_internal().await; let mut object_lock_guard = None; if opts.http_preconditions.is_some() { if !opts.no_lock { object_lock_guard = Some( self.acquire_write_lock_diag("put_object_precondition", bucket, object) .await?, ); } if let Some(err) = self.check_write_precondition(bucket, object, opts).await { return Err(err); } } let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?; let mut user_defined = opts.user_defined.clone(); if let Some(eval_metadata) = &opts.eval_metadata { for (key, value) in eval_metadata { user_defined.insert(key.clone(), value.clone()); } } if expected_restore_operation_id.is_some() { rustfs_utils::http::metadata_compat::remove_str(&mut user_defined, SUFFIX_RESTORE_OPERATION_ID); } let WriteLayout { data_drives, parity_drives, write_quorum, } = resolve_write_layout( &storage_class_config, self.pool_index, disks.len(), self.default_parity_count, user_defined.get(AMZ_STORAGE_CLASS).map(String::as_str), opts.max_parity, )?; // if filtered_online < write_quorum { // warn!( // "online disk snapshot {} below write quorum {} for {}/{}; returning erasure write quorum error", // filtered_online, write_quorum, bucket, object // ); // return Err(to_object_err(Error::ErasureWriteQuorum, vec![bucket, object])); // } let mut fi = FileInfo::new([bucket, object].join("/").as_str(), data_drives, parity_drives); fi.version_id = { if let Some(ref vid) = opts.version_id { Some(Uuid::parse_str(vid.as_str()).map_err(Error::other)?) } else { None } }; if opts.versioned && fi.version_id.is_none() { fi.version_id = Some(Uuid::new_v4()); } fi.data_dir = Some(Uuid::new_v4()); let parts_metadata = vec![fi.clone(); disks.len()]; let (mut shuffle_disks, mut parts_metadatas) = Self::shuffle_disks_and_parts_metadata(&disks, &parts_metadata, &fi); let tmp_dir = Uuid::new_v4().to_string(); let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap()); let result: Result<(ObjectInfo, Option)> = async { let erasure = erasure_from_file_info(&fi, false)?; let put_object_size = known_put_object_storage_size(data.size()); let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned); let shard_file_size = erasure.shard_file_size(put_object_size); let shard_size = erasure.shard_size(); let writer_setup_stage_start = Instant::now(); let writer_futs: Vec<_> = shuffle_disks .iter() .map(|disk_op| { let tmp_obj = tmp_object.clone(); async move { if let Some(disk) = disk_op && disk.is_online().await { match create_bitrot_writer( is_inline_buffer, Some(disk), RUSTFS_META_TMP_BUCKET, &tmp_obj, shard_file_size, shard_size, HashAlgorithm::HighwayHash256S, ) .await { Ok(writer) => (Some(writer), None), Err(err) => { warn!( event = EVENT_SET_DISK_WRITE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, disk = ?disk, state = "bitrot_writer_skipped", error = ?err, "Set disk bitrot writer skipped" ); (None, Some(err)) } } } else { (None, Some(DiskError::DiskNotFound)) } } }) .collect(); let writer_results = join_all(writer_futs).await; let mut writers = Vec::with_capacity(writer_results.len()); let mut errors = Vec::with_capacity(writer_results.len()); for (w, e) in writer_results { writers.push(w); errors.push(e); } let writer_setup_ms = writer_setup_stage_start.elapsed().as_millis() as u64; rustfs_io_metrics::record_put_object_stage_duration("set_disk_writer_setup", writer_setup_ms as f64); let nil_count = errors.iter().filter(|&e| e.is_none()).count(); if nil_count < write_quorum { error!( event = EVENT_SET_DISK_WRITE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket, object, write_quorum, available_writers = nil_count, state = "write_quorum_unavailable", error = ?errors, "Set disk write quorum unavailable" ); if let Some(write_err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, write_quorum) { return Err(to_object_err(write_err.into(), vec![bucket, object])); } return Err(Error::other(format!("not enough disks to write: {errors:?}"))); } let stream = mem::replace( &mut data.stream, HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, ); let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size); rustfs_io_metrics::record_put_object_path(write_path.metric_label()); let encode_stage_start = Instant::now(); let (reader, w_size) = match write_path { SmallWritePath::Inline => match Arc::new(erasure) .encode_inline_small(stream, &mut writers, write_quorum) .await { Ok((r, w)) => (r, w), Err(e) => { error!("encode_inline_small err {:?}", e); return Err(e.into()); } }, SmallWritePath::SingleBlockNonInline => match Arc::new(erasure) .encode_single_block_non_inline(stream, &mut writers, write_quorum) .await { Ok((r, w)) => (r, w), Err(e) => { error!("encode_single_block_non_inline err {:?}", e); return Err(e.into()); } }, SmallWritePath::PipelineBatchedLarge => { match Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await { Ok((r, w)) => (r, w), Err(e) => { error!("encode_batched err {:?}", e); return Err(e.into()); } } } SmallWritePath::Pipeline => match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { Ok((r, w)) => (r, w), Err(e) => { error!("encode err {:?}", e); return Err(e.into()); } }, }; let encode_ms = encode_stage_start.elapsed().as_millis() as u64; rustfs_io_metrics::record_put_object_stage_duration("set_disk_encode", encode_ms as f64); let _ = mem::replace(&mut data.stream, reader); // if let Err(err) = close_bitrot_writers(&mut writers).await { // error!("close_bitrot_writers err {:?}", err); // } if (w_size as i64) < data.size() { warn!( event = EVENT_SET_DISK_WRITE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket, object, written_size = w_size, expected_size = data.size(), state = "short_write", "Set disk write produced fewer bytes than expected" ); return Err(Error::other(format!( "put_object write size < data.size(), w_size={}, data.size={}", w_size, data.size() ))); } if contains_key_str(&user_defined, SUFFIX_COMPRESSION) { insert_str(&mut user_defined, SUFFIX_COMPRESSION_SIZE, w_size.to_string()); } let index_op = data .stream .try_get_index() .map(crate::io_support::rio::compression_index_storage_bytes); let mut etag = data.stream.try_resolve_etag().unwrap_or_default(); if let Some(ref tag) = opts.preserve_etag { etag = tag.clone(); } user_defined.insert("etag".to_owned(), etag.clone()); if !user_defined.contains_key("content-type") { // get content-type } let mut actual_size = data.actual_size(); if actual_size < 0 { let is_compressed = fi.is_compressed(); if !is_compressed { actual_size = w_size as i64; } } if fi.checksum.is_none() && let Some(content_hash) = data.as_hash_reader().content_hash() { fi.checksum = Some(content_hash.to_bytes(&[])); } if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS) && sc == storageclass::STANDARD { let _ = user_defined.remove(AMZ_STORAGE_CLASS); } let mod_time = if let Some(mod_time) = opts.mod_time { Some(mod_time) } else { Some(OffsetDateTime::now_utc()) }; // Drop any disk whose shard did not fully commit (offline at writer // setup, short write, or a write/shutdown error) so its truncated or // absent shard is not renamed into place and counted toward write // quorum. Otherwise redundancy is inflated: the object claims N good // shards but one is short/corrupt, so a single later disk failure can // drop it below reconstructable quorum (backlog#852 / #799 B3). // `rename_data` re-checks write quorum over the surviving disks and // rolls back if too few remain. let committed_shards = drop_failed_writer_disks(&mut shuffle_disks, &writers); if committed_shards < write_quorum { return Err(Error::other(format!( "put_object write quorum unavailable after encode: {committed_shards} shard(s) committed, need {write_quorum}" ))); } for (i, pfi) in parts_metadatas.iter_mut().enumerate() { pfi.metadata = user_defined.clone(); if is_inline_buffer { if let Some(writer) = writers[i].take() { pfi.data = Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default()); } pfi.set_inline_data(); } pfi.mod_time = mod_time; pfi.size = w_size as i64; pfi.versioned = opts.versioned || opts.version_suspended; pfi.add_object_part(1, etag.clone(), w_size, mod_time, actual_size, index_op.clone(), None); pfi.checksum = fi.checksum.clone(); if opts.data_movement { pfi.set_data_moved(); } } drop(writers); // drop writers to close all files, this is to prevent FileAccessDenied errors when renaming data if !opts.no_lock && object_lock_guard.is_none() { object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?); } // Phase 2 (backlog#899): fence the commit on lock loss. If the refresh // heartbeat has observed a refresh-quorum loss, another writer may have // re-acquired this object's lock; committing now would race a double-write. // Abort with a retryable error *before* rename_data makes the new version // durable — once rename_data returns Ok the write is committed and must // never be aborted (that would violate "a committed write is not lost"). if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { return Err(StorageError::NamespaceLockQuorumUnavailable { mode: "put_object_commit", bucket: bucket.to_string(), object: object.to_string(), required: 1, achieved: 0, }); } self.require_current_restore_operation_id(bucket, object, opts, expected_restore_operation_id, "put_object_commit") .await?; let rename_stage_start = Instant::now(); let (online_disks, _, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data( &shuffle_disks, RUSTFS_META_TMP_BUCKET, tmp_dir.as_str(), &parts_metadatas, bucket, object, write_quorum, ) .await?; let rename_stage_ms = rename_stage_start.elapsed().as_millis() as u64; rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", rename_stage_ms as f64); if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { warn!( event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, stage = "rename_data", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, duration_ms = { rename_stage_ms }, write_quorum, state = "slow", "SetDisk commit tail stage is slow" ); } let mut cleanup_stage_ms: Option = None; if let Some(old_dir) = op_old_dir { let committed_dir = fi.data_dir.unwrap_or_default().to_string(); let cleanup_stage_start = Instant::now(); // backlog#898: reclaiming the dereferenced old data dir is // best-effort and returns a receipt (never `Err`). A failed GC // here must not negate an already-committed, durable write, so we // deliberately do NOT `?`-propagate it into a 503. On residue the // report path emits the leak metric and enqueues a heal. let cleanup = self .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum) .await; let cleanup_ms = cleanup_stage_start.elapsed().as_millis() as u64; cleanup_stage_ms = Some(cleanup_ms); rustfs_io_metrics::record_put_object_stage_duration("set_disk_old_data_cleanup", cleanup_ms as f64); self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup) .await; if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { warn!( event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, stage = "commit_rename_data_dir", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, old_dir = %old_dir, duration_ms = cleanup_ms, write_quorum, state = "slow", "SetDisk commit tail stage is slow" ); } } drop(object_lock_guard); // drop object lock guard to release the lock for (i, op_disk) in online_disks.iter().enumerate() { if let Some(disk) = op_disk && disk.is_online().await { fi = parts_metadatas[i].clone(); break; } } if fi.is_compressed() { record_compression_total_memory(actual_size as u64, w_size as u64).await; } self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state())); fi.is_latest = true; if issue3031_diag_enabled() { let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count(); warn!( target: "rustfs_ecstore::set_disk", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, data_dir = ?fi.data_dir, write_quorum, online_success_count, op_old_dir = ?op_old_dir, "issue3031_put_object_commit_succeeded" ); } let total_commit_tail_ms = rename_stage_start.elapsed().as_millis(); if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS { warn!( event = EVENT_SET_DISK_COMMIT_TAIL_SLOW, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, stage = "put_object_commit_tail", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, duration_ms = total_commit_tail_ms as u64, write_quorum, state = "slow", "SetDisk commit tail is slow" ); } if issue3031_diag_enabled() { warn!( event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket = %bucket, object = %object, write_quorum, write_path = write_path.metric_label(), writer_setup_ms, encode_ms, rename_ms = rename_stage_ms, cleanup_ms = cleanup_stage_ms.unwrap_or_default(), cleanup_present = cleanup_stage_ms.is_some(), commit_tail_ms = total_commit_tail_ms as u64, result = "success", "SetDisk put_object stage summary" ); } Ok(( ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended), old_current_size, )) } .await; if issue3031_diag_enabled() && let Err(err) = &result { let stage_hint = if err.to_string().contains("not enough disks to write") { "writer_setup_or_quorum" } else { "unknown" }; warn!( event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, bucket = %bucket, object = %object, result = "error", stage_hint, error = %err, "SetDisk put_object stage summary" ); } if result.is_ok() { self.invalidate_get_object_metadata_cache(bucket, object).await; } if issue3031_diag_enabled() { warn!( target: "rustfs_ecstore::set_disk", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, result = ?result.as_ref().map(|_| ()).map_err(|err| err.to_string()), "issue3031_put_object_tmp_cleanup_start" ); } if result.is_ok() { // Success path: `rename_data` has already moved the data dir out of // the tmp workspace and removed the (empty) tmp dir where it could, // so this delete_all is a speculative safety net that normally hits // a missing path. It still must run — `rename_data`'s `remove_std` // only removes empty directories and silently ignores failures — // but it has no reason to run on the response path: under // fsync-heavy load the same-disk queueing behind it was measured to // add ~9ms average (p99 77ms) to PUT latency (backlog#924 / HP-3). // If the process dies before the spawned task runs, the stale tmp // entry is reclaimed by cleanup_stale_tmp_objects (24h expiry, // 5-minute background loop). let set_disks = self.clone(); tokio::spawn(async move { if let Err(err) = set_disks.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await { warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data"); } else if issue3031_diag_enabled() { warn!( target: "rustfs_ecstore::set_disk", tmp_dir = %tmp_dir, "issue3031_put_object_tmp_cleanup_done" ); } }); } else { // Failure path (quorum loss / rollback): keep the cleanup inline so // a failed PUT never returns while its tmp shards are still on disk // (state-residue hardening tracked by backlog#864 / backlog#898). if let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await { warn!(tmp_dir = %tmp_dir, error = ?err, "failed to cleanup put_object temporary data"); } else if issue3031_diag_enabled() { warn!( target: "rustfs_ecstore::set_disk", bucket = %bucket, object = %object, tmp_dir = %tmp_dir, "issue3031_put_object_tmp_cleanup_done" ); } } result } } struct TransitionUploadReader { inner: R, consumed: Arc, } impl TransitionUploadReader { fn new(inner: R, consumed: Arc) -> Self { Self { inner, consumed } } } impl AsyncRead for TransitionUploadReader { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { let before = buf.filled().len(); match Pin::new(&mut self.inner).poll_read(cx, buf) { Poll::Ready(Ok(())) => { let read = buf.filled().len() - before; let read = u64::try_from(read).map_err(|_| std::io::Error::other("transition upload read count exceeds u64::MAX"))?; self.consumed .fetch_update(Ordering::Release, Ordering::Relaxed, |consumed| consumed.checked_add(read)) .map_err(|_| std::io::Error::other("transition upload read count overflow"))?; Poll::Ready(Ok(())) } other => other, } } } struct TransitionUploadWriter { inner: W, produced: u64, } impl TransitionUploadWriter { fn new(inner: W) -> Self { Self { inner, produced: 0 } } fn produced(&self) -> u64 { self.produced } } impl AsyncWrite for TransitionUploadWriter { fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { match Pin::new(&mut self.inner).poll_write(cx, buf) { Poll::Ready(Ok(written)) => { let written_u64 = u64::try_from(written) .map_err(|_| std::io::Error::other("transition upload write count exceeds u64::MAX"))?; self.produced = self .produced .checked_add(written_u64) .ok_or_else(|| std::io::Error::other("transition upload write count overflow"))?; Poll::Ready(Ok(written)) } other => other, } } fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.inner).poll_flush(cx) } fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { Pin::new(&mut self.inner).poll_shutdown(cx) } } #[derive(Debug)] pub(crate) struct TransitionUploadFailure { pub(crate) error: StorageError, pub(crate) candidate: Option, } #[derive(Debug, PartialEq, Eq)] pub(crate) struct TransitionUploadCompletion { pub(crate) candidate: TransitionUploadCandidate, pub(crate) produced: u64, pub(crate) consumed: u64, } #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct TransitionUploadCandidate { remote_version: TransitionUploadRemoteVersion, } #[derive(Clone, Debug, PartialEq, Eq)] enum TransitionUploadRemoteVersion { KnownExact(String), KnownUnversioned(String), } impl TransitionUploadCandidate { pub(crate) fn from_put_response(remote_version: String) -> Self { let remote_version = if remote_version.is_empty() { TransitionUploadRemoteVersion::KnownUnversioned(remote_version) } else { TransitionUploadRemoteVersion::KnownExact(remote_version) }; Self { remote_version } } pub(crate) fn remote_version(&self) -> &str { match &self.remote_version { TransitionUploadRemoteVersion::KnownExact(remote_version) | TransitionUploadRemoteVersion::KnownUnversioned(remote_version) => remote_version, } } pub(crate) fn cleanup_version(&self) -> &str { match &self.remote_version { TransitionUploadRemoteVersion::KnownExact(remote_version) => remote_version, TransitionUploadRemoteVersion::KnownUnversioned(_) => "", } } fn cleanup_version_is_exact(&self) -> bool { matches!(&self.remote_version, TransitionUploadRemoteVersion::KnownExact(_)) } } pub(crate) async fn complete_transition_upload( remote_upload: Remote, producer: Producer, expected_size: u64, consumed: Arc, ) -> std::result::Result where Remote: Future>, Producer: Future>, { let producer = std::panic::AssertUnwindSafe(producer).catch_unwind(); let (remote_result, producer_result) = tokio::join!(remote_upload, producer); let remote_version = match remote_result { Ok(remote_version) => remote_version, Err(remote_error) => { let error = match producer_result { Ok(Err(StorageError::Io(producer_error))) if producer_error.kind() == std::io::ErrorKind::BrokenPipe => { StorageError::Io(remote_error) } Ok(Err(producer_error)) => producer_error, Err(_) => StorageError::Unexpected, Ok(Ok(_)) => StorageError::Io(remote_error), }; return Err(TransitionUploadFailure { error, candidate: None }); } }; let candidate = TransitionUploadCandidate::from_put_response(remote_version); let produced = match producer_result { Ok(Ok(produced)) => produced, Ok(Err(error)) => { return Err(TransitionUploadFailure { error, candidate: Some(candidate), }); } Err(_) => { return Err(TransitionUploadFailure { error: StorageError::Unexpected, candidate: Some(candidate), }); } }; let consumed = consumed.load(Ordering::Acquire); if produced != expected_size || consumed != expected_size { let error = if produced < expected_size || consumed < expected_size { StorageError::LessData } else { StorageError::MoreData }; return Err(TransitionUploadFailure { error, candidate: Some(candidate), }); } Ok(TransitionUploadCompletion { candidate, produced, consumed, }) } pub(crate) async fn cleanup_uncommitted_transition_upload( lease: &TierOperationLease, object: &str, cleanup_version: &str, version_id_exact: bool, ) -> std::io::Result { delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, version_id_exact).await } fn log_transition_upload_cleanup_failure(lease: &TierOperationLease, object: &str, cleanup_version: &str, err: &std::io::Error) { warn!( tier = lease.tier_name(), tier_generation = lease.generation(), object, remote_version = cleanup_version, error = ?err, "failed to clean uncommitted transition upload" ); } pub(crate) struct TransitionUploadCleanup { lease: TierOperationLease, object: String, candidate: Option, cleanup_ctx: Arc, cleanup_api: Option>, armed: bool, } impl TransitionUploadCleanup { pub(crate) fn new( lease: TierOperationLease, object: &str, cleanup_ctx: Arc, ) -> Self { Self { lease, object: object.to_string(), candidate: None, cleanup_ctx, cleanup_api: None, armed: true, } } fn cleanup_candidate(&self) -> std::io::Result<&TransitionUploadCandidate> { self.candidate .as_ref() .ok_or_else(|| std::io::Error::other("transition upload cleanup has no confirmed remote candidate")) } pub(crate) async fn cleanup(&mut self) -> std::io::Result { let candidate = self.cleanup_candidate()?; let result = cleanup_uncommitted_transition_upload( &self.lease, &self.object, candidate.cleanup_version(), candidate.cleanup_version_is_exact(), ) .await; match result { Ok(outcome) => { self.armed = false; Ok(outcome) } Err(err) => { log_transition_upload_cleanup_failure(&self.lease, &self.object, candidate.cleanup_version(), &err); Err(err) } } } async fn cleanup_rejected_upload(&mut self, api: Option>) -> std::io::Result<()> { self.cleanup_api = api.clone(); let candidate = self.cleanup_candidate()?; let result = cleanup_rejected_transition_upload_durably( &self.lease, &self.object, candidate.cleanup_version(), candidate.cleanup_version_is_exact(), api, ) .await; if result.is_ok() { self.armed = false; } result } pub(crate) fn disarm(&mut self) { self.armed = false; } } impl Drop for TransitionUploadCleanup { fn drop(&mut self) { if !self.armed { return; } let Some(candidate) = self.candidate.as_ref() else { return; }; let lease = match self.lease.try_clone() { Ok(lease) => lease, Err(err) => { warn!( tier = self.lease.tier_name(), tier_generation = self.lease.generation(), object = self.object, error = ?err, "unable to retain tier lease for cancelled transition cleanup" ); return; } }; let object = self.object.clone(); let cleanup_version = candidate.cleanup_version().to_string(); let version_id_exact = candidate.cleanup_version_is_exact(); let cleanup_api = self.cleanup_api.clone(); let cleanup_ctx = self.cleanup_ctx.clone(); if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { let api = match cleanup_api { Some(api) => Some(api), None => transition_cleanup_store(&cleanup_ctx).await, }; if let Err(err) = cleanup_rejected_transition_upload_durably(&lease, &object, &cleanup_version, version_id_exact, api).await { warn!( tier = lease.tier_name(), tier_generation = lease.generation(), object, remote_version = cleanup_version, error = ?err, "cancelled transition upload was neither deleted nor journaled" ); } }); } } } async fn cleanup_rejected_transition_upload_durably( lease: &TierOperationLease, object: &str, cleanup_version: &str, version_id_exact: bool, api: Option>, ) -> std::io::Result<()> { let journal_entry = Jentry { obj_name: object.to_string(), version_id: cleanup_version.to_string(), tier_name: lease.tier_name().to_string(), backend_identity: Some(lease.backend_identity()), version_id_exact, }; let journal_error = if let Some(api) = api.as_ref() { match persist_tier_delete_journal_entry(api.clone(), &journal_entry).await { Ok(()) => { match cleanup_uncommitted_transition_upload(lease, object, cleanup_version, version_id_exact).await { Ok(_) => { if let Err(err) = remove_tier_delete_journal_entry(api.clone(), &journal_entry).await { warn!( tier = lease.tier_name(), object, error = ?err, "rejected transition upload was deleted but its cleanup journal was retained" ); } } Err(err) => log_transition_upload_cleanup_failure(lease, object, cleanup_version, &err), } return Ok(()); } Err(err) => err, } } else { std::io::Error::other("object store unavailable for rejected transition cleanup journal") }; warn!( tier = lease.tier_name(), object, error = ?journal_error, "failed to persist rejected transition upload cleanup journal" ); let cleanup_error = match cleanup_uncommitted_transition_upload(lease, object, cleanup_version, version_id_exact).await { Ok(_) => return Ok(()), Err(err) => { log_transition_upload_cleanup_failure(lease, object, cleanup_version, &err); err } }; if let Some(api) = api { match persist_tier_delete_journal_entry(api, &journal_entry).await { Ok(()) => return Ok(()), Err(retry_error) => { return Err(std::io::Error::other(format!( "rejected transition upload was neither deleted nor journaled: initial journal error: {journal_error}; cleanup error: {cleanup_error}; journal retry error: {retry_error}" ))); } } } Err(std::io::Error::other(format!( "rejected transition upload was neither deleted nor journaled: journal error: {journal_error}; cleanup error: {cleanup_error}" ))) } async fn transition_cleanup_store(ctx: &Arc) -> Option> { #[cfg(feature = "test-util")] pause_transition_cleanup_store().await; transition_object_store(ctx).await } async fn transition_object_store(ctx: &Arc) -> Option> { if let Some(api) = runtime_sources::object_store_handle().filter(|api| Arc::ptr_eq(&api.ctx, ctx)) { return Some(api); } let metadata_sys = ctx.bucket_metadata_sys()?; let api = metadata_sys.read().await.object_store(); Arc::ptr_eq(&api.ctx, ctx).then_some(api) } fn transition_deployment_id(ctx: &crate::runtime::instance::InstanceContext) -> Result { if let Some(deployment_id) = ctx.deployment_id() { return Ok(deployment_id); } #[cfg(test)] { Ok(Uuid::new_v4()) } #[cfg(not(test))] { Err(Error::other("transition transaction requires initialized deployment id")) } } fn transition_transaction_not_after_unix_nanos() -> Result { let not_after = (OffsetDateTime::now_utc() + time::Duration::days(7)).unix_timestamp_nanos(); i64::try_from(not_after).map_err(|_| Error::other("transition transaction deadline timestamp overflow")) } fn transition_source_version_mode(opts: &ObjectOptions, fi: &FileInfo) -> TransitionSourceVersionMode { let requested_null_version = opts .version_id .as_deref() .and_then(|version_id| Uuid::parse_str(version_id).ok()) .is_some_and(|version_id| version_id.is_nil()); let source_version_id = fi.version_id.filter(|version_id| !version_id.is_nil()); if opts.versioned && (source_version_id.is_some() || (fi.versioned && !requested_null_version)) { TransitionSourceVersionMode::Versioned } else if opts.version_suspended || opts.versioned { TransitionSourceVersionMode::VersionSuspended } else { TransitionSourceVersionMode::Unversioned } } fn transition_source_identity( bucket: &str, object: &str, fi: &FileInfo, opts: &ObjectOptions, stored_etag: &str, ) -> Result { let version_mode = transition_source_version_mode(opts, fi); let mod_time = fi .mod_time .ok_or_else(|| Error::other("transition source identity requires mod_time"))? .unix_timestamp_nanos(); let mod_time_unix_nanos = i64::try_from(mod_time).map_err(|_| Error::other("transition source mod_time timestamp overflow"))?; let version_id = if version_mode == TransitionSourceVersionMode::Versioned { fi.version_id.filter(|version_id| !version_id.is_nil()) } else { None }; let data_dir = fi .data_dir .ok_or_else(|| Error::other("transition source identity requires data_dir"))?; Ok(TransitionSourceIdentity { bucket: bucket.to_string(), object: object.to_string(), version_id, data_dir, mod_time_unix_nanos, size: fi.size, etag: stored_etag.to_string(), version_mode, }) } async fn save_transition_transaction_if_available(api: Option<&Arc>, transaction: &TransitionTransaction) -> Result<()> { if let Some(api) = api { return save_transition_transaction_record(api.clone(), transaction).await; } #[cfg(test)] { Ok(()) } #[cfg(not(test))] { Err(Error::other("transition transaction store is unavailable")) } } async fn advance_and_save_transition_transaction( api: Option<&Arc>, transaction: &mut TransitionTransaction, next: TransitionTransactionState, remote_version: Option, ) -> Result<()> { transaction .advance(transaction.fence(), next, remote_version) .map_err(Error::other)?; save_transition_transaction_if_available(api, transaction).await } async fn delete_transition_transaction_if_available(api: Option<&Arc>, transaction_id: Uuid) -> Result<()> { if let Some(api) = api { return delete_transition_transaction_record(api.clone(), transaction_id).await; } Ok(()) } async fn delete_transition_transaction_after_remote_cleanup( api: Option<&Arc>, transaction_id: Uuid, bucket: &str, object: &str, ) { if let Err(err) = delete_transition_transaction_if_available(api, transaction_id).await { warn!( bucket = bucket, object = object, transaction_id = %transaction_id, error = ?err, "transition remote candidate was cleaned but transaction record cleanup failed" ); } } #[cfg(feature = "test-util")] #[derive(Default)] struct TransitionCleanupStoreBarrierState { arrived: tokio::sync::Notify, release: tokio::sync::Notify, } #[cfg(feature = "test-util")] /// One-shot test barrier placed before transition cleanup resolves its ECStore. pub(crate) struct TransitionCleanupStoreBarrier { state: Arc, } #[cfg(feature = "test-util")] static TRANSITION_CLEANUP_STORE_BARRIER: std::sync::OnceLock>>> = std::sync::OnceLock::new(); #[cfg(feature = "test-util")] impl TransitionCleanupStoreBarrier { /// Install the process-local barrier for the next cleanup-store resolution. pub(crate) fn install() -> Self { let state = Arc::new(TransitionCleanupStoreBarrierState::default()); let mut slot = TRANSITION_CLEANUP_STORE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition cleanup store barrier mutex should not poison"); assert!(slot.is_none(), "transition cleanup store barrier must be installed by one test at a time"); *slot = Some(state.clone()); drop(slot); Self { state } } /// Wait until a transition reaches the cleanup-store resolution boundary. pub(crate) async fn wait_until_paused(&self) { tokio::time::timeout(std::time::Duration::from_secs(30), self.state.arrived.notified()) .await .expect("transition should reach the cleanup store barrier"); } } #[cfg(feature = "test-util")] impl Drop for TransitionCleanupStoreBarrier { fn drop(&mut self) { self.state.release.notify_one(); let mut slot = TRANSITION_CLEANUP_STORE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition cleanup store barrier mutex should not poison"); if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { *slot = None; } } } #[cfg(feature = "test-util")] async fn pause_transition_cleanup_store() { let barrier = TRANSITION_CLEANUP_STORE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition cleanup store barrier mutex should not poison") .take(); if let Some(barrier) = barrier { barrier.arrived.notify_one(); barrier.release.notified().await; } } #[cfg(all(test, feature = "test-util"))] struct TransitionUploadCandidateBarrier { state: Arc, } #[cfg(all(test, feature = "test-util"))] static TRANSITION_UPLOAD_CANDIDATE_BARRIER: std::sync::OnceLock< std::sync::Mutex>>, > = std::sync::OnceLock::new(); #[cfg(all(test, feature = "test-util"))] impl TransitionUploadCandidateBarrier { fn install() -> Self { let state = Arc::new(TransitionCleanupStoreBarrierState::default()); let mut slot = TRANSITION_UPLOAD_CANDIDATE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition upload candidate barrier mutex should not poison"); assert!( slot.is_none(), "transition upload candidate barrier must be installed by one test at a time" ); *slot = Some(state.clone()); drop(slot); Self { state } } async fn wait_until_paused(&self) { tokio::time::timeout(std::time::Duration::from_secs(30), self.state.arrived.notified()) .await .expect("transition should record its remote upload candidate"); } } #[cfg(all(test, feature = "test-util"))] impl Drop for TransitionUploadCandidateBarrier { fn drop(&mut self) { self.state.release.notify_one(); let mut slot = TRANSITION_UPLOAD_CANDIDATE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition upload candidate barrier mutex should not poison"); if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { *slot = None; } } } #[cfg(all(test, feature = "test-util"))] async fn pause_after_transition_upload_candidate_recorded() { let barrier = TRANSITION_UPLOAD_CANDIDATE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition upload candidate barrier mutex should not poison") .take(); if let Some(barrier) = barrier { barrier.arrived.notify_one(); barrier.release.notified().await; } } #[cfg(test)] #[derive(Clone, Copy, PartialEq, Eq)] enum TransitionCommitPause { BeforeLockLost, BeforeLeaseValidation, AfterLeaseValidation, } #[cfg(test)] struct TransitionCommitBarrierState { bucket: String, object: String, pause: TransitionCommitPause, arrived: tokio::sync::Notify, release: tokio::sync::Notify, } #[cfg(test)] struct TransitionCommitBarrier { state: Arc, } #[cfg(test)] static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock>>> = std::sync::OnceLock::new(); #[cfg(test)] impl TransitionCommitBarrier { fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self { Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost) } fn install(bucket: &str, object: &str) -> Self { Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation) } fn install_after_lease_check(bucket: &str, object: &str) -> Self { Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation) } fn install_at(bucket: &str, object: &str, pause: TransitionCommitPause) -> Self { let state = Arc::new(TransitionCommitBarrierState { bucket: bucket.to_string(), object: object.to_string(), pause, arrived: tokio::sync::Notify::new(), release: tokio::sync::Notify::new(), }); let mut slot = TRANSITION_COMMIT_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition commit barrier mutex should not poison"); assert!(slot.is_none(), "transition commit barrier must be installed by one test at a time"); *slot = Some(Arc::clone(&state)); drop(slot); Self { state } } async fn wait_until_paused(&self) { tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) .await .expect("transition should reach the deterministic commit barrier"); } fn release(&self) { self.state.release.notify_one(); } } #[cfg(test)] impl Drop for TransitionCommitBarrier { fn drop(&mut self) { self.state.release.notify_one(); let mut slot = TRANSITION_COMMIT_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition commit barrier mutex should not poison"); if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { *slot = None; } } } #[cfg(test)] async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCommitPause) { let barrier = TRANSITION_COMMIT_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) .lock() .expect("transition commit barrier mutex should not poison") .as_ref() .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) .cloned(); if let Some(barrier) = barrier { barrier.arrived.notify_one(); barrier.release.notified().await; } } fn parse_transition_version_id(remote_version: &str) -> std::result::Result, uuid::Error> { if remote_version.is_empty() { return Ok(None); } Uuid::parse_str(remote_version).map(|version_id| (!version_id.is_nil()).then_some(version_id)) } #[cfg(test)] mod transition_upload_completion_tests { use super::*; fn consumed(bytes: u64) -> Arc { Arc::new(AtomicU64::new(bytes)) } #[tokio::test] async fn rejects_source_errors_at_first_middle_and_last_chunk() { let remote_version = Uuid::nil().to_string(); for consumed_bytes in [0, 512, 1023] { let result = complete_transition_upload( std::future::ready(Ok(remote_version.clone())), std::future::ready(Err(StorageError::FileCorrupt)), 1024, consumed(consumed_bytes), ) .await; let failure = result.expect_err("a source read error must fail the upload completion protocol"); assert!(matches!(failure.error, StorageError::FileCorrupt)); assert_eq!( failure.candidate.as_ref().map(TransitionUploadCandidate::remote_version), Some(remote_version.as_str()) ); } } #[tokio::test] async fn rejects_partial_body_accepted_by_remote() { let failure = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), std::future::ready(Ok(1024)), 1024, consumed(511), ) .await .expect_err("remote success must not hide a partially consumed body"); assert!(matches!(failure.error, StorageError::LessData)); assert!(failure.candidate.is_some()); } #[tokio::test] async fn maps_source_panic_cancel_and_early_close_to_failures() { let panic_failure = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), async { panic!("injected transition producer panic"); #[allow(unreachable_code)] Ok(0) }, 1, consumed(0), ) .await .expect_err("a producer panic must not enter local commit"); assert!(matches!(panic_failure.error, StorageError::Unexpected)); let cancelled = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), std::future::ready(Err(StorageError::OperationCanceled)), 1, consumed(0), ) .await .expect_err("a cancelled producer must not enter local commit"); assert!(matches!(cancelled.error, StorageError::OperationCanceled)); let early_close = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), std::future::ready(Err(StorageError::Io(std::io::Error::new( std::io::ErrorKind::BrokenPipe, "remote reader closed early", )))), 1024, consumed(16), ) .await .expect_err("an early remote close must not enter local commit"); assert!(matches!(early_close.error, StorageError::Io(ref err) if err.kind() == std::io::ErrorKind::BrokenPipe)); } #[tokio::test] async fn rejects_declared_size_mismatches_and_accepts_zero_size() { let shorter = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), std::future::ready(Ok(511)), 512, consumed(511), ) .await .expect_err("a short source must fail the declared-size check"); assert!(matches!(shorter.error, StorageError::LessData)); let longer = complete_transition_upload( std::future::ready(Ok(Uuid::new_v4().to_string())), std::future::ready(Ok(513)), 512, consumed(513), ) .await .expect_err("an oversized source must fail the declared-size check"); assert!(matches!(longer.error, StorageError::MoreData)); let exact_remote_version = Uuid::new_v4().to_string(); let exact = complete_transition_upload( std::future::ready(Ok(exact_remote_version.clone())), std::future::ready(Ok(512)), 512, consumed(512), ) .await .expect("an exact producer and consumer byte count must complete"); assert_eq!(exact.candidate.remote_version(), exact_remote_version); assert_eq!((exact.produced, exact.consumed), (512, 512)); let remote_version = Uuid::nil().to_string(); let result = complete_transition_upload(std::future::ready(Ok(remote_version.clone())), std::future::ready(Ok(0)), 0, consumed(0)) .await .expect("an empty source and empty remote body must complete"); assert_eq!(result.candidate.remote_version(), remote_version); assert_eq!((result.produced, result.consumed), (0, 0)); } #[tokio::test] async fn preserves_remote_error_when_commit_status_is_unknown() { let failure = complete_transition_upload( std::future::ready(Err(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "remote response was lost"))), std::future::ready(Err(StorageError::Io(std::io::Error::new( std::io::ErrorKind::BrokenPipe, "consumer disappeared", )))), 1024, consumed(0), ) .await .expect_err("an unknown remote commit result must fail closed"); assert!(matches!(failure.error, StorageError::Io(ref err) if err.kind() == std::io::ErrorKind::ConnectionReset)); assert!(failure.candidate.is_none(), "unknown remote versions must never enter precise cleanup"); let source_failure = complete_transition_upload( std::future::ready(Err(std::io::Error::new( std::io::ErrorKind::ConnectionReset, "remote rejected the truncated body", ))), std::future::ready(Err(StorageError::FileCorrupt)), 1024, consumed(128), ) .await .expect_err("a source integrity error must survive a concurrent remote failure"); assert!(matches!(source_failure.error, StorageError::FileCorrupt)); assert!(source_failure.candidate.is_none()); } } #[cfg(test)] mod transition_version_id_tests { use super::{TransitionUploadCandidate, parse_transition_version_id}; use uuid::Uuid; #[test] fn normalizes_persisted_unversioned_ids_and_preserves_put_constraints() { assert_eq!(parse_transition_version_id("").expect("empty remote version should be valid"), None); assert_eq!( parse_transition_version_id(&Uuid::nil().to_string()).expect("nil remote version should be valid"), None ); let nil_put_response = Uuid::nil().to_string(); let nil_candidate = TransitionUploadCandidate::from_put_response(nil_put_response.clone()); assert_eq!(nil_candidate.cleanup_version(), nil_put_response); assert!(nil_candidate.cleanup_version_is_exact()); let empty_candidate = TransitionUploadCandidate::from_put_response(String::new()); assert_eq!(empty_candidate.cleanup_version(), ""); assert!(!empty_candidate.cleanup_version_is_exact()); } #[test] fn preserves_valid_remote_id_and_rejects_invalid_text() { let version_id = Uuid::new_v4(); assert_eq!( parse_transition_version_id(&version_id.to_string()).expect("UUID remote version should be valid"), Some(version_id) ); assert_eq!( TransitionUploadCandidate::from_put_response(version_id.to_string()).cleanup_version(), version_id.to_string() ); assert_eq!( TransitionUploadCandidate::from_put_response("opaque-version-token".to_string()).cleanup_version(), "opaque-version-token" ); assert!(parse_transition_version_id("not-a-uuid").is_err()); } } #[async_trait::async_trait] impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { type Error = Error; type ObjectInfo = ObjectInfo; type ObjectOptions = ObjectOptions; type FileInfo = FileInfo; type ObjectToDelete = ObjectToDelete; type DeletedObject = DeletedObject; #[tracing::instrument(skip(self))] async fn copy_object( &self, src_bucket: &str, src_object: &str, dst_bucket: &str, dst_object: &str, src_info: &mut ObjectInfo, src_opts: &ObjectOptions, dst_opts: &ObjectOptions, ) -> Result { if !src_info.metadata_only { if path_join_buf(&[src_bucket, src_object]) != path_join_buf(&[dst_bucket, dst_object]) { return Err(StorageError::NotImplemented); } // Self-copy with a data reader: write tier data back locally (de-tiering). // Handles `mc cp --storage-class STANDARD obj obj` on a transitioned object. if let Some(mut put_reader) = src_info.put_object_reader.take() { return self.put_object(dst_bucket, dst_object, &mut put_reader, dst_opts).await; } // Same-key tiered copy without a pre-fetched reader: fall through to the metadata // path so the caller gets a disk/quorum error rather than NotImplemented. } if path_join_buf(&[src_bucket, src_object]) != path_join_buf(&[dst_bucket, dst_object]) { return Err(StorageError::NotImplemented); } let _lock_guard = if dst_opts.no_lock { None } else { Some( self.acquire_write_lock_diag("copy_object_metadata", dst_bucket, dst_object) .await?, ) }; self.invalidate_get_object_metadata_cache(dst_bucket, dst_object).await; if dst_opts.http_preconditions.is_some() && let Some(err) = self.check_write_precondition(dst_bucket, dst_object, dst_opts).await { return Err(err); } let disks = self.get_disks_internal().await; let (mut metas, errs) = { if let Some(vid) = &src_opts.version_id { Self::read_all_fileinfo(&disks, "", src_bucket, src_object, vid, true, false, false).await? } else { Self::read_all_xl(&disks, src_bucket, src_object, true, false).await } }; let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) { Ok((r, w)) => ( usize::try_from(r) .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?, usize::try_from(w) .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?, ), Err(mut err) => { if err == DiskError::ErasureReadQuorum && !src_bucket.starts_with(RUSTFS_META_BUCKET) && self .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone()) .await .is_ok() { if src_opts.version_id.is_some() { err = DiskError::FileVersionNotFound } else { err = DiskError::FileNotFound } } return Err(to_object_err(err.into(), vec![src_bucket, src_object])); } }; let src_version_id = src_opts.version_id.as_deref().unwrap_or_default(); let (online_disks, mut fi, _) = Self::select_valid_fileinfo(&disks, &metas, &errs, src_version_id, read_quorum, write_quorum) .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; if fi.deleted { if src_opts.version_id.is_none() { return Err(to_object_err(Error::FileNotFound, vec![src_bucket, src_object])); } return Err(to_object_err(Error::MethodNotAllowed, vec![src_bucket, src_object])); } let version_id = { if src_info.version_only { if let Some(vid) = &dst_opts.version_id { Some(Uuid::parse_str(vid)?) } else { Some(Uuid::new_v4()) } } else { src_info.version_id } }; fi.metadata = (*src_info.user_defined).clone(); if let Some(etag) = &src_info.etag { fi.metadata.insert("etag".to_owned(), etag.clone()); } let mod_time = OffsetDateTime::now_utc(); fi.mod_time = Some(mod_time); fi.version_id = version_id; fi.versioned = src_opts.versioned || src_opts.version_suspended; if src_info.version_only { let inline_data = fi.inline_data(); for fi in metas.iter_mut() { if fi.has_valid_erasure_geometry() { fi.metadata = (*src_info.user_defined).clone(); if let Some(etag) = &src_info.etag { fi.metadata.insert("etag".to_owned(), etag.clone()); } fi.mod_time = Some(mod_time); fi.version_id = version_id; fi.versioned = src_opts.versioned || src_opts.version_suspended; if !fi.inline_data() { fi.data = None; } if inline_data { fi.set_inline_data(); } } } Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum) .await .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; } else { self.update_object_meta_with_opts( src_bucket, src_object, fi.clone(), &online_disks, &UpdateMetadataOpts { replace_user_metadata: true, ..Default::default() }, ) .await .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?; } self.invalidate_get_object_metadata_cache(src_bucket, src_object).await; Ok(ObjectInfo::from_file_info( &fi, src_bucket, src_object, src_opts.versioned || src_opts.version_suspended, )) } #[tracing::instrument(skip(self))] async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> { let disks = self.disk_inventory().await; let write_quorum = disks.len() / 2 + 1; let rollback_dir = Uuid::new_v4(); let mut futures = Vec::with_capacity(disks.len()); let mut errs = Vec::with_capacity(disks.len()); for disk in disks.iter() { futures.push(async move { if let Some(disk) = disk { match disk .delete_version( bucket, object, fi.clone(), force_del_marker, DeleteOptions { old_data_dir: Some(rollback_dir), ..Default::default() }, ) .await { Ok(r) => Ok(r), Err(e) => Err(e), } } else { Err(DiskError::DiskNotFound) } }); } let results = join_all(futures).await; for result in results { match result { Ok(_) => { errs.push(None); } Err(e) => { errs.push(Some(e)); } } } let quorum_result = resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object); let should_rollback = quorum_result.is_err(); let mut rollback_futures = Vec::new(); for (index, err) in errs.iter().enumerate() { // backlog#1158: when rolling back, fan the idempotent undo out to every // online disk (each self-decides from its staged backup: restore if the // rollback dir is present, no-op otherwise). This covers a disk that // staged + applied the delete and *then* errored, which the plain // `err.is_some()` skip would leave deleted while its peers were restored. // On success only the successful disks' backup dirs need cleaning; errored // disks' residue is reclaimed by heal/scanner. if !should_rollback && err.is_some() { continue; } let Some(disk) = disks[index].as_ref() else { continue; }; let disk = disk.clone(); let bucket = bucket.to_string(); let object = object.to_string(); let fi = fi.clone(); rollback_futures.push(async move { if should_rollback { if let Err(err) = disk .delete_version( &bucket, &object, fi, force_del_marker, DeleteOptions { undo_write: true, undo_delete: true, old_data_dir: Some(rollback_dir), ..Default::default() }, ) .await { warn!( bucket = %bucket, object = %object, rollback_dir = %rollback_dir, error = ?err, "failed to roll back delete after write quorum failure" ); } } else { let rollback_path = format!("{object}/{rollback_dir}"); if let Err(err) = disk .delete( &bucket, &rollback_path, DeleteOptions { recursive: true, immediate: true, ..Default::default() }, ) .await && err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { warn!( bucket = %bucket, object = %object, rollback_dir = %rollback_dir, error = ?err, "failed to clean delete rollback state after quorum success" ); } } }); } join_all(rollback_futures).await; quorum_result } #[tracing::instrument(skip(self))] async fn delete_objects( &self, bucket: &str, objects: Vec, opts: ObjectOptions, ) -> (Vec, Vec>) { for object in &objects { self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await; } // Default return value let mut del_objects = vec![DeletedObject::default(); objects.len()]; let mut del_errs = Vec::with_capacity(objects.len()); for _ in 0..objects.len() { del_errs.push(None) } // Acquire locks in batch mode (best effort, matching previous behavior) let mut batch = rustfs_lock::BatchLockRequest::new(self.locker_owner.as_str()).with_all_or_nothing(false); let mut unique_objects: HashSet = HashSet::new(); for dobj in &objects { if unique_objects.insert(dobj.object_name.clone()) { batch = batch.add_write_lock(ObjectKey::new(bucket, dobj.object_name.clone())); } } let unique_lock_count = batch.requests.len(); let mut failed_map = HashMap::new(); let mut _local_batch_guards: Vec = Vec::with_capacity(batch.requests.len()); let mut locked_objects = HashSet::new(); let dist_erasure = runtime_sources::setup_is_dist_erasure().await; let mut dist_batch_lock_ids = vec![Vec::new(); self.lockers.len()]; if opts.no_lock { locked_objects = unique_objects; } else if dist_erasure { (failed_map, locked_objects, dist_batch_lock_ids) = self.acquire_dist_delete_object_locks_batch(&batch).await; } else { let batch_result = self.local_lock_manager.acquire_locks_batch(batch).await; _local_batch_guards = batch_result.guards; for key in batch_result.successful_locks { locked_objects.insert(key.object.as_ref().to_string()); } for (key, err) in batch_result.failed_locks { failed_map.insert((key.bucket.as_ref().to_string(), key.object.as_ref().to_string()), format!("{err:?}")); } } if issue3031_diag_enabled() { let failed_lock_count = failed_map.len(); let locked_object_count = locked_objects.len(); let dist_lock_id_count = dist_batch_lock_ids.iter().map(Vec::len).sum::(); warn!( target: "rustfs_ecstore::set_disk", bucket = %bucket, requested_object_count = objects.len(), unique_lock_count, locked_object_count, failed_lock_count, dist_erasure, dist_lock_id_count, failed_objects = ?failed_map.keys().collect::>(), "issue3031_delete_objects_lock_batch_context" ); } // Mark failures for objects that could not be locked for (i, dobj) in objects.iter().enumerate() { if let Some(err) = failed_map.get(&(bucket.to_string(), dobj.object_name.clone())) { del_errs[i] = Some(Error::other(err.to_string())); } } let ver_cfg = BucketVersioningSys::get(bucket).await.unwrap_or_default(); // backlog#929 (HP-8): the per-object stat below exists solely to feed // check_object_lock_delete (#4297). Resolve the bucket lock // configuration once (in-memory cache) and skip the whole stat fanout // for buckets without Object Lock; unknown metadata fails closed and // keeps the stat, so the #4297 protection is preserved verbatim for // every object-lock-enabled bucket. let object_lock_checks_required = object_lock_delete_check_required(metadata_sys::get(bucket).await.ok().as_deref()); let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new(); for (i, dobj) in objects.iter().enumerate() { if del_errs[i].is_some() { continue; } let explicit_null_version = is_explicit_null_version(dobj.version_id); let version_id = delete_file_info_version_id(dobj.version_id); if object_lock_checks_required { let check_opts = ObjectOptions { version_id: version_id.map(|version_id| version_id.to_string()), versioned: ver_cfg.prefix_enabled(dobj.object_name.as_str()), version_suspended: ver_cfg.suspended(), object_lock_delete: opts.object_lock_delete.clone(), no_lock: true, ..Default::default() }; let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await; if gerr.is_none() && let Err(err) = check_object_lock_delete(bucket, &dobj.object_name, &goi, &check_opts).await { del_errs[i] = Some(err); continue; } } let mut vr = FileInfo { name: dobj.object_name.clone(), version_id, idx: i, replication_state_internal: Some(dobj.replication_state()), ..Default::default() }; vr.set_tier_free_version_id(&Uuid::new_v4().to_string()); // Delete // del_objects[i].object_name.clone_from(&vr.name); // del_objects[i].version_id = vr.version_id.map(|v| v.to_string()); if dobj.version_id.is_none() { let (suspended, versioned) = (ver_cfg.suspended(), ver_cfg.prefix_enabled(dobj.object_name.as_str())); if suspended || versioned { vr.mod_time = Some(OffsetDateTime::now_utc()); vr.deleted = true; if versioned { vr.version_id = Some(Uuid::new_v4()); } } } let v = { if vers_map.contains_key(&dobj.object_name) { let val = vers_map.get_mut(&dobj.object_name).unwrap(); val.versions.push(vr.clone()); val.clone() } else { FileInfoVersions { name: vr.name.clone(), versions: vec![vr.clone()], ..Default::default() } } }; if vr.deleted { del_objects[i] = DeletedObject { delete_marker: vr.deleted, delete_marker_version_id: vr.version_id, delete_marker_mtime: vr.mod_time, object_name: vr.name.clone(), replication_state: vr.replication_state_internal.clone(), ..Default::default() } } else { del_objects[i] = DeletedObject { object_name: vr.name.clone(), version_id: if explicit_null_version { Some(Uuid::nil()) } else { vr.version_id }, replication_state: vr.replication_state_internal.clone(), ..Default::default() } } // Only add to vers_map if we hold the lock if locked_objects.contains(&dobj.object_name) { vers_map.insert(&dobj.object_name, v); } } let mut vers = Vec::with_capacity(vers_map.len()); for (_, mut fi_vers) in vers_map { fi_vers.versions.sort_by_key(|a| a.deleted); if let Some(index) = fi_vers.versions.iter().position(|fi| fi.deleted) { fi_vers.versions.truncate(index + 1); } vers.push(fi_vers); } let rollback_dir = Uuid::new_v4(); let disks = self.disks.read().await; let disks = disks.clone(); let mut futures = Vec::with_capacity(disks.len()); // let mut errors = Vec::with_capacity(disks.len()); for disk in disks.iter() { let vers = vers.clone(); futures.push(async move { if let Some(disk) = disk { disk.delete_versions( bucket, vers, DeleteOptions { old_data_dir: Some(rollback_dir), ..Default::default() }, ) .await } else { let mut errs = Vec::with_capacity(vers.len()); for _ in 0..vers.len() { errs.push(Some(DiskError::DiskNotFound)); } errs } }); } let results = join_all(futures).await; let mut del_obj_errs: Vec>> = vec![vec![None; objects.len()]; disks.len()]; // For each disk delete all objects for (disk_idx, errors) in results.into_iter().enumerate() { // Deletion results for all objects for idx in 0..vers.len() { if errors[idx].is_some() { for fi in vers[idx].versions.iter() { del_obj_errs[disk_idx][fi.idx] = errors[idx].clone(); } } } } for obj_idx in 0..objects.len() { let mut disk_err = vec![None; disks.len()]; for disk_idx in 0..disks.len() { if del_obj_errs[disk_idx][obj_idx].is_some() { disk_err[disk_idx] = del_obj_errs[disk_idx][obj_idx].clone(); } } let mut has_err = reduce_write_quorum_errs(&disk_err, OBJECT_OP_IGNORED_ERRS, disks.len() / 2 + 1); if let Some(err) = has_err.clone() { let er = err.into(); if (is_err_object_not_found(&er) || is_err_version_not_found(&er)) && !del_objects[obj_idx].delete_marker { has_err = None; } } else { del_objects[obj_idx].found = true; } if let Some(err) = has_err { if del_objects[obj_idx].version_id.is_some() { del_errs[obj_idx] = Some(to_object_err( err.into(), vec![ bucket, &objects[obj_idx].object_name.clone(), &objects[obj_idx].version_id.unwrap_or_default().to_string(), ], )); } else { del_errs[obj_idx] = Some(to_object_err(err.into(), vec![bucket, &objects[obj_idx].object_name.clone()])); } } } self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut rollback_futures = Vec::new(); for fi_vers in &vers { // delete_versions commits one xl.meta per object group, so rollback must use the same boundary. let should_rollback = fi_vers.versions.iter().any(|fi| del_errs[fi.idx].is_some()); for (disk_idx, disk) in disks.iter().enumerate() { // backlog#1158: on rollback, include every online disk so a disk that // staged + applied the delete and then errored is still restored (the // disk-side undo is idempotent, no-op when nothing was staged). On // success, skip the errored disks and only clean up successful ones. if !should_rollback && fi_vers.versions.iter().any(|fi| del_obj_errs[disk_idx][fi.idx].is_some()) { continue; } let Some(disk) = disk.as_ref() else { continue; }; let disk = disk.clone(); let bucket = bucket.to_string(); let object = fi_vers.name.clone(); let versions = fi_vers.clone(); rollback_futures.push(async move { if should_rollback { let errs = disk .delete_versions( &bucket, vec![versions], DeleteOptions { undo_write: true, undo_delete: true, old_data_dir: Some(rollback_dir), ..Default::default() }, ) .await; if let Some(err) = errs.into_iter().flatten().next() { warn!( bucket = %bucket, object = %object, rollback_dir = %rollback_dir, error = ?err, "failed to roll back batch delete after write quorum failure" ); } } else { let rollback_path = format!("{object}/{rollback_dir}"); if let Err(err) = disk .delete( &bucket, &rollback_path, DeleteOptions { recursive: true, immediate: true, ..Default::default() }, ) .await && err != DiskError::FileNotFound && err != DiskError::VolumeNotFound { warn!( bucket = %bucket, object = %object, rollback_dir = %rollback_dir, error = ?err, "failed to clean batch delete rollback state after quorum success" ); } } }); } } join_all(rollback_futures).await; // TODO: add_partial if dist_erasure { self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await; } for (object, err) in objects.iter().zip(del_errs.iter()) { if err.is_none() { self.invalidate_get_object_metadata_cache(bucket, &object.object_name).await; } } (del_objects, del_errs) } #[tracing::instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result { self.invalidate_get_object_metadata_cache(bucket, object).await; // Guard lock for single object delete let _lock_guard = if (!opts.delete_prefix || opts.delete_prefix_object) && !opts.no_lock { Some(self.acquire_write_lock_diag("delete_object", bucket, object).await?) } else { None }; if opts.delete_prefix { self.delete_prefix(bucket, object) .await .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; self.invalidate_all_get_object_metadata_cache(); return Ok(ObjectInfo::default()); } // TODO: Lifecycle let mut version_found = true; let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await; if let Some(err) = &gerr && goi.name.is_empty() { if should_force_delete_marker_for_missing_version(&opts) { version_found = false; } else { return Err(err.clone()); } } if version_found { check_object_lock_delete(bucket, object, &goi, &opts).await?; } if opts.transition.expire_restored { // Restore-expiry (DeleteRestoredAction / DeleteRestoredVersionAction) // must only drop the local restored copy and strip the x-amz-restore // headers; the version itself stays transitioned (status=complete) // and keeps serving GETs from the tier. Route it before delete-marker // resolution and replication dispatch: a delete marker would hide the // version, a replicated delete would remove it on the target, and a // free-version record would schedule remote tier cleanup. if !version_found { return Err(gerr.unwrap_or_else(|| StorageError::ObjectNotFound(bucket.to_string(), object.to_string()))); } let dfi = FileInfo { name: object.to_string(), version_id: goi.version_id, mod_time: Some(opts.mod_time.unwrap_or_else(OffsetDateTime::now_utc)), expire_restored: true, ..Default::default() }; self.delete_object_version(bucket, object, &dfi, false) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; self.invalidate_get_object_metadata_cache(bucket, object).await; return Ok(ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended)); } let otd = ObjectToDelete { object_name: object.to_string(), version_id: opts .version_id .clone() .map(|v| Uuid::parse_str(v.as_str()).ok().unwrap_or_default()), ..Default::default() }; let dsc = if should_preserve_delete_replication_state(&opts) { ReplicateDecision::default() } else { ReplicationObjectBridge::check_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await }; if dsc.replicate_any() { opts.set_delete_replication_state(dsc); goi.replication_decision = opts .delete_replication .as_ref() .map(|v| v.replicate_decision_str.clone()) .unwrap_or_default(); } let (mark_delete, mut delete_marker) = resolve_delete_version_state(&opts, &goi, version_found); let mod_time = if let Some(mt) = opts.mod_time { mt } else { OffsetDateTime::now_utc() }; let find_vid = Uuid::new_v4(); if mark_delete && (opts.versioned || opts.version_suspended) { if !delete_marker { delete_marker = opts.version_suspended && opts.version_id.is_none(); } let mut fi = FileInfo { name: object.to_string(), deleted: delete_marker, mark_deleted: mark_delete, mod_time: Some(mod_time), replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta), ..Default::default() // TODO: Transition }; fi.set_tier_free_version_id(&find_vid.to_string()); if opts.skip_free_version { fi.set_skip_tier_free_version(); } fi.version_id = if let Some(vid) = opts.version_id.as_ref() { Some(Uuid::parse_str(vid.as_str())?) } else if opts.versioned { Some(Uuid::new_v4()) } else { None }; self.delete_object_version(bucket, object, &fi, should_force_delete_marker_for_missing_version(&opts)) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; let disks = self.disk_inventory().await; self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); oi.replication_decision = goi.replication_decision; self.invalidate_get_object_metadata_cache(bucket, object).await; return Ok(oi); } // Create a single object deletion request let mut dfi = FileInfo { name: object.to_string(), version_id: opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok()), mark_deleted: mark_delete, deleted: delete_marker, mod_time: Some(mod_time), replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta), ..Default::default() }; dfi.set_tier_free_version_id(&find_vid.to_string()); if opts.skip_free_version { dfi.set_skip_tier_free_version(); } self.delete_object_version(bucket, object, &dfi, opts.delete_marker) .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; let disks = self.disk_inventory().await; self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended); obj_info.size = goi.size; self.invalidate_get_object_metadata_cache(bucket, object).await; Ok(obj_info) } #[tracing::instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { crate::hp_guard!("SetDisks::get_object_info"); // Acquire a shared read-lock to protect consistency during info fetch let _read_lock_guard = if !opts.no_lock { Some(self.acquire_read_lock_diag("get_object_info", bucket, object).await?) } else { None }; // 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 .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); Ok(oi) } #[tracing::instrument(skip(self))] async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { if let Err(e) = rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options( bucket.to_string(), Some(object.to_string()), false, Some(HealChannelPriority::Normal), Some(self.pool_index), Some(self.set_index), )) .await { warn!( bucket, object, version_id, error = %e, "Failed to enqueue heal request for partial object" ); } Ok(()) } #[tracing::instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.invalidate_get_object_metadata_cache(bucket, object).await; // Guard lock for metadata update let _lock_guard = if !opts.no_lock { Some(self.acquire_write_lock_diag("put_object_metadata", bucket, object).await?) } else { None }; let disks = self.get_disks_internal().await; let (metas, errs) = { if let Some(version_id) = &opts.version_id { Self::read_all_fileinfo(&disks, "", bucket, object, version_id.to_string().as_str(), false, false, false).await? } else { Self::read_all_xl(&disks, bucket, object, false, false).await } }; let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) { Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum), Err(mut err) => { if err == DiskError::ErasureReadQuorum && !bucket.starts_with(RUSTFS_META_BUCKET) && self .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone()) .await .is_ok() { if opts.version_id.is_some() { err = DiskError::FileVersionNotFound } else { err = DiskError::FileNotFound } } return Err(to_object_err(err.into(), vec![bucket, object])); } }; let read_quorum = usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?; let write_quorum = usize::try_from(write_quorum) .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?; let version_id = opts.version_id.as_deref().unwrap_or_default(); let (online_disks, mut fi, _) = Self::select_valid_fileinfo(&disks, &metas, &errs, version_id, read_quorum, write_quorum) .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; if fi.deleted { return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object])); } let obj_info = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); check_object_lock_retention_update(bucket, object, &obj_info, opts)?; for (k, v) in obj_info.user_defined.iter() { fi.metadata.insert(k.clone(), v.clone()); } if let Some(mt) = &opts.eval_metadata { for (k, v) in mt { fi.metadata.insert(k.clone(), v.clone()); } } if opts.mod_time.is_some() { fi.mod_time = opts.mod_time; } if let Some(ref version_id) = opts.version_id { fi.version_id = Uuid::parse_str(version_id).ok(); } self.update_object_meta(bucket, object, fi.clone(), &online_disks) .await .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?; self.invalidate_get_object_metadata_cache(bucket, object).await; Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } #[tracing::instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let oi = self.get_object_info(bucket, object, opts).await?; Ok((*oi.user_tags).clone()) } #[tracing::instrument(level = "debug", skip(self))] async fn transition_object(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { let tier_config_mgr = self.ctx.tier_config_mgr(); let tgt_client = match TierConfigMgr::acquire_operation_lease(&tier_config_mgr, &opts.transition.tier).await { Ok(client) => client, Err(err) => { return Err(Error::other(format!("remote tier error: {err}"))); } }; // Acquire write-lock early; hold for the whole transition operation scope // if !opts.no_lock { // let guard_opt = self // .namespace_lock // .lock_guard(object, &self.locker_owner, Duration::from_secs(5), Duration::from_secs(10)) // .await?; // if guard_opt.is_none() { // return Err(Error::other("can not get lock. please retry".to_string())); // } // _lock_guard = guard_opt; // } let (mut fi, meta_arr, online_disks) = self.get_object_fileinfo(bucket, object, opts, true, false).await?; /*if err != nil { return Err(to_object_err(err, vec![bucket, object])); }*/ /*if fi.deleted { if opts.version_id.is_none() { return Err(to_object_err(DiskError::FileNotFound, vec![bucket, object])); } return Err(to_object_err(ERR_METHOD_NOT_ALLOWED, vec![bucket, object])); }*/ // Normalize ETags by removing quotes before comparison (PR #592 compatibility) let transition_etag = rustfs_utils::path::trim_etag(&opts.transition.etag); let stored_etag = rustfs_utils::path::trim_etag(&get_raw_etag(&fi.metadata)); if let Some(mod_time1) = opts.mod_time { if let Some(mod_time2) = fi.mod_time.as_ref() { if mod_time1.unix_timestamp() != mod_time2.unix_timestamp() || (!transition_etag.is_empty() && transition_etag != stored_etag) { return Err(to_object_err(Error::other(DiskError::FileNotFound), vec![bucket, object])); } } else { return Err(Error::other("mod_time 2 error.".to_string())); } } else { return Err(Error::other("mod_time 1 error.".to_string())); } if fi.transition_status == TRANSITION_COMPLETE { return Ok(()); } /*if fi.xlv1 { if let Err(err) = self.heal_object(bucket, object, "", &HealOpts {no_lock: true, ..Default::default()}) { return err.expect("err"); } (fi, meta_arr, online_disks) = self.get_object_fileinfo(&bucket, &object, &opts, true, false); if err != nil { return to_object_err(err, vec![bucket, object]); } }*/ let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); let transaction_api = transition_object_store(&self.ctx).await; let mut transaction = TransitionTransaction::new(TransitionTransactionInit { deployment_id: transition_deployment_id(&self.ctx)?, transaction_id: Uuid::new_v4(), owner_epoch: Uuid::new_v4(), write_id: Uuid::new_v4(), source: transition_source_identity(bucket, object, &fi, opts, &stored_etag)?, tier_name: opts.transition.tier.clone(), backend_fingerprint: tgt_client.backend_identity(), not_after_unix_nanos: transition_transaction_not_after_unix_nanos()?, }) .map_err(Error::other)?; save_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await?; let transaction_id = transaction.transaction_id; let dest_obj = transaction.remote_object.clone(); let mut transition_meta = (*oi.user_defined).clone(); transition_meta.insert("name".to_string(), object.to_string()); if let Some(content_type) = oi.content_type.as_ref().filter(|value| !value.is_empty()) { transition_meta.insert(CONTENT_TYPE.to_ascii_lowercase(), content_type.clone()); } for header in [ CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_DISPOSITION, CACHE_CONTROL, EXPIRES, X_AMZ_OBJECT_LOCK_MODE.as_str(), X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str(), X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str(), ] { if let Some(value) = fi.metadata.lookup(header).filter(|value| !value.is_empty()) { transition_meta.insert(header.to_ascii_lowercase(), value.to_string()); } } let expected_size = u64::try_from(fi.size).map_err(|_| StorageError::FileCorrupt)?; let (pr, pw) = tokio::io::duplex(fi.erasure.block_size); let consumed = Arc::new(AtomicU64::new(0)); let reader = ReaderImpl::ObjectBody(GetObjectReader { stream: Box::new(TransitionUploadReader::new(pr, Arc::clone(&consumed))), object_info: oi, buffered_body: None, body_source: GetObjectBodySource::Unprobed, }); let cloned_bucket = bucket.to_string(); let cloned_object = object.to_string(); let cloned_fi = fi.clone(); let set_index = self.set_index; let pool_index = self.pool_index; let skip_verify = opts.skip_verify_bitrot; let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(cloned_fi.size); let producer = async move { let mut writer = TransitionUploadWriter::new(pw); Self::get_object_with_fileinfo( &cloned_bucket, &cloned_object, 0, cloned_fi.size, &mut writer, cloned_fi, meta_arr, &online_disks, set_index, pool_index, skip_verify, false, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, metrics_size_bucket, ) .await?; writer.shutdown().await?; Ok(writer.produced()) }; let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj, self.ctx.clone()); let remote_upload = { let lease = &upload_cleanup.lease; let recorded_candidate = &mut upload_cleanup.candidate; let remote_object = &dest_obj; async move { let remote_version = lease.put_with_meta(remote_object, reader, fi.size, transition_meta).await?; *recorded_candidate = Some(TransitionUploadCandidate::from_put_response(remote_version.clone())); #[cfg(all(test, feature = "test-util"))] pause_after_transition_upload_candidate_recorded().await; Ok(remote_version) } }; let rv = complete_transition_upload(remote_upload, producer, expected_size, consumed).await; let candidate = match rv { Ok(completion) => completion.candidate, Err(failure) => { if failure.candidate.is_some() { let cleanup_api = transition_cleanup_store(&self.ctx).await; if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await { return Err(StorageError::Io(std::io::Error::other(format!( "{}; rejected remote upload cleanup failed: {cleanup_err}", failure.error )))); } delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(failure.error); } }; if let Err(err) = upload_cleanup.lease.validate_remote_version_id(candidate.remote_version()) { let cleanup_api = transition_cleanup_store(&self.ctx).await; if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await { return Err(StorageError::Io(std::io::Error::other(format!( "{err}; rejected remote upload cleanup failed: {cleanup_err}" )))); } delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await; return Err(err.into()); } if let Err(err) = advance_and_save_transition_transaction( transaction_api.as_ref(), &mut transaction, TransitionTransactionState::Uploaded, Some(TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string())), ) .await { let cleanup_api = transition_cleanup_store(&self.ctx).await; if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await { return Err(StorageError::Io(std::io::Error::other(format!( "{err}; uploaded transition transaction persist failed and cleanup failed: {cleanup_err}" )))); } delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await; return Err(err); } let transition_version_id = match parse_transition_version_id(candidate.remote_version()) { Ok(version_id) => version_id, Err(err) => { if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(err.into()); } }; let mut commit_opts = opts.clone(); commit_opts.no_lock = true; commit_opts.metadata_cache_safe = false; let transition_lock_guard = if opts.no_lock { None } else { match self.acquire_write_lock_diag("transition_object_commit", bucket, object).await { Ok(guard) => Some(guard), Err(err) => { if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup( transaction_api.as_ref(), transaction_id, bucket, object, ) .await; } return Err(err); } } }; self.invalidate_get_object_metadata_cache(bucket, object).await; let current = self.get_object_fileinfo(bucket, object, &commit_opts, true, false).await; let (mut current_fi, _, _) = match current { Ok(current) => current, Err(err) => { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(err); } }; let source_matches = current_fi.version_id == fi.version_id && current_fi.data_dir == fi.data_dir && current_fi.mod_time == fi.mod_time && current_fi.size == fi.size && rustfs_utils::path::trim_etag(&get_raw_etag(¤t_fi.metadata)) == stored_etag; if current_fi.transition_status == TRANSITION_COMPLETE || !source_matches { let already_transitioned = current_fi.transition_status == TRANSITION_COMPLETE; drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } if already_transitioned { return Ok(()); } return Err(to_object_err(Error::other(DiskError::FileNotFound), vec![bucket, object])); } current_fi.transition_status = TRANSITION_COMPLETE.to_string(); current_fi.transitioned_objname = dest_obj; current_fi.transition_tier = opts.transition.tier.clone(); current_fi.transition_version_id = transition_version_id; rustfs_utils::http::metadata_compat::insert_str( &mut current_fi.metadata, rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID, rustfs_utils::crypto::hex(upload_cleanup.lease.backend_identity()), ); fi = current_fi; let event_name = EventName::LifecycleTransition.as_str(); #[cfg(test)] pause_transition_commit(bucket, object, TransitionCommitPause::BeforeLockLost).await; if transition_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(StorageError::NamespaceLockQuorumUnavailable { mode: "transition_object_commit", bucket: bucket.to_string(), object: object.to_string(), required: 1, achieved: 0, }); } #[cfg(test)] pause_transition_commit(bucket, object, TransitionCommitPause::BeforeLeaseValidation).await; if !upload_cleanup.lease.is_current_generation() { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(Error::other("remote tier configuration changed during transition")); } #[cfg(test)] pause_transition_commit(bucket, object, TransitionCommitPause::AfterLeaseValidation).await; if let Err(err) = advance_and_save_transition_transaction( transaction_api.as_ref(), &mut transaction, TransitionTransactionState::LocalCommitStarted, None, ) .await { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) .await; } return Err(err); } upload_cleanup.disarm(); if let Err(err) = self.delete_object_version(bucket, object, &fi, false).await { warn!( bucket = bucket, object = object, error = ?err, "transition remote upload completed but local commit failed" ); self.invalidate_get_object_metadata_cache(bucket, object).await; drop(transition_lock_guard); return Err(err); } match transaction.advance(transaction.fence(), TransitionTransactionState::Committed, None) { Ok(_) => { if let Err(err) = save_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await { warn!( bucket = bucket, object = object, transaction_id = %transaction_id, error = ?err, "transition committed locally but transaction committed-state persist failed" ); } else if let Err(err) = delete_transition_transaction_if_available(transaction_api.as_ref(), transaction_id).await { warn!( bucket = bucket, object = object, transaction_id = %transaction_id, error = ?err, "transition committed locally but transaction cleanup failed" ); } } Err(err) => { warn!( bucket = bucket, object = object, transaction_id = %transaction_id, error = ?err, "transition committed locally but transaction committed-state advance failed" ); } } // delete_object_version persisted transition_status=complete and freed the // local data, but does not touch the GET metadata cache. Drop any cached // pre-transition entry so a late duplicate transition task (or a plain GET) // re-reads the fresh state; a stale hit here defeats the TRANSITION_COMPLETE // early-return above and streams the already-deleted local data to the // remote tier again (rustfs/rustfs#4827). self.invalidate_get_object_metadata_cache(bucket, object).await; drop(transition_lock_guard); let disks = self.disk_inventory().await; self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks); for disk in disks.iter() { if let Some(disk) = disk { continue; } let _ = self .add_partial(bucket, object, opts.version_id.as_deref().unwrap_or_default()) .await; break; } let obj_info = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); send_event(EventArgs { event_name: event_name.to_string(), bucket_name: bucket.to_string(), object: obj_info, user_agent: "Internal: [ILM-Transition]".to_string(), host: runtime_sources::default_local_node_name(), ..Default::default() }); //let tags = opts.lifecycle_audit_event.tags(); //auditLogLifecycle(ctx, objInfo, ILMTransition, tags, traceFn) Ok(()) } #[tracing::instrument(level = "debug", skip(self))] async fn restore_transitioned_object(self: Arc, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { // Acquire write-lock early for the restore operation // if !opts.no_lock { // let guard_opt = self // .namespace_lock // .lock_guard(object, &self.locker_owner, Duration::from_secs(5), Duration::from_secs(10)) // .await?; // if guard_opt.is_none() { // return Err(Error::other("can not get lock. please retry".to_string())); // } // _lock_guard = guard_opt; // } let self_ = self.clone(); let set_restore_header_fn = async move |oi: &mut ObjectInfo, rerr: Option| -> Result<()> { if rerr.is_none() { return Ok(()); } self.update_restore_metadata(bucket, object, oi, opts).await?; Err(rerr.unwrap()) }; let mut oi = ObjectInfo::default(); let fi = self_.clone().get_object_fileinfo(bucket, object, opts, true, false).await; 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?; 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)?; } let mut ropts = put_restore_opts(bucket, object, &opts.transition.restore_request, &oi).await?; if let Some(expected_operation_id) = expected_operation_id { rustfs_utils::http::metadata_compat::insert_str( &mut ropts.user_defined, SUFFIX_RESTORE_OPERATION_ID, expected_operation_id.to_string(), ); } let restore_commit_metadata = if let Some(expected_operation_id) = expected_operation_id { let mut metadata = HashMap::new(); metadata.insert(X_AMZ_RESTORE.as_str().to_string(), "ongoing-request=\"false\"".to_string()); rustfs_utils::http::metadata_compat::insert_str( &mut metadata, SUFFIX_RESTORE_OPERATION_ID, expected_operation_id.to_string(), ); metadata } else { HashMap::new() }; // The restore copy-back re-writes this same object via put_object / // new_multipart_upload / complete_multipart_upload, each of which takes // the object write lock in its commit phase. The caller // (handle_restore_transitioned_object, #4877) already holds that write // lock for the whole restore and forwards no_lock=true, so the inner // writes must inherit it or they self-deadlock on the lock we already // hold and time out. put_restore_opts builds fresh options that default // no_lock=false, so propagate it explicitly here. ropts.no_lock = opts.no_lock; if oi.parts.len() == 1 { let mut opts = opts.clone(); opts.part_number = Some(1); let rs: Option = None; let gr = get_transitioned_object_reader_with_tier_manager( bucket, object, &rs, &HeaderMap::new(), &oi, &opts, &self_.ctx.tier_config_mgr(), ) .await; if let Err(err) = gr { return set_restore_header_fn(&mut oi, Some(to_object_err(err.into(), vec![bucket, object]))).await; } let gr = gr?; let reader = BufReader::new(gr.stream); let hash_reader = HashReader::from_stream(reader, gr.object_info.size, gr.object_info.size, None, None, false)?; let mut p_reader = PutObjReader::new(hash_reader); return match self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await { Ok(restored_info) => { send_event(EventArgs { event_name: EventName::ObjectRestoreCompleted.as_str().to_string(), bucket_name: bucket.to_string(), object: restored_info, user_agent: "Internal: [Restore-Completed]".to_string(), host: runtime_sources::default_local_node_name(), ..Default::default() }); Ok(()) } Err(err) => set_restore_header_fn(&mut oi, Some(to_object_err(err, vec![bucket, object]))).await, }; } let res = self_.clone().new_multipart_upload(bucket, object, &ropts).await?; //if err != nil { // return set_restore_header_fn(&mut oi, err).await; //} let mut uploaded_parts: Vec = vec![]; let parts = Arc::clone(&oi.parts); let mut part_offset: i64 = 0; for part_info in parts.iter() { let mut part_opts = opts.clone(); part_opts.part_number = Some(part_info.number); if part_info.actual_size <= 0 { return set_restore_header_fn( &mut oi, Some(Error::other(format!("invalid multipart restore part size {}", part_info.actual_size))), ) .await; } let part_end = match part_offset.checked_add(part_info.actual_size - 1) { Some(end) => end, None => { return set_restore_header_fn( &mut oi, Some(Error::other("multipart restore part range overflow".to_string())), ) .await; } }; let rs = Some(HTTPRangeSpec { is_suffix_length: false, start: part_offset, end: part_end, }); part_offset = match part_end.checked_add(1) { Some(next) => next, None => { return set_restore_header_fn( &mut oi, Some(Error::other("multipart restore part offset overflow".to_string())), ) .await; } }; let gr = match get_transitioned_object_reader_with_tier_manager( bucket, object, &rs, &HeaderMap::new(), &oi, &part_opts, &self_.ctx.tier_config_mgr(), ) .await { Ok(reader) => reader, Err(err) => { return set_restore_header_fn(&mut oi, Some(StorageError::Io(err))).await; } }; let reader = BufReader::new(gr.stream); let hash_reader = HashReader::from_stream(reader, part_info.actual_size, part_info.actual_size, None, None, false)?; let mut p_reader = PutObjReader::new(hash_reader); let p_info = self_ .clone() .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &ObjectOptions::default()) .await?; //if let Err(err) = p_info { // return set_restore_header_fn(&mut oi, err).await; //} if p_info.size as i64 != part_info.actual_size { return set_restore_header_fn( &mut oi, Some(Error::other(ObjectApiError::InvalidObjectState(GenericError { bucket: bucket.to_string(), object: object.to_string(), ..Default::default() }))), ) .await; } uploaded_parts.push(CompletePart { part_num: p_info.part_num, etag: p_info.etag, checksum_crc32: None, checksum_crc32c: None, checksum_sha1: None, checksum_sha256: None, checksum_crc64nvme: None, }); } let restored_info = match self_ .clone() .complete_multipart_upload( bucket, object, &res.upload_id, uploaded_parts, &ObjectOptions { mod_time: oi.mod_time, version_id: oi.version_id.map(|version| version.to_string()), user_defined: restore_commit_metadata, // Inherit the restore write lock (see ropts.no_lock above): // the commit phase re-acquires this object's write lock. no_lock: opts.no_lock, ..Default::default() }, ) .await { Ok(info) => info, Err(err) => return set_restore_header_fn(&mut oi, Some(err)).await, }; send_event(EventArgs { event_name: EventName::ObjectRestoreCompleted.as_str().to_string(), bucket_name: bucket.to_string(), object: restored_info, user_agent: "Internal: [Restore-Completed]".to_string(), host: runtime_sources::default_local_node_name(), ..Default::default() }); Ok(()) } #[tracing::instrument(level = "debug", skip(self))] async fn put_object_tags(&self, bucket: &str, object: &str, tags: &str, opts: &ObjectOptions) -> Result { // Acquire write-lock for tag update (metadata write) // if !opts.no_lock { // let guard_opt = self // .namespace_lock // .lock_guard(object, &self.locker_owner, Duration::from_secs(5), Duration::from_secs(10)) // .await?; // if guard_opt.is_none() { // return Err(Error::other("can not get lock. please retry".to_string())); // } // _lock_guard = guard_opt; // } // Force the full quorum fanout (allow_early_stop=false): `disks` is the // write target below, and an early-stop subset would only carry read // quorum, failing write quorum on update_object_meta (backlog#872). let (mut fi, _, disks) = self.get_object_fileinfo_gated(bucket, object, opts, false, false).await?; fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned()); self.update_object_meta(bucket, object, fi.clone(), disks.as_slice()).await?; Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)) } #[tracing::instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.put_object_tags(bucket, object, "", opts).await } #[tracing::instrument(skip(self))] async fn verify_object_integrity(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { let get_object_reader = ::get_object_reader( self, bucket, object, None, HeaderMap::new(), opts, ) .await?; // Stream to sink to avoid loading entire object into memory during verification let mut reader = get_object_reader.stream; tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?; Ok(()) } } /// Null out any disk whose shard writer failed (or was never created) so its /// truncated/absent shard is not committed by `rename_data`, and return the /// number of disks that still carry a fully-written shard. Extracted for unit /// testing the write-quorum accounting (backlog#852 / #799 B3). fn drop_failed_writer_disks(disks: &mut [Option], writers: &[Option]) -> usize { let mut committed = 0usize; for (slot, writer) in disks.iter_mut().zip(writers.iter()) { if writer.is_none() { *slot = None; } else if slot.is_some() { committed += 1; } } committed } #[cfg(test)] mod erasure_construction_tests { use super::*; use crate::erasure::coding::ErasureConstructionError; use std::error::Error as _; #[test] fn object_file_info_mapping_preserves_construction_error() { let mut fi = FileInfo::new("object", 2, 2); fi.erasure.block_size = 0; let error = match erasure_from_file_info(&fi, false) { Ok(_) => panic!("invalid object erasure metadata must be rejected"), Err(error) => error, }; assert!(error.to_string().contains("block_size must be greater than zero")); let io_source = error.source().expect("StorageError::Io must expose its io::Error source"); let construction_source = io_source .source() .expect("io::Error must expose the erasure construction error"); assert!(construction_source.is::()); } } #[cfg(test)] mod b3_write_quorum_tests { use super::drop_failed_writer_disks; #[test] fn excludes_failed_writers_and_counts_committed() { // Writer 2 failed (short write / error -> nulled). Its disk must be // dropped so the truncated shard is not renamed or counted. let mut disks = vec![Some(0u8), Some(1), Some(2), Some(3)]; let writers = vec![Some(()), Some(()), None, Some(())]; assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); assert_eq!(disks, vec![Some(0), Some(1), None, Some(3)]); } #[test] fn offline_disk_stays_excluded_and_uncounted() { // Disk 1 was offline at setup (disk None, writer None): stays None, not counted. let mut disks = vec![Some(0u8), None, Some(2)]; let writers = vec![Some(()), None, Some(())]; assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 2); assert_eq!(disks, vec![Some(0), None, Some(2)]); } #[test] fn all_writers_ok_keeps_every_disk() { let mut disks = vec![Some(0u8), Some(1), Some(2)]; let writers = vec![Some(()), Some(()), Some(())]; assert_eq!(drop_failed_writer_disks(&mut disks, &writers), 3); assert_eq!(disks, vec![Some(0), Some(1), Some(2)]); } } #[cfg(test)] pub(in crate::set_disk::ops) mod hermetic_set_disks_support { //! Shared hermetic `SetDisks` construction for the ops tests below: the //! `SetDisks` under test is built directly on formatted local disks (same //! pattern as the `ops/locking.rs` tests) so the tests stay hermetic — no //! global local-disk registry, lock clients, or ECStore instance is //! touched, and each test owns its temp workspace. use super::*; use crate::store::init_format::save_format_file; use rustfs_lock::client::LockClient; use tempfile::TempDir; use tokio::sync::RwLock; async fn make_formatted_local_disk_for_pool( disk_idx: usize, pool_index: usize, format: &FormatV3, ) -> (TempDir, Endpoint, DiskStore) { let dir = tempfile::tempdir().expect("tempdir should be created"); let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); endpoint.set_pool_index(pool_index); endpoint.set_set_index(0); endpoint.set_disk_index(disk_idx); let disk = new_disk( &endpoint, &DiskOption { cleanup: false, health_check: false, }, ) .await .expect("local disk should be created"); let mut disk_format = format.clone(); disk_format.erasure.this = format.erasure.sets[0][disk_idx]; save_format_file(&Some(disk.clone()), &Some(disk_format)) .await .expect("format should be saved"); (dir, endpoint, disk) } pub(in crate::set_disk::ops) async fn hermetic_set_disks(disk_count: usize) -> (Vec, Vec, Arc) { hermetic_set_disks_for_pool_with_default_parity(disk_count, 0, disk_count / 2).await } pub(in crate::set_disk::ops) async fn hermetic_set_disks_for_pool_with_default_parity( disk_count: usize, pool_index: usize, default_parity_count: usize, ) -> (Vec, Vec, Arc) { hermetic_set_disks_with_lockers(disk_count, pool_index, default_parity_count, Vec::new()).await } pub(in crate::set_disk::ops) async fn hermetic_set_disks_with_lockers( disk_count: usize, pool_index: usize, default_parity_count: usize, lockers: Vec>, ) -> (Vec, Vec, Arc) { let format = FormatV3::new(1, disk_count); let mut temp_dirs = Vec::with_capacity(disk_count); let mut endpoints = Vec::with_capacity(disk_count); let mut disk_stores = Vec::with_capacity(disk_count); let mut disks = Vec::with_capacity(disk_count); for disk_idx in 0..disk_count { let (temp_dir, endpoint, disk) = make_formatted_local_disk_for_pool(disk_idx, pool_index, &format).await; temp_dirs.push(temp_dir); endpoints.push(endpoint); disk_stores.push(disk.clone()); disks.push(Some(disk)); } let set_disks = SetDisks::new( "hermetic-ops-test-owner".to_string(), Arc::new(RwLock::new(disks)), disk_count, default_parity_count, 0, pool_index, endpoints, format, lockers, ) .await; (temp_dirs, disk_stores, set_disks) } } #[cfg(test)] mod metadata_mutation_generation_tests { use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; async fn put_and_prime( set_disks: &Arc, bucket: &str, object: &str, payload: &[u8], ) -> (ObjectInfo, GetObjectMetadataCacheKey) { let mut reader = PutObjReader::from_vec(payload.to_vec()); let info = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("test object should be written"); set_disks .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) .await .expect("test object metadata should resolve"); let generation = set_disks .get_object_metadata_cache_generation(bucket, object) .expect("metadata cache generation should be active"); let key = GetObjectMetadataCacheKey::new(bucket, object, generation); assert!( set_disks.get_object_metadata_cache.get(&key).await.is_some(), "metadata priming should publish the current generation" ); (info, key) } async fn assert_retired(set_disks: &SetDisks, key: &GetObjectMetadataCacheKey) { set_disks.get_object_metadata_cache.run_pending_tasks().await; assert!( set_disks.get_object_metadata_cache.get(key).await.is_none(), "the mutation must physically retire the prior metadata generation" ); } #[tokio::test] #[serial_test::serial(metadata_cache_invalidation_probe)] async fn metadata_semantic_mutation_generation_matrix_retires_cached_snapshot() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "metadata-mutation-generation-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let put_object = "put-object"; let (_, put_key) = put_and_prime(&set_disks, bucket, put_object, b"initial PUT body").await; let put_probe = MetadataCacheInvalidationProbe::install(bucket, put_object); let mut replacement = PutObjReader::from_vec(b"replacement PUT body".to_vec()); set_disks .put_object(bucket, put_object, &mut replacement, &ObjectOptions::default()) .await .expect("replacement PUT should succeed"); assert_eq!(put_probe.count(), 2, "PUT must invalidate before mutation and after commit"); assert_retired(&set_disks, &put_key).await; drop(put_probe); let delete_object = "delete-object"; let (_, delete_key) = put_and_prime(&set_disks, bucket, delete_object, b"DELETE body").await; let delete_probe = MetadataCacheInvalidationProbe::install(bucket, delete_object); set_disks .delete_object(bucket, delete_object, ObjectOptions::default()) .await .expect("DELETE should succeed"); assert_eq!(delete_probe.count(), 2, "DELETE must invalidate before mutation and after commit"); assert_retired(&set_disks, &delete_key).await; drop(delete_probe); let copy_object = "copy-object"; let (mut copy_info, copy_key) = put_and_prime(&set_disks, bucket, copy_object, b"COPY body").await; copy_info.metadata_only = true; Arc::make_mut(&mut copy_info.user_defined).insert("x-amz-meta-copy".to_string(), "updated".to_string()); let copy_probe = MetadataCacheInvalidationProbe::install(bucket, copy_object); set_disks .copy_object( bucket, copy_object, bucket, copy_object, &mut copy_info, &ObjectOptions::default(), &ObjectOptions::default(), ) .await .expect("metadata COPY should succeed"); assert_eq!( copy_probe.count(), 4, "metadata COPY must retain both outer and update-object-meta fences" ); assert_retired(&set_disks, ©_key).await; drop(copy_probe); let metadata_object = "metadata-object"; let (_, metadata_key) = put_and_prime(&set_disks, bucket, metadata_object, b"metadata body").await; let mut metadata = HashMap::new(); metadata.insert("x-amz-meta-updated".to_string(), "true".to_string()); let metadata_opts = ObjectOptions { eval_metadata: Some(metadata), ..Default::default() }; let metadata_probe = MetadataCacheInvalidationProbe::install(bucket, metadata_object); set_disks .put_object_metadata(bucket, metadata_object, &metadata_opts) .await .expect("metadata PUT should succeed"); assert_eq!( metadata_probe.count(), 4, "metadata PUT must retain both outer and update-object-meta fences" ); assert_retired(&set_disks, &metadata_key).await; } } #[cfg(all(test, feature = "test-util"))] mod transition_commit_failure_tests { use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions}; use crate::disk::DiskAPI as _; use crate::services::tier::test_util::{MockWarmBackend, register_mock_tier}; use crate::services::tier::tier::TierConfigMgr; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use http::HeaderMap; use s3s::dto::RestoreRequest; use tokio::io::AsyncReadExt; fn restore_operation_id_metadata(operation_id: Uuid) -> HashMap { let mut metadata = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut metadata, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_id.to_string(), ); metadata } fn restore_metadata(operation_id: Uuid, ongoing: bool) -> HashMap { let mut metadata = restore_operation_id_metadata(operation_id); metadata.insert(s3s::header::X_AMZ_RESTORE.as_str().to_string(), format!("ongoing-request=\"{ongoing}\"")); metadata } #[tokio::test] #[serial_test::serial] async fn local_commit_failure_returns_error_and_preserves_remote_candidate() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-commit-failure-bucket"; let object = "object.bin"; let payload = b"transition commit failure must preserve the uploaded candidate".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let (fi, parts_metadata, online_disks) = set_disks .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) .await .expect("source metadata should resolve"); let generation = set_disks .get_object_metadata_cache_generation(bucket, object) .expect("metadata cache generation should be active"); let cache_key = GetObjectMetadataCacheKey::new(bucket, object, generation); set_disks .get_object_metadata_cache .insert( cache_key.clone(), Arc::new(GetObjectMetadataCacheEntry { created_at: Instant::now(), fi: fi.clone(), parts_metadata, online_disks, read_quorum: 2, }), ) .await; assert!(set_disks.get_object_metadata_cache.get(&cache_key).await.is_some()); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let barrier = TransitionCommitBarrier::install(bucket, object); let invalidations = MetadataCacheInvalidationProbe::install(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; assert_eq!(invalidations.count(), 1, "precommit revalidation must fence the old metadata once"); let saved_disks = { let mut disks = set_disks.disks.write().await; let saved = std::mem::take(&mut *disks); *disks = vec![None; saved.len()]; saved }; barrier.release(); let result = transition.await.expect("transition task should not panic"); *set_disks.disks.write().await = saved_disks; result.expect_err("local write quorum failure must be returned to the transition worker"); assert_eq!( invalidations.count(), 2, "local commit failure must fence any partially committed metadata again" ); assert_eq!(backend.put_count().await, 1); assert_eq!( backend.remove_count().await, 0, "ambiguous local commit failure must retain the remote candidate" ); assert_eq!(backend.object_count().await, 1); assert!( set_disks.get_object_metadata_cache.get(&cache_key).await.is_none(), "local commit failure must invalidate pre-transition metadata" ); let mut restored = Vec::new(); set_disks .get_object_reader( bucket, object, None, HeaderMap::new(), &ObjectOptions { no_lock: true, ..Default::default() }, ) .await .expect("the local source must remain readable after a fail-before-commit result") .stream .read_to_end(&mut restored) .await .expect("the local source body should drain"); assert_eq!(restored, payload); } #[tokio::test] #[serial_test::serial] async fn partial_local_commit_failure_rolls_back_applied_disks_and_preserves_remote_candidate() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-partial-rollback-bucket"; let object = "object.bin"; let payload = b"partial local commit failure must roll back applied disks".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let barrier = TransitionCommitBarrier::install_after_lease_check(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; assert_eq!(backend.put_count().await, 1, "remote candidate must exist before local commit"); let saved_disks = { let mut disks = set_disks.disks.write().await; let saved = disks.clone(); for disk in disks.iter_mut().skip(2) { *disk = None; } saved }; barrier.release(); 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 .get_object_fileinfo( bucket, object, &ObjectOptions { no_lock: true, ..Default::default() }, true, false, ) .await .expect("rollback should keep the source metadata readable on applied disks") .0; assert_ne!( fi.transition_status, TRANSITION_COMPLETE, "rollback must not leave the applied disks marked as transitioned" ); let mut restored = Vec::new(); set_disks .get_object_reader( bucket, object, None, HeaderMap::new(), &ObjectOptions { no_lock: true, ..Default::default() }, ) .await .expect("rollback should keep the source object readable on applied disks") .stream .read_to_end(&mut restored) .await .expect("the local source body should drain after rollback"); assert_eq!(restored, payload); *set_disks.disks.write().await = saved_disks; assert_eq!( backend.remove_count().await, 0, "ambiguous local commit failure must retain the remote candidate for scanner reconciliation" ); assert_eq!(backend.object_count().await, 1); } #[tokio::test] #[serial_test::serial] async fn local_commit_post_apply_error_rolls_back_the_errored_disk() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-post-apply-rollback-bucket"; let object = "object.bin"; let payload = b"post-apply delete errors must still roll back the changed disk".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let barrier = TransitionCommitBarrier::install_after_lease_check(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; assert_eq!(backend.put_count().await, 1, "remote candidate must exist before local commit"); crate::disk::local::set_delete_version_fail_after_commit(disk_stores[0].path().as_path(), object); let saved_disk = { let mut disks = set_disks.disks.write().await; let saved = disks[1].take(); assert!(saved.is_some(), "test setup should start with the second disk online"); saved }; barrier.release(); let result = transition.await.expect("transition task should not panic"); result.expect_err("post-apply disk error plus one offline disk must fail write quorum"); let errored_disk_fi = disk_stores[0] .read_version("", bucket, object, "", &ReadOptions::default()) .await .expect("rollback must restore metadata on the disk that returned the post-apply error"); assert_ne!( errored_disk_fi.transition_status, TRANSITION_COMPLETE, "rollback must not skip the disk that applied delete_version but returned an error" ); disk_stores[0] .check_parts(bucket, object, &errored_disk_fi) .await .expect("rollback must restore the errored disk's staged data directory"); set_disks.disks.write().await[1] = saved_disk; assert_eq!( backend.remove_count().await, 0, "ambiguous local commit failure must retain the remote candidate for reconciliation" ); assert_eq!(backend.object_count().await, 1); let mut restored = Vec::new(); set_disks .get_object_reader( bucket, object, None, HeaderMap::new(), &ObjectOptions { no_lock: true, ..Default::default() }, ) .await .expect("source must remain readable after rolling back the post-apply error") .stream .read_to_end(&mut restored) .await .expect("the local source body should drain"); assert_eq!(restored, payload); } #[tokio::test] #[serial_test::serial] async fn production_transition_replacement_revokes_commit_and_cleans_with_old_driver() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-generation-fence-bucket"; let object = "object.bin"; let payload = b"generation replacement must fence the local transition commit".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let manager = runtime_sources::global_tier_config_mgr(); let old_backend = register_mock_tier(&manager, &tier_name).await; let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name.clone(), etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let barrier = TransitionCommitBarrier::install(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; assert_eq!(old_backend.put_count().await, 1, "remote candidate must exist before replacement"); let replacement_backend = MockWarmBackend::new(); let replacement_manager = Arc::new(RwLock::new(TierConfigMgr { driver_cache: HashMap::new(), tiers: manager.read().await.tiers.clone(), last_refreshed_at: OffsetDateTime::now_utc(), })); { let mut replacement = replacement_manager.write().await; replacement .tiers .get_mut(&tier_name) .and_then(|tier| tier.minio.as_mut()) .expect("replacement tier should exist") .prefix = "replacement/".to_string(); replacement .install_test_driver(&tier_name, Box::new(replacement_backend)) .expect("replacement driver should install"); } let replacement = match Arc::try_unwrap(replacement_manager) { Ok(manager) => manager.into_inner(), Err(_) => panic!("replacement manager should have one owner"), }; let publish_handle = manager.clone(); let publish_tier = tier_name.clone(); let publish = tokio::spawn( async move { TierConfigMgr::publish_candidate(&publish_handle, replacement, Some(&publish_tier)).await }, ); tokio::time::timeout(Duration::from_secs(5), async { loop { match TierConfigMgr::acquire_operation_lease(&manager, &tier_name).await { Err(err) if err.message.contains("being replaced") => break, Ok(lease) => drop(lease), Err(err) => panic!("unexpected lease error while replacement drains: {err}"), } tokio::task::yield_now().await; } }) .await .expect("replacement should revoke the in-flight generation"); assert!(!publish.is_finished(), "replacement must wait for the production transition lease"); barrier.release(); transition .await .expect("transition task should join") .expect_err("revoked generation must not commit tier-name metadata"); publish .await .expect("replacement task should join") .expect("replacement should finish after old cleanup releases its lease"); assert_eq!( old_backend.remove_count().await, 1, "cancelled transition must clean up with the old driver" ); assert_eq!(old_backend.object_count().await, 0); } #[tokio::test] #[serial_test::serial] async fn failed_restore_cleanup_does_not_overwrite_concurrent_unversioned_put() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "restore-cleanup-cas-bucket"; let object = "object.bin"; let original_payload = b"old transitioned body".repeat(1024); let replacement_payload = b"new visible unversioned body".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(original_payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; set_disks .transition_object( bucket, object, &ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }, ) .await .expect("source object should transition before restore"); let get_barrier = backend.arm_failing_get_barrier().await; let restore_set = Arc::clone(&set_disks); let restore = tokio::spawn(async move { let mut opts = ObjectOptions::default(); opts.transition.restore_request.days = Some(1); restore_set.restore_transitioned_object(bucket, object, &opts).await }); get_barrier.wait_until_paused().await; let mut replacement_reader = PutObjReader::from_vec(replacement_payload.clone()); let replacement = set_disks .put_object(bucket, object, &mut replacement_reader, &ObjectOptions::default()) .await .expect("concurrent unversioned PUT should commit while restore GET is paused"); get_barrier.release(); restore .await .expect("restore task should join") .expect_err("injected tier GET failure should surface"); let visible = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("replacement metadata should remain visible"); assert_eq!(visible.etag, replacement.etag, "stale restore cleanup must not republish the old ETag"); assert_ne!( visible.transitioned_object.status, TRANSITION_COMPLETE, "replacement object must not regain stale transition metadata" ); let mut body = Vec::new(); set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) .await .expect("replacement object should be readable") .stream .read_to_end(&mut body) .await .expect("replacement body should drain"); assert_eq!( body, replacement_payload, "stale restore cleanup must not make the old remote body current again" ); } #[tokio::test] #[serial_test::serial] async fn failed_restore_cleanup_ignores_replaced_operation_id_with_same_identity() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "restore-cleanup-operation-id-bucket"; let object = "object.bin"; let payload = b"same identity restore cleanup must respect operation id".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let operation_a = Uuid::new_v4(); let operation_b = Uuid::new_v4(); let metadata = restore_metadata(operation_a, true); let mut reader = PutObjReader::from_vec(payload); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(metadata), ..Default::default() }, ) .await .expect("restore operation A metadata should be installed"); let stale_operation_a = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("operation A metadata should resolve"); let mut operation_b_metadata = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut operation_b_metadata, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_b.to_string(), ); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(operation_b_metadata), ..Default::default() }, ) .await .expect("same-identity restore operation B should replace A"); let mut expected_operation_a = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut expected_operation_a, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_a.to_string(), ); set_disks .update_restore_metadata( bucket, object, &stale_operation_a, &ObjectOptions { user_defined: expected_operation_a, ..Default::default() }, ) .await .expect("stale operation A cleanup should no-op, not fail"); let current_operation_b = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("operation B metadata should remain readable"); assert!( current_operation_b .user_defined .contains_key(s3s::header::X_AMZ_RESTORE.as_str()), "stale cleanup for operation A must not remove operation B's restore header" ); assert_eq!( rustfs_utils::http::metadata_compat::get_consistent_str( current_operation_b.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ), Some(operation_b.to_string().as_str()), "stale cleanup for operation A must not remove or rewrite operation B" ); let mut expected_operation_b = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut expected_operation_b, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_b.to_string(), ); set_disks .update_restore_metadata( bucket, object, ¤t_operation_b, &ObjectOptions { user_defined: expected_operation_b, ..Default::default() }, ) .await .expect("matching operation B cleanup should remove restore markers"); let cleaned = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("cleaned object metadata should remain readable"); assert!( !cleaned.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()), "matching cleanup must remove the restore header" ); assert!( rustfs_utils::http::metadata_compat::get_consistent_str( cleaned.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ) .is_none(), "matching cleanup must remove the restore operation id" ); } #[tokio::test] #[serial_test::serial] async fn restore_worker_propagates_operation_id_to_final_put_commit() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "restore-worker-commit-operation-id-bucket"; let object = "object.bin"; let payload = b"restore worker must carry operation id to final commit".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; set_disks .transition_object( bucket, object, &ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }, ) .await .expect("source object should transition before restore"); let operation_a = Uuid::new_v4(); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(restore_metadata(operation_a, true)), ..Default::default() }, ) .await .expect("restore operation A metadata should be installed"); let get_barrier = backend.arm_get_barrier().await; let restore_set = Arc::clone(&set_disks); let restore = tokio::spawn(async move { restore_set .restore_transitioned_object( bucket, object, &ObjectOptions { transition: TransitionOptions { restore_request: RestoreRequest { days: Some(1), ..Default::default() }, ..Default::default() }, user_defined: restore_operation_id_metadata(operation_a), ..Default::default() }, ) .await }); get_barrier.wait_until_paused().await; let operation_b = Uuid::new_v4(); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(restore_operation_id_metadata(operation_b)), ..Default::default() }, ) .await .expect("operation B should replace operation A after worker starts tier GET"); get_barrier.release(); restore .await .expect("restore task should join") .expect_err("stale operation A must fail at final PUT commit"); let current = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("operation B metadata should remain visible after stale worker is rejected"); assert_eq!( rustfs_utils::http::metadata_compat::get_consistent_str( current.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ), Some(operation_b.to_string().as_str()), "stale worker must not remove or replace operation B" ); assert_eq!( current.transitioned_object.status, TRANSITION_COMPLETE, "stale worker must not publish its restored local body after operation id replacement" ); let mut body = Vec::new(); set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) .await .expect("original transitioned object should remain readable") .stream .read_to_end(&mut body) .await .expect("original remote body should drain"); assert_eq!(body, payload); } #[tokio::test] #[serial_test::serial] async fn restore_put_commit_rechecks_operation_id_and_strips_internal_marker() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "restore-put-commit-operation-id-bucket"; let object = "object.bin"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let operation_a = Uuid::new_v4(); let operation_a_metadata = restore_metadata(operation_a, true); let mut reader = PutObjReader::from_vec(b"restore source body".repeat(1024)); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(operation_a_metadata.clone()), ..Default::default() }, ) .await .expect("ongoing restore operation A should be installed"); let operation_b = Uuid::new_v4(); let mut operation_b_metadata = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut operation_b_metadata, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_b.to_string(), ); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(operation_b_metadata), ..Default::default() }, ) .await .expect("operation B should replace operation A before final commit"); let mut stale_restore_reader = PutObjReader::from_vec(b"stale A restored body".repeat(1024)); let result = set_disks .put_object( bucket, object, &mut stale_restore_reader, &ObjectOptions { user_defined: operation_a_metadata, ..Default::default() }, ) .await; result.expect_err("stale operation A must not commit after operation B replaces it"); let current = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("operation B metadata should remain current after stale commit is rejected"); assert_eq!( rustfs_utils::http::metadata_compat::get_consistent_str( current.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ), Some(operation_b.to_string().as_str()), "stale commit must not remove or replace operation B" ); let mut matching_restore_reader = PutObjReader::from_vec(b"matching B restored body".repeat(1024)); let operation_b_restore_metadata = restore_metadata(operation_b, false); set_disks .put_object( bucket, object, &mut matching_restore_reader, &ObjectOptions { user_defined: operation_b_restore_metadata, ..Default::default() }, ) .await .expect("matching operation B should be allowed to commit"); let restored = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("restored object should remain readable"); assert!( rustfs_utils::http::metadata_compat::get_consistent_str( restored.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ) .is_none(), "completed restore PUT must not persist the internal operation id" ); } #[tokio::test] #[serial_test::serial] async fn restore_multipart_complete_rechecks_operation_id_and_strips_internal_marker() { use crate::storage_api_contracts::multipart::MultipartOperations as _; let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "restore-multipart-commit-operation-id-bucket"; let object = "object.bin"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let operation_a = Uuid::new_v4(); let operation_a_metadata = restore_metadata(operation_a, true); let mut initial_reader = PutObjReader::from_vec(b"multipart restore source body".repeat(1024)); set_disks .put_object(bucket, object, &mut initial_reader, &ObjectOptions::default()) .await .expect("source object should be written"); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(operation_a_metadata.clone()), ..Default::default() }, ) .await .expect("ongoing multipart restore operation A should be installed"); let upload = set_disks .new_multipart_upload( bucket, object, &ObjectOptions { user_defined: operation_a_metadata.clone(), ..Default::default() }, ) .await .expect("restore multipart upload should be created"); let mut part_reader = PutObjReader::from_vec(b"stale multipart A restored body".repeat(1024)); let part = set_disks .put_object_part(bucket, object, &upload.upload_id, 1, &mut part_reader, &ObjectOptions::default()) .await .expect("restore multipart part should be written"); let operation_b = Uuid::new_v4(); let mut operation_b_metadata = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut operation_b_metadata, rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, operation_b.to_string(), ); set_disks .put_object_metadata( bucket, object, &ObjectOptions { eval_metadata: Some(operation_b_metadata), ..Default::default() }, ) .await .expect("operation B should replace operation A before multipart complete"); let complete_result = set_disks .clone() .complete_multipart_upload( bucket, object, &upload.upload_id, vec![CompletePart { part_num: part.part_num, etag: part.etag.clone(), ..Default::default() }], &ObjectOptions { user_defined: operation_a_metadata, ..Default::default() }, ) .await; complete_result.expect_err("stale operation A must not complete after operation B replaces it"); let current = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("operation B metadata should remain current after stale multipart completion"); assert_eq!( rustfs_utils::http::metadata_compat::get_consistent_str( current.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ), Some(operation_b.to_string().as_str()), "stale multipart completion must not remove or replace operation B" ); let operation_b_restore_metadata = restore_metadata(operation_b, false); let upload = set_disks .new_multipart_upload( bucket, object, &ObjectOptions { user_defined: operation_b_restore_metadata.clone(), ..Default::default() }, ) .await .expect("matching operation B multipart upload should be created"); let mut part_reader = PutObjReader::from_vec(b"matching multipart B restored body".repeat(1024)); let part = set_disks .put_object_part(bucket, object, &upload.upload_id, 1, &mut part_reader, &ObjectOptions::default()) .await .expect("matching restore multipart part should be written"); set_disks .clone() .complete_multipart_upload( bucket, object, &upload.upload_id, vec![CompletePart { part_num: part.part_num, etag: part.etag, ..Default::default() }], &ObjectOptions { user_defined: operation_b_restore_metadata, ..Default::default() }, ) .await .expect("matching operation B should complete"); let restored = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("completed multipart restore should remain readable"); assert!( rustfs_utils::http::metadata_compat::get_consistent_str( restored.user_defined.as_ref(), rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, ) .is_none(), "completed multipart restore must not persist the internal operation id" ); } #[tokio::test] #[serial_test::serial] async fn legacy_reload_rejects_route_change_after_local_transition_commit() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-post-check-fence-bucket"; let object = "object.bin"; let payload = b"the operation lease must cover the local transition commit".repeat(1024); for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.clone()); let original = set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let manager = runtime_sources::global_tier_config_mgr(); let old_backend = register_mock_tier(&manager, &tier_name).await; let old_prefix = manager.read().await.tiers[&tier_name] .minio .as_ref() .expect("old tier route should exist") .prefix .clone(); let old_identity = TierConfigMgr::acquire_operation_lease(&manager, &tier_name) .await .expect("old tier identity should be available") .backend_identity(); let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name.clone(), etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let barrier = TransitionCommitBarrier::install_after_lease_check(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; assert_eq!(old_backend.put_count().await, 1, "remote candidate must exist before replacement"); let replacement_manager = Arc::new(RwLock::new(TierConfigMgr { driver_cache: HashMap::new(), tiers: manager.read().await.tiers.clone(), last_refreshed_at: OffsetDateTime::now_utc(), })); { let mut replacement = replacement_manager.write().await; replacement .tiers .get_mut(&tier_name) .and_then(|tier| tier.minio.as_mut()) .expect("replacement tier should exist") .prefix = "replacement/".to_string(); } let replacement = match Arc::try_unwrap(replacement_manager) { Ok(manager) => manager.into_inner(), Err(_) => panic!("replacement manager should have one owner"), }; let publish_handle = manager.clone(); let publish = tokio::spawn(async move { publish_handle.write().await.publish_legacy_reload(replacement).await }); tokio::time::timeout(Duration::from_secs(5), async { while manager.try_read().is_ok() { tokio::task::yield_now().await; } }) .await .expect("legacy reload should hold the manager guard while the checked generation drains"); assert!( !publish.is_finished(), "replacement must wait until the local transition commit releases its lease" ); barrier.release(); transition .await .expect("transition task should join") .expect("a transition linearized before replacement should commit"); publish .await .expect("replacement task should join") .expect_err("replacement must not rebind a tier name referenced by the committed transition"); assert_eq!(old_backend.remove_count().await, 0, "committed transition data must not be cleaned up"); assert_eq!(old_backend.object_count().await, 1); let transitioned = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect("transitioned metadata should resolve"); let expected_identity = rustfs_utils::crypto::hex(old_identity); let rustfs_identity_key = format!( "{}{}", rustfs_utils::http::metadata_compat::RUSTFS_INTERNAL_PREFIX, rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID ); let minio_identity_key = format!( "{}{}", rustfs_utils::http::metadata_compat::MINIO_INTERNAL_PREFIX, rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID ); assert_eq!( transitioned.user_defined.get(&rustfs_identity_key), Some(&expected_identity), "transition commits must write the RustFS destination identity key" ); assert_eq!( transitioned.user_defined.get(&minio_identity_key), Some(&expected_identity), "transition commits must write the MinIO-compatible destination identity key" ); assert_eq!( rustfs_utils::http::metadata_compat::get_str( &transitioned.user_defined, rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID, ) .as_deref(), Some(expected_identity.as_str()) ); assert_eq!( manager .read() .await .tiers .get(&tier_name) .and_then(|tier| tier.minio.as_ref()) .expect("old tier route should remain configured") .prefix, old_prefix ); let mut restored = Vec::new(); set_disks .get_object_reader( bucket, object, None, HeaderMap::new(), &ObjectOptions { no_lock: true, ..Default::default() }, ) .await .expect("the transitioned object must remain routed through the old generation") .stream .read_to_end(&mut restored) .await .expect("the transitioned object body should drain"); assert_eq!(restored, payload); } } #[cfg(all(test, feature = "test-util"))] mod transition_upload_integrity_tests { use super::hermetic_set_disks_support::{hermetic_set_disks, hermetic_set_disks_with_lockers}; use super::*; use crate::bucket::lifecycle::lifecycle::{TRANSITION_PENDING, TransitionOptions}; use crate::disk::DiskAPI as _; use crate::layout::endpoints::SetupType; use crate::services::tier::test_util::register_mock_tier; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use http::HeaderMap; use rustfs_lock::client::local::LocalClient; use rustfs_lock::{LockClient, LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats}; use std::collections::HashSet; use std::path::PathBuf; use std::sync::atomic::AtomicUsize; struct SetupTypeGuard { previous: SetupType, } impl SetupTypeGuard { async fn switch_to(next: SetupType) -> Self { let previous = runtime_sources::current_setup_type().await; runtime_sources::set_setup_type(next).await; Self { previous } } } impl Drop for SetupTypeGuard { fn drop(&mut self) { let previous = self.previous.clone(); let handle = tokio::runtime::Handle::current(); tokio::task::block_in_place(|| { handle.block_on(async move { runtime_sources::set_setup_type(previous).await; }); }); } } #[derive(Debug)] struct FailingLockClient; #[async_trait::async_trait] impl LockClient for FailingLockClient { async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { Err(LockError::internal("simulated transition commit lock client failure")) } async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { Ok(false) } async fn refresh(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { Ok(false) } async fn force_release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { Ok(false) } async fn check_status(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { Ok(None) } async fn get_stats(&self) -> rustfs_lock::Result { Ok(LockStats::default()) } async fn close(&self) -> rustfs_lock::Result<()> { Ok(()) } async fn is_online(&self) -> bool { false } async fn is_local(&self) -> bool { false } } #[derive(Debug)] struct LockLostRefreshClient { refresh_calls: Arc, active: tokio::sync::Mutex>, } impl LockLostRefreshClient { fn new(refresh_calls: Arc) -> Self { Self { refresh_calls, active: tokio::sync::Mutex::new(HashSet::new()), } } } #[async_trait::async_trait] impl LockClient for LockLostRefreshClient { async fn acquire_lock(&self, request: &LockRequest) -> rustfs_lock::Result { self.active.lock().await.insert(request.lock_id.clone()); Ok(LockResponse::success( LockInfo { id: request.lock_id.clone(), resource: request.resource.clone(), lock_type: request.lock_type, status: rustfs_lock::LockStatus::Acquired, owner: request.owner.clone(), acquired_at: std::time::SystemTime::now(), expires_at: std::time::SystemTime::now() + request.ttl, last_refreshed: std::time::SystemTime::now(), metadata: request.metadata.clone(), priority: request.priority, wait_start_time: None, }, Duration::ZERO, )) } async fn release(&self, lock_id: &LockId) -> rustfs_lock::Result { Ok(self.active.lock().await.remove(lock_id)) } async fn refresh(&self, _lock_id: &LockId) -> rustfs_lock::Result { self.refresh_calls.fetch_add(1, Ordering::SeqCst); Ok(false) } async fn force_release(&self, lock_id: &LockId) -> rustfs_lock::Result { self.release(lock_id).await } async fn check_status(&self, _lock_id: &LockId) -> rustfs_lock::Result> { Ok(None) } async fn get_stats(&self) -> rustfs_lock::Result { Ok(LockStats::default()) } async fn close(&self) -> rustfs_lock::Result<()> { Ok(()) } async fn is_online(&self) -> bool { true } async fn is_local(&self) -> bool { false } } async fn assert_local_source_intact(set_disks: &Arc, bucket: &str, object: &str, payload: &[u8]) { let mut restored = Vec::new(); set_disks .get_object_reader( bucket, object, None, HeaderMap::new(), &ObjectOptions { no_lock: true, ..Default::default() }, ) .await .expect("failed transition must leave the local source readable") .stream .read_to_end(&mut restored) .await .expect("local source should drain after failed transition"); assert_eq!(restored, payload); let (fi, _, _) = set_disks .get_object_fileinfo( bucket, object, &ObjectOptions { no_lock: true, metadata_cache_safe: false, ..Default::default() }, true, false, ) .await .expect("local source metadata should remain available"); assert_ne!(fi.transition_status, TRANSITION_COMPLETE); } async fn write_source( set_disks: &Arc, disk_stores: &[DiskStore], bucket: &str, object: &str, payload: &[u8], ) -> ObjectInfo { for disk in disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(payload.to_vec()); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written") } fn transition_options(original: &ObjectInfo, tier_name: String) -> ObjectOptions { ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name, etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() } } async fn corrupt_shard_at(path: PathBuf, position: ShardCorruptionPosition) { let mut bytes = tokio::fs::read(&path).await.expect("committed shard should be readable"); assert!(!bytes.is_empty(), "committed shard should not be empty"); let offset = match position { ShardCorruptionPosition::First => 0, ShardCorruptionPosition::Middle => bytes.len() / 2, ShardCorruptionPosition::Last => bytes.len() - 1, }; bytes[offset] ^= 0xff; tokio::fs::write(&path, bytes) .await .expect("committed shard corruption should be written"); } #[derive(Clone, Copy, Debug)] enum ShardCorruptionPosition { First, Middle, Last, } impl ShardCorruptionPosition { fn label(self) -> &'static str { match self { ShardCorruptionPosition::First => "first", ShardCorruptionPosition::Middle => "middle", ShardCorruptionPosition::Last => "last", } } } #[tokio::test] #[serial_test::serial] async fn transition_uses_transaction_canonical_remote_object_name() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-transaction-remote-object-bucket"; let object = "object.bin"; let payload = b"transition remote object must be bound to its transaction id".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let remote_version = Uuid::new_v4().to_string(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_remote_version(Some(remote_version.clone())).await; set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect("transition should commit"); let put_versions = backend.put_versions().await; assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate"); let remote_object = &put_versions[0].0; assert!( remote_object.starts_with(crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_PREFIX), "remote object should be transaction-scoped: {remote_object}" ); let (fi, _, _) = set_disks .get_object_fileinfo( bucket, object, &ObjectOptions { no_lock: true, metadata_cache_safe: false, ..Default::default() }, true, false, ) .await .expect("committed transition metadata should be readable"); assert_eq!(fi.transition_status, TRANSITION_COMPLETE); assert_eq!(fi.transitioned_objname, *remote_object); assert_eq!( fi.transition_version_id, Some(Uuid::parse_str(&remote_version).expect("test version id should parse")) ); assert!(backend.contains(remote_object).await, "committed remote object should remain available"); } async fn corrupt_beyond_read_quorum( temp_dirs: &[tempfile::TempDir], bucket: &str, object: &str, data_dir: Uuid, parity_blocks: usize, position: ShardCorruptionPosition, ) { for temp_dir in temp_dirs.iter().take(parity_blocks + 1) { let shard = temp_dir .path() .join(bucket) .join(object) .join(data_dir.to_string()) .join("part.1"); corrupt_shard_at(shard, position).await; } } #[tokio::test] #[serial_test::serial] async fn remote_put_failure_preserves_error_without_cleanup_candidate() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-remote-put-failure-bucket"; let object = "object.bin"; let payload = b"a failed remote PUT must not manufacture a cleanup candidate".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_unreachable(true).await; let error = set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect_err("an unreachable tier must fail the remote PUT"); assert!( matches!(error, StorageError::Io(ref err) if err.kind() == std::io::ErrorKind::ConnectionRefused), "the original remote PUT error must be preserved: {error:?}" ); assert_eq!(backend.remove_count().await, 0, "an unconfirmed candidate must not be cleaned up"); assert_eq!(backend.exact_remove_count(), 0, "an unconfirmed candidate must not reach exact cleanup"); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn partial_remote_acceptance_cleans_exact_candidate_and_preserves_source() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-partial-accept-bucket"; let object = "object.bin"; let payload = vec![0x5a; 2 * 1024 * 1024]; let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let remote_version = Uuid::nil().to_string(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_read_limit(Some(4096)).await; backend.set_put_remote_version(Some(remote_version.clone())).await; let error = set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect_err("accepting only a prefix must fail transition completion"); assert!(matches!(error, StorageError::Io(_) | StorageError::LessData)); let removed_versions = backend.remove_versions().await; assert_eq!(removed_versions.len(), 1); assert_eq!( removed_versions[0].1, remote_version, "a non-empty fresh PUT response must be used as the exact cleanup constraint" ); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test(flavor = "multi_thread")] #[serial_test::serial] async fn commit_lock_acquire_failure_cleans_remote_candidate_and_preserves_source() { let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); let lockers: Vec> = vec![Arc::new(LocalClient::with_manager(manager)), Arc::new(FailingLockClient)]; let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; let bucket = "transition-lock-acquire-failure-bucket"; let object = "object.bin"; let payload = b"transition commit lock failure must clean the remote candidate".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; let mut opts = transition_options(&original, tier_name); opts.no_lock = false; let error = set_disks .transition_object(bucket, object, &opts) .await .expect_err("transition must fail when the commit namespace write lock cannot reach quorum"); assert!( matches!( error, StorageError::NamespaceLockQuorumUnavailable { .. } | StorageError::Lock(LockError::QuorumNotReached { .. }) | StorageError::Lock(LockError::Internal { .. }) ), "unexpected transition lock-acquire error: {error:?}" ); assert_eq!( backend.put_count().await, 1, "remote candidate must be uploaded before commit lock failure" ); assert_eq!( backend.remove_count().await, 1, "known remote candidate must be removed when commit lock acquisition fails" ); assert_eq!( backend.remove_versions().await, backend.put_versions().await, "lock-acquire cleanup must target the exact remote version returned by PUT" ); assert_eq!( backend.object_count().await, 0, "commit lock failure must not leave an orphan remote candidate" ); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test(flavor = "current_thread", start_paused = true)] #[serial_test::serial] async fn commit_lock_lost_after_upload_cleans_remote_candidate_and_preserves_source() { let refresh_calls = Arc::new(AtomicUsize::new(0)); let lockers: Vec> = (0..4) .map(|_| Arc::new(LockLostRefreshClient::new(Arc::clone(&refresh_calls))) as Arc) .collect(); let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; let bucket = "transition-lock-lost-bucket"; let object = "object.bin"; let payload = b"lost transition commit lock must clean the remote candidate".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let previous_setup_type = runtime_sources::current_setup_type().await; runtime_sources::set_setup_type(SetupType::DistErasure).await; let mut opts = transition_options(&original, tier_name); opts.no_lock = false; let barrier = TransitionCommitBarrier::install_before_lock_lost_check(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await }); barrier.wait_until_paused().await; tokio::time::advance(Duration::from_secs(11)).await; tokio::task::yield_now().await; assert!( refresh_calls.load(Ordering::SeqCst) > 0, "test must drive the real distributed-lock heartbeat before the commit fence" ); barrier.release(); let error = transition .await .expect("transition task should not panic") .expect_err("transition must fail after the commit namespace write lock loses refresh quorum"); assert!( matches!(error, StorageError::NamespaceLockQuorumUnavailable { .. }), "unexpected transition lock-lost error: {error:?}" ); runtime_sources::set_setup_type(previous_setup_type).await; assert_eq!(backend.put_count().await, 1); assert_eq!( backend.remove_count().await, 1, "lock-lost commit fence must remove the uploaded remote candidate" ); assert_eq!( backend.remove_versions().await, backend.put_versions().await, "lock-lost cleanup must target the exact remote version returned by PUT" ); assert_eq!( backend.object_count().await, 0, "lock-lost commit fence must not leave an orphan remote candidate" ); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn cancelled_after_remote_upload_cleans_candidate_and_preserves_source() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-cancel-after-upload-bucket"; let object = "object.bin"; let payload = b"cancelled transition after upload must clean remote candidate".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let barrier = TransitionCommitBarrier::install(bucket, object); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set .transition_object(bucket, object, &transition_options(&original, tier_name)) .await }); barrier.wait_until_paused().await; assert_eq!( backend.put_count().await, 1, "transition must upload a remote candidate before the cancellation point" ); let put_versions = backend.put_versions().await; assert_eq!(put_versions.len(), 1, "transition should expose one remote candidate/version"); assert_eq!(backend.object_count().await, 1, "remote candidate should be visible before cancellation"); transition.abort(); let join = transition .await .expect_err("aborted transition task should report cancellation"); assert!(join.is_cancelled(), "transition task should be cancelled, not panic"); drop(barrier); assert!( backend .wait_for_remote_absence(&put_versions[0].0, Duration::from_secs(5)) .await, "cancellation cleanup must remove the uploaded remote candidate" ); assert_eq!( backend.remove_versions().await, put_versions, "cancellation cleanup must target the exact remote version returned by PUT" ); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn cancelled_after_put_response_before_upload_join_cleans_candidate() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-cancel-before-upload-join-bucket"; let object = "object.bin"; let payload = b"a confirmed remote upload must survive cancellation until cleanup owns it".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let barrier = TransitionUploadCandidateBarrier::install(); let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set .transition_object(bucket, object, &transition_options(&original, tier_name)) .await }); barrier.wait_until_paused().await; let put_versions = backend.put_versions().await; assert_eq!(put_versions.len(), 1, "the remote PUT response must identify one cleanup candidate"); assert_eq!(backend.object_count().await, 1, "the candidate must exist at the cancellation point"); transition.abort(); assert!( transition .await .expect_err("aborted transition task should report cancellation") .is_cancelled() ); drop(barrier); assert!( backend .wait_for_remote_absence(&put_versions[0].0, Duration::from_secs(5)) .await, "the pre-created cleanup guard must remove a candidate recorded before upload finalization completes" ); assert_eq!(backend.remove_versions().await, put_versions); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn real_bitrot_producer_failures_do_not_commit_transition() { let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_reject_non_empty_remote_versions(true); for position in [ ShardCorruptionPosition::First, ShardCorruptionPosition::Middle, ShardCorruptionPosition::Last, ] { let bucket = format!("transition-real-bitrot-{}", position.label()); 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 .get_object_fileinfo( &bucket, &object, &ObjectOptions { no_lock: true, metadata_cache_safe: false, ..Default::default() }, true, false, ) .await .expect("source metadata should be available before shard corruption"); 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; let error = set_disks .transition_object(&bucket, &object, &transition_options(&original, tier_name.clone())) .await .expect_err("producer bitrot failure must not commit transition metadata"); assert!( matches!( error, StorageError::FileCorrupt | StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) | StorageError::LessData | StorageError::Io(_) ), "{position:?}: unexpected transition producer error: {error:?}" ); let (after, _, _) = set_disks .get_object_fileinfo( &bucket, &object, &ObjectOptions { no_lock: true, metadata_cache_safe: false, ..Default::default() }, true, false, ) .await .expect("failed transition must leave metadata readable"); assert_eq!( after.data_dir, Some(data_dir), "{position:?}: transition must not release the source data dir" ); assert_ne!( after.transition_status, TRANSITION_COMPLETE, "{position:?}: transition status must remain incomplete after producer bitrot failure" ); } assert_eq!( backend.put_count().await, 3, "each bitrot case should create one remote cleanup candidate" ); assert_eq!(backend.remove_count().await, 3, "each bitrot candidate must be removed before returning"); assert_eq!( backend.remove_versions().await, backend.put_versions().await, "bitrot cleanup must target the exact remote version returned by PUT" ); assert_eq!( backend.object_count().await, 0, "bitrot producer failures must not leave remote candidates" ); } #[tokio::test] #[serial_test::serial] async fn opaque_remote_version_is_cleaned_before_parse_failure() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-unknown-version-bucket"; let object = "object.bin"; let payload = b"unknown remote version must retain local data".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_remote_version(Some("opaque-version-token".to_string())).await; set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect_err("an unparseable remote version must fail closed"); let removed_versions = backend.remove_versions().await; assert_eq!(removed_versions.len(), 1); assert_eq!(removed_versions[0].1, "opaque-version-token"); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn authoritative_read_failure_after_upload_cleans_exact_candidate_and_preserves_source() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-authoritative-read-failure-bucket"; let object = "object.bin"; let payload = b"authoritative metadata read failure must clean remote candidate".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; let put_barrier = backend.arm_put_barrier().await; let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set .transition_object(bucket, object, &transition_options(&original, tier_name)) .await }); put_barrier.wait_until_paused().await; let saved_disks = { let mut disks = set_disks.disks.write().await; let saved = std::mem::take(&mut *disks); *disks = vec![None; saved.len()]; saved }; put_barrier.release(); let err = transition .await .expect("transition task should not panic") .expect_err("authoritative metadata read failure must fail the transition"); *set_disks.disks.write().await = saved_disks; assert!( matches!( err, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) | StorageError::Io(_) ), "unexpected authoritative read failure: {err:?}" ); assert_eq!(backend.put_count().await, 1); assert_eq!(backend.remove_count().await, 1); assert_eq!( backend.remove_versions().await, backend.put_versions().await, "authoritative read failure must remove the exact uploaded remote version" ); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn backend_version_constraint_rejects_uuid_candidate_before_commit() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-version-constraint-bucket"; let object = "object.bin"; let payload = b"version-constrained backend candidate must retain local data".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let remote_version = Uuid::new_v4().to_string(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_remote_version(Some(remote_version.clone())).await; backend.set_reject_non_empty_remote_versions(true); set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect_err("a backend requiring unversioned objects must reject a UUID-shaped version"); let put_versions = backend.put_versions().await; let removed_versions = backend.remove_versions().await; assert_eq!(removed_versions, put_versions); assert_eq!(removed_versions.len(), 1); assert_eq!(removed_versions[0].1, remote_version); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn backend_version_constraint_cleans_nil_uuid_with_exact_version() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-nil-version-constraint-bucket"; let object = "object.bin"; let payload = b"a fresh nil UUID response remains an exact cleanup constraint".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let remote_version = Uuid::nil().to_string(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_remote_version(Some(remote_version.clone())).await; backend.set_reject_non_empty_remote_versions(true); set_disks .transition_object(bucket, object, &transition_options(&original, tier_name)) .await .expect_err("a backend requiring an empty version must reject a nil UUID string"); let put_versions = backend.put_versions().await; let removed_versions = backend.remove_versions().await; assert_eq!(removed_versions.len(), 1); assert_eq!(removed_versions[0].0, put_versions[0].0); assert_eq!(removed_versions[0].1, remote_version); assert_eq!(backend.exact_remove_count(), 1); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } #[tokio::test] #[serial_test::serial] async fn remote_cleanup_failure_after_version_rejection_preserves_source_and_candidate() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-cleanup-failure-bucket"; let object = "object.bin"; let payload = b"cleanup failure must keep source and orphan evidence".repeat(1024); let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; backend.set_put_remote_version(Some("opaque-version-token".to_string())).await; let put_barrier = backend.arm_put_barrier().await; let transition_set = Arc::clone(&set_disks); let transition = tokio::spawn(async move { transition_set .transition_object(bucket, object, &transition_options(&original, tier_name)) .await }); put_barrier.wait_until_paused().await; backend.set_server_error(true).await; put_barrier.release(); transition .await .expect("transition task should not panic") .expect_err("opaque remote version must fail closed even if cleanup also fails"); assert_eq!(backend.put_count().await, 1); assert_eq!( backend.remove_count().await, 0, "cleanup failure before backend remove must not be reported as a successful exact delete" ); assert_eq!( backend.object_count().await, 1, "failed cleanup must leave the remote candidate visible for durable reconciliation" ); assert_local_source_intact(&set_disks, bucket, object, &payload).await; } } #[cfg(all(test, feature = "test-util"))] mod transition_source_identity_matrix_tests { use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::bucket::lifecycle::lifecycle::{TRANSITION_PENDING, TransitionOptions}; use crate::disk::DiskAPI as _; use crate::services::tier::test_util::register_mock_tier; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; #[test] fn transition_source_identity_treats_nil_version_as_null_source() { let fi = FileInfo { version_id: Some(Uuid::nil()), data_dir: Some(Uuid::new_v4()), mod_time: Some(OffsetDateTime::now_utc()), size: 1, ..Default::default() }; let opts = ObjectOptions { versioned: true, ..Default::default() }; let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") .expect("nil source version should build a null-version identity"); assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); assert_eq!(source.version_id, None); source.validate().expect("null-version source identity should validate"); } #[test] fn transition_source_identity_treats_requested_nil_version_as_null_source() { let fi = FileInfo { version_id: None, data_dir: Some(Uuid::new_v4()), mod_time: Some(OffsetDateTime::now_utc()), size: 1, ..Default::default() }; let opts = ObjectOptions { version_id: Some(Uuid::nil().to_string()), versioned: true, ..Default::default() }; let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") .expect("requested nil source version should build a null-version identity"); assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); assert_eq!(source.version_id, None); source .validate() .expect("requested null-version source identity should validate"); } #[test] fn transition_source_identity_treats_unversioned_fileinfo_as_null_source() { let fi = FileInfo { version_id: None, versioned: false, data_dir: Some(Uuid::new_v4()), mod_time: Some(OffsetDateTime::now_utc()), size: 1, ..Default::default() }; let opts = ObjectOptions { versioned: true, ..Default::default() }; let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") .expect("unversioned fileinfo should build a null-version identity"); assert_eq!(source.version_mode, TransitionSourceVersionMode::VersionSuspended); assert_eq!(source.version_id, None); source .validate() .expect("unversioned fileinfo null-version source identity should validate"); } #[test] fn transition_source_identity_still_rejects_missing_versioned_source_id() { let fi = FileInfo { version_id: None, versioned: true, data_dir: Some(Uuid::new_v4()), mod_time: Some(OffsetDateTime::now_utc()), size: 1, ..Default::default() }; let opts = ObjectOptions { versioned: true, ..Default::default() }; let source = transition_source_identity("bucket", "object", &fi, &opts, "etag") .expect("source identity construction should preserve missing version evidence"); assert_eq!(source.version_mode, TransitionSourceVersionMode::Versioned); assert!(matches!( source.validate(), Err(crate::bucket::lifecycle::transition_transaction::TransitionTransactionError::Corrupt( "versioned source is missing version_id" )) )); } #[tokio::test] #[serial_test::serial] async fn transition_source_identity_field_matrix_rejects_single_field_drift() { #[derive(Clone, Copy, Debug)] enum IdentityField { VersionId, DataDir, ModTime, Size, Etag, } let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "transition-identity-matrix-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; for (index, field) in [ IdentityField::VersionId, IdentityField::DataDir, IdentityField::ModTime, IdentityField::Size, IdentityField::Etag, ] .into_iter() .enumerate() { let object = format!("identity-{index}.bin"); let payload = vec![u8::try_from(index + 1).expect("matrix index should fit u8"); 1024 * 1024]; let mut reader = PutObjReader::from_vec(payload); let original = set_disks .put_object(bucket, &object, &mut reader, &ObjectOptions::default()) .await .expect("source object should be written"); let (source, _, _) = set_disks .get_object_fileinfo(bucket, &object, &ObjectOptions::default(), true, false) .await .expect("source metadata should resolve"); let opts = ObjectOptions { no_lock: true, transition: TransitionOptions { status: TRANSITION_PENDING.to_string(), tier: tier_name.clone(), etag: original.etag.clone().unwrap_or_default(), ..Default::default() }, version_id: original.version_id.map(|version| version.to_string()), mod_time: original.mod_time, ..Default::default() }; let put_barrier = backend.arm_put_barrier().await; let transition_set = Arc::clone(&set_disks); let transition_object = object.clone(); let transition = tokio::spawn(async move { transition_set.transition_object(bucket, &transition_object, &opts).await }); put_barrier.wait_until_paused().await; let mut changed = source.clone(); match field { IdentityField::VersionId => changed.version_id = Some(Uuid::new_v4()), IdentityField::DataDir => changed.data_dir = Some(Uuid::new_v4()), IdentityField::ModTime => { changed.mod_time = changed.mod_time.map(|value| value + time::Duration::nanoseconds(1)); } IdentityField::Size => changed.size += 1, IdentityField::Etag => { changed.metadata.insert("etag".to_string(), format!("changed-{index}")); } } for disk in &disk_stores { disk.write_metadata("", bucket, &object, changed.clone()) .await .expect("single-field metadata drift should be written"); } put_barrier.release(); transition .await .expect("transition task should not panic") .expect_err("transition must reject a source whose identity changed after upload"); let expected_attempts = index + 1; assert_eq!(backend.put_count().await, expected_attempts); assert_eq!(backend.remove_count().await, expected_attempts); assert_eq!( backend.remove_versions().await, backend.put_versions().await, "rejected identity drift must remove the exact uploaded version" ); match field { IdentityField::VersionId => assert_ne!(source.version_id, changed.version_id), IdentityField::DataDir => assert_ne!(source.data_dir, changed.data_dir), IdentityField::ModTime => assert_ne!(source.mod_time, changed.mod_time), IdentityField::Size => assert_ne!(source.size, changed.size), IdentityField::Etag => assert_ne!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata)), } if !matches!(field, IdentityField::VersionId) { assert_eq!(source.version_id, changed.version_id); } if !matches!(field, IdentityField::DataDir) { assert_eq!(source.data_dir, changed.data_dir); } if !matches!(field, IdentityField::ModTime) { assert_eq!(source.mod_time, changed.mod_time); } if !matches!(field, IdentityField::Size) { assert_eq!(source.size, changed.size); } if !matches!(field, IdentityField::Etag) { assert_eq!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata)); } } } } #[cfg(test)] mod heterogeneous_pool_put_tests { use super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity; use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::{DiskAPI as _, ReadOptions}; use rustfs_config::server_config::KVS; use tokio::io::AsyncReadExt; #[tokio::test] async fn second_pool_regular_put_uses_its_own_layout_and_round_trips() { // Deliberately inject the first pool's invalid scalar fallback. The // test can pass only if the production PUT uses the held [4, 2] // storage-class snapshot and resolves pool 1 to parity 1. let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_for_pool_with_default_parity(2, 1, 2).await; set_disks.set_test_storage_class_config( lookup_config_for_pools_without_env(&KVS::new(), &[4, 2]).expect("heterogeneous pool storage class should resolve"), ); let bucket = "regular-put-second-pool-bucket"; let object = "object"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let payload = vec![0x3c; 4096]; let mut reader = PutObjReader::from_vec(payload.clone()); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("second-pool regular PUT should encode without zero data shards"); for (disk_index, disk) in disk_stores.iter().enumerate() { let file_info = disk .read_version("", bucket, object, "", &ReadOptions::default()) .await .unwrap_or_else(|err| panic!("disk {disk_index} should persist valid second-pool metadata: {err}")); assert_eq!(file_info.erasure.data_blocks, 1); assert_eq!(file_info.erasure.parity_blocks, 1); } let mut object_reader = set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) .await .expect("second-pool regular PUT should be readable"); let mut restored = Vec::new(); object_reader .stream .read_to_end(&mut restored) .await .expect("second-pool regular PUT should stream"); assert_eq!(restored, payload); } } #[cfg(test)] mod put_object_tmp_cleanup_tests { //! Regression coverage for backlog#924 (HP-3): the speculative tmp-dir //! cleanup at the end of a successful PUT runs on a spawned task (off the //! response path), while a failed PUT must still clean its tmp shards //! inline before returning. use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; use std::time::Duration; use tempfile::TempDir; /// Large enough that the erasure shards are written as real tmp files /// (never inlined into xl.meta), so both tests exercise actual cleanup. const TEST_OBJECT_SIZE: usize = 1 << 20; /// Entries under `.rustfs.sys/tmp` on every disk, excluding the `.trash` /// staging directory (trash reclamation is a background concern). async fn non_trash_tmp_entries(temp_dirs: &[TempDir]) -> Vec { let mut leftovers = Vec::new(); for temp_dir in temp_dirs { let tmp_path = temp_dir.path().join(RUSTFS_META_TMP_BUCKET); let mut read_dir = match tokio::fs::read_dir(&tmp_path).await { Ok(read_dir) => read_dir, Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue, Err(err) => panic!("tmp dir {tmp_path:?} should be listable: {err}"), }; while let Some(entry) = read_dir.next_entry().await.expect("tmp dir entry should be readable") { let name = entry.file_name().to_string_lossy().to_string(); if name != ".trash" { leftovers.push(format!("{}/{name}", tmp_path.display())); } } } leftovers } #[tokio::test] async fn put_object_success_eventually_cleans_tmp_workspace() { let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "tmp-clean-ok-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(vec![7u8; TEST_OBJECT_SIZE]); set_disks .put_object(bucket, "hot-path-object", &mut reader, &ObjectOptions::default()) .await .expect("put_object should succeed"); // The speculative cleanup runs on a spawned task off the PUT response // path, so poll for the tmp workspace to drain instead of asserting // immediately. let deadline = tokio::time::Instant::now() + Duration::from_secs(10); loop { let leftovers = non_trash_tmp_entries(&temp_dirs).await; if leftovers.is_empty() { break; } assert!( tokio::time::Instant::now() < deadline, "tmp workspace should drain after a successful PUT, leftovers: {leftovers:?}" ); tokio::time::sleep(Duration::from_millis(25)).await; } drop(temp_dirs); } #[tokio::test] async fn put_object_failure_cleans_tmp_workspace_inline() { let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await; // The bucket volume is never created, so the shards are written into // the tmp workspace and the commit fails at rename_data with a quorum // error — exercising the failure-path cleanup. let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]); let err = set_disks .put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default()) .await .expect_err("put_object into a missing bucket volume must fail"); // No polling: the failure path must clean the tmp workspace inline, // before put_object returns (backlog#864 / backlog#898 hardening). let leftovers = non_trash_tmp_entries(&temp_dirs).await; assert!( leftovers.is_empty(), "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}" ); drop(temp_dirs); } } #[cfg(test)] mod put_object_tags_early_stop_regression_tests { //! Regression coverage for backlog#881: read-before-write tagging under GET //! metadata early-stop. backlog#872 enabled metadata early-stop by default; //! an early-stopped metadata fanout returns only a read-quorum subset of //! disks, and `put_object_tags` reuses that disk set as its write target, so //! it shrank the write set and failed write quorum with SlowDown. The fix //! forces a full-quorum fanout (allow_early_stop=false) inside //! `put_object_tags`. This multi-disk integration test pins that end to end: //! with early-stop enabled, the tag must land on EVERY online disk's xl.meta, //! not a read-quorum subset. use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::disk::{DiskAPI as _, ReadOptions}; #[tokio::test] async fn put_object_tags_writes_all_online_disks_under_early_stop() { // Early-stop defaults to on; pin it explicitly (enabled + full rollout) // so the regression scenario holds regardless of default/rollout drift. temp_env::async_with_vars( [ ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")), ("RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT", Some("100")), ], async { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "backlog881-early-stop-bucket"; let object = "read-before-write-tagged-object"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut reader = PutObjReader::from_vec(vec![5u8; 1024]); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("put_object should succeed"); let tags = "unit=backlog881&stage=regression"; set_disks .put_object_tags(bucket, object, tags, &ObjectOptions::default()) .await .expect("put_object_tags must not fail write quorum (SlowDown) under early-stop"); // Core assertion: the tag must be present on EVERY online disk, // proving the write target was the full set and not an // early-stopped read-quorum subset. for (idx, disk) in disk_stores.iter().enumerate() { let fi = disk .read_version("", bucket, object, "", &ReadOptions::default()) .await .unwrap_or_else(|e| panic!("disk {idx} must hold xl.meta for the tagged object: {e}")); assert_eq!( fi.metadata.get(AMZ_OBJECT_TAGGING).map(String::as_str), Some(tags), "disk {idx} must carry the tag written under early-stop (write set not shrunk to a read-quorum subset)" ); } }, ) .await; } } #[cfg(test)] mod delete_objects_lock_gating_tests { //! Regression coverage for backlog#929 (HP-8): the batch-delete per-object //! stat is gated on the bucket object-lock configuration. For buckets whose //! metadata is unknown the gate fails closed, so these hermetic tests (their //! buckets are never registered with the metadata sys) exercise the //! locked-stat path and prove the #4297 delete protection is intact end to //! end, while per-key result mapping of mixed batches stays stable. use super::hermetic_set_disks_support::hermetic_set_disks; use super::*; use crate::disk::DiskAPI as _; async fn put_plain_object(set_disks: &Arc, bucket: &str, object: &str) { let mut reader = PutObjReader::from_vec(vec![3u8; 1024]); set_disks .put_object(bucket, object, &mut reader, &ObjectOptions::default()) .await .expect("plain object should be written"); } #[tokio::test] async fn delete_objects_reports_mixed_results_per_key() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "hp8-mixed-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } put_plain_object(&set_disks, bucket, "obj-a").await; put_plain_object(&set_disks, bucket, "obj-c").await; let objects = vec![ ObjectToDelete { object_name: "obj-a".to_string(), ..Default::default() }, ObjectToDelete { object_name: "missing-b".to_string(), ..Default::default() }, ObjectToDelete { object_name: "obj-c".to_string(), ..Default::default() }, ]; let (deleted, errs) = set_disks.delete_objects(bucket, objects, ObjectOptions::default()).await; assert_eq!(deleted.len(), 3); assert_eq!(errs.len(), 3); assert!( errs.iter().all(Option::is_none), "S3 batch delete reports missing keys as deleted, not as errors: {errs:?}" ); assert_eq!(deleted[0].object_name, "obj-a"); assert_eq!(deleted[1].object_name, "missing-b"); assert_eq!(deleted[2].object_name, "obj-c"); assert!(deleted[0].found, "existing key must be reported as found"); assert!(!deleted[1].found, "missing key must be reported as not found"); assert!(deleted[2].found, "existing key must be reported as found"); for object in ["obj-a", "obj-c"] { set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await .expect_err("deleted object must be gone"); } } #[tokio::test] async fn delete_objects_blocks_locked_object_and_deletes_the_rest() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "hp8-locked-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } // COMPLIANCE retention metadata on the object; this bucket is unknown // to the metadata sys, so the fail-closed gate must keep the held-lock // stat and the #4297 rejection. let retain_until = OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24 * 30); let mut user_defined = HashMap::new(); user_defined.insert( X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(), s3s::dto::ObjectLockRetentionMode::COMPLIANCE.to_string(), ); user_defined.insert( X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(), retain_until .format(&time::format_description::well_known::Rfc3339) .expect("retain-until date should format"), ); let mut reader = PutObjReader::from_vec(vec![7u8; 512]); set_disks .put_object( bucket, "locked", &mut reader, &ObjectOptions { user_defined, ..Default::default() }, ) .await .expect("locked object should be written"); put_plain_object(&set_disks, bucket, "plain").await; let objects = vec![ ObjectToDelete { object_name: "locked".to_string(), ..Default::default() }, ObjectToDelete { object_name: "plain".to_string(), ..Default::default() }, ]; let (_deleted, errs) = set_disks.delete_objects(bucket, objects, ObjectOptions::default()).await; let lock_err = errs[0] .as_ref() .expect("COMPLIANCE retention must block the batch delete entry"); assert!( matches!(lock_err, Error::PrefixAccessDenied(_, _)), "locked entry must fail with access denied, got: {lock_err:?}" ); assert!(errs[1].is_none(), "unlocked entry must still be deleted: {:?}", errs[1]); set_disks .get_object_info(bucket, "locked", &ObjectOptions::default()) .await .expect("locked object must survive the batch delete"); set_disks .get_object_info(bucket, "plain", &ObjectOptions::default()) .await .expect_err("plain object must be deleted"); } #[tokio::test] async fn delete_objects_honors_no_lock_when_outer_write_lock_is_held() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "batch-no-lock-bucket"; for disk in &disk_stores { disk.make_volume(bucket).await.expect("bucket volume should be created"); } put_plain_object(&set_disks, bucket, "obj-a").await; let _outer_guard = set_disks .new_ns_lock(bucket, "obj-a") .await .expect("namespace lock should be created") .get_write_lock(Duration::from_secs(1)) .await .expect("outer write lock should be acquired"); let objects = vec![ObjectToDelete { object_name: "obj-a".to_string(), ..Default::default() }]; let (deleted, errs) = tokio::time::timeout( Duration::from_secs(1), set_disks.delete_objects( bucket, objects, ObjectOptions { no_lock: true, ..Default::default() }, ), ) .await .expect("no_lock batch delete path must not wait for the outer lock"); assert!(errs[0].is_none(), "no_lock batch delete should not fail with a lock error: {:?}", errs[0]); assert!(deleted[0].found, "existing object must still be deleted"); } } #[cfg(test)] mod body_cache_hook_gate_tests { //! Regression coverage for backlog#1108 (raw data-movement reads) and //! backlog#1109 (compressed objects): the app-layer body-cache hook serves //! cached full-object plaintext directly, bypassing `ReadPlan`. It must not //! be probed when the normal read path would return a different byte stream. use super::full_object_plaintext_len; use crate::object_api::{ObjectInfo, ObjectOptions}; use std::collections::HashMap; use std::sync::Arc; const STORED_SIZE: i64 = 1024; const PLAINTEXT_SIZE: i64 = 4096; fn plain_object_info() -> ObjectInfo { ObjectInfo { size: STORED_SIZE, ..Default::default() } } fn compressed_object_info() -> ObjectInfo { let mut user_defined = HashMap::new(); // is_compressed() checks for the internal compression suffix key. user_defined.insert("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string()); user_defined.insert("x-rustfs-internal-actual-size".to_string(), PLAINTEXT_SIZE.to_string()); ObjectInfo { size: STORED_SIZE, user_defined: Arc::new(user_defined), ..Default::default() } } fn encrypted_object_info() -> ObjectInfo { let mut user_defined = HashMap::new(); user_defined.insert("x-minio-encryption-key".to_string(), "opaque".to_string()); ObjectInfo { size: STORED_SIZE, user_defined: Arc::new(user_defined), ..Default::default() } } fn restore_opts() -> ObjectOptions { let mut opts = ObjectOptions::default(); opts.transition.restore_request.days = Some(1); opts } #[test] fn plain_full_object_read_yields_its_stored_size() { let len = full_object_plaintext_len(&None, &ObjectOptions::default(), &plain_object_info()); assert_eq!(len, Some(STORED_SIZE)); } #[test] fn compressed_object_yields_decompressed_size() { // ReadTransform::Compressed publishes the decompressed length as // object_info.size; the hit site must reproduce exactly this value or // UploadPartCopy truncates the copy (backlog#1109). let len = full_object_plaintext_len(&None, &ObjectOptions::default(), &compressed_object_info()); assert_eq!(len, Some(PLAINTEXT_SIZE)); } #[test] fn raw_data_movement_read_is_refused() { // Decommission reads set raw_data_movement_read: ReadPlan returns the // STORED bytes, so the cached decompressed body must not be served // (backlog#1108 — silent data corruption on the destination pool). let opts = ObjectOptions { raw_data_movement_read: true, ..Default::default() }; assert_eq!(full_object_plaintext_len(&None, &opts, &plain_object_info()), None); } #[test] fn data_movement_read_is_refused() { let opts = ObjectOptions { data_movement: true, ..Default::default() }; assert_eq!(full_object_plaintext_len(&None, &opts, &plain_object_info()), None); } #[test] fn restore_read_of_compressed_object_is_refused() { // restore_request_active forces ReadPlan down the Plain branch, so the // read yields STORED (compressed) bytes under the compressed size. assert_eq!(full_object_plaintext_len(&None, &restore_opts(), &compressed_object_info()), None); } #[test] fn restore_read_of_plain_object_is_refused() { assert_eq!(full_object_plaintext_len(&None, &restore_opts(), &plain_object_info()), None); } #[test] fn encrypted_object_is_refused() { assert_eq!( full_object_plaintext_len(&None, &ObjectOptions::default(), &encrypted_object_info()), None ); } #[test] fn compressed_object_without_actual_size_is_refused() { // A compressed object whose actual size cannot be resolved must not be // served from cache: there is no length to publish as object_info.size. let mut user_defined = HashMap::new(); user_defined.insert("x-rustfs-internal-compression".to_string(), "klauspost/compress/s2".to_string()); let info = ObjectInfo { size: STORED_SIZE, user_defined: Arc::new(user_defined), ..Default::default() }; assert_eq!(full_object_plaintext_len(&None, &ObjectOptions::default(), &info), None); } } #[cfg(test)] mod body_cache_hook_e2e_tests { //! End-to-end regression coverage for the two merged P0 data-corruption //! fixes, driving the *real* `get_object_reader` gate + probe + reader //! construction rather than the `full_object_plaintext_len` predicate in //! isolation. The predicate-only tests above cannot catch a caller that //! opens a new shortcut serving the cached body directly — which was the //! original form of both P0s (backlog#1108 / #1109) and the reason ODC-21 //! (backlog#1126) made the hook re-registrable so these tests can install a //! deterministic hook. //! //! A stand-in `GetObjectBodyCacheHook` plays the app-layer cache: the //! adapter itself lives above ecstore and cannot be reached from here, but //! the hook trait is exactly the production injection point, so exercising //! it through `get_object_reader` is a true end-to-end test of the ecstore //! side (gate decision -> probe -> served bytes and published size). //! //! The tests share one process-global hook slot, so they serialize under a //! single key and clear the slot on the way out; each uses a unique //! bucket/object and the hook returns `None` for anything else, so a stray //! concurrent GET in another test is unaffected. use crate::ecstore_validation_blackbox::make_local_set_disks; use crate::io_support::rio::{HashReader, compression_metadata_value, compression_reader}; use crate::object_api::{ GetObjectBodyCacheHook, GetObjectBodySource, ObjectInfo, ObjectOptions, PutObjReader, clear_get_object_body_cache_hook, register_get_object_body_cache_hook, }; use crate::set_disk::SetDisks; use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions}; use crate::storage_api_contracts::object::ObjectIO as _; use bytes::Bytes; use http::HeaderMap; use rustfs_utils::CompressionAlgorithm; use rustfs_utils::http::{SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, insert_str}; use std::collections::HashMap; use std::io::Cursor; use std::sync::Arc; use tokio::io::AsyncReadExt as _; /// A stand-in for the app-layer object-data cache: returns `body` only for /// the one primed key, mirroring how the real adapter answers a probe. struct PrimedBodyHook { bucket: String, object: String, body: Bytes, } #[async_trait::async_trait] impl GetObjectBodyCacheHook for PrimedBodyHook { async fn lookup(&self, bucket: &str, object: &str, _info: &ObjectInfo) -> Option { (bucket == self.bucket && object == self.object).then(|| self.body.clone()) } } /// Registers a primed hook and clears it on drop so no hook leaks into an /// unrelated test sharing the process-global slot. struct HookGuard; impl HookGuard { fn install(bucket: &str, object: &str, body: Bytes) -> Self { register_get_object_body_cache_hook(Arc::new(PrimedBodyHook { bucket: bucket.to_string(), object: object.to_string(), body, })); HookGuard } } impl Drop for HookGuard { fn drop(&mut self) { clear_get_object_body_cache_hook(); } } /// Compresses `plaintext` with the same codec the real PUT path uses, so /// the stored bytes round-trip through `get_object_reader`'s decompressor. async fn compress(plaintext: &[u8]) -> Vec { let mut reader = compression_reader(Cursor::new(plaintext.to_vec()), CompressionAlgorithm::default(), false); let mut compressed = Vec::new(); reader.read_to_end(&mut compressed).await.expect("compress plaintext"); compressed } /// Writes a genuinely compressed object: the stored data is `compressed` /// and the metadata marks it compressed with `plaintext.len()` as the /// decompressed length, exactly as the app-layer compress path records it. async fn put_compressed_object(set_disks: &Arc, bucket: &str, object: &str, plaintext: &[u8], compressed: &[u8]) { let mut user_defined = HashMap::new(); insert_str( &mut user_defined, SUFFIX_COMPRESSION, compression_metadata_value(CompressionAlgorithm::default()), ); insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, plaintext.len().to_string()); let opts = ObjectOptions { no_lock: true, user_defined, ..Default::default() }; let stream = HashReader::from_stream( Cursor::new(compressed.to_vec()), compressed.len() as i64, // stored (compressed) size plaintext.len() as i64, // actual (decompressed) size None, None, false, ) .expect("hash reader over compressed bytes"); let mut reader = PutObjReader::new(stream); set_disks .make_bucket(bucket, &MakeBucketOptions::default()) .await .expect("bucket should be created"); set_disks .put_object(bucket, object, &mut reader, &opts) .await .expect("compressed object should be written"); } async fn read_to_vec(set_disks: &Arc, bucket: &str, object: &str, opts: &ObjectOptions) -> (Vec, i64) { let mut reader = set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), opts) .await .expect("object reader should open"); let published_size = reader.object_info.size; let mut body = Vec::new(); reader.stream.read_to_end(&mut body).await.expect("object should stream"); (body, published_size) } /// A compressible payload large enough to guarantee stored != plaintext and /// to keep the object off the inline / direct-memory fast paths. fn compressible_plaintext() -> Vec { b"rustfs-body-cache-e2e-regression-".repeat(20_000) } /// Writes a plain (uncompressed, unencrypted) object of `data`, large enough /// to keep it off the inline fast path so the cache hook is actually probed. async fn put_plain_object(set_disks: &Arc, bucket: &str, object: &str, data: &[u8]) { let opts = ObjectOptions { no_lock: true, ..Default::default() }; let stream = HashReader::from_stream(Cursor::new(data.to_vec()), data.len() as i64, data.len() as i64, None, None, false) .expect("hash reader over plain bytes"); let mut reader = PutObjReader::new(stream); set_disks .make_bucket(bucket, &MakeBucketOptions::default()) .await .expect("bucket should be created"); set_disks .put_object(bucket, object, &mut reader, &opts) .await .expect("plain object should be written"); } /// ODC-16: a cache-hook hit must mark the reader `HookServed` so the app /// layer serves the buffered body without a second lookup. A large plain /// object stays off the inline fast path, so the hook is genuinely probed. #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn plain_cache_hit_marks_reader_hook_served() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let bucket = "e2e-body-cache-hook-served"; let object = "plain.bin"; let payload = b"rustfs-hook-served-payload-".repeat(40_000); put_plain_object(&set_disks, bucket, object, &payload).await; let _guard = HookGuard::install(bucket, object, Bytes::from(payload.clone())); let opts = ObjectOptions { no_lock: true, ..Default::default() }; let mut reader = set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) .await .expect("object reader should open"); assert_eq!( reader.body_source, GetObjectBodySource::HookServed, "a hook hit must mark the reader HookServed" ); assert!(reader.buffered_body.is_some(), "a hook-served reader carries the cache body"); let mut body = Vec::new(); reader.stream.read_to_end(&mut body).await.expect("object should stream"); assert_eq!(body, payload, "the hook-served body must be the primed plaintext"); } /// ODC-16: when the hook is registered but misses this object, the reader /// must be marked `HookMissed` so the app layer skips its now-redundant /// lookup (the hook's miss ran after fresh metadata resolution). #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn plain_cache_miss_marks_reader_hook_missed() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let bucket = "e2e-body-cache-hook-missed"; let object = "plain.bin"; let payload = b"rustfs-hook-missed-payload-".repeat(40_000); put_plain_object(&set_disks, bucket, object, &payload).await; // Register a hook primed for a DIFFERENT object, so this read is probed // (hook registered + eligible) but the probe misses. let _guard = HookGuard::install(bucket, "other-object", Bytes::from_static(b"unrelated")); let opts = ObjectOptions { no_lock: true, ..Default::default() }; let reader = set_disks .get_object_reader(bucket, object, None, HeaderMap::new(), &opts) .await .expect("object reader should open"); assert_eq!( reader.body_source, GetObjectBodySource::HookMissed, "a probed miss must mark the reader HookMissed" ); } /// backlog#1108: a raw data-movement read (decommission/rebalance copy) must /// yield the STORED (compressed) representation, never the cached plaintext. /// Serving the cache here writes decompressed bytes into the destination /// pool under the compressed object's metadata — silent corruption. #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn raw_data_movement_read_serves_stored_bytes_not_cached_plaintext() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let bucket = "e2e-body-cache-raw-move"; let object = "compressed.bin"; let plaintext = compressible_plaintext(); let compressed = compress(&plaintext).await; assert_ne!(compressed, plaintext, "fixture must be genuinely compressed"); put_compressed_object(&set_disks, bucket, object, &plaintext, &compressed).await; // Prime the cache with the DECOMPRESSED plaintext. let _guard = HookGuard::install(bucket, object, Bytes::from(plaintext.clone())); // decommission_object_migration_read_opts sets both raw_data_movement_read // and data_movement; ReadPlan keys the stored-bytes path on the former. let opts = ObjectOptions { no_lock: true, raw_data_movement_read: true, ..Default::default() }; let (body, published_size) = read_to_vec(&set_disks, bucket, object, &opts).await; assert_eq!(body, compressed, "raw data-movement read must return the stored compressed bytes"); assert_ne!(body, plaintext, "raw data-movement read must NOT return the cached plaintext"); assert_eq!(published_size, compressed.len() as i64, "raw read must publish the stored size"); } /// backlog#1109: on a cache hit for a compressed object the served /// `object_info.size` must be the DECOMPRESSED length (what /// ReadTransform::Compressed publishes), and the streamed body length must /// match it. UploadPartCopy reads the copy length straight off this field. #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn compressed_cache_hit_publishes_decompressed_size() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let bucket = "e2e-body-cache-compressed"; let object = "compressed.bin"; let plaintext = compressible_plaintext(); let compressed = compress(&plaintext).await; assert_ne!(compressed, plaintext, "fixture must be genuinely compressed"); put_compressed_object(&set_disks, bucket, object, &plaintext, &compressed).await; let normal_opts = ObjectOptions { no_lock: true, ..Default::default() }; // Control: with no hook the real decode path must decompress to the // plaintext and publish the decompressed length. This also proves the // fixture round-trips (stored compressed bytes -> plaintext). clear_get_object_body_cache_hook(); let (control_body, control_size) = read_to_vec(&set_disks, bucket, object, &normal_opts).await; assert_eq!(control_body, plaintext, "real read must decompress to the plaintext"); assert_eq!(control_size, plaintext.len() as i64, "real read must publish the decompressed size"); // Hit: prime with the plaintext and read normally. The probe must serve // it AND republish the decompressed length, not the stored size. let _guard = HookGuard::install(bucket, object, Bytes::from(plaintext.clone())); let (hit_body, hit_size) = read_to_vec(&set_disks, bucket, object, &normal_opts).await; assert_eq!( hit_size, plaintext.len() as i64, "cache hit must publish the decompressed size (backlog#1109)" ); assert_ne!(hit_size, compressed.len() as i64, "cache hit must not publish the stored compressed size"); assert_eq!(hit_body.len() as i64, hit_size, "streamed length must equal the published size"); assert_eq!(hit_body, plaintext, "cache hit must stream the plaintext"); } /// backlog#1146: a restore read forces ReadPlan down the Plain branch, so a /// compressed object yields its STORED bytes under the compressed size. The /// cache (holding plaintext) must not be served. #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn restore_read_serves_stored_bytes_not_cached_plaintext() { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let bucket = "e2e-body-cache-restore"; let object = "compressed.bin"; let plaintext = compressible_plaintext(); let compressed = compress(&plaintext).await; assert_ne!(compressed, plaintext, "fixture must be genuinely compressed"); put_compressed_object(&set_disks, bucket, object, &plaintext, &compressed).await; let _guard = HookGuard::install(bucket, object, Bytes::from(plaintext.clone())); let mut opts = ObjectOptions { no_lock: true, ..Default::default() }; opts.transition.restore_request.days = Some(1); let (body, published_size) = read_to_vec(&set_disks, bucket, object, &opts).await; assert_eq!(body, compressed, "restore read must return the stored compressed bytes"); assert_ne!(body, plaintext, "restore read must NOT return the cached plaintext"); assert_eq!( published_size, compressed.len() as i64, "restore read must publish the stored compressed size" ); } }