From 1042995fed4dcb69b9645cd8f314d5cf3bb8b0c1 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 11 Sep 2026 17:21:54 +0800 Subject: [PATCH] fix(ecstore): preserve multi-pool version histories --- crates/ecstore/src/store/list_objects.rs | 421 +++++++++++++++-- crates/ecstore/src/store/object.rs | 565 ++++++++++++++++++++++- crates/ecstore/src/store/rebalance.rs | 23 +- 3 files changed, 965 insertions(+), 44 deletions(-) diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 459ef0870..719c07319 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -70,7 +70,7 @@ use tokio::io::duplex; use tokio::sync::broadcast::{self}; use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio::sync::{OnceCell, RwLock}; -use tokio::task::JoinSet; +use tokio::task::{JoinHandle, JoinSet}; use tokio_util::sync::CancellationToken; use tracing::{Instrument, debug, error, info, warn}; use uuid::Uuid; @@ -4331,6 +4331,7 @@ impl ECStore { "store list_merged started" ); + let rx = rx.child_token(); let mut futures = Vec::new(); let mut inputs = Vec::new(); @@ -4346,16 +4347,10 @@ impl ECStore { } } - tokio::spawn( - async move { - if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await { - error!("merge_entry_channels err {:?}", err) - } - } - .instrument(tracing::Span::current()), - ); + let merge_task = spawn_listing_merge(rx, inputs, sender); let results = join_all(futures).await; + merge_task.await.map_err(Error::from)??; let mut all_at_eof = true; @@ -4422,6 +4417,7 @@ impl ECStore { ) -> Result<()> { check_list_objs_args(bucket, prefix, &None)?; + let rx = rx.child_token(); let mut futures = Vec::new(); let mut inputs = Vec::new(); @@ -4783,17 +4779,11 @@ impl ECStore { .instrument(tracing::Span::current()), ); - tokio::spawn( - async move { - if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await { - error!("merge_entry_channels err {:?}", err) - } - } - .instrument(tracing::Span::current()), - ); + let merge_task = spawn_listing_merge(rx, inputs, merge_tx); let walk_started = std::time::Instant::now(); let walk_results = join_all(futures).await; + merge_task.await.map_err(Error::from)??; let mut errs = Vec::new(); for walk_result in walk_results { match walk_result { @@ -5068,6 +5058,106 @@ async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender) -> Result<()> { + let name = first.name.clone(); + let mut versions: HashMap<(Option, bool), (FileMetaShallowVersion, ObjectInfo)> = HashMap::new(); + for mut entry in std::iter::once(std::mem::take(first)).chain(others) { + let meta = match entry.cached.take() { + Some(meta) => meta, + None => FileMeta::load(&entry.metadata).map_err(|_| Error::FileCorrupt)?, + }; + if meta.versions.is_empty() { + return Err(Error::FileCorrupt); + } + for version in meta.versions { + let parsed = version.parse_version_meta().map_err(|_| Error::FileCorrupt)?; + if !parsed.valid() || parsed.version_type != version.header.version_type { + return Err(Error::FileCorrupt); + } + let fi = parsed.into_fileinfo("", &name, true).map_err(|_| Error::FileCorrupt)?; + let version_id = fi.version_id.filter(|id| !id.is_nil()); + if version_id != version.header.version_id.filter(|id| !id.is_nil()) + || fi.mod_time != version.header.mod_time + || fi.tier_free_version() != version.header.free_version() + { + return Err(Error::FileCorrupt); + } + let info = ObjectInfo::from_file_info(&fi, "", &name, true); + let identity = (version_id, version.header.free_version()); + match versions.entry(identity) { + std::collections::hash_map::Entry::Vacant(slot) => { + slot.insert((version, info)); + } + std::collections::hash_map::Entry::Occupied(mut slot) => { + let (previous, previous_info) = slot.get(); + // Suspended and unversioned writes replace the one null + // slot. Distinct UUID versions never supersede each other. + if version_id.is_none() && info.mod_time != previous_info.mod_time { + if info.mod_time > previous_info.mod_time { + slot.insert((version, info)); + } + continue; + } + let equivalent = if info.delete_marker && previous_info.delete_marker { + super::object::is_equivalent_data_movement_delete_marker(&info, previous_info) + } else { + crate::data_movement::is_equivalent_data_movement_object_identity(&info, previous_info, true, true) + }; + if !equivalent { + return Err(Error::FileCorrupt); + } + // Equivalent migrated copies can have different coding or + // data directories. Choose a stable representation without + // making input order part of the S3 version order. + if version.meta < previous.meta { + slot.insert((version, info)); + } + } + } + } + } + let mut merged = FileMeta::new(); + merged.versions = versions.into_values().map(|(version, _)| version).collect(); + merged.versions.sort_by(|a, b| { + if a.header.sorts_before(&b.header) { + std::cmp::Ordering::Less + } else if b.header.sorts_before(&a.header) { + std::cmp::Ordering::Greater + } else { + std::cmp::Ordering::Equal + } + }); + let metadata = merged.marshal_msg()?; + *first = MetaCacheEntry { + name, + metadata, + cached: Some(merged), + reusable: true, + }; + Ok(()) +} + +/// `rx` is private to the producers. Cancelling it on a merge error must not +/// cancel the request token, which would suppress that error at the API edge. +fn spawn_listing_merge( + rx: CancellationToken, + inputs: Vec>, + sender: Sender, +) -> JoinHandle> { + tokio::spawn( + async move { + let result = merge_entry_channels(rx.clone(), inputs, sender, 1).await; + if result.is_err() { + rx.cancel(); + } + result + } + .instrument(tracing::Span::current()), + ) +} + async fn merge_entry_channels( rx: CancellationToken, in_channels: Vec>, @@ -5133,6 +5223,7 @@ async fn merge_entry_channels( // after anything greater has been emitted). let mut last_emitted = String::new(); let mut group: Vec> = Vec::new(); + let mut object_entries: Vec = Vec::new(); let mut refill: Vec = Vec::with_capacity(in_channels.len()); while let Some(Reverse(first)) = heap.pop() { @@ -5150,7 +5241,7 @@ async fn merge_entry_channels( // Resolve the same-name group to one winner (heads arrive in ascending // channel order): // - prefix dir vs prefix dir: the first (lowest channel) wins; - // - object vs object: the later channel wins (legacy authority rule); + // - object vs object: merge the independently resolved version stacks; // - object vs prefix dir: same-name means both end with the separator, // i.e. the object is an explicit "directory marker" for the same S3 // key — it shadows the prefix dir so the key does not surface as @@ -5168,11 +5259,27 @@ async fn merge_entry_channels( if dir_winner.is_none() { dir_winner = Some(head); } + } else if let Some(winner) = object_winner.as_ref() { + // Key-only candidates carry no version metadata and cannot + // replace a resolved stack or contribute a quorum vote. + if head.entry.is_object() { + if winner.entry.is_object() { + object_entries.push(head.entry); + } else { + object_winner = Some(head); + } + } } else { object_winner = Some(head); } } + if !object_entries.is_empty() + && let Some(winner) = object_winner.as_mut() + { + merge_object_entry_versions(&mut winner.entry, object_entries.drain(..))?; + } + if let Some(head) = object_winner.or(dir_winner) && head.entry.name != last_emitted { @@ -5605,6 +5712,7 @@ impl Sets { "sets list_merged started" ); + let rx = rx.child_token(); let mut futures = Vec::new(); let mut inputs = Vec::new(); @@ -5617,16 +5725,10 @@ impl Sets { futures.push(async move { set.list_path(rx_clone, opts, send).await }); } - tokio::spawn( - async move { - if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await { - error!("merge_entry_channels err {:?}", err); - } - } - .instrument(tracing::Span::current()), - ); + let merge_task = spawn_listing_merge(rx, inputs, sender); let results = join_all(futures).await; + merge_task.await.map_err(Error::from)??; let mut all_at_eof = true; let mut errs = Vec::new(); for result in results { @@ -5677,6 +5779,7 @@ impl Sets { ) -> Result<()> { check_list_objs_args(bucket, prefix, &None)?; + let rx = rx.child_token(); let mut futures = Vec::new(); let mut inputs = Vec::new(); @@ -6007,17 +6110,11 @@ impl Sets { .instrument(tracing::Span::current()), ); - tokio::spawn( - async move { - if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await { - error!("merge_entry_channels err {:?}", err) - } - } - .instrument(tracing::Span::current()), - ); + let merge_task = spawn_listing_merge(rx, inputs, merge_tx); let walk_started = std::time::Instant::now(); let walk_results = join_all(futures).await; + merge_task.await.map_err(Error::from)??; let mut errs = Vec::new(); for walk_result in walk_results { match walk_result { @@ -7016,7 +7113,7 @@ mod test { }; use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw}; use crate::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, error::DiskError, new_disk}; - use crate::error::StorageError; + use crate::error::{Result, StorageError}; use crate::object_api::ObjectInfo; use rustfs_filemeta::{ FileInfo, FileMeta, FileMetaVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry, MetaDeleteMarker, @@ -10399,7 +10496,7 @@ mod test { } #[tokio::test] - async fn merge_entry_channels_documents_candidate_metadata_authority_risk() { + async fn merge_entry_channels_preserves_cross_pool_delete_marker_versions() { let (tx_a, rx_a) = mpsc::channel(4); let (tx_b, rx_b) = mpsc::channel(4); let (tx_c, rx_c) = mpsc::channel(4); @@ -10428,9 +10525,13 @@ mod test { .expect("merged entry should be present"); assert_eq!(merged.name, "obj-a"); assert!( - !merged.is_latest_delete_marker(), - "current merge consumes candidate metadata bytes; future index-backed strong modes must live-verify metadata instead" + merged.is_latest_delete_marker(), + "a newer marker must remain current across independently resolved pools" ); + let versions = merged.file_info_versions("bucket").expect("merged versions should decode"); + assert_eq!(versions.versions.len(), 2, "retain the historical object and deduplicate the marker"); + assert!(versions.versions[0].deleted && versions.versions[0].is_latest); + assert!(!versions.versions[1].deleted && !versions.versions[1].is_latest); assert!( matches!(timeout(Duration::from_secs(1), out_rx.recv()).await, Ok(None)), "merge should not emit a duplicate entry for the same key" @@ -10442,6 +10543,250 @@ mod test { .expect("merge task should succeed"); } + fn rewrite_test_version(mut entry: MetaCacheEntry, change: impl FnOnce(&mut FileMetaVersion)) -> MetaCacheEntry { + let meta = entry.cached.as_mut().expect("test metadata should be decoded"); + assert_eq!(meta.versions.len(), 1); + let mut version = meta.versions[0].parse_version_meta().expect("test version should decode"); + change(&mut version); + meta.versions[0] = version.try_into().expect("test version should encode"); + entry.metadata = meta.marshal_msg().expect("test metadata should encode"); + entry + } + + async fn merge_test_object_entries(entries: Vec) -> Result { + let mut inputs = Vec::with_capacity(entries.len()); + for entry in entries { + let (sender, receiver) = mpsc::channel(1); + sender.send(entry).await.expect("fixture entry should queue"); + inputs.push(receiver); + } + let (sender, mut receiver) = mpsc::channel(1); + let task = tokio::spawn(merge_entry_channels(CancellationToken::new(), inputs, sender, 1)); + let entry = receiver.recv().await; + task.await.expect("merge must not panic")?; + Ok(entry.expect("a valid same-key group must produce an entry")) + } + + #[tokio::test] + async fn merge_entry_channels_orders_complete_histories_independently_of_pool_order() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "first", 4, 2)]); + let second = rewrite_test_version( + test_object_meta_entry_with_erasure_versions("key", &[(time, "second", 4, 2)]), + |version| version.object.as_mut().expect("object version").version_id = Some(Uuid::from_u128(2)), + ); + let marker = test_delete_marker_meta_entry("key", time + time::Duration::seconds(1)); + let inputs = [first, second, marker]; + let mut expected = None; + for order in [[0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]] { + let entry = merge_test_object_entries(order.map(|index| inputs[index].clone()).to_vec()) + .await + .expect("disjoint version chains should merge"); + let versions = entry.file_info_versions("bucket").expect("merged versions should decode"); + assert_eq!(versions.versions.len(), 3); + assert!(versions.versions[0].deleted && versions.versions[0].is_latest); + assert!( + versions.versions[1..] + .iter() + .all(|version| !version.deleted && !version.is_latest) + ); + assert!(versions.versions.iter().all(|version| version.num_versions == 3)); + let identities = versions.versions.iter().map(|version| version.version_id).collect::>(); + assert!(identities.contains(&Some(Uuid::from_u128(1)))); + assert!(identities.contains(&Some(Uuid::from_u128(2)))); + if let Some(expected) = &expected { + assert_eq!(&identities, expected, "equal-time versions must have stable pagination order"); + } else { + expected = Some(identities); + } + } + } + + #[tokio::test] + async fn merge_entry_channels_key_only_candidates_do_not_override_version_metadata() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let marker = test_delete_marker_meta_entry("key", time); + for entries in [ + vec![test_meta_entry("key"), marker.clone()], + vec![marker.clone(), test_meta_entry("key")], + ] { + let mut merged = merge_test_object_entries(entries) + .await + .expect("merge a name with resolved metadata"); + assert!(merged.is_latest_delete_marker()); + assert_eq!(merged.file_info_versions("bucket").expect("decode marker").versions.len(), 1); + } + } + + #[tokio::test] + async fn merge_entry_channels_accepts_equivalent_migrated_coding_and_data_dirs() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 4, 2)]); + let second = rewrite_test_version( + test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 6, 2)]), + |version| version.object.as_mut().expect("object version").data_dir = Some(Uuid::from_u128(42)), + ); + let forward = merge_test_object_entries(vec![first.clone(), second.clone()]) + .await + .expect("valid migration copies"); + let reverse = merge_test_object_entries(vec![second, first]) + .await + .expect("reversed migration copies"); + assert_eq!(forward.metadata, reverse.metadata, "representation must not depend on channel order"); + let versions = forward.file_info_versions("bucket").expect("merged metadata should decode"); + assert_eq!(versions.versions.len(), 1); + assert_eq!(versions.versions[0].metadata.get("etag").map(String::as_str), Some("same-etag")); + } + + #[tokio::test] + async fn merge_entry_channels_rejects_conflicting_version_identity_and_metadata() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "original", 4, 2)]); + for key in [ + "etag", + "x-amz-tagging", + "x-amz-object-lock-mode", + "x-amz-object-lock-retain-until-date", + ] { + let changed = rewrite_test_version(original.clone(), |version| { + version + .object + .as_mut() + .expect("object version") + .meta_user + .insert(key.to_string(), "changed".to_string()); + }); + for pair in [[original.clone(), changed.clone()], [changed, original.clone()]] { + let err = merge_test_object_entries(pair.to_vec()) + .await + .expect_err("conflicting copies must fail"); + assert_eq!(err, StorageError::FileCorrupt, "conflict in {key} must not become arbitrary metadata"); + } + } + let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time), |version| { + version.delete_marker.as_mut().expect("delete marker").version_id = Some(Uuid::from_u128(1)); + }); + assert_eq!( + merge_test_object_entries(vec![original, marker]) + .await + .expect_err("UUID type conflict"), + StorageError::FileCorrupt + ); + } + + #[tokio::test] + async fn merge_entry_channels_reconciles_null_overwrite_without_losing_uuid_history() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let history = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]); + let old_null = rewrite_test_version(history.clone(), |version| { + version.object.as_mut().expect("null object").version_id = None; + }); + let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time + time::Duration::seconds(1)), |version| { + version.delete_marker.as_mut().expect("null marker").version_id = Some(Uuid::nil()); + }); + for inputs in [ + vec![old_null.clone(), marker.clone(), history.clone()], + vec![history, marker, old_null], + ] { + let entry = merge_test_object_entries(inputs) + .await + .expect("new null slot should replace old null slot"); + let versions = entry.file_info_versions("bucket").expect("null versions should decode"); + assert_eq!(versions.versions.len(), 2); + assert!(versions.versions[0].deleted && versions.versions[0].is_latest); + assert!(versions.versions[0].version_id.is_none_or(|id| id.is_nil())); + assert_eq!(versions.versions[1].version_id, Some(Uuid::from_u128(1))); + } + } + + #[tokio::test] + async fn merge_entry_channels_rejects_corrupt_version_headers_and_empty_stacks() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "etag", 4, 2)]); + for empty in [false, true] { + let mut corrupt = original.clone(); + let meta = corrupt.cached.as_mut().expect("fixture metadata"); + if empty { + meta.versions.clear(); + } else { + meta.versions[0].header.version_id = Some(Uuid::from_u128(99)); + } + corrupt.metadata = meta.marshal_msg().expect("encode corrupt fixture"); + assert_eq!( + merge_test_object_entries(vec![original.clone(), corrupt]) + .await + .expect_err("corrupt candidate must fail"), + StorageError::FileCorrupt + ); + } + let mut malformed = original.clone(); + malformed.cached = None; + malformed.metadata = vec![0xff]; + assert_eq!( + merge_test_object_entries(vec![original, malformed]) + .await + .expect_err("malformed metadata must fail"), + StorageError::FileCorrupt + ); + } + + #[tokio::test] + async fn merge_entry_channels_does_not_combine_subquorum_markers_across_erasure_sets() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let old = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]); + let marked = test_object_with_delete_marker_meta_entry("key", time, time + time::Duration::seconds(1)); + let mut inputs = Vec::new(); + for marker_copies in [1, 2] { + let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 3, true, 0); + let copies = (0..3) + .map(|index| Some(if index < marker_copies { marked.clone() } else { old.clone() })) + .collect(); + let entry = resolve_listing_entries(MetaCacheEntries(copies), resolver, false) + .expect("each set independently retains its quorum-backed history"); + inputs.push(entry); + } + let merged = merge_test_object_entries(inputs).await.expect("merge resolved histories"); + let versions = merged.file_info_versions("bucket").expect("decode merged history"); + assert_eq!( + versions.versions.len(), + 1, + "three marker copies across two EC domains do not form a quorum" + ); + assert!(!versions.versions[0].deleted); + } + + #[tokio::test] + async fn listing_merge_preserves_error_after_partial_output_without_cancelling_request() { + let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); + let (first_tx, first_rx) = mpsc::channel(2); + let (second_tx, second_rx) = mpsc::channel(1); + first_tx.send(test_meta_entry("a/")).await.expect("queue preceding prefix"); + first_tx + .send(test_object_meta_entry_with_erasure_versions("b", &[(time, "one", 4, 2)])) + .await + .expect("queue first copy"); + second_tx + .send(test_object_meta_entry_with_erasure_versions("b", &[(time, "two", 4, 2)])) + .await + .expect("queue conflicting copy"); + drop(first_tx); + drop(second_tx); + let request = CancellationToken::new(); + let workers = request.child_token(); + let (sender, mut receiver) = mpsc::channel(1); + let task = super::spawn_listing_merge(workers.clone(), vec![first_rx, second_rx], sender); + assert_eq!(receiver.recv().await.expect("preceding result should arrive").name, "a/"); + assert!(receiver.recv().await.is_none()); + assert_eq!( + task.await + .expect("merge task must not panic") + .expect_err("conflict must propagate"), + StorageError::FileCorrupt + ); + assert!(workers.is_cancelled(), "failed merge must stop the disk producers"); + assert!(!request.is_cancelled(), "the API must still observe the merge error"); + } + #[tokio::test] async fn merge_entry_channels_handles_single_channel() { let (tx, rx) = mpsc::channel(4); diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index cec09eb60..a56410bce 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -2013,7 +2013,7 @@ fn effective_object_actual_size(info: &ObjectInfo) -> Option { info.get_actual_size().ok() } -fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool { +pub(super) fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool { is_data_movement_delete_marker(source) && is_data_movement_delete_marker(target) && source.version_id == target.version_id @@ -4791,7 +4791,13 @@ impl ECStore { return Ok(ObjectInfo::default()); } - let gopts = delete_pool_lookup_opts(&opts, true); + let creates_latest_marker = should_create_delete_marker_for_missing_object(&opts); + let mut gopts = delete_pool_lookup_opts(&opts, true); + if creates_latest_marker { + // An unwritable source still owns its current version. Hiding it + // during lookup would turn a rejected write into a new-pool marker. + gopts.skip_rebalancing = false; + } if opts.data_movement { let existing_pool_info = self.get_pool_info_existing_with_opts(bucket, object, &gopts).await; @@ -4917,7 +4923,12 @@ impl ECStore { } // Determine which pool contains it - let (mut pinfo, errs) = match self.get_pool_info_existing_with_opts(bucket, object, &gopts).await { + let existing_pool_info = if creates_latest_marker { + self.get_pool_info_for_delete_marker(bucket, object, &gopts).await + } else { + self.get_pool_info_existing_with_opts(bucket, object, &gopts).await + }; + let (mut pinfo, errs) = match existing_pool_info { Ok(res) => res, Err(err) if is_err_read_quorum(&err) => return Err(StorageError::ErasureWriteQuorum), Err(err) if is_err_object_not_found(&err) && should_create_delete_marker_for_missing_object(&opts) => { @@ -4954,7 +4965,7 @@ impl ECStore { } }; - if pinfo.object_info.delete_marker && opts.version_id.is_none() { + if pinfo.object_info.delete_marker && opts.version_id.is_none() && !creates_latest_marker { pinfo.object_info.name = decode_dir_object(object); return Ok(pinfo.object_info); } @@ -4976,7 +4987,13 @@ impl ECStore { } for pool in self.pools.iter() { + if creates_latest_marker && pool.pool_idx != pinfo.index { + continue; + } if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await { + if creates_latest_marker { + return Err(StorageError::SlowDown); + } continue; } @@ -5001,7 +5018,7 @@ impl ECStore { return Ok(obj); } Err(err) => { - if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { + if creates_latest_marker || (!is_err_object_not_found(&err) && !is_err_version_not_found(&err)) { return Err(err); } } @@ -8291,6 +8308,544 @@ mod tests { } } + async fn multipool_version_test_store(bucket: &str) -> (Vec, Arc) { + let ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let (mut dirs, first_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await; + let (second_dirs, second_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await; + dirs.extend(second_dirs); + let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[first_set, second_set], ctx).await); + crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + store + .handle_make_bucket( + bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("create the versioned bucket in both pools"); + (dirs, store) + } + + #[tokio::test] + async fn multipool_delete_marker_stays_with_existing_versions() { + let bucket = "multipool-marker-routing"; + let object = "history.bin"; + let (_dirs, store) = multipool_version_test_store(bucket).await; + + let mut expected_versions = Vec::new(); + for value in 1..=3_u8 { + let written = store.pools[1] + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![value; 4097]), + &ObjectOptions { + versioned: true, + user_defined: HashMap::from([ + (rustfs_utils::http::AMZ_OBJECT_TAGGING.to_string(), format!("generation={value}")), + ("x-amz-meta-generation".to_string(), value.to_string()), + ]), + ..Default::default() + }, + ) + .await + .expect("write a historical version deterministically to pool 1"); + expected_versions.push(written.version_id.expect("versioned PUT must acknowledge a UUID")); + } + let marker = store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("delete the current version"); + assert!(marker.delete_marker); + let marker_id = marker.version_id.expect("DELETE must acknowledge a marker UUID"); + assert!(!expected_versions.contains(&marker_id)); + + let local = store.pools[1] + .clone() + .inner_list_object_versions(bucket, object, None, None, None, 10) + .await + .expect("list the object-owning pool"); + assert_eq!(local.objects.len(), 4, "the marker must be committed beside the three existing versions"); + assert_eq!(local.objects[0].version_id, Some(marker_id)); + assert!(local.objects[0].delete_marker && local.objects[0].is_latest); + let versions = store + .clone() + .inner_list_object_versions(bucket, object, None, None, None, 10) + .await + .expect("list all pools"); + assert_eq!(versions.objects.len(), 4); + assert_eq!(versions.objects.iter().filter(|version| version.is_latest).count(), 1); + for (index, version_id) in expected_versions.iter().copied().enumerate() { + assert!( + versions + .objects + .iter() + .any(|version| version.version_id == Some(version_id) && !version.is_latest) + ); + let mut reader = store + .handle_get_object_reader( + bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + version_id: Some(version_id.to_string()), + ..Default::default() + }, + ) + .await + .expect("historical version should remain readable"); + let mut payload = Vec::new(); + reader + .stream + .read_to_end(&mut payload) + .await + .expect("read all historical bytes"); + let value = u8::try_from(index + 1).expect("fixture generation fits u8"); + assert_eq!(payload, vec![value; 4097]); + let listed = versions + .objects + .iter() + .find(|version| version.version_id == Some(version_id)) + .expect("listed historical version"); + assert_eq!(listed.user_tags.as_str(), format!("generation={value}")); + assert_eq!(listed.user_defined.get("x-amz-meta-generation"), Some(&value.to_string())); + } + + let repeated = store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("repeat simple DELETE"); + assert!(repeated.delete_marker); + assert_ne!( + repeated.version_id, + Some(marker_id), + "each enabled-versioning DELETE creates a new marker" + ); + store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + version_id: repeated.version_id.map(|id| id.to_string()), + ..Default::default() + }, + ) + .await + .expect("remove the newest marker by identity"); + store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + version_id: Some(marker_id.to_string()), + ..Default::default() + }, + ) + .await + .expect("remove the original marker by identity"); + let current = store + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("previous version becomes current"); + assert_eq!(current.version_id, expected_versions.last().copied()); + assert!(!current.delete_marker); + } + + #[tokio::test] + async fn multipool_existing_split_history_lists_and_paginates_without_metadata_writes() { + for marker_pool in 0..2 { + let bucket = format!("multipool-split-history-{marker_pool}"); + let object = "history.bin"; + let (dirs, store) = multipool_version_test_store(&bucket).await; + let mut expected = Vec::new(); + for value in 1..=3_u8 { + let version = store.pools[1 - marker_pool] + .put_object( + &bucket, + object, + &mut PutObjReader::from_vec(vec![value; 4097]), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed history through the owning pool's normal write path"); + expected.push(version.version_id); + } + let marker = store.pools[marker_pool] + .delete_object( + &bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("reproduce the previously committed split marker with a normal pool DELETE"); + assert!(marker.delete_marker); + expected.push(marker.version_id); + expected.reverse(); + let mut before = Vec::new(); + for dir in &dirs { + before.push( + tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta")) + .await + .expect("snapshot persisted version metadata"), + ); + } + for max_keys in [1, 2, 4, 10] { + let mut key_marker = None; + let mut version_marker = None; + let mut listed = Vec::new(); + let mut completed = false; + for _ in 0..6 { + let page = store + .clone() + .inner_list_object_versions(&bucket, object, key_marker.clone(), version_marker.clone(), None, max_keys) + .await + .expect("read a complete merged version page"); + listed.extend(page.objects); + if !page.is_truncated { + completed = true; + break; + } + assert_ne!( + (&page.next_marker, &page.next_version_idmarker), + (&key_marker, &version_marker), + "version cursor must advance" + ); + key_marker = page.next_marker; + version_marker = page.next_version_idmarker; + } + assert!(completed, "pagination must terminate"); + assert_eq!(listed.iter().map(|version| version.version_id).collect::>(), expected); + assert!(listed[0].delete_marker && listed[0].is_latest); + assert!(listed[1..].iter().all(|version| !version.delete_marker && !version.is_latest)); + } + let visible = store + .clone() + .list_objects_generic(&bucket, "", None, None, 10, false) + .await + .expect("list current objects"); + assert!(visible.objects.is_empty(), "the global current marker hides the object"); + for (dir, before) in dirs.iter().zip(before) { + assert_eq!( + tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta")) + .await + .expect("read unchanged metadata"), + before + ); + } + } + } + + #[tokio::test] + async fn multipool_conflicting_version_metadata_fails_the_listing_request() { + let bucket = "multipool-version-conflict"; + let (_dirs, store) = multipool_version_test_store(bucket).await; + store.pools[0] + .put_object( + bucket, + "a.bin", + &mut PutObjReader::from_vec(b"preceding result".to_vec()), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed an entry before the conflict"); + let version_id = Uuid::new_v4(); + let mod_time = OffsetDateTime::now_utc(); + let mut etags = Vec::new(); + for (pool_idx, value) in [(0, 1), (1, 2)] { + let written = store.pools[pool_idx] + .put_object( + bucket, + "z.bin", + &mut PutObjReader::from_vec(vec![value; 4097]), + &ObjectOptions { + versioned: true, + version_id: Some(version_id.to_string()), + mod_time: Some(mod_time), + ..Default::default() + }, + ) + .await + .expect("persist independent conflicting copies"); + assert_eq!(written.version_id, Some(version_id)); + etags.push(written.etag); + } + assert_ne!(etags[0], etags[1], "fixture must contain a semantic conflict"); + let err = store + .clone() + .inner_list_object_versions(bucket, "", None, None, None, 10) + .await + .expect_err("partial output must not hide the merge error"); + assert_eq!(err, StorageError::FileCorrupt); + + let cancellation = tokio_util::sync::CancellationToken::new(); + let (sender, mut receiver) = tokio::sync::mpsc::channel(1); + let walk = store + .clone() + .walk(cancellation.clone(), bucket, "", sender, WalkOptions::default()); + let drain = async { while receiver.recv().await.is_some() {} }; + let (walk_result, ()) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(walk, drain) }) + .await + .expect("bounded walk output must drain and terminate on a merge error"); + assert_eq!( + walk_result.expect_err("walk must report the same metadata conflict"), + StorageError::FileCorrupt + ); + assert!(!cancellation.is_cancelled(), "worker failure must not cancel the caller's request"); + } + + #[tokio::test] + async fn multipool_marker_rejects_unwritable_owner_without_falling_back() { + let bucket = "multipool-unwritable-owner"; + let object = "history.bin"; + let (_dirs, store) = multipool_version_test_store(bucket).await; + store.pools[1] + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![1; 4097]), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed the nonzero owner"); + *store.pool_meta.write().await = PoolMeta { + pools: vec![prepared_pool_test_status(0, false), prepared_pool_test_status(1, true)], + ..Default::default() + }; + let err = store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect_err("suspended owner must reject marker creation"); + assert_eq!(err, StorageError::SlowDown); + *store.pool_meta.write().await = PoolMeta::default(); + + let mut rebalancing = crate::services::rebalance::RebalanceStats::default(); + rebalancing.participating = true; + rebalancing.info.status = crate::services::rebalance::RebalStatus::Started; + *store.rebalance_meta.write().await = Some(crate::services::rebalance::RebalanceMeta { + pool_stats: vec![crate::services::rebalance::RebalanceStats::default(), rebalancing], + ..Default::default() + }); + let error = store + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await; + *store.rebalance_meta.write().await = None; + assert_eq!( + error.expect_err("rebalance must not hide the owner during marker lookup"), + StorageError::SlowDown + ); + + // Isolate object quorum from the bucket metadata preflight: disabling + // a whole pool can otherwise fail before object ownership is looked up. + let quorum_bucket = RUSTFS_META_BUCKET; + store.pools[1] + .put_object( + quorum_bucket, + object, + &mut PutObjReader::from_vec(vec![1; 4097]), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed the object-quorum fixture"); + for pool in &store.pools { + pool.put_object( + quorum_bucket, + "split.bin", + &mut PutObjReader::from_vec(vec![2; 4097]), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed another key with history in both pools"); + } + + let owner = &store.pools[1].disk_set[0]; + let healthy = owner.disks.read().await.clone(); + for disk in owner.disks.write().await.iter_mut().skip(1) { + *disk = None; + } + let error = store + .delete_object( + quorum_bucket, + object, + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await; + let split_error = store + .delete_object( + quorum_bucket, + "split.bin", + ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await; + *owner.disks.write().await = healthy; + assert_eq!( + error.expect_err("subquorum owner must not become a new-pool marker"), + StorageError::ErasureWriteQuorum + ); + assert_eq!( + split_error.expect_err("a readable older pool does not prove the global current version"), + StorageError::ErasureWriteQuorum + ); + let split = store.pools[0] + .clone() + .inner_list_object_versions(quorum_bucket, "split.bin", None, None, None, 10) + .await + .expect("inspect the readable older pool"); + assert_eq!(split.objects.len(), 1); + assert!(!split.objects[0].delete_marker); + let other = store.pools[0] + .clone() + .inner_list_object_versions(quorum_bucket, object, None, None, None, 10) + .await + .expect("inspect the other pool"); + assert!(other.objects.is_empty()); + let history = store.pools[1] + .clone() + .inner_list_object_versions(quorum_bucket, object, None, None, None, 10) + .await + .expect("inspect the restored owner"); + assert_eq!(history.objects.len(), 1); + assert!(!history.objects[0].delete_marker); + } + + #[tokio::test] + async fn multipool_suspended_and_batch_markers_keep_the_null_slot_on_the_owner() { + let bucket = "multipool-suspended-markers"; + let object = "history.bin"; + let (_dirs, store) = multipool_version_test_store(bucket).await; + let original = store.pools[1] + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![1; 4097]), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("seed the UUID history in pool 1"); + store + .update_bucket_metadata_config( + bucket, + crate::bucket::metadata::BUCKET_VERSIONING_CONFIG, + b"Suspended".to_vec(), + ) + .await + .expect("persist suspended bucket versioning"); + let null = store + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(vec![2; 4097]), + &ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + .expect("write the null version beside the history"); + assert!(null.version_id.is_none_or(|id| id.is_nil())); + for _ in 0..2 { + let marker = store + .delete_object( + bucket, + object, + ObjectOptions { + version_suspended: true, + ..Default::default() + }, + ) + .await + .expect("replace the null slot with a marker"); + assert!(marker.delete_marker); + assert!(marker.version_id.is_none_or(|id| id.is_nil())); + } + let (deleted, errors) = store + .delete_objects( + bucket, + vec![ObjectToDelete { + object_name: object.to_string(), + ..Default::default() + }], + ObjectOptions::default(), + ) + .await; + assert!(errors.iter().all(Option::is_none), "batch DELETE should succeed: {errors:?}"); + assert_eq!(deleted.len(), 1); + assert!(deleted[0].delete_marker); + let versions = store.pools[1] + .clone() + .inner_list_object_versions(bucket, object, None, None, None, 10) + .await + .expect("inspect the owner after batch DELETE"); + assert_eq!(versions.objects.len(), 2, "only one null marker and the UUID history remain"); + assert!(versions.objects[0].delete_marker && versions.objects[0].is_latest); + assert_eq!(versions.objects[1].version_id, original.version_id); + let other = store.pools[0] + .clone() + .inner_list_object_versions(bucket, object, None, None, None, 10) + .await + .expect("inspect the unused pool"); + assert!(other.objects.is_empty()); + } + async fn assert_prepared_reader_blocks_writer(store: &ECStore, bucket: &str, object: &str) { assert_pool_writer_is_blocked(store, 0, bucket, object).await; } diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 7ee4f7541..ac313f2c0 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -611,7 +611,18 @@ impl ECStore { object: &str, opts: &ObjectOptions, ) -> Result<(PoolObjInfo, Vec)> { - self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await + self.internal_get_pool_info_existing_with_opts(bucket, object, opts, false) + .await + } + + pub(super) async fn get_pool_info_for_delete_marker( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<(PoolObjInfo, Vec)> { + self.internal_get_pool_info_existing_with_opts(bucket, object, opts, true) + .await } async fn internal_get_pool_info_existing_with_opts( @@ -619,6 +630,7 @@ impl ECStore { bucket: &str, object: &str, opts: &ObjectOptions, + require_all_pool_reads: bool, ) -> Result<(PoolObjInfo, Vec)> { let mut futures = Vec::new(); for pool in self.pools.iter() { @@ -647,6 +659,15 @@ impl ECStore { }); } Err(e) => { + // A readable older pool cannot prove ownership of the + // current version while another pool is unreadable. Check + // both raw and object-scoped quorum errors before sorting. + if require_all_pool_reads && !is_err_object_not_found(&e) && !is_err_version_not_found(&e) { + return Err(match e { + Error::ErasureReadQuorum | Error::InsufficientReadQuorum(_, _) => Error::ErasureWriteQuorum, + err => err, + }); + } ress.push(PoolObjInfo { index, err: Some(e),