From 09c29dc5e39baebab5dfccf2c086d4c0730d7d9d Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 6 Sep 2026 16:05:24 +0800 Subject: [PATCH] fix(ecstore): retain namespace ownership during multipart commit --- crates/ecstore/src/set_disk/ops/multipart.rs | 304 ++++++++++++++++++- 1 file changed, 302 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 4aefdb04c..eca81e225 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -3301,13 +3301,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. Self::assign_rename_data_indexes(&mut parts_metadatas); - let mut rename_result = SetDisks::rename_data_owned( + // Disk deadlines can expire before physical publication or failure undo drains. + let mut rename_result = SetDisks::rename_data_owned_with_fence( &commit_disks, (RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path), parts_metadatas, (&commit_bucket, &commit_object), - write_quorum, commit_allows_early_ack, + crate::set_disk::core::io_primitives::RenameDataFenceOptions::new(write_quorum, None) + .with_namespace_commit_guard( + (!crate::bucket::utils::is_meta_bucketname(&commit_bucket)) + .then(|| commit_set.ctx.begin_namespace_commit()), + ), ) .await; if let Ok(rename_commit) = rename_result.as_mut() { @@ -6758,6 +6763,301 @@ mod tests { .await; } + #[tokio::test] + #[serial(capacity_dirty_scope)] + async fn complete_multipart_advances_namespace_generation_after_commit() { + let (dirs, disks, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-namespace-commit"; + let object = "completed-object"; + let body = vec![0x65; 4096]; + make_bucket_on_all(&disks, bucket).await; + let before = set_disks.ctx.namespace_commit_generation(); + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &body, &ObjectOptions::default()).await; + assert_eq!(set_disks.ctx.namespace_commit_generation(), before, "staging is not publication"); + assert!(!set_disks.ctx.namespace_commits_pending()); + tokio::time::timeout( + Duration::from_secs(10), + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &ObjectOptions::default()), + ) + .await + .expect("completion must finish") + .expect("the four real shards must commit"); + let mut reader = tokio::time::timeout( + Duration::from_secs(5), + set_disks.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()), + ) + .await + .expect("GET after completion must finish") + .expect("a successful completion must be immediately readable"); + let mut actual = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), reader.stream.read_to_end(&mut actual)) + .await + .expect("the completed body stream must finish") + .expect("read completed body"); + assert_eq!(actual, body); + let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + for dir in &dirs { + assert!(!dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path).exists()); + } + assert!(matches!( + set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await, + Err(StorageError::InvalidUploadID(..)) + )); + assert!(!set_disks.ctx.namespace_commits_pending()); + assert_eq!( + set_disks.ctx.namespace_commit_generation(), + before + 2, + "one completed MPU must invalidate snapshots at namespace admission and physical retirement" + ); + } + + #[cfg(not(windows))] + async fn assert_complete_multipart_physical_namespace_owner(undo: bool) { + use crate::disk::os::{self, prepared_publication_test_hooks as hooks}; + use crate::set_disk::core::io_primitives::rename_fault_injection; + use futures::FutureExt; + + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async { + let (dirs, disks, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-physical-namespace"; + let object = if undo { "undo-tail" } else { "publication-tail" }; + let old_body = vec![0x41; 1024]; + let new_body = vec![0x62; 4096]; + make_bucket_on_all(&disks, bucket).await; + let old = set_disks + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(old_body.clone()), + &ObjectOptions { + write_completion: crate::object_api::WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("seed a real readable old version"); + let old_etag = old.etag.expect("the committed old object must have an ETag"); + let before = set_disks.ctx.namespace_commit_generation(); + assert!(!set_disks.ctx.namespace_commits_pending()); + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &new_body, &ObjectOptions::default()).await; + let new_etag = get_complete_multipart_md5(&parts); + let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + for dir in &dirs { + assert!(dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path).exists()); + } + assert_eq!(set_disks.ctx.namespace_commit_generation(), before); + let _fault = undo.then(|| rename_fault_injection::fail_rename_on(object, &[2, 3])); + let stage = if undo { + hooks::Stage::Rename + } else { + hooks::Stage::PreparedRename + }; + let (entered_tx, mut entered_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut hooks = Vec::new(); + let mut releases = Vec::new(); + let mut mutation_paths = Vec::new(); + for (index, disk) in disks.iter().enumerate() { + let disk::Disk::Local(local) = disk.as_ref() else { + panic!("physical MPU fixture requires local disks"); + }; + let destination = local + .get_disk() + .get_object_path_for_io(bucket, object) + .expect("leased IO path"); + let entered_tx = entered_tx.clone(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + hooks.push(hooks::install_at(stage, &destination.join(STORAGE_FORMAT_FILE), move || { + let _ = entered_tx.send(index); + // Dropping senders releases every syscall on assertion failure, too. + let _ = release_rx.recv(); + })); + releases.push(release_tx); + // Canonical rename serializes the object directory; backup restore + // serializes its xl.meta destination. Drain the actual executor key. + mutation_paths.push(if undo { + destination.join(STORAGE_FORMAT_FILE) + } else { + destination + }); + } + drop(entered_tx); + let complete_set = set_disks.clone(); + let complete_upload = upload_id.clone(); + let mut complete = tokio::spawn(async move { + complete_set + .complete_multipart_upload(bucket, object, &complete_upload, parts, &ObjectOptions::default()) + .await + }); + let mut complete_joined = false; + let mut observed_counts = None; + let observations = std::panic::AssertUnwindSafe(async { + let expected_publishers = if undo { 2 } else { 4 }; + let entered = tokio::time::timeout(Duration::from_secs(10), async { + let mut entered = HashSet::new(); + while entered.len() < expected_publishers { + tokio::select! { + index = entered_rx.recv() => { + assert!(entered.insert(index.expect("physical publisher must signal entry"))); + } + result = &mut complete => { + complete_joined = true; + panic!("completion returned before physical entry: {result:?}"); + } + } + } + entered + }) + .await + .expect("all expected physical metadata operations must enter"); + let pending_at_entry = set_disks.ctx.namespace_commits_pending(); + let generation_at_entry = set_disks.ctx.namespace_commit_generation(); + for &index in &entered { + let metadata = tokio::time::timeout( + Duration::from_secs(5), + disks[index].read_version("", bucket, object, "", &ReadOptions::default()), + ) + .await + .expect("metadata observation must finish while publication is paused") + .expect("metadata before the paused physical action must be readable"); + assert_eq!( + metadata.metadata.get("etag"), + Some(if undo { &new_etag } else { &old_etag }), + "undo must follow actual publication; prepared rename must precede publication" + ); + } + + // The entry signals run inside the real blocking closures, after each + // wrapper installed its normal deadline. No quota/external guard disables it. + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let joined = tokio::time::timeout(Duration::from_secs(5), &mut complete).await; + complete_joined = joined.is_ok(); + let result = joined + .expect("ordinary MPU disk/undo deadlines must still return before physical drain") + .expect("completion task must not panic"); + assert!(result.is_err(), "a timed-out or two-shard commit cannot acknowledge success"); + let pending_after_timeout = set_disks.ctx.namespace_commits_pending(); + let generation_after_timeout = set_disks.ctx.namespace_commit_generation(); + observed_counts = Some((pending_at_entry, generation_at_entry, pending_after_timeout, generation_after_timeout)); + for &index in &entered { + assert!( + os::acquire_rename_data_mutation_lease(&disks[index].path(), bucket, &mutation_paths[index]) + .now_or_never() + .is_none(), + "timed-out physical work must still own its object serialization" + ); + } + for dir in &dirs { + assert!( + dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path).exists(), + "failed completion must not clean the upload staging" + ); + } + }) + .catch_unwind() + .await; + + // Release even after a failed observation, then finish dispatch before + // draining every physical key. No per-disk assertion may skip a later drain. + drop(releases); + drop(hooks); + let coordinator_drained = complete_joined + || tokio::time::timeout(Duration::from_secs(10), &mut complete).await.is_ok(); + if !coordinator_drained { + complete.abort(); + let _ = tokio::time::timeout(Duration::from_secs(5), &mut complete).await; + } + let drains = futures::future::join_all(disks.iter().zip(&mutation_paths).map(|(disk, destination)| async move { + tokio::time::timeout( + Duration::from_secs(5), + os::acquire_rename_data_mutation_lease(&disk.path(), bucket, destination), + ) + .await + .map(drop) + })) + .await; + let owner_drained = tokio::time::timeout(Duration::from_secs(5), async { + while set_disks.ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await; + let physical_drained = drains.iter().all(|drain| drain.is_ok()); + if !coordinator_drained || !physical_drained || owner_drained.is_err() { + // A bounded cleanup failure cannot justify deleting roots that a + // detached executor might still use. Keep them for diagnosis. + let retained = dirs.into_iter().map(TempDir::keep).collect::>(); + eprintln!("MPU cleanup incomplete: coordinator={coordinator_drained}, physical={physical_drained}, retained={retained:?}"); + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + panic!("MPU cleanup did not drain: coordinator={coordinator_drained}, physical={physical_drained}, retained={retained:?}"); + } + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + // Preserve the original drain checks after collecting every result. + for drained in drains { + drained.expect("released physical MPU work must drain"); + } + owner_drained.expect("physical retirement must finish its namespace counter decrement"); + for disk in &disks { + let metadata = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("all disks must expose the expected final metadata"); + assert_eq!(metadata.metadata.get("etag"), Some(if undo { &old_etag } else { &new_etag })); + } + let (pending_at_entry, generation_at_entry, pending_after_timeout, generation_after_timeout) = + observed_counts.expect("successful observations must record namespace counters"); + let mut reader = tokio::time::timeout( + Duration::from_secs(5), + set_disks.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()), + ) + .await + .expect("GET after physical drain must finish") + .expect("the real final object must be readable"); + let mut actual = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), reader.stream.read_to_end(&mut actual)) + .await + .expect("the final object stream must finish") + .expect("read final object bytes"); + assert_eq!(actual, if undo { old_body } else { new_body }); + assert!( + pending_at_entry && pending_after_timeout, + "physical MPU work outlived namespace accounting: undo={undo}" + ); + assert_eq!(generation_at_entry, before + 1); + assert_eq!( + generation_after_timeout, generation_at_entry, + "the blocked physical owner cannot retire early" + ); + assert_eq!(set_disks.ctx.namespace_commit_generation(), before + 2); + assert!(!set_disks.ctx.namespace_commits_pending()); + + }) + .await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial(capacity_dirty_scope)] + async fn complete_multipart_timeout_keeps_namespace_owner_until_physical_publication() { + assert_complete_multipart_physical_namespace_owner(false).await; + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial(capacity_dirty_scope)] + async fn complete_multipart_failed_quorum_keeps_namespace_owner_until_physical_undo() { + assert_complete_multipart_physical_namespace_owner(true).await; + } + #[tokio::test(flavor = "multi_thread")] #[serial] async fn complete_multipart_releases_disk_snapshot_before_cleanup() {